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                                 ident->comm.retrys[i] = -2;
 199                         }
 200                     }
 201                 }
 202                 /* Ignore ack if it isn't for most recent message. */
 203             } else {
 204                 if (msg.min_seqno > remote->incoming.seen_seqno) {
 205                     /* Entirely new information */
 206                     remote->incoming.seen_seqno = msg.seqno;
 207                     remote->incoming.saved_value = msg.value;
 208                     remote->incoming.saved_weight = msg.weight;
 209                     ident->comm.gossip.value += msg.value;
 210                     ident->comm.gossip.weight += msg.weight;
 211                     send_ack(ident, remote, msg.seqno);
 212                     remote->awol = 0;
 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                     remote->awol = 0;
 226                 } else {
 227                     /* The entire message is old news. (Duplicate). */
 228                     /* Do nothing. */
 229                 }
 230             }
 231         }
 232         break;
 233
 234         default: {
 235             printlog(LOG_CRITICAL, "ERR: Unknown identity comm fabric.\n");
 236         }
 237     }
 238
 239     pthread_mutex_unlock(&ident->comm.lock);
 240     pthread_rwlock_unlock(&limiter.limiter_lock);
 241 }
 242
 243 #if 0
 244 static void limiter_accept(comm_limiter_t *limiter) {
 245     int sock, result;
 246     struct sockaddr_in fromaddr;
 247     socklen_t fromlen = sizeof(fromaddr);
 248     remote_node_t sender;
 249     remote_limiter_t *remote;
 250     hello_t hello;
 251     comm_ident_t *ident;
 252     ident_handle *handle = NULL;
 253
 254     sock = accept(limiter->tcp_socket, (struct sockaddr *)&fromaddr, &fromlen);
 255
 256     assert(sock > 0);
 257
 258     memset(&hello, 0, sizeof(hello_t));
 259     result = recv(sock, &hello, sizeof(hello_t), 0);
 260
 261     if (result < 0) {
 262         close(sock);
 263         return; /* Failure - ignore it. */
 264     }
 265
 266     assert(result == sizeof(hello_t));
 267
 268     hello_to_hbo(&hello);
 269
 270     assert(hello.magic == MAGIC_HELLO);
 271
 272     memset(&sender, 0, sizeof(remote_node_t));
 273     sender.addr = fromaddr.sin_addr.s_addr;
 274     sender.port = ntohs(hello.port);
 275
 276     pthread_testcancel();
 277
 278     pthread_rwlock_rdlock(&limiter->rwlock);
 279
 280     handle = map_search(limiter->ident_id_to_handle, (void *) &hello.ident_id, sizeof(hello.ident_id));
 281
 282     if (handle == NULL) {
 283         printlog(LOG_WARN, "WARN:recvd hello for unknown identity.\n");
 284         pthread_rwlock_unlock(&limiter->rwlock);
 285         return;
 286     }
 287
 288     ident = limiter->identities[*handle];
 289     assert(ident != NULL);
 290
 291     pthread_mutex_lock(&ident->lock);
 292
 293     remote = map_search(ident->remote_node_map, &sender, sizeof(remote_node_t));
 294
 295     if (remote == NULL) {
 296         printlog(LOG_WARN, "WARN: Accepted connection from unknown identity.\n");
 297         pthread_mutex_unlock(&ident->lock);
 298         pthread_rwlock_unlock(&limiter->rwlock);
 299         close(sock);
 300         return;
 301     }
 302
 303     if (is_connected(remote)) {
 304         /* We are still connected, don't need the new socket. */
 305         close(sock);
 306         pthread_mutex_unlock(&ident->lock);
 307         pthread_rwlock_unlock(&limiter->rwlock);
 308         return;
 309     }
 310
 311     /* We weren't connected, but we are now... */
 312     remote->socket = sock;
 313     printf("Got connection on: %d\n", sock);
 314     FD_SET(sock, &ident->fds);
 315
 316     pthread_mutex_unlock(&ident->lock);
 317     pthread_rwlock_unlock(&limiter->rwlock);
 318 }
 319
 320 static void read_tcp_message(comm_ident_t *ident, pthread_rwlock_t *limiter_rwlock, int sock) {
 321     int result;
 322     message_t msg;
 323
 324     memset(&msg, 0, sizeof(message_t));
 325
 326     result = recv(sock, &msg, sizeof(message_t), 0);
 327
 328     if (result < 0) {
 329         pthread_rwlock_rdlock(limiter_rwlock);
 330         pthread_mutex_lock(&ident->lock);
 331         FD_CLR(sock, &ident->fds);
 332         close(sock);
 333         pthread_mutex_unlock(&ident->lock);
 334         pthread_rwlock_unlock(limiter_rwlock);
 335         return;
 336     }
 337
 338     assert(result == sizeof(message_t));
 339
 340     message_to_hbo(&msg);
 341     assert(msg.magic == MAGIC_MSG);
 342
 343     pthread_rwlock_rdlock(limiter_rwlock);
 344     pthread_mutex_lock(&ident->lock);
 345
 346     switch (ident->comm_fabric) {
 347         case COMM_GOSSIP: {
 348             ident->gossip.value += msg.value;
 349             ident->gossip.weight += msg.weight;
 350         }
 351         break;
 352
 353         default: {
 354             assert(1 == 0); /* This case shouldn't happen. Punt for now... */
 355         }
 356     }
 357     pthread_mutex_unlock(&ident->lock);
 358     pthread_rwlock_unlock(limiter_rwlock);
 359 }
 360
 361 static void ident_receive(comm_ident_t *ident, pthread_rwlock_t *limiter_rwlock) {
 362     int select_result, i;
 363     fd_set fds_copy;
 364     struct timeval timeout;
 365
 366     FD_ZERO(&fds_copy);
 367     timeout.tv_sec = 15;
 368     timeout.tv_usec = 0;
 369
 370     pthread_rwlock_rdlock(limiter_rwlock);
 371     pthread_mutex_lock(&ident->lock);
 372     memcpy(&fds_copy, &ident->fds, sizeof(fd_set));
 373     pthread_mutex_unlock(&ident->lock);
 374     pthread_rwlock_unlock(limiter_rwlock);
 375
 376     /* mask interrupt signals for this thread? */
 377
 378     select_result = select(FD_SETSIZE, &fds_copy, NULL, NULL, &timeout);
 379
 380     assert(select_result >= 0);
 381
 382     if (select_result == 0)
 383         return; /* Timed out */
 384
 385     for (i = 0; (i < FD_SETSIZE) && select_result; ++i) {
 386         if (FD_ISSET(i, &fds_copy)) {
 387             read_tcp_message(ident, limiter_rwlock, i);
 388             select_result--;
 389         }
 390     }
 391 }
 392 #endif
 393
 394 #define ALLOW_PARTITION
 395
 396 int send_udp_mesh(comm_t *comm, uint32_t id, int sock) {
 397     int result = 0;
 398     remote_limiter_t *remote;
 399     message_t msg;
 400     struct sockaddr_in toaddr;
 401     int i;
 402
 403 #ifdef ALLOW_PARTITION
 404
 405     int partition_count = 0;
 406     struct in_addr dest;
 407     char dest_ip[22];
 408
 409 #endif
 410
 411     memset(&toaddr, 0, sizeof(struct sockaddr_in));
 412     toaddr.sin_family = AF_INET;
 413
 414     memset(&msg, 0, sizeof(message_t));
 415     msg.magic = MAGIC_MSG;
 416     msg.ident_id = id;
 417     msg.value = comm->local_rate;
 418     /* Do we want seqnos for mesh?  We can get by without them. */
 419
 420     message_to_nbo(&msg);
 421
 422     /* Iterate though and send update to all remote limiters in our identity. */
 423     for (i = 0; i < comm->remote_node_count; ++i) {
 424         remote = &comm->remote_limiters[i];
 425
 426 #ifdef ALLOW_PARTITION
 427
 428         if (do_partition) {
 429             printlog(LOG_DEBUG, "Testing partition, partition set is %x, count is %d, test is %d.\n",
 430                      partition_set, partition_count, partition_set & (1 << partition_count));
 431             /* If the partition count bit isn't high in the set, don't actually send anything. */
 432             if ((partition_set & (1 << partition_count)) == 0) {
 433                 dest.s_addr = ntohl(remote->addr);
 434                 strcpy(dest_ip, inet_ntoa(dest));
 435
 436                 printlog(LOG_DEBUG, "Partition: ignoring host %s\n", dest_ip);
 437
 438                 partition_count += 1;
 439                 continue;
 440             }
 441         }
 442
 443 #endif
 444
 445         toaddr.sin_addr.s_addr = remote->addr; /* Already in network byte order. */
 446         toaddr.sin_port = remote->port;
 447         if (sendto(sock, &msg, sizeof(msg), 0, (struct sockaddr *) &toaddr, sizeof(struct sockaddr_in)) < 0) {
 448             printlog(LOG_WARN, "WARN: limiter_send_mesh: sento failed.\n");
 449             result = errno;
 450             printlog(LOG_WARN, "  - The error was |%d|\n", strerror(result));
 451             break;
 452         }
 453         partition_count += 1;
 454     }
 455
 456     return result;
 457 }
 458
 459 int send_udp_gossip(comm_t *comm, uint32_t id, int sock) {
 460     int i, j, targetid;
 461     int awol_threshold = GOSSIP_REMOTE_AWOL_THRESHOLD;
 462     int rand_count; //HACK...
 463     int result = 0;
 464     remote_limiter_t *remote;
 465     struct sockaddr_in toaddr;
 466     double msg_value, msg_weight;
 467
 468     /* This is the factor for the portion of value/weight to keep locally.
 469      * Normally this is 1, meaning that we retain the same amount of value/weight
 470      * that was sent to the peers.  In the case of not being able to send to a
 471      * peer though, we increment this to reclaim the value/weight locally. */
 472     int message_portion = 1;
 473
 474     memset(&toaddr, 0, sizeof(struct sockaddr_in));
 475     toaddr.sin_family = AF_INET;
 476
 477     msg_value = comm->gossip.value / (comm->gossip.gossip_branch + 1);
 478     msg_weight = comm->gossip.weight / (comm->gossip.gossip_branch + 1);
 479
 480     for (i = 0; i < comm->gossip.gossip_branch; ++i) {
 481         message_t msg;
 482
 483         printlog(LOG_DEBUG, "Gossip loop iteration, i=%d, branch=%d\n", i, comm->gossip.gossip_branch);
 484
 485         if (comm->retrys[i] >= 0) {
 486             remote = &comm->remote_limiters[comm->retrys[i]];
 487             targetid = comm->retrys[i];
 488
 489             if (remote->awol > awol_threshold) {
 490                 message_portion += 1;
 491                 printlog(LOG_DEBUG, "Gossip: Ignoring AWOL peer id %d.\n", comm->retrys[i]);
 492                 comm->retrys[i] = -1;
 493                 continue;
 494             }
 495         } else {
 496             targetid = -2;
 497             rand_count = 0;
 498
 499             while (targetid == -2 && rand_count < 50) {
 500                 targetid = myrand() % comm->remote_node_count;
 501                 rand_count += 1;
 502
 503                 /* Don't select an already-used index. */
 504                 for (j = 0; j < comm->gossip.gossip_branch; ++j) {
 505                     if (targetid == comm->retrys[j] || comm->remote_limiters[targetid].awol > awol_threshold) {
 506                         printlog(LOG_DEBUG, "Gossip: disqualified targetid %d.  retrys[j] is %d, and remote awol count is %d\n", targetid, comm->retrys[j], comm->remote_limiters[targetid].awol);
 507                         targetid = -2;
 508                         break;
 509                     }
 510                 }
 511             }
 512
 513             if (targetid < 0) {
 514                 /* Couldn't find a suitable peer to send to... */
 515                 message_portion += 1;
 516                 printlog(LOG_DEBUG, "Gossip: exhausted random peer search.\n");
 517                 continue;
 518             } else {
 519                 printlog(LOG_DEBUG, "Gossip: settled on peer id %d.\n", targetid);
 520             }
 521
 522             remote = &comm->remote_limiters[targetid];
 523         }
 524
 525         comm->retrys[i] = targetid;
 526
 527         toaddr.sin_addr.s_addr = remote->addr; /* Already in network byte order. */
 528         toaddr.sin_port = remote->port;
 529
 530         memset(&msg, 0, sizeof(message_t));
 531         msg.magic = MAGIC_MSG;
 532         msg.ident_id = id;
 533         msg.value = msg_value + remote->outgoing.saved_value;
 534         msg.weight = msg_weight + remote->outgoing.saved_weight;
 535         msg.seqno = remote->outgoing.next_seqno;
 536         msg.min_seqno = remote->outgoing.first_seqno;
 537         msg.type = MSG;
 538
 539         remote->outgoing.next_seqno++;
 540         remote->outgoing.saved_value += msg_value;
 541         remote->outgoing.saved_weight += msg_weight;
 542
 543 #ifdef ALLOW_PARTITION
 544
 545         if (do_partition && ((partition_set & (1 << targetid)) == 0)) {
 546             printlog(LOG_DEBUG, "Partition: Gossip: ignoring targetid %d\n", targetid);
 547             continue;
 548         }
 549
 550 #endif
 551
 552         message_to_nbo(&msg);
 553
 554         if (sendto(sock, &msg, sizeof(msg), 0, (struct sockaddr *) &toaddr, sizeof(struct sockaddr_in)) < 0) {
 555             printlog(LOG_WARN, "WARN: limiter_send_gossip: sento failed.\n");
 556             result = errno;
 557             break;
 558         }
 559     }
 560
 561     comm->gossip.value = msg_value * message_portion;
 562     comm->gossip.weight = msg_weight * message_portion;
 563
 564     return result;
 565 }
 566
 567 #if 0
 568 int send_tcp_gossip(comm_ident_t *ident, FILE *logfile, int unused) {
 569     int i, targetid, sock;
 570     int result = 0;
 571     message_t msg;
 572
 573     memset(&msg, 0, sizeof(message_t));
 574     msg.magic = MAGIC_MSG;
 575     msg.ident_id = ident->ident_id;
 576     msg.value = ident->gossip.value / (ident->gossip.gossip_branch + 1);
 577     msg.weight = ident->gossip.weight / (ident->gossip.gossip_branch + 1);
 578
 579     message_to_nbo(&msg);
 580
 581     for (i = 0; i < ident->gossip.gossip_branch; ++i) {
 582         targetid = myrand() % ident->remote_node_count;
 583         sock = ident->remote_limiters[targetid].socket;
 584
 585         result = send(sock, &msg, sizeof(message_t), 0);
 586         if (result < 0) {
 587             result = errno;
 588             FD_CLR(sock, &ident->fds);
 589             close(sock);
 590             break;
 591         }
 592
 593         assert(result == sizeof(message_t));
 594     }
 595
 596     ident->gossip.value /= (ident->gossip.gossip_branch + 1);
 597     ident->gossip.weight /= (ident->gossip.gossip_branch + 1);
 598
 599     return result;
 600 }
 601 #endif
 602
 603 #if 0
 604 void *limiter_accept_thread(void *limiter) {
 605     sigset_t signal_mask;
 606
 607     sigemptyset(&signal_mask);
 608     sigaddset(&signal_mask, SIGHUP);
 609     pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
 610
 611     pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
 612     pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
 613     while (1) {
 614         limiter_accept((comm_limiter_t *) limiter);
 615     }
 616     pthread_exit(NULL);
 617 }
 618 void *ident_receive_thread(void *recv_args) {
 619     int i, sock, result;
 620     struct recv_thread_args *args = (struct recv_thread_args *) recv_args;
 621     comm_ident_t *ident = args->ident;
 622     pthread_rwlock_t *lock = args->lock;
 623     uint16_t port = args->port;
 624     struct sockaddr_in addr;
 625     socklen_t addrlen = sizeof(addr);
 626     hello_t hello;
 627
 628     free(args);
 629
 630     while (1) {
 631         memset(&hello, 0, sizeof(hello_t));
 632
 633         /*Try to connect to all remote nodes if they aren't already connected.*/
 634         pthread_rwlock_rdlock(lock);
 635         pthread_mutex_lock(&ident->lock);
 636
 637         hello.magic = MAGIC_HELLO;
 638         hello.ident_id = ident->ident_id;
 639         hello.port = ntohs(port);
 640
 641         hello_to_nbo(&hello);
 642
 643         for (i = 0; i < ident->remote_node_count; ++i) {
 644             if (is_connected(&ident->remote_limiters[i]))
 645                 continue; /* Ignore if it's already connected */
 646
 647             sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
 648             if (sock < 0) {
 649                 perror("socket");
 650                 continue;
 651             }
 652
 653             assert(sock >= 0);
 654
 655             memset(&addr, 0, sizeof(struct sockaddr_in));
 656             addr.sin_family = AF_INET;
 657             addr.sin_port = ident->remote_limiters[i].port;
 658             addr.sin_addr.s_addr = ident->remote_limiters[i].addr;
 659
 660             result = connect(sock, (struct sockaddr *) &addr, addrlen);
 661             if (result < 0) {
 662                 close(sock);
 663                 continue;
 664             }
 665
 666             result = send(sock, &hello, sizeof(hello_t), 0);
 667             if (result < 0) {
 668                 close(sock);
 669                 continue;
 670             }
 671
 672             assert(result == sizeof(hello_t));
 673
 674             ident->remote_limiters[i].socket = sock;
 675             printf("Connected on socket: %d\n", sock);
 676             FD_SET(sock, &ident->fds);
 677         }
 678
 679         pthread_rwlock_unlock(lock);
 680         pthread_mutex_unlock(&ident->lock);
 681
 682         ident_receive(ident, lock);
 683     }
 684     pthread_exit(NULL);
 685 }
 686 #endif