Portable thread-pool/AIO, Win32 emulated AIO

This takes the eventfd emulation code from the main loop and adds it to EventNotifier. When the EventNotifier is used for the main loop too, we need this compatibility code.

Without CONFIG_EVENTFD, event_notifier_get_fd is only usable for the "read" side of the notifier, for example to set a select() handler.

Signed-off-by Paolo Bonzini <[email protected]>

---
 event_notifier.c |   83 +++++++++++++++++++++++++++++++++++++++++++-----------
 event_notifier.h |    3 +-
 2 files changed, 69 insertions(+), 17 deletions(-)

event_notifier.c +5 -5

index 2c207e1..dde2d32 100644
@@ -20,48 +20,99 @@
 
 void event_notifier_init_fd(EventNotifier *e, int fd)
 {
-    e->fd = fd;
+    e->rfd = fd;
+    e->wfd = fd;
 }
 
 int event_notifier_init(EventNotifier *e, int active)
 {
+    int fds[2];
+    int ret;
+
 #ifdef CONFIG_EVENTFD
-    int fd = eventfd(!!active, EFD_NONBLOCK | EFD_CLOEXEC);
-    if (fd < 0)
-        return -errno;
-    e->fd = fd;

event_notifier.c +10 -0

+    } else {
+        if (errno != ENOSYS) {
+            return -errno;
+        }
+        if (qemu_pipe(fds) < 0) {
+            return -errno;
+        }
+        ret = fcntl_setfl(fds[0], O_NONBLOCK);
+        if (ret < 0) {
+            goto fail;

event_notifier.c +36 -9

+    return 0;
+
+fail:
+    close(fds[0]);
+    close(fds[1]);
+    return ret;
 }
 
 void event_notifier_cleanup(EventNotifier *e)
 {
-    close(e->fd);
+    if (e->rfd != e->wfd) {
+        close(e->rfd);
+    }
+    close(e->wfd);
 }
 
 int event_notifier_get_fd(EventNotifier *e)
 {
-    return e->fd;
+    return e->rfd;
 }
 
 int event_notifier_set_handler(EventNotifier *e,
                                EventNotifierHandler *handler)
 {
-    return qemu_set_fd_handler(e->fd, (IOHandler *)handler, NULL, e);
+    return qemu_set_fd_handler(e->rfd, (IOHandler *)handler, NULL, e);
 }
 
 int event_notifier_set(EventNotifier *e)
 {
-    uint64_t value = 1;
-    int r = write(e->fd, &value, sizeof(value));
-    return r == sizeof(value);
+    static const uint64_t value = 1;
+    ssize_t ret;
+
+    do {
+        ret = write(e->wfd, &value, sizeof(value));
+    } while (ret < 0 && errno == EINTR);
+
+    /* EAGAIN is fine, a read must be pending.  */
+    if (ret < 0 && errno != EAGAIN) {
+        return -1;
+    }
+    return 0;
 }
 
 int event_notifier_test_and_clear(EventNotifier *e)
 {
-    uint64_t value;
-    int r = read(e->fd, &value, sizeof(value));
-    return r == sizeof(value);
+    int value;
+    ssize_t len;
+    char buffer[512];
+
+    /* Drain the notify pipe.  For eventfd, only 8 bytes will be read.  */
+    value = 0;
+    do {
+        len = read(e->rfd, buffer, sizeof(buffer));
+        value |= (len > 0);
+    } while ((len == -1 && errno == EINTR) || len == sizeof(buffer));
+
+    return value;
 }

event_notifier.h +2 -1

index f0ec2f2..f04d12d 100644
@@ -16,7 +16,8 @@
 #include "qemu-common.h"
 
 struct EventNotifier {
-    int fd;
+    int rfd;
+    int wfd;
 };
 
 typedef void EventNotifierHandler(EventNotifier *);

Signature · 1 line

1.7.10.4

Signed-off-by Paolo Bonzini <[email protected]>

---
 Makefile.objs |    4 +--
 main-loop.c   |  106 ++++++++-------------------------------------------------
 oslib-posix.c |   31 -----------------
 qemu-common.h |    1 -
 4 files changed, 17 insertions(+), 125 deletions(-)

Makefile.objs +2 -2

index ecdfaf9..6ed1981 100644
@@ -20,8 +20,8 @@ universal-obj-y += $(qom-obj-y)
 #######################################################################
 # oslib-obj-y is code depending on the OS (win32 vs posix)
 oslib-obj-y = osdep.o
-oslib-obj-$(CONFIG_WIN32) += oslib-win32.o qemu-thread-win32.o
-oslib-obj-$(CONFIG_POSIX) += oslib-posix.o qemu-thread-posix.o
+oslib-obj-$(CONFIG_WIN32) += oslib-win32.o qemu-thread-win32.o event_notifier-win32.o
+oslib-obj-$(CONFIG_POSIX) += oslib-posix.o qemu-thread-posix.o event_notifier-posix.o
 
 #######################################################################
 # coroutines

main-loop.c +15 -91

index eb3b6e6..81f49b3 100644
@@ -26,75 +26,12 @@
 #include "qemu-timer.h"
 #include "slirp/slirp.h"
 #include "main-loop.h"
+#include "event_notifier.h"
 
 #ifndef _WIN32
 
 #include "compatfd.h"
 
-static int io_thread_fd = -1;
-
-void qemu_notify_event(void)
-{
-    /* Write 8 bytes to be compatible with eventfd.  */
-    static const uint64_t val = 1;
-    ssize_t ret;
-
-    if (io_thread_fd == -1) {
-        return;
-    }
-    do {
-        ret = write(io_thread_fd, &val, sizeof(val));
-    } while (ret < 0 && errno == EINTR);
-
-    /* EAGAIN is fine, a read must be pending.  */
-    if (ret < 0 && errno != EAGAIN) {
-        fprintf(stderr, "qemu_notify_event: write() failed: %s\n",
-                strerror(errno));
-        exit(1);
-    }
-}
-
-static void qemu_event_read(void *opaque)
-{
-    int fd = (intptr_t)opaque;
-    ssize_t len;
-    char buffer[512];
-
-    /* Drain the notify pipe.  For eventfd, only 8 bytes will be read.  */
-    do {
-        len = read(fd, buffer, sizeof(buffer));
-    } while ((len == -1 && errno == EINTR) || len == sizeof(buffer));
-}
-
-static int qemu_event_init(void)
-{
-    int err;
-    int fds[2];
-
-    err = qemu_eventfd(fds);
-    if (err == -1) {
-        return -errno;
-    }
-    err = fcntl_setfl(fds[0], O_NONBLOCK);
-    if (err < 0) {
-        goto fail;
-    }
-    err = fcntl_setfl(fds[1], O_NONBLOCK);
-    if (err < 0) {
-        goto fail;
-    }
-    qemu_set_fd_handler2(fds[0], NULL, qemu_event_read, NULL,
-                         (void *)(intptr_t)fds[0]);
-
-    io_thread_fd = fds[1];
-    return 0;
-
-fail:
-    close(fds[0]);
-    close(fds[1]);
-    return err;
-}
-
 /* If we have signalfd, we mask out the signals we want to handle and then
  * use signalfd to listen for them.  We rely on whatever the current signal
  * handler is to dispatch the signals when we receive them.

@@ -164,40 +101,22 @@ static int qemu_signal_init(void)
 
 #else /* _WIN32 */
 
-static HANDLE qemu_event_handle = NULL;
-
-static void dummy_event_handler(void *opaque)
-{
-}
-
-static int qemu_event_init(void)
+static int qemu_signal_init(void)
 {
-    qemu_event_handle = CreateEvent(NULL, FALSE, FALSE, NULL);
-    if (!qemu_event_handle) {
-        fprintf(stderr, "Failed CreateEvent: %ld\n", GetLastError());
-        return -1;
-    }
-    qemu_add_wait_object(qemu_event_handle, dummy_event_handler, NULL);
     return 0;
 }
+#endif
+
+static EventNotifier io_thread_notifier;
+static int io_thread_initialized;
 
 void qemu_notify_event(void)
 {
-    if (!qemu_event_handle) {
+    if (!io_thread_initialized) {
         return;
     }
-    if (!SetEvent(qemu_event_handle)) {
-        fprintf(stderr, "qemu_notify_event: SetEvent failed: %ld\n",
-                GetLastError());
-        exit(1);
-    }
-}
-
-static int qemu_signal_init(void)
-{
-    return 0;
+    event_notifier_set(&io_thread_notifier);
 }
-#endif
 
 int main_loop_init(void)
 {

@@ -210,11 +129,15 @@ int main_loop_init(void)
     }
 
     /* Note eventfd must be drained before signalfd handlers run */
-    ret = qemu_event_init();
+    ret = event_notifier_init(&io_thread_notifier, 0);
     if (ret) {
         return ret;
     }
 
+    io_thread_initialized = true;
+    event_notifier_set_handler(&io_thread_notifier,
+                               (EventNotifierHandler *)
+                               event_notifier_test_and_clear);
     return 0;
 }
 

@@ -400,7 +323,8 @@ void qemu_del_wait_object(HANDLE handle, WaitObjectFunc *func, void *opaque)
 
 void qemu_fd_register(int fd)
 {
-    WSAEventSelect(fd, qemu_event_handle, FD_READ | FD_ACCEPT | FD_CLOSE |
+    WSAEventSelect(fd, event_notifier_get_handle(&io_thread_notifier),
+                   FD_READ | FD_ACCEPT | FD_CLOSE |
                    FD_CONNECT | FD_WRITE | FD_OOB);
 }
 

oslib-posix.c +0 -31

index 6b7ba64..2c6b044 100644
@@ -58,9 +58,6 @@ static int running_on_valgrind = -1;
 #ifdef CONFIG_LINUX
 #include <sys/syscall.h>
 #endif
-#ifdef CONFIG_EVENTFD
-#include <sys/eventfd.h>
-#endif
 
 int qemu_get_thread_id(void)
 {

@@ -180,34 +177,6 @@ int qemu_pipe(int pipefd[2])
     return ret;
 }
 
-/*
- * Creates an eventfd that looks like a pipe and has EFD_CLOEXEC set.
- */
-int qemu_eventfd(int fds[2])
-{
-#ifdef CONFIG_EVENTFD
-    int ret;
-
-    ret = eventfd(0, 0);
-    if (ret >= 0) {
-        fds[0] = ret;
-        fds[1] = dup(ret);
-        if (fds[1] == -1) {
-            close(ret);
-            return -1;
-        }
-        qemu_set_cloexec(ret);
-        qemu_set_cloexec(fds[1]);
-        return 0;
-    }
-    if (errno != ENOSYS) {
-        return -1;
-    }
-#endif
-
-    return qemu_pipe(fds);
-}
-
 int qemu_utimens(const char *path, const struct timespec *times)
 {
     struct timeval tv[2], tv_now;

qemu-common.h +0 -1

index 9d9e603..036d254 100644
@@ -195,7 +195,6 @@ ssize_t qemu_recv_full(int fd, void *buf, size_t count, int flags)
     QEMU_WARN_UNUSED_RESULT;
 
 #ifndef _WIN32
-int qemu_eventfd(int pipefd[2]);
 int qemu_pipe(int pipefd[2]);
 #endif
 

Signature · 1 line

1.7.10.4

Since linux-aio already uses an eventfd, converting it to use the EventNotifier-based API simplifies the code even though it is not meant to be portable.

Signed-off-by Paolo Bonzini <[email protected]>

---
 linux-aio.c |   49 +++++++++++++++++++------------------------------
 1 file changed, 19 insertions(+), 30 deletions(-)

linux-aio.c +19 -30

index fa0fbf3..779f793 100644
@@ -10,8 +10,8 @@
 #include "qemu-common.h"
 #include "qemu-aio.h"
 #include "block/raw-posix-aio.h"
+#include "event_notifier.h"
 
-#include <sys/eventfd.h>
 #include <libaio.h>
 
 /*

@@ -37,7 +37,7 @@ struct qemu_laiocb {
 
 struct qemu_laio_state {
     io_context_t ctx;
-    int efd;
+    EventNotifier e;
     int count;
 };
 

@@ -76,29 +76,17 @@ static void qemu_laio_process_completion(struct qemu_laio_state *s,
     qemu_aio_release(laiocb);
 }
 
-static void qemu_laio_completion_cb(void *opaque)
+static void qemu_laio_completion_cb(EventNotifier *e)
 {
-    struct qemu_laio_state *s = opaque;
+    struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e);
 
-    while (1) {
+    while (event_notifier_test_and_clear(&s->e)) {
         struct io_event events[MAX_EVENTS];
-        uint64_t val;
-        ssize_t ret;
         struct timespec ts = { 0 };
         int nevents, i;
 
         do {
-            ret = read(s->efd, &val, sizeof(val));
-        } while (ret == -1 && errno == EINTR);
-
-        if (ret == -1 && errno == EAGAIN)
-            break;
-
-        if (ret != 8)
-            break;
-
-        do {
-            nevents = io_getevents(s->ctx, val, MAX_EVENTS, events, &ts);
+            nevents = io_getevents(s->ctx, MAX_EVENTS, MAX_EVENTS, events, &ts);
         } while (nevents == -EINTR);
 
         for (i = 0; i < nevents; i++) {

@@ -112,9 +100,9 @@ static void qemu_laio_completion_cb(void *opaque)
     }
 }
 
-static int qemu_laio_flush_cb(void *opaque)
+static int qemu_laio_flush_cb(EventNotifier *e)
 {
-    struct qemu_laio_state *s = opaque;
+    struct qemu_laio_state *s = container_of(e, struct qemu_laio_state, e);
 
     return (s->count > 0) ? 1 : 0;
 }

@@ -146,8 +134,9 @@ static void laio_cancel(BlockDriverAIOCB *blockacb)
      * We might be able to do this slightly more optimal by removing the
      * O_NONBLOCK flag.
      */
-    while (laiocb->ret == -EINPROGRESS)
-        qemu_laio_completion_cb(laiocb->ctx);
+    while (laiocb->ret == -EINPROGRESS) {
+        qemu_laio_completion_cb(&laiocb->ctx->e);
+    }
 }
 
 static AIOPool laio_pool = {

@@ -186,7 +175,7 @@ BlockDriverAIOCB *laio_submit(BlockDriverState *bs, void *aio_ctx, int fd,
                         __func__, type);
         goto out_free_aiocb;
     }
-    io_set_eventfd(&laiocb->iocb, s->efd);
+    io_set_eventfd(&laiocb->iocb, event_notifier_get_fd(&s->e));
     s->count++;
 
     if (io_submit(s->ctx, 1, &iocbs) < 0)

@@ -205,21 +194,21 @@ void *laio_init(void)
     struct qemu_laio_state *s;
 
     s = g_malloc0(sizeof(*s));
-    s->efd = eventfd(0, 0);
-    if (s->efd == -1)
+    if (event_notifier_init(&s->e, false) < 0) {
         goto out_free_state;
-    fcntl(s->efd, F_SETFL, O_NONBLOCK);
+    }
 
-    if (io_setup(MAX_EVENTS, &s->ctx) != 0)
+    if (io_setup(MAX_EVENTS, &s->ctx) != 0) {
         goto out_close_efd;
+    }
 
-    qemu_aio_set_fd_handler(s->efd, qemu_laio_completion_cb, NULL,
-        qemu_laio_flush_cb, s);
+    qemu_aio_set_event_notifier(&s->e, qemu_laio_completion_cb,
+                                qemu_laio_flush_cb);
 
     return s;
 
 out_close_efd:
-    close(s->efd);
+    event_notifier_cleanup(&s->e);
 out_free_state:
     g_free(s);
     return NULL;

Signature · 1 line

1.7.10.4

The new thread pool will use semaphores instead of condition variables, because QemuCond does not have qemu_cond_timedwait.

(I also like it more this way, since semaphores model well the producer-consumer problem).

Quoted 2 lines

The new thread pool will use semaphores instead of condition variables, because QemuCond does not have qemu_cond_timedwait.

I'll post an updated patch (according to last round's review comments) that adds this service for POSIX. I bet you'll find a way to extend it to Win32 if that is required. ;)

Quoted 2 lines

(I also like it more this way, since semaphores model well the producer-consumer problem).

Let's not introduce another synchronization mechanism unless there is a real need. Semaphores tend to be misused for things they don't fit, so better keep them out of reach.

Also, if you do producer-consumer this way, you need a down() for every entity you dequeue. In contrast, you only interact with condition variables if there the consumer queue is empty - less atomic ops.

Il 16/07/2012 14:00, Jan Kiszka ha scritto:

Quoted 7 lines · Paolo Bonzini

Quoted 2 lines

The new thread pool will use semaphores instead of condition variables, because QemuCond does not have qemu_cond_timedwait.

I'll post an updated patch (according to last round's review comments) that adds this service for POSIX. I bet you'll find a way to extend it to Win32 if that is required. ;)

