test_threadpool.c

/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2012-2013, Digium, Inc.
 *
 * Mark Michelson <mmichelson@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.
 */

/*!
 * \file
 * \brief threadpool unit tests
 *
 * \author Mark Michelson <mmichelson@digium.com>
 *
 */

/*** MODULEINFO
	<depend>TEST_FRAMEWORK</depend>
	<support_level>core</support_level>
 ***/

#include "asterisk.h"

#include "asterisk/astobj2.h"
#include "asterisk/lock.h"
#include "asterisk/logger.h"
#include "asterisk/module.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/test.h"
#include "asterisk/threadpool.h"

struct test_listener_data {
	int num_active;
	int num_idle;
	int task_pushed;
	int num_tasks;
	int empty_notice;
	int was_empty;
	ast_mutex_t lock;
	ast_cond_t cond;
};

static struct test_listener_data *test_alloc(void)
{
	struct test_listener_data *tld = ast_calloc(1, sizeof(*tld));
	if (!tld) {
		return NULL;
	}
	ast_mutex_init(&tld->lock);
	ast_cond_init(&tld->cond, NULL);
	return tld;
}

static void test_state_changed(struct ast_threadpool *pool,
		struct ast_threadpool_listener *listener,
		int active_threads,
		int idle_threads)
{
	struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
	SCOPED_MUTEX(lock, &tld->lock);
	tld->num_active = active_threads;
	tld->num_idle = idle_threads;
	ast_log(LOG_NOTICE, "Thread state: %d active, %d idle\n", tld->num_active, tld->num_idle);
	ast_cond_signal(&tld->cond);
}

static void test_task_pushed(struct ast_threadpool *pool,
		struct ast_threadpool_listener *listener,
		int was_empty)
{
	struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
	SCOPED_MUTEX(lock, &tld->lock);
	tld->task_pushed = 1;
	++tld->num_tasks;
	tld->was_empty = was_empty;
	ast_cond_signal(&tld->cond);
}

static void test_emptied(struct ast_threadpool *pool,
		struct ast_threadpool_listener *listener)
{
	struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
	SCOPED_MUTEX(lock, &tld->lock);
	tld->empty_notice = 1;
	ast_cond_signal(&tld->cond);
}

static void test_shutdown(struct ast_threadpool_listener *listener)
{
	struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
	ast_cond_destroy(&tld->cond);
	ast_mutex_destroy(&tld->lock);
}

static const struct ast_threadpool_listener_callbacks test_callbacks = {
	.state_changed = test_state_changed,
	.task_pushed = test_task_pushed,
	.emptied = test_emptied,
	.shutdown = test_shutdown,
};

struct simple_task_data {
	int task_executed;
	ast_mutex_t lock;
	ast_cond_t cond;
};

static struct simple_task_data *simple_task_data_alloc(void)
{
	struct simple_task_data *std = ast_calloc(1, sizeof(*std));

	if (!std) {
		return NULL;
	}
	ast_mutex_init(&std->lock);
	ast_cond_init(&std->cond, NULL);
	return std;
}

static int simple_task(void *data)
{
	struct simple_task_data *std = data;
	SCOPED_MUTEX(lock, &std->lock);
	std->task_executed = 1;
	ast_cond_signal(&std->cond);
	return 0;
}

static enum ast_test_result_state wait_until_thread_state(struct ast_test *test, struct test_listener_data *tld, int num_active, int num_idle)
{
	struct timeval start = ast_tvnow();
	struct timespec end = {
		.tv_sec = start.tv_sec + 5,
		.tv_nsec = start.tv_usec * 1000
	};
	enum ast_test_result_state res = AST_TEST_PASS;
	SCOPED_MUTEX(lock, &tld->lock);

	while (!(tld->num_active == num_active && tld->num_idle == num_idle)) {
		if (ast_cond_timedwait(&tld->cond, &tld->lock, &end) == ETIMEDOUT) {
			break;
		}
	}

	if (tld->num_active != num_active && tld->num_idle != num_idle) {
		ast_test_status_update(test, "Number of active threads and idle threads not what was expected.\n");
		ast_test_status_update(test, "Expected %d active threads but got %d\n", num_active, tld->num_active);
		ast_test_status_update(test, "Expected %d idle threads but got %d\n", num_idle, tld->num_idle);
		res = AST_TEST_FAIL;
	}

	return res;
}

static void wait_for_task_pushed(struct ast_threadpool_listener *listener)
{
	struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
	struct timeval start = ast_tvnow();
	struct timespec end = {
		.tv_sec = start.tv_sec + 5,
		.tv_nsec = start.tv_usec * 1000
	};
	SCOPED_MUTEX(lock, &tld->lock);

	while (!tld->task_pushed) {
		if (ast_cond_timedwait(&tld->cond, lock, &end) == ETIMEDOUT) {
			break;
		}
	}
}

static enum ast_test_result_state wait_for_completion(struct ast_test *test, struct simple_task_data *std)
{
	struct timeval start = ast_tvnow();
	struct timespec end = {
		.tv_sec = start.tv_sec + 5,
		.tv_nsec = start.tv_usec * 1000
	};
	enum ast_test_result_state res = AST_TEST_PASS;
	SCOPED_MUTEX(lock, &std->lock);

	while (!std->task_executed) {
		if (ast_cond_timedwait(&std->cond, lock, &end) == ETIMEDOUT) {
			break;
		}
	}

	if (!std->task_executed) {
		ast_test_status_update(test, "Task execution did not occur\n");
		res = AST_TEST_FAIL;