5 Author: Pekka Riikonen <priikone@silcnet.org>
7 Copyright (C) 2007 Pekka Riikonen
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation; version 2 of the License.
13 This program is distributed in the hope that it will be useful,
14 but WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 GNU General Public License for more details.
22 /***************************** Thread Pool API *****************************/
24 /* Explanation of the thread pool execution.
26 When new call is added to thread pool by calling silc_thread_pool_run
27 it is assigned to a first free thread from the free list. If no threads
28 are available we take one from the threads list and assign the call to
29 its queue. The threads list always takes different thread finally wrapping
30 from the beginning. This way each thread will get a chance to execute
33 The thread function silc_thread_pool_run_thread executes each call. After
34 executing the current call that has been assigned to it, it will check
35 if there are any queued calls in its queue, and it will execute all calls
36 from the queue. If there aren't any calls in the queue, it will attempt
37 to steal a call from some other thread and execute it.
39 The queue list is always consumed in last-in-first-out order. The most
40 recently added call gets priority. With full utilization this helps to
41 avoid CPU cache misses. Since the queues are thread specific with full
42 utilization each thread should always be doing work for the most recent
43 (and thus most important) calls. */
45 /************************** Types and definitions ***************************/
47 /* Thread pool thread context. Each thread contains the most current call
48 to be executed, and a list of queued calls. */
49 typedef struct SilcThreadPoolThreadStruct {
50 struct SilcThreadPoolThreadStruct *next;
51 struct SilcThreadPoolThreadStruct *next2;
52 SilcThreadPool tp; /* The thread pool */
53 SilcCond thread_signal; /* Condition variable for signalling */
54 SilcMutex lock; /* Thread lock */
55 SilcList queue; /* Queue for waiting calls */
56 SilcList free_queue; /* Queue freelist */
57 SilcSchedule schedule; /* The current Scheduler, may be NULL */
58 SilcThreadPoolFunc run; /* The current call to run in a thread */
59 SilcTaskCallback completion; /* The current Completion function */
61 void *completion_context;
62 unsigned int stop : 1; /* Set to stop the thread */
63 } *SilcThreadPoolThread;
65 /* Thread pool context */
66 struct SilcThreadPoolStruct {
67 SilcStack stack; /* Stack for memory allocation */
68 SilcCond pool_signal; /* Condition variable for signalling */
69 SilcMutex lock; /* Pool lock */
70 SilcList threads; /* Threads in the pool */
71 SilcList free_threads; /* Threads freelist */
72 SilcUInt16 min_threads; /* Minimum threads in the pool */
73 SilcUInt16 max_threads; /* Maximum threads in the pool */
74 SilcUInt16 refcnt; /* Reference counter */
75 unsigned int destroy : 1; /* Set when pool is to be destroyed */
78 /************************ Static utility functions **************************/
80 /* Reference thread pool. Must be called locked. */
82 static void silc_thread_pool_ref(SilcThreadPool tp)
85 SILC_LOG_DEBUG(("Thread pool %p, refcnt %d -> %d", tp, tp->refcnt - 1,
89 /* Unreference thread pool. Must be called locked. Releases the lock. */
91 static void silc_thread_pool_unref(SilcThreadPool tp)
94 SILC_LOG_DEBUG(("Thread pool %p refcnt %d -> %d", tp, tp->refcnt + 1,
97 SilcStack stack = tp->stack;
98 silc_mutex_unlock(tp->lock);
99 silc_mutex_free(tp->lock);
100 silc_cond_free(tp->pool_signal);
101 silc_sfree(stack, tp);
102 silc_stack_free(stack);
105 silc_mutex_unlock(tp->lock);
108 /* The thread executor. Each thread in the pool is run here. They wait
109 here for something to do which is given to them by silc_thread_pool_run. */
111 static void *silc_thread_pool_run_thread(void *context)
113 SilcThreadPoolThread t = context, o, q;
114 SilcThreadPool tp = t->tp;
115 SilcMutex lock = t->lock;
116 SilcCond thread_signal = t->thread_signal;
118 silc_mutex_lock(lock);
121 /* Wait here for code to execute */
122 while (!t->run && !t->stop)
123 silc_cond_wait(thread_signal, lock);
129 silc_mutex_unlock(lock);
131 SILC_LOG_DEBUG(("Execute call %p, context %p, thread %p", t->run,
133 t->run(t->schedule, t->run_context);
135 /* If scheduler is NULL, call completion directly from here. Otherwise
136 it is called through the scheduler in the thread where the scheduler
140 SILC_LOG_DEBUG(("Run completion through scheduler %p", t->schedule));
141 if (!silc_schedule_task_add_timeout(t->schedule, t->completion,
142 t->completion_context, 0, 0)) {
143 SILC_LOG_DEBUG(("Run completion directly"));
144 t->completion(NULL, NULL, 0, 0, t->completion_context);
146 silc_schedule_wakeup(t->schedule);
148 SILC_LOG_DEBUG(("Run completion directly"));
149 t->completion(NULL, NULL, 0, 0, t->completion_context);
153 silc_mutex_lock(lock);
157 /* Check if there are calls in queue. Takes the most recently added
158 call since new ones are added at the start of the list. */
159 if (silc_list_count(t->queue) > 0) {
161 silc_list_start(t->queue);
162 q = silc_list_get(t->queue);
164 SILC_LOG_DEBUG(("Execute call from queue"));
166 /* Execute this call now */
168 t->run_context = q->run_context;
169 t->completion = q->completion;
170 t->completion_context = q->completion_context;
171 t->schedule = q->schedule;
173 silc_list_del(t->queue, q);
174 silc_list_add(t->free_queue, q);
175 silc_mutex_unlock(lock);
179 silc_mutex_unlock(lock);
180 silc_mutex_lock(tp->lock);
182 /* Nothing to do. Attempt to steal call from some other thread. */
183 o = silc_list_get(tp->threads);
185 /* List wraps around */
186 silc_list_start(tp->threads);
187 o = silc_list_get(tp->threads);
190 /* Check that the other thread is valid and has something to execute. */
191 silc_mutex_lock(o->lock);
192 if (o == t || o->stop || silc_list_count(o->queue) == 0) {
193 silc_mutex_unlock(o->lock);
198 silc_mutex_unlock(tp->lock);
199 silc_list_start(o->queue);
200 q = silc_list_get(o->queue);
202 SILC_LOG_DEBUG(("Execute call from queue from thread %p", o));
204 /* Execute this call now */
206 t->run_context = q->run_context;
207 t->completion = q->completion;
208 t->completion_context = q->completion_context;
209 t->schedule = q->schedule;
211 silc_list_del(o->queue, q);
212 silc_list_add(o->free_queue, q);
213 silc_mutex_unlock(o->lock);
217 silc_mutex_lock(lock);
219 silc_mutex_unlock(tp->lock);
223 /* Now that we have the lock back, check the queue again. */
224 if (silc_list_count(t->queue) > 0) {
225 silc_mutex_unlock(tp->lock);
229 /* The thread is now free for use again. */
231 t->completion = NULL;
233 silc_list_add(tp->free_threads, t);
234 silc_mutex_unlock(tp->lock);
238 /* Stop the thread. Remove from threads list. */
239 SILC_LOG_DEBUG(("Stop thread %p", t));
241 /* We can unlock the thread now. After we get the thread pool lock
242 no one can retrieve the thread anymore. */
243 silc_mutex_unlock(lock);
244 silc_mutex_lock(tp->lock);
246 silc_list_del(tp->threads, t);
247 silc_list_start(tp->threads);
249 /* Clear thread's call queue. */
250 silc_list_start(t->queue);
251 silc_list_start(t->free_queue);
252 while ((q = silc_list_get(t->queue)))
253 silc_sfree(tp->stack, q);
254 while ((q = silc_list_get(t->free_queue)))
255 silc_sfree(tp->stack, q);
257 /* Destroy the thread */
258 silc_mutex_free(lock);
259 silc_cond_free(thread_signal);
260 silc_sfree(tp->stack, t);
262 /* If we are last thread, signal the waiting destructor. */
263 if (silc_list_count(tp->threads) == 0)
264 silc_cond_signal(tp->pool_signal);
266 /* Release pool reference. Releases lock also. */
267 silc_thread_pool_unref(tp);
272 /* Creates new thread to thread pool */
274 static SilcThreadPoolThread silc_thread_pool_new_thread(SilcThreadPool tp)
276 SilcThreadPoolThread t;
278 t = silc_scalloc(tp->stack, 1, sizeof(*t));
282 if (!silc_mutex_alloc(&t->lock)) {
283 silc_sfree(tp->stack, t);
287 if (!silc_cond_alloc(&t->thread_signal)) {
288 silc_mutex_free(t->lock);
289 silc_sfree(tp->stack, t);
294 silc_list_init(t->queue, struct SilcThreadPoolThreadStruct, next);
295 silc_list_init(t->free_queue, struct SilcThreadPoolThreadStruct, next);
297 /* Add to thread pool */
298 silc_list_add(tp->threads, t);
299 silc_list_add(tp->free_threads, t);
300 silc_thread_pool_ref(tp);
302 SILC_LOG_DEBUG(("Start thread %p", t));
304 /* Start the thread */
305 silc_thread_create(silc_thread_pool_run_thread, t, FALSE);
310 /**************************** Thread Pool API *******************************/
312 /* Allocate thread pool */
314 SilcThreadPool silc_thread_pool_alloc(SilcStack stack,
315 SilcUInt32 min_threads,
316 SilcUInt32 max_threads,
317 SilcBool start_min_threads)
322 if (max_threads < min_threads)
328 stack = silc_stack_alloc(0, stack);
330 tp = silc_scalloc(stack, 1, sizeof(*tp));
332 silc_stack_free(stack);
336 SILC_LOG_DEBUG(("Starting thread pool %p, min threads %d, max threads %d",
337 tp, min_threads, max_threads));
340 tp->min_threads = min_threads;
341 tp->max_threads = max_threads;
344 if (!silc_mutex_alloc(&tp->lock)) {
345 silc_sfree(stack, tp);
346 silc_stack_free(stack);
350 if (!silc_cond_alloc(&tp->pool_signal)) {
351 silc_mutex_free(tp->lock);
352 silc_sfree(stack, tp);
353 silc_stack_free(stack);
357 silc_list_init(tp->threads, struct SilcThreadPoolThreadStruct, next);
358 silc_list_init(tp->free_threads, struct SilcThreadPoolThreadStruct, next2);
360 for (i = 0; i < tp->min_threads && start_min_threads; i++)
361 silc_thread_pool_new_thread(tp);
363 silc_list_start(tp->threads);
368 /* Free thread pool */
370 void silc_thread_pool_free(SilcThreadPool tp, SilcBool wait_unfinished)
372 SilcThreadPoolThread t;
374 SILC_LOG_DEBUG(("Free thread pool %p", tp));
376 silc_mutex_lock(tp->lock);
380 silc_list_start(tp->threads);
381 while ((t = silc_list_get(tp->threads))) {
382 silc_mutex_lock(t->lock);
384 silc_cond_signal(t->thread_signal);
385 silc_mutex_unlock(t->lock);
388 if (wait_unfinished) {
389 SILC_LOG_DEBUG(("Wait threads to finish"));
390 while (silc_list_count(tp->threads))
391 silc_cond_wait(tp->pool_signal, tp->lock);
394 /* Release reference. Releases lock also. */
395 silc_thread_pool_unref(tp);
398 /* Execute code in a thread in the pool */
400 SilcBool silc_thread_pool_run(SilcThreadPool tp,
402 SilcSchedule schedule,
403 SilcThreadPoolFunc run,
405 SilcTaskCallback completion,
406 void *completion_context)
408 SilcThreadPoolThread t, q;
410 silc_mutex_lock(tp->lock);
413 silc_mutex_unlock(tp->lock);
417 /* Get free thread */
418 silc_list_start(tp->free_threads);
419 t = silc_list_get(tp->free_threads);
421 if (silc_list_count(tp->threads) + 1 > tp->max_threads) {
422 /* Maximum threads reached */
424 silc_mutex_unlock(tp->lock);
428 /* User wants to queue this call until thread becomes free. Get
429 a thread to assign this call. */
430 t = silc_list_get(tp->threads);
432 /* List wraps around */
433 silc_list_start(tp->threads);
434 t = silc_list_get(tp->threads);
436 silc_mutex_unlock(tp->lock);
438 SILC_LOG_DEBUG(("Queue call %p, context %p in thread %p",
439 run, run_context, t));
441 silc_mutex_lock(t->lock);
443 /* Get free call context from the list */
444 silc_list_start(t->free_queue);
445 q = silc_list_get(t->free_queue);
447 q = silc_scalloc(tp->stack, 1, sizeof(*q));
449 silc_mutex_unlock(t->lock);
453 silc_list_del(t->free_queue, q);
457 q->run_context = run_context;
458 q->completion = completion;
459 q->completion_context = completion_context;
460 q->schedule = schedule;
462 /* Add at the start of the list. It gets executed first. */
463 silc_list_insert(t->queue, NULL, q);
464 silc_mutex_unlock(t->lock);
467 /* Create new thread */
468 t = silc_thread_pool_new_thread(tp);
470 silc_mutex_unlock(tp->lock);
476 silc_list_del(tp->free_threads, t);
477 silc_mutex_unlock(tp->lock);
479 SILC_LOG_DEBUG(("Run call %p, context %p, thread %p", run, run_context, t));
481 silc_mutex_lock(t->lock);
483 /* Mark this call to be executed in this thread */
485 t->run_context = run_context;
486 t->completion = completion;
487 t->completion_context = completion_context;
488 t->schedule = schedule;
490 /* Signal the thread */
491 silc_cond_signal(t->thread_signal);
492 silc_mutex_unlock(t->lock);
497 /* Set maximum threads in the pool */
499 void silc_thread_pool_set_max_threads(SilcThreadPool tp,
500 SilcUInt32 max_threads)
502 SILC_LOG_DEBUG(("Set thread pool %p max threads to %d", tp, max_threads));
504 silc_mutex_lock(tp->lock);
505 tp->max_threads = max_threads;
506 silc_mutex_unlock(tp->lock);
509 /* Get maximum threads in the pool */
511 SilcUInt32 silc_thread_pool_get_max_threads(SilcThreadPool tp)
513 SilcUInt32 max_threads;
515 silc_mutex_lock(tp->lock);
516 max_threads = tp->max_threads;
517 silc_mutex_unlock(tp->lock);
522 /* Get numnber of free threads in the pool */
524 SilcUInt32 silc_thread_pool_num_free_threads(SilcThreadPool tp)
526 SilcUInt32 free_threads;
528 silc_mutex_lock(tp->lock);
529 free_threads = silc_list_count(tp->free_threads);
530 silc_mutex_unlock(tp->lock);
537 void silc_thread_pool_purge(SilcThreadPool tp)
539 SilcThreadPoolThread t;
542 silc_mutex_lock(tp->lock);
544 if (silc_list_count(tp->free_threads) <= tp->min_threads) {
545 SILC_LOG_DEBUG(("No threads to purge"));
546 silc_mutex_unlock(tp->lock);
550 i = silc_list_count(tp->free_threads) - tp->min_threads;
552 SILC_LOG_DEBUG(("Purge %d threads", i));
554 silc_list_start(tp->threads);
555 while ((t = silc_list_get(tp->threads))) {
556 silc_mutex_lock(t->lock);
558 silc_mutex_unlock(t->lock);
562 /* Signal the thread to stop */
564 silc_cond_signal(t->thread_signal);
565 silc_mutex_unlock(t->lock);
567 silc_list_del(tp->free_threads, t);
574 silc_list_start(tp->threads);
575 silc_mutex_unlock(tp->lock);
578 /*************************** Thread-local Storage ***************************/
580 void silc_thread_tls_set(void *context)
582 SilcTls tls = silc_thread_get_tls();
585 /* Initialize Tls for this thread */
586 tls = silc_thread_tls_init();
591 tls->thread_context = context;
594 void *silc_thread_tls_get(void)
596 SilcTls tls = silc_thread_get_tls();
599 return tls->thread_context;