I can do that (or just use pthreads-win32), but only at the cost of making cond_wait() slower and more complex.

Quoted 7 lines

Quoted 3 lines

(I also like it more this way, since semaphores model well the producer-consumer problem).

Let's not introduce another synchronization mechanism unless there is a real need. Semaphores tend to be misused for things they don't fit, so better keep them out of reach.

That's what patch review is for...

Quoted 5 lines

Il 16/07/2012 14:00, Jan Kiszka ha scritto:

Quoted 4 lines · Paolo Bonzini

Quoted 2 lines

The new thread pool will use semaphores instead of condition variables, because QemuCond does not have qemu_cond_timedwait.

Quoted 6 lines

Quoted 3 lines

I'll post an updated patch (according to last round's review comments) that adds this service for POSIX. I bet you'll find a way to extend it to Win32 if that is required. ;)

I can do that (or just use pthreads-win32), but only at the cost of making cond_wait() slower and more complex.

Why will it affect cond_wait? WaitForSingleObject can time out as well.

Il 16/07/2012 15:34, Jan Kiszka ha scritto:

Quoted 10 lines · Paolo Bonzini

Quoted 9 lines

Il 16/07/2012 14:00, Jan Kiszka ha scritto:

Quoted 7 lines · Paolo Bonzini

Quoted 2 lines

