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 /*
  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  *  Requests can have priorities. Requests with the highest priority are served first.
  22  *  Requests of priority 0 are guaranteed to be served on first-come-first-served
  23  *  basis, requests of higher priorities are unordered.
  24  *
  25  *  When a thread pool is initialized, new_thread() is called for every thread first,
  26  *  allocating struct worker_thread (and user-defined thread context following it) for
  27  *  each thread. Then the threads are fired and each of them executes the init_thread()
  28  *  callback. These callbacks are serialized and worker_pool_init() function waits
  29  *  until all of them finish.
  30  */
  31
  32 #include "ucw/semaphore.h"
  33 #include "ucw/clists.h"
  34
  35 #include <pthread.h>
  36
  37 struct worker_thread {                          // One of threads serving requests
  38   cnode n;
  39   pthread_t thread;
  40   struct worker_pool *pool;
  41   int id;                                       // Inside the pool
  42   /* user-defined data can follow */
  43 };
  44
  45 struct raw_queue {                              // Generic queue with locking
  46   pthread_mutex_t queue_mutex;
  47   clist pri0_queue;                             // Ordinary queue for requests with priority=0
  48   struct work **pri_heap;                       // A heap for request with priority>0
  49   uns heap_cnt, heap_max;
  50   sem_t *queue_sem;                             // Number of requests queued
  51 };
  52
  53 struct worker_pool {
  54   struct raw_queue requests;
  55   uns num_threads;
  56   uns stack_size;                               // 0 for default
  57   struct worker_thread *(*new_thread)(void);    // default: xmalloc the struct
  58   void (*free_thread)(struct worker_thread *t); // default: xfree
  59   void (*init_thread)(struct worker_thread *t); // default: empty
  60   void (*cleanup_thread)(struct worker_thread *t); // default: empty
  61   clist worker_threads;
  62   sem_t *init_cleanup_sem;
  63 };
  64
  65 struct work_queue {
  66   struct worker_pool *pool;
  67   uns nr_running;                               // Number of requests in service
  68   struct raw_queue finished;                    // Finished requests queue up here
  69 };
  70
  71 struct work {                                   // A single request
  72   cnode n;
  73   uns priority;
  74   struct work_queue *reply_to;                  // Where to queue the request when it's finished
  75   void (*go)(struct worker_thread *t, struct work *w);          // Called inside the worker thread
  76 };
  77
  78 void worker_pool_init(struct worker_pool *p);
  79 void worker_pool_cleanup(struct worker_pool *p);
  80
  81 void raw_queue_init(struct raw_queue *q);
  82 void raw_queue_cleanup(struct raw_queue *q);
  83 void raw_queue_put(struct raw_queue *q, struct work *w);
  84 struct work *raw_queue_get(struct raw_queue *q);
  85 struct work *raw_queue_try_get(struct raw_queue *q);
  86
  87 void work_queue_init(struct worker_pool *p, struct work_queue *q);
  88 void work_queue_cleanup(struct work_queue *q);
  89 void work_submit(struct work_queue *q, struct work *w);
  90 struct work *work_wait(struct work_queue *q);
  91 struct work *work_try_wait(struct work_queue *q);
  92
  93 #endif  /* !_UCW_WORKQUEUE_H */