diff --git a/src/algo_rr.c b/src/algo_rr.c index 2a5c5b1..b8b7469 100644 --- a/src/algo_rr.c +++ b/src/algo_rr.c @@ -22,6 +22,7 @@ struct rr_ctx { uint8_t remote_links; int64_t mjit; uint16_t recv_id; + uint16_t recv_id_late; uint16_t sent_id; uint8_t current_link; struct timespec emit_time; @@ -128,10 +129,20 @@ int set_timeout(struct evt_core_ctx* evts, uint64_t milli_sec, struct waited_pkt 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; + char buffer[16]; + url_get_port (buffer, fdinfo->url); + int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded - //printf("Selected url %s for pkt %d to be queued for delivery\n", fdinfo->url, bp->ip.ap.str.id); + printf("Selected url %s for pkt %d to be queued for delivery\n", fdinfo->url, bp->ip.ap.str.id); - // 1. Update links I can use thanks to target feedback + // 0. Update remote links + if (ring_lt(rr->recv_id_late, bp->ip.ap.str.id) && !(rr->remote_links & 1 << link_num)) { + printf("Activate link=%d\n", link_num); + rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working + show_link_availability (rr); + } + + // 1. Update my links I can use thanks to target feedback 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; @@ -158,22 +169,22 @@ void rr_pkt_register(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, rr->wait[idx_waited].timer_fd = set_timeout(ctx, timeout, &rr->wait[idx_waited]); } - // 4. We queue the packet + // 4. If packet has not already a timer or has a wrong timer int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE; + if (rr->wait[idx_real].on && rr->wait[idx_real].id != bp->ip.ap.str.id) { + fprintf(stderr, "Waiting array is full, BUG\n"); + } else if (!rr->wait[idx_real].on) { + rr->wait[idx_real].on = 1; + rr->wait[idx_real].id = bp->ip.ap.str.id; + rr->wait[idx_real].link_num = link_num; + rr->wait[idx_real].timer_fd = set_timeout(ctx, rr->mjit + 1, &rr->wait[idx_real]); + } + + // 5. We queue the packet rr->real[idx_real].on = 1; rr->real[idx_real].idx = idx_real; rr->real[idx_real].link_fd = fdinfo->fd; 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]; - url_get_port (buffer, fdinfo->url); - int link_num = atoi(buffer) - 7500; // @FIXME Hardcoded - rr->remote_links |= 1 << link_num; // Make sure that the link is marked as working - if (!(rr->remote_links & 1 << link_num)) { - printf("Activate link=%d\n", link_num); - show_link_availability (rr); - } } void rr_deliver(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo, struct deferred_pkt* dp) { @@ -187,18 +198,19 @@ 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); - //printf("Selected url %s for pkt %d to be delivered\n", fdinfo->url, bp->ip.ap.str.id); + printf("Selected url %s for pkt %d to be delivered\n", fdinfo->url, bp->ip.ap.str.id); // 3. We update our cursor rr->recv_id = bp->ip.ap.str.id; // 4. We check that we don't have a running timeout - int idx_real = bp->ip.ap.str.id % PACKET_BUFFER_SIZE; + // We want to keep timer until the end to allow link update on multi receive + /*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); printf("Removed timer for packet %d\n",bp->ip.ap.str.id); - } + }*/ // 5. We free the buffer if it's a control packet and quit if (bp->ip.ap.str.flags & PKT_CONTROL) { @@ -227,6 +239,7 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) { struct buffer_packet* bp = NULL; while(1) { + printf("Trying to deliver %d\n", rr->recv_id+1); struct deferred_pkt* def = &rr->real[(rr->recv_id+1) % PACKET_BUFFER_SIZE]; if (!def->on) break; fdinfo = evt_core_get_from_fd (ctx, def->link_fd); @@ -237,6 +250,7 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) { } rr_deliver(ctx, fdinfo, def); + printf("Delivered %d\n", rr->recv_id); } } @@ -329,33 +343,32 @@ int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { bp->ip.ap.str.deltat = mili_sec; bp->ip.ap.str.bitfield = rr->remote_links; bp->ip.ap.str.prevlink = rr->current_link; + printf("Will send packet id=%d\n", bp->ip.ap.str.id); - int max = 10; - while(1) { - if (max-- < 0) break; // We have an error - rr->current_link = (rr->current_link + 1) % 8; - if (!(rr->my_links & (1 << rr->current_link))) continue; - //if (!(rr->remote_links & (1 << rr->current_link))) continue; - sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + rr->current_link); //@FIXME Hardcoded - //printf("-- Trying %s\n", url); - to_fdinfo = evt_core_get_from_url (ctx, url); - if (to_fdinfo != NULL) break; // We found a good candidate - } rr->emit_time = curr; rr->sent_id++; - // 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_wtof (app_ctx, fdinfo); - return 1; + int max = 10; + uint8_t sel_link = rr->current_link; + while(max-- >= 0) { + sel_link = (sel_link + 1) % 8; + sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + sel_link); //@FIXME Hardcoded + to_fdinfo = evt_core_get_from_url (ctx, url); + if (to_fdinfo == NULL) continue; // Missing link + if (rr->my_links & (1 << sel_link)) { + rr->current_link = sel_link; + mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo); + rr_on_tcp_write(ctx, to_fdinfo); + return 0; + } else { + dup_buffer_tow(app_ctx, bp, to_fdinfo); + rr_on_tcp_write(ctx, to_fdinfo); + } } - //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); - rr_on_tcp_write(ctx, to_fdinfo); + // 4. A whole packet has been read, we will find someone to write it + fprintf(stderr, "No fd for URL %s in udp-read. Dropping packet :( \n", fdinfo->url); + mv_buffer_wtof (app_ctx, fdinfo); return 0; co_error: @@ -396,7 +409,8 @@ int rr_on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { struct waited_pkt* pkt = fdinfo->other; evt_core_rm_fd(ctx, fdinfo->fd); pkt->on = 0; - if (ring_lt(pkt->id, rr->recv_id)) return 1; + if (ring_gt (pkt->id, rr->recv_id_late)) rr->recv_id_late = pkt->id; + if (ring_le (pkt->id, rr->recv_id)) return 1; printf("Timer reached for packet %d\n", pkt->id); @@ -455,6 +469,8 @@ void algo_rr(struct evt_core_ctx* evt, struct algo_skel* as) { rr->my_links = 0xff; rr->remote_links = 0xff; rr->sent_id = 1; + rr->recv_id = 0; + rr->recv_id_late = 0; ctx->misc = rr; for (int i = 0; i < sizeof(ctx->bps) / sizeof(ctx->bps[0]); i++) { g_queue_push_tail(ctx->free_buffer, &(ctx->bps[i])); diff --git a/src/algo_utils.c b/src/algo_utils.c index f8b7a04..b483c46 100644 --- a/src/algo_utils.c +++ b/src/algo_utils.c @@ -21,6 +21,19 @@ void iterate2(int* fd, struct buffer_packet *bp, gpointer user_data) { fprintf(stderr, "fd=%d has a used_buffer entry\n", *fd); } +void debug_buffer(struct algo_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) { + fprintf(stderr, "No more free buffer for fd=%d.\n", fdinfo->fd); + int waiting_count = 0; + g_hash_table_foreach(app_ctx->write_waiting, (GHFunc)iterate, &waiting_count); + g_hash_table_foreach(app_ctx->used_buffer, (GHFunc)iterate2, NULL); + fprintf(stderr, "total_buffers=%d, free_buffer=%d, used_buffers=%d, app_buffer=%d, write_buffer=%d.\n", + PACKET_BUFFER_SIZE, + app_ctx->free_buffer->length, + g_hash_table_size(app_ctx->used_buffer), + g_hash_table_size(app_ctx->application_waiting), + waiting_count); +} + /** * Returns a buffer if available, NULL otherwise */ @@ -34,16 +47,8 @@ struct buffer_packet* get_read_buffer(struct algo_ctx *app_ctx, struct evt_core_ // 2. Get a new buffer otherwise bp = g_queue_pop_head(app_ctx->free_buffer); if (bp == NULL) { - fprintf(stderr, "No more free buffer for fd=%d.\n", fdinfo->fd); - int waiting_count = 0; - g_hash_table_foreach(app_ctx->write_waiting, (GHFunc)iterate, &waiting_count); - g_hash_table_foreach(app_ctx->used_buffer, (GHFunc)iterate2, NULL); - fprintf(stderr, "total_buffers=%d, free_buffer=%d, used_buffers=%d, app_buffer=%d, write_buffer=%d.\n", - PACKET_BUFFER_SIZE, - app_ctx->free_buffer->length, - g_hash_table_size(app_ctx->used_buffer), - g_hash_table_size(app_ctx->application_waiting), - waiting_count); + debug_buffer(app_ctx, fdinfo); + // 2.1 If no buffer is available, we subscribe to be notified later g_queue_push_tail (app_ctx->read_waiting, &(fdinfo->fd)); return NULL; @@ -165,6 +170,30 @@ void mv_buffer_atof(struct algo_ctx* app_ctx, void* from) { g_queue_push_tail (app_ctx->free_buffer, bp); } +void dup_buffer_tow(struct algo_ctx* app_ctx, struct buffer_packet* bp, struct evt_core_fdinfo* to) { + GQueue* q; + + // 1. We get a free buffer + struct buffer_packet* bp_dest = g_queue_pop_head(app_ctx->free_buffer); + if (bp_dest == NULL) { + debug_buffer(app_ctx, to); + return; + } + + // 2. We duplicate the data + memcpy(bp_dest, bp, sizeof(struct buffer_packet)); + + // 3. 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); + } + + // 4. We push the content to the appropriate destination + g_queue_push_tail(q, bp_dest); +} + struct buffer_packet* get_app_buffer(struct algo_ctx *app_ctx, void* idx) { return g_hash_table_lookup (app_ctx->application_waiting, idx); } diff --git a/src/algo_utils.h b/src/algo_utils.h index 815e2b8..3fc55cc 100644 --- a/src/algo_utils.h +++ b/src/algo_utils.h @@ -24,6 +24,7 @@ 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); +void dup_buffer_tow(struct algo_ctx* app_ctx, struct buffer_packet* bp, struct evt_core_fdinfo* to); 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); diff --git a/src/net_tools.c b/src/net_tools.c index cf1c419..18b5890 100644 --- a/src/net_tools.c +++ b/src/net_tools.c @@ -44,7 +44,15 @@ int create_ip_client(char* host, char* service, int type) { } int create_tcp_client(char* host, char* service) { - return create_ip_client (host, service, SOCK_STREAM); + int sock = create_ip_client (host, service, SOCK_STREAM); + int activate = 1; + int err; + err = setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &activate, sizeof(activate)); + if (err < 0) { + perror("setsockopt TCP_NODELAY"); + exit(EXIT_FAILURE); + } + return sock; } int create_udp_client(char* host, char* service) { @@ -93,7 +101,15 @@ int create_ip_server(char* host, char* service, int type) { } int create_tcp_server(char* host, char* service) { - return create_ip_server (host, service, SOCK_STREAM); + int sock = create_ip_server (host, service, SOCK_STREAM); + int activate = 1; + int err; + err = setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &activate, sizeof(activate)); + if (err < 0) { + perror("setsockopt TCP_NODELAY"); + exit(EXIT_FAILURE); + } + return sock; } int create_udp_server(char* host, char* service) { diff --git a/src/net_tools.h b/src/net_tools.h index a847b9a..0975ebd 100644 --- a/src/net_tools.h +++ b/src/net_tools.h @@ -7,6 +7,8 @@ #include #include #include +#include +#include int create_tcp_client(char* host, char* service); int create_udp_client(char* host, char* service);