aboutsummaryrefslogtreecommitdiff
path: root/thread-pool.c
diff options
context:
space:
mode:
authorStefan Hajnoczi <stefanha@redhat.com>2013-03-07 13:41:46 +0100
committerStefan Hajnoczi <stefanha@redhat.com>2013-03-15 16:07:50 +0100
commitf7311ccc630d925e7351e9440b7ad8bc6f0a51de (patch)
tree7d0acbba48230b50a45cf73531e6c47920dff1ae /thread-pool.c
parentb811203cf2fbf83e26f8e8feb2c77784259a4cbd (diff)
threadpool: add thread_pool_new() and thread_pool_free()
ThreadPool is tied to an AioContext through its event notifier, which dictates in which AioContext the work item's callback function will be invoked. In order to support multiple AioContexts we need to support multiple ThreadPool instances. This patch adds the new/free functions. The free function deserves special attention because it quiesces remaining worker threads. This requires a new condition variable and a "stopping" flag to let workers know they should terminate once idle. We never needed to do this before since the global threadpool was not explicitly destroyed until process termination. Also stash the AioContext pointer in ThreadPool so that we can call aio_set_event_notifier() in thread_pool_free(). We didn't need to hold onto AioContext previously since there was no free function. Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com> Reviewed-by: Paolo Bonzini <pbonzini@redhat.com>
Diffstat (limited to 'thread-pool.c')
-rw-r--r--thread-pool.c52
1 files changed, 48 insertions, 4 deletions
diff --git a/thread-pool.c b/thread-pool.c
index a0aecd08fe..d1e4570829 100644
--- a/thread-pool.c
+++ b/thread-pool.c
@@ -24,8 +24,6 @@
#include "qemu/event_notifier.h"
#include "block/thread-pool.h"
-typedef struct ThreadPool ThreadPool;
-
static void do_spawn_thread(ThreadPool *pool);
typedef struct ThreadPoolElement ThreadPoolElement;
@@ -59,8 +57,10 @@ struct ThreadPoolElement {
struct ThreadPool {
EventNotifier notifier;
+ AioContext *ctx;
QemuMutex lock;
QemuCond check_cancel;
+ QemuCond worker_stopped;
QemuSemaphore sem;
int max_threads;
QEMUBH *new_thread_bh;
@@ -75,6 +75,7 @@ struct ThreadPool {
int new_threads; /* backlog of threads we need to create */
int pending_threads; /* threads created but not running yet */
int pending_cancellations; /* whether we need a cond_broadcast */
+ bool stopping;
};
/* Currently there is only one thread pool instance. */
@@ -88,7 +89,7 @@ static void *worker_thread(void *opaque)
pool->pending_threads--;
do_spawn_thread(pool);
- while (1) {
+ while (!pool->stopping) {
ThreadPoolElement *req;
int ret;
@@ -99,7 +100,7 @@ static void *worker_thread(void *opaque)
qemu_mutex_lock(&pool->lock);
pool->idle_threads--;
} while (ret == -1 && !QTAILQ_EMPTY(&pool->request_list));
- if (ret == -1) {
+ if (ret == -1 || pool->stopping) {
break;
}
@@ -124,6 +125,7 @@ static void *worker_thread(void *opaque)
}
pool->cur_threads--;
+ qemu_cond_signal(&pool->worker_stopped);
qemu_mutex_unlock(&pool->lock);
return NULL;
}
@@ -298,8 +300,10 @@ static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
memset(pool, 0, sizeof(*pool));
event_notifier_init(&pool->notifier, false);
+ pool->ctx = ctx;
qemu_mutex_init(&pool->lock);
qemu_cond_init(&pool->check_cancel);
+ 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);
@@ -311,6 +315,46 @@ static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
thread_pool_active);
}
+ThreadPool *thread_pool_new(AioContext *ctx)
+{
+ ThreadPool *pool = g_new(ThreadPool, 1);
+ thread_pool_init_one(pool, ctx);
+ return pool;
+}
+
+void thread_pool_free(ThreadPool *pool)
+{
+ if (!pool) {
+ return;
+ }
+
+ assert(QLIST_EMPTY(&pool->head));
+
+ qemu_mutex_lock(&pool->lock);
+
+ /* Stop new threads from spawning */
+ qemu_bh_delete(pool->new_thread_bh);
+ pool->cur_threads -= pool->new_threads;
+ pool->new_threads = 0;
+
+ /* Wait for worker threads to terminate */
+ pool->stopping = true;
+ while (pool->cur_threads > 0) {
+ qemu_sem_post(&pool->sem);
+ qemu_cond_wait(&pool->worker_stopped, &pool->lock);
+ }
+
+ qemu_mutex_unlock(&pool->lock);
+
+ aio_set_event_notifier(pool->ctx, &pool->notifier, NULL, NULL);
+ qemu_sem_destroy(&pool->sem);
+ qemu_cond_destroy(&pool->check_cancel);
+ qemu_cond_destroy(&pool->worker_stopped);
+ qemu_mutex_destroy(&pool->lock);
+ event_notifier_cleanup(&pool->notifier);
+ g_free(pool);
+}
+
static void thread_pool_init(void)
{
thread_pool_init_one(&global_pool, NULL);