From: Mathieu Desnoyers <mathieu.desnoyers@polymtl.ca>
Date: Wed, 23 Sep 2009 22:29:21 +0000 (-0400)
Subject: urcu-defer: ensure callbacks will never be enqueued forever
X-Git-Tag: v0.1~41
X-Git-Url: https://git.lttng.org./?a=commitdiff_plain;h=04eb9c4f11a43e4cd7307d319b2d5f7624cd2801;p=userspace-rcu.git

urcu-defer: ensure callbacks will never be enqueued forever

Even if there are no further callbacks enqueued, ensure that after a 100ms
delay, the callback queue will be dealt with.

Required proper ordering of queue vs futex.

e.g.

 The idea is to perform the "check for empty queue" between the
&defer_thread_futex decrement and the test in wait_defer. It skips the
futex call and proceed if the list is non-empty.

As I am drilling into the problem, it looks very much like an attempt to
implement efficient wait queues in userspace based on sys_futex().

/*
 * Wake-up any waiting defer thread. Called from many concurrent
 * threads.
 */
static void wake_up_defer(void)
{
        if (unlikely(atomic_read(&defer_thread_futex) == -1)) {
                atomic_set(&defer_thread_futex, 0);
                futex(&defer_thread_futex, FUTEX_WAKE, 0,
                      NULL, NULL, 0);
        }
}

/*
 * Defer thread waiting. Single thread.
 */
static void wait_defer(void)
{
        atomic_dec(&defer_thread_futex);
        smp_mb();       /* Write futex before read queue */
        if (rcu_defer_num_callbacks()) {
                smp_mb();       /* Read queue before write futex */
                /* Callbacks are queued, don't wait. */
                atomic_set(&defer_thread_futex, 0);
        } else {
                smp_rmb();      /* Read queue before read futex */
                if (atomic_read(&defer_thread_futex) == -1)
                        futex(&defer_thread_futex, FUTEX_WAIT, -1,
                              NULL, NULL, 0);
        }
}

- call_rcu():
  * queue callbacks to perform
  * smp_mb()
  * wake_up_defer()

- defer thread:
  * for (;;)
    * wait_defer()
    * sleep 100ms (wait for more callbacks to be enqueued)
    * dequeue callbacks, execute them

The goal here is that if call_rcu() enqueues a callback (even if it
races with defer thread going to sleep), there should not be a
potentially infinite delay before it gets executed. Therefore, being
blocked in sys_futex while there is a callback to execute, without any
hope to be woken up unless another callback is queued, would not meet
that design requirement. I think that checking the number of queued
callbacks within wait_defer() as I propose here should address this
situation.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@polymtl.ca>
---

diff --git a/urcu-defer.c b/urcu-defer.c
index 3efe5ba..13f94fa 100644
--- a/urcu-defer.c
+++ b/urcu-defer.c
@@ -71,29 +71,6 @@ static int num_deferers, alloc_deferers;
 
 static pthread_t tid_defer;
 
-/*
- * Wake-up any waiting defer thread. Called from many concurrent threads.
- */
-static void wake_up_defer(void)
-{
-	if (unlikely(atomic_read(&defer_thread_futex) == -1)) {
-		atomic_set(&defer_thread_futex, 0);
-		futex(&defer_thread_futex, FUTEX_WAKE, 0,
-		      NULL, NULL, 0);
-	}
-}
-
-/*
- * Defer thread waiting. Single thread.
- */
-static void wait_defer(void)
-{
-	atomic_dec(&defer_thread_futex);
-	if (atomic_read(&defer_thread_futex) == -1)
-		futex(&defer_thread_futex, FUTEX_WAIT, -1,
-		      NULL, NULL, 0);
-}
-
 static void internal_urcu_lock(pthread_mutex_t *mutex)
 {
 	int ret;
@@ -128,6 +105,51 @@ static void internal_urcu_unlock(pthread_mutex_t *mutex)
 	}
 }
 
+/*
+ * Wake-up any waiting defer thread. Called from many concurrent threads.
+ */
+static void wake_up_defer(void)
+{
+	if (unlikely(atomic_read(&defer_thread_futex) == -1)) {
+		atomic_set(&defer_thread_futex, 0);
+		futex(&defer_thread_futex, FUTEX_WAKE, 0,
+		      NULL, NULL, 0);
+	}
+}
+
+static unsigned long rcu_defer_num_callbacks(void)
+{
+	unsigned long num_items = 0, head;
+	struct deferer_registry *index;
+
+	internal_urcu_lock(&urcu_defer_mutex);
+	for (index = registry; index < registry + num_deferers; index++) {
+		head = LOAD_SHARED(index->defer_queue->head);
+		num_items += head - index->defer_queue->tail;
+	}
+	internal_urcu_unlock(&urcu_defer_mutex);
+	return num_items;
+}
+
+/*
+ * Defer thread waiting. Single thread.
+ */
+static void wait_defer(void)
+{
+	atomic_dec(&defer_thread_futex);
+	smp_mb();	/* Write futex before read queue */
+	if (rcu_defer_num_callbacks()) {
+		smp_mb();	/* Read queue before write futex */
+		/* Callbacks are queued, don't wait. */
+		atomic_set(&defer_thread_futex, 0);
+	} else {
+		smp_rmb();	/* Read queue before read futex */
+		if (atomic_read(&defer_thread_futex) == -1)
+			futex(&defer_thread_futex, FUTEX_WAIT, -1,
+			      NULL, NULL, 0);
+	}
+}
+
 /*
  * Must be called after Q.S. is reached.
  */
@@ -280,6 +302,7 @@ void _rcu_defer_queue(void (*fct)(void *p), void *p)
 	smp_wmb();	/* Publish new pointer before head */
 			/* Write q[] before head. */
 	STORE_SHARED(defer_queue.head, head);
+	smp_mb();	/* Write queue head before read futex */
 	/*
 	 * Wake-up any waiting defer thread.
 	 */