diff --git a/CMakeLists.txt b/CMakeLists.txt index 6808b0c..e15c728 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -21,6 +21,8 @@ list(APPEND CSOURCES src/algo_naive.c src/utils.h src/utils.c + src/packet.h + src/packet.c ) add_executable(donar ${CSOURCES} src/donar.c) diff --git a/src/algo_naive.c b/src/algo_naive.c index 3b45d61..f6be091 100644 --- a/src/algo_naive.c +++ b/src/algo_naive.c @@ -1,12 +1,9 @@ #include "algo_skel.h" -#define NAIVE_BUFFER 128 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; socklen_t udp_addrlen; }; @@ -41,50 +38,6 @@ 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"); @@ -109,8 +62,7 @@ void tcp_to_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { // 2. Write packet to UDP socket if (bp->mode == BP_WRITING) { - write_res = write_packet_to_udp(udp_fd, - bp, + 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; @@ -131,29 +83,27 @@ 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[NET_BUFFER_SIZE]; - int nread, nwrite, rb_free_space; + struct buffer_packet* bp = &(app_ctx->udp_to_tcp); + 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 - app_ctx->udp_addrlen = sizeof(struct sockaddr_in); - 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 - 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 + // 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)) { + + // 1. Read packet from UDP socket + if (bp->mode == BP_READING) { + read_res = read_packet_from_udp (fd, bp, + (struct sockaddr*) &(app_ctx->udp_addr), + &(app_ctx->udp_addrlen)); + if (read_res == FDS_ERR) goto co_error; } - nread = ring_buffer_read(rb, buffer, RING_BUFFER_SIZE); - nwrite = write(tcp_fd, buffer, nread); - if (nwrite == -1 && errno == EAGAIN) return; - if (nwrite == -1) goto co_error; - printf("written to tcp_fd=%d\n", nwrite); - ring_buffer_ack_read (rb, nwrite); + // 2. Write packet to TCP socket + if (bp->mode == BP_WRITING) { + write_res = write_packet_to_tcp(fd, bp); + if (write_res == FDS_ERR) goto co_error; + } } return; diff --git a/src/algo_skel.h b/src/algo_skel.h index 9f345c4..e4a21ef 100644 --- a/src/algo_skel.h +++ b/src/algo_skel.h @@ -3,48 +3,10 @@ #include #include #include +#include "packet.h" #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/packet.c b/src/packet.c new file mode 100644 index 0000000..4aca9e3 --- /dev/null +++ b/src/packet.c @@ -0,0 +1,90 @@ +#include "packet.h" + +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; + bp->awrite = 0; + + return FDS_READY; +} + +enum FD_STATE write_packet_to_tcp(int fd, struct buffer_packet* bp) { + int nwrite; + + while (bp->awrite < bp->ip.ap.str.size) { + nwrite = send(fd, &(bp->ip.ap.raw), bp->ip.ap.str.size, 0); + if (nwrite == -1 && errno == EAGAIN) return FDS_AGAIN; + if (nwrite == -1) return FDS_ERR; + bp->awrite += nwrite; + } + + bp->mode = BP_READING; + bp->aread = 0; + + 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; + + bp->mode = BP_READING; + bp->aread = 0; + + return FDS_READY; +} + +enum FD_STATE read_packet_from_udp (int fd, struct buffer_packet* bp, struct sockaddr* addr, socklen_t* addrlen) { + int nread; + if (bp->mode == BP_WRITING) return FDS_ERR; + + *addrlen = sizeof(struct sockaddr_in); + size_t pkt_size_size = sizeof(bp->ip.ap.str.size); + size_t udp_packet_size = sizeof(struct internet_packet) - pkt_size_size; + + nread = recvfrom(fd, &(bp->ip.ap.str.payload), udp_packet_size, MSG_TRUNC, + addr, addrlen); + + if (nread > udp_packet_size) return FDS_ERR; + if (nread == -1 && errno == EAGAIN) return FDS_AGAIN; + if (nread == -1) return FDS_ERR; + + bp->ip.ap.str.size = nread + pkt_size_size; + + bp->mode = BP_WRITING; + bp->awrite = 0; + + return FDS_READY; +} + diff --git a/src/packet.h b/src/packet.h new file mode 100644 index 0000000..c947efe --- /dev/null +++ b/src/packet.h @@ -0,0 +1,53 @@ +#pragma once +#include +#include +#include +#include +#include +#include +#include +#include + +/* + * 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; +}; + +enum FD_STATE read_packet_from_tcp(int fd, struct buffer_packet* bp); +enum FD_STATE write_packet_to_tcp(int fd, struct buffer_packet* bp); +enum FD_STATE write_packet_to_udp(int fd, struct buffer_packet* bp, struct sockaddr* addr, socklen_t addrlen); +enum FD_STATE read_packet_from_udp (int fd, struct buffer_packet* bp, struct sockaddr* addr, socklen_t* addrlen);