[Qemu-devel] [PATCH 7/9] iohandler: associate with main event loop via a QSource

[Michael Roth]

This introduces a GlibQContext wrapper around the main GMainContext
event loop, and associates iohandlers with it via a QSource (which
GlibQContext creates a GSource from so that it can be driven via
GLib. A subsequent patch will drive the GlibQContext directly)

We also add "QContext-aware" functionality to iohandler interfaces
so that they can be bound to other QContext event loops, and add
non-global set_fd_handler() interfaces to facilitate this. This is made
possible by simply searching a given QContext for a QSource by the name
of "iohandler" so that we can attach event handlers to the associated
IOHandlerState.

Signed-off-by: Michael Roth <address@hidden>
---
 include/qemu/main-loop.h |   31 +++++-
 iohandler.c              |  238 ++++++++++++++++++++++++++++++++--------------
 main-loop.c              |   21 +++-
 3 files changed, 213 insertions(+), 77 deletions(-)

diff --git a/include/qemu/main-loop.h b/include/qemu/main-loop.h
index 6f0200a..dbadf9f 100644
--- a/include/qemu/main-loop.h
+++ b/include/qemu/main-loop.h
@@ -26,6 +26,7 @@
 #define QEMU_MAIN_LOOP_H 1
 
 #include "block/aio.h"
+#include "qcontext/qcontext.h"
 
 #define SIG_IPI SIGUSR1
 
@@ -168,9 +169,24 @@ void qemu_del_wait_object(HANDLE handle, WaitObjectFunc 
*func, void *opaque);
 
 /* async I/O support */
 
+#define QSOURCE_IOHANDLER "iohandler"
+
 typedef void IOReadHandler(void *opaque, const uint8_t *buf, int size);
 typedef int IOCanReadHandler(void *opaque);
 
+QContext *qemu_get_qcontext(void);
+/**
+ * iohandler_attach: Attach a QSource to a QContext
+ *
+ * This enables the use of IOHandler interfaces such as
+ * set_fd_handler() on the given QContext. IOHandler lists will be
+ * tracked/handled/dispatched based on a named QSource that is added to
+ * the QContext
+ *
+ * @ctx: A QContext to add an IOHandler QSource to
+ */
+void iohandler_attach(QContext *ctx);
+
 /**
  * qemu_set_fd_handler2: Register a file descriptor with the main loop
  *
@@ -217,6 +233,13 @@ int qemu_set_fd_handler2(int fd,
                          IOHandler *fd_write,
                          void *opaque);
 
+int set_fd_handler2(QContext *ctx,
+                    int fd,
+                    IOCanReadHandler *fd_read_poll,
+                    IOHandler *fd_read,
+                    IOHandler *fd_write,
+                    void *opaque);
+
 /**
  * qemu_set_fd_handler: Register a file descriptor with the main loop
  *
@@ -250,6 +273,12 @@ int qemu_set_fd_handler(int fd,
                         IOHandler *fd_write,
                         void *opaque);
 
+int set_fd_handler(QContext *ctx,
+                   int fd,
+                   IOHandler *fd_read,
+                   IOHandler *fd_write,
+                   void *opaque);
+
 #ifdef CONFIG_POSIX
 /**
  * qemu_add_child_watch: Register a child process for reaping.
@@ -302,8 +331,6 @@ void qemu_mutex_unlock_iothread(void);
 /* internal interfaces */
 
 void qemu_fd_register(int fd);
-void qemu_iohandler_fill(GArray *pollfds);
-void qemu_iohandler_poll(GArray *pollfds, int rc);
 
 QEMUBH *qemu_bh_new(QEMUBHFunc *cb, void *opaque);
 void qemu_bh_schedule_idle(QEMUBH *bh);
diff --git a/iohandler.c b/iohandler.c
index ae2ef8f..8625272 100644
--- a/iohandler.c
+++ b/iohandler.c
@@ -41,38 +41,170 @@ typedef struct IOHandlerRecord {
     int fd;
     int pollfds_idx;
     bool deleted;
+    GPollFD pfd;
+    bool pfd_added;
 } IOHandlerRecord;
 
