* [Qemu-devel] [RFC PATCH 01/14] qemu-thread: add QemuEvent
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 02/14] rcu: add rcu library Mike Day
` (12 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
This emulates Win32 manual-reset events using futexes or conditional
variables. Typical ways to use them are with multi-producer,
single-consumer data structures, to test for a complex condition whose
elements come from different threads:
for (;;) {
qemu_event_reset(ev);
... test complex condition ...
if (condition is true) {
break;
}
qemu_event_wait(ev);
}
Alternatively:
... compute condition ...
if (condition) {
do {
qemu_event_wait(ev);
qemu_event_reset(ev);
... compute condition ...
} while(condition);
qemu_event_set(ev);
}
QemuEvent provides a very fast userspace path in the common case when
no other thread is waiting, or the event is not changing state. It
is used to report RCU quiescent states to the thread calling
synchronize_rcu (the latter being the single consumer), and to report
call_rcu invocations to the thread that receives them.
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
include/qemu/thread-posix.h | 8 +++
include/qemu/thread-win32.h | 4 ++
include/qemu/thread.h | 7 +++
util/qemu-thread-posix.c | 116 ++++++++++++++++++++++++++++++++++++++++++++
util/qemu-thread-win32.c | 26 ++++++++++
5 files changed, 161 insertions(+)
diff --git a/include/qemu/thread-posix.h b/include/qemu/thread-posix.h
index 361566a..eb5c7a1 100644
--- a/include/qemu/thread-posix.h
+++ b/include/qemu/thread-posix.h
@@ -21,6 +21,14 @@ struct QemuSemaphore {
#endif
};
+struct QemuEvent {
+#ifndef __linux__
+ pthread_mutex_t lock;
+ pthread_cond_t cond;
+#endif
+ unsigned value;
+};
+
struct QemuThread {
pthread_t thread;
};
diff --git a/include/qemu/thread-win32.h b/include/qemu/thread-win32.h
index 13adb95..3d58081 100644
--- a/include/qemu/thread-win32.h
+++ b/include/qemu/thread-win32.h
@@ -17,6 +17,10 @@ struct QemuSemaphore {
HANDLE sema;
};
+struct QemuEvent {
+ HANDLE event;
+};
+
typedef struct QemuThreadData QemuThreadData;
struct QemuThread {
QemuThreadData *data;
diff --git a/include/qemu/thread.h b/include/qemu/thread.h
index c02404b..3e32c65 100644
--- a/include/qemu/thread.h
+++ b/include/qemu/thread.h
@@ -7,6 +7,7 @@
typedef struct QemuMutex QemuMutex;
typedef struct QemuCond QemuCond;
typedef struct QemuSemaphore QemuSemaphore;
+typedef struct QemuEvent QemuEvent;
typedef struct QemuThread QemuThread;
#ifdef _WIN32
@@ -45,6 +46,12 @@ void qemu_sem_wait(QemuSemaphore *sem);
int qemu_sem_timedwait(QemuSemaphore *sem, int ms);
void qemu_sem_destroy(QemuSemaphore *sem);
+void qemu_event_init(QemuEvent *ev, bool init);
+void qemu_event_set(QemuEvent *ev);
+void qemu_event_reset(QemuEvent *ev);
+void qemu_event_wait(QemuEvent *ev);
+void qemu_event_destroy(QemuEvent *ev);
+
void qemu_thread_create(QemuThread *thread,
void *(*start_routine)(void *),
void *arg, int mode);
diff --git a/util/qemu-thread-posix.c b/util/qemu-thread-posix.c
index 4de133e..37dd298 100644
--- a/util/qemu-thread-posix.c
+++ b/util/qemu-thread-posix.c
@@ -20,7 +20,12 @@
#include <limits.h>
#include <unistd.h>
#include <sys/time.h>
+#ifdef __linux__
+#include <sys/syscall.h>
+#include <linux/futex.h>
+#endif
#include "qemu/thread.h"
+#include "qemu/atomic.h"
static void error_exit(int err, const char *msg)
{
@@ -272,6 +277,117 @@ void qemu_sem_wait(QemuSemaphore *sem)
#endif
}
+#ifdef __linux__
+#define futex(...) syscall(__NR_futex, __VA_ARGS__)
+
+static inline void futex_wake(QemuEvent *ev, int n)
+{
+ futex(ev, FUTEX_WAKE, n, NULL, NULL, 0);
+}
+
+static inline void futex_wait(QemuEvent *ev, unsigned val)
+{
+ futex(ev, FUTEX_WAIT, (int) val, NULL, NULL, 0);
+}
+#else
+static inline void futex_wake(QemuEvent *ev, int n)
+{
+ if (n == 1) {
+ pthread_cond_signal(&ev->cond);
+ } else {
+ pthread_cond_broadcast(&ev->cond);
+ }
+}
+
+static inline void futex_wait(QemuEvent *ev, unsigned val)
+{
+ pthread_mutex_lock(&ev->lock);
+ if (ev->value == val) {
+ pthread_cond_wait(&ev->cond, &ev->lock);
+ }
+ pthread_mutex_unlock(&ev->lock);
+}
+#endif
+
+/* Valid transitions:
+ * - free->set, when setting the event
+ * - busy->set, when setting the event, followed by futex_wake
+ * - set->free, when resetting the event
+ * - free->busy, when waiting
+ *
+ * set->busy does not happen (it can be observed from the outside but
+ * it really is set->free->busy).
+ *
+ * busy->free provably cannot happen; to enforce it, the set->free transition
+ * is done with an OR, which becomes a no-op if the event has concurrently
+ * transitioned to free or busy.
+ */
+
+#define EV_SET 0
+#define EV_FREE 1
+#define EV_BUSY -1
+
+void qemu_event_init(QemuEvent *ev, bool init)
+{
+#ifndef __linux__
+ pthread_mutex_init(&ev->lock, NULL);
+ pthread_cond_init(&ev->cond, NULL);
+#endif
+
+ ev->value = (init ? EV_SET : EV_FREE);
+}
+
+void qemu_event_destroy(QemuEvent *ev)
+{
+#ifndef __linux__
+ pthread_mutex_destroy(&ev->lock);
+ pthread_cond_destroy(&ev->cond);
+#endif
+}
+
+void qemu_event_set(QemuEvent *ev)
+{
+ if (atomic_mb_read(&ev->value) != EV_SET) {
+ if (atomic_xchg(&ev->value, EV_SET) == EV_BUSY) {
+ /* There were waiters, wake them up. */
+ futex_wake(ev, INT_MAX);
+ }
+ }
+}
+
+void qemu_event_reset(QemuEvent *ev)
+{
+ if (atomic_mb_read(&ev->value) == EV_SET) {
+ /*
+ * If there was a concurrent reset (or even reset+wait),
+ * do nothing. Otherwise change EV_SET->EV_FREE.
+ */
+ atomic_or(&ev->value, EV_FREE);
+ }
+}
+
+void qemu_event_wait(QemuEvent *ev)
+{
+ unsigned value;
+
+ value = atomic_mb_read(&ev->value);
+ if (value != EV_SET) {
+ if (value == EV_FREE) {
+ /*
+ * Leave the event reset and tell qemu_event_set that there
+ * are waiters. No need to retry, because there cannot be
+ * a concurent busy->free transition. After the CAS, the
+ * event will be either set or busy.
+ */
+ if (atomic_cmpxchg(&ev->value, EV_FREE, EV_BUSY) == EV_SET) {
+ return;
+ }
+ }
+ futex_wait(ev, EV_BUSY);
+ }
+}
+
+
void qemu_thread_create(QemuThread *thread,
void *(*start_routine)(void*),
void *arg, int mode)
diff --git a/util/qemu-thread-win32.c b/util/qemu-thread-win32.c
index 517878d..27a5217 100644
--- a/util/qemu-thread-win32.c
+++ b/util/qemu-thread-win32.c
@@ -227,6 +227,32 @@ void qemu_sem_wait(QemuSemaphore *sem)
}
}
+void qemu_event_init(QemuEvent *ev, bool init)
+{
+ /* Manual reset. */
+ ev->event = CreateEvent(NULL, TRUE, init, NULL);
+}
+
+void qemu_event_destroy(QemuEvent *ev)
+{
+ CloseHandle(ev->event);
+}
+
+void qemu_event_set(QemuEvent *ev)
+{
+ SetEvent(ev->event);
+}
+
+void qemu_event_reset(QemuEvent *ev)
+{
+ ResetEvent(ev->event);
+}
+
+void qemu_event_wait(QemuEvent *ev)
+{
+ WaitForSingleObject(ev->event, INFINITE);
+}
+
struct QemuThreadData {
/* Passed to win32_start_routine. */
void *(*start_routine)(void *);
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 02/14] rcu: add rcu library
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 01/14] qemu-thread: add QemuEvent Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 03/14] fix #include directive for rcu header Mike Day
` (11 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
This includes a (mangled) copy of the urcu-qsbr code from liburcu.
The main changes are: 1) removing dependencies on many other header files
in liburcu; 2) removing for simplicity the tentative busy waiting in
synchronize_rcu, which has limited performance effects; 3) replacing
futexes in synchronize_rcu with QemuEvents for Win32 portability.
The API is the same as liburcu, so it should be possible in the future
to require liburcu on POSIX systems for example and use our copy only
on Windows.
Among the various versions available I chose urcu-qsbr, which has the
fastest rcu_read_{lock,unlock} but requires the program to manually
annotate quiescent points or intervals. QEMU threads usually have easily
identified quiescent periods, so this should not be a problem.
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
docs/rcu.txt | 301 +++++++++++++++++++++++++++++++++++++++++++++
hw/9pfs/virtio-9p-synth.c | 1 +
include/qemu/queue.h | 13 ++
include/qemu/rcu-pointer.h | 110 +++++++++++++++++
include/qemu/rcu.h | 168 +++++++++++++++++++++++++
include/qemu/thread.h | 3 -
libcacard/Makefile | 3 +-
util/Makefile.objs | 1 +
util/rcu.c | 203 ++++++++++++++++++++++++++++++
9 files changed, 799 insertions(+), 4 deletions(-)
create mode 100644 docs/rcu.txt
create mode 100644 include/qemu/rcu-pointer.h
create mode 100644 include/qemu/rcu.h
create mode 100644 util/rcu.c
diff --git a/docs/rcu.txt b/docs/rcu.txt
new file mode 100644
index 0000000..19e4840
--- /dev/null
+++ b/docs/rcu.txt
@@ -0,0 +1,301 @@
+Using RCU (Read-Copy-Update) for synchronization
+================================================
+
+Read-copy update (RCU) is a synchronization mechanism that is used to
+protect read-mostly data structures. RCU is very efficient and scalable
+on the read side (it is wait-free), and thus can make the read paths
+extremely fast.
+
+RCU supports concurrency between a single writer and multiple readers,
+thus it is not used alone. Typically, the write-side will use a lock to
+serialize multiple updates, but other approaches are possible (e.g.,
+restricting updates to a single task). In QEMU, when a lock is used,
+this will often be the "iothread mutex", also known as the "big QEMU
+lock" (BQL). Also, restricting updates to a single task is done in
+QEMU using the "bottom half" API.
+
+RCU is fundamentally a "wait-to-finish" mechanism. The read side marks
+sections of code with "critical sections", and the update side will wait
+for the execution of all *currently running* critical sections before
+proceeding, or before asynchronously executing a callback.
+
+The key point here is that only the currently running critical sections
+are waited for; critical sections that are started _after_ the beginning
+of the wait do not extend the wait, despite running concurrently with
+the updater. This is the reason why RCU is more scalable than,
+for example, reader-writer locks. It is so much more scalable that
+the system will have a single instance of the RCU mechanism; a single
+mechanism can be used for an arbitrary number of "things", without
+having to worry about things such as contention or deadlocks.
+
+How is this possible? The basic idea is to split updates in two phases,
+"removal" and "reclamation". During removal, we ensure that subsequent
+readers will not be able to get a reference to the old data. After
+removal has completed, a critical section will not be able to access
+the old data. Therefore, critical sections that begin after removal
+do not matter; as soon as all previous critical sections have finished,
+there cannot be any readers who hold references to the data structure,
+which may not be safely reclaimed (e.g., freed or unref'ed).
+
+Here is a picutre:
+
+ thread 1 thread 2 thread 3
+ ------------------- ------------------------ -------------------
+ enter RCU crit.sec.
+ | finish removal phase
+ | begin wait
+ | | enter RCU crit.sec.
+ exit RCU crit.sec | |
+ complete wait |
+ begin reclamation phase |
+ exit RCU crit.sec.
+
+
+Note how thread 3 is still executing its critical section when thread 2
+starts reclaiming data. This is possible, because the old version of the
+data structure was not accessible at the time thread 3 began executing
+that critical section.
+
+
+RCU API
+=======
+
+The core RCU API is small:
+
+ void rcu_read_lock(void);
+
+ Used by a reader to inform the reclaimer that the reader is
+ entering an RCU read-side critical section.
+
+ void rcu_read_unlock(void);
+
+ Used by a reader to inform the reclaimer that the reader is
+ exiting an RCU read-side critical section. Note that RCU
+ read-side critical sections may be nested and/or overlapping.
+
+ void synchronize_rcu(void);
+
+ Blocks until all pre-existing RCU read-side critical sections
+ on all threads have completed. This marks the end of the removal
+ phase and the beginning of reclamation phase.
+
+ Note that it would be valid for another update to come while
+ synchronize_rcu is running. Because of this, it is better that
+ the updater releases any locks it may hold before calling
+ synchronize_rcu.
+
+ typeof(*p) rcu_dereference(p);
+ typeof(p) rcu_assign_pointer(p, typeof(p) v);
+
+ These macros are similar to atomic_mb_read() and atomic_mb_set()
+ respectively. However, they make some assumptions on the code
+ that calls them, which allows a more optimized implementation.
+
+ rcu_assign_pointer assumes that the update side is not going
+ to read from the data structure after "publishing" the new
+ values; that is, it assumes that all assignments happen at
+ the very end of the removal phase.
+
+ rcu_dereference assumes that whenever a single RCU critical
+ section reads multiple shared data, these reads are either
+ data-dependent or need no ordering. This is almost always the
+ case when using RCU. If this were not the case, you can use
+ atomic_mb_read() or smp_rmb().
+
+ If you are going to be fetching multiple fields from the
+ RCU-protected structure, repeated rcu_dereference() calls
+ would look ugly and incur unnecessary overhead on Alpha CPUs.
+ You can then do this:
+
+ p = &rcu_dereference(head);
+ foo = head->foo;
+ bar = head->bar;
+
+
+RCU QUIESCENT STATES
+====================
+
+An efficient implementation of rcu_read_lock() and rcu_read_unlock()
+relies on the availability of fast thread-local storage. Unfortunately,
+this is not possible on all the systems supported by QEMU (in particular
+on many POSIX systems other than Linux and Solaris).
+
+For this reason, QEMU's RCU implementation resorts to manual annotation
+of "quiescent states", i.e. points where no RCU read-side critical
+section can be active. All threads that participate in the RCU mechanism
+need to annotate such points.
+
+Marking quiescent states is done with the following three APIs:
+
+ void rcu_quiescent_state(void);
+
+ Marks a point in the execution of the current thread where no
+ RCU read-side critical section can be active.
+
+ void rcu_thread_offline(void);
+
+ Marks the beginning of an "extended quiescent state" for the
+ current thread, i.e. an interval of time during which no
+ RCU read-side critical section can be active.
+
+ void rcu_thread_online(void);
+
+ Marks the end of an extended quiescent state for the current
+ thread.
+
+
+Furthermore, threads that participate in the RCU mechanism must communicate
+this fact using the following APIs:
+
+ void rcu_register_thread(void);
+
+ Mark a thread as taking part in the RCU mechanism. Such a thread
+ will have to report quiescent points regularly, either manually
+ or through the QemuCond/QemuSemaphore/QemuEvent APIs.
+
+ void rcu_unregister_thread(void);
+
+ Mark a thread as not taking part anymore in the RCU mechanism.
+ It is not a problem if such a thread reports quiescent points,
+ either manually or by using the QemuCond/QemuSemaphore/QemuEvent
+ APIs.
+
+Note that these APIs are relatively heavyweight, and should _not_ be
+nested.
+
+
+DIFFERENCES WITH LINUX
+======================
+
+- Sleeping is possible, though discouraged, within an RCU critical section.
+
+- rcu_dereference takes a _pointer_ to the variable being accessed.
+ Wrong usage will be detected by the compiler.
+
+- Quiescent points must be marked explicitly unless the thread uses
+ condvars/semaphores/events for synchronization.
+
+
+RCU PATTERNS
+============
+
+Many patterns using read-writer locks translate directly to RCU, with
+the advantages of higher scalability and deadlock immunity.
+
+In general, RCU can be used whenever it is possible to create a new
+"version" of a data structure every time the updater runs. This may
+sound like a very strict restriction, however:
+
+- the updater does not mean "everything that writes to a data structure",
+ but rather "everything that involves a reclamation step". See the
+ array example below
+
+- in some cases, creating a new version of a data structure may actually
+ be very cheap. For example, modifying the "next" pointer of a singly
+ linked list is effectively creating a new version of the list.
+
+
+them however are worth noting.
+
+RCU list processing
+-------------------
+
+TBD (not yet used in QEMU)
+
+
+RCU reference counting
+----------------------
+
+Because grace periods are not allowed to complete while there is an RCU
+read-side critical section in progress, the RCU read-side primitives
+may be used as a restricted reference-counting mechanism. For example,
+consider the following code fragment:
+
+ rcu_read_lock();
+ p = rcu_dereference(&foo);
+ /* do something with p. */
+ rcu_read_unlock();
+
+The RCU read-side critical section ensures that the value of "p" remains
+valid until after the rcu_read_unlock(). In some sense, it is acquiring
+a reference to p that is later released when the critical section ends.
+The write side looks simply like this (with appropriate locking):
+
+ qemu_mutex_lock(&foo_mutex);
+ old = foo;
+ rcu_assign_pointer(foo, new);
+ qemu_mutex_unlock(&foo_mutex);
+ synchronize_rcu();
+ free(old);
+
+Note that the same idiom would be possible with reader/writer
+locks:
+
+ read_lock(&foo_rwlock); write_mutex_lock(&foo_rwlock);
+ p = foo; p = foo;
+ /* do something with p. */ foo = new;
+ read_unlock(&foo_rwlock); free(p);
+ write_mutex_unlock(&foo_rwlock);
+ free(p);
+
+
+RCU resizable arrays
+--------------------
+
+Resizable arrays can be used with RCU. The expensive RCU synchronization
+only needs to take place when the array is resized. The two items to
+take care of are:
+
+- ensuring that the old version of the array is available between removal
+ and reclamation;
+
+- avoiding mismatches in the read side between the array data and the
+ array size.
+
+The first problem is avoided simply by not using realloc. Instead,
+each resize will allocate a new array and copy the old data into it.
+The second problem would arise if the size and the data pointers were
+two members of a larger struct:
+
+ struct mystuff {
+ ...
+ int data_size;
+ int data_alloc;
+ T *data;
+ ...
+ };
+
+Instead, we store the size of the array with the array itself:
+
+ struct arr {
+ int size;
+ int alloc;
+ T data[];
+ };
+ struct arr *global_array;
+
+ read side:
+ rcu_read_lock();
+ struct arr *array = rcu_dereference(&global_array);
+ x = i < array->size ? array->data[i] : -1;
+ rcu_read_unlock();
+ return x;
+
+ write side (running under a lock):
+ if (global_array->size == global_array->alloc) {
+ /* Creating a new version. */
+ new_array = g_malloc(sizeof(struct arr) +
+ global_array->alloc * 2 * sizeof(T));
+ new_array->size = global_array->size;
+ new_array->alloc = global_array->alloc * 2;
+ memcpy(new_array->data, global_array->data,
+ global_array->alloc * sizeof(T));
+
+ /* Removal phase. */
+ old_array = global_array;
+ rcu_assign_pointer(new_array->data, new_array);
+ synchronize_rcu();
+
+ /* Reclamation phase. */
+ free(old_array);
+ }
diff --git a/hw/9pfs/virtio-9p-synth.c b/hw/9pfs/virtio-9p-synth.c
index 840e4eb..d5f5842 100644
--- a/hw/9pfs/virtio-9p-synth.c
+++ b/hw/9pfs/virtio-9p-synth.c
@@ -17,6 +17,7 @@
#include "virtio-9p-xattr.h"
#include "fsdev/qemu-fsdev.h"
#include "virtio-9p-synth.h"
+#include "util/rcu.h"
#include <sys/stat.h>
diff --git a/include/qemu/queue.h b/include/qemu/queue.h
index d433b90..847ddd1 100644
--- a/include/qemu/queue.h
+++ b/include/qemu/queue.h
@@ -104,6 +104,19 @@ struct { \
(head)->lh_first = NULL; \
} while (/*CONSTCOND*/0)
+#define QLIST_SWAP(dstlist, srclist, field) do { \
+ void *tmplist; \
+ tmplist = (srclist)->lh_first; \
+ (srclist)->lh_first = (dstlist)->lh_first; \
+ if ((srclist)->lh_first != NULL) { \
+ (srclist)->lh_first->field.le_prev = &(srclist)->lh_first; \
+ } \
+ (dstlist)->lh_first = tmplist; \
+ if ((dstlist)->lh_first != NULL) { \
+ (dstlist)->lh_first->field.le_prev = &(dstlist)->lh_first; \
+ } \
+} while (/*CONSTCOND*/0)
+
#define QLIST_INSERT_AFTER(listelm, elm, field) do { \
if (((elm)->field.le_next = (listelm)->field.le_next) != NULL) \
(listelm)->field.le_next->field.le_prev = \
diff --git a/include/qemu/rcu-pointer.h b/include/qemu/rcu-pointer.h
new file mode 100644
index 0000000..0e6417c
--- /dev/null
+++ b/include/qemu/rcu-pointer.h
@@ -0,0 +1,110 @@
+#ifndef _URCU_POINTER_STATIC_H
+#define _URCU_POINTER_STATIC_H
+
+/*
+ * urcu-pointer-static.h
+ *
+ * Userspace RCU header. Operations on pointers.
+ *
+ * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See urcu-pointer.h for
+ * linking dynamically with the userspace rcu library.
+ *
+ * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
+ *
+ * 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
+ *
+ * IBM's contributions to this file may be relicensed under LGPLv2 or later.
+ */
+
+#include "compiler.h"
+#include "qemu/atomic.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * rcu_dereference - reads (copy) a RCU-protected pointer to a local variable
+ * into a RCU read-side critical section. The pointer can later be safely
+ * dereferenced within the critical section.
+ *
+ * This ensures that the pointer copy is invariant thorough the whole critical
+ * section.
+ *
+ * Inserts memory barriers on architectures that require them (currently only
+ * Alpha) and documents which pointers are protected by RCU.
+ *
+ * The compiler memory barrier in atomic_read() ensures that value-speculative
+ * optimizations (e.g. VSS: Value Speculation Scheduling) does not perform the
+ * data read before the pointer read by speculating the value of the pointer.
+ * Correct ordering is ensured because the pointer is read as a volatile access.
+ * This acts as a global side-effect operation, which forbids reordering of
+ * dependent memory operations. Note that such concern about dependency-breaking
+ * optimizations will eventually be taken care of by the "memory_order_consume"
+ * addition to forthcoming C++ standard.
+ *
+ * Should match rcu_assign_pointer() or rcu_xchg_pointer().
+ */
+
+#define rcu_dereference(p) \
+ ({ \
+ typeof(p) _p1 = (p); \
+ smp_read_barrier_depends(); \
+ *(_p1); \
+ })
+
+/**
+ * rcu_cmpxchg_pointer - same as rcu_set_pointer, but tests if the pointer
+ * is as expected by "old". If succeeds, returns the previous pointer to the
+ * data structure, which can be safely freed after waiting for a quiescent state
+ * using synchronize_rcu(). If fails (unexpected value), returns old (which
+ * should not be freed !).
+ */
+
+#define rcu_cmpxchg_pointer(p, old, _new) \
+ ({ \
+ typeof(*p) _pold = (old); \
+ typeof(*p) _pnew = (_new); \
+ atomic_cmpxchg(p, _pold, _pnew); \
+ })
+
+#define rcu_set_pointer(p, v) \
+ ({ \
+ typeof(*p) _pv = (v); \
+ smp_wmb(); \
+ atomic_set(p, _pv); \
+ })
+
+/**
+ * rcu_assign_pointer - assign (publicize) a pointer to a new data structure
+ * meant to be read by RCU read-side critical sections. Returns the assigned
+ * value.
+ *
+ * Documents which pointers will be dereferenced by RCU read-side critical
+ * sections and adds the required memory barriers on architectures requiring
+ * them. It also makes sure the compiler does not reorder code initializing the
+ * data structure before its publication.
+ *
+ * Should match rcu_dereference().
+ */
+
+#define rcu_assign_pointer(p, v) rcu_set_pointer(&(p), v)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _URCU_POINTER_STATIC_H */
diff --git a/include/qemu/rcu.h b/include/qemu/rcu.h
new file mode 100644
index 0000000..b875593
--- /dev/null
+++ b/include/qemu/rcu.h
@@ -0,0 +1,168 @@
+#ifndef _URCU_QSBR_H
+#define _URCU_QSBR_H
+
+/*
+ * urcu-qsbr.h
+ *
+ * Userspace RCU QSBR header.
+ *
+ * LGPL-compatible code should include this header with :
+ *
+ * #define _LGPL_SOURCE
+ * #include <urcu.h>
+ *
+ * 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
+ *
+ * IBM's contributions to this file may be relicensed under LGPLv2 or later.
+ */
+
+#include <stdlib.h>
+#include <assert.h>
+#include <limits.h>
+#include <unistd.h>
+#include <stdint.h>
+#include <stdbool.h>
+#include <glib.h>
+
+#include "qemu/compiler.h"
+#include "qemu/rcu-pointer.h"
+#include "qemu/thread.h"
+#include "qemu/queue.h"
+#include "qemu/atomic.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/*
+ * Important !
+ *
+ * Each thread containing read-side critical sections must be registered
+ * with rcu_register_thread() before calling rcu_read_lock().
+ * rcu_unregister_thread() should be called before the thread exits.
+ */
+
+#ifdef DEBUG_RCU
+#define rcu_assert(args...) assert(args)
+#else
+#define rcu_assert(args...)
+#endif
+
+#define RCU_GP_ONLINE (1UL << 0)
+#define RCU_GP_CTR (1UL << 1)
+
+/*
+ * Global quiescent period counter with low-order bits unused.
+ * Using a int rather than a char to eliminate false register dependencies
+ * causing stalls on some architectures.
+ */
+extern unsigned long rcu_gp_ctr;
+
+extern QemuEvent rcu_gp_event;
+
+struct rcu_reader_data {
+ /* Data used by both reader and synchronize_rcu() */
+ unsigned long ctr;
+ bool waiting;
+
+ /* Data used for registry, protected by rcu_gp_lock */
+ QLIST_ENTRY(rcu_reader_data) node;
+};
+
+#ifdef __linux__
+extern __thread struct rcu_reader_data rcu_reader;
+#define DEFINE_RCU_READER() \
+ __thread struct rcu_reader_data rcu_reader
+
+static inline struct rcu_reader_data *get_rcu_reader(void)
+{
+ return &rcu_reader;
+}
+
+static inline void alloc_rcu_reader(void)
+{
+}
+#else
+extern GPrivate rcu_reader_key;
+#define DEFINE_RCU_READER() \
+ GPrivate rcu_reader_key = G_PRIVATE_INIT(g_free)
+
+static inline struct rcu_reader_data *get_rcu_reader(void)
+{
+ return g_private_get(&rcu_reader_key);
+}
+
+static inline void alloc_rcu_reader(void)
+{
+ g_private_replace(&rcu_reader_key,
+ g_malloc0(sizeof(struct rcu_reader_data)));
+}
+#endif
+
+static inline void rcu_read_lock(void)
+{
+ rcu_assert(get_rcu_reader()->ctr);
+}
+
+static inline void rcu_read_unlock(void)
+{
+ /* Ensure that the previous reads complete before starting those
+ * in another critical section.
+ */
+ smp_rmb();
+}
+
+static inline void rcu_quiescent_state(void)
+{
+ struct rcu_reader_data *p_rcu_reader = get_rcu_reader();
+
+ /* Reuses smp_rmb() in the last rcu_read_unlock(). */
+ unsigned ctr = atomic_read(&rcu_gp_ctr);
+ atomic_xchg(&p_rcu_reader->ctr, ctr);
+ if (atomic_read(&p_rcu_reader->waiting)) {
+ atomic_set(&p_rcu_reader->waiting, false);
+ qemu_event_set(&rcu_gp_event);
+ }
+}
+
+static inline void rcu_thread_offline(void)
+{
+ struct rcu_reader_data *p_rcu_reader = get_rcu_reader();
+
+ atomic_xchg(&p_rcu_reader->ctr, 0);
+ if (atomic_read(&p_rcu_reader->waiting)) {
+ atomic_set(&p_rcu_reader->waiting, false);
+ qemu_event_set(&rcu_gp_event);
+ }
+}
+
+static inline void rcu_thread_online(void)
+{
+ rcu_quiescent_state();
+}
+
+extern void synchronize_rcu(void);
+
+/*
+ * Reader thread registration.
+ */
+extern void rcu_register_thread(void);
+extern void rcu_unregister_thread(void);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _URCU_QSBR_H */
diff --git a/include/qemu/thread.h b/include/qemu/thread.h
index 3e32c65..5d64a20 100644
--- a/include/qemu/thread.h
+++ b/include/qemu/thread.h
@@ -25,9 +25,6 @@ void qemu_mutex_lock(QemuMutex *mutex);
int qemu_mutex_trylock(QemuMutex *mutex);
void qemu_mutex_unlock(QemuMutex *mutex);
-#define rcu_read_lock() do { } while (0)
-#define rcu_read_unlock() do { } while (0)
-
void qemu_cond_init(QemuCond *cond);
void qemu_cond_destroy(QemuCond *cond);
diff --git a/libcacard/Makefile b/libcacard/Makefile
index 47827a0..f7a3b07 100644
--- a/libcacard/Makefile
+++ b/libcacard/Makefile
@@ -4,7 +4,8 @@ TOOLS += vscclient$(EXESUF)
# objects linked into a shared library, built with libtool with -fPIC if required
libcacard-obj-y = $(stub-obj-y) $(libcacard-y)
-libcacard-obj-y += util/osdep.o util/cutils.o util/qemu-timer-common.o util/error.o
+libcacard-obj-y += util/osdep.o util/cutils.o util/qemu-timer-common.o
+libcacard-obj-y += util/rcu.o util/error.o
libcacard-obj-$(CONFIG_WIN32) += util/oslib-win32.o util/qemu-thread-win32.o
libcacard-obj-$(CONFIG_POSIX) += util/oslib-posix.o util/qemu-thread-posix.o
libcacard-obj-y += $(filter trace/%, $(util-obj-y))
diff --git a/util/Makefile.objs b/util/Makefile.objs
index dc72ab0..c34e5ee 100644
--- a/util/Makefile.objs
+++ b/util/Makefile.objs
@@ -11,3 +11,4 @@ util-obj-y += iov.o aes.o qemu-config.o qemu-sockets.o uri.o notify.o
util-obj-y += qemu-option.o qemu-progress.o
util-obj-y += hexdump.o
util-obj-y += crc32c.o
+util-obj-y += rcu.o
diff --git a/util/rcu.c b/util/rcu.c
new file mode 100644
index 0000000..48686a3
--- /dev/null
+++ b/util/rcu.c
@@ -0,0 +1,203 @@
+/*
+ * urcu-qsbr.c
+ *
+ * Userspace RCU QSBR library
+ *
+ * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
+ * Copyright 2013 Red Hat, Inc.
+ *
+ * Ported to QEMU by Paolo Bonzini <pbonzini@redhat.com>
+ *
+ * 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
+ *
+ * IBM's contributions to this file may be relicensed under LGPLv2 or later.
+ */
+
+#include <stdio.h>
+#include <assert.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <errno.h>
+#include "qemu/rcu.h"
+#include "qemu/atomic.h"
+
+/*
+ * Global grace period counter. Bit 0 is one if the thread is online.
+ * Bits 1 and above are defined in synchronize_rcu/update_counter_and_wait.
+ */
+#define RCU_GP_ONLINE (1UL << 0)
+#define RCU_GP_CTR (1UL << 1)
+
+unsigned long rcu_gp_ctr = RCU_GP_ONLINE;
+
+QemuEvent rcu_gp_event;
+static QemuMutex rcu_gp_lock;
+
+/*
+ * Check whether a quiescent state was crossed between the beginning of
+ * update_counter_and_wait and now.
+ */
+static inline int rcu_gp_ongoing(unsigned long *ctr)
+{
+ unsigned long v;
+
+ /* See update_counter_and_wait for the discussion of memory barriers. */
+ v = atomic_read(ctr);
+ return v && (v != rcu_gp_ctr);
+}
+
+/* Written to only by each individual reader. Read by both the reader and the
+ * writers.
+ */
+DEFINE_RCU_READER();
+
+/* Protected by rcu_gp_lock. */
+typedef QLIST_HEAD(, rcu_reader_data) ThreadList;
+static ThreadList registry = QLIST_HEAD_INITIALIZER(registry);
+
+/* Wait for previous parity/grace period to be empty of readers. */
+static void wait_for_readers(void)
+{
+ ThreadList qsreaders = QLIST_HEAD_INITIALIZER(qsreaders);
+ struct rcu_reader_data *index, *tmp;
+
+ for (;;) {
+ /* We want to be notified of changes made to rcu_gp_ongoing
+ * while we walk the list.
+ */
+ qemu_event_reset(&rcu_gp_event);
+
+ /* Instead of using atomic_mb_set for index->waiting, and
+ * atomic_mb_read for index->ctr, memory barriers are placed
+ * manually since writes to different threads are independent.
+ * atomic_mb_set has a smp_wmb before...
+ */
+ smp_wmb();
+ QLIST_FOREACH(index, ®istry, node) {
+ atomic_set(&index->waiting, true);
+ }
+
+ /* ... and a smp_mb after.
+ *
+ * This barrier also blocks stores that free old RCU-protected
+ * pointers.
+ */
+ smp_mb();
+
+ QLIST_FOREACH_SAFE(index, ®istry, node, tmp) {
+ if (!rcu_gp_ongoing(&index->ctr)) {
+ QLIST_REMOVE(index, node);
+ QLIST_INSERT_HEAD(&qsreaders, index, node);
+
+ /* No need for mb_set here, worst of all we
+ * get some extra futex wakeups.
+ */
+ atomic_set(&index->waiting, false);
+ }
+ }
+
+ /* atomic_mb_read has smp_rmb after. */
+ smp_rmb();
+
+ if (QLIST_EMPTY(®istry)) {
+ break;
+ }
+
+ /* Wait for one thread to report a quiescent state and
+ * try again.
+ */
+ qemu_event_wait(&rcu_gp_event);
+ }
+
+ /* put back the reader list in the registry */
+ QLIST_SWAP(®istry, &qsreaders, node);
+}
+
+void synchronize_rcu(void)
+{
+ unsigned long was_online;
+
+ was_online = get_rcu_reader()->ctr;
+
+ /* Mark the writer thread offline to make sure we don't wait for
+ * our own quiescent state. This allows using synchronize_rcu()
+ * in threads registered as readers.
+ *
+ * rcu_thread_offline() and rcu_thread_online() include a
+ * memory barrier.
+ */
+ if (was_online) {
+ rcu_thread_offline();
+ } else {
+ smp_mb();
+ }
+
+ qemu_mutex_lock(&rcu_gp_lock);
+
+ if (!QLIST_EMPTY(®istry)) {
+ if (sizeof(rcu_gp_ctr) < 8) {
+ /* For architectures with 32-bit longs, a two-subphases algorithm
+ * ensures we do not encounter overflow bugs.
+ *
+ * Switch parity: 0 -> 1, 1 -> 0.
+ */
+ atomic_mb_set(&rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR);
+ wait_for_readers();
+ atomic_mb_set(&rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR);
+ } else {
+ /* Increment current grace period. */
+ atomic_mb_set(&rcu_gp_ctr, rcu_gp_ctr + RCU_GP_CTR);
+ }
+
+ wait_for_readers();
+ }
+
+ qemu_mutex_unlock(&rcu_gp_lock);
+
+ if (was_online) {
+ rcu_thread_online();
+ } else {
+ smp_mb();
+ }
+}
+
+void rcu_register_thread(void)
+{
+ if (!get_rcu_reader()) {
+ alloc_rcu_reader();
+ }
+
+ assert(get_rcu_reader()->ctr == 0);
+ qemu_mutex_lock(&rcu_gp_lock);
+ QLIST_INSERT_HEAD(®istry, get_rcu_reader(), node);
+ qemu_mutex_unlock(&rcu_gp_lock);
+ rcu_quiescent_state();
+}
+
+void rcu_unregister_thread(void)
+{
+ rcu_thread_offline();
+ qemu_mutex_lock(&rcu_gp_lock);
+ QLIST_REMOVE(get_rcu_reader(), node);
+ qemu_mutex_unlock(&rcu_gp_lock);
+}
+
+static void __attribute__((__constructor__)) rcu_init(void)
+{
+ qemu_mutex_init(&rcu_gp_lock);
+ qemu_event_init(&rcu_gp_event, true);
+ rcu_register_thread();
+}
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 03/14] fix #include directive for rcu header
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 01/14] qemu-thread: add QemuEvent Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 02/14] rcu: add rcu library Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 04/14] qemu-thread: register threads with RCU Mike Day
` (10 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Mike Day
Signed-off-by: Mike Day <ncmike@ncultra.org>
---
hw/9pfs/virtio-9p-synth.c | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hw/9pfs/virtio-9p-synth.c b/hw/9pfs/virtio-9p-synth.c
index d5f5842..fdfea21 100644
--- a/hw/9pfs/virtio-9p-synth.c
+++ b/hw/9pfs/virtio-9p-synth.c
@@ -17,7 +17,7 @@
#include "virtio-9p-xattr.h"
#include "fsdev/qemu-fsdev.h"
#include "virtio-9p-synth.h"
-#include "util/rcu.h"
+#include "qemu/rcu.h"
#include <sys/stat.h>
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 04/14] qemu-thread: register threads with RCU
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (2 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 03/14] fix #include directive for rcu header Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 05/14] rcu: add call_rcu Mike Day
` (9 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
docs/rcu.txt | 13 +++++++------
util/qemu-thread-posix.c | 28 +++++++++++++++++++++++++++-
util/qemu-thread-win32.c | 2 ++
3 files changed, 36 insertions(+), 7 deletions(-)
diff --git a/docs/rcu.txt b/docs/rcu.txt
index 19e4840..5736676 100644
--- a/docs/rcu.txt
+++ b/docs/rcu.txt
@@ -122,8 +122,8 @@ on many POSIX systems other than Linux and Solaris).
For this reason, QEMU's RCU implementation resorts to manual annotation
of "quiescent states", i.e. points where no RCU read-side critical
-section can be active. All threads that participate in the RCU mechanism
-need to annotate such points.
+section can be active. All threads created with qemu_thread_create
+participate in the RCU mechanism and need to annotate such points.
Marking quiescent states is done with the following three APIs:
@@ -144,8 +144,8 @@ Marking quiescent states is done with the following three APIs:
thread.
-Furthermore, threads that participate in the RCU mechanism must communicate
-this fact using the following APIs:
+The following APIs can be used to use RCU in a thread that is not
+created with qemu_thread_create():
void rcu_register_thread(void);
@@ -160,8 +160,9 @@ this fact using the following APIs:
either manually or by using the QemuCond/QemuSemaphore/QemuEvent
APIs.
-Note that these APIs are relatively heavyweight, and should _not_ be
-nested.
+Note that these APIs are relatively heavyweight, should _not_ be
+nested, and should not be called in threads that are created with
+qemu_thread_create().
DIFFERENCES WITH LINUX
diff --git a/util/qemu-thread-posix.c b/util/qemu-thread-posix.c
index 37dd298..2371176 100644
--- a/util/qemu-thread-posix.c
+++ b/util/qemu-thread-posix.c
@@ -26,6 +26,7 @@
#endif
#include "qemu/thread.h"
#include "qemu/atomic.h"
+#include "qemu/rcu.h"
static void error_exit(int err, const char *msg)
{
@@ -388,6 +389,26 @@ void qemu_event_wait(QemuEvent *ev)
}
+typedef struct QemuThreadData {
+ /* Passed to win32_start_routine. */
+ void *(*start_routine)(void *);
+ void *arg;
+} QemuThreadData;
+
+static void *thread_start_routine(void *arg)
+{
+ QemuThreadData *data = (QemuThreadData *) arg;
+ void *(*start_routine)(void *) = data->start_routine;
+ void *thread_arg = data->arg;
+ void *ret;
+
+ rcu_register_thread();
+ g_free(data);
+ ret = start_routine(thread_arg);
+ rcu_unregister_thread();
+ return ret;
+}
+
void qemu_thread_create(QemuThread *thread,
void *(*start_routine)(void*),
void *arg, int mode)
@@ -395,6 +416,11 @@ void qemu_thread_create(QemuThread *thread,
sigset_t set, oldset;
int err;
pthread_attr_t attr;
+ QemuThreadData *data;
+
+ data = g_malloc(sizeof(*data));
+ data->start_routine = start_routine;
+ data->arg = arg;
err = pthread_attr_init(&attr);
if (err) {
@@ -410,7 +436,7 @@ void qemu_thread_create(QemuThread *thread,
/* Leave signal handling to the iothread. */
sigfillset(&set);
pthread_sigmask(SIG_SETMASK, &set, &oldset);
- err = pthread_create(&thread->thread, &attr, start_routine, arg);
+ err = pthread_create(&thread->thread, &attr, thread_start_routine, data);
if (err)
error_exit(err, __func__);
diff --git a/util/qemu-thread-win32.c b/util/qemu-thread-win32.c
index 27a5217..0c4850d 100644
--- a/util/qemu-thread-win32.c
+++ b/util/qemu-thread-win32.c
@@ -278,6 +278,7 @@ static unsigned __stdcall win32_start_routine(void *arg)
data = NULL;
}
qemu_thread_data = data;
+ rcu_register_thread();
qemu_thread_exit(start_routine(thread_arg));
abort();
}
@@ -293,6 +294,7 @@ void qemu_thread_exit(void *arg)
data->exited = true;
LeaveCriticalSection(&data->cs);
}
+ rcu_unregister_thread();
_endthreadex(0);
}
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 05/14] rcu: add call_rcu
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (3 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 04/14] qemu-thread: register threads with RCU Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 06/14] rcu: add rcutorture Mike Day
` (8 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
docs/rcu.txt | 108 +++++++++++++++++++++++++++++++++++++++++++++--
include/qemu/rcu.h | 22 ++++++++++
util/rcu.c | 120 +++++++++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 246 insertions(+), 4 deletions(-)
diff --git a/docs/rcu.txt b/docs/rcu.txt
index 5736676..d7c4f0b 100644
--- a/docs/rcu.txt
+++ b/docs/rcu.txt
@@ -82,7 +82,50 @@ The core RCU API is small:
Note that it would be valid for another update to come while
synchronize_rcu is running. Because of this, it is better that
the updater releases any locks it may hold before calling
- synchronize_rcu.
+ synchronize_rcu. If this is not possible (for example, because
+ the updater is protected by the BQL), you can use call_rcu.
+
+ void call_rcu1(struct rcu_head * head,
+ void (*func)(struct rcu_head *head));
+
+ This function invokes func(head) after all pre-existing RCU
+ read-side critical sections on all threads have completed. This
+ marks the end of the removal phase, with func taking care
+ asynchronously of the reclamation phase.
+
+ The foo struct needs to have an rcu_head structure added,
+ perhaps as follows:
+
+ struct foo {
+ struct rcu_head rcu;
+ int a;
+ char b;
+ long c;
+ };
+
+ so that the reclaimer function can fetch the struct foo address
+ and free it:
+
+ call_rcu1(foo_reclaim, &foo.rcu);
+
+ void foo_reclaim(struct rcu_head *rp)
+ {
+ struct foo *fp = container_of(rp, struct foo, rcu);
+ g_free(fp);
+ }
+
+ For the common case where the rcu_head member is the first of the
+ struct, you can use the following macro.
+
+ void call_rcu(T *p,
+ void (*func)(T *p),
+ field-name);
+
+ call_rcu1 is typically used through this macro, in the common case
+ where the "struct rcu_head" is the first field in the struct. In
+ the above case, one could have written simply:
+
+ call_rcu(foo_reclaim, g_free, rcu);
typeof(*p) rcu_dereference(p);
typeof(p) rcu_assign_pointer(p, typeof(p) v);
@@ -173,6 +216,11 @@ DIFFERENCES WITH LINUX
- rcu_dereference takes a _pointer_ to the variable being accessed.
Wrong usage will be detected by the compiler.
+- call_rcu is a macro that has an extra argument (the name of the first
+ field in the struct, which must be a struct rcu_head), and expects the
+ type of the callback's argument to be the type of the first argument.
+ call_rcu1 is the same as Linux's call_rcu.
+
- Quiescent points must be marked explicitly unless the thread uses
condvars/semaphores/events for synchronization.
@@ -229,7 +277,47 @@ The write side looks simply like this (with appropriate locking):
synchronize_rcu();
free(old);
-Note that the same idiom would be possible with reader/writer
+If the processing cannot be done purely within the critical section, it
+is possible to combine this idiom with a "real" reference count:
+
+ rcu_read_lock();
+ p = rcu_dereference(&foo);
+ foo_ref(p);
+ rcu_read_unlock();
+ /* do something with p. */
+ foo_unref(p);
+
+The write side can be like this:
+
+ qemu_mutex_lock(&foo_mutex);
+ old = foo;
+ rcu_assign_pointer(foo, new);
+ qemu_mutex_unlock(&foo_mutex);
+ synchronize_rcu();
+ foo_unref(old);
+
+or with call_rcu:
+
+ qemu_mutex_lock(&foo_mutex);
+ old = foo;
+ rcu_assign_pointer(foo, new);
+ qemu_mutex_unlock(&foo_mutex);
+ call_rcu(foo_unref, old, rcu);
+
+In both cases, the write side only performs removal. Reclamation
+happens when the last reference to a "foo" object is dropped.
+Using synchronize_rcu() is undesirably expensive, because the
+last reference may be dropped on the read side. Hence you can
+use call_rcu() instead:
+
+ foo_unref(struct foo *p) {
+ if (atomic_dec(&p->refcount) == 0) {
+ call_rcu(foo_destroy, p, rcu);
+ }
+ }
+
+
+Note that the same idioms would be possible with reader/writer
locks:
read_lock(&foo_rwlock); write_mutex_lock(&foo_rwlock);
@@ -239,13 +327,25 @@ locks:
write_mutex_unlock(&foo_rwlock);
free(p);
+ ------------------------------------------------------------------
+
+ read_lock(&foo_rwlock); write_mutex_lock(&foo_rwlock);
+ p = foo; old = foo;
+ foo_ref(p); foo = new;
+ read_unlock(&foo_rwlock); write_mutex_unlock(&foo_rwlock);
+ /* do something with p. */ foo_unref(old);
+ foo_unref(p);
+
+foo_unref could use a mechanism such as bottom halves to move deallocation
+out of hot paths.
+
RCU resizable arrays
--------------------
Resizable arrays can be used with RCU. The expensive RCU synchronization
-only needs to take place when the array is resized. The two items to
-take care of are:
+(or call_rcu) only needs to take place when the array is resized.
+The two items to take care of are:
- ensuring that the old version of the array is available between removal
and reclamation;
diff --git a/include/qemu/rcu.h b/include/qemu/rcu.h
index b875593..e43b912 100644
--- a/include/qemu/rcu.h
+++ b/include/qemu/rcu.h
@@ -161,6 +161,28 @@ extern void synchronize_rcu(void);
extern void rcu_register_thread(void);
extern void rcu_unregister_thread(void);
+struct rcu_head;
+typedef void RCUCBFunc(struct rcu_head *head);
+
+struct rcu_head {
+ struct rcu_head *next;
+ RCUCBFunc *func;
+};
+
+extern void call_rcu1(struct rcu_head *head, RCUCBFunc *func);
+
+/* The operands of the minus operator must have the same type,
+ * which must be the one that we specify in the cast.
+ */
+#define call_rcu(head, func, field) \
+ call_rcu1(({ \
+ char __attribute__((unused)) \
+ offset_must_be_zero[-offsetof(typeof(*(head)), field)], \
+ func_type_invalid = (func) - (void (*)(typeof(head)))(func); \
+ &(head)->field; \
+ }), \
+ (RCUCBFunc *)(func))
+
#ifdef __cplusplus
}
#endif
diff --git a/util/rcu.c b/util/rcu.c
index 48686a3..27fda86 100644
--- a/util/rcu.c
+++ b/util/rcu.c
@@ -26,6 +26,7 @@
* IBM's contributions to this file may be relicensed under LGPLv2 or later.
*/
+#include "qemu-common.h"
#include <stdio.h>
#include <assert.h>
#include <stdlib.h>
@@ -33,6 +34,7 @@
#include <errno.h>
#include "qemu/rcu.h"
#include "qemu/atomic.h"
+#include "qemu/thread.h"
/*
* Global grace period counter. Bit 0 is one if the thread is online.
@@ -174,6 +176,119 @@ void synchronize_rcu(void)
}
}
+
+#define RCU_CALL_MIN_SIZE 30
+
+/* Multi-producer, single-consumer queue based on urcu/static/wfqueue.h
+ * from liburcu. Note that head is only used by the consumer.
+ */
+static struct rcu_head dummy;
+static struct rcu_head *head = &dummy, **tail = &dummy.next;
+static int rcu_call_count;
+static QemuEvent rcu_call_ready_event;
+
+static void enqueue(struct rcu_head *node)
+{
+ struct rcu_head **old_tail;
+
+ node->next = NULL;
+ old_tail = atomic_xchg(&tail, &node->next);
+ atomic_mb_set(old_tail, node);
+}
+
+static struct rcu_head *try_dequeue(void)
+{
+ struct rcu_head *node, *next;
+
+retry:
+ /* Test for an empty list, which we do not expect. Note that for
+ * the consumer head and tail are always consistent. The head
+ * is consistent because only the consumer reads/writes it.
+ * The tail, because it is the first step in the enqueuing.
+ * It is only the next pointers that might be inconsistent.
+ */
+ if (head == &dummy && atomic_mb_read(&tail) == &dummy.next) {
+ abort();
+ }
+
+ /* If the head node has NULL in its next pointer, the value is
+ * wrong and we need to wait until its enqueuer finishes the update.
+ */
+ node = head;
+ next = atomic_mb_read(&head->next);
+ if (!next) {
+ return NULL;
+ }
+
+ /* Since we are the sole consumer, and can_dequeue() excludes the
+ * empty case, the queue will always have at least two nodes: the
+ * dummy node, and the one being removed. So we do not need to update
+ * the tail pointer.
+ */
+ head = next;
+
+ /* If we dequeued the dummy node, add it back at the end and retry. */
+ if (node == &dummy) {
+ enqueue(node);
+ goto retry;
+ }
+
+ return node;
+}
+
+static void *call_rcu_thread(void *opaque)
+{
+ struct rcu_head *node;
+
+ /* This thread is just a writer. */
+ rcu_thread_offline();
+
+ for (;;) {
+ int tries = 0;
+ int n = atomic_read(&rcu_call_count);
+
+ /* Heuristically wait for a decent number of callbacks to pile up.
+ * Fetch rcu_call_count now, we only must process elements that were
+ * added before synchronize_rcu() starts.
+ */
+ while (n < RCU_CALL_MIN_SIZE && ++tries <= 5) {
+ g_usleep(100000);
+ qemu_event_reset(&rcu_call_ready_event);
+ n = atomic_read(&rcu_call_count);
+ if (n < RCU_CALL_MIN_SIZE) {
+ qemu_event_wait(&rcu_call_ready_event);
+ n = atomic_read(&rcu_call_count);
+ }
+ }
+
+ atomic_sub(&rcu_call_count, n);
+ synchronize_rcu();
+ while (n > 0) {
+ node = try_dequeue();
+ while (!node) {
+ qemu_event_reset(&rcu_call_ready_event);
+ node = try_dequeue();
+ if (!node) {
+ qemu_event_wait(&rcu_call_ready_event);
+ node = try_dequeue();
+ }
+ }
+
+ n--;
+ node->func(node);
+ }
+ }
+ abort();
+}
+
+void call_rcu1(struct rcu_head *node, void (*func)(struct rcu_head *node))
+{
+ node->func = func;
+ enqueue(node);
+ atomic_inc(&rcu_call_count);
+ qemu_event_set(&rcu_call_ready_event);
+}
+
void rcu_register_thread(void)
{
if (!get_rcu_reader()) {
@@ -197,7 +312,12 @@ void rcu_unregister_thread(void)
static void __attribute__((__constructor__)) rcu_init(void)
{
+ QemuThread thread;
+
qemu_mutex_init(&rcu_gp_lock);
qemu_event_init(&rcu_gp_event, true);
+
+ qemu_event_init(&rcu_call_ready_event, false);
+ qemu_thread_create(&thread, call_rcu_thread, NULL, QEMU_THREAD_DETACHED);
rcu_register_thread();
}
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 06/14] rcu: add rcutorture
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (4 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 05/14] rcu: add call_rcu Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 07/14] rcu: allow nested calls to rcu_thread_offline/rcu_thread_online Mike Day
` (7 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
A stress test program (works, too :)).
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
tests/Makefile | 4 +-
tests/rcutorture.c | 439 +++++++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 442 insertions(+), 1 deletion(-)
create mode 100644 tests/rcutorture.c
diff --git a/tests/Makefile b/tests/Makefile
index d044908..b4a52b4 100644
--- a/tests/Makefile
+++ b/tests/Makefile
@@ -99,7 +99,8 @@ test-obj-y = tests/check-qint.o tests/check-qstring.o tests/check-qdict.o \
tests/test-string-input-visitor.o tests/test-qmp-output-visitor.o \
tests/test-qmp-input-visitor.o tests/test-qmp-input-strict.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-x86-cpuid.o tests/test-mul64.o tests/rcutortore.o \
+ tests/test-int128.o
test-qapi-obj-y = tests/test-qapi-visit.o tests/test-qapi-types.o
@@ -122,6 +123,7 @@ tests/test-hbitmap$(EXESUF): tests/test-hbitmap.o libqemuutil.a libqemustub.a
tests/test-x86-cpuid$(EXESUF): tests/test-x86-cpuid.o
tests/test-xbzrle$(EXESUF): tests/test-xbzrle.o xbzrle.o page_cache.o libqemuutil.a
tests/test-cutils$(EXESUF): tests/test-cutils.o util/cutils.o
++tests/rcutorture$(EXESUF): tests/rcutorture.o libqemuutil.a
tests/test-int128$(EXESUF): tests/test-int128.o
tests/test-qapi-types.c tests/test-qapi-types.h :\
diff --git a/tests/rcutorture.c b/tests/rcutorture.c
new file mode 100644
index 0000000..236c628
--- /dev/null
+++ b/tests/rcutorture.c
@@ -0,0 +1,439 @@
+/*
+ * rcutorture.c: simple user-level performance/stress test of RCU.
+ *
+ * Usage:
+ * ./rcu <nreaders> rperf [ <seconds> ]
+ * Run a read-side performance test with the specified
+ * number of readers for <seconds> seconds.
+ * ./rcu <nupdaters> uperf [ <seconds> ]
+ * Run an update-side performance test with the specified
+ * number of updaters and specified duration.
+ * ./rcu <nreaders> perf [ <seconds> ]
+ * Run a combined read/update performance test with the specified
+ * number of readers and one updater and specified duration.
+ *
+ * The above tests produce output as follows:
+ *
+ * n_reads: 46008000 n_updates: 146026 nreaders: 2 nupdaters: 1 duration: 1
+ * ns/read: 43.4707 ns/update: 6848.1
+ *
+ * The first line lists the total number of RCU reads and updates executed
+ * during the test, the number of reader threads, the number of updater
+ * threads, and the duration of the test in seconds. The second line
+ * lists the average duration of each type of operation in nanoseconds,
+ * or "nan" if the corresponding type of operation was not performed.
+ *
+ * ./rcu <nreaders> stress [ <seconds> ]
+ * Run a stress test with the specified number of readers and
+ * one updater.
+ *
+ * This test produces output as follows:
+ *
+ * n_reads: 114633217 n_updates: 3903415 n_mberror: 0
+ * rcu_stress_count: 114618391 14826 0 0 0 0 0 0 0 0 0
+ *
+ * The first line lists the number of RCU read and update operations
+ * executed, followed by the number of memory-ordering violations
+ * (which will be zero in a correct RCU implementation). The second
+ * line lists the number of readers observing progressively more stale
+ * data. A correct RCU implementation will have all but the first two
+ * numbers non-zero.
+ *
+ * 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) 2008 Paul E. McKenney, IBM Corporation.
+ */
+
+/*
+ * Test variables.
+ */
+
+#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/thread.h"
+
+long long n_reads = 0LL;
+long n_updates = 0L;
+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 NR_THREADS 100
+static QemuThread threads[NR_THREADS];
+static struct rcu_reader_data *data[NR_THREADS];
+static int n_threads;
+
+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], 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;
+}
+
+/*
+ * Performance test.
+ */
+
+static void *rcu_read_perf_test(void *arg)
+{
+ int i;
+ long long n_reads_local = 0;
+
+ *(struct rcu_reader_data **)arg = get_rcu_reader();
+ atomic_inc(&nthreadsrunning);
+ rcu_thread_offline();
+ while (goflag == GOFLAG_INIT) {
+ g_usleep(1000);
+ }
+ rcu_thread_online();
+ while (goflag == GOFLAG_RUN) {
+ for (i = 0; i < RCU_READ_RUN; i++) {
+ rcu_read_lock();
+ rcu_read_unlock();
+ }
+ n_reads_local += RCU_READ_RUN;
+ rcu_quiescent_state();
+ }
+ atomic_add(&n_reads, n_reads_local);
+
+ return NULL;
+}
+
+static void *rcu_update_perf_test(void *arg)
+{
+ long long n_updates_local = 0;
+
+ *(struct rcu_reader_data **)arg = get_rcu_reader();
+ atomic_inc(&nthreadsrunning);
+ while (goflag == GOFLAG_INIT) {
+ g_usleep(1000);
+ }
+ while (goflag == GOFLAG_RUN) {
+ synchronize_rcu();
+ n_updates_local++;
+ }
+ atomic_add(&n_updates, n_updates_local);
+ return NULL;
+}
+
+static void perftestinit(void)
+{
+ nthreadsrunning = 0;
+}
+
+static void perftestrun(int nthreads, int duration, int nreaders, int nupdaters)
+{
+ while (atomic_read(&nthreadsrunning) < nthreads) {
+ g_usleep(1000);
+ }
+ goflag = GOFLAG_RUN;
+ sleep(duration);
+ goflag = GOFLAG_STOP;
+ wait_all_threads();
+ printf("n_reads: %lld n_updates: %ld nreaders: %d nupdaters: %d duration: %d\n",
+ n_reads, n_updates, nreaders, nupdaters, duration);
+ printf("ns/read: %g ns/update: %g\n",
+ ((duration * 1000*1000*1000.*(double)nreaders) /
+ (double)n_reads),
+ ((duration * 1000*1000*1000.*(double)nupdaters) /
+ (double)n_updates));
+ exit(0);
+}
+
+static void perftest(int nreaders, int duration)
+{
+ int i;
+
+ perftestinit();
+ for (i = 0; i < nreaders; i++) {
+ create_thread(rcu_read_perf_test);
+ }
+ create_thread(rcu_update_perf_test);
+ perftestrun(i + 1, duration, nreaders, 1);
+}
+
+static void rperftest(int nreaders, int duration)
+{
+ int i;
+
+ perftestinit();
+ for (i = 0; i < nreaders; i++) {
+ create_thread(rcu_read_perf_test);
+ }
+ perftestrun(i, duration, nreaders, 0);
+}
+
+static void uperftest(int nupdaters, int duration)
+{
+ int i;
+
+ perftestinit();
+ for (i = 0; i < nupdaters; i++) {
+ create_thread(rcu_update_perf_test);
+ }
+ perftestrun(i, duration, 0, nupdaters);
+}
+
+/*
+ * Stress test.
+ */
+
+#define RCU_STRESS_PIPE_LEN 10
+
+struct rcu_stress {
+ int pipe_count;
+ int mbtest;
+};
+
+struct rcu_stress rcu_stress_array[RCU_STRESS_PIPE_LEN] = { { 0 } };
+struct rcu_stress *rcu_stress_current;
+int rcu_stress_idx;
+
+int n_mberror;
+long long rcu_stress_count[RCU_STRESS_PIPE_LEN + 1];
+
+
+static void *rcu_read_stress_test(void *arg)
+{
+ int i;
+ int itercnt = 0;
+ struct rcu_stress *p;
+ int pc;
+ long long n_reads_local = 0;
+ volatile int garbage;
+
+ *(struct rcu_reader_data **)arg = get_rcu_reader();
+ rcu_thread_offline();
+ while (goflag == GOFLAG_INIT) {
+ g_usleep(1000);
+ }
+ rcu_thread_online();
+ while (goflag == GOFLAG_RUN) {
+ rcu_read_lock();
+ p = rcu_dereference(&rcu_stress_current);
+ if (p->mbtest == 0) {
+ n_mberror++;
+ }
+ rcu_read_lock();
+ for (i = 0; i < 100; i++)
+ garbage++;
+ rcu_read_unlock();
+ pc = p->pipe_count;
+ rcu_read_unlock();
+ if ((pc > RCU_STRESS_PIPE_LEN) || (pc < 0)) {
+ pc = RCU_STRESS_PIPE_LEN;
+ }
+ atomic_inc(&rcu_stress_count[pc]);
+ n_reads_local++;
+ rcu_quiescent_state();
+ if ((++itercnt % 0x1000) == 0) {
+ synchronize_rcu();
+ }
+ }
+ atomic_add(&n_reads, n_reads_local);
+
+ return NULL;
+}
+
+static void *rcu_update_stress_test(void *arg)
+{
+ int i;
+ struct rcu_stress *p;
+
+ *(struct rcu_reader_data **)arg = get_rcu_reader();
+ while (goflag == GOFLAG_INIT) {
+ g_usleep(1000);
+ }
+ while (goflag == GOFLAG_RUN) {
+ i = rcu_stress_idx + 1;
+ if (i >= RCU_STRESS_PIPE_LEN) {
+ i = 0;
+ }
+ p = &rcu_stress_array[i];
+ p->mbtest = 0;
+ smp_mb();
+ p->pipe_count = 0;
+ p->mbtest = 1;
+ rcu_assign_pointer(&rcu_stress_current, &p);
+ rcu_stress_idx = i;
+ for (i = 0; i < RCU_STRESS_PIPE_LEN; i++)
+ if (i != rcu_stress_idx) {
+ rcu_stress_array[i].pipe_count++;
+ }
+ synchronize_rcu();
+ n_updates++;
+ }
+ return NULL;
+}
+
+static void *rcu_fake_update_stress_test(void *arg)
+{
+ *(struct rcu_reader_data **)arg = get_rcu_reader();
+ while (goflag == GOFLAG_INIT) {
+ g_usleep(1000);
+ }
+ while (goflag == GOFLAG_RUN) {
+ synchronize_rcu();
+ g_usleep(1000);
+ }
+ return NULL;
+}
+
+static void stresstest(int nreaders, int duration)
+{
+ int i;
+
+ rcu_stress_current = &rcu_stress_array[0];
+ rcu_stress_current->pipe_count = 0;
+ rcu_stress_current->mbtest = 1;
+ for (i = 0; i < nreaders; i++) {
+ create_thread(rcu_read_stress_test);
+ }
+ create_thread(rcu_update_stress_test);
+ for (i = 0; i < 5; i++) {
+ create_thread(rcu_fake_update_stress_test);
+ }
+ goflag = GOFLAG_RUN;
+ sleep(duration);
+ goflag = GOFLAG_STOP;
+ wait_all_threads();
+ printf("n_reads: %lld n_updates: %ld n_mberror: %d\n",
+ n_reads, n_updates, n_mberror);
+ printf("rcu_stress_count:");
+ for (i = 0; i <= RCU_STRESS_PIPE_LEN; i++) {
+ printf(" %lld", rcu_stress_count[i]);
+ }
+ printf("\n");
+ exit(0);
+}
+
+/* GTest interface */
+
+static void gtest_stress(int nreaders, int duration)
+{
+ int i;
+
+ rcu_stress_current = &rcu_stress_array[0];
+ rcu_stress_current->pipe_count = 0;
+ rcu_stress_current->mbtest = 1;
+ for (i = 0; i < nreaders; i++) {
+ create_thread(rcu_read_stress_test);
+ }
+ create_thread(rcu_update_stress_test);
+ for (i = 0; i < 5; i++) {
+ create_thread(rcu_fake_update_stress_test);
+ }
+ goflag = GOFLAG_RUN;
+ sleep(duration);
+ goflag = GOFLAG_STOP;
+ wait_all_threads();
+ g_assert_cmpint(n_mberror, ==, 0);
+ for (i = 2; i <= RCU_STRESS_PIPE_LEN; i++) {
+ g_assert_cmpint(rcu_stress_count[i], ==, 0);
+ }
+}
+
+static void gtest_stress_1_1(void)
+{
+ gtest_stress(1, 1);
+}
+
+static void gtest_stress_10_1(void)
+{
+ gtest_stress(10, 1);
+}
+
+static void gtest_stress_1_5(void)
+{
+ gtest_stress(1, 5);
+}
+
+static void gtest_stress_10_5(void)
+{
+ gtest_stress(10, 5);
+}
+
+/*
+ * Mainprogram.
+ */
+
+static void usage(int argc, char *argv[])
+{
+ fprintf(stderr, "Usage: %s [nreaders [ perf | stress ] ]\n", argv[0]);
+ exit(-1);
+}
+
+int main(int argc, char *argv[])
+{
+ int nreaders = 1;
+ int duration = 1;
+
+ /* This thread is not part of the test. */
+ rcu_thread_offline();
+
+ if (argc >= 2 && argv[1][0] == '-') {
+ g_test_init(&argc, &argv, NULL);
+ g_test_add_func("/rcutorture/short/1reader", gtest_stress_1_1);
+ g_test_add_func("/rcutorture/short/10readers", gtest_stress_10_1);
+ g_test_add_func("/rcutorture/long/1reader", gtest_stress_1_5);
+ g_test_add_func("/rcutorture/long/10readers", gtest_stress_10_5);
+ return g_test_run();
+ }
+
+ if (argc >= 2) {
+ nreaders = strtoul(argv[1], NULL, 0);
+ }
+ if (argc > 3) {
+ duration = strtoul(argv[3], NULL, 0);
+ }
+ if (argc < 3 || strcmp(argv[2], "stress") == 0) {
+ stresstest(nreaders, duration);
+ } else if (strcmp(argv[2], "rperf") == 0) {
+ rperftest(nreaders, duration);
+ } else if (strcmp(argv[2], "uperf") == 0) {
+ uperftest(nreaders, duration);
+ } else if (strcmp(argv[2], "perf") == 0) {
+ perftest(nreaders, duration);
+ }
+ usage(argc, argv);
+ return 0;
+}
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 07/14] rcu: allow nested calls to rcu_thread_offline/rcu_thread_online
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (5 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 06/14] rcu: add rcutorture Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 08/14] qemu-thread: report RCU quiescent states Mike Day
` (6 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
docs/rcu.txt | 5 +++++
include/qemu/rcu.h | 21 +++++++++++++++++++--
2 files changed, 24 insertions(+), 2 deletions(-)
diff --git a/docs/rcu.txt b/docs/rcu.txt
index d7c4f0b..4e7cde3 100644
--- a/docs/rcu.txt
+++ b/docs/rcu.txt
@@ -187,6 +187,11 @@ Marking quiescent states is done with the following three APIs:
thread.
+rcu_thread_offline() and rcu_thread_online() can be nested. The end of
+the extended quiescent state will coincide with the outermost call to
+rcu_thread_online().
+
+
The following APIs can be used to use RCU in a thread that is not
created with qemu_thread_create():
diff --git a/include/qemu/rcu.h b/include/qemu/rcu.h
index e43b912..3a55045 100644
--- a/include/qemu/rcu.h
+++ b/include/qemu/rcu.h
@@ -77,6 +77,9 @@ struct rcu_reader_data {
unsigned long ctr;
bool waiting;
+ /* Data used by reader only */
+ unsigned offline;
+
/* Data used for registry, protected by rcu_gp_lock */
QLIST_ENTRY(rcu_reader_data) node;
};
@@ -127,9 +130,14 @@ static inline void rcu_read_unlock(void)
static inline void rcu_quiescent_state(void)
{
struct rcu_reader_data *p_rcu_reader = get_rcu_reader();
+ unsigned ctr;
+
+ if (p_rcu_reader->offline > 0) {
+ return;
+ }
/* Reuses smp_rmb() in the last rcu_read_unlock(). */
- unsigned ctr = atomic_read(&rcu_gp_ctr);
+ ctr = atomic_read(&rcu_gp_ctr);
atomic_xchg(&p_rcu_reader->ctr, ctr);
if (atomic_read(&p_rcu_reader->waiting)) {
atomic_set(&p_rcu_reader->waiting, false);
@@ -141,6 +149,10 @@ static inline void rcu_thread_offline(void)
{
struct rcu_reader_data *p_rcu_reader = get_rcu_reader();
+ if (p_rcu_reader->offline++ > 0) {
+ return;
+ }
+
atomic_xchg(&p_rcu_reader->ctr, 0);
if (atomic_read(&p_rcu_reader->waiting)) {
atomic_set(&p_rcu_reader->waiting, false);
@@ -150,7 +162,12 @@ static inline void rcu_thread_offline(void)
static inline void rcu_thread_online(void)
{
- rcu_quiescent_state();
+ struct rcu_reader_data *p_rcu_reader = get_rcu_reader();
+
+ assert(p_rcu_reader->offline != 0);
+ if (--p_rcu_reader->offline == 0) {
+ rcu_quiescent_state();
+ }
}
extern void synchronize_rcu(void);
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 08/14] qemu-thread: report RCU quiescent states
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (6 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 07/14] rcu: allow nested calls to rcu_thread_offline/rcu_thread_online Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 09/14] event loop: " Mike Day
` (5 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
Most threads will use mutexes and other sleeping synchronization primitives
(condition variables, semaphores, events) periodically. For these threads,
the synchronization primitives are natural places to report a quiescent
state (possibly an extended one).
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
docs/rcu.txt | 27 +++++++++++++++++++++++++++
util/qemu-thread-posix.c | 29 +++++++++++++++++++++++++----
util/qemu-thread-win32.c | 16 +++++++++++++++-
util/rcu.c | 3 ---
4 files changed, 67 insertions(+), 8 deletions(-)
diff --git a/docs/rcu.txt b/docs/rcu.txt
index 4e7cde3..38fd8f4 100644
--- a/docs/rcu.txt
+++ b/docs/rcu.txt
@@ -168,6 +168,33 @@ of "quiescent states", i.e. points where no RCU read-side critical
section can be active. All threads created with qemu_thread_create
participate in the RCU mechanism and need to annotate such points.
+Luckily, in most cases no manual annotation is needed, because waiting
+on condition variables (qemu_cond_wait), semaphores (qemu_sem_wait,
+qemu_sem_timedwait) or events (qemu_event_wait) implicitly marks the thread
+as quiescent for the whole duration of the wait. (There is an exception
+for semaphore waits with a zero timeout).
+
+Manual annotation is still needed in the following cases:
+
+- threads that spend their sleeping time in the kernel, for example
+ in a call to select(), poll() or WaitForMultipleObjects(). The QEMU
+ I/O thread is an example of this case.
+
+- threads that perform a lot of I/O. In QEMU, the workers used for
+ aio=thread are an example of this case (see aio_worker in block/raw-*).
+
+- threads that run continuously until they exit. The migration thread
+ is an example of this case.
+
+Regarding the second case, note that the workers run in the QEMU thread
+pool. The thread pool uses semaphores for synchronization, hence it does
+report quiescent states periodically. However, in some cases (e.g. NFS
+mounted with the "hard" option) the workers can take an arbitrarily long
+amount of time. When this happens, synchronize_rcu() will not exit and
+call_rcu() callbacks will be delayed arbitrarily. It is therefore a
+good idea to mark I/O system calls as quiescence points in the worker
+functions.
+
Marking quiescent states is done with the following three APIs:
void rcu_quiescent_state(void);
diff --git a/util/qemu-thread-posix.c b/util/qemu-thread-posix.c
index 2371176..dbefff8 100644
--- a/util/qemu-thread-posix.c
+++ b/util/qemu-thread-posix.c
@@ -119,7 +119,9 @@ void qemu_cond_wait(QemuCond *cond, QemuMutex *mutex)
{
int err;
+ rcu_thread_offline();
err = pthread_cond_wait(&cond->cond, &mutex->lock);
+ rcu_thread_online();
if (err)
error_exit(err, __func__);
}
@@ -211,6 +213,10 @@ int qemu_sem_timedwait(QemuSemaphore *sem, int ms)
int rc;
struct timespec ts;
+ if (ms) {
+ rcu_thread_offline();
+ }
+
#if defined(__APPLE__) || defined(__NetBSD__)
rc = 0;
compute_abs_deadline(&ts, ms);
@@ -228,7 +234,10 @@ int qemu_sem_timedwait(QemuSemaphore *sem, int ms)
--sem->count;
}
pthread_mutex_unlock(&sem->lock);
- return (rc == ETIMEDOUT ? -1 : 0);
+ if (rc == ETIMEDOUT) {
+ rc == -1;
+ }
+
#else
if (ms <= 0) {
/* This is cheaper than sem_timedwait. */
@@ -236,7 +245,7 @@ int qemu_sem_timedwait(QemuSemaphore *sem, int ms)
rc = sem_trywait(&sem->sem);
} while (rc == -1 && errno == EINTR);
if (rc == -1 && errno == EAGAIN) {
- return -1;
+ goto out;
}
} else {
compute_abs_deadline(&ts, ms);
@@ -244,19 +253,26 @@ int qemu_sem_timedwait(QemuSemaphore *sem, int ms)
rc = sem_timedwait(&sem->sem, &ts);
} while (rc == -1 && errno == EINTR);
if (rc == -1 && errno == ETIMEDOUT) {
- return -1;
+ goto out;
}
}
if (rc < 0) {
error_exit(errno, __func__);
}
- return 0;
+
#endif
+
+out:
+ if (ms) {
+ rcu_thread_online();
+ }
+ return rc;
}
void qemu_sem_wait(QemuSemaphore *sem)
{
int rc;
+ rcu_thread_offline();
#if defined(__APPLE__) || defined(__NetBSD__)
pthread_mutex_lock(&sem->lock);
@@ -276,6 +292,7 @@ void qemu_sem_wait(QemuSemaphore *sem)
error_exit(errno, __func__);
}
#endif
+ rcu_thread_online();
}
#ifdef __linux__
@@ -384,7 +401,11 @@ void qemu_event_wait(QemuEvent *ev)
return;
}
}
+ rcu_thread_offline();
futex_wait(ev, EV_BUSY);
+ rcu_thread_online();
+ } else {
+ rcu_quiescent_state();
}
}
diff --git a/util/qemu-thread-win32.c b/util/qemu-thread-win32.c
index 0c4850d..7bc07b3 100644
--- a/util/qemu-thread-win32.c
+++ b/util/qemu-thread-win32.c
@@ -12,6 +12,7 @@
*/
#include "qemu-common.h"
#include "qemu/thread.h"
+#include "qemu/rcu.h"
#include <process.h>
#include <assert.h>
#include <limits.h>
@@ -170,7 +171,9 @@ void qemu_cond_wait(QemuCond *cond, QemuMutex *mutex)
* leaving mutex unlocked before we wait on semaphore.
*/
qemu_mutex_unlock(mutex);
+ rcu_thread_offline();
WaitForSingleObject(cond->sema, INFINITE);
+ rcu_thread_online();
/* Now waiters must rendez-vous with the signaling thread and
* let it continue. For cond_broadcast this has heavy contention
@@ -210,7 +213,16 @@ void qemu_sem_post(QemuSemaphore *sem)
int qemu_sem_timedwait(QemuSemaphore *sem, int ms)
{
- int rc = WaitForSingleObject(sem->sema, ms);
+ int rc;
+
+ if (ms) {
+ rcu_thread_offline();
+ }
+ rc = WaitForSingleObject(sem->sema, ms);
+ if (ms) {
+ rcu_thread_offline();
+ }
+
if (rc == WAIT_OBJECT_0) {
return 0;
}
@@ -250,7 +262,9 @@ void qemu_event_reset(QemuEvent *ev)
void qemu_event_wait(QemuEvent *ev)
{
+ rcu_thread_offline();
WaitForSingleObject(ev->event, INFINITE);
+ rcu_thread_online();
}
struct QemuThreadData {
diff --git a/util/rcu.c b/util/rcu.c
index 27fda86..91d6ae2 100644
--- a/util/rcu.c
+++ b/util/rcu.c
@@ -240,9 +240,6 @@ static void *call_rcu_thread(void *opaque)
{
struct rcu_head *node;
- /* This thread is just a writer. */
- rcu_thread_offline();
-
for (;;) {
int tries = 0;
int n = atomic_read(&rcu_call_count);
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 09/14] event loop: report RCU quiescent states
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (7 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 08/14] qemu-thread: report RCU quiescent states Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 10/14] cpus: " Mike Day
` (4 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
Threads that run event loops also have places that can sleep for an extended
time. Place an extended quiescent state there.
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
aio-posix.c | 9 ++++++++-
aio-win32.c | 7 +++++++
main-loop.c | 7 ++++++-
3 files changed, 21 insertions(+), 2 deletions(-)
diff --git a/aio-posix.c b/aio-posix.c
index b68eccd..562add6 100644
--- a/aio-posix.c
+++ b/aio-posix.c
@@ -17,6 +17,7 @@
#include "block/block.h"
#include "qemu/queue.h"
#include "qemu/sockets.h"
+#include "qemu/rcu.h"
struct AioHandler
{
@@ -232,10 +233,16 @@ bool aio_poll(AioContext *ctx, bool blocking)
}
/* wait until next event */
+ if (blocking) {
+ rcu_thread_offline();
+ }
ret = g_poll((GPollFD *)ctx->pollfds->data,
ctx->pollfds->len,
blocking ? -1 : 0);
-
+ if (blocking) {
+ rcu_thread_online();
+ }
+
/* if we have any readable fds, dispatch event */
if (ret > 0) {
QLIST_FOREACH(node, &ctx->aio_handlers, node) {
diff --git a/aio-win32.c b/aio-win32.c
index 38723bf..8a6abb0 100644
--- a/aio-win32.c
+++ b/aio-win32.c
@@ -175,7 +175,14 @@ bool aio_poll(AioContext *ctx, bool blocking)
/* wait until next event */
while (count > 0) {
int timeout = blocking ? INFINITE : 0;
+
+ if (timeout) {
+ rcu_thread_offline();
+ }
int ret = WaitForMultipleObjects(count, events, FALSE, timeout);
+ if (timeout) {
+ rcu_thread_online();
+ }
/* if we have any signaled events, dispatch event */
if ((DWORD) (ret - WAIT_OBJECT_0) >= count) {
diff --git a/main-loop.c b/main-loop.c
index a44fff6..79f43da 100644
--- a/main-loop.c
+++ b/main-loop.c
@@ -27,6 +27,7 @@
#include "qemu/sockets.h" // struct in_addr needed for libslirp.h
#include "slirp/libslirp.h"
#include "qemu/main-loop.h"
+#include "qemu/rcu.h"
#include "block/aio.h"
#ifndef _WIN32
@@ -220,6 +221,7 @@ static int os_host_main_loop_wait(uint32_t timeout)
if (timeout > 0) {
spin_counter = 0;
qemu_mutex_unlock_iothread();
+ rcu_thread_offline();
} else {
spin_counter++;
}
@@ -227,7 +229,8 @@ static int os_host_main_loop_wait(uint32_t timeout)
ret = g_poll((GPollFD *)gpollfds->data, gpollfds->len, timeout);
if (timeout > 0) {
- qemu_mutex_lock_iothread();
+ rcu_thread_online();
+ qemu_mutex_lock_iothread();
}
glib_pollfds_poll();
@@ -424,7 +427,9 @@ static int os_host_main_loop_wait(uint32_t timeout)
}
qemu_mutex_unlock_iothread();
+ rcu_thread_offline();
g_poll_ret = g_poll(poll_fds, n_poll_fds + w->num, poll_timeout);
+ rcu_thread_online();
qemu_mutex_lock_iothread();
if (g_poll_ret > 0) {
for (i = 0; i < w->num; i++) {
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 10/14] cpus: report RCU quiescent states
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (8 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 09/14] event loop: " Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 11/14] block: " Mike Day
` (3 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
CPU threads have extended quiescent states while relinquishing control
to the accelerator (except TCG).
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
cpus.c | 3 +++
kvm-all.c | 3 +++
2 files changed, 6 insertions(+)
diff --git a/cpus.c b/cpus.c
index 0f65e76..624658e 100644
--- a/cpus.c
+++ b/cpus.c
@@ -37,6 +37,7 @@
#include "sysemu/qtest.h"
#include "qemu/main-loop.h"
#include "qemu/bitmap.h"
+#include "qemu/rcu.h"
#ifndef _WIN32
#include "qemu/compatfd.h"
@@ -818,6 +819,7 @@ static void *qemu_dummy_cpu_thread_fn(void *arg)
while (1) {
current_cpu = NULL;
qemu_mutex_unlock_iothread();
+ rcu_thread_offline();
do {
int sig;
r = sigwait(&waitset, &sig);
@@ -826,6 +828,7 @@ static void *qemu_dummy_cpu_thread_fn(void *arg)
perror("sigwait");
exit(1);
}
+ rcu_thread_online();
qemu_mutex_lock_iothread();
current_cpu = cpu;
qemu_wait_io_event_common(cpu);
diff --git a/kvm-all.c b/kvm-all.c
index 716860f..a47573d 100644
--- a/kvm-all.c
+++ b/kvm-all.c
@@ -33,6 +33,7 @@
#include "exec/memory.h"
#include "exec/address-spaces.h"
#include "qemu/event_notifier.h"
+#include "qemu/rcu.h"
#include "trace.h"
/* This check must be after config-host.h is included */
@@ -1641,7 +1642,9 @@ int kvm_cpu_exec(CPUState *cpu)
}
qemu_mutex_unlock_iothread();
+ rcu_thread_offline();
run_ret = kvm_vcpu_ioctl(cpu, KVM_RUN, 0);
+ rcu_thread_online();
qemu_mutex_lock_iothread();
kvm_arch_post_run(cpu, run);
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 11/14] block: report RCU quiescent states
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (9 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 10/14] cpus: " Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 12/14] migration: " Mike Day
` (2 subsequent siblings)
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
The aio workers may spend a long time executing I/O operations;
mark that time as an extended quiescent state.
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
block/raw-posix.c | 3 +++
block/raw-win32.c | 3 +++
2 files changed, 6 insertions(+)
diff --git a/block/raw-posix.c b/block/raw-posix.c
index ba721d3..962833e 100644
--- a/block/raw-posix.c
+++ b/block/raw-posix.c
@@ -26,6 +26,7 @@
#include "qemu/log.h"
#include "block/block_int.h"
#include "qemu/module.h"
+#include "qemu/rcu.h"
#include "trace.h"
#include "block/thread-pool.h"
#include "qemu/iov.h"
@@ -735,6 +736,7 @@ static int aio_worker(void *arg)
RawPosixAIOData *aiocb = arg;
ssize_t ret = 0;
+ rcu_thread_offline();
switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
case QEMU_AIO_READ:
ret = handle_aiocb_rw(aiocb);
@@ -774,6 +776,7 @@ static int aio_worker(void *arg)
}
g_slice_free(RawPosixAIOData, aiocb);
+ rcu_thread_online();
return ret;
}
diff --git a/block/raw-win32.c b/block/raw-win32.c
index 9b5b2af..5f88452 100644
--- a/block/raw-win32.c
+++ b/block/raw-win32.c
@@ -25,6 +25,7 @@
#include "qemu/timer.h"
#include "block/block_int.h"
#include "qemu/module.h"
+#include "qemu/rcu.h"
#include "raw-aio.h"
#include "trace.h"
#include "block/thread-pool.h"
@@ -99,6 +100,7 @@ static int aio_worker(void *arg)
ssize_t ret = 0;
size_t count;
+ rcu_thread_offline();
switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
case QEMU_AIO_READ:
count = handle_aiocb_rw(aiocb);
@@ -136,6 +138,7 @@ static int aio_worker(void *arg)
}
g_slice_free(RawWin32AIOData, aiocb);
+ rcu_thread_online();
return ret;
}
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [RFC PATCH 12/14] migration: report RCU quiescent states
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (10 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 11/14] block: " Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [PATCH 13/14] include osdep.h for definition of glue(a, b) Mike Day
2013-08-14 15:50 ` [Qemu-devel] [PATCH 14/14] fix pointer reference to rcu_assign_pointer Mike Day
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Paolo Bonzini
From: Paolo Bonzini <pbonzini@redhat.com>
The migration thread polls s->state periodically, it does not
use a mutex or condition variable, so it has to report quiescent
states manually.
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Reviewed-by: Mike Day <ncmike@ncultra.org>
---
migration.c | 2 ++
1 file changed, 2 insertions(+)
diff --git a/migration.c b/migration.c
index 1402fa7..9d0950e 100644
--- a/migration.c
+++ b/migration.c
@@ -22,6 +22,7 @@
#include "qemu/sockets.h"
#include "migration/block.h"
#include "qemu/thread.h"
+#include "qemu/rcu.h"
#include "qmp-commands.h"
#include "trace.h"
@@ -563,6 +564,7 @@ static void *migration_thread(void *opaque)
int64_t current_time;
uint64_t pending_size;
+ rcu_quiescent_state();
if (!qemu_file_rate_limit(s->file)) {
DPRINTF("iterate\n");
pending_size = qemu_savevm_state_pending(s->file, max_size);
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [PATCH 13/14] include osdep.h for definition of glue(a, b)
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (11 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [RFC PATCH 12/14] migration: " Mike Day
@ 2013-08-14 15:50 ` Mike Day
2013-08-14 15:50 ` [Qemu-devel] [PATCH 14/14] fix pointer reference to rcu_assign_pointer Mike Day
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Mike Day
diff --git a/include/qemu/rcu.h b/include/qemu/rcu.h
index 3a55045..f398e6c 100644
--- a/include/qemu/rcu.h
+++ b/include/qemu/rcu.h
@@ -39,6 +39,7 @@
#include "qemu/compiler.h"
#include "qemu/rcu-pointer.h"
#include "qemu/thread.h"
+#include "qemu/osdep.h"
#include "qemu/queue.h"
#include "qemu/atomic.h"
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread* [Qemu-devel] [PATCH 14/14] fix pointer reference to rcu_assign_pointer
2013-08-14 15:50 [Qemu-devel] [RFC PATCH 00/14] RCU Implementation for Qemu Mike Day
` (12 preceding siblings ...)
2013-08-14 15:50 ` [Qemu-devel] [PATCH 13/14] include osdep.h for definition of glue(a, b) Mike Day
@ 2013-08-14 15:50 ` Mike Day
13 siblings, 0 replies; 15+ messages in thread
From: Mike Day @ 2013-08-14 15:50 UTC (permalink / raw)
To: qemu-devel; +Cc: Mike Day
Signed-off-by: Mike Day <ncmike@ncultra.org>
---
tests/rcutorture.c | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/tests/rcutorture.c b/tests/rcutorture.c
index 236c628..b219ff5 100644
--- a/tests/rcutorture.c
+++ b/tests/rcutorture.c
@@ -293,7 +293,7 @@ static void *rcu_update_stress_test(void *arg)
smp_mb();
p->pipe_count = 0;
p->mbtest = 1;
- rcu_assign_pointer(&rcu_stress_current, &p);
+ rcu_assign_pointer(rcu_stress_current, p);
rcu_stress_idx = i;
for (i = 0; i < RCU_STRESS_PIPE_LEN; i++)
if (i != rcu_stress_idx) {
--
1.8.3.1
^ permalink raw reply related [flat|nested] 15+ messages in thread