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, 7 insertions, 231 deletions
diff --git a/lib/pthreadpool/pthreadpool_tevent.c b/lib/pthreadpool/pthreadpool_tevent.c index db20ddc1530..3eca605d0c4 100644 --- a/lib/pthreadpool/pthreadpool_tevent.c +++ b/lib/pthreadpool/pthreadpool_tevent.c @@ -18,43 +18,11 @@ */ #include "replace.h" -#include "system/threads.h" #include "system/filesys.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; @@ -108,70 +76,6 @@ 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); @@ -396,11 +300,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 needs_fence.orphaned == false. + * We should never be called with state->state != NULL. * Only pthreadpool_tevent_job_orphan() will call TALLOC_FREE(job) * after detaching from the request state and pool list. */ - if (!job->needs_fence.orphaned) { + if (job->state != NULL) { abort(); } @@ -425,17 +329,6 @@ 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 @@ -459,15 +352,11 @@ 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; @@ -588,7 +477,6 @@ 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; @@ -605,76 +493,13 @@ 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, @@ -689,12 +514,8 @@ static int pthreadpool_tevent_job_signal(int jobid, struct tevent_threaded_context *tctx = NULL; struct pthreadpool_tevent_glue *g = NULL; - job->needs_fence.finished = true; - PTHREAD_TEVENT_JOB_THREAD_FENCE(job); - if (job->needs_fence.orphaned) { + if (state == NULL) { /* Request already gone */ - job->needs_fence.dropped = true; - PTHREAD_TEVENT_JOB_THREAD_FENCE(job); return 0; } @@ -723,8 +544,6 @@ static int pthreadpool_tevent_job_signal(int jobid, job); } - job->needs_fence.signaled = true; - PTHREAD_TEVENT_JOB_THREAD_FENCE(job); return 0; } @@ -747,17 +566,9 @@ static void pthreadpool_tevent_job_done(struct tevent_context *ctx, /* * pthreadpool_tevent_job_cleanup() - * (called by tevent_req_done() or - * tevent_req_error()) will destroy the job. + * will destroy the job. */ - - if (job->needs_fence.executed) { - tevent_req_done(state->req); - return; - } - - tevent_req_error(state->req, ENOEXEC); - return; + tevent_req_done(state->req); } static bool pthreadpool_tevent_job_cancel(struct tevent_req *req) @@ -772,19 +583,6 @@ 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 37e491e17c4..fdb86e23757 100644 --- a/lib/pthreadpool/pthreadpool_tevent.h +++ b/lib/pthreadpool/pthreadpool_tevent.h @@ -32,20 +32,6 @@ 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, @@ -279,18 +279,10 @@ def configure(conf): conf.DEFINE('WITH_NTVFS_FILESERVER', 1) if Options.options.with_pthreadpool: - if conf.CONFIG_SET('HAVE_PTHREAD') and \ - conf.CONFIG_SET('HAVE___THREAD') and \ - conf.CONFIG_SET('HAVE_ATOMIC_THREAD_FENCE_SUPPORT'): + if conf.CONFIG_SET('HAVE_PTHREAD'): conf.DEFINE('WITH_PTHREADPOOL', '1') else: - 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") + Logs.warn("pthreadpool support cannot be enabled when pthread support was not found") conf.undefine('WITH_PTHREADPOOL') conf.SET_TARGET_TYPE('jansson', 'EMPTY') |