epicsThreadPool.h File Reference

#include <stdlib.h>
#include <stdio.h>
#include "libComAPI.h"
#include "errMdef.h"

Include dependency graph for epicsThreadPool.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Classes
struct	epicsThreadPoolConfig

Macros
#define	S_pool_jobBusy   (M_pool| 1) /*Job already queued or running*/
#define	S_pool_jobIdle   (M_pool| 2) /*Job was not queued or running*/
#define	S_pool_noPool   (M_pool| 3) /*Job not associated with a pool*/
#define	S_pool_paused   (M_pool| 4) /*Pool not currently accepting jobs*/
#define	S_pool_noThreads   (M_pool| 5) /*Can't create worker thread*/
#define	S_pool_timeout   (M_pool| 6) /*Pool still busy after timeout*/
#define	EPICSJOB_SELF   epicsJobArgSelfMagic

Typedefs
typedef struct epicsThreadPool	epicsThreadPool
typedef void(*	epicsJobFunction) (void *arg, epicsJobMode mode)
typedef struct epicsJob	epicsJob

Enumerations
enum	epicsJobMode { epicsJobModeRun, epicsJobModeCleanup }
enum	epicsThreadPoolOption { epicsThreadPoolQueueAdd, epicsThreadPoolQueueRun }

Functions
LIBCOM_API void	epicsThreadPoolConfigDefaults (epicsThreadPoolConfig *)
LIBCOM_API epicsThreadPool *	epicsThreadPoolGetShared (epicsThreadPoolConfig *opts)
LIBCOM_API void	epicsThreadPoolReleaseShared (epicsThreadPool *pool)
LIBCOM_API epicsThreadPool *	epicsThreadPoolCreate (epicsThreadPoolConfig *opts)
LIBCOM_API void	epicsThreadPoolDestroy (epicsThreadPool *)
LIBCOM_API void	epicsThreadPoolControl (epicsThreadPool *pool, epicsThreadPoolOption opt, unsigned int val)
LIBCOM_API int	epicsThreadPoolWait (epicsThreadPool *pool, double timeout)
LIBCOM_API epicsJob *	epicsJobCreate (epicsThreadPool *pool, epicsJobFunction cb, void *user)
LIBCOM_API void	epicsJobDestroy (epicsJob *)
LIBCOM_API int	epicsJobMove (epicsJob *job, epicsThreadPool *pool)
LIBCOM_API int	epicsJobQueue (epicsJob *)
LIBCOM_API int	epicsJobUnqueue (epicsJob *)
LIBCOM_API void	epicsThreadPoolReport (epicsThreadPool *pool, FILE *fd)
LIBCOM_API unsigned int	epicsThreadPoolNThreads (epicsThreadPool *)

Variables
LIBCOM_API void *	epicsJobArgSelfMagic

Macro Definition Documentation

#define EPICSJOB_SELF   epicsJobArgSelfMagic

Definition at line 107 of file epicsThreadPool.h.

#define S_pool_jobBusy   (M_pool| 1) /*Job already queued or running*/

Definition at line 19 of file epicsThreadPool.h.

#define S_pool_jobIdle   (M_pool| 2) /*Job was not queued or running*/

Definition at line 20 of file epicsThreadPool.h.

#define S_pool_noPool   (M_pool| 3) /*Job not associated with a pool*/

Definition at line 21 of file epicsThreadPool.h.

#define S_pool_noThreads   (M_pool| 5) /*Can't create worker thread*/

Definition at line 23 of file epicsThreadPool.h.

#define S_pool_paused   (M_pool| 4) /*Pool not currently accepting jobs*/

Definition at line 22 of file epicsThreadPool.h.

#define S_pool_timeout   (M_pool| 6) /*Pool still busy after timeout*/

Definition at line 24 of file epicsThreadPool.h.

Typedef Documentation

typedef struct epicsJob epicsJob

Definition at line 54 of file epicsThreadPool.h.

typedef void(* epicsJobFunction) (void *arg, epicsJobMode mode)

Definition at line 52 of file epicsThreadPool.h.

typedef struct epicsThreadPool epicsThreadPool

Definition at line 37 of file epicsThreadPool.h.

Enumeration Type Documentation

enum epicsJobMode

