diff --git a/sched/wqueue/kwork_cancel.c b/sched/wqueue/kwork_cancel.c index 8a8bc6036b1..d18511ba68a 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,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 * ****************************************************************************/ diff --git a/sched/wqueue/kwork_queue.c b/sched/wqueue/kwork_queue.c index 75b5a7f5fd9..4e36159a66b 100644 --- a/sched/wqueue/kwork_queue.c +++ b/sched/wqueue/kwork_queue.c @@ -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); diff --git a/sched/wqueue/kwork_thread.c b/sched/wqueue/kwork_thread.c index 4cb84ffb4ea..020b6fc3b77 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 3ac74072906..5b210153ab5 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.