1 /* See the DRL-LICENSE file for this file's software license. */
3 #define _XOPEN_SOURCE 600
9 /* Socket functions. */
10 #include <sys/types.h>
11 #include <sys/socket.h>
13 /* Byte ordering and address structures. */
14 #include <arpa/inet.h>
19 /* close() & usleep */
22 /* Mutex lock/unlock. */
28 /* select() w/ timeout */
29 #include <sys/select.h>
35 /* sigaddset(), sigemptyset(), SIGHUP, etc. */
38 /* DRL data structures. */
39 #include "raterouter.h"
40 #include "ratetypes.h"
41 #include "drl_state.h"
42 #include "peer_comm.h"
45 /* Artifically makes a network partition. */
47 int partition_set = 0xfffffff;
49 extern limiter_t limiter;
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;
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 */
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 */
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);
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);
88 static int is_connected(remote_limiter_t *remote) {
89 struct sockaddr_in addr;
90 socklen_t addrlen = sizeof(addr);
92 if (getpeername(remote->socket, (struct sockaddr *) &addr, &addrlen) == 0)
98 static int send_ack(identity_t *ident, remote_limiter_t *remote, uint32_t seqno) {
101 struct sockaddr_in toaddr;
103 memset(&toaddr, 0, sizeof(struct sockaddr_in));
104 toaddr.sin_family = AF_INET;
106 toaddr.sin_addr.s_addr = remote->addr; /* Already in network byte order. */
107 toaddr.sin_port = remote->port;
109 memset(&msg, 0, sizeof(msg));
110 msg.magic = MAGIC_MSG;
111 msg.ident_id = ident->id;
115 message_to_nbo(&msg);
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");
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;
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");
138 memset(&sender, 0, sizeof(remote_node_t));
139 sender.addr = fromaddr.sin_addr.s_addr;
140 sender.port = fromaddr.sin_port;
142 message_to_hbo(&msg);
144 assert(msg.magic == MAGIC_MSG);
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));
150 pthread_testcancel();
152 pthread_rwlock_rdlock(&limiter.limiter_lock);
154 ident = map_search(limiter.stable_instance.ident_map, &msg.ident_id,
155 sizeof(msg.ident_id));
158 printlog(LOG_WARN, "WARN:recvd message for unknown identity.\n");
159 pthread_rwlock_unlock(&limiter.limiter_lock);
163 pthread_mutex_lock(&ident->comm.lock);
165 remote = map_search(ident->comm.remote_node_map, &sender, sizeof(remote_node_t));
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);
174 switch (ident->comm.comm_fabric) {
176 /* Use the message's value to be our new GRDrate/FPSweight for the
177 * message's sender. */
178 remote->rate = msg.value;
180 /* Reset the AWOL counter to zero since we received an update. */
186 if (msg.type == ACK) {
187 if (msg.seqno == remote->outgoing.next_seqno - 1) {
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;
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;
203 /* Ignore ack if it isn't for most recent message. */
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;
218 remote->incoming.seen_seqno = msg.seqno;
219 remote->incoming.saved_value = msg.value;
220 remote->incoming.saved_weight = msg.weight;
222 ident->comm.gossip.value += diff_value;
223 ident->comm.gossip.weight += diff_weight;
224 send_ack(ident, remote, msg.seqno);
226 /* The entire message is old news. (Duplicate). */
234 printlog(LOG_CRITICAL, "ERR: Unknown identity comm fabric.\n");
238 pthread_mutex_unlock(&ident->comm.lock);
239 pthread_rwlock_unlock(&limiter.limiter_lock);
243 static void limiter_accept(comm_limiter_t *limiter) {
245 struct sockaddr_in fromaddr;
246 socklen_t fromlen = sizeof(fromaddr);
247 remote_node_t sender;
248 remote_limiter_t *remote;
251 ident_handle *handle = NULL;
253 sock = accept(limiter->tcp_socket, (struct sockaddr *)&fromaddr, &fromlen);
257 memset(&hello, 0, sizeof(hello_t));
258 result = recv(sock, &hello, sizeof(hello_t), 0);
262 return; /* Failure - ignore it. */
265 assert(result == sizeof(hello_t));
267 hello_to_hbo(&hello);
269 assert(hello.magic == MAGIC_HELLO);
271 memset(&sender, 0, sizeof(remote_node_t));
272 sender.addr = fromaddr.sin_addr.s_addr;
273 sender.port = ntohs(hello.port);
275 pthread_testcancel();
277 pthread_rwlock_rdlock(&limiter->rwlock);
279 handle = map_search(limiter->ident_id_to_handle, (void *) &hello.ident_id, sizeof(hello.ident_id));
281 if (handle == NULL) {
282 printlog(LOG_WARN, "WARN:recvd hello for unknown identity.\n");
283 pthread_rwlock_unlock(&limiter->rwlock);
287 ident = limiter->identities[*handle];
288 assert(ident != NULL);
290 pthread_mutex_lock(&ident->lock);
292 remote = map_search(ident->remote_node_map, &sender, sizeof(remote_node_t));
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);
302 if (is_connected(remote)) {
303 /* We are still connected, don't need the new socket. */
305 pthread_mutex_unlock(&ident->lock);
306 pthread_rwlock_unlock(&limiter->rwlock);
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);
315 pthread_mutex_unlock(&ident->lock);
316 pthread_rwlock_unlock(&limiter->rwlock);
319 static void read_tcp_message(comm_ident_t *ident, pthread_rwlock_t *limiter_rwlock, int sock) {
323 memset(&msg, 0, sizeof(message_t));
325 result = recv(sock, &msg, sizeof(message_t), 0);
328 pthread_rwlock_rdlock(limiter_rwlock);
329 pthread_mutex_lock(&ident->lock);
330 FD_CLR(sock, &ident->fds);
332 pthread_mutex_unlock(&ident->lock);
333 pthread_rwlock_unlock(limiter_rwlock);
337 assert(result == sizeof(message_t));
339 message_to_hbo(&msg);
340 assert(msg.magic == MAGIC_MSG);
342 pthread_rwlock_rdlock(limiter_rwlock);
343 pthread_mutex_lock(&ident->lock);
345 switch (ident->comm_fabric) {
347 ident->gossip.value += msg.value;
348 ident->gossip.weight += msg.weight;
353 assert(1 == 0); /* This case shouldn't happen. Punt for now... */
356 pthread_mutex_unlock(&ident->lock);
357 pthread_rwlock_unlock(limiter_rwlock);
360 static void ident_receive(comm_ident_t *ident, pthread_rwlock_t *limiter_rwlock) {
361 int select_result, i;
363 struct timeval timeout;
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);
375 /* mask interrupt signals for this thread? */
377 select_result = select(FD_SETSIZE, &fds_copy, NULL, NULL, &timeout);
379 assert(select_result >= 0);
381 if (select_result == 0)
382 return; /* Timed out */
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);
393 #define ALLOW_PARTITION
395 int send_udp_mesh(comm_t *comm, uint32_t id, int sock) {
397 remote_limiter_t *remote;
399 struct sockaddr_in toaddr;
402 #ifdef ALLOW_PARTITION
404 int partition_count = 0;
410 memset(&toaddr, 0, sizeof(struct sockaddr_in));
411 toaddr.sin_family = AF_INET;
413 memset(&msg, 0, sizeof(message_t));
414 msg.magic = MAGIC_MSG;
416 msg.value = comm->local_rate;
417 /* Do we want seqnos for mesh? We can get by without them. */
419 message_to_nbo(&msg);
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];
425 #ifdef ALLOW_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));
435 printlog(LOG_DEBUG, "Partition: ignoring host %s\n", dest_ip);
437 partition_count += 1;
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");
449 printlog(LOG_WARN, " - The error was |%d|\n", strerror(result));
452 partition_count += 1;
458 int send_udp_gossip(comm_t *comm, uint32_t id, int sock) {
461 remote_limiter_t *remote;
462 struct sockaddr_in toaddr;
463 double msg_value, msg_weight;
465 memset(&toaddr, 0, sizeof(struct sockaddr_in));
466 toaddr.sin_family = AF_INET;
468 msg_value = comm->gossip.value / (comm->gossip.gossip_branch + 1);
469 msg_weight = comm->gossip.weight / (comm->gossip.gossip_branch + 1);
471 for (i = 0; i < comm->gossip.gossip_branch; ++i) {
474 printlog(LOG_DEBUG, "Gossip loop iteration, i=%d, branch=%d\n", i, comm->gossip.gossip_branch);
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]);
483 while (targetid == -2) {
484 targetid = myrand() % comm->remote_node_count;
486 for (j = 0; j < comm->gossip.gossip_branch; ++j) {
487 if (targetid == comm->retrys[j]) {
494 remote = &comm->remote_limiters[targetid];
495 //printf("%d:r:%d, ", i, targetid);
498 toaddr.sin_addr.s_addr = remote->addr; /* Already in network byte order. */
499 toaddr.sin_port = remote->port;
501 memset(&msg, 0, sizeof(message_t));
502 msg.magic = MAGIC_MSG;
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;
510 remote->outgoing.next_seqno++;
511 remote->outgoing.saved_value += msg_value;
512 remote->outgoing.saved_weight += msg_weight;
514 message_to_nbo(&msg);
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");
522 comm->retrys[i] = targetid;
526 comm->gossip.value = msg_value;
527 comm->gossip.weight = msg_weight;
533 int send_tcp_gossip(comm_ident_t *ident, FILE *logfile, int unused) {
534 int i, targetid, sock;
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);
544 message_to_nbo(&msg);
546 for (i = 0; i < ident->gossip.gossip_branch; ++i) {
547 targetid = myrand() % ident->remote_node_count;
548 sock = ident->remote_limiters[targetid].socket;
550 result = send(sock, &msg, sizeof(message_t), 0);
553 FD_CLR(sock, &ident->fds);
558 assert(result == sizeof(message_t));
561 ident->gossip.value /= (ident->gossip.gossip_branch + 1);
562 ident->gossip.weight /= (ident->gossip.gossip_branch + 1);
569 void *limiter_accept_thread(void *limiter) {
570 sigset_t signal_mask;
572 sigemptyset(&signal_mask);
573 sigaddset(&signal_mask, SIGHUP);
574 pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
576 pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
577 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
579 limiter_accept((comm_limiter_t *) limiter);
583 void *ident_receive_thread(void *recv_args) {
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);
596 memset(&hello, 0, sizeof(hello_t));
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);
602 hello.magic = MAGIC_HELLO;
603 hello.ident_id = ident->ident_id;
604 hello.port = ntohs(port);
606 hello_to_nbo(&hello);
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 */
612 sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
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;
625 result = connect(sock, (struct sockaddr *) &addr, addrlen);
631 result = send(sock, &hello, sizeof(hello_t), 0);
637 assert(result == sizeof(hello_t));
639 ident->remote_limiters[i].socket = sock;
640 printf("Connected on socket: %d\n", sock);
641 FD_SET(sock, &ident->fds);
644 pthread_rwlock_unlock(lock);
645 pthread_mutex_unlock(&ident->lock);
647 ident_receive(ident, lock);