Reimplement timeout RR

src/algo_rr.c

@@ -5,14 +5,14 @@
 #include "proxy.h"
 #include "timer.h"
 
-struct waited_pkt {
+struct timer_info {
-  uint16_t id;
+  uint16_t health_id;
-  int link_num;
+  uint8_t prevlink;
-  uint8_t on;
+  uint16_t min_blocked_pkt;
   struct algo_ctx* algo;
 };
 
-struct deferred_pkt {
+struct queued_pkt {
   int link_fd;
   int idx;
   uint16_t id;
@@ -25,14 +25,15 @@ struct rr_ctx {
   uint16_t my_links_ver;
   uint8_t remote_links;
   int64_t mjit;
-  uint16_t recv_id;
+  uint16_t health_id;
-  uint16_t recv_id_late;
+  uint16_t health_id_late;
+  uint16_t content_id;
   uint16_t sent_id;
   uint8_t current_link;
   struct internet_packet prev_packet;
   struct timespec emit_time;
-  struct deferred_pkt real[PACKET_BUFFER_SIZE];
+  struct queued_pkt real[PACKET_BUFFER_SIZE];
-  struct waited_pkt wait[PACKET_BUFFER_SIZE];
+  struct timer_info wait[PACKET_BUFFER_SIZE];
 };
 
 void show_link_availability(struct rr_ctx* rr) {
@@ -51,6 +52,7 @@ void show_link_availability(struct rr_ctx* rr) {
 
 void expired_wait (struct evt_core_ctx* ctx, void* user);
 void expired_late(struct evt_core_ctx* ctx, void* user);
+void on_timeout_health (struct evt_core_ctx* ctx, void* user);
 
 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;
@@ -58,8 +60,8 @@ void rr_pkt_register(struct evt_core_ctx* ctx,  struct evt_core_fdinfo* fdinfo,
   char buffer[16];
   url_get_port (buffer, fdinfo->url);
   int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded
-  uint16_t real_idx = bp->ip.ap.str.id % PACKET_BUFFER_SIZE;
+  uint16_t real_idx = bp->ip.ap.content.health.id % PACKET_BUFFER_SIZE;
-  uint16_t wait_idx = (bp->ip.ap.str.id - 1) % PACKET_BUFFER_SIZE;
+  uint16_t wait_idx = (bp->ip.ap.content.health.id - 1) % PACKET_BUFFER_SIZE;
 
   //printf("Selected url %s for pkt %d to be queued for delivery\n", fdinfo->url, bp->ip.ap.str.id);
 
@@ -147,12 +149,13 @@ 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);
+  assert(bp->ip.ap.headers.cmd == CMD_CLEAR);
 
   // 3. We update our cursor
-  rr->recv_id = bp->ip.ap.str.id;
+  rr->recv_id = bp->ip.ap.content.clear.id;
 
   // 4. Find its target
-  sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port);
+  sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.content.clear.port);
   to_fdinfo = evt_core_get_from_url (ctx, url);
   if (to_fdinfo == NULL) {
     fprintf(stderr, "No fd for URL %s in udp:write for tcp-read. Dropping packet :( \n", url);
@@ -166,6 +169,27 @@ void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct
   main_on_udp_write(ctx, to_fdinfo);
 }
 
+void rr_pkt_manage_links(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;
+
+  assert(bp->ip.ap.headers.cmd == CMD_HEALTH);
+  if (ring_le(app_ctx->health_id, rr->health_id_late)) {
+    return;
+  }
+
+  int64_t timeout = rr->mjit - (int64_t) bp->ip.ap.content.health.deltat;
+  if (timeout <= 0) timeout = 0;
+  uint64_t idx = bp->ip.ap.content.health.id % PACKET_BUFFER_SIZE;
+
+  rr->wait[idx].health_id = bp->ip.content.health.id;
+  rr->wait[idx].prevlink = bp->ip.ap.content.health.prevlink;
+  rr->wait[idx].min_blocked_pkt = bp->ip.ap.content.health.min_blocked_pkt;
+  rr->wait[idx].algo = app_ctx;
+
+  set_timeout (ctx, timeout, &rr->wait[idx], on_timeout_health);
+}
+
 void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) {
   struct rr_ctx* rr = app_ctx->misc;
   struct evt_core_fdinfo* fdinfo = NULL;
@@ -193,11 +217,15 @@ int algo_rr_on_stream(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo,
   struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
   struct rr_ctx* rr = app_ctx->misc;
 
-  // 1. Register packet in our queue
+  if (bp->ip.ap.headers.cmd == CMD_CLEAR) {
-  rr_pkt_register(ctx, fdinfo, bp);
+    // 1. Register packet in our queue
+    rr_pkt_register(ctx, fdinfo, bp);
 
-  // 2. Process queue
+    // 2. Process queue
-  rr_pkt_unroll (ctx, app_ctx);
+    rr_pkt_unroll (ctx, app_ctx);
+  } else if (bp->ip.ap.headers.cmd == CMD_HEALTH) {
+    rr_pkt_manage_links(ctx, fdinfo, bp);
+  }
 
   return 0;
 co_error:
@@ -273,6 +301,31 @@ co_error:
   exit(EXIT_FAILURE);
 }
 
+void on_timeout_health (struct evt_core_ctx* ctx, void* raw) {
+  struct timer_info* t = raw;
+  struct algo_ctx* app_ctx = t->algo->cat->app_ctx;
+  struct rr_ctx* rr = app_ctx->misc;
+
+  // 1. Update link recovery window if needed
+  if (ring_gt(t->health_id, rr->health_id_late)) rr->health_id_late = t->health_id;
+
+  // 2. Blacklist previous link if needed
+  uint16_t prev_health_id = (t->health_id - 1);
+  uint16_t prev_health_idx = prev_health_id % PACKET_BUFFER_SIZE;
+  struct timer_info* t_old = rr->wait[prev_health_idx];
+  if (t_old->health_id != prev_health_id) {
+    printf("Blacklist link=%d | ", t->prevlink);
+    rr->remote_links &= 0xff ^ 1 << t->prevlink;
+    show_link_availability (rr);
+  }
+
+  // 3. Deliver blocked packets
+  while (ring_gt(t->min_blocked_pkt, rr->content_id)) {
+    rr->content_id++;
+    rr_pkt_unroll (ctx, app_ctx);
+  }
+}
+
 void expired_wait(struct evt_core_ctx* ctx, void* user) {
   struct waited_pkt* pkt = user;
   struct rr_ctx* rr = pkt->algo->misc;

src/packet.h

@@ -49,6 +49,7 @@ struct abstract_packet {
       uint8_t bitfield;
       uint8_t prevlink;
       uint16_t deltat;
+      uint16_t min_blocked_pkt;
     } health;
     struct {
       uint16_t id;