threadpool.c

Go to the documentation of this file.

/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2012-2013, Digium, Inc.
 *
 * Mark Michelson <[email protected]>
 *
 * See http://www.asterisk.org for more information about
 * the Asterisk project. Please do not directly contact
 * any of the maintainers of this project for assistance;
 * the project provides a web site, mailing lists and IRC
 * channels for your use.
 *
 * This program is free software, distributed under the terms of
 * the GNU General Public License Version 2. See the LICENSE file
 * at the top of the source tree.
 */


#include "asterisk.h"

#include "asterisk/threadpool.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/astobj2.h"
#include "asterisk/utils.h"

/* Needs to stay prime if increased */
#define THREAD_BUCKETS 89

/*!
 * \brief An opaque threadpool structure
 *
 * A threadpool is a collection of threads that execute
 * tasks from a common queue.
 */
struct ast_threadpool {
   /*! Threadpool listener */
   struct ast_threadpool_listener *listener;
   /*!
    * \brief The container of active threads.
    * Active threads are those that are currently running tasks
    */
   struct ao2_container *active_threads;
   /*!
    * \brief The container of idle threads.
    * Idle threads are those that are currenly waiting to run tasks
    */
   struct ao2_container *idle_threads;
   /*!
    * \brief The container of zombie threads.
    * Zombie threads may be running tasks, but they are scheduled to die soon
    */
   struct ao2_container *zombie_threads;
   /*!
    * \brief The main taskprocessor
    *
    * Tasks that are queued in this taskprocessor are
    * doled out to the worker threads. Worker threads that
    * execute tasks from the threadpool are executing tasks
    * in this taskprocessor.
    *
    * The threadpool itself is actually the private data for
    * this taskprocessor's listener. This way, as taskprocessor
    * changes occur, the threadpool can alert its listeners
    * appropriately.
    */
   struct ast_taskprocessor *tps;
   /*!
    * \brief The control taskprocessor
    *
    * This is a standard taskprocessor that uses the default
    * taskprocessor listener. In other words, all tasks queued to
    * this taskprocessor have a single thread that executes the
    * tasks.
    *
    * All tasks that modify the state of the threadpool and all tasks
    * that call out to threadpool listeners are pushed to this
    * taskprocessor.
    *
    * For instance, when the threadpool changes sizes, a task is put
    * into this taskprocessor to do so. When it comes time to tell the
    * threadpool listener that worker threads have changed state,
    * the task is placed in this taskprocessor.
    *
    * This is done for three main reasons
    * 1) It ensures that listeners are given an accurate portrayal
    * of the threadpool's current state. In other words, when a listener
    * gets told a count of active, idle and zombie threads, it does not
    * need to worry that internal state of the threadpool might be different
    * from what it has been told.
    * 2) It minimizes the locking required in both the threadpool and in
    * threadpool listener's callbacks.
    * 3) It ensures that listener callbacks are called in the same order
    * that the threadpool had its state change.
    */
   struct ast_taskprocessor *control_tps;
   /*! True if the threadpool is in the process of shutting down */
   int shutting_down;
   /*! Threadpool-specific options */
   struct ast_threadpool_options options;
};

/*!
 * \brief listener for a threadpool
 *
 * The listener is notified of changes in a threadpool. It can
 * react by doing things like increasing the number of threads
 * in the pool
 */
struct ast_threadpool_listener {
   /*! Callbacks called by the threadpool */
   const struct ast_threadpool_listener_callbacks *callbacks;
   /*! User data for the listener */
   void *user_data;
};

/*!
 * \brief states for worker threads
 */
enum worker_state {
   /*! The worker is either active or idle */
   ALIVE,
   /*!
    * The worker has been asked to shut down but
    * may still be in the process of executing tasks.
    * This transition happens when the threadpool needs
    * to shrink and needs to kill active threads in order
    * to do so.
    */
   ZOMBIE,
   /*!
    * The worker has been asked to shut down. Typically
    * only idle threads go to this state directly, but
    * active threads may go straight to this state when
    * the threadpool is shut down.
    */
   DEAD,
};

/*!
 * A thread that executes threadpool tasks
 */
struct worker_thread {
   /*! A unique (within a run of Asterisk) ID for the thread. Used for hashing and searching */
   int id;
   /*! Condition used in conjunction with state changes */
   ast_cond_t cond;
   /*! Lock used alongside the condition for state changes */
   ast_mutex_t lock;
   /*! The actual thread that is executing tasks */
   pthread_t thread;
   /*! A pointer to the threadpool. Needed to be able to execute tasks */
   struct ast_threadpool *pool;
   /*! The current state of the worker thread */
   enum worker_state state;
   /*! A boolean used to determine if an idle thread should become active */
   int wake_up;
   /*! Options for this threadpool */
   struct ast_threadpool_options options;
};

/* Worker thread forward declarations. See definitions for documentation */
static int worker_thread_hash(const void *obj, int flags);
static int worker_thread_cmp(void *obj, void *arg, int flags);
static void worker_thread_destroy(void *obj);
static void worker_active(struct worker_thread *worker);
static void *worker_start(void *arg);
static struct worker_thread *worker_thread_alloc(struct ast_threadpool *pool);
static int worker_thread_start(struct worker_thread *worker);
static int worker_idle(struct worker_thread *worker);
static int worker_set_state(struct worker_thread *worker, enum worker_state state);
static void worker_shutdown(struct worker_thread *worker);

