[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Qemu-devel] [PULL v2 09/59] thread-pool: Convert thread_pool_aiocb_info
From: |
Stefan Hajnoczi |
Subject: |
[Qemu-devel] [PULL v2 09/59] thread-pool: Convert thread_pool_aiocb_info.cancel to cancel_async |
Date: |
Mon, 22 Sep 2014 12:41:40 +0100 |
From: Fam Zheng <address@hidden>
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 <address@hidden>
Signed-off-by: Stefan Hajnoczi <address@hidden>
---
tests/test-thread-pool.c | 34 ++++++++++++++++++++++++++--------
thread-pool.c | 32 ++++++++++++++------------------
2 files changed, 40 insertions(+), 26 deletions(-)
diff --git a/tests/test-thread-pool.c b/tests/test-thread-pool.c
index f40b7fc..ed2b25b 100644
--- a/tests/test-thread-pool.c
+++ b/tests/test-thread-pool.c
@@ -33,7 +33,7 @@ static int long_cb(void *opaque)
static void done_cb(void *opaque, int ret)
{
WorkerTestData *data = opaque;
- g_assert_cmpint(data->ret, ==, -EINPROGRESS);
+ g_assert(data->ret == -EINPROGRESS || data->ret == -ECANCELED);
data->ret = ret;
data->aiocb = NULL;
@@ -132,7 +132,7 @@ static void test_submit_many(void)
}
}
-static void test_cancel(void)
+static void do_test_cancel(bool sync)
{
WorkerTestData data[100];
int num_canceled;
@@ -170,18 +170,25 @@ static void test_cancel(void)
for (i = 0; i < 100; i++) {
if (atomic_cmpxchg(&data[i].n, 0, 3) == 0) {
data[i].ret = -ECANCELED;
- bdrv_aio_cancel(data[i].aiocb);
- active--;
+ if (sync) {
+ bdrv_aio_cancel(data[i].aiocb);
+ } else {
+ bdrv_aio_cancel_async(data[i].aiocb);
+ }
num_canceled++;
}
}
g_assert_cmpint(active, >, 0);
g_assert_cmpint(num_canceled, <, 100);
- /* Canceling the others will be a blocking operation. */
for (i = 0; i < 100; i++) {
if (data[i].aiocb && data[i].n != 3) {
- bdrv_aio_cancel(data[i].aiocb);
+ if (sync) {
+ /* Canceling the others will be a blocking operation. */
+ bdrv_aio_cancel(data[i].aiocb);
+ } else {
+ bdrv_aio_cancel_async(data[i].aiocb);
+ }
}
}
@@ -193,15 +200,25 @@ static void test_cancel(void)
for (i = 0; i < 100; i++) {
if (data[i].n == 3) {
g_assert_cmpint(data[i].ret, ==, -ECANCELED);
- g_assert(data[i].aiocb != NULL);
+ g_assert(data[i].aiocb == NULL);
} else {
g_assert_cmpint(data[i].n, ==, 2);
- g_assert_cmpint(data[i].ret, ==, 0);
+ g_assert(data[i].ret == 0 || data[i].ret == -ECANCELED);
g_assert(data[i].aiocb == NULL);
}
}
}
+static void test_cancel(void)
+{
+ do_test_cancel(true);
+}
+
+static void test_cancel_async(void)
+{
+ do_test_cancel(false);
+}
+
int main(int argc, char **argv)
{
int ret;
@@ -217,6 +234,7 @@ int main(int argc, char **argv)
g_test_add_func("/thread-pool/submit-co", test_submit_co);
g_test_add_func("/thread-pool/submit-many", test_submit_many);
g_test_add_func("/thread-pool/cancel", test_cancel);
+ g_test_add_func("/thread-pool/cancel-async", test_cancel_async);
ret = g_test_run();
diff --git a/thread-pool.c b/thread-pool.c
index bc07d7a..bb1395c 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);
--
1.9.3
- [Qemu-devel] [PULL v2 00/59] Block patches, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 01/59] block/vhdx.c: Mark parent_vhdx_guid variable as unused, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 02/59] ide/atapi: Mark non-data commands as complete, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 04/59] ide/ahci: Check for -ECANCELED in aio callbacks, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 05/59] block: Add refcnt in BlockDriverAIOCB, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 03/59] aio-win32: fix uninitialized use of have_select_revents, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 06/59] block: Add bdrv_aio_cancel_async, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 07/59] block: Drop bdrv_em_co_aiocb_info.cancel, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 08/59] block: Drop bdrv_em_aiocb_info.cancel, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 09/59] thread-pool: Convert thread_pool_aiocb_info.cancel to cancel_async,
Stefan Hajnoczi <=
- [Qemu-devel] [PULL v2 10/59] linux-aio: Convert laio_aiocb_info.cancel to .cancel_async, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 11/59] dma: Convert dma_aiocb_info.cancel to .cancel_async, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 13/59] archipelago: Drop archipelago_aiocb_info.cancel, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 14/59] blkdebug: Drop blkdebug_aiocb_info.cancel, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 12/59] iscsi: Convert iscsi_aiocb_info.cancel to .cancel_async, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 16/59] curl: Drop curl_aiocb_info.cancel, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 15/59] blkverify: Drop blkverify_aiocb_info.cancel, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 17/59] qed: Drop qed_aiocb_info.cancel, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 19/59] quorum: Convert quorum_aiocb_info.cancel to .cancel_async, Stefan Hajnoczi, 2014/09/22
- [Qemu-devel] [PULL v2 21/59] sheepdog: Convert sd_aiocb_info.cancel to .cancel_async, Stefan Hajnoczi, 2014/09/22