diff options
| -rw-r--r-- | lib/pthreadpool/pthreadpool_tevent.c | 212 | ||||
| -rw-r--r-- | lib/pthreadpool/pthreadpool_tevent.h | 14 | ||||
| -rw-r--r-- | wscript | 12 |
3 files changed, 231 insertions, 7 deletions
diff --git a/lib/pthreadpool/pthreadpool_tevent.c b/lib/pthreadpool/pthreadpool_tevent.c index e7e17d3bf0f..fbf9c0e835a 100644 --- a/lib/pthreadpool/pthreadpool_tevent.c +++ b/lib/pthreadpool/pthreadpool_tevent.c @@ -18,10 +18,42 @@ */ #include "replace.h" +#include "system/threads.h" #include "pthreadpool_tevent.h" #include "pthreadpool.h" #include "lib/util/tevent_unix.h" #include "lib/util/dlinklist.h" +#include "lib/util/attr.h" + +#define PTHREAD_TEVENT_JOB_THREAD_FENCE_INIT(__job) do { \ + _UNUSED_ const struct pthreadpool_tevent_job *__j = __job; \ +} while(0); + +#ifdef WITH_PTHREADPOOL +/* + * configure checked we have pthread and atomic_thread_fence() available + */ +#define __PTHREAD_TEVENT_JOB_THREAD_FENCE(__order) do { \ + atomic_thread_fence(__order); \ +} while(0) +#else +/* + * we're using lib/pthreadpool/pthreadpool_sync.c ... + */ +#define __PTHREAD_TEVENT_JOB_THREAD_FENCE(__order) do { } while(0) +#ifndef HAVE___THREAD +#define __thread +#endif +#endif + +#define PTHREAD_TEVENT_JOB_THREAD_FENCE(__job) do { \ + _UNUSED_ const struct pthreadpool_tevent_job *__j = __job; \ + __PTHREAD_TEVENT_JOB_THREAD_FENCE(memory_order_seq_cst); \ +} while(0); + +#define PTHREAD_TEVENT_JOB_THREAD_FENCE_FINI(__job) do { \ + _UNUSED_ const struct pthreadpool_tevent_job *__j = __job; \ +} while(0); struct pthreadpool_tevent_job_state; @@ -75,6 +107,70 @@ struct pthreadpool_tevent_job { void (*fn)(void *private_data); void *private_data; + + /* + * Coordination between threads + * + * There're only one side writing each element + * either the main process or the job thread. + * + * The coordination is done by a full memory + * barrier using atomic_thread_fence(memory_order_seq_cst) + * wrapped in PTHREAD_TEVENT_JOB_THREAD_FENCE() + */ + struct { + /* + * 'maycancel' + * set when tevent_req_cancel() is called. + * (only written by main thread!) + */ + bool maycancel; + + /* + * 'orphaned' + * set when talloc_free is called on the job request, + * tevent_context or pthreadpool_tevent. + * (only written by main thread!) + */ + bool orphaned; + + /* + * 'started' + * set when the job is picked up by a worker thread + * (only written by job thread!) + */ + bool started; + + /* + * 'executed' + * set once the job function returned. + * (only written by job thread!) + */ + bool executed; + + /* + * 'finished' + * set when pthreadpool_tevent_job_signal() is entered + * (only written by job thread!) + */ + bool finished; + + /* + * 'dropped' + * set when pthreadpool_tevent_job_signal() leaves with + * orphaned already set. + * (only written by job thread!) + */ + bool dropped; + + /* + * 'signaled' + * set when pthreadpool_tevent_job_signal() leaves normal + * and the immediate event was scheduled. + * (only written by job thread!) + */ + bool signaled; + } needs_fence; }; static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool); @@ -299,11 +395,11 @@ static bool pthreadpool_tevent_job_cancel(struct tevent_req *req); static int pthreadpool_tevent_job_destructor(struct pthreadpool_tevent_job *job) { /* - * We should never be called with state->state != NULL. + * We should never be called with needs_fence.orphaned == false. * Only pthreadpool_tevent_job_orphan() will call TALLOC_FREE(job) * after detaching from the request state and pool list. */ - if (job->state != NULL) { + if (!job->needs_fence.orphaned) { abort(); } @@ -328,6 +424,17 @@ static int pthreadpool_tevent_job_destructor(struct pthreadpool_tevent_job *job) } } + PTHREAD_TEVENT_JOB_THREAD_FENCE(job); + if (job->needs_fence.dropped) { + /* + * The signal function saw job->needs_fence.orphaned + * before it started the signaling via the immediate + * event. So we'll never geht triggered and can + * remove job->im and let the whole job go... + */ + TALLOC_FREE(job->im); + } + /* * pthreadpool_tevent_job_orphan() already removed * it from pool->jobs. And we don't need try @@ -351,11 +458,15 @@ static int pthreadpool_tevent_job_destructor(struct pthreadpool_tevent_job *job) */ DLIST_REMOVE(orphaned_jobs, job); + PTHREAD_TEVENT_JOB_THREAD_FENCE_FINI(job); return 0; } static void pthreadpool_tevent_job_orphan(struct pthreadpool_tevent_job *job) { + job->needs_fence.orphaned = true; + PTHREAD_TEVENT_JOB_THREAD_FENCE(job); + /* * We're the only function that sets * job->state = NULL; @@ -476,6 +587,7 @@ struct tevent_req *pthreadpool_tevent_job_send( if (tevent_req_nomem(job->im, req)) { return tevent_req_post(req, ev); } + PTHREAD_TEVENT_JOB_THREAD_FENCE_INIT(job); talloc_set_destructor(job, pthreadpool_tevent_job_destructor); DLIST_ADD_END(job->pool->jobs, job); job->state = state; @@ -492,13 +604,76 @@ struct tevent_req *pthreadpool_tevent_job_send( return req; } +static __thread struct pthreadpool_tevent_job *current_job; + +bool pthreadpool_tevent_current_job_canceled(void) +{ + if (current_job == NULL) { + /* + * Should only be called from within + * the job function. + */ + abort(); + return false; + } + + PTHREAD_TEVENT_JOB_THREAD_FENCE(current_job); + return current_job->needs_fence.maycancel; +} + +bool pthreadpool_tevent_current_job_orphaned(void) +{ + if (current_job == NULL) { + /* + * Should only be called from within + * the job function. + */ + abort(); + return false; + } + + PTHREAD_TEVENT_JOB_THREAD_FENCE(current_job); + return current_job->needs_fence.orphaned; +} + +bool pthreadpool_tevent_current_job_continue(void) +{ + if (current_job == NULL) { + /* + * Should only be called from within + * the job function. + */ + abort(); + return false; + } + + PTHREAD_TEVENT_JOB_THREAD_FENCE(current_job); + if (current_job->needs_fence.maycancel) { + return false; + } + PTHREAD_TEVENT_JOB_THREAD_FENCE(current_job); + if (current_job->needs_fence.orphaned) { + return false; + } + + return true; +} + static void pthreadpool_tevent_job_fn(void *private_data) { struct pthreadpool_tevent_job *job = talloc_get_type_abort(private_data, struct pthreadpool_tevent_job); + current_job = job; + job->needs_fence.started = true; + PTHREAD_TEVENT_JOB_THREAD_FENCE(job); + job->fn(job->private_data); + + job->needs_fence.executed = true; + PTHREAD_TEVENT_JOB_THREAD_FENCE(job); + current_job = NULL; } static int pthreadpool_tevent_job_signal(int jobid, @@ -513,8 +688,12 @@ static int pthreadpool_tevent_job_signal(int jobid, struct tevent_threaded_context *tctx = NULL; struct pthreadpool_tevent_glue *g = NULL; - if (state == NULL) { + job->needs_fence.finished = true; + PTHREAD_TEVENT_JOB_THREAD_FENCE(job); + if (job->needs_fence.orphaned) { /* Request already gone */ + job->needs_fence.dropped = true; + PTHREAD_TEVENT_JOB_THREAD_FENCE(job); return 0; } @@ -543,6 +722,8 @@ static int pthreadpool_tevent_job_signal(int jobid, job); } + job->needs_fence.signaled = true; + PTHREAD_TEVENT_JOB_THREAD_FENCE(job); return 0; } @@ -565,9 +746,17 @@ static void pthreadpool_tevent_job_done(struct tevent_context *ctx, /* * pthreadpool_tevent_job_cleanup() - * will destroy the job. + * (called by tevent_req_done() or + * tevent_req_error()) will destroy the job. */ - tevent_req_done(state->req); + + if (job->needs_fence.executed) { + tevent_req_done(state->req); + return; + } + + tevent_req_error(state->req, ENOEXEC); + return; } static bool pthreadpool_tevent_job_cancel(struct tevent_req *req) @@ -582,6 +771,19 @@ static bool pthreadpool_tevent_job_cancel(struct tevent_req *req) return false; } + job->needs_fence.maycancel = true; + PTHREAD_TEVENT_JOB_THREAD_FENCE(job); + if (job->needs_fence.started) { + /* + * It was too late to cancel the request. + * + * The job still has the chance to look + * at pthreadpool_tevent_current_job_canceled() + * or pthreadpool_tevent_current_job_continue() + */ + return false; + } + num = pthreadpool_cancel_job(job->pool->pool, 0, pthreadpool_tevent_job_fn, job); diff --git a/lib/pthreadpool/pthreadpool_tevent.h b/lib/pthreadpool/pthreadpool_tevent.h index fdb86e23757..37e491e17c4 100644 --- a/lib/pthreadpool/pthreadpool_tevent.h +++ b/lib/pthreadpool/pthreadpool_tevent.h @@ -32,6 +32,20 @@ int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads, size_t pthreadpool_tevent_max_threads(struct pthreadpool_tevent *pool); size_t pthreadpool_tevent_queued_jobs(struct pthreadpool_tevent *pool); +/* + * return true - if tevent_req_cancel() was called. + */ +bool pthreadpool_tevent_current_job_canceled(void); +/* + * return true - if talloc_free() was called on the job request, + * tevent_context or pthreadpool_tevent. + */ +bool pthreadpool_tevent_current_job_orphaned(void); +/* + * return true if canceled and orphaned are both false. + */ +bool pthreadpool_tevent_current_job_continue(void); + struct tevent_req *pthreadpool_tevent_job_send( TALLOC_CTX *mem_ctx, struct tevent_context *ev, struct pthreadpool_tevent *pool, @@ -259,10 +259,18 @@ def configure(conf): conf.DEFINE('WITH_NTVFS_FILESERVER', 1) if Options.options.with_pthreadpool: - if conf.CONFIG_SET('HAVE_PTHREAD'): + if conf.CONFIG_SET('HAVE_PTHREAD') and \ + conf.CONFIG_SET('HAVE___THREAD') and \ + conf.CONFIG_SET('HAVE_ATOMIC_THREAD_FENCE_SUPPORT'): conf.DEFINE('WITH_PTHREADPOOL', '1') else: - Logs.warn("pthreadpool support cannot be enabled when pthread support was not found") + if not conf.CONFIG_SET('HAVE_PTHREAD'): + Logs.warn("pthreadpool support cannot be enabled when pthread support was not found") + if not conf.CONFIG_SET('HAVE_ATOMIC_THREAD_FENCE_SUPPORT'): + Logs.warn("""pthreadpool support cannot be enabled when there is + no support for atomic_thead_fence()""") + if not conf.CONFIG_SET('HAVE___THREAD'): + Logs.warn("pthreadpool support cannot be enabled when __thread support was not found") conf.undefine('WITH_PTHREADPOOL') conf.RECURSE('source3') |