diff --git a/src/algo_naive.c b/src/algo_naive.c index cc8538a..2ae0af9 100644 --- a/src/algo_naive.c +++ b/src/algo_naive.c @@ -65,13 +65,13 @@ int on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { 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); + mv_buffer_wtof (app_ctx, fdinfo); 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); + mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo); on_udp_write(ctx, to_fdinfo); return 0; @@ -97,7 +97,7 @@ int on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { // 3. A whole packet has been written // Release the buffer and notify - mv_buffer_wtor(app_ctx, fdinfo, bp); + mv_buffer_wtof(app_ctx, fdinfo); notify_read(ctx, app_ctx); return 0; @@ -127,13 +127,13 @@ int on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { to_fdinfo = evt_core_get_from_url (ctx, url); if (to_fdinfo == NULL) { fprintf(stderr, "No fd for URL %s in udp-read. Dropping packet :( \n", url); - mv_buffer_wtor (app_ctx, fdinfo, bp); + mv_buffer_wtof (app_ctx, fdinfo); 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); + mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo); on_tcp_write(ctx, to_fdinfo); return 0; @@ -158,7 +158,7 @@ int on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { // 3. A whole packet has been written // Release the buffer and notify - mv_buffer_wtor(app_ctx, fdinfo, bp); + mv_buffer_wtof(app_ctx, fdinfo); notify_read(ctx, app_ctx); return 0; @@ -199,6 +199,7 @@ void algo_naive(struct evt_core_ctx* evt, struct algo_skel* as) { memset(ctx, 0, sizeof(struct algo_ctx)); ctx->free_buffer = g_queue_new (); ctx->read_waiting = g_queue_new (); + ctx->application_waiting = g_hash_table_new (NULL, NULL); ctx->used_buffer = g_hash_table_new(g_int_hash, g_int_equal); ctx->write_waiting = g_hash_table_new_full (g_int_hash, g_int_equal, NULL, naive_free_simple); for (int i = 0; i < sizeof(ctx->bps) / sizeof(ctx->bps[0]); i++) { diff --git a/src/algo_rr.c b/src/algo_rr.c index ae6dc8e..65fc13f 100644 --- a/src/algo_rr.c +++ b/src/algo_rr.c @@ -12,7 +12,8 @@ struct waited_pkt { struct deferred_pkt { int link_fd; - struct buffer_packet* bp; + int idx; + uint8_t on; }; struct rr_ctx { @@ -125,7 +126,7 @@ void rr_pkt_register(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, if (ring_gt(rr->recv_id, bp->ip.ap.str.id - 1)) { // Packet has already been delivered or dropped, we free the buffer fprintf(stderr, "Packet %d arrived too late (current: %d)\n", bp->ip.ap.str.id, rr->recv_id); - mv_buffer_wtor (app_ctx, fdinfo, bp); + mv_buffer_wtof (app_ctx, fdinfo); return; } @@ -144,9 +145,10 @@ void rr_pkt_register(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, // 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].on = 1; + rr->real[idx_real].idx = idx_real; rr->real[idx_real].link_fd = fdinfo->fd; - g_hash_table_remove(app_ctx->used_buffer, &fdinfo->fd); // We remove the packet from the reading buffer + mv_buffer_rtoa(app_ctx, fdinfo, &rr->real[idx_real].idx); // 5. We make sure that the remote link is set to up char buffer[16]; @@ -155,18 +157,23 @@ void rr_pkt_register(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working } -void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { +void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct deferred_pkt* dp) { 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. Marked the packet as handled + dp->on = 0; + + // 2. Get the buffer + struct buffer_packet* bp = get_app_buffer (app_ctx, &dp->idx); //printf("Selected url %s for pkt %d to be delivered\n", fdinfo->url, bp->ip.ap.str.id); - // 0. We update our cursor + // 3. We update our cursor rr->recv_id = bp->ip.ap.str.id; - // 1. We check that we don't have a running timeout + // 4. 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; @@ -174,22 +181,24 @@ void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct printf("Removed timer for packet %d\n",bp->ip.ap.str.id); } - // 2. We free the buffer if it's a control packet and quit + // 5. 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); + mv_buffer_atof (app_ctx, &dp->idx); return; } - // 3. A whole packet has been read, we will find its target + // 6. 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); + //mv_buffer_wtor (app_ctx, fdinfo, bp); + mv_buffer_atof (app_ctx, &dp->idx); } // 4. We move the buffer and notify the target - mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp); + //mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp); + mv_buffer_atow (app_ctx, &dp->idx, to_fdinfo); rr_on_udp_write(ctx, to_fdinfo); } @@ -200,9 +209,7 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) { 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; + if (!def->on) break; 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"); @@ -210,7 +217,7 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) { continue; } - rr_deliver(ctx, fdinfo, bp); + rr_deliver(ctx, fdinfo, def); } } @@ -257,7 +264,7 @@ int rr_on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { // 3. A whole packet has been written // Release the buffer and notify - mv_buffer_wtor(app_ctx, fdinfo, bp); + mv_buffer_wtof(app_ctx, fdinfo); notify_read(ctx, app_ctx); return 0; @@ -323,13 +330,13 @@ int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { // 4. A whole packet has been read, we will find someone to write it if (to_fdinfo == NULL) { fprintf(stderr, "No fd for URL %s in udp-read. Dropping packet :( \n", fdinfo->url); - mv_buffer_wtor (app_ctx, fdinfo, bp); + mv_buffer_wtof (app_ctx, fdinfo); return 1; } //printf("Pass packet from %s to %s\n", fdinfo->url, url); // 5. We move the buffer and notify the target - mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp); + mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo); rr_on_tcp_write(ctx, to_fdinfo); return 0; @@ -356,7 +363,7 @@ int rr_on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { // 3. A whole packet has been written // Release the buffer and notify - mv_buffer_wtor(app_ctx, fdinfo, bp); + mv_buffer_wtof(app_ctx, fdinfo); notify_read(ctx, app_ctx); return 0; @@ -419,6 +426,7 @@ void algo_rr(struct evt_core_ctx* evt, struct algo_skel* as) { memset(ctx, 0, sizeof(struct algo_ctx)); ctx->free_buffer = g_queue_new (); ctx->read_waiting = g_queue_new (); + ctx->application_waiting = g_hash_table_new (NULL, NULL); ctx->used_buffer = g_hash_table_new(g_int_hash, g_int_equal); ctx->write_waiting = g_hash_table_new_full (g_int_hash, g_int_equal, NULL, naive_free_simple); struct rr_ctx* rr = malloc(sizeof(struct rr_ctx)); diff --git a/src/algo_utils.c b/src/algo_utils.c index 4f2b5d4..a9e6c12 100644 --- a/src/algo_utils.c +++ b/src/algo_utils.c @@ -76,31 +76,98 @@ struct buffer_packet* get_write_buffer(struct algo_ctx *app_ctx, struct evt_core return bp; } -void mv_buffer_rtow(struct algo_ctx* app_ctx, - struct evt_core_fdinfo* from, - struct evt_core_fdinfo* to, - struct buffer_packet* bp) { - - // 1. We get the target writing queue +void mv_buffer_rtow(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, struct evt_core_fdinfo* to) { GQueue* q; + struct buffer_packet* bp; + + // 1. We get the packet buffer + bp = g_hash_table_lookup (app_ctx->used_buffer, &from->fd); + if (bp == NULL) { + fprintf(stderr, "Unable to find a buffer for fd=%d url=%s", from->fd, from->url); + exit(EXIT_FAILURE); + } + + // 2. We get the target writing queue q = g_hash_table_lookup(app_ctx->write_waiting, &(to->fd)); if (q == NULL) { q = g_queue_new (); g_hash_table_insert(app_ctx->write_waiting, &(to->fd), q); } - // 2. We move the buffer to the target queue + // 3. We move the data g_hash_table_remove(app_ctx->used_buffer, &from->fd); g_queue_push_tail(q, bp); } -void mv_buffer_wtor(struct algo_ctx* app_ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { +void mv_buffer_wtof(struct algo_ctx* app_ctx, struct evt_core_fdinfo* fdinfo) { + struct buffer_packet* bp = g_hash_table_lookup (app_ctx->used_buffer, &(fdinfo->fd)); + if (bp == NULL) { + fprintf(stderr, "Unable to find a buffer for fd=%d url=%s", fdinfo->fd, fdinfo->url); + exit(EXIT_FAILURE); + } bp->mode = BP_READING; bp->aread = 0; g_queue_push_tail (app_ctx->free_buffer, bp); g_hash_table_remove(app_ctx->used_buffer, &(fdinfo->fd)); } +void mv_buffer_rtoa(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, void* to) { + struct buffer_packet* bp; + bp = g_hash_table_lookup (app_ctx->used_buffer, &from->fd); + if (bp == NULL) { + fprintf(stderr, "Unable to find a buffer for fd=%d url=%s\n", from->fd, from->url); + exit(EXIT_FAILURE); + } + g_hash_table_remove(app_ctx->used_buffer, &from->fd); + if (g_hash_table_contains(app_ctx->application_waiting, to)) { + fprintf(stderr, "Data already exist for this entry\n"); + exit(EXIT_FAILURE); + } + g_hash_table_insert(app_ctx->application_waiting, to, bp); +} + +void mv_buffer_atow(struct algo_ctx* app_ctx, void* from, struct evt_core_fdinfo* to) { + GQueue* q; + struct buffer_packet* bp; + + // 1. We get the buffer + bp = g_hash_table_lookup (app_ctx->application_waiting, from); + if (bp == NULL) { + fprintf(stderr, "Unable to find this application buffer\n"); + exit(EXIT_FAILURE); + } + + // 2. We get the target writing queue + q = g_hash_table_lookup(app_ctx->write_waiting, &(to->fd)); + if (q == NULL) { + q = g_queue_new (); + g_hash_table_insert(app_ctx->write_waiting, &(to->fd), q); + } + + // 3. We move the buffer + g_hash_table_remove (app_ctx->application_waiting, from); + g_queue_push_tail(q, bp); +} + +void mv_buffer_atof(struct algo_ctx* app_ctx, void* from) { + struct buffer_packet* bp; + + // 1. We get the buffer + bp = g_hash_table_lookup (app_ctx->application_waiting, from); + if (bp == NULL) { + fprintf(stderr, "Unable to find this application buffer\n"); + exit(EXIT_FAILURE); + } + + // 2. We move it + g_hash_table_remove (app_ctx->application_waiting, from); + g_queue_push_tail (app_ctx->free_buffer, bp); +} + +struct buffer_packet* get_app_buffer(struct algo_ctx *app_ctx, void* idx) { + return g_hash_table_lookup (app_ctx->application_waiting, idx); +} + void notify_read(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) { struct evt_core_fdinfo* next_fdinfo = NULL; while (next_fdinfo == NULL) { diff --git a/src/algo_utils.h b/src/algo_utils.h index f515190..815e2b8 100644 --- a/src/algo_utils.h +++ b/src/algo_utils.h @@ -10,22 +10,27 @@ typedef void (*algo_ctx_free_misc)(void*); struct algo_ctx { int ref_count; struct buffer_packet bps[PACKET_BUFFER_SIZE]; - GQueue* free_buffer; // Available buffers - GHashTable* used_buffer; // Buffers used for reading or writing - GQueue* read_waiting; // Who wait to be notified for a read - GHashTable* write_waiting; // Structure to track packets waiting to be written - void* misc; // Additional structures - algo_ctx_free_misc free_misc; // Fx ptr to free misc + GQueue* free_buffer; // Available buffers + GHashTable* used_buffer; // Buffers used for reading or writing + GQueue* read_waiting; // Who wait to be notified for a read + GHashTable* application_waiting; // Structure that can be used by the algo for its internal logic + GHashTable* write_waiting; // Structure to track packets waiting to be written + void* misc; // Additional structures + algo_ctx_free_misc free_misc; // Fx ptr to free misc }; -void mv_buffer_rtow(struct algo_ctx* app_ctx, - struct evt_core_fdinfo* from, - struct evt_core_fdinfo* to, - struct buffer_packet* bp); -void mv_buffer_wtor(struct algo_ctx* app_ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp); +void mv_buffer_rtow(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, struct evt_core_fdinfo* to); +void mv_buffer_wtof(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from); +void mv_buffer_rtoa(struct algo_ctx* app_ctx, struct evt_core_fdinfo* from, void* to); +void mv_buffer_atow(struct algo_ctx* app_ctx, void* from, struct evt_core_fdinfo* to); +void mv_buffer_atof(struct algo_ctx* app_ctx, void* from); + struct buffer_packet* get_write_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo); struct buffer_packet* get_read_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo); +struct buffer_packet* get_app_buffer(struct algo_ctx *app_ctx, void* idx); + +void notify_read(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx); + void free_naive(void* app_ctx); void free_nothing(void* app_ctx); -void notify_read(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx); void naive_free_simple(void* v);