Enumerator
epicsJobModeRun
epicsJobModeCleanup

Definition at line 40 of file epicsThreadPool.h.

             {
    /* Normal run of job */
    epicsJobModeRun,

    /* Thread pool is being destroyed.
     * A chance to cleanup the job immediately with epicsJobDestroy().
     * If ignored, the job is orphaned (dissociated from the thread pool)
     * and epicsJobDestroy() must be called later.
     */
    epicsJobModeCleanup
} epicsJobMode;

enum epicsThreadPoolOption

Enumerator
epicsThreadPoolQueueAdd
epicsThreadPoolQueueRun

Definition at line 82 of file epicsThreadPool.h.

             {
    epicsThreadPoolQueueAdd, /* val==0 causes epicsJobQueue to fail, 1 is default */
    epicsThreadPoolQueueRun /* val==0 prevents workers from running jobs, 1 is default */
} epicsThreadPoolOption;

Function Documentation

LIBCOM_API epicsJob* epicsJobCreate	(	epicsThreadPool *	pool,
		epicsJobFunction	cb,
		void *	user
	)

Definition at line 134 of file poolJob.c.

{
    epicsJob *job = calloc(1, sizeof(*job));

    if (!job)
        return NULL;

    if (arg == &epicsJobArgSelfMagic)
        arg = job;

    job->pool = NULL;
    job->func = func;
    job->arg = arg;

    epicsJobMove(job, pool);

    return job;
}

LIBCOM_API void epicsJobDestroy

(

epicsJob *

)

Definition at line 155 of file poolJob.c.

{
    epicsThreadPool *pool;
    if (!job || !job->pool) {
        free(job);
        return;
    }
    pool = job->pool;

    epicsMutexMustLock(pool->guard);

    assert(!job->dead);

    epicsJobUnqueue(job);

    if (job->running || job->freewhendone) {
        job->freewhendone = 1;
    }
    else {
        ellDelete(&pool->owned, &job->jobnode);
        job->dead = 1;
        free(job);
    }

    epicsMutexUnlock(pool->guard);
}

LIBCOM_API int epicsJobMove	(	epicsJob *	job,
		epicsThreadPool *	pool
	)

Definition at line 182 of file poolJob.c.

{
    epicsThreadPool *pool = job->pool;

    /* remove from current pool */
    if (pool) {
        epicsMutexMustLock(pool->guard);

        if (job->queued || job->running) {
            epicsMutexUnlock(pool->guard);
            return S_pool_jobBusy;
        }

        ellDelete(&pool->owned, &job->jobnode);

        epicsMutexUnlock(pool->guard);
    }

    pool = job->pool = newpool;

    /* add to new pool */
    if (pool) {
        epicsMutexMustLock(pool->guard);

        ellAdd(&pool->owned, &job->jobnode);

        epicsMutexUnlock(pool->guard);
    }

    return 0;
}

LIBCOM_API int epicsJobQueue

(

epicsJob *

)

Definition at line 214 of file poolJob.c.

{
    int ret = 0;
    epicsThreadPool *pool = job->pool;

    if (!pool)
        return S_pool_noPool;

    epicsMutexMustLock(pool->guard);

    assert(!job->dead);

    if (pool->pauseadd) {
        ret = S_pool_paused;
        goto done;
    }
    else if (job->freewhendone) {
        ret = S_pool_jobBusy;
        goto done;
    }
    else if (job->queued) {
        goto done;
    }

    job->queued = 1;
    /* Job may be queued from within a callback */
    if (!job->running) {
        ellDelete(&pool->owned, &job->jobnode);
        ellAdd(&pool->jobs, &job->jobnode);
    }
    else {
        /* some worker will find it again before sleeping */
        goto done;
    }

    /* Since we hold the lock, we can be certain that all awake worker are
     * executing work functions.  The current thread may be a worker.
     * We prefer to wakeup a new worker rather then wait for a busy worker to
     * finish.  However, after we initiate a wakeup there will be a race
     * between the worker waking up, and a busy worker finishing.
     * Thus we can't avoid spurious wakeups.
     */

    if (pool->threadsRunning >= pool->conf.maxThreads) {
        /* all workers created... */
        /* ... but some are sleeping, so wake one up */
        if (pool->threadsWaking < pool->threadsSleeping) {
            pool->threadsWaking++;
            epicsEventSignal(pool->workerWakeup);
        }
        /*else one of the running workers will find this job before sleeping */
        CHECKCOUNT(pool);

    }
    else {
        /* could create more workers so
         * will either create a new worker, or wakeup an existing worker
         */

        if (pool->threadsWaking >= pool->threadsSleeping) {
            /* all sleeping workers have already been woken.
             * start a new worker for this job
             */
            if (createPoolThread(pool) && pool->threadsRunning == 0) {
                /* oops, we couldn't lazy create our first worker
                 * so this job would never run!
                 */
                ret = S_pool_noThreads;
                job->queued = 0;
                /* if threadsRunning==0 then no jobs can be running */
                assert(!job->running);
                ellDelete(&pool->jobs, &job->jobnode);
                ellAdd(&pool->owned, &job->jobnode);
            }
        }
        if (ret == 0) {
            pool->threadsWaking++;
            epicsEventSignal(pool->workerWakeup);
        }
        CHECKCOUNT(pool);
    }

done:
    epicsMutexUnlock(pool->guard);
    return ret;
}

