[PATCH v2 0/4] util/thread-pool: Expose minimun and maximum size

[PATCH v2 0/4] util/thread-pool: Expose minimun and maximum size

[Nicolas Saenz Julienne]

As discussed on the previous RFC[1] the thread-pool's dynamic thread
management doesn't play well with real-time and latency sensitive
systems. This series introduces a set of controls that'll permit
achieving more deterministic behaviours, for example by fixing the
pool's size.

We first introduce a new common interface to event loop configuration by
moving iothread's already available properties into an abstract class
called 'EventLooopBackend' and have both 'IOThread' and the newly
created 'MainLoop' inherit the properties from that class.

With this new configuration interface in place it's relatively simple to
introduce new options to fix the even loop's thread pool sizes. The
resulting QAPI looks like this:

    -object main-loop,id=main-loop,thread-pool-min=1,thread-pool-max=1

Note that all patches are bisect friendly and pass all the tests.

[1] https://patchwork.ozlabs.org/project/qemu-devel/patch/20220202175234.656711-1-nsaenzju@redhat.com/

---
Changes since v1:
 - Address all Stefan's comments
 - Introduce new fix

Nicolas Saenz Julienne (4):
  util/thread-pool: Fix thread pool freeing locking
  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

 event-loop-base.c                | 140 +++++++++++++++++++++++++++++++
 include/block/aio.h              |  10 +++
 include/block/thread-pool.h      |   3 +
 include/qemu/main-loop.h         |  10 +++
 include/sysemu/event-loop-base.h |  41 +++++++++
 include/sysemu/iothread.h        |   6 +-
 iothread.c                       |  68 +++++----------
 meson.build                      |  26 +++---
 qapi/qom.json                    |  36 +++++++-
 util/aio-posix.c                 |   1 +
 util/async.c                     |  20 +++++
 util/main-loop.c                 |  65 ++++++++++++++
 util/thread-pool.c               |  61 +++++++++++++-
 13 files changed, 422 insertions(+), 65 deletions(-)
 create mode 100644 event-loop-base.c
 create mode 100644 include/sysemu/event-loop-base.h

-- 
2.35.1

[PATCH v2 1/4] util/thread-pool: Fix thread pool freeing locking

[Nicolas Saenz Julienne]

Upon freeing a thread pool we need to get rid of any remaining worker.
This is achieved by setting the thread pool's topping flag, waking the
workers up, and waiting for them to exit one by one. The problem is that
currently all this process happens with the thread pool lock held,
effectively blocking the workers from exiting.

So let's release the thread pool lock after signaling a worker thread
that it's time to exit to give it a chance to do so.

Fixes: f7311ccc63 ("threadpool: add thread_pool_new() and thread_pool_free()")
Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
---
 util/thread-pool.c | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/util/thread-pool.c b/util/thread-pool.c
index d763cea505..fdb43c2d3b 100644
--- a/util/thread-pool.c
+++ b/util/thread-pool.c
@@ -339,7 +339,9 @@ void thread_pool_free(ThreadPool *pool)
     pool->stopping = true;
     while (pool->cur_threads > 0) {
         qemu_sem_post(&pool->sem);
+        qemu_mutex_unlock(&pool->lock);
         qemu_cond_wait(&pool->worker_stopped, &pool->lock);
+        qemu_mutex_lock(&pool->lock);
     }
 
     qemu_mutex_unlock(&pool->lock);
-- 
2.35.1

[PATCH v2 2/4] Introduce event-loop-base abstract class

[Nicolas Saenz Julienne]

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, and it
imposes new compilation dependencies:

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

It is built with on its own using the link_whole option as there are no
direct function dependencies between the class and its children
(everything happens through the 'contructor' attribute). All this forced
some amount of reordering in meson.build, among other things the 'hw'
subdir is processed earlier as it introduces files into the 'qom' source
set.

No functional changes intended.

Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
---

