pthreadpool: add some lockless coordination between the main and job threads

In the direction from the main process to the job thread, we have:

- 'maycancel', which is set when tevent_req_cancel() is called,
- 'orphaned' is the job request, tevent_context or pthreadpool_tevent
  was talloc_free'ed.

The job function can consume these by using:

   /*
    * 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);

In the other direction we remember the following points
in the job execution:

- 'started'  - set when the job is picked up by a worker thread
- 'executed' - set once the job function returned.
- 'finished' - set when pthreadpool_tevent_job_signal() is entered
- 'dropped'  - set when pthreadpool_tevent_job_signal() leaves with orphaned
- 'signaled' - set when pthreadpool_tevent_job_signal() leaves normal

There're only one side writing each element,
either the main process or the job thread.

This means we can do the coordination with a full memory
barrier using atomic_thread_fence(memory_order_seq_cst).
lib/replace provides fallbacks if C11 stdatomic.h is not available.

A real pthreadpool requires pthread and atomic_thread_fence() (or an
replacement) to be available, otherwise we only have pthreadpool_sync.c.
But this should not make a real difference, as at least
__sync_synchronize() is availabe since 2005 in gcc.
We also require __thread which is available since 2002.

Signed-off-by: Stefan Metzmacher <metze@samba.org>
Reviewed-by: Ralph Boehme <slow@samba.org>

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);

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,

wscript

@@ -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')