[Qemu-devel] [RFC PATCH 4/3] qemu-timer: use RCU to preserve the timers during lockless lookup

[Paolo Bonzini]

This patch uses RCU to access the active timers list with no lock.  The
access can race with concurrent timer_mod calls, but most of our read-side
critical sections are racy anyway and rely on aio_notify being called
whenever the head of the list changes.  RCU is used simply to ensure that
the timer head does not go away while it is being examined lockless; thus
timer_mod is unchanged and only timer_free has to wait for a grace period.

The only case where we have to take care is in qemu_run_timers.  In this
case, "upgrading" the critical section by taking the mutex is not enough,
because the code relies on the timer being at the head of the list.  Thus,
we fetch the head again after taking the mutex.  The cost is two extra
memory reads for fired timer, and the savings is one mutex lock/unlock
per call to qemu_run_timers, so there is a net advantage.

Signed-off-by: Paolo Bonzini <address@hidden>
---
 qemu-timer.c | 49 +++++++++++++++++++++++++++++++++++++------------
 1 file changed, 37 insertions(+), 12 deletions(-)

diff --git a/qemu-timer.c b/qemu-timer.c
index 807319a..9a9f52d 100644
--- a/qemu-timer.c
+++ b/qemu-timer.c
@@ -167,14 +167,16 @@ bool qemu_clock_has_timers(QEMUClockType type)
 bool timerlist_expired(QEMUTimerList *timer_list)
 {
     int64_t expire_time;
+    QEMUTimer *timer;
 
-    qemu_mutex_lock(&timer_list->active_timers_lock);
-    if (!timer_list->active_timers) {
-        qemu_mutex_unlock(&timer_list->active_timers_lock);
+    rcu_read_lock();
+    timer = atomic_rcu_read(&timer_list->active_timers);
+    if (!timer) {
+        rcu_read_unlock();
         return false;
     }
-    expire_time = timer_list->active_timers->expire_time;
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
+    expire_time = timer->expire_time;
+    rcu_read_unlock();
 
     return expire_time < qemu_clock_get_ns(timer_list->clock->type);
 }
@@ -192,6 +195,7 @@ bool qemu_clock_expired(QEMUClockType type)
 
 int64_t timerlist_deadline_ns(QEMUTimerList *timer_list)
 {
+    QEMUTimer *timer;
     int64_t delta;
     int64_t expire_time;
 
@@ -203,13 +207,14 @@ 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();
+    timer = atomic_rcu_read(&timer_list->active_timers);
+    if (!timer) {
+        rcu_read_unlock();
         return -1;
     }
-    expire_time = timer_list->active_timers->expire_time;
-    qemu_mutex_unlock(&timer_list->active_timers_lock);
+    expire_time = timer->expire_time;
+    rcu_read_unlock();
 
     delta = expire_time - qemu_clock_get_ns(timer_list->clock->type);
 
@@ -318,6 +323,8 @@ void timer_init(QEMUTimer *ts,
 void timer_free(QEMUTimer *ts)
 {
     timer_del(ts);
+
+    /* TODO: use call_rcu to free.  */
     g_free(ts);
 }
 
@@ -370,7 +377,7 @@ void timer_mod_ns(QEMUTimer *ts, int64_t expire_time)
     }
     ts->expire_time = MAX(expire_time, 0);
     ts->next = *pt;
-    *pt = ts;
+    atomic_rcu_set(pt, ts);
     qemu_mutex_unlock(&timer_list->active_timers_lock);
 
     /* Rearm if necessary  */
@@ -410,6 +417,22 @@ bool timerlist_run_timers(QEMUTimerList *timer_list)
 
     current_time = qemu_clock_get_ns(timer_list->clock->type);
     for(;;) {
+        /* Check if the head of the list might be expired.  If a
+         * deletion or modification happens concurrently, we will
+         * detect it below with the lock taken.
+         */
+        rcu_read_lock();
+        ts = atomic_rcu_read(&timer_list->active_timers);
+        if (!timer_expired_ns(ts, current_time)) {
+            rcu_read_unlock();
+            break;
+        }
+        rcu_read_unlock();
+
+        /* A timer might be ready to fire.  Check it again with the lock
+         * taken to be safe against concurrent timer_mod; but in the common
+         * case no timer is ready and we can do everything lockless.
+         */
         qemu_mutex_lock(&timer_list->active_timers_lock);
         ts = timer_list->active_timers;
         if (!timer_expired_ns(ts, current_time)) {
-- 
1.8.3.1