[Qemu-devel] [PATCHv2] [RFC 6/7] aio / timers: Switch to ppoll, run AioContext timers in aio_poll/aio_dispatch

[Alex Bligh]

Switch to ppoll (or rather qemu_g_poll_ns which will use ppoll if available).

Set timeouts for aio, g_source, and mainloop from earliest timer deadline.

Run timers for AioContext (only) in aio_poll/aio_dispatch.

Signed-off-by: Alex Bligh <address@hidden>
---
 aio-posix.c |   20 +++++++++++++-------
 aio-win32.c |   20 ++++++++++++++++++--
 async.c     |   16 ++++++++++++++--
 main-loop.c |   43 ++++++++++++++++++++++++++++++++-----------
 4 files changed, 77 insertions(+), 22 deletions(-)

diff --git a/aio-posix.c b/aio-posix.c
index b68eccd..5bdb9fe 100644
--- a/aio-posix.c
+++ b/aio-posix.c
@@ -166,6 +166,10 @@ static bool aio_dispatch(AioContext *ctx)
             g_free(tmp);
         }
     }
+
+    /* Run our timers */
+    progress |= qemu_run_timers(ctx->clock);
+
     return progress;
 }
 
@@ -232,9 +236,9 @@ bool aio_poll(AioContext *ctx, bool blocking)
     }
 
     /* wait until next event */
-    ret = g_poll((GPollFD *)ctx->pollfds->data,
-                 ctx->pollfds->len,
-                 blocking ? -1 : 0);
+    ret = qemu_g_poll_ns((GPollFD *)ctx->pollfds->data,
+                         ctx->pollfds->len,
+                         blocking ? qemu_clock_deadline_all_ns() : 0);
 
     /* if we have any readable fds, dispatch event */
     if (ret > 0) {
@@ -245,11 +249,13 @@ bool aio_poll(AioContext *ctx, bool blocking)
                 node->pfd.revents = pfd->revents;
             }
         }
-        if (aio_dispatch(ctx)) {
-            progress = true;
-        }
+    }
+
+    /* Run dispatch even if there were no readable fds to run timers */
+    if (aio_dispatch(ctx)) {
+        progress = true;
     }
 
     assert(progress || busy);
-    return true;
+    return progress;
 }
diff --git a/aio-win32.c b/aio-win32.c
index 38723bf..68343ba 100644
--- a/aio-win32.c
+++ b/aio-win32.c
@@ -98,6 +98,7 @@ bool aio_poll(AioContext *ctx, bool blocking)
     HANDLE events[MAXIMUM_WAIT_OBJECTS + 1];
     bool busy, progress;
     int count;
+    int timeout;
 
     progress = false;
 
@@ -111,6 +112,9 @@ bool aio_poll(AioContext *ctx, bool blocking)
         progress = true;
     }
 
+    /* Run timers */
+    progress |= qemu_run_timers(ctx->clock);
+
     /*
      * Then dispatch any pending callbacks from the GSource.
      *
@@ -174,8 +178,11 @@ bool aio_poll(AioContext *ctx, bool blocking)
 
     /* wait until next event */
     while (count > 0) {
-        int timeout = blocking ? INFINITE : 0;
-        int ret = WaitForMultipleObjects(count, events, FALSE, timeout);
+        int ret;
+
+        timeout = blocking ?
+            qemu_timeout_ns_to_ms(qemu_clock_deadline_all_ns()) : 0;
+        ret = WaitForMultipleObjects(count, events, FALSE, timeout);
 
         /* if we have any signaled events, dispatch event */
         if ((DWORD) (ret - WAIT_OBJECT_0) >= count) {
@@ -214,6 +221,15 @@ bool aio_poll(AioContext *ctx, bool blocking)
         events[ret - WAIT_OBJECT_0] = events[--count];
     }
 
+    if (blocking) {
+        /* Run the timers a second time. We do this because otherwise aio_wait
+         * will not note progress - and will stop a drain early - if we have
+         * a timer that was not ready to run entering g_poll but is ready
+         * after g_poll. This will only do anything if a timer has expired.
+         */
+        progress |= qemu_run_timers(ctx->clock);
+    }
+
     assert(progress || busy);
     return true;
 }
diff --git a/async.c b/async.c
index 0d41431..cb6b1d4 100644
--- a/async.c
+++ b/async.c
@@ -123,13 +123,16 @@ aio_ctx_prepare(GSource *source, gint    *timeout)
 {
     AioContext *ctx = (AioContext *) source;
     QEMUBH *bh;
+    int deadline;
 
     for (bh = ctx->first_bh; bh; bh = bh->next) {
         if (!bh->deleted && bh->scheduled) {
             if (bh->idle) {
                 /* idle bottom halves will be polled at least
                  * every 10ms */
-                *timeout = 10;
+                if ((*timeout < 0) || (*timeout > 10)) {
+                    *timeout = 10;
+                }
             } else {
                 /* non-idle bottom halves will be executed
                  * immediately */
@@ -139,6 +142,15 @@ aio_ctx_prepare(GSource *source, gint    *timeout)
         }
     }
 
+    deadline = qemu_timeout_ns_to_ms(qemu_clock_deadline_ns(ctx->clock));
+    if (deadline == 0) {
+        *timeout = 0;
+        return true;
+    } else if ((deadline > 0) &&
+               ((*timeout < 0) || (deadline < *timeout))) {
+        *timeout = deadline;
+    }
+
     return false;
 }
 
@@ -153,7 +165,7 @@ aio_ctx_check(GSource *source)
             return true;
        }
     }
