Add timeout for the classifier
This commit is contained in:
parent
6e3eb95778
commit
298d0f7f26
1 changed files with 30 additions and 3 deletions
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in a new issue