[1/2] rcu: Introduce force_rcu notifier

Commit Message

Greg Kurz Oct. 15, 2021, 4:12 p.m. UTC

The drain_rcu_call() function can be blocked as long as an RCU reader
stays in a read-side critical section. This is typically what happens
when a TCG vCPU is executing a busy loop. It can deadlock the QEMU
monitor as reported in https://gitlab.com/qemu-project/qemu/-/issues/650 .

This can be avoided by allowing drain_rcu_call() to enforce an RCU grace
period. Since each reader might need to do specific actions to end a
read-side critical section, do it with notifiers.

Prepare ground for this by adding a NotifierList and use it in
wait_for_readers() if drain_rcu_call() is in progress. Readers can
now optionally specify a Notifier to be called in this case at
thread registration time. The current rcu_register_thread() API is
preserved for readers that don't need this. The notifier is removed
automatically when the thread unregisters.

This is largely based on a draft from Paolo Bonzini.

Suggested-by: Paolo Bonzini <[e-mail]>
Signed-off-by: Greg Kurz <[e-mail]>
---
 include/qemu/rcu.h | 21 ++++++++++++++++++++-
 util/rcu.c         | 23 +++++++++++++++++++++--
 2 files changed, 41 insertions(+), 3 deletions(-)

Comments

Paolo Bonzini Oct. 18, 2021, 9:17 a.m. UTC | #1

On 15/10/21 18:12, Greg Kurz wrote:
> +/*
> + * NotifierList used to force an RCU grace period.  Accessed under
> + * rcu_registry_lock.
> + */
> +static NotifierList force_rcu_notifiers =
> +    NOTIFIER_LIST_INITIALIZER(force_rcu_notifiers);
> +
>   /*
>    * Check whether a quiescent state was crossed between the beginning of
>    * update_counter_and_wait and now.
> @@ -107,6 +115,8 @@ static void wait_for_readers(void)
>                    * get some extra futex wakeups.
>                    */
>                   qatomic_set(&index->waiting, false);
> +            } else if (qatomic_read(&in_drain_call_rcu)) {
> +                notifier_list_notify(&force_rcu_notifiers, NULL);
>               }
>           }
>   

You can put the notifier in struct rcu_reader_data---this way it doesn't 
call all the notifiers but only those that are necessary to make progress.

While at it, I have a slight preference for a separate 
rcu_add_force_rcu_notifier API, but I can be convinced otherwise. :)

Paolo

Patch

diff --git a/include/qemu/rcu.h b/include/qemu/rcu.h
index 515d327cf11c..498e4e5e3479 100644
--- a/include/qemu/rcu.h
+++ b/include/qemu/rcu.h
@@ -27,6 +27,7 @@ 
 #include "qemu/thread.h"
 #include "qemu/queue.h"
 #include "qemu/atomic.h"
+#include "qemu/notify.h"
 #include "qemu/sys_membarrier.h"
 
 #ifdef __cplusplus
@@ -66,6 +67,13 @@  struct rcu_reader_data {
 
     /* Data used for registry, protected by rcu_registry_lock */
     QLIST_ENTRY(rcu_reader_data) node;
+
+    /*
+     * Notifier used to force an RCU grace period.  Accessed under
+     * rcu_registry_lock.  Note that the notifier is called _outside_
+     * the thread!
+     */
+    Notifier *force_rcu;
 };
 
 extern __thread struct rcu_reader_data rcu_reader;
@@ -114,8 +122,19 @@  extern void synchronize_rcu(void);
 
 /*
  * Reader thread registration.
+ *
+ * The caller can specify an optional notifier if it wants RCU
+ * to enforce grace periods. This is needed by drain_call_rcu().
+ * Note that the notifier is executed in the context of the RCU
+ * thread.
  */
-extern void rcu_register_thread(void);
+extern void rcu_register_thread_with_force_rcu(Notifier *n);
+
+static inline void rcu_register_thread(void)
+{
+    rcu_register_thread_with_force_rcu(NULL);
+}
+
 extern void rcu_unregister_thread(void);
 
 /*
diff --git a/util/rcu.c b/util/rcu.c
index 13ac0f75cb2a..da3506917fa8 100644
--- a/util/rcu.c
+++ b/util/rcu.c
@@ -46,9 +46,17 @@ 
 unsigned long rcu_gp_ctr = RCU_GP_LOCKED;
 
 QemuEvent rcu_gp_event;
+static int in_drain_call_rcu;
 static QemuMutex rcu_registry_lock;
 static QemuMutex rcu_sync_lock;
 
+/*
+ * NotifierList used to force an RCU grace period.  Accessed under
+ * rcu_registry_lock.
+ */
+static NotifierList force_rcu_notifiers =
+    NOTIFIER_LIST_INITIALIZER(force_rcu_notifiers);
+
 /*
  * Check whether a quiescent state was crossed between the beginning of
  * update_counter_and_wait and now.
@@ -107,6 +115,8 @@  static void wait_for_readers(void)
                  * get some extra futex wakeups.
                  */
                 qatomic_set(&index->waiting, false);
+            } else if (qatomic_read(&in_drain_call_rcu)) {
+                notifier_list_notify(&force_rcu_notifiers, NULL);
             }
         }
 
@@ -293,7 +303,6 @@  void call_rcu1(struct rcu_head *node, void (*func)(struct rcu_head *node))
     qemu_event_set(&rcu_call_ready_event);
 }
 
-
 struct rcu_drain {
     struct rcu_head rcu;
     QemuEvent drain_complete_event;
@@ -339,8 +348,10 @@  void drain_call_rcu(void)
      * assumed.
      */
 
+    qatomic_inc(&in_drain_call_rcu);
     call_rcu1(&rcu_drain.rcu, drain_rcu_callback);
     qemu_event_wait(&rcu_drain.drain_complete_event);
+    qatomic_dec(&in_drain_call_rcu);
 
     if (locked) {
         qemu_mutex_lock_iothread();
@@ -348,10 +359,14 @@  void drain_call_rcu(void)
 
 }
 
-void rcu_register_thread(void)
+void rcu_register_thread_with_force_rcu(Notifier *n)
 {
     assert(rcu_reader.ctr == 0);
     qemu_mutex_lock(&rcu_registry_lock);
+    if (n) {
+        rcu_reader.force_rcu = n;
+        notifier_list_add(&force_rcu_notifiers, rcu_reader.force_rcu);
+    }
     QLIST_INSERT_HEAD(&registry, &rcu_reader, node);
     qemu_mutex_unlock(&rcu_registry_lock);
 }
@@ -360,6 +375,10 @@  void rcu_unregister_thread(void)
 {
     qemu_mutex_lock(&rcu_registry_lock);
     QLIST_REMOVE(&rcu_reader, node);
+    if (rcu_reader.force_rcu) {
+        notifier_remove(rcu_reader.force_rcu);
+        rcu_reader.force_rcu = NULL;
+    }
     qemu_mutex_unlock(&rcu_registry_lock);
 }