Changes since v1:
 - Rename to event-loop-base
 - Move event-loop-base into root directory
 - Build event-loop-base on its own, use link_whole to avoid the problem
   of the object file not being linked due to lacking direct calls from
   dependencies.
 - Move poll parameters into iothread, as main loop can't poll
 - Update Authorship (I took what iothread.c had and added myself, I
   hope that's fine)
 - Introduce update_params() callback

 event-loop-base.c                | 104 +++++++++++++++++++++++++++++++
 include/sysemu/event-loop-base.h |  36 +++++++++++
 include/sysemu/iothread.h        |   6 +-
 iothread.c                       |  65 ++++++-------------
 meson.build                      |  23 ++++---
 5 files changed, 175 insertions(+), 59 deletions(-)
 create mode 100644 event-loop-base.c
 create mode 100644 include/sysemu/event-loop-base.h

diff --git a/event-loop-base.c b/event-loop-base.c
new file mode 100644
index 0000000000..a924c73a7c
--- /dev/null
+++ b/event-loop-base.c
@@ -0,0 +1,104 @@
+/*
+ * 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/include/sysemu/event-loop-base.h b/include/sysemu/event-loop-base.h
new file mode 100644
index 0000000000..8e77d8b69f
--- /dev/null
+++ b/include/sysemu/event-loop-base.h
@@ -0,0 +1,36 @@
+/*
+ * 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 7f714bd136..8f8601d6ab 100644
--- a/include/sysemu/iothread.h
+++ b/include/sysemu/iothread.h
@@ -17,11 +17,12 @@
 #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;
@@ -37,9 +38,6 @@ 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/iothread.c b/iothread.c
index 0f98af0f2a..8fa2f3bfb8 100644
--- a/iothread.c
+++ b/iothread.c
@@ -17,6 +17,7 @@
 #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"
@@ -152,10 +153,15 @@ 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,
@@ -166,14 +172,15 @@ 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;
@@ -189,7 +196,7 @@ 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);
@@ -201,7 +208,7 @@ 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);
@@ -226,9 +233,6 @@ 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)
@@ -288,35 +292,12 @@ 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,
@@ -330,23 +311,15 @@ 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)
@@ -383,7 +356,7 @@ 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;
diff --git a/meson.build b/meson.build
index 8df40bfac4..3c90b4f2aa 100644
--- a/meson.build
+++ b/meson.build
@@ -2717,6 +2717,7 @@ subdir('qom')
 subdir('authz')
 subdir('crypto')
 subdir('ui')
+subdir('hw')
 
 
 if enable_modules
@@ -2724,6 +2725,18 @@ 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,
+                                 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)
@@ -2810,7 +2823,6 @@ subdir('monitor')
 subdir('net')
 subdir('replay')
 subdir('semihosting')
-subdir('hw')
 subdir('tcg')
 subdir('fpu')
 subdir('accel')
@@ -2935,13 +2947,6 @@ 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()],
@@ -2994,7 +2999,7 @@ 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,
-- 
2.35.1

[PATCH v2 3/4] util/main-loop: Introduce the main loop into QOM

[Nicolas Saenz Julienne]

'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.

Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>

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

Changes since v1:
 - Fix json files to differentiate between iothread and main-loop
 - Use OBJECT_DECLARE_TYPE()
 - Fix build dependencies

 event-loop-base.c                | 13 ++++++++
 include/qemu/main-loop.h         | 10 ++++++
 include/sysemu/event-loop-base.h |  1 +
 meson.build                      |  3 +-
 qapi/qom.json                    | 16 +++++++++
 util/main-loop.c                 | 56 ++++++++++++++++++++++++++++++++
 6 files changed, 98 insertions(+), 1 deletion(-)

diff --git a/event-loop-base.c b/event-loop-base.c
index a924c73a7c..e7f99a6ec8 100644
--- a/event-loop-base.c
+++ b/event-loop-base.c
@@ -73,10 +73,23 @@ 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/include/qemu/main-loop.h b/include/qemu/main-loop.h
index 8dbc6fcb89..71231729c2 100644
--- a/include/qemu/main-loop.h
+++ b/include/qemu/main-loop.h
@@ -26,9 +26,19 @@
 #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 8e77d8b69f..fced4c9fea 100644
--- a/include/sysemu/event-loop-base.h
+++ b/include/sysemu/event-loop-base.h
@@ -25,6 +25,7 @@ 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/meson.build b/meson.build
index 3c90b4f2aa..2b9b7202ac 100644
--- a/meson.build
+++ b/meson.build
@@ -2745,7 +2745,8 @@ 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/qapi/qom.json b/qapi/qom.json
index eeb5395ff3..c8eb192772 100644
--- a/qapi/qom.json
+++ b/qapi/qom.json
@@ -528,6 +528,20 @@
             '*poll-shrink': 'int',
             '*aio-max-batch': 'int' } }
 
+##
+# @MainLoopProperties:
+#
+# Properties for the main-loop object.
+#
+# @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)
+#
+# Since: 2.0
+##
+{ 'struct': 'MainLoopProperties',
+  'data': { '*aio-max-batch': 'int' } }
+
 ##
 # @MemoryBackendProperties:
 #
@@ -818,6 +832,7 @@
     { 'name': 'input-linux',
       'if': 'CONFIG_LINUX' },
     'iothread',
+    'main-loop',
     { 'name': 'memory-backend-epc',
       'if': 'CONFIG_LINUX' },
     'memory-backend-file',
@@ -883,6 +898,7 @@
       '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/util/main-loop.c b/util/main-loop.c
index 4d5a5b9943..3bf5709374 100644
--- a/util/main-loop.c
+++ b/util/main-loop.c
@@ -33,6 +33,7 @@
 #include "qemu/error-report.h"
 #include "qemu/queue.h"
 #include "qemu/compiler.h"
+#include "qom/object.h"
 
 #ifndef _WIN32
 #include <sys/wait.h>
@@ -184,6 +185,61 @@ 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

[PATCH v2 4/4] util/event-loop-base: Introduce options to set the thread pool size

[Nicolas Saenz Julienne]

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 'EventLoopBaase'
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>
---

Changes since v1:
 - Add INT_MAX check
 - Have copy of thread pool sizes in AioContext to properly decouple
   both instances
 - More coherent variable naming
 - Handle case where max_threads decreases
 - Code comments

 event-loop-base.c                | 23 +++++++++++++
 include/block/aio.h              | 10 ++++++
 include/block/thread-pool.h      |  3 ++
 include/sysemu/event-loop-base.h |  4 +++
 iothread.c                       |  3 ++
 qapi/qom.json                    | 22 ++++++++++--
 util/aio-posix.c                 |  1 +
 util/async.c                     | 20 +++++++++++
 util/main-loop.c                 |  9 +++++
 util/thread-pool.c               | 59 +++++++++++++++++++++++++++++---
 10 files changed, 148 insertions(+), 6 deletions(-)

diff --git a/event-loop-base.c b/event-loop-base.c
index e7f99a6ec8..d5be4dc6fc 100644
--- a/event-loop-base.c
+++ b/event-loop-base.c
@@ -14,6 +14,7 @@
 #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 {
@@ -21,9 +22,22 @@ 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)
@@ -95,12 +109,21 @@ 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/include/block/aio.h b/include/block/aio.h
index 5634173b12..d128558f1d 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -192,6 +192,8 @@ 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.
      */
@@ -769,4 +771,12 @@ 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 7dd7d730a0..2020bcc92d 100644
--- a/include/block/thread-pool.h
+++ b/include/block/thread-pool.h
@@ -20,6 +20,8 @@
 
 #include "block/block.h"
 
+#define THREAD_POOL_MAX_THREADS_DEFAULT         64
+
 typedef int ThreadPoolFunc(void *opaque);
 
 typedef struct ThreadPool ThreadPool;
@@ -33,5 +35,6 @@ 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 fced4c9fea..2748bf6ae1 100644
--- a/include/sysemu/event-loop-base.h
+++ b/include/sysemu/event-loop-base.h
@@ -33,5 +33,9 @@ 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/iothread.c b/iothread.c
index 8fa2f3bfb8..529194a566 100644
--- a/iothread.c
+++ b/iothread.c
@@ -174,6 +174,9 @@ 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/qapi/qom.json b/qapi/qom.json
index c8eb192772..54ef28b4a9 100644
--- a/qapi/qom.json
+++ b/qapi/qom.json
@@ -520,13 +520,22 @@
 #                 0 means that the engine will use its default
 #                 (default:0, since 6.1)
 #
+# @thread-pool-min: minimum number of threads readily available in the thread
+#                   pool
+#                   (default:0, since 6.2)
+#
+# @thread-pool-max: maximum number of threads the thread pool can contain
+#                   (default:64, since 6.2)
+#
 # Since: 2.0
 ##
 { 'struct': 'IothreadProperties',
   'data': { '*poll-max-ns': 'int',
             '*poll-grow': 'int',
             '*poll-shrink': 'int',
-            '*aio-max-batch': 'int' } }
+            '*aio-max-batch': 'int',
+            '*thread-pool-min': 'int',
+            '*thread-pool-max': 'int' } }
 
 ##
 # @MainLoopProperties:
@@ -537,10 +546,19 @@
 #                 0 means that the engine will use its default
 #                 (default:0, since 6.1)
 #
+# @thread-pool-min: minimum number of threads readily available in the thread
+#                   pool
+#                   (default:0, since 6.2)
+#
+# @thread-pool-max: maximum number of threads the thread pool can contain
+#                   (default:64, since 6.2)
+#
 # Since: 2.0
 ##
 { 'struct': 'MainLoopProperties',
-  'data': { '*aio-max-batch': 'int' } }
+  'data': { '*aio-max-batch': 'int',
+            '*thread-pool-min': 'int',
+            '*thread-pool-max': 'int' } }
 
 ##
 # @MemoryBackendProperties:
diff --git a/util/aio-posix.c b/util/aio-posix.c
index 7b9f629218..926e6dafba 100644
--- a/util/aio-posix.c
+++ b/util/aio-posix.c
@@ -15,6 +15,7 @@
 
 #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 08d25feef5..93469a4668 100644
--- a/util/async.c
+++ b/util/async.c
@@ -562,6 +562,9 @@ 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);
@@ -694,3 +697,20 @@ void qemu_set_current_aio_context(AioContext *ctx)
     assert(!my_aiocontext);
     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 3bf5709374..ae02afb0f8 100644