/*!
 * \brief Notify the threadpool listener that the state has changed.
 *
 * This notifies the threadpool listener via its state_changed callback.
 * \param pool The threadpool whose state has changed
 */
static void threadpool_send_state_changed(struct ast_threadpool *pool)
{
   int active_size = ao2_container_count(pool->active_threads);
   int idle_size = ao2_container_count(pool->idle_threads);

   if (pool->listener && pool->listener->callbacks->state_changed) {
      pool->listener->callbacks->state_changed(pool, pool->listener, active_size, idle_size);
   }
}

/*!
 * \brief Struct used for queued operations involving worker state changes
 */
struct thread_worker_pair {
   /*! Threadpool that contains the worker whose state has changed */
   struct ast_threadpool *pool;
   /*! Worker whose state has changed */
   struct worker_thread *worker;
};

/*!
 * \brief Destructor for thread_worker_pair
 */
static void thread_worker_pair_free(struct thread_worker_pair *pair)
{
   ao2_ref(pair->worker, -1);
   ast_free(pair);
}

/*!
 * \brief Allocate and initialize a thread_worker_pair
 * \param pool Threadpool to assign to the thread_worker_pair
 * \param worker Worker thread to assign to the thread_worker_pair
 */
static struct thread_worker_pair *thread_worker_pair_alloc(struct ast_threadpool *pool,
      struct worker_thread *worker)
{
   struct thread_worker_pair *pair = ast_malloc(sizeof(*pair));
   if (!pair) {
      return NULL;
   }
   pair->pool = pool;
   ao2_ref(worker, +1);
   pair->worker = worker;

   return pair;
}

/*!
 * \brief Move a worker thread from the active container to the idle container.
 *
 * This function is called from the threadpool's control taskprocessor thread.
 * \param data A thread_worker_pair containing the threadpool and the worker to move.
 * \return 0
 */
static int queued_active_thread_idle(void *data)
{
   struct thread_worker_pair *pair = data;

   ao2_link(pair->pool->idle_threads, pair->worker);
   ao2_unlink(pair->pool->active_threads, pair->worker);

   threadpool_send_state_changed(pair->pool);

   thread_worker_pair_free(pair);
   return 0;
}

/*!
 * \brief Queue a task to move a thread from the active list to the idle list
 *
 * This is called by a worker thread when it runs out of tasks to perform and
 * goes idle.
 * \param pool The threadpool to which the worker belongs
 * \param worker The worker thread that has gone idle
 */
static void threadpool_active_thread_idle(struct ast_threadpool *pool,
      struct worker_thread *worker)
{
   struct thread_worker_pair *pair;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   pair = thread_worker_pair_alloc(pool, worker);
   if (!pair) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_active_thread_idle, pair)) {
      thread_worker_pair_free(pair);
   }
}

/*!
 * \brief Kill a zombie thread
 *
 * This runs from the threadpool's control taskprocessor thread.
 *
 * \param data A thread_worker_pair containing the threadpool and the zombie thread
 * \return 0
 */
static int queued_zombie_thread_dead(void *data)
{
   struct thread_worker_pair *pair = data;

   ao2_unlink(pair->pool->zombie_threads, pair->worker);
   threadpool_send_state_changed(pair->pool);

   thread_worker_pair_free(pair);
   return 0;
}

/*!
 * \brief Queue a task to kill a zombie thread
 *
 * This is called by a worker thread when it acknowledges that it is time for
 * it to die.
 */
static void threadpool_zombie_thread_dead(struct ast_threadpool *pool,
      struct worker_thread *worker)
{
   struct thread_worker_pair *pair;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   pair = thread_worker_pair_alloc(pool, worker);
   if (!pair) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_zombie_thread_dead, pair)) {
      thread_worker_pair_free(pair);
   }
}

static int queued_idle_thread_dead(void *data)
{
   struct thread_worker_pair *pair = data;

   ao2_unlink(pair->pool->idle_threads, pair->worker);
   threadpool_send_state_changed(pair->pool);

   thread_worker_pair_free(pair);
   return 0;
}

static void threadpool_idle_thread_dead(struct ast_threadpool *pool,
      struct worker_thread *worker)
{
   struct thread_worker_pair *pair;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   pair = thread_worker_pair_alloc(pool, worker);
   if (!pair) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_idle_thread_dead, pair)) {
      thread_worker_pair_free(pair);
   }
}

/*!
 * \brief Execute a task in the threadpool
 *
 * This is the function that worker threads call in order to execute tasks
 * in the threadpool
 *
 * \param pool The pool to which the tasks belong.
 * \retval 0 Either the pool has been shut down or there are no tasks.
 * \retval 1 There are still tasks remaining in the pool.
 */
static int threadpool_execute(struct ast_threadpool *pool)
{
   ao2_lock(pool);
   if (!pool->shutting_down) {
      ao2_unlock(pool);
      return ast_taskprocessor_execute(pool->tps);
   }
   ao2_unlock(pool);
   return 0;
}