-    return aio_pending(ctx);
+    return aio_pending(ctx) || (qemu_clock_deadline_ns(ctx->clock) >= 0);
 }
 
 static gboolean
diff --git a/main-loop.c b/main-loop.c
index 8918dd1..1bd10e8 100644
--- a/main-loop.c
+++ b/main-loop.c
@@ -151,10 +151,11 @@ static int max_priority;
 static int glib_pollfds_idx;
 static int glib_n_poll_fds;
 
-static void glib_pollfds_fill(uint32_t *cur_timeout)
+static void glib_pollfds_fill(int64_t *cur_timeout)
 {
     GMainContext *context = g_main_context_default();
     int timeout = 0;
+    int64_t timeout_ns;
     int n;
 
     g_main_context_prepare(context, &max_priority);
@@ -170,9 +171,13 @@ static void glib_pollfds_fill(uint32_t *cur_timeout)
                                  glib_n_poll_fds);
     } while (n != glib_n_poll_fds);
 
-    if (timeout >= 0 && timeout < *cur_timeout) {
-        *cur_timeout = timeout;
+    if (timeout < 0) {
+        timeout_ns = -1;
+    } else {
+      timeout_ns = (int64_t)timeout * (int64_t)SCALE_MS;
     }
+
+    *cur_timeout = qemu_soonest_timeout(timeout_ns, *cur_timeout);
 }
 
 static void glib_pollfds_poll(void)
@@ -187,7 +192,7 @@ static void glib_pollfds_poll(void)
 
 #define MAX_MAIN_LOOP_SPIN (1000)
 
-static int os_host_main_loop_wait(uint32_t timeout)
+static int os_host_main_loop_wait(int64_t timeout)
 {
     int ret;
     static int spin_counter;
@@ -210,7 +215,7 @@ static int os_host_main_loop_wait(uint32_t timeout)
             notified = true;
         }
 
-        timeout = 1;
+        timeout = SCALE_MS;
     }
 
     if (timeout > 0) {
@@ -220,7 +225,7 @@ static int os_host_main_loop_wait(uint32_t timeout)
         spin_counter++;
     }
 
-    ret = g_poll((GPollFD *)gpollfds->data, gpollfds->len, timeout);
+    ret = qemu_g_poll_ns((GPollFD *)gpollfds->data, gpollfds->len, timeout);
 
     if (timeout > 0) {
         qemu_mutex_lock_iothread();
@@ -369,7 +374,7 @@ static void pollfds_poll(GArray *pollfds, int nfds, fd_set 
*rfds,
     }
 }
 
-static int os_host_main_loop_wait(uint32_t timeout)
+static int os_host_main_loop_wait(int64_t timeout)
 {
     GMainContext *context = g_main_context_default();
     GPollFD poll_fds[1024 * 2]; /* this is probably overkill */
@@ -378,6 +383,7 @@ static int os_host_main_loop_wait(uint32_t timeout)
     PollingEntry *pe;
     WaitObjects *w = &wait_objects;
     gint poll_timeout;
+    int64_t poll_timeout_ns;
     static struct timeval tv0;
     fd_set rfds, wfds, xfds;
     int nfds;
@@ -415,12 +421,17 @@ static int os_host_main_loop_wait(uint32_t timeout)
         poll_fds[n_poll_fds + i].events = G_IO_IN;
     }
 
-    if (poll_timeout < 0 || timeout < poll_timeout) {
-        poll_timeout = timeout;
+    if (poll_timeout < 0) {
+        poll_timeout_ns = -1;
+    } else {
+        poll_timeout_ns = (int64_t)poll_timeout * (int64_t)SCALE_MS;
     }
 
+    poll_timeout_ns = qemu_soonest_timeout(poll_timeout_ns, timeout);
+
     qemu_mutex_unlock_iothread();
-    g_poll_ret = g_poll(poll_fds, n_poll_fds + w->num, poll_timeout);
+    g_poll_ret = qemu_g_poll_ns(poll_fds, n_poll_fds + w->num, 
poll_timeout_ns);
+
     qemu_mutex_lock_iothread();
     if (g_poll_ret > 0) {
         for (i = 0; i < w->num; i++) {
@@ -445,6 +456,7 @@ int main_loop_wait(int nonblocking)
 {
     int ret;
     uint32_t timeout = UINT32_MAX;
+    int64_t timeout_ns;
 
     if (nonblocking) {
         timeout = 0;
@@ -458,7 +470,16 @@ int main_loop_wait(int nonblocking)
     slirp_pollfds_fill(gpollfds);
 #endif
     qemu_iohandler_fill(gpollfds);
-    ret = os_host_main_loop_wait(timeout);
+
+    if (timeout == UINT32_MAX) {
+        timeout_ns = -1;
+    } else {
+        timeout_ns = (uint64_t)timeout * (int64_t)(SCALE_MS);
+    }
+
+    timeout_ns = qemu_soonest_timeout(timeout_ns, 
qemu_clock_deadline_all_ns());
+
+    ret = os_host_main_loop_wait(timeout_ns);
     qemu_iohandler_poll(gpollfds, ret);
 #ifdef CONFIG_SLIRP
     slirp_pollfds_poll(gpollfds, (ret < 0));
-- 
1.7.9.5