--- a/util/main-loop.c
+++ b/util/main-loop.c
@@ -30,6 +30,7 @@
 #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"
@@ -187,12 +188,20 @@ 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 fdb43c2d3b..749bcdcfad 100644
--- a/util/thread-pool.c
+++ b/util/thread-pool.c
@@ -21,6 +21,7 @@
 #include "trace.h"
 #include "block/thread-pool.h"
 #include "qemu/main-loop.h"
+#include "qapi/error.h"
 
 static void do_spawn_thread(ThreadPool *pool);
 
@@ -58,7 +59,6 @@ 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. */
@@ -71,8 +71,27 @@ 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;
@@ -91,8 +110,9 @@ 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;
         }
 
@@ -294,6 +314,36 @@ 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);
+    }
+
+    while (pool->cur_threads > pool->max_threads) {
+        qemu_sem_post(&pool->sem);
+        qemu_mutex_unlock(&pool->lock);
+        qemu_cond_wait(&pool->worker_stopped, &pool->lock);
+        qemu_mutex_lock(&pool->lock);
+    }
+
+    qemu_mutex_unlock(&pool->lock);
+}
+
 static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
 {
     if (!ctx) {
@@ -306,11 +356,12 @@ 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

Re: [PATCH v2 1/4] util/thread-pool: Fix thread pool freeing locking

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1949 bytes --]

On Thu, Mar 03, 2022 at 03:58:19PM +0100, Nicolas Saenz Julienne wrote:
> Upon freeing a thread pool we need to get rid of any remaining worker.
> This is achieved by setting the thread pool's topping flag, waking the

s/topping/stopping/

> workers up, and waiting for them to exit one by one. The problem is that
> currently all this process happens with the thread pool lock held,
> effectively blocking the workers from exiting.
> 
> So let's release the thread pool lock after signaling a worker thread
> that it's time to exit to give it a chance to do so.
> 
> Fixes: f7311ccc63 ("threadpool: add thread_pool_new() and thread_pool_free()")
> Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
> ---
>  util/thread-pool.c | 2 ++
>  1 file changed, 2 insertions(+)
> 
> diff --git a/util/thread-pool.c b/util/thread-pool.c
> index d763cea505..fdb43c2d3b 100644
> --- a/util/thread-pool.c
> +++ b/util/thread-pool.c
> @@ -339,7 +339,9 @@ void thread_pool_free(ThreadPool *pool)
>      pool->stopping = true;
>      while (pool->cur_threads > 0) {
>          qemu_sem_post(&pool->sem);
> +        qemu_mutex_unlock(&pool->lock);
>          qemu_cond_wait(&pool->worker_stopped, &pool->lock);
> +        qemu_mutex_lock(&pool->lock);

This patch looks incorrect. qemu_cond_wait() (and pthread_cond_wait())
take a mutex as the second argument. This mutex is released and the
thread blocks to wait (this is done atomically with respect to the
condvar check so there are no races).

In other words, qemu_cond_wait(&pool->worker_stopped, &pool->lock)
already releases pool->lock. It is unnecessary to release/re-acquire it
manually plus it probably suffers from a race condition in case the
other thread signals the condvar between the time when we unlock and
before we block on the condvar.

Please check that this patch really solves a problem and if so, please
explain the root cause.

Stefan

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 1/4] util/thread-pool: Fix thread pool freeing locking

[Nicolas Saenz Julienne]

On Thu, 2022-03-10 at 09:20 +0000, Stefan Hajnoczi wrote:
> On Thu, Mar 03, 2022 at 03:58:19PM +0100, Nicolas Saenz Julienne wrote:
> > Upon freeing a thread pool we need to get rid of any remaining worker.
> > This is achieved by setting the thread pool's topping flag, waking the
> 
> s/topping/stopping/
> 
> > workers up, and waiting for them to exit one by one. The problem is that
> > currently all this process happens with the thread pool lock held,
> > effectively blocking the workers from exiting.
> > 
> > So let's release the thread pool lock after signaling a worker thread
> > that it's time to exit to give it a chance to do so.
> > 
> > Fixes: f7311ccc63 ("threadpool: add thread_pool_new() and thread_pool_free()")
> > Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
> > ---
> >  util/thread-pool.c | 2 ++
> >  1 file changed, 2 insertions(+)
> > 
> > diff --git a/util/thread-pool.c b/util/thread-pool.c
> > index d763cea505..fdb43c2d3b 100644
> > --- a/util/thread-pool.c
> > +++ b/util/thread-pool.c
> > @@ -339,7 +339,9 @@ void thread_pool_free(ThreadPool *pool)
> >      pool->stopping = true;
> >      while (pool->cur_threads > 0) {
> >          qemu_sem_post(&pool->sem);
> > +        qemu_mutex_unlock(&pool->lock);
> >          qemu_cond_wait(&pool->worker_stopped, &pool->lock);
> > +        qemu_mutex_lock(&pool->lock);
> 
> This patch looks incorrect. qemu_cond_wait() (and pthread_cond_wait())
> take a mutex as the second argument. This mutex is released and the
> thread blocks to wait (this is done atomically with respect to the
> condvar check so there are no races).

Sorry I completely missed that. It was a late addition, should've thought it
trough. Patch #4 also needs to take this into account as I follow the same
pattern.

Thanks,

-- 
Nicolás Sáenz

Re: [PATCH v2 2/4] Introduce event-loop-base abstract class

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 758 bytes --]

On Thu, Mar 03, 2022 at 03:58:20PM +0100, Nicolas Saenz Julienne wrote:
> @@ -2935,13 +2947,6 @@ 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)
> -

