From 5b25f89d00433f1c11576d1ea19304797dc60d1b Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Sat, 31 Oct 2026 18:48:00 +0100 Subject: Refactor ot_workqueue out of ot_mutex, no behaviour change --- Makefile | 8 +- opentracker.c | 3 +- ot_fullscrape.c | 25 +++--- ot_fullscrape.h | 1 + ot_http.c | 3 +- ot_mutex.c | 221 ----------------------------------------------- ot_mutex.h | 57 ------------ ot_stats.c | 9 +- ot_sync.c | 7 +- ot_workqueue.c | 243 ++++++++++++++++++++++++++++++++++++++++++++++++++++ ot_workqueue.h | 78 +++++++++++++++++ tests/runner.c | 2 +- tests/t_mutex.c | 107 ----------------------- tests/t_workqueue.c | 107 +++++++++++++++++++++++ tests/test.h | 2 +- trackerlogic.c | 3 + 16 files changed, 464 insertions(+), 412 deletions(-) create mode 100644 ot_workqueue.c create mode 100644 ot_workqueue.h delete mode 100644 tests/t_mutex.c create mode 100644 tests/t_workqueue.c diff --git a/Makefile b/Makefile index 8f4dcf6..61eb872 100644 --- a/Makefile +++ b/Makefile @@ -69,8 +69,8 @@ CFLAGS_libowfat=-I$(LIBOWFAT_HEADERS) LDFLAGS_libowfat=-L$(LIBOWFAT_LIBRARY) -lowfat BINARY =opentracker -HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.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 ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.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 +HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.h ot_workqueue.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 ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.c ot_workqueue.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_fdstate.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c all: $(BINARY) $(BINARY).debug @@ -96,8 +96,8 @@ proxy.debug: $(SOURCES_proxy) $(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_urlencoded.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 tests/t_fdstate.c tests/t_event.c tests/t_fmt.c -TEST_UNITS=ot_urlencoded.c ot_fmt.c ot_event.c ot_batch.c ot_fdstate.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_urlencoded.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_workqueue.c tests/t_batch.c tests/t_fdstate.c tests/t_event.c tests/t_fmt.c +TEST_UNITS=ot_urlencoded.c ot_fmt.c ot_event.c ot_batch.c ot_fdstate.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 ot_workqueue.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 6e073e9..2f2746d 100644 --- a/opentracker.c +++ b/opentracker.c @@ -29,6 +29,7 @@ #include "ot_ip.h" #include "ot_livesync.h" #include "ot_mutex.h" +#include "ot_workqueue.h" #include "ot_random.h" #include "ot_scan.h" #include "ot_event.h" @@ -148,7 +149,7 @@ static void *server_mainloop(void *args) { } } - while ((sock = mutex_workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1) + while ((sock = workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1) http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial); if (time_now_seconds() > next_timeout_check) { diff --git a/ot_fullscrape.c b/ot_fullscrape.c index a86d6d7..ae29308 100644 --- a/ot_fullscrape.c +++ b/ot_fullscrape.c @@ -23,6 +23,7 @@ #include "ot_fullscrape.h" #include "ot_iovec.h" #include "ot_mutex.h" +#include "ot_workqueue.h" #include "ot_urlencoded.h" #include "trackerlogic.h" @@ -54,7 +55,7 @@ static void *fullscrape_worker(void *args) { while (g_opentracker_running) { ot_tasktype tasktype = TASK_FULLSCRAPE; - ot_taskid taskid = mutex_workqueue_poptask(&tasktype); + ot_taskid taskid = workqueue_poptask(&tasktype); if (!taskid) break; #ifdef WANT_COMPRESSION_ZSTD @@ -68,7 +69,7 @@ static void *fullscrape_worker(void *args) { else #endif fullscrape_make(taskid, tasktype); - mutex_workqueue_pushchunked(taskid, NULL); + workqueue_pushchunked(taskid, NULL); } return NULL; } @@ -79,12 +80,12 @@ void fullscrape_init(void) { } void fullscrape_deinit(void) { - mutex_workqueue_wakeall(); + workqueue_wakeall(); pthread_join( thread_id, NULL ); } void fullscrape_deliver( int64_t sock, ot_tasktype tasktype ) { - mutex_workqueue_pushtask( sock, TASK_FULLSCRAPE | tasktype ); + workqueue_pushtask( sock, TASK_FULLSCRAPE | tasktype ); } static void fullscrape_write_one( ot_tasktype mode, char **cursor, ot_torrent *torrent, ot_hash *hash ) { @@ -170,7 +171,7 @@ static void fullscrape_make(int taskid, ot_tasktype mode) { if (r > re) { iovector.iov_len = r - (char *)iovector.iov_base; - if (mutex_workqueue_pushchunked(taskid, &iovector)) { + if (workqueue_pushchunked(taskid, &iovector)) { free(iovector.iov_base); mutex_bucket_unlock(bucket, 0); return; @@ -200,7 +201,7 @@ static void fullscrape_make(int taskid, ot_tasktype mode) { /* Send rest of data */ iovector.iov_len = r - (char *)iovector.iov_base; - if (mutex_workqueue_pushchunked(taskid, &iovector)) + if (workqueue_pushchunked(taskid, &iovector)) free(iovector.iov_base); } @@ -251,7 +252,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) { while (!strm.avail_out) { iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base; - if (mutex_workqueue_pushchunked(taskid, &iovector)) { + if (workqueue_pushchunked(taskid, &iovector)) { free(iovector.iov_base); mutex_bucket_unlock(bucket, 0); return; @@ -291,7 +292,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) { fprintf(stderr, "deflate() failed while in fullscrape_make()'s endgame.\n"); iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base; - if (mutex_workqueue_pushchunked(taskid, &iovector)) { + if (workqueue_pushchunked(taskid, &iovector)) { free(iovector.iov_base); deflateEnd(&strm); return; @@ -314,7 +315,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) { /* Only pass the new buffer if there actually was some data left in the buffer */ iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base; - if (!iovector.iov_len || mutex_workqueue_pushchunked(taskid, &iovector)) + if (!iovector.iov_len || workqueue_pushchunked(taskid, &iovector)) free(iovector.iov_base); } @@ -380,7 +381,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) { while (outbuf.pos + OT_SCRAPE_MAXENTRYLEN > outbuf.size) { iovector.iov_len = outbuf.size; - if (mutex_workqueue_pushchunked(taskid, &iovector)) { + if (workqueue_pushchunked(taskid, &iovector)) { free(iovector.iov_base); ZSTD_freeCCtx(zstream); return mutex_bucket_unlock(bucket, 0); @@ -418,7 +419,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) { more_bytes = ZSTD_compressStream2(zstream, &outbuf, &inbuf, ZSTD_e_end); iovector.iov_len = outbuf.pos; - if (mutex_workqueue_pushchunked(taskid, &iovector)) { + if (workqueue_pushchunked(taskid, &iovector)) { free(iovector.iov_base); ZSTD_freeCCtx(zstream); return; @@ -443,7 +444,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) { /* Only pass the new buffer if there actually was some data left in the buffer */ iovector.iov_len = outbuf.pos; - if (!iovector.iov_len || mutex_workqueue_pushchunked(taskid, &iovector)) + if (!iovector.iov_len || workqueue_pushchunked(taskid, &iovector)) free(iovector.iov_base); } diff --git a/ot_fullscrape.h b/ot_fullscrape.h index 0e7690c..b759a27 100644 --- a/ot_fullscrape.h +++ b/ot_fullscrape.h @@ -9,6 +9,7 @@ #ifdef WANT_FULLSCRAPE #include "ot_mutex.h" +#include "ot_workqueue.h" void fullscrape_init(void); void fullscrape_deinit(void); diff --git a/ot_http.c b/ot_http.c index 602ff94..b0e1de3 100644 --- a/ot_http.c +++ b/ot_http.c @@ -22,6 +22,7 @@ #include "ot_iovec.h" #include "ot_ip.h" #include "ot_mutex.h" +#include "ot_workqueue.h" #include "ot_scan.h" #include "ot_event.h" #include "ot_fdstate.h" @@ -77,7 +78,7 @@ void http_close(const int64_t sock) { free(cookie->batch); buf_reset(&cookie->request); if (cookie->flag & (STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER)) - mutex_workqueue_canceltask(sock); + workqueue_canceltask(sock); if (!(cookie->flag & STRUCT_HTTP_FLAG_ONSTACK)) free(cookie); } diff --git a/ot_mutex.c b/ot_mutex.c index 5392264..d856d5f 100644 --- a/ot_mutex.c +++ b/ot_mutex.c @@ -31,7 +31,6 @@ static pthread_mutex_t bucket_mutex[OT_BUCKET_COUNT]; static _Atomic size_t g_torrent_count; /* Self pipe from opentracker.c */ -extern int g_self_pipe[2]; ot_vector *mutex_bucket_lock(int bucket) { pthread_mutex_lock(bucket_mutex + bucket); @@ -60,237 +59,17 @@ void mutex_bucket_unlock_by_hash(ot_hash const hash, int delta_torrentcount) { size_t mutex_get_torrent_count(void) { return atomic_load_explicit(&g_torrent_count, memory_order_relaxed); } -/* TaskQueue Magic */ - -struct ot_task { - ot_taskid taskid; - ot_tasktype tasktype; - int64_t sock; - size_t iovec_entries; - struct iovec *iovec; - struct ot_task *next; -}; - -static ot_taskid next_free_taskid = 1; -static struct ot_task *tasklist; -static pthread_mutex_t tasklist_mutex; -static pthread_cond_t tasklist_being_filled; - -int mutex_workqueue_pushtask(int64_t sock, ot_tasktype tasktype) { - struct ot_task **tmptask, *task; - - task = malloc(sizeof(struct ot_task)); - if (!task) - return -1; - - task->taskid = 0; - task->tasktype = tasktype; - task->sock = sock; - task->iovec_entries = 0; - task->iovec = NULL; - task->next = 0; - - /* Want exclusive access to tasklist */ - pthread_mutex_lock(&tasklist_mutex); - - /* Skip to end of list */ - tmptask = &tasklist; - while (*tmptask) - tmptask = &(*tmptask)->next; - *tmptask = task; - - /* Inform waiting workers and release lock */ - pthread_cond_broadcast(&tasklist_being_filled); - pthread_mutex_unlock(&tasklist_mutex); - return 0; -} - -void mutex_workqueue_canceltask(int64_t sock) { - struct ot_task **task; - - /* Want exclusive access to tasklist */ - pthread_mutex_lock(&tasklist_mutex); - - for (task = &tasklist; *task; task = &((*task)->next)) - if ((*task)->sock == sock) { - struct iovec *iovec = (*task)->iovec; - struct ot_task *ptask = *task; - size_t i; - - /* Free task's iovec, the buffers it points at and the vector itself */ - for (i = 0; i < (*task)->iovec_entries; ++i) - free(iovec[i].iov_base); - free(iovec); - - *task = (*task)->next; - free(ptask); - break; - } - - /* Release lock */ - pthread_mutex_unlock(&tasklist_mutex); -} - -ot_taskid mutex_workqueue_poptask(ot_tasktype *tasktype) { - struct ot_task *task; - ot_taskid taskid = 0; - - /* Want exclusive access to tasklist */ - pthread_mutex_lock(&tasklist_mutex); - - while (!taskid) { - /* Skip to the first unassigned task this worker wants to do */ - for (task = tasklist; task; task = task->next) - if (!task->taskid && (TASK_CLASS_MASK & task->tasktype) == *tasktype) { - /* If we found an outstanding task, assign a taskid to it - and leave the loop */ - task->taskid = taskid = ++next_free_taskid; - *tasktype = task->tasktype; - break; - } - - /* Wait until the next task is being fed, unless the tracker is on its - way out, in which case the worker is meant to leave rather than wait */ - if (!taskid) { - if (!g_opentracker_running) - break; - pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex); - } - } - - /* Release lock */ - pthread_mutex_unlock(&tasklist_mutex); - - return taskid; -} - -void mutex_workqueue_pushsuccess(ot_taskid taskid) { - struct ot_task **task; - - /* Want exclusive access to tasklist */ - pthread_mutex_lock(&tasklist_mutex); - - for (task = &tasklist; *task; task = &((*task)->next)) - if ((*task)->taskid == taskid) { - struct ot_task *ptask = *task; - *task = (*task)->next; - free(ptask); - break; - } - - /* Release lock */ - pthread_mutex_unlock(&tasklist_mutex); -} - -int mutex_workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovec) { - struct ot_task *task; - const char byte = 'o'; - - /* Want exclusive access to tasklist */ - pthread_mutex_lock(&tasklist_mutex); - - for (task = tasklist; task; task = task->next) - if (task->taskid == taskid) { - task->iovec_entries = iovec_entries; - task->iovec = iovec; - task->tasktype = TASK_DONE; - break; - } - - /* Release lock */ - pthread_mutex_unlock(&tasklist_mutex); - - /* A full pipe has already done this nudge's job, so a failure is fine */ - (void)!write(g_self_pipe[1], &byte, 1); - - /* Indicate whether the worker has to throw away results */ - return task ? 0 : -1; -} - -int mutex_workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec) { - struct ot_task *task; - const char byte = 'o'; - - /* Want exclusive access to tasklist */ - pthread_mutex_lock(&tasklist_mutex); - - for (task = tasklist; task; task = task->next) - if (task->taskid == taskid) { - if (iovec) { - if (iovec_append(&task->iovec_entries, &task->iovec, iovec)) - task->tasktype = TASK_DONE_PARTIAL; - else - task = NULL; - } else - task->tasktype = TASK_DONE; - break; - } - - /* Release lock */ - pthread_mutex_unlock(&tasklist_mutex); - - /* A full pipe has already done this nudge's job, so a failure is fine */ - (void)!write(g_self_pipe[1], &byte, 1); - - /* Indicate whether the worker has to throw away results */ - return task ? 0 : -1; -} - -int64_t mutex_workqueue_popresult(size_t *iovec_entries, struct iovec **iovec, int *is_partial) { - struct ot_task **task; - int64_t sock = -1; - - *is_partial = 0; - - /* Want exclusive access to tasklist */ - pthread_mutex_lock(&tasklist_mutex); - - for (task = &tasklist; *task; task = &((*task)->next)) - if (((*task)->tasktype & TASK_CLASS_MASK) == TASK_DONE) { - struct ot_task *ptask = *task; - *iovec_entries = ptask->iovec_entries; - *iovec = ptask->iovec; - sock = ptask->sock; - - if ((*task)->tasktype == TASK_DONE) { - *task = ptask->next; - free(ptask); - } else { - ptask->iovec_entries = 0; - ptask->iovec = NULL; - *is_partial = 1; - /* Prevent task from showing up immediately again unless new data was added */ - (*task)->tasktype = TASK_FULLSCRAPE; - } - break; - } - - /* Release lock */ - pthread_mutex_unlock(&tasklist_mutex); - return sock; -} - void mutex_init(void) { size_t i; - pthread_mutex_init(&tasklist_mutex, NULL); - pthread_cond_init(&tasklist_being_filled, NULL); for (i = 0; i < OT_BUCKET_COUNT; ++i) pthread_mutex_init(bucket_mutex + i, NULL); memset(all_torrents, 0, sizeof(all_torrents)); } /* Let every worker parked on the queue see that the tracker is stopping */ -void mutex_workqueue_wakeall(void) { - pthread_mutex_lock(&tasklist_mutex); - pthread_cond_broadcast(&tasklist_being_filled); - pthread_mutex_unlock(&tasklist_mutex); -} - void mutex_deinit(void) { size_t i; for (i = 0; i < OT_BUCKET_COUNT; ++i) pthread_mutex_destroy(bucket_mutex + i); - pthread_mutex_destroy(&tasklist_mutex); - pthread_cond_destroy(&tasklist_being_filled); memset(all_torrents, 0, sizeof(all_torrents)); } diff --git a/ot_mutex.h b/ot_mutex.h index c048fdb..c1218d7 100644 --- a/ot_mutex.h +++ b/ot_mutex.h @@ -10,7 +10,6 @@ #include "trackerlogic.h" void mutex_init(void); -void mutex_workqueue_wakeall(void); void mutex_deinit(void); ot_vector *mutex_bucket_lock(int bucket); @@ -21,60 +20,4 @@ void mutex_bucket_unlock_by_hash(ot_hash const hash, int delta_torrentcoun size_t mutex_get_torrent_count(void); -typedef enum { - TASK_STATS_CONNS = 0x0001, - TASK_STATS_TCP = 0x0002, - TASK_STATS_UDP = 0x0003, - TASK_STATS_SCRAPE = 0x0004, - TASK_STATS_FULLSCRAPE = 0x0005, - TASK_STATS_TPB = 0x0006, - TASK_STATS_HTTPERRORS = 0x0007, - TASK_STATS_VERSION = 0x0008, - TASK_STATS_RENEW = 0x000a, - TASK_STATS_SYNCS = 0x000b, - TASK_STATS_COMPLETED = 0x000c, - TASK_STATS_NUMWANTS = 0x000d, - - TASK_STATS = 0x0100, /* Mask */ - TASK_STATS_TORRENTS = 0x0101, - TASK_STATS_PEERS = 0x0102, - TASK_STATS_SLASH24S = 0x0103, - TASK_STATS_TOP10 = 0x0104, - TASK_STATS_TOP100 = 0x0105, - TASK_STATS_EVERYTHING = 0x0106, - TASK_STATS_FULLLOG = 0x0107, - TASK_STATS_WOODPECKERS = 0x0108, - - TASK_FULLSCRAPE = 0x0200, /* Default mode */ - TASK_FULLSCRAPE_TPB_BINARY = 0x0201, - TASK_FULLSCRAPE_TPB_ASCII = 0x0202, - TASK_FULLSCRAPE_TPB_ASCII_PLUS = 0x0203, - TASK_FULLSCRAPE_TPB_URLENCODED = 0x0204, - TASK_FULLSCRAPE_TRACKERSTATE = 0x0205, - - TASK_DMEM = 0x0300, - - TASK_DONE = 0x0f00, - TASK_DONE_PARTIAL = 0x0f01, - - TASK_FLAG_GZIP = 0x1000, - TASK_FLAG_BZIP2 = 0x2000, - TASK_FLAG_ZSTD = 0x4000, - TASK_FLAG_CHUNKED = 0x8000, - - TASK_TASK_MASK = 0x0fff, - TASK_CLASS_MASK = 0x0f00, - TASK_FLAGS_MASK = 0xf000 -} ot_tasktype; - -typedef unsigned long ot_taskid; - -int mutex_workqueue_pushtask(int64_t sock, ot_tasktype tasktype); -void mutex_workqueue_canceltask(int64_t sock); -void mutex_workqueue_pushsuccess(ot_taskid taskid); -ot_taskid mutex_workqueue_poptask(ot_tasktype *tasktype); -int mutex_workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovector); -int mutex_workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec); -int64_t mutex_workqueue_popresult(size_t *iovec_entries, struct iovec **iovector, int *is_partial); - #endif diff --git a/ot_stats.c b/ot_stats.c index 5f97a2f..d38355a 100644 --- a/ot_stats.c +++ b/ot_stats.c @@ -23,6 +23,7 @@ #include "ot_iovec.h" #include "ot_ip.h" #include "ot_mutex.h" +#include "ot_workqueue.h" #include "ot_fmt.h" #include "ot_stats.h" #include "trackerlogic.h" @@ -803,17 +804,17 @@ static void *stats_worker(void *args) { while (g_opentracker_running) { ot_tasktype tasktype = TASK_STATS; - ot_taskid taskid = mutex_workqueue_poptask(&tasktype); + ot_taskid taskid = workqueue_poptask(&tasktype); if (!taskid) break; stats_make(&iovec_entries, &iovector, tasktype); - if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector)) + if (workqueue_pushresult(taskid, iovec_entries, iovector)) iovec_free(&iovec_entries, &iovector); } return NULL; } -void stats_deliver(int64_t sock, int tasktype) { mutex_workqueue_pushtask(sock, tasktype); } +void stats_deliver(int64_t sock, int tasktype) { workqueue_pushtask(sock, tasktype); } static pthread_t thread_id; void stats_init(void) { @@ -822,6 +823,6 @@ void stats_init(void) { } void stats_deinit(void) { - mutex_workqueue_wakeall(); + workqueue_wakeall(); pthread_join(thread_id, NULL); } diff --git a/ot_sync.c b/ot_sync.c index 3668aa1..c19cf36 100644 --- a/ot_sync.c +++ b/ot_sync.c @@ -14,6 +14,7 @@ #include "ot_fmt.h" #include "ot_iovec.h" #include "ot_mutex.h" +#include "ot_workqueue.h" #include "ot_scan.h" #include "ot_stats.h" #include "ot_sync.h" @@ -139,10 +140,10 @@ static void *sync_worker(void *args) { while (1) { ot_tasktype tasktype = TASK_SYNC_OUT; - ot_taskid taskid = mutex_workqueue_poptask(&tasktype); + ot_taskid taskid = workqueue_poptask(&tasktype); sync_make(&iovec_entries, &iovector); stats_issue_event(ws, EVENT_SYNC_OUT, FLAG_TCP, iovec_length(&iovec_entries, &iovector)); - if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector)) + if (workqueue_pushresult(taskid, iovec_entries, iovector)) iovec_free(&iovec_entries, &iovector); } return NULL; @@ -158,7 +159,7 @@ void sync_deinit( ) { } void sync_deliver( int64_t socket ) { - mutex_workqueue_pushtask( socket, TASK_SYNC_OUT ); + workqueue_pushtask( socket, TASK_SYNC_OUT ); } #endif diff --git a/ot_workqueue.c b/ot_workqueue.c new file mode 100644 index 0000000..aac366c --- /dev/null +++ b/ot_workqueue.c @@ -0,0 +1,243 @@ +/* This software was written by Dirk Engling + It is considered beerware. Prost. Skol. Cheers or whatever. + + $id$ */ + +/* System */ +#include +#include +#include +#include +#include + +/* Opentracker */ +#include "ot_iovec.h" +#include "ot_workqueue.h" +#include "trackerlogic.h" + +/* Pushing a result wakes the thread that will send it */ +extern int g_self_pipe[2]; + +struct ot_task { + ot_taskid taskid; + ot_tasktype tasktype; + int64_t sock; + size_t iovec_entries; + struct iovec *iovec; + struct ot_task *next; +}; + +static ot_taskid next_free_taskid = 1; +static struct ot_task *tasklist; +static pthread_mutex_t tasklist_mutex; +static pthread_cond_t tasklist_being_filled; + +int workqueue_pushtask(int64_t sock, ot_tasktype tasktype) { + struct ot_task **tmptask, *task; + + task = malloc(sizeof(struct ot_task)); + if (!task) + return -1; + + task->taskid = 0; + task->tasktype = tasktype; + task->sock = sock; + task->iovec_entries = 0; + task->iovec = NULL; + task->next = 0; + + /* Want exclusive access to tasklist */ + pthread_mutex_lock(&tasklist_mutex); + + /* Skip to end of list */ + tmptask = &tasklist; + while (*tmptask) + tmptask = &(*tmptask)->next; + *tmptask = task; + + /* Inform waiting workers and release lock */ + pthread_cond_broadcast(&tasklist_being_filled); + pthread_mutex_unlock(&tasklist_mutex); + return 0; +} + +void workqueue_canceltask(int64_t sock) { + struct ot_task **task; + + /* Want exclusive access to tasklist */ + pthread_mutex_lock(&tasklist_mutex); + + for (task = &tasklist; *task; task = &((*task)->next)) + if ((*task)->sock == sock) { + struct iovec *iovec = (*task)->iovec; + struct ot_task *ptask = *task; + size_t i; + + /* Free task's iovec, the buffers it points at and the vector itself */ + for (i = 0; i < (*task)->iovec_entries; ++i) + free(iovec[i].iov_base); + free(iovec); + + *task = (*task)->next; + free(ptask); + break; + } + + /* Release lock */ + pthread_mutex_unlock(&tasklist_mutex); +} + +ot_taskid workqueue_poptask(ot_tasktype *tasktype) { + struct ot_task *task; + ot_taskid taskid = 0; + + /* Want exclusive access to tasklist */ + pthread_mutex_lock(&tasklist_mutex); + + while (!taskid) { + /* Skip to the first unassigned task this worker wants to do */ + for (task = tasklist; task; task = task->next) + if (!task->taskid && (TASK_CLASS_MASK & task->tasktype) == *tasktype) { + /* If we found an outstanding task, assign a taskid to it + and leave the loop */ + task->taskid = taskid = ++next_free_taskid; + *tasktype = task->tasktype; + break; + } + + /* Wait until the next task is being fed, unless the tracker is on its + way out, in which case the worker is meant to leave rather than wait */ + if (!taskid) { + if (!g_opentracker_running) + break; + pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex); + } + } + + /* Release lock */ + pthread_mutex_unlock(&tasklist_mutex); + + return taskid; +} + +void workqueue_pushsuccess(ot_taskid taskid) { + struct ot_task **task; + + /* Want exclusive access to tasklist */ + pthread_mutex_lock(&tasklist_mutex); + + for (task = &tasklist; *task; task = &((*task)->next)) + if ((*task)->taskid == taskid) { + struct ot_task *ptask = *task; + *task = (*task)->next; + free(ptask); + break; + } + + /* Release lock */ + pthread_mutex_unlock(&tasklist_mutex); +} + +int workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovec) { + struct ot_task *task; + const char byte = 'o'; + + /* Want exclusive access to tasklist */ + pthread_mutex_lock(&tasklist_mutex); + + for (task = tasklist; task; task = task->next) + if (task->taskid == taskid) { + task->iovec_entries = iovec_entries; + task->iovec = iovec; + task->tasktype = TASK_DONE; + break; + } + + /* Release lock */ + pthread_mutex_unlock(&tasklist_mutex); + + /* A full pipe has already done this nudge's job, so a failure is fine */ + (void)!write(g_self_pipe[1], &byte, 1); + + /* Indicate whether the worker has to throw away results */ + return task ? 0 : -1; +} + +int workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec) { + struct ot_task *task; + const char byte = 'o'; + + /* Want exclusive access to tasklist */ + pthread_mutex_lock(&tasklist_mutex); + + for (task = tasklist; task; task = task->next) + if (task->taskid == taskid) { + if (iovec) { + if (iovec_append(&task->iovec_entries, &task->iovec, iovec)) + task->tasktype = TASK_DONE_PARTIAL; + else + task = NULL; + } else + task->tasktype = TASK_DONE; + break; + } + + /* Release lock */ + pthread_mutex_unlock(&tasklist_mutex); + + /* A full pipe has already done this nudge's job, so a failure is fine */ + (void)!write(g_self_pipe[1], &byte, 1); + + /* Indicate whether the worker has to throw away results */ + return task ? 0 : -1; +} + +int64_t workqueue_popresult(size_t *iovec_entries, struct iovec **iovec, int *is_partial) { + struct ot_task **task; + int64_t sock = -1; + + *is_partial = 0; + + /* Want exclusive access to tasklist */ + pthread_mutex_lock(&tasklist_mutex); + + for (task = &tasklist; *task; task = &((*task)->next)) + if (((*task)->tasktype & TASK_CLASS_MASK) == TASK_DONE) { + struct ot_task *ptask = *task; + *iovec_entries = ptask->iovec_entries; + *iovec = ptask->iovec; + sock = ptask->sock; + + if ((*task)->tasktype == TASK_DONE) { + *task = ptask->next; + free(ptask); + } else { + ptask->iovec_entries = 0; + ptask->iovec = NULL; + *is_partial = 1; + /* Prevent task from showing up immediately again unless new data was added */ + (*task)->tasktype = TASK_FULLSCRAPE; + } + break; + } + + /* Release lock */ + pthread_mutex_unlock(&tasklist_mutex); + return sock; +} + +void workqueue_wakeall(void) { + pthread_mutex_lock(&tasklist_mutex); + pthread_cond_broadcast(&tasklist_being_filled); + pthread_mutex_unlock(&tasklist_mutex); +} + +void workqueue_init(void) { + pthread_mutex_init(&tasklist_mutex, NULL); + pthread_cond_init(&tasklist_being_filled, NULL); +} + +void workqueue_deinit(void) { + pthread_mutex_destroy(&tasklist_mutex); + pthread_cond_destroy(&tasklist_being_filled); +} diff --git a/ot_workqueue.h b/ot_workqueue.h new file mode 100644 index 0000000..f99a7b5 --- /dev/null +++ b/ot_workqueue.h @@ -0,0 +1,78 @@ +/* This software was written by Dirk Engling + It is considered beerware. Prost. Skol. Cheers or whatever. + + $id$ */ + +#ifndef OT_WORKQUEUE_H__ +#define OT_WORKQUEUE_H__ + +#include +#include + +#include "trackerlogic.h" + +/* Work the tracker hands to a thread of its own: a full scrape or an + expensive stats page, which take too long to build while a client waits. + The task goes in here, the reply comes back out. */ + +typedef enum { + TASK_STATS_CONNS = 0x0001, + TASK_STATS_TCP = 0x0002, + TASK_STATS_UDP = 0x0003, + TASK_STATS_SCRAPE = 0x0004, + TASK_STATS_FULLSCRAPE = 0x0005, + TASK_STATS_TPB = 0x0006, + TASK_STATS_HTTPERRORS = 0x0007, + TASK_STATS_VERSION = 0x0008, + TASK_STATS_RENEW = 0x000a, + TASK_STATS_SYNCS = 0x000b, + TASK_STATS_COMPLETED = 0x000c, + TASK_STATS_NUMWANTS = 0x000d, + + TASK_STATS = 0x0100, /* Mask */ + TASK_STATS_TORRENTS = 0x0101, + TASK_STATS_PEERS = 0x0102, + TASK_STATS_SLASH24S = 0x0103, + TASK_STATS_TOP10 = 0x0104, + TASK_STATS_TOP100 = 0x0105, + TASK_STATS_EVERYTHING = 0x0106, + TASK_STATS_FULLLOG = 0x0107, + TASK_STATS_WOODPECKERS = 0x0108, + + TASK_FULLSCRAPE = 0x0200, /* Default mode */ + TASK_FULLSCRAPE_TPB_BINARY = 0x0201, + TASK_FULLSCRAPE_TPB_ASCII = 0x0202, + TASK_FULLSCRAPE_TPB_ASCII_PLUS = 0x0203, + TASK_FULLSCRAPE_TPB_URLENCODED = 0x0204, + TASK_FULLSCRAPE_TRACKERSTATE = 0x0205, + + TASK_DMEM = 0x0300, + + TASK_DONE = 0x0f00, + TASK_DONE_PARTIAL = 0x0f01, + + TASK_FLAG_GZIP = 0x1000, + TASK_FLAG_BZIP2 = 0x2000, + TASK_FLAG_ZSTD = 0x4000, + TASK_FLAG_CHUNKED = 0x8000, + + TASK_TASK_MASK = 0x0fff, + TASK_CLASS_MASK = 0x0f00, + TASK_FLAGS_MASK = 0xf000 +} ot_tasktype; + +typedef unsigned long ot_taskid; + +int workqueue_pushtask(int64_t sock, ot_tasktype tasktype); +void workqueue_canceltask(int64_t sock); +void workqueue_pushsuccess(ot_taskid taskid); +ot_taskid workqueue_poptask(ot_tasktype *tasktype); +int workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovector); +int workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec); +int64_t workqueue_popresult(size_t *iovec_entries, struct iovec **iovector, int *is_partial); + +void workqueue_init(void); +void workqueue_deinit(void); +void workqueue_wakeall(void); + +#endif diff --git a/tests/runner.c b/tests/runner.c index 71a5e2a..38bfb86 100644 --- a/tests/runner.c +++ b/tests/runner.c @@ -27,7 +27,7 @@ static const struct module modules[] = {{"ot_urlencoded", test_urlencoded}, {"ot_buf", test_buf}, {"ot_ip", test_ip}, {"ot_socket", test_socket}, - {"ot_mutex", test_mutex}, + {"ot_workqueue", test_workqueue}, {"ot_batch", test_batch}, {"ot_fdstate", test_fdstate}, {"ot_event", test_event}, diff --git a/tests/t_mutex.c b/tests/t_mutex.c deleted file mode 100644 index 3307d70..0000000 --- a/tests/t_mutex.c +++ /dev/null @@ -1,107 +0,0 @@ -/* This software was written by Dirk Engling - It is considered beerware. Prost. Skol. Cheers or whatever. - - $id$ */ - -/* System */ -#include -#include - -/* Opentracker */ -#include "ot_mutex.h" -#include "test.h" - -/* A worker asks for a class of task and blocks until one turns up, so a task - whose class matches no worker is never claimed and whoever is waiting for - its answer waits forever. These cases pin the matching. */ - -static void case_task_goes_to_its_own_class(void) { - ot_tasktype asked; - ot_taskid taskid; - - test_case("a worker is handed the task of the class it asked for"); - CHECK_INT(mutex_workqueue_pushtask(41, TASK_FULLSCRAPE_TPB_ASCII), 0); - asked = TASK_FULLSCRAPE; - taskid = mutex_workqueue_poptask(&asked); - CHECK(taskid != 0); - - test_case("and the full task type, not just its class"); - CHECK_INT(asked, TASK_FULLSCRAPE_TPB_ASCII); - mutex_workqueue_canceltask(41); -} - -static void case_flags_survive_the_queue(void) { - ot_tasktype asked; - - test_case("flags outside the class mask are carried through"); - CHECK_INT(mutex_workqueue_pushtask(42, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP), 0); - asked = TASK_FULLSCRAPE; - CHECK(mutex_workqueue_poptask(&asked) != 0); - CHECK_INT(asked, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP); - mutex_workqueue_canceltask(42); -} - -static void case_classes_do_not_cross(void) { - ot_tasktype asked; - ot_taskid taskid; - - test_case("a stats worker passes over a fullscrape task"); - CHECK_INT(mutex_workqueue_pushtask(43, TASK_FULLSCRAPE_TPB_BINARY), 0); - CHECK_INT(mutex_workqueue_pushtask(44, TASK_STATS_PEERS), 0); - - asked = TASK_STATS; - taskid = mutex_workqueue_poptask(&asked); - CHECK(taskid != 0); - CHECK_INT(asked, TASK_STATS_PEERS); - - test_case("and leaves it for the one that wants it"); - asked = TASK_FULLSCRAPE; - taskid = mutex_workqueue_poptask(&asked); - CHECK(taskid != 0); - CHECK_INT(asked, TASK_FULLSCRAPE_TPB_BINARY); - - mutex_workqueue_canceltask(43); - mutex_workqueue_canceltask(44); -} - -static void case_every_task_gets_its_own_id(void) { - ot_tasktype asked; - ot_taskid first, second; - - test_case("two queued tasks are handed out under different ids"); - CHECK_INT(mutex_workqueue_pushtask(45, TASK_FULLSCRAPE), 0); - CHECK_INT(mutex_workqueue_pushtask(46, TASK_FULLSCRAPE), 0); - - asked = TASK_FULLSCRAPE; - first = mutex_workqueue_poptask(&asked); - asked = TASK_FULLSCRAPE; - second = mutex_workqueue_poptask(&asked); - CHECK(first != 0); - CHECK(second != 0); - CHECK(first != second); - - mutex_workqueue_canceltask(45); - mutex_workqueue_canceltask(46); -} - -static void case_cancelled_task_is_not_handed_out(void) { - ot_tasktype asked; - - test_case("a cancelled task is gone before any worker sees it"); - CHECK_INT(mutex_workqueue_pushtask(47, TASK_FULLSCRAPE_TPB_ASCII_PLUS), 0); - CHECK_INT(mutex_workqueue_pushtask(48, TASK_FULLSCRAPE), 0); - mutex_workqueue_canceltask(47); - - asked = TASK_FULLSCRAPE; - CHECK(mutex_workqueue_poptask(&asked) != 0); - CHECK_INT(asked, TASK_FULLSCRAPE); - mutex_workqueue_canceltask(48); -} - -void test_mutex(void) { - case_task_goes_to_its_own_class(); - case_flags_survive_the_queue(); - case_classes_do_not_cross(); - case_every_task_gets_its_own_id(); - case_cancelled_task_is_not_handed_out(); -} diff --git a/tests/t_workqueue.c b/tests/t_workqueue.c new file mode 100644 index 0000000..13e0b2e --- /dev/null +++ b/tests/t_workqueue.c @@ -0,0 +1,107 @@ +/* This software was written by Dirk Engling + It is considered beerware. Prost. Skol. Cheers or whatever. + + $id$ */ + +/* System */ +#include +#include + +/* Opentracker */ +#include "ot_workqueue.h" +#include "test.h" + +/* A worker asks for a class of task and blocks until one turns up, so a task + whose class matches no worker is never claimed and whoever is waiting for + its answer waits forever. These cases pin the matching. */ + +static void case_task_goes_to_its_own_class(void) { + ot_tasktype asked; + ot_taskid taskid; + + test_case("a worker is handed the task of the class it asked for"); + CHECK_INT(workqueue_pushtask(41, TASK_FULLSCRAPE_TPB_ASCII), 0); + asked = TASK_FULLSCRAPE; + taskid = workqueue_poptask(&asked); + CHECK(taskid != 0); + + test_case("and the full task type, not just its class"); + CHECK_INT(asked, TASK_FULLSCRAPE_TPB_ASCII); + workqueue_canceltask(41); +} + +static void case_flags_survive_the_queue(void) { + ot_tasktype asked; + + test_case("flags outside the class mask are carried through"); + CHECK_INT(workqueue_pushtask(42, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP), 0); + asked = TASK_FULLSCRAPE; + CHECK(workqueue_poptask(&asked) != 0); + CHECK_INT(asked, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP); + workqueue_canceltask(42); +} + +static void case_classes_do_not_cross(void) { + ot_tasktype asked; + ot_taskid taskid; + + test_case("a stats worker passes over a fullscrape task"); + CHECK_INT(workqueue_pushtask(43, TASK_FULLSCRAPE_TPB_BINARY), 0); + CHECK_INT(workqueue_pushtask(44, TASK_STATS_PEERS), 0); + + asked = TASK_STATS; + taskid = workqueue_poptask(&asked); + CHECK(taskid != 0); + CHECK_INT(asked, TASK_STATS_PEERS); + + test_case("and leaves it for the one that wants it"); + asked = TASK_FULLSCRAPE; + taskid = workqueue_poptask(&asked); + CHECK(taskid != 0); + CHECK_INT(asked, TASK_FULLSCRAPE_TPB_BINARY); + + workqueue_canceltask(43); + workqueue_canceltask(44); +} + +static void case_every_task_gets_its_own_id(void) { + ot_tasktype asked; + ot_taskid first, second; + + test_case("two queued tasks are handed out under different ids"); + CHECK_INT(workqueue_pushtask(45, TASK_FULLSCRAPE), 0); + CHECK_INT(workqueue_pushtask(46, TASK_FULLSCRAPE), 0); + + asked = TASK_FULLSCRAPE; + first = workqueue_poptask(&asked); + asked = TASK_FULLSCRAPE; + second = workqueue_poptask(&asked); + CHECK(first != 0); + CHECK(second != 0); + CHECK(first != second); + + workqueue_canceltask(45); + workqueue_canceltask(46); +} + +static void case_cancelled_task_is_not_handed_out(void) { + ot_tasktype asked; + + test_case("a cancelled task is gone before any worker sees it"); + CHECK_INT(workqueue_pushtask(47, TASK_FULLSCRAPE_TPB_ASCII_PLUS), 0); + CHECK_INT(workqueue_pushtask(48, TASK_FULLSCRAPE), 0); + workqueue_canceltask(47); + + asked = TASK_FULLSCRAPE; + CHECK(workqueue_poptask(&asked) != 0); + CHECK_INT(asked, TASK_FULLSCRAPE); + workqueue_canceltask(48); +} + +void test_workqueue(void) { + case_task_goes_to_its_own_class(); + case_flags_survive_the_queue(); + case_classes_do_not_cross(); + case_every_task_gets_its_own_id(); + case_cancelled_task_is_not_handed_out(); +} diff --git a/tests/test.h b/tests/test.h index 0eddc82..d58742c 100644 --- a/tests/test.h +++ b/tests/test.h @@ -86,7 +86,7 @@ void test_scan(void); void test_buf(void); void test_ip(void); void test_socket(void); -void test_mutex(void); +void test_workqueue(void); void test_batch(void); void test_fdstate(void); void test_event(void); diff --git a/trackerlogic.c b/trackerlogic.c index fc0fa56..b0b7f4e 100644 --- a/trackerlogic.c +++ b/trackerlogic.c @@ -20,6 +20,7 @@ #include "ot_ip.h" #include "ot_livesync.h" #include "ot_mutex.h" +#include "ot_workqueue.h" #include "ot_random.h" #include "ot_stats.h" #include "ot_udp.h" @@ -560,6 +561,7 @@ void trackerlogic_init(void) { /* Initialise background worker threads */ mutex_init(); + workqueue_init(); clean_init(); fullscrape_init(); accesslist_init(); @@ -623,5 +625,6 @@ void trackerlogic_deinit(void) { } /* Release mutexes */ + workqueue_deinit(); mutex_deinit(); } -- cgit v1.3