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_mutex.c | 221 ------------------------------------------------------------- 1 file changed, 221 deletions(-) (limited to 'ot_mutex.c') 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)); } -- cgit v1.3