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