diff --git a/src/algo_rr.c b/src/algo_rr.c index 834c6d1..5bbc8e4 100644 --- a/src/algo_rr.c +++ b/src/algo_rr.c @@ -5,14 +5,14 @@ #include "proxy.h" #include "timer.h" -struct waited_pkt { - uint16_t id; - int link_num; - uint8_t on; +struct timer_info { + uint16_t health_id; + uint8_t prevlink; + uint16_t min_blocked_pkt; struct algo_ctx* algo; }; -struct deferred_pkt { +struct queued_pkt { int link_fd; int idx; uint16_t id; @@ -25,14 +25,15 @@ struct rr_ctx { uint16_t my_links_ver; uint8_t remote_links; int64_t mjit; - uint16_t recv_id; - uint16_t recv_id_late; + uint16_t health_id; + uint16_t health_id_late; + uint16_t content_id; uint16_t sent_id; uint8_t current_link; struct internet_packet prev_packet; struct timespec emit_time; - struct deferred_pkt real[PACKET_BUFFER_SIZE]; - struct waited_pkt wait[PACKET_BUFFER_SIZE]; + struct queued_pkt real[PACKET_BUFFER_SIZE]; + struct timer_info wait[PACKET_BUFFER_SIZE]; }; void show_link_availability(struct rr_ctx* rr) { @@ -51,6 +52,7 @@ void show_link_availability(struct rr_ctx* rr) { void expired_wait (struct evt_core_ctx* ctx, void* user); void expired_late(struct evt_core_ctx* ctx, void* user); +void on_timeout_health (struct evt_core_ctx* ctx, void* user); void rr_pkt_register(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; @@ -58,8 +60,8 @@ void rr_pkt_register(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, char buffer[16]; url_get_port (buffer, fdinfo->url); int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded - uint16_t real_idx = bp->ip.ap.str.id % PACKET_BUFFER_SIZE; - uint16_t wait_idx = (bp->ip.ap.str.id - 1) % PACKET_BUFFER_SIZE; + uint16_t real_idx = bp->ip.ap.content.health.id % PACKET_BUFFER_SIZE; + uint16_t wait_idx = (bp->ip.ap.content.health.id - 1) % PACKET_BUFFER_SIZE; //printf("Selected url %s for pkt %d to be queued for delivery\n", fdinfo->url, bp->ip.ap.str.id); @@ -147,12 +149,13 @@ void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct // 2. Get the buffer struct buffer_packet* bp = get_app_buffer (app_ctx, &dp->idx); + assert(bp->ip.ap.headers.cmd == CMD_CLEAR); // 3. We update our cursor - rr->recv_id = bp->ip.ap.str.id; + rr->recv_id = bp->ip.ap.content.clear.id; // 4. Find its target - sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port); + sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.content.clear.port); to_fdinfo = evt_core_get_from_url (ctx, url); if (to_fdinfo == NULL) { fprintf(stderr, "No fd for URL %s in udp:write for tcp-read. Dropping packet :( \n", url); @@ -166,6 +169,27 @@ void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct main_on_udp_write(ctx, to_fdinfo); } +void rr_pkt_manage_links(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { + struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; + struct rr_ctx* rr = app_ctx->misc; + + assert(bp->ip.ap.headers.cmd == CMD_HEALTH); + if (ring_le(app_ctx->health_id, rr->health_id_late)) { + return; + } + + int64_t timeout = rr->mjit - (int64_t) bp->ip.ap.content.health.deltat; + if (timeout <= 0) timeout = 0; + uint64_t idx = bp->ip.ap.content.health.id % PACKET_BUFFER_SIZE; + + rr->wait[idx].health_id = bp->ip.content.health.id; + rr->wait[idx].prevlink = bp->ip.ap.content.health.prevlink; + rr->wait[idx].min_blocked_pkt = bp->ip.ap.content.health.min_blocked_pkt; + rr->wait[idx].algo = app_ctx; + + set_timeout (ctx, timeout, &rr->wait[idx], on_timeout_health); +} + void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) { struct rr_ctx* rr = app_ctx->misc; struct evt_core_fdinfo* fdinfo = NULL; @@ -193,11 +217,15 @@ int algo_rr_on_stream(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; struct rr_ctx* rr = app_ctx->misc; - // 1. Register packet in our queue - rr_pkt_register(ctx, fdinfo, bp); + if (bp->ip.ap.headers.cmd == CMD_CLEAR) { + // 1. Register packet in our queue + rr_pkt_register(ctx, fdinfo, bp); - // 2. Process queue - rr_pkt_unroll (ctx, app_ctx); + // 2. Process queue + rr_pkt_unroll (ctx, app_ctx); + } else if (bp->ip.ap.headers.cmd == CMD_HEALTH) { + rr_pkt_manage_links(ctx, fdinfo, bp); + } return 0; co_error: @@ -273,6 +301,31 @@ co_error: exit(EXIT_FAILURE); } +void on_timeout_health (struct evt_core_ctx* ctx, void* raw) { + struct timer_info* t = raw; + struct algo_ctx* app_ctx = t->algo->cat->app_ctx; + struct rr_ctx* rr = app_ctx->misc; + + // 1. Update link recovery window if needed + if (ring_gt(t->health_id, rr->health_id_late)) rr->health_id_late = t->health_id; + + // 2. Blacklist previous link if needed + uint16_t prev_health_id = (t->health_id - 1); + uint16_t prev_health_idx = prev_health_id % PACKET_BUFFER_SIZE; + struct timer_info* t_old = rr->wait[prev_health_idx]; + if (t_old->health_id != prev_health_id) { + printf("Blacklist link=%d | ", t->prevlink); + rr->remote_links &= 0xff ^ 1 << t->prevlink; + show_link_availability (rr); + } + + // 3. Deliver blocked packets + while (ring_gt(t->min_blocked_pkt, rr->content_id)) { + rr->content_id++; + rr_pkt_unroll (ctx, app_ctx); + } +} + void expired_wait(struct evt_core_ctx* ctx, void* user) { struct waited_pkt* pkt = user; struct rr_ctx* rr = pkt->algo->misc; diff --git a/src/packet.h b/src/packet.h index dc229ca..80dc4ff 100644 --- a/src/packet.h +++ b/src/packet.h @@ -49,6 +49,7 @@ struct abstract_packet { uint8_t bitfield; uint8_t prevlink; uint16_t deltat; + uint16_t min_blocked_pkt; } health; struct { uint16_t id;