WIP link healing

src/algo_rr.c

@@ -22,6 +22,7 @@ struct rr_ctx {
   uint8_t remote_links;
   int64_t mjit;
   uint16_t recv_id;
+  uint16_t recv_id_late;
   uint16_t sent_id;
   uint8_t current_link;
   struct timespec emit_time;
@@ -128,10 +129,20 @@ int set_timeout(struct evt_core_ctx* evts, uint64_t milli_sec, struct waited_pkt
 void rr_pkt_register(struct evt_core_ctx* ctx,  struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
   struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
   struct rr_ctx* rr = app_ctx->misc;
+  char buffer[16];
+  url_get_port (buffer, fdinfo->url);
+  int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded
 
-  //printf("Selected url %s for pkt %d to be queued for delivery\n", fdinfo->url, bp->ip.ap.str.id);
+  printf("Selected url %s for pkt %d to be queued for delivery\n", fdinfo->url, bp->ip.ap.str.id);
 
-  // 1. Update links I can use thanks to target feedback
+  // 0. Update remote links
+  if (ring_lt(rr->recv_id_late, bp->ip.ap.str.id) && !(rr->remote_links & 1 << link_num)) {
+    printf("Activate link=%d\n", link_num);
+    rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working
+    show_link_availability (rr);
+  }
+
+  // 1. Update my links I can use thanks to target feedback
   if (bp->ip.ap.str.id > rr->my_links_ver) {
     rr->my_links = bp->ip.ap.str.bitfield;
     rr->my_links_ver = bp->ip.ap.str.id;
@@ -158,22 +169,22 @@ void rr_pkt_register(struct evt_core_ctx* ctx,  struct evt_core_fdinfo* fdinfo,
     rr->wait[idx_waited].timer_fd = set_timeout(ctx, timeout, &rr->wait[idx_waited]);
   }
 
-  // 4. We queue the packet
+  // 4. If packet has not already a timer or has a wrong timer
   int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
+  if (rr->wait[idx_real].on && rr->wait[idx_real].id != bp->ip.ap.str.id) {
+    fprintf(stderr, "Waiting array is full, BUG\n");
+  } else if (!rr->wait[idx_real].on) {
+    rr->wait[idx_real].on = 1;
+    rr->wait[idx_real].id = bp->ip.ap.str.id;
+    rr->wait[idx_real].link_num = link_num;
+    rr->wait[idx_real].timer_fd = set_timeout(ctx, rr->mjit + 1, &rr->wait[idx_real]);
+  }
+
+  // 5. We queue the packet
   rr->real[idx_real].on = 1;
   rr->real[idx_real].idx = idx_real;
   rr->real[idx_real].link_fd = fdinfo->fd;
   mv_buffer_rtoa(app_ctx, fdinfo, &rr->real[idx_real].idx);
-
-  // 5. We make sure that the remote link is set to up
-  char buffer[16];
-  url_get_port (buffer, fdinfo->url);
-  int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded
-  rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working
-  if (!(rr->remote_links & 1 << link_num)) {
-    printf("Activate link=%d\n", link_num);
-    show_link_availability (rr);
-  }
 }
 
 void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct deferred_pkt* dp) {
@@ -187,18 +198,19 @@ void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct
 
   // 2. Get the buffer
   struct buffer_packet* bp = get_app_buffer (app_ctx, &dp->idx);
-  //printf("Selected url %s for pkt %d to be delivered\n", fdinfo->url, bp->ip.ap.str.id);
+  printf("Selected url %s for pkt %d to be delivered\n", fdinfo->url, bp->ip.ap.str.id);
 
   // 3. We update our cursor
   rr->recv_id = bp->ip.ap.str.id;
 
   // 4. We check that we don't have a running timeout
-  int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
+  // We want to keep timer until the end to allow link update on multi receive
+  /*int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
   if (rr->wait[idx_real].on) {
     rr->wait[idx_real].on = 0;
     evt_core_rm_fd (ctx, rr->wait[idx_real].timer_fd);
     printf("Removed timer for packet %d\n",bp->ip.ap.str.id);
-  }
+  }*/
 
   // 5. We free the buffer if it's a control packet and quit
   if (bp->ip.ap.str.flags & PKT_CONTROL) {
@@ -227,6 +239,7 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
   struct buffer_packet* bp = NULL;
 
   while(1) {
+    printf("Trying to deliver %d\n", rr->recv_id+1);
     struct deferred_pkt* def = &rr->real[(rr->recv_id+1) % PACKET_BUFFER_SIZE];
     if (!def->on) break;
     fdinfo = evt_core_get_from_fd (ctx, def->link_fd);
@@ -237,6 +250,7 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
     }
 
     rr_deliver(ctx, fdinfo, def);
+    printf("Delivered %d\n", rr->recv_id);
   }
 }
 
@@ -329,33 +343,32 @@ int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   bp->ip.ap.str.deltat = mili_sec;
   bp->ip.ap.str.bitfield = rr->remote_links;
   bp->ip.ap.str.prevlink = rr->current_link;
+  printf("Will send packet id=%d\n", bp->ip.ap.str.id);
 
-  int max = 10;
-  while(1) {
-    if (max-- < 0) break; // We have an error
-    rr->current_link = (rr->current_link + 1) % 8;
-    if (!(rr->my_links & (1 << rr->current_link))) continue;
-    //if (!(rr->remote_links & (1 << rr->current_link))) continue;
-    sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + rr->current_link); //@FIXME Hardcoded
-    //printf("-- Trying %s\n", url);
-    to_fdinfo = evt_core_get_from_url (ctx, url);
-    if (to_fdinfo != NULL) break; // We found a good candidate
-  }
   rr->emit_time = curr;
   rr->sent_id++;
 
-  // 4. A whole packet has been read, we will find someone to write it
+  int max = 10;
-  if (to_fdinfo == NULL) {
+  uint8_t sel_link = rr->current_link;
-    fprintf(stderr, "No fd for URL %s in udp-read. Dropping packet :( \n", fdinfo->url);
+  while(max-- >= 0) {
-    mv_buffer_wtof (app_ctx, fdinfo);
+    sel_link = (sel_link + 1) % 8;
-    return 1;
+    sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + sel_link); //@FIXME Hardcoded
+    to_fdinfo = evt_core_get_from_url (ctx, url);
+    if (to_fdinfo == NULL) continue; // Missing link
+    if (rr->my_links & (1 << sel_link)) {
+      rr->current_link = sel_link;
+      mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo);
+      rr_on_tcp_write(ctx, to_fdinfo);
+      return 0;
+    } else {
+      dup_buffer_tow(app_ctx, bp, to_fdinfo);
+      rr_on_tcp_write(ctx, to_fdinfo);
+    }
   }