/*!
 * \brief Destroy a threadpool's components.
 *
 * This is the destructor called automatically when the threadpool's
 * reference count reaches zero. This is not to be confused with
 * threadpool_destroy.
 *
 * By the time this actually gets called, most of the cleanup has already
 * been done in the pool. The only thing left to do is to release the
 * final reference to the threadpool listener.
 *
 * \param obj The pool to destroy
 */
static void threadpool_destructor(void *obj)
{
   struct ast_threadpool *pool = obj;
   ao2_cleanup(pool->listener);
}

/*
 * \brief Allocate a threadpool
 *
 * This is implemented as a taskprocessor listener's alloc callback. This
 * is because the threadpool exists as the private data on a taskprocessor
 * listener.
 *
 * \param name The name of the threadpool.
 * \param options The options the threadpool uses.
 * \retval NULL Could not initialize threadpool properly
 * \retval non-NULL The newly-allocated threadpool
 */
static struct ast_threadpool *threadpool_alloc(const char *name, const struct ast_threadpool_options *options)
{
   RAII_VAR(struct ast_threadpool *, pool, NULL, ao2_cleanup);
   struct ast_str *control_tps_name;

   pool = ao2_alloc(sizeof(*pool), threadpool_destructor);
   control_tps_name = ast_str_create(64);
   if (!pool || !control_tps_name) {
      ast_free(control_tps_name);
      return NULL;
   }

   ast_str_set(&control_tps_name, 0, "%s/pool-control", name);

   pool->control_tps = ast_taskprocessor_get(ast_str_buffer(control_tps_name), TPS_REF_DEFAULT);
   ast_free(control_tps_name);
   if (!pool->control_tps) {
      return NULL;
   }
   pool->active_threads = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
      THREAD_BUCKETS, worker_thread_hash, NULL, worker_thread_cmp);
   if (!pool->active_threads) {
      return NULL;
   }
   pool->idle_threads = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
      THREAD_BUCKETS, worker_thread_hash, NULL, worker_thread_cmp);
   if (!pool->idle_threads) {
      return NULL;
   }
   pool->zombie_threads = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
      THREAD_BUCKETS, worker_thread_hash, NULL, worker_thread_cmp);
   if (!pool->zombie_threads) {
      return NULL;
   }
   pool->options = *options;

   ao2_ref(pool, +1);
   return pool;
}

static int threadpool_tps_start(struct ast_taskprocessor_listener *listener)
{
   return 0;
}

/*!
 * \brief helper used for queued task when tasks are pushed
 */
struct task_pushed_data {
   /*! Pool into which a task was pushed */
   struct ast_threadpool *pool;
   /*! Indicator of whether the pool had no tasks prior to the new task being added */
   int was_empty;
};

/*!
 * \brief Allocate and initialize a task_pushed_data
 * \param pool The threadpool to set in the task_pushed_data
 * \param was_empty The was_empty value to set in the task_pushed_data
 * \retval NULL Unable to allocate task_pushed_data
 * \retval non-NULL The newly-allocated task_pushed_data
 */
static struct task_pushed_data *task_pushed_data_alloc(struct ast_threadpool *pool,
      int was_empty)
{
   struct task_pushed_data *tpd = ast_malloc(sizeof(*tpd));

   if (!tpd) {
      return NULL;
   }
   tpd->pool = pool;
   tpd->was_empty = was_empty;
   return tpd;
}

/*!
 * \brief Activate idle threads
 *
 * This function always returns CMP_MATCH because all workers that this
 * function acts on need to be seen as matches so they are unlinked from the
 * list of idle threads.
 *
 * Called as an ao2_callback in the threadpool's control taskprocessor thread.
 * \param obj The worker to activate
 * \param arg The pool where the worker belongs
 * \retval CMP_MATCH
 */
static int activate_thread(void *obj, void *arg, int flags)
{
   struct worker_thread *worker = obj;
   struct ast_threadpool *pool = arg;

   if (!ao2_link(pool->active_threads, worker)) {
      /* If we can't link the idle thread into the active container, then
       * we'll just leave the thread idle and not wake it up.
       */
      ast_log(LOG_WARNING, "Failed to activate thread %d. Remaining idle\n",
            worker->id);
      return 0;
   }

   if (worker_set_state(worker, ALIVE)) {
      ast_debug(1, "Failed to activate thread %d. It is dead\n",
            worker->id);
      /* The worker thread will no longer exist in the active threads or
       * idle threads container after this.
       */
      ao2_unlink(pool->active_threads, worker);
   }

   return CMP_MATCH;
}

/*!
 * \brief Add threads to the threadpool
 *
 * This function is called from the threadpool's control taskprocessor thread.
 * \param pool The pool that is expanding
 * \delta The number of threads to add to the pool
 */
static void grow(struct ast_threadpool *pool, int delta)
{
   int i;

   int current_size = ao2_container_count(pool->active_threads) +
      ao2_container_count(pool->idle_threads);

   if (pool->options.max_size && current_size + delta > pool->options.max_size) {
      delta = pool->options.max_size - current_size;
   }

   ast_debug(3, "Increasing threadpool %s's size by %d\n",
         ast_taskprocessor_name(pool->tps), delta);

   for (i = 0; i < delta; ++i) {
      struct worker_thread *worker = worker_thread_alloc(pool);
      if (!worker) {
         return;
      }
      if (ao2_link(pool->idle_threads, worker)) {
         if (worker_thread_start(worker)) {
            ast_log(LOG_ERROR, "Unable to start worker thread %d. Destroying.\n", worker->id);
            ao2_unlink(pool->active_threads, worker);
         }
      } else {
         ast_log(LOG_WARNING, "Failed to activate worker thread %d. Destroying.\n", worker->id);
      }
      ao2_ref(worker, -1);
   }
}