LIBCOM_API int epicsJobUnqueue

(

epicsJob *

)

Definition at line 301 of file poolJob.c.

{
    int ret = S_pool_jobIdle;
    epicsThreadPool *pool = job->pool;

    if (!pool)
        return S_pool_noPool;

    epicsMutexMustLock(pool->guard);

    assert(!job->dead);

    if (job->queued) {
        if (!job->running) {
            ellDelete(&pool->jobs, &job->jobnode);
            ellAdd(&pool->owned, &job->jobnode);
        }
        job->queued = 0;
        ret = 0;
    }

    epicsMutexUnlock(pool->guard);

    return ret;
}

LIBCOM_API void epicsThreadPoolConfigDefaults

(

epicsThreadPoolConfig *

)

Definition at line 26 of file threadPool.c.

{
    memset(opts, 0, sizeof(*opts));
    opts->maxThreads = epicsThreadGetCPUs();
    opts->workerStack = epicsThreadGetStackSize(epicsThreadStackSmall);

    if (epicsThreadLowestPriorityLevelAbove(epicsThreadPriorityCAServerHigh, &opts->workerPriority)
            != epicsThreadBooleanStatusSuccess)
        opts->workerPriority = epicsThreadPriorityMedium;
}

LIBCOM_API void epicsThreadPoolControl	(	epicsThreadPool *	pool,
		epicsThreadPoolOption	opt,
		unsigned int	val
	)

Definition at line 152 of file threadPool.c.

{
    epicsMutexMustLock(pool->guard);
    epicsThreadPoolControlImpl(pool, opt, val);
    epicsMutexUnlock(pool->guard);
}

LIBCOM_API epicsThreadPool* epicsThreadPoolCreate

(

epicsThreadPoolConfig *

opts

)

Definition at line 37 of file threadPool.c.

{
    size_t i;
    epicsThreadPool *pool;

    /* caller likely didn't initialize the options structure */
    if (opts && opts->maxThreads == 0) {
        errlogMessage("Error: epicsThreadPoolCreate() options provided, but not initialized");
        return NULL;
    }

    pool = calloc(1, sizeof(*pool));
    if (!pool)
        return NULL;

    if (opts)
        memcpy(&pool->conf, opts, sizeof(*opts));
    else
        epicsThreadPoolConfigDefaults(&pool->conf);

    if (pool->conf.initialThreads > pool->conf.maxThreads)
        pool->conf.initialThreads = pool->conf.maxThreads;

    pool->workerWakeup = epicsEventCreate(epicsEventEmpty);
    pool->shutdownEvent = epicsEventCreate(epicsEventEmpty);
    pool->observerWakeup = epicsEventCreate(epicsEventEmpty);
    pool->guard = epicsMutexCreate();

    if (!pool->workerWakeup || !pool->shutdownEvent ||
       !pool->observerWakeup || !pool->guard)
        goto cleanup;

    ellInit(&pool->jobs);
    ellInit(&pool->owned);

    epicsMutexMustLock(pool->guard);

    for (i = 0; i < pool->conf.initialThreads; i++) {
        createPoolThread(pool);
    }

    if (pool->threadsRunning == 0 && pool->conf.initialThreads != 0) {
        epicsMutexUnlock(pool->guard);
        errlogPrintf("Error: Unable to create any threads for thread pool\n");
        goto cleanup;

    }
    else if (pool->threadsRunning < pool->conf.initialThreads) {
        errlogPrintf("Warning: Unable to create all threads for thread pool (%u/%u)\n",
                     pool->threadsRunning, pool->conf.initialThreads);
    }

    epicsMutexUnlock(pool->guard);

    return pool;

cleanup:
    if (pool->workerWakeup)
        epicsEventDestroy(pool->workerWakeup);
    if (pool->shutdownEvent)
        epicsEventDestroy(pool->shutdownEvent);
    if (pool->observerWakeup)
        epicsEventDestroy(pool->observerWakeup);
    if (pool->guard)
        epicsMutexDestroy(pool->guard);

    free(pool);
    return NULL;
}

