[Qemu-devel] Make thread pool implementation modular

[Qemu-devel] Make thread pool implementation modular

[Matthias Brugger]

This patch series makes the thread pool implementation modular.
This allows each drive to use a special implementation.
The patch series prepares qemu to be able to include thread pools different
the one actually implemented. It will allow to implement approaches like 
paravirtualized block requests [1].

 async.c                     |  4 ++--
 block/raw-posix.c           | 15 +++++++++++----
 block/raw-win32.c           |  9 +++++++--
 blockdev.c                  | 13 +++++++++++++
 include/block/aio.h         |  2 +-
 include/block/thread-pool.h |  9 +++++++++
 include/qemu-common.h       |  2 ++
 qemu-options.hx             |  2 +-
 thread-pool.c               | 32 ++++++++++++++++++++++++++++++++
 9 files changed, 78 insertions(+), 10 deletions(-)

[1] http://www.linux-kvm.org/wiki/images/5/53/2012-forum-Brugger-lightningtalk.pdf

[Qemu-devel] [PATCH 1/3] Make thread pool implementation modular

[Matthias Brugger]

This patch introduces function pointers for the thread pool, so that
it's implementation can be set at run-time.

Signed-off-by: Matthias Brugger <matthias.bgg@gmail.com>
---
 include/block/thread-pool.h |  9 +++++++++
 thread-pool.c               | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 41 insertions(+)

diff --git a/include/block/thread-pool.h b/include/block/thread-pool.h
index 32afcdd..53a779f 100644
--- a/include/block/thread-pool.h
+++ b/include/block/thread-pool.h
@@ -38,4 +38,13 @@ int coroutine_fn thread_pool_submit_co(ThreadPool *pool,
         ThreadPoolFunc *func, void *arg);
 void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg);
 
+ThreadPoolFuncArr *thread_pool_probe(void);
+void thread_pool_delete(ThreadPoolFuncArr *tpf);
+
+struct ThreadPoolFuncArr {
+    BlockDriverAIOCB *(*thread_pool_submit_aio)(ThreadPool *pool, ThreadPoolFunc *func, void *arg, BlockDriverCompletionFunc *cb, void *opaque);
+    ThreadPool *(*thread_pool_new)(AioContext *ctx);
+};
+
+
 #endif
diff --git a/thread-pool.c b/thread-pool.c
index 3735fd3..8c5d1a2 100644
--- a/thread-pool.c
+++ b/thread-pool.c
@@ -26,6 +26,7 @@
 #include "qemu/main-loop.h"
 
 static void do_spawn_thread(ThreadPool *pool);
+void thread_pool_aio_free(ThreadPool *pool);
 
 typedef struct ThreadPoolElement ThreadPoolElement;
 
@@ -77,6 +78,7 @@ struct ThreadPool {
     int pending_threads; /* threads created but not running yet */
     int pending_cancellations; /* whether we need a cond_broadcast */
     bool stopping;
+    void (*thread_pool_free)(ThreadPool *pool);
 };
 
 static void *worker_thread(void *opaque)
@@ -300,6 +302,7 @@ static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
     qemu_sem_init(&pool->sem, 0);
     pool->max_threads = 64;
     pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);
+    pool->thread_pool_free = &thread_pool_aio_free;
 
     QLIST_INIT(&pool->head);
     QTAILQ_INIT(&pool->request_list);
@@ -316,6 +319,11 @@ ThreadPool *thread_pool_new(AioContext *ctx)
 
 void thread_pool_free(ThreadPool *pool)
 {
+    pool->thread_pool_free(pool);
+}
+
+void thread_pool_aio_free(ThreadPool *pool)
+{
     if (!pool) {
         return;
     }
@@ -346,3 +354,27 @@ void thread_pool_free(ThreadPool *pool)
     event_notifier_cleanup(&pool->notifier);
     g_free(pool);
 }
+
+ThreadPoolFuncArr *thread_pool_probe(void)
+{
+    ThreadPoolFuncArr *tpf_pool = NULL;
+
+    if (tpf_pool)
+        return tpf_pool;
+
+    tpf_pool = g_new(ThreadPoolFuncArr, 1); //TODO right now, this leaks!
+    if (!tpf_pool) {
+        printf("error allocating thread pool\n");
+        return NULL;
+    }
+
+    tpf_pool->thread_pool_submit_aio = thread_pool_submit_aio;
+    tpf_pool->thread_pool_new = thread_pool_new;
+
+    return tpf_pool;
+}
+
+void thread_pool_delete(ThreadPoolFuncArr *tpf)
+{
+    g_free(tpf);
+}
-- 
1.8.1.2

[Qemu-devel] [PATCH 2/3] Block layer uses modular thread pool

[Matthias Brugger]

With this patch, the calls to the thread pool functions pass through the
new modular thread pool implementation.

