lib/silcutil/silcthread.c

   1 /*
   2
   3   silcthread.c
   4
   5   Author: Pekka Riikonen <priikone@silcnet.org>
   6
   7   Copyright (C) 2007 Pekka Riikonen
   8
   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.
  12
  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.
  17
  18 */
  19
  20 #include "silc.h"
  21
  22 /***************************** Thread Pool API *****************************/
  23
  24 /* Explanation of the thread pool execution.
  25
  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
  31    queued calls.
  32
  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.
  38
  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. */
  44
  45 /************************** Types and definitions ***************************/
  46
  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 */
  60   void *run_context;
  61   void *completion_context;
  62   unsigned int stop        : 1;     /* Set to stop the thread */
  63 } *SilcThreadPoolThread;
  64
  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 */
  76 };
  77
  78 /************************ Static utility functions **************************/
  79
  80 /* Reference thread pool.  Must be called locked. */
  81
  82 static void silc_thread_pool_ref(SilcThreadPool tp)
  83 {
  84   tp->refcnt++;
  85   SILC_LOG_DEBUG(("Thread pool %p, refcnt %d -> %d", tp, tp->refcnt - 1,
  86                   tp->refcnt));
  87 }
  88
  89 /* Unreference thread pool.  Must be called locked.  Releases the lock. */
  90
  91 static void silc_thread_pool_unref(SilcThreadPool tp)
  92 {
  93   tp->refcnt--;
  94   SILC_LOG_DEBUG(("Thread pool %p refcnt %d -> %d", tp, tp->refcnt + 1,
  95                   tp->refcnt));
  96   if (!tp->refcnt) {
  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);
 103     return;
 104   }
 105   silc_mutex_unlock(tp->lock);
 106 }
 107
 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. */
 110
 111 static void *silc_thread_pool_run_thread(void *context)
 112 {
 113   SilcThreadPoolThread t = context, o, q;
 114   SilcThreadPool tp = t->tp;
 115   SilcMutex lock = t->lock;
 116   SilcCond thread_signal = t->thread_signal;
 117
 118   silc_mutex_lock(lock);
 119
 120   while (1) {
 121     /* Wait here for code to execute */
 122     while (!t->run && !t->stop)
 123       silc_cond_wait(thread_signal, lock);
 124
 125     if (t->stop)
 126       goto stop;
 127
 128     /* Execute code */
 129     silc_mutex_unlock(lock);
 130   execute:
 131     SILC_LOG_DEBUG(("Execute call %p, context %p, thread %p", t->run,
 132                     t->run_context, t));
 133     t->run(t->schedule, t->run_context);
 134
 135     /* If scheduler is NULL, call completion directly from here.  Otherwise
 136        it is called through the scheduler in the thread where the scheduler
 137        is running. */
 138     if (t->completion) {
 139       if (t->schedule) {
 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);
 145         }
 146         silc_schedule_wakeup(t->schedule);
 147       } else {
 148         SILC_LOG_DEBUG(("Run completion directly"));
 149         t->completion(NULL, NULL, 0, 0, t->completion_context);
 150       }
 151     }
 152
 153     silc_mutex_lock(lock);
 154     if (t->stop)
 155       goto stop;
 156
 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) {
 160     execute_queue:
 161       silc_list_start(t->queue);
 162       q = silc_list_get(t->queue);
 163
 164       SILC_LOG_DEBUG(("Execute call from queue"));
 165
 166       /* Execute this call now */
 167       t->run = q->run;
 168       t->run_context = q->run_context;
 169       t->completion = q->completion;
 170       t->completion_context = q->completion_context;
 171       t->schedule = q->schedule;
 172
 173       silc_list_del(t->queue, q);
 174       silc_list_add(t->free_queue, q);
 175       silc_mutex_unlock(lock);
 176       goto execute;
 177     }
 178
 179     silc_mutex_unlock(lock);
 180     silc_mutex_lock(tp->lock);
 181
 182     /* Nothing to do.  Attempt to steal call from some other thread. */
 183     o = silc_list_get(tp->threads);
 184     if (!o) {
 185       /* List wraps around */
 186       silc_list_start(tp->threads);
 187       o = silc_list_get(tp->threads);
 188     }
 189
 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);
 194       o = NULL;
 195     }
 196
 197     if (o) {
 198       silc_mutex_unlock(tp->lock);
 199       silc_list_start(o->queue);
 200       q = silc_list_get(o->queue);
 201
 202       SILC_LOG_DEBUG(("Execute call from queue from thread %p", o));
 203
 204       /* Execute this call now */
 205       t->run = q->run;
 206       t->run_context = q->run_context;
 207       t->completion = q->completion;
 208       t->completion_context = q->completion_context;
 209       t->schedule = q->schedule;
 210
 211       silc_list_del(o->queue, q);
 212       silc_list_add(o->free_queue, q);
 213       silc_mutex_unlock(o->lock);
 214       goto execute;
 215     }
 216
 217     silc_mutex_lock(lock);
 218     if (t->stop) {
 219       silc_mutex_unlock(tp->lock);
 220       goto stop;
 221     }
 222
 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);
 226       goto execute_queue;
 227     }
 228
 229     /* The thread is now free for use again. */
 230     t->run = NULL;
 231     t->completion = NULL;
 232     t->schedule = NULL;
 233     silc_list_add(tp->free_threads, t);
 234     silc_mutex_unlock(tp->lock);
 235   }
 236
 237  stop:
 238   /* Stop the thread.  Remove from threads list. */
 239   SILC_LOG_DEBUG(("Stop thread %p", t));
 240
 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);
 245
 246   silc_list_del(tp->threads, t);
 247   silc_list_start(tp->threads);
 248
 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);
 256
 257   /* Destroy the thread */
 258   silc_mutex_free(lock);
 259   silc_cond_free(thread_signal);
 260   silc_sfree(tp->stack, t);
 261
 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);
 265
 266   /* Release pool reference.  Releases lock also. */
 267   silc_thread_pool_unref(tp);
 268
 269   return NULL;
 270 }
 271
 272 /* Creates new thread to thread pool */
 273
 274 static SilcThreadPoolThread silc_thread_pool_new_thread(SilcThreadPool tp)
 275 {
 276   SilcThreadPoolThread t;
 277
 278   t = silc_scalloc(tp->stack, 1, sizeof(*t));
 279   if (!t)
 280     return NULL;
 281
 282   if (!silc_mutex_alloc(&t->lock)) {
 283     silc_sfree(tp->stack, t);
 284     return NULL;
 285   }
 286
 287   if (!silc_cond_alloc(&t->thread_signal)) {
 288     silc_mutex_free(t->lock);
 289     silc_sfree(tp->stack, t);
 290     return NULL;
 291   }
 292
 293   t->tp = tp;
 294   silc_list_init(t->queue, struct SilcThreadPoolThreadStruct, next);
 295   silc_list_init(t->free_queue, struct SilcThreadPoolThreadStruct, next);
 296
 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);
 301
 302   SILC_LOG_DEBUG(("Start thread %p", t));
 303
 304   /* Start the thread */
 305   silc_thread_create(silc_thread_pool_run_thread, t, FALSE);
 306
 307   return t;
 308 }
 309
 310 /**************************** Thread Pool API *******************************/
 311
 312 /* Allocate thread pool */
 313
 314 SilcThreadPool silc_thread_pool_alloc(SilcStack stack,
 315                                       SilcUInt32 min_threads,
 316                                       SilcUInt32 max_threads,
 317                                       SilcBool start_min_threads)
 318 {
 319   SilcThreadPool tp;
 320   int i;
 321
 322   if (max_threads < min_threads)
 323     return NULL;
 324   if (!max_threads)
 325     return NULL;
 326
 327   if (stack)
 328     stack = silc_stack_alloc(0, stack);
 329
 330   tp = silc_scalloc(stack, 1, sizeof(*tp));
 331   if (!tp) {
 332     silc_stack_free(stack);
 333     return NULL;
 334   }
 335
 336   SILC_LOG_DEBUG(("Starting thread pool %p, min threads %d, max threads %d",
 337                   tp, min_threads, max_threads));
 338
 339   tp->stack = stack;
 340   tp->min_threads = min_threads;
 341   tp->max_threads = max_threads;
 342   tp->refcnt++;
 343
 344   if (!silc_mutex_alloc(&tp->lock)) {
 345     silc_sfree(stack, tp);
 346     silc_stack_free(stack);
 347     return NULL;
 348   }
 349
 350   if (!silc_cond_alloc(&tp->pool_signal)) {
 351     silc_mutex_free(tp->lock);
 352     silc_sfree(stack, tp);
 353     silc_stack_free(stack);
 354     return NULL;
 355   }
 356
 357   silc_list_init(tp->threads, struct SilcThreadPoolThreadStruct, next);
 358   silc_list_init(tp->free_threads, struct SilcThreadPoolThreadStruct, next2);
 359
 360   for (i = 0; i < tp->min_threads && start_min_threads; i++)
 361     silc_thread_pool_new_thread(tp);
 362
 363   silc_list_start(tp->threads);
 364
 365   return tp;
 366 }
 367
 368 /* Free thread pool */
 369
 370 void silc_thread_pool_free(SilcThreadPool tp, SilcBool wait_unfinished)
 371 {
 372   SilcThreadPoolThread t;
 373
 374   SILC_LOG_DEBUG(("Free thread pool %p", tp));
 375
 376   silc_mutex_lock(tp->lock);
 377   tp->destroy = TRUE;
 378
 379   /* Stop threads */
 380   silc_list_start(tp->threads);
 381   while ((t = silc_list_get(tp->threads))) {
 382     silc_mutex_lock(t->lock);
 383     t->stop = TRUE;
 384     silc_cond_signal(t->thread_signal);
 385     silc_mutex_unlock(t->lock);
 386   }
 387
 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);
 392   }
 393
 394   /* Release reference.  Releases lock also. */
 395   silc_thread_pool_unref(tp);
 396 }
 397
 398 /* Execute code in a thread in the pool */
 399
 400 SilcBool silc_thread_pool_run(SilcThreadPool tp,
 401                               SilcBool queuable,
 402                               SilcSchedule schedule,
 403                               SilcThreadPoolFunc run,
 404                               void *run_context,
 405                               SilcTaskCallback completion,
 406                               void *completion_context)
 407 {
 408   SilcThreadPoolThread t, q;
 409
 410   silc_mutex_lock(tp->lock);
 411
 412   if (tp->destroy) {
 413     silc_mutex_unlock(tp->lock);
 414     return FALSE;
 415   }
 416
 417   /* Get free thread */
 418   silc_list_start(tp->free_threads);
 419   t = silc_list_get(tp->free_threads);
 420   if (!t || t->stop) {
 421     if (silc_list_count(tp->threads) + 1 > tp->max_threads) {
 422       /* Maximum threads reached */
 423       if (!queuable) {
 424         silc_mutex_unlock(tp->lock);
 425         return FALSE;
 426       }
 427
 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);
 431       if (!t) {
 432         /* List wraps around */
 433         silc_list_start(tp->threads);
 434         t = silc_list_get(tp->threads);
 435       }
 436       silc_mutex_unlock(tp->lock);
 437
 438       SILC_LOG_DEBUG(("Queue call %p, context %p in thread %p",
 439                       run, run_context, t));
 440
 441       silc_mutex_lock(t->lock);
 442
 443       /* Get free call context from the list */
 444       silc_list_start(t->free_queue);
 445       q = silc_list_get(t->free_queue);
 446       if (!q) {
 447         q = silc_scalloc(tp->stack, 1, sizeof(*q));
 448         if (!q) {
 449           silc_mutex_unlock(t->lock);
 450           return FALSE;
 451         }
 452       } else {
 453         silc_list_del(t->free_queue, q);
 454       }
 455
 456       q->run = run;
 457       q->run_context = run_context;
 458       q->completion = completion;
 459       q->completion_context = completion_context;
 460       q->schedule = schedule;
 461
 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);
 465       return TRUE;
 466     } else {
 467       /* Create new thread */
 468       t = silc_thread_pool_new_thread(tp);
 469       if (!t) {
 470         silc_mutex_unlock(tp->lock);
 471         return FALSE;
 472       }
 473     }
 474   }
 475
 476   silc_list_del(tp->free_threads, t);
 477   silc_mutex_unlock(tp->lock);
 478
 479   SILC_LOG_DEBUG(("Run call %p, context %p, thread %p", run, run_context, t));
 480
 481   silc_mutex_lock(t->lock);
 482
 483   /* Mark this call to be executed in this thread */
 484   t->run = run;
 485   t->run_context = run_context;
 486   t->completion = completion;
 487   t->completion_context = completion_context;
 488   t->schedule = schedule;
 489
 490   /* Signal the thread */
 491   silc_cond_signal(t->thread_signal);
 492   silc_mutex_unlock(t->lock);
 493
 494   return TRUE;
 495 }
 496
 497 /* Set maximum threads in the pool */
 498
 499 void silc_thread_pool_set_max_threads(SilcThreadPool tp,
 500                                       SilcUInt32 max_threads)
 501 {
 502   SILC_LOG_DEBUG(("Set thread pool %p max threads to %d", tp, max_threads));
 503
 504   silc_mutex_lock(tp->lock);
 505   tp->max_threads = max_threads;
 506   silc_mutex_unlock(tp->lock);
 507 }
 508
 509 /* Get maximum threads in the pool */
 510
 511 SilcUInt32 silc_thread_pool_get_max_threads(SilcThreadPool tp)
 512 {
 513   SilcUInt32 max_threads;
 514
 515   silc_mutex_lock(tp->lock);
 516   max_threads = tp->max_threads;
 517   silc_mutex_unlock(tp->lock);
 518
 519   return max_threads;
 520 }
 521
 522 /* Get numnber of free threads in the pool */
 523
 524 SilcUInt32 silc_thread_pool_num_free_threads(SilcThreadPool tp)
 525 {
 526   SilcUInt32 free_threads;
 527
 528   silc_mutex_lock(tp->lock);
 529   free_threads = silc_list_count(tp->free_threads);
 530   silc_mutex_unlock(tp->lock);
 531
 532   return free_threads;
 533 }
 534
 535 /* Purge pool */
 536
 537 void silc_thread_pool_purge(SilcThreadPool tp)
 538 {
 539   SilcThreadPoolThread t;
 540   int i;
 541
 542   silc_mutex_lock(tp->lock);
 543
 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);
 547     return;
 548   }
 549
 550   i = silc_list_count(tp->free_threads) - tp->min_threads;
 551
 552   SILC_LOG_DEBUG(("Purge %d threads", i));
 553
 554   silc_list_start(tp->threads);
 555   while ((t = silc_list_get(tp->threads))) {
 556     silc_mutex_lock(t->lock);
 557     if (t->run) {
 558       silc_mutex_unlock(t->lock);
 559       continue;
 560     }
 561
 562     /* Signal the thread to stop */
 563     t->stop = TRUE;
 564     silc_cond_signal(t->thread_signal);
 565     silc_mutex_unlock(t->lock);
 566
 567     silc_list_del(tp->free_threads, t);
 568
 569     i--;
 570     if (!i)
 571       break;
 572   }
 573
 574   silc_list_start(tp->threads);
 575   silc_mutex_unlock(tp->lock);
 576 }
 577
 578 /*************************** Thread-local Storage ***************************/
 579
 580 void silc_thread_tls_set(void *context)
 581 {
 582   SilcTls tls = silc_thread_get_tls();
 583
 584   if (!tls) {
 585     /* Initialize Tls for this thread */
 586     tls = silc_thread_tls_init();
 587     if (!tls)
 588       return;
 589   }
 590
 591   tls->thread_context = context;
 592 }
 593
 594 void *silc_thread_tls_get(void)
 595 {
 596   SilcTls tls = silc_thread_get_tls();
 597   if (!tls)
 598     return NULL;
 599   return tls->thread_context;
 600 }