LIBCOM_API void epicsThreadPoolDestroy

(

epicsThreadPool *

)

Definition at line 199 of file threadPool.c.

{
    unsigned int nThr;
    ELLLIST notify;
    ELLNODE *cur;

    if (!pool)
        return;

    ellInit(&notify);

    epicsMutexMustLock(pool->guard);

    /* run remaining queued jobs */
    epicsThreadPoolControlImpl(pool, epicsThreadPoolQueueAdd, 0);
    epicsThreadPoolControlImpl(pool, epicsThreadPoolQueueRun, 1);
    nThr = pool->threadsRunning;
    pool->freezeopt = 1;

    epicsMutexUnlock(pool->guard);

    epicsThreadPoolWait(pool, -1.0);
    /* At this point all queued jobs have run */

    epicsMutexMustLock(pool->guard);

    pool->shutdown = 1;
    /* wakeup all */
    if (pool->threadsWaking < pool->threadsSleeping) {
        pool->threadsWaking = pool->threadsSleeping;
        epicsEventSignal(pool->workerWakeup);
    }

    ellConcat(&notify, &pool->owned);
    ellConcat(&notify, &pool->jobs);

    epicsMutexUnlock(pool->guard);

    if (nThr && epicsEventWait(pool->shutdownEvent) != epicsEventWaitOK){
        errlogMessage("epicsThreadPoolDestroy: wait error");
        return;
    }

    /* all workers are now shutdown */

    /* notify remaining jobs that pool is being destroyed */
    while ((cur = ellGet(&notify)) != NULL) {
        epicsJob *job = CONTAINER(cur, epicsJob, jobnode);

        job->running = 1;
        job->func(job->arg, epicsJobModeCleanup);
        job->running = 0;
        if (job->freewhendone)
            free(job);
        else
            job->pool = NULL; /* orphan */
    }

    epicsEventDestroy(pool->workerWakeup);
    epicsEventDestroy(pool->shutdownEvent);
    epicsEventDestroy(pool->observerWakeup);
    epicsMutexDestroy(pool->guard);

    free(pool);
}

LIBCOM_API epicsThreadPool* epicsThreadPoolGetShared

(

epicsThreadPoolConfig *

opts

)

Definition at line 328 of file threadPool.c.

{
    ELLNODE *node;
    epicsThreadPool *cur;
    epicsThreadPoolConfig defopts;
    size_t N = epicsThreadGetCPUs();

    if (!opts) {
        epicsThreadPoolConfigDefaults(&defopts);
        opts = &defopts;
    }
    /* shared pools must have a minimum allowed number of workers.
     * Use the number of CPU cores
     */
    if (opts->maxThreads < N)
        opts->maxThreads = N;

    epicsThreadOnce(&sharedPoolsOnce, &sharedPoolsInit, NULL);

    epicsMutexMustLock(sharedPoolsGuard);

    for (node = ellFirst(&sharedPools); node; node = ellNext(node)) {
        cur = CONTAINER(node, epicsThreadPool, sharedNode);

        /* Must have exactly the requested priority
         * At least the requested max workers
         * and at least the requested stack size
         */
        if (cur->conf.workerPriority != opts->workerPriority)
            continue;
        if (cur->conf.maxThreads < opts->maxThreads)
            continue;
        if (cur->conf.workerStack < opts->workerStack)
            continue;

        cur->sharedCount++;
        assert(cur->sharedCount > 0);
        epicsMutexUnlock(sharedPoolsGuard);

        epicsMutexMustLock(cur->guard);
        *opts = cur->conf;
        epicsMutexUnlock(cur->guard);
        return cur;
    }

    cur = epicsThreadPoolCreate(opts);
    if (!cur) {
        epicsMutexUnlock(sharedPoolsGuard);
        return NULL;
    }
    cur->sharedCount = 1;

    ellAdd(&sharedPools, &cur->sharedNode);
    epicsMutexUnlock(sharedPoolsGuard);
    return cur;
}