/*!
 * \brief Queued task called when tasks are pushed into the threadpool
 *
 * This function first calls into the threadpool's listener to let it know
 * that a task has been pushed. It then wakes up all idle threads and moves
 * them into the active thread container.
 * \param data A task_pushed_data
 * \return 0
 */
static int queued_task_pushed(void *data)
{
   struct task_pushed_data *tpd = data;
   struct ast_threadpool *pool = tpd->pool;
   int was_empty = tpd->was_empty;
   unsigned int existing_active;

   ast_free(tpd);

   if (pool->listener && pool->listener->callbacks->task_pushed) {
      pool->listener->callbacks->task_pushed(pool, pool->listener, was_empty);
   }

   existing_active = ao2_container_count(pool->active_threads);

   /* The first pass transitions any existing idle threads to be active, and
    * will also remove any worker threads that have recently entered the dead
    * state.
    */
   ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA,
         activate_thread, pool);

   /* If no idle threads could be transitioned to active grow the pool as permitted. */
   if (ao2_container_count(pool->active_threads) == existing_active) {
      if (!pool->options.auto_increment) {
         return 0;
      }
      grow(pool, pool->options.auto_increment);
      /* An optional second pass transitions any newly added threads. */
      ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA,
            activate_thread, pool);
   }

   threadpool_send_state_changed(pool);
   return 0;
}

/*!
 * \brief Taskprocessor listener callback called when a task is added
 *
 * The threadpool uses this opportunity to queue a task on its control taskprocessor
 * in order to activate idle threads and notify the threadpool listener that the
 * task has been pushed.
 * \param listener The taskprocessor listener. The threadpool is the listener's private data
 * \param was_empty True if the taskprocessor was empty prior to the task being pushed
 */
static void threadpool_tps_task_pushed(struct ast_taskprocessor_listener *listener,
      int was_empty)
{
   struct ast_threadpool *pool = ast_taskprocessor_listener_get_user_data(listener);
   struct task_pushed_data *tpd;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   tpd = task_pushed_data_alloc(pool, was_empty);
   if (!tpd) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_task_pushed, tpd)) {
      ast_free(tpd);
   }
}

/*!
 * \brief Queued task that handles the case where the threadpool's taskprocessor is emptied
 *
 * This simply lets the threadpool's listener know that the threadpool is devoid of tasks
 * \param data The pool that has become empty
 * \return 0
 */
static int queued_emptied(void *data)
{
   struct ast_threadpool *pool = data;

   /* We already checked for existence of this callback when this was queued */
   pool->listener->callbacks->emptied(pool, pool->listener);
   return 0;
}

/*!
 * \brief Taskprocessor listener emptied callback
 *
 * The threadpool queues a task to let the threadpool listener know that
 * the threadpool no longer contains any tasks.
 * \param listener The taskprocessor listener. The threadpool is the listener's private data.
 */
static void threadpool_tps_emptied(struct ast_taskprocessor_listener *listener)
{
   struct ast_threadpool *pool = ast_taskprocessor_listener_get_user_data(listener);
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   if (pool->listener && pool->listener->callbacks->emptied) {
      if (ast_taskprocessor_push(pool->control_tps, queued_emptied, pool)) {
         /* Nothing to do here but we need the check to keep the compiler happy. */
      }
   }
}

/*!
 * \brief Taskprocessor listener shutdown callback
 *
 * The threadpool will shut down and destroy all of its worker threads when
 * this is called back. By the time this gets called, the taskprocessor's
 * control taskprocessor has already been destroyed. Therefore there is no risk
 * in outright destroying the worker threads here.
 * \param listener The taskprocessor listener. The threadpool is the listener's private data.
 */
static void threadpool_tps_shutdown(struct ast_taskprocessor_listener *listener)
{
   struct ast_threadpool *pool = ast_taskprocessor_listener_get_user_data(listener);

   if (pool->listener && pool->listener->callbacks->shutdown) {
      pool->listener->callbacks->shutdown(pool->listener);
   }
   ao2_cleanup(pool->active_threads);
   ao2_cleanup(pool->idle_threads);
   ao2_cleanup(pool->zombie_threads);
   ao2_cleanup(pool);
}

/*!
 * \brief Table of taskprocessor listener callbacks for threadpool's main taskprocessor
 */
static struct ast_taskprocessor_listener_callbacks threadpool_tps_listener_callbacks = {
   .start = threadpool_tps_start,
   .task_pushed = threadpool_tps_task_pushed,
   .emptied = threadpool_tps_emptied,
   .shutdown = threadpool_tps_shutdown,
};