-static QLIST_HEAD(, IOHandlerRecord) io_handlers =
-    QLIST_HEAD_INITIALIZER(io_handlers);
+typedef struct IOHandlerState {
+    QLIST_HEAD(, IOHandlerRecord) io_handlers;
+} IOHandlerState;
 
+static bool iohandler_prepare(QSource *qsource, int *timeout)
+{
+    QSourceClass *qsourcek = QSOURCE_GET_CLASS(qsource);
+    IOHandlerState *s = qsourcek->get_user_data(qsource);
+    IOHandlerRecord *ioh;
 
-/* XXX: fd_read_poll should be suppressed, but an API change is
-   necessary in the character devices to suppress fd_can_read(). */
-int qemu_set_fd_handler2(int fd,
-                         IOCanReadHandler *fd_read_poll,
-                         IOHandler *fd_read,
-                         IOHandler *fd_write,
-                         void *opaque)
+    QLIST_FOREACH(ioh, &s->io_handlers, next) {
+        int events = 0;
+
+        if (ioh->deleted)
+            continue;
+
+        if (ioh->fd_read &&
+            (!ioh->fd_read_poll ||
+             ioh->fd_read_poll(ioh->opaque) != 0)) {
+            events |= G_IO_IN | G_IO_HUP | G_IO_ERR;
+        }
+        if (ioh->fd_write) {
+            events |= G_IO_OUT | G_IO_ERR;
+        }
+        if (events) {
+            ioh->pfd.fd = ioh->fd;
+            ioh->pfd.events = events;
+            if (!ioh->pfd_added) {
+                qsourcek->add_poll(qsource, &ioh->pfd);
+                ioh->pfd_added = true;
+            }
+        } else {
+            ioh->pfd.events = 0;
+            ioh->pfd.revents = 0;
+        }
+    }
+    *timeout = 10;
+    return false;
+}
+
+static bool iohandler_check(QSource *qsource)
 {
+    QSourceClass *sourcek = QSOURCE_GET_CLASS(qsource);
+    IOHandlerState *s = sourcek->get_user_data(qsource);
     IOHandlerRecord *ioh;
 
+    QLIST_FOREACH(ioh, &s->io_handlers, next) {
+        if (ioh->deleted) {
+            continue;
+        }
+        if (ioh->pfd.revents) {
+            return true;
+        }
+    }
+
+    return false;
+}
+
+static bool iohandler_dispatch(QSource *qsource)
+{
+    QSourceClass *sourcek = QSOURCE_GET_CLASS(qsource);
+    IOHandlerState *s = sourcek->get_user_data(qsource);
+    IOHandlerRecord *pioh, *ioh;
+
+    QLIST_FOREACH_SAFE(ioh, &s->io_handlers, next, pioh) {
+        int revents = ioh->pfd.revents;
+        if (!ioh->deleted && ioh->fd_read &&
+            (revents && (G_IO_IN | G_IO_HUP | G_IO_ERR))) {
+            ioh->fd_read(ioh->opaque);
+        }
+
+        if (!ioh->deleted && ioh->fd_write &&
+            (revents & (G_IO_OUT | G_IO_ERR))) {
+            ioh->fd_write(ioh->opaque);
+        }
+
+        /* Do this last in case read/write handlers marked it for deletion */
+        if (ioh->deleted) {
+            if (ioh->pfd_added) {
+                sourcek->remove_poll(qsource, &ioh->pfd);
+            }
+            QLIST_REMOVE(ioh, next);
+            g_free(ioh);
+        }
+    }
+
+    return true;
+}
+
+static void iohandler_finalize(QSource *qsource)
+{
+    QSourceClass *sourcek = QSOURCE_GET_CLASS(qsource);
+    IOHandlerState *s = sourcek->get_user_data(qsource);
+    IOHandlerRecord *pioh, *ioh;
+
+    QLIST_FOREACH_SAFE(ioh, &s->io_handlers, next, pioh) {
+        if (ioh->pfd_added) {
+            sourcek->remove_poll(qsource, &ioh->pfd);
+        }
+        QLIST_REMOVE(ioh, next);
+        g_free(ioh);
+    }
+
+    g_free(s);
+}
+
+static const QSourceFuncs iohandler_funcs = {
+    iohandler_prepare,
+    iohandler_check,
+    iohandler_dispatch,
+    iohandler_finalize,
+};
+
+void iohandler_attach(QContext *ctx)
+{
+    IOHandlerState *s;
+    QSource *qsource;
+
+    s = g_new0(IOHandlerState, 1);
+    QLIST_INIT(&s->io_handlers);
+
+    qsource = qsource_new(iohandler_funcs, NULL, QSOURCE_IOHANDLER, s);
+    qcontext_attach(ctx, qsource, NULL);
+}
+
+int set_fd_handler2(QContext *ctx,
+                    int fd,
+                    IOCanReadHandler *fd_read_poll,
+                    IOHandler *fd_read,
+                    IOHandler *fd_write,
+                    void *opaque)
+{
+    QSourceClass *sourcek;
+    QSource *source;
+    IOHandlerState *s;
+    IOHandlerRecord *ioh;
+
+    source = qcontext_find_source_by_name(ctx, QSOURCE_IOHANDLER);
+    if (!source) {
+        assert(0);
+    }
+    sourcek = QSOURCE_GET_CLASS(source);
+    s = sourcek->get_user_data(source);
+
     assert(fd >= 0);
 
     if (!fd_read && !fd_write) {
-        QLIST_FOREACH(ioh, &io_handlers, next) {
+        QLIST_FOREACH(ioh, &s->io_handlers, next) {
             if (ioh->fd == fd) {
                 ioh->deleted = 1;
                 break;
             }
         }
     } else {
-        QLIST_FOREACH(ioh, &io_handlers, next) {
+        QLIST_FOREACH(ioh, &s->io_handlers, next) {
             if (ioh->fd == fd)
                 goto found;
         }
         ioh = g_malloc0(sizeof(IOHandlerRecord));
-        QLIST_INSERT_HEAD(&io_handlers, ioh, next);
+        QLIST_INSERT_HEAD(&s->io_handlers, ioh, next);
     found:
         ioh->fd = fd;
         ioh->fd_read_poll = fd_read_poll;
@@ -86,74 +218,34 @@ int qemu_set_fd_handler2(int fd,
     return 0;
 }
 
-int qemu_set_fd_handler(int fd,
-                        IOHandler *fd_read,
-                        IOHandler *fd_write,
-                        void *opaque)
+int set_fd_handler(QContext *ctx,
+                   int fd,
+                   IOHandler *fd_read,
+                   IOHandler *fd_write,
+                   void *opaque)
 {
-    return qemu_set_fd_handler2(fd, NULL, fd_read, fd_write, opaque);
+    return set_fd_handler2(ctx, fd, NULL, fd_read, fd_write, opaque);
 }
 
-void qemu_iohandler_fill(GArray *pollfds)
+/* XXX: fd_read_poll should be suppressed, but an API change is
+   necessary in the character devices to suppress fd_can_read(). */
+int qemu_set_fd_handler2(int fd,
+                         IOCanReadHandler *fd_read_poll,
+                         IOHandler *fd_read,
+                         IOHandler *fd_write,
+                         void *opaque)
 {
-    IOHandlerRecord *ioh;
-
-    QLIST_FOREACH(ioh, &io_handlers, next) {
-        int events = 0;
-
-        if (ioh->deleted)
-            continue;
-        if (ioh->fd_read &&
-            (!ioh->fd_read_poll ||
-             ioh->fd_read_poll(ioh->opaque) != 0)) {
-            events |= G_IO_IN | G_IO_HUP | G_IO_ERR;
-        }
-        if (ioh->fd_write) {
-            events |= G_IO_OUT | G_IO_ERR;
-        }
-        if (events) {
-            GPollFD pfd = {
-                .fd = ioh->fd,
-                .events = events,
-            };
-            ioh->pollfds_idx = pollfds->len;
-            g_array_append_val(pollfds, pfd);
-        } else {
-            ioh->pollfds_idx = -1;
-        }
-    }
+    return set_fd_handler2(qemu_get_qcontext(), fd,
+                           fd_read_poll, fd_read, fd_write,
+                           opaque);
 }
 
-void qemu_iohandler_poll(GArray *pollfds, int ret)
+int qemu_set_fd_handler(int fd,
+                        IOHandler *fd_read,
+                        IOHandler *fd_write,
+                        void *opaque)
 {
-    if (ret > 0) {
-        IOHandlerRecord *pioh, *ioh;
-
-        QLIST_FOREACH_SAFE(ioh, &io_handlers, next, pioh) {
-            int revents = 0;
-
-            if (!ioh->deleted && ioh->pollfds_idx != -1) {
-                GPollFD *pfd = &g_array_index(pollfds, GPollFD,
-                                              ioh->pollfds_idx);
-                revents = pfd->revents;
-            }
-
-            if (!ioh->deleted && ioh->fd_read &&
-                (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR))) {
-                ioh->fd_read(ioh->opaque);
-            }
-            if (!ioh->deleted && ioh->fd_write &&
-                (revents & (G_IO_OUT | G_IO_ERR))) {
-                ioh->fd_write(ioh->opaque);
-            }
-
-            /* Do this last in case read/write handlers marked it for deletion 
*/
-            if (ioh->deleted) {
-                QLIST_REMOVE(ioh, next);
-                g_free(ioh);
-            }
-        }
-    }
+    return qemu_set_fd_handler2(fd, NULL, fd_read, fd_write, opaque);
 }
 
 /* reaping of zombies.  right now we're not passing the status to
diff --git a/main-loop.c b/main-loop.c
index f46aece..ae284a6 100644
--- a/main-loop.c
+++ b/main-loop.c
@@ -27,6 +27,8 @@
 #include "slirp/slirp.h"
 #include "qemu/main-loop.h"
 #include "block/aio.h"
+#include "qcontext/qcontext.h"
+#include "qcontext/glib-qcontext.h"
 
 #ifndef _WIN32
 
@@ -107,6 +109,13 @@ static int qemu_signal_init(void)
 }
 #endif
 
+static QContext *qemu_qcontext;
+
+QContext *qemu_get_qcontext(void)
+{
+    return qemu_qcontext;
+}
+
 static AioContext *qemu_aio_context;
 
 AioContext *qemu_get_aio_context(void)
@@ -128,6 +137,7 @@ int qemu_init_main_loop(void)
 {
     int ret;
     GSource *src;
+    Error *err = NULL;
 
     init_clocks();
     if (init_timer_alarm() < 0) {
@@ -135,6 +145,15 @@ int qemu_init_main_loop(void)
         exit(1);
     }
 
+    qemu_qcontext = QCONTEXT(glib_qcontext_new("main", false, &err));
+    if (err) {
+        g_warning("error initializing main qcontext");
+        error_free(err);
+        return -1;
+    }
+
+    iohandler_attach(qemu_qcontext);
+
     ret = qemu_signal_init();
     if (ret) {
         return ret;
@@ -464,9 +483,7 @@ int main_loop_wait(int nonblocking)
     slirp_update_timeout(&timeout);
     slirp_pollfds_fill(gpollfds);
 #endif
-    qemu_iohandler_fill(gpollfds);
     ret = os_host_main_loop_wait(timeout);
-    qemu_iohandler_poll(gpollfds, ret);
 #ifdef CONFIG_SLIRP
     slirp_pollfds_poll(gpollfds, (ret < 0));
 #endif
-- 
1.7.9.5