LIBCOM_API unsigned int epicsThreadPoolNThreads

(

epicsThreadPool *

)

Definition at line 302 of file threadPool.c.

{
    unsigned int ret;

    epicsMutexMustLock(pool->guard);
    ret = pool->threadsRunning;
    epicsMutexUnlock(pool->guard);

    return ret;
}

LIBCOM_API void epicsThreadPoolReleaseShared

(

epicsThreadPool *

pool

)

Definition at line 385 of file threadPool.c.

{
    if (!pool)
        return;

    epicsMutexMustLock(sharedPoolsGuard);

    assert(pool->sharedCount > 0);

    pool->sharedCount--;

    if (pool->sharedCount == 0) {
        ellDelete(&sharedPools, &pool->sharedNode);
        epicsThreadPoolDestroy(pool);
    }

    epicsMutexUnlock(sharedPoolsGuard);
}

LIBCOM_API void epicsThreadPoolReport	(	epicsThreadPool *	pool,
		FILE *	fd
	)

Definition at line 266 of file threadPool.c.

{
    ELLNODE *cur;
    epicsMutexMustLock(pool->guard);

    fprintf(fd, "Thread Pool with %u/%u threads\n"
            " running %d jobs with %u threads\n",
            pool->threadsRunning,
            pool->conf.maxThreads,
            ellCount(&pool->jobs),
            pool->threadsAreAwake);
    if (pool->pauseadd)
        fprintf(fd, "  Inhibit queueing\n");
    if (pool->pauserun)
        fprintf(fd, "  Pause workers\n");
    if (pool->shutdown)
        fprintf(fd, "  Shutdown in progress\n");

    for (cur = ellFirst(&pool->jobs); cur; cur = ellNext(cur)) {
        epicsJob *job = CONTAINER(cur, epicsJob, jobnode);

        fprintf(fd, "  job %p func: %p, arg: %p ",
                job, job->func,
                job->arg);
        if (job->queued)
            fprintf(fd, "Queued ");
        if (job->running)
            fprintf(fd, "Running ");
        if (job->freewhendone)
            fprintf(fd, "Free ");
        fprintf(fd, "\n");
    }

    epicsMutexUnlock(pool->guard);
}

LIBCOM_API int epicsThreadPoolWait	(	epicsThreadPool *	pool,
		double	timeout
	)

Definition at line 159 of file threadPool.c.

{
    int ret = 0;
    epicsMutexMustLock(pool->guard);

    while (ellCount(&pool->jobs) > 0 || pool->threadsAreAwake > 0) {
        pool->observerCount++;
        epicsMutexUnlock(pool->guard);

        if (timeout < 0.0) {
            epicsEventMustWait(pool->observerWakeup);
        }
        else {
            switch (epicsEventWaitWithTimeout(pool->observerWakeup, timeout)) {
            case epicsEventWaitError:
                cantProceed("epicsThreadPoolWait: failed to wait for Event");
                break;
            case epicsEventWaitTimeout:
                ret = S_pool_timeout;
                break;
            case epicsEventWaitOK:
                ret = 0;
                break;
            }
        }

        epicsMutexMustLock(pool->guard);
        pool->observerCount--;

        if (pool->observerCount)
            epicsEventSignal(pool->observerWakeup);

        if (ret != 0)
            break;
    }

    epicsMutexUnlock(pool->guard);
    return ret;
}

Variable Documentation

LIBCOM_API void* epicsJobArgSelfMagic

Definition at line 24 of file poolJob.c.