/*!
 * \brief ao2 callback to kill a set number of threads.
 *
 * Threads will be unlinked from the container as long as the
 * counter has not reached zero. The counter is decremented with
 * each thread that is removed.
 * \param obj The worker thread up for possible destruction
 * \param arg The counter
 * \param flags Unused
 * \retval CMP_MATCH The counter has not reached zero, so this flag should be removed.
 * \retval CMP_STOP The counter has reached zero so no more threads should be removed.
 */
static int kill_threads(void *obj, void *arg, int flags)
{
   int *num_to_kill = arg;

   if (*num_to_kill > 0) {
      --(*num_to_kill);
      return CMP_MATCH;
   } else {
      return CMP_STOP;
   }
}

/*!
 * \brief ao2 callback to zombify a set number of threads.
 *
 * Threads will be zombified as long as the counter has not reached
 * zero. The counter is decremented with each thread that is zombified.
 *
 * Zombifying a thread involves removing it from its current container,
 * adding it to the zombie container, and changing the state of the
 * worker to a zombie
 *
 * This callback is called from the threadpool control taskprocessor thread.
 *
 * \param obj The worker thread that may be zombified
 * \param arg The pool to which the worker belongs
 * \param data The counter
 * \param flags Unused
 * \retval CMP_MATCH The zombified thread should be removed from its current container
 * \retval CMP_STOP Stop attempting to zombify threads
 */
static int zombify_threads(void *obj, void *arg, void *data, int flags)
{
   struct worker_thread *worker = obj;
   struct ast_threadpool *pool = arg;
   int *num_to_zombify = data;

   if ((*num_to_zombify)-- > 0) {
      if (!ao2_link(pool->zombie_threads, worker)) {
         ast_log(LOG_WARNING, "Failed to zombify active thread %d. Thread will remain active\n", worker->id);
         return 0;
      }
      worker_set_state(worker, ZOMBIE);
      return CMP_MATCH;
   } else {
      return CMP_STOP;
   }
}

/*!
 * \brief Remove threads from the threadpool
 *
 * The preference is to kill idle threads. However, if there are
 * more threads to remove than there are idle threads, then active
 * threads will be zombified instead.
 *
 * This function is called from the threadpool control taskprocessor thread.
 *
 * \param pool The threadpool to remove threads from
 * \param delta The number of threads to remove
 */
static void shrink(struct ast_threadpool *pool, int delta)
{
   /*
    * Preference is to kill idle threads, but
    * we'll move on to deactivating active threads
    * if we have to
    */
   int idle_threads = ao2_container_count(pool->idle_threads);
   int idle_threads_to_kill = MIN(delta, idle_threads);
   int active_threads_to_zombify = delta - idle_threads_to_kill;

   ast_debug(3, "Destroying %d idle threads in threadpool %s\n", idle_threads_to_kill,
         ast_taskprocessor_name(pool->tps));

   ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
         kill_threads, &idle_threads_to_kill);

   ast_debug(3, "Destroying %d active threads in threadpool %s\n", active_threads_to_zombify,
         ast_taskprocessor_name(pool->tps));

   ao2_callback_data(pool->active_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
         zombify_threads, pool, &active_threads_to_zombify);
}

/*!
 * \brief Helper struct used for queued operations that change the size of the threadpool
 */
struct set_size_data {
   /*! The pool whose size is to change */
   struct ast_threadpool *pool;
   /*! The requested new size of the pool */
   unsigned int size;
};

/*!
 * \brief Allocate and initialize a set_size_data
 * \param pool The pool for the set_size_data
 * \param size The size to store in the set_size_data
 */
static struct set_size_data *set_size_data_alloc(struct ast_threadpool *pool,
      unsigned int size)
{
   struct set_size_data *ssd = ast_malloc(sizeof(*ssd));
   if (!ssd) {
      return NULL;
   }

   ssd->pool = pool;
   ssd->size = size;
   return ssd;
}

/*!
 * \brief Change the size of the threadpool
 *
 * This can either result in shrinking or growing the threadpool depending
 * on the new desired size and the current size.
 *
 * This function is run from the threadpool control taskprocessor thread
 *
 * \param data A set_size_data used for determining how to act
 * \return 0
 */
static int queued_set_size(void *data)
{
   struct set_size_data *ssd = data;
   struct ast_threadpool *pool = ssd->pool;
   unsigned int num_threads = ssd->size;

   /* We don't count zombie threads as being "live" when potentially resizing */
   unsigned int current_size = ao2_container_count(pool->active_threads) +
         ao2_container_count(pool->idle_threads);

   ast_free(ssd);

   if (current_size == num_threads) {
      ast_debug(3, "Not changing threadpool size since new size %u is the same as current %u\n",
           num_threads, current_size);
      return 0;
   }

   if (current_size < num_threads) {
      ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
            activate_thread, pool);

      /* As the above may have altered the number of current threads update it */
      current_size = ao2_container_count(pool->active_threads) +
            ao2_container_count(pool->idle_threads);
      grow(pool, num_threads - current_size);
      ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
            activate_thread, pool);
   } else {
      shrink(pool, current_size - num_threads);
   }

   threadpool_send_state_changed(pool);
   return 0;
}

void ast_threadpool_set_size(struct ast_threadpool *pool, unsigned int size)
{
   struct set_size_data *ssd;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   ssd = set_size_data_alloc(pool, size);
   if (!ssd) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_set_size, ssd)) {
      ast_free(ssd);
   }
}

