[Qemu-devel] [PATCH 0/2][RFC] Convert Clock lists to RCU (V3)

[Qemu-devel] [PATCH 0/2][RFC] Convert Clock lists to RCU (V3)

[Mike Day]

The first patch in the series converts the
clock->timerlists->active_timers list to RCU. It also creates
QemuMutex in the parent QemuClock data structure.

The second patch converts clock->timerlists to RCU.

Mike Day (2):
  [RFC] Convert active timers list to use RCU V3
  [RFC] Convert Clock Timerlists to RCU V3

 include/qemu/timer.h |  19 ++++---
 qemu-timer.c         | 142 ++++++++++++++++++++++++++++++++-------------------
 2 files changed, 98 insertions(+), 63 deletions(-)

-- 
1.9.0

[Qemu-devel] [PATCH 1/2] [RFC] Convert active timers list to use RCU V3

[Mike Day]

active_timers is a list of timer lists, associated with a Qemu Clock,
that is read-mostly. This patch converts read accesses to the list to
use RCU, which should hopefully mitigate most of the synchronization
overhead. Write accesses are now via a mutex in the parent data
structure. Functions that change the change the parent data structure
are also protected by the mutex.

This patch applies against Paolo Bonzini's rcu branch:
https://github.com/bonzini/qemu/tree/rcu

V3:
- Address comments from Alex Bligh and Paolo Bonzini
- Move the mutex into the parent QemuClock structure

Signed-off-by: Mike Day <ncmike@ncultra.org>
---
 include/qemu/timer.h |  19 +++++----
 qemu-timer.c         | 111 +++++++++++++++++++++++++++++----------------------
 2 files changed, 72 insertions(+), 58 deletions(-)

diff --git a/include/qemu/timer.h b/include/qemu/timer.h
index 5afcffc..f69ec49 100644
--- a/include/qemu/timer.h
+++ b/include/qemu/timer.h
@@ -5,8 +5,15 @@
 #include "qemu-common.h"
 #include "qemu/notify.h"
 
-/* timers */
+#ifdef __GNUC__
+#ifndef __ATOMIC_RELEASE
+#define __ATOMIC_RELEASE
+#endif
+#endif
+#include "qemu/atomic.h"
+#include "qemu/rcu.h"
 
+/* timers */
 #define SCALE_MS 1000000
 #define SCALE_US 1000
 #define SCALE_NS 1
@@ -61,6 +68,7 @@ struct QEMUTimer {
     void *opaque;
     QEMUTimer *next;
     int scale;
+    struct rcu_head rcu;
 };
 
 extern QEMUTimerListGroup main_loop_tlg;
@@ -189,12 +197,6 @@ void qemu_clock_notify(QEMUClockType type);
  * @enabled: true to enable, false to disable
  *
  * Enable or disable a clock
- * Disabling the clock will wait for related timerlists to stop
- * executing qemu_run_timers.  Thus, this functions should not
- * be used from the callback of a timer that is based on @clock.
- * Doing so would cause a deadlock.
- *
- * Caller should hold BQL.
  */
 void qemu_clock_enable(QEMUClockType type, bool enabled);
 
@@ -543,7 +545,6 @@ void timer_del(QEMUTimer *ts);
  * freed while this function is running.
  */
 void timer_mod_ns(QEMUTimer *ts, int64_t expire_time);
-
 /**
  * timer_mod_anticipate_ns:
  * @ts: the timer
@@ -685,9 +686,7 @@ static inline int64_t qemu_soonest_timeout(int64_t timeout1, int64_t timeout2)
 void init_clocks(void);
 
 int64_t cpu_get_ticks(void);
-/* Caller must hold BQL */
 void cpu_enable_ticks(void);
-/* Caller must hold BQL */
 void cpu_disable_ticks(void);
 
 static inline int64_t get_ticks_per_sec(void)
diff --git a/qemu-timer.c b/qemu-timer.c
index fb9e680..57a1545 100644
--- a/qemu-timer.c
+++ b/qemu-timer.c
@@ -29,6 +29,7 @@
 #include "hw/hw.h"
 
 #include "qemu/timer.h"
