[Qemu-devel] [PATCH 10/10] aio: change aio_poll constraints

["Denis V. Lunev" <den@openvz.org>]

There are 2 versions of the aio_poll: blocking and non-blocking.

Non-blocking version is called at the moment from 3 places:
    - iothread_run
    - bdrv_drain
    - bdrv_drain_all
iothread_run and bdrv_drain_all properly acquires AioContext by their own.
bdrv_drain (according to the description) MUST be called with pre-acquired
context. This is perfect.

Blocking version of aio_poll is called mostly using the following syntax:
    AioContext *aio_context = bdrv_get_aio_context(bs);
    co = qemu_coroutine_create(bdrv_rw_co_entry);
    qemu_coroutine_enter(co, &rwco);
    while (rwco.ret == NOT_DONE) {
        aio_poll(aio_context, true);
    }
There is no problem if this is called from iothread, when AioContext is
properly acquired. Unfortunately, this code is called from HMP thread
and this leads to a disaster.

        HMP thread                     IO thread (in aio_poll)
            |                                    |
    qemu_coroutine_enter                         |
    while (rwco.ret == NOT_DONE)                 |
        aio_poll                                 |
            aio_context_acquire                  |
            |                         ret from qemu_poll_ns
            |                         aio_context_acquire (nested = 2)
            |                         process bdrv_rw_co_entry, set rwco.ret
            |                       aio_context_release (nested = )
            |                       reenters aio_poll, clear events
            |                       aio_context_release
            aio_context_release
            qemu_poll_ns

In this case HMP thread will be never waked up. Alas.

This means that all such patterns MUST be guarded with aio_context_is_owner
checks, but this is terrible as if we'll find all such places we can fix
them with ease.

This patch proposes different solution: aio_poll MUST be called with
AioContext acquired. Non-blocking places are perfectly OK already, blocking
places MUST be guarded anyway to avoid above deadlock.

Another approach would be to take the lock at the very top (at the beginning
of the operation) but this is much more difficult and leads to spreading
of aio_context_acquire to a lot of unrelated pieces.

Signed-off-by: Denis V. Lunev <den@openvz.org>
CC: Stefan Hajnoczi <stefanha@redhat.com>
CC: Kevin Wolf <kwolf@redhat.com>
---
 aio-posix.c              | 11 +----------
 aio-win32.c              |  9 +--------
 include/block/aio.h      |  2 ++
 tests/test-aio.c         | 11 +++++++++++
 tests/test-thread-pool.c | 15 +++++++++++++++
 5 files changed, 30 insertions(+), 18 deletions(-)

diff --git a/aio-posix.c b/aio-posix.c
index 0467f23..735d272 100644
--- a/aio-posix.c
+++ b/aio-posix.c
@@ -241,7 +241,7 @@ bool aio_poll(AioContext *ctx, bool blocking)
     bool progress;
     int64_t timeout;
 
-    aio_context_acquire(ctx);
+    assert(aio_context_is_owner(ctx));
     progress = false;
 
     /* aio_notify can avoid the expensive event_notifier_set if
@@ -269,17 +269,10 @@ bool aio_poll(AioContext *ctx, bool blocking)
 
     timeout = blocking ? aio_compute_timeout(ctx) : 0;
 
-    /* wait until next event */
-    if (timeout) {
-        aio_context_release(ctx);
-    }
     ret = qemu_poll_ns((GPollFD *)pollfds, npfd, timeout);
     if (blocking) {
         atomic_sub(&ctx->notify_me, 2);
     }
-    if (timeout) {
-        aio_context_acquire(ctx);
-    }
 
     aio_notify_accept(ctx);
 
@@ -298,7 +291,5 @@ bool aio_poll(AioContext *ctx, bool blocking)
         progress = true;
     }
 
-    aio_context_release(ctx);
-
     return progress;
 }
diff --git a/aio-win32.c b/aio-win32.c
index 43c4c79..ce45b98 100644
--- a/aio-win32.c
+++ b/aio-win32.c
@@ -288,7 +288,7 @@ bool aio_poll(AioContext *ctx, bool blocking)
     int count;
     int timeout;
 
