ucw/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 #ifdef CONFIG_UCW_CLEAN_ABI
  14 #define raw_queue_cleanup ucw_raw_queue_cleanup
  15 #define raw_queue_get ucw_raw_queue_get
  16 #define raw_queue_init ucw_raw_queue_init
  17 #define raw_queue_put ucw_raw_queue_put
  18 #define raw_queue_try_get ucw_raw_queue_try_get
  19 #define work_queue_cleanup ucw_work_queue_cleanup
  20 #define work_queue_init ucw_work_queue_init
  21 #define work_submit ucw_work_submit
  22 #define work_try_wait ucw_work_try_wait
  23 #define work_wait ucw_work_wait
  24 #define worker_pool_cleanup ucw_worker_pool_cleanup
  25 #define worker_pool_init ucw_worker_pool_init
  26 #endif
  27
  28 /*
  29  *  A thread pool is a set of threads receiving work requests from a common queue,
  30  *  each work request contains a pointer to a function inside the thread.
  31  *
  32  *  A work queue is an interface for submitting work requests. It's bound to a single
  33  *  thread pool, it remembers running requests and gathers replies. A single work queue
  34  *  should not be used by multiple threads simultaneously.
  35  *
  36  *  Requests can have priorities. Requests with the highest priority are served first.
  37  *  Requests of priority 0 are guaranteed to be served on first-come-first-served
  38  *  basis, requests of higher priorities are unordered.
  39  *
  40  *  When a thread pool is initialized, new_thread() is called for every thread first,
  41  *  allocating struct worker_thread (and user-defined thread context following it) for
  42  *  each thread. Then the threads are fired and each of them executes the init_thread()
  43  *  callback. These callbacks are serialized and worker_pool_init() function waits
  44  *  until all of them finish.
  45  */
  46
  47 #include <ucw/semaphore.h>
  48 #include <ucw/clists.h>
  49
  50 #include <pthread.h>
  51
  52 struct worker_thread {                          // One of threads serving requests
  53   cnode n;
  54   pthread_t thread;
  55   struct worker_pool *pool;
  56   int id;                                       // Inside the pool
  57   /* user-defined data can follow */
  58 };
  59
  60 struct raw_queue {                              // Generic queue with locking
  61   pthread_mutex_t queue_mutex;
  62   clist pri0_queue;                             // Ordinary queue for requests with priority=0
  63   struct work **pri_heap;                       // A heap for request with priority>0
  64   uns heap_cnt, heap_max;
  65   sem_t *queue_sem;                             // Number of requests queued
  66 };
  67
  68 struct worker_pool {
  69   struct raw_queue requests;
  70   uns num_threads;
  71   uns stack_size;                               // 0 for default
  72   struct worker_thread *(*new_thread)(void);    // default: xmalloc the struct
  73   void (*free_thread)(struct worker_thread *t); // default: xfree
  74   void (*init_thread)(struct worker_thread *t); // default: empty
  75   void (*cleanup_thread)(struct worker_thread *t); // default: empty
  76   clist worker_threads;
  77   sem_t *init_cleanup_sem;
  78 };
  79
  80 struct work_queue {
  81   struct worker_pool *pool;
  82   uns nr_running;                               // Number of requests in service
  83   struct raw_queue finished;                    // Finished requests queue up here
  84 };
  85
  86 struct work {                                   // A single request
  87   cnode n;
  88   uns priority;
  89   struct work_queue *reply_to;                  // Where to queue the request when it's finished
  90   void (*go)(struct worker_thread *t, struct work *w);          // Called inside the worker thread
  91 };
  92
  93 void worker_pool_init(struct worker_pool *p);
  94 void worker_pool_cleanup(struct worker_pool *p);
  95
  96 void raw_queue_init(struct raw_queue *q);
  97 void raw_queue_cleanup(struct raw_queue *q);
  98 void raw_queue_put(struct raw_queue *q, struct work *w);
  99 struct work *raw_queue_get(struct raw_queue *q);
 100 struct work *raw_queue_try_get(struct raw_queue *q);
 101
 102 void work_queue_init(struct worker_pool *p, struct work_queue *q);
 103 void work_queue_cleanup(struct work_queue *q);
 104 void work_submit(struct work_queue *q, struct work *w);
 105 struct work *work_wait(struct work_queue *q);
 106 struct work *work_try_wait(struct work_queue *q);
 107
 108 #endif  /* !_UCW_WORKQUEUE_H */