Series comparison

-[PULL 0/9] Block patches
+[PULL 0/8] Block patches
-The following changes since commit 554623226f800acf48a2ed568900c1c968ec9a8b:
+The following changes since commit c6a5fc2ac76c5ab709896ee1b0edd33685a67ed1:
-  Merge tag 'qemu-sparc-20220508' of https://github.com/mcayland/qemu into staging (2022-05-08 17:03:26 -0500)
+  decodetree: Add --output-null for meson testing (2023-05-31 19:56:42 -0700)
 are available in the Git repository at:
   https://gitlab.com/stefanha/qemu.git tags/block-pull-request
-for you to fetch changes up to 3dc584abeef0e1277c2de8c1c1974cb49444eb0a:
+for you to fetch changes up to 98b126f5e3228a346c774e569e26689943b401dd:
-  virtio-scsi: move request-related items from .h to .c (2022-05-09 10:45:04 +0100)
+  qapi: add '@fdset' feature for BlockdevOptionsVirtioBlkVhostVdpa (2023-06-01 11:08:21 -0400)
 ----------------------------------------------------------------
 Pull request
-- Add new thread-pool-min/thread-pool-max parameters to control the thread pool
+- Stefano Garzarella's blkio block driver 'fd' parameter
-  used for async I/O.
+- My thread-local blk_io_plug() series
 - Fix virtio-scsi IOThread 100% CPU consumption QEMU 7.0 regression.
 ----------------------------------------------------------------
-Nicolas Saenz Julienne (3):
+Stefan Hajnoczi (6):
-  Introduce event-loop-base abstract class
+  block: add blk_io_plug_call() API
-  util/main-loop: Introduce the main loop into QOM
+  block/nvme: convert to blk_io_plug_call() API
-  util/event-loop-base: Introduce options to set the thread pool size
+  block/blkio: convert to blk_io_plug_call() API
   block/io_uring: convert to blk_io_plug_call() API
   block/linux-aio: convert to blk_io_plug_call() API
   block: remove bdrv_co_io_plug() API
-Stefan Hajnoczi (6):
+Stefano Garzarella (2):
-  virtio-scsi: fix ctrl and event handler functions in dataplane mode
+  block/blkio: use qemu_open() to support fd passing for virtio-blk
-  virtio-scsi: don't waste CPU polling the event virtqueue
+  qapi: add '@fdset' feature for BlockdevOptionsVirtioBlkVhostVdpa
   virtio-scsi: clean up virtio_scsi_handle_event_vq()
   virtio-scsi: clean up virtio_scsi_handle_ctrl_vq()
   virtio-scsi: clean up virtio_scsi_handle_cmd_vq()
   virtio-scsi: move request-related items from .h to .c
- qapi/qom.json                    |  43 ++++++++--
+ MAINTAINERS                       |   1 +
- meson.build                      |  26 +++---
+ qapi/block-core.json              |   6 ++
- include/block/aio.h              |  10 +++
+ meson.build                       |   4 +
- include/block/thread-pool.h      |   3 +
+ include/block/block-io.h          |   3 -
- include/hw/virtio/virtio-scsi.h  |  43 ----------
+ include/block/block_int-common.h  |  11 ---
- include/hw/virtio/virtio.h       |   1 +
+ include/block/raw-aio.h           |  14 ---
- include/qemu/main-loop.h         |  10 +++
+ include/sysemu/block-backend-io.h |  13 +--
- include/sysemu/event-loop-base.h |  41 +++++++++
+ block/blkio.c                     |  96 ++++++++++++------
- include/sysemu/iothread.h        |   6 +-
+ block/block-backend.c             |  22 -----
- event-loop-base.c                | 140 +++++++++++++++++++++++++++++++
+ block/file-posix.c                |  38 -------
- hw/scsi/virtio-scsi-dataplane.c  |   2 +-
+ block/io.c                        |  37 -------
- hw/scsi/virtio-scsi.c            | 101 +++++++++++++++-------
+ block/io_uring.c                  |  44 ++++-----
- hw/virtio/virtio.c               |  13 +++
+ block/linux-aio.c                 |  41 +++-----
- iothread.c                       |  68 +++++----------
+ block/nvme.c                      |  44 +++------
- util/aio-posix.c                 |   1 +
+ block/plug.c                      | 159 ++++++++++++++++++++++++++++++
- util/async.c                     |  20 +++++
+ hw/block/dataplane/xen-block.c    |   8 +-
- util/main-loop.c                 |  65 ++++++++++++++
+ hw/block/virtio-blk.c             |   4 +-
- util/thread-pool.c               |  55 +++++++++++-
+ hw/scsi/virtio-scsi.c             |   6 +-
-files changed, 505 insertions(+), 143 deletions(-)
+ block/meson.build                 |   1 +
- create mode 100644 include/sysemu/event-loop-base.h
+ block/trace-events                |   6 +-
- create mode 100644 event-loop-base.c
+files changed, 293 insertions(+), 265 deletions(-)
  create mode 100644 block/plug.c
 --
-.35.1
+.40.1

-[PULL 8/9] virtio-scsi: clean up virtio_scsi_handle_cmd_vq()
+[PULL 1/8] block: add blk_io_plug_call() API
-virtio_scsi_handle_cmd_vq() is only called from hw/scsi/virtio-scsi.c
+Introduce a new API for thread-local blk_io_plug() that does not
-now and its return value is no longer used. Remove the function
+traverse the block graph. The goal is to make blk_io_plug() multi-queue
-prototype from virtio-scsi.h and drop the return value.
+friendly.
 Instead of having block drivers track whether or not we're in a plugged
 section, provide an API that allows them to defer a function call until
 we're unplugged: blk_io_plug_call(fn, opaque). If blk_io_plug_call() is
 called multiple times with the same fn/opaque pair, then fn() is only
 called once at the end of the function - resulting in batching.
 This patch introduces the API and changes blk_io_plug()/blk_io_unplug().
 blk_io_plug()/blk_io_unplug() no longer require a BlockBackend argument
 because the plug state is now thread-local.
 Later patches convert block drivers to blk_io_plug_call() and then we
 can finally remove .bdrv_co_io_plug() once all block drivers have been
 converted.
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
-Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
+Reviewed-by: Eric Blake <eblake@redhat.com>
-Message-id: 20220427143541.119567-6-stefanha@redhat.com
+Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
 Acked-by: Kevin Wolf <kwolf@redhat.com>
 Message-id: 20230530180959.1108766-2-stefanha@redhat.com
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
 ---
- include/hw/virtio/virtio-scsi.h | 1 -
+ MAINTAINERS                       |   1 +
- hw/scsi/virtio-scsi.c           | 5 +----
+ include/sysemu/block-backend-io.h |  13 +--
-files changed, 1 insertion(+), 5 deletions(-)
+ block/block-backend.c             |  22 -----
+ block/plug.c                      | 159 ++++++++++++++++++++++++++++++
-diff --git a/include/hw/virtio/virtio-scsi.h b/include/hw/virtio/virtio-scsi.h
+ hw/block/dataplane/xen-block.c    |   8 +-
-index XXXXXXX..XXXXXXX 100644
+ hw/block/virtio-blk.c             |   4 +-
---- a/include/hw/virtio/virtio-scsi.h
+ hw/scsi/virtio-scsi.c             |   6 +-
-+++ b/include/hw/virtio/virtio-scsi.h
+ block/meson.build                 |   1 +
-@@ -XXX,XX +XXX,XX @@ void virtio_scsi_common_realize(DeviceState *dev,
+files changed, 173 insertions(+), 41 deletions(-)
-                                 Error **errp);
+ create mode 100644 block/plug.c
- void virtio_scsi_common_unrealize(DeviceState *dev);
+diff --git a/MAINTAINERS b/MAINTAINERS
--bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq);
+index XXXXXXX..XXXXXXX 100644
- void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req);
+--- a/MAINTAINERS
- void virtio_scsi_free_req(VirtIOSCSIReq *req);
++++ b/MAINTAINERS
- void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
+@@ -XXX,XX +XXX,XX @@ F: util/aio-*.c
  F: util/aio-*.h
  F: util/fdmon-*.c
  F: block/io.c
 +F: block/plug.c
  F: migration/block*
  F: include/block/aio.h
  F: include/block/aio-wait.h
 diff --git a/include/sysemu/block-backend-io.h b/include/sysemu/block-backend-io.h
 index XXXXXXX..XXXXXXX 100644
 --- a/include/sysemu/block-backend-io.h
 +++ b/include/sysemu/block-backend-io.h
@@ -XXX,XX +XXX,XX @@ void blk_iostatus_set_err(BlockBackend *blk, int error);
  int blk_get_max_iov(BlockBackend *blk);
  int blk_get_max_hw_iov(BlockBackend *blk);
 -/*
 - * blk_io_plug/unplug are thread-local operations. This means that multiple
 - * IOThreads can simultaneously call plug/unplug, but the caller must ensure
 - * that each unplug() is called in the same IOThread of the matching plug().
 - */
 -void coroutine_fn blk_co_io_plug(BlockBackend *blk);
 -void co_wrapper blk_io_plug(BlockBackend *blk);
 -
 -void coroutine_fn blk_co_io_unplug(BlockBackend *blk);
 -void co_wrapper blk_io_unplug(BlockBackend *blk);
 +void blk_io_plug(void);
 +void blk_io_unplug(void);
 +void blk_io_plug_call(void (*fn)(void *), void *opaque);
  AioContext *blk_get_aio_context(BlockBackend *blk);
  BlockAcctStats *blk_get_stats(BlockBackend *blk);
 diff --git a/block/block-backend.c b/block/block-backend.c
 index XXXXXXX..XXXXXXX 100644
 --- a/block/block-backend.c
 +++ b/block/block-backend.c
@@ -XXX,XX +XXX,XX @@ void blk_add_insert_bs_notifier(BlockBackend *blk, Notifier *notify)
      notifier_list_add(&blk->insert_bs_notifiers, notify);
  }
 -void coroutine_fn blk_co_io_plug(BlockBackend *blk)
 -{
 -    BlockDriverState *bs = blk_bs(blk);
 -    IO_CODE();
 -    GRAPH_RDLOCK_GUARD();
 -
 -    if (bs) {
 -        bdrv_co_io_plug(bs);
 -    }
 -}
 -
 -void coroutine_fn blk_co_io_unplug(BlockBackend *blk)
 -{
 -    BlockDriverState *bs = blk_bs(blk);
 -    IO_CODE();
 -    GRAPH_RDLOCK_GUARD();
 -
 -    if (bs) {
 -        bdrv_co_io_unplug(bs);
 -    }
 -}
 -
  BlockAcctStats *blk_get_stats(BlockBackend *blk)
  {
      IO_CODE();
 diff --git a/block/plug.c b/block/plug.c
 new file mode 100644
 index XXXXXXX..XXXXXXX
 --- /dev/null
 +++ b/block/plug.c
@@ -XXX,XX +XXX,XX @@
 +/* SPDX-License-Identifier: GPL-2.0-or-later */
 +/*
 + * Block I/O plugging
 + *
 + * Copyright Red Hat.
 + *
 + * This API defers a function call within a blk_io_plug()/blk_io_unplug()
 + * section, allowing multiple calls to batch up. This is a performance
 + * optimization that is used in the block layer to submit several I/O requests
 + * at once instead of individually:
 + *
 + *   blk_io_plug(); <-- start of plugged region
 + *   ...
 + *   blk_io_plug_call(my_func, my_obj); <-- deferred my_func(my_obj) call
 + *   blk_io_plug_call(my_func, my_obj); <-- another
 + *   blk_io_plug_call(my_func, my_obj); <-- another
 + *   ...
 + *   blk_io_unplug(); <-- end of plugged region, my_func(my_obj) is called once
 + *
 + * This code is actually generic and not tied to the block layer. If another
 + * subsystem needs this functionality, it could be renamed.
 + */
 +
 +#include "qemu/osdep.h"
 +#include "qemu/coroutine-tls.h"
 +#include "qemu/notify.h"
 +#include "qemu/thread.h"
 +#include "sysemu/block-backend.h"
 +
 +/* A function call that has been deferred until unplug() */
 +typedef struct {
 +    void (*fn)(void *);
 +    void *opaque;
 +} UnplugFn;
 +
 +/* Per-thread state */
 +typedef struct {
 +    unsigned count;       /* how many times has plug() been called? */
 +    GArray *unplug_fns;   /* functions to call at unplug time */
 +} Plug;
 +
 +/* Use get_ptr_plug() to fetch this thread-local value */
 +QEMU_DEFINE_STATIC_CO_TLS(Plug, plug);
 +
 +/* Called at thread cleanup time */
 +static void blk_io_plug_atexit(Notifier *n, void *value)
 +{
 +    Plug *plug = get_ptr_plug();
 +    g_array_free(plug->unplug_fns, TRUE);
 +}
 +
 +/* This won't involve coroutines, so use __thread */
 +static __thread Notifier blk_io_plug_atexit_notifier;
 +
 +/**
 + * blk_io_plug_call:
 + * @fn: a function pointer to be invoked
 + * @opaque: a user-defined argument to @fn()
 + *
 + * Call @fn(@opaque) immediately if not within a blk_io_plug()/blk_io_unplug()
 + * section.
 + *
 + * Otherwise defer the call until the end of the outermost
 + * blk_io_plug()/blk_io_unplug() section in this thread. If the same
 + * @fn/@opaque pair has already been deferred, it will only be called once upon
 + * blk_io_unplug() so that accumulated calls are batched into a single call.
 + *
 + * The caller must ensure that @opaque is not freed before @fn() is invoked.
 + */
 +void blk_io_plug_call(void (*fn)(void *), void *opaque)
 +{
 +    Plug *plug = get_ptr_plug();
 +
 +    /* Call immediately if we're not plugged */
 +    if (plug->count == 0) {
 +        fn(opaque);
 +        return;
 +    }
 +
 +    GArray *array = plug->unplug_fns;
 +    if (!array) {
 +        array = g_array_new(FALSE, FALSE, sizeof(UnplugFn));
 +        plug->unplug_fns = array;
 +        blk_io_plug_atexit_notifier.notify = blk_io_plug_atexit;
 +        qemu_thread_atexit_add(&blk_io_plug_atexit_notifier);
 +    }
 +
 +    UnplugFn *fns = (UnplugFn *)array->data;
 +    UnplugFn new_fn = {
 +        .fn = fn,
 +        .opaque = opaque,
 +    };
 +
 +    /*
 +     * There won't be many, so do a linear search. If this becomes a bottleneck
 +     * then a binary search (glib 2.62+) or different data structure could be
 +     * used.
 +     */
 +    for (guint i = 0; i < array->len; i++) {
 +        if (memcmp(&fns[i], &new_fn, sizeof(new_fn)) == 0) {
 +            return; /* already exists */
 +        }
 +    }
 +
 +    g_array_append_val(array, new_fn);
 +}
 +
 +/**
 + * blk_io_plug: Defer blk_io_plug_call() functions until blk_io_unplug()
 + *
 + * blk_io_plug/unplug are thread-local operations. This means that multiple
 + * threads can simultaneously call plug/unplug, but the caller must ensure that
 + * each unplug() is called in the same thread of the matching plug().
 + *
 + * Nesting is supported. blk_io_plug_call() functions are only called at the
 + * outermost blk_io_unplug().
 + */
 +void blk_io_plug(void)
 +{
 +    Plug *plug = get_ptr_plug();
 +
 +    assert(plug->count < UINT32_MAX);
 +
 +    plug->count++;
 +}
 +
 +/**
 + * blk_io_unplug: Run any pending blk_io_plug_call() functions
 + *
 + * There must have been a matching blk_io_plug() call in the same thread prior
 + * to this blk_io_unplug() call.
 + */
 +void blk_io_unplug(void)
 +{
 +    Plug *plug = get_ptr_plug();
 +
 +    assert(plug->count > 0);
 +
 +    if (--plug->count > 0) {
 +        return;
 +    }
 +
 +    GArray *array = plug->unplug_fns;
 +    if (!array) {
 +        return;
 +    }
 +
 +    UnplugFn *fns = (UnplugFn *)array->data;
 +
 +    for (guint i = 0; i < array->len; i++) {
 +        fns[i].fn(fns[i].opaque);
 +    }
 +
 +    /*
 +     * This resets the array without freeing memory so that appending is cheap
 +     * in the future.
 +     */
 +    g_array_set_size(array, 0);
 +}
 diff --git a/hw/block/dataplane/xen-block.c b/hw/block/dataplane/xen-block.c
 index XXXXXXX..XXXXXXX 100644
 --- a/hw/block/dataplane/xen-block.c
 +++ b/hw/block/dataplane/xen-block.c
@@ -XXX,XX +XXX,XX @@ static bool xen_block_handle_requests(XenBlockDataPlane *dataplane)
       * is below us.
       */
      if (inflight_atstart > IO_PLUG_THRESHOLD) {
 -        blk_io_plug(dataplane->blk);
 +        blk_io_plug();
      }
      while (rc != rp) {
          /* pull request from ring */
@@ -XXX,XX +XXX,XX @@ static bool xen_block_handle_requests(XenBlockDataPlane *dataplane)
          if (inflight_atstart > IO_PLUG_THRESHOLD &&
              batched >= inflight_atstart) {
 -            blk_io_unplug(dataplane->blk);
 +            blk_io_unplug();
          }
          xen_block_do_aio(request);
          if (inflight_atstart > IO_PLUG_THRESHOLD) {
              if (batched >= inflight_atstart) {
 -                blk_io_plug(dataplane->blk);
 +                blk_io_plug();
                  batched = 0;
              } else {
                  batched++;
@@ -XXX,XX +XXX,XX @@ static bool xen_block_handle_requests(XenBlockDataPlane *dataplane)
          }
      }
      if (inflight_atstart > IO_PLUG_THRESHOLD) {
 -        blk_io_unplug(dataplane->blk);
 +        blk_io_unplug();
      }
      return done_something;
 diff --git a/hw/block/virtio-blk.c b/hw/block/virtio-blk.c
 index XXXXXXX..XXXXXXX 100644
 --- a/hw/block/virtio-blk.c
 +++ b/hw/block/virtio-blk.c
@@ -XXX,XX +XXX,XX @@ void virtio_blk_handle_vq(VirtIOBlock *s, VirtQueue *vq)
      bool suppress_notifications = virtio_queue_get_notification(vq);
      aio_context_acquire(blk_get_aio_context(s->blk));
 -    blk_io_plug(s->blk);
 +    blk_io_plug();
      do {
          if (suppress_notifications) {
@@ -XXX,XX +XXX,XX @@ void virtio_blk_handle_vq(VirtIOBlock *s, VirtQueue *vq)
          virtio_blk_submit_multireq(s, &mrb);
      }
 -    blk_io_unplug(s->blk);
 +    blk_io_unplug();
      aio_context_release(blk_get_aio_context(s->blk));
  }
 diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
 index XXXXXXX..XXXXXXX 100644
 --- a/hw/scsi/virtio-scsi.c
 +++ b/hw/scsi/virtio-scsi.c
+@@ -XXX,XX +XXX,XX @@ static int virtio_scsi_handle_cmd_req_prepare(VirtIOSCSI *s, VirtIOSCSIReq *req)
+         return -ENOBUFS;
+     }
+     scsi_req_ref(req->sreq);
+-    blk_io_plug(d->conf.blk);
++    blk_io_plug();
+     object_unref(OBJECT(d));
+     return 0;
+ }
 @@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_cmd_req_submit(VirtIOSCSI *s, VirtIOSCSIReq *req)
+     if (scsi_req_enqueue(sreq)) {
+         scsi_req_continue(sreq);
+     }
+-    blk_io_unplug(sreq->dev->conf.blk);
++    blk_io_unplug();
      scsi_req_unref(sreq);
  }
--bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
+@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
-+static void virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
+                 while (!QTAILQ_EMPTY(&reqs)) {
- {
+                     req = QTAILQ_FIRST(&reqs);
-     VirtIOSCSIReq *req, *next;
+                     QTAILQ_REMOVE(&reqs, req, next);
-     int ret = 0;
+-                    blk_io_unplug(req->sreq->dev->conf.blk);
-     bool suppress_notifications = virtio_queue_get_notification(vq);
++                    blk_io_unplug();
--    bool progress = false;
+                     scsi_req_unref(req->sreq);
+                     virtqueue_detach_element(req->vq, &req->elem, 0);
-     QTAILQ_HEAD(, VirtIOSCSIReq) reqs = QTAILQ_HEAD_INITIALIZER(reqs);
+                     virtio_scsi_free_req(req);
+diff --git a/block/meson.build b/block/meson.build
-@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
+index XXXXXXX..XXXXXXX 100644
-         }
+--- a/block/meson.build
++++ b/block/meson.build
-         while ((req = virtio_scsi_pop_req(s, vq))) {
+@@ -XXX,XX +XXX,XX @@ block_ss.add(files(
--            progress = true;
+   'mirror.c',
-             ret = virtio_scsi_handle_cmd_req_prepare(s, req);
+   'nbd.c',
-             if (!ret) {
+   'null.c',
-                 QTAILQ_INSERT_TAIL(&reqs, req, next);
++  'plug.c',
-@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
+   'qapi.c',
-     QTAILQ_FOREACH_SAFE(req, &reqs, next, next) {
+   'qcow2-bitmap.c',
-         virtio_scsi_handle_cmd_req_submit(s, req);
+   'qcow2-cache.c',
      }
 -    return progress;
  }
  static void virtio_scsi_handle_cmd(VirtIODevice *vdev, VirtQueue *vq)
 --
-.35.1
+.40.1

-[PULL 1/9] Introduce event-loop-base abstract class
+[PULL 2/8] block/nvme: convert to blk_io_plug_call() API
-From: Nicolas Saenz Julienne <nsaenzju@redhat.com>
+Stop using the .bdrv_co_io_plug() API because it is not multi-queue
 block layer friendly. Use the new blk_io_plug_call() API to batch I/O
 submission instead.
-Introduce the 'event-loop-base' abstract class, it'll hold the
+Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
-properties common to all event loops and provide the necessary hooks for
+Reviewed-by: Eric Blake <eblake@redhat.com>
-their creation and maintenance. Then have iothread inherit from it.
+Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
+Acked-by: Kevin Wolf <kwolf@redhat.com>
-EventLoopBaseClass is defined as user creatable and provides a hook for
+Message-id: 20230530180959.1108766-3-stefanha@redhat.com
 its children to attach themselves to the user creatable class 'complete'
 function. It also provides an update_params() callback to propagate
 property changes onto its children.
 The new 'event-loop-base' class will live in the root directory. It is
 built on its own using the 'link_whole' option (there are no direct
 function dependencies between the class and its children, it all happens
 trough 'constructor' magic). And also imposes new compilation
 dependencies:
     qom <- event-loop-base <- blockdev (iothread.c)
 And in subsequent patches:
     qom <- event-loop-base <- qemuutil (util/main-loop.c)
 All this forced some amount of reordering in meson.build:
  - Moved qom build definition before qemuutil. Doing it the other way
    around (i.e. moving qemuutil after qom) isn't possible as a lot of
    core libraries that live in between the two depend on it.
  - Process the 'hw' subdir earlier, as it introduces files into the
    'qom' source set.
 No functional changes intended.
 Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
 Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
 Acked-by: Markus Armbruster <armbru@redhat.com>
 Message-id: 20220425075723.20019-2-nsaenzju@redhat.com
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
 ---
- qapi/qom.json                    |  22 +++++--
+ block/nvme.c       | 44 ++++++++++++--------------------------------
- meson.build                      |  23 ++++---
+ block/trace-events |  1 -
- include/sysemu/event-loop-base.h |  36 +++++++++++
+files changed, 12 insertions(+), 33 deletions(-)
  include/sysemu/iothread.h        |   6 +-
  event-loop-base.c                | 104 +++++++++++++++++++++++++++++++
  iothread.c                       |  65 ++++++-------------
 files changed, 192 insertions(+), 64 deletions(-)
  create mode 100644 include/sysemu/event-loop-base.h
  create mode 100644 event-loop-base.c
-diff --git a/qapi/qom.json b/qapi/qom.json
+diff --git a/block/nvme.c b/block/nvme.c
 index XXXXXXX..XXXXXXX 100644
---- a/qapi/qom.json
+--- a/block/nvme.c
-+++ b/qapi/qom.json
++++ b/block/nvme.c
 @@ -XXX,XX +XXX,XX @@
-             '*repeat': 'bool',
+ #include "qemu/vfio-helpers.h"
-             '*grab-toggle': 'GrabToggleKeys' } }
+ #include "block/block-io.h"
+ #include "block/block_int.h"
-+##
++#include "sysemu/block-backend.h"
-+# @EventLoopBaseProperties:
+ #include "sysemu/replay.h"
-+#
+ #include "trace.h"
-+# Common properties for event loops
-+#
+@@ -XXX,XX +XXX,XX @@ struct BDRVNVMeState {
-+# @aio-max-batch: maximum number of requests in a batch for the AIO engine,
+     int blkshift;
-+#                 0 means that the engine will use its default.
-+#                 (default: 0)
+     uint64_t max_transfer;
-+#
+-    bool plugged;
-+# Since: 7.1
-+##
+     bool supports_write_zeroes;
-+{ 'struct': 'EventLoopBaseProperties',
+     bool supports_discard;
-+  'data': { '*aio-max-batch': 'int' } }
+@@ -XXX,XX +XXX,XX @@ static void nvme_kick(NVMeQueuePair *q)
  {
      BDRVNVMeState *s = q->s;
 -    if (s->plugged || !q->need_kick) {
 +    if (!q->need_kick) {
          return;
      }
      trace_nvme_kick(s, q->index);
@@ -XXX,XX +XXX,XX @@ static bool nvme_process_completion(NVMeQueuePair *q)
      NvmeCqe *c;
      trace_nvme_process_completion(s, q->index, q->inflight);
 -    if (s->plugged) {
 -        trace_nvme_process_completion_queue_plugged(s, q->index);
 -        return false;
 -    }
      /*
       * Support re-entrancy when a request cb() function invokes aio_poll().
@@ -XXX,XX +XXX,XX @@ static void nvme_trace_command(const NvmeCmd *cmd)
      }
  }
 +static void nvme_unplug_fn(void *opaque)
 +{
 +    NVMeQueuePair *q = opaque;
 +
- ##
++    QEMU_LOCK_GUARD(&q->lock);
- # @IothreadProperties:
++    nvme_kick(q);
- #
++    nvme_process_completion(q);
@@ -XXX,XX +XXX,XX @@
  #               algorithm detects it is spending too long polling without
  #               encountering events. 0 selects a default behaviour (default: 0)
  #
 -# @aio-max-batch: maximum number of requests in a batch for the AIO engine,
 -#                 0 means that the engine will use its default
 -#                 (default:0, since 6.1)
 +# The @aio-max-batch option is available since 6.1.
  #
  # Since: 2.0
  ##
  { 'struct': 'IothreadProperties',
 +  'base': 'EventLoopBaseProperties',
    'data': { '*poll-max-ns': 'int',
              '*poll-grow': 'int',
 -            '*poll-shrink': 'int',
 -            '*aio-max-batch': 'int' } }
 +            '*poll-shrink': 'int' } }
  ##
  # @MemoryBackendProperties:
 diff --git a/meson.build b/meson.build
 index XXXXXXX..XXXXXXX 100644
 --- a/meson.build
 +++ b/meson.build
@@ -XXX,XX +XXX,XX @@ subdir('qom')
  subdir('authz')
  subdir('crypto')
  subdir('ui')
 +subdir('hw')
  if enable_modules
@@ -XXX,XX +XXX,XX @@ if enable_modules
    modulecommon = declare_dependency(link_whole: libmodulecommon, compile_args: '-DBUILD_DSO')
  endif
 +qom_ss = qom_ss.apply(config_host, strict: false)
 +libqom = static_library('qom', qom_ss.sources() + genh,
 +                        dependencies: [qom_ss.dependencies()],
 +                        name_suffix: 'fa')
 +qom = declare_dependency(link_whole: libqom)
 +
 +event_loop_base = files('event-loop-base.c')
 +event_loop_base = static_library('event-loop-base', sources: event_loop_base + genh,
 +                                 build_by_default: true)
 +event_loop_base = declare_dependency(link_whole: event_loop_base,
 +                                     dependencies: [qom])
 +
  stub_ss = stub_ss.apply(config_all, strict: false)
  util_ss.add_all(trace_ss)
@@ -XXX,XX +XXX,XX @@ subdir('monitor')
  subdir('net')
  subdir('replay')
  subdir('semihosting')
 -subdir('hw')
  subdir('tcg')
  subdir('fpu')
  subdir('accel')
@@ -XXX,XX +XXX,XX @@ qemu_syms = custom_target('qemu.syms', output: 'qemu.syms',
                               capture: true,
                               command: [undefsym, nm, '@INPUT@'])
 -qom_ss = qom_ss.apply(config_host, strict: false)
 -libqom = static_library('qom', qom_ss.sources() + genh,
 -                        dependencies: [qom_ss.dependencies()],
 -                        name_suffix: 'fa')
 -
 -qom = declare_dependency(link_whole: libqom)
 -
  authz_ss = authz_ss.apply(config_host, strict: false)
  libauthz = static_library('authz', authz_ss.sources() + genh,
                            dependencies: [authz_ss.dependencies()],
@@ -XXX,XX +XXX,XX @@ libblockdev = static_library('blockdev', blockdev_ss.sources() + genh,
                               build_by_default: false)
  blockdev = declare_dependency(link_whole: [libblockdev],
 -                              dependencies: [block])
 +                              dependencies: [block, event_loop_base])
  qmp_ss = qmp_ss.apply(config_host, strict: false)
  libqmp = static_library('qmp', qmp_ss.sources() + genh,
 diff --git a/include/sysemu/event-loop-base.h b/include/sysemu/event-loop-base.h
 new file mode 100644
 index XXXXXXX..XXXXXXX
 --- /dev/null
 +++ b/include/sysemu/event-loop-base.h
@@ -XXX,XX +XXX,XX @@
 +/*
 + * QEMU event-loop backend
 + *
 + * Copyright (C) 2022 Red Hat Inc
 + *
 + * Authors:
 + *  Nicolas Saenz Julienne <nsaenzju@redhat.com>
 + *
 + * This work is licensed under the terms of the GNU GPL, version 2 or later.
 + * See the COPYING file in the top-level directory.
 + */
 +#ifndef QEMU_EVENT_LOOP_BASE_H
 +#define QEMU_EVENT_LOOP_BASE_H
 +
 +#include "qom/object.h"
 +#include "block/aio.h"
 +#include "qemu/typedefs.h"
 +
 +#define TYPE_EVENT_LOOP_BASE         "event-loop-base"
 +OBJECT_DECLARE_TYPE(EventLoopBase, EventLoopBaseClass,
 +                    EVENT_LOOP_BASE)
 +
 +struct EventLoopBaseClass {
 +    ObjectClass parent_class;
 +
 +    void (*init)(EventLoopBase *base, Error **errp);
 +    void (*update_params)(EventLoopBase *base, Error **errp);
 +};
 +
 +struct EventLoopBase {
 +    Object parent;
 +
 +    /* AioContext AIO engine parameters */
 +    int64_t aio_max_batch;
 +};
 +#endif
 diff --git a/include/sysemu/iothread.h b/include/sysemu/iothread.h
 index XXXXXXX..XXXXXXX 100644
 --- a/include/sysemu/iothread.h
 +++ b/include/sysemu/iothread.h
@@ -XXX,XX +XXX,XX @@
  #include "block/aio.h"
  #include "qemu/thread.h"
  #include "qom/object.h"
 +#include "sysemu/event-loop-base.h"
  #define TYPE_IOTHREAD "iothread"
  struct IOThread {
 -    Object parent_obj;
 +    EventLoopBase parent_obj;
      QemuThread thread;
      AioContext *ctx;
@@ -XXX,XX +XXX,XX @@ struct IOThread {
      int64_t poll_max_ns;
      int64_t poll_grow;
      int64_t poll_shrink;
 -
 -    /* AioContext AIO engine parameters */
 -    int64_t aio_max_batch;
  };
  typedef struct IOThread IOThread;
 diff --git a/event-loop-base.c b/event-loop-base.c
 new file mode 100644
 index XXXXXXX..XXXXXXX
 --- /dev/null
 +++ b/event-loop-base.c
@@ -XXX,XX +XXX,XX @@
 +/*
 + * QEMU event-loop base
 + *
 + * Copyright (C) 2022 Red Hat Inc
 + *
 + * Authors:
 + *  Stefan Hajnoczi <stefanha@redhat.com>
 + *  Nicolas Saenz Julienne <nsaenzju@redhat.com>
 + *
 + * This work is licensed under the terms of the GNU GPL, version 2 or later.
 + * See the COPYING file in the top-level directory.
 + */
 +
 +#include "qemu/osdep.h"
 +#include "qom/object_interfaces.h"
 +#include "qapi/error.h"
 +#include "sysemu/event-loop-base.h"
 +
 +typedef struct {
 +    const char *name;
 +    ptrdiff_t offset; /* field's byte offset in EventLoopBase struct */
 +} EventLoopBaseParamInfo;
 +
 +static EventLoopBaseParamInfo aio_max_batch_info = {
 +    "aio-max-batch", offsetof(EventLoopBase, aio_max_batch),
 +};
 +
 +static void event_loop_base_get_param(Object *obj, Visitor *v,
 +        const char *name, void *opaque, Error **errp)
 +{
 +    EventLoopBase *event_loop_base = EVENT_LOOP_BASE(obj);
 +    EventLoopBaseParamInfo *info = opaque;
 +    int64_t *field = (void *)event_loop_base + info->offset;
 +
 +    visit_type_int64(v, name, field, errp);
 +}
 +
-+static void event_loop_base_set_param(Object *obj, Visitor *v,
+ static void nvme_submit_command(NVMeQueuePair *q, NVMeRequest *req,
-+        const char *name, void *opaque, Error **errp)
+                                 NvmeCmd *cmd, BlockCompletionFunc cb,
-+{
+                                 void *opaque)
-+    EventLoopBaseClass *bc = EVENT_LOOP_BASE_GET_CLASS(obj);
+@@ -XXX,XX +XXX,XX @@ static void nvme_submit_command(NVMeQueuePair *q, NVMeRequest *req,
-+    EventLoopBase *base = EVENT_LOOP_BASE(obj);
+            q->sq.tail * NVME_SQ_ENTRY_BYTES, cmd, sizeof(*cmd));
-+    EventLoopBaseParamInfo *info = opaque;
+     q->sq.tail = (q->sq.tail + 1) % NVME_QUEUE_SIZE;
-+    int64_t *field = (void *)base + info->offset;
+     q->need_kick++;
-+    int64_t value;
+-    nvme_kick(q);
-+
+-    nvme_process_completion(q);
-+    if (!visit_type_int64(v, name, &value, errp)) {
++    blk_io_plug_call(nvme_unplug_fn, q);
-+        return;
+     qemu_mutex_unlock(&q->lock);
 +    }
 +
 +    if (value < 0) {
 +        error_setg(errp, "%s value must be in range [0, %" PRId64 "]",
 +                   info->name, INT64_MAX);
 +        return;
 +    }
 +
 +    *field = value;
 +
 +    if (bc->update_params) {
 +        bc->update_params(base, errp);
 +    }
 +
 +    return;
 +}
 +
 +static void event_loop_base_complete(UserCreatable *uc, Error **errp)
 +{
 +    EventLoopBaseClass *bc = EVENT_LOOP_BASE_GET_CLASS(uc);
 +    EventLoopBase *base = EVENT_LOOP_BASE(uc);
 +
 +    if (bc->init) {
 +        bc->init(base, errp);
 +    }
 +}
 +
 +static void event_loop_base_class_init(ObjectClass *klass, void *class_data)
 +{
 +    UserCreatableClass *ucc = USER_CREATABLE_CLASS(klass);
 +    ucc->complete = event_loop_base_complete;
 +
 +    object_class_property_add(klass, "aio-max-batch", "int",
 +                              event_loop_base_get_param,
 +                              event_loop_base_set_param,
 +                              NULL, &aio_max_batch_info);
 +}
 +
 +static const TypeInfo event_loop_base_info = {
 +    .name = TYPE_EVENT_LOOP_BASE,
 +    .parent = TYPE_OBJECT,
 +    .instance_size = sizeof(EventLoopBase),
 +    .class_size = sizeof(EventLoopBaseClass),
 +    .class_init = event_loop_base_class_init,
 +    .abstract = true,
 +    .interfaces = (InterfaceInfo[]) {
 +        { TYPE_USER_CREATABLE },
 +        { }
 +    }
 +};
 +
 +static void register_types(void)
 +{
 +    type_register_static(&event_loop_base_info);
 +}
 +type_init(register_types);
 diff --git a/iothread.c b/iothread.c
 index XXXXXXX..XXXXXXX 100644
 --- a/iothread.c
 +++ b/iothread.c
@@ -XXX,XX +XXX,XX @@
  #include "qemu/module.h"
  #include "block/aio.h"
  #include "block/block.h"
 +#include "sysemu/event-loop-base.h"
  #include "sysemu/iothread.h"
  #include "qapi/error.h"
  #include "qapi/qapi-commands-misc.h"
@@ -XXX,XX +XXX,XX @@ static void iothread_init_gcontext(IOThread *iothread)
      iothread->main_loop = g_main_loop_new(iothread->worker_context, TRUE);
  }
--static void iothread_set_aio_context_params(IOThread *iothread, Error **errp)
+@@ -XXX,XX +XXX,XX @@ static void nvme_attach_aio_context(BlockDriverState *bs,
 +static void iothread_set_aio_context_params(EventLoopBase *base, Error **errp)
  {
 +    IOThread *iothread = IOTHREAD(base);
      ERRP_GUARD();
 +    if (!iothread->ctx) {
 +        return;
 +    }
 +
      aio_context_set_poll_params(iothread->ctx,
                                  iothread->poll_max_ns,
                                  iothread->poll_grow,
@@ -XXX,XX +XXX,XX @@ static void iothread_set_aio_context_params(IOThread *iothread, Error **errp)
      }
      aio_context_set_aio_params(iothread->ctx,
 -                               iothread->aio_max_batch,
 +                               iothread->parent_obj.aio_max_batch,
                                 errp);
  }
 -static void iothread_complete(UserCreatable *obj, Error **errp)
 +
 +static void iothread_init(EventLoopBase *base, Error **errp)
  {
      Error *local_error = NULL;
 -    IOThread *iothread = IOTHREAD(obj);
 +    IOThread *iothread = IOTHREAD(base);
      char *thread_name;
      iothread->stopping = false;
@@ -XXX,XX +XXX,XX @@ static void iothread_complete(UserCreatable *obj, Error **errp)
       */
      iothread_init_gcontext(iothread);
 -    iothread_set_aio_context_params(iothread, &local_error);
 +    iothread_set_aio_context_params(base, &local_error);
      if (local_error) {
          error_propagate(errp, local_error);
          aio_context_unref(iothread->ctx);
@@ -XXX,XX +XXX,XX @@ static void iothread_complete(UserCreatable *obj, Error **errp)
       * to inherit.
       */
      thread_name = g_strdup_printf("IO %s",
 -                        object_get_canonical_path_component(OBJECT(obj)));
 +                        object_get_canonical_path_component(OBJECT(base)));
      qemu_thread_create(&iothread->thread, thread_name, iothread_run,
                         iothread, QEMU_THREAD_JOINABLE);
      g_free(thread_name);
@@ -XXX,XX +XXX,XX @@ static IOThreadParamInfo poll_grow_info = {
  static IOThreadParamInfo poll_shrink_info = {
      "poll-shrink", offsetof(IOThread, poll_shrink),
  };
 -static IOThreadParamInfo aio_max_batch_info = {
 -    "aio-max-batch", offsetof(IOThread, aio_max_batch),
 -};
  static void iothread_get_param(Object *obj, Visitor *v,
          const char *name, IOThreadParamInfo *info, Error **errp)
@@ -XXX,XX +XXX,XX @@ static void iothread_set_poll_param(Object *obj, Visitor *v,
      }
  }
--static void iothread_get_aio_param(Object *obj, Visitor *v,
+-static void coroutine_fn nvme_co_io_plug(BlockDriverState *bs)
 -        const char *name, void *opaque, Error **errp)
 -{
--    IOThreadParamInfo *info = opaque;
+-    BDRVNVMeState *s = bs->opaque;
--
+-    assert(!s->plugged);
--    iothread_get_param(obj, v, name, info, errp);
+-    s->plugged = true;
 -}
 -
--static void iothread_set_aio_param(Object *obj, Visitor *v,
+-static void coroutine_fn nvme_co_io_unplug(BlockDriverState *bs)
 -        const char *name, void *opaque, Error **errp)
 -{
--    IOThread *iothread = IOTHREAD(obj);
+-    BDRVNVMeState *s = bs->opaque;
--    IOThreadParamInfo *info = opaque;
+-    assert(s->plugged);
--
+-    s->plugged = false;
--    if (!iothread_set_param(obj, v, name, info, errp)) {
+-    for (unsigned i = INDEX_IO(0); i < s->queue_count; i++) {
--        return;
+-        NVMeQueuePair *q = s->queues[i];
--    }
+-        qemu_mutex_lock(&q->lock);
--
+-        nvme_kick(q);
--    if (iothread->ctx) {
+-        nvme_process_completion(q);
--        aio_context_set_aio_params(iothread->ctx,
+-        qemu_mutex_unlock(&q->lock);
 -                                   iothread->aio_max_batch,
 -                                   errp);
 -    }
 -}
 -
- static void iothread_class_init(ObjectClass *klass, void *class_data)
+ static bool nvme_register_buf(BlockDriverState *bs, void *host, size_t size,
                                Error **errp)
  {
--    UserCreatableClass *ucc = USER_CREATABLE_CLASS(klass);
+@@ -XXX,XX +XXX,XX @@ static BlockDriver bdrv_nvme = {
--    ucc->complete = iothread_complete;
+     .bdrv_detach_aio_context  = nvme_detach_aio_context,
-+    EventLoopBaseClass *bc = EVENT_LOOP_BASE_CLASS(klass);
+     .bdrv_attach_aio_context  = nvme_attach_aio_context,
-+
-+    bc->init = iothread_init;
+-    .bdrv_co_io_plug          = nvme_co_io_plug,
-+    bc->update_params = iothread_set_aio_context_params;
+-    .bdrv_co_io_unplug        = nvme_co_io_unplug,
+-
-     object_class_property_add(klass, "poll-max-ns", "int",
+     .bdrv_register_buf        = nvme_register_buf,
-                               iothread_get_poll_param,
+     .bdrv_unregister_buf      = nvme_unregister_buf,
@@ -XXX,XX +XXX,XX @@ static void iothread_class_init(ObjectClass *klass, void *class_data)
                                iothread_get_poll_param,
                                iothread_set_poll_param,
                                NULL, &poll_shrink_info);
 -    object_class_property_add(klass, "aio-max-batch", "int",
 -                              iothread_get_aio_param,
 -                              iothread_set_aio_param,
 -                              NULL, &aio_max_batch_info);
  }
  static const TypeInfo iothread_info = {
      .name = TYPE_IOTHREAD,
 -    .parent = TYPE_OBJECT,
 +    .parent = TYPE_EVENT_LOOP_BASE,
      .class_init = iothread_class_init,
      .instance_size = sizeof(IOThread),
      .instance_init = iothread_instance_init,
      .instance_finalize = iothread_instance_finalize,
 -    .interfaces = (InterfaceInfo[]) {
 -        {TYPE_USER_CREATABLE},
 -        {}
 -    },
  };
+diff --git a/block/trace-events b/block/trace-events
- static void iothread_register_types(void)
+index XXXXXXX..XXXXXXX 100644
-@@ -XXX,XX +XXX,XX @@ static int query_one_iothread(Object *object, void *opaque)
+--- a/block/trace-events
-     info->poll_max_ns = iothread->poll_max_ns;
++++ b/block/trace-events
-     info->poll_grow = iothread->poll_grow;
+@@ -XXX,XX +XXX,XX @@ nvme_kick(void *s, unsigned q_index) "s %p q #%u"
-     info->poll_shrink = iothread->poll_shrink;
+ nvme_dma_flush_queue_wait(void *s) "s %p"
--    info->aio_max_batch = iothread->aio_max_batch;
+ nvme_error(int cmd_specific, int sq_head, int sqid, int cid, int status) "cmd_specific %d sq_head %d sqid %d cid %d status 0x%x"
-+    info->aio_max_batch = iothread->parent_obj.aio_max_batch;
+ nvme_process_completion(void *s, unsigned q_index, int inflight) "s %p q #%u inflight %d"
+-nvme_process_completion_queue_plugged(void *s, unsigned q_index) "s %p q #%u"
-     QAPI_LIST_APPEND(*tail, info);
+ nvme_complete_command(void *s, unsigned q_index, int cid) "s %p q #%u cid %d"
-     return 0;
+ nvme_submit_command(void *s, unsigned q_index, int cid) "s %p q #%u cid %d"
  nvme_submit_command_raw(int c0, int c1, int c2, int c3, int c4, int c5, int c6, int c7) "%02x %02x %02x %02x %02x %02x %02x %02x"
 --
-.35.1
+.40.1

-[PULL 5/9] virtio-scsi: don't waste CPU polling the event virtqueue
+[PULL 3/8] block/blkio: convert to blk_io_plug_call() API
-The virtio-scsi event virtqueue is not emptied by its handler function.
+Stop using the .bdrv_co_io_plug() API because it is not multi-queue
-This is typical for rx virtqueues where the device uses buffers when
+block layer friendly. Use the new blk_io_plug_call() API to batch I/O
-some event occurs (e.g. a packet is received, an error condition
+submission instead.
 happens, etc).
-Polling non-empty virtqueues wastes CPU cycles. We are not waiting for
-new buffers to become available, we are waiting for an event to occur,
-so it's a misuse of CPU resources to poll for buffers.
-Introduce the new virtio_queue_aio_attach_host_notifier_no_poll() API,
-which is identical to virtio_queue_aio_attach_host_notifier() except
-that it does not poll the virtqueue.
-Before this patch the following command-line consumed 100% CPU in the
-IOThread polling and calling virtio_scsi_handle_event():
-  $ qemu-system-x86_64 -M accel=kvm -m 1G -cpu host \
-      --object iothread,id=iothread0 \
-      --device virtio-scsi-pci,iothread=iothread0 \
-      --blockdev file,filename=test.img,aio=native,cache.direct=on,node-name=drive0 \
-      --device scsi-hd,drive=drive0
-After this patch CPU is no longer wasted.
-Reported-by: Nir Soffer <nsoffer@redhat.com>
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
-Tested-by: Nir Soffer <nsoffer@redhat.com>
+Reviewed-by: Eric Blake <eblake@redhat.com>
-Message-id: 20220427143541.119567-3-stefanha@redhat.com
+Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
 Acked-by: Kevin Wolf <kwolf@redhat.com>
 Message-id: 20230530180959.1108766-4-stefanha@redhat.com
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
 ---
- include/hw/virtio/virtio.h      |  1 +
+ block/blkio.c | 43 ++++++++++++++++++++++++-------------------
- hw/scsi/virtio-scsi-dataplane.c |  2 +-
+file changed, 24 insertions(+), 19 deletions(-)
  hw/virtio/virtio.c              | 13 +++++++++++++
 files changed, 15 insertions(+), 1 deletion(-)
-diff --git a/include/hw/virtio/virtio.h b/include/hw/virtio/virtio.h
+diff --git a/block/blkio.c b/block/blkio.c
 index XXXXXXX..XXXXXXX 100644
---- a/include/hw/virtio/virtio.h
+--- a/block/blkio.c
-+++ b/include/hw/virtio/virtio.h
++++ b/block/blkio.c
-@@ -XXX,XX +XXX,XX @@ EventNotifier *virtio_queue_get_host_notifier(VirtQueue *vq);
+@@ -XXX,XX +XXX,XX @@
- void virtio_queue_set_host_notifier_enabled(VirtQueue *vq, bool enabled);
+ #include "qemu/error-report.h"
- void virtio_queue_host_notifier_read(EventNotifier *n);
+ #include "qapi/qmp/qdict.h"
- void virtio_queue_aio_attach_host_notifier(VirtQueue *vq, AioContext *ctx);
+ #include "qemu/module.h"
-+void virtio_queue_aio_attach_host_notifier_no_poll(VirtQueue *vq, AioContext *ctx);
++#include "sysemu/block-backend.h"
- void virtio_queue_aio_detach_host_notifier(VirtQueue *vq, AioContext *ctx);
+ #include "exec/memory.h" /* for ram_block_discard_disable() */
- VirtQueue *virtio_vector_first_queue(VirtIODevice *vdev, uint16_t vector);
- VirtQueue *virtio_vector_next_queue(VirtQueue *vq);
+ #include "block/block-io.h"
-diff --git a/hw/scsi/virtio-scsi-dataplane.c b/hw/scsi/virtio-scsi-dataplane.c
+@@ -XXX,XX +XXX,XX @@ static void blkio_detach_aio_context(BlockDriverState *bs)
-index XXXXXXX..XXXXXXX 100644
+                        NULL, NULL, NULL);
 --- a/hw/scsi/virtio-scsi-dataplane.c
 +++ b/hw/scsi/virtio-scsi-dataplane.c
@@ -XXX,XX +XXX,XX @@ int virtio_scsi_dataplane_start(VirtIODevice *vdev)
      aio_context_acquire(s->ctx);
      virtio_queue_aio_attach_host_notifier(vs->ctrl_vq, s->ctx);
 -    virtio_queue_aio_attach_host_notifier(vs->event_vq, s->ctx);
 +    virtio_queue_aio_attach_host_notifier_no_poll(vs->event_vq, s->ctx);
      for (i = 0; i < vs->conf.num_queues; i++) {
          virtio_queue_aio_attach_host_notifier(vs->cmd_vqs[i], s->ctx);
 diff --git a/hw/virtio/virtio.c b/hw/virtio/virtio.c
 index XXXXXXX..XXXXXXX 100644
 --- a/hw/virtio/virtio.c
 +++ b/hw/virtio/virtio.c
@@ -XXX,XX +XXX,XX @@ void virtio_queue_aio_attach_host_notifier(VirtQueue *vq, AioContext *ctx)
                                  virtio_queue_host_notifier_aio_poll_end);
  }
+-/* Call with s->blkio_lock held to submit I/O after enqueuing a new request */
+-static void blkio_submit_io(BlockDriverState *bs)
 +/*
-+ * Same as virtio_queue_aio_attach_host_notifier() but without polling. Use
++ * Called by blk_io_unplug() or immediately if not plugged. Called without
-+ * this for rx virtqueues and similar cases where the virtqueue handler
++ * blkio_lock.
 + * function does not pop all elements. When the virtqueue is left non-empty
 + * polling consumes CPU cycles and should not be used.
 + */
-+void virtio_queue_aio_attach_host_notifier_no_poll(VirtQueue *vq, AioContext *ctx)
++static void blkio_unplug_fn(void *opaque)
  {
 -    if (qatomic_read(&bs->io_plugged) == 0) {
 -        BDRVBlkioState *s = bs->opaque;
 +    BDRVBlkioState *s = opaque;
 +    WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
          blkioq_do_io(s->blkioq, NULL, 0, 0, NULL);
      }
  }
 +/*
 + * Schedule I/O submission after enqueuing a new request. Called without
 + * blkio_lock.
 + */
 +static void blkio_submit_io(BlockDriverState *bs)
 +{
-+    aio_set_event_notifier(ctx, &vq->host_notifier, true,
++    BDRVBlkioState *s = bs->opaque;
-+                           virtio_queue_host_notifier_read,
++
-+                           NULL, NULL);
++    blk_io_plug_call(blkio_unplug_fn, s);
 +}
 +
- void virtio_queue_aio_detach_host_notifier(VirtQueue *vq, AioContext *ctx)
+ static int coroutine_fn
  blkio_co_pdiscard(BlockDriverState *bs, int64_t offset, int64_t bytes)
  {
-     aio_set_event_notifier(ctx, &vq->host_notifier, true, NULL, NULL, NULL);
+@@ -XXX,XX +XXX,XX @@ blkio_co_pdiscard(BlockDriverState *bs, int64_t offset, int64_t bytes)
      WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
          blkioq_discard(s->blkioq, offset, bytes, &cod, 0);
 -        blkio_submit_io(bs);
      }
 +    blkio_submit_io(bs);
      qemu_coroutine_yield();
      return cod.ret;
  }
@@ -XXX,XX +XXX,XX @@ blkio_co_preadv(BlockDriverState *bs, int64_t offset, int64_t bytes,
      WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
          blkioq_readv(s->blkioq, offset, iov, iovcnt, &cod, 0);
 -        blkio_submit_io(bs);
      }
 +    blkio_submit_io(bs);
      qemu_coroutine_yield();
      if (use_bounce_buffer) {
@@ -XXX,XX +XXX,XX @@ static int coroutine_fn blkio_co_pwritev(BlockDriverState *bs, int64_t offset,
      WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
          blkioq_writev(s->blkioq, offset, iov, iovcnt, &cod, blkio_flags);
 -        blkio_submit_io(bs);
      }
 +    blkio_submit_io(bs);
      qemu_coroutine_yield();
      if (use_bounce_buffer) {
@@ -XXX,XX +XXX,XX @@ static int coroutine_fn blkio_co_flush(BlockDriverState *bs)
      WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
          blkioq_flush(s->blkioq, &cod, 0);
 -        blkio_submit_io(bs);
      }
 +    blkio_submit_io(bs);
      qemu_coroutine_yield();
      return cod.ret;
  }
@@ -XXX,XX +XXX,XX @@ static int coroutine_fn blkio_co_pwrite_zeroes(BlockDriverState *bs,
      WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
          blkioq_write_zeroes(s->blkioq, offset, bytes, &cod, blkio_flags);
 -        blkio_submit_io(bs);
      }
 +    blkio_submit_io(bs);
      qemu_coroutine_yield();
      return cod.ret;
  }
 -static void coroutine_fn blkio_co_io_unplug(BlockDriverState *bs)
 -{
 -    BDRVBlkioState *s = bs->opaque;
 -
 -    WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
 -        blkio_submit_io(bs);
 -    }
 -}
 -
  typedef enum {
      BMRR_OK,
      BMRR_SKIP,
@@ -XXX,XX +XXX,XX @@ static void blkio_refresh_limits(BlockDriverState *bs, Error **errp)
          .bdrv_co_pwritev         = blkio_co_pwritev, \
          .bdrv_co_flush_to_disk   = blkio_co_flush, \
          .bdrv_co_pwrite_zeroes   = blkio_co_pwrite_zeroes, \
 -        .bdrv_co_io_unplug       = blkio_co_io_unplug, \
          .bdrv_refresh_limits     = blkio_refresh_limits, \
          .bdrv_register_buf       = blkio_register_buf, \
          .bdrv_unregister_buf     = blkio_unregister_buf, \
 --
-.35.1
+.40.1

-[PULL 7/9] virtio-scsi: clean up virtio_scsi_handle_ctrl_vq()
+[PULL 4/8] block/io_uring: convert to blk_io_plug_call() API
-virtio_scsi_handle_ctrl_vq() is only called from hw/scsi/virtio-scsi.c
+Stop using the .bdrv_co_io_plug() API because it is not multi-queue
-now and its return value is no longer used. Remove the function
+block layer friendly. Use the new blk_io_plug_call() API to batch I/O
-prototype from virtio-scsi.h and drop the return value.
+submission instead.
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
-Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
+Reviewed-by: Eric Blake <eblake@redhat.com>
-Message-id: 20220427143541.119567-5-stefanha@redhat.com
+Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
 Acked-by: Kevin Wolf <kwolf@redhat.com>
 Message-id: 20230530180959.1108766-5-stefanha@redhat.com
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
 ---
- include/hw/virtio/virtio-scsi.h | 1 -
+ include/block/raw-aio.h |  7 -------
- hw/scsi/virtio-scsi.c           | 5 +----
+ block/file-posix.c      | 10 ----------
-files changed, 1 insertion(+), 5 deletions(-)
+ block/io_uring.c        | 44 ++++++++++++++++-------------------------
  block/trace-events      |  5 ++---
 files changed, 19 insertions(+), 47 deletions(-)
-diff --git a/include/hw/virtio/virtio-scsi.h b/include/hw/virtio/virtio-scsi.h
+diff --git a/include/block/raw-aio.h b/include/block/raw-aio.h
 index XXXXXXX..XXXXXXX 100644
---- a/include/hw/virtio/virtio-scsi.h
+--- a/include/block/raw-aio.h
-+++ b/include/hw/virtio/virtio-scsi.h
++++ b/include/block/raw-aio.h
-@@ -XXX,XX +XXX,XX @@ void virtio_scsi_common_realize(DeviceState *dev,
+@@ -XXX,XX +XXX,XX @@ int coroutine_fn luring_co_submit(BlockDriverState *bs, int fd, uint64_t offset,
+                                   QEMUIOVector *qiov, int type);
- void virtio_scsi_common_unrealize(DeviceState *dev);
+ void luring_detach_aio_context(LuringState *s, AioContext *old_context);
- bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq);
+ void luring_attach_aio_context(LuringState *s, AioContext *new_context);
--bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq);
+-
- void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req);
+-/*
- void virtio_scsi_free_req(VirtIOSCSIReq *req);
+- * luring_io_plug/unplug work in the thread's current AioContext, therefore the
- void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
+- * caller must ensure that they are paired in the same IOThread.
-diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
+- */
 -void luring_io_plug(void);
 -void luring_io_unplug(void);
  #endif
  #ifdef _WIN32
 diff --git a/block/file-posix.c b/block/file-posix.c
 index XXXXXXX..XXXXXXX 100644
---- a/hw/scsi/virtio-scsi.c
+--- a/block/file-posix.c
-+++ b/hw/scsi/virtio-scsi.c
++++ b/block/file-posix.c
-@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_ctrl_req(VirtIOSCSI *s, VirtIOSCSIReq *req)
+@@ -XXX,XX +XXX,XX @@ static void coroutine_fn raw_co_io_plug(BlockDriverState *bs)
          laio_io_plug();
      }
  #endif
 -#ifdef CONFIG_LINUX_IO_URING
 -    if (s->use_linux_io_uring) {
 -        luring_io_plug();
 -    }
 -#endif
  }
  static void coroutine_fn raw_co_io_unplug(BlockDriverState *bs)
@@ -XXX,XX +XXX,XX @@ static void coroutine_fn raw_co_io_unplug(BlockDriverState *bs)
          laio_io_unplug(s->aio_max_batch);
      }
  #endif
 -#ifdef CONFIG_LINUX_IO_URING
 -    if (s->use_linux_io_uring) {
 -        luring_io_unplug();
 -    }
 -#endif
  }
  static int coroutine_fn raw_co_flush_to_disk(BlockDriverState *bs)
 diff --git a/block/io_uring.c b/block/io_uring.c
 index XXXXXXX..XXXXXXX 100644
 --- a/block/io_uring.c
 +++ b/block/io_uring.c
