@@ -1136,7 +1136,9 @@ static BlockAIOCB *paio_submit(BlockDriverState *bs, int fd,
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);
+ return thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque,
+ qiov ? qiov->replay : false,
+ qiov ? qiov->replay_step : 0);
}
static BlockAIOCB *raw_aio_submit(BlockDriverState *bs,
@@ -2062,7 +2064,7 @@ static BlockAIOCB *hdev_aio_ioctl(BlockDriverState *bs,
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);
+ return thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque, false, 0);
}
#elif defined(__FreeBSD__) || defined(__FreeBSD_kernel__)
@@ -158,7 +158,9 @@ static BlockAIOCB *paio_submit(BlockDriverState *bs, HANDLE hfile,
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);
+ return thread_pool_submit_aio(pool, aio_worker, acb, cb, opaque,
+ qiov ? qiov->replay : false,
+ qiov ? qiov->replay_step : 0);
}
int qemu_ftruncate64(int fd, int64_t length)
@@ -29,9 +29,11 @@ void thread_pool_free(ThreadPool *pool);
BlockAIOCB *thread_pool_submit_aio(ThreadPool *pool,
ThreadPoolFunc *func, void *arg,
- BlockCompletionFunc *cb, void *opaque);
+ BlockCompletionFunc *cb, void *opaque,
+ bool replay, uint64_t replay_step);
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_work(ThreadPool *pool, void *r);
#endif
@@ -13,6 +13,7 @@
#include "qemu/error-report.h"
#include "replay.h"
#include "replay-internal.h"
+#include "block/thread-pool.h"
typedef struct Event {
ReplayAsyncEventKind event_kind;
@@ -38,6 +39,9 @@ static void replay_run_event(Event *event)
case REPLAY_ASYNC_EVENT_BH:
aio_bh_call(event->opaque);
break;
+ case REPLAY_ASYNC_EVENT_THREAD:
+ thread_pool_work((ThreadPool *)event->opaque, event->opaque2);
+ break;
default:
error_report("Replay: invalid async event ID (%d) in the queue",
event->event_kind);
@@ -137,6 +141,7 @@ static void replay_save_event(Event *event, int checkpoint)
/* save event-specific data */
switch (event->event_kind) {
case REPLAY_ASYNC_EVENT_BH:
+ case REPLAY_ASYNC_EVENT_THREAD:
replay_put_qword(event->id);
break;
}
@@ -148,6 +153,11 @@ void replay_add_bh_event(void *bh, uint64_t id)
replay_add_event_internal(REPLAY_ASYNC_EVENT_BH, bh, NULL, id);
}
+void replay_add_thread_event(void *opaque, void *opaque2, uint64_t id)
+{
+ replay_add_event_internal(REPLAY_ASYNC_EVENT_THREAD, opaque, opaque2, id);
+}
+
/* Called with replay mutex locked */
void replay_save_events(int checkpoint)
{
@@ -180,6 +190,7 @@ static Event *replay_read_event(int checkpoint)
/* Events that has not to be in the queue */
switch (read_event_kind) {
case REPLAY_ASYNC_EVENT_BH:
+ case REPLAY_ASYNC_EVENT_THREAD:
if (read_id == -1) {
read_id = replay_get_qword();
}
@@ -40,6 +40,7 @@ enum ReplayEvents {
enum ReplayAsyncEventKind {
REPLAY_ASYNC_EVENT_BH,
+ REPLAY_ASYNC_EVENT_THREAD,
REPLAY_ASYNC_COUNT
};
@@ -97,5 +97,7 @@ void replay_disable_events(void);
bool replay_events_enabled(void);
/*! Adds BH event to the queue */
void replay_add_bh_event(void *bh, uint64_t id);
+/*! Adds thread event to the queue */
+void replay_add_thread_event(void *pool, void *req, uint64_t id);
#endif
@@ -30,3 +30,7 @@ uint64_t replay_get_current_step(void)
{
return 0;
}
+
+void replay_add_thread_event(void *opaque, void *opaque2, uint64_t id)
+{
+}
@@ -56,7 +56,7 @@ static void test_submit_aio(void)
{
WorkerTestData data = { .n = 0, .ret = -EINPROGRESS };
data.aiocb = thread_pool_submit_aio(pool, worker_cb, &data,
- done_cb, &data);
+ done_cb, &data, false, 0);
/* The callbacks are not called until after the first wait. */
active = 1;
@@ -120,7 +120,8 @@ static void test_submit_many(void)
for (i = 0; i < 100; i++) {
data[i].n = 0;
data[i].ret = -EINPROGRESS;
- thread_pool_submit_aio(pool, worker_cb, &data[i], done_cb, &data[i]);
+ thread_pool_submit_aio(pool, worker_cb, &data[i], done_cb, &data[i],
+ false, 0);
}
active = 100;
@@ -149,7 +150,7 @@ static void do_test_cancel(bool sync)
data[i].n = 0;
data[i].ret = -EINPROGRESS;
data[i].aiocb = thread_pool_submit_aio(pool, long_cb, &data[i],
- done_cb, &data[i]);
+ done_cb, &data[i], false, 0);
}
/* Starting the threads may be left to a bottom half. Let it
@@ -22,6 +22,7 @@
#include "trace.h"
#include "block/thread-pool.h"
#include "qemu/main-loop.h"
+#include "replay/replay.h"
static void do_spawn_thread(ThreadPool *pool);
@@ -74,6 +75,27 @@ struct ThreadPool {
bool stopping;
};
+void thread_pool_work(ThreadPool *pool, void *r)
+{
+ ThreadPoolElement *req = (ThreadPoolElement *)r;
+ int ret;
+ if (replay_mode == REPLAY_MODE_NONE) {
+ qemu_mutex_unlock(&pool->lock);
+ }
+
+ ret = req->func(req->arg);
+ req->ret = ret;
+ /* Write ret before state. */
+ smp_wmb();
+ req->state = THREAD_DONE;
+
+ if (replay_mode == REPLAY_MODE_NONE) {
+ qemu_mutex_lock(&pool->lock);
+ }
+
+ qemu_bh_schedule(pool->completion_bh);
+}
+
static void *worker_thread(void *opaque)
{
ThreadPool *pool = opaque;
@@ -100,18 +122,12 @@ static void *worker_thread(void *opaque)
req = QTAILQ_FIRST(&pool->request_list);
QTAILQ_REMOVE(&pool->request_list, req, reqs);
req->state = THREAD_ACTIVE;
- qemu_mutex_unlock(&pool->lock);
-
- ret = req->func(req->arg);
-
- req->ret = ret;
- /* Write ret before state. */
- smp_wmb();
- req->state = THREAD_DONE;
-
- qemu_mutex_lock(&pool->lock);
- qemu_bh_schedule(pool->completion_bh);
+ if (replay_mode != REPLAY_MODE_NONE && req->common.replay) {
+ replay_add_thread_event(pool, req, req->common.replay_step);
+ } else {
+ thread_pool_work(pool, req);
+ }
}
pool->cur_threads--;
@@ -235,7 +251,8 @@ static const AIOCBInfo thread_pool_aiocb_info = {
BlockAIOCB *thread_pool_submit_aio(ThreadPool *pool,
ThreadPoolFunc *func, void *arg,
- BlockCompletionFunc *cb, void *opaque)
+ BlockCompletionFunc *cb, void *opaque,
+ bool replay, uint64_t replay_step)
{
ThreadPoolElement *req;
@@ -244,6 +261,8 @@ BlockAIOCB *thread_pool_submit_aio(ThreadPool *pool,
req->arg = arg;
req->state = THREAD_QUEUED;
req->pool = pool;
+ req->common.replay = replay;
+ req->common.replay_step = replay_step;
QLIST_INSERT_HEAD(&pool->head, req, all);
@@ -254,8 +273,8 @@ BlockAIOCB *thread_pool_submit_aio(ThreadPool *pool,
spawn_thread(pool);
}
QTAILQ_INSERT_TAIL(&pool->request_list, req, reqs);
- qemu_mutex_unlock(&pool->lock);
qemu_sem_post(&pool->sem);
+ qemu_mutex_unlock(&pool->lock);
return &req->common;
}
@@ -277,14 +296,14 @@ int coroutine_fn thread_pool_submit_co(ThreadPool *pool, ThreadPoolFunc *func,
{
ThreadPoolCo tpc = { .co = qemu_coroutine_self(), .ret = -EINPROGRESS };
assert(qemu_in_coroutine());
- thread_pool_submit_aio(pool, func, arg, thread_pool_co_cb, &tpc);
+ thread_pool_submit_aio(pool, func, arg, thread_pool_co_cb, &tpc, false, 0);
qemu_coroutine_yield();
return tpc.ret;
}
void thread_pool_submit(ThreadPool *pool, ThreadPoolFunc *func, void *arg)
{
- thread_pool_submit_aio(pool, func, arg, NULL, NULL);
+ thread_pool_submit_aio(pool, func, arg, NULL, NULL, false, 0);
}
static void thread_pool_init_one(ThreadPool *pool, AioContext *ctx)
This patch modifies thread pool to allow replaying asynchronous thread tasks synchronously in replay mode. Signed-off-by: Pavel Dovgalyuk <pavel.dovgaluk@ispras.ru> --- block/raw-posix.c | 6 ++++- block/raw-win32.c | 4 +++- include/block/thread-pool.h | 4 +++- replay/replay-events.c | 11 ++++++++++ replay/replay-internal.h | 1 + replay/replay.h | 2 ++ stubs/replay.c | 4 ++++ tests/test-thread-pool.c | 7 ++++-- thread-pool.c | 49 ++++++++++++++++++++++++++++++------------- 9 files changed, 66 insertions(+), 22 deletions(-)