The new thread pool will use semaphores instead of condition variables, because QemuCond does not have qemu_cond_timedwait.

I'll post an updated patch (according to last round's review comments) that adds this service for POSIX. I bet you'll find a way to extend it to Win32 if that is required. ;)

Quoted 4 lines

Quoted 2 lines

I can do that (or just use pthreads-win32), but only at the cost of making cond_wait() slower and more complex.

Why will it affect cond_wait? WaitForSingleObject can time out as well.

qemu_cond_wait only uses WaitForSingleObject with INFINITE timeout, and the algorithm relies on that.

Quoted 3 lines

Quoted 3 lines

qemu_cond_wait only uses WaitForSingleObject with INFINITE timeout, and the algorithm relies on that.

I see. But this doesn't look complex awfully. Just move the waker signaling from within cond_wait under the mutex as well, maybe add another flag if there is really someone waiting, and that's it. The costs should be hard to measure, even in line of code.

Il 16/07/2012 15:53, Jan Kiszka ha scritto:

Quoted 3 lines

Quoted 3 lines

Quoted 3 lines

qemu_cond_wait only uses WaitForSingleObject with INFINITE timeout, and the algorithm relies on that.

Quoted 4 lines

I see. But this doesn't look complex awfully. Just move the waker signaling from within cond_wait under the mutex as well, maybe add another flag if there is really someone waiting, and that's it. The costs should be hard to measure, even in line of code.