@@ -XXX,XX +XXX,XX @@
  #include "block/raw-aio.h"
  #include "qemu/coroutine.h"
  #include "qapi/error.h"
 +#include "sysemu/block-backend.h"
  #include "trace.h"
  /* Only used for assertions.  */
@@ -XXX,XX +XXX,XX @@ typedef struct LuringAIOCB {
  } LuringAIOCB;
  typedef struct LuringQueue {
 -    int plugged;
      unsigned int in_queue;
      unsigned int in_flight;
      bool blocked;
@@ -XXX,XX +XXX,XX @@ static void luring_process_completions_and_submit(LuringState *s)
  {
      luring_process_completions(s);
 -    if (!s->io_q.plugged && s->io_q.in_queue > 0) {
 +    if (s->io_q.in_queue > 0) {
          ioq_submit(s);
      }
  }
+@@ -XXX,XX +XXX,XX @@ static void qemu_luring_poll_ready(void *opaque)
--bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq)
+ static void ioq_init(LuringQueue *io_q)
 +static void virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq)
  {
-     VirtIOSCSIReq *req;
+     QSIMPLEQ_INIT(&io_q->submit_queue);
--    bool progress = false;
+-    io_q->plugged = 0;
+     io_q->in_queue = 0;
-     while ((req = virtio_scsi_pop_req(s, vq))) {
+     io_q->in_flight = 0;
--        progress = true;
+     io_q->blocked = false;
-         virtio_scsi_handle_ctrl_req(s, req);
+ }
 -void luring_io_plug(void)
 +static void luring_unplug_fn(void *opaque)
  {
 -    AioContext *ctx = qemu_get_current_aio_context();
 -    LuringState *s = aio_get_linux_io_uring(ctx);
 -    trace_luring_io_plug(s);
 -    s->io_q.plugged++;
 -}
 -
 -void luring_io_unplug(void)
 -{
 -    AioContext *ctx = qemu_get_current_aio_context();
 -    LuringState *s = aio_get_linux_io_uring(ctx);
 -    assert(s->io_q.plugged);
 -    trace_luring_io_unplug(s, s->io_q.blocked, s->io_q.plugged,
 -                           s->io_q.in_queue, s->io_q.in_flight);
 -    if (--s->io_q.plugged == 0 &&
 -        !s->io_q.blocked && s->io_q.in_queue > 0) {
 +    LuringState *s = opaque;
 +    trace_luring_unplug_fn(s, s->io_q.blocked, s->io_q.in_queue,
 +                           s->io_q.in_flight);
 +    if (!s->io_q.blocked && s->io_q.in_queue > 0) {
          ioq_submit(s);
      }
--    return progress;
  }
+@@ -XXX,XX +XXX,XX @@ static int luring_do_submit(int fd, LuringAIOCB *luringcb, LuringState *s,
- /*
      QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
      s->io_q.in_queue++;
 -    trace_luring_do_submit(s, s->io_q.blocked, s->io_q.plugged,
 -                           s->io_q.in_queue, s->io_q.in_flight);
 -    if (!s->io_q.blocked &&
 -        (!s->io_q.plugged ||
 -         s->io_q.in_flight + s->io_q.in_queue >= MAX_ENTRIES)) {
 -        ret = ioq_submit(s);
 -        trace_luring_do_submit_done(s, ret);
 -        return ret;
 +    trace_luring_do_submit(s, s->io_q.blocked, s->io_q.in_queue,
 +                           s->io_q.in_flight);
 +    if (!s->io_q.blocked) {
 +        if (s->io_q.in_flight + s->io_q.in_queue >= MAX_ENTRIES) {
 +            ret = ioq_submit(s);
 +            trace_luring_do_submit_done(s, ret);
 +            return ret;
 +        }
 +
 +        blk_io_plug_call(luring_unplug_fn, s);
      }
      return 0;
  }
 diff --git a/block/trace-events b/block/trace-events
 index XXXXXXX..XXXXXXX 100644
 --- a/block/trace-events
 +++ b/block/trace-events
@@ -XXX,XX +XXX,XX @@ file_paio_submit(void *acb, void *opaque, int64_t offset, int count, int type) "
  # io_uring.c
  luring_init_state(void *s, size_t size) "s %p size %zu"
  luring_cleanup_state(void *s) "%p freed"
 -luring_io_plug(void *s) "LuringState %p plug"
 -luring_io_unplug(void *s, int blocked, int plugged, int queued, int inflight) "LuringState %p blocked %d plugged %d queued %d inflight %d"
 -luring_do_submit(void *s, int blocked, int plugged, int queued, int inflight) "LuringState %p blocked %d plugged %d queued %d inflight %d"
 +luring_unplug_fn(void *s, int blocked, int queued, int inflight) "LuringState %p blocked %d queued %d inflight %d"
 +luring_do_submit(void *s, int blocked, int queued, int inflight) "LuringState %p blocked %d queued %d inflight %d"
  luring_do_submit_done(void *s, int ret) "LuringState %p submitted to kernel %d"
  luring_co_submit(void *bs, void *s, void *luringcb, int fd, uint64_t offset, size_t nbytes, int type) "bs %p s %p luringcb %p fd %d offset %" PRId64 " nbytes %zd type %d"
  luring_process_completion(void *s, void *aiocb, int ret) "LuringState %p luringcb %p ret %d"
 --
-.35.1
+.40.1

-[PULL 9/9] virtio-scsi: move request-related items from .h to .c
+[PULL 5/8] block/linux-aio: convert to blk_io_plug_call() API
-There is no longer a need to expose the request and related APIs in
+Stop using the .bdrv_co_io_plug() API because it is not multi-queue
-virtio-scsi.h since there are no callers outside virtio-scsi.c.
+block layer friendly. Use the new blk_io_plug_call() API to batch I/O
+submission instead.
-Note the block comment in VirtIOSCSIReq has been adjusted to meet the
-coding style.
+Note that a dev_max_batch check is dropped in laio_io_unplug() because
 the semantics of unplug_fn() are different from .bdrv_co_unplug():
 . unplug_fn() is only called when the last blk_io_unplug() call occurs,
    not every time blk_io_unplug() is called.
 . unplug_fn() is per-thread, not per-BlockDriverState, so there is no
    way to get per-BlockDriverState fields like dev_max_batch.
 Therefore this condition cannot be moved to laio_unplug_fn(). It is not
 obvious that this condition affects performance in practice, so I am
 removing it instead of trying to come up with a more complex mechanism
 to preserve the condition.
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
-Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
+Reviewed-by: Eric Blake <eblake@redhat.com>
-Message-id: 20220427143541.119567-7-stefanha@redhat.com
+Acked-by: Kevin Wolf <kwolf@redhat.com>
 Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
 Message-id: 20230530180959.1108766-6-stefanha@redhat.com
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
 ---
- include/hw/virtio/virtio-scsi.h | 40 -----------------------------
+ include/block/raw-aio.h |  7 -------
- hw/scsi/virtio-scsi.c           | 45 ++++++++++++++++++++++++++++++---
+ block/file-posix.c      | 28 ----------------------------
-files changed, 41 insertions(+), 44 deletions(-)
+ block/linux-aio.c       | 41 +++++++++++------------------------------
+files changed, 11 insertions(+), 65 deletions(-)
-diff --git a/include/hw/virtio/virtio-scsi.h b/include/hw/virtio/virtio-scsi.h
 diff --git a/include/block/raw-aio.h b/include/block/raw-aio.h
 index XXXXXXX..XXXXXXX 100644
---- a/include/hw/virtio/virtio-scsi.h
+--- a/include/block/raw-aio.h
-+++ b/include/hw/virtio/virtio-scsi.h
++++ b/include/block/raw-aio.h
-@@ -XXX,XX +XXX,XX @@ struct VirtIOSCSI {
+@@ -XXX,XX +XXX,XX @@ int coroutine_fn laio_co_submit(int fd, uint64_t offset, QEMUIOVector *qiov,
-     uint32_t host_features;
  void laio_detach_aio_context(LinuxAioState *s, AioContext *old_context);
  void laio_attach_aio_context(LinuxAioState *s, AioContext *new_context);
 -
 -/*
 - * laio_io_plug/unplug work in the thread's current AioContext, therefore the
 - * caller must ensure that they are paired in the same IOThread.
 - */
 -void laio_io_plug(void);
 -void laio_io_unplug(uint64_t dev_max_batch);
  #endif
  /* io_uring.c - Linux io_uring implementation */
  #ifdef CONFIG_LINUX_IO_URING
 diff --git a/block/file-posix.c b/block/file-posix.c
 index XXXXXXX..XXXXXXX 100644
 --- a/block/file-posix.c
 +++ b/block/file-posix.c
@@ -XXX,XX +XXX,XX @@ static int coroutine_fn raw_co_pwritev(BlockDriverState *bs, int64_t offset,
      return raw_co_prw(bs, offset, bytes, qiov, QEMU_AIO_WRITE);
  }
 -static void coroutine_fn raw_co_io_plug(BlockDriverState *bs)
 -{
 -    BDRVRawState __attribute__((unused)) *s = bs->opaque;
 -#ifdef CONFIG_LINUX_AIO
 -    if (s->use_linux_aio) {
 -        laio_io_plug();
 -    }
 -#endif
 -}
 -
 -static void coroutine_fn raw_co_io_unplug(BlockDriverState *bs)
 -{
 -    BDRVRawState __attribute__((unused)) *s = bs->opaque;
 -#ifdef CONFIG_LINUX_AIO
 -    if (s->use_linux_aio) {
 -        laio_io_unplug(s->aio_max_batch);
 -    }
 -#endif
 -}
 -
  static int coroutine_fn raw_co_flush_to_disk(BlockDriverState *bs)
  {
      BDRVRawState *s = bs->opaque;
@@ -XXX,XX +XXX,XX @@ BlockDriver bdrv_file = {
      .bdrv_co_copy_range_from = raw_co_copy_range_from,
      .bdrv_co_copy_range_to  = raw_co_copy_range_to,
      .bdrv_refresh_limits = raw_refresh_limits,
 -    .bdrv_co_io_plug        = raw_co_io_plug,
 -    .bdrv_co_io_unplug      = raw_co_io_unplug,
      .bdrv_attach_aio_context = raw_aio_attach_aio_context,
      .bdrv_co_truncate                   = raw_co_truncate,
@@ -XXX,XX +XXX,XX @@ static BlockDriver bdrv_host_device = {
      .bdrv_co_copy_range_from = raw_co_copy_range_from,
      .bdrv_co_copy_range_to  = raw_co_copy_range_to,
      .bdrv_refresh_limits = raw_refresh_limits,
 -    .bdrv_co_io_plug        = raw_co_io_plug,
 -    .bdrv_co_io_unplug      = raw_co_io_unplug,
      .bdrv_attach_aio_context = raw_aio_attach_aio_context,
      .bdrv_co_truncate                   = raw_co_truncate,
@@ -XXX,XX +XXX,XX @@ static BlockDriver bdrv_host_cdrom = {
      .bdrv_co_pwritev        = raw_co_pwritev,
      .bdrv_co_flush_to_disk  = raw_co_flush_to_disk,
      .bdrv_refresh_limits    = cdrom_refresh_limits,
 -    .bdrv_co_io_plug        = raw_co_io_plug,
 -    .bdrv_co_io_unplug      = raw_co_io_unplug,
      .bdrv_attach_aio_context = raw_aio_attach_aio_context,
      .bdrv_co_truncate                   = raw_co_truncate,
@@ -XXX,XX +XXX,XX @@ static BlockDriver bdrv_host_cdrom = {
      .bdrv_co_pwritev        = raw_co_pwritev,
      .bdrv_co_flush_to_disk  = raw_co_flush_to_disk,
      .bdrv_refresh_limits    = cdrom_refresh_limits,
 -    .bdrv_co_io_plug        = raw_co_io_plug,
 -    .bdrv_co_io_unplug      = raw_co_io_unplug,
      .bdrv_attach_aio_context = raw_aio_attach_aio_context,
      .bdrv_co_truncate                   = raw_co_truncate,
 diff --git a/block/linux-aio.c b/block/linux-aio.c
 index XXXXXXX..XXXXXXX 100644
 --- a/block/linux-aio.c
 +++ b/block/linux-aio.c
@@ -XXX,XX +XXX,XX @@
  #include "qemu/event_notifier.h"
  #include "qemu/coroutine.h"
  #include "qapi/error.h"
 +#include "sysemu/block-backend.h"
  /* Only used for assertions.  */
  #include "qemu/coroutine_int.h"
@@ -XXX,XX +XXX,XX @@ struct qemu_laiocb {
  };
--typedef struct VirtIOSCSIReq {
+ typedef struct {
--    /* Note:
+-    int plugged;
--     * - fields up to resp_iov are initialized by virtio_scsi_init_req;
+     unsigned int in_queue;
--     * - fields starting at vring are zeroed by virtio_scsi_init_req.
+     unsigned int in_flight;
--     * */
+     bool blocked;
--    VirtQueueElement elem;
+@@ -XXX,XX +XXX,XX @@ static void qemu_laio_process_completions_and_submit(LinuxAioState *s)
--
+ {
--    VirtIOSCSI *dev;
+     qemu_laio_process_completions(s);
--    VirtQueue *vq;
--    QEMUSGList qsgl;
+-    if (!s->io_q.plugged && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
--    QEMUIOVector resp_iov;
++    if (!QSIMPLEQ_EMPTY(&s->io_q.pending)) {
--
+         ioq_submit(s);
--    union {
+     }
--        /* Used for two-stage request submission */
+ }
--        QTAILQ_ENTRY(VirtIOSCSIReq) next;
+@@ -XXX,XX +XXX,XX @@ static void qemu_laio_poll_ready(EventNotifier *opaque)
--
+ static void ioq_init(LaioQueue *io_q)
--        /* Used for cancellation of request during TMFs */
+ {
--        int remaining;
+     QSIMPLEQ_INIT(&io_q->pending);
--    };
+-    io_q->plugged = 0;
--
+     io_q->in_queue = 0;
--    SCSIRequest *sreq;
+     io_q->in_flight = 0;
--    size_t resp_size;
+     io_q->blocked = false;
--    enum SCSIXferMode mode;
+@@ -XXX,XX +XXX,XX @@ static uint64_t laio_max_batch(LinuxAioState *s, uint64_t dev_max_batch)
--    union {
+     return max_batch;
--        VirtIOSCSICmdResp     cmd;
+ }
--        VirtIOSCSICtrlTMFResp tmf;
--        VirtIOSCSICtrlANResp  an;
+-void laio_io_plug(void)
--        VirtIOSCSIEvent       event;
++static void laio_unplug_fn(void *opaque)
--    } resp;
+ {
--    union {
+-    AioContext *ctx = qemu_get_current_aio_context();
--        VirtIOSCSICmdReq      cmd;
+-    LinuxAioState *s = aio_get_linux_aio(ctx);
--        VirtIOSCSICtrlTMFReq  tmf;
++    LinuxAioState *s = opaque;
--        VirtIOSCSICtrlANReq   an;
--    } req;
+-    s->io_q.plugged++;
--} VirtIOSCSIReq;
+-}
 -
- static inline void virtio_scsi_acquire(VirtIOSCSI *s)
+-void laio_io_unplug(uint64_t dev_max_batch)
- {
+-{
-     if (s->ctx) {
+-    AioContext *ctx = qemu_get_current_aio_context();
-@@ -XXX,XX +XXX,XX @@ void virtio_scsi_common_realize(DeviceState *dev,
+-    LinuxAioState *s = aio_get_linux_aio(ctx);
-                                 Error **errp);
+-
+-    assert(s->io_q.plugged);
- void virtio_scsi_common_unrealize(DeviceState *dev);
+-    s->io_q.plugged--;
--void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req);
+-
--void virtio_scsi_free_req(VirtIOSCSIReq *req);
+-    /*
--void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
+-     * Why max batch checking is performed here:
--                            uint32_t event, uint32_t reason);
+-     * Another BDS may have queued requests with a higher dev_max_batch and
+-     * therefore in_queue could now exceed our dev_max_batch. Re-check the max
- void virtio_scsi_dataplane_setup(VirtIOSCSI *s, Error **errp);
+-     * batch so we can honor our device's dev_max_batch.
- int virtio_scsi_dataplane_start(VirtIODevice *s);
+-     */
-diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
+-    if (s->io_q.in_queue >= laio_max_batch(s, dev_max_batch) ||
-index XXXXXXX..XXXXXXX 100644
+-        (!s->io_q.plugged &&
---- a/hw/scsi/virtio-scsi.c
+-         !s->io_q.blocked && !QSIMPLEQ_EMPTY(&s->io_q.pending))) {
-+++ b/hw/scsi/virtio-scsi.c
++    if (!s->io_q.blocked && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
-@@ -XXX,XX +XXX,XX @@
+         ioq_submit(s);
- #include "hw/virtio/virtio-access.h"
+     }
- #include "trace.h"
+ }
+@@ -XXX,XX +XXX,XX @@ static int laio_do_submit(int fd, struct qemu_laiocb *laiocb, off_t offset,
-+typedef struct VirtIOSCSIReq {
-+    /*
+     QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
-+     * Note:
+     s->io_q.in_queue++;
-+     * - fields up to resp_iov are initialized by virtio_scsi_init_req;
+-    if (!s->io_q.blocked &&
-+     * - fields starting at vring are zeroed by virtio_scsi_init_req.
+-        (!s->io_q.plugged ||
-+     */
+-         s->io_q.in_queue >= laio_max_batch(s, dev_max_batch))) {
-+    VirtQueueElement elem;
+-        ioq_submit(s);
-+
++    if (!s->io_q.blocked) {
-+    VirtIOSCSI *dev;
++        if (s->io_q.in_queue >= laio_max_batch(s, dev_max_batch)) {
-+    VirtQueue *vq;
++            ioq_submit(s);
-+    QEMUSGList qsgl;
++        } else {
-+    QEMUIOVector resp_iov;
++            blk_io_plug_call(laio_unplug_fn, s);
-+
++        }
-+    union {
+     }
-+        /* Used for two-stage request submission */
-+        QTAILQ_ENTRY(VirtIOSCSIReq) next;
+     return 0;
 +
 +        /* Used for cancellation of request during TMFs */
 +        int remaining;
 +    };
 +
 +    SCSIRequest *sreq;
 +    size_t resp_size;
 +    enum SCSIXferMode mode;
 +    union {
 +        VirtIOSCSICmdResp     cmd;
 +        VirtIOSCSICtrlTMFResp tmf;
 +        VirtIOSCSICtrlANResp  an;
 +        VirtIOSCSIEvent       event;
 +    } resp;
 +    union {
 +        VirtIOSCSICmdReq      cmd;
 +        VirtIOSCSICtrlTMFReq  tmf;
 +        VirtIOSCSICtrlANReq   an;
 +    } req;
 +} VirtIOSCSIReq;
 +
  static inline int virtio_scsi_get_lun(uint8_t *lun)
  {
      return ((lun[2] << 8) | lun[3]) & 0x3FFF;
@@ -XXX,XX +XXX,XX @@ static inline SCSIDevice *virtio_scsi_device_get(VirtIOSCSI *s, uint8_t *lun)
      return scsi_device_get(&s->bus, 0, lun[1], virtio_scsi_get_lun(lun));
  }
 -void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req)
 +static void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req)
  {
      VirtIODevice *vdev = VIRTIO_DEVICE(s);
      const size_t zero_skip =
@@ -XXX,XX +XXX,XX @@ void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req)
      memset((uint8_t *)req + zero_skip, 0, sizeof(*req) - zero_skip);
  }
 -void virtio_scsi_free_req(VirtIOSCSIReq *req)
 +static void virtio_scsi_free_req(VirtIOSCSIReq *req)
  {
      qemu_iovec_destroy(&req->resp_iov);
      qemu_sglist_destroy(&req->qsgl);
@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_reset(VirtIODevice *vdev)
      s->events_dropped = false;
  }
 -void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
 -                            uint32_t event, uint32_t reason)
 +static void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
 +                                   uint32_t event, uint32_t reason)
  {
      VirtIOSCSICommon *vs = VIRTIO_SCSI_COMMON(s);
      VirtIOSCSIReq *req;
 --
-.35.1
+.40.1

-[PULL 6/9] virtio-scsi: clean up virtio_scsi_handle_event_vq()
+[PULL 6/8] block: remove bdrv_co_io_plug() API
-virtio_scsi_handle_event_vq() is only called from hw/scsi/virtio-scsi.c
+No block driver implements .bdrv_co_io_plug() anymore. Get rid of the
-now and its return value is no longer used. Remove the function
+function pointers.
 prototype from virtio-scsi.h and drop the return value.
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
-Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
+Reviewed-by: Eric Blake <eblake@redhat.com>
-Message-id: 20220427143541.119567-4-stefanha@redhat.com
+Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
 Acked-by: Kevin Wolf <kwolf@redhat.com>
 Message-id: 20230530180959.1108766-7-stefanha@redhat.com
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
 ---
- include/hw/virtio/virtio-scsi.h | 1 -
+ include/block/block-io.h         |  3 ---
- hw/scsi/virtio-scsi.c           | 4 +---
+ include/block/block_int-common.h | 11 ----------
-files changed, 1 insertion(+), 4 deletions(-)
+ block/io.c                       | 37 --------------------------------
 files changed, 51 deletions(-)
-diff --git a/include/hw/virtio/virtio-scsi.h b/include/hw/virtio/virtio-scsi.h
+diff --git a/include/block/block-io.h b/include/block/block-io.h
 index XXXXXXX..XXXXXXX 100644
---- a/include/hw/virtio/virtio-scsi.h
+--- a/include/block/block-io.h
-+++ b/include/hw/virtio/virtio-scsi.h
++++ b/include/block/block-io.h
-@@ -XXX,XX +XXX,XX @@ void virtio_scsi_common_realize(DeviceState *dev,
+@@ -XXX,XX +XXX,XX @@ void coroutine_fn bdrv_co_leave(BlockDriverState *bs, AioContext *old_ctx);
-                                 Error **errp);
+ AioContext *child_of_bds_get_parent_aio_context(BdrvChild *c);
- void virtio_scsi_common_unrealize(DeviceState *dev);
--bool virtio_scsi_handle_event_vq(VirtIOSCSI *s, VirtQueue *vq);
+-void coroutine_fn GRAPH_RDLOCK bdrv_co_io_plug(BlockDriverState *bs);
- bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq);
+-void coroutine_fn GRAPH_RDLOCK bdrv_co_io_unplug(BlockDriverState *bs);
- bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq);
+-
- void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req);
+ bool coroutine_fn GRAPH_RDLOCK
-diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
+ bdrv_co_can_store_new_dirty_bitmap(BlockDriverState *bs, const char *name,
                                     uint32_t granularity, Error **errp);
 diff --git a/include/block/block_int-common.h b/include/block/block_int-common.h
 index XXXXXXX..XXXXXXX 100644
---- a/hw/scsi/virtio-scsi.c
+--- a/include/block/block_int-common.h
-+++ b/hw/scsi/virtio-scsi.c
++++ b/include/block/block_int-common.h
-@@ -XXX,XX +XXX,XX @@ void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
+@@ -XXX,XX +XXX,XX @@ struct BlockDriver {
-     virtio_scsi_complete_req(req);
+     void coroutine_fn GRAPH_RDLOCK_PTR (*bdrv_co_debug_event)(
          BlockDriverState *bs, BlkdebugEvent event);
 -    /* io queue for linux-aio */
 -    void coroutine_fn GRAPH_RDLOCK_PTR (*bdrv_co_io_plug)(BlockDriverState *bs);
 -    void coroutine_fn GRAPH_RDLOCK_PTR (*bdrv_co_io_unplug)(
 -        BlockDriverState *bs);
 -
      bool (*bdrv_supports_persistent_dirty_bitmap)(BlockDriverState *bs);
      bool coroutine_fn GRAPH_RDLOCK_PTR (*bdrv_co_can_store_new_dirty_bitmap)(
@@ -XXX,XX +XXX,XX @@ struct BlockDriverState {
      unsigned int in_flight;
      unsigned int serialising_in_flight;
 -    /*
 -     * counter for nested bdrv_io_plug.
 -     * Accessed with atomic ops.
 -     */
 -    unsigned io_plugged;
 -
      /* do we need to tell the quest if we have a volatile write cache? */
      int enable_write_cache;
 diff --git a/block/io.c b/block/io.c
 index XXXXXXX..XXXXXXX 100644
 --- a/block/io.c
 +++ b/block/io.c
@@ -XXX,XX +XXX,XX @@ void *qemu_try_blockalign0(BlockDriverState *bs, size_t size)
      return mem;
  }
--bool virtio_scsi_handle_event_vq(VirtIOSCSI *s, VirtQueue *vq)
+-void coroutine_fn bdrv_co_io_plug(BlockDriverState *bs)
-+static void virtio_scsi_handle_event_vq(VirtIOSCSI *s, VirtQueue *vq)
+-{
- {
+-    BdrvChild *child;
-     if (s->events_dropped) {
+-    IO_CODE();
-         virtio_scsi_push_event(s, NULL, VIRTIO_SCSI_T_NO_EVENT, 0);
+-    assert_bdrv_graph_readable();
--        return true;
+-
-     }
+-    QLIST_FOREACH(child, &bs->children, next) {
--    return false;
+-        bdrv_co_io_plug(child->bs);
- }
+-    }
+-
- static void virtio_scsi_handle_event(VirtIODevice *vdev, VirtQueue *vq)
+-    if (qatomic_fetch_inc(&bs->io_plugged) == 0) {
 -        BlockDriver *drv = bs->drv;
 -        if (drv && drv->bdrv_co_io_plug) {
 -            drv->bdrv_co_io_plug(bs);
 -        }
 -    }
 -}
 -
 -void coroutine_fn bdrv_co_io_unplug(BlockDriverState *bs)
 -{
 -    BdrvChild *child;
 -    IO_CODE();
 -    assert_bdrv_graph_readable();
 -
 -    assert(bs->io_plugged);
 -    if (qatomic_fetch_dec(&bs->io_plugged) == 1) {
 -        BlockDriver *drv = bs->drv;
 -        if (drv && drv->bdrv_co_io_unplug) {
 -            drv->bdrv_co_io_unplug(bs);
 -        }
 -    }
 -
 -    QLIST_FOREACH(child, &bs->children, next) {
 -        bdrv_co_io_unplug(child->bs);
 -    }
 -}
 -
  /* Helper that undoes bdrv_register_buf() when it fails partway through */
  static void GRAPH_RDLOCK
  bdrv_register_buf_rollback(BlockDriverState *bs, void *host, size_t size,
 --
-.35.1
+.40.1

-[PULL 3/9] util/event-loop-base: Introduce options to set the thread pool size
+[PULL 7/8] block/blkio: use qemu_open() to support fd passing for virtio-blk
-From: Nicolas Saenz Julienne <nsaenzju@redhat.com>
+From: Stefano Garzarella <sgarzare@redhat.com>
-The thread pool regulates itself: when idle, it kills threads until
+Some virtio-blk drivers (e.g. virtio-blk-vhost-vdpa) supports the fd
-empty, when in demand, it creates new threads until full. This behaviour
+passing. Let's expose this to the user, so the management layer
-doesn't play well with latency sensitive workloads where the price of
+can pass the file descriptor of an already opened path.
 creating a new thread is too high. For example, when paired with qemu's
 '-mlock', or using safety features like SafeStack, creating a new thread
 has been measured take multiple milliseconds.
-In order to mitigate this let's introduce a new 'EventLoopBase'
+If the libblkio virtio-blk driver supports fd passing, let's always
-property to set the thread pool size. The threads will be created during
+use qemu_open() to open the `path`, so we can handle fd passing
-the pool's initialization or upon updating the property's value, remain
+from the management layer through the "/dev/fdset/N" special path.
 available during its lifetime regardless of demand, and destroyed upon
 freeing it. A properly characterized workload will then be able to
 configure the pool to avoid any latency spikes.
-Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
 Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
-Acked-by: Markus Armbruster <armbru@redhat.com>
+Signed-off-by: Stefano Garzarella <sgarzare@redhat.com>
-Message-id: 20220425075723.20019-4-nsaenzju@redhat.com
+Message-id: 20230530071941.8954-2-sgarzare@redhat.com
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
 ---
- qapi/qom.json                    | 10 +++++-
+ block/blkio.c | 53 ++++++++++++++++++++++++++++++++++++++++++---------
- include/block/aio.h              | 10 ++++++
+file changed, 44 insertions(+), 9 deletions(-)
  include/block/thread-pool.h      |  3 ++
  include/sysemu/event-loop-base.h |  4 +++
  event-loop-base.c                | 23 +++++++++++++
  iothread.c                       |  3 ++
  util/aio-posix.c                 |  1 +
  util/async.c                     | 20 ++++++++++++
  util/main-loop.c                 |  9 ++++++
  util/thread-pool.c               | 55 +++++++++++++++++++++++++++++---
 files changed, 133 insertions(+), 5 deletions(-)
-diff --git a/qapi/qom.json b/qapi/qom.json
+diff --git a/block/blkio.c b/block/blkio.c
 index XXXXXXX..XXXXXXX 100644
---- a/qapi/qom.json
+--- a/block/blkio.c
-+++ b/qapi/qom.json
++++ b/block/blkio.c
-@@ -XXX,XX +XXX,XX @@
+@@ -XXX,XX +XXX,XX @@ static int blkio_virtio_blk_common_open(BlockDriverState *bs,
- #                 0 means that the engine will use its default.
+ {
- #                 (default: 0)
+     const char *path = qdict_get_try_str(options, "path");
- #
+     BDRVBlkioState *s = bs->opaque;
-+# @thread-pool-min: minimum number of threads reserved in the thread pool
+-    int ret;
-+#                   (default:0)
++    bool fd_supported = false;
-+#
++    int fd, ret;
-+# @thread-pool-max: maximum number of threads the thread pool can contain
-+#                   (default:64)
+     if (!path) {
-+#
+         error_setg(errp, "missing 'path' option");
- # Since: 7.1
+         return -EINVAL;
- ##
+     }
- { 'struct': 'EventLoopBaseProperties',
--  'data': { '*aio-max-batch': 'int' } }
+-    ret = blkio_set_str(s->blkio, "path", path);
-+  'data': { '*aio-max-batch': 'int',
+-    qdict_del(options, "path");
-+            '*thread-pool-min': 'int',
+-    if (ret < 0) {
-+            '*thread-pool-max': 'int' } }
+-        error_setg_errno(errp, -ret, "failed to set path: %s",
+-                         blkio_get_error_msg());
- ##
+-        return ret;
- # @IothreadProperties:
+-    }
-diff --git a/include/block/aio.h b/include/block/aio.h
+-
-index XXXXXXX..XXXXXXX 100644
+     if (!(flags & BDRV_O_NOCACHE)) {
---- a/include/block/aio.h
+         error_setg(errp, "cache.direct=off is not supported");
-+++ b/include/block/aio.h
+         return -EINVAL;
-@@ -XXX,XX +XXX,XX @@ struct AioContext {
+     }
      QSLIST_HEAD(, Coroutine) scheduled_coroutines;
      QEMUBH *co_schedule_bh;
 +    int thread_pool_min;
 +    int thread_pool_max;
      /* Thread pool for performing work and receiving completion callbacks.
       * Has its own locking.
       */
@@ -XXX,XX +XXX,XX @@ void aio_context_set_poll_params(AioContext *ctx, int64_t max_ns,
  void aio_context_set_aio_params(AioContext *ctx, int64_t max_batch,
                                  Error **errp);
 +/**
 + * aio_context_set_thread_pool_params:
 + * @ctx: the aio context
 + * @min: min number of threads to have readily available in the thread pool
 + * @min: max number of threads the thread pool can contain
 + */
 +void aio_context_set_thread_pool_params(AioContext *ctx, int64_t min,
 +                                        int64_t max, Error **errp);
  #endif
 diff --git a/include/block/thread-pool.h b/include/block/thread-pool.h
 index XXXXXXX..XXXXXXX 100644
 --- a/include/block/thread-pool.h
 +++ b/include/block/thread-pool.h
@@ -XXX,XX +XXX,XX @@
  #include "block/block.h"
 +#define THREAD_POOL_MAX_THREADS_DEFAULT         64
 +
- typedef int ThreadPoolFunc(void *opaque);
++    if (blkio_get_int(s->blkio, "fd", &fd) == 0) {
++        fd_supported = true;
  typedef struct ThreadPool ThreadPool;
@@ -XXX,XX +XXX,XX @@ BlockAIOCB *thread_pool_submit_aio(ThreadPool *pool,
  int coroutine_fn thread_pool_submit_co(ThreadPool *pool,
          ThreadPoolFunc *func, void *arg);
  void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg);
 +void thread_pool_update_params(ThreadPool *pool, struct AioContext *ctx);
  #endif
 diff --git a/include/sysemu/event-loop-base.h b/include/sysemu/event-loop-base.h
 index XXXXXXX..XXXXXXX 100644
 --- a/include/sysemu/event-loop-base.h
 +++ b/include/sysemu/event-loop-base.h
@@ -XXX,XX +XXX,XX @@ struct EventLoopBase {
      /* AioContext AIO engine parameters */
      int64_t aio_max_batch;
 +
 +    /* AioContext thread pool parameters */
 +    int64_t thread_pool_min;
 +    int64_t thread_pool_max;
  };
  #endif
 diff --git a/event-loop-base.c b/event-loop-base.c
 index XXXXXXX..XXXXXXX 100644
 --- a/event-loop-base.c
 +++ b/event-loop-base.c
@@ -XXX,XX +XXX,XX @@
  #include "qemu/osdep.h"
  #include "qom/object_interfaces.h"
  #include "qapi/error.h"
 +#include "block/thread-pool.h"
  #include "sysemu/event-loop-base.h"
  typedef struct {
@@ -XXX,XX +XXX,XX @@ typedef struct {
      ptrdiff_t offset; /* field's byte offset in EventLoopBase struct */
  } EventLoopBaseParamInfo;
 +static void event_loop_base_instance_init(Object *obj)
 +{
 +    EventLoopBase *base = EVENT_LOOP_BASE(obj);
 +
 +    base->thread_pool_max = THREAD_POOL_MAX_THREADS_DEFAULT;
 +}
 +
  static EventLoopBaseParamInfo aio_max_batch_info = {
      "aio-max-batch", offsetof(EventLoopBase, aio_max_batch),
  };
 +static EventLoopBaseParamInfo thread_pool_min_info = {
 +    "thread-pool-min", offsetof(EventLoopBase, thread_pool_min),
 +};
 +static EventLoopBaseParamInfo thread_pool_max_info = {
 +    "thread-pool-max", offsetof(EventLoopBase, thread_pool_max),
 +};
  static void event_loop_base_get_param(Object *obj, Visitor *v,
          const char *name, void *opaque, Error **errp)
@@ -XXX,XX +XXX,XX @@ static void event_loop_base_class_init(ObjectClass *klass, void *class_data)
                                event_loop_base_get_param,
                                event_loop_base_set_param,
                                NULL, &aio_max_batch_info);
 +    object_class_property_add(klass, "thread-pool-min", "int",
 +                              event_loop_base_get_param,
 +                              event_loop_base_set_param,
 +                              NULL, &thread_pool_min_info);
 +    object_class_property_add(klass, "thread-pool-max", "int",
 +                              event_loop_base_get_param,
 +                              event_loop_base_set_param,
 +                              NULL, &thread_pool_max_info);
  }
  static const TypeInfo event_loop_base_info = {
      .name = TYPE_EVENT_LOOP_BASE,
      .parent = TYPE_OBJECT,
      .instance_size = sizeof(EventLoopBase),
 +    .instance_init = event_loop_base_instance_init,
      .class_size = sizeof(EventLoopBaseClass),
      .class_init = event_loop_base_class_init,
      .abstract = true,
 diff --git a/iothread.c b/iothread.c
 index XXXXXXX..XXXXXXX 100644
 --- a/iothread.c
 +++ b/iothread.c
@@ -XXX,XX +XXX,XX @@ static void iothread_set_aio_context_params(EventLoopBase *base, Error **errp)
      aio_context_set_aio_params(iothread->ctx,
                                 iothread->parent_obj.aio_max_batch,
                                 errp);
 +
 +    aio_context_set_thread_pool_params(iothread->ctx, base->thread_pool_min,
 +                                       base->thread_pool_max, errp);
  }
 diff --git a/util/aio-posix.c b/util/aio-posix.c
 index XXXXXXX..XXXXXXX 100644
 --- a/util/aio-posix.c
 +++ b/util/aio-posix.c
@@ -XXX,XX +XXX,XX @@
  #include "qemu/osdep.h"
  #include "block/block.h"
 +#include "block/thread-pool.h"
  #include "qemu/main-loop.h"
  #include "qemu/rcu.h"
  #include "qemu/rcu_queue.h"
 diff --git a/util/async.c b/util/async.c
 index XXXXXXX..XXXXXXX 100644
 --- a/util/async.c
 +++ b/util/async.c
@@ -XXX,XX +XXX,XX @@ AioContext *aio_context_new(Error **errp)
      ctx->aio_max_batch = 0;
 +    ctx->thread_pool_min = 0;
 +    ctx->thread_pool_max = THREAD_POOL_MAX_THREADS_DEFAULT;
 +
      return ctx;
  fail:
      g_source_destroy(&ctx->source);
@@ -XXX,XX +XXX,XX @@ void qemu_set_current_aio_context(AioContext *ctx)
      assert(!get_my_aiocontext());
      set_my_aiocontext(ctx);
  }
 +
 +void aio_context_set_thread_pool_params(AioContext *ctx, int64_t min,
 +                                        int64_t max, Error **errp)
 +{
 +
 +    if (min > max || !max || min > INT_MAX || max > INT_MAX) {
 +        error_setg(errp, "bad thread-pool-min/thread-pool-max values");
 +        return;
 +    }
 +
-+    ctx->thread_pool_min = min;
++    /*
-+    ctx->thread_pool_max = max;
++     * If the libblkio driver supports fd passing, let's always use qemu_open()
 +     * to open the `path`, so we can handle fd passing from the management
 +     * layer through the "/dev/fdset/N" special path.
 +     */
 +    if (fd_supported) {
 +        int open_flags;
 +
-+    if (ctx->thread_pool) {
++        if (flags & BDRV_O_RDWR) {
-+        thread_pool_update_params(ctx->thread_pool, ctx);
++            open_flags = O_RDWR;
-+    }
++        } else {
-+}
++            open_flags = O_RDONLY;
-diff --git a/util/main-loop.c b/util/main-loop.c
++        }
 index XXXXXXX..XXXXXXX 100644
 --- a/util/main-loop.c
 +++ b/util/main-loop.c
@@ -XXX,XX +XXX,XX @@
  #include "sysemu/replay.h"
  #include "qemu/main-loop.h"
  #include "block/aio.h"
 +#include "block/thread-pool.h"
  #include "qemu/error-report.h"
  #include "qemu/queue.h"
  #include "qemu/compiler.h"
@@ -XXX,XX +XXX,XX @@ int qemu_init_main_loop(Error **errp)
  static void main_loop_update_params(EventLoopBase *base, Error **errp)
  {
 +    ERRP_GUARD();
 +
-     if (!qemu_aio_context) {
++        fd = qemu_open(path, open_flags, errp);
-         error_setg(errp, "qemu aio context not ready");
++        if (fd < 0) {
-         return;
++            return -EINVAL;
-     }
++        }
++
-     aio_context_set_aio_params(qemu_aio_context, base->aio_max_batch, errp);
++        ret = blkio_set_int(s->blkio, "fd", fd);
-+    if (*errp) {
++        if (ret < 0) {
-+        return;
++            error_setg_errno(errp, -ret, "failed to set fd: %s",
 +                             blkio_get_error_msg());
 +            qemu_close(fd);
 +            return ret;
 +        }
 +    } else {
 +        ret = blkio_set_str(s->blkio, "path", path);
 +        if (ret < 0) {
 +            error_setg_errno(errp, -ret, "failed to set path: %s",
 +                             blkio_get_error_msg());
 +            return ret;
 +        }
 +    }
 +
-+    aio_context_set_thread_pool_params(qemu_aio_context, base->thread_pool_min,
++    qdict_del(options, "path");
-+                                       base->thread_pool_max, errp);
++
      return 0;
  }
- MainLoop *mloop;
-diff --git a/util/thread-pool.c b/util/thread-pool.c
-index XXXXXXX..XXXXXXX 100644
---- a/util/thread-pool.c
-+++ b/util/thread-pool.c
-@@ -XXX,XX +XXX,XX @@ struct ThreadPool {
-     QemuMutex lock;
-     QemuCond worker_stopped;
-     QemuSemaphore sem;
--    int max_threads;
-     QEMUBH *new_thread_bh;
-     /* The following variables are only accessed from one AioContext. */
-@@ -XXX,XX +XXX,XX @@ struct ThreadPool {
-     int new_threads;     /* backlog of threads we need to create */
-     int pending_threads; /* threads created but not running yet */
-     bool stopping;
-+    int min_threads;
-+    int max_threads;
- };
-+static inline bool back_to_sleep(ThreadPool *pool, int ret)
-+{
-+    /*
-+     * The semaphore timed out, we should exit the loop except when:
-+     *  - There is work to do, we raced with the signal.
-+     *  - The max threads threshold just changed, we raced with the signal.
-+     *  - The thread pool forces a minimum number of readily available threads.
-+     */
-+    if (ret == -1 && (!QTAILQ_EMPTY(&pool->request_list) ||
-+            pool->cur_threads > pool->max_threads ||
-+            pool->cur_threads <= pool->min_threads)) {
-+            return true;
-+    }
-+
-+    return false;
-+}
-+
- static void *worker_thread(void *opaque)
- {
-     ThreadPool *pool = opaque;
-@@ -XXX,XX +XXX,XX @@ static void *worker_thread(void *opaque)
-             ret = qemu_sem_timedwait(&pool->sem, 10000);
-             qemu_mutex_lock(&pool->lock);
-             pool->idle_threads--;
--        } while (ret == -1 && !QTAILQ_EMPTY(&pool->request_list));
--        if (ret == -1 || pool->stopping) {
-+        } while (back_to_sleep(pool, ret));
-+        if (ret == -1 || pool->stopping ||
-+            pool->cur_threads > pool->max_threads) {
-             break;
-         }
-@@ -XXX,XX +XXX,XX @@ void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg)
-     thread_pool_submit_aio(pool, func, arg, NULL, NULL);
- }
-+void thread_pool_update_params(ThreadPool *pool, AioContext *ctx)
-+{
-+    qemu_mutex_lock(&pool->lock);
-+
-+    pool->min_threads = ctx->thread_pool_min;
-+    pool->max_threads = ctx->thread_pool_max;
-+
-+    /*
-+     * We either have to:
-+     *  - Increase the number available of threads until over the min_threads
-+     *    threshold.
-+     *  - Decrease the number of available threads until under the max_threads
-+     *    threshold.
-+     *  - Do nothing. The current number of threads fall in between the min and
-+     *    max thresholds. We'll let the pool manage itself.
-+     */
-+    for (int i = pool->cur_threads; i < pool->min_threads; i++) {
-+        spawn_thread(pool);
-+    }
-+
-+    for (int i = pool->cur_threads; i > pool->max_threads; i--) {
-+        qemu_sem_post(&pool->sem);
-+    }
-+
-+    qemu_mutex_unlock(&pool->lock);
-+}
-+
- static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
- {
-     if (!ctx) {
-@@ -XXX,XX +XXX,XX @@ static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
-     qemu_mutex_init(&pool->lock);
-     qemu_cond_init(&pool->worker_stopped);
-     qemu_sem_init(&pool->sem, 0);
--    pool->max_threads = 64;
-     pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);
-     QLIST_INIT(&pool->head);
-     QTAILQ_INIT(&pool->request_list);
-+
-+    thread_pool_update_params(pool, ctx);
- }
- ThreadPool *thread_pool_new(AioContext *ctx)
 --
-.35.1
+.40.1

-[PULL 2/9] util/main-loop: Introduce the main loop into QOM
+[PULL 8/8] qapi: add '@fdset' feature for BlockdevOptionsVirtioBlkVhostVdpa
-From: Nicolas Saenz Julienne <nsaenzju@redhat.com>
+From: Stefano Garzarella <sgarzare@redhat.com>
-'event-loop-base' provides basic property handling for all 'AioContext'
+The virtio-blk-vhost-vdpa driver in libblkio 1.3.0 supports the fd
-based event loops. So let's define a new 'MainLoopClass' that inherits
+passing through the new 'fd' property.
 from it. This will permit tweaking the main loop's properties through
 qapi as well as through the command line using the '-object' keyword[1].
 Only one instance of 'MainLoopClass' might be created at any time.
-'EventLoopBaseClass' learns a new callback, 'can_be_deleted()' so as to
+Since now we are using qemu_open() on '@path' if the virtio-blk driver
-mark 'MainLoop' as non-deletable.
+supports the fd passing, let's announce it.
 In this way, the management layer can pass the file descriptor of an
 already opened vhost-vdpa character device. This is useful especially
 when the device can only be accessed with certain privileges.
-[1] For example:
+Add the '@fdset' feature only when the virtio-blk-vhost-vdpa driver
-      -object main-loop,id=main-loop,aio-max-batch=<value>
+in libblkio supports it.
-Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
+Suggested-by: Markus Armbruster <armbru@redhat.com>
 Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
-Acked-by: Markus Armbruster <armbru@redhat.com>
+Signed-off-by: Stefano Garzarella <sgarzare@redhat.com>
-Message-id: 20220425075723.20019-3-nsaenzju@redhat.com
+Message-id: 20230530071941.8954-3-sgarzare@redhat.com
 Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
 ---
- qapi/qom.json                    | 13 ++++++++
+ qapi/block-core.json | 6 ++++++
- meson.build                      |  3 +-
+ meson.build          | 4 ++++
- include/qemu/main-loop.h         | 10 ++++++
+files changed, 10 insertions(+)
  include/sysemu/event-loop-base.h |  1 +
  event-loop-base.c                | 13 ++++++++
  util/main-loop.c                 | 56 ++++++++++++++++++++++++++++++++
 files changed, 95 insertions(+), 1 deletion(-)
-diff --git a/qapi/qom.json b/qapi/qom.json
+diff --git a/qapi/block-core.json b/qapi/block-core.json
 index XXXXXXX..XXXXXXX 100644
---- a/qapi/qom.json
+--- a/qapi/block-core.json
-+++ b/qapi/qom.json
++++ b/qapi/block-core.json
 @@ -XXX,XX +XXX,XX @@
-             '*poll-grow': 'int',
+ #
-             '*poll-shrink': 'int' } }
+ # @path: path to the vhost-vdpa character device.
+ #
-+##
++# Features:
-+# @MainLoopProperties:
++# @fdset: Member @path supports the special "/dev/fdset/N" path
 +#     (since 8.1)
 +#
-+# Properties for the main-loop object.
+ # Since: 7.2
 +#
 +# Since: 7.1
 +##
 +{ 'struct': 'MainLoopProperties',
 +  'base': 'EventLoopBaseProperties',
 +  'data': {} }
 +
  ##
- # @MemoryBackendProperties:
+ { 'struct': 'BlockdevOptionsVirtioBlkVhostVdpa',
- #
+   'data': { 'path': 'str' },
-@@ -XXX,XX +XXX,XX @@
++  'features': [ { 'name' :'fdset',
-     { 'name': 'input-linux',
++                  'if': 'CONFIG_BLKIO_VHOST_VDPA_FD' } ],
-       'if': 'CONFIG_LINUX' },
+   'if': 'CONFIG_BLKIO' }
-     'iothread',
-+    'main-loop',
+ ##
      { 'name': 'memory-backend-epc',
        'if': 'CONFIG_LINUX' },
      'memory-backend-file',
@@ -XXX,XX +XXX,XX @@
        'input-linux':                { 'type': 'InputLinuxProperties',
                                        'if': 'CONFIG_LINUX' },
        'iothread':                   'IothreadProperties',
 +      'main-loop':                  'MainLoopProperties',
        'memory-backend-epc':         { 'type': 'MemoryBackendEpcProperties',
                                        'if': 'CONFIG_LINUX' },
        'memory-backend-file':        'MemoryBackendFileProperties',
 diff --git a/meson.build b/meson.build
 index XXXXXXX..XXXXXXX 100644
 --- a/meson.build
 +++ b/meson.build
-@@ -XXX,XX +XXX,XX @@ libqemuutil = static_library('qemuutil',
+@@ -XXX,XX +XXX,XX @@ config_host_data.set('CONFIG_LZO', lzo.found())
-                              sources: util_ss.sources() + stub_ss.sources() + genh,
+ config_host_data.set('CONFIG_MPATH', mpathpersist.found())
-                              dependencies: [util_ss.dependencies(), libm, threads, glib, socket, malloc, pixman])
+ config_host_data.set('CONFIG_MPATH_NEW_API', mpathpersist_new_api)
- qemuutil = declare_dependency(link_with: libqemuutil,
+ config_host_data.set('CONFIG_BLKIO', blkio.found())
--                              sources: genh + version_res)
++if blkio.found()
-+                              sources: genh + version_res,
++  config_host_data.set('CONFIG_BLKIO_VHOST_VDPA_FD',
-+                              dependencies: [event_loop_base])
++                       blkio.version().version_compare('>=1.3.0'))
++endif
- if have_system or have_user
+ config_host_data.set('CONFIG_CURL', curl.found())
-   decodetree = generator(find_program('scripts/decodetree.py'),
+ config_host_data.set('CONFIG_CURSES', curses.found())
-diff --git a/include/qemu/main-loop.h b/include/qemu/main-loop.h
+ config_host_data.set('CONFIG_GBM', gbm.found())
 index XXXXXXX..XXXXXXX 100644
 --- a/include/qemu/main-loop.h
 +++ b/include/qemu/main-loop.h
@@ -XXX,XX +XXX,XX @@
  #define QEMU_MAIN_LOOP_H
  #include "block/aio.h"
 +#include "qom/object.h"
 +#include "sysemu/event-loop-base.h"
  #define SIG_IPI SIGUSR1
 +#define TYPE_MAIN_LOOP  "main-loop"
 +OBJECT_DECLARE_TYPE(MainLoop, MainLoopClass, MAIN_LOOP)
 +
 +struct MainLoop {
 +    EventLoopBase parent_obj;
 +};
 +typedef struct MainLoop MainLoop;
 +
  /**
   * qemu_init_main_loop: Set up the process so that it can run the main loop.
   *
 diff --git a/include/sysemu/event-loop-base.h b/include/sysemu/event-loop-base.h
 index XXXXXXX..XXXXXXX 100644
 --- a/include/sysemu/event-loop-base.h
 +++ b/include/sysemu/event-loop-base.h
@@ -XXX,XX +XXX,XX @@ struct EventLoopBaseClass {
      void (*init)(EventLoopBase *base, Error **errp);
      void (*update_params)(EventLoopBase *base, Error **errp);
 +    bool (*can_be_deleted)(EventLoopBase *base);
  };
  struct EventLoopBase {
 diff --git a/event-loop-base.c b/event-loop-base.c
 index XXXXXXX..XXXXXXX 100644
 --- a/event-loop-base.c
 +++ b/event-loop-base.c
@@ -XXX,XX +XXX,XX @@ static void event_loop_base_complete(UserCreatable *uc, Error **errp)
      }
  }
 +static bool event_loop_base_can_be_deleted(UserCreatable *uc)
 +{
 +    EventLoopBaseClass *bc = EVENT_LOOP_BASE_GET_CLASS(uc);
 +    EventLoopBase *backend = EVENT_LOOP_BASE(uc);
 +
 +    if (bc->can_be_deleted) {
 +        return bc->can_be_deleted(backend);
 +    }
 +
 +    return true;
 +}
 +
  static void event_loop_base_class_init(ObjectClass *klass, void *class_data)
  {
      UserCreatableClass *ucc = USER_CREATABLE_CLASS(klass);
      ucc->complete = event_loop_base_complete;
 +    ucc->can_be_deleted = event_loop_base_can_be_deleted;
      object_class_property_add(klass, "aio-max-batch", "int",
                                event_loop_base_get_param,
 diff --git a/util/main-loop.c b/util/main-loop.c
 index XXXXXXX..XXXXXXX 100644
 --- a/util/main-loop.c
 +++ b/util/main-loop.c
@@ -XXX,XX +XXX,XX @@
  #include "qemu/error-report.h"
  #include "qemu/queue.h"
  #include "qemu/compiler.h"
 +#include "qom/object.h"
  #ifndef _WIN32
  #include <sys/wait.h>
@@ -XXX,XX +XXX,XX @@ int qemu_init_main_loop(Error **errp)
      return 0;
  }
 +static void main_loop_update_params(EventLoopBase *base, Error **errp)
 +{
 +    if (!qemu_aio_context) {
 +        error_setg(errp, "qemu aio context not ready");
 +        return;
 +    }
 +
 +    aio_context_set_aio_params(qemu_aio_context, base->aio_max_batch, errp);
 +}
 +
 +MainLoop *mloop;
 +
 +static void main_loop_init(EventLoopBase *base, Error **errp)
 +{
 +    MainLoop *m = MAIN_LOOP(base);
 +
 +    if (mloop) {
 +        error_setg(errp, "only one main-loop instance allowed");
 +        return;
 +    }
 +
 +    main_loop_update_params(base, errp);
 +
 +    mloop = m;
 +    return;
 +}
 +
 +static bool main_loop_can_be_deleted(EventLoopBase *base)
 +{
 +    return false;
 +}
 +
 +static void main_loop_class_init(ObjectClass *oc, void *class_data)
 +{
 +    EventLoopBaseClass *bc = EVENT_LOOP_BASE_CLASS(oc);
 +
 +    bc->init = main_loop_init;
 +    bc->update_params = main_loop_update_params;
 +    bc->can_be_deleted = main_loop_can_be_deleted;
 +}
 +
 +static const TypeInfo main_loop_info = {
 +    .name = TYPE_MAIN_LOOP,
 +    .parent = TYPE_EVENT_LOOP_BASE,
 +    .class_init = main_loop_class_init,
 +    .instance_size = sizeof(MainLoop),
 +};
 +
 +static void main_loop_register_types(void)
 +{
 +    type_register_static(&main_loop_info);
 +}
 +
 +type_init(main_loop_register_types)
 +
  static int max_priority;
  #ifndef _WIN32
 --
-.35.1
+.40.1

-[PULL 4/9] virtio-scsi: fix ctrl and event handler functions in dataplane mode
+Deleted patch
-Commit f34e8d8b8d48d73f36a67b6d5e492ef9784b5012 ("virtio-scsi: prepare
-virtio_scsi_handle_cmd for dataplane") prepared the virtio-scsi cmd
-virtqueue handler function to be used in both the dataplane and
-non-datpalane code paths.
-It failed to convert the ctrl and event virtqueue handler functions,
-which are not designed to be called from the dataplane code path but
-will be since the ioeventfd is set up for those virtqueues when
-dataplane starts.
-Convert the ctrl and event virtqueue handler functions now so they
-operate correctly when called from the dataplane code path. Avoid code
-duplication by extracting this code into a helper function.
-Fixes: f34e8d8b8d48d73f36a67b6d5e492ef9784b5012 ("virtio-scsi: prepare virtio_scsi_handle_cmd for dataplane")
-Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
-Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
-Message-id: 20220427143541.119567-2-stefanha@redhat.com
-[Fixed s/by used/be used/ typo pointed out by Michael Tokarev
-<mjt@tls.msk.ru>.
---Stefan]
-Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
----
- hw/scsi/virtio-scsi.c | 42 +++++++++++++++++++++++++++---------------
-file changed, 27 insertions(+), 15 deletions(-)
-diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
-index XXXXXXX..XXXXXXX 100644
---- a/hw/scsi/virtio-scsi.c
-+++ b/hw/scsi/virtio-scsi.c
-@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq)
-     return progress;
- }
-+/*
-+ * If dataplane is configured but not yet started, do so now and return true on
-+ * success.
-+ *
-+ * Dataplane is started by the core virtio code but virtqueue handler functions
-+ * can also be invoked when a guest kicks before DRIVER_OK, so this helper
-+ * function helps us deal with manually starting ioeventfd in that case.
-+ */
-+static bool virtio_scsi_defer_to_dataplane(VirtIOSCSI *s)
-+{
-+    if (!s->ctx || s->dataplane_started) {
-+        return false;
-+    }
-+
-+    virtio_device_start_ioeventfd(&s->parent_obj.parent_obj);
-+    return !s->dataplane_fenced;
-+}
-+
- static void virtio_scsi_handle_ctrl(VirtIODevice *vdev, VirtQueue *vq)
- {
-     VirtIOSCSI *s = (VirtIOSCSI *)vdev;
--    if (s->ctx) {
--        virtio_device_start_ioeventfd(vdev);
--        if (!s->dataplane_fenced) {
--            return;
--        }
-+    if (virtio_scsi_defer_to_dataplane(s)) {
-+        return;
-     }
-+
-     virtio_scsi_acquire(s);
-     virtio_scsi_handle_ctrl_vq(s, vq);
-     virtio_scsi_release(s);
-@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_cmd(VirtIODevice *vdev, VirtQueue *vq)
-     /* use non-QOM casts in the data path */
-     VirtIOSCSI *s = (VirtIOSCSI *)vdev;
--    if (s->ctx && !s->dataplane_started) {
--        virtio_device_start_ioeventfd(vdev);
--        if (!s->dataplane_fenced) {
--            return;
--        }
-+    if (virtio_scsi_defer_to_dataplane(s)) {
-+        return;
-     }
-+
-     virtio_scsi_acquire(s);
-     virtio_scsi_handle_cmd_vq(s, vq);
-     virtio_scsi_release(s);
-@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_event(VirtIODevice *vdev, VirtQueue *vq)
- {
-     VirtIOSCSI *s = VIRTIO_SCSI(vdev);
--    if (s->ctx) {
--        virtio_device_start_ioeventfd(vdev);
--        if (!s->dataplane_fenced) {
--            return;
--        }
-+    if (virtio_scsi_defer_to_dataplane(s)) {
-+        return;
-     }
-+
-     virtio_scsi_acquire(s);
-     virtio_scsi_handle_event_vq(s, vq);
-     virtio_scsi_release(s);
---
-.35.1

The following changes since commit 554623226f800acf48a2ed568900c1c968ec9a8b:

Merge tag 'qemu-sparc-20220508' of https://github.com/mcayland/qemu into staging (2022-05-08 17:03:26 -0500)

are available in the Git repository at:

https://gitlab.com/stefanha/qemu.git tags/block-pull-request

for you to fetch changes up to 3dc584abeef0e1277c2de8c1c1974cb49444eb0a:

virtio-scsi: move request-related items from .h to .c (2022-05-09 10:45:04 +0100)

----------------------------------------------------------------
Pull request

- Add new thread-pool-min/thread-pool-max parameters to control the thread pool
  used for async I/O.

- Fix virtio-scsi IOThread 100% CPU consumption QEMU 7.0 regression.

----------------------------------------------------------------

Nicolas Saenz Julienne (3):
  Introduce event-loop-base abstract class
  util/main-loop: Introduce the main loop into QOM
  util/event-loop-base: Introduce options to set the thread pool size

Stefan Hajnoczi (6):
  virtio-scsi: fix ctrl and event handler functions in dataplane mode
  virtio-scsi: don't waste CPU polling the event virtqueue
  virtio-scsi: clean up virtio_scsi_handle_event_vq()
  virtio-scsi: clean up virtio_scsi_handle_ctrl_vq()
  virtio-scsi: clean up virtio_scsi_handle_cmd_vq()
  virtio-scsi: move request-related items from .h to .c

-- 
2.35.1

From: Nicolas Saenz Julienne <nsaenzju@redhat.com>

Introduce the 'event-loop-base' abstract class, it'll hold the
properties common to all event loops and provide the necessary hooks for
their creation and maintenance. Then have iothread inherit from it.

EventLoopBaseClass is defined as user creatable and provides a hook for
its children to attach themselves to the user creatable class 'complete'
function. It also provides an update_params() callback to propagate
property changes onto its children.

The new 'event-loop-base' class will live in the root directory. It is
built on its own using the 'link_whole' option (there are no direct
function dependencies between the class and its children, it all happens
trough 'constructor' magic). And also imposes new compilation
dependencies:

qom <- event-loop-base <- blockdev (iothread.c)

And in subsequent patches:

qom <- event-loop-base <- qemuutil (util/main-loop.c)

All this forced some amount of reordering in meson.build:

- Moved qom build definition before qemuutil. Doing it the other way
   around (i.e. moving qemuutil after qom) isn't possible as a lot of
   core libraries that live in between the two depend on it.

- Process the 'hw' subdir earlier, as it introduces files into the
   'qom' source set.

No functional changes intended.

Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Acked-by: Markus Armbruster <armbru@redhat.com>
Message-id: 20220425075723.20019-2-nsaenzju@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 qapi/qom.json                    |  22 +++++--
 meson.build                      |  23 ++++---
 include/sysemu/event-loop-base.h |  36 +++++++++++
 include/sysemu/iothread.h        |   6 +-
 event-loop-base.c                | 104 +++++++++++++++++++++++++++++++
 iothread.c                       |  65 ++++++-------------
 6 files changed, 192 insertions(+), 64 deletions(-)
 create mode 100644 include/sysemu/event-loop-base.h
 create mode 100644 event-loop-base.c

diff --git a/qapi/qom.json b/qapi/qom.json
index XXXXXXX..XXXXXXX 100644
--- a/qapi/qom.json
+++ b/qapi/qom.json
@@ -XXX,XX +XXX,XX @@
             '*repeat': 'bool',
             '*grab-toggle': 'GrabToggleKeys' } }
 
+##
+# @EventLoopBaseProperties:
+#
+# Common properties for event loops
+#
+# @aio-max-batch: maximum number of requests in a batch for the AIO engine,
+#                 0 means that the engine will use its default.
+#                 (default: 0)
+#
+# Since: 7.1
+##
+{ 'struct': 'EventLoopBaseProperties',
+  'data': { '*aio-max-batch': 'int' } }
+
 ##
 # @IothreadProperties:
 #
@@ -XXX,XX +XXX,XX @@
 #               algorithm detects it is spending too long polling without
 #               encountering events. 0 selects a default behaviour (default: 0)
 #
-# @aio-max-batch: maximum number of requests in a batch for the AIO engine,
-#                 0 means that the engine will use its default
-#                 (default:0, since 6.1)
+# The @aio-max-batch option is available since 6.1.
 #
 # Since: 2.0
 ##
 { 'struct': 'IothreadProperties',
+  'base': 'EventLoopBaseProperties',
   'data': { '*poll-max-ns': 'int',
             '*poll-grow': 'int',
-            '*poll-shrink': 'int',
-            '*aio-max-batch': 'int' } }
+            '*poll-shrink': 'int' } }
 
 ##
 # @MemoryBackendProperties:
diff --git a/meson.build b/meson.build
index XXXXXXX..XXXXXXX 100644
--- a/meson.build
+++ b/meson.build
@@ -XXX,XX +XXX,XX @@ subdir('qom')
 subdir('authz')
 subdir('crypto')
 subdir('ui')
+subdir('hw')
 
 
 if enable_modules
@@ -XXX,XX +XXX,XX @@ if enable_modules
   modulecommon = declare_dependency(link_whole: libmodulecommon, compile_args: '-DBUILD_DSO')
 endif
 
+qom_ss = qom_ss.apply(config_host, strict: false)
+libqom = static_library('qom', qom_ss.sources() + genh,
+                        dependencies: [qom_ss.dependencies()],
+                        name_suffix: 'fa')
+qom = declare_dependency(link_whole: libqom)
+
+event_loop_base = files('event-loop-base.c')
+event_loop_base = static_library('event-loop-base', sources: event_loop_base + genh,
+                                 build_by_default: true)
+event_loop_base = declare_dependency(link_whole: event_loop_base,
+                                     dependencies: [qom])
+
 stub_ss = stub_ss.apply(config_all, strict: false)
 
 util_ss.add_all(trace_ss)
@@ -XXX,XX +XXX,XX @@ subdir('monitor')
 subdir('net')
 subdir('replay')
 subdir('semihosting')
-subdir('hw')
 subdir('tcg')
 subdir('fpu')
 subdir('accel')
@@ -XXX,XX +XXX,XX @@ qemu_syms = custom_target('qemu.syms', output: 'qemu.syms',
                              capture: true,
                              command: [undefsym, nm, '@INPUT@'])
 
-qom_ss = qom_ss.apply(config_host, strict: false)
-libqom = static_library('qom', qom_ss.sources() + genh,
-                        dependencies: [qom_ss.dependencies()],
-                        name_suffix: 'fa')
-
-qom = declare_dependency(link_whole: libqom)
-
 authz_ss = authz_ss.apply(config_host, strict: false)
 libauthz = static_library('authz', authz_ss.sources() + genh,
                           dependencies: [authz_ss.dependencies()],
@@ -XXX,XX +XXX,XX @@ libblockdev = static_library('blockdev', blockdev_ss.sources() + genh,
                              build_by_default: false)
 
 blockdev = declare_dependency(link_whole: [libblockdev],
-                              dependencies: [block])
+                              dependencies: [block, event_loop_base])
 
 qmp_ss = qmp_ss.apply(config_host, strict: false)
 libqmp = static_library('qmp', qmp_ss.sources() + genh,
diff --git a/include/sysemu/event-loop-base.h b/include/sysemu/event-loop-base.h
new file mode 100644
index XXXXXXX..XXXXXXX
--- /dev/null
+++ b/include/sysemu/event-loop-base.h
@@ -XXX,XX +XXX,XX @@
+/*
+ * QEMU event-loop backend
+ *
+ * Copyright (C) 2022 Red Hat Inc
+ *
+ * Authors:
+ *  Nicolas Saenz Julienne <nsaenzju@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or later.
+ * See the COPYING file in the top-level directory.
+ */
+#ifndef QEMU_EVENT_LOOP_BASE_H
+#define QEMU_EVENT_LOOP_BASE_H
+
+#include "qom/object.h"
+#include "block/aio.h"
+#include "qemu/typedefs.h"
+
+#define TYPE_EVENT_LOOP_BASE         "event-loop-base"
+OBJECT_DECLARE_TYPE(EventLoopBase, EventLoopBaseClass,
+                    EVENT_LOOP_BASE)
+
+struct EventLoopBaseClass {
+    ObjectClass parent_class;
+
+    void (*init)(EventLoopBase *base, Error **errp);
+    void (*update_params)(EventLoopBase *base, Error **errp);
+};
+
+struct EventLoopBase {
+    Object parent;
+
+    /* AioContext AIO engine parameters */
+    int64_t aio_max_batch;
+};
+#endif
diff --git a/include/sysemu/iothread.h b/include/sysemu/iothread.h
index XXXXXXX..XXXXXXX 100644
--- a/include/sysemu/iothread.h
+++ b/include/sysemu/iothread.h
@@ -XXX,XX +XXX,XX @@
 #include "block/aio.h"
 #include "qemu/thread.h"
 #include "qom/object.h"
+#include "sysemu/event-loop-base.h"
 
 #define TYPE_IOTHREAD "iothread"
 
 struct IOThread {
-    Object parent_obj;
+    EventLoopBase parent_obj;
 
     QemuThread thread;
     AioContext *ctx;
@@ -XXX,XX +XXX,XX @@ struct IOThread {
     int64_t poll_max_ns;
     int64_t poll_grow;
     int64_t poll_shrink;
-
-    /* AioContext AIO engine parameters */
-    int64_t aio_max_batch;
 };
 typedef struct IOThread IOThread;
 
diff --git a/event-loop-base.c b/event-loop-base.c
new file mode 100644
index XXXXXXX..XXXXXXX
--- /dev/null
+++ b/event-loop-base.c
@@ -XXX,XX +XXX,XX @@
+/*
+ * QEMU event-loop base
+ *
+ * Copyright (C) 2022 Red Hat Inc
+ *
+ * Authors:
+ *  Stefan Hajnoczi <stefanha@redhat.com>
+ *  Nicolas Saenz Julienne <nsaenzju@redhat.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or later.
+ * See the COPYING file in the top-level directory.
+ */
+
+#include "qemu/osdep.h"
+#include "qom/object_interfaces.h"
+#include "qapi/error.h"
+#include "sysemu/event-loop-base.h"
+
+typedef struct {
+    const char *name;
+    ptrdiff_t offset; /* field's byte offset in EventLoopBase struct */
+} EventLoopBaseParamInfo;
+
+static EventLoopBaseParamInfo aio_max_batch_info = {
+    "aio-max-batch", offsetof(EventLoopBase, aio_max_batch),
+};
+
+static void event_loop_base_get_param(Object *obj, Visitor *v,
+        const char *name, void *opaque, Error **errp)
+{
+    EventLoopBase *event_loop_base = EVENT_LOOP_BASE(obj);
+    EventLoopBaseParamInfo *info = opaque;
+    int64_t *field = (void *)event_loop_base + info->offset;
+
+    visit_type_int64(v, name, field, errp);
+}
+
+static void event_loop_base_set_param(Object *obj, Visitor *v,
+        const char *name, void *opaque, Error **errp)
+{
+    EventLoopBaseClass *bc = EVENT_LOOP_BASE_GET_CLASS(obj);
+    EventLoopBase *base = EVENT_LOOP_BASE(obj);
+    EventLoopBaseParamInfo *info = opaque;
+    int64_t *field = (void *)base + info->offset;
+    int64_t value;
+
+    if (!visit_type_int64(v, name, &value, errp)) {
+        return;
+    }
+
+    if (value < 0) {
+        error_setg(errp, "%s value must be in range [0, %" PRId64 "]",
+                   info->name, INT64_MAX);
+        return;
+    }
+
+    *field = value;
+
+    if (bc->update_params) {
+        bc->update_params(base, errp);
+    }
+
+    return;
+}
+
+static void event_loop_base_complete(UserCreatable *uc, Error **errp)
+{
+    EventLoopBaseClass *bc = EVENT_LOOP_BASE_GET_CLASS(uc);
+    EventLoopBase *base = EVENT_LOOP_BASE(uc);
+
+    if (bc->init) {
+        bc->init(base, errp);
+    }
+}
+
+static void event_loop_base_class_init(ObjectClass *klass, void *class_data)
+{
+    UserCreatableClass *ucc = USER_CREATABLE_CLASS(klass);
+    ucc->complete = event_loop_base_complete;
+
+    object_class_property_add(klass, "aio-max-batch", "int",
+                              event_loop_base_get_param,
+                              event_loop_base_set_param,
+                              NULL, &aio_max_batch_info);
+}
+
+static const TypeInfo event_loop_base_info = {
+    .name = TYPE_EVENT_LOOP_BASE,
+    .parent = TYPE_OBJECT,
+    .instance_size = sizeof(EventLoopBase),
+    .class_size = sizeof(EventLoopBaseClass),
+    .class_init = event_loop_base_class_init,
+    .abstract = true,
+    .interfaces = (InterfaceInfo[]) {
+        { TYPE_USER_CREATABLE },
+        { }
+    }
+};
+
+static void register_types(void)
+{
+    type_register_static(&event_loop_base_info);
+}
+type_init(register_types);
diff --git a/iothread.c b/iothread.c
index XXXXXXX..XXXXXXX 100644
--- a/iothread.c
+++ b/iothread.c
@@ -XXX,XX +XXX,XX @@
 #include "qemu/module.h"
 #include "block/aio.h"
 #include "block/block.h"
+#include "sysemu/event-loop-base.h"
 #include "sysemu/iothread.h"
 #include "qapi/error.h"
 #include "qapi/qapi-commands-misc.h"
@@ -XXX,XX +XXX,XX @@ static void iothread_init_gcontext(IOThread *iothread)
     iothread->main_loop = g_main_loop_new(iothread->worker_context, TRUE);
 }
 
-static void iothread_set_aio_context_params(IOThread *iothread, Error **errp)
+static void iothread_set_aio_context_params(EventLoopBase *base, Error **errp)
 {
+    IOThread *iothread = IOTHREAD(base);
     ERRP_GUARD();
 
+    if (!iothread->ctx) {
+        return;
+    }
+
     aio_context_set_poll_params(iothread->ctx,
                                 iothread->poll_max_ns,
                                 iothread->poll_grow,
@@ -XXX,XX +XXX,XX @@ static void iothread_set_aio_context_params(IOThread *iothread, Error **errp)
     }
 
     aio_context_set_aio_params(iothread->ctx,
-                               iothread->aio_max_batch,
+                               iothread->parent_obj.aio_max_batch,
                                errp);
 }
 
-static void iothread_complete(UserCreatable *obj, Error **errp)
+
+static void iothread_init(EventLoopBase *base, Error **errp)
 {
     Error *local_error = NULL;
-    IOThread *iothread = IOTHREAD(obj);
+    IOThread *iothread = IOTHREAD(base);
     char *thread_name;
 
     iothread->stopping = false;
@@ -XXX,XX +XXX,XX @@ static void iothread_complete(UserCreatable *obj, Error **errp)
      */
     iothread_init_gcontext(iothread);
 
-    iothread_set_aio_context_params(iothread, &local_error);
+    iothread_set_aio_context_params(base, &local_error);
     if (local_error) {
         error_propagate(errp, local_error);
         aio_context_unref(iothread->ctx);
@@ -XXX,XX +XXX,XX @@ static void iothread_complete(UserCreatable *obj, Error **errp)
      * to inherit.
      */
     thread_name = g_strdup_printf("IO %s",
-                        object_get_canonical_path_component(OBJECT(obj)));
+                        object_get_canonical_path_component(OBJECT(base)));
     qemu_thread_create(&iothread->thread, thread_name, iothread_run,
                        iothread, QEMU_THREAD_JOINABLE);
     g_free(thread_name);
@@ -XXX,XX +XXX,XX @@ static IOThreadParamInfo poll_grow_info = {
 static IOThreadParamInfo poll_shrink_info = {
     "poll-shrink", offsetof(IOThread, poll_shrink),
 };
-static IOThreadParamInfo aio_max_batch_info = {
-    "aio-max-batch", offsetof(IOThread, aio_max_batch),
-};
 
 static void iothread_get_param(Object *obj, Visitor *v,
         const char *name, IOThreadParamInfo *info, Error **errp)
@@ -XXX,XX +XXX,XX @@ static void iothread_set_poll_param(Object *obj, Visitor *v,
     }
 }
 
-static void iothread_get_aio_param(Object *obj, Visitor *v,
-        const char *name, void *opaque, Error **errp)
-{
-    IOThreadParamInfo *info = opaque;
-
-    iothread_get_param(obj, v, name, info, errp);
-}
-
-static void iothread_set_aio_param(Object *obj, Visitor *v,
-        const char *name, void *opaque, Error **errp)
-{
-    IOThread *iothread = IOTHREAD(obj);
-    IOThreadParamInfo *info = opaque;
-
-    if (!iothread_set_param(obj, v, name, info, errp)) {
-        return;
-    }
-
-    if (iothread->ctx) {
-        aio_context_set_aio_params(iothread->ctx,
-                                   iothread->aio_max_batch,
-                                   errp);
-    }
-}
-
 static void iothread_class_init(ObjectClass *klass, void *class_data)
 {
-    UserCreatableClass *ucc = USER_CREATABLE_CLASS(klass);
-    ucc->complete = iothread_complete;
+    EventLoopBaseClass *bc = EVENT_LOOP_BASE_CLASS(klass);
+
+    bc->init = iothread_init;
+    bc->update_params = iothread_set_aio_context_params;
 
     object_class_property_add(klass, "poll-max-ns", "int",
                               iothread_get_poll_param,
@@ -XXX,XX +XXX,XX @@ static void iothread_class_init(ObjectClass *klass, void *class_data)
                               iothread_get_poll_param,
                               iothread_set_poll_param,
                               NULL, &poll_shrink_info);
-    object_class_property_add(klass, "aio-max-batch", "int",
-                              iothread_get_aio_param,
-                              iothread_set_aio_param,
-                              NULL, &aio_max_batch_info);
 }
 
 static const TypeInfo iothread_info = {
     .name = TYPE_IOTHREAD,
-    .parent = TYPE_OBJECT,
+    .parent = TYPE_EVENT_LOOP_BASE,
     .class_init = iothread_class_init,
     .instance_size = sizeof(IOThread),
     .instance_init = iothread_instance_init,
     .instance_finalize = iothread_instance_finalize,
-    .interfaces = (InterfaceInfo[]) {
-        {TYPE_USER_CREATABLE},
-        {}
-    },
 };
 
 static void iothread_register_types(void)
@@ -XXX,XX +XXX,XX @@ static int query_one_iothread(Object *object, void *opaque)
     info->poll_max_ns = iothread->poll_max_ns;
     info->poll_grow = iothread->poll_grow;
     info->poll_shrink = iothread->poll_shrink;
-    info->aio_max_batch = iothread->aio_max_batch;
+    info->aio_max_batch = iothread->parent_obj.aio_max_batch;
 
     QAPI_LIST_APPEND(*tail, info);
     return 0;
-- 
2.35.1

From: Nicolas Saenz Julienne <nsaenzju@redhat.com>

'event-loop-base' provides basic property handling for all 'AioContext'
based event loops. So let's define a new 'MainLoopClass' that inherits
from it. This will permit tweaking the main loop's properties through
qapi as well as through the command line using the '-object' keyword[1].
Only one instance of 'MainLoopClass' might be created at any time.

'EventLoopBaseClass' learns a new callback, 'can_be_deleted()' so as to
mark 'MainLoop' as non-deletable.

[1] For example:
      -object main-loop,id=main-loop,aio-max-batch=<value>

Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Acked-by: Markus Armbruster <armbru@redhat.com>
Message-id: 20220425075723.20019-3-nsaenzju@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 qapi/qom.json                    | 13 ++++++++
 meson.build                      |  3 +-
 include/qemu/main-loop.h         | 10 ++++++
 include/sysemu/event-loop-base.h |  1 +
 event-loop-base.c                | 13 ++++++++
 util/main-loop.c                 | 56 ++++++++++++++++++++++++++++++++
 6 files changed, 95 insertions(+), 1 deletion(-)

diff --git a/qapi/qom.json b/qapi/qom.json
index XXXXXXX..XXXXXXX 100644
--- a/qapi/qom.json
+++ b/qapi/qom.json
@@ -XXX,XX +XXX,XX @@
             '*poll-grow': 'int',
             '*poll-shrink': 'int' } }
 
+##
+# @MainLoopProperties:
+#
+# Properties for the main-loop object.
+#
+# Since: 7.1
+##
+{ 'struct': 'MainLoopProperties',
+  'base': 'EventLoopBaseProperties',
+  'data': {} }
+
 ##
 # @MemoryBackendProperties:
 #
@@ -XXX,XX +XXX,XX @@
     { 'name': 'input-linux',
       'if': 'CONFIG_LINUX' },
     'iothread',
+    'main-loop',
     { 'name': 'memory-backend-epc',
       'if': 'CONFIG_LINUX' },
     'memory-backend-file',
@@ -XXX,XX +XXX,XX @@
       'input-linux':                { 'type': 'InputLinuxProperties',
                                       'if': 'CONFIG_LINUX' },
       'iothread':                   'IothreadProperties',
+      'main-loop':                  'MainLoopProperties',
       'memory-backend-epc':         { 'type': 'MemoryBackendEpcProperties',
                                       'if': 'CONFIG_LINUX' },
       'memory-backend-file':        'MemoryBackendFileProperties',
diff --git a/meson.build b/meson.build
index XXXXXXX..XXXXXXX 100644
--- a/meson.build
+++ b/meson.build
@@ -XXX,XX +XXX,XX @@ libqemuutil = static_library('qemuutil',
                              sources: util_ss.sources() + stub_ss.sources() + genh,
                              dependencies: [util_ss.dependencies(), libm, threads, glib, socket, malloc, pixman])
 qemuutil = declare_dependency(link_with: libqemuutil,
-                              sources: genh + version_res)
+                              sources: genh + version_res,
+                              dependencies: [event_loop_base])
 
 if have_system or have_user
   decodetree = generator(find_program('scripts/decodetree.py'),
diff --git a/include/qemu/main-loop.h b/include/qemu/main-loop.h
index XXXXXXX..XXXXXXX 100644
--- a/include/qemu/main-loop.h
+++ b/include/qemu/main-loop.h
@@ -XXX,XX +XXX,XX @@
 #define QEMU_MAIN_LOOP_H
 
 #include "block/aio.h"
+#include "qom/object.h"
+#include "sysemu/event-loop-base.h"
 
 #define SIG_IPI SIGUSR1
 
+#define TYPE_MAIN_LOOP  "main-loop"
+OBJECT_DECLARE_TYPE(MainLoop, MainLoopClass, MAIN_LOOP)
+
+struct MainLoop {
+    EventLoopBase parent_obj;
+};
+typedef struct MainLoop MainLoop;
+
 /**
  * qemu_init_main_loop: Set up the process so that it can run the main loop.
  *
diff --git a/include/sysemu/event-loop-base.h b/include/sysemu/event-loop-base.h
index XXXXXXX..XXXXXXX 100644
--- a/include/sysemu/event-loop-base.h
+++ b/include/sysemu/event-loop-base.h
@@ -XXX,XX +XXX,XX @@ struct EventLoopBaseClass {
 
     void (*init)(EventLoopBase *base, Error **errp);
     void (*update_params)(EventLoopBase *base, Error **errp);
+    bool (*can_be_deleted)(EventLoopBase *base);
 };
 
 struct EventLoopBase {
diff --git a/event-loop-base.c b/event-loop-base.c
index XXXXXXX..XXXXXXX 100644
--- a/event-loop-base.c
+++ b/event-loop-base.c
@@ -XXX,XX +XXX,XX @@ static void event_loop_base_complete(UserCreatable *uc, Error **errp)
     }
 }
 
+static bool event_loop_base_can_be_deleted(UserCreatable *uc)
+{
+    EventLoopBaseClass *bc = EVENT_LOOP_BASE_GET_CLASS(uc);
+    EventLoopBase *backend = EVENT_LOOP_BASE(uc);
+
+    if (bc->can_be_deleted) {
+        return bc->can_be_deleted(backend);
+    }
+
+    return true;
+}
+
 static void event_loop_base_class_init(ObjectClass *klass, void *class_data)
 {
     UserCreatableClass *ucc = USER_CREATABLE_CLASS(klass);
     ucc->complete = event_loop_base_complete;
+    ucc->can_be_deleted = event_loop_base_can_be_deleted;
 
     object_class_property_add(klass, "aio-max-batch", "int",
                               event_loop_base_get_param,
diff --git a/util/main-loop.c b/util/main-loop.c
index XXXXXXX..XXXXXXX 100644
--- a/util/main-loop.c
+++ b/util/main-loop.c
@@ -XXX,XX +XXX,XX @@
 #include "qemu/error-report.h"
 #include "qemu/queue.h"
 #include "qemu/compiler.h"
+#include "qom/object.h"
 
 #ifndef _WIN32
 #include <sys/wait.h>
@@ -XXX,XX +XXX,XX @@ int qemu_init_main_loop(Error **errp)
     return 0;
 }
 
+static void main_loop_update_params(EventLoopBase *base, Error **errp)
+{
+    if (!qemu_aio_context) {
+        error_setg(errp, "qemu aio context not ready");
+        return;
+    }
+
+    aio_context_set_aio_params(qemu_aio_context, base->aio_max_batch, errp);
+}
+
+MainLoop *mloop;
+
+static void main_loop_init(EventLoopBase *base, Error **errp)
+{
+    MainLoop *m = MAIN_LOOP(base);
+
+    if (mloop) {
+        error_setg(errp, "only one main-loop instance allowed");
+        return;
+    }
+
+    main_loop_update_params(base, errp);
+
+    mloop = m;
+    return;
+}
+
+static bool main_loop_can_be_deleted(EventLoopBase *base)
+{
+    return false;
+}
+
+static void main_loop_class_init(ObjectClass *oc, void *class_data)
+{
+    EventLoopBaseClass *bc = EVENT_LOOP_BASE_CLASS(oc);
+
+    bc->init = main_loop_init;
+    bc->update_params = main_loop_update_params;
+    bc->can_be_deleted = main_loop_can_be_deleted;
+}
+
+static const TypeInfo main_loop_info = {
+    .name = TYPE_MAIN_LOOP,
+    .parent = TYPE_EVENT_LOOP_BASE,
+    .class_init = main_loop_class_init,
+    .instance_size = sizeof(MainLoop),
+};
+
+static void main_loop_register_types(void)
+{
+    type_register_static(&main_loop_info);
+}
+
+type_init(main_loop_register_types)
+
 static int max_priority;
 
 #ifndef _WIN32
-- 
2.35.1

From: Nicolas Saenz Julienne <nsaenzju@redhat.com>

The thread pool regulates itself: when idle, it kills threads until
empty, when in demand, it creates new threads until full. This behaviour
doesn't play well with latency sensitive workloads where the price of
creating a new thread is too high. For example, when paired with qemu's
'-mlock', or using safety features like SafeStack, creating a new thread
has been measured take multiple milliseconds.

In order to mitigate this let's introduce a new 'EventLoopBase'
property to set the thread pool size. The threads will be created during
the pool's initialization or upon updating the property's value, remain
available during its lifetime regardless of demand, and destroyed upon
freeing it. A properly characterized workload will then be able to
configure the pool to avoid any latency spikes.

Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Acked-by: Markus Armbruster <armbru@redhat.com>
Message-id: 20220425075723.20019-4-nsaenzju@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 qapi/qom.json                    | 10 +++++-
 include/block/aio.h              | 10 ++++++
 include/block/thread-pool.h      |  3 ++
 include/sysemu/event-loop-base.h |  4 +++
 event-loop-base.c                | 23 +++++++++++++
 iothread.c                       |  3 ++
 util/aio-posix.c                 |  1 +
 util/async.c                     | 20 ++++++++++++
 util/main-loop.c                 |  9 ++++++
 util/thread-pool.c               | 55 +++++++++++++++++++++++++++++---
 10 files changed, 133 insertions(+), 5 deletions(-)

diff --git a/qapi/qom.json b/qapi/qom.json
index XXXXXXX..XXXXXXX 100644
--- a/qapi/qom.json
+++ b/qapi/qom.json
@@ -XXX,XX +XXX,XX @@
 #                 0 means that the engine will use its default.
 #                 (default: 0)
 #
+# @thread-pool-min: minimum number of threads reserved in the thread pool
+#                   (default:0)
+#
+# @thread-pool-max: maximum number of threads the thread pool can contain
+#                   (default:64)
+#
 # Since: 7.1
 ##
 { 'struct': 'EventLoopBaseProperties',
-  'data': { '*aio-max-batch': 'int' } }
+  'data': { '*aio-max-batch': 'int',
+            '*thread-pool-min': 'int',
+            '*thread-pool-max': 'int' } }
 
 ##
 # @IothreadProperties:
diff --git a/include/block/aio.h b/include/block/aio.h
index XXXXXXX..XXXXXXX 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -XXX,XX +XXX,XX @@ struct AioContext {
     QSLIST_HEAD(, Coroutine) scheduled_coroutines;
     QEMUBH *co_schedule_bh;
 
+    int thread_pool_min;
+    int thread_pool_max;
     /* Thread pool for performing work and receiving completion callbacks.
      * Has its own locking.
      */
@@ -XXX,XX +XXX,XX @@ void aio_context_set_poll_params(AioContext *ctx, int64_t max_ns,
 void aio_context_set_aio_params(AioContext *ctx, int64_t max_batch,
                                 Error **errp);
 
+/**
+ * aio_context_set_thread_pool_params:
+ * @ctx: the aio context
+ * @min: min number of threads to have readily available in the thread pool
+ * @min: max number of threads the thread pool can contain
+ */
+void aio_context_set_thread_pool_params(AioContext *ctx, int64_t min,
+                                        int64_t max, Error **errp);
 #endif
diff --git a/include/block/thread-pool.h b/include/block/thread-pool.h
index XXXXXXX..XXXXXXX 100644
--- a/include/block/thread-pool.h
+++ b/include/block/thread-pool.h
@@ -XXX,XX +XXX,XX @@
 
 #include "block/block.h"
 
+#define THREAD_POOL_MAX_THREADS_DEFAULT         64
+
 typedef int ThreadPoolFunc(void *opaque);
 
 typedef struct ThreadPool ThreadPool;
@@ -XXX,XX +XXX,XX @@ BlockAIOCB *thread_pool_submit_aio(ThreadPool *pool,
 int coroutine_fn thread_pool_submit_co(ThreadPool *pool,
         ThreadPoolFunc *func, void *arg);
 void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg);
+void thread_pool_update_params(ThreadPool *pool, struct AioContext *ctx);
 
 #endif
diff --git a/include/sysemu/event-loop-base.h b/include/sysemu/event-loop-base.h
index XXXXXXX..XXXXXXX 100644
--- a/include/sysemu/event-loop-base.h
+++ b/include/sysemu/event-loop-base.h
@@ -XXX,XX +XXX,XX @@ struct EventLoopBase {
 
     /* AioContext AIO engine parameters */
     int64_t aio_max_batch;
+
+    /* AioContext thread pool parameters */
+    int64_t thread_pool_min;
+    int64_t thread_pool_max;
 };
 #endif
diff --git a/event-loop-base.c b/event-loop-base.c
index XXXXXXX..XXXXXXX 100644
--- a/event-loop-base.c
+++ b/event-loop-base.c
@@ -XXX,XX +XXX,XX @@
 #include "qemu/osdep.h"
 #include "qom/object_interfaces.h"
 #include "qapi/error.h"
+#include "block/thread-pool.h"
 #include "sysemu/event-loop-base.h"
 
 typedef struct {
@@ -XXX,XX +XXX,XX @@ typedef struct {
     ptrdiff_t offset; /* field's byte offset in EventLoopBase struct */
 } EventLoopBaseParamInfo;
 
+static void event_loop_base_instance_init(Object *obj)
+{
+    EventLoopBase *base = EVENT_LOOP_BASE(obj);
+
+    base->thread_pool_max = THREAD_POOL_MAX_THREADS_DEFAULT;
+}
+
 static EventLoopBaseParamInfo aio_max_batch_info = {
     "aio-max-batch", offsetof(EventLoopBase, aio_max_batch),
 };
+static EventLoopBaseParamInfo thread_pool_min_info = {
+    "thread-pool-min", offsetof(EventLoopBase, thread_pool_min),
+};
+static EventLoopBaseParamInfo thread_pool_max_info = {
+    "thread-pool-max", offsetof(EventLoopBase, thread_pool_max),
+};
 
 static void event_loop_base_get_param(Object *obj, Visitor *v,
         const char *name, void *opaque, Error **errp)
@@ -XXX,XX +XXX,XX @@ static void event_loop_base_class_init(ObjectClass *klass, void *class_data)
                               event_loop_base_get_param,
                               event_loop_base_set_param,
                               NULL, &aio_max_batch_info);
+    object_class_property_add(klass, "thread-pool-min", "int",
+                              event_loop_base_get_param,
+                              event_loop_base_set_param,
+                              NULL, &thread_pool_min_info);
+    object_class_property_add(klass, "thread-pool-max", "int",
+                              event_loop_base_get_param,
+                              event_loop_base_set_param,
+                              NULL, &thread_pool_max_info);
 }
 
 static const TypeInfo event_loop_base_info = {
     .name = TYPE_EVENT_LOOP_BASE,
     .parent = TYPE_OBJECT,
     .instance_size = sizeof(EventLoopBase),
+    .instance_init = event_loop_base_instance_init,
     .class_size = sizeof(EventLoopBaseClass),
     .class_init = event_loop_base_class_init,
     .abstract = true,
diff --git a/iothread.c b/iothread.c
index XXXXXXX..XXXXXXX 100644
--- a/iothread.c
+++ b/iothread.c
@@ -XXX,XX +XXX,XX @@ static void iothread_set_aio_context_params(EventLoopBase *base, Error **errp)
     aio_context_set_aio_params(iothread->ctx,
                                iothread->parent_obj.aio_max_batch,
                                errp);
+
+    aio_context_set_thread_pool_params(iothread->ctx, base->thread_pool_min,
+                                       base->thread_pool_max, errp);
 }
 
 
diff --git a/util/aio-posix.c b/util/aio-posix.c
index XXXXXXX..XXXXXXX 100644
--- a/util/aio-posix.c
+++ b/util/aio-posix.c
@@ -XXX,XX +XXX,XX @@
 
 #include "qemu/osdep.h"
 #include "block/block.h"
+#include "block/thread-pool.h"
 #include "qemu/main-loop.h"
 #include "qemu/rcu.h"
 #include "qemu/rcu_queue.h"
diff --git a/util/async.c b/util/async.c
index XXXXXXX..XXXXXXX 100644
--- a/util/async.c
+++ b/util/async.c
@@ -XXX,XX +XXX,XX @@ AioContext *aio_context_new(Error **errp)
 
     ctx->aio_max_batch = 0;
 
+    ctx->thread_pool_min = 0;
+    ctx->thread_pool_max = THREAD_POOL_MAX_THREADS_DEFAULT;
+
     return ctx;
 fail:
     g_source_destroy(&ctx->source);
@@ -XXX,XX +XXX,XX @@ void qemu_set_current_aio_context(AioContext *ctx)
     assert(!get_my_aiocontext());
     set_my_aiocontext(ctx);
 }
+
+void aio_context_set_thread_pool_params(AioContext *ctx, int64_t min,
+                                        int64_t max, Error **errp)
+{
+
+    if (min > max || !max || min > INT_MAX || max > INT_MAX) {
+        error_setg(errp, "bad thread-pool-min/thread-pool-max values");
+        return;
+    }
+
+    ctx->thread_pool_min = min;
+    ctx->thread_pool_max = max;
+
+    if (ctx->thread_pool) {
+        thread_pool_update_params(ctx->thread_pool, ctx);
+    }
+}
diff --git a/util/main-loop.c b/util/main-loop.c
index XXXXXXX..XXXXXXX 100644
--- a/util/main-loop.c
+++ b/util/main-loop.c
@@ -XXX,XX +XXX,XX @@
 #include "sysemu/replay.h"
 #include "qemu/main-loop.h"
 #include "block/aio.h"
+#include "block/thread-pool.h"
 #include "qemu/error-report.h"
 #include "qemu/queue.h"
 #include "qemu/compiler.h"
@@ -XXX,XX +XXX,XX @@ int qemu_init_main_loop(Error **errp)
 
 static void main_loop_update_params(EventLoopBase *base, Error **errp)
 {
+    ERRP_GUARD();
+
     if (!qemu_aio_context) {
         error_setg(errp, "qemu aio context not ready");
         return;
     }
 
     aio_context_set_aio_params(qemu_aio_context, base->aio_max_batch, errp);
+    if (*errp) {
+        return;
+    }
+
+    aio_context_set_thread_pool_params(qemu_aio_context, base->thread_pool_min,
+                                       base->thread_pool_max, errp);
 }
 
 MainLoop *mloop;
diff --git a/util/thread-pool.c b/util/thread-pool.c
index XXXXXXX..XXXXXXX 100644
--- a/util/thread-pool.c
+++ b/util/thread-pool.c
@@ -XXX,XX +XXX,XX @@ struct ThreadPool {
     QemuMutex lock;
     QemuCond worker_stopped;
     QemuSemaphore sem;
-    int max_threads;
     QEMUBH *new_thread_bh;
 
     /* The following variables are only accessed from one AioContext. */
@@ -XXX,XX +XXX,XX @@ struct ThreadPool {
     int new_threads;     /* backlog of threads we need to create */
     int pending_threads; /* threads created but not running yet */
     bool stopping;
+    int min_threads;
+    int max_threads;
 };
 
+static inline bool back_to_sleep(ThreadPool *pool, int ret)
+{
+    /*
+     * The semaphore timed out, we should exit the loop except when:
+     *  - There is work to do, we raced with the signal.
+     *  - The max threads threshold just changed, we raced with the signal.
+     *  - The thread pool forces a minimum number of readily available threads.
+     */
+    if (ret == -1 && (!QTAILQ_EMPTY(&pool->request_list) ||
+            pool->cur_threads > pool->max_threads ||
+            pool->cur_threads <= pool->min_threads)) {
+            return true;
+    }
+
+    return false;
+}
+
 static void *worker_thread(void *opaque)
 {
     ThreadPool *pool = opaque;
@@ -XXX,XX +XXX,XX @@ static void *worker_thread(void *opaque)
             ret = qemu_sem_timedwait(&pool->sem, 10000);
             qemu_mutex_lock(&pool->lock);
             pool->idle_threads--;
-        } while (ret == -1 && !QTAILQ_EMPTY(&pool->request_list));
-        if (ret == -1 || pool->stopping) {
+        } while (back_to_sleep(pool, ret));
+        if (ret == -1 || pool->stopping ||
+            pool->cur_threads > pool->max_threads) {
             break;
         }
 
@@ -XXX,XX +XXX,XX @@ void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg)
     thread_pool_submit_aio(pool, func, arg, NULL, NULL);
 }
 
+void thread_pool_update_params(ThreadPool *pool, AioContext *ctx)
+{
+    qemu_mutex_lock(&pool->lock);
+
+    pool->min_threads = ctx->thread_pool_min;
+    pool->max_threads = ctx->thread_pool_max;
+
+    /*
+     * We either have to:
+     *  - Increase the number available of threads until over the min_threads
+     *    threshold.
+     *  - Decrease the number of available threads until under the max_threads
+     *    threshold.
+     *  - Do nothing. The current number of threads fall in between the min and
+     *    max thresholds. We'll let the pool manage itself.
+     */
+    for (int i = pool->cur_threads; i < pool->min_threads; i++) {
+        spawn_thread(pool);
+    }
+
+    for (int i = pool->cur_threads; i > pool->max_threads; i--) {
+        qemu_sem_post(&pool->sem);
+    }
+
+    qemu_mutex_unlock(&pool->lock);
+}
+
 static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
 {
     if (!ctx) {
@@ -XXX,XX +XXX,XX @@ static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
     qemu_mutex_init(&pool->lock);
     qemu_cond_init(&pool->worker_stopped);
     qemu_sem_init(&pool->sem, 0);
-    pool->max_threads = 64;
     pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);
 
     QLIST_INIT(&pool->head);
     QTAILQ_INIT(&pool->request_list);
+
+    thread_pool_update_params(pool, ctx);
 }
 
 ThreadPool *thread_pool_new(AioContext *ctx)
-- 
2.35.1

Commit f34e8d8b8d48d73f36a67b6d5e492ef9784b5012 ("virtio-scsi: prepare
virtio_scsi_handle_cmd for dataplane") prepared the virtio-scsi cmd
virtqueue handler function to be used in both the dataplane and
non-datpalane code paths.

It failed to convert the ctrl and event virtqueue handler functions,
which are not designed to be called from the dataplane code path but
will be since the ioeventfd is set up for those virtqueues when
dataplane starts.

Convert the ctrl and event virtqueue handler functions now so they
operate correctly when called from the dataplane code path. Avoid code
duplication by extracting this code into a helper function.

Fixes: f34e8d8b8d48d73f36a67b6d5e492ef9784b5012 ("virtio-scsi: prepare virtio_scsi_handle_cmd for dataplane")
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
Message-id: 20220427143541.119567-2-stefanha@redhat.com
[Fixed s/by used/be used/ typo pointed out by Michael Tokarev
<mjt@tls.msk.ru>.
--Stefan]
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 hw/scsi/virtio-scsi.c | 42 +++++++++++++++++++++++++++---------------
 1 file changed, 27 insertions(+), 15 deletions(-)

diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/scsi/virtio-scsi.c
+++ b/hw/scsi/virtio-scsi.c
@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq)
     return progress;
 }
 
+/*
+ * If dataplane is configured but not yet started, do so now and return true on
+ * success.
+ *
+ * Dataplane is started by the core virtio code but virtqueue handler functions
+ * can also be invoked when a guest kicks before DRIVER_OK, so this helper
+ * function helps us deal with manually starting ioeventfd in that case.
+ */
+static bool virtio_scsi_defer_to_dataplane(VirtIOSCSI *s)
+{
+    if (!s->ctx || s->dataplane_started) {
+        return false;
+    }
+
+    virtio_device_start_ioeventfd(&s->parent_obj.parent_obj);
+    return !s->dataplane_fenced;
+}
+
 static void virtio_scsi_handle_ctrl(VirtIODevice *vdev, VirtQueue *vq)
 {
     VirtIOSCSI *s = (VirtIOSCSI *)vdev;
 
-    if (s->ctx) {
-        virtio_device_start_ioeventfd(vdev);
-        if (!s->dataplane_fenced) {
-            return;
-        }
+    if (virtio_scsi_defer_to_dataplane(s)) {
+        return;
     }
+
     virtio_scsi_acquire(s);
     virtio_scsi_handle_ctrl_vq(s, vq);
     virtio_scsi_release(s);
@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_cmd(VirtIODevice *vdev, VirtQueue *vq)
     /* use non-QOM casts in the data path */
     VirtIOSCSI *s = (VirtIOSCSI *)vdev;
 
-    if (s->ctx && !s->dataplane_started) {
-        virtio_device_start_ioeventfd(vdev);
-        if (!s->dataplane_fenced) {
-            return;
-        }
+    if (virtio_scsi_defer_to_dataplane(s)) {
+        return;
     }
+
     virtio_scsi_acquire(s);
     virtio_scsi_handle_cmd_vq(s, vq);
     virtio_scsi_release(s);
@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_event(VirtIODevice *vdev, VirtQueue *vq)
 {
     VirtIOSCSI *s = VIRTIO_SCSI(vdev);
 
-    if (s->ctx) {
-        virtio_device_start_ioeventfd(vdev);
-        if (!s->dataplane_fenced) {
-            return;
-        }
+    if (virtio_scsi_defer_to_dataplane(s)) {
+        return;
     }
+
     virtio_scsi_acquire(s);
     virtio_scsi_handle_event_vq(s, vq);
     virtio_scsi_release(s);
-- 
2.35.1

The virtio-scsi event virtqueue is not emptied by its handler function.
This is typical for rx virtqueues where the device uses buffers when
some event occurs (e.g. a packet is received, an error condition
happens, etc).

Polling non-empty virtqueues wastes CPU cycles. We are not waiting for
new buffers to become available, we are waiting for an event to occur,
so it's a misuse of CPU resources to poll for buffers.

Introduce the new virtio_queue_aio_attach_host_notifier_no_poll() API,
which is identical to virtio_queue_aio_attach_host_notifier() except
that it does not poll the virtqueue.

Before this patch the following command-line consumed 100% CPU in the
IOThread polling and calling virtio_scsi_handle_event():

$ qemu-system-x86_64 -M accel=kvm -m 1G -cpu host \
      --object iothread,id=iothread0 \
      --device virtio-scsi-pci,iothread=iothread0 \
      --blockdev file,filename=test.img,aio=native,cache.direct=on,node-name=drive0 \
      --device scsi-hd,drive=drive0

After this patch CPU is no longer wasted.

Reported-by: Nir Soffer <nsoffer@redhat.com>
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Tested-by: Nir Soffer <nsoffer@redhat.com>
Message-id: 20220427143541.119567-3-stefanha@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/hw/virtio/virtio.h      |  1 +
 hw/scsi/virtio-scsi-dataplane.c |  2 +-
 hw/virtio/virtio.c              | 13 +++++++++++++
 3 files changed, 15 insertions(+), 1 deletion(-)

diff --git a/include/hw/virtio/virtio.h b/include/hw/virtio/virtio.h
index XXXXXXX..XXXXXXX 100644
--- a/include/hw/virtio/virtio.h
+++ b/include/hw/virtio/virtio.h
@@ -XXX,XX +XXX,XX @@ EventNotifier *virtio_queue_get_host_notifier(VirtQueue *vq);
 void virtio_queue_set_host_notifier_enabled(VirtQueue *vq, bool enabled);
 void virtio_queue_host_notifier_read(EventNotifier *n);
 void virtio_queue_aio_attach_host_notifier(VirtQueue *vq, AioContext *ctx);
+void virtio_queue_aio_attach_host_notifier_no_poll(VirtQueue *vq, AioContext *ctx);
 void virtio_queue_aio_detach_host_notifier(VirtQueue *vq, AioContext *ctx);
 VirtQueue *virtio_vector_first_queue(VirtIODevice *vdev, uint16_t vector);
 VirtQueue *virtio_vector_next_queue(VirtQueue *vq);
diff --git a/hw/scsi/virtio-scsi-dataplane.c b/hw/scsi/virtio-scsi-dataplane.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/scsi/virtio-scsi-dataplane.c
+++ b/hw/scsi/virtio-scsi-dataplane.c
@@ -XXX,XX +XXX,XX @@ int virtio_scsi_dataplane_start(VirtIODevice *vdev)
 
     aio_context_acquire(s->ctx);
     virtio_queue_aio_attach_host_notifier(vs->ctrl_vq, s->ctx);
-    virtio_queue_aio_attach_host_notifier(vs->event_vq, s->ctx);
+    virtio_queue_aio_attach_host_notifier_no_poll(vs->event_vq, s->ctx);
 
     for (i = 0; i < vs->conf.num_queues; i++) {
         virtio_queue_aio_attach_host_notifier(vs->cmd_vqs[i], s->ctx);
diff --git a/hw/virtio/virtio.c b/hw/virtio/virtio.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/virtio/virtio.c
+++ b/hw/virtio/virtio.c
@@ -XXX,XX +XXX,XX @@ void virtio_queue_aio_attach_host_notifier(VirtQueue *vq, AioContext *ctx)
                                 virtio_queue_host_notifier_aio_poll_end);
 }
 
+/*
+ * Same as virtio_queue_aio_attach_host_notifier() but without polling. Use
+ * this for rx virtqueues and similar cases where the virtqueue handler
+ * function does not pop all elements. When the virtqueue is left non-empty
+ * polling consumes CPU cycles and should not be used.
+ */
+void virtio_queue_aio_attach_host_notifier_no_poll(VirtQueue *vq, AioContext *ctx)
+{
+    aio_set_event_notifier(ctx, &vq->host_notifier, true,
+                           virtio_queue_host_notifier_read,
+                           NULL, NULL);
+}
+
 void virtio_queue_aio_detach_host_notifier(VirtQueue *vq, AioContext *ctx)
 {
     aio_set_event_notifier(ctx, &vq->host_notifier, true, NULL, NULL, NULL);
-- 
2.35.1

virtio_scsi_handle_event_vq() is only called from hw/scsi/virtio-scsi.c
now and its return value is no longer used. Remove the function
prototype from virtio-scsi.h and drop the return value.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
Message-id: 20220427143541.119567-4-stefanha@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/hw/virtio/virtio-scsi.h | 1 -
 hw/scsi/virtio-scsi.c           | 4 +---
 2 files changed, 1 insertion(+), 4 deletions(-)

diff --git a/include/hw/virtio/virtio-scsi.h b/include/hw/virtio/virtio-scsi.h
index XXXXXXX..XXXXXXX 100644
--- a/include/hw/virtio/virtio-scsi.h
+++ b/include/hw/virtio/virtio-scsi.h
@@ -XXX,XX +XXX,XX @@ void virtio_scsi_common_realize(DeviceState *dev,
                                 Error **errp);
 
 void virtio_scsi_common_unrealize(DeviceState *dev);
-bool virtio_scsi_handle_event_vq(VirtIOSCSI *s, VirtQueue *vq);
 bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq);
 bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq);
 void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req);
diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/scsi/virtio-scsi.c
+++ b/hw/scsi/virtio-scsi.c
@@ -XXX,XX +XXX,XX @@ void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
     virtio_scsi_complete_req(req);
 }
 
-bool virtio_scsi_handle_event_vq(VirtIOSCSI *s, VirtQueue *vq)
+static void virtio_scsi_handle_event_vq(VirtIOSCSI *s, VirtQueue *vq)
 {
     if (s->events_dropped) {
         virtio_scsi_push_event(s, NULL, VIRTIO_SCSI_T_NO_EVENT, 0);
-        return true;
     }
-    return false;
 }
 
 static void virtio_scsi_handle_event(VirtIODevice *vdev, VirtQueue *vq)
-- 
2.35.1

virtio_scsi_handle_ctrl_vq() is only called from hw/scsi/virtio-scsi.c
now and its return value is no longer used. Remove the function
prototype from virtio-scsi.h and drop the return value.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
Message-id: 20220427143541.119567-5-stefanha@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/hw/virtio/virtio-scsi.h | 1 -
 hw/scsi/virtio-scsi.c           | 5 +----
 2 files changed, 1 insertion(+), 5 deletions(-)

diff --git a/include/hw/virtio/virtio-scsi.h b/include/hw/virtio/virtio-scsi.h
index XXXXXXX..XXXXXXX 100644
--- a/include/hw/virtio/virtio-scsi.h
+++ b/include/hw/virtio/virtio-scsi.h
@@ -XXX,XX +XXX,XX @@ void virtio_scsi_common_realize(DeviceState *dev,
 
 void virtio_scsi_common_unrealize(DeviceState *dev);
 bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq);
-bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq);
 void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req);
 void virtio_scsi_free_req(VirtIOSCSIReq *req);
 void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/scsi/virtio-scsi.c
+++ b/hw/scsi/virtio-scsi.c
@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_ctrl_req(VirtIOSCSI *s, VirtIOSCSIReq *req)
     }
 }
 
-bool virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq)
+static void virtio_scsi_handle_ctrl_vq(VirtIOSCSI *s, VirtQueue *vq)
 {
     VirtIOSCSIReq *req;
-    bool progress = false;
 
     while ((req = virtio_scsi_pop_req(s, vq))) {
-        progress = true;
         virtio_scsi_handle_ctrl_req(s, req);
     }
-    return progress;
 }
 
 /*
-- 
2.35.1

virtio_scsi_handle_cmd_vq() is only called from hw/scsi/virtio-scsi.c
now and its return value is no longer used. Remove the function
prototype from virtio-scsi.h and drop the return value.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
Message-id: 20220427143541.119567-6-stefanha@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/hw/virtio/virtio-scsi.h | 1 -
 hw/scsi/virtio-scsi.c           | 5 +----
 2 files changed, 1 insertion(+), 5 deletions(-)

diff --git a/include/hw/virtio/virtio-scsi.h b/include/hw/virtio/virtio-scsi.h
index XXXXXXX..XXXXXXX 100644
--- a/include/hw/virtio/virtio-scsi.h
+++ b/include/hw/virtio/virtio-scsi.h
@@ -XXX,XX +XXX,XX @@ void virtio_scsi_common_realize(DeviceState *dev,
                                 Error **errp);
 
 void virtio_scsi_common_unrealize(DeviceState *dev);
-bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq);
 void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req);
 void virtio_scsi_free_req(VirtIOSCSIReq *req);
 void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/scsi/virtio-scsi.c
+++ b/hw/scsi/virtio-scsi.c
@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_cmd_req_submit(VirtIOSCSI *s, VirtIOSCSIReq *req)
     scsi_req_unref(sreq);
 }
 
-bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
+static void virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
 {
     VirtIOSCSIReq *req, *next;
     int ret = 0;
     bool suppress_notifications = virtio_queue_get_notification(vq);
-    bool progress = false;
 
     QTAILQ_HEAD(, VirtIOSCSIReq) reqs = QTAILQ_HEAD_INITIALIZER(reqs);
 
@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
         }
 
         while ((req = virtio_scsi_pop_req(s, vq))) {
-            progress = true;
             ret = virtio_scsi_handle_cmd_req_prepare(s, req);
             if (!ret) {
                 QTAILQ_INSERT_TAIL(&reqs, req, next);
@@ -XXX,XX +XXX,XX @@ bool virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
     QTAILQ_FOREACH_SAFE(req, &reqs, next, next) {
         virtio_scsi_handle_cmd_req_submit(s, req);
     }
-    return progress;
 }
 
 static void virtio_scsi_handle_cmd(VirtIODevice *vdev, VirtQueue *vq)
-- 
2.35.1

There is no longer a need to expose the request and related APIs in
virtio-scsi.h since there are no callers outside virtio-scsi.c.

Note the block comment in VirtIOSCSIReq has been adjusted to meet the
coding style.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
Message-id: 20220427143541.119567-7-stefanha@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/hw/virtio/virtio-scsi.h | 40 -----------------------------
 hw/scsi/virtio-scsi.c           | 45 ++++++++++++++++++++++++++++++---
 2 files changed, 41 insertions(+), 44 deletions(-)

diff --git a/include/hw/virtio/virtio-scsi.h b/include/hw/virtio/virtio-scsi.h
index XXXXXXX..XXXXXXX 100644
--- a/include/hw/virtio/virtio-scsi.h
+++ b/include/hw/virtio/virtio-scsi.h
@@ -XXX,XX +XXX,XX @@ struct VirtIOSCSI {
     uint32_t host_features;
 };
 
-typedef struct VirtIOSCSIReq {
-    /* Note:
-     * - fields up to resp_iov are initialized by virtio_scsi_init_req;
-     * - fields starting at vring are zeroed by virtio_scsi_init_req.
-     * */
-    VirtQueueElement elem;
-
-    VirtIOSCSI *dev;
-    VirtQueue *vq;
-    QEMUSGList qsgl;
-    QEMUIOVector resp_iov;
-
-    union {
-        /* Used for two-stage request submission */
-        QTAILQ_ENTRY(VirtIOSCSIReq) next;
-
-        /* Used for cancellation of request during TMFs */
-        int remaining;
-    };
-
-    SCSIRequest *sreq;
-    size_t resp_size;
-    enum SCSIXferMode mode;
-    union {
-        VirtIOSCSICmdResp     cmd;
-        VirtIOSCSICtrlTMFResp tmf;
-        VirtIOSCSICtrlANResp  an;
-        VirtIOSCSIEvent       event;
-    } resp;
-    union {
-        VirtIOSCSICmdReq      cmd;
-        VirtIOSCSICtrlTMFReq  tmf;
-        VirtIOSCSICtrlANReq   an;
-    } req;
-} VirtIOSCSIReq;
-
 static inline void virtio_scsi_acquire(VirtIOSCSI *s)
 {
     if (s->ctx) {
@@ -XXX,XX +XXX,XX @@ void virtio_scsi_common_realize(DeviceState *dev,
                                 Error **errp);
 
 void virtio_scsi_common_unrealize(DeviceState *dev);
-void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req);
-void virtio_scsi_free_req(VirtIOSCSIReq *req);
-void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
-                            uint32_t event, uint32_t reason);
 
 void virtio_scsi_dataplane_setup(VirtIOSCSI *s, Error **errp);
 int virtio_scsi_dataplane_start(VirtIODevice *s);
diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/scsi/virtio-scsi.c
+++ b/hw/scsi/virtio-scsi.c
@@ -XXX,XX +XXX,XX @@
 #include "hw/virtio/virtio-access.h"
 #include "trace.h"
 
+typedef struct VirtIOSCSIReq {
+    /*
+     * Note:
+     * - fields up to resp_iov are initialized by virtio_scsi_init_req;
+     * - fields starting at vring are zeroed by virtio_scsi_init_req.
+     */
+    VirtQueueElement elem;
+
+    VirtIOSCSI *dev;
+    VirtQueue *vq;
+    QEMUSGList qsgl;
+    QEMUIOVector resp_iov;
+
+    union {
+        /* Used for two-stage request submission */
+        QTAILQ_ENTRY(VirtIOSCSIReq) next;
+
+        /* Used for cancellation of request during TMFs */
+        int remaining;
+    };
+
+    SCSIRequest *sreq;
+    size_t resp_size;
+    enum SCSIXferMode mode;
+    union {
+        VirtIOSCSICmdResp     cmd;
+        VirtIOSCSICtrlTMFResp tmf;
+        VirtIOSCSICtrlANResp  an;
+        VirtIOSCSIEvent       event;
+    } resp;
+    union {
+        VirtIOSCSICmdReq      cmd;
+        VirtIOSCSICtrlTMFReq  tmf;
+        VirtIOSCSICtrlANReq   an;
+    } req;
+} VirtIOSCSIReq;
+
 static inline int virtio_scsi_get_lun(uint8_t *lun)
 {
     return ((lun[2] << 8) | lun[3]) & 0x3FFF;
@@ -XXX,XX +XXX,XX @@ static inline SCSIDevice *virtio_scsi_device_get(VirtIOSCSI *s, uint8_t *lun)
     return scsi_device_get(&s->bus, 0, lun[1], virtio_scsi_get_lun(lun));
 }
 
-void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req)
+static void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req)
 {
     VirtIODevice *vdev = VIRTIO_DEVICE(s);
     const size_t zero_skip =
@@ -XXX,XX +XXX,XX @@ void virtio_scsi_init_req(VirtIOSCSI *s, VirtQueue *vq, VirtIOSCSIReq *req)
     memset((uint8_t *)req + zero_skip, 0, sizeof(*req) - zero_skip);
 }
 
