Add thread pool

Included if pthreads are found, can be disabled with --enable-pthreadpool=no

Tim, Steven, I haven't yet seen comments from you. You have been asking for
such a thing at SambaXP. Do you like this? :-)

4 files changed, 562 insertions, 2 deletions

diff --git a/source3/Makefile.in b/source3/Makefile.in
index 8bf29ddfda..eb255c2061 100644
--- a/source3/Makefile.in
+++ b/source3/Makefile.in
@@ -385,7 +385,7 @@ LIB_OBJ = $(LIBSAMBAUTIL_OBJ) $(UTIL_OBJ) $(CRYPTO_OBJ) \
 	  lib/module.o lib/events.o @LIBTEVENT_OBJ0@ \
 	  lib/ldap_escape.o @CHARSET_STATIC@ \
 	  lib/secdesc.o lib/util_seaccess.o ../libcli/security/secace.o \
-	  ../libcli/security/secacl.o \
+	  ../libcli/security/secacl.o @PTHREADPOOL_OBJ@ \
 	  libads/krb5_errs.o lib/system_smbd.o lib/audit.o $(LIBNDR_OBJ) \
 	  lib/file_id.o lib/idmap_cache.o \
 	  ../libcli/security/dom_sid.o ../libcli/security/security_descriptor.o
diff --git a/source3/configure.in b/source3/configure.in
index 4ed9e68d0d..fc925ee736 100644
--- a/source3/configure.in
+++ b/source3/configure.in
@@ -5895,7 +5895,9 @@ else
 fi
 
 AC_CHECK_LIB(pthread, pthread_mutex_lock, [WINBIND_NSS_PTHREAD="-lpthread"
-			AC_DEFINE(HAVE_PTHREAD, 1, [whether pthread exists])])
+			AC_DEFINE(HAVE_PTHREAD, 1, [whether pthread exists])
+			samba_cv_HAVE_PTHREAD=yes],
+			samba_cv_HAVE_PTHREAD=no)
 
 AC_SUBST(WINBIND_NSS_PTHREAD)
 AC_SUBST(WINBIND_NSS)
@@ -6051,6 +6053,17 @@ if test x"$enable_avahi" != x"no"; then
 fi
 
 #################################################
+# Check if user wants pthreadpool support
+
+AC_ARG_ENABLE(pthreadpool,
+[AS_HELP_STRING([--enable-pthreadpool], [Enable pthreads pool helper support (default=auto)])])
+
+if test x"$enable_pthreadpool" != x"no" -a x"$samba_cv_HAVE_PTHREAD" = x"yes"; then
+    AC_DEFINE(WITH_PTHREADPOOL, 1, [Whether to include pthreadpool helpers])
+    AC_SUBST(PTHREADPOOL_OBJ, "lib/pthreadpool.o")
+fi
+
+#################################################
 # Check to see if we should use the included iniparser
 
 AC_ARG_WITH(included-iniparser,
diff --git a/source3/include/pthreadpool.h b/source3/include/pthreadpool.h
new file mode 100644
index 0000000000..7ef7ddf419
--- /dev/null
+++ b/source3/include/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 <http://www.gnu.org/licenses/>.
+ */
+
+#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.c b/source3/lib/pthreadpool.c
new file mode 100644
index 0000000000..c5f2b41265
--- /dev/null
+++ b/source3/lib/pthreadpool.c
@@ -0,0 +1,505 @@
+/*
+ * 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;
+			}
+		}
+
+		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) && (ret1 == 0)) {
+		free(pool);
+	}
+
+	if (ret != 0) {
+		return ret;
+	}
+	return ret1;
+}
+
+/*
+ * 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 timespec timeout;
+		struct pthreadpool_job *job;
+
+		/*
+		 * idle-wait at most 1 second. If nothing happens in that
+		 * time, exit this thread.
+		 */
+
+		clock_gettime(CLOCK_REALTIME, &timeout);
+		timeout.tv_sec += 1;
+
+		while ((pool->jobs == NULL) && (pool->shutdown == 0)) {
+
+			pool->num_idle += 1;
+			res = pthread_cond_timedwait(
+				&pool->condvar, &pool->mutex, &timeout);
+			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) {
+			int fd = pool->sig_pipe[1];
+			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 = sizeof(int);
+
+			res = pthread_mutex_lock(&pool->mutex);
+			assert(res == 0);
+
+			if (fd != -1) {
+				written = write(fd, &job->id, sizeof(int));
+			}
+
+			free(job);
+
+			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;
+	}
+
+	/*
+	 * 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;
+}