There is still a race after WaitForSingleObject times out. You need to catch the mutex before decreasing the number of waiters, and during that window somebody can broadcast the condition variable.

I'm not saying it's impossible, just that it's hard and I dislike qemu_cond_timedwait as much as you dislike semaphores. :)

Quoted 9 lines

Il 16/07/2012 15:53, Jan Kiszka ha scritto:

Quoted 7 lines

Quoted 3 lines

Quoted 3 lines

qemu_cond_wait only uses WaitForSingleObject with INFINITE timeout, and the algorithm relies on that.

I see. But this doesn't look complex awfully. Just move the waker signaling from within cond_wait under the mutex as well, maybe add another flag if there is really someone waiting, and that's it. The costs should be hard to measure, even in line of code.

Quoted 3 lines

There is still a race after WaitForSingleObject times out. You need to catch the mutex before decreasing the number of waiters, and during that window somebody can broadcast the condition variable.

...and that's why you check what needs to be done to handle this race after grabbing the mutex. IOW, replicate the state information that the Windows semaphore contains into the emulated condition variable object.

Quoted 2 lines

I'm not saying it's impossible, just that it's hard and I dislike qemu_cond_timedwait as much as you dislike semaphores. :)

As you cannot eliminate our use cases for condition variables, let's focus on the existing synchronization patterns instead of introducing new ones + new mechanisms.