-void virtio_scsi_free_req(VirtIOSCSIReq *req)
+static void virtio_scsi_free_req(VirtIOSCSIReq *req)
 {
     qemu_iovec_destroy(&req->resp_iov);
     qemu_sglist_destroy(&req->qsgl);
@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_reset(VirtIODevice *vdev)
     s->events_dropped = false;
 }
 
-void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
-                            uint32_t event, uint32_t reason)
+static void virtio_scsi_push_event(VirtIOSCSI *s, SCSIDevice *dev,
+                                   uint32_t event, uint32_t reason)
 {
     VirtIOSCSICommon *vs = VIRTIO_SCSI_COMMON(s);
     VirtIOSCSIReq *req;
-- 
2.35.1

The following changes since commit c6a5fc2ac76c5ab709896ee1b0edd33685a67ed1:

decodetree: Add --output-null for meson testing (2023-05-31 19:56:42 -0700)

are available in the Git repository at:

https://gitlab.com/stefanha/qemu.git tags/block-pull-request

for you to fetch changes up to 98b126f5e3228a346c774e569e26689943b401dd:

qapi: add '@fdset' feature for BlockdevOptionsVirtioBlkVhostVdpa (2023-06-01 11:08:21 -0400)

----------------------------------------------------------------
Pull request

- Stefano Garzarella's blkio block driver 'fd' parameter
- My thread-local blk_io_plug() series

----------------------------------------------------------------

Stefan Hajnoczi (6):
  block: add blk_io_plug_call() API
  block/nvme: convert to blk_io_plug_call() API
  block/blkio: convert to blk_io_plug_call() API
  block/io_uring: convert to blk_io_plug_call() API
  block/linux-aio: convert to blk_io_plug_call() API
  block: remove bdrv_co_io_plug() API

Stefano Garzarella (2):
  block/blkio: use qemu_open() to support fd passing for virtio-blk
  qapi: add '@fdset' feature for BlockdevOptionsVirtioBlkVhostVdpa

-- 
2.40.1

Introduce a new API for thread-local blk_io_plug() that does not
traverse the block graph. The goal is to make blk_io_plug() multi-queue
friendly.

Instead of having block drivers track whether or not we're in a plugged
section, provide an API that allows them to defer a function call until
we're unplugged: blk_io_plug_call(fn, opaque). If blk_io_plug_call() is
called multiple times with the same fn/opaque pair, then fn() is only
called once at the end of the function - resulting in batching.

This patch introduces the API and changes blk_io_plug()/blk_io_unplug().
blk_io_plug()/blk_io_unplug() no longer require a BlockBackend argument
because the plug state is now thread-local.

Later patches convert block drivers to blk_io_plug_call() and then we
can finally remove .bdrv_co_io_plug() once all block drivers have been
converted.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Eric Blake <eblake@redhat.com>
Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
Acked-by: Kevin Wolf <kwolf@redhat.com>
Message-id: 20230530180959.1108766-2-stefanha@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 MAINTAINERS                       |   1 +
 include/sysemu/block-backend-io.h |  13 +--
 block/block-backend.c             |  22 -----
 block/plug.c                      | 159 ++++++++++++++++++++++++++++++
 hw/block/dataplane/xen-block.c    |   8 +-
 hw/block/virtio-blk.c             |   4 +-
 hw/scsi/virtio-scsi.c             |   6 +-
 block/meson.build                 |   1 +
 8 files changed, 173 insertions(+), 41 deletions(-)
 create mode 100644 block/plug.c

diff --git a/MAINTAINERS b/MAINTAINERS
index XXXXXXX..XXXXXXX 100644
--- a/MAINTAINERS
+++ b/MAINTAINERS
@@ -XXX,XX +XXX,XX @@ F: util/aio-*.c
 F: util/aio-*.h
 F: util/fdmon-*.c
 F: block/io.c
+F: block/plug.c
 F: migration/block*
 F: include/block/aio.h
 F: include/block/aio-wait.h
diff --git a/include/sysemu/block-backend-io.h b/include/sysemu/block-backend-io.h
index XXXXXXX..XXXXXXX 100644
--- a/include/sysemu/block-backend-io.h
+++ b/include/sysemu/block-backend-io.h
@@ -XXX,XX +XXX,XX @@ void blk_iostatus_set_err(BlockBackend *blk, int error);
 int blk_get_max_iov(BlockBackend *blk);
 int blk_get_max_hw_iov(BlockBackend *blk);
 
-/*
- * blk_io_plug/unplug are thread-local operations. This means that multiple
- * IOThreads can simultaneously call plug/unplug, but the caller must ensure
- * that each unplug() is called in the same IOThread of the matching plug().
- */
-void coroutine_fn blk_co_io_plug(BlockBackend *blk);
-void co_wrapper blk_io_plug(BlockBackend *blk);
-
-void coroutine_fn blk_co_io_unplug(BlockBackend *blk);
-void co_wrapper blk_io_unplug(BlockBackend *blk);
+void blk_io_plug(void);
+void blk_io_unplug(void);
+void blk_io_plug_call(void (*fn)(void *), void *opaque);
 
 AioContext *blk_get_aio_context(BlockBackend *blk);
 BlockAcctStats *blk_get_stats(BlockBackend *blk);
diff --git a/block/block-backend.c b/block/block-backend.c
index XXXXXXX..XXXXXXX 100644
--- a/block/block-backend.c
+++ b/block/block-backend.c
@@ -XXX,XX +XXX,XX @@ void blk_add_insert_bs_notifier(BlockBackend *blk, Notifier *notify)
     notifier_list_add(&blk->insert_bs_notifiers, notify);
 }
 
