threadPool.c File Reference

#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include "dbDefs.h"
#include "errlog.h"
#include "ellLib.h"
#include "epicsThread.h"
#include "epicsMutex.h"
#include "epicsEvent.h"
#include "epicsInterrupt.h"
#include "cantProceed.h"
#include "epicsThreadPool.h"
#include "poolPriv.h"

Include dependency graph for threadPool.c:

Go to the source code of this file.

Functions
void	epicsThreadPoolConfigDefaults (epicsThreadPoolConfig *opts)
epicsThreadPool *	epicsThreadPoolCreate (epicsThreadPoolConfig *opts)
void	epicsThreadPoolControl (epicsThreadPool *pool, epicsThreadPoolOption opt, unsigned int val)
int	epicsThreadPoolWait (epicsThreadPool *pool, double timeout)
void	epicsThreadPoolDestroy (epicsThreadPool *pool)
void	epicsThreadPoolReport (epicsThreadPool *pool, FILE *fd)
unsigned int	epicsThreadPoolNThreads (epicsThreadPool *pool)
LIBCOM_API epicsThreadPool *	epicsThreadPoolGetShared (epicsThreadPoolConfig *opts)
LIBCOM_API void	epicsThreadPoolReleaseShared (epicsThreadPool *pool)

Function Documentation

void epicsThreadPoolConfigDefaults

(

epicsThreadPoolConfig *

opts

)

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;
}

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);
}

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;
}

void epicsThreadPoolDestroy

(

epicsThreadPool *

pool

)

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;
}

unsigned int epicsThreadPoolNThreads

(

epicsThreadPool *

pool

)

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);
}

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);
}

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;
}