[Qemu-devel] [PATCH 0/3] aio-posix: polling mode bug fixes

[Qemu-devel] [PATCH 0/3] aio-posix: polling mode bug fixes

[Paolo Bonzini]

Patch 1 fixes a too-strict assertion that could fire when aio_poll
is called in parallel with aio_set_fd_handler.

Patch 2 and 3 reinstate the performance benefits of polling, which were
essentially disabled by commit 70232b5253 ("aio-posix: Don't count
ctx->notifier as progress when polling", 2018-08-15).

Please review carefully! :)

Thanks,

Paolo

Paolo Bonzini (3):
  aio-posix: fix concurrent access to poll_disable_cnt
  aio-posix: compute timeout before polling
  aio-posix: do skip system call if ctx->notifier polling succeeds

 util/aio-posix.c  | 88 +++++++++++++++++++++++++++--------------------
 util/trace-events |  4 +--
 2 files changed, 53 insertions(+), 39 deletions(-)

-- 
2.17.1

[Qemu-devel] [PATCH 1/3] aio-posix: fix concurrent access to poll_disable_cnt

[Paolo Bonzini]

It is valid for an aio_set_fd_handler to happen concurrently with
aio_poll.  In that case, poll_disable_cnt can change under the heels
of aio_poll, and the assertion on poll_disable_cnt can fail in
run_poll_handlers.

Therefore, this patch simply checks the counter on every polling
iteration.  There are no particular needs for ordering, since the
polling loop is terminated anyway by aio_notify at the end of
aio_set_fd_handler.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 util/aio-posix.c | 26 +++++++++++++++-----------
 1 file changed, 15 insertions(+), 11 deletions(-)

