From 0e3a2189f65354962fc71af41f5aa52261a782e8 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Sun, 25 Oct 2026 22:30:04 +0100 Subject: Replace libowfat's iobatch with our own smaller (and probably only temporary) ot_batch --- Makefile | 10 +- opentracker.c | 30 ++--- ot_batch.c | 158 +++++++++++++++++++++++++ ot_batch.h | 48 ++++++++ ot_http.c | 55 +++------ ot_http.h | 7 +- proxy.c | 14 +-- tests/runner.c | 1 + tests/t_batch.c | 348 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ tests/test.h | 1 + trackerlogic.c | 3 - 11 files changed, 592 insertions(+), 83 deletions(-) create mode 100644 ot_batch.c create mode 100644 ot_batch.h create mode 100644 tests/t_batch.c diff --git a/Makefile b/Makefile index 00a9f93..c7105d4 100644 --- a/Makefile +++ b/Makefile @@ -68,9 +68,9 @@ CFLAGS+=-I$(LIBOWFAT_HEADERS) -DGIT_VERSION=$(GIT_VERSION) -Wall -pipe -pthread LDFLAGS+=-L$(LIBOWFAT_LIBRARY) -lowfat -pthread BINARY =opentracker -HEADERS=trackerlogic.h scan_urlencoded_query.h ot_buf.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h -SOURCES=opentracker.c trackerlogic.c scan_urlencoded_query.c ot_buf.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c -SOURCES_proxy=proxy.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c +HEADERS=trackerlogic.h scan_urlencoded_query.h ot_batch.h ot_buf.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h +SOURCES=opentracker.c trackerlogic.c scan_urlencoded_query.c ot_batch.c ot_buf.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c +SOURCES_proxy=proxy.c ot_batch.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c OBJECTS = $(SOURCES:%.c=%.o) OBJECTS_debug = $(SOURCES:%.c=%.debug.o) @@ -119,8 +119,8 @@ $(OBJECTS_proxy_debug): $(HEADERS) # has to. tests/stubs.c supplies the few symbols those modules reference but # the tests never reach. TESTBIN=tests/runner -TEST_SOURCES=tests/runner.c tests/test.c tests/stubs.c tests/t_scan_urlencoded_query.c tests/t_iovec.c tests/t_vector.c tests/t_random.c tests/t_hex.c tests/t_accesslist.c tests/t_scan.c tests/t_buf.c tests/t_ip.c tests/t_socket.c tests/t_mutex.c -TEST_UNITS=scan_urlencoded_query.c ot_buf.c ot_ip.c ot_socket.c ot_iovec.c ot_vector.c ot_random.c ot_hex.c ot_scan.c ot_accesslist.c ot_mutex.c +TEST_SOURCES=tests/runner.c tests/test.c tests/stubs.c tests/t_scan_urlencoded_query.c tests/t_iovec.c tests/t_vector.c tests/t_random.c tests/t_hex.c tests/t_accesslist.c tests/t_scan.c tests/t_buf.c tests/t_ip.c tests/t_socket.c tests/t_mutex.c tests/t_batch.c +TEST_UNITS=scan_urlencoded_query.c ot_batch.c ot_buf.c ot_ip.c ot_socket.c ot_iovec.c ot_vector.c ot_random.c ot_hex.c ot_scan.c ot_accesslist.c ot_mutex.c TEST_HEADERS=tests/test.h $(HEADERS) # Tests reach code that sits behind feature flags, so the units under test are # built with those features enabled. A test binary is not a production binary. diff --git a/opentracker.c b/opentracker.c index 96a265a..50fff29 100644 --- a/opentracker.c +++ b/opentracker.c @@ -23,7 +23,6 @@ /* Libowfat */ #include "io.h" -#include "iob.h" /* Opentracker */ #include "ot_accesslist.h" @@ -120,9 +119,8 @@ static ssize_t header_complete(char *request, ssize_t byte_count) { static void handle_dead(const int64_t sock) { struct http_data *cookie = io_getcookie(sock); if (cookie) { - size_t i; - for (i = 0; i < cookie->batches; ++i) - iob_reset(cookie->batch + i); + if (cookie->batch) + batch_reset(cookie->batch); free(cookie->batch); buf_reset(&cookie->request); if (cookie->flag & (STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER)) @@ -178,33 +176,23 @@ static void handle_read(const int64_t sock, struct ot_workstruct *ws) { static void handle_write(const int64_t sock) { struct http_data *cookie = io_getcookie(sock); - size_t i; int chunked = 0; - /* Look for the first io_batch still containing bytes to write */ if (cookie) { if (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER) chunked = 1; - for (i = 0; i < cookie->batches; ++i) { - if (cookie->batch[i].bytesleft) { - int64_t res = iob_send(sock, cookie->batch + i); - - if (res == -3) { - handle_dead(sock); - return; - } - - if (!cookie->batch[i].bytesleft) - continue; - - if (res == -1 || res > 0 || i < cookie->batches - 1) - return; + if (cookie->batch) { + if (batch_send(sock, cookie->batch) == BATCH_FAILED) { + handle_dead(sock); + return; } + if (batch_pending(cookie->batch)) + return; } } - /* In a chunked transfer after all batches accumulated have been sent, wait for the next one */ + /* In a chunked transfer, wait for the next piece once everything queued has gone out */ if (chunked) io_dontwantwrite(sock); else diff --git a/ot_batch.c b/ot_batch.c new file mode 100644 index 0000000..84eb4a8 --- /dev/null +++ b/ot_batch.c @@ -0,0 +1,158 @@ +/* This software was written by Dirk Engling + It is considered beerware. Prost. Skol. Cheers or whatever. + + $id$ */ + +/* System */ +#include +#include +#include +#include +#include +#include +#include + +/* Opentracker */ +#include "ot_batch.h" + +#ifndef IOV_MAX +#define IOV_MAX 1024 +#endif + +#define OT_BATCH_MIN_ENTRIES 16 + +static int batch_room_for(ot_batch *batch, size_t wanted) { + size_t space = batch->space ? batch->space : OT_BATCH_MIN_ENTRIES; + struct iovec *grown; + + if (batch->count + wanted <= batch->space) + return 0; + + /* During a chunked transfer entries are appended while earlier ones are + still going out, so the front of the array frees up as we go. Slide the + rest down before asking for more. */ + if (batch->first && batch->count - batch->first + wanted <= batch->space) { + memmove(batch->entries, batch->entries + batch->first, (batch->count - batch->first) * sizeof(struct iovec)); + batch->count -= batch->first; + batch->first = 0; + return 0; + } + + while (space < batch->count + wanted) + space *= 2; + + if (!(grown = realloc(batch->entries, space * sizeof(struct iovec)))) + return -1; + + batch->entries = grown; + batch->space = space; + return 0; +} + +int batch_add(ot_batch *batch, void *buf, size_t len) { + if (batch_room_for(batch, 1)) { + free(buf); + return -1; + } + + batch->entries[batch->count].iov_base = buf; + batch->entries[batch->count].iov_len = len; + ++batch->count; + return 0; +} + +int batch_add_vector(ot_batch *batch, const struct iovec *entries, size_t count) { + if (batch_room_for(batch, count)) { + while (count--) + free(entries[count].iov_base); + return -1; + } + + memcpy(batch->entries + batch->count, entries, count * sizeof(struct iovec)); + batch->count += count; + return 0; +} + +/* Drop everything the kernel accepted, freeing each buffer as it goes */ +static void batch_consume(ot_batch *batch, size_t written) { + while (written) { + struct iovec *entry = batch->entries + batch->first; + size_t left = entry->iov_len - batch->offset; + + if (written < left) { + batch->offset += written; + return; + } + + free(entry->iov_base); + written -= left; + batch->offset = 0; + ++batch->first; + } + + /* Nothing is outstanding, so the array can start over from the front */ + if (batch->first == batch->count) + batch->first = batch->count = 0; +} + +/* A peer that has gone away must come back as an error rather than as a + signal that kills the tracker, so ask the kernel not to raise one where it + can be asked. macOS cannot, and relies on SIGPIPE being blocked. */ +static ssize_t batch_writev(int s, struct iovec *entries, int count) { +#ifdef MSG_NOSIGNAL + struct msghdr msg; + + memset(&msg, 0, sizeof(msg)); + msg.msg_iov = entries; + msg.msg_iovlen = count; + return sendmsg(s, &msg, MSG_NOSIGNAL); +#else + return writev(s, entries, count); +#endif +} + +ssize_t batch_send(int s, ot_batch *batch) { + size_t entries = batch->count - batch->first; + ssize_t written; + + if (!batch_pending(batch)) + return BATCH_DONE; + + if (entries > IOV_MAX) + entries = IOV_MAX; + + /* The first entry may be half way out already */ + batch->entries[batch->first].iov_base = (char *)batch->entries[batch->first].iov_base + batch->offset; + batch->entries[batch->first].iov_len -= batch->offset; + written = batch_writev(s, batch->entries + batch->first, (int)entries); + batch->entries[batch->first].iov_base = (char *)batch->entries[batch->first].iov_base - batch->offset; + batch->entries[batch->first].iov_len += batch->offset; + + if (written < 0) + return (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) ? BATCH_AGAIN : BATCH_FAILED; + if (!written) + return BATCH_FAILED; + + batch_consume(batch, (size_t)written); + return written; +} + +void batch_reset(ot_batch *batch) { + while (batch->first < batch->count) + free(batch->entries[batch->first++].iov_base); + + free(batch->entries); + batch->entries = NULL; + batch->count = batch->first = batch->offset = batch->space = 0; +} + +/* For the few constants the chunked encoding needs, which the batch frees + along with everything else */ +int batch_add_copy(ot_batch *batch, const void *buf, size_t len) { + void *copy = malloc(len); + + if (!copy) + return -1; + memcpy(copy, buf, len); + return batch_add(batch, copy, len); +} diff --git a/ot_batch.h b/ot_batch.h new file mode 100644 index 0000000..c855671 --- /dev/null +++ b/ot_batch.h @@ -0,0 +1,48 @@ +/* This software was written by Dirk Engling + It is considered beerware. Prost. Skol. Cheers or whatever. + + $id$ */ + +#ifndef OT_BATCH_H__ +#define OT_BATCH_H__ + +#include +#include +#include + +/* A queue of buffers on their way to one socket. The batch owns every buffer + handed to it and frees each one as soon as it has left, so a client that + reads slowly costs memory only for what it has not received yet. + + A zeroed ot_batch is ready to use. */ +typedef struct { + struct iovec *entries; + size_t space; + size_t count; + size_t first; /* entries before this one are sent and freed */ + size_t offset; /* bytes of entries[first] already sent */ +} ot_batch; + +static inline int batch_pending(const ot_batch *batch) { return batch->first != batch->count; } + +/* What batch_send reports instead of a byte count */ +enum { BATCH_DONE = 0, BATCH_AGAIN = -1, BATCH_FAILED = -2 }; + +/* Queue buf and take ownership of it. Returns 0, or -1 having freed buf. */ +int batch_add(ot_batch *batch, void *buf, size_t len); + +/* Write as much as the socket takes. Returns the number of bytes written, or + BATCH_DONE, BATCH_AGAIN or BATCH_FAILED. */ +ssize_t batch_send(int s, ot_batch *batch); + +/* Queue every entry of a vector, taking ownership of each buffer but not of + the vector itself. Returns 0, or -1 having freed all of them. */ +int batch_add_vector(ot_batch *batch, const struct iovec *entries, size_t count); + +/* Queue a copy of buf, which the batch then owns. Returns 0, or -1. */ +int batch_add_copy(ot_batch *batch, const void *buf, size_t len); + +/* Free whatever has not left yet, along with the batch's own array */ +void batch_reset(ot_batch *batch); + +#endif diff --git a/ot_http.c b/ot_http.c index 4c4121b..4675efb 100644 --- a/ot_http.c +++ b/ot_http.c @@ -15,7 +15,7 @@ #include /* Libowfat */ -#include "iob.h" +#include "io.h" /* Opentracker */ #include "ot_accesslist.h" @@ -29,14 +29,7 @@ #include "scan_urlencoded_query.h" #include "trackerlogic.h" -#ifdef WANT_NO_AUTO_FREE -#define OT_IOB_INIT(B) (bzero(B, sizeof(io_batch)), 0) -#else -#define OT_IOB_INIT(B) iob_init_autofree(B, 0) -#endif - #define OT_MAXMULTISCRAPE_COUNT 64 -#define OT_BATCH_LIMIT (1024 * 1024 * 16) extern const char *g_redirecturl; const char *g_stats_path; @@ -90,20 +83,17 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) { memcpy(outbuf, ws->reply + written_size, ws->reply_size - written_size); if (!cookie->batch) { - cookie->batch = malloc(sizeof(io_batch)); - if (!cookie->batch || OT_IOB_INIT(cookie->batch) == -1) { - free(cookie->batch); + cookie->batch = calloc(1, sizeof(ot_batch)); + if (!cookie->batch) { free(outbuf); buf_reset(&cookie->request); free(cookie); io_close(sock); return; } - - cookie->batches = 1; } - iob_addbuf_free(cookie->batch, outbuf, ws->reply_size - written_size); + batch_add(cookie->batch, outbuf, ws->reply_size - written_size); /* writeable short data sockets just have a tcp timeout */ if (!ws->keep_alive) { @@ -146,10 +136,9 @@ ssize_t http_issue_error(const int64_t sock, struct ot_workstruct *ws, int code) ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t iovec_entries, struct iovec *iovector, int is_partial) { struct http_data *cookie = io_getcookie(sock); - io_batch *current; char *header; const char *encoding = ""; - size_t i, header_size, size = iovec_length(&iovec_entries, (const struct iovec **)&iovector); + size_t header_size, size = iovec_length(&iovec_entries, (const struct iovec **)&iovector); tai6464 t; /* No cookie? Bad socket. Leave. */ @@ -199,42 +188,24 @@ ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t } if (!cookie->batch) { - cookie->batch = malloc(sizeof(io_batch)); - if (!cookie->batch || OT_IOB_INIT(cookie->batch) == -1) { - free(cookie->batch); + cookie->batch = calloc(1, sizeof(ot_batch)); + if (!cookie->batch) { free(header); iovec_free(&iovec_entries, &iovector); HTTPERROR_500; } - cookie->batches = 1; - } - current = cookie->batch + cookie->batches - 1; - iob_addbuf_free(current, header, header_size); - - /* Split huge iovectors into separate io_batches */ - for (i = 0; i < (size_t)iovec_entries; ++i) { - /* If the current batch's limit is reached, try to reallocate a new batch to work on */ - if (current->bytesleft > OT_BATCH_LIMIT) { - io_batch *new_batch = realloc(cookie->batch, (cookie->batches + 1) * sizeof(io_batch)); - - if (new_batch) { - cookie->batch = new_batch; - if (OT_IOB_INIT(cookie->batch + cookie->batches) != -1) - ++cookie->batches; - } - /* realloc may have moved the batches, so current is stale either way */ - current = cookie->batch + cookie->batches - 1; - } - iob_addbuf_free(current, iovector[i].iov_base, iovector[i].iov_len); } + + batch_add(cookie->batch, header, header_size); + batch_add_vector(cookie->batch, iovector, iovec_entries); free(iovector); + if (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER) - iob_addbuf(current, "\r\n", 2); + batch_add_copy(cookie->batch, "\r\n", 2); } if ((cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER) && cookie->batch && !is_partial) { - current = cookie->batch + cookie->batches - 1; - iob_addbuf(current, "0\r\n\r\n", 5); + batch_add_copy(cookie->batch, "0\r\n\r\n", 5); cookie->flag &= ~STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER; } diff --git a/ot_http.h b/ot_http.h index abad83a..003b545 100644 --- a/ot_http.h +++ b/ot_http.h @@ -10,9 +10,7 @@ #include #include -/* Libowfat */ -#include "iob.h" - +#include "ot_batch.h" #include "ot_buf.h" #include "trackerlogic.h" @@ -27,8 +25,7 @@ typedef enum { struct http_data { ot_buf request; - io_batch *batch; - size_t batches; + ot_batch *batch; ot_ip6 ip; STRUCT_HTTP_FLAG flag; }; diff --git a/proxy.c b/proxy.c index 219d926..63b4c4f 100644 --- a/proxy.c +++ b/proxy.c @@ -16,9 +16,9 @@ /* Libowfat */ #include "io.h" -#include "iob.h" /* Opentracker */ +#include "ot_batch.h" #include "ot_ip.h" #include "ot_mutex.h" #include "ot_random.h" @@ -228,7 +228,7 @@ typedef struct { size_t indata_length; /* Length of unprocessed data */ uint32_t tracker_id; /* How the other end greeted */ int64_t fd; /* A file handle, if connected, <= 0 is disconnected (0 initially, -1 else) */ - io_batch outdata; /* The iobatch containing our sync data */ + ot_batch outdata; /* The batch containing our sync data */ size_t packet_tcount; /* Number of unprocessed torrents in packet we currently receive */ uint8_t packet_tprefix; /* Prefix byte for all torrents in current packet */ @@ -243,7 +243,7 @@ void reset_info_block(proxy_peer *peer) { peer->tracker_id = 0; peer->fd = -1; peer->packet_tcount = 0; - iob_reset(&peer->outdata); + batch_reset(&peer->outdata); PROXYPEER_SETDISCONNECTED(peer->state); } @@ -424,11 +424,11 @@ static void handle_write(int64_t peersocket) { } break; case FLAG_CONNECTED: - switch (iob_send(peersocket, &peer->outdata)) { - case 0: /* all data sent */ + switch (batch_send(peersocket, &peer->outdata)) { + case BATCH_DONE: io_dontwantwrite(peersocket); break; - case -3: /* an error occured */ + case BATCH_FAILED: io_close(peersocket); reset_info_block(peer); break; @@ -728,7 +728,7 @@ static void *streamsync_worker(void *args) { void *tmp = malloc(mem); if (tmp) { memcpy(tmp, ptr, mem); - iob_addbuf_free(&g_connections[i].outdata, tmp, mem); + batch_add(&g_connections[i].outdata, tmp, mem); io_wantwrite(g_connections[i].fd); } } diff --git a/tests/runner.c b/tests/runner.c index 972a56a..148b928 100644 --- a/tests/runner.c +++ b/tests/runner.c @@ -24,6 +24,7 @@ static const struct module modules[] = {{"scan_urlencoded_query", test_scan_urle {"ot_ip", test_ip}, {"ot_socket", test_socket}, {"ot_mutex", test_mutex}, + {"ot_batch", test_batch}, {"ot_accesslist", test_accesslist}, {NULL, NULL}}; diff --git a/tests/t_batch.c b/tests/t_batch.c new file mode 100644 index 0000000..02743f1 --- /dev/null +++ b/tests/t_batch.c @@ -0,0 +1,348 @@ +/* This software was written by Dirk Engling + It is considered beerware. Prost. Skol. Cheers or whatever. + + $id$ */ + +/* System */ +#include +#include +#include +#include +#include +#include + +/* Opentracker */ +#include "ot_batch.h" +#include "test.h" + +/* The batch writes to a real socket pair, because what is worth pinning down + is what it does when the kernel takes only part of what it was offered. */ + +static char *dup_of(const char *text) { + size_t len = strlen(text); + char *copy = malloc(len); + + memcpy(copy, text, len); + return copy; +} + +static void drain(int s, char *into, size_t len) { + size_t got = 0; + + while (got < len) { + ssize_t r = read(s, into + got, len - got); + + if (r <= 0) + break; + got += (size_t)r; + } + into[got] = 0; +} + +static void case_queue_accounting(void) { + ot_batch batch; + + memset(&batch, 0, sizeof(batch)); + + test_case("a zeroed batch is empty and sends nothing"); + CHECK_INT(batch_pending(&batch), 0); + CHECK_INT(batch_send(-1, &batch), BATCH_DONE); + + test_case("anything queued makes it pending"); + CHECK_INT(batch_add(&batch, dup_of("hello "), 6), 0); + CHECK_INT(batch_add_copy(&batch, "world", 5), 0); + CHECK_INT(batch_pending(&batch), 1); + CHECK_INT(batch.count, 2); + + batch_reset(&batch); + test_case("and a reset empties it"); + CHECK_INT(batch_pending(&batch), 0); + CHECK_PTR(batch.entries, NULL); +} + +static void case_send_in_one_go(void) { + ot_batch batch; + int sv[2]; + char got[32]; + + memset(&batch, 0, sizeof(batch)); + CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0); + + test_case("several buffers arrive as one stream, in order"); + CHECK_INT(batch_add(&batch, dup_of("opentracker"), 11), 0); + CHECK_INT(batch_add_copy(&batch, " says ", 6), 0); + CHECK_INT(batch_add(&batch, dup_of("hello"), 5), 0); + CHECK_INT(batch_send(sv[0], &batch), 22); + CHECK_INT(batch_pending(&batch), 0); + + drain(sv[1], got, 22); + CHECK_STR(got, "opentracker says hello"); + + test_case("and an empty batch reports it has nothing left"); + CHECK_INT(batch_send(sv[0], &batch), BATCH_DONE); + + batch_reset(&batch); + close(sv[0]); + close(sv[1]); +} + +/* Send everything the batch holds, reading the other end as we go so the + socket never stays full, and return what arrived. */ +static size_t pump(int writer, int reader, ot_batch *batch, char *into, size_t room) { + size_t got = 0; + + while (batch_pending(batch) || got < room) { + ssize_t r; + + if (batch_pending(batch)) { + ssize_t res = batch_send(writer, batch); + + if (res == BATCH_FAILED) + break; + if (res > 0) + continue; + } + + r = read(reader, into + got, room - got); + if (r <= 0) + break; + got += (size_t)r; + } + return got; +} + +static void case_partial_write_resumes(void) { + const size_t size = 200000; + ot_batch batch; + int sv[2], sndbuf = 4096; + char *sent, *got; + size_t i, arrived; + + memset(&batch, 0, sizeof(batch)); + CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0); + setsockopt(sv[0], SOL_SOCKET, SO_SNDBUF, &sndbuf, sizeof(sndbuf)); + fcntl(sv[0], F_SETFL, fcntl(sv[0], F_GETFL, 0) | O_NONBLOCK); + + sent = malloc(size); + got = malloc(size); + for (i = 0; i < size; ++i) + sent[i] = (char)(i % 251); + + test_case("a buffer the kernel takes in pieces arrives whole and in order"); + CHECK_INT(batch_add(&batch, memcpy(malloc(size), sent, size), size), 0); + arrived = pump(sv[0], sv[1], &batch, got, size); + CHECK_INT(arrived, size); + CHECK_INT(batch_pending(&batch), 0); + CHECK_MEM(got, sent, size); + + free(sent); + free(got); + batch_reset(&batch); + close(sv[0]); + close(sv[1]); +} + +/* writev refuses more than IOV_MAX entries at once, so a long batch has to be + sent across several calls. */ +static void case_more_entries_than_one_writev(void) { + const size_t entries = 3000, each = 4; + ot_batch batch; + int sv[2], sndbuf = 4096; + char *sent, *got; + size_t i, arrived, refused = 0; + + memset(&batch, 0, sizeof(batch)); + CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0); + setsockopt(sv[0], SOL_SOCKET, SO_SNDBUF, &sndbuf, sizeof(sndbuf)); + fcntl(sv[0], F_SETFL, fcntl(sv[0], F_GETFL, 0) | O_NONBLOCK); + + sent = malloc(entries * each); + got = malloc(entries * each); + + test_case("a batch of more entries than writev accepts still arrives whole"); + for (i = 0; i < entries; ++i) { + char piece[4]; + + piece[0] = (char)(i), piece[1] = (char)(i >> 8), piece[2] = 'x', piece[3] = 'y'; + memcpy(sent + i * each, piece, each); + if (batch_add(&batch, memcpy(malloc(each), piece, each), each)) + ++refused; + } + CHECK_INT(refused, 0); + CHECK_INT(batch.count, entries); + + arrived = pump(sv[0], sv[1], &batch, got, entries * each); + CHECK_INT(arrived, entries * each); + CHECK_MEM(got, sent, entries * each); + + free(sent); + free(got); + batch_reset(&batch); + close(sv[0]); + close(sv[1]); +} + +static void case_reset_frees_the_unsent(void) { + ot_batch batch; + int sv[2]; + + memset(&batch, 0, sizeof(batch)); + CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0); + + test_case("a batch torn down mid flight keeps no bytes outstanding"); + CHECK_INT(batch_add(&batch, dup_of("first"), 5), 0); + CHECK_INT(batch_add(&batch, dup_of("second"), 6), 0); + CHECK_INT(batch_send(sv[0], &batch), 11); + CHECK_INT(batch_add(&batch, dup_of("never sent"), 10), 0); + CHECK_INT(batch_pending(&batch), 1); + + /* Whether the unsent buffer is freed is for the sanitizer to say */ + batch_reset(&batch); + CHECK_INT(batch_pending(&batch), 0); + CHECK_INT(batch.count, 0); + + close(sv[0]); + close(sv[1]); +} + +static void case_closed_socket_is_fatal(void) { + ot_batch batch; + int sv[2]; + + memset(&batch, 0, sizeof(batch)); + CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0); + close(sv[1]); + + test_case("writing to a socket whose peer is gone reports a dead connection"); + CHECK_INT(batch_add(&batch, dup_of("nobody is listening"), 19), 0); + CHECK_INT(batch_send(sv[0], &batch), BATCH_FAILED); + + batch_reset(&batch); + close(sv[0]); +} + +static void case_vector_is_taken_whole(void) { + ot_batch batch; + struct iovec *vector = malloc(3 * sizeof(struct iovec)); + int sv[2]; + char got[32]; + + memset(&batch, 0, sizeof(batch)); + CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0); + + test_case("a whole vector is queued in one go, in order"); + vector[0].iov_base = dup_of("one "), vector[0].iov_len = 4; + vector[1].iov_base = dup_of("two "), vector[1].iov_len = 4; + vector[2].iov_base = dup_of("three"), vector[2].iov_len = 5; + CHECK_INT(batch_add_vector(&batch, vector, 3), 0); + free(vector); + CHECK_INT(batch.count, 3); + CHECK_INT(batch_send(sv[0], &batch), 13); + + drain(sv[1], got, 13); + CHECK_STR(got, "one two three"); + + batch_reset(&batch); + close(sv[0]); + close(sv[1]); +} + +/* The entry array starts small and doubles, so crossing that boundary one + buffer at a time is worth its own case. */ +static void case_growing_one_entry_at_a_time(void) { + const size_t entries = 100; + ot_batch batch; + int sv[2]; + char sent[400], got[401]; + size_t i, refused = 0; + + memset(&batch, 0, sizeof(batch)); + CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0); + + test_case("a batch grown one buffer at a time keeps them all, in order"); + for (i = 0; i < entries; ++i) { + char piece[4]; + + piece[0] = (char)('a' + i % 26), piece[1] = (char)('0' + i % 10); + piece[2] = (char)('A' + i % 26), piece[3] = '.'; + memcpy(sent + i * 4, piece, 4); + if (batch_add(&batch, memcpy(malloc(4), piece, 4), 4)) + ++refused; + } + CHECK_INT(refused, 0); + CHECK_INT(batch.count, entries); + CHECK_INT(batch_send(sv[0], &batch), (ssize_t)(entries * 4)); + + drain(sv[1], got, entries * 4); + CHECK_MEM(got, sent, entries * 4); + + batch_reset(&batch); + close(sv[0]); + close(sv[1]); +} + +/* Appending while earlier entries are still on their way is what a chunked + transfer does. Rather than depend on when a kernel decides to block, this + takes entries off the front the way a completed send would and checks that + the array reuses the space instead of growing. */ +static void case_append_while_sending(void) { + const size_t per_round = 12, each = 8; + ot_batch batch; + int sv[2]; + char expected[16], got[17]; + size_t i, round, refused = 0, space_after_first = 0; + + memset(&batch, 0, sizeof(batch)); + CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0); + + test_case("an array whose front has gone out is reused, not grown"); + for (round = 0; round < 8; ++round) { + for (i = 0; i < per_round; ++i) { + /* Distinct per entry, so a move from the wrong index shows up */ + char *piece = malloc(each); + + memset(piece, (int)('a' + (round * per_round + i) % 26), each); + piece[0] = (char)('0' + i); + piece[1] = (char)('0' + round); + if (batch_add(&batch, piece, each)) + ++refused; + } + + /* All but the last entry have left and been freed */ + while (batch.count - batch.first > 1) + free(batch.entries[batch.first++].iov_base); + + if (!round) + space_after_first = batch.space; + } + CHECK_INT(refused, 0); + CHECK_INT(batch.space, space_after_first); + + test_case("and what is still queued survives the move, in order"); + CHECK_INT(batch.count - batch.first, 1); + CHECK_INT(batch_add(&batch, memset(malloc(each), 'Z', each), each), 0); + CHECK_INT(batch_send(sv[0], &batch), (ssize_t)(2 * each)); + + memset(expected, (int)('a' + (7 * per_round + per_round - 1) % 26), each); + expected[0] = (char)('0' + per_round - 1); + expected[1] = '7'; + memset(expected + each, 'Z', each); + drain(sv[1], got, 2 * each); + CHECK_MEM(got, expected, 2 * each); + + batch_reset(&batch); + close(sv[0]); + close(sv[1]); +} + +void test_batch(void) { + case_queue_accounting(); + case_send_in_one_go(); + case_vector_is_taken_whole(); + case_growing_one_entry_at_a_time(); + case_append_while_sending(); + case_partial_write_resumes(); + case_more_entries_than_one_writev(); + case_reset_frees_the_unsent(); + case_closed_socket_is_fatal(); +} diff --git a/tests/test.h b/tests/test.h index 9677524..fd574c2 100644 --- a/tests/test.h +++ b/tests/test.h @@ -87,5 +87,6 @@ void test_buf(void); void test_ip(void); void test_socket(void); void test_mutex(void); +void test_batch(void); #endif diff --git a/trackerlogic.c b/trackerlogic.c index 0fa7bfb..74fea6d 100644 --- a/trackerlogic.c +++ b/trackerlogic.c @@ -10,9 +10,6 @@ #include #include -/* Libowfat */ -#include "io.h" -#include "iob.h" /* Opentracker */ #include "ot_accesslist.h" -- cgit v1.3