From 298d0f7f26a469d6299b17a731d4add03cc22591 Mon Sep 17 00:00:00 2001
From: Quentin Dufour <quentin@deuxfleurs.fr>
Date: Wed, 28 Aug 2019 10:50:34 +0200
Subject: [PATCH] Add timeout for the classifier

---
 src/algo_thunder.c | 33 ++++++++++++++++++++++++++++++---
 1 file changed, 30 insertions(+), 3 deletions(-)

diff --git a/src/algo_thunder.c b/src/algo_thunder.c
index cce83e7..9e7072c 100644
--- a/src/algo_thunder.c
+++ b/src/algo_thunder.c
@@ -7,6 +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 200
 
 struct thunder_ctx {
   uint16_t recv_id;
@@ -39,6 +40,10 @@ uint64_t compute_delta(struct timespec* prev_time, uint64_t max) {
   return mili_sec;
 }
 
+int is_blacklisted(struct thunder_ctx* thunderc, int link_id) {
+  return thunderc->blacklisted[link_id] >= thunderc->received_pkts_on_link[link_id];
+}
+
 void prepare(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;
@@ -120,7 +125,7 @@ int schedule(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct bu
     li[thunderc->selected_link].delta_t = 0;
 
     main_on_tcp_write(ctx, to_fdinfo);
-  } while (thunderc->blacklisted[thunderc->selected_link]);
+  } while (is_blacklisted (thunderc, thunderc->selected_link));
 
   if (ctx->verbose > 1) fprintf(stderr, "    [algo_thunder] Packets sent\n");
 
@@ -144,6 +149,21 @@ void algo_thunder_init(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx, struc
   thunderc->monit_pkt_size = sizeof(links.fmt.headers) + sizeof(links.fmt.content.link_monitoring_thunder) + sizeof(struct link_info) * (thunderc->total_links - 1);
 }
 
+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;
+  struct thunder_ctx* thunderc = bi->app_ctx->misc;
+
+  if (thunderc->received_pkts_on_link[bi->i] >= bi->missing) goto release;
+  if (thunderc->blacklisted[bi->i] >= bi->missing) goto release;
+
+  thunderc->blacklisted[bi->i] = bi->missing;
+
+release:
+  free(bi);
+}
+
 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;
@@ -159,10 +179,17 @@ void classify(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct b
   thunderc->received_pkts_on_link[link_id]++;
 
   struct link_info *li = &ap->fmt.content.link_monitoring_thunder.links_status;
-  for (int i = 0; i < thunderc->total_links; i++) {
+  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
-    set_timeout ()
+
+    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;
+
+    set_timeout (ctx, timeout, &bi, on_block);
   }
 }