[Qemu-devel] [PATCH 0/2] aio: add aio_context_acquire() and aio_context_release()

[Qemu-devel] [PATCH 0/2] aio: add aio_context_acquire() and aio_context_release()

[Stefan Hajnoczi]

aio_context_acquire() and aio_context_release() make it possible for multiple
threads to safely operate on a shared AioContext.  This is a prerequisite for
using the block layer outside the QEMU global mutex.

Imagine that a dataplane thread is performing I/O on behalf of the guest when
the user issues a monitor command that needs to access the BlockDriverState.
The monitor thread needs to acquire the AioContext before calling bdrv_*()
functions on it.  This prevents the dataplane thread from interfering with the
monitor thread.

There was a discussion in the RFC email thread about how to implement fairness:
each thread should get a turn to acquire the AioContext so that starvation is
avoided.

Paolo suggested doing what the QEMU main loop does today: drop the lock before
invoking ppoll(2).  It turns out that this is tricky since we want both threads
to be able to call aio_poll() simultaneously.  AioContext->pollfds[] currently
prevents two simultaneous aio_poll() calls since it is a shared resource.  It's
also tricky to avoid deadlocks if two threads execute aio_poll() simulatenously
except by waking all threads each time *any* thread makes any progress.

I found the simplest solution is to implement RFifoLock, a recursive lock with
FIFO ordering.  This lock supports the semantics we need for the following
pattern:

  /* Event loop thread */
  while (running) {
      aio_context_acquire(ctx);
      aio_poll(ctx, true);
      aio_context_release(ctx);
  }

  /* Another thread */
  aio_context_acquire(ctx);
  bdrv_read(bs, 0x1000, buf, 1);
  aio_context_release(ctx);

When the other thread wants to acquire the AioContext but the lock is
contended, it uses aio_notify(ctx) to bump the event loop thread out of
aio_poll().  Fairness ensures everyone gets an opportunity to use the
AioContext.

Stefan Hajnoczi (2):
  rfifolock: add recursive FIFO lock
  aio: add aio_context_acquire() and aio_context_release()

 async.c                  | 18 ++++++++++
 include/block/aio.h      | 18 ++++++++++
 include/qemu/rfifolock.h | 54 +++++++++++++++++++++++++++++
 tests/Makefile           |  2 ++
 tests/test-aio.c         | 58 +++++++++++++++++++++++++++++++
 tests/test-rfifolock.c   | 90 ++++++++++++++++++++++++++++++++++++++++++++++++
 util/Makefile.objs       |  1 +
 util/rfifolock.c         | 79 ++++++++++++++++++++++++++++++++++++++++++
 8 files changed, 320 insertions(+)
 create mode 100644 include/qemu/rfifolock.h
 create mode 100644 tests/test-rfifolock.c
 create mode 100644 util/rfifolock.c

-- 
1.8.3.1

[Qemu-devel] [PATCH 1/2] rfifolock: add recursive FIFO lock

[Stefan Hajnoczi]

QemuMutex does not guarantee fairness and cannot be acquired
recursively:

Fairness means each locker gets a turn and the scheduler cannot cause
starvation.

Recursive locking is useful for composition, it allows a sequence of
locking operations to be invoked atomically by acquiring the lock around
them.

This patch adds RFifoLock, a recursive lock that guarantees FIFO order.
Its first user is added in the next patch.

RFifoLock has one additional feature: it can be initialized with an
optional contention callback.  The callback is invoked whenever a thread
must wait for the lock.  For example, it can be used to poke the current
owner so that they release the lock soon.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/qemu/rfifolock.h | 54 +++++++++++++++++++++++++++++
 tests/Makefile           |  2 ++
 tests/test-rfifolock.c   | 90 ++++++++++++++++++++++++++++++++++++++++++++++++
 util/Makefile.objs       |  1 +
 util/rfifolock.c         | 79 ++++++++++++++++++++++++++++++++++++++++++
 5 files changed, 226 insertions(+)
 create mode 100644 include/qemu/rfifolock.h
 create mode 100644 tests/test-rfifolock.c
 create mode 100644 util/rfifolock.c

diff --git a/include/qemu/rfifolock.h b/include/qemu/rfifolock.h
new file mode 100644
index 0000000..b23ab53
--- /dev/null
+++ b/include/qemu/rfifolock.h
@@ -0,0 +1,54 @@
+/*
+ * Recursive FIFO lock
+ *
+ * Copyright Red Hat, Inc. 2013
+ *
+ * Authors:
+ *  Stefan Hajnoczi   <stefanha@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or later.
+ * See the COPYING file in the top-level directory.
+ *
+ */
+
+#ifndef QEMU_RFIFOLOCK_H
+#define QEMU_RFIFOLOCK_H
+
+#include "qemu/thread.h"
+
+/* Recursive FIFO lock
+ *
+ * This lock provides more features than a plain mutex:
+ *
+ * 1. Fairness - enforces FIFO order.
+ * 2. Nesting - can be taken recursively.
+ * 3. Contention callback - optional, called when thread must wait.
+ *
+ * The recursive FIFO lock is heavyweight so prefer other synchronization
+ * primitives if you do not need its features.
+ */
+typedef struct {
+    QemuMutex lock;             /* protects all fields */
+
+    /* FIFO order */
+    unsigned int head;          /* active ticket number */
+    unsigned int tail;          /* waiting ticket number */
+    QemuCond cond;              /* used to wait for our ticket number */
+
+    /* Nesting */
+    QemuThread owner_thread;    /* thread that currently has ownership */
+    unsigned int nesting;       /* amount of nesting levels */
+
+    /* Contention callback */
+    void (*cb)(void *);         /* called when thread must wait, with ->lock
+                                 * held so it may not recursively lock/unlock
+                                 */
+    void *cb_opaque;
+} RFifoLock;
+
+void rfifolock_init(RFifoLock *r, void (*cb)(void *), void *opaque);
+void rfifolock_destroy(RFifoLock *r);
+void rfifolock_lock(RFifoLock *r);
+void rfifolock_unlock(RFifoLock *r);
+
+#endif /* QEMU_RFIFOLOCK_H */
diff --git a/tests/Makefile b/tests/Makefile
index 994fef1..2959ed1 100644
--- a/tests/Makefile
+++ b/tests/Makefile
@@ -31,6 +31,7 @@ check-unit-y += tests/test-visitor-serialization$(EXESUF)
 check-unit-y += tests/test-iov$(EXESUF)
 gcov-files-test-iov-y = util/iov.c
 check-unit-y += tests/test-aio$(EXESUF)