Quoted 3 lines

Quoted 3 lines

...and that's why you check what needs to be done to handle this race after grabbing the mutex. IOW, replicate the state information that the Windows semaphore contains into the emulated condition variable object.

It is already there (cv->waiters), but it is accessed atomically. To do what you suggest I would need to add a mutex.

The thread pool can be used under Win32 in the same way as in raw-posix.c. Move the existing synchronous code into callbacks, and pass the return code back.

Signed-off-by Paolo Bonzini <[email protected]>

---
 block/raw-win32.c |  189 +++++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 140 insertions(+), 49 deletions(-)

block/raw-win32.c +3 -0

index e4b0b75..a50d636 100644
@@ -25,6 +25,9 @@
 #include "qemu-timer.h"
 #include "block_int.h"
 #include "module.h"
+#include "raw-aio.h"
+#include "trace.h"
+#include "thread-pool.h"
 #include <windows.h>
 #include <winioctl.h>
 

block/raw-win32.c +78 -0

+    BlockDriverState *bs;
+    HANDLE hfile;
+    struct iovec *aio_iov;
+    int aio_niov;
+    size_t aio_nbytes;
+    off_t aio_offset;
+    int aio_type;
+};
+
 typedef struct BDRVRawState {
     HANDLE hfile;
     int type;
     char drive_path[16]; /* format: "d:\" */
 } BDRVRawState;
 