Why was it necessary to move qom_ss and subdir('hw') up? Can
event_loop_base be defined down here instead?

(The benefit of less code churn is it reduces the risk of patch conflicts.)

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 3/4] util/main-loop: Introduce the main loop into QOM

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1299 bytes --]

On Thu, Mar 03, 2022 at 03:58:21PM +0100, Nicolas Saenz Julienne wrote:
> '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.
> 
> Signed-off-by: Nicolas Saenz Julienne <nsaenzju@redhat.com>
> 
> [1] For example:
>       -object main-loop,id=main-loop,aio-max-batch=<value>
> ---
> 
> Changes since v1:
>  - Fix json files to differentiate between iothread and main-loop
>  - Use OBJECT_DECLARE_TYPE()
>  - Fix build dependencies
> 
>  event-loop-base.c                | 13 ++++++++
>  include/qemu/main-loop.h         | 10 ++++++
>  include/sysemu/event-loop-base.h |  1 +
>  meson.build                      |  3 +-
>  qapi/qom.json                    | 16 +++++++++
>  util/main-loop.c                 | 56 ++++++++++++++++++++++++++++++++
>  6 files changed, 98 insertions(+), 1 deletion(-)

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 4/4] util/event-loop-base: Introduce options to set the thread pool size

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1937 bytes --]