+check-unit-y += tests/test-rfifolock$(EXESUF)
 check-unit-y += tests/test-throttle$(EXESUF)
 gcov-files-test-aio-$(CONFIG_WIN32) = aio-win32.c
 gcov-files-test-aio-$(CONFIG_POSIX) = aio-posix.c
@@ -121,6 +122,7 @@ tests/check-qfloat$(EXESUF): tests/check-qfloat.o libqemuutil.a
 tests/check-qjson$(EXESUF): tests/check-qjson.o libqemuutil.a libqemustub.a
 tests/test-coroutine$(EXESUF): tests/test-coroutine.o $(block-obj-y) libqemuutil.a libqemustub.a
 tests/test-aio$(EXESUF): tests/test-aio.o $(block-obj-y) libqemuutil.a libqemustub.a
+tests/test-rfifolock$(EXESUF): tests/test-rfifolock.o libqemuutil.a libqemustub.a
 tests/test-throttle$(EXESUF): tests/test-throttle.o $(block-obj-y) libqemuutil.a libqemustub.a
 tests/test-thread-pool$(EXESUF): tests/test-thread-pool.o $(block-obj-y) libqemuutil.a libqemustub.a
 tests/test-iov$(EXESUF): tests/test-iov.o libqemuutil.a
diff --git a/tests/test-rfifolock.c b/tests/test-rfifolock.c
new file mode 100644
index 0000000..440dbcb
--- /dev/null
+++ b/tests/test-rfifolock.c
@@ -0,0 +1,90 @@
+/*
+ * RFifoLock tests
+ *
+ * Copyright Red Hat, Inc. 2013
+ *
+ * Authors:
+ *  Stefan Hajnoczi    <stefanha@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU LGPL, version 2 or later.
+ * See the COPYING.LIB file in the top-level directory.
+ */
+
+#include <glib.h>
+#include "qemu-common.h"
+#include "qemu/rfifolock.h"
+
+static void test_nesting(void)
+{
+    RFifoLock lock;
+
+    /* Trivial test, ensure the lock is recursive */
+    rfifolock_init(&lock, NULL, NULL);
+    rfifolock_lock(&lock);
+    rfifolock_lock(&lock);
+    rfifolock_lock(&lock);
+    rfifolock_unlock(&lock);
+    rfifolock_unlock(&lock);
+    rfifolock_unlock(&lock);
+    rfifolock_destroy(&lock);
+}
+
+typedef struct {
+    RFifoLock lock;
+    int fd[2];
+} CallbackTestData;
+
+static void rfifolock_cb(void *opaque)
+{
+    CallbackTestData *data = opaque;
+    int ret;
+    char c = 0;
+
+    ret = write(data->fd[1], &c, sizeof(c));
+    g_assert(ret == 1);
+}
+
+static void *callback_thread(void *opaque)
+{
+    CallbackTestData *data = opaque;
+
+    /* The other thread holds the lock so the contention callback will be
+     * invoked...
+     */
+    rfifolock_lock(&data->lock);
+    rfifolock_unlock(&data->lock);
+    return NULL;
+}
+
+static void test_callback(void)
+{
+    CallbackTestData data;
+    QemuThread thread;
+    int ret;
+    char c;
+
+    rfifolock_init(&data.lock, rfifolock_cb, &data);
+    ret = qemu_pipe(data.fd);
+    g_assert(ret == 0);
+
+    /* Hold lock but allow the callback to kick us by writing to the pipe */
+    rfifolock_lock(&data.lock);
+    qemu_thread_create(&thread, callback_thread, &data, QEMU_THREAD_JOINABLE);
+    ret = read(data.fd[0], &c, sizeof(c));
+    g_assert(ret == 1);
+    rfifolock_unlock(&data.lock);
+    /* If we got here then the callback was invoked, as expected */
+
+    qemu_thread_join(&thread);
+    close(data.fd[0]);
+    close(data.fd[1]);
+    rfifolock_destroy(&data.lock);
+}
+
+int main(int argc, char **argv)
+{
+    g_test_init(&argc, &argv, NULL);
+    g_test_add_func("/nesting", test_nesting);
+    g_test_add_func("/callback", test_callback);
+    return g_test_run();
+}
diff --git a/util/Makefile.objs b/util/Makefile.objs
index 2bb13a2..9bc3515 100644
--- a/util/Makefile.objs
+++ b/util/Makefile.objs
@@ -12,3 +12,4 @@ util-obj-y += qemu-option.o qemu-progress.o
 util-obj-y += hexdump.o
 util-obj-y += crc32c.o
 util-obj-y += throttle.o