-    aio_context_acquire(ctx);
+    assert(aio_context_is_owner(ctx));
     progress = false;
 
     /* aio_notify can avoid the expensive event_notifier_set if
@@ -331,17 +331,11 @@ bool aio_poll(AioContext *ctx, bool blocking)
 
         timeout = blocking && !have_select_revents
             ? qemu_timeout_ns_to_ms(aio_compute_timeout(ctx)) : 0;
-        if (timeout) {
-            aio_context_release(ctx);
-        }
         ret = WaitForMultipleObjects(count, events, FALSE, timeout);
         if (blocking) {
             assert(first);
             atomic_sub(&ctx->notify_me, 2);
         }
-        if (timeout) {
-            aio_context_acquire(ctx);
-        }
 
         if (first) {
             aio_notify_accept(ctx);
@@ -366,6 +360,5 @@ bool aio_poll(AioContext *ctx, bool blocking)
 
     progress |= timerlistgroup_run_timers(&ctx->tlg);
 
-    aio_context_release(ctx);
     return progress;
 }
diff --git a/include/block/aio.h b/include/block/aio.h
index d8cd41a..c8dc7ea 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -292,6 +292,8 @@ bool aio_dispatch(AioContext *ctx);
  * blocking.  If @blocking is true, this function will wait until one
  * or more AIO events have completed, to ensure something has moved
  * before returning.
+ *
+ * The caller must have AioContext owned.
  */
 bool aio_poll(AioContext *ctx, bool blocking);
 
diff --git a/tests/test-aio.c b/tests/test-aio.c
index 1623803..87b2dfd 100644
--- a/tests/test-aio.c
+++ b/tests/test-aio.c
@@ -16,6 +16,17 @@
 #include "qemu/sockets.h"
 #include "qemu/error-report.h"
 
+static int aio_poll_debug(AioContext *ctx, bool blocking)
+{
+    int ret;
+    aio_context_acquire(ctx);
+    ret = aio_poll(ctx, blocking);
+    aio_context_release(ctx);
+
+    return ret;
+}
+#define aio_poll(ctx, blocking) aio_poll_debug(ctx, blocking)
+
 static AioContext *ctx;
 
 typedef struct {
diff --git a/tests/test-thread-pool.c b/tests/test-thread-pool.c
index 6a0b981..3180335 100644
--- a/tests/test-thread-pool.c
+++ b/tests/test-thread-pool.c
@@ -6,6 +6,17 @@
 #include "qemu/timer.h"
 #include "qemu/error-report.h"
 
+static int aio_poll_debug(AioContext *ctx, bool blocking)
+{
+    int ret;
+    aio_context_acquire(ctx);
+    ret = aio_poll(ctx, blocking);
+    aio_context_release(ctx);
+
+    return ret;
+}
+#define aio_poll(ctx, blocking) aio_poll_debug(ctx, blocking)
+
 static AioContext *ctx;
 static ThreadPool *pool;
 static int active;
@@ -172,7 +183,9 @@ static void do_test_cancel(bool sync)
         if (atomic_cmpxchg(&data[i].n, 0, 3) == 0) {
             data[i].ret = -ECANCELED;
             if (sync) {
+                aio_context_acquire(ctx);
                 bdrv_aio_cancel(data[i].aiocb);
+                aio_context_release(ctx);
             } else {
                 bdrv_aio_cancel_async(data[i].aiocb);
             }
@@ -186,7 +199,9 @@ static void do_test_cancel(bool sync)
         if (data[i].aiocb && data[i].n != 3) {
             if (sync) {
                 /* Canceling the others will be a blocking operation.  */
+                aio_context_acquire(ctx);
                 bdrv_aio_cancel(data[i].aiocb);
+                aio_context_release(ctx);
             } else {
                 bdrv_aio_cancel_async(data[i].aiocb);
             }
-- 
2.5.0