source: vendor/current/source3/lib/pthreadpool.c@ 414

/*
 * Unix SMB/CIFS implementation.
 * thread pool implementation
 * Copyright (C) Volker Lendecke 2009
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <errno.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <signal.h>
#include <assert.h>
#include <fcntl.h>

#include "pthreadpool.h"

struct pthreadpool_job {
        struct pthreadpool_job *next;
        int id;
        void (*fn)(void *private_data);
        void *private_data;
};

struct pthreadpool {
        /*
         * Control access to this struct
         */
        pthread_mutex_t mutex;

        /*
         * Threads waiting for work do so here
         */
        pthread_cond_t condvar;

        /*
         * List of work jobs
         */
        struct pthreadpool_job *jobs, *last_job;

        /*
         * pipe for signalling
         */
        int sig_pipe[2];

        /*
         * indicator to worker threads that they should shut down
         */
        int shutdown;

        /*
         * maximum number of threads
         */
        int max_threads;

        /*
         * Number of threads
         */
        int num_threads;

        /*
         * Number of idle threads
         */
        int num_idle;

        /*
         * An array of threads that require joining, the array has
         * "max_threads" elements. It contains "num_exited" ids.
         */
        int                     num_exited;
        pthread_t               exited[1]; /* We alloc more */
};

/*
 * Initialize a thread pool
 */

int pthreadpool_init(unsigned max_threads, struct pthreadpool **presult)
{
        struct pthreadpool *pool;
        size_t size;
        int ret;

        size = sizeof(struct pthreadpool) + max_threads * sizeof(pthread_t);

        pool = (struct pthreadpool *)malloc(size);
        if (pool == NULL) {
                return ENOMEM;
        }

        ret = pthread_mutex_init(&pool->mutex, NULL);
        if (ret != 0) {
                free(pool);
                return ret;
        }

        ret = pthread_cond_init(&pool->condvar, NULL);
        if (ret != 0) {
                pthread_mutex_destroy(&pool->mutex);
                free(pool);
                return ret;
        }

        pool->shutdown = 0;
        pool->jobs = pool->last_job = NULL;
        pool->num_threads = 0;
        pool->num_exited = 0;
        pool->max_threads = max_threads;
        pool->num_idle = 0;
        pool->sig_pipe[0] = -1;
        pool->sig_pipe[1] = -1;

        *presult = pool;
        return 0;
}

/*
 * Create and return a file descriptor which becomes readable when a job has
 * finished
 */

int pthreadpool_sig_fd(struct pthreadpool *pool)
{
        int result, ret;

        ret = pthread_mutex_lock(&pool->mutex);
        if (ret != 0) {
                errno = ret;
                return -1;
        }

        if (pool->sig_pipe[0] != -1) {
                result = pool->sig_pipe[0];
                goto done;
        }

        ret = pipe(pool->sig_pipe);
        if (ret == -1) {
                result = -1;
                goto done;
        }

        result = pool->sig_pipe[0];
done:
        ret = pthread_mutex_unlock(&pool->mutex);
        assert(ret == 0);
        return result;
}

/*
 * Do a pthread_join() on all children that have exited, pool->mutex must be
 * locked
 */
static void pthreadpool_join_children(struct pthreadpool *pool)
{
        int i;

        for (i=0; i<pool->num_exited; i++) {
                pthread_join(pool->exited[i], NULL);
        }
        pool->num_exited = 0;
}

/*
 * Fetch a finished job number from the signal pipe
 */

int pthreadpool_finished_job(struct pthreadpool *pool)
{
        int result, ret, fd;
        ssize_t nread;

        ret = pthread_mutex_lock(&pool->mutex);
        if (ret != 0) {
                errno = ret;
                return -1;
        }

        /*
         * Just some cleanup under the mutex
         */
        pthreadpool_join_children(pool);

        fd = pool->sig_pipe[0];

        ret = pthread_mutex_unlock(&pool->mutex);
        assert(ret == 0);

        if (fd == -1) {
                errno = EINVAL;
                return -1;
        }

        nread = -1;
        errno = EINTR;

        while ((nread == -1) && (errno == EINTR)) {
                nread = read(fd, &result, sizeof(int));
        }

        /*
         * TODO: handle nread > 0 && nread < sizeof(int)
         */

        /*
         * Lock the mutex to provide a memory barrier for data from the worker
         * thread to the main thread. The pipe access itself does not have to
         * be locked, for sizeof(int) the write to a pipe is atomic, and only
         * one thread reads from it. But we need to lock the mutex briefly
         * even if we don't do anything under the lock, to make sure we can
         * see all memory the helper thread has written.
         */

        ret = pthread_mutex_lock(&pool->mutex);
        if (ret == -1) {
                errno = ret;
                return -1;
        }

        ret = pthread_mutex_unlock(&pool->mutex);
        assert(ret == 0);

        return result;
}

/*
 * Destroy a thread pool, finishing all threads working for it
 */

int pthreadpool_destroy(struct pthreadpool *pool)
{
        int ret, ret1;

        ret = pthread_mutex_lock(&pool->mutex);
        if (ret != 0) {
                return ret;
        }

        if (pool->num_threads > 0) {
                /*
                 * We have active threads, tell them to finish, wait for that.
                 */

                pool->shutdown = 1;

                if (pool->num_idle > 0) {
                        /*
                         * Wake the idle threads. They will find pool->quit to
                         * be set and exit themselves
                         */
                        ret = pthread_cond_broadcast(&pool->condvar);
                        if (ret != 0) {
                                pthread_mutex_unlock(&pool->mutex);
                                return ret;
                        }
                }