From 588219f0f0754c10e005709a500f0accdd4f1435 Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Mon, 18 Mar 2019 17:58:40 +0100 Subject: [PATCH] WIP algo rr, TCP read logic --- src/algo_rr.c | 128 ++++++++++++++++++++++++++++++++++++-------------- src/utils.c | 20 ++++++++ src/utils.h | 5 ++ 3 files changed, 118 insertions(+), 35 deletions(-) diff --git a/src/algo_rr.c b/src/algo_rr.c index 72c500e..cdc62c6 100644 --- a/src/algo_rr.c +++ b/src/algo_rr.c @@ -1,7 +1,7 @@ #include #include "algo_skel.h" #include "algo_utils.h" - +#include "utils.h" struct deferred_pkt { int link_fd; @@ -15,10 +15,15 @@ struct rr_ctx { uint16_t mjit; uint16_t recv_id; uint16_t sent_id; + uint8_t current_link; struct deferred_pkt real[LINK_COUNT]; struct deferred_pkt stub[LINK_COUNT]; }; +int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); +int rr_on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); +int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); +int rr_on_udp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); int rr_on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { int conn_sock1, conn_sock2; @@ -92,54 +97,99 @@ void set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec) { evt_core_add_fd (evts, &fdinfo); } -void rr_handle_recv(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) { +int rr_update_states(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) { struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; struct rr_ctx* rr = app_ctx->misc; + char buffer[16]; + url_get_port (buffer, fdinfo->url); + int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded // 1. Update links I can use thanks to target feedback - if (bp->ip.ap.str.bitfield ^ rr->my_links) { - update_my_bitfield(bp, rr); + if (bp->ip.ap.str.id > rr->my_links_ver) { + rr->my_links = bp->ip.ap.str.bitfield; + rr->my_links_ver = bp->ip.ap.str.id; } // 2. If packet arrived too late - if (rr->recv_id > bp->ip.ap.str.id - 1) { + if (ring_gt(rr->recv_id, bp->ip.ap.str.id - 1)) { // 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; } // 3. If packet arrived too early - else if (rr->recv_id < bp->ip.ap.str.id - 1) { + 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); + // 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 % LINK_COUNT; rr->real[idx].bp = bp; rr->real[idx].link_fd = fdinfo->fd; + return 0; } // 4. If we were waiting this packet - else { - if (bp->ip.ap.str.flags & PKT_TIMEOUT) broken_rlink(rr, fdinfo); - else if (bp->ip.ap.str.flags & PKT_CONTROL) working_rlink(rr, fdinfo); - else deliver(bp); - rr->recv_id = bp->ip.ap.str.id; - struct deferred_pkt* def = &rr->real[rr->recv_id+1]; - if (def->bp != NULL) { - struct evt_core_fdinfo* next = evt_core_get_from_fd (ctx, def->link_fd); - if (next == NULL) return; - rr_handle_recv(ctx, def->bp, next); - } + rr->recv_id = bp->ip.ap.str.id; + // 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; +} + +void rr_deliver(struct evt_core_ctx* ctx, struct buffer_packet* bp, struct evt_core_fdinfo* fdinfo) { + struct evt_core_fdinfo *to_fdinfo = NULL; + struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; + char url[255]; + + // 1. 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) { + fprintf(stderr, "No fd for URL %s in udp:write for tcp-read. Dropping packet :( \n", url); + mv_buffer_wtor (app_ctx, fdinfo, bp); + } + + // 2. 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; + struct rr_ctx* rr = app_ctx->misc; + + do { + if(rr_update_states(ctx, bp, fdinfo)) rr_deliver(ctx, bp, fdinfo); + do { + struct deferred_pkt* def = &rr->real[rr->recv_id+1 % LINK_COUNT]; + 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 link\n"); + rr->recv_id++; + } + } while (fdinfo == NULL); + } while(1); } int rr_on_tcp_read(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; int read_res = FDS_READY; - char url[255]; // 1. Get current read buffer OR a new read buffer OR subscribe to be notified later if ((bp = get_read_buffer(app_ctx, fdinfo)) == NULL) return 1; @@ -152,29 +202,37 @@ int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { } // 3. Logic on packet - rr_handle_recv(ctx, bp, fdinfo); + rr_pkt_recv (ctx, fdinfo, bp); - /* - // 3. A whole packet has been read, we will find someone to write it - 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) { - fprintf(stderr, "No fd for URL %s in tcp-read. Dropping packet :( \n", url); - mv_buffer_wtor (app_ctx, fdinfo, bp); - return 1; - } - //printf("Pass packet from %s to %s\n", fdinfo->url, url); - - // 4. We move the buffer and notify the target - mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp); - on_udp_write(ctx, to_fdinfo); -*/ return 0; co_error: perror("Failed to TCP read"); exit(EXIT_FAILURE); } +int rr_on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { + struct buffer_packet* bp; + struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; + int write_res = FDS_READY; + + // 1. Get current write buffer OR a buffer from the waiting queue OR leave + if ((bp = get_write_buffer(app_ctx, fdinfo)) == NULL) return 1; + + // 2. Write buffer + write_res = write_packet_to_udp(fdinfo->fd, bp, fdinfo->other); + if (write_res == FDS_ERR) goto co_error; + if (write_res == FDS_AGAIN) return 1; + + // 3. A whole packet has been written + // Release the buffer and notify + mv_buffer_wtor(app_ctx, fdinfo, bp); + notify_read(ctx, app_ctx); + + return 0; +co_error: + perror("Failed to UDP write"); + exit(EXIT_FAILURE); +} void algo_rr(struct algo_skel* as) { struct algo_ctx* ctx = malloc(sizeof(struct algo_ctx)); diff --git a/src/utils.c b/src/utils.c index 57ddc11..281238c 100644 --- a/src/utils.c +++ b/src/utils.c @@ -60,3 +60,23 @@ int ring_buffer_free_space(struct ring_buffer* rb) { int ring_buffer_used_space(struct ring_buffer* rb) { return RING_BUFFER_SIZE - ring_buffer_free_space (rb); } + +int ring_gt(uint16_t v1, uint16_t v2) { + int32_t vv1 = (int32_t) v1, vv2 = (int32_t) v2; + return v1 != v2 && (vv1 - vv2) % UINT16_MAX <= UINT16_MAX / 2; +} + +int ring_ge(uint16_t v1, uint16_t v2) { + int32_t vv1 = (int32_t) v1, vv2 = (int32_t) v2; + return (vv1 - vv2) % UINT16_MAX <= UINT16_MAX / 2; +} + +int ring_lt(uint16_t v1, uint16_t v2) { + int32_t vv1 = (int32_t) v1, vv2 = (int32_t) v2; + return v1 != v2 && (vv1 - vv2) % UINT16_MAX > UINT16_MAX / 2; +} + +int ring_le(uint16_t v1, uint16_t v2) { + int32_t vv1 = (int32_t) v1, vv2 = (int32_t) v2; + return (vv1 - vv2) % UINT16_MAX > UINT16_MAX / 2; +} diff --git a/src/utils.h b/src/utils.h index d30250f..4aba838 100644 --- a/src/utils.h +++ b/src/utils.h @@ -18,3 +18,8 @@ void ring_buffer_ack_read(struct ring_buffer* rb, int size); int ring_buffer_write(struct ring_buffer* rb, char* source, int size); int ring_buffer_free_space(struct ring_buffer* rb); int ring_buffer_used_space(struct ring_buffer* rb); + +int ring_gt(uint16_t v1, uint16_t v2); +int ring_ge(uint16_t v1, uint16_t v2); +int ring_lt(uint16_t v1, uint16_t v2); +int ring_le(uint16_t v1, uint16_t v2);