Aksal/nuttx
3
1
Fork 1
mirror of https://github.com/apache/nuttx.git synced 2026-09-06 08:56:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

sched/wqueue: harden custom queue lifecycle

Browse source 
Prevent work_queue_free() from destroying predefined queues or freeing a
custom queue from one of its own callbacks.  Mark teardown under the queue
lock, reject new submissions, return pending work to its owner, and wait for
every worker before releasing queue resources.

Clean up partially created worker pools, reject invalid delays, safely
replace pending periodic work, and make synchronous cancellation wait for
every concurrent callback using the same work structure.

Tested on an STM32H7 PX4 FMUv6C with the matching ostest suite in Flat and
Protected kernel builds.

Assisted-by: Codex:GPT-5
Signed-off-by: DuoYuWang <thirteenking.wang@gmail.com>
This commit is contained in:
DuoYuWang 2026-08-26 22:04:15 +08:00 • committed by Xiang Xiao
parent 673b4245f7
commit d2e01b9055
4 changed files with 201 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

98
sched/wqueue/kwork_cancel.c
View file

@ -32,6 +32,7 @@
#include <nuttx/irq.h>
#include <nuttx/arch.h>
#include <nuttx/list.h>
#include <nuttx/sched.h>
#include <nuttx/wqueue.h>
#include "wqueue/wqueue.h"
@ -46,63 +47,71 @@ 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;
int ret;
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;
/* 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.
*/
if (work_remove(wqueue, work))
flags = spin_lock_irqsave(&wqueue->lock);
if (!work_available(work))
{
work_timer_reset(wqueue);
}
}
/* If the head of the pending queue has changed, reset the timer. */
/* Note that cancel_sync can not be called in the interrupt
* context and the idletask context.
*/
if (sync)
{
int wndx;
pid_t pid = nxsched_gettid();
FAR struct kworker_s *worker = wq_get_worker(wqueue);
/* Wait until the worker thread finished the work. */
for (wndx = 0; wndx < wqueue->nthreads; wndx++)
{
if (worker[wndx].work == work && worker[wndx].pid != pid)
if (work_remove(wqueue, work))
{
worker[wndx].wait_count++;
sync_wait = &worker[wndx].wait;
break;
work_timer_reset(wqueue);
}
}
if (sync)
{
for (wndx = 0; wndx < wqueue->nthreads; wndx++)
{
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);
if (sync_wait == NULL)
{
return OK;
}
do
{
ret = nxsem_wait(sync_wait);
}
while (ret == -EINTR);
}
spin_unlock_irqrestore(&wqueue->lock, flags);
if (sync_wait)
{
nxsem_wait_uninterruptible(sync_wait);
}
return 0;
}
/****************************************************************************
@ -125,7 +134,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 +153,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 +167,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
*
****************************************************************************/

37
sched/wqueue/kwork_queue.c
View file

@ -72,21 +72,33 @@ int work_queue_next_wq(FAR struct kwork_wqueue_s *wqueue,
FAR void *arg, clock_t delay)
{
irqstate_t flags;
bool retimer;
int ret = OK;
if (wqueue == NULL || work == NULL || worker == NULL ||
delay > WDOG_MAX_DELAY)
delay < 0 || delay > WDOG_MAX_DELAY)
{
return -EINVAL;
}
flags = spin_lock_irqsave(&wqueue->lock);
if (wqueue->exit)
{
ret = -ESHUTDOWN;
goto out;
}
/* 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 (delay)
{
/* Insert to the pending list of the wqueue. */
@ -95,6 +107,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 +119,22 @@ int work_queue_next_wq(FAR struct kwork_wqueue_s *wqueue,
list_add_tail(&wqueue->expired, &work->node);
}
if (retimer)
{
work_timer_reset(wqueue);
}
out:
spin_unlock_irqrestore(&wqueue->lock, flags);
if (!delay)
if (ret == OK && !delay)
{
/* Immediately wake up the worker thread. */
nxsem_post(&wqueue->sem);
}
return 0;
return ret;
}
int work_queue_next(int qid, FAR struct work_s *work, worker_t worker,
@ -163,7 +182,7 @@ int work_queue_wq(FAR struct kwork_wqueue_s *wqueue,
bool retimer;
if (wqueue == NULL || work == NULL || worker == NULL ||
delay > WDOG_MAX_DELAY)
delay < 0 || delay > WDOG_MAX_DELAY)
{
return -EINVAL;
}
@ -176,6 +195,12 @@ int work_queue_wq(FAR struct kwork_wqueue_s *wqueue,
flags = spin_lock_irqsave(&wqueue->lock);
if (wqueue->exit)
{
spin_unlock_irqrestore(&wqueue->lock, flags);
return -ESHUTDOWN;
}
/* Ensure the work has been removed. */
retimer = work_available(work) ? false : work_remove(wqueue, work);

138
sched/wqueue/kwork_thread.c
View file

@ -28,6 +28,7 @@
#include <unistd.h>
#include <sched.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
@ -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);

6
sched/wqueue/wqueue.h
View file

@ -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.