From 62689d8166b8e070f855e6910470796dd7e1b2c8 Mon Sep 17 00:00:00 2001 From: Volker Lendecke Date: Fri, 22 Apr 2011 11:47:11 +0200 Subject: s3: Many pthreadpool fixes In particular, this makes it fork-safe --- source3/lib/pthreadpool/Makefile | 9 + source3/lib/pthreadpool/pthreadpool.c | 592 ++++++++++++++++++++++++++++++++++ source3/lib/pthreadpool/pthreadpool.h | 42 +++ source3/lib/pthreadpool/tests.c | 362 +++++++++++++++++++++ 4 files changed, 1005 insertions(+) create mode 100644 source3/lib/pthreadpool/Makefile create mode 100644 source3/lib/pthreadpool/pthreadpool.c create mode 100644 source3/lib/pthreadpool/pthreadpool.h create mode 100644 source3/lib/pthreadpool/tests.c (limited to 'source3/lib/pthreadpool') diff --git a/source3/lib/pthreadpool/Makefile b/source3/lib/pthreadpool/Makefile new file mode 100644 index 0000000000..48626bd2c0 --- /dev/null +++ b/source3/lib/pthreadpool/Makefile @@ -0,0 +1,9 @@ +all: tests + +CFLAGS=-O3 -g -Wall + +pthreadpool.o: pthreadpool.c pthreadpool.h + gcc -c -O3 -o pthreadpool.o pthreadpool.c -I../../.. + +tests: tests.o pthreadpool.o + gcc -o tests tests.o pthreadpool.o -lpthread \ No newline at end of file diff --git a/source3/lib/pthreadpool/pthreadpool.c b/source3/lib/pthreadpool/pthreadpool.c new file mode 100644 index 0000000000..4605538cf2 --- /dev/null +++ b/source3/lib/pthreadpool/pthreadpool.c @@ -0,0 +1,592 @@ +/* + * 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 . + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "pthreadpool.h" +#include "lib/util/dlinklist.h" + +struct pthreadpool_job { + struct pthreadpool_job *next; + int id; + void (*fn)(void *private_data); + void *private_data; +}; + +struct pthreadpool { + /* + * List pthreadpools for fork safety + */ + struct pthreadpool *prev, *next; + + /* + * 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 */ +}; + +static pthread_mutex_t pthreadpools_mutex = PTHREAD_MUTEX_INITIALIZER; +static struct pthreadpool *pthreadpools = NULL; +static pthread_once_t pthreadpool_atfork_initialized = PTHREAD_ONCE_INIT; + +static void pthreadpool_prep_atfork(void); + +/* + * 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-1) * sizeof(pthread_t); + + pool = (struct pthreadpool *)malloc(size); + if (pool == NULL) { + return ENOMEM; + } + + ret = pipe(pool->sig_pipe); + if (ret == -1) { + int err = errno; + free(pool); + return err; + } + + 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; + + ret = pthread_mutex_lock(&pthreadpools_mutex); + if (ret != 0) { + pthread_cond_destroy(&pool->condvar); + pthread_mutex_destroy(&pool->mutex); + free(pool); + return ret; + } + DLIST_ADD(pthreadpools, pool); + + ret = pthread_mutex_unlock(&pthreadpools_mutex); + assert(ret == 0); + + pthread_once(&pthreadpool_atfork_initialized, pthreadpool_prep_atfork); + + *presult = pool; + + return 0; +} + +static void pthreadpool_prepare(void) +{ + int ret; + struct pthreadpool *pool; + + ret = pthread_mutex_lock(&pthreadpools_mutex); + assert(ret == 0); + + pool = pthreadpools; + + while (pool != NULL) { + ret = pthread_mutex_lock(&pool->mutex); + assert(ret == 0); + pool = pool->next; + } +} + +static void pthreadpool_parent(void) +{ + int ret; + struct pthreadpool *pool; + + pool = DLIST_TAIL(pthreadpools); + + while (1) { + ret = pthread_mutex_unlock(&pool->mutex); + assert(ret == 0); + + if (pool == pthreadpools) { + break; + } + pool = pool->prev; + } + + ret = pthread_mutex_unlock(&pthreadpools_mutex); + assert(ret == 0); +} + +static void pthreadpool_child(void) +{ + int ret; + struct pthreadpool *pool; + + pool = DLIST_TAIL(pthreadpools); + + while (1) { + close(pool->sig_pipe[0]); + close(pool->sig_pipe[1]); + + ret = pipe(pool->sig_pipe); + assert(ret == 0); + + pool->num_threads = 0; + pool->num_exited = 0; + pool->num_idle = 0; + + while (pool->jobs != NULL) { + struct pthreadpool_job *job; + job = pool->jobs; + pool->jobs = job->next; + free(job); + } + pool->last_job = NULL; + + ret = pthread_mutex_unlock(&pool->mutex); + assert(ret == 0); + + if (pool == pthreadpools) { + break; + } + pool = pool->prev; + } + + ret = pthread_mutex_unlock(&pthreadpools_mutex); + assert(ret == 0); +} + +static void pthreadpool_prep_atfork(void) +{ + pthread_atfork(pthreadpool_prepare, pthreadpool_parent, + pthreadpool_child); +} + +/* + * Return the file descriptor which becomes readable when a job has + * finished + */ + +int pthreadpool_sig_fd(struct pthreadpool *pool) +{ + return pool->sig_pipe[0]; +} + +/* + * 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; inum_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; + ssize_t nread; + + nread = -1; + errno = EINTR; + + while ((nread == -1) && (errno == EINTR)) { + nread = read(pool->sig_pipe[0], &result, sizeof(int)); + } + if (nread == -1) { + return errno; + } + if (nread != sizeof(int)) { + return EINVAL; + } + 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->jobs != NULL) || pool->shutdown) { + ret = pthread_mutex_unlock(&pool->mutex); + assert(ret == 0); + return EBUSY; + } + + 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; + } + } + + while ((pool->num_threads > 0) || (pool->num_exited > 0)) { + + if (pool->num_exited > 0) { + pthreadpool_join_children(pool); + continue; + } + /* + * A thread that shuts down will also signal + * pool->condvar + */ + ret = pthread_cond_wait(&pool->condvar, &pool->mutex); + if (ret != 0) { + pthread_mutex_unlock(&pool->mutex); + return ret; + } + } + } + + ret = pthread_mutex_unlock(&pool->mutex); + if (ret != 0) { + return ret; + } + ret = pthread_mutex_destroy(&pool->mutex); + ret1 = pthread_cond_destroy(&pool->condvar); + + if (ret != 0) { + return ret; + } + if (ret1 != 0) { + return ret1; + } + + ret = pthread_mutex_lock(&pthreadpools_mutex); + if (ret != 0) { + return ret; + } + DLIST_REMOVE(pthreadpools, pool); + ret = pthread_mutex_unlock(&pthreadpools_mutex); + assert(ret == 0); + + close(pool->sig_pipe[0]); + pool->sig_pipe[0] = -1; + + close(pool->sig_pipe[1]); + pool->sig_pipe[1] = -1; + + free(pool); + + return 0; +} + +/* + * Prepare for pthread_exit(), pool->mutex must be locked + */ +static void pthreadpool_server_exit(struct pthreadpool *pool) +{ + pool->num_threads -= 1; + pool->exited[pool->num_exited] = pthread_self(); + pool->num_exited += 1; +} + +static void *pthreadpool_server(void *arg) +{ + struct pthreadpool *pool = (struct pthreadpool *)arg; + int res; + + res = pthread_mutex_lock(&pool->mutex); + if (res != 0) { + return NULL; + } + + while (1) { + struct timeval tv; + struct timespec ts; + struct pthreadpool_job *job; + + /* + * idle-wait at most 1 second. If nothing happens in that + * time, exit this thread. + */ + + gettimeofday(&tv, NULL); + ts.tv_sec = tv.tv_sec + 1; + ts.tv_nsec = tv.tv_usec*1000; + + while ((pool->jobs == NULL) && (pool->shutdown == 0)) { + + pool->num_idle += 1; + res = pthread_cond_timedwait( + &pool->condvar, &pool->mutex, &ts); + pool->num_idle -= 1; + + if (res == ETIMEDOUT) { + + if (pool->jobs == NULL) { + /* + * we timed out and still no work for + * us. Exit. + */ + pthreadpool_server_exit(pool); + pthread_mutex_unlock(&pool->mutex); + return NULL; + } + + break; + } + assert(res == 0); + } + + job = pool->jobs; + + if (job != NULL) { + ssize_t written; + + /* + * Ok, there's work for us to do, remove the job from + * the pthreadpool list + */ + pool->jobs = job->next; + if (pool->last_job == job) { + pool->last_job = NULL; + } + + /* + * Do the work with the mutex unlocked + */ + + res = pthread_mutex_unlock(&pool->mutex); + assert(res == 0); + + job->fn(job->private_data); + + written = write(pool->sig_pipe[1], &job->id, + sizeof(int)); + + free(job); + + res = pthread_mutex_lock(&pool->mutex); + assert(res == 0); + + if (written != sizeof(int)) { + pthreadpool_server_exit(pool); + pthread_mutex_unlock(&pool->mutex); + return NULL; + } + } + + if ((pool->jobs == NULL) && (pool->shutdown != 0)) { + /* + * No more work to do and we're asked to shut down, so + * exit + */ + pthreadpool_server_exit(pool); + + if (pool->num_threads == 0) { + /* + * Ping the main thread waiting for all of us + * workers to have quit. + */ + pthread_cond_broadcast(&pool->condvar); + } + + pthread_mutex_unlock(&pool->mutex); + return NULL; + } + } +} + +int pthreadpool_add_job(struct pthreadpool *pool, int job_id, + void (*fn)(void *private_data), void *private_data) +{ + struct pthreadpool_job *job; + pthread_t thread_id; + int res; + sigset_t mask, omask; + + job = (struct pthreadpool_job *)malloc(sizeof(struct pthreadpool_job)); + if (job == NULL) { + return ENOMEM; + } + + job->fn = fn; + job->private_data = private_data; + job->id = job_id; + job->next = NULL; + + res = pthread_mutex_lock(&pool->mutex); + if (res != 0) { + free(job); + return res; + } + + if (pool->shutdown) { + /* + * Protect against the pool being shut down while + * trying to add a job + */ + res = pthread_mutex_unlock(&pool->mutex); + assert(res == 0); + free(job); + return EINVAL; + } + + /* + * Just some cleanup under the mutex + */ + pthreadpool_join_children(pool); + + /* + * Add job to the end of the queue + */ + if (pool->jobs == NULL) { + pool->jobs = job; + } + else { + pool->last_job->next = job; + } + pool->last_job = job; + + if (pool->num_idle > 0) { + /* + * We have idle threads, wake one. + */ + res = pthread_cond_signal(&pool->condvar); + pthread_mutex_unlock(&pool->mutex); + return res; + } + + if (pool->num_threads >= pool->max_threads) { + /* + * No more new threads, we just queue the request + */ + pthread_mutex_unlock(&pool->mutex); + return 0; + } + + /* + * Create a new worker thread. It should not receive any signals. + */ + + sigfillset(&mask); + + res = pthread_sigmask(SIG_BLOCK, &mask, &omask); + if (res != 0) { + pthread_mutex_unlock(&pool->mutex); + return res; + } + + res = pthread_create(&thread_id, NULL, pthreadpool_server, + (void *)pool); + if (res == 0) { + pool->num_threads += 1; + } + + assert(pthread_sigmask(SIG_SETMASK, &omask, NULL) == 0); + + pthread_mutex_unlock(&pool->mutex); + return res; +} diff --git a/source3/lib/pthreadpool/pthreadpool.h b/source3/lib/pthreadpool/pthreadpool.h new file mode 100644 index 0000000000..7ef7ddf419 --- /dev/null +++ b/source3/lib/pthreadpool/pthreadpool.h @@ -0,0 +1,42 @@ +/* + * Unix SMB/CIFS implementation. + * threadpool implementation based on pthreads + * 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 . + */ + +#ifndef __PTHREADPOOL_H__ +#define __PTHREADPOOL_H__ + +struct pthreadpool; + +int pthreadpool_init(unsigned max_threads, struct pthreadpool **presult); +int pthreadpool_destroy(struct pthreadpool *pool); + +/* + * Add a job to a pthreadpool. + */ +int pthreadpool_add_job(struct pthreadpool *pool, int job_id, + void (*fn)(void *private_data), void *private_data); + +/* + * Get the signalling fd out of a thread pool. This fd will become readable + * when a job is finished. The job that finished can be retrieved via + * pthreadpool_finished_job(). + */ +int pthreadpool_sig_fd(struct pthreadpool *pool); +int pthreadpool_finished_job(struct pthreadpool *pool); + +#endif diff --git a/source3/lib/pthreadpool/tests.c b/source3/lib/pthreadpool/tests.c new file mode 100644 index 0000000000..d365fbd5b6 --- /dev/null +++ b/source3/lib/pthreadpool/tests.c @@ -0,0 +1,362 @@ +#include +#include +#include +#include +#include +#include +#include +#include "pthreadpool.h" + +static int test_init(void) +{ + struct pthreadpool *p; + int ret; + + ret = pthreadpool_init(1, &p); + if (ret != 0) { + fprintf(stderr, "pthreadpool_init failed: %s\n", + strerror(ret)); + return -1; + } + ret = pthreadpool_destroy(p); + if (ret != 0) { + fprintf(stderr, "pthreadpool_init failed: %s\n", + strerror(ret)); + return -1; + } + return 0; +} + +static void test_sleep(void *ptr) +{ + int *ptimeout = (int *)ptr; + int ret; + ret = poll(NULL, 0, *ptimeout); + if (ret != 0) { + fprintf(stderr, "poll returned %d (%s)\n", + ret, strerror(errno)); + } +} + +static int test_jobs(int num_threads, int num_jobs) +{ + char *finished; + struct pthreadpool *p; + int timeout = 1; + int i, ret; + + finished = (char *)calloc(1, num_jobs); + if (finished == NULL) { + fprintf(stderr, "calloc failed\n"); + return -1; + } + + ret = pthreadpool_init(num_threads, &p); + if (ret != 0) { + fprintf(stderr, "pthreadpool_init failed: %s\n", + strerror(ret)); + return -1; + } + + for (i=0; i= num_jobs)) { + fprintf(stderr, "invalid job number %d\n", ret); + return -1; + } + finished[ret] += 1; + } + + for (i=0; inum_jobs; i++) { + int ret = pthreadpool_add_job(state->p, state->start_job + i, + test_sleep, &state->timeout); + if (ret != 0) { + fprintf(stderr, "pthreadpool_add_job failed: %s\n", + strerror(ret)); + return NULL; + } + } + return NULL; +} + +static int test_threaded_addjob(int num_pools, int num_threads, int poolsize, + int num_jobs) +{ + struct pthreadpool **pools; + struct threaded_state *states; + struct threaded_state *state; + struct pollfd *pfds; + char *finished; + pid_t child; + int i, ret, poolnum; + int received; + + states = calloc(num_threads, sizeof(struct threaded_state)); + if (states == NULL) { + fprintf(stderr, "calloc failed\n"); + return -1; + } + + finished = calloc(num_threads * num_jobs, 1); + if (finished == NULL) { + fprintf(stderr, "calloc failed\n"); + return -1; + } + + pools = calloc(num_pools, sizeof(struct pthreadpool *)); + if (pools == NULL) { + fprintf(stderr, "calloc failed\n"); + return -1; + } + + pfds = calloc(num_pools, sizeof(struct pollfd)); + if (pfds == NULL) { + fprintf(stderr, "calloc failed\n"); + return -1; + } + + for (i=0; ip = pools[poolnum]; + poolnum = (poolnum + 1) % num_pools; + + state->num_jobs = num_jobs; + state->timeout = 1; + state->start_job = i * num_jobs; + + ret = pthread_create(&state->tid, NULL, test_threaded_worker, + state); + if (ret != 0) { + fprintf(stderr, "pthread_create failed: %s\n", + strerror(ret)); + return -1; + } + } + + if (random() % 1) { + poll(NULL, 0, 1); + } + + child = fork(); + if (child < 0) { + fprintf(stderr, "fork failed: %s\n", strerror(errno)); + return -1; + } + if (child == 0) { + for (i=0; i= num_jobs * num_threads)) { + fprintf(stderr, "invalid job number %d\n", + ret); + return -1; + } + finished[ret] += 1; + received += 1; + } + } + + for (i=0; i