diff --git a/Documentation/reference/os/wqueue.rst b/Documentation/reference/os/wqueue.rst index 020c94cb7a2aa..c5277a3d61f7b 100644 --- a/Documentation/reference/os/wqueue.rst +++ b/Documentation/reference/os/wqueue.rst @@ -149,10 +149,28 @@ and the user-mode work queue is functionally equivalent to the high priority work queue. It differs in that its implementation does not depend on internal, kernel-space facilities. +**Custom User Work Queues**. Applications can use +``work_queue_create()`` to create additional user-mode queues with a +configurable priority and worker pool. The returned handle is passed to +the ``*_wq()`` interfaces and to ``work_queue_free()``. The predefined +``USRWORK`` queue remains available through the queue-ID interfaces. Custom +user-mode queues require pthread support; the predefined ``USRWORK`` queue +does not require pthread support in a protected build. + +**Execution Context**. The user-mode implementation uses mutexes and +semaphores for synchronization. Its queue, cancel, create, priority, and +destroy interfaces must therefore only be called from task context and +must not be called from an interrupt handler. Kernel-mode and flat-build +``work_queue()``, ``work_queue_wq()``, ``work_cancel()``, and +``work_cancel_wq()`` remain safe for interrupt handlers. Creation, +destruction, and synchronous cancellation are task-context operations in +all build modes. + **Configuration Options**. - ``CONFIG_LIBC_USRWORK``. If CONFIG_LIBC_USRWORK is also defined - then the user-mode work queue will be enabled. + then the user-mode work queue will be enabled. Dynamically allocated + user-mode work queues require pthread support. - ``CONFIG_LIBC_USRWORKPRIORITY``. The execution priority of the user-mode priority worker thread. Default: 100 - ``CONFIG_LIBC_USRWORKSTACKSIZE``. The stack size allocated for @@ -202,7 +220,7 @@ Work Queue Interfaces --------------------- .. c:function:: int work_queue(int qid, FAR struct work_s *work, worker_t worker, \ - FAR void *arg, uint32_t delay) + FAR void *arg, clock_t delay) Queue work to be performed at a later time. All queued work will be performed on the worker thread of execution @@ -230,6 +248,56 @@ Work Queue Interfaces :return: Zero is returned on success; a negated errno is returned on failure. +.. c:function:: FAR struct kwork_wqueue_s *work_queue_create( \ + FAR const char *name, int priority, FAR void *stack_addr, \ + int stack_size, int nthreads) + + Create a custom work queue containing ``nthreads`` workers. All + workers use the requested name, priority, and stack size. If + ``stack_addr`` is ``NULL``, each worker stack is allocated by the + thread creation logic. Otherwise, ``stack_addr`` must identify storage + for ``nthreads * stack_size`` bytes. + + This interface must only be called from task context. + + :return: A work queue handle on success; ``NULL`` on failure. + +.. c:function:: int work_queue_free(FAR struct kwork_wqueue_s *wqueue) + + Destroy a custom queue, discard pending work, and wait for all running + callbacks and worker threads to finish. Pending work structures become + available for reuse before the function returns. The predefined + ``HPWORK``, ``LPWORK``, and ``USRWORK`` queues cannot be destroyed. + + This interface must only be called from task context and cannot be + called from one of the queue's own callbacks. + + :return: Zero on success, ``-EINVAL`` for an invalid or predefined + queue, or ``-EDEADLK`` when called by one of the queue's workers. + +.. c:function:: int work_queue_wq(FAR struct kwork_wqueue_s *wqueue, \ + FAR struct work_s *work, worker_t worker, FAR void *arg, \ + clock_t delay) + + Queue work on a custom queue. If the work structure is already pending + on the same queue, the pending instance is replaced. A work structure + must be cancelled before it is moved to another queue. + + :return: Zero on success, ``-EINVAL`` for invalid arguments, or + ``-ESHUTDOWN`` after queue destruction starts. + +.. c:function:: int work_queue_next_wq( \ + FAR struct kwork_wqueue_s *wqueue, \ + FAR struct work_s *work, worker_t worker, FAR void *arg, \ + clock_t delay) + + Queue the next invocation relative to the work structure's previous + expiration time. This avoids accumulating callback execution time in a + periodic schedule. It is normally called from the work callback. + + :return: Zero on success, ``-EINVAL`` for invalid arguments, or + ``-ESHUTDOWN`` after queue destruction starts. + .. c:function:: int work_cancel(int qid, FAR struct work_s *work) Cancel previously queued work. This removes work @@ -240,11 +308,37 @@ Work Queue Interfaces :param work: The previously queued work structure to cancel. :return: Zero is returned on success; a negated ``errno`` is returned on - failure. + failure. Cancelling work that is not queued is a successful no-op. - - ``ENOENT``: There is no such work queued. - ``EINVAL``: An invalid work queue was specified. +.. c:function:: int work_cancel_wq(FAR struct kwork_wqueue_s *wqueue, \ + FAR struct work_s *work) + + Cancel pending work on a custom queue. Cancelling work that is not + queued is a successful no-op. + + :return: Zero on success or ``-EINVAL`` for an invalid argument. + +.. c:function:: int work_cancel_sync_wq( \ + FAR struct kwork_wqueue_s *wqueue, \ + FAR struct work_s *work) + + Cancel pending work and wait for callbacks already using the same work + structure to finish. If called from that work's own callback, the caller + is excluded from the wait to avoid self-deadlock. + + This interface must only be called from task context. + + :return: Zero on success or ``-EINVAL`` for an invalid argument. + +.. c:function:: int work_queue_priority_wq( \ + FAR struct kwork_wqueue_s *wqueue) + + Return the common scheduling priority of a custom queue's worker pool. + + :return: The worker priority on success or a negated errno on failure. + .. c:function:: int work_signal(int qid) Signal the worker thread to process the work @@ -295,4 +389,3 @@ Work Queue Interfaces :param reqprio: Previously requested minimum worker thread priority to be "unboosted". - diff --git a/boards/arm64/imx9/imx93-evk/configs/bootloader/defconfig b/boards/arm64/imx9/imx93-evk/configs/bootloader/defconfig index 5f1b05724e79f..8e0db27c1b76b 100644 --- a/boards/arm64/imx9/imx93-evk/configs/bootloader/defconfig +++ b/boards/arm64/imx9/imx93-evk/configs/bootloader/defconfig @@ -107,10 +107,6 @@ CONFIG_READLINE_CMD_HISTORY=y CONFIG_READLINE_CMD_HISTORY_LEN=16 CONFIG_READLINE_CMD_HISTORY_LINELEN=80 CONFIG_RR_INTERVAL=200 -CONFIG_SCHED_HPWORK=y -CONFIG_SCHED_HPWORKPRIORITY=192 -CONFIG_SCHED_LPWORK=y -CONFIG_SCHED_LPWORKPRIORITY=50 CONFIG_SIG_PREALLOC_IRQ_ACTIONS=8 CONFIG_SPINLOCK=y CONFIG_STACK_COLORATION=y diff --git a/include/nuttx/wqueue.h b/include/nuttx/wqueue.h index 0bda2d0acbeef..97becc82d7969 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 0dd9cf425f91f..72f7fb7af8f1a 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 7d4188e974fcb..65893daae311c 100644 --- a/libs/libc/wqueue/work_cancel.c +++ b/libs/libc/wqueue/work_cancel.c @@ -26,6 +26,7 @@ #include +#include #include #include @@ -51,59 +52,81 @@ * * 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); + + if (sync_wait == NULL) + { + return OK; + } - nxmutex_unlock(&wqueue->lock); - return ret; + nxsem_wait_uninterruptible(sync_wait); + } } /**************************************************************************** @@ -126,7 +149,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 +157,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 +165,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 104e10ff40361..bc10986ead856 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 f4a3663ab1049..f74632c2e5d4e 100644 --- a/libs/libc/wqueue/work_usrthread.c +++ b/libs/libc/wqueue/work_usrthread.c @@ -30,6 +30,7 @@ #include #include #include +#include #include #include @@ -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 eab017c6125b9..141c78e5bdd4f 100644 --- a/libs/libc/wqueue/wqueue.h +++ b/libs/libc/wqueue/wqueue.h @@ -29,7 +29,9 @@ #include -#include +#include +#include +#include #include #include @@ -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 */ diff --git a/sched/wqueue/kwork_cancel.c b/sched/wqueue/kwork_cancel.c index 8a8bc6036b158..2329abaf70824 100644 --- a/sched/wqueue/kwork_cancel.c +++ b/sched/wqueue/kwork_cancel.c @@ -32,6 +32,7 @@ #include #include #include +#include #include #include "wqueue/wqueue.h" @@ -46,63 +47,66 @@ static int work_qcancel(FAR struct kwork_wqueue_s *wqueue, bool sync, FAR struct work_s *work) { irqstate_t flags; - FAR sem_t *sync_wait = NULL; + pid_t self = sync ? nxsched_gettid() : INVALID_PROCESS_ID; 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. */ - flags = spin_lock_irqsave(&wqueue->lock); - - if (!work_available(work)) + for (; ; ) { - /* If the head of the pending queue has changed, we should reset - * the wqueue timer. - */ + FAR struct kworker_s *worker = wq_get_worker(wqueue); + FAR sem_t *sync_wait = NULL; + int wndx; - if (work_remove(wqueue, work)) - { - work_timer_reset(wqueue); - } - } + /* 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 from interrupt + * handlers. + */ - /* Note that cancel_sync can not be called in the interrupt - * context and the idletask context. - */ + flags = spin_lock_irqsave(&wqueue->lock); - if (sync) - { - int wndx; - pid_t pid = nxsched_gettid(); - FAR struct kworker_s *worker = wq_get_worker(wqueue); + if (!work_available(work)) + { + /* If the head of the pending queue has changed, reset the timer. */ - /* Wait until the worker thread finished the work. */ + if (work_remove(wqueue, work)) + { + work_timer_reset(wqueue); + } + } - for (wndx = 0; wndx < wqueue->nthreads; wndx++) + if (sync) { - if (worker[wndx].work == work && worker[wndx].pid != pid) + for (wndx = 0; wndx < wqueue->nthreads; wndx++) { - worker[wndx].wait_count++; - sync_wait = &worker[wndx].wait; - break; + if (worker[wndx].work == work && worker[wndx].pid != self) + { + worker[wndx].wait_count++; + sync_wait = &worker[wndx].wait; + break; + } } } - } - spin_unlock_irqrestore(&wqueue->lock, flags); + spin_unlock_irqrestore(&wqueue->lock, flags); + + if (sync_wait == NULL) + { + return OK; + } - if (sync_wait) - { nxsem_wait_uninterruptible(sync_wait); } - - return 0; } /**************************************************************************** @@ -125,7 +129,6 @@ static int work_qcancel(FAR struct kwork_wqueue_s *wqueue, bool sync, * Returned Value: * Zero on success, a negated errno on failure * - * -ENOENT - There is no such work queued. * -EINVAL - An invalid work queue was specified * ****************************************************************************/ @@ -145,9 +148,10 @@ 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 work queue and waits for callbacks that are already running. After + * work has been cancelled, it may be requeued by calling work_queue() + * again. * * Input Parameters: * qid - The work queue ID (must be HPWORK or LPWORK) @@ -158,7 +162,6 @@ int work_cancel_wq(FAR struct kwork_wqueue_s *wqueue, * Zero means the work was successfully cancelled. * A negated errno value is returned on any failure: * - * -ENOENT - There is no such work queued. * -EINVAL - An invalid work queue was specified * ****************************************************************************/ diff --git a/sched/wqueue/kwork_queue.c b/sched/wqueue/kwork_queue.c index 75b5a7f5fd9fa..6e9b4ac1a0738 100644 --- a/sched/wqueue/kwork_queue.c +++ b/sched/wqueue/kwork_queue.c @@ -41,18 +41,18 @@ #ifdef CONFIG_SCHED_WORKQUEUE /**************************************************************************** - * Public Functions + * Private Functions ****************************************************************************/ /**************************************************************************** - * Name: work_queue_next/work_queue_next_wq + * Name: work_qqueue * * Description: - * Queue work to be performed at a later time based on the last expiration - * time. This function must be called in the workqueue callback. + * Queue work on a kernel-mode work queue. Regular work uses an absolute + * expiration calculated before taking the queue lock. Periodic work + * advances the previous expiration while holding the lock. * * Input Parameters: - * qid - The work queue ID (must be HPWORK or LPWORK) * wqueue - The work queue handle * work - The work structure to queue * worker - The worker callback to be invoked. The callback will be @@ -61,33 +61,65 @@ * it is invoked. * delay - Delay (in clock ticks) from the time queue until the worker * is invoked. Zero means to perform the work immediately. + * period - Use the previous expiration as the scheduling reference * * Returned Value: * Zero on success, a negated errno on failure * ****************************************************************************/ -int work_queue_next_wq(FAR struct kwork_wqueue_s *wqueue, +static int work_qqueue(FAR struct kwork_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) { irqstate_t flags; + clock_t expected = 0; + bool retimer; if (wqueue == NULL || work == NULL || worker == NULL || - delay > WDOG_MAX_DELAY) + delay < 0 || delay > WDOG_MAX_DELAY) { return -EINVAL; } + /* Preserve regular queue timing across lock contention. */ + + if (!period) + { + expected = clock_delay2abstick(delay); + } + + /* Interrupts are disabled so that this logic can be called from task + * logic or interrupt handling logic. + */ + + flags = spin_lock_irqsave(&wqueue->lock); + + if (wqueue->exit) + { + spin_unlock_irqrestore(&wqueue->lock, flags); + return -ESHUTDOWN; + } + + /* Remove a previous pending instance before requeueing it. */ + + retimer = work_available(work) ? false : work_remove(wqueue, work); + /* Initialize the work structure. */ work->worker = worker; /* Work callback. non-NULL means queued */ work->arg = arg; /* Callback argument */ - work->qtime += delay; /* Expected time based on last expiration time */ - flags = spin_lock_irqsave(&wqueue->lock); + if (period) + { + work->qtime += delay; + } + else + { + work->qtime = expected; + } - if (delay) + if (delay > 0) { /* Insert to the pending list of the wqueue. */ @@ -95,6 +127,7 @@ int work_queue_next_wq(FAR struct kwork_wqueue_s *wqueue, { /* Start the timer if the work is the earliest expired work. */ + retimer = false; wd_start_abstick(&wqueue->timer, work->qtime, work_timer_expired, (wdparm_t)wqueue); } @@ -106,16 +139,55 @@ int work_queue_next_wq(FAR struct kwork_wqueue_s *wqueue, list_add_tail(&wqueue->expired, &work->node); } + if (retimer) + { + work_timer_reset(wqueue); + } + spin_unlock_irqrestore(&wqueue->lock, flags); - if (!delay) + if (delay == 0) { /* Immediately wake up the worker thread. */ nxsem_post(&wqueue->sem); } - return 0; + return OK; +} + +/**************************************************************************** + * Public Functions + ****************************************************************************/ + +/**************************************************************************** + * Name: work_queue_next/work_queue_next_wq + * + * Description: + * Queue work to be performed at a later time based on the last expiration + * time. This function must be called in the workqueue callback. + * + * Input Parameters: + * qid - The work queue ID (must be HPWORK or LPWORK) + * wqueue - The work queue handle + * work - The work structure to queue + * 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 + * it is invoked. + * delay - Delay (in clock ticks) from the time queue until the worker + * is invoked. Zero means to perform the work immediately. + * + * Returned Value: + * Zero on success, a negated errno on failure + * + ****************************************************************************/ + +int work_queue_next_wq(FAR struct kwork_wqueue_s *wqueue, + FAR struct work_s *work, worker_t worker, + FAR void *arg, clock_t delay) +{ + return work_qqueue(wqueue, work, worker, arg, delay, true); } int work_queue_next(int qid, FAR struct work_s *work, worker_t worker, @@ -158,69 +230,7 @@ int work_queue_wq(FAR struct kwork_wqueue_s *wqueue, FAR struct work_s *work, worker_t worker, FAR void *arg, clock_t delay) { - irqstate_t flags; - clock_t expected; - bool retimer; - - if (wqueue == NULL || work == NULL || worker == NULL || - delay > WDOG_MAX_DELAY) - { - return -EINVAL; - } - - expected = clock_delay2abstick(delay); - - /* Interrupts are disabled so that this logic can be called from with - * task logic or from interrupt handling logic. - */ - - flags = spin_lock_irqsave(&wqueue->lock); - - /* Ensure the work has been removed. */ - - retimer = work_available(work) ? false : work_remove(wqueue, work); - - /* Initialize the work structure. */ - - work->worker = worker; /* Work callback. non-NULL means queued */ - work->arg = arg; /* Callback argument */ - work->qtime = expected; /* Expected time */ - - if (delay) - { - /* Insert to the pending list of the wqueue. */ - - if (work_insert_pending(wqueue, work)) - { - /* Start the timer if the work is the earliest expired work. */ - - retimer = false; - wd_start_abstick(&wqueue->timer, work->qtime, - work_timer_expired, (wdparm_t)wqueue); - } - } - else - { - /* Insert to the expired list of the wqueue. */ - - list_add_tail(&wqueue->expired, &work->node); - } - - if (retimer) - { - work_timer_reset(wqueue); - } - - spin_unlock_irqrestore(&wqueue->lock, flags); - - if (!delay) - { - /* Immediately wake up the worker thread. */ - - nxsem_post(&wqueue->sem); - } - - return 0; + return work_qqueue(wqueue, work, worker, arg, delay, false); } int work_queue(int qid, FAR struct work_s *work, worker_t worker, diff --git a/sched/wqueue/kwork_thread.c b/sched/wqueue/kwork_thread.c index 4cb84ffb4ea8a..020b6fc3b77f5 100644 --- a/sched/wqueue/kwork_thread.c +++ b/sched/wqueue/kwork_thread.c @@ -28,6 +28,7 @@ #include #include +#include #include #include #include @@ -199,19 +200,20 @@ static int work_thread(int argc, FAR char *argv[]) kworker = (FAR struct kworker_s *) ((uintptr_t)strtoul(argv[2], NULL, 16)); - /* Loop until wqueue->exit != 0. - * Since the only way to set wqueue->exit is to call work_queue_free(), - * there is no need for entering the critical section. - */ - - while (!wqueue->exit) + for (; ; ) { /* And check first entry in the work queue. Since we have disabled * interrupts we know: (1) we will not be suspended unless we do * so ourselves, and (2) there will be no changes to the work queue */ - flags = spin_lock_irqsave_nopreempt(&wqueue->lock); + flags = spin_lock_irqsave_nopreempt(&wqueue->lock); + + if (wqueue->exit) + { + spin_unlock_irqrestore_nopreempt(&wqueue->lock, flags); + break; + } /* If the wqueue timer is expired and non-active, it indicates that * there might be expired work in the pending queue. @@ -266,6 +268,12 @@ static int work_thread(int argc, FAR char *argv[]) kworker->wait_count--; nxsem_post(&kworker->wait); } + + if (wqueue->exit) + { + spin_unlock_irqrestore_nopreempt(&wqueue->lock, flags); + break; + } } spin_unlock_irqrestore_nopreempt(&wqueue->lock, flags); @@ -306,6 +314,9 @@ static int work_thread_create(FAR const char *name, int priority, FAR char *argv[3]; char arg0[32]; char arg1[32]; + irqstate_t flags; + int created = 0; + int initialized = 0; int wndx; int pid; FAR void *stack = NULL; @@ -319,6 +330,7 @@ static int work_thread_create(FAR const char *name, int priority, for (wndx = 0; wndx < wqueue->nthreads; wndx++) { nxsem_init(&worker[wndx].wait, 0, 0); + initialized++; snprintf(arg0, sizeof(arg0), "%p", wqueue); snprintf(arg1, sizeof(arg1), "%p", &worker[wndx]); @@ -336,19 +348,51 @@ static int work_thread_create(FAR const char *name, int priority, pid = kthread_create_with_stack(name, priority, stack, stack_size, work_thread, argv); - DEBUGASSERT(pid > 0); - if (pid < 0) + if (pid <= 0) { + if (pid == 0) + { + pid = -EIO; + } + serr("ERROR: work_thread_create %d failed: %d\n", wndx, pid); - sched_unlock(); - return pid; + goto errout_with_threads; } worker[wndx].pid = pid; + created++; } sched_unlock(); return OK; + +errout_with_threads: + flags = spin_lock_irqsave_nopreempt(&wqueue->lock); + wqueue->exit = true; + spin_unlock_irqrestore_nopreempt(&wqueue->lock, flags); + + sched_unlock(); + + for (wndx = 0; wndx < created; wndx++) + { + nxsem_post(&wqueue->sem); + } + + for (wndx = 0; wndx < created; wndx++) + { + nxsem_wait_uninterruptible(&wqueue->exsem); + } + + for (wndx = 0; wndx < initialized; wndx++) + { + worker[wndx].pid = INVALID_PROCESS_ID; + nxsem_destroy(&worker[wndx].wait); + } + + nxsem_reset(&wqueue->sem, 0); + nxsem_reset(&wqueue->exsem, 0); + + return pid; } /**************************************************************************** @@ -373,6 +417,7 @@ void work_timer_expired(wdparm_t arg) */ FAR struct kwork_wqueue_s *wq = (FAR struct kwork_wqueue_s *)arg; + nxsem_post(&wq->sem); } @@ -380,18 +425,15 @@ void work_timer_expired(wdparm_t arg) * 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. This function creates + * a pool containing nthreads workers. * * 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 @@ -406,7 +448,8 @@ FAR struct kwork_wqueue_s *work_queue_create(FAR const char *name, FAR struct kwork_wqueue_s *wqueue; int ret; - if (nthreads < 1) + if (name == NULL || stack_size <= 0 || nthreads < 1 || + nthreads > (SIZE_MAX - sizeof(*wqueue)) / sizeof(struct kworker_s)) { return NULL; } @@ -428,6 +471,7 @@ FAR struct kwork_wqueue_s *work_queue_create(FAR const char *name, nxsem_init(&wqueue->sem, 0, 0); nxsem_init(&wqueue->exsem, 0, 0); wqueue->nthreads = nthreads; + wqueue->dynamic = true; spin_lock_init(&wqueue->lock); /* Create the work queue thread pool */ @@ -435,6 +479,8 @@ FAR struct kwork_wqueue_s *work_queue_create(FAR const char *name, ret = work_thread_create(name, priority, stack_addr, stack_size, wqueue); if (ret < 0) { + nxsem_destroy(&wqueue->sem); + nxsem_destroy(&wqueue->exsem); kmm_free(wqueue); return NULL; } @@ -446,12 +492,11 @@ 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. * * Input Parameters: - * qid - The work queue ID + * wqueue - The custom work queue handle * * Returned Value: * Zero on success, a negated errno value on failure. @@ -460,19 +505,57 @@ FAR struct kwork_wqueue_s *work_queue_create(FAR const char *name, int work_queue_free(FAR struct kwork_wqueue_s *wqueue) { + FAR struct work_s *work; + FAR struct work_s *next; + FAR struct kworker_s *worker; + irqstate_t flags; + pid_t self; int wndx; - if (wqueue == NULL) + if (wqueue == NULL || !wqueue->dynamic) { return -EINVAL; } - wd_cancel(&wqueue->timer); + worker = wq_get_worker(wqueue); + self = nxsched_gettid(); - /* Mark the work queue as exiting */ + for (wndx = 0; wndx < wqueue->nthreads; wndx++) + { + if (worker[wndx].pid == self) + { + return -EDEADLK; + } + } + + /* Mark the work queue as exiting and return all queued work structures + * to their owners before the queue storage is released. + */ + + flags = spin_lock_irqsave_nopreempt(&wqueue->lock); wqueue->exit = true; + list_for_every_entry_safe(&wqueue->expired, work, next, + struct work_s, node) + { + list_delete(&work->node); + work->worker = NULL; + } + + list_for_every_entry_safe(&wqueue->pending, work, next, + struct work_s, node) + { + list_delete(&work->node); + work->worker = NULL; + } + + spin_unlock_irqrestore_nopreempt(&wqueue->lock, flags); + + /* Stop delayed dispatch after new submissions have been disabled. */ + + wd_cancel(&wqueue->timer); + /* Queue a exit work for all threads */ for (wndx = 0; wndx < wqueue->nthreads; wndx++) @@ -485,6 +568,11 @@ int work_queue_free(FAR struct kwork_wqueue_s *wqueue) nxsem_wait_uninterruptible(&wqueue->exsem); } + for (wndx = 0; wndx < wqueue->nthreads; wndx++) + { + nxsem_destroy(&worker[wndx].wait); + } + nxsem_destroy(&wqueue->sem); nxsem_destroy(&wqueue->exsem); kmm_free(wqueue); diff --git a/sched/wqueue/wqueue.h b/sched/wqueue/wqueue.h index 3ac740729068a..5b210153ab556 100644 --- a/sched/wqueue/wqueue.h +++ b/sched/wqueue/wqueue.h @@ -78,8 +78,9 @@ struct kwork_wqueue_s sem_t sem; /* The counting semaphore of the wqueue */ sem_t exsem; /* Sync waiting for thread exit */ spinlock_t lock; /* Spinlock */ - uint8_t nthreads; /* Number of worker threads */ + int nthreads; /* Number of worker threads */ bool exit; /* A flag to request the thread to exit */ + bool dynamic; /* Dynamically allocated queue */ struct wdog_s timer; /* Timer to pending. */ }; @@ -214,11 +215,12 @@ bool work_insert_pending(FAR struct kwork_wqueue_s *wqueue, * * Description: * Internal public function to remove the work from the workqueue. + * The caller must hold wqueue->lock, and work must be queued on wqueue. * Require wqueue != NULL and work != NULL. * * Input Parameters: * wqueue - The work queue. - * work - The work to be inserted. + * work - The work to be removed. * * Returned Value: * Return whether the head of the pending queue has changed.