+util-obj-y += rfifolock.o
diff --git a/util/rfifolock.c b/util/rfifolock.c
new file mode 100644
index 0000000..13b08de
--- /dev/null
+++ b/util/rfifolock.c
@@ -0,0 +1,79 @@
+/*
+ * Recursive FIFO lock
+ *
+ * Copyright Red Hat, Inc. 2013
+ *
+ * Authors:
+ *  Stefan Hajnoczi   <stefanha@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU LGPL, version 2 or later.
+ * See the COPYING.LIB file in the top-level directory.
+ *
+ */
+
+#include <assert.h>
+#include "qemu/rfifolock.h"
+
+void rfifolock_init(RFifoLock *r, void (*cb)(void *), void *opaque)
+{
+    qemu_mutex_init(&r->lock);
+    r->head = 0;
+    r->tail = 0;
+    qemu_cond_init(&r->cond);
+    r->nesting = 0;
+    r->cb = cb;
+    r->cb_opaque = opaque;
+}
+
+void rfifolock_destroy(RFifoLock *r)
+{
+    qemu_cond_destroy(&r->cond);
+    qemu_mutex_destroy(&r->lock);
+}
+
+/*
+ * Theory of operation:
+ *
+ * In order to ensure FIFO ordering, implement a ticketlock.  Threads acquiring
+ * the lock enqueue themselves by incrementing the tail index.  When the lock
+ * is unlocked, the head is incremented and waiting threads are notified.
+ *
+ * Recursive locking does not take a ticket since the head is only incremented
+ * when the outermost recursive caller unlocks.
+ */
+void rfifolock_lock(RFifoLock *r)
+{
+    qemu_mutex_lock(&r->lock);
+
+    /* Take a ticket */
+    unsigned int ticket = r->tail++;
+
+    if (r->nesting > 0) {
+        if (qemu_thread_is_self(&r->owner_thread)) {
+            r->tail--; /* put ticket back, we're nesting */
+        } else {
+            while (ticket != r->head) {
+                /* Invoke optional contention callback */
+                if (r->cb) {
+                    r->cb(r->cb_opaque);
+                }
+                qemu_cond_wait(&r->cond, &r->lock);
+            }
+        }
+    }
+    qemu_thread_get_self(&r->owner_thread);
+    r->nesting++;
+    qemu_mutex_unlock(&r->lock);
+}
+
+void rfifolock_unlock(RFifoLock *r)
+{
+    qemu_mutex_lock(&r->lock);
+    assert(r->nesting > 0);
+    assert(qemu_thread_is_self(&r->owner_thread));
+    if (--r->nesting == 0) {
+        r->head++;
+        qemu_cond_broadcast(&r->cond);
+    }
+    qemu_mutex_unlock(&r->lock);
+}
-- 
1.8.3.1

Re: [Qemu-devel] [PATCH 1/2] rfifolock: add recursive FIFO lock

[Paolo Bonzini]