struct ast_threadpool_listener *ast_threadpool_listener_alloc(
      const struct ast_threadpool_listener_callbacks *callbacks, void *user_data)
{
   struct ast_threadpool_listener *listener = ao2_alloc(sizeof(*listener), NULL);
   if (!listener) {
      return NULL;
   }
   listener->callbacks = callbacks;
   listener->user_data = user_data;
   return listener;
}

void *ast_threadpool_listener_get_user_data(const struct ast_threadpool_listener *listener)
{
   return listener->user_data;
}

struct pool_options_pair {
   struct ast_threadpool *pool;
   struct ast_threadpool_options options;
};

struct ast_threadpool *ast_threadpool_create(const char *name,
      struct ast_threadpool_listener *listener,
      const struct ast_threadpool_options *options)
{
   struct ast_taskprocessor *tps;
   RAII_VAR(struct ast_taskprocessor_listener *, tps_listener, NULL, ao2_cleanup);
   RAII_VAR(struct ast_threadpool *, pool, NULL, ao2_cleanup);
   char *fullname;

   pool = threadpool_alloc(name, options);
   if (!pool) {
      return NULL;
   }

   tps_listener = ast_taskprocessor_listener_alloc(&threadpool_tps_listener_callbacks, pool);
   if (!tps_listener) {
      return NULL;
   }

   if (options->version != AST_THREADPOOL_OPTIONS_VERSION) {
      ast_log(LOG_WARNING, "Incompatible version of threadpool options in use.\n");
      return NULL;
   }

   fullname = ast_alloca(strlen(name) + strlen("/pool") + 1);
   sprintf(fullname, "%s/pool", name); /* Safe */
   tps = ast_taskprocessor_create_with_listener(fullname, tps_listener);
   if (!tps) {
      return NULL;
   }

   pool->tps = tps;
   if (listener) {
      ao2_ref(listener, +1);
      pool->listener = listener;
   }
   ast_threadpool_set_size(pool, pool->options.initial_size);
   ao2_ref(pool, +1);
   return pool;
}

int ast_threadpool_push(struct ast_threadpool *pool, int (*task)(void *data), void *data)
{
   SCOPED_AO2LOCK(lock, pool);
   if (!pool->shutting_down) {
      return ast_taskprocessor_push(pool->tps, task, data);
   }
   return -1;
}

void ast_threadpool_shutdown(struct ast_threadpool *pool)
{
   if (!pool) {
      return;
   }
   /* Shut down the taskprocessors and everything else just
    * takes care of itself via the taskprocessor callbacks
    */
   ao2_lock(pool);
   pool->shutting_down = 1;
   ao2_unlock(pool);
   ast_taskprocessor_unreference(pool->control_tps);
   ast_taskprocessor_unreference(pool->tps);
}

/*!
 * A monotonically increasing integer used for worker
 * thread identification.
 */
static int worker_id_counter;

static int worker_thread_hash(const void *obj, int flags)
{
   const struct worker_thread *worker = obj;

   return worker->id;
}

static int worker_thread_cmp(void *obj, void *arg, int flags)
{
   struct worker_thread *worker1 = obj;
   struct worker_thread *worker2 = arg;

   return worker1->id == worker2->id ? CMP_MATCH : 0;
}

/*!
 * \brief shut a worker thread down
 *
 * Set the worker dead and then wait for its thread
 * to finish executing.
 *
 * \param worker The worker thread to shut down
 */
static void worker_shutdown(struct worker_thread *worker)
{
   worker_set_state(worker, DEAD);
   if (worker->thread != AST_PTHREADT_NULL) {
      pthread_join(worker->thread, NULL);
      worker->thread = AST_PTHREADT_NULL;
   }
}

/*!
 * \brief Worker thread destructor
 *
 * Called automatically when refcount reaches 0. Shuts
 * down the worker thread and destroys its component
 * parts
 */
static void worker_thread_destroy(void *obj)
{
   struct worker_thread *worker = obj;
   ast_debug(3, "Destroying worker thread %d\n", worker->id);
   worker_shutdown(worker);
   ast_mutex_destroy(&worker->lock);
   ast_cond_destroy(&worker->cond);
}

/*!
 * \brief start point for worker threads
 *
 * Worker threads start in the active state but may
 * immediately go idle if there is no work to be
 * done
 *
 * \param arg The worker thread
 * \retval NULL
 */
static void *worker_start(void *arg)
{
   struct worker_thread *worker = arg;
   enum worker_state saved_state;

   if (worker->options.thread_start) {
      worker->options.thread_start();
   }

   ast_mutex_lock(&worker->lock);
   while (worker_idle(worker)) {
      ast_mutex_unlock(&worker->lock);
      worker_active(worker);
      ast_mutex_lock(&worker->lock);
      if (worker->state != ALIVE) {
         break;
      }
      threadpool_active_thread_idle(worker->pool, worker);
   }
   saved_state = worker->state;
   ast_mutex_unlock(&worker->lock);

   /* Reaching this portion means the thread is
    * on death's door. It may have been killed while
    * it was idle, in which case it can just die
    * peacefully. If it's a zombie, though, then
    * it needs to let the pool know so
    * that the thread can be removed from the
    * list of zombie threads.
    */
   if (saved_state == ZOMBIE) {
      threadpool_zombie_thread_dead(worker->pool, worker);
   }

   if (worker->options.thread_end) {
      worker->options.thread_end();
   }
   return NULL;
}