+/*
+ * Read/writes the data to/from a given linear buffer.
+ *
+ * Returns the number of bytes handles or -errno in case of an error. Short
+ * reads are only returned if the end of the file is reached.
+ */
+static size_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
+{
+    size_t offset = 0;
+    int i;
+
+    for (i = 0; i < aiocb->aio_niov; i++) {
+        OVERLAPPED ov;
+        DWORD ret, ret_count, len;
+
+        memset(&ov, 0, sizeof(ov));
+        ov.Offset = (aiocb->aio_offset + offset);
+        ov.OffsetHigh = (aiocb->aio_offset + offset) >> 32;
+        len = aiocb->aio_iov[i].iov_len;
+        if (aiocb->aio_type & QEMU_AIO_WRITE) {
+            ret = WriteFile(aiocb->hfile, aiocb->aio_iov[i].iov_base,
+                            len, &ret_count, &ov);
+        } else {
+            ret = ReadFile(aiocb->hfile, aiocb->aio_iov[i].iov_base,
+                           len, &ret_count, &ov);
+        }
+        if (!ret) {
+            ret_count = 0;
+        }
+        if (ret_count != len) {
+            break;
+        }
+        offset += len;
+    }
+
+    return offset;
+}
+
+static int aio_worker(void *arg)
+{
+    struct qemu_paiocb *aiocb = arg;
+    ssize_t ret = 0;
+    size_t count;
+
+    switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
+    case QEMU_AIO_READ:
+        count = handle_aiocb_rw(aiocb);
+        if (count < aiocb->aio_nbytes && aiocb->bs->growable) {
+            /* A short read means that we have reached EOF. Pad the buffer
+             * with zeros for bytes after EOF. */
+            QEMUIOVector qiov;
+
+            qemu_iovec_init_external(&qiov, aiocb->aio_iov,
+                                     aiocb->aio_niov);
+            qemu_iovec_memset_skip(&qiov, 0, aiocb->aio_nbytes - count, count);
+
+            count = aiocb->aio_nbytes;
+        }
+        if (count == aiocb->aio_nbytes) {
+            ret = 0;
+        } else {
+            ret = -EINVAL;
+        }
+        break;
+    case QEMU_AIO_WRITE:
+        count = handle_aiocb_rw(aiocb);
+        if (count == aiocb->aio_nbytes) {
+            count = 0;
+        } else {

block/raw-win32.c +48 -50

+        ret = -EINVAL;
+        break;
+    }
+
+    g_slice_free(struct qemu_paiocb, aiocb);
+    return ret;
+}
+
+static BlockDriverAIOCB *paio_submit(BlockDriverState *bs, HANDLE hfile,
+        int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
+        BlockDriverCompletionFunc *cb, void *opaque, int type)
+{
+    struct qemu_paiocb *acb = g_slice_new(struct qemu_paiocb);
+
+    acb->bs = bs;
+    acb->hfile = hfile;
+    acb->aio_type = type;
+
+    if (qiov) {
+        acb->aio_iov = qiov->iov;
+        acb->aio_niov = qiov->niov;
+    }
+    acb->aio_nbytes = nb_sectors * 512;
+    acb->aio_offset = sector_num * 512;
+
+    trace_paio_submit(acb, opaque, sector_num, nb_sectors, type);
+    return thread_pool_submit_aio(aio_worker, acb, cb, opaque);
+}
+
 int qemu_ftruncate64(int fd, int64_t length)
 {
     LARGE_INTEGER li;

@@ -109,59 +230,29 @@ static int raw_open(BlockDriverState *bs, const char *filename, int flags)
     return 0;
 }
 
