aboutsummaryrefslogtreecommitdiff
path: root/thread-pool.c
diff options
context:
space:
mode:
authorFam Zheng <famz@redhat.com>2014-09-11 13:41:12 +0800
committerStefan Hajnoczi <stefanha@redhat.com>2014-09-22 11:39:02 +0100
commit3391f5e51c4b55bf72fd33d9bc542a49114e48cd (patch)
tree96f6d6f6977058d8f05dbaa861abaab3a341db05 /thread-pool.c
parentf600ac19027e67dd73f1adad50a67a2cdb4f3218 (diff)
downloadqemu-arm-3391f5e51c4b55bf72fd33d9bc542a49114e48cd.tar.gz
thread-pool: Convert thread_pool_aiocb_info.cancel to cancel_async
The .cancel_async shares the same the first half with .cancel: try to steal the request if not submitted yet. In this case set the elem to THREAD_DONE status and ret to -ECANCELED, which means thread_pool_completion_bh will call the cb with -ECANCELED. If the request is already submitted, do nothing, as we know the normal completion will happen in the future. Testing code update: Before, done_cb is only called if the request is already submitted by thread pool. Now done_cb is always called, even before it is submitted, because we emulate bdrv_aio_cancel with bdrv_aio_cancel_async. So also update the test criteria accordingly. Signed-off-by: Fam Zheng <famz@redhat.com> Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
Diffstat (limited to 'thread-pool.c')
-rw-r--r--thread-pool.c32
1 files changed, 14 insertions, 18 deletions
diff --git a/thread-pool.c b/thread-pool.c
index bc07d7a1c9..bb1395cff8 100644
--- a/thread-pool.c
+++ b/thread-pool.c
@@ -31,7 +31,6 @@ enum ThreadState {
THREAD_QUEUED,
THREAD_ACTIVE,
THREAD_DONE,
- THREAD_CANCELED,
};
struct ThreadPoolElement {
@@ -58,7 +57,6 @@ struct ThreadPool {
AioContext *ctx;
QEMUBH *completion_bh;
QemuMutex lock;
- QemuCond check_cancel;
QemuCond worker_stopped;
QemuSemaphore sem;
int max_threads;
@@ -73,7 +71,6 @@ struct ThreadPool {
int idle_threads;
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;
};
@@ -113,9 +110,6 @@ static void *worker_thread(void *opaque)
req->state = THREAD_DONE;
qemu_mutex_lock(&pool->lock);
- if (pool->pending_cancellations) {
- qemu_cond_broadcast(&pool->check_cancel);
- }
qemu_bh_schedule(pool->completion_bh);
}
@@ -173,7 +167,7 @@ static void thread_pool_completion_bh(void *opaque)
restart:
QLIST_FOREACH_SAFE(elem, &pool->head, all, next) {
- if (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
+ if (elem->state != THREAD_DONE) {
continue;
}
if (elem->state == THREAD_DONE) {
@@ -217,22 +211,26 @@ static void thread_pool_cancel(BlockDriverAIOCB *acb)
*/
qemu_sem_timedwait(&pool->sem, 0) == 0) {
QTAILQ_REMOVE(&pool->request_list, elem, reqs);
- elem->state = THREAD_CANCELED;
qemu_bh_schedule(pool->completion_bh);
- } else {
- pool->pending_cancellations++;
- while (elem->state != THREAD_CANCELED && elem->state != THREAD_DONE) {
- qemu_cond_wait(&pool->check_cancel, &pool->lock);
- }
- pool->pending_cancellations--;
+
+ elem->state = THREAD_DONE;
+ elem->ret = -ECANCELED;
}
+
qemu_mutex_unlock(&pool->lock);
- thread_pool_completion_bh(pool);
+}
+
+static AioContext *thread_pool_get_aio_context(BlockDriverAIOCB *acb)
+{
+ ThreadPoolElement *elem = (ThreadPoolElement *)acb;
+ ThreadPool *pool = elem->pool;
+ return pool->ctx;
}
static const AIOCBInfo thread_pool_aiocb_info = {
.aiocb_size = sizeof(ThreadPoolElement),
- .cancel = thread_pool_cancel,
+ .cancel_async = thread_pool_cancel,
+ .get_aio_context = thread_pool_get_aio_context,
};
BlockDriverAIOCB *thread_pool_submit_aio(ThreadPool *pool,
@@ -299,7 +297,6 @@ static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
pool->ctx = ctx;
pool->completion_bh = aio_bh_new(ctx, thread_pool_completion_bh, pool);
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;
@@ -342,7 +339,6 @@ void thread_pool_free(ThreadPool *pool)
qemu_bh_delete(pool->completion_bh);
qemu_sem_destroy(&pool->sem);
- qemu_cond_destroy(&pool->check_cancel);
qemu_cond_destroy(&pool->worker_stopped);
qemu_mutex_destroy(&pool->lock);
g_free(pool);