From 3b3c5b6dbd2d7d783887f38d86cd287548bc674f Mon Sep 17 00:00:00 2001 From: Quentin Dufour Date: Thu, 14 Feb 2019 11:16:38 +0100 Subject: [PATCH] Huge refactor work in progress --- src/algo_naive.c | 91 ++++++++++++++++++++++++++++++++++++++---------- src/algo_skel.h | 39 +++++++++++++++++++++ src/utils.h | 2 +- 3 files changed, 113 insertions(+), 19 deletions(-) diff --git a/src/algo_naive.c b/src/algo_naive.c index 6657d4a..3b45d61 100644 --- a/src/algo_naive.c +++ b/src/algo_naive.c @@ -3,6 +3,8 @@ struct naive_ctx { int ref_count; + struct buffer_packet tcp_to_udp; + struct buffer_packet udp_to_tcp; struct ring_buffer tcp_to_udp_rb; struct ring_buffer udp_to_tcp_rb; struct sockaddr_in udp_addr; @@ -39,6 +41,50 @@ co_error: exit(EXIT_FAILURE); } +enum FD_STATE read_packet_from_tcp(int fd, struct buffer_packet* bp) { + int nread; + size_t pkt_size_size = sizeof(bp->ip.ap.str.size); + if (bp->mode == BP_WRITING) return FDS_ERR; + + while (bp->aread < pkt_size_size) { + nread = read(fd, &(bp->ip.ap.raw) + bp->aread, pkt_size_size - bp->aread); + if (nread == 0) return FDS_AGAIN; + if (nread == -1 && errno == EAGAIN) return FDS_AGAIN; + if (nread == -1) return FDS_ERR; + bp->aread += nread; + } + + while (bp->aread < bp->ip.ap.str.size) { + nread = read(fd, &(bp->ip.ap.raw) + bp->aread, bp->ip.ap.str.size - bp->aread); + if (nread == 0) return FDS_AGAIN; + if (nread == -1 && errno == EAGAIN) return FDS_AGAIN; + if (nread == -1) return FDS_ERR; + bp->aread += nread; + } + bp->mode = BP_WRITING; + + return FDS_READY; +} + +enum FD_STATE write_packet_to_udp(int fd, struct buffer_packet* bp, struct sockaddr* addr, socklen_t addrlen) { + int nwrite, bytes_to_send; + size_t pkt_size_size = sizeof(bp->ip.ap.str.size); + + if (bp->mode == BP_READING) return FDS_ERR; + + bytes_to_send = bp->ip.ap.str.size - pkt_size_size; + nwrite = sendto(fd, + &(bp->ip.ap.str.payload), + bytes_to_send, + 0, + addr, + addrlen); + if (nwrite == -1 && errno == EAGAIN) return FDS_AGAIN; + if (nwrite != bytes_to_send) return FDS_ERR; + + return FDS_READY; +} + void tcp_to_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { // Get target file descriptor struct evt_core_cat* udp = g_hash_table_lookup (ctx->catlist, "udp-write"); @@ -47,23 +93,28 @@ 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->tcp_to_udp_rb); - char buffer[RING_BUFFER_SIZE]; - int nread, nwrite, rb_free_space; + struct buffer_packet* bp = &(app_ctx->tcp_to_udp); + int read_res = FDS_READY; + int write_res = FDS_READY; - 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 - 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 - ring_buffer_write(rb, buffer, nread); // Persist read data in our buffer + // Either we can read something, either we can write something + while ((read_res != FDS_AGAIN && bp->mode == BP_READING) + || (write_res != FDS_AGAIN && bp->mode == BP_WRITING)) { - nread = ring_buffer_read(rb, buffer, RING_BUFFER_SIZE); - 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); + // 1. Read packet from TCP socket + if (bp->mode == BP_READING) { + read_res = read_packet_from_tcp (fd, bp); + if (read_res == FDS_ERR) goto co_error; + } + + // 2. Write packet to UDP socket + if (bp->mode == BP_WRITING) { + write_res = write_packet_to_udp(udp_fd, + bp, + (struct sockaddr*) &(app_ctx->udp_addr), + app_ctx->udp_addrlen); + if (read_res == FDS_ERR) goto co_error; + } } return; @@ -81,17 +132,21 @@ 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->udp_to_tcp_rb); - char buffer[RING_BUFFER_SIZE]; + char buffer[NET_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 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 + nread = recvfrom(fd, buffer, rb_free_space, MSG_TRUNC, (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 - ring_buffer_write(rb, buffer, nread); // Persist read data in our buffer + if (nread > rb_free_space) { + fprintf(stderr, "[warn!] we partially read a UDP packet\n"); + } else { + ring_buffer_write(rb, buffer, nread); // Persist read data in our buffer + } nread = ring_buffer_read(rb, buffer, RING_BUFFER_SIZE); nwrite = write(tcp_fd, buffer, nread); diff --git a/src/algo_skel.h b/src/algo_skel.h index 4dbdfba..9f345c4 100644 --- a/src/algo_skel.h +++ b/src/algo_skel.h @@ -6,6 +6,45 @@ #include "evt_core.h" #include "utils.h" +/* + * man 7 udp about receive operation on UDP sockets: + * + * > All receive operations return only one packet. When the packet is smaller than the passed + * > buffer, only that much data is returned; when it is bigger, the packet is truncated and the + * > MSG_TRUNC flag is set. MSG_WAITALL is not supported. + */ + +enum FD_STATE { + FDS_READY, + FDS_AGAIN, + FDS_ERR +}; + +enum BP_MODE { + BP_READING, + BP_WRITING +}; + +union abstract_packet { + char raw; + struct { + uint16_t size; + char payload; + } str; +}; + +struct internet_packet { + union abstract_packet ap; + char rest[1499]; // MTU = 1500, 1 byte in the union +}; + +struct buffer_packet { + uint8_t mode; + uint16_t aread; + uint16_t awrite; + struct internet_packet ip; +}; + struct algo_skel { struct evt_core_cat on_udp_read; struct evt_core_cat on_tcp_read; diff --git a/src/utils.h b/src/utils.h index a787232..d30250f 100644 --- a/src/utils.h +++ b/src/utils.h @@ -5,7 +5,7 @@ #include // 1500 = internet MTU -#define RING_BUFFER_SIZE 1500 +#define RING_BUFFER_SIZE 1500*10 struct ring_buffer { char buffer[RING_BUFFER_SIZE];