This is an automated email from the ASF dual-hosted git repository. xiaoxiang781216 pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/nuttx.git
commit ae5997eef7c731979c6012523956b7868e936b1d Author: DuoYuWang <[email protected]> AuthorDate: Thu Aug 27 20:52:26 2026 +0800 libc/wqueue: support custom user work queues Implement the handle-based create, queue, priority, cancellation, and teardown APIs for CONFIG_LIBC_USRWORK. Custom queues use configurable pthread worker pools while the predefined USRWORK queue remains available. Match scheduler-backend delay, replacement, cancellation, and lifecycle semantics. Restrict the libc backend to task context because it uses blocking synchronization. Tested on an STM32H7 PX4 FMUv6C with ostest wqueue in Protected user space. Assisted-by: Codex:GPT-5 Signed-off-by: DuoYuWang <[email protected]> --- include/nuttx/wqueue.h | 89 ++++-- libs/libc/wqueue/Kconfig | 7 +- libs/libc/wqueue/work_cancel.c | 111 ++++++-- libs/libc/wqueue/work_queue.c | 120 +++++++-- libs/libc/wqueue/work_usrthread.c | 553 ++++++++++++++++++++++++++++---------- libs/libc/wqueue/wqueue.h | 76 +++++- 6 files changed, 723 insertions(+), 233 deletions(-) diff --git a/include/nuttx/wqueue.h b/include/nuttx/wqueue.h index 0bda2d0acbe..97becc82d79 100644 --- a/include/nuttx/wqueue.h +++ b/include/nuttx/wqueue.h @@ -74,11 +74,15 @@ * priority worker thread. Default: 2048. * * The user-mode work queue is only available in the protected or kernel - * builds. This those configurations, the user-mode work queue provides the - * same (non-standard) facility for use by applications. + * builds. In those configurations, the user-mode work queue provides the + * same (non-standard) facility for use by applications. User-mode work + * queue APIs use blocking synchronization and must only be called from task + * context. They must not be called from an interrupt handler. * * CONFIG_LIBC_USRWORK. If CONFIG_LIBC_USRWORK is also defined then the - * user-mode work queue will be created. + * user-mode work queue will be created. Dynamically allocated user-mode + * work queues require pthread support. The predefined protected-build + * USRWORK queue does not require pthread support. * CONFIG_LIBC_USRWORKPRIORITY - The minimum execution priority of the lower * priority worker thread. Default: 100 * CONFIG_LIBC_USRWORKSTACKSIZE - The stack size allocated for the lower @@ -332,18 +336,18 @@ int work_usrstart(void); * Name: work_queue_create * * Description: - * Create a new work queue. The work queue is identified by its work - * queue ID, which is used to queue works to the work queue and to - * perform other operations on the work queue. - * This function will create a work thread pool with nthreads threads. - * The work queue ID is returned on success. + * Create a custom work queue and return its handle. The handle is used + * to queue and cancel work, query the worker priority, and destroy the + * queue. This function creates a pool containing nthreads workers. + * This function must only be called from task context. + * User-mode custom queues require pthread support. * * Input Parameters: * name - Name of the new task * priority - Priority of the new task * stack_addr - Stack buffer of the new task * stack_size - size (in bytes) of the stack needed - * nthreads - Number of work thread should be created + * nthreads - Number of worker threads to create * * Returned Value: * The work queue handle returned on success. Otherwise, NULL @@ -359,9 +363,12 @@ FAR struct kwork_wqueue_s *work_queue_create(FAR const char *name, * Name: work_queue_free * * Description: - * Destroy a work queue. The work queue is identified by its work queue ID. - * All worker threads will be destroyed and the work queue will be freed. - * The work queue ID is invalid after this function returns. + * Destroy a custom work queue. All worker threads are stopped and the + * queue is freed. The handle is invalid after this function returns. + * Only a custom queue returned by work_queue_create() may be destroyed; + * the predefined HPWORK, LPWORK, and USRWORK queues cannot be destroyed. + * This function must only be called from task context and must not be + * called by one of the queue's own worker threads. * * Input Parameters: * wqueue - The work queue handle @@ -369,6 +376,9 @@ FAR struct kwork_wqueue_s *work_queue_create(FAR const char *name, * Returned Value: * Zero on success, a negated errno value on failure. * + * -EDEADLK - Called by one of the queue's own worker threads. + * -EINVAL - The handle is NULL or does not identify a custom queue. + * ****************************************************************************/ int work_queue_free(FAR struct kwork_wqueue_s *wqueue); @@ -384,8 +394,14 @@ int work_queue_free(FAR struct kwork_wqueue_s *wqueue); * the caller. Otherwise, the work structure is completely managed by the * work queue logic. The caller should never modify the contents of the * work queue structure directly. If work_queue() is called before the - * previous work has been performed and removed from the queue, then any - * pending work will be canceled and lost. + * previous work has been performed and removed from the same queue, then + * any pending work will be canceled and replaced. A queued work structure + * must be cancelled before it is moved to a different work queue. + * + * work_queue_wq() may be called from interrupt context for a kernel-mode + * or flat-build queue. A user-mode custom queue uses blocking + * synchronization, so work_queue_wq() must only be called from task + * context in user space. * * Input Parameters: * qid - The work queue ID (must be HPWORK or LPWORK) @@ -399,7 +415,10 @@ int work_queue_free(FAR struct kwork_wqueue_s *wqueue); * is invoked. Zero means to perform the work immediately. * * Returned Value: - * Zero on success, a negated errno on failure + * Zero on success, a negated errno on failure. + * + * -EINVAL - An argument or delay is invalid. + * -ESHUTDOWN - The custom work queue is being destroyed. * ****************************************************************************/ @@ -420,6 +439,10 @@ int work_queue_wq(FAR struct kwork_wqueue_s *wqueue, * Note that calling this function outside the work callback requires * the work->qtime being set. * + * A user-mode custom queue uses blocking synchronization, so + * work_queue_next_wq() must only be called from task context in user + * space. + * * Input Parameters: * qid - The work queue ID (must be HPWORK or LPWORK) * wqueue - The work queue handle @@ -432,7 +455,10 @@ int work_queue_wq(FAR struct kwork_wqueue_s *wqueue, * is invoked. Zero means to perform the work immediately. * * Returned Value: - * Zero on success, a negated errno on failure + * Zero on success, a negated errno on failure. + * + * -EINVAL - An argument or delay is invalid. + * -ESHUTDOWN - The custom work queue is being destroyed. * ****************************************************************************/ @@ -443,7 +469,7 @@ int work_queue_next_wq(FAR struct kwork_wqueue_s *wqueue, FAR void *arg, clock_t delay); /**************************************************************************** - * Name: work_queue_pri + * Name: work_queue_priority/work_queue_priority_wq * * Description: Get priority of the wqueue. We believe that all worker * threads have the same priority. @@ -466,7 +492,12 @@ int work_queue_priority_wq(FAR struct kwork_wqueue_s *wqueue); * Description: * Cancel previously queued work. This removes work from the work queue. * After work has been cancelled, it may be requeued by calling - * work_queue() again. + * work_queue() again. Cancelling work that is not queued is a successful + * no-op. + * + * work_cancel_wq() may be called from interrupt context for a kernel-mode + * or flat-build queue. It must only be called from task context for a + * user-mode custom queue. * * Input Parameters: * qid - The work queue ID (must be HPWORK or LPWORK) @@ -476,8 +507,7 @@ int work_queue_priority_wq(FAR struct kwork_wqueue_s *wqueue); * Returned Value: * Zero on success, a negated errno on failure * - * -ENOENT - There is no such work queued. - * -EINVAL - An invalid work queue was specified + * -EINVAL - An invalid work queue was specified. * ****************************************************************************/ @@ -489,9 +519,11 @@ int work_cancel_wq(FAR struct kwork_wqueue_s *wqueue, * Name: work_cancel_sync/work_cancel_sync_wq * * Description: - * Blocked cancel previously queued user-mode work. This removes work - * from the user mode work queue. After work has been cancelled, it may - * be requeued by calling work_queue() again. + * Synchronously cancel previously queued work. This removes work from + * the queue and waits for callbacks that are already running. After work + * has been cancelled, it may be requeued by calling work_queue() again. + * Cancelling work that is not queued is a successful no-op. + * This function must only be called from task context. * * Input Parameters: * qid - The work queue ID (must be HPWORK or LPWORK) @@ -499,13 +531,12 @@ int work_cancel_wq(FAR struct kwork_wqueue_s *wqueue, * work - The previously queued work structure to cancel * * Returned Value: - * Zero means the work was successfully cancelled. - * One means the work was not cancelled because it is currently being - * processed by work thread, but wait for it to finish. - * A negated errno value is returned on any failure: + * Zero means that queued work was cancelled and all callbacks using the + * work structure have finished, except for a callback running in the + * caller's own worker thread. A negated errno value is returned on any + * failure: * - * -ENOENT - There is no such work queued. - * -EINVAL - An invalid work queue was specified + * -EINVAL - An invalid work queue was specified. * ****************************************************************************/ diff --git a/libs/libc/wqueue/Kconfig b/libs/libc/wqueue/Kconfig index 0dd9cf425f9..72f7fb7af8f 100644 --- a/libs/libc/wqueue/Kconfig +++ b/libs/libc/wqueue/Kconfig @@ -9,9 +9,12 @@ menu "User Work Queue Support" config LIBC_USRWORK bool "User mode worker thread" default n + depends on BUILD_PROTECTED || !DISABLE_PTHREAD ---help--- - User space work queues can also be made available for deferred - processing in the NuttX kernel build. + User-space work queues provide deferred processing in protected and + kernel builds. Dynamically allocated user-mode work queues require + pthread support. The predefined protected-build USRWORK queue does + not require pthread support. if LIBC_USRWORK diff --git a/libs/libc/wqueue/work_cancel.c b/libs/libc/wqueue/work_cancel.c index 7d4188e974f..cd23c783204 100644 --- a/libs/libc/wqueue/work_cancel.c +++ b/libs/libc/wqueue/work_cancel.c @@ -26,6 +26,7 @@ #include <nuttx/config.h> +#include <unistd.h> #include <assert.h> #include <errno.h> @@ -51,59 +52,85 @@ * * Input Parameters: * wqueue - The work queue - * work - The previously queue work structure to cancel + * work - The previously queued work structure to cancel * * Returned Value: * Zero (OK) on success, a negated errno on failure. This error may be * reported: * - * -ENOENT - There is no such work queued. * -EINVAL - An invalid work queue was specified * ****************************************************************************/ -static int work_qcancel(FAR struct usr_wqueue_s *wqueue, +static int work_qcancel(FAR struct usr_wqueue_s *wqueue, bool sync, FAR struct work_s *work) { - int ret = -ENOENT; - int semcount; - - DEBUGASSERT(work != NULL); - - /* Get exclusive access to the work queue */ + pid_t self = sync ? gettid() : 0; + int ret; - while (nxmutex_lock(&wqueue->lock) < 0); + if (wqueue == NULL || work == NULL) + { + return -EINVAL; + } - /* Cancelling the work is simply a matter of removing the work structure - * from the work queue. This must be done with interrupts disabled because - * new work is typically added to the work queue from interrupt handlers. + /* A work structure becomes available for requeue after it is dequeued, + * before its callback returns. Multiple workers can therefore execute + * callbacks using the same work structure. Find one such worker and + * repeat after it finishes until no callback remains. Exclude the + * calling worker to avoid self-deadlock. */ - if (work->worker != NULL) + for (; ; ) { - bool is_head = list_is_head(&wqueue->q, &work->node); + FAR sem_t *sync_wait = NULL; + int wndx; + + /* Get exclusive access to the work queue */ - /* Now, remove the work from the work queue */ + do + { + ret = nxmutex_lock(&wqueue->lock); + } + while (ret < 0); - list_delete(&work->node); + /* Remove a pending instance from the queue. */ - if (is_head) + if (work->worker != NULL) { - /* Remove the work at the head of the queue */ + if (work_remove(wqueue, work)) + { + work_wake(wqueue); + } + } - nxsem_get_value(&wqueue->wake, &semcount); - if (semcount < 1) + if (sync) + { + for (wndx = 0; wndx < wqueue->nthreads; wndx++) { - nxsem_post(&wqueue->wake); + FAR struct usr_worker_s *worker = &wqueue->worker[wndx]; + + if (worker->work == work && self != worker->tid) + { + worker->wait_count++; + sync_wait = &worker->wait; + break; + } } } - work->worker = NULL; - ret = OK; - } + nxmutex_unlock(&wqueue->lock); - nxmutex_unlock(&wqueue->lock); - return ret; + if (sync_wait == NULL) + { + return OK; + } + + do + { + ret = nxsem_wait(sync_wait); + } + while (ret == -EINTR); + } } /**************************************************************************** @@ -126,7 +153,7 @@ static int work_qcancel(FAR struct usr_wqueue_s *wqueue, * Zero (OK) on success, a negated errno on failure. This error may be * reported: * - * -ENOENT - There is no such work queued. + * -EINVAL - An invalid work queue was specified * ****************************************************************************/ @@ -134,7 +161,7 @@ int work_cancel(int qid, FAR struct work_s *work) { if (qid == USRWORK) { - return work_qcancel(&g_usrwork, work); + return work_qcancel(&g_usrwork, false, work); } else { @@ -142,4 +169,30 @@ int work_cancel(int qid, FAR struct work_s *work) } } +#ifndef CONFIG_DISABLE_PTHREAD +int work_cancel_wq(FAR struct kwork_wqueue_s *handle, + FAR struct work_s *work) +{ + return work_qcancel((FAR struct usr_wqueue_s *)handle, false, work); +} +#endif + +int work_cancel_sync(int qid, FAR struct work_s *work) +{ + if (qid == USRWORK) + { + return work_qcancel(&g_usrwork, true, work); + } + + return -EINVAL; +} + +#ifndef CONFIG_DISABLE_PTHREAD +int work_cancel_sync_wq(FAR struct kwork_wqueue_s *handle, + FAR struct work_s *work) +{ + return work_qcancel((FAR struct usr_wqueue_s *)handle, true, work); +} +#endif + #endif /* CONFIG_LIBC_USRWORK && !__KERNEL__ */ diff --git a/libs/libc/wqueue/work_queue.c b/libs/libc/wqueue/work_queue.c index 104e10ff403..bc10986ead8 100644 --- a/libs/libc/wqueue/work_queue.c +++ b/libs/libc/wqueue/work_queue.c @@ -52,10 +52,8 @@ * * The work structure is allocated by caller, but completely managed by * the work queue logic. The caller should never modify the contents of - * the work queue structure; the caller should not call work_qqueue() - * again until either (1) the previous work has been performed and removed - * from the queue, or (2) work_cancel() has been called to cancel the work - * and remove it from the work queue. + * the work queue structure. Calling work_qqueue() while the work is + * pending on the same queue cancels and replaces the pending instance. * * Input Parameters: * wqueue - The work queue @@ -63,7 +61,7 @@ * worker - The worker callback to be invoked. The callback will be * invoked on the worker thread of execution. * arg - The argument that will be passed to the worker callback when - * int is invoked. + * it is invoked. * delay - Delay (in clock ticks) from the time queue until the worker * is invoked. Zero means to perform the work immediately. * @@ -74,21 +72,57 @@ static int work_qqueue(FAR struct usr_wqueue_s *wqueue, FAR struct work_s *work, worker_t worker, - FAR void *arg, clock_t delay) + FAR void *arg, clock_t delay, bool period) { FAR struct work_s *curr; FAR struct work_s *head; - int semcount; + bool wake = false; + int ret; + + if (wqueue == NULL || work == NULL || worker == NULL || + delay < 0 || delay > WDOG_MAX_DELAY) + { + return -EINVAL; + } /* Get exclusive access to the work queue */ - while (nxmutex_lock(&wqueue->lock) < 0); + do + { + ret = nxmutex_lock(&wqueue->lock); + } + while (ret < 0); + + if (wqueue->exit) + { + nxmutex_unlock(&wqueue->lock); + return -ESHUTDOWN; + } + + /* Remove a previous pending instance before requeueing it. */ + + if (work->worker != NULL) + { + wake = work_remove(wqueue, work); + } /* Initialize the work structure */ - work->worker = worker; /* Work callback. non-NULL means queued */ - work->arg = arg; /* Callback argument */ - work->qtime = clock() + delay; /* Delay until work performed */ + work->worker = worker; /* Work callback. non-NULL means queued */ + work->arg = arg; /* Callback argument */ + + if (period) + { + work->qtime += delay; + } + else if (delay > 0) + { + work->qtime = clock() + delay + 1; + } + else + { + work->qtime = clock(); + } /* Insert the work into the wait queue sorted by the expired time. */ @@ -111,17 +145,13 @@ static int work_qqueue(FAR struct usr_wqueue_s *wqueue, list_add_before(&curr->node, &work->node); - /* If the current work is the head of the wait queue. - * We should wake up the worker thread. + /* Wake if this work becomes the new head. Immediate work may be queued + * behind other ready work, so wake another worker in the pool as well. */ - if (curr == head) + if (wake || delay == 0 || curr == head) { - nxsem_get_value(&wqueue->wake, &semcount); - if (semcount < 1) - { - nxsem_post(&wqueue->wake); - } + work_wake(wqueue); } nxmutex_unlock(&wqueue->lock); @@ -152,7 +182,7 @@ static int work_qqueue(FAR struct usr_wqueue_s *wqueue, * worker - The worker callback to be invoked. The callback will be * invoked on the worker thread of execution. * arg - The argument that will be passed to the worker callback when - * int is invoked. + * it is invoked. * delay - Delay (in clock ticks) from the time queue until the worker * is invoked. Zero means to perform the work immediately. * @@ -166,11 +196,7 @@ int work_queue(int qid, FAR struct work_s *work, worker_t worker, { if (qid == USRWORK) { - /* Is there already pending work? */ - - work_cancel(qid, work); - - return work_qqueue(&g_usrwork, work, worker, arg, delay); + return work_qqueue(&g_usrwork, work, worker, arg, delay, false); } else { @@ -178,4 +204,48 @@ int work_queue(int qid, FAR struct work_s *work, worker_t worker, } } +/**************************************************************************** + * Name: work_queue_wq + * + * Description: + * Queue work on a user-mode custom work queue. This function must only + * be called from task context. + * + ****************************************************************************/ + +#ifndef CONFIG_DISABLE_PTHREAD +int work_queue_wq(FAR struct kwork_wqueue_s *handle, + FAR struct work_s *work, worker_t worker, + FAR void *arg, clock_t delay) +{ + return work_qqueue((FAR struct usr_wqueue_s *)handle, work, + worker, arg, delay, false); +} +#endif + +/**************************************************************************** + * Name: work_queue_next/work_queue_next_wq + ****************************************************************************/ + +int work_queue_next(int qid, FAR struct work_s *work, worker_t worker, + FAR void *arg, clock_t delay) +{ + if (qid == USRWORK) + { + return work_qqueue(&g_usrwork, work, worker, arg, delay, true); + } + + return -EINVAL; +} + +#ifndef CONFIG_DISABLE_PTHREAD +int work_queue_next_wq(FAR struct kwork_wqueue_s *handle, + FAR struct work_s *work, worker_t worker, + FAR void *arg, clock_t delay) +{ + return work_qqueue((FAR struct usr_wqueue_s *)handle, work, + worker, arg, delay, true); +} +#endif + #endif /* CONFIG_LIBC_USRWORK && !__KERNEL__ */ diff --git a/libs/libc/wqueue/work_usrthread.c b/libs/libc/wqueue/work_usrthread.c index f4a3663ab10..f74632c2e5d 100644 --- a/libs/libc/wqueue/work_usrthread.c +++ b/libs/libc/wqueue/work_usrthread.c @@ -30,6 +30,7 @@ #include <unistd.h> #include <pthread.h> #include <sched.h> +#include <stdlib.h> #include <errno.h> #include <assert.h> @@ -57,11 +58,24 @@ /* The state of the user mode work queue. */ +static struct usr_worker_s g_usrworker = +{ + 0, + NULL, + &g_usrwork, + SEM_INITIALIZER(0), + 0, +}; + struct usr_wqueue_s g_usrwork = { LIST_INITIAL_VALUE(g_usrwork.q), NXMUTEX_INITIALIZER, SEM_INITIALIZER(0), + &g_usrworker, + 1, + false, + false, }; /**************************************************************************** @@ -69,205 +83,475 @@ struct usr_wqueue_s g_usrwork = ****************************************************************************/ /**************************************************************************** - * Name: work_process + * Name: work_pthread * * Description: - * This is the logic that performs actions placed on any work list. This - * logic is the common underlying logic to all work queues. This logic is - * part of the internal implementation of each work queue; it should not - * be called from application level logic. + * This is the worker thread that performs the actions placed on the user + * work queue. + * + * This is a user-mode work queue. The predefined queue is started by + * application start-up logic through work_usrstart(); custom queues are + * started by work_queue_create(). * * Input Parameters: - * wqueue - Describes the work queue to be processed + * arg - Describes this worker and its parent queue * * Returned Value: - * None + * NULL * ****************************************************************************/ -static void work_process(FAR struct usr_wqueue_s *wqueue) +static pthread_addr_t work_pthread(pthread_addr_t arg) { + FAR struct usr_worker_s *worker = + (FAR struct usr_worker_s *)arg; + FAR struct usr_wqueue_s *wqueue = worker->wqueue; FAR struct work_s *work; - worker_t worker; - FAR void *arg; + worker_t callback; + FAR void *work_arg; clock_t tick; clock_t next; int ret; - /* Then process queued work. Lock the work queue while we process items - * in the work list. - */ - - next = WORK_DELAY_MAX; - ret = nxmutex_lock(&wqueue->lock); - if (ret < 0) + for (; ; ) { - /* Break out earlier if we were awakened by a signal */ + /* Then process queued work. Lock the work queue while we process + * items in the work list. + */ - return; - } + next = WORK_DELAY_MAX; + ret = nxmutex_lock(&wqueue->lock); + if (ret < 0) + { + /* Restart if we were awakened by a signal. */ - /* And check each entry in the work queue. Since we have locked the - * work queue we know: (1) we will not be suspended unless we do - * so ourselves, and (2) there will be no changes to the work queue - */ + continue; + } - while (!list_is_empty(&wqueue->q)) - { - work = list_first_entry(&wqueue->q, struct work_s, node); + if (wqueue->exit) + { + nxmutex_unlock(&wqueue->lock); + break; + } - /* Is this work ready? It is ready if there is no delay or if - * the delay has elapsed. is the time that the work was added - * to the work queue. Therefore a delay of equal or less than - * zero will always execute immediately. + /* And check each entry in the work queue. Since we have locked the + * work queue we know: (1) we will not be suspended unless we do + * so ourselves, and (2) there will be no changes to the work queue */ - tick = clock(); - - /* Is this delay work ready? */ - - if (clock_compare(work->qtime, tick)) + while (!list_is_empty(&wqueue->q)) { - /* Remove the ready-to-execute work from the list */ + work = list_first_entry(&wqueue->q, struct work_s, node); - list_delete(&work->node); - - /* Extract the work description from the entry (in case the work - * instance by the reused after it has been de-queued). + /* Is this work ready? It is ready if there is no delay or if + * the delay has elapsed. is the time that the work was added + * to the work queue. Therefore a delay of equal or less than + * zero will always execute immediately. */ - worker = work->worker; + tick = clock(); - /* Check for a race condition where the work may be nullified - * before it is removed from the queue. - */ + /* Is this delay work ready? */ - if (worker != NULL) + if (clock_compare(work->qtime, tick)) { - /* Extract the work argument before unlocking the work queue */ - - arg = work->arg; + /* Remove the ready-to-execute work from the list */ - /* Mark the work as no longer being queued */ + list_delete(&work->node); - work->worker = NULL; - - /* Do the work. Unlock the work queue while the work is being - * performed... we don't have any idea how long this will take! + /* Extract the work description from the entry (in case the + * work instance may be reused after it has been de-queued). */ - nxmutex_unlock(&wqueue->lock); - worker(arg); + callback = work->worker; - /* Now, unfortunately, since we unlocked the work queue we - * don't know the state of the work list and we will have to - * start back at the head of the list. + /* Check for a race condition where the work may be nullified + * before it is removed from the queue. */ - ret = nxmutex_lock(&wqueue->lock); - if (ret < 0) + if (callback != NULL) { - /* Break out earlier if we were awakened by a signal */ + /* Extract the work argument before unlocking the queue. */ + + work_arg = work->arg; + + /* Mark the work as no longer being queued */ - return; + work->worker = NULL; + worker->work = work; + + /* Let another worker process the next ready entry. */ + + if (!list_is_empty(&wqueue->q)) + { + FAR struct work_s *next_work = + list_first_entry(&wqueue->q, struct work_s, node); + + if (clock_compare(next_work->qtime, tick)) + { + work_wake(wqueue); + } + } + + /* Do the work. Unlock the work queue while the work is + * being performed... we don't have any idea how long + * this will take! + */ + + nxmutex_unlock(&wqueue->lock); + callback(work_arg); + + /* Now, unfortunately, since we unlocked the work queue + * we don't know the state of the work list and we will + * have to start back at the head of the list. + */ + + do + { + ret = nxmutex_lock(&wqueue->lock); + } + while (ret < 0); + + worker->work = NULL; + + while (worker->wait_count > 0) + { + worker->wait_count--; + nxsem_post(&worker->wait); + } + + if (wqueue->exit) + { + nxmutex_unlock(&wqueue->lock); + return NULL; + } } } + else + { + next = work->qtime - tick; + break; + } + } + + /* Unlock the work queue before waiting. */ + + nxmutex_unlock(&wqueue->lock); + + if (next == WORK_DELAY_MAX) + { + /* Wait indefinitely until work_queue has new items */ + + nxsem_wait(&wqueue->wake); } else { - next = work->qtime - clock(); - break; + struct timespec now; + struct timespec delay; + struct timespec rqtp; + + /* Wait awhile to check the work list. We will wait here until + * either the time elapses or until we are awakened by a semaphore. + * Interrupts will be re-enabled while we wait. + */ + + clock_gettime(CLOCK_REALTIME, &now); + clock_ticks2time(&delay, next); + clock_timespec_add(&now, &delay, &rqtp); + + nxsem_timedwait(&wqueue->wake, &rqtp); } } - /* Unlock the work queue before waiting. */ + return NULL; +} - nxmutex_unlock(&wqueue->lock); +#ifdef CONFIG_BUILD_PROTECTED +static int work_usrtask(int argc, char *argv[]) +{ + work_pthread(&g_usrworker); + return OK; +} +#endif + +#ifndef CONFIG_DISABLE_PTHREAD +/**************************************************************************** + * Name: work_thread_create + * + * Description: + * Create the worker threads for a dynamically allocated user work queue. + * + ****************************************************************************/ + +static int work_thread_create(FAR const char *name, int priority, + FAR void *stack_addr, int stack_size, + FAR struct usr_wqueue_s *wqueue) +{ + pthread_attr_t attr; + struct sched_param param; + int created = 0; + int lockret; + int ret; + int wndx; - if (next == WORK_DELAY_MAX) + ret = pthread_attr_init(&attr); + if (ret != 0) { - /* Wait indefinitely until work_queue has new items */ + return -ret; + } - nxsem_wait(&wqueue->wake); + ret = pthread_attr_setstacksize(&attr, stack_size); + if (ret != 0) + { + goto errout_with_attr; } - else + + ret = pthread_attr_setschedpolicy(&attr, SCHED_FIFO); + if (ret != 0) { - struct timespec now; - struct timespec delay; - struct timespec rqtp; + goto errout_with_attr; + } - /* Wait awhile to check the work list. We will wait here until - * either the time elapses or until we are awakened by a semaphore. - * Interrupts will be re-enabled while we wait. - */ + ret = pthread_attr_getschedparam(&attr, ¶m); + if (ret != 0) + { + goto errout_with_attr; + } - clock_gettime(CLOCK_REALTIME, &now); - clock_ticks2time(&delay, next); - clock_timespec_add(&now, &delay, &rqtp); + param.sched_priority = priority; + ret = pthread_attr_setschedparam(&attr, ¶m); + if (ret != 0) + { + goto errout_with_attr; + } - nxsem_timedwait(&wqueue->wake, &rqtp); + ret = pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED); + if (ret != 0) + { + goto errout_with_attr; } + + for (wndx = 0; wndx < wqueue->nthreads; wndx++) + { + FAR struct usr_worker_s *worker = &wqueue->worker[wndx]; + + if (stack_addr != NULL) + { + FAR void *stack = (FAR void *) + ((uintptr_t)stack_addr + wndx * stack_size); + + ret = pthread_attr_setstack(&attr, stack, stack_size); + if (ret != 0) + { + goto errout_with_threads; + } + } + + ret = pthread_create(&worker->tid, &attr, work_pthread, worker); + if (ret != 0) + { + goto errout_with_threads; + } + + created++; + pthread_setname_np(worker->tid, name); + } + + pthread_attr_destroy(&attr); + return OK; + +errout_with_threads: + do + { + lockret = nxmutex_lock(&wqueue->lock); + } + while (lockret < 0); + + wqueue->exit = true; + nxmutex_unlock(&wqueue->lock); + + for (wndx = 0; wndx < created; wndx++) + { + nxsem_post(&wqueue->wake); + } + + for (wndx = 0; wndx < created; wndx++) + { + pthread_join(wqueue->worker[wndx].tid, NULL); + } + +errout_with_attr: + pthread_attr_destroy(&attr); + return -ret; } /**************************************************************************** - * Name: work_usrthread + * Public Functions + ****************************************************************************/ + +/**************************************************************************** + * Name: work_queue_create * * Description: - * This is the worker thread that performs the actions placed on the user - * work queue. - * - * This is a user mode work queue. It must be used by applications for - * miscellaneous operations. The user work thread must be started by - * application start-up logic by calling work_usrstart(). + * Create a user-mode custom work queue. This function must only be + * called from task context. * - * Input Parameters: - * argc, argv (not used) + ****************************************************************************/ + +FAR struct kwork_wqueue_s *work_queue_create(FAR const char *name, + int priority, + FAR void *stack_addr, + int stack_size, int nthreads) +{ + FAR struct usr_wqueue_s *wqueue; + size_t allocsize; + int ret; + int wndx; + + if (name == NULL || stack_size <= 0 || nthreads < 1 || + nthreads > (SIZE_MAX - sizeof(*wqueue)) / sizeof(struct usr_worker_s)) + { + return NULL; + } + + allocsize = sizeof(*wqueue) + + nthreads * sizeof(struct usr_worker_s); + wqueue = calloc(1, allocsize); + if (wqueue == NULL) + { + return NULL; + } + + list_initialize(&wqueue->q); + nxmutex_init(&wqueue->lock); + nxsem_init(&wqueue->wake, 0, 0); + wqueue->worker = (FAR struct usr_worker_s *)(wqueue + 1); + wqueue->nthreads = nthreads; + wqueue->dynamic = true; + + for (wndx = 0; wndx < nthreads; wndx++) + { + wqueue->worker[wndx].wqueue = wqueue; + nxsem_init(&wqueue->worker[wndx].wait, 0, 0); + } + + ret = work_thread_create(name, priority, stack_addr, stack_size, wqueue); + if (ret < 0) + { + for (wndx = 0; wndx < nthreads; wndx++) + { + nxsem_destroy(&wqueue->worker[wndx].wait); + } + + nxsem_destroy(&wqueue->wake); + nxmutex_destroy(&wqueue->lock); + free(wqueue); + return NULL; + } + + return (FAR struct kwork_wqueue_s *)wqueue; +} + +/**************************************************************************** + * Name: work_queue_free * - * Returned Value: - * Does not return + * Description: + * Destroy a user-mode custom work queue. This function must only be + * called from task context and must not be called by one of the queue's + * own worker threads. * ****************************************************************************/ -#ifdef CONFIG_BUILD_PROTECTED -static int work_usrthread(int argc, char *argv[]) -#else -static pthread_addr_t work_usrthread(pthread_addr_t arg) -#endif +int work_queue_free(FAR struct kwork_wqueue_s *handle) { - /* Loop forever */ + FAR struct usr_wqueue_s *wqueue = (FAR struct usr_wqueue_s *)handle; + FAR struct work_s *work; + FAR struct work_s *next; + pid_t self = gettid(); + int ret; + int wndx; - for (; ; ) + if (wqueue == NULL || !wqueue->dynamic) { - /* Then process queued work. We need to keep the work queue locked - * while we process items in the work list. - */ + return -EINVAL; + } + + for (wndx = 0; wndx < wqueue->nthreads; wndx++) + { + if (self == wqueue->worker[wndx].tid) + { + return -EDEADLK; + } + } - work_process(&g_usrwork); + do + { + ret = nxmutex_lock(&wqueue->lock); } + while (ret < 0); -#ifdef CONFIG_BUILD_PROTECTED - return OK; /* To keep some compilers happy */ -#else - return NULL; /* To keep some compilers happy */ -#endif + wqueue->exit = true; + + list_for_every_entry_safe(&wqueue->q, work, next, struct work_s, node) + { + list_delete(&work->node); + work->worker = NULL; + } + + nxmutex_unlock(&wqueue->lock); + + for (wndx = 0; wndx < wqueue->nthreads; wndx++) + { + nxsem_post(&wqueue->wake); + } + + for (wndx = 0; wndx < wqueue->nthreads; wndx++) + { + pthread_join(wqueue->worker[wndx].tid, NULL); + nxsem_destroy(&wqueue->worker[wndx].wait); + } + + nxsem_destroy(&wqueue->wake); + nxmutex_destroy(&wqueue->lock); + free(wqueue); + return OK; } +#endif /**************************************************************************** - * Public Functions + * Name: work_queue_priority_wq ****************************************************************************/ +int work_queue_priority_wq(FAR struct kwork_wqueue_s *handle) +{ + FAR struct usr_wqueue_s *wqueue = (FAR struct usr_wqueue_s *)handle; + struct sched_param param; + int ret; + + if (wqueue == NULL || wqueue->nthreads < 1) + { + return -EINVAL; + } + + ret = sched_getparam(wqueue->worker[0].tid, ¶m); + return ret == OK ? param.sched_priority : -get_errno(); +} + +int work_queue_priority(int qid) +{ + if (qid != USRWORK) + { + return -EINVAL; + } + + return work_queue_priority_wq((FAR struct kwork_wqueue_s *)&g_usrwork); +} + /**************************************************************************** * Name: work_usrstart * * Description: - * Start the user mode work queue. - * - * Input Parameters: - * None - * - * Returned Value: - * The task ID of the worker thread is returned on success. A negated - * errno value is returned on failure. + * Start the predefined user work queue. * ****************************************************************************/ @@ -275,55 +559,42 @@ int work_usrstart(void) { int ret; #ifndef CONFIG_BUILD_PROTECTED - pthread_t usrwork; pthread_attr_t attr; struct sched_param param; #endif - /* Initialize the work queue */ - - list_initialize(&g_usrwork.q); - #ifdef CONFIG_BUILD_PROTECTED - - /* Start a user-mode worker thread for use by applications. */ - ret = task_create("uwork", CONFIG_LIBC_USRWORKPRIORITY, CONFIG_LIBC_USRWORKSTACKSIZE, - work_usrthread, NULL); + work_usrtask, NULL); if (ret < 0) { int errcode = get_errno(); + DEBUGASSERT(errcode > 0); return -errcode; } - return ret; + g_usrworker.tid = ret; #else - /* Start a user-mode worker thread for use by applications. */ - pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); pthread_attr_setstacksize(&attr, CONFIG_LIBC_USRWORKSTACKSIZE); - pthread_attr_getschedparam(&attr, ¶m); param.sched_priority = CONFIG_LIBC_USRWORKPRIORITY; pthread_attr_setschedparam(&attr, ¶m); - ret = pthread_create(&usrwork, &attr, work_usrthread, NULL); + ret = pthread_create(&g_usrworker.tid, &attr, work_pthread, + &g_usrworker); + pthread_attr_destroy(&attr); if (ret != 0) { return -ret; } - - /* Detach because the return value and completion status will not be - * requested. - */ - - pthread_detach(usrwork); - - return (pid_t)usrwork; #endif + + return g_usrworker.tid; } -#endif /* CONFIG_LIBC_USRWORK && !__KERNEL__*/ +#endif /* CONFIG_LIBC_USRWORK && !__KERNEL__ */ diff --git a/libs/libc/wqueue/wqueue.h b/libs/libc/wqueue/wqueue.h index eab017c6125..141c78e5bdd 100644 --- a/libs/libc/wqueue/wqueue.h +++ b/libs/libc/wqueue/wqueue.h @@ -29,7 +29,9 @@ #include <nuttx/config.h> -#include <pthread.h> +#include <stdbool.h> +#include <stdint.h> +#include <sys/types.h> #include <nuttx/mutex.h> #include <nuttx/semaphore.h> @@ -46,13 +48,32 @@ * Public Type Definitions ****************************************************************************/ -/* This structure defines the state of one user-modework queue. */ +/* Forward reference */ + +struct usr_wqueue_s; + +/* This structure describes one user-mode worker thread. */ + +struct usr_worker_s +{ + pid_t tid; /* Worker thread ID */ + FAR struct work_s *work; /* Work currently being processed */ + FAR struct usr_wqueue_s *wqueue; /* Parent work queue */ + sem_t wait; /* Wait for the current work */ + uint16_t wait_count; /* Number of synchronous waiters */ +}; + +/* This structure defines the state of one user-mode work queue. */ struct usr_wqueue_s { - struct list_node q; /* The queue of pending work */ - mutex_t lock; /* exclusive access to user-mode work queue */ - sem_t wake; /* The wake-up semaphore of the usrthread */ + struct list_node q; /* The queue of pending work */ + mutex_t lock; /* Exclusive access to the queue */ + sem_t wake; /* Wake-up semaphore */ + FAR struct usr_worker_s *worker; /* Worker thread state array */ + int nthreads; /* Number of worker threads */ + bool exit; /* Request worker thread exit */ + bool dynamic; /* Dynamically allocated queue */ }; /**************************************************************************** @@ -64,8 +85,49 @@ struct usr_wqueue_s extern struct usr_wqueue_s g_usrwork; /**************************************************************************** - * Public Function Prototypes + * Inline Functions ****************************************************************************/ -#endif /* CONFIG_LIBC_USRWORK && !__KERNEL__*/ +/**************************************************************************** + * Name: work_remove + * + * Description: + * Remove work from a user-mode work queue. The caller must hold + * wqueue->lock, and work must be queued on wqueue. + * + * Returned Value: + * true if removing work changed the head of the queue; otherwise false. + * + ****************************************************************************/ + +static inline_function bool +work_remove(FAR struct usr_wqueue_s *wqueue, + FAR struct work_s *work) +{ + FAR struct work_s *head; + + head = list_first_entry(&wqueue->q, struct work_s, node); + + work->worker = NULL; + list_delete(&work->node); + + return head == work; +} + +static inline_function void work_wake(FAR struct usr_wqueue_s *wqueue) +{ + int semcount; + + /* Keep enough wake tokens for the worker pool while bounding stale + * tokens when workers are already running. + */ + + nxsem_get_value(&wqueue->wake, &semcount); + if (semcount < wqueue->nthreads) + { + nxsem_post(&wqueue->wake); + } +} + +#endif /* CONFIG_LIBC_USRWORK && !__KERNEL__ */ #endif /* __LIBS_LIBC_WQUEUE_WQUEUE_H */