On Thu, Mar 03, 2022 at 04:13:07PM +0100, Nicolas Saenz Julienne wrote:
> @@ -537,10 +546,19 @@
>  #                 0 means that the engine will use its default
>  #                 (default:0, since 6.1)
>  #
> +# @thread-pool-min: minimum number of threads readily available in the thread
> +#                   pool
> +#                   (default:0, since 6.2)
> +#
> +# @thread-pool-max: maximum number of threads the thread pool can contain
> +#                   (default:64, since 6.2)

Here and elsewhere:
s/6.2/7.1/

> @@ -294,6 +314,36 @@ 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);
> +    }
> +
> +    while (pool->cur_threads > pool->max_threads) {
> +        qemu_sem_post(&pool->sem);
> +        qemu_mutex_unlock(&pool->lock);
> +        qemu_cond_wait(&pool->worker_stopped, &pool->lock);
> +        qemu_mutex_lock(&pool->lock);

Same question as Patch 1. This looks incorrect because qemu_cond_wait()
already drops pool->lock if it needs to block.

Also, why wait? If worker threads are blocked for some reason then our
thread will block too.

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 2/4] Introduce event-loop-base abstract class

[Nicolas Saenz Julienne]

On Thu, 2022-03-10 at 10:25 +0000, Stefan Hajnoczi wrote:
> On Thu, Mar 03, 2022 at 03:58:20PM +0100, Nicolas Saenz Julienne wrote:
> > @@ -2935,13 +2947,6 @@ 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)
> > -
> 
> Why was it necessary to move qom_ss and subdir('hw') up? Can
> event_loop_base be defined down here instead?

The way I setup it up, qemuutil now depdens on event_loop_base which in turn
depends on qom. IIUC I can't declare dependencies without declaring first the
libraries and source sets. All has to happen sequencially. With this in mind,
almost all libraries depend on libqemuutil so moving it down isn't possible.

subdir('hw') was also moved up, as 'hw/nvram/meson.build' is introducing files
into qom_ss. This operation has to be performed before declaring libqom.

> (The benefit of less code churn is it reduces the risk of patch conflicts.)

Agree, I know how painful it can be for backports.

Thanks,

-- 
Nicolás Sáenz

Re: [PATCH v2 4/4] util/event-loop-base: Introduce options to set the thread pool size

[Nicolas Saenz Julienne]

On Thu, 2022-03-10 at 10:45 +0000, Stefan Hajnoczi wrote:
> On Thu, Mar 03, 2022 at 04:13:07PM +0100, Nicolas Saenz Julienne wrote:
> > @@ -537,10 +546,19 @@
> >  #                 0 means that the engine will use its default
> >  #                 (default:0, since 6.1)
> >  #
> > +# @thread-pool-min: minimum number of threads readily available in the thread
> > +#                   pool
> > +#                   (default:0, since 6.2)
> > +#
> > +# @thread-pool-max: maximum number of threads the thread pool can contain
> > +#                   (default:64, since 6.2)
> 
> Here and elsewhere:
> s/6.2/7.1/

