diff --git a/src/algo_rr.c b/src/algo_rr.c index b748948..d54e165 100644 --- a/src/algo_rr.c +++ b/src/algo_rr.c @@ -23,6 +23,7 @@ struct rr_ctx { uint16_t recv_id; uint16_t sent_id; uint8_t current_link; + struct timespec emit_time; struct deferred_pkt real[PACKET_BUFFER_SIZE]; struct waited_pkt wait[PACKET_BUFFER_SIZE]; }; @@ -202,6 +203,8 @@ void rr_pkt_unroll(struct evt_core_ctx* ctx, struct algo_ctx* app_ctx) { } } +//------ + int rr_on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { struct buffer_packet* bp; struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; @@ -252,6 +255,97 @@ co_error: exit(EXIT_FAILURE); } +int rr_on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { + struct buffer_packet* bp; + struct evt_core_fdinfo *to_fdinfo; + struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; + struct rr_ctx* rr = app_ctx->misc; + 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. Read packet from socket + bp->ip.ap.str.port = url_get_port_int (fdinfo->url); + read_res = read_packet_from_udp (fdinfo->fd, bp, fdinfo->other); + if (read_res == FDS_ERR) goto co_error; + if (read_res == FDS_AGAIN) return 1; + + // 3. Prepare RR state and packet values + struct timespec curr; + int secs, nsecs; + uint64_t micro_sec; + + if (clock_gettime(CLOCK_MONOTONIC, &curr) == -1){ + perror("clock_gettime error"); + exit(EXIT_FAILURE); + } + + secs = curr.tv_sec - rr->emit_time.tv_sec; + nsecs = curr.tv_nsec - rr->emit_time.tv_nsec; + micro_sec = secs * 1000000 + nsecs / 1000; + if (micro_sec > rr->mjit) micro_sec = rr->mjit; + + bp->ip.ap.str.id = rr->sent_id; + bp->ip.ap.str.flags = 0; + bp->ip.ap.str.deltat = micro_sec; + bp->ip.ap.str.bitfield = rr->remote_links; + bp->ip.ap.str.prevlink = rr->current_link; + + do { + rr->current_link = (rr->current_link) + 1 % 10; + } while (!(rr->my_links & (1 << rr->current_link))); + rr->emit_time = curr; + rr->sent_id++; + + // 4. A whole packet has been read, we will find someone to write it + sprintf(url, "tcp:write:127.0.0.1:%d", 7500 + rr->current_link); //@FIXME Hardcoded + 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); + 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); + rr_on_tcp_write(ctx, to_fdinfo); + + return 0; + +co_error: + perror("Failed to UDP read"); + exit(EXIT_FAILURE); +} + +int rr_on_tcp_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 data from the buffer to the socket + while (bp->mode == BP_WRITING) { + write_res = write_packet_to_tcp(fdinfo->fd, bp); + 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 TCP write"); + exit(EXIT_FAILURE); +} + int rr_on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { struct algo_ctx* app_ctx = fdinfo->cat->app_ctx; struct rr_ctx* rr = app_ctx->misc; @@ -327,23 +421,22 @@ void algo_rr(struct evt_core_ctx* evt, struct algo_skel* as) { as->on_tcp_read.err_cb = rr_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; + as->on_udp_read.cb = rr_on_udp_read; + as->on_udp_read.err_cb = rr_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; + as->on_tcp_write.cb = rr_on_tcp_write; + as->on_tcp_write.err_cb = rr_on_err; ctx->ref_count++; -*/ as->on_udp_write.name = "udp-write"; as->on_udp_write.flags = EPOLLOUT | EPOLLET;