+#include "qemu/rcu_queue.h"
 #ifdef CONFIG_POSIX
 #include <pthread.h>
 #endif
@@ -45,12 +46,10 @@
 /* timers */
 
 typedef struct QEMUClock {
-    /* We rely on BQL to protect the timerlists */
     QLIST_HEAD(, QEMUTimerList) timerlists;
-
+    QemuMutex timer_lock;
     NotifierList reset_notifiers;
     int64_t last;
-
     QEMUClockType type;
     bool enabled;
 
@@ -70,11 +69,11 @@ QEMUClock qemu_clocks[QEMU_CLOCK_MAX];
 
 struct QEMUTimerList {
     QEMUClock *clock;
-    QemuMutex active_timers_lock;
     QEMUTimer *active_timers;
     QLIST_ENTRY(QEMUTimerList) list;
     QEMUTimerListNotifyCB *notify_cb;
     void *notify_opaque;
+    QemuEvent timers_done_ev;
 };
 
 /**
@@ -87,6 +86,7 @@ struct QEMUTimerList {
  */
 static inline QEMUClock *qemu_clock_ptr(QEMUClockType type)
 {
+    smp_rmb();
     return &qemu_clocks[type];
 }
 
@@ -107,7 +107,6 @@ QEMUTimerList *timerlist_new(QEMUClockType type,
     timer_list->clock = clock;
     timer_list->notify_cb = cb;
     timer_list->notify_opaque = opaque;
-    qemu_mutex_init(&timer_list->active_timers_lock);
     QLIST_INSERT_HEAD(&clock->timerlists, timer_list, list);
     return timer_list;
 }
@@ -118,7 +117,7 @@ void timerlist_free(QEMUTimerList *timer_list)
     if (timer_list->clock) {
         QLIST_REMOVE(timer_list, list);
     }
-    qemu_mutex_destroy(&timer_list->active_timers_lock);
+    qemu_event_destroy(&timer_list->timers_done_ev);
     g_free(timer_list);
 }
 
@@ -129,6 +128,7 @@ static void qemu_clock_init(QEMUClockType type)
     clock->type = type;
     clock->enabled = true;
     clock->last = INT64_MIN;
+    qemu_mutex_init(&clock->timer_lock);
     QLIST_INIT(&clock->timerlists);
     notifier_list_init(&clock->reset_notifiers);
     main_loop_tlg.tl[type] = timerlist_new(type, NULL, NULL);
@@ -148,13 +148,6 @@ void qemu_clock_notify(QEMUClockType type)
     }
 }
 
