From 5853e99e7b10251f812dc227efb7aa6b1955c202 Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Tue, 19 Mar 2019 15:39:05 +0100 Subject: [PATCH] Rework algorithm --- src/algo_rr.c | 137 +++++++++++++++++++++++------------------------ src/algo_utils.h | 2 +- src/packet.h | 3 +- 3 files changed, 70 insertions(+), 72 deletions(-) diff --git a/src/algo_rr.c b/src/algo_rr.c index 7e9ad51..1c613a0 100644 --- a/src/algo_rr.c +++ b/src/algo_rr.c @@ -3,7 +3,12 @@ #include "algo_utils.h" #include "utils.h" -#define WAITING 10 +struct waited_pkt { + uint16_t id; + int link_num; + int timer_fd; + uint8_t on; +}; struct deferred_pkt { int link_fd; @@ -18,8 +23,8 @@ struct rr_ctx { uint16_t recv_id; uint16_t sent_id; uint8_t current_link; - struct deferred_pkt real[WAITING]; - struct buffer_packet stub_bp; + struct deferred_pkt real[PACKET_BUFFER_SIZE]; + struct waited_pkt wait[PACKET_BUFFER_SIZE]; }; int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); @@ -64,7 +69,7 @@ co_error: exit(EXIT_FAILURE); } -void set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec, int64_t link) { +int set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec, struct waited_pkt* wpkt) { struct timespec now; struct itimerspec timer_config; char url[1024]; @@ -93,13 +98,14 @@ void set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec, int64_t link) { exit(EXIT_FAILURE); } fdinfo.cat->name = "timeout"; - fdinfo.other = (void*)link; // Should put the link number and the id + fdinfo.other = wpkt; // Should put the link number and the id fdinfo.free_other = NULL; sprintf(fdinfo.url, "timer:%ld:1", micro_sec); evt_core_add_fd (evts, &fdinfo); + return fdinfo.fd; } -int rr_update_states(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) { +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; struct rr_ctx* rr = app_ctx->misc; @@ -109,57 +115,59 @@ int rr_update_states(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct rr->my_links_ver = bp->ip.ap.str.id; } - // 2. If packet arrived too late + // 2. If packet arrived too late, we discard it if (ring_gt(rr->recv_id, bp->ip.ap.str.id - 1)) { - if (bp->ip.ap.str.flags & PKT_TIMEOUT) return 0; // We don't use real buffer pkt for PKT_TIMEOUT // Packet has already been delivered or dropped, we free the buffer - g_hash_table_remove (app_ctx->used_buffer, &(fdinfo->fd)); - memset(bp, 0, sizeof(struct buffer_packet)); - g_queue_push_tail(app_ctx->free_buffer, bp); - return 0; + mv_buffer_wtor (app_ctx, fdinfo, bp); + return; } - // 3. If packet arrived too early + // 3. If packet arrived too early, we register a timer if (ring_lt(rr->recv_id, bp->ip.ap.str.id - 1)) { int64_t timeout = rr->mjit - (int64_t) bp->ip.ap.str.deltat; if (timeout <= 0) timeout = 0; - set_timeout(ctx, timeout, bp->ip.ap.str.prevlink); - // Add a buffer to stub too - // Bitfield can be anything as a greater packet has triggered the bitfield update before - int idx = bp->ip.ap.str.id % PACKET_BUFFER_SIZE; - rr->real[idx].bp = bp; - rr->real[idx].link_fd = fdinfo->fd; - return 0; + int idx_waited = (bp->ip.ap.str.id - 1) % PACKET_BUFFER_SIZE; + rr->wait[idx_waited].on = 1; + rr->wait[idx_waited].id = bp->ip.ap.str.id; + rr->wait[idx_waited].link_num = bp->ip.ap.str.prevlink; + set_timeout(ctx, timeout, &rr->wait[idx_waited]); } - // 4. If we were waiting this packet - rr->recv_id = bp->ip.ap.str.id; + // 4. We queue the packet + int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE; + rr->real[idx_real].bp = bp; + rr->real[idx_real].link_fd = fdinfo->fd; + // 5. We make sure that the remote link is set to up char buffer[16]; url_get_port (buffer, fdinfo->url); int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded - - // 4.1 This is a timeout packet, we set the link as dead - if (bp->ip.ap.str.flags & PKT_TIMEOUT) { - rr->remote_links &= (1 << link_num) ^ UINT16_MAX; - return 0; - } - - // 4.2 This is a control packet, we set the link as alive - if (bp->ip.ap.str.flags & PKT_CONTROL) { - rr->remote_links |= 1 << link_num; - return 0; - } - - return 1; + rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working } -void rr_deliver(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) { +void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { struct evt_core_fdinfo *to_fdinfo = NULL; struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; + struct rr_ctx* rr = app_ctx->misc; char url[255]; - // 1. A whole packet has been read, we will find its target + // 0. We update our cursor + rr->recv_id = bp->ip.ap.str.id; + + // 1. We check that we don't have a running timeout + int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE; + if (rr->wait[idx_real].on) { + rr->wait[idx_real].on = 0; + evt_core_rm_fd (ctx, rr->wait[idx_real].timer_fd); + } + + // 2. We free the buffer if it's a control packet and quit + if (bp->ip.ap.str.flags & PKT_CONTROL) { + mv_buffer_wtor (app_ctx, fdinfo, bp); + return; + } + + // 3. A whole packet has been read, we will find its target sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port); to_fdinfo = evt_core_get_from_url (ctx, url); if (to_fdinfo == NULL) { @@ -167,33 +175,36 @@ void rr_deliver(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_c mv_buffer_wtor (app_ctx, fdinfo, bp); } - // 2. We move the buffer and notify the target + // 4. We move the buffer and notify the target mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp); rr_on_udp_write(ctx, to_fdinfo); } -void rr_pkt_recv(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { - struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; +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; + struct buffer_packet* bp = NULL; - do { - if(rr_update_states(ctx, bp, fdinfo)) rr_deliver(ctx, bp, fdinfo); - do { - struct deferred_pkt* def = &rr->real[(rr->recv_id+1) % PACKET_BUFFER_SIZE]; - if (def->bp == NULL) break; - def->bp = NULL; - struct evt_core_fdinfo* fdinfo = evt_core_get_from_fd (ctx, def->link_fd); - if (fdinfo == NULL) { - fprintf(stderr, "An error occured as the link seems to be closed for the requested fd\n"); - rr->recv_id++; - } - } while (fdinfo == NULL); - } while(1); + while(1) { + struct deferred_pkt* def = &rr->real[(rr->recv_id+1) % PACKET_BUFFER_SIZE]; + if (def->bp == NULL) break; + bp = def->bp; + def->bp = NULL; + fdinfo = evt_core_get_from_fd (ctx, def->link_fd); + if (fdinfo == NULL) { + fprintf(stderr, "An error occured as the link seems to be closed for the requested fd\n"); + rr->recv_id++; + continue; + } + + rr_deliver(ctx, fdinfo, bp); + } } int rr_on_tcp_read(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; int read_res = FDS_READY; // 1. Get current read buffer OR a new read buffer OR subscribe to be notified later @@ -207,7 +218,8 @@ int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { } // 3. Logic on packet - rr_pkt_recv (ctx, fdinfo, bp); + rr_pkt_register(ctx, fdinfo, bp); + rr_pkt_unroll (ctx, app_ctx); return 0; co_error: @@ -242,22 +254,9 @@ co_error: int rr_on_timer(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; - uint16_t id; - evt_core_rm_fd (ctx, fdinfo->fd); // We use the timeout only once - int fd = (int)fdinfo->other; - struct evt_core_fdinfo* tfdinfo = evt_core_get_from_fd (ctx, fd); - if (tfdinfo == NULL) { - fprintf(stderr, "An error occured as the link seems to be closed for the requested fd\n"); - rr->recv_id++; - return 1; - } - rr->stub_bp.ip.ap.str.id = 0; //? - rr->stub_bp.ip.ap.str.flags = PKT_TIMEOUT; - rr->stub_bp.ip.ap.str.deltat = rr->mjit; - rr->stub_bp.ip.ap.str.prevlink = UINT8_MAX; - rr_pkt_recv (ctx, tfdinfo, &rr->stub_bp); + return 1; } diff --git a/src/algo_utils.h b/src/algo_utils.h index 0c4d93e..820dfef 100644 --- a/src/algo_utils.h +++ b/src/algo_utils.h @@ -3,7 +3,7 @@ #include #include #include -#define PACKET_BUFFER_SIZE 10 +#define PACKET_BUFFER_SIZE 100 typedef void (*algo_ctx_free_misc)(void*); diff --git a/src/packet.h b/src/packet.h index 29e13b8..2928836 100644 --- a/src/packet.h +++ b/src/packet.h @@ -29,8 +29,7 @@ enum BP_MODE { }; enum PKT_FLAGS { - PKT_TIMEOUT = 1 << 0, - PKT_CONTROL = 1 << 1 + PKT_CONTROL = 1 << 0 }; union abstract_packet {