Il 09/10/2013 11:55, Stefan Hajnoczi ha scritto:
> +    /* Take a ticket */
> +    unsigned int ticket = r->tail++;
> +
> +    if (r->nesting > 0) {
> +        if (qemu_thread_is_self(&r->owner_thread)) {
> +            r->tail--; /* put ticket back, we're nesting */
> +        } else {

ticket is dead in the "nested" path, why not move it (and the increment)
directly after the else?

Otherwise, the code is very nice.  It's very interesting that no pthread
mutex is held in the rfifolock critical section, so that the thread that
holds the next ticket has a chance to be woken up.

Paolo

Re: [Qemu-devel] [PATCH 1/2] rfifolock: add recursive FIFO lock

[Stefan Hajnoczi]

On Wed, Oct 09, 2013 at 12:05:43PM +0200, Paolo Bonzini wrote:
> Il 09/10/2013 11:55, Stefan Hajnoczi ha scritto:
> > +    /* Take a ticket */
> > +    unsigned int ticket = r->tail++;
> > +
> > +    if (r->nesting > 0) {
> > +        if (qemu_thread_is_self(&r->owner_thread)) {
> > +            r->tail--; /* put ticket back, we're nesting */
> > +        } else {
> 
> ticket is dead in the "nested" path, why not move it (and the increment)
> directly after the else?

The increment cannot be moved because there are 3 cases:

1. Uncontended lock: r->tail++
2. Recursive lock: do not change tail
3. Wait for contended lock: r->tail++

Case #1 needs r->tail++ too.

I find this approach clearest:

if (r->nesting == 0) {
    /* Uncontended, just take a ticket */
    r->tail++;
} else if (qemu_thread_is_self(&r->owner_thread)) {
    /* Recursive lock, no need to take a ticket */
} else {
    /* Contended case, take a ticket and wait for it */
    unsigned int ticket = r->tail++;

    while (ticket != r->head) {
        ...
    }
}

There's also a shorter approach which collapses everything into a single
if statement, but I find the if expression a harder to understand:

if (r->nesting == 0 || !qemu_thread_is_self(&r->owner_thread)) {
    unsigned int ticket = r->tail++;
    while (ticket != r->head) {
        ...
    }
}

Any preferences?

Stefan

Re: [Qemu-devel] [PATCH 1/2] rfifolock: add recursive FIFO lock

[Paolo Bonzini]

Il 09/10/2013 13:26, Stefan Hajnoczi ha scritto:
> On Wed, Oct 09, 2013 at 12:05:43PM +0200, Paolo Bonzini wrote:
>> Il 09/10/2013 11:55, Stefan Hajnoczi ha scritto:
>>> +    /* Take a ticket */
>>> +    unsigned int ticket = r->tail++;
>>> +
>>> +    if (r->nesting > 0) {
>>> +        if (qemu_thread_is_self(&r->owner_thread)) {
>>> +            r->tail--; /* put ticket back, we're nesting */
>>> +        } else {
>>
>> ticket is dead in the "nested" path, why not move it (and the increment)
>> directly after the else?
> 
> The increment cannot be moved because there are 3 cases:
> 
> 1. Uncontended lock: r->tail++
> 2. Recursive lock: do not change tail
> 3. Wait for contended lock: r->tail++
> 
> Case #1 needs r->tail++ too.
> 
> I find this approach clearest:
> 
> if (r->nesting == 0) {
>     /* Uncontended, just take a ticket */
>     r->tail++;
> } else if (qemu_thread_is_self(&r->owner_thread)) {
>     /* Recursive lock, no need to take a ticket */
> } else {
>     /* Contended case, take a ticket and wait for it */
>     unsigned int ticket = r->tail++;
> 
>     while (ticket != r->head) {
>         ...
>     }
> }

I like this one too.

Paolo

Re: [Qemu-devel] [PATCH 1/2] rfifolock: add recursive FIFO lock

[Wenchao Xia]

Logic looks fine to me, only a minor comments.

> QemuMutex does not guarantee fairness and cannot be acquired
> recursively:
>
> Fairness means each locker gets a turn and the scheduler cannot cause
> starvation.
>
> Recursive locking is useful for composition, it allows a sequence of
> locking operations to be invoked atomically by acquiring the lock around
> them.
>
> This patch adds RFifoLock, a recursive lock that guarantees FIFO order.
> Its first user is added in the next patch.
>
> RFifoLock has one additional feature: it can be initialized with an
> optional contention callback.  The callback is invoked whenever a thread
> must wait for the lock.  For example, it can be used to poke the current
> owner so that they release the lock soon.
>
> Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
> ---
>  include/qemu/rfifolock.h | 54 +++++++++++++++++++++++++++++
>  tests/Makefile           |  2 ++
>  tests/test-rfifolock.c   | 90 ++++++++++++++++++++++++++++++++++++++++++++++++
>  util/Makefile.objs       |  1 +
>  util/rfifolock.c         | 79 ++++++++++++++++++++++++++++++++++++++++++
>  5 files changed, 226 insertions(+)
>  create mode 100644 include/qemu/rfifolock.h
>  create mode 100644 tests/test-rfifolock.c
>  create mode 100644 util/rfifolock.c
>
> diff --git a/include/qemu/rfifolock.h b/include/qemu/rfifolock.h
> new file mode 100644
> index 0000000..b23ab53
> --- /dev/null
> +++ b/include/qemu/rfifolock.h
> @@ -0,0 +1,54 @@
> +/*
> + * Recursive FIFO lock
> + *
> + * Copyright Red Hat, Inc. 2013
> + *
> + * Authors:
> + *  Stefan Hajnoczi   <stefanha@redhat.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or later.
> + * See the COPYING file in the top-level directory.
> + *
> + */
> +
> +#ifndef QEMU_RFIFOLOCK_H
> +#define QEMU_RFIFOLOCK_H
> +
> +#include "qemu/thread.h"
> +
> +/* Recursive FIFO lock
> + *
> + * This lock provides more features than a plain mutex:
> + *
> + * 1. Fairness - enforces FIFO order.
> + * 2. Nesting - can be taken recursively.
> + * 3. Contention callback - optional, called when thread must wait.
> + *
> + * The recursive FIFO lock is heavyweight so prefer other synchronization
> + * primitives if you do not need its features.
> + */
> +typedef struct {
> +    QemuMutex lock;             /* protects all fields */
> +
> +    /* FIFO order */
> +    unsigned int head;          /* active ticket number */
> +    unsigned int tail;          /* waiting ticket number */
> +    QemuCond cond;              /* used to wait for our ticket number */
> +
> +    /* Nesting */
> +    QemuThread owner_thread;    /* thread that currently has ownership */
> +    unsigned int nesting;       /* amount of nesting levels */
> +
> +    /* Contention callback */
> +    void (*cb)(void *);         /* called when thread must wait, with ->lock
> +                                 * held so it may not recursively lock/unlock
> +                                 */
> +    void *cb_opaque;
> +} RFifoLock;
> +
If you respin, the define can be moved to util/rfifolock.c, leave
typedef struct RFifoLock RFifoLock;
in header.

> +void rfifolock_init(RFifoLock *r, void (*cb)(void *), void *opaque);
> +void rfifolock_destroy(RFifoLock *r);
> +void rfifolock_lock(RFifoLock *r);
> +void rfifolock_unlock(RFifoLock *r);
> +
> +#endif /* QEMU_RFIFOLOCK_H */
> diff --git a/tests/Makefile b/tests/Makefile
> index 994fef1..2959ed1 100644
> --- a/tests/Makefile
> +++ b/tests/Makefile
> @@ -31,6 +31,7 @@ check-unit-y += tests/test-visitor-serialization$(EXESUF)
>  check-unit-y += tests/test-iov$(EXESUF)
>  gcov-files-test-iov-y = util/iov.c
>  check-unit-y += tests/test-aio$(EXESUF)
> +check-unit-y += tests/test-rfifolock$(EXESUF)
>  check-unit-y += tests/test-throttle$(EXESUF)
>  gcov-files-test-aio-$(CONFIG_WIN32) = aio-win32.c
>  gcov-files-test-aio-$(CONFIG_POSIX) = aio-posix.c
> @@ -121,6 +122,7 @@ tests/check-qfloat$(EXESUF): tests/check-qfloat.o libqemuutil.a
>  tests/check-qjson$(EXESUF): tests/check-qjson.o libqemuutil.a libqemustub.a
>  tests/test-coroutine$(EXESUF): tests/test-coroutine.o $(block-obj-y) libqemuutil.a libqemustub.a
>  tests/test-aio$(EXESUF): tests/test-aio.o $(block-obj-y) libqemuutil.a libqemustub.a
> +tests/test-rfifolock$(EXESUF): tests/test-rfifolock.o libqemuutil.a libqemustub.a
>  tests/test-throttle$(EXESUF): tests/test-throttle.o $(block-obj-y) libqemuutil.a libqemustub.a
>  tests/test-thread-pool$(EXESUF): tests/test-thread-pool.o $(block-obj-y) libqemuutil.a libqemustub.a
>  tests/test-iov$(EXESUF): tests/test-iov.o libqemuutil.a
> diff --git a/tests/test-rfifolock.c b/tests/test-rfifolock.c
> new file mode 100644
> index 0000000..440dbcb
> --- /dev/null
> +++ b/tests/test-rfifolock.c
> @@ -0,0 +1,90 @@
> +/*
> + * RFifoLock tests
> + *
> + * Copyright Red Hat, Inc. 2013
> + *
> + * Authors:
> + *  Stefan Hajnoczi    <stefanha@redhat.com>
> + *
> + * This work is licensed under the terms of the GNU LGPL, version 2 or later.
> + * See the COPYING.LIB file in the top-level directory.
> + */
> +
> +#include <glib.h>
> +#include "qemu-common.h"
> +#include "qemu/rfifolock.h"
> +
> +static void test_nesting(void)
> +{
> +    RFifoLock lock;
> +
> +    /* Trivial test, ensure the lock is recursive */
> +    rfifolock_init(&lock, NULL, NULL);
> +    rfifolock_lock(&lock);
> +    rfifolock_lock(&lock);
> +    rfifolock_lock(&lock);
> +    rfifolock_unlock(&lock);
> +    rfifolock_unlock(&lock);
> +    rfifolock_unlock(&lock);
> +    rfifolock_destroy(&lock);
> +}
> +
> +typedef struct {
> +    RFifoLock lock;
> +    int fd[2];
> +} CallbackTestData;
> +
> +static void rfifolock_cb(void *opaque)
> +{
> +    CallbackTestData *data = opaque;
> +    int ret;
> +    char c = 0;
> +
> +    ret = write(data->fd[1], &c, sizeof(c));
> +    g_assert(ret == 1);
> +}
> +
> +static void *callback_thread(void *opaque)
> +{
> +    CallbackTestData *data = opaque;
> +
> +    /* The other thread holds the lock so the contention callback will be
> +     * invoked...
> +     */
> +    rfifolock_lock(&data->lock);
> +    rfifolock_unlock(&data->lock);
> +    return NULL;
> +}
> +
> +static void test_callback(void)
> +{
> +    CallbackTestData data;
> +    QemuThread thread;
> +    int ret;
> +    char c;
> +
> +    rfifolock_init(&data.lock, rfifolock_cb, &data);
> +    ret = qemu_pipe(data.fd);
> +    g_assert(ret == 0);
> +
> +    /* Hold lock but allow the callback to kick us by writing to the pipe */
> +    rfifolock_lock(&data.lock);
> +    qemu_thread_create(&thread, callback_thread, &data, QEMU_THREAD_JOINABLE);
> +    ret = read(data.fd[0], &c, sizeof(c));
> +    g_assert(ret == 1);
> +    rfifolock_unlock(&data.lock);
> +    /* If we got here then the callback was invoked, as expected */
> +
> +    qemu_thread_join(&thread);
> +    close(data.fd[0]);
> +    close(data.fd[1]);
> +    rfifolock_destroy(&data.lock);
> +}
> +
> +int main(int argc, char **argv)
> +{
> +    g_test_init(&argc, &argv, NULL);
> +    g_test_add_func("/nesting", test_nesting);
> +    g_test_add_func("/callback", test_callback);
> +    return g_test_run();
> +}
> diff --git a/util/Makefile.objs b/util/Makefile.objs
> index 2bb13a2..9bc3515 100644
> --- a/util/Makefile.objs
> +++ b/util/Makefile.objs
> @@ -12,3 +12,4 @@ util-obj-y += qemu-option.o qemu-progress.o
>  util-obj-y += hexdump.o
>  util-obj-y += crc32c.o
>  util-obj-y += throttle.o
> +util-obj-y += rfifolock.o
> diff --git a/util/rfifolock.c b/util/rfifolock.c
> new file mode 100644
> index 0000000..13b08de
> --- /dev/null
> +++ b/util/rfifolock.c
> @@ -0,0 +1,79 @@
> +/*
> + * Recursive FIFO lock
> + *
> + * Copyright Red Hat, Inc. 2013
> + *
> + * Authors:
> + *  Stefan Hajnoczi   <stefanha@redhat.com>
> + *
> + * This work is licensed under the terms of the GNU LGPL, version 2 or later.
> + * See the COPYING.LIB file in the top-level directory.
> + *
> + */
> +
> +#include <assert.h>
> +#include "qemu/rfifolock.h"
> +
> +void rfifolock_init(RFifoLock *r, void (*cb)(void *), void *opaque)
> +{
> +    qemu_mutex_init(&r->lock);
> +    r->head = 0;
> +    r->tail = 0;
> +    qemu_cond_init(&r->cond);
> +    r->nesting = 0;
> +    r->cb = cb;
> +    r->cb_opaque = opaque;
> +}
> +
> +void rfifolock_destroy(RFifoLock *r)
> +{
> +    qemu_cond_destroy(&r->cond);
> +    qemu_mutex_destroy(&r->lock);
> +}
> +
> +/*
> + * Theory of operation:
> + *
> + * In order to ensure FIFO ordering, implement a ticketlock.  Threads acquiring
> + * the lock enqueue themselves by incrementing the tail index.  When the lock
> + * is unlocked, the head is incremented and waiting threads are notified.
> + *
> + * Recursive locking does not take a ticket since the head is only incremented
> + * when the outermost recursive caller unlocks.
> + */
> +void rfifolock_lock(RFifoLock *r)
> +{
> +    qemu_mutex_lock(&r->lock);
> +
> +    /* Take a ticket */
> +    unsigned int ticket = r->tail++;
> +
> +    if (r->nesting > 0) {
> +        if (qemu_thread_is_self(&r->owner_thread)) {
> +            r->tail--; /* put ticket back, we're nesting */
> +        } else {
> +            while (ticket != r->head) {
> +                /* Invoke optional contention callback */
> +                if (r->cb) {
> +                    r->cb(r->cb_opaque);
> +                }
> +                qemu_cond_wait(&r->cond, &r->lock);
> +            }
> +        }
> +    }
> +    qemu_thread_get_self(&r->owner_thread);
> +    r->nesting++;
> +    qemu_mutex_unlock(&r->lock);
> +}
> +
> +void rfifolock_unlock(RFifoLock *r)
> +{
> +    qemu_mutex_lock(&r->lock);
> +    assert(r->nesting > 0);
> +    assert(qemu_thread_is_self(&r->owner_thread));
> +    if (--r->nesting == 0) {
> +        r->head++;
> +        qemu_cond_broadcast(&r->cond);
> +    }
> +    qemu_mutex_unlock(&r->lock);
> +}

Re: [Qemu-devel] [PATCH 1/2] rfifolock: add recursive FIFO lock

[Stefan Hajnoczi]

On Fri, Oct 11, 2013 at 04:55:31PM +0800, Wenchao Xia wrote:
> > +/* Recursive FIFO lock
> > + *
> > + * This lock provides more features than a plain mutex:
> > + *
> > + * 1. Fairness - enforces FIFO order.
> > + * 2. Nesting - can be taken recursively.
> > + * 3. Contention callback - optional, called when thread must wait.
> > + *
> > + * The recursive FIFO lock is heavyweight so prefer other synchronization
> > + * primitives if you do not need its features.
> > + */
> > +typedef struct {
> > +    QemuMutex lock;             /* protects all fields */
> > +
> > +    /* FIFO order */
> > +    unsigned int head;          /* active ticket number */
> > +    unsigned int tail;          /* waiting ticket number */
> > +    QemuCond cond;              /* used to wait for our ticket number */
> > +
> > +    /* Nesting */
> > +    QemuThread owner_thread;    /* thread that currently has ownership */
> > +    unsigned int nesting;       /* amount of nesting levels */
> > +
> > +    /* Contention callback */
> > +    void (*cb)(void *);         /* called when thread must wait, with ->lock
> > +                                 * held so it may not recursively lock/unlock
> > +                                 */
> > +    void *cb_opaque;
> > +} RFifoLock;
> > +
> If you respin, the define can be moved to util/rfifolock.c, leave
> typedef struct RFifoLock RFifoLock;
> in header.

Then the struct cannot be embedded as a field, it would require heap
allocation of all RFifoLocks.

This is why I chose to include the definition in the header file.

Stefan

Re: [Qemu-devel] [PATCH 1/2] rfifolock: add recursive FIFO lock

[Wenchao Xia]

于 2013/10/11 21:35, Stefan Hajnoczi 写道:
> On Fri, Oct 11, 2013 at 04:55:31PM +0800, Wenchao Xia wrote:
>>> +/* Recursive FIFO lock
>>> + *
>>> + * This lock provides more features than a plain mutex:
>>> + *
>>> + * 1. Fairness - enforces FIFO order.
>>> + * 2. Nesting - can be taken recursively.
>>> + * 3. Contention callback - optional, called when thread must wait.
>>> + *
>>> + * The recursive FIFO lock is heavyweight so prefer other synchronization
>>> + * primitives if you do not need its features.
>>> + */
>>> +typedef struct {
>>> +    QemuMutex lock;             /* protects all fields */
>>> +
>>> +    /* FIFO order */
>>> +    unsigned int head;          /* active ticket number */
>>> +    unsigned int tail;          /* waiting ticket number */
>>> +    QemuCond cond;              /* used to wait for our ticket number */
>>> +
>>> +    /* Nesting */
>>> +    QemuThread owner_thread;    /* thread that currently has ownership */
>>> +    unsigned int nesting;       /* amount of nesting levels */
>>> +
>>> +    /* Contention callback */
>>> +    void (*cb)(void *);         /* called when thread must wait, with ->lock
>>> +                                 * held so it may not recursively lock/unlock
>>> +                                 */
>>> +    void *cb_opaque;
>>> +} RFifoLock;
>>> +
>> If you respin, the define can be moved to util/rfifolock.c, leave
>> typedef struct RFifoLock RFifoLock;
>> in header.
> Then the struct cannot be embedded as a field, it would require heap
> allocation of all RFifoLocks.
>
> This is why I chose to include the definition in the header file.
>
> Stefan
>
OK, it is not serious problem, the patch looks good to me.

[Qemu-devel] [PATCH 2/2] aio: add aio_context_acquire() and aio_context_release()

[Stefan Hajnoczi]

It can be useful to run an AioContext from a thread which normally does
not "own" the AioContext.  For example, request draining can be
implemented by acquiring the AioContext and looping aio_poll() until all
requests have been completed.

The following pattern should work:

  /* Event loop thread */
  while (running) {
      aio_context_acquire(ctx);
      aio_poll(ctx, true);
      aio_context_release(ctx);
  }

  /* Another thread */
  aio_context_acquire(ctx);
  bdrv_read(bs, 0x1000, buf, 1);
  aio_context_release(ctx);

This patch implements aio_context_acquire() and aio_context_release().

Note that existing aio_poll() callers do not need to worry about
acquiring and releasing - it is only needed when multiple threads will
call aio_poll() on the same AioContext.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 async.c             | 18 +++++++++++++++++
 include/block/aio.h | 18 +++++++++++++++++
 tests/test-aio.c    | 58 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 94 insertions(+)

diff --git a/async.c b/async.c
index 5fb3fa6..6930185 100644
--- a/async.c
+++ b/async.c
@@ -214,6 +214,7 @@ aio_ctx_finalize(GSource     *source)
     thread_pool_free(ctx->thread_pool);
     aio_set_event_notifier(ctx, &ctx->notifier, NULL);
     event_notifier_cleanup(&ctx->notifier);
+    rfifolock_destroy(&ctx->lock);
     qemu_mutex_destroy(&ctx->bh_lock);
     g_array_free(ctx->pollfds, TRUE);
     timerlistgroup_deinit(&ctx->tlg);
@@ -250,6 +251,12 @@ static void aio_timerlist_notify(void *opaque)
     aio_notify(opaque);
 }
 
+static void aio_rfifolock_cb(void *opaque)
+{
+    /* Kick owner thread in case they are blocked in aio_poll() */
+    aio_notify(opaque);
+}
+
 AioContext *aio_context_new(void)
 {
     AioContext *ctx;
@@ -257,6 +264,7 @@ AioContext *aio_context_new(void)
     ctx->pollfds = g_array_new(FALSE, FALSE, sizeof(GPollFD));
     ctx->thread_pool = NULL;
     qemu_mutex_init(&ctx->bh_lock);
+    rfifolock_init(&ctx->lock, aio_rfifolock_cb, ctx);
     event_notifier_init(&ctx->notifier, false);
     aio_set_event_notifier(ctx, &ctx->notifier, 
                            (EventNotifierHandler *)
@@ -275,3 +283,13 @@ void aio_context_unref(AioContext *ctx)
 {
     g_source_unref(&ctx->source);
 }
+
+void aio_context_acquire(AioContext *ctx)
+{
+    rfifolock_lock(&ctx->lock);
+}
+
+void aio_context_release(AioContext *ctx)
+{
+    rfifolock_unlock(&ctx->lock);
+}
diff --git a/include/block/aio.h b/include/block/aio.h
index 2efdf41..4aaa5d5 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -19,6 +19,7 @@
 #include "qemu/queue.h"
 #include "qemu/event_notifier.h"
 #include "qemu/thread.h"
+#include "qemu/rfifolock.h"
 #include "qemu/timer.h"
 
 typedef struct BlockDriverAIOCB BlockDriverAIOCB;
@@ -47,6 +48,9 @@ typedef void IOHandler(void *opaque);
 struct AioContext {
     GSource source;
 
+    /* Protects all fields from multi-threaded access */
+    RFifoLock lock;
+
     /* The list of registered AIO handlers */
     QLIST_HEAD(, AioHandler) aio_handlers;
 
@@ -104,6 +108,20 @@ void aio_context_ref(AioContext *ctx);
  */
 void aio_context_unref(AioContext *ctx);
 
+/* Take ownership of the AioContext.  If the AioContext will be shared between
+ * threads, a thread must have ownership when calling aio_poll().
+ *
+ * Note that multiple threads calling aio_poll() means timers, BHs, and
+ * callbacks may be invoked from a different thread than they were registered
+ * from.  Therefore, code must use AioContext acquire/release or use
+ * fine-grained synchronization to protect shared state if other threads will
+ * be accessing it simultaneously.
+ */
+void aio_context_acquire(AioContext *ctx);
+
+/* Reliquinish ownership of the AioContext. */
+void aio_context_release(AioContext *ctx);
+
 /**
  * aio_bh_new: Allocate a new bottom half structure.
  *
diff --git a/tests/test-aio.c b/tests/test-aio.c
index c4fe0fc..c971c07 100644
--- a/tests/test-aio.c
+++ b/tests/test-aio.c
@@ -112,6 +112,63 @@ static void test_notify(void)
     g_assert(!aio_poll(ctx, false));
 }
 
+typedef struct {
+    QemuMutex start_lock;
+    bool thread_acquired;
+} AcquireTestData;
+
+static void *test_acquire_thread(void *opaque)
+{
+    AcquireTestData *data = opaque;
+
+    /* Wait for other thread to let us start */
+    qemu_mutex_lock(&data->start_lock);
+    qemu_mutex_unlock(&data->start_lock);
+
+    aio_context_acquire(ctx);
+    aio_context_release(ctx);
+
+    data->thread_acquired = true; /* success, we got here */
+
+    return NULL;
+}
+
+static void dummy_notifier_read(EventNotifier *unused)
+{
+    g_assert(false); /* should never be invoked */
+}
+
+static void test_acquire(void)
+{
+    QemuThread thread;
+    EventNotifier notifier;
+    AcquireTestData data;
+
+    /* Dummy event notifier ensures aio_poll() will block */
+    event_notifier_init(&notifier, false);
+    aio_set_event_notifier(ctx, &notifier, dummy_notifier_read);
+    g_assert(!aio_poll(ctx, false)); /* consume aio_notify() */
+
+    qemu_mutex_init(&data.start_lock);
+    qemu_mutex_lock(&data.start_lock);
+    data.thread_acquired = false;
+
+    qemu_thread_create(&thread, test_acquire_thread,
+                       &data, QEMU_THREAD_JOINABLE);
+
+    /* Block in aio_poll(), let other thread kick us and acquire context */
+    aio_context_acquire(ctx);
+    qemu_mutex_unlock(&data.start_lock); /* let the thread run */
+    g_assert(!aio_poll(ctx, true));
+    aio_context_release(ctx);
+
+    qemu_thread_join(&thread);
+    aio_set_event_notifier(ctx, &notifier, NULL);
+    event_notifier_cleanup(&notifier);
+
+    g_assert(data.thread_acquired);
+}
+
 static void test_bh_schedule(void)
 {
     BHTestData data = { .n = 0 };
@@ -776,6 +833,7 @@ int main(int argc, char **argv)
 
     g_test_init(&argc, &argv, NULL);
     g_test_add_func("/aio/notify",                  test_notify);
+    g_test_add_func("/aio/acquire",                 test_acquire);
     g_test_add_func("/aio/bh/schedule",             test_bh_schedule);
     g_test_add_func("/aio/bh/schedule10",           test_bh_schedule10);
     g_test_add_func("/aio/bh/cancel",               test_bh_cancel);
-- 
1.8.3.1

Re: [Qemu-devel] [PATCH 2/2] aio: add aio_context_acquire() and aio_context_release()

[Wenchao Xia]

Reviewed-by: Wenchao Xia <xiawenc@linux.vnet.ibm.com>

Re: [Qemu-devel] [PATCH 0/2] aio: add aio_context_acquire() and aio_context_release()

[Paolo Bonzini]

Il 09/10/2013 11:55, Stefan Hajnoczi ha scritto:
> aio_context_acquire() and aio_context_release() make it possible for multiple
> threads to safely operate on a shared AioContext.  This is a prerequisite for
> using the block layer outside the QEMU global mutex.
> 
> Imagine that a dataplane thread is performing I/O on behalf of the guest when
> the user issues a monitor command that needs to access the BlockDriverState.
> The monitor thread needs to acquire the AioContext before calling bdrv_*()
> functions on it.  This prevents the dataplane thread from interfering with the
> monitor thread.
> 
> There was a discussion in the RFC email thread about how to implement fairness:
> each thread should get a turn to acquire the AioContext so that starvation is
> avoided.
> 
> Paolo suggested doing what the QEMU main loop does today: drop the lock before
> invoking ppoll(2).  It turns out that this is tricky since we want both threads
> to be able to call aio_poll() simultaneously.  AioContext->pollfds[] currently
> prevents two simultaneous aio_poll() calls since it is a shared resource.  It's
> also tricky to avoid deadlocks if two threads execute aio_poll() simulatenously
> except by waking all threads each time *any* thread makes any progress.
> 
> I found the simplest solution is to implement RFifoLock, a recursive lock with
> FIFO ordering.  This lock supports the semantics we need for the following
> pattern:
> 
>   /* Event loop thread */
>   while (running) {
>       aio_context_acquire(ctx);
>       aio_poll(ctx, true);
>       aio_context_release(ctx);
>   }
> 
>   /* Another thread */
>   aio_context_acquire(ctx);
>   bdrv_read(bs, 0x1000, buf, 1);
>   aio_context_release(ctx);

Should bdrv_read itself call aio_context_acquire?  If not, what's the
use of recursion?

> When the other thread wants to acquire the AioContext but the lock is
> contended, it uses aio_notify(ctx) to bump the event loop thread out of
> aio_poll().  Fairness ensures everyone gets an opportunity to use the
> AioContext.

The implementation of fairness is very nice.  The callbacks are an
interesting solution, too.  They ensure that the lock is released fast,
but at the same time some progress is allowed.

I don't like recursive mutexes very much, because it's easier to get
AB-BA deadlocks with them.  However, it's not easy to avoid them except
by adding a lot of acquire/release pairs.  And there's also a precedent
(the mmap_lock used by user-level emulation).

We can avoid the problem for now by heavily limiting the coarse-grained
mutexes we have---basically only the iothread mutex and this new lock.
We should aim at making locks fine-grained enough that we hardly have
any nesting, or use RCU so that the read-side is independent and the
write-side can just keep using the iothread mutex.

Still, in the long run it would be nice to push acquire/release pairs up
to all the users and avoid recursion...

Acked-by: Paolo Bonzini <pbonzini@redhat.com>

Paolo

Re: [Qemu-devel] [PATCH 0/2] aio: add aio_context_acquire() and aio_context_release()

[Stefan Hajnoczi]

On Wed, Oct 09, 2013 at 12:16:47PM +0200, Paolo Bonzini wrote:
> Il 09/10/2013 11:55, Stefan Hajnoczi ha scritto:
> > aio_context_acquire() and aio_context_release() make it possible for multiple
> > threads to safely operate on a shared AioContext.  This is a prerequisite for
> > using the block layer outside the QEMU global mutex.
> > 
> > Imagine that a dataplane thread is performing I/O on behalf of the guest when
> > the user issues a monitor command that needs to access the BlockDriverState.
> > The monitor thread needs to acquire the AioContext before calling bdrv_*()
> > functions on it.  This prevents the dataplane thread from interfering with the
> > monitor thread.
> > 
> > There was a discussion in the RFC email thread about how to implement fairness:
> > each thread should get a turn to acquire the AioContext so that starvation is
> > avoided.
> > 
> > Paolo suggested doing what the QEMU main loop does today: drop the lock before
> > invoking ppoll(2).  It turns out that this is tricky since we want both threads
> > to be able to call aio_poll() simultaneously.  AioContext->pollfds[] currently
> > prevents two simultaneous aio_poll() calls since it is a shared resource.  It's
> > also tricky to avoid deadlocks if two threads execute aio_poll() simulatenously
> > except by waking all threads each time *any* thread makes any progress.
> > 
> > I found the simplest solution is to implement RFifoLock, a recursive lock with
> > FIFO ordering.  This lock supports the semantics we need for the following
> > pattern:
> > 
> >   /* Event loop thread */
> >   while (running) {
> >       aio_context_acquire(ctx);
> >       aio_poll(ctx, true);
> >       aio_context_release(ctx);
> >   }
> > 
> >   /* Another thread */
> >   aio_context_acquire(ctx);
> >   bdrv_read(bs, 0x1000, buf, 1);
> >   aio_context_release(ctx);
> 
> Should bdrv_read itself call aio_context_acquire?  If not, what's the
> use of recursion?

Recursion is an artifact of our previous discussions.  In a few days I
should have explored enough use cases to know whether we need it or not
- I'm converting code to use acquire/release right now.

Will let you know once I've figured out if we really need it or not.

Stefan