[virtio-9p] Add infrastructure to support glib threads and coroutines.

This patch is originally made by Arun Bharadwaj for glib support.
Later Harsh Prateek Bora added coroutines support.
This version implemented with suggestions from
Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>.

Signed-off-by: Arun R Bharadwaj <arun@linux.vnet.ibm.com>
Signed-off-by: Harsh Prateek Bora <harsh@linux.vnet.ibm.com>
Signed-off-by: Venkateswararao Jujjuri "<jvrao@linux.vnet.ibm.com>
Signed-off-by: Aneesh Kumar K.V <aneesh.kumar@linux.vnet.ibm.com>

3 files changed, 165 insertions, 2 deletions

diff --git a/hw/9pfs/virtio-9p-coth.c b/hw/9pfs/virtio-9p-coth.c
new file mode 100644
index 0000000000..ae05658632
--- /dev/null
+++ b/hw/9pfs/virtio-9p-coth.c
@@ -0,0 +1,102 @@
+/*
+ * Virtio 9p backend
+ *
+ * Copyright IBM, Corp. 2010
+ *
+ * Authors:
+ *  Harsh Prateek Bora <harsh@linux.vnet.ibm.com>
+ *  Venkateswararao Jujjuri(JV) <jvrao@linux.vnet.ibm.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2.  See
+ * the COPYING file in the top-level directory.
+ *
+ */
+
+#include "qemu-char.h"
+#include "fsdev/qemu-fsdev.h"
+#include "qemu-thread.h"
+#include "qemu-coroutine.h"
+#include "virtio-9p-coth.h"
+
+/* v9fs glib thread pool */
+static V9fsThPool v9fs_pool;
+
+void co_run_in_worker_bh(void *opaque)
+{
+    Coroutine *co = opaque;
+    g_thread_pool_push(v9fs_pool.pool, co, NULL);
+}
+
+static void v9fs_qemu_process_req_done(void *arg)
+{
+    char byte;
+    ssize_t len;
+    Coroutine *co;
+
+    do {
+        len = read(v9fs_pool.rfd, &byte, sizeof(byte));
+    } while (len == -1 &&  errno == EINTR);
+
+    while ((co = g_async_queue_try_pop(v9fs_pool.completed)) != NULL) {
+        qemu_coroutine_enter(co, NULL);
+    }
+}
+
+static void v9fs_thread_routine(gpointer data, gpointer user_data)
+{
+    ssize_t len;
+    char byte = 0;
+    Coroutine *co = data;
+
+    qemu_coroutine_enter(co, NULL);
+
+    g_async_queue_push(v9fs_pool.completed, co);
+    do {
+        len = write(v9fs_pool.wfd, &byte, sizeof(byte));
+    } while (len == -1 && errno == EINTR);
+}
+
+int v9fs_init_worker_threads(void)
+{
+    int ret = 0;
+    int notifier_fds[2];
+    V9fsThPool *p = &v9fs_pool;
+    sigset_t set, oldset;
+
+    sigfillset(&set);
+    /* Leave signal handling to the iothread.  */
+    pthread_sigmask(SIG_SETMASK, &set, &oldset);
+
+    /* init thread system if not already initialized */
+    if (!g_thread_get_initialized()) {
+        g_thread_init(NULL);
+    }
+    if (qemu_pipe(notifier_fds) == -1) {
+        ret = -1;
+        goto err_out;
+    }
+    p->pool = g_thread_pool_new(v9fs_thread_routine, p, -1, FALSE, NULL);
+    if (!p->pool) {
+        ret = -1;
+        goto err_out;
+    }
+    p->completed = g_async_queue_new();
+    if (!p->completed) {
+        /*
+         * We are going to terminate.
+         * So don't worry about cleanup
+         */
+        ret = -1;
+        goto err_out;
+    }
+    p->rfd = notifier_fds[0];
+    p->wfd = notifier_fds[1];
+
+    fcntl(p->rfd, F_SETFL, O_NONBLOCK);
+    fcntl(p->wfd, F_SETFL, O_NONBLOCK);
+
+    qemu_set_fd_handler(p->rfd, v9fs_qemu_process_req_done, NULL, NULL);
+err_out:
+    pthread_sigmask(SIG_SETMASK, &oldset, NULL);
+    return ret;
+}
diff --git a/hw/9pfs/virtio-9p-coth.h b/hw/9pfs/virtio-9p-coth.h
new file mode 100644
index 0000000000..8445d29bc4
--- /dev/null
+++ b/hw/9pfs/virtio-9p-coth.h
@@ -0,0 +1,58 @@
+/*
+ * Virtio 9p backend
+ *
+ * Copyright IBM, Corp. 2010
+ *
+ * Authors:
+ *  Harsh Prateek Bora <harsh@linux.vnet.ibm.com>
+ *  Venkateswararao Jujjuri(JV) <jvrao@linux.vnet.ibm.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2.  See
+ * the COPYING file in the top-level directory.
+ *
+ */
+
+#ifndef _QEMU_VIRTIO_9P_COTH_H
+#define _QEMU_VIRTIO_9P_COTH_H
+
+#include "qemu-thread.h"
+#include "qemu-coroutine.h"
+#include <glib.h>
+
+typedef struct V9fsThPool {
+    int rfd;
+    int wfd;
+    GThreadPool *pool;
+    GAsyncQueue *completed;
+} V9fsThPool;
+
+/*
+ * we want to use bottom half because we want to make sure the below
+ * sequence of events.
+ *
+ *   1. Yield the coroutine in the QEMU thread.
+ *   2. Submit the coroutine to a worker thread.
+ *   3. Enter the coroutine in the worker thread.
+ * we cannot swap step 1 and 2, because that would imply worker thread
+ * can enter coroutine while step1 is still running
+ */
+#define v9fs_co_run_in_worker(code_block)                               \
+    do {                                                                \
+        QEMUBH *co_bh;                                                  \
+        co_bh = qemu_bh_new(co_run_in_worker_bh,                        \
+                            qemu_coroutine_self());                     \
+        qemu_bh_schedule(co_bh);                                        \
+        /*                                                              \
+         * yeild in qemu thread and re-enter back                       \
+         * in glib worker thread                                        \
+         */                                                             \
+        qemu_coroutine_yield();                                         \
+        qemu_bh_delete(co_bh);                                          \
+        code_block;                                                     \
+        /* re-enter back to qemu thread */                              \
+        qemu_coroutine_yield();                                         \
+    } while (0)
+
+extern void co_run_in_worker_bh(void *);
+extern int v9fs_init_worker_threads(void);
+#endif
diff --git a/hw/9pfs/virtio-9p-device.c b/hw/9pfs/virtio-9p-device.c
index f235236ea0..f4bf471871 100644
--- a/hw/9pfs/virtio-9p-device.c
+++ b/hw/9pfs/virtio-9p-device.c
@@ -18,6 +18,7 @@
 #include "virtio-9p.h"
 #include "fsdev/qemu-fsdev.h"
 #include "virtio-9p-xattr.h"
+#include "virtio-9p-coth.h"
 
 static uint32_t virtio_9p_get_features(VirtIODevice *vdev, uint32_t features)
 {
@@ -50,13 +51,11 @@ VirtIODevice *virtio_9p_init(DeviceState *dev, V9fsConf *conf)
     struct stat stat;
     FsTypeEntry *fse;
 
-
     s = (V9fsState *)virtio_common_init("virtio-9p",
                                     VIRTIO_ID_9P,
                                     sizeof(struct virtio_9p_config)+
                                     MAX_TAG_LEN,
                                     sizeof(V9fsState));
-
     /* initialize pdu allocator */
     QLIST_INIT(&s->free_list);
     for (i = 0; i < (MAX_REQ - 1); i++) {
@@ -132,6 +131,10 @@ VirtIODevice *virtio_9p_init(DeviceState *dev, V9fsConf *conf)
                         s->tag_len;
     s->vdev.get_config = virtio_9p_get_config;
 
+    if (v9fs_init_worker_threads() < 0) {
+        fprintf(stderr, "worker thread initialization failed\n");
+        exit(1);
+    }
     return &s->vdev;
 }