Write scheduler

src/algo_thunder.c

@@ -8,11 +8,11 @@
 struct thunder_ctx {
   uint16_t recv_id;
   uint16_t emit_id;
-  uint64_t lid_sent;
   uint8_t selected_link;
   uint8_t total_links;
-  uint8_t lid_sent_per_link[64];
   uint8_t delta_t_per_link[64];
+  uint8_t blacklisted[64];
+  struct timespec prev_link_time;
 };
 
 
@@ -26,7 +26,7 @@ void prepare(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct bu
     .fmt.headers.size = sizeof(metadata.fmt.headers) + sizeof(metadata.fmt.content.udp_metadata_thunder),
     .fmt.headers.flags = 0,
     .fmt.content.udp_metadata_thunder.id = thunderc->emit_id,
-    .fmt.content.udp_metadata_thunder.deltat = 0 //@FIXME
+    .fmt.content.udp_metadata_thunder.deltat = 0 //@FIXME delta t must be set
   };
   buffer_append_ap (bp, &metadata);
 
@@ -42,32 +42,50 @@ int schedule(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct bu
   struct evt_core_fdinfo *to_fdinfo = NULL;
   struct evt_core_cat* cat = evt_core_get_from_cat (ctx, "tcp-write");
 
+  struct timespec curr;
+  int secs, nsecs;
+  uint64_t mili_sec;
+
+
   do {
-    thunderc->selected_link = thunderc->selected_link + 1 % cat->socklist->len;
-
-    if (thunderc->selected_link == 0) {
-      thunderc->lid_sent++;
-      for (int i = 0; i < thunderc->total_links; i++) {
-        thunderc->lid_sent_per_link[i]++;
-      }
-    }
+    // 1. We choose the link
+    thunderc->selected_link = (thunderc->selected_link + 1) % cat->socklist->len;
 
+    // 2. We create the packet template
     union abstract_packet links = {
       .fmt.headers.cmd = CMD_LINK_MONITORING_THUNDER,
-      .fmt.headers.size = sizeof(links.fmt.headers) + sizeof(links.fmt.content.link_monitoring_thunder),
+      .fmt.headers.size = sizeof(links.fmt.headers) + sizeof(links.fmt.content.link_monitoring_thunder) + sizeof(struct link_info) * (thunderc->total_links - 1),
       .fmt.headers.flags = 0,
-      .fmt.content.link_monitoring_thunder.lid = thunderc->lid_sent,
-      .fmt.content.link_monitoring_thunder.links_status = NULL
+      .fmt.content.link_monitoring_thunder.links_status = {}
     };
 
     to_fdinfo = g_array_index(cat->socklist, struct evt_core_fdinfo*, thunderc->selected_link);
 
-    // We move the buffer and notify the target
+    // 3. We append the template to the buffer
     struct buffer_packet* bp_dup = dup_buffer_tow (&app_ctx->br, bp, to_fdinfo);
-    buffer_append_ap (bp_dup, &links);
+    union abstract_packet *new_ap = buffer_append_ap (bp_dup, &links);
+
+    // 4. We compute the time difference
+    if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){
+      perror("clock_gettime error");
+      exit(EXIT_FAILURE);
+    }
+    secs = curr.tv_sec - thunderc->prev_link_time.tv_sec;
+    nsecs = curr.tv_nsec - thunderc->prev_link_time.tv_nsec;
+    thunderc->prev_link_time = curr;
+    mili_sec = secs * 1000 + nsecs / 1000000;
+    if (mili_sec > 200) mili_sec = 200;
+
+    // 5. We create the array
+    struct link_info *li = &new_ap->fmt.content.link_monitoring_thunder.links_status;
+    for (int i = 0; i < thunderc->total_links; i++) {
+      thunderc->delta_t_per_link[i] += mili_sec;
+      li[i].delta_t = thunderc->delta_t_per_link[i];
+    }
+    li[thunderc->selected_link].delta_t = 0;
 
     main_on_tcp_write(ctx, to_fdinfo);
-  } while (1);
+  } while (thunderc->blacklisted[thunderc->selected_link]);
 
   if (ctx->verbose > 1) fprintf(stderr, "    [algo_thunder] Packets sent\n");
 
@@ -85,7 +103,7 @@ void algo_thunder_init(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx, struc
   }
   memset(app_ctx->misc, 0, sizeof(struct thunder_ctx));
   struct thunder_ctx* thunderc = app_ctx->misc;
-
+  thunderc->selected_link = UINT8_MAX - 1;
 }
 
 int algo_thunder_on_stream(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {

src/packet.c

@@ -32,10 +32,12 @@ size_t get_full_size(struct buffer_packet* bp) {
   return &(buffer_free_ap (bp))->raw - &bp->ip[0];
 }
 
-void buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap) {
+union abstract_packet* buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap) {
   buffer_last_ap(bp)->fmt.headers.flags |= FLAG_READ_NEXT;
-  memcpy(buffer_last_ap(bp), ap, ap->fmt.headers.size);
+  union abstract_packet *new_ap = buffer_last_ap(bp);
+  memcpy(new_ap, ap, ap->fmt.headers.size);
   bp->ap_count++;
+  return new_ap;
 }
 
 enum FD_STATE read_packet_from_tcp(struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {

src/packet.h

@@ -43,7 +43,6 @@ enum PKT_FLAGS {
 
 struct link_info {
   uint8_t delta_t;
-  uint8_t delta_lid;
 };
 
 union abstract_packet {
@@ -57,8 +56,7 @@ union abstract_packet {
 
     union {
       struct {
-        uint16_t lid;
-        struct link_info *links_status;
+        struct link_info links_status;
       } link_monitoring_thunder;
       struct {
         uint16_t id;
@@ -90,7 +88,7 @@ struct udp_target {
 
 size_t get_full_size(struct buffer_packet* bp);
 
-void buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap);
+union abstract_packet* buffer_append_ap(struct buffer_packet* bp, union abstract_packet* ap);
 union abstract_packet* buffer_free_ptr(struct buffer_packet* bp);
 union abstract_packet* buffer_last_ptr(struct buffer_packet* bp);
 union abstract_packet* ap_next(union abstract_packet* ap);