#include "tevent.h"
#include "tevent_internal.h"
#include "tevent_util.h"
+#include "lib/util/dlinklist.h"
#ifdef HAVE_PTHREAD
#include "system/threads.h"
}
}
+/*********************************************************************
+ * Raw threadpool implementation
+ *********************************************************************/
+
+struct threadpool_job {
+ /* List of jobs */
+ struct threadpool_job *prev, *next;
+
+ int id;
+ void (*fn)(void *private_data);
+ void *private_data;
+ bool cancel;
+};
+
+struct threadpool {
+ /*
+ * List threadpools for fork safety
+ */
+ struct threadpool *prev, *next;
+
+ /*
+ * Control access to this struct
+ */
+ pthread_mutex_t mutex;
+
+ /*
+ * Threads waiting for work do so here
+ */
+ pthread_cond_t condvar;
+
+ /*
+ * List of jobs
+ */
+ struct threadpool_job *jobs;
+
+ /*
+ * 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 threadpools_mutex = PTHREAD_MUTEX_INITIALIZER;
+static struct threadpool *threadpools = NULL;
+static pthread_once_t threadpool_atfork_initialized = PTHREAD_ONCE_INIT;
+
+static void threadpool_prep_atfork(void);
+
+/*
+ * Initialize a thread pool
+ */
+
+static struct threadpool *threadpool_init(TALLOC_CTX *mem_ctx,
+ unsigned max_threads)
+{
+ struct threadpool *pool;
+ int ret;
+
+ pool = talloc_zero(mem_ctx, struct threadpool);
+ if (pool == NULL) {
+ return NULL;
+ }
+
+ ret = pipe(pool->sig_pipe);
+ if (ret == -1) {
+ TALLOC_FREE(pool);
+ return NULL;
+ }
+
+ ret = pthread_mutex_init(&pool->mutex, NULL);
+ if (ret != 0) {
+ close(pool->sig_pipe[0]);
+ close(pool->sig_pipe[1]);
+ TALLOC_FREE(pool);
+ return NULL;
+ }
+
+ 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]);
+ TALLOC_FREE(pool);
+ return NULL;
+ }
+
+ pool->shutdown = 0;
+ pool->num_threads = 0;
+ pool->num_exited = 0;
+ pool->exited = NULL;
+ pool->max_threads = max_threads;
+ pool->num_idle = 0;
+
+ ret = pthread_mutex_lock(&threadpools_mutex);
+ if (ret != 0) {
+ pthread_cond_destroy(&pool->condvar);
+ pthread_mutex_destroy(&pool->mutex);
+ close(pool->sig_pipe[0]);
+ close(pool->sig_pipe[1]);
+ TALLOC_FREE(pool);
+ return NULL;
+ }
+ DLIST_ADD(threadpools, pool);
+
+ ret = pthread_mutex_unlock(&threadpools_mutex);
+ assert(ret == 0);
+
+ pthread_once(&threadpool_atfork_initialized, threadpool_prep_atfork);
+
+ return pool;
+}
+
+static void threadpool_prepare(void)
+{
+ int ret;
+ struct threadpool *pool;
+
+ ret = pthread_mutex_lock(&threadpools_mutex);
+ assert(ret == 0);
+
+ pool = threadpools;
+
+ while (pool != NULL) {
+ ret = pthread_mutex_lock(&pool->mutex);
+ assert(ret == 0);
+ pool = pool->next;
+ }
+}
+
+static void threadpool_parent(void)
+{
+ int ret;
+ struct threadpool *pool;
+
+ for (pool = DLIST_TAIL(threadpools);
+ pool != NULL;
+ pool = DLIST_PREV(pool)) {
+ ret = pthread_mutex_unlock(&pool->mutex);
+ assert(ret == 0);
+ }
+
+ ret = pthread_mutex_unlock(&threadpools_mutex);
+ assert(ret == 0);
+}
+
+static void threadpool_child(void)
+{
+ int ret;
+ struct threadpool *pool;
+
+ for (pool = DLIST_TAIL(threadpools);
+ pool != NULL;
+ pool = DLIST_PREV(pool)) {
+
+ 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;
+
+ ret = pthread_mutex_unlock(&pool->mutex);
+ assert(ret == 0);
+ }
+
+ ret = pthread_mutex_unlock(&threadpools_mutex);
+ assert(ret == 0);
+}
+
+static void threadpool_prep_atfork(void)
+{
+ pthread_atfork(threadpool_prepare, threadpool_parent,
+ threadpool_child);
+}
+
+/*
+ * Return the file descriptor which becomes readable when a job has
+ * finished
+ */
+
+static int threadpool_signal_fd(struct threadpool *pool)
+{
+ return pool->sig_pipe[0];
+}
+
+static void threadpool_job_cancel(struct threadpool_job *job)
+{
+ job->cancel = true;
+}
+
+static bool threadpool_job_is_cancelled(struct threadpool_job *job)
+{
+ return job->cancel;
+}
+
+/*
+ * Do a pthread_join() on all children that have exited, pool->mutex must be
+ * locked
+ */
+static void threadpool_join_children(struct threadpool *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
+ */
+
+static int threadpool_finished_jobs(struct threadpool *pool, int *jobids,
+ unsigned num_jobids)
+{
+ ssize_t to_read, nread;
+
+ nread = -1;
+ errno = EINTR;
+
+ to_read = sizeof(int) * num_jobids;
+
+ while ((nread == -1) && (errno == EINTR)) {
+ nread = read(pool->sig_pipe[0], jobids, to_read);
+ }
+ if (nread == -1) {
+ return -errno;
+ }
+ if ((nread % sizeof(int)) != 0) {
+ return -EINVAL;
+ }
+ return nread / sizeof(int);
+}
+
+/*
+ * Destroy a thread pool, finishing all threads working for it
+ */
+
+static int threadpool_destroy(struct threadpool *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->shutdown 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) {
+ threadpool_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(&threadpools_mutex);
+ if (ret != 0) {
+ return ret;
+ }
+ DLIST_REMOVE(threadpools, pool);
+ ret = pthread_mutex_unlock(&threadpools_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);
+ TALLOC_FREE(pool);
+
+ return 0;
+}
+
+/*
+ * Prepare for pthread_exit(), pool->mutex must be locked
+ */
+static void threadpool_server_exit(struct threadpool *pool)
+{
+ pthread_t *exited;
+
+ pool->num_threads--;
+
+ 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++;
+}
+
+static struct threadpool_job *threadpool_get_job(struct threadpool *p)
+{
+ struct threadpool_job *job = p->jobs;
+
+ if (job) {
+ DLIST_REMOVE(p->jobs, job);
+ }
+ return job;
+}
+
+static struct threadpool_job *threadpool_put_job(TALLOC_CTX *mem_ctx,
+ struct threadpool *p,
+ int id,
+ void (*fn)(void *private_data),
+ void *private_data)
+{
+ struct threadpool_job *job;
+
+ job = talloc_zero(mem_ctx, struct threadpool_job);
+ if (job == NULL) {
+ return NULL;
+ }
+
+ job->id = id;
+ job->fn = fn;
+ job->private_data = private_data;
+ DLIST_ADD_END(p->jobs, job, struct threadpool_job *);
+
+ return job;
+}
+
+static void *threadpool_server(void *arg)
+{
+ struct threadpool *pool = (struct threadpool *)arg;
+ int res;
+
+ res = pthread_mutex_lock(&pool->mutex);
+ if (res != 0) {
+ return NULL;
+ }
+
+ while (1) {
+ struct timespec ts;
+ struct threadpool_job *job;
+
+ /*
+ * idle-wait at most 1 second. If nothing happens in that
+ * time, exit this thread.
+ */
+
+ clock_gettime(CLOCK_MONOTONIC, &ts);
+ ts.tv_sec += 1;
+
+ while ((pool->jobs == NULL) && (pool->shutdown == 0)) {
+
+ pool->num_idle++;
+ res = pthread_cond_timedwait(
+ &pool->condvar, &pool->mutex, &ts);
+ pool->num_idle--;
+
+ if (res == ETIMEDOUT) {
+ if (pool->jobs == NULL) {
+ /*
+ * we timed out and still no work for
+ * us. Exit.
+ */
+ threadpool_server_exit(pool);
+ pthread_mutex_unlock(&pool->mutex);
+ return NULL;
+ }
+
+ break;
+ }
+ assert(res == 0);
+ }
+
+ job = threadpool_get_job(pool);
+ if (job != NULL) {
+ ssize_t written;
+ int sig_pipe = pool->sig_pipe[1];
+
+ printf("threadpool_server: pulled job %d\n", job->id);
+ /*
+ * Do the work with the mutex unlocked
+ */
+
+ res = pthread_mutex_unlock(&pool->mutex);
+ assert(res == 0);
+
+ if (likely(!job->cancel)) {
+ printf("threadpool_server: executing job %d\n", job->id);
+ job->fn(job->private_data);
+ } else {
+ printf("threadpool_server: skipping cancelled job %d\n", job->id);
+ }
+
+ /*
+ * Do the write without the lock, otherwise we
+ * can deadlock with the scheduler
+ */
+ written = write(sig_pipe, &job->id, sizeof(job->id));
+
+ res = pthread_mutex_lock(&pool->mutex);
+ assert(res == 0);
+
+ if (written != sizeof(int)) {
+ threadpool_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
+ */
+ threadpool_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;
+ }
+ }
+}
+
+static struct threadpool_job *threadpool_add_job(TALLOC_CTX *mem_ctx,
+ struct threadpool *pool,
+ int job_id,
+ void (*fn)(void *private_data),
+ void *private_data)
+{
+ pthread_t thread_id;
+ int res;
+ sigset_t mask, omask;
+ struct threadpool_job *job;
+
+ res = pthread_mutex_lock(&pool->mutex);
+ assert(res == 0);
+
+ if (pool->shutdown) {
+ /*
+ * Protect against the pool being shut down while
+ * trying to add a job
+ */
+ printf("threadpool_add_job: pool is shutting down\n");
+ res = pthread_mutex_unlock(&pool->mutex);
+ assert(res == 0);
+ return NULL;
+ }
+
+ /*
+ * Just some cleanup under the mutex
+ */
+ threadpool_join_children(pool);
+
+ /*
+ * Add job to the end of the queue
+ */
+ job = threadpool_put_job(mem_ctx, pool, job_id, fn, private_data);
+ if (job == NULL) {
+ printf("threadpool_add_job: threadpool_put_job failed\n");
+ pthread_mutex_unlock(&pool->mutex);
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ if (pool->num_idle > 0) {
+ /*
+ * We have idle threads, wake one.
+ */
+ res = pthread_cond_signal(&pool->condvar);
+ pthread_mutex_unlock(&pool->mutex);
+ return job;
+ }
+
+ 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 job;
+ }
+
+ /*
+ * Create a new worker thread. It should not receive any signals.
+ */
+
+ sigfillset(&mask);
+
+ res = pthread_sigmask(SIG_BLOCK, &mask, &omask);
+ assert(res == 0);
+
+ res = pthread_create(&thread_id, NULL, threadpool_server,
+ (void *)pool);
+ if (res == 0) {
+ pool->num_threads += 1;
+ }
+
+ res = pthread_sigmask(SIG_SETMASK, &omask, NULL);
+ assert(res == 0);
+
+ pthread_mutex_unlock(&pool->mutex);
+ return job;
+}
+
+/*******************************************************************
+ * tvent_threadpool_send()/recv() sugar
+ *******************************************************************/
+
+struct tevent_threadpool_job {
+ tevent_threadpool_fn_t fn;
+ void *private_data;
+};
+
struct tevent_threadpool_state {
struct tevent_threadpool *pool;
int job_id;
bool done;
- void *private_parent;
- void *job_private;
- struct threadpool_job *job;
+ struct tevent_threadpool_job *job;
+ struct threadpool_job *threadpool_job;
};
struct tevent_threadpool {
&req->data);
}
-#if 0
-typedef (*tevent_threadpool_fn_t)(struct tevent_threadpool_fn *fn, void *args);
-struct tevent_threadpool_job(TALLOC_CTX *mem_ctx, tevent_threadpool_fn_t fn)
+struct tevent_threadpool_job *_tevent_threadpool_job_create(TALLOC_CTX *mem_ctx,
+ tevent_threadpool_fn_t fn,
+ void *pdata,
+ size_t data_size,
+ const char *type,
+ const char *location)
{
- struct tevent_threadpool_fn *f;
+ struct tevent_threadpool_job *job;
+ void **ppdata = (void **)pdata;
+ void *data;
+
+ job = talloc_pooled_object(mem_ctx, tevent_threadpool_job, 1,
+ data_size);
+ if (job == NULL) {
+ return NULL;
+ }
+ ZERO_STRUCTP(job);
- f = talloc_zero(mem_ctx, struct tevent_threadpool_fn);
- if (f == NULL) {
+ job->fn = fn;
+
+ data = talloc_zero_size(job, data_size);
+ if (data == NULL) {
+ talloc_free(job);
return NULL;
}
+ talloc_set_name_const(job, type);
- f->fn = fn;
+ job->data = data;
- talloc_set_destructor(f, ....);
+ talloc_set_destructor(job, tevent_threaded_job_destructor);
- return f;
+ *ppdata = data;
+ return job;
}
-#endif
struct tevent_req *tevent_threadpool_send(struct tevent_threadpool *pool,
- void (*fn)(void *private_data),
- void *private_data)
+ struct tevent_threadpool_job *job)
{
struct tevent_req *req;
struct tevent_threadpool_state *state;
state->pool = pool;
state->job_id = tevent_threadpool_next_job_id(state->pool);
state->done = false;
- state->job_private = private_data;
+ state->job = job;
printf("tevent_threadpool_send: sheduling job %d\n", state->job_id);
- state->job = threadpool_add_job(state,
- state->pool->threadpool,
- state->job_id,
- fn,
- state->job_private);
- if (state->job == NULL) {
+ state->threadpool_job = threadpool_add_job(state,
+ state->pool->threadpool,
+ state->job_id,
+ state->job->fn,
+ state->job->data);
+ if (state->threadpool_job == NULL) {
printf("tevent_threadpool_send: job %d state->job == NULL\n", state->job_id);
tevent_req_error(req, errno);
return tevent_req_post(req, pool->ev);
git.samba.org / metze / samba / wip.git / commitdiff