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); } }