Signed-off-by: Matthias Brugger <matthias.bgg@gmail.com>
---
 async.c               |  4 ++--
 block/raw-posix.c     | 15 +++++++++++----
 block/raw-win32.c     |  9 +++++++--
 include/block/aio.h   |  2 +-
 include/qemu-common.h |  2 ++
 5 files changed, 23 insertions(+), 9 deletions(-)

diff --git a/async.c b/async.c
index 5fb3fa6..e66f70f 100644
--- a/async.c
+++ b/async.c
@@ -232,10 +232,10 @@ GSource *aio_get_g_source(AioContext *ctx)
     return &ctx->source;
 }
 
-ThreadPool *aio_get_thread_pool(AioContext *ctx)
+ThreadPool *aio_get_thread_pool(AioContext *ctx, ThreadPoolFuncArr *tpf)
 {
     if (!ctx->thread_pool) {
-        ctx->thread_pool = thread_pool_new(ctx);
+        ctx->thread_pool = tpf->thread_pool_new(ctx);
     }
     return ctx->thread_pool;
 }
diff --git a/block/raw-posix.c b/block/raw-posix.c
index 6f03fbf..22842ed 100644
--- a/block/raw-posix.c
+++ b/block/raw-posix.c
@@ -142,6 +142,7 @@ typedef struct BDRVRawState {
     bool is_xfs : 1;
 #endif
     bool has_discard : 1;
+    ThreadPoolFuncArr *tpf;
 } BDRVRawState;
 
 typedef struct BDRVRawReopenState {
@@ -345,6 +346,9 @@ static int raw_open(BlockDriverState *bs, QDict *options, int flags,
     int ret;
 
     s->type = FTYPE_FILE;
+
+    s->tpf = thread_pool_probe();
+
     ret = raw_open_common(bs, options, flags, 0, &local_err);
     if (error_is_set(&local_err)) {
         error_propagate(errp, local_err);
@@ -792,6 +796,7 @@ static BlockDriverAIOCB *paio_submit(BlockDriverState *bs, int fd,
         int64_t sector_num, QEMUIOVector *qiov, int nb_sectors,
         BlockDriverCompletionFunc *cb, void *opaque, int type)
 {
+    BDRVRawState *s = bs->opaque;
     RawPosixAIOData *acb = g_slice_new(RawPosixAIOData);
     ThreadPool *pool;
 
@@ -807,8 +812,8 @@ static BlockDriverAIOCB *paio_submit(BlockDriverState *bs, int fd,
     acb->aio_offset = sector_num * 512;
 
     trace_paio_submit(acb, opaque, sector_num, nb_sectors, type);
-    pool = aio_get_thread_pool(bdrv_get_aio_context(bs));
-    return thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque);
+    pool = aio_get_thread_pool(bdrv_get_aio_context(bs), s->tpf);
+    return s->tpf->thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque);
 }
 
 static BlockDriverAIOCB *raw_aio_submit(BlockDriverState *bs,
@@ -874,6 +879,8 @@ static void raw_close(BlockDriverState *bs)
         qemu_close(s->fd);
         s->fd = -1;
     }
+
+    thread_pool_delete(s->tpf);
 }
 
 static int raw_truncate(BlockDriverState *bs, int64_t offset)
@@ -1490,8 +1497,8 @@ static BlockDriverAIOCB *hdev_aio_ioctl(BlockDriverState *bs,
     acb->aio_offset = 0;
     acb->aio_ioctl_buf = buf;
     acb->aio_ioctl_cmd = req;
-    pool = aio_get_thread_pool(bdrv_get_aio_context(bs));
-    return thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque);
+    pool = aio_get_thread_pool(bdrv_get_aio_context(bs), s->tpf);
+    return s->tpf->thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque);
 }
 
 #elif defined(__FreeBSD__) || defined(__FreeBSD_kernel__)
diff --git a/block/raw-win32.c b/block/raw-win32.c
index 676b570..dad1255 100644
--- a/block/raw-win32.c
+++ b/block/raw-win32.c
@@ -46,6 +46,7 @@ typedef struct RawWin32AIOData {
     size_t aio_nbytes;
     off64_t aio_offset;
     int aio_type;
+    ThreadPoolFuncArr *tpf;
 } RawWin32AIOData;
 
 typedef struct BDRVRawState {
@@ -159,8 +160,8 @@ static BlockDriverAIOCB *paio_submit(BlockDriverState *bs, HANDLE hfile,
     acb->aio_offset = sector_num * 512;
 
     trace_paio_submit(acb, opaque, sector_num, nb_sectors, type);
-    pool = aio_get_thread_pool(bdrv_get_aio_context(bs));
-    return thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque);
+    pool = aio_get_thread_pool(bdrv_get_aio_context(bs), s->tpf);
+    return s->tpf->thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque);
 }
 
 int qemu_ftruncate64(int fd, int64_t length)
