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