-static int raw_read(BlockDriverState *bs, int64_t sector_num,
-                    uint8_t *buf, int nb_sectors)
+static BlockDriverAIOCB *raw_aio_readv(BlockDriverState *bs,
+                         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
+                         BlockDriverCompletionFunc *cb, void *opaque)
 {
     BDRVRawState *s = bs->opaque;
-    OVERLAPPED ov;
-    DWORD ret_count;
-    int ret;
-    int64_t offset = sector_num * 512;
-    int count = nb_sectors * 512;
-
-    memset(&ov, 0, sizeof(ov));
-    ov.Offset = offset;
-    ov.OffsetHigh = offset >> 32;
-    ret = ReadFile(s->hfile, buf, count, &ret_count, &ov);
-    if (!ret)
-        return ret_count;
-    if (ret_count == count)
-        ret_count = 0;
-    return ret_count;
+    return paio_submit(bs, s->hfile, sector_num, qiov, nb_sectors,
+                       cb, opaque, QEMU_AIO_READ);
 }
 
-static int raw_write(BlockDriverState *bs, int64_t sector_num,
-                     const uint8_t *buf, int nb_sectors)
+static BlockDriverAIOCB *raw_aio_writev(BlockDriverState *bs,
+                          int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
+                          BlockDriverCompletionFunc *cb, void *opaque)
 {
     BDRVRawState *s = bs->opaque;
-    OVERLAPPED ov;
-    DWORD ret_count;
-    int ret;
-    int64_t offset = sector_num * 512;
-    int count = nb_sectors * 512;
-
-    memset(&ov, 0, sizeof(ov));
-    ov.Offset = offset;
-    ov.OffsetHigh = offset >> 32;
-    ret = WriteFile(s->hfile, buf, count, &ret_count, &ov);
-    if (!ret)
-        return ret_count;
-    if (ret_count == count)
-        ret_count = 0;
-    return ret_count;
+    return paio_submit(bs, s->hfile, sector_num, qiov, nb_sectors,
+                       cb, opaque, QEMU_AIO_WRITE);
 }
 
-static int raw_flush(BlockDriverState *bs)
+static BlockDriverAIOCB *raw_aio_flush(BlockDriverState *bs,
+                         BlockDriverCompletionFunc *cb, void *opaque)
 {
     BDRVRawState *s = bs->opaque;
-    int ret;
-
-    ret = FlushFileBuffers(s->hfile);
-    if (ret == 0) {
-        return -EIO;
-    }
-
-    return 0;
+    return paio_submit(bs, s->hfile, 0, NULL, 0, cb, opaque, QEMU_AIO_FLUSH);
 }
 
 static void raw_close(BlockDriverState *bs)

