xiaoxiang781216 commented on code in PR #19982:
URL: https://github.com/apache/nuttx/pull/19982#discussion_r3870817035
##########
sched/wqueue/kwork_cancel.c:
##########
@@ -46,63 +47,65 @@ 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.
- */
-
- 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 sem_t *sync_wait = NULL;
+ int wndx;
+ FAR struct kworker_s *worker = wq_get_worker(wqueue);
Review Comment:
move before line 59
##########
libs/libc/wqueue/work_usrthread.c:
##########
@@ -78,17 +89,18 @@ struct usr_wqueue_s g_usrwork =
* be called from application level logic.
*
* Input Parameters:
- * wqueue - Describes the work queue to be processed
+ * wqworker - Describes this worker and its parent queue
*
* Returned Value:
- * None
+ * true to continue processing; false to exit
*
****************************************************************************/
-static void work_process(FAR struct usr_wqueue_s *wqueue)
+static bool work_process(FAR struct usr_worker_s *wqworker)
Review Comment:
wqworker->worker
##########
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",
CONFIG_LIBC_USRWORKPRIORITY,
CONFIG_LIBC_USRWORKSTACKSIZE,
- work_usrthread, NULL);
+ work_usrtask, NULL);
if (ret < 0)
{
int errcode = get_errno();
+
DEBUGASSERT(errcode > 0);
return -errcode;
}
+ g_usrworker.tid = (pthread_t)ret;
return ret;
#else
- /* Start a user-mode worker thread for use by applications. */
-
pthread_attr_init(&attr);
pthread_attr_setstacksize(&attr, CONFIG_LIBC_USRWORKSTACKSIZE);
-
Review Comment:
add pthread_attr_setdetachstate and remove line 617
##########
libs/libc/wqueue/work_usrthread.c:
##########
@@ -57,17 +59,26 @@
/* 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,
};
-/****************************************************************************
Review Comment:
why remove
##########
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);
Review Comment:
move the init to the variable definition
##########
libs/libc/wqueue/work_queue.c:
##########
@@ -115,13 +144,9 @@ static int work_qqueue(FAR struct usr_wqueue_s *wqueue,
* We should wake up the worker thread.
*/
- if (curr == head)
+ if (delay == 0 || curr == head)
Review Comment:
why need check delay
##########
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)
Review Comment:
merge into work_pthread
##########
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);
Review Comment:
work_pthread
##########
libs/libc/wqueue/work_queue.c:
##########
@@ -39,10 +39,6 @@
#if defined(CONFIG_LIBC_USRWORK) && !defined(__KERNEL__)
-/****************************************************************************
Review Comment:
why remove
##########
libs/libc/wqueue/work_queue.c:
##########
@@ -74,21 +68,56 @@
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;
+ 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)
+ {
+ list_delete(&work->node);
Review Comment:
it's wrong to modify the list without lock, let's call cancel directly
##########
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 ||
+ if (wqueue == NULL || work == NULL || worker == NULL || delay < 0 ||
Review Comment:
move to `delay < 0 ||` next line
##########
libs/libc/wqueue/work_queue.c:
##########
@@ -74,21 +68,56 @@
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;
+ int ret;
+
+ if (wqueue == NULL || work == NULL || worker == NULL || delay < 0 ||
Review Comment:
move `delay < 0` to next line
##########
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:
what about other thread
##########
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;
Review Comment:
why need zero again
##########
sched/wqueue/kwork_queue.c:
##########
@@ -162,7 +181,7 @@ int work_queue_wq(FAR struct kwork_wqueue_s *wqueue,
clock_t expected;
bool retimer;
- if (wqueue == NULL || work == NULL || worker == NULL ||
+ if (wqueue == NULL || work == NULL || worker == NULL || delay < 0 ||
Review Comment:
move `delay < 0 ||` to next line
##########
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",
CONFIG_LIBC_USRWORKPRIORITY,
CONFIG_LIBC_USRWORKSTACKSIZE,
- work_usrthread, NULL);
+ work_usrtask, NULL);
if (ret < 0)
{
int errcode = get_errno();
+
DEBUGASSERT(errcode > 0);
return -errcode;
}
+ g_usrworker.tid = (pthread_t)ret;
return ret;
#else
- /* Start a user-mode worker thread for use by applications. */
-
pthread_attr_init(&attr);
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;
+ pthread_detach(g_usrworker.tid);
+ return (pid_t)g_usrworker.tid;
Review Comment:
move after line 619 and remove line 601
--
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]
