13022591351 commented on code in PR #19982:
URL: https://github.com/apache/nuttx/pull/19982#discussion_r3885587238
##########
libs/libc/wqueue/work_usrthread.c:
##########
@@ -207,122 +251,371 @@ static void work_process(FAR struct usr_wqueue_s
*wqueue)
nxsem_timedwait(&wqueue->wake, &rqtp);
}
+
+ return true;
}
/****************************************************************************
- * Name: work_usrthread
+ * Name: work_thread
*
* 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().
+ * 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:
- * argc, argv (not used)
+ * worker - Describes this worker and its parent queue
*
* Returned Value:
- * Does not return
+ * None
*
****************************************************************************/
+static void work_thread(FAR struct usr_worker_s *worker)
+{
+ while (work_process(worker))
+ {
+ }
+}
+
+static pthread_addr_t work_pthread(pthread_addr_t arg)
+{
+ work_thread((FAR struct usr_worker_s *)arg);
+ return NULL;
+}
+
#ifdef CONFIG_BUILD_PROTECTED
-static int work_usrthread(int argc, char *argv[])
-#else
-static pthread_addr_t work_usrthread(pthread_addr_t arg)
+static int work_usrtask(int argc, char *argv[])
+{
+ work_thread(&g_usrworker);
+ return OK;
+}
#endif
+
+/****************************************************************************
+ * 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)
{
- /* Loop forever */
+ pthread_attr_t attr;
+ struct sched_param param;
+ int created = 0;
+ int lockret;
+ int ret;
+ int wndx;
+
+ ret = pthread_attr_init(&attr);
+ if (ret != 0)
+ {
+ return -ret;
+ }
- for (; ; )
+ ret = pthread_attr_setstacksize(&attr, stack_size);
+ if (ret != 0)
{
- /* Then process queued work. We need to keep the work queue locked
- * while we process items in the work list.
- */
+ goto errout_with_attr;
+ }
- work_process(&g_usrwork);
+ ret = pthread_attr_setschedpolicy(&attr, SCHED_FIFO);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
}
-#ifdef CONFIG_BUILD_PROTECTED
- return OK; /* To keep some compilers happy */
-#else
- return NULL; /* To keep some compilers happy */
-#endif
+ param.sched_priority = priority;
+ ret = pthread_attr_setschedparam(&attr, ¶m);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ 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;
}
/****************************************************************************
* Public Functions
****************************************************************************/
/****************************************************************************
- * Name: work_usrstart
+ * Name: work_queue_create
*
* Description:
- * Start the user mode work queue.
+ * Create a user-mode custom work queue. This function must only be
+ * called from task context.
*
- * Input Parameters:
- * None
+ ****************************************************************************/
+
+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 ||
+ (size_t)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:
- * The task ID of the worker thread is returned on success. A negated
- * errno value is returned on failure.
+ * 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.
+ *
+ ****************************************************************************/
+
+int work_queue_free(FAR struct kwork_wqueue_s *handle)
+{
+ FAR struct usr_wqueue_s *wqueue = (FAR struct usr_wqueue_s *)handle;
+ FAR struct work_s *work;
+ FAR struct work_s *next;
+ pthread_t self = pthread_self();
+ int ret;
+ int wndx;
+
+ if (wqueue == NULL || !wqueue->dynamic)
+ {
+ return -EINVAL;
+ }
+
+ for (wndx = 0; wndx < wqueue->nthreads; wndx++)
+ {
+ if (pthread_equal(self, wqueue->worker[wndx].tid))
+ {
+ return -EDEADLK;
+ }
+ }
+
+ do
+ {
+ ret = nxmutex_lock(&wqueue->lock);
+ }
+ while (ret < 0);
+
+ 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;
+}
+
+/****************************************************************************
+ * 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 policy;
+ int ret;
+
+ if (wqueue == NULL || wqueue->nthreads < 1)
+ {
+ return -EINVAL;
+ }
+
+ ret = pthread_getschedparam(wqueue->worker[0].tid, &policy, ¶m);
+ return ret == 0 ? param.sched_priority : -ret;
+}
+
+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 predefined user work queue.
*
****************************************************************************/
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);
+ g_usrwork.exit = false;
+ g_usrworker.work = NULL;
+ g_usrworker.wait_count = 0;
#ifdef CONFIG_BUILD_PROTECTED
-
- /* Start a user-mode worker thread for use by applications. */
-
ret = task_create("uwork",
Review Comment:
USRWORK is the predefined user-mode work queue, analogous to the predefined
HPWORK and LPWORK queues. Its existing configuration provides priority and
stack size only, and it has one predefined worker. Additional multi-worker user
queues are custom queues created by work_queue_create() with the requested
nthreads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
