drl/peer_comm.c

   1 /* See the DRL-LICENSE file for this file's software license. */
   2
   3 #define _XOPEN_SOURCE 600
   4
   5 /* Debug output. */
   6 #include <stdio.h>
   7 #include <stdlib.h>
   8
   9 /* Socket functions. */
  10 #include <sys/types.h>
  11 #include <sys/socket.h>
  12
  13 /* Byte ordering and address structures. */
  14 #include <arpa/inet.h>
  15
  16 /* memset() */
  17 #include <string.h>
  18
  19 /* close() & usleep */
  20 #include <unistd.h>
  21
  22 /* Mutex lock/unlock. */
  23 #include <pthread.h>
  24
  25 /* perror() */
  26 #include <errno.h>
  27
  28 /* select() w/ timeout */
  29 #include <sys/select.h>
  30 #include <sys/time.h>
  31
  32 /* assert() */
  33 #include <assert.h>
  34
  35 /* sigaddset(), sigemptyset(), SIGHUP, etc. */
  36 #include <signal.h>
  37
  38 /* DRL data structures. */
  39 #include "raterouter.h"
  40 #include "ratetypes.h"
  41 #include "drl_state.h"
  42 #include "peer_comm.h"
  43 #include "logging.h"
  44
  45 /* Artifically makes a network partition. */
  46 int do_partition = 0;
  47 int partition_set = 0xfffffff;
  48
  49 extern limiter_t limiter;
  50
  51 static const uint32_t MAGIC_MSG = 0x123123;
  52 static const uint32_t MAGIC_HELLO = 0x456456;
  53 static const uint16_t MSG = 1;
  54 static const uint16_t ACK = 2;
  55
  56 static void message_to_hbo(message_t *msg) {
  57     msg->magic = ntohl(msg->magic);
  58     msg->ident_id = ntohl(msg->ident_id);
  59     msg->seqno = ntohl(msg->seqno);
  60     msg->min_seqno = ntohl(msg->min_seqno);
  61     msg->type = ntohs(msg->type);
  62     /* value is a double */
  63     /* weight is a double */
  64 }
  65
  66 static void message_to_nbo(message_t *msg) {
  67     msg->magic = htonl(msg->magic);
  68     msg->ident_id = htonl(msg->ident_id);
  69     msg->seqno = htonl(msg->seqno);
  70     msg->min_seqno = htonl(msg->min_seqno);
  71     msg->type = htons(msg->type);
  72     /* value is a double */
  73     /* weight is a double */
  74 }
  75
  76 static void hello_to_hbo(hello_t *hello) {
  77     hello->magic = ntohl(hello->magic);
  78     hello->ident_id = ntohl(hello->ident_id);
  79     hello->port = ntohs(hello->port);
  80 }
  81
  82 static void hello_to_nbo(hello_t *hello) {
  83     hello->magic = htonl(hello->magic);
  84     hello->ident_id = htonl(hello->ident_id);
  85     hello->port = htons(hello->port);
  86 }
  87
  88 static int is_connected(remote_limiter_t *remote) {
  89     struct sockaddr_in addr;
  90     socklen_t addrlen = sizeof(addr);
  91
  92     if (getpeername(remote->socket, (struct sockaddr *) &addr, &addrlen) == 0)
  93         return 1;
  94     else
  95         return 0;
  96 }
  97
  98 static int send_ack(identity_t *ident, remote_limiter_t *remote, uint32_t seqno) {
  99     int result = 0;
 100     message_t msg;
 101     struct sockaddr_in toaddr;
 102
 103     memset(&toaddr, 0, sizeof(struct sockaddr_in));
 104     toaddr.sin_family = AF_INET;
 105
 106     toaddr.sin_addr.s_addr = remote->addr; /* Already in network byte order. */
 107     toaddr.sin_port = remote->port;
 108
 109     memset(&msg, 0, sizeof(msg));
 110     msg.magic = MAGIC_MSG;
 111     msg.ident_id = ident->id;
 112     msg.type = ACK;
 113     msg.seqno = seqno;
 114
 115     message_to_nbo(&msg);
 116
 117     if (sendto(limiter.udp_socket, &msg, sizeof(msg), 0, (struct sockaddr *) &toaddr, sizeof(struct sockaddr_in)) < 0) {
 118         printlog(LOG_WARN, "send_ack: sento failed.\n");
 119         result = errno;
 120     }
 121
 122     return result;
 123 }
 124
 125 void limiter_receive() {
 126     struct sockaddr_in fromaddr;
 127     remote_node_t sender;
 128     socklen_t fromlen = sizeof(fromaddr);
 129     identity_t *ident = NULL;
 130     remote_limiter_t *remote = NULL;
 131     message_t msg;
 132
 133     if (recvfrom(limiter.udp_socket, &msg, sizeof(msg), MSG_WAITALL, (struct sockaddr *) &fromaddr, (socklen_t *) &fromlen) != sizeof(msg)) {
 134         /* recv failed.  Log and continue. */
 135         printlog(LOG_WARN, "recv failed to read full message.\n");
 136         return;
 137     }
 138     memset(&sender, 0, sizeof(remote_node_t));
 139     sender.addr = fromaddr.sin_addr.s_addr;
 140     sender.port = fromaddr.sin_port;
 141
 142     message_to_hbo(&msg);
 143
 144     assert(msg.magic == MAGIC_MSG);
 145
 146 #if 0
 147     printlog(LOG_WARN, "Rcvd (value, weight) : (%f, %f) from ident %d (net order host 0x%x port %d) key size(%d)\n",
 148             msg.value, msg.weight, msg.ident_id, sender.addr,sender.port,sizeof(remote_node_t));
 149 #endif
 150     pthread_testcancel();
 151
 152     pthread_rwlock_rdlock(&limiter.limiter_lock);
 153
 154     ident = map_search(limiter.stable_instance.ident_map, &msg.ident_id,
 155                        sizeof(msg.ident_id));
 156
 157     if (ident == NULL) {
 158         printlog(LOG_WARN, "WARN:recvd message for unknown identity.\n");
 159         pthread_rwlock_unlock(&limiter.limiter_lock);
 160         return;
 161     }
 162
 163     pthread_mutex_lock(&ident->comm.lock);
 164
 165     remote = map_search(ident->comm.remote_node_map, &sender, sizeof(remote_node_t));
 166
 167     if (remote == NULL) {
 168         printlog(LOG_WARN, "WARN: recvd msg from unknown entity.\n");
 169         pthread_mutex_unlock(&ident->comm.lock);
 170         pthread_rwlock_unlock(&limiter.limiter_lock);
 171         return;
 172     }
 173
 174     switch (ident->comm.comm_fabric) {
 175         case COMM_MESH: {
 176             /* Use the message's value to be our new GRDrate/FPSweight for the
 177              * message's sender. */
 178             remote->rate = msg.value;
 179
 180             /* Reset the AWOL counter to zero since we received an update. */
 181             remote->awol = 0;
 182         }
 183         break;
 184
 185         case COMM_GOSSIP: {
 186             if (msg.type == ACK) {
 187                 if (msg.seqno == remote->outgoing.next_seqno - 1) {
 188                     int i;
 189
 190                     /* Ack for most recent message.  Clear saved state. */
 191                     remote->outgoing.first_seqno = remote->outgoing.next_seqno;
 192                     remote->outgoing.saved_value = 0;
 193                     remote->outgoing.saved_weight = 0;
 194
 195                     for (i = 0; i < ident->comm.gossip.gossip_branch; ++i) {
 196                         if (ident->comm.retrys[i] >= 0 &&
 197                             remote == &ident->comm.remote_limiters[ident->comm.retrys[i]]) {
 198                                 //printf("clearing spot %d, it was %d\n", i, ident->retrys[i]);
 199                                 ident->comm.retrys[i] = -2;
 200                         }
 201                     }
 202                 }
 203                 /* Ignore ack if it isn't for most recent message. */
 204             } else {
 205                 if (msg.min_seqno > remote->incoming.seen_seqno) {
 206                     /* Entirely new information */
 207                     remote->incoming.seen_seqno = msg.seqno;
 208                     remote->incoming.saved_value = msg.value;
 209                     remote->incoming.saved_weight = msg.weight;
 210                     ident->comm.gossip.value += msg.value;
 211                     ident->comm.gossip.weight += msg.weight;
 212                     send_ack(ident, remote, msg.seqno);
 213                 } else if (msg.seqno > remote->incoming.seen_seqno) {
 214                     /* Only some of the message is old news. */
 215                     double diff_value = msg.value - remote->incoming.saved_value;
 216                     double diff_weight = msg.weight - remote->incoming.saved_weight;
 217
 218                     remote->incoming.seen_seqno = msg.seqno;
 219                     remote->incoming.saved_value = msg.value;
 220                     remote->incoming.saved_weight = msg.weight;
 221
 222                     ident->comm.gossip.value += diff_value;
 223                     ident->comm.gossip.weight += diff_weight;
 224                     send_ack(ident, remote, msg.seqno);
 225                 } else {
 226                     /* The entire message is old news. (Duplicate). */
 227                     /* Do nothing. */
 228                 }
 229             }
 230         }
 231         break;
 232
 233         default: {
 234             printlog(LOG_CRITICAL, "ERR: Unknown identity comm fabric.\n");
 235         }
 236     }
 237
 238     pthread_mutex_unlock(&ident->comm.lock);
 239     pthread_rwlock_unlock(&limiter.limiter_lock);
 240 }
 241
 242 #if 0
 243 static void limiter_accept(comm_limiter_t *limiter) {
 244     int sock, result;
 245     struct sockaddr_in fromaddr;
 246     socklen_t fromlen = sizeof(fromaddr);
 247     remote_node_t sender;
 248     remote_limiter_t *remote;
 249     hello_t hello;
 250     comm_ident_t *ident;
 251     ident_handle *handle = NULL;
 252
 253     sock = accept(limiter->tcp_socket, (struct sockaddr *)&fromaddr, &fromlen);
 254
 255     assert(sock > 0);
 256
 257     memset(&hello, 0, sizeof(hello_t));
 258     result = recv(sock, &hello, sizeof(hello_t), 0);
 259
 260     if (result < 0) {
 261         close(sock);
 262         return; /* Failure - ignore it. */
 263     }
 264
 265     assert(result == sizeof(hello_t));
 266
 267     hello_to_hbo(&hello);
 268
 269     assert(hello.magic == MAGIC_HELLO);
 270
 271     memset(&sender, 0, sizeof(remote_node_t));
 272     sender.addr = fromaddr.sin_addr.s_addr;
 273     sender.port = ntohs(hello.port);
 274
 275     pthread_testcancel();
 276
 277     pthread_rwlock_rdlock(&limiter->rwlock);
 278
 279     handle = map_search(limiter->ident_id_to_handle, (void *) &hello.ident_id, sizeof(hello.ident_id));
 280
 281     if (handle == NULL) {
 282         printlog(LOG_WARN, "WARN:recvd hello for unknown identity.\n");
 283         pthread_rwlock_unlock(&limiter->rwlock);
 284         return;
 285     }
 286
 287     ident = limiter->identities[*handle];
 288     assert(ident != NULL);
 289
 290     pthread_mutex_lock(&ident->lock);
 291
 292     remote = map_search(ident->remote_node_map, &sender, sizeof(remote_node_t));
 293
 294     if (remote == NULL) {
 295         printlog(LOG_WARN, "WARN: Accepted connection from unknown identity.\n");
 296         pthread_mutex_unlock(&ident->lock);
 297         pthread_rwlock_unlock(&limiter->rwlock);
 298         close(sock);
 299         return;
 300     }
 301
 302     if (is_connected(remote)) {
 303         /* We are still connected, don't need the new socket. */
 304         close(sock);
 305         pthread_mutex_unlock(&ident->lock);
 306         pthread_rwlock_unlock(&limiter->rwlock);
 307         return;
 308     }
 309
 310     /* We weren't connected, but we are now... */
 311     remote->socket = sock;
 312     printf("Got connection on: %d\n", sock);
 313     FD_SET(sock, &ident->fds);
 314
 315     pthread_mutex_unlock(&ident->lock);
 316     pthread_rwlock_unlock(&limiter->rwlock);
 317 }
 318
 319 static void read_tcp_message(comm_ident_t *ident, pthread_rwlock_t *limiter_rwlock, int sock) {
 320     int result;
 321     message_t msg;
 322
 323     memset(&msg, 0, sizeof(message_t));
 324
 325     result = recv(sock, &msg, sizeof(message_t), 0);
 326
 327     if (result < 0) {
 328         pthread_rwlock_rdlock(limiter_rwlock);
 329         pthread_mutex_lock(&ident->lock);
 330         FD_CLR(sock, &ident->fds);
 331         close(sock);
 332         pthread_mutex_unlock(&ident->lock);
 333         pthread_rwlock_unlock(limiter_rwlock);
 334         return;
 335     }
 336
 337     assert(result == sizeof(message_t));
 338
 339     message_to_hbo(&msg);
 340     assert(msg.magic == MAGIC_MSG);
 341
 342     pthread_rwlock_rdlock(limiter_rwlock);
 343     pthread_mutex_lock(&ident->lock);
 344
 345     switch (ident->comm_fabric) {
 346         case COMM_GOSSIP: {
 347             ident->gossip.value += msg.value;
 348             ident->gossip.weight += msg.weight;
 349         }
 350         break;
 351
 352         default: {
 353             assert(1 == 0); /* This case shouldn't happen. Punt for now... */
 354         }
 355     }
 356     pthread_mutex_unlock(&ident->lock);
 357     pthread_rwlock_unlock(limiter_rwlock);
 358 }
 359
 360 static void ident_receive(comm_ident_t *ident, pthread_rwlock_t *limiter_rwlock) {
 361     int select_result, i;
 362     fd_set fds_copy;
 363     struct timeval timeout;
 364
 365     FD_ZERO(&fds_copy);
 366     timeout.tv_sec = 15;
 367     timeout.tv_usec = 0;
 368
 369     pthread_rwlock_rdlock(limiter_rwlock);
 370     pthread_mutex_lock(&ident->lock);
 371     memcpy(&fds_copy, &ident->fds, sizeof(fd_set));
 372     pthread_mutex_unlock(&ident->lock);
 373     pthread_rwlock_unlock(limiter_rwlock);
 374
 375     /* mask interrupt signals for this thread? */
 376
 377     select_result = select(FD_SETSIZE, &fds_copy, NULL, NULL, &timeout);
 378
 379     assert(select_result >= 0);
 380
 381     if (select_result == 0)
 382         return; /* Timed out */
 383
 384     for (i = 0; (i < FD_SETSIZE) && select_result; ++i) {
 385         if (FD_ISSET(i, &fds_copy)) {
 386             read_tcp_message(ident, limiter_rwlock, i);
 387             select_result--;
 388         }
 389     }
 390 }
 391 #endif
 392
 393 #define ALLOW_PARTITION
 394
 395 int send_udp_mesh(comm_t *comm, uint32_t id, int sock) {
 396     int result = 0;
 397     remote_limiter_t *remote;
 398     message_t msg;
 399     struct sockaddr_in toaddr;
 400     int i;
 401
 402 #ifdef ALLOW_PARTITION
 403
 404     int partition_count = 0;
 405     struct in_addr dest;
 406     char dest_ip[22];
 407
 408 #endif
 409
 410     memset(&toaddr, 0, sizeof(struct sockaddr_in));
 411     toaddr.sin_family = AF_INET;
 412
 413     memset(&msg, 0, sizeof(message_t));
 414     msg.magic = MAGIC_MSG;
 415     msg.ident_id = id;
 416     msg.value = comm->local_rate;
 417     /* Do we want seqnos for mesh?  We can get by without them. */
 418
 419     message_to_nbo(&msg);
 420
 421     /* Iterate though and send update to all remote limiters in our identity. */
 422     for (i = 0; i < comm->remote_node_count; ++i) {
 423         remote = &comm->remote_limiters[i];
 424
 425 #ifdef ALLOW_PARTITION
 426
 427         if (do_partition) {
 428             printlog(LOG_DEBUG, "Testing partition, partition set is %x, count is %d, test is %d.\n",
 429                      partition_set, partition_count, partition_set & (1 << partition_count));
 430             /* If the partition count bit isn't high in the set, don't actually send anything. */
 431             if ((partition_set & (1 << partition_count)) == 0) {
 432                 dest.s_addr = ntohl(remote->addr);
 433                 strcpy(dest_ip, inet_ntoa(dest));
 434
 435                 printlog(LOG_DEBUG, "Partition: ignoring host %s\n", dest_ip);
 436
 437                 partition_count += 1;
 438                 continue;
 439             }
 440         }
 441
 442 #endif
 443
 444         toaddr.sin_addr.s_addr = remote->addr; /* Already in network byte order. */
 445         toaddr.sin_port = remote->port;
 446         if (sendto(sock, &msg, sizeof(msg), 0, (struct sockaddr *) &toaddr, sizeof(struct sockaddr_in)) < 0) {
 447             printlog(LOG_WARN, "WARN: limiter_send_mesh: sento failed.\n");
 448             result = errno;
 449             printlog(LOG_WARN, "  - The error was |%d|\n", strerror(result));
 450             break;
 451         }
 452         partition_count += 1;
 453     }
 454
 455     return result;
 456 }
 457
 458 int send_udp_gossip(comm_t *comm, uint32_t id, int sock) {
 459     int i, j, targetid;
 460     int result = 0;
 461     remote_limiter_t *remote;
 462     struct sockaddr_in toaddr;
 463     double msg_value, msg_weight;
 464
 465     memset(&toaddr, 0, sizeof(struct sockaddr_in));
 466     toaddr.sin_family = AF_INET;
 467
 468     msg_value = comm->gossip.value / (comm->gossip.gossip_branch + 1);
 469     msg_weight = comm->gossip.weight / (comm->gossip.gossip_branch + 1);
 470
 471     for (i = 0; i < comm->gossip.gossip_branch; ++i) {
 472         message_t msg;
 473
 474         printlog(LOG_DEBUG, "Gossip loop iteration, i=%d, branch=%d\n", i, comm->gossip.gossip_branch);
 475
 476         if (comm->retrys[i] >= 0) {
 477             remote = &comm->remote_limiters[comm->retrys[i]];
 478             targetid = comm->retrys[i];
 479             //printf("%d:d:%d, ", i, comm->retrys[i]);
 480         } else {
 481             targetid = -2;
 482
 483             while (targetid == -2) {
 484                 targetid = myrand() % comm->remote_node_count;
 485
 486                 for (j = 0; j < comm->gossip.gossip_branch; ++j) {
 487                     if (targetid == comm->retrys[j]) {
 488                         targetid = -2;
 489                         break;
 490                     }
 491                 }
 492             }
 493
 494             remote = &comm->remote_limiters[targetid];
 495             //printf("%d:r:%d, ", i, targetid);
 496         }
 497
 498         toaddr.sin_addr.s_addr = remote->addr; /* Already in network byte order. */
 499         toaddr.sin_port = remote->port;
 500
 501         memset(&msg, 0, sizeof(message_t));
 502         msg.magic = MAGIC_MSG;
 503         msg.ident_id = id;
 504         msg.value = msg_value + remote->outgoing.saved_value;
 505         msg.weight = msg_weight + remote->outgoing.saved_weight;
 506         msg.seqno = remote->outgoing.next_seqno;
 507         msg.min_seqno = remote->outgoing.first_seqno;
 508         msg.type = MSG;
 509
 510         remote->outgoing.next_seqno++;
 511         remote->outgoing.saved_value += msg_value;
 512         remote->outgoing.saved_weight += msg_weight;
 513
 514         message_to_nbo(&msg);
 515
 516         if (sendto(sock, &msg, sizeof(msg), 0, (struct sockaddr *) &toaddr, sizeof(struct sockaddr_in)) < 0) {
 517             printlog(LOG_WARN, "WARN: limiter_send_gossip: sento failed.\n");
 518             result = errno;
 519             break;
 520         }
 521
 522         comm->retrys[i] = targetid;
 523     }
 524     //printf("\n");
 525
 526     comm->gossip.value = msg_value;
 527     comm->gossip.weight = msg_weight;
 528
 529     return result;
 530 }
 531
 532 #if 0
 533 int send_tcp_gossip(comm_ident_t *ident, FILE *logfile, int unused) {
 534     int i, targetid, sock;
 535     int result = 0;
 536     message_t msg;
 537
 538     memset(&msg, 0, sizeof(message_t));
 539     msg.magic = MAGIC_MSG;
 540     msg.ident_id = ident->ident_id;
 541     msg.value = ident->gossip.value / (ident->gossip.gossip_branch + 1);
 542     msg.weight = ident->gossip.weight / (ident->gossip.gossip_branch + 1);
 543
 544     message_to_nbo(&msg);
 545
 546     for (i = 0; i < ident->gossip.gossip_branch; ++i) {
 547         targetid = myrand() % ident->remote_node_count;
 548         sock = ident->remote_limiters[targetid].socket;
 549
 550         result = send(sock, &msg, sizeof(message_t), 0);
 551         if (result < 0) {
 552             result = errno;
 553             FD_CLR(sock, &ident->fds);
 554             close(sock);
 555             break;
 556         }
 557
 558         assert(result == sizeof(message_t));
 559     }
 560
 561     ident->gossip.value /= (ident->gossip.gossip_branch + 1);
 562     ident->gossip.weight /= (ident->gossip.gossip_branch + 1);
 563
 564     return result;
 565 }
 566 #endif
 567
 568 #if 0
 569 void *limiter_accept_thread(void *limiter) {
 570     sigset_t signal_mask;
 571
 572     sigemptyset(&signal_mask);
 573     sigaddset(&signal_mask, SIGHUP);
 574     pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
 575
 576     pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
 577     pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
 578     while (1) {
 579         limiter_accept((comm_limiter_t *) limiter);
 580     }
 581     pthread_exit(NULL);
 582 }
 583 void *ident_receive_thread(void *recv_args) {
 584     int i, sock, result;
 585     struct recv_thread_args *args = (struct recv_thread_args *) recv_args;
 586     comm_ident_t *ident = args->ident;
 587     pthread_rwlock_t *lock = args->lock;
 588     uint16_t port = args->port;
 589     struct sockaddr_in addr;
 590     socklen_t addrlen = sizeof(addr);
 591     hello_t hello;
 592
 593     free(args);
 594
 595     while (1) {
 596         memset(&hello, 0, sizeof(hello_t));
 597
 598         /*Try to connect to all remote nodes if they aren't already connected.*/
 599         pthread_rwlock_rdlock(lock);
 600         pthread_mutex_lock(&ident->lock);
 601
 602         hello.magic = MAGIC_HELLO;
 603         hello.ident_id = ident->ident_id;
 604         hello.port = ntohs(port);
 605
 606         hello_to_nbo(&hello);
 607
 608         for (i = 0; i < ident->remote_node_count; ++i) {
 609             if (is_connected(&ident->remote_limiters[i]))
 610                 continue; /* Ignore if it's already connected */
 611
 612             sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
 613             if (sock < 0) {
 614                 perror("socket");
 615                 continue;
 616             }
 617
 618             assert(sock >= 0);
 619
 620             memset(&addr, 0, sizeof(struct sockaddr_in));
 621             addr.sin_family = AF_INET;
 622             addr.sin_port = ident->remote_limiters[i].port;
 623             addr.sin_addr.s_addr = ident->remote_limiters[i].addr;
 624
 625             result = connect(sock, (struct sockaddr *) &addr, addrlen);
 626             if (result < 0) {
 627                 close(sock);
 628                 continue;
 629             }
 630
 631             result = send(sock, &hello, sizeof(hello_t), 0);
 632             if (result < 0) {
 633                 close(sock);
 634                 continue;
 635             }
 636
 637             assert(result == sizeof(hello_t));
 638
 639             ident->remote_limiters[i].socket = sock;
 640             printf("Connected on socket: %d\n", sock);
 641             FD_SET(sock, &ident->fds);
 642         }
 643
 644         pthread_rwlock_unlock(lock);
 645         pthread_mutex_unlock(&ident->lock);
 646
 647         ident_receive(ident, lock);
 648     }
 649     pthread_exit(NULL);
 650 }
 651 #endif