main/threadpool.c

   1 /*
   2  * Asterisk -- An open source telephony toolkit.
   3  *
   4  * Copyright (C) 2012-2013, Digium, Inc.
   5  *
   6  * Mark Michelson <mmmichelson@digium.com>
   7  *
   8  * See http://www.asterisk.org for more information about
   9  * the Asterisk project. Please do not directly contact
  10  * any of the maintainers of this project for assistance;
  11  * the project provides a web site, mailing lists and IRC
  12  * channels for your use.
  13  *
  14  * This program is free software, distributed under the terms of
  15  * the GNU General Public License Version 2. See the LICENSE file
  16  * at the top of the source tree.
  17  */
  18
  19
  20 #include "asterisk.h"
  21
  22 #include "asterisk/threadpool.h"
  23 #include "asterisk/taskprocessor.h"
  24 #include "asterisk/astobj2.h"
  25 #include "asterisk/utils.h"
  26
  27 /* Needs to stay prime if increased */
  28 #define THREAD_BUCKETS 89
  29
  30 /*!
  31  * \brief An opaque threadpool structure
  32  *
  33  * A threadpool is a collection of threads that execute
  34  * tasks from a common queue.
  35  */
  36 struct ast_threadpool {
  37         /*! Threadpool listener */
  38         struct ast_threadpool_listener *listener;
  39         /*!
  40          * \brief The container of active threads.
  41          * Active threads are those that are currently running tasks
  42          */
  43         struct ao2_container *active_threads;
  44         /*!
  45          * \brief The container of idle threads.
  46          * Idle threads are those that are currenly waiting to run tasks
  47          */
  48         struct ao2_container *idle_threads;
  49         /*!
  50          * \brief The container of zombie threads.
  51          * Zombie threads may be running tasks, but they are scheduled to die soon
  52          */
  53         struct ao2_container *zombie_threads;
  54         /*!
  55          * \brief The main taskprocessor
  56          *
  57          * Tasks that are queued in this taskprocessor are
  58          * doled out to the worker threads. Worker threads that
  59          * execute tasks from the threadpool are executing tasks
  60          * in this taskprocessor.
  61          *
  62          * The threadpool itself is actually the private data for
  63          * this taskprocessor's listener. This way, as taskprocessor
  64          * changes occur, the threadpool can alert its listeners
  65          * appropriately.
  66          */
  67         struct ast_taskprocessor *tps;
  68         /*!
  69          * \brief The control taskprocessor
  70          *
  71          * This is a standard taskprocessor that uses the default
  72          * taskprocessor listener. In other words, all tasks queued to
  73          * this taskprocessor have a single thread that executes the
  74          * tasks.
  75          *
  76          * All tasks that modify the state of the threadpool and all tasks
  77          * that call out to threadpool listeners are pushed to this
  78          * taskprocessor.
  79          *
  80          * For instance, when the threadpool changes sizes, a task is put
  81          * into this taskprocessor to do so. When it comes time to tell the
  82          * threadpool listener that worker threads have changed state,
  83          * the task is placed in this taskprocessor.
  84          *
  85          * This is done for three main reasons
  86          * 1) It ensures that listeners are given an accurate portrayal
  87          * of the threadpool's current state. In other words, when a listener
  88          * gets told a count of active, idle and zombie threads, it does not
  89          * need to worry that internal state of the threadpool might be different
  90          * from what it has been told.
  91          * 2) It minimizes the locking required in both the threadpool and in
  92          * threadpool listener's callbacks.
  93          * 3) It ensures that listener callbacks are called in the same order
  94          * that the threadpool had its state change.
  95          */
  96         struct ast_taskprocessor *control_tps;
  97         /*! True if the threadpool is in the process of shutting down */
  98         int shutting_down;
  99         /*! Threadpool-specific options */
 100         struct ast_threadpool_options options;
 101 };
 102
 103 /*!
 104  * \brief listener for a threadpool
 105  *
 106  * The listener is notified of changes in a threadpool. It can
 107  * react by doing things like increasing the number of threads
 108  * in the pool
 109  */
 110 struct ast_threadpool_listener {
 111         /*! Callbacks called by the threadpool */
 112         const struct ast_threadpool_listener_callbacks *callbacks;
 113         /*! User data for the listener */
 114         void *user_data;
 115 };
 116
 117 /*!
 118  * \brief states for worker threads
 119  */
 120 enum worker_state {
 121         /*! The worker is either active or idle */
 122         ALIVE,
 123         /*!
 124          * The worker has been asked to shut down but
 125          * may still be in the process of executing tasks.
 126          * This transition happens when the threadpool needs
 127          * to shrink and needs to kill active threads in order
 128          * to do so.
 129          */
 130         ZOMBIE,
 131         /*!
 132          * The worker has been asked to shut down. Typically
 133          * only idle threads go to this state directly, but
 134          * active threads may go straight to this state when
 135          * the threadpool is shut down.
 136          */
 137         DEAD,
 138 };
 139
 140 /*!
 141  * A thread that executes threadpool tasks
 142  */
 143 struct worker_thread {
 144         /*! A unique (within a run of Asterisk) ID for the thread. Used for hashing and searching */
 145         int id;
 146         /*! Condition used in conjunction with state changes */
 147         ast_cond_t cond;
 148         /*! Lock used alongside the condition for state changes */
 149         ast_mutex_t lock;
 150         /*! The actual thread that is executing tasks */
 151         pthread_t thread;
 152         /*! A pointer to the threadpool. Needed to be able to execute tasks */
 153         struct ast_threadpool *pool;
 154         /*! The current state of the worker thread */
 155         enum worker_state state;
 156         /*! A boolean used to determine if an idle thread should become active */
 157         int wake_up;
 158         /*! Options for this threadpool */
 159         struct ast_threadpool_options options;
 160 };
 161
 162 /* Worker thread forward declarations. See definitions for documentation */
 163 static int worker_thread_hash(const void *obj, int flags);
 164 static int worker_thread_cmp(void *obj, void *arg, int flags);
 165 static void worker_thread_destroy(void *obj);
 166 static void worker_active(struct worker_thread *worker);
 167 static void *worker_start(void *arg);
 168 static struct worker_thread *worker_thread_alloc(struct ast_threadpool *pool);
 169 static int worker_thread_start(struct worker_thread *worker);
 170 static int worker_idle(struct worker_thread *worker);
 171 static void worker_set_state(struct worker_thread *worker, enum worker_state state);
 172 static void worker_shutdown(struct worker_thread *worker);
 173
 174 /*!
 175  * \brief Notify the threadpool listener that the state has changed.
 176  *
 177  * This notifies the threadpool listener via its state_changed callback.
 178  * \param pool The threadpool whose state has changed
 179  */
 180 static void threadpool_send_state_changed(struct ast_threadpool *pool)
 181 {
 182         int active_size = ao2_container_count(pool->active_threads);
 183         int idle_size = ao2_container_count(pool->idle_threads);
 184
 185         if (pool->listener && pool->listener->callbacks->state_changed) {
 186                 pool->listener->callbacks->state_changed(pool, pool->listener, active_size, idle_size);
 187         }
 188 }
 189
 190 /*!
 191  * \brief Struct used for queued operations involving worker state changes
 192  */
 193 struct thread_worker_pair {
 194         /*! Threadpool that contains the worker whose state has changed */
 195         struct ast_threadpool *pool;
 196         /*! Worker whose state has changed */
 197         struct worker_thread *worker;
 198 };
 199
 200 /*!
 201  * \brief Destructor for thread_worker_pair
 202  */
 203 static void thread_worker_pair_destructor(void *obj)
 204 {
 205         struct thread_worker_pair *pair = obj;
 206         ao2_ref(pair->worker, -1);
 207 }
 208
 209 /*!
 210  * \brief Allocate and initialize a thread_worker_pair
 211  * \param pool Threadpool to assign to the thread_worker_pair
 212  * \param worker Worker thread to assign to the thread_worker_pair
 213  */
 214 static struct thread_worker_pair *thread_worker_pair_alloc(struct ast_threadpool *pool,
 215                 struct worker_thread *worker)
 216 {
 217         struct thread_worker_pair *pair = ao2_alloc(sizeof(*pair), thread_worker_pair_destructor);
 218         if (!pair) {
 219                 return NULL;
 220         }
 221         pair->pool = pool;
 222         ao2_ref(worker, +1);
 223         pair->worker = worker;
 224         return pair;
 225 }
 226
 227 /*!
 228  * \brief Move a worker thread from the active container to the idle container.
 229  *
 230  * This function is called from the threadpool's control taskprocessor thread.
 231  * \param data A thread_worker_pair containing the threadpool and the worker to move.
 232  * \return 0
 233  */
 234 static int queued_active_thread_idle(void *data)
 235 {
 236         struct thread_worker_pair *pair = data;
 237
 238         ao2_link(pair->pool->idle_threads, pair->worker);
 239         ao2_unlink(pair->pool->active_threads, pair->worker);
 240
 241         threadpool_send_state_changed(pair->pool);
 242
 243         ao2_ref(pair, -1);
 244         return 0;
 245 }
 246
 247 /*!
 248  * \brief Queue a task to move a thread from the active list to the idle list
 249  *
 250  * This is called by a worker thread when it runs out of tasks to perform and
 251  * goes idle.
 252  * \param pool The threadpool to which the worker belongs
 253  * \param worker The worker thread that has gone idle
 254  */
 255 static void threadpool_active_thread_idle(struct ast_threadpool *pool,
 256                 struct worker_thread *worker)
 257 {
 258         struct thread_worker_pair *pair;
 259         SCOPED_AO2LOCK(lock, pool);
 260         if (pool->shutting_down) {
 261                 return;
 262         }
 263         pair = thread_worker_pair_alloc(pool, worker);
 264         if (!pair) {
 265                 return;
 266         }
 267         ast_taskprocessor_push(pool->control_tps, queued_active_thread_idle, pair);
 268 }
 269
 270 /*!
 271  * \brief Kill a zombie thread
 272  *
 273  * This runs from the threadpool's control taskprocessor thread.
 274  *
 275  * \param data A thread_worker_pair containing the threadpool and the zombie thread
 276  * \return 0
 277  */
 278 static int queued_zombie_thread_dead(void *data)
 279 {
 280         struct thread_worker_pair *pair = data;
 281
 282         ao2_unlink(pair->pool->zombie_threads, pair->worker);
 283         threadpool_send_state_changed(pair->pool);
 284
 285         ao2_ref(pair, -1);
 286         return 0;
 287 }
 288
 289 /*!
 290  * \brief Queue a task to kill a zombie thread
 291  *
 292  * This is called by a worker thread when it acknowledges that it is time for
 293  * it to die.
 294  */
 295 static void threadpool_zombie_thread_dead(struct ast_threadpool *pool,
 296                 struct worker_thread *worker)
 297 {
 298         struct thread_worker_pair *pair;
 299         SCOPED_AO2LOCK(lock, pool);
 300         if (pool->shutting_down) {
 301                 return;
 302         }
 303         pair = thread_worker_pair_alloc(pool, worker);
 304         if (!pair) {
 305                 return;
 306         }
 307         ast_taskprocessor_push(pool->control_tps, queued_zombie_thread_dead, pair);
 308 }
 309
 310 static int queued_idle_thread_dead(void *data)
 311 {
 312         struct thread_worker_pair *pair = data;
 313
 314         ao2_unlink(pair->pool->idle_threads, pair->worker);
 315         threadpool_send_state_changed(pair->pool);
 316
 317         ao2_ref(pair, -1);
 318         return 0;
 319 }
 320
 321 static void threadpool_idle_thread_dead(struct ast_threadpool *pool,
 322                 struct worker_thread *worker)
 323 {
 324         struct thread_worker_pair *pair;
 325         SCOPED_AO2LOCK(lock, pool);
 326         if (pool->shutting_down) {
 327                 return;
 328         }
 329         pair = thread_worker_pair_alloc(pool, worker);
 330         if (!pair) {
 331                 return;
 332         }
 333         ast_taskprocessor_push(pool->control_tps, queued_idle_thread_dead, pair);
 334 }
 335
 336 /*!
 337  * \brief Execute a task in the threadpool
 338  *
 339  * This is the function that worker threads call in order to execute tasks
 340  * in the threadpool
 341  *
 342  * \param pool The pool to which the tasks belong.
 343  * \retval 0 Either the pool has been shut down or there are no tasks.
 344  * \retval 1 There are still tasks remaining in the pool.
 345  */
 346 static int threadpool_execute(struct ast_threadpool *pool)
 347 {
 348         ao2_lock(pool);
 349         if (!pool->shutting_down) {
 350                 ao2_unlock(pool);
 351                 return ast_taskprocessor_execute(pool->tps);
 352         }
 353         ao2_unlock(pool);
 354         return 0;
 355 }
 356
 357 /*!
 358  * \brief Destroy a threadpool's components.
 359  *
 360  * This is the destructor called automatically when the threadpool's
 361  * reference count reaches zero. This is not to be confused with
 362  * threadpool_destroy.
 363  *
 364  * By the time this actually gets called, most of the cleanup has already
 365  * been done in the pool. The only thing left to do is to release the
 366  * final reference to the threadpool listener.
 367  *
 368  * \param obj The pool to destroy
 369  */
 370 static void threadpool_destructor(void *obj)
 371 {
 372         struct ast_threadpool *pool = obj;
 373         ao2_cleanup(pool->listener);
 374 }
 375
 376 /*
 377  * \brief Allocate a threadpool
 378  *
 379  * This is implemented as a taskprocessor listener's alloc callback. This
 380  * is because the threadpool exists as the private data on a taskprocessor
 381  * listener.
 382  *
 383  * \param name The name of the threadpool.
 384  * \param options The options the threadpool uses.
 385  * \retval NULL Could not initialize threadpool properly
 386  * \retval non-NULL The newly-allocated threadpool
 387  */
 388 static void *threadpool_alloc(const char *name, const struct ast_threadpool_options *options)
 389 {
 390         RAII_VAR(struct ast_threadpool *, pool,
 391                         ao2_alloc(sizeof(*pool), threadpool_destructor), ao2_cleanup);
 392         struct ast_str *control_tps_name = ast_str_create(64);
 393
 394         if (!control_tps_name) {
 395                 return NULL;
 396         }
 397
 398         ast_str_set(&control_tps_name, 0, "%s-control", name);
 399
 400         pool->control_tps = ast_taskprocessor_get(ast_str_buffer(control_tps_name), TPS_REF_DEFAULT);
 401         ast_free(control_tps_name);
 402         if (!pool->control_tps) {
 403                 return NULL;
 404         }
 405         pool->active_threads = ao2_container_alloc(THREAD_BUCKETS, worker_thread_hash, worker_thread_cmp);
 406         if (!pool->active_threads) {
 407                 return NULL;
 408         }
 409         pool->idle_threads = ao2_container_alloc(THREAD_BUCKETS, worker_thread_hash, worker_thread_cmp);
 410         if (!pool->idle_threads) {
 411                 return NULL;
 412         }
 413         pool->zombie_threads = ao2_container_alloc(THREAD_BUCKETS, worker_thread_hash, worker_thread_cmp);
 414         if (!pool->zombie_threads) {
 415                 return NULL;
 416         }
 417         pool->options = *options;
 418
 419         ao2_ref(pool, +1);
 420         return pool;
 421 }
 422
 423 static int threadpool_tps_start(struct ast_taskprocessor_listener *listener)
 424 {
 425         return 0;
 426 }
 427
 428 /*!
 429  * \brief helper used for queued task when tasks are pushed
 430  */
 431 struct task_pushed_data {
 432         /*! Pool into which a task was pushed */
 433         struct ast_threadpool *pool;
 434         /*! Indicator of whether the pool had no tasks prior to the new task being added */
 435         int was_empty;
 436 };
 437
 438 /*!
 439  * \brief Allocate and initialize a task_pushed_data
 440  * \param pool The threadpool to set in the task_pushed_data
 441  * \param was_empty The was_empty value to set in the task_pushed_data
 442  * \retval NULL Unable to allocate task_pushed_data
 443  * \retval non-NULL The newly-allocated task_pushed_data
 444  */
 445 static struct task_pushed_data *task_pushed_data_alloc(struct ast_threadpool *pool,
 446                 int was_empty)
 447 {
 448         struct task_pushed_data *tpd = ao2_alloc(sizeof(*tpd), NULL);
 449
 450         if (!tpd) {
 451                 return NULL;
 452         }
 453         tpd->pool = pool;
 454         tpd->was_empty = was_empty;
 455         return tpd;
 456 }
 457
 458 /*!
 459  * \brief Activate idle threads
 460  *
 461  * This function always returns CMP_MATCH because all workers that this
 462  * function acts on need to be seen as matches so they are unlinked from the
 463  * list of idle threads.
 464  *
 465  * Called as an ao2_callback in the threadpool's control taskprocessor thread.
 466  * \param obj The worker to activate
 467  * \param arg The pool where the worker belongs
 468  * \retval CMP_MATCH
 469  */
 470 static int activate_thread(void *obj, void *arg, int flags)
 471 {
 472         struct worker_thread *worker = obj;
 473         struct ast_threadpool *pool = arg;
 474
 475         if (!ao2_link(pool->active_threads, worker)) {
 476                 /* If we can't link the idle thread into the active container, then
 477                  * we'll just leave the thread idle and not wake it up.
 478                  */
 479                 ast_log(LOG_WARNING, "Failed to activate thread %d. Remaining idle\n",
 480                                 worker->id);
 481                 return 0;
 482         }
 483         worker_set_state(worker, ALIVE);
 484         return CMP_MATCH;
 485 }
 486
 487 /*!
 488  * \brief Add threads to the threadpool
 489  *
 490  * This function is called from the threadpool's control taskprocessor thread.
 491  * \param pool The pool that is expanding
 492  * \delta The number of threads to add to the pool
 493  */
 494 static void grow(struct ast_threadpool *pool, int delta)
 495 {
 496         int i;
 497
 498         int current_size = ao2_container_count(pool->active_threads) +
 499                 ao2_container_count(pool->idle_threads);
 500
 501         if (pool->options.max_size && current_size + delta > pool->options.max_size) {
 502                 delta = pool->options.max_size - current_size;
 503         }
 504
 505         ast_debug(3, "Increasing threadpool %s's size by %d\n",
 506                         ast_taskprocessor_name(pool->tps), delta);
 507
 508         for (i = 0; i < delta; ++i) {
 509                 struct worker_thread *worker = worker_thread_alloc(pool);
 510                 if (!worker) {
 511                         return;
 512                 }
 513                 if (ao2_link(pool->active_threads, worker)) {
 514                         if (worker_thread_start(worker)) {
 515                                 ast_log(LOG_ERROR, "Unable to start worker thread %d. Destroying.\n", worker->id);
 516                                 ao2_unlink(pool->active_threads, worker);
 517                         }
 518                 } else {
 519                         ast_log(LOG_WARNING, "Failed to activate worker thread %d. Destroying.\n", worker->id);
 520                 }
 521                 ao2_ref(worker, -1);
 522         }
 523 }
 524
 525 /*!
 526  * \brief Queued task called when tasks are pushed into the threadpool
 527  *
 528  * This function first calls into the threadpool's listener to let it know
 529  * that a task has been pushed. It then wakes up all idle threads and moves
 530  * them into the active thread container.
 531  * \param data A task_pushed_data
 532  * \return 0
 533  */
 534 static int queued_task_pushed(void *data)
 535 {
 536         struct task_pushed_data *tpd = data;
 537         struct ast_threadpool *pool = tpd->pool;
 538         int was_empty = tpd->was_empty;
 539         int state_changed;
 540
 541         if (pool->listener && pool->listener->callbacks->task_pushed) {
 542                 pool->listener->callbacks->task_pushed(pool, pool->listener, was_empty);
 543         }
 544         if (ao2_container_count(pool->idle_threads) == 0) {
 545                 if (pool->options.auto_increment > 0) {
 546                         grow(pool, pool->options.auto_increment);
 547                         state_changed = 1;
 548                 }
 549         } else {
 550                 ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA,
 551                                 activate_thread, pool);
 552                 state_changed = 1;
 553         }
 554         if (state_changed) {
 555                 threadpool_send_state_changed(pool);
 556         }
 557         ao2_ref(tpd, -1);
 558         return 0;
 559 }
 560
 561 /*!
 562  * \brief Taskprocessor listener callback called when a task is added
 563  *
 564  * The threadpool uses this opportunity to queue a task on its control taskprocessor
 565  * in order to activate idle threads and notify the threadpool listener that the
 566  * task has been pushed.
 567  * \param listener The taskprocessor listener. The threadpool is the listener's private data
 568  * \param was_empty True if the taskprocessor was empty prior to the task being pushed
 569  */
 570 static void threadpool_tps_task_pushed(struct ast_taskprocessor_listener *listener,
 571                 int was_empty)
 572 {
 573         struct ast_threadpool *pool = ast_taskprocessor_listener_get_user_data(listener);
 574         struct task_pushed_data *tpd;
 575         SCOPED_AO2LOCK(lock, pool);
 576
 577         if (pool->shutting_down) {
 578                 return;
 579         }
 580         tpd = task_pushed_data_alloc(pool, was_empty);
 581         if (!tpd) {
 582                 return;
 583         }
 584
 585         ast_taskprocessor_push(pool->control_tps, queued_task_pushed, tpd);
 586 }
 587
 588 /*!
 589  * \brief Queued task that handles the case where the threadpool's taskprocessor is emptied
 590  *
 591  * This simply lets the threadpool's listener know that the threadpool is devoid of tasks
 592  * \param data The pool that has become empty
 593  * \return 0
 594  */
 595 static int queued_emptied(void *data)
 596 {
 597         struct ast_threadpool *pool = data;
 598
 599         /* We already checked for existence of this callback when this was queued */
 600         pool->listener->callbacks->emptied(pool, pool->listener);
 601         return 0;
 602 }
 603
 604 /*!
 605  * \brief Taskprocessor listener emptied callback
 606  *
 607  * The threadpool queues a task to let the threadpool listener know that
 608  * the threadpool no longer contains any tasks.
 609  * \param listener The taskprocessor listener. The threadpool is the listener's private data.
 610  */
 611 static void threadpool_tps_emptied(struct ast_taskprocessor_listener *listener)
 612 {
 613         struct ast_threadpool *pool = ast_taskprocessor_listener_get_user_data(listener);
 614         SCOPED_AO2LOCK(lock, pool);
 615
 616         if (pool->shutting_down) {
 617                 return;
 618         }
 619
 620         if (pool->listener && pool->listener->callbacks->emptied) {
 621                 ast_taskprocessor_push(pool->control_tps, queued_emptied, pool);
 622         }
 623 }
 624
 625 /*!
 626  * \brief Taskprocessor listener shutdown callback
 627  *
 628  * The threadpool will shut down and destroy all of its worker threads when
 629  * this is called back. By the time this gets called, the taskprocessor's
 630  * control taskprocessor has already been destroyed. Therefore there is no risk
 631  * in outright destroying the worker threads here.
 632  * \param listener The taskprocessor listener. The threadpool is the listener's private data.
 633  */
 634 static void threadpool_tps_shutdown(struct ast_taskprocessor_listener *listener)
 635 {
 636         struct ast_threadpool *pool = ast_taskprocessor_listener_get_user_data(listener);
 637
 638         if (pool->listener && pool->listener->callbacks->shutdown) {
 639                 pool->listener->callbacks->shutdown(pool->listener);
 640         }
 641         ao2_cleanup(pool->active_threads);
 642         ao2_cleanup(pool->idle_threads);
 643         ao2_cleanup(pool->zombie_threads);
 644         ao2_cleanup(pool);
 645 }
 646
 647 /*!
 648  * \brief Table of taskprocessor listener callbacks for threadpool's main taskprocessor
 649  */
 650 static struct ast_taskprocessor_listener_callbacks threadpool_tps_listener_callbacks = {
 651         .start = threadpool_tps_start,
 652         .task_pushed = threadpool_tps_task_pushed,
 653         .emptied = threadpool_tps_emptied,
 654         .shutdown = threadpool_tps_shutdown,
 655 };
 656
 657 /*!
 658  * \brief ao2 callback to kill a set number of threads.
 659  *
 660  * Threads will be unlinked from the container as long as the
 661  * counter has not reached zero. The counter is decremented with
 662  * each thread that is removed.
 663  * \param obj The worker thread up for possible destruction
 664  * \param arg The counter
 665  * \param flags Unused
 666  * \retval CMP_MATCH The counter has not reached zero, so this flag should be removed.
 667  * \retval CMP_STOP The counter has reached zero so no more threads should be removed.
 668  */
 669 static int kill_threads(void *obj, void *arg, int flags)
 670 {
 671         int *num_to_kill = arg;
 672
 673         if (*num_to_kill > 0) {
 674                 --(*num_to_kill);
 675                 return CMP_MATCH;
 676         } else {
 677                 return CMP_STOP;
 678         }
 679 }
 680
 681 /*!
 682  * \brief ao2 callback to zombify a set number of threads.
 683  *
 684  * Threads will be zombified as long as as the counter has not reached
 685  * zero. The counter is decremented with each thread that is zombified.
 686  *
 687  * Zombifying a thread involves removing it from its current container,
 688  * adding it to the zombie container, and changing the state of the
 689  * worker to a zombie
 690  *
 691  * This callback is called from the threadpool control taskprocessor thread.
 692  *
 693  * \param obj The worker thread that may be zombified
 694  * \param arg The pool to which the worker belongs
 695  * \param data The counter
 696  * \param flags Unused
 697  * \retval CMP_MATCH The zombified thread should be removed from its current container
 698  * \retval CMP_STOP Stop attempting to zombify threads
 699  */
 700 static int zombify_threads(void *obj, void *arg, void *data, int flags)
 701 {
 702         struct worker_thread *worker = obj;
 703         struct ast_threadpool *pool = arg;
 704         int *num_to_zombify = data;
 705
 706         if ((*num_to_zombify)-- > 0) {
 707                 if (!ao2_link(pool->zombie_threads, worker)) {
 708                         ast_log(LOG_WARNING, "Failed to zombify active thread %d. Thread will remain active\n", worker->id);
 709                         return 0;
 710                 }
 711                 worker_set_state(worker, ZOMBIE);
 712                 return CMP_MATCH;
 713         } else {
 714                 return CMP_STOP;
 715         }
 716 }
 717
 718 /*!
 719  * \brief Remove threads from the threadpool
 720  *
 721  * The preference is to kill idle threads. However, if there are
 722  * more threads to remove than there are idle threads, then active
 723  * threads will be zombified instead.
 724  *
 725  * This function is called from the threadpool control taskprocessor thread.
 726  *
 727  * \param pool The threadpool to remove threads from
 728  * \param delta The number of threads to remove
 729  */
 730 static void shrink(struct ast_threadpool *pool, int delta)
 731 {
 732         /*
 733          * Preference is to kill idle threads, but
 734          * we'll move on to deactivating active threads
 735          * if we have to
 736          */
 737         int idle_threads = ao2_container_count(pool->idle_threads);
 738         int idle_threads_to_kill = MIN(delta, idle_threads);
 739         int active_threads_to_zombify = delta - idle_threads_to_kill;
 740
 741         ast_debug(3, "Destroying %d idle threads in threadpool %s\n", idle_threads_to_kill,
 742                         ast_taskprocessor_name(pool->tps));
 743
 744         ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
 745                         kill_threads, &idle_threads_to_kill);
 746
 747         ast_debug(3, "Destroying %d active threads in threadpool %s\n", active_threads_to_zombify,
 748                         ast_taskprocessor_name(pool->tps));
 749
 750         ao2_callback_data(pool->active_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
 751                         zombify_threads, pool, &active_threads_to_zombify);
 752 }
 753
 754 /*!
 755  * \brief Helper struct used for queued operations that change the size of the threadpool
 756  */
 757 struct set_size_data {
 758         /*! The pool whose size is to change */
 759         struct ast_threadpool *pool;
 760         /*! The requested new size of the pool */
 761         unsigned int size;
 762 };
 763
 764 /*!
 765  * \brief Allocate and initialize a set_size_data
 766  * \param pool The pool for the set_size_data
 767  * \param size The size to store in the set_size_data
 768  */
 769 static struct set_size_data *set_size_data_alloc(struct ast_threadpool *pool,
 770                 unsigned int size)
 771 {
 772         struct set_size_data *ssd = ao2_alloc(sizeof(*ssd), NULL);
 773         if (!ssd) {
 774                 return NULL;
 775         }
 776
 777         ssd->pool = pool;
 778         ssd->size = size;
 779         return ssd;
 780 }
 781
 782 /*!
 783  * \brief Change the size of the threadpool
 784  *
 785  * This can either result in shrinking or growing the threadpool depending
 786  * on the new desired size and the current size.
 787  *
 788  * This function is run from the threadpool control taskprocessor thread
 789  *
 790  * \param data A set_size_data used for determining how to act
 791  * \return 0
 792  */
 793 static int queued_set_size(void *data)
 794 {
 795         struct set_size_data *ssd = data;
 796         struct ast_threadpool *pool = ssd->pool;
 797         unsigned int num_threads = ssd->size;
 798
 799         /* We don't count zombie threads as being "live" when potentially resizing */
 800         unsigned int current_size = ao2_container_count(pool->active_threads) +
 801                 ao2_container_count(pool->idle_threads);
 802
 803         if (current_size == num_threads) {
 804                 ast_log(LOG_NOTICE, "Not changing threadpool size since new size %u is the same as current %u\n",
 805                                 num_threads, current_size);
 806                 return 0;
 807         }
 808
 809         if (current_size < num_threads) {
 810                 grow(pool, num_threads - current_size);
 811         } else {
 812                 shrink(pool, current_size - num_threads);
 813         }
 814
 815         threadpool_send_state_changed(pool);
 816         ao2_ref(ssd, -1);
 817         return 0;
 818 }
 819
 820 void ast_threadpool_set_size(struct ast_threadpool *pool, unsigned int size)
 821 {
 822         struct set_size_data *ssd;
 823         SCOPED_AO2LOCK(lock, pool);
 824         if (pool->shutting_down) {
 825                 return;
 826         }
 827
 828         ssd = set_size_data_alloc(pool, size);
 829         if (!ssd) {
 830                 return;
 831         }
 832
 833         ast_taskprocessor_push(pool->control_tps, queued_set_size, ssd);
 834 }
 835
 836 struct ast_threadpool_listener *ast_threadpool_listener_alloc(
 837                 const struct ast_threadpool_listener_callbacks *callbacks, void *user_data)
 838 {
 839         struct ast_threadpool_listener *listener = ao2_alloc(sizeof(*listener), NULL);
 840         if (!listener) {
 841                 return NULL;
 842         }
 843         listener->callbacks = callbacks;
 844         listener->user_data = user_data;
 845         return listener;
 846 }
 847
 848 void *ast_threadpool_listener_get_user_data(const struct ast_threadpool_listener *listener)
 849 {
 850         return listener->user_data;
 851 }
 852
 853 struct pool_options_pair {
 854         struct ast_threadpool *pool;
 855         struct ast_threadpool_options options;
 856 };
 857
 858 struct ast_threadpool *ast_threadpool_create(const char *name,
 859                 struct ast_threadpool_listener *listener,
 860                 const struct ast_threadpool_options *options)
 861 {
 862         struct ast_taskprocessor *tps;
 863         RAII_VAR(struct ast_taskprocessor_listener *, tps_listener, NULL, ao2_cleanup);
 864         RAII_VAR(struct ast_threadpool *, pool, threadpool_alloc(name, options), ao2_cleanup);
 865
 866         if (!pool) {
 867                 return NULL;
 868         }
 869
 870         tps_listener = ast_taskprocessor_listener_alloc(&threadpool_tps_listener_callbacks, pool);
 871         if (!tps_listener) {
 872                 return NULL;
 873         }
 874
 875         if (options->version != AST_THREADPOOL_OPTIONS_VERSION) {
 876                 ast_log(LOG_WARNING, "Incompatible version of threadpool options in use.\n");
 877                 return NULL;
 878         }
 879
 880         tps = ast_taskprocessor_create_with_listener(name, tps_listener);
 881         if (!tps) {
 882                 return NULL;
 883         }
 884
 885         pool->tps = tps;
 886         if (listener) {
 887                 ao2_ref(listener, +1);
 888                 pool->listener = listener;
 889         }
 890         ast_threadpool_set_size(pool, pool->options.initial_size);
 891         ao2_ref(pool, +1);
 892         return pool;
 893 }
 894
 895 int ast_threadpool_push(struct ast_threadpool *pool, int (*task)(void *data), void *data)
 896 {
 897         SCOPED_AO2LOCK(lock, pool);
 898         if (!pool->shutting_down) {
 899                 return ast_taskprocessor_push(pool->tps, task, data);
 900         }
 901         return 0;
 902 }
 903
 904 void ast_threadpool_shutdown(struct ast_threadpool *pool)
 905 {
 906         if (!pool) {
 907                 return;
 908         }
 909         /* Shut down the taskprocessors and everything else just
 910          * takes care of itself via the taskprocessor callbacks
 911          */
 912         ao2_lock(pool);
 913         pool->shutting_down = 1;
 914         ao2_unlock(pool);
 915         ast_taskprocessor_unreference(pool->control_tps);
 916         ast_taskprocessor_unreference(pool->tps);
 917 }
 918
 919 /*!
 920  * A monotonically increasing integer used for worker
 921  * thread identification.
 922  */
 923 static int worker_id_counter;
 924
 925 static int worker_thread_hash(const void *obj, int flags)
 926 {
 927         const struct worker_thread *worker = obj;
 928
 929         return worker->id;
 930 }
 931
 932 static int worker_thread_cmp(void *obj, void *arg, int flags)
 933 {
 934         struct worker_thread *worker1 = obj;
 935         struct worker_thread *worker2 = arg;
 936
 937         return worker1->id == worker2->id ? CMP_MATCH : 0;
 938 }
 939
 940 /*!
 941  * \brief shut a worker thread down
 942  *
 943  * Set the worker dead and then wait for its thread
 944  * to finish executing.
 945  *
 946  * \param worker The worker thread to shut down
 947  */
 948 static void worker_shutdown(struct worker_thread *worker)
 949 {
 950         worker_set_state(worker, DEAD);
 951         if (worker->thread != AST_PTHREADT_NULL) {
 952                 pthread_join(worker->thread, NULL);
 953                 worker->thread = AST_PTHREADT_NULL;
 954         }
 955 }
 956
 957 /*!
 958  * \brief Worker thread destructor
 959  *
 960  * Called automatically when refcount reaches 0. Shuts
 961  * down the worker thread and destroys its component
 962  * parts
 963  */
 964 static void worker_thread_destroy(void *obj)
 965 {
 966         struct worker_thread *worker = obj;
 967         ast_debug(3, "Destroying worker thread %d\n", worker->id);
 968         worker_shutdown(worker);
 969         ast_mutex_destroy(&worker->lock);
 970         ast_cond_destroy(&worker->cond);
 971 }
 972
 973 /*!
 974  * \brief start point for worker threads
 975  *
 976  * Worker threads start in the active state but may
 977  * immediately go idle if there is no work to be
 978  * done
 979  *
 980  * \param arg The worker thread
 981  * \retval NULL
 982  */
 983 static void *worker_start(void *arg)
 984 {
 985         struct worker_thread *worker = arg;
 986
 987         worker_active(worker);
 988         return NULL;
 989 }
 990
 991 /*!
 992  * \brief Allocate and initialize a new worker thread
 993  *
 994  * This will create, initialize, and start the thread.
 995  *
 996  * \param pool The threadpool to which the worker will be added
 997  * \retval NULL Failed to allocate or start the worker thread
 998  * \retval non-NULL The newly-created worker thread
 999  */
