Better classification

src/algo_thunder.c

@@ -7,7 +7,7 @@
 
 // A Tor cell size is 512 bytes but handle only 498 bytes of data
 #define TOR_CELL_SIZE 498
-#define ALLOWED_JITTER_MS 100
+#define ALLOWED_JITTER_MS 200
 #define MAX_LINKS 64
 
 struct thunder_ctx {
@@ -16,10 +16,11 @@ struct thunder_ctx {
   uint8_t selected_link;
   uint8_t total_links;
   uint64_t delta_t_per_link[MAX_LINKS];
+  uint64_t rcv_delta_t_per_link[MAX_LINKS];
   uint64_t received_pkts_on_link[MAX_LINKS];
   uint64_t blacklisted[MAX_LINKS];
   size_t monit_pkt_size;
-  struct timespec prev_link_time, prev_packet_time;
+  struct timespec prev_link_time, prev_rcv_link_time;
 };
 
 uint64_t compute_delta(struct timespec* prev_time, uint64_t max) {
@@ -160,7 +161,7 @@ int schedule(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct bu
   return 0;
 }
 
-struct block_info { uint8_t i; struct algo_ctx* app_ctx; uint64_t missing; };
+struct block_info { uint8_t i; struct algo_ctx* app_ctx; uint64_t missing;};
 
 void on_block (struct evt_core_ctx* ctx, void* raw) {
   struct block_info* bi = raw;
@@ -176,6 +177,16 @@ release:
   free(bi);
 }
 
+int is_in_order(struct thunder_ctx* thunderc, uint8_t link_id) {
+  uint64_t ref = thunderc->received_pkts_on_link[link_id];
+  for (int i = 0; i < thunderc->total_links; i++) {
+    uint64_t expected = link_id > i ? ref - 1 : ref;
+    if (thunderc->received_pkts_on_link[i] > expected) return 0;
+  }
+
+  return 1;
+}
+
 void classify(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) {
   struct algo_ctx* app_ctx = fdinfo->cat->app_ctx;
   struct thunder_ctx* thunderc = app_ctx->misc;
@@ -187,27 +198,67 @@ void classify(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct b
     exit(EXIT_FAILURE);
   }
 
+  /*
   if (ap->fmt.headers.flags & FLAG_RESET) {
     for (int i = 0; i < MAX_LINKS; i++) thunderc->received_pkts_on_link[i] = 1;
   }
+  */
 
+  // 1. Update link info
   int link_id = url_get_port_int(fdinfo->url) - 7500;
   thunderc->received_pkts_on_link[link_id]++;
   printf("Received %ld packets on link %d\n", thunderc->received_pkts_on_link[link_id], link_id);
-
   struct link_info *li = &ap->fmt.content.link_monitoring_thunder.links_status;
+
+  uint64_t mili_sec = compute_delta (&thunderc->prev_rcv_link_time, UINT16_MAX);
+  for (int i = 0; i < thunderc->total_links; i++) {
+    thunderc->rcv_delta_t_per_link[i] += mili_sec;
+  }
+  thunderc->rcv_delta_t_per_link[link_id] = 0;
+
+  // 2. Disable links that have received packets too late
+  if (is_in_order (thunderc, link_id)) {
+    printf("Local: ");
+    for (int i = 0; i < thunderc->total_links; i++) {
+      printf("%ld ", thunderc->rcv_delta_t_per_link[i]);
+    }
+    printf("\n");
+    printf("Packet: ");
+    for (int i = 0; i < thunderc->total_links; i++) {
+      printf("%d ", li[i].delta_t);
+    }
+    printf("\n");
+
+    for (int i = 0; i < thunderc->total_links; i++) {
+      if (ALLOWED_JITTER_MS >= li[i].delta_t) continue;
+      if (li[i].delta_t - ALLOWED_JITTER_MS <= thunderc->rcv_delta_t_per_link[i]) continue;
+
+      struct block_info *bi = malloc(sizeof(struct block_info));
+      bi->i = i; bi->app_ctx = app_ctx; bi->missing = thunderc->received_pkts_on_link[i]+1;
+
+      printf("  Packet Too Late - Blocked link %d (expected: at least %dms ago, received: %ldms ago)\n", i, li[i].delta_t - ALLOWED_JITTER_MS, thunderc->rcv_delta_t_per_link[i]);
+      on_block(ctx, bi);
+    }
+  }
+
+  // 3. Disable links that miss packets
   for (uint8_t i = 0; i < thunderc->total_links; i++) {
     uint64_t expected = i <= link_id ? thunderc->received_pkts_on_link[link_id] : thunderc->received_pkts_on_link[link_id] - 1;
     if (thunderc->received_pkts_on_link[i] >= expected) continue; // Nothing to do, all packets have been received
 
     int64_t timeout = ALLOWED_JITTER_MS - li[i].delta_t;
-    if (timeout < 0) timeout = 0;
 
     struct block_info *bi = malloc(sizeof(struct block_info));
     bi->i = i; bi->app_ctx = app_ctx; bi->missing = expected;
 
+    if (timeout <= 0) {
+      on_block(ctx, bi);
+      printf("  Missing Packet - Blocked link %d (expected: %ld, seen: %ld, )\n", i, expected, thunderc->received_pkts_on_link[i]);
+      continue;
+    }
+
     set_timeout (ctx, timeout, bi, on_block);
-    printf("  Triggered timeout for link %d in %ldms (expected: %ld, seen: %ld)\n", i, timeout, expected, thunderc->received_pkts_on_link[i]);
+    printf("  Missing Packet - Triggered timeout for link %d in %ldms (expected: %ld, seen: %ld)\n", i, timeout, expected, thunderc->received_pkts_on_link[i]);
     if (ctx->verbose > 1) {
       fprintf(stderr, "    [algo_thunder] Set timeout on link %d of %ld ms (packets expected: %ld, seen: %ld)\n",
               i, timeout, expected, thunderc->received_pkts_on_link[i]);