-void coroutine_fn blk_co_io_plug(BlockBackend *blk)
-{
-    BlockDriverState *bs = blk_bs(blk);
-    IO_CODE();
-    GRAPH_RDLOCK_GUARD();
-
-    if (bs) {
-        bdrv_co_io_plug(bs);
-    }
-}
-
-void coroutine_fn blk_co_io_unplug(BlockBackend *blk)
-{
-    BlockDriverState *bs = blk_bs(blk);
-    IO_CODE();
-    GRAPH_RDLOCK_GUARD();
-
-    if (bs) {
-        bdrv_co_io_unplug(bs);
-    }
-}
-
 BlockAcctStats *blk_get_stats(BlockBackend *blk)
 {
     IO_CODE();
diff --git a/block/plug.c b/block/plug.c
new file mode 100644
index XXXXXXX..XXXXXXX
--- /dev/null
+++ b/block/plug.c
@@ -XXX,XX +XXX,XX @@
+/* SPDX-License-Identifier: GPL-2.0-or-later */
+/*
+ * Block I/O plugging
+ *
+ * Copyright Red Hat.
+ *
+ * This API defers a function call within a blk_io_plug()/blk_io_unplug()
+ * section, allowing multiple calls to batch up. This is a performance
+ * optimization that is used in the block layer to submit several I/O requests
+ * at once instead of individually:
+ *
+ *   blk_io_plug(); <-- start of plugged region
+ *   ...
+ *   blk_io_plug_call(my_func, my_obj); <-- deferred my_func(my_obj) call
+ *   blk_io_plug_call(my_func, my_obj); <-- another
+ *   blk_io_plug_call(my_func, my_obj); <-- another
+ *   ...
+ *   blk_io_unplug(); <-- end of plugged region, my_func(my_obj) is called once
+ *
+ * This code is actually generic and not tied to the block layer. If another
+ * subsystem needs this functionality, it could be renamed.
+ */
+
+#include "qemu/osdep.h"
+#include "qemu/coroutine-tls.h"
+#include "qemu/notify.h"
+#include "qemu/thread.h"
+#include "sysemu/block-backend.h"
+
+/* A function call that has been deferred until unplug() */
+typedef struct {
+    void (*fn)(void *);
+    void *opaque;
+} UnplugFn;
+
+/* Per-thread state */
+typedef struct {
+    unsigned count;       /* how many times has plug() been called? */
+    GArray *unplug_fns;   /* functions to call at unplug time */
+} Plug;
+
+/* Use get_ptr_plug() to fetch this thread-local value */
+QEMU_DEFINE_STATIC_CO_TLS(Plug, plug);
+
+/* Called at thread cleanup time */
+static void blk_io_plug_atexit(Notifier *n, void *value)
+{
+    Plug *plug = get_ptr_plug();
+    g_array_free(plug->unplug_fns, TRUE);
+}
+
+/* This won't involve coroutines, so use __thread */
+static __thread Notifier blk_io_plug_atexit_notifier;
+
+/**
+ * blk_io_plug_call:
+ * @fn: a function pointer to be invoked
+ * @opaque: a user-defined argument to @fn()
+ *
+ * Call @fn(@opaque) immediately if not within a blk_io_plug()/blk_io_unplug()
+ * section.
+ *
+ * Otherwise defer the call until the end of the outermost
+ * blk_io_plug()/blk_io_unplug() section in this thread. If the same
+ * @fn/@opaque pair has already been deferred, it will only be called once upon
+ * blk_io_unplug() so that accumulated calls are batched into a single call.
+ *
+ * The caller must ensure that @opaque is not freed before @fn() is invoked.
+ */
+void blk_io_plug_call(void (*fn)(void *), void *opaque)
+{
+    Plug *plug = get_ptr_plug();
+
+    /* Call immediately if we're not plugged */
+    if (plug->count == 0) {
+        fn(opaque);
+        return;
+    }
+
+    GArray *array = plug->unplug_fns;
+    if (!array) {
+        array = g_array_new(FALSE, FALSE, sizeof(UnplugFn));
+        plug->unplug_fns = array;
+        blk_io_plug_atexit_notifier.notify = blk_io_plug_atexit;
+        qemu_thread_atexit_add(&blk_io_plug_atexit_notifier);
+    }
+
+    UnplugFn *fns = (UnplugFn *)array->data;
+    UnplugFn new_fn = {
+        .fn = fn,
+        .opaque = opaque,
+    };
+
+    /*
+     * There won't be many, so do a linear search. If this becomes a bottleneck
+     * then a binary search (glib 2.62+) or different data structure could be
+     * used.
+     */
+    for (guint i = 0; i < array->len; i++) {
+        if (memcmp(&fns[i], &new_fn, sizeof(new_fn)) == 0) {
+            return; /* already exists */
+        }
+    }
+
+    g_array_append_val(array, new_fn);
+}
+
+/**
+ * blk_io_plug: Defer blk_io_plug_call() functions until blk_io_unplug()
+ *
+ * blk_io_plug/unplug are thread-local operations. This means that multiple
+ * threads can simultaneously call plug/unplug, but the caller must ensure that
+ * each unplug() is called in the same thread of the matching plug().
+ *
+ * Nesting is supported. blk_io_plug_call() functions are only called at the
+ * outermost blk_io_unplug().
+ */
+void blk_io_plug(void)
+{
+    Plug *plug = get_ptr_plug();
+
+    assert(plug->count < UINT32_MAX);
+
+    plug->count++;
+}
+
+/**
+ * blk_io_unplug: Run any pending blk_io_plug_call() functions
+ *
+ * There must have been a matching blk_io_plug() call in the same thread prior
+ * to this blk_io_unplug() call.
+ */
+void blk_io_unplug(void)
+{
+    Plug *plug = get_ptr_plug();
+
+    assert(plug->count > 0);
+
+    if (--plug->count > 0) {
+        return;
+    }
+
+    GArray *array = plug->unplug_fns;
+    if (!array) {
+        return;
+    }
+
+    UnplugFn *fns = (UnplugFn *)array->data;
+
+    for (guint i = 0; i < array->len; i++) {
+        fns[i].fn(fns[i].opaque);
+    }
+
+    /*
+     * This resets the array without freeing memory so that appending is cheap
+     * in the future.
+     */
+    g_array_set_size(array, 0);
+}
diff --git a/hw/block/dataplane/xen-block.c b/hw/block/dataplane/xen-block.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/block/dataplane/xen-block.c
+++ b/hw/block/dataplane/xen-block.c
@@ -XXX,XX +XXX,XX @@ static bool xen_block_handle_requests(XenBlockDataPlane *dataplane)
      * is below us.
      */
     if (inflight_atstart > IO_PLUG_THRESHOLD) {
-        blk_io_plug(dataplane->blk);
+        blk_io_plug();
     }
     while (rc != rp) {
         /* pull request from ring */
@@ -XXX,XX +XXX,XX @@ static bool xen_block_handle_requests(XenBlockDataPlane *dataplane)
 
         if (inflight_atstart > IO_PLUG_THRESHOLD &&
             batched >= inflight_atstart) {
-            blk_io_unplug(dataplane->blk);
+            blk_io_unplug();
         }
         xen_block_do_aio(request);
         if (inflight_atstart > IO_PLUG_THRESHOLD) {
             if (batched >= inflight_atstart) {
-                blk_io_plug(dataplane->blk);
+                blk_io_plug();
                 batched = 0;
             } else {
                 batched++;
@@ -XXX,XX +XXX,XX @@ static bool xen_block_handle_requests(XenBlockDataPlane *dataplane)
         }
     }
     if (inflight_atstart > IO_PLUG_THRESHOLD) {
-        blk_io_unplug(dataplane->blk);
+        blk_io_unplug();
     }
 
     return done_something;
diff --git a/hw/block/virtio-blk.c b/hw/block/virtio-blk.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/block/virtio-blk.c
+++ b/hw/block/virtio-blk.c
@@ -XXX,XX +XXX,XX @@ void virtio_blk_handle_vq(VirtIOBlock *s, VirtQueue *vq)
     bool suppress_notifications = virtio_queue_get_notification(vq);
 
     aio_context_acquire(blk_get_aio_context(s->blk));
-    blk_io_plug(s->blk);
+    blk_io_plug();
 
     do {
         if (suppress_notifications) {
@@ -XXX,XX +XXX,XX @@ void virtio_blk_handle_vq(VirtIOBlock *s, VirtQueue *vq)
         virtio_blk_submit_multireq(s, &mrb);
     }
 
-    blk_io_unplug(s->blk);
+    blk_io_unplug();
     aio_context_release(blk_get_aio_context(s->blk));
 }
 
diff --git a/hw/scsi/virtio-scsi.c b/hw/scsi/virtio-scsi.c
index XXXXXXX..XXXXXXX 100644
--- a/hw/scsi/virtio-scsi.c
+++ b/hw/scsi/virtio-scsi.c
@@ -XXX,XX +XXX,XX @@ static int virtio_scsi_handle_cmd_req_prepare(VirtIOSCSI *s, VirtIOSCSIReq *req)
         return -ENOBUFS;
     }
     scsi_req_ref(req->sreq);
-    blk_io_plug(d->conf.blk);
+    blk_io_plug();
     object_unref(OBJECT(d));
     return 0;
 }
