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

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

[Mike Day]

The first patch in the series convers the
clock->timerlists->active_timers list to RCU. The second patch
converts clock->timerlists to RCU and also protects access to
timerlists->active_timers->timer_list.

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

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

-- 
1.8.5.3

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

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

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

V2:
- Addresses comments from Alex Bligh

Signed-off-by: Mike Day <ncmike@ncultra.org>
---
 include/qemu/timer.h | 19 +++++++--------
 qemu-timer.c         | 69 ++++++++++++++++++++++++++--------------------------
 2 files changed, 44 insertions(+), 44 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..dad30a3 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;
 
     NotifierList reset_notifiers;
     int64_t last;
-
     QEMUClockType type;
     bool enabled;
 
@@ -75,6 +74,7 @@ struct QEMUTimerList {
     QLIST_ENTRY(QEMUTimerList) list;
     QEMUTimerListNotifyCB *notify_cb;
     void *notify_opaque;
+    QemuEvent timers_done_ev;
 };
 
 /**
@@ -87,6 +87,7 @@ struct QEMUTimerList {
  */
 static inline QEMUClock *qemu_clock_ptr(QEMUClockType type)
 {
+    smp_rmb();
     return &qemu_clocks[type];
 }
 
@@ -148,13 +149,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 +166,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 +178,17 @@ 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;
     }
     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);
+    ret = (expire_time < qemu_clock_get_ns(timer_list->clock->type));
+    rcu_read_unlock();
+    return ret;
 }
 
 bool qemu_clock_expired(QEMUClockType type)
@@ -220,16 +215,16 @@ 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;
     }
     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);
 
+    rcu_read_unlock();
     if (delta <= 0) {
         return 0;
     }
@@ -332,11 +327,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 +346,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 +376,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;
 }
 
@@ -416,16 +419,14 @@ void timer_mod_ns(QEMUTimer *ts, int64_t expire_time)
 void timer_mod_anticipate_ns(QEMUTimer *ts, int64_t expire_time)
 {
     QEMUTimerList *timer_list = ts->timer_list;
-    bool rearm;
+    bool rearm = false;
 
     qemu_mutex_lock(&timer_list->active_timers_lock);
     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);
 
@@ -461,12 +462,13 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
     bool progress = false;
     QEMUTimerCB *cb;
     void *opaque;
+    bool enabled;
 
-    qemu_event_reset(&timer_list->timers_done_ev);
-    if (!timer_list->clock->enabled) {
-        goto out;
+    enabled = atomic_rcu_read(&timer_list->clock->enabled);
+    if (!enabled) {
+        return progress;
     }
-
+    qemu_event_reset(&timer_list->timers_done_ev);
     current_time = qemu_clock_get_ns(timer_list->clock->type);
     for(;;) {
         qemu_mutex_lock(&timer_list->active_timers_lock);
@@ -482,14 +484,13 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
         ts->expire_time = -1;
         cb = ts->cb;
         opaque = ts->opaque;
+        rcu_read_lock();
         qemu_mutex_unlock(&timer_list->active_timers_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.8.5.3

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

[Alex Bligh]

Mike,

This one's nice and clear. A few comments in line.

On 27 Feb 2014, at 19:35, Mike Day wrote:

> @@ -184,16 +178,17 @@ 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;
>     }
>     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);
> +    ret = (expire_time < qemu_clock_get_ns(timer_list->clock->type));
> +    rcu_read_unlock();
> +    return ret;
> }

I think it probably makes little difference, but why not:

       int type = timerlist->clock->type;
       rcu_read_unlock();
       return expire_time < qemu_clock_get_ns(type);



> bool qemu_clock_expired(QEMUClockType type)
> @@ -220,16 +215,16 @@ 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;
>     }
>     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);
> 
> +    rcu_read_unlock();
>     if (delta <= 0) {
>         return 0;
>     }

