]> mj.ucw.cz Git - libucw.git/blob - lib/workqueue.h
Added a library module for distributing work between threads.
[libucw.git] / 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;                               // 0 for default
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 struct work *raw_queue_try_get(struct raw_queue *q);
80
81 void work_queue_init(struct worker_pool *p, struct work_queue *q);
82 void work_queue_cleanup(struct work_queue *q);
83 void work_submit(struct work_queue *q, struct work *w);
84 struct work *work_wait(struct work_queue *q);
85 struct work *work_try_wait(struct work_queue *q);
86
87 #endif  /* !_UCW_WORKQUEUE_H */