@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_cmd_req_submit(VirtIOSCSI *s, VirtIOSCSIReq *req)
     if (scsi_req_enqueue(sreq)) {
         scsi_req_continue(sreq);
     }
-    blk_io_unplug(sreq->dev->conf.blk);
+    blk_io_unplug();
     scsi_req_unref(sreq);
 }
 
@@ -XXX,XX +XXX,XX @@ static void virtio_scsi_handle_cmd_vq(VirtIOSCSI *s, VirtQueue *vq)
                 while (!QTAILQ_EMPTY(&reqs)) {
                     req = QTAILQ_FIRST(&reqs);
                     QTAILQ_REMOVE(&reqs, req, next);
-                    blk_io_unplug(req->sreq->dev->conf.blk);
+                    blk_io_unplug();
                     scsi_req_unref(req->sreq);
                     virtqueue_detach_element(req->vq, &req->elem, 0);
                     virtio_scsi_free_req(req);
diff --git a/block/meson.build b/block/meson.build
index XXXXXXX..XXXXXXX 100644
--- a/block/meson.build
+++ b/block/meson.build
@@ -XXX,XX +XXX,XX @@ block_ss.add(files(
   'mirror.c',
   'nbd.c',
   'null.c',
+  'plug.c',
   'qapi.c',
   'qcow2-bitmap.c',
   'qcow2-cache.c',
-- 
2.40.1

Stop using the .bdrv_co_io_plug() API because it is not multi-queue
block layer friendly. Use the new blk_io_plug_call() API to batch I/O
submission instead.

diff --git a/block/nvme.c b/block/nvme.c
index XXXXXXX..XXXXXXX 100644
--- a/block/nvme.c
+++ b/block/nvme.c
@@ -XXX,XX +XXX,XX @@
 #include "qemu/vfio-helpers.h"
 #include "block/block-io.h"
 #include "block/block_int.h"
+#include "sysemu/block-backend.h"
 #include "sysemu/replay.h"
 #include "trace.h"
 
@@ -XXX,XX +XXX,XX @@ struct BDRVNVMeState {
     int blkshift;
 
     uint64_t max_transfer;
-    bool plugged;
 
     bool supports_write_zeroes;
     bool supports_discard;
@@ -XXX,XX +XXX,XX @@ static void nvme_kick(NVMeQueuePair *q)
 {
     BDRVNVMeState *s = q->s;
 
-    if (s->plugged || !q->need_kick) {
+    if (!q->need_kick) {
         return;
     }
     trace_nvme_kick(s, q->index);
@@ -XXX,XX +XXX,XX @@ static bool nvme_process_completion(NVMeQueuePair *q)
     NvmeCqe *c;
 
     trace_nvme_process_completion(s, q->index, q->inflight);
-    if (s->plugged) {
-        trace_nvme_process_completion_queue_plugged(s, q->index);
-        return false;
-    }
 
     /*
      * Support re-entrancy when a request cb() function invokes aio_poll().
@@ -XXX,XX +XXX,XX @@ static void nvme_trace_command(const NvmeCmd *cmd)
     }
 }
 
+static void nvme_unplug_fn(void *opaque)
+{
+    NVMeQueuePair *q = opaque;
+
+    QEMU_LOCK_GUARD(&q->lock);
+    nvme_kick(q);
+    nvme_process_completion(q);
+}
+
 static void nvme_submit_command(NVMeQueuePair *q, NVMeRequest *req,
                                 NvmeCmd *cmd, BlockCompletionFunc cb,
                                 void *opaque)
@@ -XXX,XX +XXX,XX @@ static void nvme_submit_command(NVMeQueuePair *q, NVMeRequest *req,
            q->sq.tail * NVME_SQ_ENTRY_BYTES, cmd, sizeof(*cmd));
     q->sq.tail = (q->sq.tail + 1) % NVME_QUEUE_SIZE;
     q->need_kick++;
-    nvme_kick(q);
-    nvme_process_completion(q);
+    blk_io_plug_call(nvme_unplug_fn, q);
     qemu_mutex_unlock(&q->lock);
 }
 
@@ -XXX,XX +XXX,XX @@ static void nvme_attach_aio_context(BlockDriverState *bs,
     }
 }
 
-static void coroutine_fn nvme_co_io_plug(BlockDriverState *bs)
-{
-    BDRVNVMeState *s = bs->opaque;
-    assert(!s->plugged);
-    s->plugged = true;
-}
-
-static void coroutine_fn nvme_co_io_unplug(BlockDriverState *bs)
-{
-    BDRVNVMeState *s = bs->opaque;
-    assert(s->plugged);
-    s->plugged = false;
-    for (unsigned i = INDEX_IO(0); i < s->queue_count; i++) {
-        NVMeQueuePair *q = s->queues[i];
-        qemu_mutex_lock(&q->lock);
-        nvme_kick(q);
-        nvme_process_completion(q);
-        qemu_mutex_unlock(&q->lock);
-    }
-}
-
 static bool nvme_register_buf(BlockDriverState *bs, void *host, size_t size,
                               Error **errp)
 {
@@ -XXX,XX +XXX,XX @@ static BlockDriver bdrv_nvme = {
     .bdrv_detach_aio_context  = nvme_detach_aio_context,
     .bdrv_attach_aio_context  = nvme_attach_aio_context,
 
-    .bdrv_co_io_plug          = nvme_co_io_plug,
-    .bdrv_co_io_unplug        = nvme_co_io_unplug,
-
     .bdrv_register_buf        = nvme_register_buf,
     .bdrv_unregister_buf      = nvme_unregister_buf,
 };
diff --git a/block/trace-events b/block/trace-events
index XXXXXXX..XXXXXXX 100644
--- a/block/trace-events
+++ b/block/trace-events
@@ -XXX,XX +XXX,XX @@ nvme_kick(void *s, unsigned q_index) "s %p q #%u"
 nvme_dma_flush_queue_wait(void *s) "s %p"
 nvme_error(int cmd_specific, int sq_head, int sqid, int cid, int status) "cmd_specific %d sq_head %d sqid %d cid %d status 0x%x"
 nvme_process_completion(void *s, unsigned q_index, int inflight) "s %p q #%u inflight %d"
-nvme_process_completion_queue_plugged(void *s, unsigned q_index) "s %p q #%u"
 nvme_complete_command(void *s, unsigned q_index, int cid) "s %p q #%u cid %d"
 nvme_submit_command(void *s, unsigned q_index, int cid) "s %p q #%u cid %d"
 nvme_submit_command_raw(int c0, int c1, int c2, int c3, int c4, int c5, int c6, int c7) "%02x %02x %02x %02x %02x %02x %02x %02x"
-- 
2.40.1

Stop using the .bdrv_co_io_plug() API because it is not multi-queue
block layer friendly. Use the new blk_io_plug_call() API to batch I/O
submission instead.

diff --git a/block/blkio.c b/block/blkio.c
index XXXXXXX..XXXXXXX 100644
--- a/block/blkio.c
+++ b/block/blkio.c
@@ -XXX,XX +XXX,XX @@
 #include "qemu/error-report.h"
 #include "qapi/qmp/qdict.h"
 #include "qemu/module.h"
+#include "sysemu/block-backend.h"
 #include "exec/memory.h" /* for ram_block_discard_disable() */
 
 #include "block/block-io.h"
@@ -XXX,XX +XXX,XX @@ static void blkio_detach_aio_context(BlockDriverState *bs)
                        NULL, NULL, NULL);
 }
 
-/* Call with s->blkio_lock held to submit I/O after enqueuing a new request */
-static void blkio_submit_io(BlockDriverState *bs)
+/*
+ * Called by blk_io_unplug() or immediately if not plugged. Called without
+ * blkio_lock.
+ */
+static void blkio_unplug_fn(void *opaque)
 {
-    if (qatomic_read(&bs->io_plugged) == 0) {
-        BDRVBlkioState *s = bs->opaque;
+    BDRVBlkioState *s = opaque;
 
+    WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
         blkioq_do_io(s->blkioq, NULL, 0, 0, NULL);
     }
 }
 
+/*
+ * Schedule I/O submission after enqueuing a new request. Called without
+ * blkio_lock.
+ */
+static void blkio_submit_io(BlockDriverState *bs)
+{
+    BDRVBlkioState *s = bs->opaque;
+
+    blk_io_plug_call(blkio_unplug_fn, s);
+}
+
 static int coroutine_fn
 blkio_co_pdiscard(BlockDriverState *bs, int64_t offset, int64_t bytes)
 {
@@ -XXX,XX +XXX,XX @@ blkio_co_pdiscard(BlockDriverState *bs, int64_t offset, int64_t bytes)
 
     WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
         blkioq_discard(s->blkioq, offset, bytes, &cod, 0);
-        blkio_submit_io(bs);
     }
 
+    blkio_submit_io(bs);
     qemu_coroutine_yield();
     return cod.ret;
 }
@@ -XXX,XX +XXX,XX @@ blkio_co_preadv(BlockDriverState *bs, int64_t offset, int64_t bytes,
 
     WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
         blkioq_readv(s->blkioq, offset, iov, iovcnt, &cod, 0);
-        blkio_submit_io(bs);
     }
 
+    blkio_submit_io(bs);
     qemu_coroutine_yield();
 
     if (use_bounce_buffer) {
@@ -XXX,XX +XXX,XX @@ static int coroutine_fn blkio_co_pwritev(BlockDriverState *bs, int64_t offset,
 
     WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
         blkioq_writev(s->blkioq, offset, iov, iovcnt, &cod, blkio_flags);
-        blkio_submit_io(bs);
     }
 
+    blkio_submit_io(bs);
     qemu_coroutine_yield();
 
     if (use_bounce_buffer) {
@@ -XXX,XX +XXX,XX @@ static int coroutine_fn blkio_co_flush(BlockDriverState *bs)
 
     WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
         blkioq_flush(s->blkioq, &cod, 0);
-        blkio_submit_io(bs);
     }
 
+    blkio_submit_io(bs);
     qemu_coroutine_yield();
     return cod.ret;
 }
@@ -XXX,XX +XXX,XX @@ static int coroutine_fn blkio_co_pwrite_zeroes(BlockDriverState *bs,
 
     WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
         blkioq_write_zeroes(s->blkioq, offset, bytes, &cod, blkio_flags);
-        blkio_submit_io(bs);
     }
 
+    blkio_submit_io(bs);
     qemu_coroutine_yield();
     return cod.ret;
 }
 
-static void coroutine_fn blkio_co_io_unplug(BlockDriverState *bs)
-{
-    BDRVBlkioState *s = bs->opaque;
-
-    WITH_QEMU_LOCK_GUARD(&s->blkio_lock) {
-        blkio_submit_io(bs);
-    }
-}
-
 typedef enum {
     BMRR_OK,
     BMRR_SKIP,
@@ -XXX,XX +XXX,XX @@ static void blkio_refresh_limits(BlockDriverState *bs, Error **errp)
         .bdrv_co_pwritev         = blkio_co_pwritev, \
         .bdrv_co_flush_to_disk   = blkio_co_flush, \
         .bdrv_co_pwrite_zeroes   = blkio_co_pwrite_zeroes, \
-        .bdrv_co_io_unplug       = blkio_co_io_unplug, \
         .bdrv_refresh_limits     = blkio_refresh_limits, \
         .bdrv_register_buf       = blkio_register_buf, \
         .bdrv_unregister_buf     = blkio_unregister_buf, \
-- 
2.40.1

Stop using the .bdrv_co_io_plug() API because it is not multi-queue
block layer friendly. Use the new blk_io_plug_call() API to batch I/O
submission instead.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Eric Blake <eblake@redhat.com>
Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
Acked-by: Kevin Wolf <kwolf@redhat.com>
Message-id: 20230530180959.1108766-5-stefanha@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/block/raw-aio.h |  7 -------
 block/file-posix.c      | 10 ----------
 block/io_uring.c        | 44 ++++++++++++++++-------------------------
 block/trace-events      |  5 ++---
 4 files changed, 19 insertions(+), 47 deletions(-)

diff --git a/include/block/raw-aio.h b/include/block/raw-aio.h
index XXXXXXX..XXXXXXX 100644
--- a/include/block/raw-aio.h
+++ b/include/block/raw-aio.h
@@ -XXX,XX +XXX,XX @@ int coroutine_fn luring_co_submit(BlockDriverState *bs, int fd, uint64_t offset,
                                   QEMUIOVector *qiov, int type);
 void luring_detach_aio_context(LuringState *s, AioContext *old_context);
 void luring_attach_aio_context(LuringState *s, AioContext *new_context);
-
-/*
- * luring_io_plug/unplug work in the thread's current AioContext, therefore the
- * caller must ensure that they are paired in the same IOThread.
- */
-void luring_io_plug(void);
-void luring_io_unplug(void);
 #endif
 
 #ifdef _WIN32
diff --git a/block/file-posix.c b/block/file-posix.c
index XXXXXXX..XXXXXXX 100644
--- a/block/file-posix.c
+++ b/block/file-posix.c
@@ -XXX,XX +XXX,XX @@ static void coroutine_fn raw_co_io_plug(BlockDriverState *bs)
         laio_io_plug();
     }
 #endif
-#ifdef CONFIG_LINUX_IO_URING
-    if (s->use_linux_io_uring) {
-        luring_io_plug();
-    }
-#endif
 }
 
 static void coroutine_fn raw_co_io_unplug(BlockDriverState *bs)
@@ -XXX,XX +XXX,XX @@ static void coroutine_fn raw_co_io_unplug(BlockDriverState *bs)
         laio_io_unplug(s->aio_max_batch);
     }
 #endif