1000 static struct worker_thread *worker_thread_alloc(struct ast_threadpool *pool)
1001 {
1002         struct worker_thread *worker = ao2_alloc(sizeof(*worker), worker_thread_destroy);
1003         if (!worker) {
1004                 return NULL;
1005         }
1006         worker->id = ast_atomic_fetchadd_int(&worker_id_counter, 1);
1007         ast_mutex_init(&worker->lock);
1008         ast_cond_init(&worker->cond, NULL);
1009         worker->pool = pool;
1010         worker->thread = AST_PTHREADT_NULL;
1011         worker->state = ALIVE;
1012         worker->options = pool->options;
1013         return worker;
1014 }
1015
1016 static int worker_thread_start(struct worker_thread *worker)
1017 {
1018         return ast_pthread_create(&worker->thread, NULL, worker_start, worker);
1019 }
1020
1021 /*!
1022  * \brief Active loop for worker threads
1023  *
1024  * The worker will stay in this loop for its lifetime,
1025  * executing tasks as they become available. If there
1026  * are no tasks currently available, then the thread
1027  * will go idle.
1028  *
1029  * \param worker The worker thread executing tasks.
1030  */
1031 static void worker_active(struct worker_thread *worker)
1032 {
1033         int alive = 1;
1034         while (alive) {
1035                 if (!threadpool_execute(worker->pool)) {
1036                         alive = worker_idle(worker);
1037                 }
1038         }
1039
1040         /* Reaching this portion means the thread is
1041          * on death's door. It may have been killed while
1042          * it was idle, in which case it can just die
1043          * peacefully. If it's a zombie, though, then
1044          * it needs to let the pool know so
1045          * that the thread can be removed from the
1046          * list of zombie threads.
1047          */
1048         if (worker->state == ZOMBIE) {
1049                 threadpool_zombie_thread_dead(worker->pool, worker);
1050         }
1051 }
1052
1053 /*!
1054  * \brief Idle function for worker threads
1055  *
1056  * The worker waits here until it gets told by the threadpool
1057  * to wake up.
1058  *
1059  * \param worker The idle worker
1060  * \retval 0 The thread is being woken up so that it can conclude.
1061  * \retval non-zero The thread is being woken up to do more work.
1062  */
1063 static int worker_idle(struct worker_thread *worker)
1064 {
1065         struct timeval start = ast_tvnow();
1066         struct timespec end = {
1067                 .tv_sec = start.tv_sec + worker->options.idle_timeout,
1068                 .tv_nsec = start.tv_usec * 1000,
1069         };
1070         SCOPED_MUTEX(lock, &worker->lock);
1071         if (worker->state != ALIVE) {
1072                 return 0;
1073         }
1074         threadpool_active_thread_idle(worker->pool, worker);
1075         while (!worker->wake_up) {
1076                 if (worker->options.idle_timeout <= 0) {
1077                         ast_cond_wait(&worker->cond, lock);
1078                 } else if (ast_cond_timedwait(&worker->cond, lock, &end) == ETIMEDOUT) {
1079                         break;
1080                 }
1081         }
1082
1083         if (!worker->wake_up) {
1084                 ast_debug(1, "Worker thread idle timeout reached. Dying.\n");
1085                 threadpool_idle_thread_dead(worker->pool, worker);
1086                 worker->state = DEAD;
1087         }
1088         worker->wake_up = 0;
1089         return worker->state == ALIVE;
1090 }
1091
1092 /*!
1093  * \brief Change a worker's state
1094  *
1095  * The threadpool calls into this function in order to let a worker know
1096  * how it should proceed.
1097  */
1098 static void worker_set_state(struct worker_thread *worker, enum worker_state state)
1099 {
1100         SCOPED_MUTEX(lock, &worker->lock);
1101         worker->state = state;
1102         worker->wake_up = 1;
1103         ast_cond_signal(&worker->cond);
1104 }
1105
1106 struct serializer {
1107         struct ast_threadpool *pool;
1108 };
1109
1110 static void serializer_dtor(void *obj)
1111 {
1112         struct serializer *ser = obj;
1113         ao2_cleanup(ser->pool);
1114         ser->pool = NULL;
1115 }
1116
1117 static struct serializer *serializer_create(struct ast_threadpool *pool)
1118 {
1119         struct serializer *ser = ao2_alloc(sizeof(*ser), serializer_dtor);
1120         if (!ser) {
1121                 return NULL;
1122         }
1123         ao2_ref(pool, +1);
1124         ser->pool = pool;
1125         return ser;
1126 }
1127
1128 static int execute_tasks(void *data)
1129 {
1130         struct ast_taskprocessor *tps = data;
1131
1132         while (ast_taskprocessor_execute(tps)) {
1133                 /* No-op */
1134         }
1135
1136         ast_taskprocessor_unreference(tps);
1137         return 0;
1138 }
1139
1140 static void serializer_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
1141 {
1142         if (was_empty) {
1143                 struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
1144                 struct ast_taskprocessor *tps = ast_taskprocessor_listener_get_tps(listener);
1145
1146                 ast_threadpool_push(ser->pool, execute_tasks, tps);
1147         }
1148 }
1149
1150 static int serializer_start(struct ast_taskprocessor_listener *listener)
1151 {
1152         /* No-op */
1153         return 0;
1154 }
1155
1156 static void serializer_shutdown(struct ast_taskprocessor_listener *listener)
1157 {
1158         struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
1159         ao2_cleanup(ser);
1160 }
1161
1162 static struct ast_taskprocessor_listener_callbacks serializer_tps_listener_callbacks = {
1163         .task_pushed = serializer_task_pushed,
1164         .start = serializer_start,
1165         .shutdown = serializer_shutdown,
1166 };
1167
1168 struct ast_taskprocessor *ast_threadpool_serializer(const char *name, struct ast_threadpool *pool)
1169 {
1170         RAII_VAR(struct serializer *, ser, NULL, ao2_cleanup);
1171         RAII_VAR(struct ast_taskprocessor_listener *, listener, NULL, ao2_cleanup);
1172         struct ast_taskprocessor *tps = NULL;
1173
1174         ser = serializer_create(pool);
1175         if (!ser) {
1176                 return NULL;
1177         }
1178
1179         listener = ast_taskprocessor_listener_alloc(&serializer_tps_listener_callbacks, ser);
1180         if (!listener) {
1181                 return NULL;
1182         }
1183         ser = NULL; /* ownership transferred to listener */
1184
1185         tps = ast_taskprocessor_create_with_listener(name, listener);
1186         if (!tps) {
1187                 return NULL;
1188         }
1189         listener = NULL; /* ownership transferred to tps */
1190
1191         return tps;
1192 }