MINOR: task: introduce work lists

Sometimes we need to delegate some list processing to a function running
on another thread. In this case the list element will simply be queued
into a dedicated self-locked list and the task responsible for this list
will be woken up, calling the associated function which will run over the
list.

This is what work_list does. Such lists will be dedicated to a limited
type of work but will significantly ease such remote handling. A function
is provided to create these per-thread lists, their tasks and to properly
bind each task to a distinct thread, so that the caller only has to store
the resulting pointer to the start of the structure.

These structures should not be abused though as each head will consume
4 pointers per thread, hence 32 bytes per thread or 2 kB for 64 threads.

diff --git a/include/proto/task.h b/include/proto/task.h
index 424e8ba66775ba0069bc50335124e58f4eaa16d6..4044bd459c0c7b579061dc2221572f9d428d5cd3 100644 (file)
--- a/include/proto/task.h
+++ b/include/proto/task.h
@@ -541,6 +541,19 @@ static inline int thread_has_tasks(void)
                !LIST_ISEMPTY(&task_per_thread[tid].task_list));
 }
 
+/* adds list item <item> to work list <work> and wake up the associated task */
+static inline void work_list_add(struct work_list *work, struct list *item)
+{
+       LIST_ADDQ_LOCKED(&work->head, item);
+       task_wakeup(work->task, TASK_WOKEN_OTHER);
+}
+
+struct work_list *work_list_create(int nbthread,
+                                   struct task *(*fct)(struct task *, void *, unsigned short),
+                                   void *arg);
+
+void work_list_destroy(struct work_list *work, int nbthread);
+
 /*
  * This does 3 things :
  *   - wake up all expired tasks

diff --git a/include/types/task.h b/include/types/task.h
index ab909d0f8f33f809a5ca766620c2a6200329f3f5..a94905893998b1099d7259e9aa7285736b88ac51 100644 (file)
--- a/include/types/task.h
+++ b/include/types/task.h
@@ -106,6 +106,23 @@ struct tasklet {
  * expire timer. The scheduler will requeue the task at the proper location.
  */
 
+
+/* A work_list is a thread-safe way to enqueue some work to be run on another
+ * thread. It consists of a list, a task and a general-purpose argument.
+ * A work is appended to the list by atomically adding a list element to the
+ * list and waking up the associated task, which is done using work_add(). The
+ * caller must be careful about how operations are run as it will definitely
+ * happen that the element being enqueued is processed by the other thread
+ * before the call returns. Some locking conventions between the caller and the
+ * callee might sometimes be necessary. The task is always woken up with reason
+ * TASK_WOKEN_OTHER and a context pointing to the work_list entry.
+ */
+struct work_list {
+       struct list head;
+       struct task *task;
+       void *arg;
+};
+
 #endif /* _TYPES_TASK_H */
 
 /*

diff --git a/src/task.c b/src/task.c
index 2176a91905942e40e557ebe3925107f1fc5d92a2..9a55a604f83528e47b044d1fc380893060610808 100644 (file)
--- a/src/task.c
+++ b/src/task.c
@@ -447,6 +447,51 @@ void process_runnable_tasks()
                activity[tid].long_rq++;
 }
 
+/* create a work list array for <nbthread> threads, using tasks made of
+ * function <fct>. The context passed to the function will be the pointer to
+ * the thread's work list, which will contain a copy of argument <arg>. The
+ * wake up reason will be TASK_WOKEN_OTHER. The pointer to the work_list array
+ * is returned on success, otherwise NULL on failure.
+ */
+struct work_list *work_list_create(int nbthread,
+                                   struct task *(*fct)(struct task *, void *, unsigned short),
+                                   void *arg)
+{
+       struct work_list *wl;
+       int i;
+
+       wl = calloc(nbthread, sizeof(*wl));
+       if (!wl)
+               goto fail;
+
+       for (i = 0; i < nbthread; i++) {
+               LIST_INIT(&wl[i].head);
+               wl[i].task = task_new(1UL << i);
+               if (!wl[i].task)
+                       goto fail;
+               wl[i].task->process = fct;
+               wl[i].task->context = &wl[i];
+               wl[i].arg = arg;
+       }
+       return wl;
+
+ fail:
+       work_list_destroy(wl, nbthread);
+       return NULL;
+}
+
+/* destroy work list <work> */
+void work_list_destroy(struct work_list *work, int nbthread)
+{
+       int t;
+
+       if (!work)
+               return;
+       for (t = 0; t < nbthread; t++)
+               task_destroy(work[t].task);
+       free(work);
+}
+
 /*
  * Delete every tasks before running the master polling loop
  */