Mirrors_Framework/nuttx
1
0
Fork 0
mirror of https://github.com/apache/nuttx.git synced 2026-06-06 16:50:55 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

User-mode work queue logic should not disable interrupts

Browse Source
This commit is contained in:
Gregory Nutt
2014-10-10 14:52:04 -06:00
parent 16a3e83258
commit cf59a195ba
14 changed files with 654 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libc/wqueue/work_usrthread.c
+269 -22
View File
@@ -39,13 +39,19 @@
#include <nuttx/config.h>
#include <stdint.h>
#include <unistd.h>
#include <pthread.h>
#include <sched.h>
#include <errno.h>
#include <assert.h>
#include <queue.h>
#include <debug.h>
#include <nuttx/wqueue.h>
#include <nuttx/clock.h>
#include "wqueue/wqueue.h"
#if defined(CONFIG_SCHED_WORKQUEUE) && defined(CONFIG_SCHED_USRWORK) && \
!defined(__KERNEL__)
@@ -53,9 +59,19 @@
* Pre-processor Definitions
****************************************************************************/
/* The state of the user mode work queue. */
/* Use CLOCK_MONOTONIC if it is available. CLOCK_REALTIME can cause bad
* delays if the time is changed.
*/
extern struct wqueue_s g_usrwork;
#ifdef CONFIG_CLOCK_MONOTONIC
# define WORK_CLOCK CLOCK_MONOTONIC
#else
# define WORK_CLOCK CLOCK_REALTIME
#endif
#ifndef MIN
# define MIN(a,b) ((a) < (b) ? (a) : (b))
#endif
/****************************************************************************
* Private Type Declarations
@@ -65,6 +81,18 @@ extern struct wqueue_s g_usrwork;
* Public Data
****************************************************************************/
/* The state of the user mode work queue. */
struct wqueue_s g_usrwork;
/* This semaphore supports exclusive access to the user-mode work queue */
#ifdef CONFIG_BUILD_PROTECTED
extern sem_t g_usrsem;
#else
extern pthread_mutex_t g_usrmutex;
#endif
/****************************************************************************
* Private Data
****************************************************************************/
@@ -73,6 +101,179 @@ extern struct wqueue_s g_usrwork;
* Private Functions
****************************************************************************/
/****************************************************************************
* Name: work_process
*
* Description:
* This is the logic that performs actions placed on any work list. This
* logic is the common underlying logic to all work queues. This logic is
* part of the internal implementation of each work queue; it should not
* be called from application level logic.
*
* Input parameters:
* wqueue - Describes the work queue to be processed
*
* Returned Value:
* None
*
****************************************************************************/
void work_process(FAR struct wqueue_s *wqueue)
{
volatile FAR struct work_s *work;
worker_t worker;
FAR void *arg;
uint32_t elapsed;
uint32_t remaining;
uint32_t stick;
uint32_t ctick;
uint32_t next;
int ret;
/* Then process queued work. Lock the work queue while we process items
* in the work list.
*/
next = wqueue->delay;
ret = work_lock();
if (ret < 0)
{
/* Break out earlier if we were awakened by a signal */
return;
}
/* Get the time that we started this polling cycle in clock ticks. */
stick = clock_systimer();
/* And check each entry in the work queue. Since we have locked the
* work queue we know: (1) we will not be suspended unless we do
* so ourselves, and (2) there will be no changes to the work queue
*/
work = (FAR struct work_s *)wqueue->q.head;
while (work)
{
DEBUGASSERT(wqueue->wq_sem.count > 0);
/* Is this work ready? It is ready if there is no delay or if
* the delay has elapsed. qtime is the time that the work was added
* to the work queue. It will always be greater than or equal to
* zero. Therefore a delay of zero will always execute immediately.
*/
ctick = clock_systimer();
elapsed = ctick - work->qtime;
if (elapsed >= work->delay)
{
/* Remove the ready-to-execute work from the list */
(void)dq_rem((struct dq_entry_s *)work, &wqueue->q);
/* Extract the work description from the entry (in case the work
* instance by the re-used after it has been de-queued).
*/
worker = work->worker;
/* Check for a race condition where the work may be nullified
* before it is removed from the queue.
*/
if (worker != NULL)
{
/* Extract the work argument (before unlocking the work queue) */
arg = work->arg;
/* Mark the work as no longer being queued */
work->worker = NULL;
/* Do the work. Unlock the the work queue while the work is being
* performed... we don't have any idea how long this will take!
*/
work_unlock();
worker(arg);
/* Now, unfortunately, since we unlocked the work queue we don't
* know the state of the work list and we will have to start
* back at the head of the list.
*/
ret = work_lock();
if (ret < 0)
{
/* Break out earlier if we were awakened by a signal */
return;
}
work = (FAR struct work_s *)wqueue->q.head;
}
else
{
/* Cancelled.. Just move to the next work in the list with
* the work queue still locked.
*/
work = (FAR struct work_s *)work->dq.flink;
}
}
else
{
/* This one is not ready.. will it be ready before the next
* scheduled wakeup interval?
*
* NOTE that elapsed is relative to the the current time,
* not the time of beginning of this queue processing pass.
* So it may need an adjustment.
*/
elapsed += (ctick - stick);
remaining = elapsed - work->delay;
if (remaining < next)
{
/* Yes.. Then schedule to wake up when the work is ready */
next = remaining;
}
/* Then try the next in the list. */
work = (FAR struct work_s *)work->dq.flink;
}
}
/* Get the delay (in clock ticks) since we started the sampling */
elapsed = clock_systimer() - work->qtime;
if (elapsed <= wqueue->delay)
{
/* How must time would we need to delay to get to the end of the
* sampling period? The amount of time we delay should be the smaller
* of the time to the end of the sampling period and the time to the
* next work expiry.
*/
remaining = wqueue->delay - elapsed;
next = MIN(next, remaining);
if (next > 0)
{
/* Wait awhile to check the work list. We will wait here until
* either the time elapses or until we are awakened by a signal.
* Interrupts will be re-enabled while we wait.
*/
usleep(next * USEC_PER_TICK);
}
}
work_unlock();
}
/****************************************************************************
* Name: work_usrthread
*
@@ -92,24 +293,32 @@ extern struct wqueue_s g_usrwork;
*
****************************************************************************/
#ifdef CONFIG_BUILD_PROTECTED
static int work_usrthread(int argc, char *argv[])
#else
static pthread_addr_t work_usrthread(pthread_addr_t arg)
#endif
{
/* Loop forever */
for (;;)
{
/* Then process queued work. We need to keep interrupts disabled while
* we process items in the work list.
/* Then process queued work. We need to keep the work queue locked
* while we process items in the work list.
*/
work_process(&g_usrwork);
}
#ifdef CONFIG_BUILD_PROTECTED
return OK; /* To keep some compilers happy */
#else
return NULL; /* To keep some compilers happy */
#endif
}
/****************************************************************************
* Private Functions
* Public Functions
****************************************************************************/
/****************************************************************************
@@ -134,27 +343,65 @@ int work_usrstart(void)
g_usrwork.delay = CONFIG_SCHED_USRWORKPERIOD / USEC_PER_TICK;
dq_init(&g_usrwork.q);
/* Start a user-mode worker thread for use by applications. */
#ifdef CONFIG_BUILD_PROTECTED
{
/* Set up the work queue lock */
svdbg("Starting user-mode worker thread\n");
(void)sem_init(&g_usrsem, 0, 1);
g_usrwork.pid[0] = task_create("uwork",
CONFIG_SCHED_USRWORKPRIORITY,
CONFIG_SCHED_USRWORKSTACKSIZE,
(main_t)work_usrthread,
(FAR char * const *)NULL);
/* Start a user-mode worker thread for use by applications. */
DEBUGASSERT(g_usrwork.pid[0] > 0);
if (g_usrwork.pid[0] < 0)
{
int errcode = errno;
DEBUGASSERT(errcode > 0);
g_usrwork.pid[0] = task_create("uwork",
CONFIG_SCHED_USRWORKPRIORITY,
CONFIG_SCHED_USRWORKSTACKSIZE,
(main_t)work_usrthread,
(FAR char * const *)NULL);
sdbg("task_create failed: %d\n", errcode);
return -errcode;
}
DEBUGASSERT(g_usrwork.pid[0] > 0);
if (g_usrwork.pid[0] < 0)
{
int errcode = errno;
DEBUGASSERT(errcode > 0);
return -errcode;
}
return g_usrwork.pid[0];
return g_usrwork.pid[0];
}
#else
{
pthread_t usrwork;
pthread_attr_t attr;
struct sched_param param;
int status;
/* Set up the work queue lock */
(void)pthread_mutex_init(&g_usrmutex, NULL);
/* Start a user-mode worker thread for use by applications. */
(void)pthread_attr_init(&attr);
(void)pthread_attr_setstacksize(&attr, CONFIG_SCHED_USRWORKSTACKSIZE);
param.sched_priority = CONFIG_SCHED_USRWORKPRIORITY;
(void)pthread_attr_setschedparam(&attr, &param);
status = pthread_create(&usrwork, &attr, work_usrthread, NULL);
if (status != 0)
{
return -status;
}
/* Detach because the return value and completion status will not be
* requested.
*/
(void)pthread_detach(usrwork);
g_usrwork.pid[0] = (pid_t)usrwork;
return g_usrwork.pid[0];
}
#endif
}
#endif /* CONFIG_SCHED_WORKQUEUE && CONFIG_SCHED_USRWORK && !__KERNEL__*/