/*!
 * \brief Allocate and initialize a new worker thread
 *
 * This will create, initialize, and start the thread.
 *
 * \param pool The threadpool to which the worker will be added
 * \retval NULL Failed to allocate or start the worker thread
 * \retval non-NULL The newly-created worker thread
 */
static struct worker_thread *worker_thread_alloc(struct ast_threadpool *pool)
{
   struct worker_thread *worker = ao2_alloc(sizeof(*worker), worker_thread_destroy);
   if (!worker) {
      return NULL;
   }
   worker->id = ast_atomic_fetchadd_int(&worker_id_counter, 1);
   ast_mutex_init(&worker->lock);
   ast_cond_init(&worker->cond, NULL);
   worker->pool = pool;
   worker->thread = AST_PTHREADT_NULL;
   worker->state = ALIVE;
   worker->options = pool->options;
   return worker;
}

static int worker_thread_start(struct worker_thread *worker)
{
   return ast_pthread_create(&worker->thread, NULL, worker_start, worker);
}

/*!
 * \brief Active loop for worker threads
 *
 * The worker will stay in this loop for its lifetime,
 * executing tasks as they become available. If there
 * are no tasks currently available, then the thread
 * will go idle.
 *
 * \param worker The worker thread executing tasks.
 */
static void worker_active(struct worker_thread *worker)
{
   int alive;

   /* The following is equivalent to
    *
    * while (threadpool_execute(worker->pool));
    *
    * However, reviewers have suggested in the past
    * doing that can cause optimizers to (wrongly)
    * optimize the code away.
    */
   do {
      alive = threadpool_execute(worker->pool);
   } while (alive);
}

/*!
 * \brief Idle function for worker threads
 *
 * The worker waits here until it gets told by the threadpool
 * to wake up.
 *
 * worker is locked before entering this function.
 *
 * \param worker The idle worker
 * \retval 0 The thread is being woken up so that it can conclude.
 * \retval non-zero The thread is being woken up to do more work.
 */
static int worker_idle(struct worker_thread *worker)
{
   struct timeval start = ast_tvnow();
   struct timespec end = {
      .tv_sec = start.tv_sec + worker->options.idle_timeout,
      .tv_nsec = start.tv_usec * 1000,
   };
   while (!worker->wake_up) {
      if (worker->options.idle_timeout <= 0) {
         ast_cond_wait(&worker->cond, &worker->lock);
      } else if (ast_cond_timedwait(&worker->cond, &worker->lock, &end) == ETIMEDOUT) {
         break;
      }
   }

   if (!worker->wake_up) {
      ast_debug(1, "Worker thread idle timeout reached. Dying.\n");
      threadpool_idle_thread_dead(worker->pool, worker);
      worker->state = DEAD;
   }
   worker->wake_up = 0;
   return worker->state == ALIVE;
}

/*!
 * \brief Change a worker's state
 *
 * The threadpool calls into this function in order to let a worker know
 * how it should proceed.
 *
 * \retval -1 failure (state transition not permitted)
 * \retval 0 success
 */
static int worker_set_state(struct worker_thread *worker, enum worker_state state)
{
   SCOPED_MUTEX(lock, &worker->lock);

   switch (state) {
   case ALIVE:
      /* This can occur due to a race condition between being told to go active
       * and an idle timeout happening.
       */
      if (worker->state == DEAD) {
         return -1;
      }
      ast_assert(worker->state != ZOMBIE);
      break;
   case DEAD:
      break;
   case ZOMBIE:
      ast_assert(worker->state != DEAD);
      break;
   }

   worker->state = state;
   worker->wake_up = 1;
   ast_cond_signal(&worker->cond);

   return 0;
}

/*! Serializer group shutdown control object. */
struct ast_serializer_shutdown_group {
   /*! Shutdown thread waits on this conditional. */
   ast_cond_t cond;
   /*! Count of serializers needing to shutdown. */
   int count;
};

static void serializer_shutdown_group_dtor(void *vdoomed)
{
   struct ast_serializer_shutdown_group *doomed = vdoomed;

   ast_cond_destroy(&doomed->cond);
}

struct ast_serializer_shutdown_group *ast_serializer_shutdown_group_alloc(void)
{
   struct ast_serializer_shutdown_group *shutdown_group;

   shutdown_group = ao2_alloc(sizeof(*shutdown_group), serializer_shutdown_group_dtor);
   if (!shutdown_group) {
      return NULL;
   }
   ast_cond_init(&shutdown_group->cond, NULL);
   return shutdown_group;
}

