[Qemu-devel] [RFC/ PATCH 3/4] virtio-9p: Add async helper functions

[Gautham R Shenoy <ego@in.ibm.com>]

Add helper functions to enable virtio-9p make use of the generic asynchronous
threading framework for offloading blocking tasks such as making posix calls on
to the asynchronous threads and handle the post_posix_operations() from the
context of the iothread. This frees the vcpu thread to process any other guest
operations while the processing of v9fs_io is in progress.

Signed-off-by: Gautham R Shenoy <ego@in.ibm.com>
---
 hw/virtio-9p.c |  167 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 167 insertions(+), 0 deletions(-)

diff --git a/hw/virtio-9p.c b/hw/virtio-9p.c
index 82bb663..f8f60d3 100644
--- a/hw/virtio-9p.c
+++ b/hw/virtio-9p.c
@@ -17,10 +17,147 @@
 #include "virtio-9p.h"
 #include "fsdev/qemu-fsdev.h"
 #include "virtio-9p-debug.h"
+#include "async-work.h"
 
 int dotu = 1;
 int debug_9p_pdu;
 
+struct v9fs_post_op {
+    QTAILQ_ENTRY(v9fs_post_op) node;
+    void (*func)(void *arg);
+    void *arg;
+};
+
+static struct {
+    struct async_queue virtio_9p_aqueue;
+    int rfd;
+    int wfd;
+    pthread_mutex_t lock;
+    QTAILQ_HEAD(, v9fs_post_op) post_op_list;
+} v9fs_async_struct;
+
+static void die2(int err, const char *what)
+{
+    fprintf(stderr, "%s failed: %s\n", what, strerror(err));
+    abort();
+}
+
+static void die(const char *what)
+{
+    die2(errno, what);
+}
+
+/**
+ * v9fs_signal_handler: Handle the SIGUSR1 signal.
+ * @signum: Is SIGUSR1 in this case.
+ *
+ * This function is used to inform the iothread that a particular
+ * async-operation pertaining to v9fs has been completed and that the io thread
+ * can handle the v9fs_post_posix_operation.
+ *
+ * This is based on the aio_signal_handler
+ */
+static void v9fs_signal_handler(int signum)
+{
+    char byte = 0;
+    ssize_t ret;
+
+    printf("Writing to file descriptor %d\n", v9fs_async_struct.wfd);
+    ret = write(v9fs_async_struct.wfd, &byte, sizeof(byte));
+
+    if (ret < 0 && errno != EAGAIN)
+        die("write() in v9fs");
+
+    qemu_service_io();
+}
+
+#define ASYNC_MAX_PROCESS   5
+
+/**
+ * v9fs_process_post_ops: Process any pending v9fs_post_posix_operation
+ * @arg: Not used.
+ *
+ * This function serves as a callback to the iothread to be called into whenever
+ * the v9fs_async_struct.wfd is written into. This thread goes through the list
+ * of v9fs_post_posix_operations() and executes them. In the process, it might
+ * queue more job on the asynchronous thread pool.
+ */
+static void v9fs_process_post_ops(void *arg)
+{
+    int count = 0;
+    struct v9fs_post_op *post_op;
+
+    pthread_mutex_lock(&v9fs_async_struct.lock);
+    for (count = 0; count < ASYNC_MAX_PROCESS; count++) {
+        if (QTAILQ_EMPTY(&(v9fs_async_struct.post_op_list))) {
+            break;
+        }
+        post_op = QTAILQ_FIRST(&(v9fs_async_struct.post_op_list));
+        QTAILQ_REMOVE(&(v9fs_async_struct.post_op_list), post_op, node);
+
+        pthread_mutex_unlock(&v9fs_async_struct.lock);
+        post_op->func(post_op->arg);
+        qemu_free(post_op);
+        pthread_mutex_lock(&v9fs_async_struct.lock);
+    }
+    pthread_mutex_unlock(&v9fs_async_struct.lock);
+}
+
+/**
+ * v9fs_async_signal: Send a signal to the iothread.
+ * @func: v9fs_post_posix_func() to be called by the iothread.
+ * @arg: Argument to func.
+ *
+ * This function is called from the context of one of the asynchronous threads
+ * in the thread pool. This is called when the asynchronous thread has finished
+ * executing a v9fs_posix_operation. It's purpose is to initiate the process of
+ * informing the iothread that the v9fs_posix_operation has completed.
+ *
+ * This code follows the suit of the aio_thread() and uses SIGUSR1 to notify the
+ * iothread.
+ */
+static void v9fs_async_signal(void (*func)(void *arg), void *arg)
+{
+    struct v9fs_post_op *post_op;
+    pid_t pid = getpid();
+
+    post_op = qemu_mallocz(sizeof(*post_op));
+    post_op->func = func;
+    post_op->arg = arg;
+
+    pthread_mutex_lock(&v9fs_async_struct.lock);
+    QTAILQ_INSERT_TAIL(&(v9fs_async_struct.post_op_list), post_op, node);
+    pthread_mutex_unlock(&v9fs_async_struct.lock);
+
+    if(kill(pid, SIGUSR1)) die("v9fs kill failed");
+}
+
+/**
+ * v9fs_do_async_posix: Offload v9fs_posix_operation onto async thread.
+ * @vs: V9fsOPState variable for the OP operation.
+ * @posix_fn: The posix function which has to be offloaded onto async thread.
+ * @post_fn_ptr: Address of the location to hold the post_fn corresponding to
+ * the posix_fn
+ * @post_fn: The post_fn corresponding to the posix_fn.
+ *
+ * This function is a helper to offload posix_operation on to the asynchronous
+ * thread pool. It sets up the associations with the post_function that needs to
+ * be invoked by from the context of the iothread once the posix_fn has been
+ * executed.
+ */
+static void v9fs_do_async_posix(void *vs ,
+                                void (*posix_fn)(struct work_item *work),
+                                void (**post_fn_ptr)(void *arg),
+                                void (*post_fn)(void *arg))
+{
+    struct work_item *work;
+
+    *post_fn_ptr = post_fn;
+    work = async_work_init(&v9fs_async_struct.virtio_9p_aqueue,
+                            posix_fn, vs);
+    qemu_async_submit(&v9fs_async_struct.virtio_9p_aqueue, work);
+}
+
 enum {
     Oread   = 0x00,
     Owrite  = 0x01,
@@ -2321,6 +2458,8 @@ VirtIODevice *virtio_9p_init(DeviceState *dev, V9fsConf *conf)
     int i, len;
     struct stat stat;
     FsTypeEntry *fse;
+    int fds[2];
+    struct sigaction act;
 
 
     s = (V9fsState *)virtio_common_init("virtio-9p",
@@ -2395,5 +2534,33 @@ VirtIODevice *virtio_9p_init(DeviceState *dev, V9fsConf *conf)
                         s->tag_len;
     s->vdev.get_config = virtio_9p_get_config;
 
+    sigfillset(&act.sa_mask);
+    act.sa_flags = 0;
+    act.sa_handler = v9fs_signal_handler;
+    sigaction(SIGUSR1, &act, NULL);
+
+    if (qemu_pipe(fds) == -1) {
+        fprintf(stderr, "failed to create fd's for virtio-9p\n");
+        exit(1);
+    }
+
+    v9fs_async_struct.rfd = fds[0];
+    v9fs_async_struct.wfd = fds[1];
+
+    printf("v9fs: rfd: %d\n", v9fs_async_struct.rfd);
+    printf("v9fs: wfd: %d\n", v9fs_async_struct.wfd);
+
+    fcntl(fds[0], F_SETFL, O_NONBLOCK);
+    fcntl(fds[1], F_SETFL, O_NONBLOCK);
+
+    qemu_set_fd_handler(fds[0], v9fs_process_post_ops, NULL, NULL);
+    QTAILQ_INIT(&v9fs_async_struct.post_op_list);
+    pthread_mutex_init(&(v9fs_async_struct.lock), NULL);
+    /* Create async queue. */
+    async_queue_init(&v9fs_async_struct.virtio_9p_aqueue, 10, 3);
+
+    (void)v9fs_do_async_posix;
+    (void)v9fs_async_signal;
+
     return &s->vdev;
 }