port/tpool/thread_pool.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

#pragma ident   "%Z%%M% %I% %E% SMI"

#include "lint.h"
#include "thr_uberdata.h"
#include <stdlib.h>
#include <signal.h>
#include <errno.h>
#include "thread_pool_impl.h"

static mutex_t thread_pool_lock = DEFAULTMUTEX;
static tpool_t *thread_pools = NULL;

static void
delete_pool(tpool_t *tpool)
{
    tpool_job_t *job;

    ASSERT(tpool->tp_current == 0 && tpool->tp_active == NULL);

    /*
     * Unlink the pool from the global list of all pools.
     */
    lmutex_lock(&thread_pool_lock);
    if (thread_pools == tpool)
        thread_pools = tpool->tp_forw;
    if (thread_pools == tpool)
        thread_pools = NULL;
    else {
        tpool->tp_back->tp_forw = tpool->tp_forw;
        tpool->tp_forw->tp_back = tpool->tp_back;
    }
    lmutex_unlock(&thread_pool_lock);

    /*
     * There should be no pending jobs, but just in case...
     */
    for (job = tpool->tp_head; job != NULL; job = tpool->tp_head) {
        tpool->tp_head = job->tpj_next;
        lfree(job, sizeof (*job));
    }
    (void) pthread_attr_destroy(&tpool->tp_attr);
    lfree(tpool, sizeof (*tpool));
}

/*
 * Worker thread is terminating.
 */
static void
worker_cleanup(tpool_t *tpool)
{
    ASSERT(MUTEX_HELD(&tpool->tp_mutex));

    if (--tpool->tp_current == 0 &&
        (tpool->tp_flags & (TP_DESTROY | TP_ABANDON))) {
        if (tpool->tp_flags & TP_ABANDON) {
            sig_mutex_unlock(&tpool->tp_mutex);
            delete_pool(tpool);
            return;
        }
        if (tpool->tp_flags & TP_DESTROY)
            (void) cond_broadcast(&tpool->tp_busycv);
    }
    sig_mutex_unlock(&tpool->tp_mutex);
}

static void
notify_waiters(tpool_t *tpool)
{
    if (tpool->tp_head == NULL && tpool->tp_active == NULL) {
        tpool->tp_flags &= ~TP_WAIT;
        (void) cond_broadcast(&tpool->tp_waitcv);
    }
}

/*
 * Called by a worker thread on return from a tpool_dispatch()d job.
 */
static void
job_cleanup(tpool_t *tpool)
{
    pthread_t my_tid = pthread_self();
    tpool_active_t *activep;
    tpool_active_t **activepp;

    sig_mutex_lock(&tpool->tp_mutex);
    /* CSTYLED */
    for (activepp = &tpool->tp_active;; activepp = &activep->tpa_next) {
        activep = *activepp;
        if (activep->tpa_tid == my_tid) {
            *activepp = activep->tpa_next;
            break;
        }
    }
    if (tpool->tp_flags & TP_WAIT)
        notify_waiters(tpool);
}