-/* Disabling the clock will wait for related timerlists to stop
- * executing qemu_run_timers.  Thus, this functions should not
- * be used from the callback of a timer that is based on @clock.
- * Doing so would cause a deadlock.
- *
- * Caller should hold BQL.
- */
 void qemu_clock_enable(QEMUClockType type, bool enabled)
 {
     QEMUClock *clock = qemu_clock_ptr(type);
@@ -172,7 +165,7 @@ void qemu_clock_enable(QEMUClockType type, bool enabled)
 
 bool timerlist_has_timers(QEMUTimerList *timer_list)
 {
-    return !!timer_list->active_timers;
+    return !!atomic_rcu_read(&timer_list->active_timers);
 }
 
 bool qemu_clock_has_timers(QEMUClockType type)
@@ -184,16 +177,18 @@ bool qemu_clock_has_timers(QEMUClockType type)
 bool timerlist_expired(QEMUTimerList *timer_list)
 {
     int64_t expire_time;
+    bool ret;
 
-    qemu_mutex_lock(&timer_list->active_timers_lock);
-    if (!timer_list->active_timers) {
-        qemu_mutex_unlock(&timer_list->active_timers_lock);
+    rcu_read_lock();
+    if (!atomic_rcu_read(&timer_list->active_timers)) {
+        rcu_read_unlock();
         return false;
     }
+    int type = timer_list->clock->type;
     expire_time = timer_list->active_timers->expire_time;
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
-
-    return expire_time < qemu_clock_get_ns(timer_list->clock->type);
+    rcu_read_unlock();
+    ret = (expire_time < qemu_clock_get_ns(type));
+    return ret;
 }
 
 bool qemu_clock_expired(QEMUClockType type)
@@ -220,16 +215,17 @@ int64_t timerlist_deadline_ns(QEMUTimerList *timer_list)
      * value but ->notify_cb() is called when the deadline changes.  Therefore
      * the caller should notice the change and there is no race condition.
      */
-    qemu_mutex_lock(&timer_list->active_timers_lock);
-    if (!timer_list->active_timers) {
-        qemu_mutex_unlock(&timer_list->active_timers_lock);
+
+    rcu_read_lock();
+    if (!atomic_rcu_read(&timer_list->active_timers)) {
+        rcu_read_unlock();
         return -1;
     }
+    int type = timer_list->clock->type;
     expire_time = timer_list->active_timers->expire_time;
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
-
-    delta = expire_time - qemu_clock_get_ns(timer_list->clock->type);
+    delta = expire_time - qemu_clock_get_ns(type);
 
+    rcu_read_unlock();
     if (delta <= 0) {
         return 0;
     }
@@ -332,11 +328,18 @@ void timer_init(QEMUTimer *ts,
     ts->expire_time = -1;
 }
 
-void timer_free(QEMUTimer *ts)
+static void reclaim_timer(struct rcu_head *rcu)
 {
+    QEMUTimer *ts = container_of(rcu, QEMUTimer, rcu);
     g_free(ts);
 }
 
+void timer_free(QEMUTimer *ts)
+{
+    call_rcu1(&ts->rcu, reclaim_timer);
+}
+
+
 static void timer_del_locked(QEMUTimerList *timer_list, QEMUTimer *ts)
 {
     QEMUTimer **pt, *t;
@@ -344,6 +347,8 @@ static void timer_del_locked(QEMUTimerList *timer_list, QEMUTimer *ts)
     ts->expire_time = -1;
     pt = &timer_list->active_timers;
     for(;;) {
+        /* caller's lock causes cache updates, so we don't need to use */
+        /* atomic_rcu_read or atomic_rcu_set */
         t = *pt;
         if (!t)
             break;
@@ -372,7 +377,6 @@ static bool timer_mod_ns_locked(QEMUTimerList *timer_list,
     ts->expire_time = MAX(expire_time, 0);
     ts->next = *pt;
     *pt = ts;
-
     return pt == &timer_list->active_timers;
 }
 
@@ -386,24 +390,30 @@ static void timerlist_rearm(QEMUTimerList *timer_list)
 /* stop a timer, but do not dealloc it */
 void timer_del(QEMUTimer *ts)
 {
-    QEMUTimerList *timer_list = ts->timer_list;
 
-    qemu_mutex_lock(&timer_list->active_timers_lock);
+    rcu_read_lock();
+    QEMUTimerList *timer_list = ts->timer_list;
+    QemuMutex *lock = &timer_list->clock->timer_lock;
+    qemu_mutex_lock(lock);
+    rcu_read_unlock();
     timer_del_locked(timer_list, ts);
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
+    qemu_mutex_unlock(lock);
 }
 
 /* modify the current timer so that it will be fired when current_time
    >= expire_time. The corresponding callback will be called. */
 void timer_mod_ns(QEMUTimer *ts, int64_t expire_time)
 {
-    QEMUTimerList *timer_list = ts->timer_list;
-    bool rearm;
 
-    qemu_mutex_lock(&timer_list->active_timers_lock);
+    bool rearm;
+    rcu_read_lock();
+    QEMUTimerList *timer_list = ts->timer_list;
+    QemuMutex *lock = &timer_list->clock->timer_lock;
+    qemu_mutex_lock(lock);
+    rcu_read_unlock();
     timer_del_locked(timer_list, ts);
     rearm = timer_mod_ns_locked(timer_list, ts, expire_time);
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
+    qemu_mutex_unlock(lock);
 
     if (rearm) {
         timerlist_rearm(timer_list);
@@ -415,19 +425,19 @@ void timer_mod_ns(QEMUTimer *ts, int64_t expire_time)
    The corresponding callback will be called. */
 void timer_mod_anticipate_ns(QEMUTimer *ts, int64_t expire_time)
 {
+    bool rearm = false;
+    rcu_read_lock();
     QEMUTimerList *timer_list = ts->timer_list;
-    bool rearm;
-
-    qemu_mutex_lock(&timer_list->active_timers_lock);
+    QemuMutex *lock = &timer_list->clock->timer_lock;
+    qemu_mutex_lock(lock);
+    rcu_read_unlock();
     if (ts->expire_time == -1 || ts->expire_time > expire_time) {
         if (ts->expire_time != -1) {
             timer_del_locked(timer_list, ts);
+            rearm = timer_mod_ns_locked(timer_list, ts, expire_time);
         }
-        rearm = timer_mod_ns_locked(timer_list, ts, expire_time);
-    } else {
-        rearm = false;
     }
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
+    qemu_mutex_unlock(lock);
 
     if (rearm) {
         timerlist_rearm(timer_list);
@@ -462,17 +472,22 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
     QEMUTimerCB *cb;
     void *opaque;
 
+    rcu_read_lock();
+    QemuMutex *lock = &timer_list->clock->timer_lock;
     qemu_event_reset(&timer_list->timers_done_ev);
+    rcu_read_unlock();
     if (!timer_list->clock->enabled) {
         goto out;
     }
-
+    /* timer_list should also be protected here. */
+    /* that is the subject of the next patch. */
+    /* (which will also remove this comment). */
     current_time = qemu_clock_get_ns(timer_list->clock->type);
     for(;;) {
-        qemu_mutex_lock(&timer_list->active_timers_lock);
+        qemu_mutex_lock(lock);
         ts = timer_list->active_timers;
         if (!timer_expired_ns(ts, current_time)) {
-            qemu_mutex_unlock(&timer_list->active_timers_lock);
+            qemu_mutex_unlock(lock);
             break;
         }
 
@@ -482,13 +497,13 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
         ts->expire_time = -1;
         cb = ts->cb;
         opaque = ts->opaque;
-        qemu_mutex_unlock(&timer_list->active_timers_lock);
-
+        rcu_read_lock();
+        qemu_mutex_unlock(lock);
         /* run the callback (the timer list can be modified) */
         cb(opaque);
+        rcu_read_unlock();
         progress = true;
     }
-
 out:
     qemu_event_set(&timer_list->timers_done_ev);
     return progress;
-- 
1.9.0

[Qemu-devel] [PATCH 2/2] [RFC] Convert Clock Timerlists to RCU V3

[Mike Day]

timerlists is a list of lists that holds active timers, among other
items. It is read-mostly. This patch converts read access to the
timerlists to use RCU.

Rather than introduce a second mutex for timerlists, which would
require nested mutexes to in order to write to the timerlists, use one
QemuMutex in the QemuClock structure for all write access to any list
hanging off the QemuClock structure.

This patch applies against Paolo Bonzini's rcu branch:
https://github.com/bonzini/qemu/tree/rcu and also requires the
previously submitted patch ae11e1c "Convert active timers list to use
RCU for read operations V3."

V3:
- timerlists modifications split to a separate patch (this one).
- Addressed comments from Alex Bligh and Paolo Bonzini.

Signed-off-by: Mike Day <ncmike@ncultra.org>
---
 qemu-timer.c | 31 ++++++++++++++++++++++++++-----
 1 file changed, 26 insertions(+), 5 deletions(-)

diff --git a/qemu-timer.c b/qemu-timer.c
index 57a1545..4144e54 100644
--- a/qemu-timer.c
+++ b/qemu-timer.c
@@ -74,6 +74,7 @@ struct QEMUTimerList {
     QEMUTimerListNotifyCB *notify_cb;
     void *notify_opaque;
     QemuEvent timers_done_ev;
+    struct rcu_head rcu;
 };
 
 /**
@@ -111,6 +112,13 @@ QEMUTimerList *timerlist_new(QEMUClockType type,
     return timer_list;
 }
 
+static void reclaim_timerlist(struct rcu_head *rcu)
+{
+    QEMUTimerList *tl = container_of(rcu, QEMUTimerList, rcu);
+    g_free(tl);
+}
+
+
 void timerlist_free(QEMUTimerList *timer_list)
 {
     assert(!timerlist_has_timers(timer_list));
@@ -118,7 +126,7 @@ void timerlist_free(QEMUTimerList *timer_list)
         QLIST_REMOVE(timer_list, list);
     }
     qemu_event_destroy(&timer_list->timers_done_ev);
-    g_free(timer_list);
+    call_rcu1(&timer_list->rcu, reclaim_timerlist);
 }
 
 static void qemu_clock_init(QEMUClockType type)
@@ -143,9 +151,11 @@ void qemu_clock_notify(QEMUClockType type)
 {
     QEMUTimerList *timer_list;
     QEMUClock *clock = qemu_clock_ptr(type);
-    QLIST_FOREACH(timer_list, &clock->timerlists, list) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(timer_list, &clock->timerlists, list) {
         timerlist_notify(timer_list);
     }
+    rcu_read_unlock();
 }
 
 void qemu_clock_enable(QEMUClockType type, bool enabled)
@@ -157,9 +167,11 @@ void qemu_clock_enable(QEMUClockType type, bool enabled)
     if (enabled && !old) {
         qemu_clock_notify(type);
     } else if (!enabled && old) {
-        QLIST_FOREACH(tl, &clock->timerlists, list) {
+        rcu_read_lock();
+        QLIST_FOREACH_RCU(tl, &clock->timerlists, list) {
             qemu_event_wait(&tl->timers_done_ev);
         }
+        rcu_read_unlock();
     }
 }
 
@@ -243,10 +255,12 @@ int64_t qemu_clock_deadline_ns_all(QEMUClockType type)
     int64_t deadline = -1;
     QEMUTimerList *timer_list;
     QEMUClock *clock = qemu_clock_ptr(type);
-    QLIST_FOREACH(timer_list, &clock->timerlists, list) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(timer_list, &clock->timerlists, list) {
         deadline = qemu_soonest_timeout(deadline,
                                         timerlist_deadline_ns(timer_list));
     }
+    rcu_read_unlock();
     return deadline;
 }
 
@@ -262,11 +276,13 @@ QEMUTimerList *qemu_clock_get_main_loop_timerlist(QEMUClockType type)
 
 void timerlist_notify(QEMUTimerList *timer_list)
 {
-    if (timer_list->notify_cb) {
+    rcu_read_lock();
+    if (atomic_rcu_read(&timer_list->notify_cb)) {
         timer_list->notify_cb(timer_list->notify_opaque);
     } else {
         qemu_notify_event();
     }
+    rcu_read_unlock();
 }
 
 /* Transition function to convert a nanosecond timeout to ms
@@ -585,13 +601,18 @@ void qemu_clock_register_reset_notifier(QEMUClockType type,
                                         Notifier *notifier)
 {
     QEMUClock *clock = qemu_clock_ptr(type);
+    qemu_mutex_lock(&clock->timer_lock);
     notifier_list_add(&clock->reset_notifiers, notifier);
+    qemu_mutex_unlock(&clock->timer_lock);
 }
 
 void qemu_clock_unregister_reset_notifier(QEMUClockType type,
                                           Notifier *notifier)
 {
+    QEMUClock *clock = qemu_clock_ptr(type);
+    qemu_mutex_lock(&clock->timer_lock);
     notifier_remove(notifier);
+    qemu_mutex_unlock(&clock->timer_lock);
 }
 
 void init_clocks(void)
-- 
1.9.0