Yes, forgot to mention it was a placeholder, as I wasn't sure what version to
target.

> > @@ -294,6 +314,36 @@ 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);
> > +    }
> > +
> > +    while (pool->cur_threads > pool->max_threads) {
> > +        qemu_sem_post(&pool->sem);
> > +        qemu_mutex_unlock(&pool->lock);
> > +        qemu_cond_wait(&pool->worker_stopped, &pool->lock);
> > +        qemu_mutex_lock(&pool->lock);
> 
> Same question as Patch 1. This looks incorrect because qemu_cond_wait()
> already drops pool->lock if it needs to block.

Yes, I'll fix that.

> Also, why wait? If worker threads are blocked for some reason then our
> thread will block too.

Exiting thread_pool_update_params() before honoring the new constraints is a
source of potential race conditions (having to worry for situations where
cur_threads > max_threads), and on systems where determinism is important it's
crucial to have a clear boundary between 'unsafe' and 'safe' states.

Thanks,

-- 
Nicolás Sáenz

Re: [PATCH v2 2/4] Introduce event-loop-base abstract class

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1362 bytes --]

On Fri, Mar 11, 2022 at 11:17:22AM +0100, Nicolas Saenz Julienne wrote:
> On Thu, 2022-03-10 at 10:25 +0000, Stefan Hajnoczi wrote:
> > On Thu, Mar 03, 2022 at 03:58:20PM +0100, Nicolas Saenz Julienne wrote:
> > > @@ -2935,13 +2947,6 @@ 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)
> > > -
> > 
> > Why was it necessary to move qom_ss and subdir('hw') up? Can
> > event_loop_base be defined down here instead?
> 
> The way I setup it up, qemuutil now depdens on event_loop_base which in turn
> depends on qom. IIUC I can't declare dependencies without declaring first the
> libraries and source sets. All has to happen sequencially. With this in mind,
> almost all libraries depend on libqemuutil so moving it down isn't possible.

I see now. The qemuutil dependency on event_loop_base is introduced in
the next patch so the reason wasn't clear at this point in the patch
series.

Stefan

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 2/4] Introduce event-loop-base abstract class

[Nicolas Saenz Julienne]

On Mon, 2022-03-14 at 13:33 +0000, Stefan Hajnoczi wrote:
> On Fri, Mar 11, 2022 at 11:17:22AM +0100, Nicolas Saenz Julienne wrote:
> > On Thu, 2022-03-10 at 10:25 +0000, Stefan Hajnoczi wrote:
> > > On Thu, Mar 03, 2022 at 03:58:20PM +0100, Nicolas Saenz Julienne wrote:
> > > > @@ -2935,13 +2947,6 @@ 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)
> > > > -
> > > 
> > > Why was it necessary to move qom_ss and subdir('hw') up? Can
> > > event_loop_base be defined down here instead?
> > 
> > The way I setup it up, qemuutil now depdens on event_loop_base which in turn
> > depends on qom. IIUC I can't declare dependencies without declaring first the
> > libraries and source sets. All has to happen sequencially. With this in mind,
> > almost all libraries depend on libqemuutil so moving it down isn't possible.
> 
> I see now. The qemuutil dependency on event_loop_base is introduced in
> the next patch so the reason wasn't clear at this point in the patch
> series.

I'll mention it in the commit message.

-- 
Nicolás Sáenz

Re: [PATCH v2 4/4] util/event-loop-base: Introduce options to set the thread pool size

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 2912 bytes --]

