1 files changed, 241 insertions, 0 deletions

diff --git a/src/ot_workqueue.c b/src/ot_workqueue.c
new file mode 100644
index 0000000..56a7cd2
--- /dev/null
+++ b/src/ot_workqueue.c
@@ -0,0 +1,241 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever. */
+
+/* System */
+#include <pthread.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/uio.h>
+#include <unistd.h>
+
+/* 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);
+}