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 --- ot_workqueue.c | 243 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 243 insertions(+) create mode 100644 ot_workqueue.c (limited to 'ot_workqueue.c') 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); +} -- cgit v1.3