[RFC PATCH liburcu 1/2] Implement urcu workqueues internal API

[RFC PATCH liburcu 1/2] Implement urcu workqueues internal API

[Mathieu Desnoyers]

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
---
 src/Makefile.am |   2 +-
 src/workqueue.c | 507 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/workqueue.h | 104 ++++++++++++
 3 files changed, 612 insertions(+), 1 deletion(-)
 create mode 100644 src/workqueue.c
 create mode 100644 src/workqueue.h

diff --git a/src/Makefile.am b/src/Makefile.am
index a801020..60b833d 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -54,7 +54,7 @@ liburcu_bp_la_SOURCES = urcu-bp.c urcu-pointer.c $(COMPAT)
 liburcu_bp_la_LIBADD = liburcu-common.la
 
 liburcu_cds_la_SOURCES = rculfqueue.c rculfstack.c lfstack.c \
-	$(RCULFHASH) $(COMPAT)
+	workqueue.c workqueue.h $(RCULFHASH) $(COMPAT)
 liburcu_cds_la_LIBADD = liburcu-common.la
 
 pkgconfigdir = $(libdir)/pkgconfig
diff --git a/src/workqueue.c b/src/workqueue.c
new file mode 100644
index 0000000..891a8fc
--- /dev/null
+++ b/src/workqueue.c
@@ -0,0 +1,507 @@
+/*
+ * workqueue.c
+ *
+ * Userspace RCU library - Userspace workqeues
+ *
+ * Copyright (c) 2010 Paul E. McKenney <paulmck@linux.vnet.ibm.com>
+ * Copyright (c) 2017 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#define _LGPL_SOURCE
+#include <stdio.h>
+#include <pthread.h>
+#include <signal.h>
+#include <assert.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <string.h>
+#include <errno.h>
+#include <poll.h>
+#include <sys/time.h>
+#include <unistd.h>
+#include <sched.h>
+
+#include "compat-getcpu.h"
+#include "urcu/wfcqueue.h"
+#include "urcu-call-rcu.h"
+#include "urcu-pointer.h"
+#include "urcu/list.h"
+#include "urcu/futex.h"
+#include "urcu/tls-compat.h"
+#include "urcu/ref.h"
+#include "urcu-die.h"
+
+#include "workqueue.h"
+
+#define SET_AFFINITY_CHECK_PERIOD		(1U << 8)	/* 256 */
+#define SET_AFFINITY_CHECK_PERIOD_MASK		(SET_AFFINITY_CHECK_PERIOD - 1)
+
+/* Data structure that identifies a workqueue. */
+
+struct urcu_workqueue {
+	/*
+	 * We do not align head on a different cache-line than tail
+	 * mainly because call_rcu callback-invocation threads use
+	 * batching ("splice") to get an entire list of callbacks, which
+	 * effectively empties the queue, and requires to touch the tail
+	 * anyway.
+	 */
+	struct cds_wfcq_tail cbs_tail;
+	struct cds_wfcq_head cbs_head;
+	unsigned long flags;
+	int32_t futex;
+	unsigned long qlen; /* maintained for debugging. */
+	pthread_t tid;
+	int cpu_affinity;
+	unsigned long loop_count;
+	void *priv;
+	void (*grace_period_fct)(struct urcu_workqueue *workqueue, void *priv);
+	void (*initialize_worker_fct)(struct urcu_workqueue *workqueue, void *priv);
+	void (*finalize_worker_fct)(struct urcu_workqueue *workqueue, void *priv);
+	void (*worker_before_pause_fct)(struct urcu_workqueue *workqueue, void *priv);
+	void (*worker_after_resume_fct)(struct urcu_workqueue *workqueue, void *priv);
+	void (*worker_before_wait_fct)(struct urcu_workqueue *workqueue, void *priv);
+	void (*worker_after_wake_up_fct)(struct urcu_workqueue *workqueue, void *priv);
+} __attribute__((aligned(CAA_CACHE_LINE_SIZE)));
+
+struct urcu_workqueue_completion {
+	int barrier_count;
+	int32_t futex;
+	struct urcu_ref ref;
+};
+
+struct urcu_workqueue_completion_work {
+	struct urcu_work work;
+	struct urcu_workqueue_completion *completion;
+};
+
+/*
+ * Periodically retry setting CPU affinity if we migrate.
+ * Losing affinity can be caused by CPU hotunplug/hotplug, or by
+ * cpuset(7).
+ */
+#if HAVE_SCHED_SETAFFINITY
+static int set_thread_cpu_affinity(struct urcu_workqueue *workqueue)
+{
+	cpu_set_t mask;
+	int ret;
+
+	if (workqueue->cpu_affinity < 0)
+		return 0;
+	if (++workqueue->loop_count & SET_AFFINITY_CHECK_PERIOD_MASK)
+		return 0;
+	if (urcu_sched_getcpu() == workqueue->cpu_affinity)
+		return 0;
+
+	CPU_ZERO(&mask);
+	CPU_SET(workqueue->cpu_affinity, &mask);
+#if SCHED_SETAFFINITY_ARGS == 2
+	ret = sched_setaffinity(0, &mask);
+#else
+	ret = sched_setaffinity(0, sizeof(mask), &mask);
+#endif
+	/*
+	 * EINVAL is fine: can be caused by hotunplugged CPUs, or by
+	 * cpuset(7). This is why we should always retry if we detect
+	 * migration.
+	 */
+	if (ret && errno == EINVAL) {
+		ret = 0;
+		errno = 0;
+	}
+	return ret;
+}
+#else
+static int set_thread_cpu_affinity(struct urcu_workqueue *workqueue)
+{
+	return 0;
+}
+#endif
+
+static void workqueue_wait(struct urcu_workqueue *workqueue)
+{
+	/* Read workqueue before read futex */
+	cmm_smp_mb();
+	if (uatomic_read(&workqueue->futex) != -1)
+		return;
+	while (futex_async(&workqueue->futex, FUTEX_WAIT, -1,
+			NULL, NULL, 0)) {
+		switch (errno) {
+		case EWOULDBLOCK:
+			/* Value already changed. */
+			return;
+		case EINTR:
+			/* Retry if interrupted by signal. */
+			break;	/* Get out of switch. */
+		default:
+			/* Unexpected error. */
+			urcu_die(errno);
+		}
+	}
+}
+
+static void workqueue_wake_up(struct urcu_workqueue *workqueue)
+{
+	/* Write to workqueue before reading/writing futex */
+	cmm_smp_mb();
+	if (caa_unlikely(uatomic_read(&workqueue->futex) == -1)) {
+		uatomic_set(&workqueue->futex, 0);
+		if (futex_async(&workqueue->futex, FUTEX_WAKE, 1,
+				NULL, NULL, 0) < 0)
+			urcu_die(errno);
+	}
+}
+
+static void __urcu_workqueue_wait_completion(struct urcu_workqueue_completion *completion)
+{
+	/* Read completion barrier count before read futex */
+	cmm_smp_mb();
+	if (uatomic_read(&completion->futex) != -1)
+		return;
+	while (futex_async(&completion->futex, FUTEX_WAIT, -1,
+			NULL, NULL, 0)) {
+		switch (errno) {
+		case EWOULDBLOCK:
+			/* Value already changed. */
+			return;
+		case EINTR:
+			/* Retry if interrupted by signal. */
+			break;	/* Get out of switch. */
+		default:
+			/* Unexpected error. */
+			urcu_die(errno);
+		}
+	}
+}
+
+static void urcu_workqueue_completion_wake_up(struct urcu_workqueue_completion *completion)
+{
+	/* Write to completion barrier count before reading/writing futex */
+	cmm_smp_mb();
+	if (caa_unlikely(uatomic_read(&completion->futex) == -1)) {
+		uatomic_set(&completion->futex, 0);
+		if (futex_async(&completion->futex, FUTEX_WAKE, 1,
+				NULL, NULL, 0) < 0)
+			urcu_die(errno);
+	}
+}
+
+/* This is the code run by each worker thread. */
+
+static void *workqueue_thread(void *arg)
+{
+	unsigned long cbcount;
+	struct urcu_workqueue *workqueue = (struct urcu_workqueue *) arg;
+	int rt = !!(uatomic_read(&workqueue->flags) & URCU_WORKQUEUE_RT);
+
+	if (set_thread_cpu_affinity(workqueue))
+		urcu_die(errno);
+
+	if (workqueue->initialize_worker_fct)
+		workqueue->initialize_worker_fct(workqueue, workqueue->priv);
+
+	if (!rt) {
+		uatomic_dec(&workqueue->futex);
+		/* Decrement futex before reading workqueue */
+		cmm_smp_mb();
+	}
+	for (;;) {
+		struct cds_wfcq_head cbs_tmp_head;
+		struct cds_wfcq_tail cbs_tmp_tail;
+		struct cds_wfcq_node *cbs, *cbs_tmp_n;
+		enum cds_wfcq_ret splice_ret;
+
+		if (set_thread_cpu_affinity(workqueue))
+			urcu_die(errno);
+
+		if (uatomic_read(&workqueue->flags) & URCU_WORKQUEUE_PAUSE) {
+			/*
+			 * Pause requested. Become quiescent: remove
+			 * ourself from all global lists, and don't
+			 * process any callback. The callback lists may
+			 * still be non-empty though.
+			 */
+			if (workqueue->worker_before_pause_fct)
+				workqueue->worker_before_pause_fct(workqueue, workqueue->priv);
+			cmm_smp_mb__before_uatomic_or();
+			uatomic_or(&workqueue->flags, URCU_WORKQUEUE_PAUSED);
+			while ((uatomic_read(&workqueue->flags) & URCU_WORKQUEUE_PAUSE) != 0)
+				(void) poll(NULL, 0, 1);
+			uatomic_and(&workqueue->flags, ~URCU_WORKQUEUE_PAUSED);
+			cmm_smp_mb__after_uatomic_and();
+			if (workqueue->worker_after_resume_fct)
+				workqueue->worker_after_resume_fct(workqueue, workqueue->priv);
+		}
+
+		cds_wfcq_init(&cbs_tmp_head, &cbs_tmp_tail);
+		splice_ret = __cds_wfcq_splice_blocking(&cbs_tmp_head,
+			&cbs_tmp_tail, &workqueue->cbs_head, &workqueue->cbs_tail);
+		assert(splice_ret != CDS_WFCQ_RET_WOULDBLOCK);
+		assert(splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
+		if (splice_ret != CDS_WFCQ_RET_SRC_EMPTY) {
+			if (workqueue->grace_period_fct)
+				workqueue->grace_period_fct(workqueue, workqueue->priv);
+			cbcount = 0;
+			__cds_wfcq_for_each_blocking_safe(&cbs_tmp_head,
+					&cbs_tmp_tail, cbs, cbs_tmp_n) {
+				struct rcu_head *rhp;
+
+				rhp = caa_container_of(cbs,
+					struct rcu_head, next);
+				rhp->func(rhp);
+				cbcount++;
+			}
+			uatomic_sub(&workqueue->qlen, cbcount);
+		}
+		if (uatomic_read(&workqueue->flags) & URCU_WORKQUEUE_STOP)
+			break;
+		if (workqueue->worker_before_wait_fct)
+			workqueue->worker_before_wait_fct(workqueue, workqueue->priv);
+		if (!rt) {
+			if (cds_wfcq_empty(&workqueue->cbs_head,
+					&workqueue->cbs_tail)) {
+				workqueue_wait(workqueue);
+				(void) poll(NULL, 0, 10);
+				uatomic_dec(&workqueue->futex);
+				/*
+				 * Decrement futex before reading
+				 * call_rcu list.
+				 */
+				cmm_smp_mb();
+			} else {
+				(void) poll(NULL, 0, 10);
+			}
+		} else {
+			(void) poll(NULL, 0, 10);
+		}
+		if (workqueue->worker_after_wake_up_fct)
+			workqueue->worker_after_wake_up_fct(workqueue, workqueue->priv);
+	}
+	if (!rt) {
+		/*
+		 * Read call_rcu list before write futex.
+		 */
+		cmm_smp_mb();
+		uatomic_set(&workqueue->futex, 0);
+	}
+	if (workqueue->finalize_worker_fct)
+		workqueue->finalize_worker_fct(workqueue, workqueue->priv);
+	return NULL;
+}
+
+struct urcu_workqueue *urcu_workqueue_create(unsigned long flags,
+		int cpu_affinity, void *priv,
+		void (*grace_period_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*initialize_worker_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*finalize_worker_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*worker_before_wait_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*worker_after_wake_up_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*worker_before_pause_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*worker_after_resume_fct)(struct urcu_workqueue *workqueue, void *priv))
+{
+	struct urcu_workqueue *workqueue;
+	int ret;
+
+	workqueue = malloc(sizeof(*workqueue));
+	if (workqueue == NULL)
+		urcu_die(errno);
+	memset(workqueue, '\0', sizeof(*workqueue));
+	cds_wfcq_init(&workqueue->cbs_head, &workqueue->cbs_tail);
+	workqueue->qlen = 0;
+	workqueue->futex = 0;
+	workqueue->flags = flags;
+	workqueue->priv = priv;
+	workqueue->grace_period_fct = grace_period_fct;
+	workqueue->initialize_worker_fct = initialize_worker_fct;
+	workqueue->finalize_worker_fct = finalize_worker_fct;
+	workqueue->worker_before_wait_fct = worker_before_wait_fct;
+	workqueue->worker_after_wake_up_fct = worker_after_wake_up_fct;
+	workqueue->worker_before_pause_fct = worker_before_pause_fct;
+	workqueue->worker_after_resume_fct = worker_after_resume_fct;
+	workqueue->cpu_affinity = cpu_affinity;
+	workqueue->loop_count = 0;
+	cmm_smp_mb();  /* Structure initialized before pointer is planted. */
+	ret = pthread_create(&workqueue->tid, NULL, workqueue_thread, workqueue);
+	if (ret) {
+		urcu_die(ret);
+	}
+	return workqueue;
+}
+
+static void wake_worker_thread(struct urcu_workqueue *workqueue)
+{
+	if (!(_CMM_LOAD_SHARED(workqueue->flags) & URCU_CALL_RCU_RT))
+		workqueue_wake_up(workqueue);
+}
+
+static int urcu_workqueue_destroy_worker(struct urcu_workqueue *workqueue)
+{
+	int ret;
+	void *retval;
+
+	uatomic_or(&workqueue->flags, URCU_WORKQUEUE_STOP);
+	wake_worker_thread(workqueue);
+
+	ret = pthread_join(workqueue->tid, &retval);
+	if (ret) {
+		urcu_die(ret);
+	}
+	if (retval != NULL) {
+		urcu_die(EINVAL);
+	}
+	workqueue->flags &= ~URCU_WORKQUEUE_STOP;
+	workqueue->tid = 0;
+	return 0;
+}
+
+void urcu_workqueue_destroy(struct urcu_workqueue *workqueue)
+{
+	if (workqueue == NULL) {
+		return;
+	}
+	if (urcu_workqueue_destroy_worker(workqueue)) {
+		urcu_die(errno);
+	}
+	assert(cds_wfcq_empty(&workqueue->cbs_head, &workqueue->cbs_tail));
+	free(workqueue);
+}
+
+void urcu_workqueue_queue_work(struct urcu_workqueue *workqueue,
+		      struct urcu_work *work,
+		      void (*func)(struct urcu_work *work))
+{
+	cds_wfcq_node_init(&work->next);
+	work->func = func;
+	cds_wfcq_enqueue(&workqueue->cbs_head, &workqueue->cbs_tail, &work->next);
+	uatomic_inc(&workqueue->qlen);
+	wake_worker_thread(workqueue);
+}
+
+static
+void free_completion(struct urcu_ref *ref)
+{
+	struct urcu_workqueue_completion *completion;
+
+	completion = caa_container_of(ref, struct urcu_workqueue_completion, ref);
+	free(completion);
+}
+
+static
+void _urcu_workqueue_wait_complete(struct urcu_work *work)
+{
+	struct urcu_workqueue_completion_work *completion_work;
+	struct urcu_workqueue_completion *completion;
+
+	completion_work = caa_container_of(work, struct urcu_workqueue_completion_work, work);
+	completion = completion_work->completion;
+	if (!uatomic_sub_return(&completion->barrier_count, 1))
+		urcu_workqueue_completion_wake_up(completion);
+	urcu_ref_put(&completion->ref, free_completion);
+	free(completion_work);
+}
+
+struct urcu_workqueue_completion *urcu_workqueue_create_completion(void)
+{
+	struct urcu_workqueue_completion *completion;
+
+	completion = calloc(sizeof(*completion), 1);
+	if (!completion)
+		urcu_die(errno);
+	urcu_ref_set(&completion->ref, 1);
+	completion->barrier_count = 0;
+	return completion;
+}
+
+void urcu_workqueue_destroy_completion(struct urcu_workqueue_completion *completion)
+{
+	urcu_ref_put(&completion->ref, free_completion);
+}
+
+void urcu_workqueue_wait_completion(struct urcu_workqueue_completion *completion)
+{
+	/* Wait for them */
+	for (;;) {
+		uatomic_dec(&completion->futex);
+		/* Decrement futex before reading barrier_count */
+		cmm_smp_mb();
+		if (!uatomic_read(&completion->barrier_count))
+			break;
+		__urcu_workqueue_wait_completion(completion);
+	}
+}
+
+void urcu_workqueue_queue_completion(struct urcu_workqueue *workqueue,
+		struct urcu_workqueue_completion *completion)
+{
+	struct urcu_workqueue_completion_work *work;
+
+	work = calloc(sizeof(*work), 1);
+	if (!work)
+		urcu_die(errno);
+	work->completion = completion;
+	urcu_ref_get(&completion->ref);
+	uatomic_inc(&completion->barrier_count);
+	urcu_workqueue_queue_work(workqueue, &work->work, _urcu_workqueue_wait_complete);
+}
+
+/*
+ * Wait for all in-flight work to complete execution.
+ */
+void urcu_workqueue_flush_queued_work(struct urcu_workqueue *workqueue)
+{
+	struct urcu_workqueue_completion *completion;
+
+	completion = urcu_workqueue_create_completion();
+	if (!completion)
+		urcu_die(ENOMEM);
+	urcu_workqueue_queue_completion(workqueue, completion);
+	urcu_workqueue_wait_completion(completion);
+	urcu_workqueue_destroy_completion(completion);
+}
+
+/* To be used in before fork handler. */
+void urcu_workqueue_pause_worker(struct urcu_workqueue *workqueue)
+{
+	uatomic_or(&workqueue->flags, URCU_WORKQUEUE_PAUSE);
+	cmm_smp_mb__after_uatomic_or();
+	wake_worker_thread(workqueue);
+
+	while ((uatomic_read(&workqueue->flags) & URCU_WORKQUEUE_PAUSED) == 0)
+		(void) poll(NULL, 0, 1);
+}
+
+/* To be used in after fork parent handler. */
+void urcu_workqueue_resume_worker(struct urcu_workqueue *workqueue)
+{
+	uatomic_and(&workqueue->flags, ~URCU_WORKQUEUE_PAUSE);
+	while ((uatomic_read(&workqueue->flags) & URCU_WORKQUEUE_PAUSED) != 0)
+		(void) poll(NULL, 0, 1);
+}
+
+void urcu_workqueue_create_worker(struct urcu_workqueue *workqueue)
+{
+	int ret;
+
+	/* Clear workqueue state from parent. */
+	workqueue->flags &= ~URCU_WORKQUEUE_PAUSED;
+	workqueue->flags &= ~URCU_WORKQUEUE_PAUSE;
+	workqueue->tid = 0;
+	ret = pthread_create(&workqueue->tid, NULL, workqueue_thread, workqueue);
+	if (ret) {
+		urcu_die(ret);
+	}
+}
diff --git a/src/workqueue.h b/src/workqueue.h
new file mode 100644
index 0000000..52b6973
--- /dev/null
+++ b/src/workqueue.h
@@ -0,0 +1,104 @@
+#ifndef _URCU_WORKQUEUE_H
+#define _URCU_WORKQUEUE_H
+
+/*
+ * workqueue.h
+ *
+ * Userspace RCU header - Userspace workqueues
+ *
+ * Copyright (c) 2009,2017 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include <stdlib.h>
+#include <pthread.h>
+
+#include <urcu/wfcqueue.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/* Note that struct urcu_workqueue is opaque to callers. */
+
+struct urcu_workqueue;
+struct urcu_workqueue_completion;
+
+/* Flag values. */
+
+#define URCU_WORKQUEUE_RT	(1U << 0)
+#define URCU_WORKQUEUE_STOP	(1U << 1)
+#define URCU_WORKQUEUE_PAUSE	(1U << 2)
+#define URCU_WORKQUEUE_PAUSED	(1U << 3)
+
+/*
+ * The urcu_work data structure is placed in the structure to be acted
+ * upon via urcu_workqueue_queue_work().
+ */
+
+struct urcu_work {
+	struct cds_wfcq_node next;
+	void (*func)(struct urcu_work *head);
+};
+
+/*
+ * Exported functions
+ */
+
+struct urcu_workqueue *urcu_workqueue_create(unsigned long flags,
+		int cpu_affinity, void *priv,
+		void (*grace_period_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*initialize_worker_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*finalize_worker_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*worker_before_wait_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*worker_after_wake_up_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*worker_before_pause_fct)(struct urcu_workqueue *workqueue, void *priv),
+		void (*worker_after_resume_fct)(struct urcu_workqueue *workqueue, void *priv));
+void urcu_workqueue_destroy(struct urcu_workqueue *workqueue);
+
+/*
+ * Never fails. Should not be used to enqueue work from worker threads
+ * after the application invokes urcu_workqueue_free.
+ */
+void urcu_workqueue_queue_work(struct urcu_workqueue *workqueue,
+		struct urcu_work *work,
+		void (*func)(struct urcu_work *work));
+
+struct urcu_workqueue_completion *urcu_workqueue_create_completion(void);
+void urcu_workqueue_destroy_completion(struct urcu_workqueue_completion *completion);
+
+void urcu_workqueue_queue_completion(struct urcu_workqueue *workqueue,
+		struct urcu_workqueue_completion *completion);
+void urcu_workqueue_wait_completion(struct urcu_workqueue_completion *completion);
+
+void urcu_workqueue_flush_queued_work(struct urcu_workqueue *workqueue);
+
+/*
+ * pause/resume/create worker threads. Can be used to pause worker
+ * threads across fork/clone while keeping the workqueue in place.
+ * Pause is used in parent pre-fork, resume in parent post-fork, create
+ * in child after-fork.
+ */
+void urcu_workqueue_pause_worker(struct urcu_workqueue *workqueue);
+void urcu_workqueue_resume_worker(struct urcu_workqueue *workqueue);
+void urcu_workqueue_create_worker(struct urcu_workqueue *workqueue);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _URCU_WORKQUEUE_H */
-- 
2.1.4

_______________________________________________
lttng-dev mailing list
lttng-dev@lists.lttng.org
https://lists.lttng.org/cgi-bin/mailman/listinfo/lttng-dev

[RCU PATCH liburcu 2/2] Use workqueue in rculfhash

[Mathieu Desnoyers]

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
---
 include/urcu/rculfhash.h |  15 ++++--
 src/rculfhash-internal.h |   2 +-
 src/rculfhash.c          | 124 +++++++++++++++++++++++------------------------
 3 files changed, 74 insertions(+), 67 deletions(-)

diff --git a/include/urcu/rculfhash.h b/include/urcu/rculfhash.h
index 9934422..0789aa5 100644
--- a/include/urcu/rculfhash.h
+++ b/include/urcu/rculfhash.h
@@ -176,10 +176,17 @@ struct cds_lfht *cds_lfht_new(unsigned long init_size,
  *        need to be informed of the value passed to cds_lfht_new().
  *
  * Return 0 on success, negative error value on error.
- * Threads calling this API need to be registered RCU read-side threads.
- * cds_lfht_destroy should *not* be called from a RCU read-side critical
- * section. It should *not* be called from a call_rcu thread context
- * neither.
+
+ * Prior to liburcu 0.10:
+ * - Threads calling this API need to be registered RCU read-side
+ *   threads.
+ * - cds_lfht_destroy should *not* be called from a RCU read-side
+ *   critical section. It should *not* be called from a call_rcu thread
+ *   context neither.
+ *
+ * Starting from liburcu 0.10, rculfhash implements its own worker
+ * thread to handle resize operations, which removes RCU requirements on
+ * cds_lfht_destroy.
  */
 extern
 int cds_lfht_destroy(struct cds_lfht *ht, pthread_attr_t **attr);
diff --git a/src/rculfhash-internal.h b/src/rculfhash-internal.h
index d7cec95..0f8df97 100644
--- a/src/rculfhash-internal.h
+++ b/src/rculfhash-internal.h
@@ -82,7 +82,7 @@ struct cds_lfht {
 	 */
 	pthread_mutex_t resize_mutex;	/* resize mutex: add/del mutex */
 	pthread_attr_t *resize_attr;	/* Resize threads attributes */
-	unsigned int in_progress_resize, in_progress_destroy;
+	unsigned int in_progress_destroy;
 	unsigned long resize_target;
 	int resize_initiated;
 
diff --git a/src/rculfhash.c b/src/rculfhash.c
index d7a1f23..b7b8f95 100644
--- a/src/rculfhash.c
+++ b/src/rculfhash.c
@@ -64,7 +64,7 @@
  * - Split-counters are used to keep track of the number of
  *   nodes within the hash table for automatic resize triggering.
  * - Resize operation initiated by long chain detection is executed by a
- *   call_rcu thread, which keeps lock-freedom of add and remove.
+ *   worker thread, which keeps lock-freedom of add and remove.
  * - Resize operations are protected by a mutex.
  * - The removal operation is split in two parts: first, a "removed"
  *   flag is set in the next pointer within the node to remove. Then,
@@ -276,6 +276,8 @@
 #include <rculfhash-internal.h>
 #include <stdio.h>
 #include <pthread.h>
+#include "workqueue.h"
+#include "urcu-die.h"
 
 /*
  * Split-counters lazily update the global counter each 1024
@@ -335,11 +337,11 @@ struct ht_items_count {
 } __attribute__((aligned(CAA_CACHE_LINE_SIZE)));
 
 /*
- * rcu_resize_work: Contains arguments passed to RCU worker thread
+ * resize_work: Contains arguments passed to worker thread
  * responsible for performing lazy resize.
  */
-struct rcu_resize_work {
-	struct rcu_head head;
+struct resize_work {
+	struct urcu_work work;
 	struct cds_lfht *ht;
 };
 
@@ -356,6 +358,8 @@ struct partition_resize_work {
 		    unsigned long start, unsigned long len);
 };
 
+static struct urcu_workqueue *cds_lfht_workqueue;
+
 /*
  * Algorithm to reverse bits in a word by lookup table, extended to
  * 64-bit words.
@@ -1224,14 +1228,12 @@ void partition_resize_helper(struct cds_lfht *ht, unsigned long i,
 	if (start == 0 && nr_threads > 0)
 		return;
 fallback:
-	ht->flavor->thread_online();
 	fct(ht, i, start, len);
-	ht->flavor->thread_offline();
 }
 
 /*
  * Holding RCU read lock to protect _cds_lfht_add against memory
- * reclaim that could be performed by other call_rcu worker threads (ABA
+ * reclaim that could be performed by other worker threads (ABA
  * problem).
  *
  * When we reach a certain length, we can split this population phase over
@@ -1308,7 +1310,7 @@ void init_table(struct cds_lfht *ht,
 
 /*
  * Holding RCU read lock to protect _cds_lfht_remove against memory
- * reclaim that could be performed by other call_rcu worker threads (ABA
+ * reclaim that could be performed by other worker threads (ABA
  * problem).
  * For a single level, we logically remove and garbage collect each node.
  *
@@ -1320,8 +1322,9 @@ void init_table(struct cds_lfht *ht,
  *
  * Concurrent removal and add operations are helping us perform garbage
  * collection of logically removed nodes. We guarantee that all logically
- * removed nodes have been garbage-collected (unlinked) before call_rcu is
- * invoked to free a hole level of bucket nodes (after a grace period).
+ * removed nodes have been garbage-collected (unlinked) before work
+ * enqueue is invoked to free a hole level of bucket nodes (after a
+ * grace period).
  *
  * Logical removal and garbage collection can therefore be done in batch
  * or on a node-per-node basis, as long as the guarantee above holds.
@@ -1772,25 +1775,12 @@ int cds_lfht_delete_bucket(struct cds_lfht *ht)
  */
 int cds_lfht_destroy(struct cds_lfht *ht, pthread_attr_t **attr)
 {
-	int ret, was_online;
+	int ret;
 
-	/* Wait for in-flight resize operations to complete */
+	/* Cancel ongoing resize operations. */
 	_CMM_STORE_SHARED(ht->in_progress_destroy, 1);
-	cmm_smp_mb();	/* Store destroy before load resize */
-	was_online = ht->flavor->read_ongoing();
-	if (was_online)
-		ht->flavor->thread_offline();
-	/* Calling with RCU read-side held is an error. */
-	if (ht->flavor->read_ongoing()) {
-		ret = -EINVAL;
-		if (was_online)
-			ht->flavor->thread_online();
-		goto end;
-	}
-	while (uatomic_read(&ht->in_progress_resize))
-		poll(NULL, 0, 100);	/* wait for 100ms */
-	if (was_online)
-		ht->flavor->thread_online();
+	/* Wait for in-flight resize operations to complete */
+	urcu_workqueue_flush_queued_work(cds_lfht_workqueue);
 	ret = cds_lfht_delete_bucket(ht);
 	if (ret)
 		return ret;
@@ -1801,7 +1791,6 @@ int cds_lfht_destroy(struct cds_lfht *ht, pthread_attr_t **attr)
 	if (ret)
 		ret = -EBUSY;
 	poison_free(ht);
-end:
 	return ret;
 }
 
@@ -1897,7 +1886,6 @@ void _do_cds_lfht_resize(struct cds_lfht *ht)
 	 * Resize table, re-do if the target size has changed under us.
 	 */
 	do {
-		assert(uatomic_read(&ht->in_progress_resize));
 		if (CMM_LOAD_SHARED(ht->in_progress_destroy))
 			break;
 		ht->resize_initiated = 1;
@@ -1930,71 +1918,47 @@ void resize_target_update_count(struct cds_lfht *ht,
 
 void cds_lfht_resize(struct cds_lfht *ht, unsigned long new_size)
 {
-	int was_online;
-
-	was_online = ht->flavor->read_ongoing();
-	if (was_online)
-		ht->flavor->thread_offline();
-	/* Calling with RCU read-side held is an error. */
-	if (ht->flavor->read_ongoing()) {
-		static int print_once;
-
-		if (!CMM_LOAD_SHARED(print_once))
-			fprintf(stderr, "[error] rculfhash: cds_lfht_resize "
-				"called with RCU read-side lock held.\n");
-		CMM_STORE_SHARED(print_once, 1);
-		assert(0);
-		goto end;
-	}
 	resize_target_update_count(ht, new_size);
 	CMM_STORE_SHARED(ht->resize_initiated, 1);
 	pthread_mutex_lock(&ht->resize_mutex);
 	_do_cds_lfht_resize(ht);
 	pthread_mutex_unlock(&ht->resize_mutex);
-end:
-	if (was_online)
-		ht->flavor->thread_online();
 }
 
 static
-void do_resize_cb(struct rcu_head *head)
+void do_resize_cb(struct urcu_work *work)
 {
-	struct rcu_resize_work *work =
-		caa_container_of(head, struct rcu_resize_work, head);
-	struct cds_lfht *ht = work->ht;
+	struct resize_work *resize_work =
+		caa_container_of(work, struct resize_work, work);
+	struct cds_lfht *ht = resize_work->ht;
 
-	ht->flavor->thread_offline();
+	ht->flavor->register_thread();
 	pthread_mutex_lock(&ht->resize_mutex);
 	_do_cds_lfht_resize(ht);
 	pthread_mutex_unlock(&ht->resize_mutex);
-	ht->flavor->thread_online();
+	ht->flavor->unregister_thread();
 	poison_free(work);
-	cmm_smp_mb();	/* finish resize before decrement */
-	uatomic_dec(&ht->in_progress_resize);
 }
 
 static
 void __cds_lfht_resize_lazy_launch(struct cds_lfht *ht)
 {
-	struct rcu_resize_work *work;
+	struct resize_work *work;
 
 	/* Store resize_target before read resize_initiated */
 	cmm_smp_mb();
 	if (!CMM_LOAD_SHARED(ht->resize_initiated)) {
-		uatomic_inc(&ht->in_progress_resize);
-		cmm_smp_mb();	/* increment resize count before load destroy */
 		if (CMM_LOAD_SHARED(ht->in_progress_destroy)) {
-			uatomic_dec(&ht->in_progress_resize);
 			return;
 		}
 		work = malloc(sizeof(*work));
 		if (work == NULL) {
 			dbg_printf("error allocating resize work, bailing out\n");
-			uatomic_dec(&ht->in_progress_resize);
 			return;
 		}
 		work->ht = ht;
-		ht->flavor->update_call_rcu(&work->head, do_resize_cb);
+		urcu_workqueue_queue_work(cds_lfht_workqueue,
+			&work->work, do_resize_cb);
 		CMM_STORE_SHARED(ht->resize_initiated, 1);
 	}
 }
@@ -2045,3 +2009,39 @@ void cds_lfht_resize_lazy_count(struct cds_lfht *ht, unsigned long size,
 	}
 	__cds_lfht_resize_lazy_launch(ht);
 }
+
+static void cds_lfht_fork_prepare(void)
+{
+	urcu_workqueue_pause_worker(cds_lfht_workqueue);
+}
+
+static void cds_lfht_fork_parent(void)
+{
+	urcu_workqueue_resume_worker(cds_lfht_workqueue);
+}
+
+static void cds_lfht_fork_child(void)
+{
+	urcu_workqueue_create_worker(cds_lfht_workqueue);
+}
+
+static void __attribute__((constructor)) cds_lfht_init_worker(void)
+{
+	int ret;
+
+	if (cds_lfht_workqueue)
+		return;
+	cds_lfht_workqueue = urcu_workqueue_create(0, -1, NULL,
+		NULL, NULL, NULL, NULL, NULL, NULL, NULL);
+	ret = pthread_atfork(cds_lfht_fork_prepare,
+		cds_lfht_fork_parent, cds_lfht_fork_child);
+	if (ret) {
+		urcu_die(ret);
+	}
+}
+
+static void __attribute__((destructor)) cds_lfht_fini_worker(void)
+{
+	urcu_workqueue_destroy(cds_lfht_workqueue);
+	cds_lfht_workqueue = NULL;
+}
-- 
2.1.4

_______________________________________________
lttng-dev mailing list
lttng-dev@lists.lttng.org
https://lists.lttng.org/cgi-bin/mailman/listinfo/lttng-dev

Re: [RFC PATCH liburcu 0/2] Remove RCU requirements on hash table destroy

[Paul E. McKenney]

On Tue, May 30, 2017 at 05:10:18PM -0400, Mathieu Desnoyers wrote:
> The RCU lock-free hash table currently requires that the destroy
> function should not be called from within RCU read-side critical
> sections. This is caused by the lazy resize, which uses the call_rcu
> worker thread, even though all it really needs is a workqueue/worker
> thread scheme.
> 
> Implement an internal workqueue API in liburcu, and use it instead of
> call_rcu in rculfhash to overcome this limitation.

Took a quick look, and it appears plausible.

Some opportunity to share CPU-affinity code between this and the
call_rcu() code, FWIW.  Two of the system-call stubs look to be identical
other than the system call (EINTR checks and soforth), but I am not sure
that it is worth combining them.

								Thanx, Paul

> Mathieu Desnoyers (2):
>   Implement urcu workqueues internal API
>   Use workqueue in rculfhash
> 
>  include/urcu/rculfhash.h |  15 +-
>  src/Makefile.am          |   2 +-
>  src/rculfhash-internal.h |   2 +-
>  src/rculfhash.c          | 124 ++++++------
>  src/workqueue.c          | 507 +++++++++++++++++++++++++++++++++++++++++++++++
>  src/workqueue.h          | 104 ++++++++++
>  6 files changed, 686 insertions(+), 68 deletions(-)
>  create mode 100644 src/workqueue.c
>  create mode 100644 src/workqueue.h
> 
> -- 
> 2.1.4
> 

_______________________________________________
lttng-dev mailing list
lttng-dev@lists.lttng.org
https://lists.lttng.org/cgi-bin/mailman/listinfo/lttng-dev

Re: [RFC PATCH liburcu 0/2] Remove RCU requirements on hash table destroy

[Mathieu Desnoyers]

[-- Attachment #1: Type: text/plain, Size: 1406 bytes --]

----- On Jun 5, 2017, at 6:56 PM, Paul E. McKenney paulmck@linux.vnet.ibm.com wrote:

> On Tue, May 30, 2017 at 05:10:18PM -0400, Mathieu Desnoyers wrote:
>> The RCU lock-free hash table currently requires that the destroy
>> function should not be called from within RCU read-side critical
>> sections. This is caused by the lazy resize, which uses the call_rcu
>> worker thread, even though all it really needs is a workqueue/worker
>> thread scheme.
>> 
>> Implement an internal workqueue API in liburcu, and use it instead of
>> call_rcu in rculfhash to overcome this limitation.
> 
> Took a quick look, and it appears plausible.
> 
> Some opportunity to share CPU-affinity code between this and the
> call_rcu() code, FWIW.

Given that I plan to reimplement the call_rcu code using this new
internal workqueue API, I don't think it is useful to try to lift
out the duplicated code between call_rcu and workqueue. When call_rcu
is reimplemented, the duplicated cpu affinity code will vanish.

>  Two of the system-call stubs look to be identical
> other than the system call (EINTR checks and soforth), but I am not sure
> that it is worth combining them.

Are you talking about the futex wait/wakeup ? If so, would the
attached patch that combine those work for you ? I noticed that
even the error handling is identical.

Thanks,

Mathieu


-- 
Mathieu Desnoyers
EfficiOS Inc.
http://www.efficios.com

[-- Attachment #2: patch --]
[-- Type: application/octet-stream, Size: 3972 bytes --]

commit 16355b70504149028d27b60e3c8839ce590ca1ef
Author: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Date:   Tue Jun 6 13:59:11 2017 -0400

    workqueue: combine futex wait/wakeup code
    
    Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>

diff --git a/src/workqueue.c b/src/workqueue.c
index 891a8fc..17ea835 100644
--- a/src/workqueue.c
+++ b/src/workqueue.c
@@ -132,14 +132,13 @@ static int set_thread_cpu_affinity(struct urcu_workqueue *workqueue)
 }
 #endif
 
-static void workqueue_wait(struct urcu_workqueue *workqueue)
+static void futex_wait(int32_t *futex)
 {
-	/* Read workqueue before read futex */
+	/* Read condition before read futex */
 	cmm_smp_mb();
-	if (uatomic_read(&workqueue->futex) != -1)
+	if (uatomic_read(futex) != -1)
 		return;
-	while (futex_async(&workqueue->futex, FUTEX_WAIT, -1,
-			NULL, NULL, 0)) {
+	while (futex_async(futex, FUTEX_WAIT, -1, NULL, NULL, 0)) {
 		switch (errno) {
 		case EWOULDBLOCK:
 			/* Value already changed. */
@@ -154,47 +153,13 @@ static void workqueue_wait(struct urcu_workqueue *workqueue)
 	}
 }
 
-static void workqueue_wake_up(struct urcu_workqueue *workqueue)
+static void futex_wake_up(int32_t *futex)
 {
-	/* Write to workqueue before reading/writing futex */
+	/* Write to condition before reading/writing futex */
 	cmm_smp_mb();
-	if (caa_unlikely(uatomic_read(&workqueue->futex) == -1)) {
-		uatomic_set(&workqueue->futex, 0);
-		if (futex_async(&workqueue->futex, FUTEX_WAKE, 1,
-				NULL, NULL, 0) < 0)
-			urcu_die(errno);
-	}
-}
-
-static void __urcu_workqueue_wait_completion(struct urcu_workqueue_completion *completion)
-{
-	/* Read completion barrier count before read futex */
-	cmm_smp_mb();
-	if (uatomic_read(&completion->futex) != -1)
-		return;
-	while (futex_async(&completion->futex, FUTEX_WAIT, -1,
-			NULL, NULL, 0)) {
-		switch (errno) {
-		case EWOULDBLOCK:
-			/* Value already changed. */
-			return;
-		case EINTR:
-			/* Retry if interrupted by signal. */
-			break;	/* Get out of switch. */
-		default:
-			/* Unexpected error. */
-			urcu_die(errno);
-		}
-	}
-}
-
-static void urcu_workqueue_completion_wake_up(struct urcu_workqueue_completion *completion)
-{
-	/* Write to completion barrier count before reading/writing futex */
-	cmm_smp_mb();
-	if (caa_unlikely(uatomic_read(&completion->futex) == -1)) {
-		uatomic_set(&completion->futex, 0);
-		if (futex_async(&completion->futex, FUTEX_WAKE, 1,
+	if (caa_unlikely(uatomic_read(futex) == -1)) {
+		uatomic_set(futex, 0);
+		if (futex_async(futex, FUTEX_WAKE, 1,
 				NULL, NULL, 0) < 0)
 			urcu_die(errno);
 	}
@@ -274,7 +239,7 @@ static void *workqueue_thread(void *arg)
 		if (!rt) {
 			if (cds_wfcq_empty(&workqueue->cbs_head,
 					&workqueue->cbs_tail)) {
-				workqueue_wait(workqueue);
+				futex_wait(&workqueue->futex);
 				(void) poll(NULL, 0, 10);
 				uatomic_dec(&workqueue->futex);
 				/*
@@ -345,7 +310,7 @@ struct urcu_workqueue *urcu_workqueue_create(unsigned long flags,
 static void wake_worker_thread(struct urcu_workqueue *workqueue)
 {
 	if (!(_CMM_LOAD_SHARED(workqueue->flags) & URCU_CALL_RCU_RT))
-		workqueue_wake_up(workqueue);
+		futex_wake_up(&workqueue->futex);
 }
 
 static int urcu_workqueue_destroy_worker(struct urcu_workqueue *workqueue)
@@ -409,7 +374,7 @@ void _urcu_workqueue_wait_complete(struct urcu_work *work)
 	completion_work = caa_container_of(work, struct urcu_workqueue_completion_work, work);
 	completion = completion_work->completion;
 	if (!uatomic_sub_return(&completion->barrier_count, 1))
-		urcu_workqueue_completion_wake_up(completion);
+		futex_wake_up(&completion->futex);
 	urcu_ref_put(&completion->ref, free_completion);
 	free(completion_work);
 }
@@ -440,7 +405,7 @@ void urcu_workqueue_wait_completion(struct urcu_workqueue_completion *completion
 		cmm_smp_mb();
 		if (!uatomic_read(&completion->barrier_count))
 			break;
-		__urcu_workqueue_wait_completion(completion);
+		futex_wait(&completion->futex);
 	}
 }
 

[-- Attachment #3: Type: text/plain, Size: 156 bytes --]

_______________________________________________
lttng-dev mailing list
lttng-dev@lists.lttng.org
https://lists.lttng.org/cgi-bin/mailman/listinfo/lttng-dev

[RFC PATCH liburcu 0/2] Remove RCU requirements on hash table destroy

[Mathieu Desnoyers]

The RCU lock-free hash table currently requires that the destroy
function should not be called from within RCU read-side critical
sections. This is caused by the lazy resize, which uses the call_rcu
worker thread, even though all it really needs is a workqueue/worker
thread scheme.

Implement an internal workqueue API in liburcu, and use it instead of
call_rcu in rculfhash to overcome this limitation.

Mathieu Desnoyers (2):
  Implement urcu workqueues internal API
  Use workqueue in rculfhash

 include/urcu/rculfhash.h |  15 +-
 src/Makefile.am          |   2 +-
 src/rculfhash-internal.h |   2 +-
 src/rculfhash.c          | 124 ++++++------
 src/workqueue.c          | 507 +++++++++++++++++++++++++++++++++++++++++++++++
 src/workqueue.h          | 104 ++++++++++
 6 files changed, 686 insertions(+), 68 deletions(-)
 create mode 100644 src/workqueue.c
 create mode 100644 src/workqueue.h

-- 
2.1.4

_______________________________________________
lttng-dev mailing list
lttng-dev@lists.lttng.org
https://lists.lttng.org/cgi-bin/mailman/listinfo/lttng-dev