From e5e05b42cff1690dc78ddc01004d346bc2fc2fe5 Mon Sep 17 00:00:00 2001 From: DuoYuWang Date: Mon, 7 Sep 2026 16:54:52 +0800 Subject: [PATCH] testing/ostest: exercise custom work queues Add a focused wqueue entry that exercises predefined and dynamically created work queues without changing the no-argument ostest flow. Share the semaphore wait and basic worker helpers across configurations. Flat builds exercise scheduler queues, Protected builds exercise USRWORK, and pthread-enabled builds additionally cover configurable custom worker pools. When pthread support is disabled, only predefined USRWORK cases are compiled. Cover invalid arguments, priority lookup, immediate and delayed work, pending replacement and cancellation, periodic requeue, synchronous and parallel cancellation, concurrent queues, callback self-destruction, and queue teardown. Assisted-by: Codex:GPT-5 Signed-off-by: DuoYuWang --- testing/ostest/CMakeLists.txt | 8 +- testing/ostest/Kconfig | 10 + testing/ostest/Makefile | 8 +- testing/ostest/ostest.h | 2 +- testing/ostest/ostest_main.c | 36 +- testing/ostest/wqueue.c | 1124 +++++++++++++++++++++++++++++---- 6 files changed, 1056 insertions(+), 132 deletions(-) diff --git a/testing/ostest/CMakeLists.txt b/testing/ostest/CMakeLists.txt index 7b8a034c374..c5225610559 100644 --- a/testing/ostest/CMakeLists.txt +++ b/testing/ostest/CMakeLists.txt @@ -126,15 +126,15 @@ if(CONFIG_TESTING_OSTEST) list(APPEND SRCS sporadic.c sporadic2.c) endif() - if(CONFIG_SCHED_WORKQUEUE) - list(APPEND SRCS wqueue.c) - endif() - if(CONFIG_PRIORITY_INHERITANCE) list(APPEND SRCS prioinherit.c) endif() # CONFIG_PRIORITY_INHERITANCE endif() # CONFIG_DISABLE_PTHREAD + if(CONFIG_TESTING_OSTEST_WQUEUE) + list(APPEND SRCS wqueue.c) + endif() + if(NOT CONFIG_DISABLE_MQUEUE) if(NOT CONFIG_DISABLE_PTHREAD) list(APPEND SRCS timedmqueue.c) diff --git a/testing/ostest/Kconfig b/testing/ostest/Kconfig index aa4425edc60..35738246369 100644 --- a/testing/ostest/Kconfig +++ b/testing/ostest/Kconfig @@ -51,6 +51,16 @@ config TESTING_OSTEST_AIOPATH endif +config TESTING_OSTEST_WQUEUE + bool "Work queue tests" + default y + depends on (BUILD_FLAT && !DISABLE_PTHREAD && SCHED_WORKQUEUE) || LIBC_USRWORK + ---help--- + Enables the work queue tests. In non-flat builds, the tests run in + user space and require the libc user work queue implementation. If + pthread support is disabled, only the predefined USRWORK queue is + tested because dynamically allocated queues require pthread support. + config TESTING_OSTEST_RR_RANGE int "Round-robin test - end of search range" default 30000 diff --git a/testing/ostest/Makefile b/testing/ostest/Makefile index 9ddd8fffbac..12afab653c0 100644 --- a/testing/ostest/Makefile +++ b/testing/ostest/Makefile @@ -127,15 +127,15 @@ ifeq ($(CONFIG_SCHED_SPORADIC),y) CSRCS += sporadic.c sporadic2.c endif -ifeq ($(CONFIG_SCHED_WORKQUEUE),y) -CSRCS += wqueue.c -endif - ifeq ($(CONFIG_PRIORITY_INHERITANCE),y) CSRCS += prioinherit.c endif endif # CONFIG_DISABLE_PTHREAD +ifeq ($(CONFIG_TESTING_OSTEST_WQUEUE),y) +CSRCS += wqueue.c +endif + ifneq ($(CONFIG_DISABLE_MQUEUE),y) ifneq ($(CONFIG_DISABLE_PTHREAD),y) CSRCS += timedmqueue.c diff --git a/testing/ostest/ostest.h b/testing/ostest/ostest.h index e9b50981116..38e95d1fb7a 100644 --- a/testing/ostest/ostest.h +++ b/testing/ostest/ostest.h @@ -146,7 +146,7 @@ int chroot_test(void); /* wqueue.c *****************************************************************/ -#if defined(CONFIG_SCHED_LPWORK) || defined(CONFIG_SCHED_HPWORK) +#ifdef CONFIG_TESTING_OSTEST_WQUEUE void wqueue_test(void); #endif diff --git a/testing/ostest/ostest_main.c b/testing/ostest/ostest_main.c index 523b6aa4af1..095b63c0245 100644 --- a/testing/ostest/ostest_main.c +++ b/testing/ostest/ostest_main.c @@ -155,6 +155,7 @@ static void show_variable(const char *var_name, const char *exptd_value, bool var_valid) { char *actual_value = getenv(var_name); + if (actual_value) { if (var_valid) @@ -302,19 +303,19 @@ static int user_main(int argc, char *argv[]) #if defined(CONFIG_SCHED_HAVE_PARENT) && defined(CONFIG_SCHED_CHILD_STATUS) && \ defined(CONFIG_ENABLE_ALL_SIGNALS) - { - struct sigaction sa; - int ret; + { + struct sigaction sa; + int ret; - sa.sa_handler = SIG_IGN; - sa.sa_flags = SA_NOCLDWAIT; - ret = sigaction(SIGCHLD, &sa, NULL); - if (ret < 0) - { - printf("user_main: ERROR: sigaction failed: %d\n", errno); - ASSERT(false); - } - } + sa.sa_handler = SIG_IGN; + sa.sa_flags = SA_NOCLDWAIT; + ret = sigaction(SIGCHLD, &sa, NULL); + if (ret < 0) + { + printf("user_main: ERROR: sigaction failed: %d\n", errno); + ASSERT(false); + } + } #endif #ifndef CONFIG_DISABLE_ENVIRON @@ -430,8 +431,7 @@ static int user_main(int argc, char *argv[]) check_test_memory_usage(); #endif -#if !defined(CONFIG_DISABLE_PTHREAD) && defined(CONFIG_BUILD_FLAT) && \ - defined(CONFIG_SCHED_WORKQUEUE) +#if defined(CONFIG_TESTING_OSTEST_WQUEUE) && defined(CONFIG_BUILD_FLAT) /* Check work queues */ printf("\nuser_main: wqueue test\n"); @@ -770,6 +770,14 @@ int main(int argc, FAR char **argv) } #endif +#ifdef CONFIG_TESTING_OSTEST_WQUEUE + if (argc == 2 && strcmp(argv[1], "wqueue") == 0) + { + wqueue_test(); + return EXIT_SUCCESS; + } +#endif + /* Verify that stdio works first */ stdio_test(); diff --git a/testing/ostest/wqueue.c b/testing/ostest/wqueue.c index 29c3271fd7c..73acdef8b21 100644 --- a/testing/ostest/wqueue.c +++ b/testing/ostest/wqueue.c @@ -27,75 +27,831 @@ #include #include +#include +#ifndef CONFIG_DISABLE_PTHREAD #include +#endif #include #include +#include +#include #include +#include +#include #include +#include #include #include "ostest.h" -#ifdef CONFIG_SCHED_WORKQUEUE - /**************************************************************************** * Pre-processor Definitions ****************************************************************************/ -#define SLEEP_TIME (100 * 1000) -#define TEST_COUNT (100) -#define VERIFY_COUNT (100) +#define SLEEP_TIME (100 * 1000) +#define TEST_COUNT 100 +#define VERIFY_COUNT 100 +#define WQUEUE_TEST_TIMEOUT_SEC 2 -#ifdef CONFIG_SCHED_LPWORK -# define TEST_QUEUE LPWORK -# define TEST_QUEUE_PRIORITY CONFIG_SCHED_LPWORKPRIORITY -#else -# define TEST_QUEUE HPWORK -# define TEST_QUEUE_PRIORITY CONFIG_SCHED_HPWORKPRIORITY -#endif +#define MULTI_QUEUE_COUNT 4 +#define MULTI_QUEUE_LOOPS 4 +#define MULTI_WORK_PER_QUEUE 8 +#define CUSTOM_PRIORITY 100 + +/**************************************************************************** + * Private Types + ****************************************************************************/ + +struct sync_cancel_s +{ + sem_t started; + sem_t finished; +}; + +struct replace_s +{ + sem_t done; + int total; +}; + +struct replace_arg_s +{ + FAR struct replace_s *test; + int value; +}; + +#ifdef CONFIG_LIBC_USRWORK + +struct usrwork_periodic_s +{ + struct work_s work; + sem_t done; + int calls; + int result; +}; + +/**************************************************************************** + * Private Data + ****************************************************************************/ + +static int g_usrwork_errors; + +#endif /* CONFIG_LIBC_USRWORK */ /**************************************************************************** * Private Functions ****************************************************************************/ +static int wait_sem(FAR sem_t *sem) +{ + struct timespec abstime; + int ret; + + ret = clock_gettime(CLOCK_REALTIME, &abstime); + if (ret < 0) + { + return -errno; + } + + abstime.tv_sec += WQUEUE_TEST_TIMEOUT_SEC; + + do + { + ret = sem_timedwait(sem, &abstime); + } + while (ret < 0 && errno == EINTR); + + return ret < 0 ? -errno : OK; +} + static void empty_worker(FAR void *arg) { + UNUSED(arg); } -static void sleep_worker(FAR void *arg) +static void count_worker(FAR void *arg) { FAR sem_t *sem = arg; - usleep(SLEEP_TIME); sem_post(sem); } -static void count_worker(FAR void *arg) +static void replace_worker(FAR void *arg) +{ + FAR struct replace_arg_s *replace = arg; + + replace->test->total += replace->value; + sem_post(&replace->test->done); +} + +static void sync_worker(FAR void *arg) +{ + FAR struct sync_cancel_s *test = arg; + + sem_post(&test->started); + usleep(SLEEP_TIME); + sem_post(&test->finished); +} + +#ifdef CONFIG_LIBC_USRWORK + +static void usrwork_check_result(FAR const char *name, int actual, + int expected) +{ + if (actual != expected) + { + printf("wqueue_test: ERROR %s: got %d, expected %d\n", + name, actual, expected); + g_usrwork_errors++; + } +} + +static void usrwork_check_true(FAR const char *name, bool result) +{ + if (!result) + { + printf("wqueue_test: ERROR %s\n", name); + g_usrwork_errors++; + } +} + +static void usrwork_periodic_worker(FAR void *arg) +{ + FAR struct usrwork_periodic_s *test = arg; + + test->calls++; + if (test->calls < 3) + { + test->result = work_queue_next(USRWORK, &test->work, + usrwork_periodic_worker, test, 1); + if (test->result < 0) + { + sem_post(&test->done); + } + } + else + { + sem_post(&test->done); + } +} + +static void usrwork_api_validation_test(void) +{ + struct work_s work; + clock_t excessive_delay = WDOG_MAX_DELAY + 1; + + printf("wqueue_test: API validation\n"); + memset(&work, 0, sizeof(work)); + + usrwork_check_result("null work", + work_queue(USRWORK, NULL, empty_worker, NULL, 0), -EINVAL); + usrwork_check_result("null worker", + work_queue(USRWORK, &work, NULL, NULL, 0), -EINVAL); + usrwork_check_result("negative delay", + work_queue(USRWORK, &work, empty_worker, NULL, -1), -EINVAL); + usrwork_check_result("excessive delay", + work_queue(USRWORK, &work, empty_worker, NULL, + excessive_delay), -EINVAL); + usrwork_check_result("negative periodic delay", + work_queue_next(USRWORK, &work, empty_worker, NULL, -1), + -EINVAL); + usrwork_check_result("excessive periodic delay", + work_queue_next(USRWORK, &work, empty_worker, NULL, + excessive_delay), -EINVAL); + usrwork_check_result("invalid queue", + work_queue(-1, &work, empty_worker, NULL, 0), -EINVAL); + usrwork_check_result("invalid periodic queue", + work_queue_next(-1, &work, empty_worker, NULL, 0), -EINVAL); + usrwork_check_result("invalid cancel", work_cancel(-1, &work), -EINVAL); + usrwork_check_result("invalid sync cancel", + work_cancel_sync(-1, &work), -EINVAL); + usrwork_check_result("null cancel", work_cancel(USRWORK, NULL), -EINVAL); + usrwork_check_result("null sync cancel", + work_cancel_sync(USRWORK, NULL), -EINVAL); + usrwork_check_result("idle cancel", work_cancel(USRWORK, &work), OK); + usrwork_check_result("idle sync cancel", + work_cancel_sync(USRWORK, &work), OK); + usrwork_check_true("idle work available", work_available(&work)); + printf("wqueue_test: API validation done\n"); +} + +static void usrwork_priority_test(void) +{ + int priority; + + priority = work_queue_priority(USRWORK); + printf("wqueue_test: priority = %d, expect = %d\n", + priority, CONFIG_LIBC_USRWORKPRIORITY); + usrwork_check_result("USRWORK priority", priority, + CONFIG_LIBC_USRWORKPRIORITY); +} + +static void usrwork_queue_test(clock_t delay) +{ + struct work_s work; + sem_t called; + int ret; + + memset(&work, 0, sizeof(work)); + ret = sem_init(&called, 0, 0); + usrwork_check_result( + delay == 0 ? "immediate sem init" : "delayed sem init", ret, OK); + if (ret < 0) + { + return; + } + + ret = work_queue(USRWORK, &work, count_worker, &called, delay); + usrwork_check_result( + delay == 0 ? "immediate queue" : "delayed queue", ret, OK); + if (ret == OK) + { + ret = wait_sem(&called); + usrwork_check_result( + delay == 0 ? "immediate callback" : "delayed callback", ret, OK); + } + + usrwork_check_result("queue cleanup", + work_cancel_sync(USRWORK, &work), OK); + usrwork_check_true("queued work available", work_available(&work)); + usrwork_check_result("queue sem destroy", sem_destroy(&called), OK); +} + +static void usrwork_pending_replace_test(void) +{ + struct replace_arg_s first; + struct replace_arg_s second; + struct replace_s test; + struct work_s work; + int ret; + + printf("wqueue_test: pending replacement\n"); + memset(&test, 0, sizeof(test)); + memset(&work, 0, sizeof(work)); + first.test = &test; + first.value = 1; + second.test = &test; + second.value = 2; + + ret = sem_init(&test.done, 0, 0); + usrwork_check_result("replacement sem init", ret, OK); + if (ret < 0) + { + return; + } + + ret = work_queue(USRWORK, &work, replace_worker, &first, + MSEC2TICK(100)); + usrwork_check_result("replacement first queue", ret, OK); + if (ret == OK) + { + ret = work_queue(USRWORK, &work, replace_worker, &second, 1); + usrwork_check_result("replacement second queue", ret, OK); + if (ret == OK) + { + usrwork_check_result("replacement callback", + wait_sem(&test.done), OK); + } + } + + usrwork_check_result("replacement cleanup", + work_cancel_sync(USRWORK, &work), OK); + usrwork_check_result("replacement total", test.total, 2); + usrwork_check_true("replacement work available", work_available(&work)); + usrwork_check_result("replacement sem destroy", + sem_destroy(&test.done), OK); +} + +static void usrwork_pending_cancel_test(void) +{ + struct work_s work; + sem_t called; + int count = -1; + int ret; + + printf("wqueue_test: pending cancel\n"); + memset(&work, 0, sizeof(work)); + ret = sem_init(&called, 0, 0); + usrwork_check_result("pending sem init", ret, OK); + if (ret < 0) + { + return; + } + + ret = work_queue(USRWORK, &work, count_worker, &called, + MSEC2TICK(100)); + usrwork_check_result("pending queue", ret, OK); + if (ret == OK) + { + usrwork_check_result("pending cancel", + work_cancel(USRWORK, &work), OK); + usleep(150 * 1000); + usrwork_check_result("pending sem value", + sem_getvalue(&called, &count), OK); + usrwork_check_result("pending callback count", count, 0); + } + + usrwork_check_result("pending cleanup", + work_cancel_sync(USRWORK, &work), OK); + usrwork_check_true("pending work available", work_available(&work)); + usrwork_check_result("pending sem destroy", sem_destroy(&called), OK); +} + +static void usrwork_sync_cancel_test(void) +{ + struct sync_cancel_s test; + struct work_s work; + int count = -1; + int ret; + + printf("wqueue_test: synchronous cancel\n"); + memset(&work, 0, sizeof(work)); + ret = sem_init(&test.started, 0, 0); + usrwork_check_result("sync started sem init", ret, OK); + if (ret < 0) + { + return; + } + + ret = sem_init(&test.finished, 0, 0); + usrwork_check_result("sync finished sem init", ret, OK); + if (ret < 0) + { + sem_destroy(&test.started); + return; + } + + ret = work_queue(USRWORK, &work, sync_worker, &test, 0); + usrwork_check_result("sync queue", ret, OK); + if (ret == OK) + { + ret = wait_sem(&test.started); + usrwork_check_result("sync callback start", ret, OK); + if (ret == OK) + { + usrwork_check_result("sync cancel", + work_cancel_sync(USRWORK, &work), OK); + usrwork_check_result("sync finished value", + sem_getvalue(&test.finished, &count), OK); + usrwork_check_result("sync finished count", count, 1); + } + } + + usrwork_check_result("sync cleanup", + work_cancel_sync(USRWORK, &work), OK); + usrwork_check_true("sync work available", work_available(&work)); + usrwork_check_result("sync finished sem destroy", + sem_destroy(&test.finished), OK); + usrwork_check_result("sync started sem destroy", + sem_destroy(&test.started), OK); +} + +static void usrwork_periodic_test(void) +{ + struct usrwork_periodic_s test; + int ret; + + printf("wqueue_test: periodic requeue\n"); + memset(&test, 0, sizeof(test)); + ret = sem_init(&test.done, 0, 0); + usrwork_check_result("periodic sem init", ret, OK); + if (ret < 0) + { + return; + } + + test.result = work_queue(USRWORK, &test.work, usrwork_periodic_worker, + &test, 1); + usrwork_check_result("periodic queue", test.result, OK); + if (test.result == OK) + { + usrwork_check_result("periodic callback", wait_sem(&test.done), OK); + } + + usrwork_check_result("periodic cleanup", + work_cancel_sync(USRWORK, &test.work), OK); + usrwork_check_result("periodic result", test.result, OK); + usrwork_check_result("periodic calls", test.calls, 3); + usrwork_check_true("periodic work available", work_available(&test.work)); + usrwork_check_result("periodic sem destroy", + sem_destroy(&test.done), OK); +} + +/**************************************************************************** + * Private Functions + ****************************************************************************/ + +static void usrwork_test(void) +{ + g_usrwork_errors = 0; + + printf("wqueue_test: backend = predefined USRWORK\n"); + usrwork_api_validation_test(); + usrwork_priority_test(); + usrwork_queue_test(0); + usrwork_queue_test(MSEC2TICK(20)); + usrwork_pending_replace_test(); + usrwork_pending_cancel_test(); + usrwork_sync_cancel_test(); + usrwork_periodic_test(); + + if (g_usrwork_errors == 0) + { + printf("wqueue_test: PASS\n"); + } + else + { + printf("wqueue_test: FAIL (%d errors)\n", g_usrwork_errors); + } + + ASSERT(g_usrwork_errors == 0); +} + +#endif /* CONFIG_LIBC_USRWORK */ + +#ifndef CONFIG_DISABLE_PTHREAD + +/**************************************************************************** + * Private Types + ****************************************************************************/ + +typedef FAR void *(*test_thread_entry_t)(FAR void *arg); + +/**************************************************************************** + * Private Functions + ****************************************************************************/ + +static void run_test_thread(test_thread_entry_t entry, FAR void *arg, + int priority, int stacksize) +{ + pthread_t thread; + pthread_attr_t attr; + struct sched_param sparam; + int status; + + status = pthread_attr_init(&attr); + ASSERT(status == OK); + + status = pthread_attr_setschedpolicy(&attr, SCHED_FIFO); + ASSERT(status == OK); + + memset(&sparam, 0, sizeof(sparam)); + sparam.sched_priority = priority; + status = pthread_attr_setschedparam(&attr, &sparam); + ASSERT(status == OK); + + status = pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED); + ASSERT(status == OK); + + if (stacksize > 0) + { + status = pthread_attr_setstacksize(&attr, stacksize); + ASSERT(status == OK); + } + + status = pthread_create(&thread, &attr, entry, arg); + ASSERT(status == OK); + status = pthread_join(thread, NULL); + ASSERT(status == OK); + status = pthread_attr_destroy(&attr); + ASSERT(status == OK); +} + +static void sleep_worker(FAR void *arg) { FAR sem_t *sem = arg; + usleep(SLEEP_TIME); sem_post(sem); } +struct requeue_s +{ + FAR struct kwork_wqueue_s *wqueue; + FAR struct work_s *work; + sem_t started; + int next_result; + int queue_result; +}; + +struct self_free_s +{ + FAR struct kwork_wqueue_s *wqueue; + sem_t done; + int result; +}; + +struct periodic_s +{ + FAR struct kwork_wqueue_s *wqueue; + struct work_s work; + sem_t done; + int calls; + int result; +}; + +struct parallel_cancel_s; + +struct parallel_worker_s +{ + FAR struct parallel_cancel_s *test; + sem_t started; + sem_t release; + bool short_delay; +}; + +struct parallel_cancel_s +{ + struct parallel_worker_s worker[2]; + sem_t finished; +}; + +static void requeue_worker(FAR void *arg) +{ + FAR struct requeue_s *requeue = arg; + + sem_post(&requeue->started); + usleep(SLEEP_TIME); + requeue->next_result = work_queue_next_wq(requeue->wqueue, + requeue->work, + empty_worker, NULL, 1); + requeue->queue_result = work_queue_wq(requeue->wqueue, requeue->work, + empty_worker, NULL, 0); +} + +static void self_free_worker(FAR void *arg) +{ + FAR struct self_free_s *self_free = arg; + + self_free->result = work_queue_free(self_free->wqueue); + sem_post(&self_free->done); +} + +static void periodic_worker(FAR void *arg) +{ + FAR struct periodic_s *periodic = arg; + + periodic->calls++; + + if (periodic->calls < 3) + { + periodic->result = work_queue_next_wq(periodic->wqueue, + &periodic->work, + periodic_worker, + periodic, 1); + + if (periodic->result < 0) + { + sem_post(&periodic->done); + } + } + else + { + sem_post(&periodic->done); + } +} + +static void parallel_worker(FAR void *arg) +{ + FAR struct parallel_worker_s *worker = arg; + + sem_post(&worker->started); + ASSERT(wait_sem(&worker->release) == OK); + + if (worker->short_delay) + { + usleep(SLEEP_TIME / 10); + } + else + { + usleep(SLEEP_TIME); + } + + sem_post(&worker->test->finished); +} + +static FAR void *release_thread(FAR void *arg) +{ + FAR struct parallel_cancel_s *test = arg; + + usleep(SLEEP_TIME / 10); + sem_post(&test->worker[0].release); + sem_post(&test->worker[1].release); + return NULL; +} + +static void sync_cancel_test(FAR void *wq) +{ + struct sync_cancel_s sync; + struct work_s work; + int count; + int ret; + + ASSERT(sem_init(&sync.started, 0, 0) == OK); + ASSERT(sem_init(&sync.finished, 0, 0) == OK); + memset(&work, 0, sizeof(work)); + + ret = work_queue_wq(wq, &work, sync_worker, &sync, 0); + ASSERT(ret == OK); + + ASSERT(wait_sem(&sync.started) == OK); + ret = work_cancel_sync_wq(wq, &work); + ASSERT(ret == OK); + + sem_getvalue(&sync.finished, &count); + printf("wqueue_test: sync cancel finished = %d, expect = 1\n", count); + ASSERT(count == 1); + + ASSERT(sem_destroy(&sync.finished) == OK); + ASSERT(sem_destroy(&sync.started) == OK); +} + +static void parallel_cancel_test(FAR void *wq) +{ + struct parallel_cancel_s test; + struct work_s work; + pthread_t releaser; + int count; + int ret; + + printf("wqueue_test: parallel sync cancel\n"); + memset(&test, 0, sizeof(test)); + memset(&work, 0, sizeof(work)); + ASSERT(sem_init(&test.finished, 0, 0) == OK); + + test.worker[0].test = &test; + test.worker[1].test = &test; + test.worker[0].short_delay = true; + ASSERT(sem_init(&test.worker[0].started, 0, 0) == OK); + ASSERT(sem_init(&test.worker[0].release, 0, 0) == OK); + ASSERT(sem_init(&test.worker[1].started, 0, 0) == OK); + ASSERT(sem_init(&test.worker[1].release, 0, 0) == OK); + + ret = work_queue_wq(wq, &work, parallel_worker, &test.worker[0], 0); + ASSERT(ret == OK); + ASSERT(wait_sem(&test.worker[0].started) == OK); + + ret = work_queue_wq(wq, &work, parallel_worker, &test.worker[1], 0); + ASSERT(ret == OK); + ASSERT(wait_sem(&test.worker[1].started) == OK); + + ret = pthread_create(&releaser, NULL, release_thread, &test); + ASSERT(ret == OK); + ret = work_cancel_sync_wq(wq, &work); + ASSERT(ret == OK); + ASSERT(pthread_join(releaser, NULL) == OK); + + ASSERT(sem_getvalue(&test.finished, &count) == OK); + printf("wqueue_test: parallel callbacks = %d, expect = 2\n", count); + ASSERT(count == 2); + ASSERT(work_available(&work)); + + ASSERT(sem_destroy(&test.worker[1].release) == OK); + ASSERT(sem_destroy(&test.worker[1].started) == OK); + ASSERT(sem_destroy(&test.worker[0].release) == OK); + ASSERT(sem_destroy(&test.worker[0].started) == OK); + ASSERT(sem_destroy(&test.finished) == OK); +} + +static void self_free_test(void) +{ + struct self_free_s self_free; + struct work_s work; + int ret; + + printf("wqueue_test: self free\n"); + memset(&work, 0, sizeof(work)); + ASSERT(sem_init(&self_free.done, 0, 0) == OK); + self_free.wqueue = work_queue_create("test", CUSTOM_PRIORITY, NULL, + STACKSIZE, 1); + ASSERT(self_free.wqueue != NULL); + self_free.result = OK; + + ret = work_queue_wq(self_free.wqueue, &work, self_free_worker, + &self_free, 0); + ASSERT(ret == OK); + ASSERT(wait_sem(&self_free.done) == OK); + printf("wqueue_test: self free result = %d, expect = %d\n", + self_free.result, -EDEADLK); + ASSERT(self_free.result == -EDEADLK); + ASSERT(work_queue_free(self_free.wqueue) == OK); + ASSERT(work_available(&work)); + ASSERT(sem_destroy(&self_free.done) == OK); +} + +static void api_validation_test(FAR struct kwork_wqueue_s *wqueue) +{ + struct work_s work; + clock_t excessive_delay = WDOG_MAX_DELAY + 1; + + printf("wqueue_test: API validation\n"); + memset(&work, 0, sizeof(work)); + + ASSERT(work_queue_wq(NULL, &work, empty_worker, NULL, 0) == -EINVAL); + ASSERT(work_queue_wq(wqueue, NULL, empty_worker, NULL, 0) == -EINVAL); + ASSERT(work_queue_wq(wqueue, &work, NULL, NULL, 0) == -EINVAL); + ASSERT(work_queue_wq(wqueue, &work, empty_worker, NULL, -1) == -EINVAL); + ASSERT(work_queue_wq(wqueue, &work, empty_worker, NULL, + excessive_delay) == -EINVAL); + ASSERT(work_queue_next_wq(wqueue, &work, empty_worker, NULL, -1) == + -EINVAL); + ASSERT(work_queue_next_wq(wqueue, &work, empty_worker, NULL, + excessive_delay) == -EINVAL); + ASSERT(work_queue(-1, &work, empty_worker, NULL, 0) == -EINVAL); + ASSERT(work_queue_next(-1, &work, empty_worker, NULL, 0) == -EINVAL); + ASSERT(work_cancel(-1, &work) == -EINVAL); + ASSERT(work_cancel_sync(-1, &work) == -EINVAL); + ASSERT(work_cancel_wq(NULL, &work) == -EINVAL); + ASSERT(work_cancel_wq(wqueue, NULL) == -EINVAL); + ASSERT(work_cancel_sync_wq(NULL, &work) == -EINVAL); + ASSERT(work_cancel_sync_wq(wqueue, NULL) == -EINVAL); + ASSERT(work_queue_priority_wq(NULL) == -EINVAL); + + /* Cancelling idle work is intentionally idempotent. */ + + ASSERT(work_cancel_wq(wqueue, &work) == OK); + ASSERT(work_cancel_sync_wq(wqueue, &work) == OK); + ASSERT(work_available(&work)); + printf("wqueue_test: API validation done\n"); +} + +static void periodic_test(FAR struct kwork_wqueue_s *wqueue) +{ + struct periodic_s periodic; + + printf("wqueue_test: periodic requeue\n"); + memset(&periodic, 0, sizeof(periodic)); + periodic.wqueue = wqueue; + periodic.result = OK; + ASSERT(sem_init(&periodic.done, 0, 0) == OK); + ASSERT(work_queue_wq(wqueue, &periodic.work, periodic_worker, + &periodic, 1) == OK); + ASSERT(wait_sem(&periodic.done) == OK); + ASSERT(periodic.result == OK); + ASSERT(periodic.calls == 3); + ASSERT(work_available(&periodic.work)); + ASSERT(sem_destroy(&periodic.done) == OK); + printf("wqueue_test: periodic calls = %d, expect = 3\n", + periodic.calls); +} + +static void pending_replace_test(FAR struct kwork_wqueue_s *wqueue) +{ + struct replace_arg_s first; + struct replace_arg_s second; + struct replace_s replace; + struct work_s work; + int count; + + printf("wqueue_test: pending replacement\n"); + memset(&replace, 0, sizeof(replace)); + memset(&work, 0, sizeof(work)); + first.test = &replace; + first.value = 1; + second.test = &replace; + second.value = 2; + ASSERT(sem_init(&replace.done, 0, 0) == OK); + + ASSERT(work_queue_wq(wqueue, &work, replace_worker, &first, + MSEC2TICK(200)) == OK); + ASSERT(work_queue_next_wq(wqueue, &work, replace_worker, + &second, 1) == OK); + ASSERT(wait_sem(&replace.done) == OK); + usleep(300 * 1000); + ASSERT(sem_getvalue(&replace.done, &count) == OK); + ASSERT(count == 0); + ASSERT(replace.total == 2); + ASSERT(work_available(&work)); + ASSERT(sem_destroy(&replace.done) == OK); + printf("wqueue_test: replacement total = %d, expect = 2\n", + replace.total); +} + static FAR void *tester(FAR void *arg) { FAR void **val = arg; struct work_s work; int i; + int ret; memset(&work, 0, sizeof(work)); for (i = 0; i < TEST_COUNT; i++) { if (val[1] != NULL) { - work_queue_wq(val[1], &work, empty_worker, NULL, 0); - work_cancel_wq(val[1], &work); + ret = work_queue_wq(val[1], &work, empty_worker, NULL, 0); + ASSERT(ret == OK); + ret = work_cancel_wq(val[1], &work); + ASSERT(ret == OK); } else { - work_queue((int)(uintptr_t)val[0], &work, empty_worker, NULL, 0); - work_cancel((int)(uintptr_t)val[0], &work); + ret = work_queue((int)(uintptr_t)val[0], &work, + empty_worker, NULL, 0); + ASSERT(ret == OK); + ret = work_cancel((int)(uintptr_t)val[0], &work); + ASSERT(ret == OK); } usleep((int)(uintptr_t)val[2]); @@ -111,141 +867,92 @@ static FAR void *verifier(FAR void *arg) sem_t sem; sem_t call_sem; int call_count; + int extra_count; int i; + int ret; struct work_s work[VERIFY_COUNT + 1]; - sem_init(&sem, 0, 0); - sem_init(&call_sem, 0, 0); + ASSERT(sem_init(&sem, 0, 0) == OK); + ASSERT(sem_init(&call_sem, 0, 0) == OK); memset(&work, 0, sizeof(work)); /* Queue sleep worker. */ if (val[1] != NULL) { - work_queue_wq(val[1], &work[0], sleep_worker, &sem, 0); + ret = work_queue_wq(val[1], &work[0], sleep_worker, &sem, 0); } else { - work_queue((int)(uintptr_t)val[0], &work[0], sleep_worker, &sem, 0); + ret = work_queue((int)(uintptr_t)val[0], &work[0], + sleep_worker, &sem, 0); } + ASSERT(ret == OK); + /* Queue count workers when qid is busy. */ for (i = 1; i <= VERIFY_COUNT; i++) { if (val[1] != NULL) { - work_queue_wq(val[1], &work[i], count_worker, &call_sem, 0); + ret = work_queue_wq(val[1], &work[i], count_worker, + &call_sem, 0); } else { - work_queue((int)(uintptr_t)val[0], &work[i], - count_worker, &call_sem, 0); + ret = work_queue((int)(uintptr_t)val[0], &work[i], + count_worker, &call_sem, 0); } + + ASSERT(ret == OK); } /* Wait for sleep worker to run. */ - sem_wait(&sem); + ASSERT(wait_sem(&sem) == OK); /* Wait for count workers to run. */ - do + for (call_count = 0; call_count < VERIFY_COUNT; call_count++) { - usleep(SLEEP_TIME); - sem_getvalue(&call_sem, &call_count); + ASSERT(wait_sem(&call_sem) == OK); } - while (call_count != VERIFY_COUNT); - sem_getvalue(&call_sem, &call_count); + usleep(SLEEP_TIME); + ASSERT(sem_getvalue(&call_sem, &extra_count) == OK); + ASSERT(extra_count == 0); + printf("wqueue_test: call = %d, expect = %d\n", call_count, VERIFY_COUNT); - for (i = 0; i < VERIFY_COUNT; i++) + for (i = 0; i <= VERIFY_COUNT; i++) { ASSERT(work[i].worker == NULL); } ASSERT(call_count == VERIFY_COUNT); + ASSERT(sem_destroy(&call_sem) == OK); + ASSERT(sem_destroy(&sem) == OK); return NULL; } static void run_once(int qid, FAR void *wq, int interval, int priority_test, int priority_verify) { - pthread_t thread; - pthread_attr_t attr; - struct sched_param sparam; - int status; FAR void *val[3]; - status = pthread_attr_init(&attr); - if (status != 0) - { - printf("wqueue_test: pthread_attr_init failed, status=%d\n", status); - } - - memset(&sparam, 0, sizeof(sparam)); - /* Tester: try race conditions. */ - sparam.sched_priority = priority_test; - status = pthread_attr_setschedparam(&attr, &sparam); - if (status != 0) - { - printf("wqueue_test: pthread_attr_setschedparam failed for tester, " - "status=%d\n", status); - } - val[0] = (FAR void *)(uintptr_t)qid; val[1] = wq; val[2] = (FAR void *)(uintptr_t)interval; - status = pthread_create(&thread, &attr, tester, val); - if (status != 0) - { - printf("wqueue_test: pthread_create failed for tester, " - "status=%d\n", status); - } - - status = pthread_join(thread, NULL); - if (status != 0) - { - printf("wqueue_test: pthread_join failed for tester, " - "status=%d\n", status); - } + run_test_thread(tester, val, priority_test, CONFIG_PTHREAD_STACK_DEFAULT); /* Verifier: make sure queue is still working properly. */ - sparam.sched_priority = priority_verify; - status = pthread_attr_setschedparam(&attr, &sparam); - if (status != 0) - { - printf("wqueue_test: pthread_attr_setschedparam failed for verifier, " - "status=%d\n", status); - } - - status = pthread_attr_setstacksize(&attr, - VERIFY_COUNT * sizeof(struct work_s) + CONFIG_PTHREAD_STACK_DEFAULT); - if (status != 0) - { - printf("wqueue_test: pthread_attr_setstacksize failed for verifier, " - "status=%d\n", status); - } - - val[0] = (FAR void *)(uintptr_t)qid; - val[1] = wq; - status = pthread_create(&thread, &attr, verifier, val); - if (status != 0) - { - printf("wqueue_test: pthread_create failed for verifier, " - "status=%d\n", status); - } - - status = pthread_join(thread, NULL); - if (status != 0) - { - printf("wqueue_test: pthread_join failed for verifier, " - "status=%d\n", status); - } + run_test_thread(verifier, val, priority_verify, + VERIFY_COUNT * sizeof(struct work_s) + + CONFIG_PTHREAD_STACK_DEFAULT); } void wqueue_priority_test(int qid, FAR void *wq, int prio) @@ -270,15 +977,166 @@ void wqueue_priority_test(int qid, FAR void *wq, int prio) } } -/**************************************************************************** - * Public Functions - ****************************************************************************/ +static void multiple_queue_test(void) +{ + FAR void *wqueue[MULTI_QUEUE_COUNT]; + struct work_s work[MULTI_QUEUE_COUNT][MULTI_WORK_PER_QUEUE]; + char name[MULTI_QUEUE_COUNT][16]; + sem_t finished; + int loop; + int q; + int w; + int ret; -void wqueue_test(void) + printf("wqueue_test: multiple custom queues\n"); + ASSERT(work_queue_create(NULL, CUSTOM_PRIORITY, NULL, + STACKSIZE, 1) == NULL); + ASSERT(work_queue_create("test", CUSTOM_PRIORITY, NULL, 0, 1) == NULL); + ASSERT(work_queue_create("test", CUSTOM_PRIORITY, NULL, + STACKSIZE, 0) == NULL); + ASSERT(work_queue_free(NULL) < 0); + ASSERT(sem_init(&finished, 0, 0) == OK); + + for (loop = 0; loop < MULTI_QUEUE_LOOPS; loop++) + { + memset(work, 0, sizeof(work)); + + for (q = 0; q < MULTI_QUEUE_COUNT; q++) + { + snprintf(name[q], sizeof(name[q]), "ostest-wq%d", q); + wqueue[q] = work_queue_create(name[q], CUSTOM_PRIORITY + q, + NULL, STACKSIZE, q + 1); + ASSERT(wqueue[q] != NULL); + ASSERT(work_queue_priority_wq(wqueue[q]) == + CUSTOM_PRIORITY + q); + } + + for (q = 0; q < MULTI_QUEUE_COUNT; q++) + { + for (w = 0; w < MULTI_WORK_PER_QUEUE; w++) + { + ret = work_queue_wq(wqueue[q], &work[q][w], count_worker, + &finished, (w & 1) != 0 ? 1 : 0); + ASSERT(ret == OK); + } + } + + for (q = 0; q < MULTI_QUEUE_COUNT * MULTI_WORK_PER_QUEUE; q++) + { + ASSERT(wait_sem(&finished) == OK); + } + + for (q = 0; q < MULTI_QUEUE_COUNT; q++) + { + for (w = 0; w < MULTI_WORK_PER_QUEUE; w++) + { + ASSERT(work_available(&work[q][w])); + } + } + + for (q = MULTI_QUEUE_COUNT - 1; q >= 0; q--) + { + ASSERT(work_queue_free(wqueue[q]) == OK); + } + + printf("wqueue_test: multiple queues loop %d/%d done\n", + loop + 1, MULTI_QUEUE_LOOPS); + } + + ASSERT(sem_destroy(&finished) == OK); + printf("wqueue_test: multiple custom queues done\n"); +} + +static void teardown_test(void) +{ + FAR void *wqueue; + struct requeue_s requeue; + struct sync_cancel_s sync; + struct work_s work; + sem_t called; + int count; + int ret; + + printf("wqueue_test: teardown\n"); + + /* Free a queue while delayed work is still pending. */ + + memset(&work, 0, sizeof(work)); + ASSERT(sem_init(&called, 0, 0) == OK); + wqueue = work_queue_create("test", CUSTOM_PRIORITY, NULL, + STACKSIZE, 1); + ASSERT(wqueue != NULL); + ret = work_queue_wq(wqueue, &work, count_worker, &called, + MSEC2TICK(500)); + ASSERT(ret == OK); + ASSERT(work_queue_free(wqueue) == OK); + ASSERT(work_available(&work)); + usleep(SLEEP_TIME); + ASSERT(sem_getvalue(&called, &count) == OK); + printf("wqueue_test: pending callback = %d, expect = 0\n", count); + ASSERT(count == 0); + ASSERT(sem_destroy(&called) == OK); + + /* Free a queue while a callback is running. */ + + memset(&work, 0, sizeof(work)); + ASSERT(sem_init(&sync.started, 0, 0) == OK); + ASSERT(sem_init(&sync.finished, 0, 0) == OK); + wqueue = work_queue_create("test", CUSTOM_PRIORITY, NULL, + STACKSIZE, 1); + ASSERT(wqueue != NULL); + ret = work_queue_wq(wqueue, &work, sync_worker, &sync, 0); + ASSERT(ret == OK); + ASSERT(wait_sem(&sync.started) == OK); + ASSERT(work_queue_free(wqueue) == OK); + ASSERT(work_available(&work)); + ASSERT(sem_getvalue(&sync.finished, &count) == OK); + printf("wqueue_test: running callback = %d, expect = 1\n", count); + ASSERT(count == 1); + ASSERT(sem_destroy(&sync.finished) == OK); + ASSERT(sem_destroy(&sync.started) == OK); + + /* Reject attempts by a running callback to requeue work after teardown + * starts. + */ + + memset(&work, 0, sizeof(work)); + ASSERT(sem_init(&requeue.started, 0, 0) == OK); + wqueue = work_queue_create("test", CUSTOM_PRIORITY, NULL, + STACKSIZE, 1); + ASSERT(wqueue != NULL); + requeue.wqueue = wqueue; + requeue.work = &work; + requeue.next_result = OK; + requeue.queue_result = OK; + ret = work_queue_wq(wqueue, &work, requeue_worker, &requeue, 0); + ASSERT(ret == OK); + ASSERT(wait_sem(&requeue.started) == OK); + ASSERT(work_queue_free(wqueue) == OK); + ASSERT(requeue.next_result == -ESHUTDOWN); + ASSERT(requeue.queue_result == -ESHUTDOWN); + ASSERT(work_available(&work)); + printf("wqueue_test: teardown requeue rejected\n"); + ASSERT(sem_destroy(&requeue.started) == OK); + + printf("wqueue_test: teardown done\n"); +} + +static FAR void *wqueue_test_entry(FAR void *arg) { FAR void *wq; + int priority; int i; + UNUSED(arg); + +#ifdef CONFIG_BUILD_FLAT + printf("wqueue_test: backend = flat\n"); +#else + printf("wqueue_test: backend = libc user\n"); +#endif + +#ifdef CONFIG_BUILD_FLAT #ifdef CONFIG_SCHED_HPWORK printf("wqueue_test: HPWORK\n"); wqueue_priority_test(HPWORK, NULL, CONFIG_SCHED_HPWORKPRIORITY); @@ -288,18 +1146,66 @@ void wqueue_test(void) #ifdef CONFIG_SCHED_LPWORK printf("wqueue_test: LPWORK\n"); wqueue_priority_test(LPWORK, NULL, CONFIG_SCHED_LPWORKPRIORITY); - printf("wqueue_test: HPWORK done\n"); + printf("wqueue_test: LPWORK done\n"); +#endif #endif for (i = 1; i < 3; i++) { - printf("wqueue_test: test %d\n", i); - wq = work_queue_create("test", 100, NULL, STACKSIZE, i); - DEBUGASSERT(wq != NULL); - wqueue_priority_test(0, wq, 100); - work_queue_free(wq); - printf("wqueue_test: test %d done\n", i); + printf("wqueue_test: custom queue, threads = %d\n", i); + wq = work_queue_create("test", CUSTOM_PRIORITY, NULL, + STACKSIZE, i); + ASSERT(wq != NULL); + + priority = work_queue_priority_wq(wq); + printf("wqueue_test: priority = %d, expect = %d\n", + priority, CUSTOM_PRIORITY); + ASSERT(priority == CUSTOM_PRIORITY); + + if (i == 1) + { + api_validation_test(wq); + periodic_test(wq); + pending_replace_test(wq); + } + + sync_cancel_test(wq); + + if (i == 2) + { + parallel_cancel_test(wq); + } + + wqueue_priority_test(0, wq, CUSTOM_PRIORITY); + ASSERT(work_queue_free(wq) == OK); + printf("wqueue_test: custom queue, threads = %d done\n", i); } + + multiple_queue_test(); + self_free_test(); + teardown_test(); + + return NULL; } -#endif /* CONFIG_SCHED_WORKQUEUE */ +static void custom_wqueue_test(void) +{ + run_test_thread(wqueue_test_entry, NULL, CUSTOM_PRIORITY, STACKSIZE); +} + +#endif /* CONFIG_DISABLE_PTHREAD */ + +/**************************************************************************** + * Public Functions + ****************************************************************************/ + +void wqueue_test(void) +{ +#ifdef CONFIG_LIBC_USRWORK + usrwork_test(); +#endif + +#ifndef CONFIG_DISABLE_PTHREAD + custom_wqueue_test(); +#endif +}