Similarly perhaps save 'type' and read the clock outside the
rcu read-sde lock.
> 
> static void timer_del_locked(QEMUTimerList *timer_list, QEMUTimer *ts)
> {
>     QEMUTimer **pt, *t;
> @@ -344,6 +346,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;

I'm showing my RCU ignorance here, but does that mean we should be doing
an rmb() after taking the lock in (e.g.) timer_mod_ns? Or are mutexes
guaranteed to rmb()?

> @@ -461,12 +462,13 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
>     bool progress = false;
>     QEMUTimerCB *cb;
>     void *opaque;
> +    bool enabled;
> 
> -    qemu_event_reset(&timer_list->timers_done_ev);
> -    if (!timer_list->clock->enabled) {
> -        goto out;
> +    enabled = atomic_rcu_read(&timer_list->clock->enabled);
> +    if (!enabled) {
> +        return progress;
>     }
> -
> +    qemu_event_reset(&timer_list->timers_done_ev);
>     current_time = qemu_clock_get_ns(timer_list->clock->type);
>     for(;;) {
>         qemu_mutex_lock(&timer_list->active_timers_lock);

What's the introduction of 'enabled' for? Why not simply

    if (!atomic_rcu_read(&timer_list->clock->enabled)) {
        return progress;
    }

Also, previously if called on a clock that was disabled, this would set and
reset the qemu_event, which would wake up any waiters. It no longer
does that. Is this change intended?

> @@ -482,14 +484,13 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
>         ts->expire_time = -1;
>         cb = ts->cb;
>         opaque = ts->opaque;
> +        rcu_read_lock();
>         qemu_mutex_unlock(&timer_list->active_timers_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;
> }

Again showing my RCU ignorance here, but do qemu_mutex_lock() and
qemu_mutex_unlock() automatically take and drop the rcu read lock?
If not, we'd need to hold that rcu_read_lock whenever the mutex
is locked.

-- 
Alex Bligh

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

[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 orderwrite to the timerlists, use one
QemuMutex in the QemuClock structure for all write access to any list
hanging off the QemuClock structure.

This patch also protects timerlists->active_timers->timer_list by the
new clock mutex for write access and by RCU for read access.

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

V2:
- timerlists modifications split to a separate patch (this one).
- Addressed Alex Blighs comments.

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

diff --git a/qemu-timer.c b/qemu-timer.c
index dad30a3..e335a7a 100644
--- a/qemu-timer.c
+++ b/qemu-timer.c
@@ -47,7 +47,7 @@
 
 typedef struct QEMUClock {
     QLIST_HEAD(, QEMUTimerList) timerlists;
-
+    QemuMutex clock_mutex;
     NotifierList reset_notifiers;
     int64_t last;
     QEMUClockType type;
@@ -69,12 +69,12 @@ 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;
+    struct rcu_head rcu;
 };
 
 /**
@@ -108,19 +108,24 @@ 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);
+    qemu_mutex_init(&clock->clock_mutex);
     QLIST_INSERT_HEAD(&clock->timerlists, timer_list, list);
     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));
     if (timer_list->clock) {
         QLIST_REMOVE(timer_list, list);
     }
-    qemu_mutex_destroy(&timer_list->active_timers_lock);
-    g_free(timer_list);
+    call_rcu1(&timer_list->rcu, reclaim_timerlist);
 }
 
 static void qemu_clock_init(QEMUClockType type)
@@ -144,9 +149,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)
@@ -154,13 +161,15 @@ void qemu_clock_enable(QEMUClockType type, bool enabled)
     QEMUClock *clock = qemu_clock_ptr(type);
     QEMUTimerList *tl;
     bool old = clock->enabled;
-    clock->enabled = enabled;
+    atomic_rcu_set(&clock->enabled, 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();
     }
 }
 
@@ -242,16 +251,18 @@ 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;
 }
 
 QEMUClockType timerlist_get_clock(QEMUTimerList *timer_list)
 {
-    return timer_list->clock->type;
+    return atomic_rcu_read(&timer_list->clock->type);
 }
 
 QEMUTimerList *qemu_clock_get_main_loop_timerlist(QEMUClockType type)
@@ -261,11 +272,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
@@ -389,24 +402,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;
+    QEMUTimerList *timer_list;
 
-    qemu_mutex_lock(&timer_list->active_timers_lock);
+    timer_list = atomic_rcu_read(&ts->timer_list);
+    rcu_read_lock();
+    qemu_mutex_lock(&timer_list->clock->clock_mutex);
+    rcu_read_unlock();
     timer_del_locked(timer_list, ts);
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
+    qemu_mutex_unlock(&timer_list->clock->clock_mutex);
 }
 
 /* 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;
+    QEMUTimerList *timer_list;
     bool rearm;
 
-    qemu_mutex_lock(&timer_list->active_timers_lock);
+    timer_list = atomic_rcu_read(&ts->timer_list);
+    rcu_read_lock();
+    qemu_mutex_lock(&timer_list->clock->clock_mutex);
+    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(&timer_list->clock->clock_mutex);
 
     if (rearm) {
         timerlist_rearm(timer_list);
@@ -418,18 +437,20 @@ 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)
 {
-    QEMUTimerList *timer_list = ts->timer_list;
+    QEMUTimerList *timer_list;
     bool rearm = false;
 
-    qemu_mutex_lock(&timer_list->active_timers_lock);
+    timer_list = atomic_rcu_read(&ts->timer_list);
+    rcu_read_lock();
+    qemu_mutex_lock(&timer_list->clock->clock_mutex);
+    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);
         }
     }
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
-
+    qemu_mutex_unlock(&timer_list->clock->clock_mutex);
     if (rearm) {
         timerlist_rearm(timer_list);
     }
@@ -447,7 +468,10 @@ void timer_mod_anticipate(QEMUTimer *ts, int64_t expire_time)
 
 bool timer_pending(QEMUTimer *ts)
 {
-    return ts->expire_time >= 0;
+    int64 expire_time;
+
+    expire_time = atomic_rcu_read(&ts->expire_time);
+    return expire_time >= 0;
 }
 
 bool timer_expired(QEMUTimer *timer_head, int64_t current_time)
@@ -471,10 +495,10 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
     qemu_event_reset(&timer_list->timers_done_ev);
     current_time = qemu_clock_get_ns(timer_list->clock->type);
     for(;;) {
-        qemu_mutex_lock(&timer_list->active_timers_lock);
+        qemu_mutex_lock(&timer_list->clock->clock_mutex);
         ts = timer_list->active_timers;
         if (!timer_expired_ns(ts, current_time)) {
-            qemu_mutex_unlock(&timer_list->active_timers_lock);
+            qemu_mutex_unlock(&timer_list->clock->clock_mutex);
             break;
         }
 
@@ -485,7 +509,7 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
         cb = ts->cb;
         opaque = ts->opaque;
         rcu_read_lock();
-        qemu_mutex_unlock(&timer_list->active_timers_lock);
+        qemu_mutex_unlock(&timer_list->clock->clock_mutex);
         /* run the callback (the timer list can be modified) */
         cb(opaque);
         rcu_read_unlock();
@@ -571,13 +595,18 @@ void qemu_clock_register_reset_notifier(QEMUClockType type,
                                         Notifier *notifier)
 {
     QEMUClock *clock = qemu_clock_ptr(type);
+    qemu_mutex_lock(&clock->clock_mutex);
     notifier_list_add(&clock->reset_notifiers, notifier);
+    qemu_mutex_unlock(&clock->clock_mutex);
 }
 
 void qemu_clock_unregister_reset_notifier(QEMUClockType type,
                                           Notifier *notifier)
 {
+    QEMUClock *clock = qemu_clock_ptr(type);
+    qemu_mutex_lock(&clock->clock_mutex);
     notifier_remove(notifier);
+    qemu_mutex_unlock(&clock->clock_mutex);
 }
 
 void init_clocks(void)
-- 
1.8.5.3

Re: [Qemu-devel] [PATCH 2/2] [RFC] Convert Clock Timerlists to RCU V2

[Alex Bligh]

Mike

On 27 Feb 2014, at 19:35, Mike Day wrote:

> 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 orderwrite to the timerlists, use one
> QemuMutex in the QemuClock structure for all write access to any list
> hanging off the QemuClock structure.

I sort of wonder why you don't just use one Mutex across the whole
of QEMU rather than one per clock.

This would save worrying about whether when you do things like:
  qemu_mutex_lock(&timer_list->clock->clock_mutex);

timer_list 'disappears' as (prior to taking the mutex) it's outside
the rcu read lock. 

> @@ -108,19 +108,24 @@ 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);
> +    qemu_mutex_init(&clock->clock_mutex);
>     QLIST_INSERT_HEAD(&clock->timerlists, timer_list, list);
>     return timer_list;
> }

That can't be right, you are calling qemu_mutex_init for each
timerlist, but the timerlists share the mutex which is now in the
clock. The mutex should therefore be initialized in qemu_clock_init.

Also, clock_mutex appears never to get destroyed.

> QEMUClockType timerlist_get_clock(QEMUTimerList *timer_list)
> {
> -    return timer_list->clock->type;
> +    return atomic_rcu_read(&timer_list->clock->type);
> }

I don't think this works because of the double indirection. It's
The '&' means it's not actually dereferencing clock->type, but it
is dereferencing timer_list->clock outside of either an rcu read
lock or an atomic read. I think you'd be better with than
rcu_read_lock() etc. (sadly).

> QEMUTimerList *qemu_clock_get_main_loop_timerlist(QEMUClockType type)
> @@ -261,11 +272,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();
> }

If you have the rcu_read_lock why do you need the atomic_rcu_read?
And if you need it, why do you not need it on the following line?

However, as any QEMUTimerList can (now) be reclaimed, surely all
callers of this function are going to need to hold the rcu_read_lock
anyway?

I think this is used only by timerlist_rearm and qemu_clock_notify.
Provided these call this function holding the rcu_read_lock
I don't think this function needs changing.

> /* Transition function to convert a nanosecond timeout to ms
> @@ -389,24 +402,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;
> +    QEMUTimerList *timer_list;
> 
> -    qemu_mutex_lock(&timer_list->active_timers_lock);
> +    timer_list = atomic_rcu_read(&ts->timer_list);
> +    rcu_read_lock();
> +    qemu_mutex_lock(&timer_list->clock->clock_mutex);
> +    rcu_read_unlock();
>     timer_del_locked(timer_list, ts);
> -    qemu_mutex_unlock(&timer_list->active_timers_lock);
> +    qemu_mutex_unlock(&timer_list->clock->clock_mutex);
> }

Again in my ignorance of RCU I don't know whether taking a mutex
implicitly takes an rcu read lock. If not, that rcu_read_unlock
should be after the mutex unlock.

> /* 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;
> +    QEMUTimerList *timer_list;
>     bool rearm;
> 
> -    qemu_mutex_lock(&timer_list->active_timers_lock);
> +    timer_list = atomic_rcu_read(&ts->timer_list);
> +    rcu_read_lock();
> +    qemu_mutex_lock(&timer_list->clock->clock_mutex);
> +    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(&timer_list->clock->clock_mutex);
> 
>     if (rearm) {
>         timerlist_rearm(timer_list);

Ditto

> @@ -418,18 +437,20 @@ 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)
> {
> -    QEMUTimerList *timer_list = ts->timer_list;
> +    QEMUTimerList *timer_list;
>     bool rearm = false;
> 
> -    qemu_mutex_lock(&timer_list->active_timers_lock);
> +    timer_list = atomic_rcu_read(&ts->timer_list);
> +    rcu_read_lock();
> +    qemu_mutex_lock(&timer_list->clock->clock_mutex);
> +    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);
>         }
>     }
> -    qemu_mutex_unlock(&timer_list->active_timers_lock);
> -
> +    qemu_mutex_unlock(&timer_list->clock->clock_mutex);
>     if (rearm) {
>         timerlist_rearm(timer_list);
>     }

Ditto

> @@ -447,7 +468,10 @@ void timer_mod_anticipate(QEMUTimer *ts, int64_t expire_time)
> 
> bool timer_pending(QEMUTimer *ts)
> {
> -    return ts->expire_time >= 0;
> +    int64 expire_time;
> +
> +    expire_time = atomic_rcu_read(&ts->expire_time);
> +    return expire_time >= 0;
> }

Why not just

       return atomic_rcu_read(&ts->expire_time) >= 0;

-- 
Alex Bligh

Re: [Qemu-devel] [PATCH 2/2] [RFC] Convert Clock Timerlists to RCU V2

[Mike Day]

On Fri, Feb 28, 2014 at 3:05 PM, Alex Bligh <alex@alex.org.uk> wrote:

>> Rather than introduce a second mutex for timerlists, which would
>> require nested mutexes to in orderwrite to the timerlists, use one
>> QemuMutex in the QemuClock structure for all write access to any list
>> hanging off the QemuClock structure.
>
> I sort of wonder why you don't just use one Mutex across the whole
> of QEMU rather than one per clock.
>
> This would save worrying about whether when you do things like:
>   qemu_mutex_lock(&timer_list->clock->clock_mutex);

I like it, it solves another problem I spotted after I submitted the patch.

>
>> @@ -108,19 +108,24 @@ 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);
>> +    qemu_mutex_init(&clock->clock_mutex);
>>     QLIST_INSERT_HEAD(&clock->timerlists, timer_list, list);
>>     return timer_list;
>> }
>
> That can't be right, you are calling qemu_mutex_init for each
> timerlist, but the timerlists share the mutex which is now in the
> clock. The mutex should therefore be initialized in qemu_clock_init.

> Also, clock_mutex appears never to get destroyed.

Yes, thank you, on both points.

>
>> QEMUClockType timerlist_get_clock(QEMUTimerList *timer_list)
>> {
>> -    return timer_list->clock->type;
>> +    return atomic_rcu_read(&timer_list->clock->type);
>> }
>
> I don't think this works because of the double indirection. It's
> The '&' means it's not actually dereferencing clock->type, but it
> is dereferencing timer_list->clock outside of either an rcu read
> lock or an atomic read. I think you'd be better with than
> rcu_read_lock() etc. (sadly).

I should fix this with parenthesis as in &(timer_list->clock->type)

>> QEMUTimerList *qemu_clock_get_main_loop_timerlist(QEMUClockType type)
>> @@ -261,11 +272,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();
>> }
>
> If you have the rcu_read_lock why do you need the atomic_rcu_read?
> And if you need it, why do you not need it on the following line?

rcu_read_lock/unlock and atomic_rcu_read/set do different things and
frequently need to be used together. rcu_read_lock/unlock causes the
thread to enter an RCU critical section by getting a counter out of
TLS. atomic_rcu_read/set will use memory barriers to flush caches
(depending on the memory model and platform) and ensure that reads and
writes are ordered. You typically would use atomic_rcu_read on the
first read, thereafter the memory is up-to-date and writes have been
flushed. And you typically use atomic_rcu_set on the last write, when
the data structure is complete. You don't need to use them on every
update. Just for completeness, rcu_read_unlock ends the rcu critical
section but its usually a no-op.

> However, as any QEMUTimerList can (now) be reclaimed, surely all
> callers of this function are going to need to hold the rcu_read_lock
> anyway?

Yes

> I think this is used only by timerlist_rearm and qemu_clock_notify.
> Provided these call this function holding the rcu_read_lock
> I don't think this function needs changing.
>
>> /* Transition function to convert a nanosecond timeout to ms
>> @@ -389,24 +402,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;
>> +    QEMUTimerList *timer_list;
>>
>> -    qemu_mutex_lock(&timer_list->active_timers_lock);
>> +    timer_list = atomic_rcu_read(&ts->timer_list);
>> +    rcu_read_lock();
>> +    qemu_mutex_lock(&timer_list->clock->clock_mutex);
>> +    rcu_read_unlock();
>>     timer_del_locked(timer_list, ts);
>> -    qemu_mutex_unlock(&timer_list->active_timers_lock);
>> +    qemu_mutex_unlock(&timer_list->clock->clock_mutex);
>> }
>
> Again in my ignorance of RCU I don't know whether taking a mutex
> implicitly takes an rcu read lock. If not, that rcu_read_unlock
> should be after the mutex unlock.

I am going to change this because of a race condition, To answer your
question, holding a mutex does not imply holding an rcu read lock. It
does indicate the potential need for readers to use rcu to read the
list because they can do so when the mutex holder is updating the
list. And, I'm working under the assumption that holding a mutex
implies a memory barrier so there is no need for atomic_rcu_read/set.

>
>> @@ -447,7 +468,10 @@ void timer_mod_anticipate(QEMUTimer *ts, int64_t expire_time)
>>
>> bool timer_pending(QEMUTimer *ts)
>> {
>> -    return ts->expire_time >= 0;
>> +    int64 expire_time;
>> +
>> +    expire_time = atomic_rcu_read(&ts->expire_time);
>> +    return expire_time >= 0;
>> }
>
> Why not just
>
>        return atomic_rcu_read(&ts->expire_time) >= 0;

That's better.

Thanks Alex!

Re: [Qemu-devel] [PATCH 2/2] [RFC] Convert Clock Timerlists to RCU V2

[Paolo Bonzini]

Il 28/02/2014 21:05, Alex Bligh ha scritto:
> Mike
>
> On 27 Feb 2014, at 19:35, Mike Day wrote:
>
>> 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 orderwrite to the timerlists, use one
>> QemuMutex in the QemuClock structure for all write access to any list
>> hanging off the QemuClock structure.
>
> I sort of wonder why you don't just use one Mutex across the whole
> of QEMU rather than one per clock.

I think this is not enough; separate iothreads could have separate 
timerlist, and higher-priority iothreads should not be slowed down by 
lower-priority iothreads.

>> QEMUClockType timerlist_get_clock(QEMUTimerList *timer_list)
>> {
>> -    return timer_list->clock->type;
>> +    return atomic_rcu_read(&timer_list->clock->type);
>> }
>
> I don't think this works because of the double indirection.

It doesn't but you can of course do this:

	QEMUClock *clock = atomic_rcu_read(&timer_list->clock);
	return atomic_rcu_read(&clock->type);

> It's The '&' means it's not actually dereferencing clock->type, but it
> is dereferencing timer_list->clock outside of either an rcu read
> lock or an atomic read. I think you'd be better with than
> rcu_read_lock() etc. (sadly).

You could also assume that the caller does it.  Right now, this function 
has no user, so you just have to document it.

>> QEMUTimerList *qemu_clock_get_main_loop_timerlist(QEMUClockType type)
>> @@ -261,11 +272,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();
>> }
>
> If you have the rcu_read_lock why do you need the atomic_rcu_read?
> And if you need it, why do you not need it on the following line?

Read the documentation. :)  It was also posted on the list.

atomic_rcu_read() is only about blocking invalid compiler optimizations; 
it is not required if you are in the *write* side, but it is always 
required in the read side.

However, I agree that it is not needed here.  atomic_rcu_read() is not 
needed when reading a "leaf" element of the data structure.

> However, as any QEMUTimerList can (now) be reclaimed, surely all
> callers of this function are going to need to hold the rcu_read_lock
> anyway?
>
> I think this is used only by timerlist_rearm and qemu_clock_notify.
> Provided these call this function holding the rcu_read_lock
> I don't think this function needs changing.

Yes.

Paolo