/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2012-2013, Digium, Inc.
 *
 * Mark Michelson <mmmichelson@digium.com>
 *
 * 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 currently 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
 * \param flags
 * \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
 * \param 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
 */
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.
 */
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.
 */
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);
}