static void *
tpool_worker(void *arg)
{
    tpool_t *tpool = (tpool_t *)arg;
    int elapsed;
    tpool_job_t *job;
    void (*func)(void *);
    tpool_active_t active;

    sig_mutex_lock(&tpool->tp_mutex);
    pthread_cleanup_push(worker_cleanup, tpool);

    /*
     * This is the worker's main loop.
     * It will only be left if a timeout or an error has occured.
     */
    active.tpa_tid = pthread_self();
    for (;;) {
        elapsed = 0;
        tpool->tp_idle++;
        if (tpool->tp_flags & TP_WAIT)
            notify_waiters(tpool);
        while ((tpool->tp_head == NULL ||
            (tpool->tp_flags & TP_SUSPEND)) &&
            !(tpool->tp_flags & (TP_DESTROY | TP_ABANDON))) {
            if (tpool->tp_current <= tpool->tp_minimum ||
                tpool->tp_linger == 0) {
                (void) sig_cond_wait(&tpool->tp_workcv,
                    &tpool->tp_mutex);
            } else {
                timestruc_t timeout;

                timeout.tv_sec = tpool->tp_linger;
                timeout.tv_nsec = 0;
                if (sig_cond_reltimedwait(&tpool->tp_workcv,
                    &tpool->tp_mutex, &timeout) != 0) {
                    elapsed = 1;
                    break;
                }
            }
        }
        tpool->tp_idle--;
        if (tpool->tp_flags & TP_DESTROY)
            break;
        if (tpool->tp_flags & TP_ABANDON) {
            /* can't abandon a suspended pool */
            if (tpool->tp_flags & TP_SUSPEND) {
                tpool->tp_flags &= ~TP_SUSPEND;
                (void) cond_broadcast(&tpool->tp_workcv);
            }
            if (tpool->tp_head == NULL)
                break;
        }
        if ((job = tpool->tp_head) != NULL &&
            !(tpool->tp_flags & TP_SUSPEND)) {
            elapsed = 0;
            func = job->tpj_func;
            arg = job->tpj_arg;
            tpool->tp_head = job->tpj_next;
            if (job == tpool->tp_tail)
                tpool->tp_tail = NULL;
            tpool->tp_njobs--;
            active.tpa_next = tpool->tp_active;
            tpool->tp_active = &active;
            sig_mutex_unlock(&tpool->tp_mutex);
            pthread_cleanup_push(job_cleanup, tpool);
            lfree(job, sizeof (*job));
            /*
             * Call the specified function.
             */
            func(arg);
            /*
             * We don't know what this thread has been doing,
             * so we reset its signal mask and cancellation
             * state back to the initial values.
             */
            (void) pthread_sigmask(SIG_SETMASK, &maskset, NULL);
            (void) pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,
                NULL);
            (void) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,
                NULL);
            pthread_cleanup_pop(1);
        }
        if (elapsed && tpool->tp_current > tpool->tp_minimum) {
            /*
             * We timed out and there is no work to be done
             * and the number of workers exceeds the minimum.
             * Exit now to reduce the size of the pool.
             */
            break;
        }
    }
    pthread_cleanup_pop(1);
    return (arg);
}

/*
 * Create a worker thread, with all signals blocked.
 */
static int
create_worker(tpool_t *tpool)
{
    sigset_t oset;
    int error;

    (void) pthread_sigmask(SIG_SETMASK, &maskset, &oset);
    error = pthread_create(NULL, &tpool->tp_attr, tpool_worker, tpool);
    (void) pthread_sigmask(SIG_SETMASK, &oset, NULL);
    return (error);
}

tpool_t *
tpool_create(uint_t min_threads, uint_t max_threads, uint_t linger,
    pthread_attr_t *attr)
{
    tpool_t *tpool;
    void *stackaddr;
    size_t stacksize;
    size_t minstack;
    int error;

    if (min_threads > max_threads || max_threads < 1) {
        errno = EINVAL;
        return (NULL);
    }
    if (attr != NULL) {
        if (pthread_attr_getstack(attr, &stackaddr, &stacksize) != 0) {
            errno = EINVAL;
            return (NULL);
        }
        /*
         * Allow only one thread in the pool with a specified stack.
         * Require threads to have at least the minimum stack size.
         */
        minstack = thr_min_stack();
        if (stackaddr != NULL) {
            if (stacksize < minstack || max_threads != 1) {
                errno = EINVAL;
                return (NULL);
            }
        } else if (stacksize != 0 && stacksize < minstack) {
            errno = EINVAL;
            return (NULL);
        }
    }

    tpool = lmalloc(sizeof (*tpool));
    if (tpool == NULL) {
        errno = ENOMEM;
        return (NULL);
    }
    (void) mutex_init(&tpool->tp_mutex, USYNC_THREAD, NULL);
    (void) cond_init(&tpool->tp_busycv, USYNC_THREAD, NULL);
    (void) cond_init(&tpool->tp_workcv, USYNC_THREAD, NULL);
    (void) cond_init(&tpool->tp_waitcv, USYNC_THREAD, NULL);
    tpool->tp_minimum = min_threads;
    tpool->tp_maximum = max_threads;
    tpool->tp_linger = linger;

    /*
     * We cannot just copy the attribute pointer.
     * We need to initialize a new pthread_attr_t structure
     * with the values from the user-supplied pthread_attr_t.
     * If the attribute pointer is NULL, we need to initialize
     * the new pthread_attr_t structure with default values.
     */
    error = pthread_attr_clone(&tpool->tp_attr, attr);
    if (error) {
        lfree(tpool, sizeof (*tpool));
        errno = error;
        return (NULL);
    }

    /* make all pool threads be detached daemon threads */
    (void) pthread_attr_setdetachstate(&tpool->tp_attr,
        PTHREAD_CREATE_DETACHED);
    (void) pthread_attr_setdaemonstate_np(&tpool->tp_attr,
        PTHREAD_CREATE_DAEMON_NP);

    /* insert into the global list of all thread pools */
    lmutex_lock(&thread_pool_lock);
    if (thread_pools == NULL) {
        tpool->tp_forw = tpool;
        tpool->tp_back = tpool;
        thread_pools = tpool;
    } else {
        thread_pools->tp_back->tp_forw = tpool;
        tpool->tp_forw = thread_pools;
        tpool->tp_back = thread_pools->tp_back;
        thread_pools->tp_back = tpool;
    }
    lmutex_unlock(&thread_pool_lock);

    return (tpool);
}

