diff --git a/src/algo_naive.c b/src/algo_naive.c index a622061..6657d4a 100644 --- a/src/algo_naive.c +++ b/src/algo_naive.c @@ -2,13 +2,18 @@ #define NAIVE_BUFFER 128 struct naive_ctx { - struct ring_buffer rb; + int ref_count; + struct ring_buffer tcp_to_udp_rb; + struct ring_buffer udp_to_tcp_rb; + struct sockaddr_in udp_addr; + socklen_t udp_addrlen; }; void free_nothing(void* app_ctx) {} - void free_naive(void* app_ctx) { - if (app_ctx != NULL) free(app_ctx); + struct naive_ctx* ctx = (struct naive_ctx*) app_ctx; + ctx->ref_count--; + if (ctx->ref_count <= 0) free(ctx); } void on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { @@ -42,7 +47,7 @@ void tcp_to_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { // Init data structures for the transfer struct naive_ctx* app_ctx = cat->app_ctx; - struct ring_buffer* rb = &(app_ctx->rb); + struct ring_buffer* rb = &(app_ctx->tcp_to_udp_rb); char buffer[RING_BUFFER_SIZE]; int nread, nwrite, rb_free_space; @@ -55,7 +60,7 @@ void tcp_to_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { ring_buffer_write(rb, buffer, nread); // Persist read data in our buffer nread = ring_buffer_read(rb, buffer, RING_BUFFER_SIZE); - nwrite = write(udp_fd, buffer, nread); + nwrite = sendto(udp_fd, buffer, nread, 0, (struct sockaddr*)&(app_ctx->udp_addr), app_ctx->udp_addrlen); if (nwrite == -1 && errno == EAGAIN) return; if (nwrite == -1) goto co_error; ring_buffer_ack_read (rb, nwrite); @@ -75,13 +80,14 @@ void udp_to_tcp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { // Init data structures for the transfer struct naive_ctx* app_ctx = cat->app_ctx; - struct ring_buffer* rb = &(app_ctx->rb); + struct ring_buffer* rb = &(app_ctx->udp_to_tcp_rb); char buffer[RING_BUFFER_SIZE]; int nread, nwrite, rb_free_space; while (1) { rb_free_space = ring_buffer_free_space (rb); // We can't afford to read more - nread = read(fd, buffer, rb_free_space); // Effective read + app_ctx->udp_addrlen = sizeof(struct sockaddr_in); + nread = recvfrom(fd, buffer, rb_free_space, 0, (struct sockaddr*)&(app_ctx->udp_addr), &(app_ctx->udp_addrlen)); // Effective read if (nread == 0) return; // End of file if (nread == -1 && errno == EAGAIN) return; // No more data to read if (nread == -1) goto co_error; // A bad error @@ -103,44 +109,49 @@ co_error: } void algo_naive(struct algo_skel* as) { + struct naive_ctx* ctx = malloc(sizeof(struct naive_ctx)); + if (ctx == NULL) goto init_err; + memset(ctx, 0, sizeof(struct naive_ctx)); + ctx->udp_addrlen = sizeof(struct sockaddr_in); + as->on_tcp_co.name = "tcp-listen"; as->on_tcp_co.flags = EPOLLIN; as->on_tcp_co.app_ctx = NULL; - as->on_tcp_co.free_app_ctx = free_naive; + as->on_tcp_co.free_app_ctx = free_nothing; as->on_tcp_co.cb = on_tcp_co; as->on_tcp_co.socklist = NULL; as->on_tcp_read.name = "tcp-read"; as->on_tcp_read.flags = EPOLLIN | EPOLLET | EPOLLRDHUP; - as->on_tcp_read.app_ctx = malloc(sizeof(struct naive_ctx)); + as->on_tcp_read.app_ctx = ctx; as->on_tcp_read.free_app_ctx = free_naive; as->on_tcp_read.cb = tcp_to_udp; as->on_tcp_read.socklist = NULL; - if (as->on_tcp_read.app_ctx == NULL) goto init_err; - memset(as->on_tcp_read.app_ctx, 0, sizeof(struct naive_ctx)); + ctx->ref_count++; as->on_udp_read.name = "udp-read"; as->on_udp_read.flags = EPOLLIN | EPOLLET; - as->on_udp_read.app_ctx = malloc(sizeof(struct naive_ctx)); + as->on_udp_read.app_ctx = ctx; as->on_udp_read.free_app_ctx = free_naive; as->on_udp_read.cb = udp_to_tcp; as->on_udp_read.socklist = NULL; - if (as->on_udp_read.app_ctx == NULL) goto init_err; - memset(as->on_udp_read.app_ctx, 0, sizeof(struct naive_ctx)); + ctx->ref_count++; as->on_tcp_write.name = "tcp-write"; as->on_tcp_write.flags = EPOLLOUT | EPOLLET | EPOLLRDHUP; - as->on_tcp_write.app_ctx = as->on_udp_read.app_ctx; - as->on_tcp_write.free_app_ctx = free_nothing; + as->on_tcp_write.app_ctx = ctx; + as->on_tcp_write.free_app_ctx = free_naive; as->on_tcp_write.cb = udp_to_tcp; as->on_tcp_write.socklist = NULL; + ctx->ref_count++; as->on_udp_write.name = "udp-write"; as->on_udp_write.flags = EPOLLOUT | EPOLLET; - as->on_udp_write.app_ctx = as->on_tcp_read.app_ctx; - as->on_udp_write.free_app_ctx = free_nothing; + as->on_udp_write.app_ctx = ctx; + as->on_udp_write.free_app_ctx = free_naive; as->on_udp_write.cb = tcp_to_udp; as->on_udp_write.socklist = NULL; + ctx->ref_count++; return; init_err: