bind/lib/isc/tests/task_test.c

/*
 * Copyright (C) Internet Systems Consortium, Inc. ("ISC")
 *
 * SPDX-License-Identifier: MPL-2.0
 *
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, you can obtain one at https://mozilla.org/MPL/2.0/.
 *
 * See the COPYRIGHT file distributed with this work for additional
 * information regarding copyright ownership.
 */

#if HAVE_CMOCKA

#include <inttypes.h>
#include <sched.h> /* IWYU pragma: keep */
#include <setjmp.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define UNIT_TESTING

#include <cmocka.h>

#include <isc/atomic.h>
#include <isc/cmocka.h>
#include <isc/commandline.h>
#include <isc/condition.h>
#include <isc/managers.h>
#include <isc/mem.h>
#include <isc/print.h>
#include <isc/task.h>
#include <isc/time.h>
#include <isc/timer.h>
#include <isc/util.h>

#include "isctest.h"

/* Set to true (or use -v option) for verbose output */
static bool verbose = false;

static isc_mutex_t lock;
static isc_condition_t cv;

atomic_int_fast32_t counter;
static int active[10];
static atomic_bool done;

static int
_setup(void **state) {
	isc_result_t result;

	UNUSED(state);

	isc_mutex_init(&lock);

	isc_condition_init(&cv);

	result = isc_test_begin(NULL, true, 0);
	assert_int_equal(result, ISC_R_SUCCESS);

	return (0);
}

static int
_setup2(void **state) {
	isc_result_t result;

	UNUSED(state);

	isc_mutex_init(&lock);

	isc_condition_init(&cv);

	/* Two worker threads */
	result = isc_test_begin(NULL, true, 2);
	assert_int_equal(result, ISC_R_SUCCESS);

	return (0);
}

static int
_setup4(void **state) {
	isc_result_t result;

	UNUSED(state);

	isc_mutex_init(&lock);

	isc_condition_init(&cv);

	/* Four worker threads */
	result = isc_test_begin(NULL, true, 4);
	assert_int_equal(result, ISC_R_SUCCESS);

	return (0);
}

static int
_teardown(void **state) {
	UNUSED(state);

	isc_test_end();
	isc_condition_destroy(&cv);

	return (0);
}

static void
set(isc_task_t *task, isc_event_t *event) {
	atomic_int_fast32_t *value = (atomic_int_fast32_t *)event->ev_arg;

	UNUSED(task);

	isc_event_free(&event);
	atomic_store(value, atomic_fetch_add(&counter, 1));
}

#include <isc/thread.h>

/* Create a task */
static void
create_task(void **state) {
	isc_result_t result;
	isc_task_t *task = NULL;

	UNUSED(state);

	result = isc_task_create(taskmgr, 0, &task);
	assert_int_equal(result, ISC_R_SUCCESS);

	isc_task_destroy(&task);
	assert_null(task);
}

/* Process events */
static void
all_events(void **state) {
	isc_result_t result;
	isc_task_t *task = NULL;
	isc_event_t *event = NULL;
	atomic_int_fast32_t a, b;
	int i = 0;

	UNUSED(state);

	atomic_init(&counter, 1);
	atomic_init(&a, 0);
	atomic_init(&b, 0);

	result = isc_task_create(taskmgr, 0, &task);
	assert_int_equal(result, ISC_R_SUCCESS);

	/* First event */
	event = isc_event_allocate(test_mctx, task, ISC_TASKEVENT_TEST, set, &a,
				   sizeof(isc_event_t));
	assert_non_null(event);

	assert_int_equal(atomic_load(&a), 0);
	isc_task_send(task, &event);

	event = isc_event_allocate(test_mctx, task, ISC_TASKEVENT_TEST, set, &b,
				   sizeof(isc_event_t));
	assert_non_null(event);

	assert_int_equal(atomic_load(&b), 0);
	isc_task_send(task, &event);

	while ((atomic_load(&a) == 0 || atomic_load(&b) == 0) && i++ < 5000) {
		isc_test_nap(1000);
	}

	assert_int_not_equal(atomic_load(&a), 0);
	assert_int_not_equal(atomic_load(&b), 0);

	isc_task_destroy(&task);
	assert_null(task);
}

/*
 * Basic task functions:
 */
static void
basic_cb(isc_task_t *task, isc_event_t *event) {
	int i, j;

	UNUSED(task);

	j = 0;
	for (i = 0; i < 1000000; i++) {
		j += 100;
	}

	UNUSED(j);

	if (verbose) {
		print_message("# task %s\n", (char *)event->ev_arg);
	}

	isc_event_free(&event);
}

static void
basic_shutdown(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	if (verbose) {
		print_message("# shutdown %s\n", (char *)event->ev_arg);
	}

	isc_event_free(&event);
}

static void
basic_tick(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	if (verbose) {
		print_message("# %s\n", (char *)event->ev_arg);
	}

	isc_event_free(&event);
}

static char one[] = "1";
static char two[] = "2";
static char three[] = "3";
static char four[] = "4";
static char tick[] = "tick";
static char tock[] = "tock";

static void
basic(void **state) {
	isc_result_t result;
	isc_task_t *task1 = NULL;
	isc_task_t *task2 = NULL;
	isc_task_t *task3 = NULL;
	isc_task_t *task4 = NULL;
	isc_event_t *event = NULL;
	isc_timer_t *ti1 = NULL;
	isc_timer_t *ti2 = NULL;
	isc_interval_t interval;
	char *testarray[] = { one, one, one,   one,  one, one,	 one,  one,
			      one, two, three, four, two, three, four, NULL };
	int i;

	UNUSED(state);

	result = isc_task_create(taskmgr, 0, &task1);
	assert_int_equal(result, ISC_R_SUCCESS);
	result = isc_task_create(taskmgr, 0, &task2);
	assert_int_equal(result, ISC_R_SUCCESS);
	result = isc_task_create(taskmgr, 0, &task3);
	assert_int_equal(result, ISC_R_SUCCESS);
	result = isc_task_create(taskmgr, 0, &task4);
	assert_int_equal(result, ISC_R_SUCCESS);

	result = isc_task_onshutdown(task1, basic_shutdown, one);
	assert_int_equal(result, ISC_R_SUCCESS);
	result = isc_task_onshutdown(task2, basic_shutdown, two);
	assert_int_equal(result, ISC_R_SUCCESS);
	result = isc_task_onshutdown(task3, basic_shutdown, three);
	assert_int_equal(result, ISC_R_SUCCESS);
	result = isc_task_onshutdown(task4, basic_shutdown, four);
	assert_int_equal(result, ISC_R_SUCCESS);

	isc_interval_set(&interval, 1, 0);
	isc_timer_create(timermgr, task1, basic_tick, tick, &ti1);
	result = isc_timer_reset(ti1, isc_timertype_ticker, &interval, false);
	assert_int_equal(result, ISC_R_SUCCESS);

	ti2 = NULL;
	isc_interval_set(&interval, 1, 0);
	isc_timer_create(timermgr, task2, basic_tick, tock, &ti2);
	result = isc_timer_reset(ti2, isc_timertype_ticker, &interval, false);
	assert_int_equal(result, ISC_R_SUCCESS);

	sleep(2);

	for (i = 0; testarray[i] != NULL; i++) {
		/*
		 * Note:  (void *)1 is used as a sender here, since some
		 * compilers don't like casting a function pointer to a
		 * (void *).
		 *
		 * In a real use, it is more likely the sender would be a
		 * structure (socket, timer, task, etc) but this is just a
		 * test program.
		 */
		event = isc_event_allocate(test_mctx, (void *)1, 1, basic_cb,
					   testarray[i], sizeof(*event));
		assert_non_null(event);
		isc_task_send(task1, &event);
	}

	isc_task_detach(&task1);
	isc_task_detach(&task2);
	isc_task_detach(&task3);
	isc_task_detach(&task4);

	sleep(10);
	isc_timer_detach(&ti1);
	isc_timer_detach(&ti2);
}

/*
 * Exclusive mode test:
 * When one task enters exclusive mode, all other active
 * tasks complete first.
 */
static int
spin(int n) {
	int i;
	int r = 0;
	for (i = 0; i < n; i++) {
		r += i;
		if (r > 1000000) {
			r = 0;
		}
	}
	return (r);
}

static void
exclusive_cb(isc_task_t *task, isc_event_t *event) {
	int taskno = *(int *)(event->ev_arg);

	if (verbose) {
		print_message("# task enter %d\n", taskno);
	}

	/* task chosen from the middle of the range */
	if (taskno == 6) {
		isc_result_t result;
		int i;

		result = isc_task_beginexclusive(task);
		assert_int_equal(result, ISC_R_SUCCESS);

		for (i = 0; i < 10; i++) {
			assert_int_equal(active[i], 0);
		}

		isc_task_endexclusive(task);
		atomic_store(&done, true);
	} else {
		active[taskno]++;
		(void)spin(10000000);
		active[taskno]--;
	}

	if (verbose) {
		print_message("# task exit %d\n", taskno);
	}

	if (atomic_load(&done)) {
		isc_mem_put(event->ev_destroy_arg, event->ev_arg, sizeof(int));
		isc_event_free(&event);
		atomic_fetch_sub(&counter, 1);
	} else {
		isc_task_send(task, &event);
	}
}

static void
task_exclusive(void **state) {
	isc_task_t *tasks[10];
	isc_result_t result;
	int i;

	UNUSED(state);

	atomic_init(&counter, 0);

	for (i = 0; i < 10; i++) {
		isc_event_t *event = NULL;
		int *v;

		tasks[i] = NULL;

		if (i == 6) {
			/* task chosen from the middle of the range */
			result = isc_task_create_bound(taskmgr, 0, &tasks[i],
						       0);
			assert_int_equal(result, ISC_R_SUCCESS);

			isc_taskmgr_setexcltask(taskmgr, tasks[6]);
		} else {
			result = isc_task_create(taskmgr, 0, &tasks[i]);
			assert_int_equal(result, ISC_R_SUCCESS);
		}

		v = isc_mem_get(test_mctx, sizeof *v);
		assert_non_null(v);

		*v = i;

		event = isc_event_allocate(test_mctx, NULL, 1, exclusive_cb, v,
					   sizeof(*event));
		assert_non_null(event);

		isc_task_send(tasks[i], &event);
		atomic_fetch_add(&counter, 1);
	}

	for (i = 0; i < 10; i++) {
		isc_task_detach(&tasks[i]);
	}

	while (atomic_load(&counter) > 0) {
		isc_test_nap(1000);
	}
}

/*
 * Max tasks test:
 * The task system can create and execute many tasks. Tests with 10000.
 */
static void
maxtask_shutdown(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	if (event->ev_arg != NULL) {
		isc_task_destroy((isc_task_t **)&event->ev_arg);
	} else {
		LOCK(&lock);
		atomic_store(&done, true);
		SIGNAL(&cv);
		UNLOCK(&lock);
	}

	isc_event_free(&event);
}

static void
maxtask_cb(isc_task_t *task, isc_event_t *event) {
	isc_result_t result;

	if (event->ev_arg != NULL) {
		isc_task_t *newtask = NULL;

		event->ev_arg = (void *)(((uintptr_t)event->ev_arg) - 1);

		/*
		 * Create a new task and forward the message.
		 */
		result = isc_task_create(taskmgr, 0, &newtask);
		assert_int_equal(result, ISC_R_SUCCESS);

		result = isc_task_onshutdown(newtask, maxtask_shutdown,
					     (void *)task);
		assert_int_equal(result, ISC_R_SUCCESS);

		isc_task_send(newtask, &event);
	} else if (task != NULL) {
		isc_task_destroy(&task);
		isc_event_free(&event);
	}
}

static void
manytasks(void **state) {
	isc_mem_t *mctx = NULL;
	isc_event_t *event = NULL;
	uintptr_t ntasks = 10000;

	UNUSED(state);

	if (verbose) {
		print_message("# Testing with %lu tasks\n",
			      (unsigned long)ntasks);
	}

	isc_mutex_init(&lock);
	isc_condition_init(&cv);

	isc_mem_debugging = ISC_MEM_DEBUGRECORD;
	isc_mem_create(&mctx);

	isc_managers_create(mctx, 4, 0, &netmgr, &taskmgr, NULL);

	atomic_init(&done, false);

	event = isc_event_allocate(mctx, (void *)1, 1, maxtask_cb,
				   (void *)ntasks, sizeof(*event));
	assert_non_null(event);

	LOCK(&lock);
	maxtask_cb(NULL, event);
	while (!atomic_load(&done)) {
		WAIT(&cv, &lock);
	}
	UNLOCK(&lock);

	isc_managers_destroy(&netmgr, &taskmgr, NULL);

	isc_mem_destroy(&mctx);
	isc_condition_destroy(&cv);
	isc_mutex_destroy(&lock);
}

/*
 * Shutdown test:
 * When isc_task_shutdown() is called, shutdown events are posted
 * in LIFO order.
 */

static int nevents = 0;
static int nsdevents = 0;
static int senders[4];
atomic_bool ready, all_done;

static void
sd_sde1(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	assert_int_equal(nevents, 256);
	assert_int_equal(nsdevents, 1);
	++nsdevents;

	if (verbose) {
		print_message("# shutdown 1\n");
	}

	isc_event_free(&event);

	atomic_store(&all_done, true);
}

static void
sd_sde2(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	assert_int_equal(nevents, 256);
	assert_int_equal(nsdevents, 0);
	++nsdevents;

	if (verbose) {
		print_message("# shutdown 2\n");
	}

	isc_event_free(&event);
}

static void
sd_event1(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	LOCK(&lock);
	while (!atomic_load(&ready)) {
		WAIT(&cv, &lock);
	}
	UNLOCK(&lock);

	if (verbose) {
		print_message("# event 1\n");
	}

	isc_event_free(&event);
}

static void
sd_event2(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	++nevents;

	if (verbose) {
		print_message("# event 2\n");
	}

	isc_event_free(&event);
}

static void
task_shutdown(void **state) {
	isc_result_t result;
	isc_eventtype_t event_type;
	isc_event_t *event = NULL;
	isc_task_t *task = NULL;
	int i;

	UNUSED(state);

	nevents = nsdevents = 0;
	event_type = 3;
	atomic_init(&ready, false);
	atomic_init(&all_done, false);

	LOCK(&lock);

	result = isc_task_create(taskmgr, 0, &task);
	assert_int_equal(result, ISC_R_SUCCESS);

	/*
	 * This event causes the task to wait on cv.
	 */
	event = isc_event_allocate(test_mctx, &senders[1], event_type,
				   sd_event1, NULL, sizeof(*event));
	assert_non_null(event);
	isc_task_send(task, &event);

	/*
	 * Now we fill up the task's event queue with some events.
	 */
	for (i = 0; i < 256; ++i) {
		event = isc_event_allocate(test_mctx, &senders[1], event_type,
					   sd_event2, NULL, sizeof(*event));
		assert_non_null(event);
		isc_task_send(task, &event);
	}

	/*
	 * Now we register two shutdown events.
	 */
	result = isc_task_onshutdown(task, sd_sde1, NULL);
	assert_int_equal(result, ISC_R_SUCCESS);

	result = isc_task_onshutdown(task, sd_sde2, NULL);
	assert_int_equal(result, ISC_R_SUCCESS);

	isc_task_shutdown(task);
	isc_task_detach(&task);

	/*
	 * Now we free the task by signaling cv.
	 */
	atomic_store(&ready, true);
	SIGNAL(&cv);
	UNLOCK(&lock);

	while (!atomic_load(&all_done)) {
		isc_test_nap(1000);
	}

	assert_int_equal(nsdevents, 2);
}

/*
 * Post-shutdown test:
 * After isc_task_shutdown() has been called, any call to
 * isc_task_onshutdown() will return ISC_R_SHUTTINGDOWN.
 */
static void
psd_event1(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	LOCK(&lock);

	while (!atomic_load(&done)) {
		WAIT(&cv, &lock);
	}

	UNLOCK(&lock);

	isc_event_free(&event);
}

static void
psd_sde(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	isc_event_free(&event);
}

static void
post_shutdown(void **state) {
	isc_result_t result;
	isc_eventtype_t event_type;
	isc_event_t *event;
	isc_task_t *task;

	UNUSED(state);

	atomic_init(&done, false);
	event_type = 4;

	isc_condition_init(&cv);

	LOCK(&lock);

	task = NULL;
	result = isc_task_create(taskmgr, 0, &task);
	assert_int_equal(result, ISC_R_SUCCESS);

	/*
	 * This event causes the task to wait on cv.
	 */
	event = isc_event_allocate(test_mctx, &senders[1], event_type,
				   psd_event1, NULL, sizeof(*event));
	assert_non_null(event);
	isc_task_send(task, &event);

	isc_task_shutdown(task);

	result = isc_task_onshutdown(task, psd_sde, NULL);
	assert_int_equal(result, ISC_R_SHUTTINGDOWN);

	/*
	 * Release the task.
	 */
	atomic_store(&done, true);

	SIGNAL(&cv);
	UNLOCK(&lock);

	isc_task_detach(&task);
}

/*
 * Helper for the purge tests below:
 */

#define SENDERCNT 3
#define TYPECNT	  4
#define TAGCNT	  5
#define NEVENTS	  (SENDERCNT * TYPECNT * TAGCNT)

static int eventcnt;

atomic_bool started;

/*
 * Helpers for purge event tests
 */
static void
pge_event1(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	LOCK(&lock);
	while (!atomic_load(&started)) {
		WAIT(&cv, &lock);
	}
	UNLOCK(&lock);

	isc_event_free(&event);
}

static void
pge_event2(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	++eventcnt;
	isc_event_free(&event);
}

static void
pge_sde(isc_task_t *task, isc_event_t *event) {
	UNUSED(task);

	LOCK(&lock);
	atomic_store(&done, true);
	SIGNAL(&cv);
	UNLOCK(&lock);

	isc_event_free(&event);
}

static void
try_purgeevent(void) {
	isc_result_t result;
	isc_task_t *task = NULL;
	bool purged;
	isc_event_t *event1 = NULL;
	isc_event_t *event2 = NULL;
	isc_event_t *event2_clone = NULL;
	isc_time_t now;
	isc_interval_t interval;

	atomic_init(&started, false);
	atomic_init(&done, false);
	eventcnt = 0;

	isc_condition_init(&cv);

	result = isc_task_create(taskmgr, 0, &task);
	assert_int_equal(result, ISC_R_SUCCESS);

	result = isc_task_onshutdown(task, pge_sde, NULL);
	assert_int_equal(result, ISC_R_SUCCESS);

	/*
	 * Block the task on cv.
	 */
	event1 = isc_event_allocate(test_mctx, (void *)1, (isc_eventtype_t)1,
				    pge_event1, NULL, sizeof(*event1));
	assert_non_null(event1);
	isc_task_send(task, &event1);

	event2 = isc_event_allocate(test_mctx, (void *)1, (isc_eventtype_t)1,
				    pge_event2, NULL, sizeof(*event2));
	assert_non_null(event2);

	event2_clone = event2;

	isc_task_send(task, &event2);

	purged = isc_task_purgeevent(task, event2_clone);

	assert_true(purged);

	/*
	 * Unblock the task, allowing event processing.
	 */
	LOCK(&lock);
	atomic_store(&started, true);
	SIGNAL(&cv);

	isc_task_shutdown(task);

	isc_interval_set(&interval, 5, 0);

	/*
	 * Wait for shutdown processing to complete.
	 */
	while (!atomic_load(&done)) {
		result = isc_time_nowplusinterval(&now, &interval);
		assert_int_equal(result, ISC_R_SUCCESS);

		WAITUNTIL(&cv, &lock, &now);
	}

	UNLOCK(&lock);

	isc_task_detach(&task);
}

/*
 * Purge event test:
 * When the event is marked as purgeable, a call to
 * isc_task_purgeevent(task, event) purges the event 'event' from the
 * task's queue and returns true.
 */

static void
purgeevent(void **state) {
	UNUSED(state);

	try_purgeevent();
}

int
main(int argc, char **argv) {
	const struct CMUnitTest tests[] = {
		cmocka_unit_test(manytasks),
		cmocka_unit_test_setup_teardown(all_events, _setup, _teardown),
		cmocka_unit_test_setup_teardown(basic, _setup2, _teardown),
		cmocka_unit_test_setup_teardown(create_task, _setup, _teardown),
		cmocka_unit_test_setup_teardown(post_shutdown, _setup2,
						_teardown),
		cmocka_unit_test_setup_teardown(purgeevent, _setup2, _teardown),
		cmocka_unit_test_setup_teardown(task_shutdown, _setup4,
						_teardown),
		cmocka_unit_test_setup_teardown(task_exclusive, _setup4,
						_teardown),
	};
	struct CMUnitTest selected[sizeof(tests) / sizeof(tests[0])];
	size_t i;
	int c;

	memset(selected, 0, sizeof(selected));

	while ((c = isc_commandline_parse(argc, argv, "lt:v")) != -1) {
		switch (c) {
		case 'l':
			for (i = 0; i < (sizeof(tests) / sizeof(tests[0])); i++)
			{
				if (tests[i].name != NULL) {
					fprintf(stdout, "%s\n", tests[i].name);
				}
			}
			return (0);
		case 't':
			if (!cmocka_add_test_byname(
				    tests, isc_commandline_argument, selected))
			{
				fprintf(stderr, "unknown test '%s'\n",
					isc_commandline_argument);
				exit(1);
			}
			break;
		case 'v':
			verbose = true;
			break;
		default:
			break;
		}
	}

	if (selected[0].name != NULL) {
		return (cmocka_run_group_tests(selected, NULL, NULL));
	} else {
		return (cmocka_run_group_tests(tests, NULL, NULL));
	}
}

#else /* HAVE_CMOCKA */

#include <stdio.h>

int
main(void) {
	printf("1..0 # Skipped: cmocka not available\n");
	return (SKIPPED_TEST_EXIT_CODE);
}

#endif /* if HAVE_CMOCKA */