xiaoxiang781216 commented on code in PR #19982:
URL: https://github.com/apache/nuttx/pull/19982#discussion_r3887237047
##########
libs/libc/wqueue/work_usrthread.c:
##########
@@ -207,123 +255,385 @@ static void work_process(FAR struct usr_wqueue_s
*wqueue)
nxsem_timedwait(&wqueue->wake, &rqtp);
}
+
+ return true;
}
/****************************************************************************
- * Name: work_usrthread
+ * Name: work_priority
+ *
+ * Description:
+ * Return the common scheduling priority of a user-mode work queue.
+ *
+ ****************************************************************************/
+
+static int work_priority(FAR struct usr_wqueue_s *wqueue)
+{
+ 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();
+}
+
+/****************************************************************************
+ * Name: work_pthread
*
* 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)
+ * arg - Describes this worker and its parent queue
*
* Returned Value:
- * Does not return
+ * NULL
*
****************************************************************************/
+static pthread_addr_t work_pthread(pthread_addr_t arg)
+{
+ FAR struct usr_worker_s *worker =
+ (FAR struct usr_worker_s *)arg;
+
+ while (work_process(worker))
+ {
+ }
+
+ 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_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)
{
- /* Loop forever */
+ pthread_attr_t attr;
+ struct sched_param param;
+ int created = 0;
+ int lockret;
+ int ret;
+ int wndx;
- for (; ; )
+ ret = pthread_attr_init(&attr);
+ if (ret != 0)
{
- /* Then process queued work. We need to keep the work queue locked
- * while we process items in the work list.
- */
+ return -ret;
+ }
- work_process(&g_usrwork);
+ ret = pthread_attr_setstacksize(&attr, stack_size);
+ 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
+ ret = pthread_attr_setschedpolicy(&attr, SCHED_FIFO);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ ret = pthread_attr_getschedparam(&attr, ¶m);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ 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;
+ pid_t self = gettid();
+ int ret;
+ int wndx;
+
+ if (wqueue == NULL || !wqueue->dynamic)
+ {
+ return -EINVAL;
+ }
+
+ for (wndx = 0; wndx < wqueue->nthreads; wndx++)
+ {
+ if (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;
+}
+#endif
+
+/****************************************************************************
+ * Name: work_queue_priority_wq
+ ****************************************************************************/
+
+#ifndef CONFIG_DISABLE_PTHREAD
+int work_queue_priority_wq(FAR struct kwork_wqueue_s *handle)
+{
+ return work_priority((FAR struct usr_wqueue_s *)handle);
Review Comment:
merge work_priority here and remove work_priority
##########
libs/libc/wqueue/work_cancel.c:
##########
@@ -57,53 +58,78 @@
* 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;
+ pid_t self = gettid();
+ int ret;
- DEBUGASSERT(work != NULL);
+ if (wqueue == NULL || work == NULL)
+ {
+ return -EINVAL;
+ }
- /* Get exclusive access to the work queue */
+ for (; ; )
Review Comment:
why need for loop
##########
sched/wqueue/kwork_thread.c:
##########
@@ -336,19 +348,54 @@ 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);
+
+ flags = spin_lock_irqsave_nopreempt(&wqueue->lock);
Review Comment:
why need lock
##########
libs/libc/wqueue/work_usrthread.c:
##########
@@ -207,123 +255,385 @@ static void work_process(FAR struct usr_wqueue_s
*wqueue)
nxsem_timedwait(&wqueue->wake, &rqtp);
}
+
+ return true;
}
/****************************************************************************
- * Name: work_usrthread
+ * Name: work_priority
+ *
+ * Description:
+ * Return the common scheduling priority of a user-mode work queue.
+ *
+ ****************************************************************************/
+
+static int work_priority(FAR struct usr_wqueue_s *wqueue)
+{
+ 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();
+}
+
+/****************************************************************************
+ * Name: work_pthread
*
* 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)
+ * arg - Describes this worker and its parent queue
*
* Returned Value:
- * Does not return
+ * NULL
*
****************************************************************************/
+static pthread_addr_t work_pthread(pthread_addr_t arg)
+{
+ FAR struct usr_worker_s *worker =
+ (FAR struct usr_worker_s *)arg;
+
+ while (work_process(worker))
+ {
+ }
+
+ 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_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)
{
- /* Loop forever */
+ pthread_attr_t attr;
+ struct sched_param param;
+ int created = 0;
+ int lockret;
+ int ret;
+ int wndx;
- for (; ; )
+ ret = pthread_attr_init(&attr);
+ if (ret != 0)
{
- /* Then process queued work. We need to keep the work queue locked
- * while we process items in the work list.
- */
+ return -ret;
+ }
- work_process(&g_usrwork);
+ ret = pthread_attr_setstacksize(&attr, stack_size);
+ 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
+ ret = pthread_attr_setschedpolicy(&attr, SCHED_FIFO);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ ret = pthread_attr_getschedparam(&attr, ¶m);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ 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;
+ pid_t self = gettid();
+ int ret;
+ int wndx;
+
+ if (wqueue == NULL || !wqueue->dynamic)
+ {
+ return -EINVAL;
+ }
+
+ for (wndx = 0; wndx < wqueue->nthreads; wndx++)
+ {
+ if (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;
+}
+#endif
+
+/****************************************************************************
+ * Name: work_queue_priority_wq
+ ****************************************************************************/
+
+#ifndef CONFIG_DISABLE_PTHREAD
+int work_queue_priority_wq(FAR struct kwork_wqueue_s *handle)
+{
+ return work_priority((FAR struct usr_wqueue_s *)handle);
+}
+#endif
+
+int work_queue_priority(int qid)
+{
+ if (qid != USRWORK)
+ {
+ return -EINVAL;
+ }
+
+ return work_priority(&g_usrwork);
Review Comment:
call work_queue_priority_wq
##########
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 (; ; )
Review Comment:
why need for loop here?
##########
libs/libc/wqueue/work_usrthread.c:
##########
@@ -207,123 +255,385 @@ static void work_process(FAR struct usr_wqueue_s
*wqueue)
nxsem_timedwait(&wqueue->wake, &rqtp);
}
+
+ return true;
}
/****************************************************************************
- * Name: work_usrthread
+ * Name: work_priority
+ *
+ * Description:
+ * Return the common scheduling priority of a user-mode work queue.
+ *
+ ****************************************************************************/
+
+static int work_priority(FAR struct usr_wqueue_s *wqueue)
+{
+ 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();
+}
+
+/****************************************************************************
+ * Name: work_pthread
*
* 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)
+ * arg - Describes this worker and its parent queue
*
* Returned Value:
- * Does not return
+ * NULL
*
****************************************************************************/
+static pthread_addr_t work_pthread(pthread_addr_t arg)
+{
+ FAR struct usr_worker_s *worker =
+ (FAR struct usr_worker_s *)arg;
+
+ while (work_process(worker))
+ {
+ }
+
+ 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_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)
{
- /* Loop forever */
+ pthread_attr_t attr;
+ struct sched_param param;
+ int created = 0;
+ int lockret;
+ int ret;
+ int wndx;
- for (; ; )
+ ret = pthread_attr_init(&attr);
+ if (ret != 0)
{
- /* Then process queued work. We need to keep the work queue locked
- * while we process items in the work list.
- */
+ return -ret;
+ }
- work_process(&g_usrwork);
+ ret = pthread_attr_setstacksize(&attr, stack_size);
+ 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
+ ret = pthread_attr_setschedpolicy(&attr, SCHED_FIFO);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ ret = pthread_attr_getschedparam(&attr, ¶m);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ 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)) /
Review Comment:
remove the cast and merge the next line
##########
libs/libc/wqueue/work_usrthread.c:
##########
@@ -207,123 +255,385 @@ static void work_process(FAR struct usr_wqueue_s
*wqueue)
nxsem_timedwait(&wqueue->wake, &rqtp);
}
+
+ return true;
}
/****************************************************************************
- * Name: work_usrthread
+ * Name: work_priority
+ *
+ * Description:
+ * Return the common scheduling priority of a user-mode work queue.
+ *
+ ****************************************************************************/
+
+static int work_priority(FAR struct usr_wqueue_s *wqueue)
+{
+ 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();
+}
+
+/****************************************************************************
+ * Name: work_pthread
*
* 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)
+ * arg - Describes this worker and its parent queue
*
* Returned Value:
- * Does not return
+ * NULL
*
****************************************************************************/
+static pthread_addr_t work_pthread(pthread_addr_t arg)
+{
+ FAR struct usr_worker_s *worker =
+ (FAR struct usr_worker_s *)arg;
+
+ while (work_process(worker))
Review Comment:
merge work_process here and remove work_process
##########
libs/libc/wqueue/work_usrthread.c:
##########
@@ -207,123 +255,385 @@ static void work_process(FAR struct usr_wqueue_s
*wqueue)
nxsem_timedwait(&wqueue->wake, &rqtp);
}
+
+ return true;
}
/****************************************************************************
- * Name: work_usrthread
+ * Name: work_priority
+ *
+ * Description:
+ * Return the common scheduling priority of a user-mode work queue.
+ *
+ ****************************************************************************/
+
+static int work_priority(FAR struct usr_wqueue_s *wqueue)
+{
+ 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();
+}
+
+/****************************************************************************
+ * Name: work_pthread
*
* 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)
+ * arg - Describes this worker and its parent queue
*
* Returned Value:
- * Does not return
+ * NULL
*
****************************************************************************/
+static pthread_addr_t work_pthread(pthread_addr_t arg)
+{
+ FAR struct usr_worker_s *worker =
+ (FAR struct usr_worker_s *)arg;
+
+ while (work_process(worker))
+ {
+ }
+
+ 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_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)
{
- /* Loop forever */
+ pthread_attr_t attr;
+ struct sched_param param;
+ int created = 0;
+ int lockret;
+ int ret;
+ int wndx;
- for (; ; )
+ ret = pthread_attr_init(&attr);
+ if (ret != 0)
{
- /* Then process queued work. We need to keep the work queue locked
- * while we process items in the work list.
- */
+ return -ret;
+ }
- work_process(&g_usrwork);
+ ret = pthread_attr_setstacksize(&attr, stack_size);
+ 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
+ ret = pthread_attr_setschedpolicy(&attr, SCHED_FIFO);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ ret = pthread_attr_getschedparam(&attr, ¶m);
+ if (ret != 0)
+ {
+ goto errout_with_attr;
+ }
+
+ 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;
+ pid_t self = gettid();
+ int ret;
+ int wndx;
+
+ if (wqueue == NULL || !wqueue->dynamic)
+ {
+ return -EINVAL;
+ }
+
+ for (wndx = 0; wndx < wqueue->nthreads; wndx++)
+ {
+ if (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;
+}
+#endif
+
+/****************************************************************************
+ * Name: work_queue_priority_wq
+ ****************************************************************************/
+
+#ifndef CONFIG_DISABLE_PTHREAD
Review Comment:
remove
--
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]
