lib/workqueue.h

   1 /*
   2  *      UCW Library -- Thread Pools and Work Queues
   3  *
   4  *      (c) 2006 Martin Mares <mj@ucw.cz>
   5  *
   6  *      This software may be freely distributed and used according to the terms
   7  *      of the GNU Lesser General Public License.
   8  */
   9
  10 #ifndef _UCW_WORKQUEUE_H
  11 #define _UCW_WORKQUEUE_H
  12
  13 /*
  14  *  A thread pool is a set of threads receiving work requests from a common queue,
  15  *  each work request contains a pointer to a function inside the thread.
  16  *
  17  *  A work queue is an interface for submitting work requests. It's bound to a single
  18  *  thread pool, it remembers running requests and gathers replies. A single work queue
  19  *  should not be used by multiple threads simultaneously.
  20  *
  21  *  When a thread pool is initialized, new_thread() is called for every thread first,
  22  *  allocating struct worker_thread (and user-defined thread context following it) for
  23  *  each thread. Then the threads are fired and each of them executes the init_thread()
  24  *  callback. These callbacks are serialized and worker_pool_init() function waits
  25  *  until all of them finish.
  26  */
  27
  28 #include "lib/semaphore.h"
  29 #include "lib/clists.h"
  30
  31 #include <pthread.h>
  32
  33 struct worker_thread {                          // One of threads serving requests
  34   cnode n;
  35   pthread_t thread;
  36   struct worker_pool *pool;
  37   int id;                                       // Inside the pool
  38   /* user-defined data can follow */
  39 };
  40
  41 struct raw_queue {                              // Generic queue with locking
  42   pthread_mutex_t queue_mutex;
  43   clist queue;
  44   sem_t *queue_sem;                             // Number of requests queued
  45 };
  46
  47 struct worker_pool {
  48   struct raw_queue requests;
  49   uns num_threads;
  50   uns stack_size;
  51   struct worker_thread *(*new_thread)(void);    // default: xmalloc the struct
  52   void (*free_thread)(struct worker_thread *t); // default: xfree
  53   void (*init_thread)(struct worker_thread *t); // default: empty
  54   void (*cleanup_thread)(struct worker_thread *t); // default: empty
  55   clist worker_threads;
  56   sem_t *init_cleanup_sem;
  57 };
  58
  59 struct work_queue {
  60   struct worker_pool *pool;
  61   uns nr_running;                               // Number of requests in service
  62   struct raw_queue finished;                    // Finished requests queue up here
  63 };
  64
  65 struct work {                                   // A single request
  66   cnode n;
  67   struct work_queue *reply_to;                  // Where to queue the request when it's finished
  68   void (*go)(struct worker_thread *t, struct work *w);          // Called inside the worker thread
  69   void (*returned)(struct work_queue *q, struct work *w);       // Called when returned back, NULL if work_wait should return
  70 };
  71
  72 void worker_pool_init(struct worker_pool *p);
  73 void worker_pool_cleanup(struct worker_pool *p);
  74
  75 void raw_queue_init(struct raw_queue *q);
  76 void raw_queue_cleanup(struct raw_queue *q);
  77 void raw_queue_put(struct raw_queue *q, struct work *w);
  78 struct work *raw_queue_get(struct raw_queue *q);
  79
  80 void work_queue_init(struct worker_pool *p, struct work_queue *q);
  81 void work_queue_cleanup(struct work_queue *q);
  82 void work_submit(struct work_queue *q, struct work *w);
  83 struct work *work_wait(struct work_queue *q);
  84
  85 #endif  /* !_UCW_WORKQUEUE_H */