@@ -248,6 +249,8 @@ static int raw_open(BlockDriverState *bs, QDict *options, int flags,
 
     s->type = FTYPE_FILE;
 
+    s->tpf = thread_pool_probe();
+
     opts = qemu_opts_create_nofail(&raw_runtime_opts);
     qemu_opts_absorb_qdict(opts, options, &local_err);
     if (error_is_set(&local_err)) {
@@ -339,6 +342,8 @@ static void raw_close(BlockDriverState *bs)
 {
     BDRVRawState *s = bs->opaque;
     CloseHandle(s->hfile);
+
+    thread_pool_delete(s->tpf);
 }
 
 static int raw_truncate(BlockDriverState *bs, int64_t offset)
diff --git a/include/block/aio.h b/include/block/aio.h
index 2efdf41..22d6fa0 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -231,7 +231,7 @@ void aio_set_event_notifier(AioContext *ctx,
 GSource *aio_get_g_source(AioContext *ctx);
 
 /* Return the ThreadPool bound to this AioContext */
-struct ThreadPool *aio_get_thread_pool(AioContext *ctx);
+struct ThreadPool *aio_get_thread_pool(AioContext *ctx, ThreadPoolFuncArr *tpf);
 
 /* Functions to operate on the main QEMU AioContext.  */
 
diff --git a/include/qemu-common.h b/include/qemu-common.h
index 5054836..594a53b 100644
--- a/include/qemu-common.h
+++ b/include/qemu-common.h
@@ -478,4 +478,6 @@ size_t buffer_find_nonzero_offset(const void *buf, size_t len);
  */
 int parse_debug_env(const char *name, int max, int initial);
 
+typedef struct ThreadPoolFuncArr ThreadPoolFuncArr;
+
 #endif
-- 
1.8.1.2

[Qemu-devel] [PATCH 3/3] Add workerthreads configuration option

[Matthias Brugger]

This patch allows the definition which thread  pool will be used by
every block device. The defintion of the workerthreads option
allows this at the command line level.

At the moment only the thread pool implementation "pool" can be chosen.

Signed-off-by: Matthias Brugger <matthias.bgg@gmail.com>
---
 blockdev.c      | 13 +++++++++++++
 qemu-options.hx |  2 +-
 2 files changed, 14 insertions(+), 1 deletion(-)

diff --git a/blockdev.c b/blockdev.c
index b260477..8b83611 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -388,6 +388,15 @@ static DriveInfo *blockdev_init(QDict *bs_opts,
     }
 #endif
 
+    if ((buf = qemu_opt_get(opts, "workerthreads")) != NULL) {
+        if (!strcmp(buf, "pool")) {
+            /* this is the default */
+        } else {
+            error_report("invalid workerthreads option");
+            return NULL;
+        }
+    }
+
     if ((buf = qemu_opt_get(opts, "format")) != NULL) {
         if (is_help_option(buf)) {
             error_printf("Supported formats:");
@@ -2269,6 +2278,10 @@ QemuOptsList qemu_common_drive_opts = {
             .type = QEMU_OPT_STRING,
             .help = "disk serial number",
         },{
+            .name = "workerthreads",
+            .type = QEMU_OPT_STRING,
+            .help = "type of worker threads (pool)",
+        },{
             .name = "rerror",
             .type = QEMU_OPT_STRING,
             .help = "read error action",
diff --git a/qemu-options.hx b/qemu-options.hx
index 5dc8b75..6f22242 100644
--- a/qemu-options.hx
+++ b/qemu-options.hx
@@ -408,7 +408,7 @@ DEF("drive", HAS_ARG, QEMU_OPTION_drive,
     "       [,cyls=c,heads=h,secs=s[,trans=t]][,snapshot=on|off]\n"
     "       [,cache=writethrough|writeback|none|directsync|unsafe][,format=f]\n"
     "       [,serial=s][,addr=A][,id=name][,aio=threads|native]\n"
-    "       [,readonly=on|off][,copy-on-read=on|off]\n"
+    "       [,workerthreads=pool][,readonly=on|off][,copy-on-read=on|off]\n"
     "       [[,bps=b]|[[,bps_rd=r][,bps_wr=w]]]\n"
     "       [[,iops=i]|[[,iops_rd=r][,iops_wr=w]]]\n"
     "       [[,bps_max=bm]|[[,bps_rd_max=rm][,bps_wr_max=wm]]]\n"
-- 
1.8.1.2

Re: [Qemu-devel] Make thread pool implementation modular

[Stefan Weil]

Am 02.11.2013 00:34, schrieb Matthias Brugger:
> This patch series makes the thread pool implementation modular.
> This allows each drive to use a special implementation.
> The patch series prepares qemu to be able to include thread pools different
> the one actually implemented. It will allow to implement approaches like 
> paravirtualized block requests [1].
>

Please use scripts/checkpatch.pl to check any patch before sending it to
the list.
There are currently errors and warnings for patch 1 and patch 3 which
should be
fixed.

In patch 3 you can also fix "defintion" in the patch description.

Regards,
Stefan