diff --git a/util/aio-posix.c b/util/aio-posix.c
index 131ba6b4a8..5c29380575 100644
--- a/util/aio-posix.c
+++ b/util/aio-posix.c
@@ -211,6 +211,7 @@ void aio_set_fd_handler(AioContext *ctx,
     AioHandler *node;
     bool is_new = false;
     bool deleted = false;
+    int poll_disable_change;
 
     qemu_lockcnt_lock(&ctx->list_lock);
 
@@ -244,11 +245,9 @@ void aio_set_fd_handler(AioContext *ctx,
             QLIST_REMOVE(node, node);
             deleted = true;
         }
-
-        if (!node->io_poll) {
-            ctx->poll_disable_cnt--;
-        }
+        poll_disable_change = -!node->io_poll;
     } else {
+        poll_disable_change = !io_poll - (node && !node->io_poll);
         if (node == NULL) {
             /* Alloc and insert if it's not already there */
             node = g_new0(AioHandler, 1);
@@ -257,10 +256,6 @@ void aio_set_fd_handler(AioContext *ctx,
 
             g_source_add_poll(&ctx->source, &node->pfd);
             is_new = true;
-
-            ctx->poll_disable_cnt += !io_poll;
-        } else {
-            ctx->poll_disable_cnt += !io_poll - !node->io_poll;
         }
 
         /* Update handler with latest information */
@@ -274,6 +269,15 @@ void aio_set_fd_handler(AioContext *ctx,
         node->pfd.events |= (io_write ? G_IO_OUT | G_IO_ERR : 0);
     }
 
+    /* No need to order poll_disable_cnt writes against other updates;
+     * the counter is only used to avoid wasting time and latency on
+     * iterated polling when the system call will be ultimately necessary.
+     * Changing handlers is a rare event, and a little wasted polling until
+     * the aio_notify below is not an issue.
+     */
+    atomic_set(&ctx->poll_disable_cnt,
+               atomic_read(&ctx->poll_disable_cnt) + poll_disable_change);
+
     aio_epoll_update(ctx, node, is_new);
     qemu_lockcnt_unlock(&ctx->list_lock);
     aio_notify(ctx);
@@ -525,7 +529,6 @@ static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
 
     assert(ctx->notify_me);
     assert(qemu_lockcnt_count(&ctx->list_lock) > 0);
-    assert(ctx->poll_disable_cnt == 0);
 
     trace_run_poll_handlers_begin(ctx, max_ns);
 
@@ -533,7 +536,8 @@ static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
 
     do {
         progress = run_poll_handlers_once(ctx);
-    } while (!progress && qemu_clock_get_ns(QEMU_CLOCK_REALTIME) < end_time);
+    } while (!progress && qemu_clock_get_ns(QEMU_CLOCK_REALTIME) < end_time
+             && !atomic_read(&ctx->poll_disable_cnt));
 
     trace_run_poll_handlers_end(ctx, progress);
 
@@ -552,7 +556,7 @@ static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
  */
 static bool try_poll_mode(AioContext *ctx, bool blocking)
 {
-    if (blocking && ctx->poll_max_ns && ctx->poll_disable_cnt == 0) {
+    if (blocking && ctx->poll_max_ns && !atomic_read(&ctx->poll_disable_cnt)) {
         /* See qemu_soonest_timeout() uint64_t hack */
         int64_t max_ns = MIN((uint64_t)aio_compute_timeout(ctx),
                              (uint64_t)ctx->poll_ns);
-- 
2.17.1

[Qemu-devel] [PATCH 2/3] aio-posix: compute timeout before polling

[Paolo Bonzini]

This is a preparation for the next patch, and also a very small
optimization.  Compute the timeout only once, before invoking
try_poll_mode, and adjust it in run_poll_handlers.  The adjustment
is the polling time when polling fails, or zero (non-blocking) if
polling succeeds.

Fixes: 70232b5253a3c4e03ed1ac47ef9246a8ac66c6fa
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 util/aio-posix.c  | 59 +++++++++++++++++++++++++++--------------------
 util/trace-events |  4 ++--
 2 files changed, 36 insertions(+), 27 deletions(-)

diff --git a/util/aio-posix.c b/util/aio-posix.c
index 5c29380575..21d8987179 100644
--- a/util/aio-posix.c
+++ b/util/aio-posix.c
@@ -490,7 +490,7 @@ static void add_pollfd(AioHandler *node)
     npfd++;
 }
 
-static bool run_poll_handlers_once(AioContext *ctx)
+static bool run_poll_handlers_once(AioContext *ctx, int64_t *timeout)
 {
     bool progress = false;
     AioHandler *node;
@@ -500,6 +500,7 @@ static bool run_poll_handlers_once(AioContext *ctx)
             aio_node_check(ctx, node->is_external) &&
             node->io_poll(node->opaque) &&
             node->opaque != &ctx->notifier) {
+            *timeout = 0;
             progress = true;
         }
 
@@ -522,31 +523,38 @@ static bool run_poll_handlers_once(AioContext *ctx)
  *
  * Returns: true if progress was made, false otherwise
  */
-static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
+static bool run_poll_handlers(AioContext *ctx, int64_t max_ns, int64_t *timeout)
 {
     bool progress;
-    int64_t end_time;
+    int64_t start_time, elapsed_time;
 
     assert(ctx->notify_me);
     assert(qemu_lockcnt_count(&ctx->list_lock) > 0);
 
-    trace_run_poll_handlers_begin(ctx, max_ns);
-
-    end_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + max_ns;
+    trace_run_poll_handlers_begin(ctx, max_ns, *timeout);
 
+    start_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
     do {
-        progress = run_poll_handlers_once(ctx);
-    } while (!progress && qemu_clock_get_ns(QEMU_CLOCK_REALTIME) < end_time
+        progress = run_poll_handlers_once(ctx, timeout);
+        elapsed_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time;
+    } while (!progress && elapsed_time < max_ns
              && !atomic_read(&ctx->poll_disable_cnt));
 
-    trace_run_poll_handlers_end(ctx, progress);
+    /* If time has passed with no successful polling, adjust *timeout to
+     * keep the same ending time.
+     */
+    if (*timeout != -1) {
+        *timeout -= MIN(*timeout, elapsed_time);
+    }
 
+    trace_run_poll_handlers_end(ctx, progress, *timeout);
     return progress;
 }
 
 /* try_poll_mode:
  * @ctx: the AioContext
- * @blocking: busy polling is only attempted when blocking is true
+ * @timeout: timeout for blocking wait, computed by the caller and updated if
+ *    polling succeeds.
  *
  * ctx->notify_me must be non-zero so this function can detect aio_notify().
  *
@@ -554,19 +562,16 @@ static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
  *
  * Returns: true if progress was made, false otherwise
  */
-static bool try_poll_mode(AioContext *ctx, bool blocking)
+static bool try_poll_mode(AioContext *ctx, int64_t *timeout)
 {
-    if (blocking && ctx->poll_max_ns && !atomic_read(&ctx->poll_disable_cnt)) {
-        /* See qemu_soonest_timeout() uint64_t hack */
-        int64_t max_ns = MIN((uint64_t)aio_compute_timeout(ctx),
-                             (uint64_t)ctx->poll_ns);
+    /* See qemu_soonest_timeout() uint64_t hack */
+    int64_t max_ns = MIN((uint64_t)*timeout, (uint64_t)ctx->poll_ns);
 
-        if (max_ns) {
-            poll_set_started(ctx, true);
+    if (max_ns && !atomic_read(&ctx->poll_disable_cnt)) {
+        poll_set_started(ctx, true);
 
-            if (run_poll_handlers(ctx, max_ns)) {
-                return true;
-            }
+        if (run_poll_handlers(ctx, max_ns, timeout)) {
+            return true;
         }
     }
 
@@ -575,7 +580,7 @@ static bool try_poll_mode(AioContext *ctx, bool blocking)
     /* Even if we don't run busy polling, try polling once in case it can make
      * progress and the caller will be able to avoid ppoll(2)/epoll_wait(2).
      */
-    return run_poll_handlers_once(ctx);
+    return run_poll_handlers_once(ctx, timeout);
 }
 
 bool aio_poll(AioContext *ctx, bool blocking)
@@ -605,8 +610,14 @@ bool aio_poll(AioContext *ctx, bool blocking)
         start = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
     }
 
-    progress = try_poll_mode(ctx, blocking);
-    if (!progress) {
+    timeout = blocking ? aio_compute_timeout(ctx) : 0;
+    progress = try_poll_mode(ctx, &timeout);
+    assert(!(timeout && progress));
+
+    /* If polling is allowed, non-blocking aio_poll does not need the
+     * system call---a single round of run_poll_handlers_once suffices.
+     */
+    if (timeout || atomic_read(&ctx->poll_disable_cnt)) {
         assert(npfd == 0);
 
         /* fill pollfds */
@@ -620,8 +631,6 @@ bool aio_poll(AioContext *ctx, bool blocking)
             }
         }
 
-        timeout = blocking ? aio_compute_timeout(ctx) : 0;
-
         /* wait until next event */
         if (aio_epoll_check_poll(ctx, pollfds, npfd, timeout)) {
             AioHandler epoll_handler;
diff --git a/util/trace-events b/util/trace-events
index 4822434c89..79569b7fdf 100644
--- a/util/trace-events
+++ b/util/trace-events
@@ -1,8 +1,8 @@
 # See docs/devel/tracing.txt for syntax documentation.
 
 # util/aio-posix.c
-run_poll_handlers_begin(void *ctx, int64_t max_ns) "ctx %p max_ns %"PRId64
-run_poll_handlers_end(void *ctx, bool progress) "ctx %p progress %d"
+run_poll_handlers_begin(void *ctx, int64_t max_ns, int64_t timeout) "ctx %p max_ns %"PRId64 " timeout %"PRId64
+run_poll_handlers_end(void *ctx, bool progress, int64_t timeout) "ctx %p progress %d new timeout %"PRId64
 poll_shrink(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64
 poll_grow(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64
 
-- 
2.17.1

[Qemu-devel] [PATCH 3/3] aio-posix: do skip system call if ctx->notifier polling succeeds

[Paolo Bonzini]

Commit 70232b5253 ("aio-posix: Don't count ctx->notifier as progress when
2018-08-15), by not reporting progress, causes aio_poll to execute the
system call when polling succeeds because of ctx->notifier.  This introduces
latency before the call to aio_bh_poll() and negates the advantages of
polling, unfortunately.

The fix builds on the previous patch, separating the effect of polling on
the timeout from the progress reported to aio_poll().  ctx->notifier
does zero the timeout, causing the caller to skip the system call,
but it does not report progress, so that the bug fix of commit 70232b5253
still stands.

Fixes: 70232b5253a3c4e03ed1ac47ef9246a8ac66c6fa
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 util/aio-posix.c | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/util/aio-posix.c b/util/aio-posix.c
index 21d8987179..efa79b9e62 100644
--- a/util/aio-posix.c
+++ b/util/aio-posix.c
@@ -498,10 +498,11 @@ static bool run_poll_handlers_once(AioContext *ctx, int64_t *timeout)
     QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
         if (!node->deleted && node->io_poll &&
             aio_node_check(ctx, node->is_external) &&
-            node->io_poll(node->opaque) &&
-            node->opaque != &ctx->notifier) {
+            node->io_poll(node->opaque)) {
             *timeout = 0;
-            progress = true;
+            if (node->opaque != &ctx->notifier) {
+                progress = true;
+            }
         }
 
         /* Caller handles freeing deleted nodes.  Don't do it here. */
-- 
2.17.1

Re: [Qemu-devel] [PATCH 1/3] aio-posix: fix concurrent access to poll_disable_cnt

[Fam Zheng]

On Wed, 09/12 19:10, Paolo Bonzini wrote:
> It is valid for an aio_set_fd_handler to happen concurrently with
> aio_poll.  In that case, poll_disable_cnt can change under the heels
> of aio_poll, and the assertion on poll_disable_cnt can fail in
> run_poll_handlers.
> 
> Therefore, this patch simply checks the counter on every polling
> iteration.  There are no particular needs for ordering, since the
> polling loop is terminated anyway by aio_notify at the end of
> aio_set_fd_handler.
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  util/aio-posix.c | 26 +++++++++++++++-----------
>  1 file changed, 15 insertions(+), 11 deletions(-)
> 
> diff --git a/util/aio-posix.c b/util/aio-posix.c
> index 131ba6b4a8..5c29380575 100644
> --- a/util/aio-posix.c
> +++ b/util/aio-posix.c
> @@ -211,6 +211,7 @@ void aio_set_fd_handler(AioContext *ctx,
>      AioHandler *node;
>      bool is_new = false;
>      bool deleted = false;
> +    int poll_disable_change;
>  
>      qemu_lockcnt_lock(&ctx->list_lock);
>  
> @@ -244,11 +245,9 @@ void aio_set_fd_handler(AioContext *ctx,
>              QLIST_REMOVE(node, node);
>              deleted = true;
>          }
> -
> -        if (!node->io_poll) {
> -            ctx->poll_disable_cnt--;
> -        }
> +        poll_disable_change = -!node->io_poll;
>      } else {
> +        poll_disable_change = !io_poll - (node && !node->io_poll);
>          if (node == NULL) {
>              /* Alloc and insert if it's not already there */
>              node = g_new0(AioHandler, 1);
> @@ -257,10 +256,6 @@ void aio_set_fd_handler(AioContext *ctx,
>  
>              g_source_add_poll(&ctx->source, &node->pfd);
>              is_new = true;
> -
> -            ctx->poll_disable_cnt += !io_poll;
> -        } else {
> -            ctx->poll_disable_cnt += !io_poll - !node->io_poll;
>          }
>  
>          /* Update handler with latest information */
> @@ -274,6 +269,15 @@ void aio_set_fd_handler(AioContext *ctx,
>          node->pfd.events |= (io_write ? G_IO_OUT | G_IO_ERR : 0);
>      }
>  
> +    /* No need to order poll_disable_cnt writes against other updates;
> +     * the counter is only used to avoid wasting time and latency on
> +     * iterated polling when the system call will be ultimately necessary.
> +     * Changing handlers is a rare event, and a little wasted polling until
> +     * the aio_notify below is not an issue.
> +     */
> +    atomic_set(&ctx->poll_disable_cnt,
> +               atomic_read(&ctx->poll_disable_cnt) + poll_disable_change);

Why not atomic_add?

> +
>      aio_epoll_update(ctx, node, is_new);
>      qemu_lockcnt_unlock(&ctx->list_lock);
>      aio_notify(ctx);
> @@ -525,7 +529,6 @@ static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
>  
>      assert(ctx->notify_me);
>      assert(qemu_lockcnt_count(&ctx->list_lock) > 0);
> -    assert(ctx->poll_disable_cnt == 0);
>  
>      trace_run_poll_handlers_begin(ctx, max_ns);
>  
> @@ -533,7 +536,8 @@ static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
>  
>      do {
>          progress = run_poll_handlers_once(ctx);
> -    } while (!progress && qemu_clock_get_ns(QEMU_CLOCK_REALTIME) < end_time);
> +    } while (!progress && qemu_clock_get_ns(QEMU_CLOCK_REALTIME) < end_time
> +             && !atomic_read(&ctx->poll_disable_cnt));
>  
>      trace_run_poll_handlers_end(ctx, progress);
>  
> @@ -552,7 +556,7 @@ static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
>   */
>  static bool try_poll_mode(AioContext *ctx, bool blocking)
>  {
> -    if (blocking && ctx->poll_max_ns && ctx->poll_disable_cnt == 0) {
> +    if (blocking && ctx->poll_max_ns && !atomic_read(&ctx->poll_disable_cnt)) {
>          /* See qemu_soonest_timeout() uint64_t hack */
>          int64_t max_ns = MIN((uint64_t)aio_compute_timeout(ctx),
>                               (uint64_t)ctx->poll_ns);
> -- 
> 2.17.1
> 
> 

Reviewed-by: Fam Zheng <famz@redhat.com>

Re: [Qemu-devel] [PATCH 2/3] aio-posix: compute timeout before polling

[Fam Zheng]

On Wed, 09/12 19:10, Paolo Bonzini wrote:
> This is a preparation for the next patch, and also a very small
> optimization.  Compute the timeout only once, before invoking
> try_poll_mode, and adjust it in run_poll_handlers.  The adjustment
> is the polling time when polling fails, or zero (non-blocking) if
> polling succeeds.
> 
> Fixes: 70232b5253a3c4e03ed1ac47ef9246a8ac66c6fa
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  util/aio-posix.c  | 59 +++++++++++++++++++++++++++--------------------
>  util/trace-events |  4 ++--
>  2 files changed, 36 insertions(+), 27 deletions(-)
> 
> diff --git a/util/aio-posix.c b/util/aio-posix.c
> index 5c29380575..21d8987179 100644
> --- a/util/aio-posix.c
> +++ b/util/aio-posix.c
> @@ -490,7 +490,7 @@ static void add_pollfd(AioHandler *node)
>      npfd++;
>  }
>  
> -static bool run_poll_handlers_once(AioContext *ctx)
> +static bool run_poll_handlers_once(AioContext *ctx, int64_t *timeout)
>  {
>      bool progress = false;
>      AioHandler *node;
> @@ -500,6 +500,7 @@ static bool run_poll_handlers_once(AioContext *ctx)
>              aio_node_check(ctx, node->is_external) &&
>              node->io_poll(node->opaque) &&
>              node->opaque != &ctx->notifier) {
> +            *timeout = 0;
>              progress = true;
>          }
>  
> @@ -522,31 +523,38 @@ static bool run_poll_handlers_once(AioContext *ctx)
>   *
>   * Returns: true if progress was made, false otherwise
>   */
> -static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
> +static bool run_poll_handlers(AioContext *ctx, int64_t max_ns, int64_t *timeout)
>  {
>      bool progress;
> -    int64_t end_time;
> +    int64_t start_time, elapsed_time;
>  
>      assert(ctx->notify_me);
>      assert(qemu_lockcnt_count(&ctx->list_lock) > 0);
>  
> -    trace_run_poll_handlers_begin(ctx, max_ns);
> -
> -    end_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + max_ns;
> +    trace_run_poll_handlers_begin(ctx, max_ns, *timeout);
>  
> +    start_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
>      do {
> -        progress = run_poll_handlers_once(ctx);
> -    } while (!progress && qemu_clock_get_ns(QEMU_CLOCK_REALTIME) < end_time
> +        progress = run_poll_handlers_once(ctx, timeout);
> +        elapsed_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time;
> +    } while (!progress && elapsed_time < max_ns
>               && !atomic_read(&ctx->poll_disable_cnt));
>  
> -    trace_run_poll_handlers_end(ctx, progress);
> +    /* If time has passed with no successful polling, adjust *timeout to
> +     * keep the same ending time.
> +     */
> +    if (*timeout != -1) {
> +        *timeout -= MIN(*timeout, elapsed_time);
> +    }
>  
> +    trace_run_poll_handlers_end(ctx, progress, *timeout);
>      return progress;
>  }
>  
>  /* try_poll_mode:
>   * @ctx: the AioContext
> - * @blocking: busy polling is only attempted when blocking is true
> + * @timeout: timeout for blocking wait, computed by the caller and updated if
> + *    polling succeeds.
>   *
>   * ctx->notify_me must be non-zero so this function can detect aio_notify().
>   *
> @@ -554,19 +562,16 @@ static bool run_poll_handlers(AioContext *ctx, int64_t max_ns)
>   *
>   * Returns: true if progress was made, false otherwise
>   */
> -static bool try_poll_mode(AioContext *ctx, bool blocking)
> +static bool try_poll_mode(AioContext *ctx, int64_t *timeout)
>  {
> -    if (blocking && ctx->poll_max_ns && !atomic_read(&ctx->poll_disable_cnt)) {
> -        /* See qemu_soonest_timeout() uint64_t hack */
> -        int64_t max_ns = MIN((uint64_t)aio_compute_timeout(ctx),
> -                             (uint64_t)ctx->poll_ns);
> +    /* See qemu_soonest_timeout() uint64_t hack */
> +    int64_t max_ns = MIN((uint64_t)*timeout, (uint64_t)ctx->poll_ns);
>  
> -        if (max_ns) {
> -            poll_set_started(ctx, true);
> +    if (max_ns && !atomic_read(&ctx->poll_disable_cnt)) {
> +        poll_set_started(ctx, true);
>  
> -            if (run_poll_handlers(ctx, max_ns)) {
> -                return true;
> -            }
> +        if (run_poll_handlers(ctx, max_ns, timeout)) {
> +            return true;
>          }
>      }
>  
> @@ -575,7 +580,7 @@ static bool try_poll_mode(AioContext *ctx, bool blocking)
>      /* Even if we don't run busy polling, try polling once in case it can make
>       * progress and the caller will be able to avoid ppoll(2)/epoll_wait(2).
>       */
> -    return run_poll_handlers_once(ctx);
> +    return run_poll_handlers_once(ctx, timeout);
>  }
>  
>  bool aio_poll(AioContext *ctx, bool blocking)
> @@ -605,8 +610,14 @@ bool aio_poll(AioContext *ctx, bool blocking)
>          start = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
>      }
>  
> -    progress = try_poll_mode(ctx, blocking);
> -    if (!progress) {
> +    timeout = blocking ? aio_compute_timeout(ctx) : 0;
> +    progress = try_poll_mode(ctx, &timeout);
> +    assert(!(timeout && progress));
> +
> +    /* If polling is allowed, non-blocking aio_poll does not need the
> +     * system call---a single round of run_poll_handlers_once suffices.
> +     */
> +    if (timeout || atomic_read(&ctx->poll_disable_cnt)) {
>          assert(npfd == 0);
>  
>          /* fill pollfds */
> @@ -620,8 +631,6 @@ bool aio_poll(AioContext *ctx, bool blocking)
>              }
>          }
>  
> -        timeout = blocking ? aio_compute_timeout(ctx) : 0;
> -
>          /* wait until next event */
>          if (aio_epoll_check_poll(ctx, pollfds, npfd, timeout)) {
>              AioHandler epoll_handler;
> diff --git a/util/trace-events b/util/trace-events
> index 4822434c89..79569b7fdf 100644
> --- a/util/trace-events
> +++ b/util/trace-events
> @@ -1,8 +1,8 @@
>  # See docs/devel/tracing.txt for syntax documentation.
>  
>  # util/aio-posix.c
> -run_poll_handlers_begin(void *ctx, int64_t max_ns) "ctx %p max_ns %"PRId64
> -run_poll_handlers_end(void *ctx, bool progress) "ctx %p progress %d"
> +run_poll_handlers_begin(void *ctx, int64_t max_ns, int64_t timeout) "ctx %p max_ns %"PRId64 " timeout %"PRId64
> +run_poll_handlers_end(void *ctx, bool progress, int64_t timeout) "ctx %p progress %d new timeout %"PRId64
>  poll_shrink(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64
>  poll_grow(void *ctx, int64_t old, int64_t new) "ctx %p old %"PRId64" new %"PRId64
>  
> -- 
> 2.17.1
> 
> 

Reviewed-by: Fam Zheng <famz@redhat.com>

Re: [Qemu-devel] [PATCH 3/3] aio-posix: do skip system call if ctx->notifier polling succeeds

[Fam Zheng]

On Wed, 09/12 19:10, Paolo Bonzini wrote:
> Commit 70232b5253 ("aio-posix: Don't count ctx->notifier as progress when
> 2018-08-15), by not reporting progress, causes aio_poll to execute the
> system call when polling succeeds because of ctx->notifier.  This introduces
> latency before the call to aio_bh_poll() and negates the advantages of
> polling, unfortunately.
> 
> The fix builds on the previous patch, separating the effect of polling on
> the timeout from the progress reported to aio_poll().  ctx->notifier
> does zero the timeout, causing the caller to skip the system call,
> but it does not report progress, so that the bug fix of commit 70232b5253
> still stands.
> 
> Fixes: 70232b5253a3c4e03ed1ac47ef9246a8ac66c6fa
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  util/aio-posix.c | 7 ++++---
>  1 file changed, 4 insertions(+), 3 deletions(-)
> 
> diff --git a/util/aio-posix.c b/util/aio-posix.c
> index 21d8987179..efa79b9e62 100644
> --- a/util/aio-posix.c
> +++ b/util/aio-posix.c
> @@ -498,10 +498,11 @@ static bool run_poll_handlers_once(AioContext *ctx, int64_t *timeout)
>      QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
>          if (!node->deleted && node->io_poll &&
>              aio_node_check(ctx, node->is_external) &&
> -            node->io_poll(node->opaque) &&
> -            node->opaque != &ctx->notifier) {
> +            node->io_poll(node->opaque)) {
>              *timeout = 0;
> -            progress = true;
> +            if (node->opaque != &ctx->notifier) {
> +                progress = true;
> +            }
>          }
>  
>          /* Caller handles freeing deleted nodes.  Don't do it here. */
> -- 
> 2.17.1
> 

Reviewed-by: Fam Zheng <famz@redhat.com>

Re: [Qemu-devel] [Qemu-block] [PATCH 1/3] aio-posix: fix concurrent access to poll_disable_cnt

[Paolo Bonzini]

On 13/09/2018 08:56, Fam Zheng wrote:
>> +    /* No need to order poll_disable_cnt writes against other updates;
>> +     * the counter is only used to avoid wasting time and latency on
>> +     * iterated polling when the system call will be ultimately necessary.
>> +     * Changing handlers is a rare event, and a little wasted polling until
>> +     * the aio_notify below is not an issue.
>> +     */
>> +    atomic_set(&ctx->poll_disable_cnt,
>> +               atomic_read(&ctx->poll_disable_cnt) + poll_disable_change);
>
> Why not atomic_add?

This is not lockless, it's protected by list_lock, so there's no race
condition involved.  I'm just mimicking what is done for other similar
cases, for example involving seqlocks.

The alternative would be to add a full set of
atomic_{add,sub,...}_relaxed atomics.

Paolo

Re: [Qemu-devel] [Qemu-block] [PATCH 1/3] aio-posix: fix concurrent access to poll_disable_cnt

[Fam Zheng]

On Thu, 09/13 10:29, Paolo Bonzini wrote:
> On 13/09/2018 08:56, Fam Zheng wrote:
> >> +    /* No need to order poll_disable_cnt writes against other updates;
> >> +     * the counter is only used to avoid wasting time and latency on
> >> +     * iterated polling when the system call will be ultimately necessary.
> >> +     * Changing handlers is a rare event, and a little wasted polling until
> >> +     * the aio_notify below is not an issue.
> >> +     */
> >> +    atomic_set(&ctx->poll_disable_cnt,
> >> +               atomic_read(&ctx->poll_disable_cnt) + poll_disable_change);
> >
> > Why not atomic_add?
> 
> This is not lockless, it's protected by list_lock, so there's no race
> condition involved.  I'm just mimicking what is done for other similar
> cases, for example involving seqlocks.

Yeah, no doubt about the correctness. Just curious about the preference since
atomic_add is less typing and one less atomic_* op.

Fam

> 
> The alternative would be to add a full set of
> atomic_{add,sub,...}_relaxed atomics.
> 
> Paolo