/*
 * Dispatch a work request to the thread pool.
 * If there are idle workers, awaken one.
 * Else, if the maximum number of workers has
 * not been reached, spawn a new worker thread.
 * Else just return with the job added to the queue.
 */
int
tpool_dispatch(tpool_t *tpool, void (*func)(void *), void *arg)
{
    tpool_job_t *job;

    ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));

    if ((job = lmalloc(sizeof (*job))) == NULL)
        return (-1);
    job->tpj_next = NULL;
    job->tpj_func = func;
    job->tpj_arg = arg;

    sig_mutex_lock(&tpool->tp_mutex);

    if (tpool->tp_head == NULL)
        tpool->tp_head = job;
    else
        tpool->tp_tail->tpj_next = job;
    tpool->tp_tail = job;
    tpool->tp_njobs++;

    if (!(tpool->tp_flags & TP_SUSPEND)) {
        if (tpool->tp_idle > 0)
            (void) cond_signal(&tpool->tp_workcv);
        else if (tpool->tp_current < tpool->tp_maximum &&
            create_worker(tpool) == 0)
            tpool->tp_current++;
    }

    sig_mutex_unlock(&tpool->tp_mutex);
    return (0);
}

/*
 * Assumes: by the time tpool_destroy() is called no one will use this
 * thread pool in any way and no one will try to dispatch entries to it.
 * Calling tpool_destroy() from a job in the pool will cause deadlock.
 */
void
tpool_destroy(tpool_t *tpool)
{
    tpool_active_t *activep;

    ASSERT(!tpool_member(tpool));
    ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));

    sig_mutex_lock(&tpool->tp_mutex);
    pthread_cleanup_push(sig_mutex_unlock, &tpool->tp_mutex);

    /* mark the pool as being destroyed; wakeup idle workers */
    tpool->tp_flags |= TP_DESTROY;
    tpool->tp_flags &= ~TP_SUSPEND;
    (void) cond_broadcast(&tpool->tp_workcv);

    /* cancel all active workers */
    for (activep = tpool->tp_active; activep; activep = activep->tpa_next)
        (void) pthread_cancel(activep->tpa_tid);

    /* wait for all active workers to finish */
    while (tpool->tp_active != NULL) {
        tpool->tp_flags |= TP_WAIT;
        (void) sig_cond_wait(&tpool->tp_waitcv, &tpool->tp_mutex);
    }

    /* the last worker to terminate will wake us up */
    while (tpool->tp_current != 0)
        (void) sig_cond_wait(&tpool->tp_busycv, &tpool->tp_mutex);

    pthread_cleanup_pop(1); /* sig_mutex_unlock(&tpool->tp_mutex); */
    delete_pool(tpool);
}

/*
 * Like tpool_destroy(), but don't cancel workers or wait for them to finish.
 * The last worker to terminate will delete the pool.
 */
void
tpool_abandon(tpool_t *tpool)
{
    ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));

    sig_mutex_lock(&tpool->tp_mutex);
    if (tpool->tp_current == 0) {
        /* no workers, just delete the pool */
        sig_mutex_unlock(&tpool->tp_mutex);
        delete_pool(tpool);
    } else {
        /* wake up all workers, last one will delete the pool */
        tpool->tp_flags |= TP_ABANDON;
        tpool->tp_flags &= ~TP_SUSPEND;
        (void) cond_broadcast(&tpool->tp_workcv);
        sig_mutex_unlock(&tpool->tp_mutex);
    }
}

/*
 * Wait for all jobs to complete.
 * Calling tpool_wait() from a job in the pool will cause deadlock.
 */