-#ifdef CONFIG_LINUX_IO_URING
-    if (s->use_linux_io_uring) {
-        luring_io_unplug();
-    }
-#endif
 }
 
 static int coroutine_fn raw_co_flush_to_disk(BlockDriverState *bs)
diff --git a/block/io_uring.c b/block/io_uring.c
index XXXXXXX..XXXXXXX 100644
--- a/block/io_uring.c
+++ b/block/io_uring.c
@@ -XXX,XX +XXX,XX @@
 #include "block/raw-aio.h"
 #include "qemu/coroutine.h"
 #include "qapi/error.h"
+#include "sysemu/block-backend.h"
 #include "trace.h"
 
 /* Only used for assertions.  */
@@ -XXX,XX +XXX,XX @@ typedef struct LuringAIOCB {
 } LuringAIOCB;
 
 typedef struct LuringQueue {
-    int plugged;
     unsigned int in_queue;
     unsigned int in_flight;
     bool blocked;
@@ -XXX,XX +XXX,XX @@ static void luring_process_completions_and_submit(LuringState *s)
 {
     luring_process_completions(s);
 
-    if (!s->io_q.plugged && s->io_q.in_queue > 0) {
+    if (s->io_q.in_queue > 0) {
         ioq_submit(s);
     }
 }
@@ -XXX,XX +XXX,XX @@ static void qemu_luring_poll_ready(void *opaque)
 static void ioq_init(LuringQueue *io_q)
 {
     QSIMPLEQ_INIT(&io_q->submit_queue);
-    io_q->plugged = 0;
     io_q->in_queue = 0;
     io_q->in_flight = 0;
     io_q->blocked = false;
 }
 
-void luring_io_plug(void)
+static void luring_unplug_fn(void *opaque)
 {
-    AioContext *ctx = qemu_get_current_aio_context();
-    LuringState *s = aio_get_linux_io_uring(ctx);
-    trace_luring_io_plug(s);
-    s->io_q.plugged++;
-}
-
-void luring_io_unplug(void)
-{
-    AioContext *ctx = qemu_get_current_aio_context();
-    LuringState *s = aio_get_linux_io_uring(ctx);
-    assert(s->io_q.plugged);
-    trace_luring_io_unplug(s, s->io_q.blocked, s->io_q.plugged,
-                           s->io_q.in_queue, s->io_q.in_flight);
-    if (--s->io_q.plugged == 0 &&
-        !s->io_q.blocked && s->io_q.in_queue > 0) {
+    LuringState *s = opaque;
+    trace_luring_unplug_fn(s, s->io_q.blocked, s->io_q.in_queue,
+                           s->io_q.in_flight);
+    if (!s->io_q.blocked && s->io_q.in_queue > 0) {
         ioq_submit(s);
     }
 }
@@ -XXX,XX +XXX,XX @@ static int luring_do_submit(int fd, LuringAIOCB *luringcb, LuringState *s,
 
     QSIMPLEQ_INSERT_TAIL(&s->io_q.submit_queue, luringcb, next);
     s->io_q.in_queue++;
-    trace_luring_do_submit(s, s->io_q.blocked, s->io_q.plugged,
-                           s->io_q.in_queue, s->io_q.in_flight);
-    if (!s->io_q.blocked &&
-        (!s->io_q.plugged ||
-         s->io_q.in_flight + s->io_q.in_queue >= MAX_ENTRIES)) {
-        ret = ioq_submit(s);
-        trace_luring_do_submit_done(s, ret);
-        return ret;
+    trace_luring_do_submit(s, s->io_q.blocked, s->io_q.in_queue,
+                           s->io_q.in_flight);
+    if (!s->io_q.blocked) {
+        if (s->io_q.in_flight + s->io_q.in_queue >= MAX_ENTRIES) {
+            ret = ioq_submit(s);
+            trace_luring_do_submit_done(s, ret);
+            return ret;
+        }
+
+        blk_io_plug_call(luring_unplug_fn, s);
     }
     return 0;
 }
diff --git a/block/trace-events b/block/trace-events
index XXXXXXX..XXXXXXX 100644
--- a/block/trace-events
+++ b/block/trace-events
@@ -XXX,XX +XXX,XX @@ file_paio_submit(void *acb, void *opaque, int64_t offset, int count, int type) "
 # io_uring.c
 luring_init_state(void *s, size_t size) "s %p size %zu"
 luring_cleanup_state(void *s) "%p freed"
-luring_io_plug(void *s) "LuringState %p plug"
-luring_io_unplug(void *s, int blocked, int plugged, int queued, int inflight) "LuringState %p blocked %d plugged %d queued %d inflight %d"
-luring_do_submit(void *s, int blocked, int plugged, int queued, int inflight) "LuringState %p blocked %d plugged %d queued %d inflight %d"
+luring_unplug_fn(void *s, int blocked, int queued, int inflight) "LuringState %p blocked %d queued %d inflight %d"
+luring_do_submit(void *s, int blocked, int queued, int inflight) "LuringState %p blocked %d queued %d inflight %d"
 luring_do_submit_done(void *s, int ret) "LuringState %p submitted to kernel %d"
 luring_co_submit(void *bs, void *s, void *luringcb, int fd, uint64_t offset, size_t nbytes, int type) "bs %p s %p luringcb %p fd %d offset %" PRId64 " nbytes %zd type %d"
 luring_process_completion(void *s, void *aiocb, int ret) "LuringState %p luringcb %p ret %d"
-- 
2.40.1

Stop using the .bdrv_co_io_plug() API because it is not multi-queue
block layer friendly. Use the new blk_io_plug_call() API to batch I/O
submission instead.

Note that a dev_max_batch check is dropped in laio_io_unplug() because
the semantics of unplug_fn() are different from .bdrv_co_unplug():
1. unplug_fn() is only called when the last blk_io_unplug() call occurs,
   not every time blk_io_unplug() is called.
2. unplug_fn() is per-thread, not per-BlockDriverState, so there is no
   way to get per-BlockDriverState fields like dev_max_batch.

Therefore this condition cannot be moved to laio_unplug_fn(). It is not
obvious that this condition affects performance in practice, so I am
removing it instead of trying to come up with a more complex mechanism
to preserve the condition.

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Reviewed-by: Eric Blake <eblake@redhat.com>
Acked-by: Kevin Wolf <kwolf@redhat.com>
Reviewed-by: Stefano Garzarella <sgarzare@redhat.com>
Message-id: 20230530180959.1108766-6-stefanha@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 include/block/raw-aio.h |  7 -------
 block/file-posix.c      | 28 ----------------------------
 block/linux-aio.c       | 41 +++++++++++------------------------------
 3 files changed, 11 insertions(+), 65 deletions(-)

diff --git a/include/block/raw-aio.h b/include/block/raw-aio.h
index XXXXXXX..XXXXXXX 100644
--- a/include/block/raw-aio.h
+++ b/include/block/raw-aio.h
@@ -XXX,XX +XXX,XX @@ int coroutine_fn laio_co_submit(int fd, uint64_t offset, QEMUIOVector *qiov,
 
 void laio_detach_aio_context(LinuxAioState *s, AioContext *old_context);
 void laio_attach_aio_context(LinuxAioState *s, AioContext *new_context);
-
-/*
- * laio_io_plug/unplug work in the thread's current AioContext, therefore the
- * caller must ensure that they are paired in the same IOThread.
- */
-void laio_io_plug(void);
-void laio_io_unplug(uint64_t dev_max_batch);
 #endif
 /* io_uring.c - Linux io_uring implementation */
 #ifdef CONFIG_LINUX_IO_URING
diff --git a/block/file-posix.c b/block/file-posix.c
index XXXXXXX..XXXXXXX 100644
--- a/block/file-posix.c
+++ b/block/file-posix.c
@@ -XXX,XX +XXX,XX @@ static int coroutine_fn raw_co_pwritev(BlockDriverState *bs, int64_t offset,
     return raw_co_prw(bs, offset, bytes, qiov, QEMU_AIO_WRITE);
 }
 
-static void coroutine_fn raw_co_io_plug(BlockDriverState *bs)
-{
-    BDRVRawState __attribute__((unused)) *s = bs->opaque;
-#ifdef CONFIG_LINUX_AIO
-    if (s->use_linux_aio) {
-        laio_io_plug();
-    }
-#endif
-}
-
-static void coroutine_fn raw_co_io_unplug(BlockDriverState *bs)
-{
-    BDRVRawState __attribute__((unused)) *s = bs->opaque;
-#ifdef CONFIG_LINUX_AIO
-    if (s->use_linux_aio) {
-        laio_io_unplug(s->aio_max_batch);
-    }
-#endif
-}
-
 static int coroutine_fn raw_co_flush_to_disk(BlockDriverState *bs)
 {
     BDRVRawState *s = bs->opaque;
@@ -XXX,XX +XXX,XX @@ BlockDriver bdrv_file = {
     .bdrv_co_copy_range_from = raw_co_copy_range_from,
     .bdrv_co_copy_range_to  = raw_co_copy_range_to,
     .bdrv_refresh_limits = raw_refresh_limits,
-    .bdrv_co_io_plug        = raw_co_io_plug,
-    .bdrv_co_io_unplug      = raw_co_io_unplug,
     .bdrv_attach_aio_context = raw_aio_attach_aio_context,
 
     .bdrv_co_truncate                   = raw_co_truncate,
@@ -XXX,XX +XXX,XX @@ static BlockDriver bdrv_host_device = {
     .bdrv_co_copy_range_from = raw_co_copy_range_from,
     .bdrv_co_copy_range_to  = raw_co_copy_range_to,
     .bdrv_refresh_limits = raw_refresh_limits,
-    .bdrv_co_io_plug        = raw_co_io_plug,
-    .bdrv_co_io_unplug      = raw_co_io_unplug,
     .bdrv_attach_aio_context = raw_aio_attach_aio_context,
 
     .bdrv_co_truncate                   = raw_co_truncate,
@@ -XXX,XX +XXX,XX @@ static BlockDriver bdrv_host_cdrom = {
     .bdrv_co_pwritev        = raw_co_pwritev,
     .bdrv_co_flush_to_disk  = raw_co_flush_to_disk,
     .bdrv_refresh_limits    = cdrom_refresh_limits,
-    .bdrv_co_io_plug        = raw_co_io_plug,
-    .bdrv_co_io_unplug      = raw_co_io_unplug,
     .bdrv_attach_aio_context = raw_aio_attach_aio_context,
 
     .bdrv_co_truncate                   = raw_co_truncate,
@@ -XXX,XX +XXX,XX @@ static BlockDriver bdrv_host_cdrom = {
     .bdrv_co_pwritev        = raw_co_pwritev,
     .bdrv_co_flush_to_disk  = raw_co_flush_to_disk,
     .bdrv_refresh_limits    = cdrom_refresh_limits,
-    .bdrv_co_io_plug        = raw_co_io_plug,
-    .bdrv_co_io_unplug      = raw_co_io_unplug,
     .bdrv_attach_aio_context = raw_aio_attach_aio_context,
 
     .bdrv_co_truncate                   = raw_co_truncate,
diff --git a/block/linux-aio.c b/block/linux-aio.c
index XXXXXXX..XXXXXXX 100644
--- a/block/linux-aio.c
+++ b/block/linux-aio.c
@@ -XXX,XX +XXX,XX @@
 #include "qemu/event_notifier.h"
 #include "qemu/coroutine.h"
 #include "qapi/error.h"
+#include "sysemu/block-backend.h"
 
 /* Only used for assertions.  */
 #include "qemu/coroutine_int.h"
@@ -XXX,XX +XXX,XX @@ struct qemu_laiocb {
 };
 
 typedef struct {
-    int plugged;
     unsigned int in_queue;
     unsigned int in_flight;
     bool blocked;
@@ -XXX,XX +XXX,XX @@ static void qemu_laio_process_completions_and_submit(LinuxAioState *s)
 {
     qemu_laio_process_completions(s);
 
-    if (!s->io_q.plugged && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
+    if (!QSIMPLEQ_EMPTY(&s->io_q.pending)) {
         ioq_submit(s);
     }
 }
@@ -XXX,XX +XXX,XX @@ static void qemu_laio_poll_ready(EventNotifier *opaque)
 static void ioq_init(LaioQueue *io_q)
 {
     QSIMPLEQ_INIT(&io_q->pending);
-    io_q->plugged = 0;
     io_q->in_queue = 0;
     io_q->in_flight = 0;
     io_q->blocked = false;
@@ -XXX,XX +XXX,XX @@ static uint64_t laio_max_batch(LinuxAioState *s, uint64_t dev_max_batch)
     return max_batch;
 }
 
-void laio_io_plug(void)
+static void laio_unplug_fn(void *opaque)
 {
-    AioContext *ctx = qemu_get_current_aio_context();
-    LinuxAioState *s = aio_get_linux_aio(ctx);
+    LinuxAioState *s = opaque;
 
-    s->io_q.plugged++;
-}
-
-void laio_io_unplug(uint64_t dev_max_batch)
-{
-    AioContext *ctx = qemu_get_current_aio_context();
-    LinuxAioState *s = aio_get_linux_aio(ctx);
-
-    assert(s->io_q.plugged);
-    s->io_q.plugged--;
-
-    /*
-     * Why max batch checking is performed here:
-     * Another BDS may have queued requests with a higher dev_max_batch and
-     * therefore in_queue could now exceed our dev_max_batch. Re-check the max
-     * batch so we can honor our device's dev_max_batch.
-     */
-    if (s->io_q.in_queue >= laio_max_batch(s, dev_max_batch) ||
-        (!s->io_q.plugged &&
-         !s->io_q.blocked && !QSIMPLEQ_EMPTY(&s->io_q.pending))) {
+    if (!s->io_q.blocked && !QSIMPLEQ_EMPTY(&s->io_q.pending)) {
         ioq_submit(s);
     }
 }
@@ -XXX,XX +XXX,XX @@ static int laio_do_submit(int fd, struct qemu_laiocb *laiocb, off_t offset,
 
     QSIMPLEQ_INSERT_TAIL(&s->io_q.pending, laiocb, next);
     s->io_q.in_queue++;
-    if (!s->io_q.blocked &&
-        (!s->io_q.plugged ||
-         s->io_q.in_queue >= laio_max_batch(s, dev_max_batch))) {
-        ioq_submit(s);
+    if (!s->io_q.blocked) {
+        if (s->io_q.in_queue >= laio_max_batch(s, dev_max_batch)) {
+            ioq_submit(s);
+        } else {
+            blk_io_plug_call(laio_unplug_fn, s);
+        }
     }
 
     return 0;
-- 
2.40.1

No block driver implements .bdrv_co_io_plug() anymore. Get rid of the
function pointers.

diff --git a/include/block/block-io.h b/include/block/block-io.h
index XXXXXXX..XXXXXXX 100644
--- a/include/block/block-io.h
+++ b/include/block/block-io.h
@@ -XXX,XX +XXX,XX @@ void coroutine_fn bdrv_co_leave(BlockDriverState *bs, AioContext *old_ctx);
 
 AioContext *child_of_bds_get_parent_aio_context(BdrvChild *c);
 
-void coroutine_fn GRAPH_RDLOCK bdrv_co_io_plug(BlockDriverState *bs);
-void coroutine_fn GRAPH_RDLOCK bdrv_co_io_unplug(BlockDriverState *bs);
-
 bool coroutine_fn GRAPH_RDLOCK
 bdrv_co_can_store_new_dirty_bitmap(BlockDriverState *bs, const char *name,
                                    uint32_t granularity, Error **errp);
diff --git a/include/block/block_int-common.h b/include/block/block_int-common.h
index XXXXXXX..XXXXXXX 100644
--- a/include/block/block_int-common.h
+++ b/include/block/block_int-common.h
@@ -XXX,XX +XXX,XX @@ struct BlockDriver {
     void coroutine_fn GRAPH_RDLOCK_PTR (*bdrv_co_debug_event)(
         BlockDriverState *bs, BlkdebugEvent event);
 
-    /* io queue for linux-aio */
-    void coroutine_fn GRAPH_RDLOCK_PTR (*bdrv_co_io_plug)(BlockDriverState *bs);
-    void coroutine_fn GRAPH_RDLOCK_PTR (*bdrv_co_io_unplug)(
-        BlockDriverState *bs);
-
     bool (*bdrv_supports_persistent_dirty_bitmap)(BlockDriverState *bs);
 
     bool coroutine_fn GRAPH_RDLOCK_PTR (*bdrv_co_can_store_new_dirty_bitmap)(
@@ -XXX,XX +XXX,XX @@ struct BlockDriverState {
     unsigned int in_flight;
     unsigned int serialising_in_flight;
 
-    /*
-     * counter for nested bdrv_io_plug.
-     * Accessed with atomic ops.
-     */
-    unsigned io_plugged;
-
     /* do we need to tell the quest if we have a volatile write cache? */
     int enable_write_cache;
 
diff --git a/block/io.c b/block/io.c
index XXXXXXX..XXXXXXX 100644
--- a/block/io.c
+++ b/block/io.c
@@ -XXX,XX +XXX,XX @@ void *qemu_try_blockalign0(BlockDriverState *bs, size_t size)
     return mem;
 }
 
-void coroutine_fn bdrv_co_io_plug(BlockDriverState *bs)
-{
-    BdrvChild *child;
-    IO_CODE();
-    assert_bdrv_graph_readable();
-
-    QLIST_FOREACH(child, &bs->children, next) {
-        bdrv_co_io_plug(child->bs);
-    }
-
-    if (qatomic_fetch_inc(&bs->io_plugged) == 0) {
-        BlockDriver *drv = bs->drv;
-        if (drv && drv->bdrv_co_io_plug) {
-            drv->bdrv_co_io_plug(bs);
-        }
-    }
-}
-
-void coroutine_fn bdrv_co_io_unplug(BlockDriverState *bs)
-{
-    BdrvChild *child;
-    IO_CODE();
-    assert_bdrv_graph_readable();
-
-    assert(bs->io_plugged);
-    if (qatomic_fetch_dec(&bs->io_plugged) == 1) {
-        BlockDriver *drv = bs->drv;
-        if (drv && drv->bdrv_co_io_unplug) {
-            drv->bdrv_co_io_unplug(bs);
-        }
-    }
-
-    QLIST_FOREACH(child, &bs->children, next) {
-        bdrv_co_io_unplug(child->bs);
-    }
-}
-
 /* Helper that undoes bdrv_register_buf() when it fails partway through */
 static void GRAPH_RDLOCK
 bdrv_register_buf_rollback(BlockDriverState *bs, void *host, size_t size,
-- 
2.40.1

From: Stefano Garzarella <sgarzare@redhat.com>

Some virtio-blk drivers (e.g. virtio-blk-vhost-vdpa) supports the fd
passing. Let's expose this to the user, so the management layer
can pass the file descriptor of an already opened path.

If the libblkio virtio-blk driver supports fd passing, let's always
use qemu_open() to open the `path`, so we can handle fd passing
from the management layer through the "/dev/fdset/N" special path.

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Signed-off-by: Stefano Garzarella <sgarzare@redhat.com>
Message-id: 20230530071941.8954-2-sgarzare@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 block/blkio.c | 53 ++++++++++++++++++++++++++++++++++++++++++---------
 1 file changed, 44 insertions(+), 9 deletions(-)

diff --git a/block/blkio.c b/block/blkio.c
index XXXXXXX..XXXXXXX 100644
--- a/block/blkio.c
+++ b/block/blkio.c
@@ -XXX,XX +XXX,XX @@ static int blkio_virtio_blk_common_open(BlockDriverState *bs,
 {
     const char *path = qdict_get_try_str(options, "path");
     BDRVBlkioState *s = bs->opaque;
-    int ret;
+    bool fd_supported = false;
+    int fd, ret;
 
     if (!path) {
         error_setg(errp, "missing 'path' option");
         return -EINVAL;
     }
 
-    ret = blkio_set_str(s->blkio, "path", path);
-    qdict_del(options, "path");
-    if (ret < 0) {
-        error_setg_errno(errp, -ret, "failed to set path: %s",
-                         blkio_get_error_msg());
-        return ret;
-    }
-
     if (!(flags & BDRV_O_NOCACHE)) {
         error_setg(errp, "cache.direct=off is not supported");
         return -EINVAL;
     }
+
+    if (blkio_get_int(s->blkio, "fd", &fd) == 0) {
+        fd_supported = true;
+    }
+
+    /*
+     * If the libblkio driver supports fd passing, let's always use qemu_open()
+     * to open the `path`, so we can handle fd passing from the management
+     * layer through the "/dev/fdset/N" special path.
+     */
+    if (fd_supported) {
+        int open_flags;
+
+        if (flags & BDRV_O_RDWR) {
+            open_flags = O_RDWR;
+        } else {
+            open_flags = O_RDONLY;
+        }
+
+        fd = qemu_open(path, open_flags, errp);
+        if (fd < 0) {
+            return -EINVAL;
+        }
+
+        ret = blkio_set_int(s->blkio, "fd", fd);
+        if (ret < 0) {
+            error_setg_errno(errp, -ret, "failed to set fd: %s",
+                             blkio_get_error_msg());
+            qemu_close(fd);
+            return ret;
+        }
+    } else {
+        ret = blkio_set_str(s->blkio, "path", path);
+        if (ret < 0) {
+            error_setg_errno(errp, -ret, "failed to set path: %s",
+                             blkio_get_error_msg());
+            return ret;
+        }
+    }
+
+    qdict_del(options, "path");
+
     return 0;
 }
 
-- 
2.40.1

From: Stefano Garzarella <sgarzare@redhat.com>

The virtio-blk-vhost-vdpa driver in libblkio 1.3.0 supports the fd
passing through the new 'fd' property.

Since now we are using qemu_open() on '@path' if the virtio-blk driver
supports the fd passing, let's announce it.
In this way, the management layer can pass the file descriptor of an
already opened vhost-vdpa character device. This is useful especially
when the device can only be accessed with certain privileges.

Add the '@fdset' feature only when the virtio-blk-vhost-vdpa driver
in libblkio supports it.

Suggested-by: Markus Armbruster <armbru@redhat.com>
Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Signed-off-by: Stefano Garzarella <sgarzare@redhat.com>
Message-id: 20230530071941.8954-3-sgarzare@redhat.com
Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
---
 qapi/block-core.json | 6 ++++++
 meson.build          | 4 ++++
 2 files changed, 10 insertions(+)

diff --git a/qapi/block-core.json b/qapi/block-core.json
index XXXXXXX..XXXXXXX 100644
--- a/qapi/block-core.json
+++ b/qapi/block-core.json
@@ -XXX,XX +XXX,XX @@
 #
 # @path: path to the vhost-vdpa character device.
 #
+# Features:
+# @fdset: Member @path supports the special "/dev/fdset/N" path
+#     (since 8.1)
+#
 # Since: 7.2
 ##
 { 'struct': 'BlockdevOptionsVirtioBlkVhostVdpa',
   'data': { 'path': 'str' },
+  'features': [ { 'name' :'fdset',
+                  'if': 'CONFIG_BLKIO_VHOST_VDPA_FD' } ],
   'if': 'CONFIG_BLKIO' }
 
 ##
diff --git a/meson.build b/meson.build
index XXXXXXX..XXXXXXX 100644
--- a/meson.build
+++ b/meson.build
@@ -XXX,XX +XXX,XX @@ config_host_data.set('CONFIG_LZO', lzo.found())
 config_host_data.set('CONFIG_MPATH', mpathpersist.found())
 config_host_data.set('CONFIG_MPATH_NEW_API', mpathpersist_new_api)
 config_host_data.set('CONFIG_BLKIO', blkio.found())
+if blkio.found()
+  config_host_data.set('CONFIG_BLKIO_VHOST_VDPA_FD',
+                       blkio.version().version_compare('>=1.3.0'))
+endif
 config_host_data.set('CONFIG_CURL', curl.found())
 config_host_data.set('CONFIG_CURSES', curses.found())
 config_host_data.set('CONFIG_GBM', gbm.found())
-- 
2.40.1