@@ -282,9 +373,9 @@ static BlockDriver bdrv_file = {
     .bdrv_close                = raw_close,
     .bdrv_create       = raw_create,
 
-    .bdrv_read              = raw_read,
-    .bdrv_write             = raw_write,
-    .bdrv_co_flush_to_disk  = raw_flush,
+    .bdrv_aio_readv     = raw_aio_readv,
+    .bdrv_aio_writev    = raw_aio_writev,
+    .bdrv_aio_flush     = raw_aio_flush,
 
     .bdrv_truncate     = raw_truncate,
     .bdrv_getlength    = raw_getlength,

@@ -413,9 +504,9 @@ static BlockDriver bdrv_host_device = {
     .bdrv_close                = raw_close,
     .bdrv_has_zero_init = hdev_has_zero_init,
 
-    .bdrv_read              = raw_read,
-    .bdrv_write             = raw_write,
-    .bdrv_co_flush_to_disk  = raw_flush,
+    .bdrv_aio_readv     = raw_aio_readv,
+    .bdrv_aio_writev    = raw_aio_writev,
+    .bdrv_aio_flush     = raw_aio_flush,
 
     .bdrv_getlength    = raw_getlength,
     .bdrv_get_allocated_file_size
--

1.7.10.4

Quoted 1 line

Quoted 1 line

+struct qemu_paiocb {

QEMUPAIOCB

Quoted 87 lines

Quoted 87 lines

block/raw-win32.c +81 -0

+    BlockDriverState *bs;
+    HANDLE hfile;
+    struct iovec *aio_iov;
+    int aio_niov;
+    size_t aio_nbytes;
+    off_t aio_offset;
+    int aio_type;
+};
+
 typedef struct BDRVRawState {
     HANDLE hfile;
     int type;
     char drive_path[16]; /* format: "d:\" */
 } BDRVRawState;
 
+/*
+ * Read/writes the data to/from a given linear buffer.
+ *
+ * Returns the number of bytes handles or -errno in case of an error. Short
+ * reads are only returned if the end of the file is reached.
+ */
+static size_t handle_aiocb_rw(struct qemu_paiocb *aiocb)
+{
+    size_t offset = 0;
+    int i;
+
+    for (i = 0; i < aiocb->aio_niov; i++) {
+        OVERLAPPED ov;
+        DWORD ret, ret_count, len;
+
+        memset(&ov, 0, sizeof(ov));
+        ov.Offset = (aiocb->aio_offset + offset);
+        ov.OffsetHigh = (aiocb->aio_offset + offset) >> 32;
+        len = aiocb->aio_iov[i].iov_len;
+        if (aiocb->aio_type & QEMU_AIO_WRITE) {
+            ret = WriteFile(aiocb->hfile, aiocb->aio_iov[i].iov_base,
+                            len, &ret_count, &ov);
+        } else {
+            ret = ReadFile(aiocb->hfile, aiocb->aio_iov[i].iov_base,
+                           len, &ret_count, &ov);
+        }
+        if (!ret) {
+            ret_count = 0;
+        }
+        if (ret_count != len) {
+            break;
+        }
+        offset += len;
+    }
+
+    return offset;
+}
+
+static int aio_worker(void *arg)
+{
+    struct qemu_paiocb *aiocb = arg;
+    ssize_t ret = 0;
+    size_t count;
+
+    switch (aiocb->aio_type & QEMU_AIO_TYPE_MASK) {
+    case QEMU_AIO_READ:
+        count = handle_aiocb_rw(aiocb);
+        if (count < aiocb->aio_nbytes && aiocb->bs->growable) {
+            /* A short read means that we have reached EOF. Pad the buffer
+             * with zeros for bytes after EOF. */
+            QEMUIOVector qiov;
+
+            qemu_iovec_init_external(&qiov, aiocb->aio_iov,
+                                     aiocb->aio_niov);
+            qemu_iovec_memset_skip(&qiov, 0, aiocb->aio_nbytes - count, count);
+
+            count = aiocb->aio_nbytes;
+        }
+        if (count == aiocb->aio_nbytes) {
+            ret = 0;
+        } else {
+            ret = -EINVAL;
+        }
+        break;
+    case QEMU_AIO_WRITE:
+        count = handle_aiocb_rw(aiocb);
+        if (count == aiocb->aio_nbytes) {
+            count = 0;
+        } else {
+            count = -EINVAL;
+        }
+        break;

Quoted 7 lines

block/raw-win32.c line 132

Quoted 7 lines

block/raw-win32.c +7 -0

+    case QEMU_AIO_FLUSH:
+        if (!FlushFileBuffers(aiocb->hfile)) {
+            return -EIO;
+        }
+        break;
+    default:
+        fprintf(stderr, "invalid aio request (0x%x)\n", aiocb->aio_type);

Assert instead?