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

[Mathieu Desnoyers <mathieu.desnoyers@efficios.com>]

[-- 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