Re: [Qemu-devel] [PATCH 4/9] rcu: introduce RCU-enabled QLIST

[Fam Zheng]

On Wed, 02/04 13:46, Paolo Bonzini wrote:
> 
> 
> On 04/02/2015 04:42, Fam Zheng wrote:
> > On Tue, 02/03 13:52, Paolo Bonzini wrote:
> >> From: Mike Day <address@hidden>
> >>
> >> Add RCU-enabled variants on the existing bsd DQ facility. Each
> >> operation has the same interface as the existing (non-RCU)
> >> version. Also, each operation is implemented as macro.
> >>
> >> Using the RCU-enabled QLIST, existing QLIST users will be able to
> >> convert to RCU without using a different list interface.
> >>
> >> Signed-off-by: Mike Day <address@hidden>
> >> Signed-off-by: Paolo Bonzini <address@hidden>
> >> ---
> >>  hw/9pfs/virtio-9p-synth.c |   2 +-
> >>  include/qemu/queue.h      |  11 --
> >>  include/qemu/rcu_queue.h  | 134 ++++++++++++++++++++
> >>  tests/Makefile            |   5 +-
> >>  tests/test-rcu-list.c     | 306 
> >> ++++++++++++++++++++++++++++++++++++++++++++++
> >>  5 files changed, 445 insertions(+), 13 deletions(-)
> >>  create mode 100644 include/qemu/rcu_queue.h
> >>  create mode 100644 tests/test-rcu-list.c
> >>
> >> diff --git a/hw/9pfs/virtio-9p-synth.c b/hw/9pfs/virtio-9p-synth.c
> >> index e75aa87..a0ab9a8 100644
> >> --- a/hw/9pfs/virtio-9p-synth.c
> >> +++ b/hw/9pfs/virtio-9p-synth.c
> >> @@ -18,7 +18,7 @@
> >>  #include "fsdev/qemu-fsdev.h"
> >>  #include "virtio-9p-synth.h"
> >>  #include "qemu/rcu.h"
> >> -
> >> +#include "qemu/rcu_queue.h"
> >>  #include <sys/stat.h>
> >>  
> >>  /* Root node for synth file system */
> >> diff --git a/include/qemu/queue.h b/include/qemu/queue.h
> >> index c602797..8094150 100644
> >> --- a/include/qemu/queue.h
> >> +++ b/include/qemu/queue.h
> >> @@ -139,17 +139,6 @@ struct {                                              
> >>                   \
> >>          (elm)->field.le_prev = &(head)->lh_first;                       \
> >>  } while (/*CONSTCOND*/0)
> >>  
> >> -#define QLIST_INSERT_HEAD_RCU(head, elm, field) do {                    \
> >> -        (elm)->field.le_prev = &(head)->lh_first;                       \
> >> -        (elm)->field.le_next = (head)->lh_first;                        \
> >> -        smp_wmb(); /* fill elm before linking it */                     \
> >> -        if ((head)->lh_first != NULL)  {                                \
> >> -            (head)->lh_first->field.le_prev = &(elm)->field.le_next;    \
> >> -        }                                                               \
> >> -        (head)->lh_first = (elm);                                       \
> >> -        smp_wmb();                                                      \
> >> -} while (/* CONSTCOND*/0)
> >> -
> >>  #define QLIST_REMOVE(elm, field) do {                                   \
> >>          if ((elm)->field.le_next != NULL)                               \
> >>                  (elm)->field.le_next->field.le_prev =                   \
> >> diff --git a/include/qemu/rcu_queue.h b/include/qemu/rcu_queue.h
> >> new file mode 100644
> >> index 0000000..3aca7a5
> >> --- /dev/null
> >> +++ b/include/qemu/rcu_queue.h
> >> @@ -0,0 +1,134 @@
> >> +#ifndef QEMU_RCU_QUEUE_H
> >> +#define QEMU_RCU_QUEUE_H
> >> +
> >> +/*
> >> + * rcu_queue.h
> >> + *
> >> + * RCU-friendly versions of the queue.h primitives.
> >> + *
> >> + * This library is free software; you can redistribute it and/or
> >> + * modify it under the terms of the GNU Lesser General Public
> >> + * License as published by the Free Software Foundation; either
> >> + * version 2.1 of the License, or (at your option) any later version.
> >> + *
> >> + * This library is distributed in the hope that it will be useful,
> >> + * but WITHOUT ANY WARRANTY; without even the implied warranty of
> >> + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
> >> + * Lesser General Public License for more details.
> >> + *
> >> + * You should have received a copy of the GNU Lesser General Public
> >> + * License along with this library; if not, write to the Free Software
> >> + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 
> >> 02110-1301 USA
> >> + *
> >> + * Copyright (c) 2013 Mike D. Day, IBM Corporation.
> >> + *
> >> + * IBM's contributions to this file may be relicensed under LGPLv2 or 
> >> later.
> >> + */
> >> +
> >> +#include "qemu/queue.h"
> >> +#include "qemu/atomic.h"
> >> +
> >> +#ifdef __cplusplus
> >> +extern "C" {
> >> +#endif
> >> +
> >> +
> >> +/*
> >> + * List access methods.
> >> + */
> >> +#define QLIST_EMPTY_RCU(head) (atomic_rcu_read(&(head)->lh_first) == NULL)
> >> +#define QLIST_FIRST_RCU(head) (atomic_rcu_read(&(head)->lh_first))
> >> +#define QLIST_NEXT_RCU(elm, field) 
> >> (atomic_rcu_read(&(elm)->field.le_next))
> >> +
> >> +/*
> >> + * List functions.
> >> + */
> >> +
> >> +
> >> +/*
> >> + *  The difference between atomic_read/set and atomic_rcu_read/set
> >> + *  is in the including of a read/write memory barrier to the volatile
> >> + *  access. atomic_rcu_* macros include the memory barrier, the
> >> + *  plain atomic macros do not. Therefore, it should be correct to
> >> + *  issue a series of reads or writes to the same element using only
> >> + *  the atomic_* macro, until the last read or write, which should be
> >> + *  atomic_rcu_* to introduce a read or write memory barrier as
> >> + *  appropriate.
> >> + */
> >> +
> >> +/* Upon publication of the listelm->next value, list readers
> >> + * will see the new node when following next pointers from
> >> + * antecedent nodes, but may not see the new node when following
> >> + * prev pointers from subsequent nodes until after the RCU grace
> >> + * period expires.
> >> + * see linux/include/rculist.h __list_add_rcu(new, prev, next)
> >> + */
> >> +#define QLIST_INSERT_AFTER_RCU(listelm, elm, field) do {    \
> >> +    (elm)->field.le_next = (listelm)->field.le_next;        \
> >> +    (elm)->field.le_prev = &(listelm)->field.le_next;       \
> >> +    atomic_rcu_set(&(listelm)->field.le_next, (elm));       \
> >> +    if ((elm)->field.le_next != NULL) {                     \
> >> +       (elm)->field.le_next->field.le_prev =                \
> >> +        &(elm)->field.le_next;                              \
> >> +    }                                                       \
> >> +} while (/*CONSTCOND*/0)
> >> +
> >> +/* Upon publication of the listelm->prev->next value, list
> >> + * readers will see the new element when following prev pointers
> >> + * from subsequent elements, but may not see the new element
> >> + * when following next pointers from antecedent elements
> >> + * until after the RCU grace period expires.
> >> + */
> >> +#define QLIST_INSERT_BEFORE_RCU(listelm, elm, field) do {   \
> >> +    (elm)->field.le_prev = (listelm)->field.le_prev;        \
> >> +    (elm)->field.le_next = (listelm);                       \
> >> +    atomic_rcu_set((listelm)->field.le_prev, (elm));        \
> >> +    (listelm)->field.le_prev = &(elm)->field.le_next;       \
> >> +} while (/*CONSTCOND*/0)
> >> +
> >> +/* Upon publication of the head->first value, list readers
> >> + * will see the new element when following the head, but may
> >> + * not see the new element when following prev pointers from
> >> + * subsequent elements until after the RCU grace period has
> >> + * expired.
> >> + */
> >> +#define QLIST_INSERT_HEAD_RCU(head, elm, field) do {    \
> >> +    (elm)->field.le_prev = &(head)->lh_first;           \
> >> +    (elm)->field.le_next = (head)->lh_first;            \
> >> +    atomic_rcu_set((&(head)->lh_first), (elm));         \
> >> +    if ((elm)->field.le_next != NULL) {                 \
> >> +       (elm)->field.le_next->field.le_prev =            \
> >> +        &(elm)->field.le_next;                          \
> >> +    }                                                   \
> >> +} while (/*CONSTCOND*/0)
> >> +
> >> +
> >> +/* prior to publication of the elm->prev->next value, some list
> >> + * readers may still see the removed element when following
> >> + * the antecedent's next pointer.
> >> + */
> >> +#define QLIST_REMOVE_RCU(elm, field) do {           \
> >> +    if ((elm)->field.le_next != NULL) {             \
> >> +       (elm)->field.le_next->field.le_prev =        \
> >> +        (elm)->field.le_prev;                       \
> >> +    }                                               \
> >> +    *(elm)->field.le_prev =  (elm)->field.le_next;  \
> >> +} while (/*CONSTCOND*/0)
> >> +
> >> +/* List traversal must occur within an RCU critical section.  */
> >> +#define QLIST_FOREACH_RCU(var, head, field)                 \
> >> +        for ((var) = atomic_rcu_read(&(head)->lh_first);    \
> >> +                (var);                                      \
> >> +                (var) = atomic_rcu_read(&(var)->field.le_next))
> >> +
> >> +/* List traversal must occur within an RCU critical section.  */
> >> +#define QLIST_FOREACH_SAFE_RCU(var, head, field, next_var)           \
> >> +    for ((var) = (atomic_rcu_read(&(head)->lh_first));               \
> >> +      (var) &&                                                       \
> >> +          ((next_var) = atomic_rcu_read(&(var)->field.le_next), 1);  \
> >> +           (var) = (next_var))
> >> +
> >> +#ifdef __cplusplus
> >> +}
> >> +#endif
> >> +#endif /* QEMU_RCU_QUEUE.H */
> >> diff --git a/tests/Makefile b/tests/Makefile
> >> index db5b3c3..ef8d589 100644
> >> --- a/tests/Makefile
> >> +++ b/tests/Makefile
> >> @@ -62,6 +62,8 @@ check-unit-y += tests/test-int128$(EXESUF)
> >>  gcov-files-test-int128-y =
> >>  check-unit-y += tests/rcutorture$(EXESUF)
> >>  gcov-files-rcutorture-y = util/rcu.c
> >> +check-unit-y += tests/test-rcu-list$(EXESUF)
> >> +gcov-files-test-rcu-list-y = util/rcu.c
> >>  check-unit-y += tests/test-bitops$(EXESUF)
> >>  check-unit-$(CONFIG_HAS_GLIB_SUBPROCESS_TESTS) += 
> >> tests/test-qdev-global-props$(EXESUF)
> >>  check-unit-y += tests/check-qom-interface$(EXESUF)
> >> @@ -226,7 +228,7 @@ test-obj-y = tests/check-qint.o tests/check-qstring.o 
> >> tests/check-qdict.o \
> >>    tests/test-qmp-commands.o tests/test-visitor-serialization.o \
> >>    tests/test-x86-cpuid.o tests/test-mul64.o tests/test-int128.o \
> >>    tests/test-opts-visitor.o tests/test-qmp-event.o \
> >> -  tests/rcutorture.o
> >> +  tests/rcutorture.o tests/test-rcu-list.o
> >>  
> >>  test-qapi-obj-y = tests/test-qapi-visit.o tests/test-qapi-types.o \
> >>              tests/test-qapi-event.o
> >> @@ -256,6 +258,7 @@ tests/test-xbzrle$(EXESUF): tests/test-xbzrle.o 
> >> migration/xbzrle.o page_cache.o
> >>  tests/test-cutils$(EXESUF): tests/test-cutils.o util/cutils.o
> >>  tests/test-int128$(EXESUF): tests/test-int128.o
> >>  tests/rcutorture$(EXESUF): tests/rcutorture.o libqemuutil.a
> >> +tests/test-rcu-list$(EXESUF): tests/test-rcu-list.o libqemuutil.a
> >>  
> >>  tests/test-qdev-global-props$(EXESUF): tests/test-qdev-global-props.o \
> >>    hw/core/qdev.o hw/core/qdev-properties.o hw/core/hotplug.o\
> >> diff --git a/tests/test-rcu-list.c b/tests/test-rcu-list.c
> >> new file mode 100644
> >> index 0000000..5ce55e9
> >> --- /dev/null
> >> +++ b/tests/test-rcu-list.c
> >> @@ -0,0 +1,306 @@
> >> +/*
> >> + * rcuq_test.c
> >> + *
> >> + * usage: rcuq_test <readers> <duration>
> >> + *
> >> + * This program is free software; you can redistribute it and/or modify
> >> + * it under the terms of the GNU General Public License as published by
> >> + * the Free Software Foundation; either version 2 of the License, or
> >> + * (at your option) any later version.
> >> + *
> >> + * This program is distributed in the hope that it will be useful,
> >> + * but WITHOUT ANY WARRANTY; without even the implied warranty of
> >> + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
> >> + * GNU General Public License for more details.
> >> + *
> >> + * You should have received a copy of the GNU General Public License
> >> + * along with this program; if not, write to the Free Software
> >> + * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, 
> >> USA.
> >> + *
> >> + * Copyright (c) 2013 Mike D. Day, IBM Corporation.
> >> + */
> >> +
> >> +/*
> >> + * Test variables.
> >> + */
> > 
> > Move below the includes?
> > 
> >> +
> >> +#include <glib.h>
> >> +#include <stdlib.h>
> >> +#include <stdio.h>
> >> +#include <string.h>
> >> +#include "qemu/atomic.h"
> >> +#include "qemu/rcu.h"
> >> +#include "qemu/compiler.h"
> >> +#include "qemu/osdep.h"
> >> +#include "qemu/thread.h"
> >> +#include "qemu/rcu_queue.h"
> >> +
> >> +long long n_reads = 0LL;
> >> +long long n_updates = 0LL;
> >> +long long n_reclaims = 0LL;
> >> +long long n_nodes_removed = 0LL;
> >> +long long n_nodes = 0LL;
> >> +int g_test_in_charge = 0;
> >> +
> >> +int nthreadsrunning;
> >> +
> >> +char argsbuf[64];
> >> +
> >> +#define GOFLAG_INIT 0
> >> +#define GOFLAG_RUN  1
> >> +#define GOFLAG_STOP 2
> >> +
> >> +static volatile int goflag = GOFLAG_INIT;
> >> +
> >> +#define RCU_READ_RUN 1000
> >> +#define RCU_UPDATE_RUN 10
> >> +#define NR_THREADS 100
> >> +#define RCU_Q_LEN 100
> >> +
> >> +static QemuThread threads[NR_THREADS];
> >> +static struct rcu_reader_data *data[NR_THREADS];
> >> +static int n_threads;
> >> +
> >> +static int select_random_el(int max)
> >> +{
> >> +    return (rand() % max);
> >> +}
> >> +
> >> +
> >> +static void create_thread(void *(*func)(void *))
> >> +{
> >> +    if (n_threads >= NR_THREADS) {
> >> +        fprintf(stderr, "Thread limit of %d exceeded!\n", NR_THREADS);
> >> +        exit(-1);
> >> +    }
> >> +    qemu_thread_create(&threads[n_threads], "test", func, 
> >> &data[n_threads],
> >> +                       QEMU_THREAD_JOINABLE);
> >> +    n_threads++;
> >> +}
> >> +
> >> +static void wait_all_threads(void)
> >> +{
> >> +    int i;
> >> +
> >> +    for (i = 0; i < n_threads; i++) {
> >> +        qemu_thread_join(&threads[i]);
> >> +    }
> >> +    n_threads = 0;
> >> +}
> >> +
> >> +
> >> +struct list_element {
> >> +    QLIST_ENTRY(list_element) entry;
> >> +    struct rcu_head rcu;
> >> +    long long val;
> >> +};
> >> +
> >> +static void reclaim_list_el(struct rcu_head *prcu)
> >> +{
> >> +    struct list_element *el = container_of(prcu, struct list_element, 
> >> rcu);
> >> +    g_free(el);
> >> +    atomic_add(&n_reclaims, 1);
> >> +}
> >> +
> >> +static QLIST_HEAD(q_list_head, list_element) Q_list_head;
> >> +
> >> +static void *rcu_q_reader(void *arg)
> >> +{
> >> +    long long j, n_reads_local = 0;
> >> +    struct list_element *el;
> >> +
> >> +    *(struct rcu_reader_data **)arg = &rcu_reader;
> >> +    atomic_inc(&nthreadsrunning);
> >> +    while (goflag == GOFLAG_INIT) {
> >> +        g_usleep(1000);
> >> +    }
> >> +
> >> +    while (goflag == GOFLAG_RUN) {
> >> +        rcu_read_lock();
> >> +        QLIST_FOREACH_RCU(el, &Q_list_head, entry) {
> >> +            j = atomic_read(&el->val);
> >> +            (void)j;
> >> +            n_reads_local++;
> >> +            if (goflag == GOFLAG_STOP) {
> >> +                break;
> >> +            }
> >> +        }
> >> +        rcu_read_unlock();
> >> +
> >> +        g_usleep(100);
> >> +    }
> >> +    atomic_add(&n_reads, n_reads_local);
> >> +    return NULL;
> >> +}
> >> +
> >> +
> >> +static void *rcu_q_updater(void *arg)
> >> +{
> >> +    int j, target_el;
> >> +    long long n_updates_local = 0;
> >> +    long long n_removed_local = 0;
> >> +    struct list_element *el, *prev_el;
> >> +
> >> +    *(struct rcu_reader_data **)arg = &rcu_reader;
> >> +    atomic_inc(&nthreadsrunning);
> >> +    while (goflag == GOFLAG_INIT) {
> >> +        g_usleep(1000);
> >> +    }
> >> +
> >> +    while (goflag == GOFLAG_RUN) {
> >> +        target_el = select_random_el(RCU_Q_LEN);
> >> +        j = 0;
> >> +        /* FOREACH_RCU could work here but let's use both macros */
> >> +        QLIST_FOREACH_SAFE_RCU(prev_el, &Q_list_head, entry, el) {
> >> +            j++;
> >> +            if (target_el == j) {
> >> +                QLIST_REMOVE_RCU(prev_el, entry);
> >> +                /* may be more than one updater in the future */
> >> +                call_rcu1(&prev_el->rcu, reclaim_list_el);
> >> +                n_removed_local++;
> >> +                break;
> >> +            }
> >> +        }
> >> +        if (goflag == GOFLAG_STOP) {
> >> +            break;
> >> +        }
> >> +        target_el = select_random_el(RCU_Q_LEN);
> >> +        j = 0;
> >> +        QLIST_FOREACH_RCU(el, &Q_list_head, entry) {
> >> +            j++;
> >> +            if (target_el == j) {
> >> +                prev_el = g_new(struct list_element, 1);
> >> +                atomic_add(&n_nodes, 1);
> >> +                prev_el->val = atomic_read(&n_nodes);
> >> +                QLIST_INSERT_BEFORE_RCU(el, prev_el, entry);
> >> +                break;
> >> +            }
> >> +        }
> >> +
> >> +        n_updates_local += 2;
> >> +        synchronize_rcu();
> >> +    }
> >> +    synchronize_rcu();
> > 
> > Is one of two synchronize_rcu's superfluous? Otherwise looks good:
> 
> There is a "break"s in the loop, in that case the second synchronize_rcu
> is necessary.

Oh yes. So my rev-by holds.

Fam

> 
> Paolo
> 
> > 
> > Reviewed-by: Fam Zheng <address@hidden>
> > 
> >> +    atomic_add(&n_updates, n_updates_local);
> >> +    atomic_add(&n_nodes_removed, n_removed_local);
> >> +    return NULL;
> >> +}
> > 
> > <...>
> > 
> > 
>