void
tpool_wait(tpool_t *tpool)
{
    ASSERT(!tpool_member(tpool));
    ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));

    sig_mutex_lock(&tpool->tp_mutex);
    pthread_cleanup_push(sig_mutex_unlock, &tpool->tp_mutex);
    while (tpool->tp_head != NULL || tpool->tp_active != NULL) {
        tpool->tp_flags |= TP_WAIT;
        (void) sig_cond_wait(&tpool->tp_waitcv, &tpool->tp_mutex);
        ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));
    }
    pthread_cleanup_pop(1); /* sig_mutex_unlock(&tpool->tp_mutex); */
}

void
tpool_suspend(tpool_t *tpool)
{
    ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));

    sig_mutex_lock(&tpool->tp_mutex);
    tpool->tp_flags |= TP_SUSPEND;
    sig_mutex_unlock(&tpool->tp_mutex);
}

int
tpool_suspended(tpool_t *tpool)
{
    int suspended;

    ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));

    sig_mutex_lock(&tpool->tp_mutex);
    suspended = (tpool->tp_flags & TP_SUSPEND) != 0;
    sig_mutex_unlock(&tpool->tp_mutex);

    return (suspended);
}

void
tpool_resume(tpool_t *tpool)
{
    int excess;

    ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));

    sig_mutex_lock(&tpool->tp_mutex);
    if (!(tpool->tp_flags & TP_SUSPEND)) {
        sig_mutex_unlock(&tpool->tp_mutex);
        return;
    }
    tpool->tp_flags &= ~TP_SUSPEND;
    (void) cond_broadcast(&tpool->tp_workcv);
    excess = tpool->tp_njobs - tpool->tp_idle;
    while (excess-- > 0 && tpool->tp_current < tpool->tp_maximum) {
        if (create_worker(tpool) != 0)
            break;      /* pthread_create() failed */
        tpool->tp_current++;
    }
    sig_mutex_unlock(&tpool->tp_mutex);
}

int
tpool_member(tpool_t *tpool)
{
    pthread_t my_tid = pthread_self();
    tpool_active_t *activep;

    ASSERT(!(tpool->tp_flags & (TP_DESTROY | TP_ABANDON)));

    sig_mutex_lock(&tpool->tp_mutex);
    for (activep = tpool->tp_active; activep; activep = activep->tpa_next) {
        if (activep->tpa_tid == my_tid) {
            sig_mutex_unlock(&tpool->tp_mutex);
            return (1);
        }
    }
    sig_mutex_unlock(&tpool->tp_mutex);
    return (0);
}

void
postfork1_child_tpool(void)
{
    pthread_t my_tid = pthread_self();
    tpool_t *tpool;
    tpool_job_t *job;

    /*
     * All of the thread pool workers are gone, except possibly
     * for the current thread, if it is a thread pool worker thread.
     * Retain the thread pools, but make them all empty.  Whatever
     * jobs were queued or running belong to the parent process.
     */
top:
    if ((tpool = thread_pools) == NULL)
        return;

    do {
        tpool_active_t *activep;

        (void) mutex_init(&tpool->tp_mutex, USYNC_THREAD, NULL);
        (void) cond_init(&tpool->tp_busycv, USYNC_THREAD, NULL);
        (void) cond_init(&tpool->tp_workcv, USYNC_THREAD, NULL);
        (void) cond_init(&tpool->tp_waitcv, USYNC_THREAD, NULL);
        for (job = tpool->tp_head; job; job = tpool->tp_head) {
            tpool->tp_head = job->tpj_next;
            lfree(job, sizeof (*job));
        }
        tpool->tp_tail = NULL;
        tpool->tp_njobs = 0;
        for (activep = tpool->tp_active; activep;
            activep = activep->tpa_next) {
            if (activep->tpa_tid == my_tid) {
                activep->tpa_next = NULL;
                break;
            }
        }
        tpool->tp_idle = 0;
        tpool->tp_current = 0;
        if ((tpool->tp_active = activep) != NULL)
            tpool->tp_current = 1;
        tpool->tp_flags &= ~TP_WAIT;
        if (tpool->tp_flags & (TP_DESTROY | TP_ABANDON)) {
            tpool->tp_flags &= ~TP_DESTROY;
            tpool->tp_flags |= TP_ABANDON;
            if (tpool->tp_current == 0) {
                delete_pool(tpool);
                goto top;   /* start over */
            }
        }
    } while ((tpool = tpool->tp_forw) != thread_pools);
}