On Fri, Mar 11, 2022 at 11:40:30AM +0100, Nicolas Saenz Julienne wrote:
> On Thu, 2022-03-10 at 10:45 +0000, Stefan Hajnoczi wrote:
> > On Thu, Mar 03, 2022 at 04:13:07PM +0100, Nicolas Saenz Julienne wrote:
> > > @@ -537,10 +546,19 @@
> > >  #                 0 means that the engine will use its default
> > >  #                 (default:0, since 6.1)
> > >  #
> > > +# @thread-pool-min: minimum number of threads readily available in the thread
> > > +#                   pool
> > > +#                   (default:0, since 6.2)
> > > +#
> > > +# @thread-pool-max: maximum number of threads the thread pool can contain
> > > +#                   (default:64, since 6.2)
> > 
> > Here and elsewhere:
> > s/6.2/7.1/
> 
> Yes, forgot to mention it was a placeholder, as I wasn't sure what version to
> target.
> 
> > > @@ -294,6 +314,36 @@ 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);
> > > +    }
> > > +
> > > +    while (pool->cur_threads > pool->max_threads) {
> > > +        qemu_sem_post(&pool->sem);
> > > +        qemu_mutex_unlock(&pool->lock);
> > > +        qemu_cond_wait(&pool->worker_stopped, &pool->lock);
> > > +        qemu_mutex_lock(&pool->lock);
> > 
> > Same question as Patch 1. This looks incorrect because qemu_cond_wait()
> > already drops pool->lock if it needs to block.
> 
> Yes, I'll fix that.
> 
> > Also, why wait? If worker threads are blocked for some reason then our
> > thread will block too.
> 
> Exiting thread_pool_update_params() before honoring the new constraints is a
> source of potential race conditions (having to worry for situations where
> cur_threads > max_threads), and on systems where determinism is important it's
> crucial to have a clear boundary between 'unsafe' and 'safe' states.

On the other hand it creates a reliability problem where a random worker
thread can block all of QEMU. Maybe it's better to let a blocked worker
thread terminate eventually than to hang QEMU?

Stefan

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 4/4] util/event-loop-base: Introduce options to set the thread pool size

[Nicolas Saenz Julienne]

On Mon, 2022-03-14 at 13:35 +0000, Stefan Hajnoczi wrote:
> On Fri, Mar 11, 2022 at 11:40:30AM +0100, Nicolas Saenz Julienne wrote:
> > On Thu, 2022-03-10 at 10:45 +0000, Stefan Hajnoczi wrote:
> > > On Thu, Mar 03, 2022 at 04:13:07PM +0100, Nicolas Saenz Julienne wrote:
> > > > @@ -537,10 +546,19 @@
> > > >  #                 0 means that the engine will use its default
> > > >  #                 (default:0, since 6.1)
> > > >  #
> > > > +# @thread-pool-min: minimum number of threads readily available in the thread
> > > > +#                   pool
> > > > +#                   (default:0, since 6.2)
> > > > +#
> > > > +# @thread-pool-max: maximum number of threads the thread pool can contain
> > > > +#                   (default:64, since 6.2)
> > > 
> > > Here and elsewhere:
> > > s/6.2/7.1/
> > 
> > Yes, forgot to mention it was a placeholder, as I wasn't sure what version to
> > target.
> > 
> > > > @@ -294,6 +314,36 @@ 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);
> > > > +    }
> > > > +
> > > > +    while (pool->cur_threads > pool->max_threads) {
> > > > +        qemu_sem_post(&pool->sem);
> > > > +        qemu_mutex_unlock(&pool->lock);
> > > > +        qemu_cond_wait(&pool->worker_stopped, &pool->lock);
> > > > +        qemu_mutex_lock(&pool->lock);
> > > 
> > > Same question as Patch 1. This looks incorrect because qemu_cond_wait()
> > > already drops pool->lock if it needs to block.
> > 
> > Yes, I'll fix that.
> > 
> > > Also, why wait? If worker threads are blocked for some reason then our
> > > thread will block too.
> > 
> > Exiting thread_pool_update_params() before honoring the new constraints is a
> > source of potential race conditions (having to worry for situations where
> > cur_threads > max_threads), and on systems where determinism is important it's
> > crucial to have a clear boundary between 'unsafe' and 'safe' states.
> 
> On the other hand it creates a reliability problem where a random worker
> thread can block all of QEMU. Maybe it's better to let a blocked worker
> thread terminate eventually than to hang QEMU?

OK, fair enough. I'll switch to that behaviour.

Thanks!

-- 
Nicolás Sáenz