diff --git a/src/algo_rr.c b/src/algo_rr.c index 6404eca..165ac81 100644 --- a/src/algo_rr.c +++ b/src/algo_rr.c @@ -1,6 +1,145 @@ +#include #include "algo_skel.h" #include "algo_utils.h" +int rr_on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { + int conn_sock1, conn_sock2; + struct sockaddr_in addr; + socklen_t in_len; + char url[1024], port[6]; + struct evt_core_cat local_cat = {0}; + struct evt_core_fdinfo to_fdinfo = {0}; + to_fdinfo.cat = &local_cat; + to_fdinfo.url = url; + + in_len = sizeof(addr); + conn_sock1 = accept(fdinfo->fd, (struct sockaddr*)&addr, &in_len); + + if (conn_sock1 == -1) goto co_error; + conn_sock2 = dup(conn_sock1); + if (conn_sock2 == -1) goto co_error; + //printf("fd=%d accepts, creating fds=%d,%d\n", fd, conn_sock1, conn_sock2); + + url_get_port(port, fdinfo->url); + + to_fdinfo.fd = conn_sock1; + to_fdinfo.cat->name = "tcp-read"; + sprintf(to_fdinfo.url, "tcp:read:127.0.0.1:%s", port); + evt_core_add_fd (ctx, &to_fdinfo); + + to_fdinfo.fd = conn_sock2; + to_fdinfo.cat->name = "tcp-write"; + sprintf(to_fdinfo.url, "tcp:write:127.0.0.1:%s", port); + evt_core_add_fd (ctx, &to_fdinfo); + + return 1; + +co_error: + perror("Failed to handle new connection"); + exit(EXIT_FAILURE); +} + +void set_timeout(struct evt_core_ctx* evts, uint64_t micro_sec) { + struct timespec now; + struct itimerspec timer_config; + char url[1024]; + struct evt_core_cat cat = {0}; + struct evt_core_fdinfo fdinfo = {0}; + fdinfo.cat = &cat; + fdinfo.url = url; + + if (clock_gettime(CLOCK_REALTIME, &now) == -1) { + perror("clock_gettime"); + exit(EXIT_FAILURE); + } + + timer_config.it_value.tv_sec = now.tv_sec + micro_sec / 1000; + timer_config.it_value.tv_nsec = now.tv_nsec + micro_sec % 1000 * 1000000; + timer_config.it_interval.tv_sec = 60; + timer_config.it_interval.tv_nsec = 0; + + fdinfo.fd = timerfd_create(CLOCK_REALTIME, 0); + if (fdinfo.fd == -1) { + perror("Unable to timerfd_create"); + exit(EXIT_FAILURE); + } + if (timerfd_settime (fdinfo.fd, TFD_TIMER_ABSTIME, &timer_config, NULL) == -1) { + perror("Unable to timerfd_time"); + exit(EXIT_FAILURE); + } + fdinfo.cat->name = "timeout"; + fdinfo.other = NULL; + fdinfo.free_other = NULL; + sprintf(fdinfo.url, "timer:%ld:1", 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) { + struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; + + // 1. Update links I can used thanks to target feedback + if (bp->ip.ap.str.bitfield ^ app_ctx->my_bitfield) { + update_my_bitfield(bp, app_ctx); + } + + // 2. If packet arrived too late + if (app_ctx->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); + } + + // 3. If packet arrived too early + else if (app_ctx->recv_id < bp->ip.ap.str.id - 1) { + int64_t timeout = app_ctx->mjit - (int64_t) bp->ip.ap.str.deltat; + if (timeout <= 0) timeout = 0; + //bp->ip.ap.str.id; + set_timeout(ctx, timeout); + } +} + +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; + + // 2. Try to read a whole packet in the buffer + while (bp->mode == BP_READING) { + read_res = read_packet_from_tcp (fdinfo->fd, bp); + if (read_res == FDS_ERR) goto co_error; + if (read_res == FDS_AGAIN) return 1; + } + + // 3. Logic on packet + rr_handle_recv(bp, app_ctx, fdinfo); + + /* + // 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); +} + void algo_rr(struct algo_skel* as) { struct algo_ctx* ctx = malloc(sizeof(struct algo_ctx)); if (ctx == NULL) goto init_err; diff --git a/src/packet.h b/src/packet.h index eb3dc29..7a0c72a 100644 --- a/src/packet.h +++ b/src/packet.h @@ -34,6 +34,9 @@ union abstract_packet { uint16_t size; uint16_t port; uint8_t id; + uint8_t bitfield; + uint8_t prevlink; + uint16_t deltat; char payload; } str; };