summaryrefslogtreecommitdiff
path: root/source3/lib/pthreadpool
diff options
context:
space:
mode:
Diffstat (limited to 'source3/lib/pthreadpool')
-rw-r--r--source3/lib/pthreadpool/Makefile9
-rw-r--r--source3/lib/pthreadpool/pthreadpool.c615
-rw-r--r--source3/lib/pthreadpool/pthreadpool.h97
-rw-r--r--source3/lib/pthreadpool/tests.c362
-rw-r--r--source3/lib/pthreadpool/wscript_build11
5 files changed, 1094 insertions, 0 deletions
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..9981ed25ae
--- /dev/null
+++ b/source3/lib/pthreadpool/pthreadpool.c
@@ -0,0 +1,615 @@
+/*
+ * 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 <sys/time.h>
+
+#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.
+ */
+ int num_exited;
+ pthread_t *exited; /* 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;
+ int ret;
+
+ pool = (struct pthreadpool *)malloc(sizeof(struct pthreadpool));
+ 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) {
+ close(pool->sig_pipe[0]);
+ close(pool->sig_pipe[1]);
+ free(pool);
+ return ret;
+ }
+
+ ret = pthread_cond_init(&pool->condvar, NULL);
+ if (ret != 0) {
+ pthread_mutex_destroy(&pool->mutex);
+ close(pool->sig_pipe[0]);
+ close(pool->sig_pipe[1]);
+ free(pool);
+ return ret;
+ }
+
+ pool->shutdown = 0;
+ pool->jobs = pool->last_job = NULL;
+ pool->num_threads = 0;
+ pool->num_exited = 0;
+ pool->exited = NULL;
+ 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);
+ close(pool->sig_pipe[0]);
+ close(pool->sig_pipe[1]);
+ 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;
+ free(pool->exited);
+ pool->exited = NULL;
+
+ 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_signal_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; i<pool->num_exited; i++) {
+ pthread_join(pool->exited[i], NULL);
+ }
+ pool->num_exited = 0;
+
+ /*
+ * Deliberately not free and NULL pool->exited. That will be
+ * re-used by realloc later.
+ */
+}
+
+/*
+ * 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->exited);
+ free(pool);
+
+ return 0;
+}
+
+/*
+ * Prepare for pthread_exit(), pool->mutex must be locked
+ */
+static void pthreadpool_server_exit(struct pthreadpool *pool)
+{
+ pthread_t *exited;
+
+ pool->num_threads -= 1;
+
+ exited = (pthread_t *)realloc(
+ pool->exited, sizeof(pthread_t *) * (pool->num_exited + 1));
+
+ if (exited == NULL) {
+ /* lost a thread status */
+ return;
+ }
+ pool->exited = exited;
+
+ 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 timespec ts;
+ struct pthreadpool_job *job;
+
+ /*
+ * idle-wait at most 1 second. If nothing happens in that
+ * time, exit this thread.
+ */
+
+ clock_gettime(CLOCK_REALTIME, &ts);
+ ts.tv_sec += 1;
+
+ 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->max_threads != 0) &&
+ (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..79704ea385
--- /dev/null
+++ b/source3/lib/pthreadpool/pthreadpool.h
@@ -0,0 +1,97 @@
+/*
+ * Unix SMB/CIFS implementation.
+ * threadpool implementation based on pthreads
+ * Copyright (C) Volker Lendecke 2009,2011
+ *
+ * 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/>.
+ */
+
+#ifndef __PTHREADPOOL_H__
+#define __PTHREADPOOL_H__
+
+struct pthreadpool;
+
+/**
+ * @defgroup pthreadpool The pthreadpool API
+ *
+ * This API provides a way to run threadsafe functions in a helper
+ * thread. It is initially intended to run getaddrinfo asynchronously.
+ */
+
+
+/**
+ * @brief Create a pthreadpool
+ *
+ * A struct pthreadpool is the basis for for running threads in the
+ * background.
+ *
+ * @param[in] max_threads Maximum parallelism in this pool
+ * @param[out] presult Pointer to the threadpool returned
+ * @return success: 0, failure: errno
+ *
+ * max_threads=0 means unlimited parallelism. The caller has to take
+ * care to not overload the system.
+ */
+int pthreadpool_init(unsigned max_threads, struct pthreadpool **presult);
+
+/**
+ * @brief Destroy a pthreadpool
+ *
+ * Destroy a pthreadpool. If jobs are still active, this returns
+ * EBUSY.
+ *
+ * @param[in] pool The pool to destroy
+ * @return success: 0, failure: errno
+ */
+int pthreadpool_destroy(struct pthreadpool *pool);
+
+/**
+ * @brief Add a job to a pthreadpool
+ *
+ * This adds a job to a pthreadpool. The job can be identified by
+ * job_id. This integer will be returned from
+ * pthreadpool_finished_job() then the job is completed.
+ *
+ * @param[in] pool The pool to run the job on
+ * @param[in] job_id A custom identifier
+ * @param[in] fn The function to run asynchronously
+ * @param[in] private_data Pointer passed to fn
+ * @return success: 0, failure: errno
+ */
+int pthreadpool_add_job(struct pthreadpool *pool, int job_id,
+ void (*fn)(void *private_data), void *private_data);
+
+/**
+ * @brief Get the signalling fd from a pthreadpool
+ *
+ * Completion of a job is indicated by readability of the fd retuned
+ * by pthreadpool_signal_fd().
+ *
+ * @param[in] pool The pool in question
+ * @return The fd to listen on for readability
+ */
+int pthreadpool_signal_fd(struct pthreadpool *pool);
+
+/**
+ * @brief Get the job_id of a finished job
+ *
+ * This blocks until a job has finished unless the fd returned by
+ * pthreadpool_signal_fd() is readable.
+ *
+ * @param[in] pool The pool to query for finished jobs
+ * @return The job_id of the finished job
+ */
+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..667ee01784
--- /dev/null
+++ b/source3/lib/pthreadpool/tests.c
@@ -0,0 +1,362 @@
+#include <stdio.h>
+#include <string.h>
+#include <poll.h>
+#include <errno.h>
+#include <stdlib.h>
+#include <pthread.h>
+#include <unistd.h>
+#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; i++) {
+ ret = pthreadpool_add_job(p, i, test_sleep, &timeout);
+ if (ret != 0) {
+ fprintf(stderr, "pthreadpool_add_job failed: %s\n",
+ strerror(ret));
+ return -1;
+ }
+ }
+
+ for (i=0; i<num_jobs; i++) {
+ ret = pthreadpool_finished_job(p);
+ if ((ret < 0) || (ret >= num_jobs)) {
+ fprintf(stderr, "invalid job number %d\n", ret);
+ return -1;
+ }
+ finished[ret] += 1;
+ }
+
+ for (i=0; i<num_jobs; i++) {
+ if (finished[i] != 1) {
+ fprintf(stderr, "finished[%d] = %d\n",
+ i, finished[i]);
+ return -1;
+ }
+ }
+
+ ret = pthreadpool_destroy(p);
+ if (ret != 0) {
+ fprintf(stderr, "pthreadpool_destroy failed: %s\n",
+ strerror(ret));
+ return -1;
+ }
+
+ free(finished);
+ return 0;
+}
+
+static int test_busydestroy(void)
+{
+ struct pthreadpool *p;
+ int timeout = 50;
+ struct pollfd pfd;
+ int ret;
+
+ ret = pthreadpool_init(1, &p);
+ if (ret != 0) {
+ fprintf(stderr, "pthreadpool_init failed: %s\n",
+ strerror(ret));
+ return -1;
+ }
+ ret = pthreadpool_add_job(p, 1, test_sleep, &timeout);
+ if (ret != 0) {
+ fprintf(stderr, "pthreadpool_add_job failed: %s\n",
+ strerror(ret));
+ return -1;
+ }
+ ret = pthreadpool_destroy(p);
+ if (ret != EBUSY) {
+ fprintf(stderr, "Could destroy a busy pool\n");
+ return -1;
+ }
+
+ pfd.fd = pthreadpool_signal_fd(p);
+ pfd.events = POLLIN|POLLERR;
+
+ poll(&pfd, 1, -1);
+
+ ret = pthreadpool_destroy(p);
+ if (ret != 0) {
+ fprintf(stderr, "pthreadpool_destroy failed: %s\n",
+ strerror(ret));
+ return -1;
+ }
+ return 0;
+}
+
+struct threaded_state {
+ pthread_t tid;
+ struct pthreadpool *p;
+ int start_job;
+ int num_jobs;
+ int timeout;
+};
+
+static void *test_threaded_worker(void *p)
+{
+ struct threaded_state *state = (struct threaded_state *)p;
+ int i;
+
+ for (i=0; i<state->num_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; i<num_pools; i++) {
+ ret = pthreadpool_init(poolsize, &pools[i]);
+ if (ret != 0) {
+ fprintf(stderr, "pthreadpool_init failed: %s\n",
+ strerror(ret));
+ return -1;
+ }
+ pfds[i].fd = pthreadpool_signal_fd(pools[i]);
+ pfds[i].events = POLLIN|POLLHUP;
+ }
+
+ poolnum = 0;
+
+ for (i=0; i<num_threads; i++) {
+ state = &states[i];
+
+ state->p = 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_pools; i++) {
+ ret = pthreadpool_destroy(pools[i]);
+ if (ret != 0) {
+ fprintf(stderr, "pthreadpool_destroy failed: "
+ "%s\n", strerror(ret));
+ exit(1);
+ }
+ }
+ /* child */
+ exit(0);
+ }
+
+ for (i=0; i<num_threads; i++) {
+ ret = pthread_join(states[i].tid, NULL);
+ if (ret != 0) {
+ fprintf(stderr, "pthread_join(%d) failed: %s\n",
+ i, strerror(ret));
+ return -1;
+ }
+ }
+
+ received = 0;
+
+ while (received < num_threads*num_jobs) {
+ int j;
+
+ ret = poll(pfds, num_pools, 1000);
+ if (ret == -1) {
+ fprintf(stderr, "poll failed: %s\n",
+ strerror(errno));
+ return -1;
+ }
+ if (ret == 0) {
+ fprintf(stderr, "\npoll timed out\n");
+ break;
+ }
+
+ for (j=0; j<num_pools; j++) {
+
+ if ((pfds[j].revents & (POLLIN|POLLHUP)) == 0) {
+ continue;
+ }
+
+ ret = pthreadpool_finished_job(pools[j]);
+ if ((ret < 0) || (ret >= num_jobs * num_threads)) {
+ fprintf(stderr, "invalid job number %d\n",
+ ret);
+ return -1;
+ }
+ finished[ret] += 1;
+ received += 1;
+ }
+ }
+
+ for (i=0; i<num_threads*num_jobs; i++) {
+ if (finished[i] != 1) {
+ fprintf(stderr, "finished[%d] = %d\n",
+ i, finished[i]);
+ return -1;
+ }
+ }
+
+ for (i=0; i<num_pools; i++) {
+ ret = pthreadpool_destroy(pools[i]);
+ if (ret != 0) {
+ fprintf(stderr, "pthreadpool_destroy failed: %s\n",
+ strerror(ret));
+ return -1;
+ }
+ }
+
+ free(pfds);
+ free(pools);
+ free(states);
+ free(finished);
+
+ return 0;
+}
+
+int main(void)
+{
+ int ret;
+
+ ret = test_init();
+ if (ret != 0) {
+ fprintf(stderr, "test_init failed\n");
+ return 1;
+ }
+
+ ret = test_jobs(10, 10000);
+ if (ret != 0) {
+ fprintf(stderr, "test_jobs failed\n");
+ return 1;
+ }
+
+ ret = test_busydestroy();
+ if (ret != 0) {
+ fprintf(stderr, "test_busydestroy failed\n");
+ return 1;
+ }
+
+ /*
+ * Test 10 threads adding jobs on a single pool
+ */
+ ret = test_threaded_addjob(1, 10, 5, 5000);
+ if (ret != 0) {
+ fprintf(stderr, "test_jobs failed\n");
+ return 1;
+ }
+
+ /*
+ * Test 10 threads on 3 pools to verify our fork handling
+ * works right.
+ */
+ ret = test_threaded_addjob(3, 10, 5, 5000);
+ if (ret != 0) {
+ fprintf(stderr, "test_jobs failed\n");
+ return 1;
+ }
+
+ printf("success\n");
+ return 0;
+}
diff --git a/source3/lib/pthreadpool/wscript_build b/source3/lib/pthreadpool/wscript_build
new file mode 100644
index 0000000000..611ff5a433
--- /dev/null
+++ b/source3/lib/pthreadpool/wscript_build
@@ -0,0 +1,11 @@
+#!/usr/bin/env python
+
+bld.SAMBA3_SUBSYSTEM('PTHREADPOOL',
+ source='pthreadpool.c',
+ deps='pthread rt',
+ enabled=bld.env.WITH_PTHREADPOOL)
+
+bld.SAMBA3_BINARY('pthreadpooltest',
+ source='tests.c',
+ deps='PTHREADPOOL',
+ enabled=bld.env.WITH_PTHREADPOOL)
generated by cgit (git 2.34.1) at 2024-11-18 03:04:39 +0000