-  //printf("Pass packet from %s to %s\n", fdinfo->url, url);
-
-  // 5. We move the buffer and notify the target
-  mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo);
-  rr_on_tcp_write(ctx, to_fdinfo);
 
+  // 4. A whole packet has been read, we will find someone to write it
+  fprintf(stderr, "No fd for URL %s in udp-read. Dropping packet :( \n", fdinfo->url);
+  mv_buffer_wtof (app_ctx, fdinfo);
   return 0;
 
 co_error:
@@ -396,7 +409,8 @@ int rr_on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) {
   struct waited_pkt* pkt = fdinfo->other;
   evt_core_rm_fd(ctx, fdinfo->fd);
   pkt->on = 0;
-  if (ring_lt(pkt->id, rr->recv_id)) return 1;
+  if (ring_gt (pkt->id, rr->recv_id_late)) rr->recv_id_late = pkt->id;
+  if (ring_le (pkt->id, rr->recv_id)) return 1;
 
   printf("Timer reached for packet %d\n", pkt->id);
 
@@ -455,6 +469,8 @@ void algo_rr(struct evt_core_ctx* evt, struct algo_skel* as) {
   rr->my_links = 0xff;
   rr->remote_links = 0xff;
   rr->sent_id = 1;
+  rr->recv_id = 0;
+  rr->recv_id_late = 0;
   ctx->misc = rr;
   for (int i = 0; i < sizeof(ctx->bps) / sizeof(ctx->bps[0]); i++) {
     g_queue_push_tail(ctx->free_buffer, &(ctx->bps[i]));

src/algo_utils.c

@@ -21,6 +21,19 @@ void iterate2(int* fd, struct buffer_packet *bp, gpointer user_data) {
   fprintf(stderr, "fd=%d has a used_buffer entry\n", *fd);
 }
 
+void debug_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) {
+  fprintf(stderr, "No more free buffer for fd=%d.\n", fdinfo->fd);
+  int waiting_count = 0;
+  g_hash_table_foreach(app_ctx->write_waiting, (GHFunc)iterate, &waiting_count);
+  g_hash_table_foreach(app_ctx->used_buffer, (GHFunc)iterate2, NULL);
+  fprintf(stderr, "total_buffers=%d, free_buffer=%d, used_buffers=%d, app_buffer=%d, write_buffer=%d.\n",
+          PACKET_BUFFER_SIZE,
+          app_ctx->free_buffer->length,
+          g_hash_table_size(app_ctx->used_buffer),
+          g_hash_table_size(app_ctx->application_waiting),
+          waiting_count);
+}
+
 /**
  * Returns a buffer if available, NULL otherwise
  */
@@ -34,16 +47,8 @@ struct buffer_packet* get_read_buffer(struct algo_ctx *app_ctx, struct evt_core_
   // 2. Get a new buffer otherwise
   bp = g_queue_pop_head(app_ctx->free_buffer);
   if (bp == NULL) {
-    fprintf(stderr, "No more free buffer for fd=%d.\n", fdinfo->fd);
+    debug_buffer(app_ctx, fdinfo);
-    int waiting_count = 0;
+
-    g_hash_table_foreach(app_ctx->write_waiting, (GHFunc)iterate, &waiting_count);
-    g_hash_table_foreach(app_ctx->used_buffer, (GHFunc)iterate2, NULL);
-    fprintf(stderr, "total_buffers=%d, free_buffer=%d, used_buffers=%d, app_buffer=%d, write_buffer=%d.\n",
-            PACKET_BUFFER_SIZE,
-            app_ctx->free_buffer->length,
-            g_hash_table_size(app_ctx->used_buffer),
-            g_hash_table_size(app_ctx->application_waiting),
-            waiting_count);
     // 2.1 If no buffer is available, we subscribe to be notified later
     g_queue_push_tail (app_ctx->read_waiting, &(fdinfo->fd));
     return NULL;
@@ -165,6 +170,30 @@ void mv_buffer_atof(struct algo_ctx* app_ctx, void* from) {
   g_queue_push_tail (app_ctx->free_buffer, bp);
 }
 
+void dup_buffer_tow(struct algo_ctx* app_ctx, struct buffer_packet* bp, struct evt_core_fdinfo* to) {
+  GQueue* q;
+
+  // 1. We get a free buffer
+  struct buffer_packet* bp_dest = g_queue_pop_head(app_ctx->free_buffer);
+  if (bp_dest == NULL) {
+    debug_buffer(app_ctx, to);
+    return;
+  }
+
+  // 2. We duplicate the data
+  memcpy(bp_dest, bp, sizeof(struct buffer_packet));
+
+  // 3. We get the target writing queue
+  q = g_hash_table_lookup(app_ctx->write_waiting, &(to->fd));
+  if (q == NULL) {
+    q = g_queue_new ();
+    g_hash_table_insert(app_ctx->write_waiting, &(to->fd), q);
+  }
+
+  // 4. We push the content to the appropriate destination
+  g_queue_push_tail(q, bp_dest);
+}
+
 struct buffer_packet* get_app_buffer(struct algo_ctx *app_ctx, void* idx) {
   return g_hash_table_lookup (app_ctx->application_waiting, idx);
 }

src/algo_utils.h

@@ -24,6 +24,7 @@ void mv_buffer_wtof(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from);
 void mv_buffer_rtoa(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, void* to);
 void mv_buffer_atow(struct algo_ctx* app_ctx, void* from, struct evt_core_fdinfo* to);
 void mv_buffer_atof(struct algo_ctx* app_ctx, void* from);
+void dup_buffer_tow(struct algo_ctx* app_ctx, struct buffer_packet* bp, struct evt_core_fdinfo* to);
 
 struct buffer_packet* get_write_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo);
 struct buffer_packet* get_read_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo);

src/net_tools.c

@@ -44,7 +44,15 @@ int create_ip_client(char* host, char* service, int type) {
 }
 
 int create_tcp_client(char* host, char* service) {
-  return create_ip_client (host, service, SOCK_STREAM);
+  int sock = create_ip_client (host, service, SOCK_STREAM);
+  int activate = 1;
+  int err;
+  err = setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &activate, sizeof(activate));
+  if (err < 0) {
+    perror("setsockopt TCP_NODELAY");
+    exit(EXIT_FAILURE);
+  }
+  return sock;
 }
 
 int create_udp_client(char* host, char* service) {
@@ -93,7 +101,15 @@ int create_ip_server(char* host, char* service, int type) {
 }
 
 int create_tcp_server(char* host, char* service) {
-  return create_ip_server (host, service, SOCK_STREAM);
+  int sock = create_ip_server (host, service, SOCK_STREAM);
+  int activate = 1;
+  int err;
+  err = setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &activate, sizeof(activate));
+  if (err < 0) {
+    perror("setsockopt TCP_NODELAY");
+    exit(EXIT_FAILURE);
+  }
+  return sock;
 }
 
 int create_udp_server(char* host, char* service) {

src/net_tools.h

@@ -7,6 +7,8 @@
 #include <stdlib.h>
 #include <string.h>
 #include <sys/epoll.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
 
 int create_tcp_client(char* host, char* service);
 int create_udp_client(char* host, char* service);