#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; // Should put the duration, the file descriptor, the packet id 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; set_timeout(ctx, timeout); app_ctx->waiting[bp->ip.ap.str.id % LINK_COUNT] = bp; // should store more than that } // 4. If we were waiting this packet else { if (bp->ip.ap.str.flags & PKT_TIMEOUT) broken_rlink(fdinfo); else if (bp->ip.ap.str.flags & PKT_CONTROL) working_rlink(fdinfo); else deliver(bp); app_ctx->recv_id = bp->ip.ap.str.id; int next = app_ctx->recv_id+1; if (app_ctx->waiting[next % LINK_COUNT] != NULL) { bp = app_ctx->waiting[next % LINK_COUNT]; app_ctx->waiting[next % LINK_COUNT] = NULL; rr_handle_recv(ctx,bp,NULL); } } } 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); } struct deferred_pkt { int link_fd; struct buffer_packet* bp; }; struct rr_ctx { uint16_t mjit; uint16_t recv_id; uint16_t sent_id; struct deferred_pkt real[LINK_COUNT]; struct deferred_pkt stub[LINK_COUNT]; }; void algo_rr(struct algo_skel* as) { struct algo_ctx* ctx = malloc(sizeof(struct algo_ctx)); if (ctx == NULL) goto init_err; memset(ctx, 0, sizeof(struct algo_ctx)); ctx->free_buffer = g_queue_new (); ctx->read_waiting = g_queue_new (); 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); ctx->misc = malloc(sizeof(struct rr_ctx)); for (int i = 0; i < sizeof(ctx->bps) / sizeof(ctx->bps[0]); i++) { g_queue_push_tail(ctx->free_buffer, &(ctx->bps[i])); } /* as->on_tcp_co.name = "tcp-listen"; as->on_tcp_co.flags = EPOLLIN; as->on_tcp_co.free_app_ctx = free_nothing; as->on_tcp_co.cb = on_tcp_co; as->on_tcp_read.name = "tcp-read"; as->on_tcp_read.flags = EPOLLIN | EPOLLET | EPOLLRDHUP; as->on_tcp_read.app_ctx = ctx; as->on_tcp_read.free_app_ctx = free_naive; as->on_tcp_read.cb = on_tcp_read; as->on_tcp_read.err_cb = on_err; ctx->ref_count++; as->on_udp_read.name = "udp-read"; as->on_udp_read.flags = EPOLLIN | EPOLLET; as->on_udp_read.app_ctx = ctx; as->on_udp_read.free_app_ctx = free_naive; as->on_udp_read.cb = on_udp_read; as->on_udp_read.err_cb = on_err; ctx->ref_count++; as->on_tcp_write.name = "tcp-write"; as->on_tcp_write.flags = EPOLLOUT | EPOLLET | EPOLLRDHUP; as->on_tcp_write.app_ctx = ctx; as->on_tcp_write.free_app_ctx = free_naive; as->on_tcp_write.cb = on_tcp_write; as->on_tcp_write.err_cb = on_err; ctx->ref_count++; as->on_udp_write.name = "udp-write"; as->on_udp_write.flags = EPOLLOUT | EPOLLET; as->on_udp_write.app_ctx = ctx; as->on_udp_write.free_app_ctx = free_naive; as->on_udp_write.cb = on_udp_write; as->on_udp_write.err_cb = on_err; ctx->ref_count++; */ return; init_err: fprintf(stderr, "Failed to init algo naive\n"); exit(EXIT_FAILURE); }