int ast_serializer_shutdown_group_join(struct ast_serializer_shutdown_group *shutdown_group, int timeout)
{
   int remaining;
   ast_mutex_t *lock;

   if (!shutdown_group) {
      return 0;
   }

   lock = ao2_object_get_lockaddr(shutdown_group);
   ast_assert(lock != NULL);

   ao2_lock(shutdown_group);
   if (timeout) {
      struct timeval start;
      struct timespec end;

      start = ast_tvnow();
      end.tv_sec = start.tv_sec + timeout;
      end.tv_nsec = start.tv_usec * 1000;
      while (shutdown_group->count) {
         if (ast_cond_timedwait(&shutdown_group->cond, lock, &end)) {
            /* Error or timed out waiting for the count to reach zero. */
            break;
         }
      }
   } else {
      while (shutdown_group->count) {
         if (ast_cond_wait(&shutdown_group->cond, lock)) {
            /* Error */
            break;
         }
      }
   }
   remaining = shutdown_group->count;
   ao2_unlock(shutdown_group);
   return remaining;
}

/*!
 * \internal
 * \brief Increment the number of serializer members in the group.
 * \since 13.5.0
 *
 * \param shutdown_group Group shutdown controller.
 *
 * \return Nothing
 */
static void serializer_shutdown_group_inc(struct ast_serializer_shutdown_group *shutdown_group)
{
   ao2_lock(shutdown_group);
   ++shutdown_group->count;
   ao2_unlock(shutdown_group);
}

/*!
 * \internal
 * \brief Decrement the number of serializer members in the group.
 * \since 13.5.0
 *
 * \param shutdown_group Group shutdown controller.
 *
 * \return Nothing
 */
static void serializer_shutdown_group_dec(struct ast_serializer_shutdown_group *shutdown_group)
{
   ao2_lock(shutdown_group);
   --shutdown_group->count;
   if (!shutdown_group->count) {
      ast_cond_signal(&shutdown_group->cond);
   }
   ao2_unlock(shutdown_group);
}

struct serializer {
   /*! Threadpool the serializer will use to process the jobs. */
   struct ast_threadpool *pool;
   /*! Which group will wait for this serializer to shutdown. */
   struct ast_serializer_shutdown_group *shutdown_group;
};

static void serializer_dtor(void *obj)
{
   struct serializer *ser = obj;

   ao2_cleanup(ser->pool);
   ser->pool = NULL;
   ao2_cleanup(ser->shutdown_group);
   ser->shutdown_group = NULL;
}

static struct serializer *serializer_create(struct ast_threadpool *pool,
   struct ast_serializer_shutdown_group *shutdown_group)
{
   struct serializer *ser;

   ser = ao2_alloc_options(sizeof(*ser), serializer_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
   if (!ser) {
      return NULL;
   }
   ao2_ref(pool, +1);
   ser->pool = pool;
   ser->shutdown_group = ao2_bump(shutdown_group);
   return ser;
}

AST_THREADSTORAGE_RAW(current_serializer);

static int execute_tasks(void *data)
{
   struct ast_taskprocessor *tps = data;

   ast_threadstorage_set_ptr(&current_serializer, tps);
   while (ast_taskprocessor_execute(tps)) {
      /* No-op */
   }
   ast_threadstorage_set_ptr(&current_serializer, NULL);

   ast_taskprocessor_unreference(tps);
   return 0;
}

static void serializer_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
{
   if (was_empty) {
      struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
      struct ast_taskprocessor *tps = ast_taskprocessor_listener_get_tps(listener);

      if (ast_threadpool_push(ser->pool, execute_tasks, tps)) {
         ast_taskprocessor_unreference(tps);
      }
   }
}

static int serializer_start(struct ast_taskprocessor_listener *listener)
{
   /* No-op */
   return 0;
}

static void serializer_shutdown(struct ast_taskprocessor_listener *listener)
{
   struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);

   if (ser->shutdown_group) {
      serializer_shutdown_group_dec(ser->shutdown_group);
   }
   ao2_cleanup(ser);
}

static struct ast_taskprocessor_listener_callbacks serializer_tps_listener_callbacks = {
   .task_pushed = serializer_task_pushed,
   .start = serializer_start,
   .shutdown = serializer_shutdown,
};

struct ast_taskprocessor *ast_threadpool_serializer_get_current(void)
{
   return ast_threadstorage_get_ptr(&current_serializer);
}

struct ast_taskprocessor *ast_threadpool_serializer_group(const char *name,
   struct ast_threadpool *pool, struct ast_serializer_shutdown_group *shutdown_group)
{
   struct serializer *ser;
   struct ast_taskprocessor_listener *listener;
   struct ast_taskprocessor *tps;

   ser = serializer_create(pool, shutdown_group);
   if (!ser) {
      return NULL;
   }

   listener = ast_taskprocessor_listener_alloc(&serializer_tps_listener_callbacks, ser);
   if (!listener) {
      ao2_ref(ser, -1);
      return NULL;
   }

   tps = ast_taskprocessor_create_with_listener(name, listener);
   if (!tps) {
      /* ser ref transferred to listener but not cleaned without tps */
      ao2_ref(ser, -1);
   } else if (shutdown_group) {
      serializer_shutdown_group_inc(shutdown_group);
   }

   ao2_ref(listener, -1);
   return tps;
}

struct ast_taskprocessor *ast_threadpool_serializer(const char *name, struct ast_threadpool *pool)
{
   return ast_threadpool_serializer_group(name, pool, NULL);
}

long ast_threadpool_queue_size(struct ast_threadpool *pool)
{
   return ast_taskprocessor_size(pool->tps);
}