izzy/xemu
0
0
Fork 0
mirror of https://github.com/izzy2lost/xemu.git synced 2026-07-06 00:20:22 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge tag 'for-upstream' of https://repo.or.cz/qemu/kevin into staging

Browse Source 
Block layer patches

- Multi-threading fixes in several block drivers

# -----BEGIN PGP SIGNATURE-----
#
# iQJFBAABCgAvFiEE3D3rFZqa+V09dFb+fwmycsiPL9YFAmkcpowRHGt3b2xmQHJl
# ZGhhdC5jb20ACgkQfwmycsiPL9bnrA/9HpvuLovahyZH+zke9FAzE9EcZ8eLgT4t
# JM5ijkrfZ8KoyvW9zmUiT/T/BhewFY4if6FqIUvQoCVIQAN+Y3Z8Us/WcZpb3xG7
# EMUr/CeiRl7Ka0SGbxZWR6H10Enuq4hoesfkRPXGV33CKuef09HvgE0184aazZLs
# bfeBn562zUcYzn/HqNFcPqLh6goeHWhsnQuxfJJeu7i05HofeM/3T7WUcCmxhn3V
# 4Pk6o0toYONuZpfUZityieEI54ID95gVmllaE8QT5f0rBXHp6ae5P6x+fnqbj306
# DA5RKBxM5nJ5Md2mo2tJtJ0eUNzbxXpAnT+wMkcDO4OQq8F6KngOTokPPKv13z1f
# uUSwZW5QXV8pvU6AZouUlPStz0MFtfXSVqLZaV51BB7MrIGF79ina9vwMYoteTNJ
# CGi32Ca0dl3PjLm//avmZiHYAeczpK34nPImRLZnhh78zQal/yDpdpi5No/tm4Yf
# /0OAES0legGC3hTsuDh9sXqkmRn3qNvUd40IuDRyvrNYxMUuF+2IQymD5Hcm/6CO
# uIf+/61OPHLHwKFYHyyK44XmcYQHCE0LxilmWVrNaULs4a3nvlRz3GTZ09R/ebAl
# ddtedo9eHN6KfViTa95YWuu53iXzQRPUp3CrSC1gPB04zphpYsCy7eJ/t3kJqCpP
# 9CVf3PMOmZA=
# =FNy0
# -----END PGP SIGNATURE-----
# gpg: Signature made Tue 18 Nov 2025 06:02:04 PM CET
# gpg:                using RSA key DC3DEB159A9AF95D3D7456FE7F09B272C88F2FD6
# gpg:                issuer "kwolf@redhat.com"
# gpg: Good signature from "Kevin Wolf <kwolf@redhat.com>" [unknown]
# gpg: WARNING: The key's User ID is not certified with a trusted signature!
# gpg:          There is no indication that the signature belongs to the owner.
# Primary key fingerprint: DC3D EB15 9A9A F95D 3D74  56FE 7F09 B272 C88F 2FD6

* tag 'for-upstream' of https://repo.or.cz/qemu/kevin:
  win32-aio: Run CB in original context
  null-aio: Run CB in original AioContext
  iscsi: Create AIO BH in original AioContext
  block: Note in which AioContext AIO CBs are called
  blkreplay: Run BH in coroutine’s AioContext
  ssh: Run restart_coroutine in current AioContext
  qcow2: Schedule cache-clean-timer in realtime
  qcow2: Fix cache_clean_timer
  qcow2: Re-initialize lock in invalidate_cache
  block/io: Take reqs_lock for tracked_requests
  nvme: Note in which AioContext some functions run
  nvme: Fix coroutine waking
  nvme: Kick and check completions in BDS context
  gluster: Do not move coroutine into BDS context
  curl: Fix coroutine waking
  nfs: Run co BH CB in the coroutine’s AioContext
  iscsi: Run co BH CB in the coroutine’s AioContext
  rbd: Run co BH CB in the coroutine’s AioContext
  block: Note on aio_co_wake use if not yet yielding

Signed-off-by: Richard Henderson <richard.henderson@linaro.org>
This commit is contained in:
Richard Henderson
2025-11-19 07:38:13 +01:00
parent 711a1ddf89 837c04e9fc
commit 6327540d92
15 changed files with 346 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
block/blkreplay.c
+2 -1
View File
@@ -63,9 +63,10 @@ static void block_request_create(uint64_t reqid, BlockDriverState *bs,
Coroutine *co)
{
Request *req = g_new(Request, 1);
AioContext *ctx = qemu_coroutine_get_aio_context(co);
*req = (Request) {
.co = co,
.bh = aio_bh_new(bdrv_get_aio_context(bs), blkreplay_bh_cb, req),
.bh = aio_bh_new(ctx, blkreplay_bh_cb, req),
};
replay_block_event(req->bh, reqid);
}
block/curl.c
+31 -14
View File
@@ -258,8 +258,8 @@ read_end:
}
/* Called with s->mutex held. */
static bool curl_find_buf(BDRVCURLState *s, uint64_t start, uint64_t len,
CURLAIOCB *acb)
static bool coroutine_fn
curl_find_buf(BDRVCURLState *s, uint64_t start, uint64_t len, CURLAIOCB *acb)
{
int i;
uint64_t end = start + len;
@@ -307,6 +307,10 @@ static bool curl_find_buf(BDRVCURLState *s, uint64_t start, uint64_t len,
for (j=0; j<CURL_NUM_ACB; j++) {
if (!state->acb[j]) {
state->acb[j] = acb;
/* Await ongoing request */
qemu_mutex_unlock(&s->mutex);
qemu_coroutine_yield();
qemu_mutex_lock(&s->mutex);
return true;
}
}
@@ -378,6 +382,16 @@ static void curl_multi_check_completion(BDRVCURLState *s)
acb->ret = error ? -EIO : 0;
state->acb[i] = NULL;
qemu_mutex_unlock(&s->mutex);
/*
* Current AioContext is the BDS context, which may or may not
* be the request (coroutine) context.
* - If it is, the coroutine must have yielded or the FD handler
* (curl_multi_do()/curl_multi_timeout_do()) could not have
* been called and we would not be here
* - If it is not, it doesn't matter whether it has already
* yielded or not; it will be scheduled once it does yield
* So aio_co_wake() is safe to call.
*/
aio_co_wake(acb->co);
qemu_mutex_lock(&s->mutex);
}
@@ -868,7 +882,7 @@ out_noclean:
return -EINVAL;
}
static void coroutine_fn curl_setup_preadv(BlockDriverState *bs, CURLAIOCB *acb)
static void coroutine_fn curl_do_preadv(BlockDriverState *bs, CURLAIOCB *acb)
{
CURLState *state;
int running;
@@ -880,10 +894,13 @@ static void coroutine_fn curl_setup_preadv(BlockDriverState *bs, CURLAIOCB *acb)
qemu_mutex_lock(&s->mutex);
// In case we have the requested data already (e.g. read-ahead),
// we can just call the callback and be done.
/*
* In case we have the requested data already (e.g. read-ahead),
* we can just call the callback and be done. This may have to
* await an ongoing request, in which case it itself will yield.
*/
if (curl_find_buf(s, start, acb->bytes, acb)) {
goto out;
goto dont_yield;
}
// No cache found, so let's start a new request
@@ -898,7 +915,7 @@ static void coroutine_fn curl_setup_preadv(BlockDriverState *bs, CURLAIOCB *acb)
if (curl_init_state(s, state) < 0) {
curl_clean_state(state);
acb->ret = -EIO;
goto out;
goto dont_yield;
}
acb->start = 0;
@@ -913,7 +930,7 @@ static void coroutine_fn curl_setup_preadv(BlockDriverState *bs, CURLAIOCB *acb)
if (state->buf_len && state->orig_buf == NULL) {
curl_clean_state(state);
acb->ret = -ENOMEM;
goto out;
goto dont_yield;
}
state->acb[0] = acb;
@@ -925,13 +942,16 @@ static void coroutine_fn curl_setup_preadv(BlockDriverState *bs, CURLAIOCB *acb)
acb->ret = -EIO;
curl_clean_state(state);
goto out;
goto dont_yield;
}
/* Tell curl it needs to kick things off */
curl_multi_socket_action(s->multi, CURL_SOCKET_TIMEOUT, 0, &running);
qemu_mutex_unlock(&s->mutex);
qemu_coroutine_yield();
return;
out:
dont_yield:
qemu_mutex_unlock(&s->mutex);
}
@@ -947,10 +967,7 @@ static int coroutine_fn curl_co_preadv(BlockDriverState *bs,
.bytes = bytes
};
curl_setup_preadv(bs, &acb);
while (acb.ret == -EINPROGRESS) {
qemu_coroutine_yield();
}
curl_do_preadv(bs, &acb);
return acb.ret;
}
block/gluster.c
+11 -6
View File
@@ -56,7 +56,6 @@ typedef struct GlusterAIOCB {
int64_t size;
int ret;
Coroutine *coroutine;
AioContext *aio_context;
} GlusterAIOCB;
typedef struct BDRVGlusterState {
@@ -743,7 +742,17 @@ static void gluster_finish_aiocb(struct glfs_fd *fd, ssize_t ret,
acb->ret = -EIO; /* Partial read/write - fail it */
}
aio_co_schedule(acb->aio_context, acb->coroutine);
/*
* Safe to call: The coroutine will yield exactly once awaiting this
* scheduling, and the context is its own context, so it will be scheduled
* once it does yield.
*
* (aio_co_wake() would call qemu_get_current_aio_context() to check whether
* we are in the same context, but we are not in a qemu thread, so we cannot
* do that. Use aio_co_schedule() directly.)
*/
aio_co_schedule(qemu_coroutine_get_aio_context(acb->coroutine),
acb->coroutine);
}
static void qemu_gluster_parse_flags(int bdrv_flags, int *open_flags)
@@ -1006,7 +1015,6 @@ static coroutine_fn int qemu_gluster_co_pwrite_zeroes(BlockDriverState *bs,
acb.size = bytes;
acb.ret = 0;
acb.coroutine = qemu_coroutine_self();
acb.aio_context = bdrv_get_aio_context(bs);
ret = glfs_zerofill_async(s->fd, offset, bytes, gluster_finish_aiocb, &acb);
if (ret < 0) {
@@ -1184,7 +1192,6 @@ static coroutine_fn int qemu_gluster_co_rw(BlockDriverState *bs,
acb.size = size;
acb.ret = 0;
acb.coroutine = qemu_coroutine_self();
acb.aio_context = bdrv_get_aio_context(bs);
if (write) {
ret = glfs_pwritev_async(s->fd, qiov->iov, qiov->niov, offset, 0,
@@ -1251,7 +1258,6 @@ static coroutine_fn int qemu_gluster_co_flush_to_disk(BlockDriverState *bs)
acb.size = 0;
acb.ret = 0;
acb.coroutine = qemu_coroutine_self();
acb.aio_context = bdrv_get_aio_context(bs);
ret = glfs_fsync_async(s->fd, gluster_finish_aiocb, &acb);
if (ret < 0) {
@@ -1299,7 +1305,6 @@ static coroutine_fn int qemu_gluster_co_pdiscard(BlockDriverState *bs,
acb.size = 0;
acb.ret = 0;
acb.coroutine = qemu_coroutine_self();
acb.aio_context = bdrv_get_aio_context(bs);
ret = glfs_discard_async(s->fd, offset, bytes, gluster_finish_aiocb, &acb);
if (ret < 0) {
block/io.c
+3
View File
@@ -718,11 +718,14 @@ BdrvTrackedRequest *coroutine_fn bdrv_co_get_self_request(BlockDriverState *bs)
Coroutine *self = qemu_coroutine_self();
IO_CODE();
qemu_mutex_lock(&bs->reqs_lock);
QLIST_FOREACH(req, &bs->tracked_requests, list) {
if (req->co == self) {
qemu_mutex_unlock(&bs->reqs_lock);
return req;
}
}
qemu_mutex_unlock(&bs->reqs_lock);
return NULL;
}
block/iscsi.c
+25 -38
View File
@@ -107,7 +107,6 @@ typedef struct IscsiLun {
typedef struct IscsiTask {
int status;
int complete;
int retries;
int do_retry;
struct scsi_task *task;
@@ -120,6 +119,7 @@ typedef struct IscsiTask {
typedef struct IscsiAIOCB {
BlockAIOCB common;
AioContext *ctx;
QEMUBH *bh;
IscsiLun *iscsilun;
struct scsi_task *task;
@@ -174,27 +174,16 @@ iscsi_schedule_bh(IscsiAIOCB *acb)
if (acb->bh) {
return;
}
acb->bh = aio_bh_new(acb->iscsilun->aio_context, iscsi_bh_cb, acb);
acb->bh = aio_bh_new(acb->ctx, iscsi_bh_cb, acb);
qemu_bh_schedule(acb->bh);
}
#endif
static void iscsi_co_generic_bh_cb(void *opaque)
{
struct IscsiTask *iTask = opaque;
iTask->complete = 1;
aio_co_wake(iTask->co);
}
static void iscsi_retry_timer_expired(void *opaque)
{
struct IscsiTask *iTask = opaque;
iTask->complete = 1;
if (iTask->co) {
aio_co_wake(iTask->co);
}
aio_co_wake(iTask->co);
}
static inline unsigned exp_random(double mean)
@@ -239,6 +228,8 @@ iscsi_co_generic_cb(struct iscsi_context *iscsi, int status,
{
struct IscsiTask *iTask = opaque;
struct scsi_task *task = command_data;
IscsiLun *iscsilun = iTask->iscsilun;
AioContext *itask_ctx = qemu_coroutine_get_aio_context(iTask->co);
iTask->status = status;
iTask->do_retry = 0;
@@ -263,9 +254,9 @@ iscsi_co_generic_cb(struct iscsi_context *iscsi, int status,
" (retry #%u in %u ms): %s",
iTask->retries, retry_time,
iscsi_get_error(iscsi));
aio_timer_init(iTask->iscsilun->aio_context,
&iTask->retry_timer, QEMU_CLOCK_REALTIME,
SCALE_MS, iscsi_retry_timer_expired, iTask);
aio_timer_init(itask_ctx, &iTask->retry_timer,
QEMU_CLOCK_REALTIME, SCALE_MS,
iscsi_retry_timer_expired, iTask);
timer_mod(&iTask->retry_timer,
qemu_clock_get_ms(QEMU_CLOCK_REALTIME) + retry_time);
iTask->do_retry = 1;
@@ -284,12 +275,17 @@ iscsi_co_generic_cb(struct iscsi_context *iscsi, int status,
}
}
if (iTask->co) {
replay_bh_schedule_oneshot_event(iTask->iscsilun->aio_context,
iscsi_co_generic_bh_cb, iTask);
} else {
iTask->complete = 1;
}
/*
* aio_co_wake() is safe to call: iscsi_service(), which called us, is only
* run from the event_timer and/or the FD handlers, never from the request
* coroutine. The request coroutine in turn will yield unconditionally.
* We must release the lock, though, in case we enter the coroutine
* directly. (Note that if do we enter the coroutine, iTask will probably
* be dangling once aio_co_wake() returns.)
*/
qemu_mutex_unlock(&iscsilun->mutex);
aio_co_wake(iTask->co);
qemu_mutex_lock(&iscsilun->mutex);
}
static void coroutine_fn
@@ -592,12 +588,10 @@ static inline bool iscsi_allocmap_is_valid(IscsiLun *iscsilun,
static void coroutine_fn iscsi_co_wait_for_task(IscsiTask *iTask,
IscsiLun *iscsilun)
{
while (!iTask->complete) {
iscsi_set_events(iscsilun);
qemu_mutex_unlock(&iscsilun->mutex);
qemu_coroutine_yield();
qemu_mutex_lock(&iscsilun->mutex);
}
iscsi_set_events(iscsilun);
qemu_mutex_unlock(&iscsilun->mutex);
qemu_coroutine_yield();
qemu_mutex_lock(&iscsilun->mutex);
}
static int coroutine_fn
@@ -669,7 +663,6 @@ retry:
}
if (iTask.do_retry) {
iTask.complete = 0;
goto retry;
}
@@ -740,7 +733,6 @@ retry:
scsi_free_scsi_task(iTask.task);
iTask.task = NULL;
}
iTask.complete = 0;
goto retry;
}
@@ -902,7 +894,6 @@ retry:
}
if (iTask.do_retry) {
iTask.complete = 0;
goto retry;
}
@@ -940,7 +931,6 @@ retry:
}
if (iTask.do_retry) {
iTask.complete = 0;
goto retry;
}
@@ -1023,8 +1013,7 @@ static void iscsi_ioctl_handle_emulated(IscsiAIOCB *acb, int req, void *buf)
ret = -EINVAL;
}
assert(!acb->bh);
acb->bh = aio_bh_new(bdrv_get_aio_context(bs),
iscsi_ioctl_bh_completion, acb);
acb->bh = aio_bh_new(acb->ctx, iscsi_ioctl_bh_completion, acb);
acb->ret = ret;
qemu_bh_schedule(acb->bh);
}
@@ -1041,6 +1030,7 @@ static BlockAIOCB *iscsi_aio_ioctl(BlockDriverState *bs,
acb = qemu_aio_get(&iscsi_aiocb_info, bs, cb, opaque);
acb->iscsilun = iscsilun;
acb->ctx = qemu_get_current_aio_context();
acb->bh = NULL;
acb->status = -EINPROGRESS;
acb->ioh = buf;
@@ -1184,7 +1174,6 @@ retry:
}
if (iTask.do_retry) {
iTask.complete = 0;
goto retry;
}
@@ -1301,7 +1290,6 @@ retry:
}
if (iTask.do_retry) {
iTask.complete = 0;
goto retry;
}
@@ -2390,7 +2378,6 @@ retry:
iscsi_co_wait_for_task(&iscsi_task, dst_lun);
if (iscsi_task.do_retry) {
iscsi_task.complete = 0;
goto retry;
}
block/nfs.c
+16 -25
View File
@@ -69,7 +69,6 @@ typedef struct NFSClient {
typedef struct NFSRPC {
BlockDriverState *bs;
int ret;
int complete;
QEMUIOVector *iov;
struct stat *st;
Coroutine *co;
@@ -230,14 +229,6 @@ static void coroutine_fn nfs_co_init_task(BlockDriverState *bs, NFSRPC *task)
};
}
static void nfs_co_generic_bh_cb(void *opaque)
{
NFSRPC *task = opaque;
task->complete = 1;
aio_co_wake(task->co);
}
/* Called (via nfs_service) with QemuMutex held. */
static void
nfs_co_generic_cb(int ret, struct nfs_context *nfs, void *data,
@@ -256,8 +247,16 @@ nfs_co_generic_cb(int ret, struct nfs_context *nfs, void *data,
if (task->ret < 0) {
error_report("NFS Error: %s", nfs_get_error(nfs));
}
replay_bh_schedule_oneshot_event(task->client->aio_context,
nfs_co_generic_bh_cb, task);
/*
* Safe to call: nfs_service(), which called us, is only run from the FD
* handlers, never from the request coroutine. The request coroutine in
* turn will yield unconditionally.
* No need to release the lock, even if we directly enter the coroutine, as
* the lock is never re-taken after yielding. (Note: If we do enter the
* coroutine, @task will probably be dangling once aio_co_wake() returns.)
*/
aio_co_wake(task->co);
}
static int coroutine_fn nfs_co_preadv(BlockDriverState *bs, int64_t offset,
@@ -278,9 +277,7 @@ static int coroutine_fn nfs_co_preadv(BlockDriverState *bs, int64_t offset,
nfs_set_events(client);
}
while (!task.complete) {
qemu_coroutine_yield();
}
qemu_coroutine_yield();
if (task.ret < 0) {
return task.ret;
@@ -328,9 +325,7 @@ static int coroutine_fn nfs_co_pwritev(BlockDriverState *bs, int64_t offset,
nfs_set_events(client);
}
while (!task.complete) {
qemu_coroutine_yield();
}
qemu_coroutine_yield();
if (my_buffer) {
g_free(buf);
@@ -358,9 +353,7 @@ static int coroutine_fn nfs_co_flush(BlockDriverState *bs)
nfs_set_events(client);
}
while (!task.complete) {
qemu_coroutine_yield();
}
qemu_coroutine_yield();
return task.ret;
}
@@ -723,8 +716,8 @@ nfs_get_allocated_file_size_cb(int ret, struct nfs_context *nfs, void *data,
if (task->ret < 0) {
error_report("NFS Error: %s", nfs_get_error(nfs));
}
replay_bh_schedule_oneshot_event(task->client->aio_context,
nfs_co_generic_bh_cb, task);
/* Safe to call, see nfs_co_generic_cb() */
aio_co_wake(task->co);
}
static int64_t coroutine_fn nfs_co_get_allocated_file_size(BlockDriverState *bs)
@@ -748,9 +741,7 @@ static int64_t coroutine_fn nfs_co_get_allocated_file_size(BlockDriverState *bs)
nfs_set_events(client);
}
while (!task.complete) {
qemu_coroutine_yield();
}
qemu_coroutine_yield();
return (task.ret < 0 ? task.ret : st.st_blocks * 512);
}
block/null.c
+3 -4
View File
@@ -173,18 +173,17 @@ static inline BlockAIOCB *null_aio_common(BlockDriverState *bs,
{
NullAIOCB *acb;
BDRVNullState *s = bs->opaque;
AioContext *ctx = qemu_get_current_aio_context();
acb = qemu_aio_get(&null_aiocb_info, bs, cb, opaque);
/* Only emulate latency after vcpu is running. */
if (s->latency_ns) {
aio_timer_init(bdrv_get_aio_context(bs), &acb->timer,
QEMU_CLOCK_REALTIME, SCALE_NS,
aio_timer_init(ctx, &acb->timer, QEMU_CLOCK_REALTIME, SCALE_NS,
null_timer_cb, acb);
timer_mod_ns(&acb->timer,
qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + s->latency_ns);
} else {
replay_bh_schedule_oneshot_event(bdrv_get_aio_context(bs),
null_bh_cb, acb);
replay_bh_schedule_oneshot_event(ctx, null_bh_cb, acb);
}
return &acb->common;
}
block/nvme.c
+76 -37
View File
@@ -65,6 +65,7 @@ typedef struct {
} NVMeQueue;
typedef struct {
/* Called from nvme_process_completion() in the BDS's main AioContext */
BlockCompletionFunc *cb;
void *opaque;
int cid;
@@ -84,6 +85,7 @@ typedef struct {
uint8_t *prp_list_pages;
/* Fields protected by @lock */
/* Coroutines in this queue are woken in their own context */
CoQueue free_req_queue;
NVMeQueue sq, cq;
int cq_phase;
@@ -92,7 +94,7 @@ typedef struct {
int need_kick;
int inflight;
/* Thread-safe, no lock necessary */
/* Thread-safe, no lock necessary; runs in the BDS's main context */
QEMUBH *completion_bh;
} NVMeQueuePair;
@@ -206,11 +208,13 @@ static void nvme_free_queue_pair(NVMeQueuePair *q)
g_free(q);
}
/* Runs in the BDS's main AioContext */
static void nvme_free_req_queue_cb(void *opaque)
{
NVMeQueuePair *q = opaque;
qemu_mutex_lock(&q->lock);
/* qemu_co_enter_next() wakes the coroutine in its own AioContext */
while (q->free_req_head != -1 &&
qemu_co_enter_next(&q->free_req_queue, &q->lock)) {
/* Retry waiting requests */
@@ -281,7 +285,7 @@ fail:
return NULL;
}
/* With q->lock */
/* With q->lock, must be run in the BDS's main AioContext */
static void nvme_kick(NVMeQueuePair *q)
{
BDRVNVMeState *s = q->s;
@@ -308,7 +312,10 @@ static NVMeRequest *nvme_get_free_req_nofail_locked(NVMeQueuePair *q)
return req;
}
/* Return a free request element if any, otherwise return NULL. */
/*
* Return a free request element if any, otherwise return NULL.
* May be run from any AioContext.
*/
static NVMeRequest *nvme_get_free_req_nowait(NVMeQueuePair *q)
{
QEMU_LOCK_GUARD(&q->lock);
@@ -321,6 +328,7 @@ static NVMeRequest *nvme_get_free_req_nowait(NVMeQueuePair *q)
/*
* Wait for a free request to become available if necessary, then
* return it.
* May be called in any AioContext.
*/
static coroutine_fn NVMeRequest *nvme_get_free_req(NVMeQueuePair *q)
{
@@ -328,20 +336,21 @@ static coroutine_fn NVMeRequest *nvme_get_free_req(NVMeQueuePair *q)
while (q->free_req_head == -1) {
trace_nvme_free_req_queue_wait(q->s, q->index);
/* nvme_free_req_queue_cb() wakes us in our own AioContext */
qemu_co_queue_wait(&q->free_req_queue, &q->lock);
}
return nvme_get_free_req_nofail_locked(q);
}
/* With q->lock */
/* With q->lock, may be called in any AioContext */
static void nvme_put_free_req_locked(NVMeQueuePair *q, NVMeRequest *req)
{
req->free_req_next = q->free_req_head;
q->free_req_head = req - q->reqs;
}
/* With q->lock */
/* With q->lock, may be called in any AioContext */
static void nvme_wake_free_req_locked(NVMeQueuePair *q)
{
if (!qemu_co_queue_empty(&q->free_req_queue)) {
@@ -350,7 +359,7 @@ static void nvme_wake_free_req_locked(NVMeQueuePair *q)
}
}
/* Insert a request in the freelist and wake waiters */
/* Insert a request in the freelist and wake waiters (from any AioContext) */
static void nvme_put_free_req_and_wake(NVMeQueuePair *q, NVMeRequest *req)
{
qemu_mutex_lock(&q->lock);
@@ -381,7 +390,7 @@ static inline int nvme_translate_error(const NvmeCqe *c)
}
}
/* With q->lock */
/* With q->lock, must be run in the BDS's main AioContext */
static bool nvme_process_completion(NVMeQueuePair *q)
{
BDRVNVMeState *s = q->s;
@@ -451,6 +460,7 @@ static bool nvme_process_completion(NVMeQueuePair *q)
return progress;
}
/* As q->completion_bh, runs in the BDS's main AioContext */
static void nvme_process_completion_bh(void *opaque)
{
NVMeQueuePair *q = opaque;
@@ -481,7 +491,8 @@ static void nvme_trace_command(const NvmeCmd *cmd)
}
}
static void nvme_deferred_fn(void *opaque)
/* Must be run in the BDS's main AioContext */
static void nvme_kick_and_check_completions(void *opaque)
{
NVMeQueuePair *q = opaque;
@@ -490,6 +501,20 @@ static void nvme_deferred_fn(void *opaque)
nvme_process_completion(q);
}
/* Runs in nvme_submit_command()'s AioContext */
static void nvme_deferred_fn(void *opaque)
{
NVMeQueuePair *q = opaque;
if (qemu_get_current_aio_context() == q->s->aio_context) {
nvme_kick_and_check_completions(q);
} else {
aio_bh_schedule_oneshot(q->s->aio_context,
nvme_kick_and_check_completions, q);
}
}
/* May be run in any AioContext */
static void nvme_submit_command(NVMeQueuePair *q, NVMeRequest *req,
NvmeCmd *cmd, BlockCompletionFunc cb,
void *opaque)
@@ -511,6 +536,7 @@ static void nvme_submit_command(NVMeQueuePair *q, NVMeRequest *req,
defer_call(nvme_deferred_fn, q);
}
/* Put into NVMeRequest.cb, so runs in the BDS's main AioContext */
static void nvme_admin_cmd_sync_cb(void *opaque, int ret)
{
int *pret = opaque;
@@ -518,6 +544,7 @@ static void nvme_admin_cmd_sync_cb(void *opaque, int ret)
aio_wait_kick();
}
/* Must be run in the BDS's or qemu's main AioContext */
static int nvme_admin_cmd_sync(BlockDriverState *bs, NvmeCmd *cmd)
{
BDRVNVMeState *s = bs->opaque;
@@ -626,6 +653,7 @@ out:
return ret;
}
/* Must be run in the BDS's main AioContext */
static void nvme_poll_queue(NVMeQueuePair *q)
{
const size_t cqe_offset = q->cq.head * NVME_CQ_ENTRY_BYTES;
@@ -648,6 +676,7 @@ static void nvme_poll_queue(NVMeQueuePair *q)
qemu_mutex_unlock(&q->lock);
}
/* Must be run in the BDS's main AioContext */
static void nvme_poll_queues(BDRVNVMeState *s)
{
int i;
@@ -657,6 +686,7 @@ static void nvme_poll_queues(BDRVNVMeState *s)
}
}
/* Run as an event notifier in the BDS's main AioContext */
static void nvme_handle_event(EventNotifier *n)
{
BDRVNVMeState *s = container_of(n, BDRVNVMeState,
@@ -710,6 +740,7 @@ out_error:
return false;
}
/* Run as an event notifier in the BDS's main AioContext */
static bool nvme_poll_cb(void *opaque)
{
EventNotifier *e = opaque;
@@ -733,6 +764,7 @@ static bool nvme_poll_cb(void *opaque)
return false;
}
/* Run as an event notifier in the BDS's main AioContext */
static void nvme_poll_ready(EventNotifier *e)
{
BDRVNVMeState *s = container_of(e, BDRVNVMeState,
@@ -1038,7 +1070,7 @@ static int nvme_probe_blocksizes(BlockDriverState *bs, BlockSizes *bsz)
return 0;
}
/* Called with s->dma_map_lock */
/* Called with s->dma_map_lock, may be run in any AioContext */
static coroutine_fn int nvme_cmd_unmap_qiov(BlockDriverState *bs,
QEMUIOVector *qiov)
{
@@ -1049,13 +1081,17 @@ static coroutine_fn int nvme_cmd_unmap_qiov(BlockDriverState *bs,
if (!s->dma_map_count && !qemu_co_queue_empty(&s->dma_flush_queue)) {
r = qemu_vfio_dma_reset_temporary(s->vfio);
if (!r) {
/*
* Queue access is protected by the dma_map_lock, and all
* coroutines are woken in their own AioContext
*/
qemu_co_queue_restart_all(&s->dma_flush_queue);
}
}
return r;
}
/* Called with s->dma_map_lock */
/* Called with s->dma_map_lock, may be run in any AioContext */
static coroutine_fn int nvme_cmd_map_qiov(BlockDriverState *bs, NvmeCmd *cmd,
NVMeRequest *req, QEMUIOVector *qiov)
{
@@ -1164,25 +1200,36 @@ fail:
typedef struct {
Coroutine *co;
bool skip_yield;
int ret;
AioContext *ctx;
} NVMeCoData;
static void nvme_rw_cb_bh(void *opaque)
{
NVMeCoData *data = opaque;
qemu_coroutine_enter(data->co);
}
/* Put into NVMeRequest.cb, so runs in the BDS's main AioContext */
static void nvme_rw_cb(void *opaque, int ret)
{
NVMeCoData *data = opaque;
data->ret = ret;
if (!data->co) {
/* The rw coroutine hasn't yielded, don't try to enter. */
return;
if (data->co == qemu_coroutine_self()) {
/*
* Fast path: We are inside of the request coroutine (through
* nvme_submit_command, nvme_deferred_fn, nvme_process_completion).
* We can set data->skip_yield here to keep the coroutine from
* yielding, and then we don't need to schedule a BH to wake it.
*/
data->skip_yield = true;
} else {
/*
* Safe to call: The case where we run in the request coroutine is
* handled above, so we must be independent of it; and without
* skip_yield set, the coroutine will yield.
* No need to release NVMeQueuePair.lock (we are called without it
* held). (Note: If we enter the coroutine here, @data will
* probably be dangling once aio_co_wake() returns.)
*/
aio_co_wake(data->co);
}
replay_bh_schedule_oneshot_event(data->ctx, nvme_rw_cb_bh, data);
}
static coroutine_fn int nvme_co_prw_aligned(BlockDriverState *bs,
@@ -1206,7 +1253,7 @@ static coroutine_fn int nvme_co_prw_aligned(BlockDriverState *bs,
.cdw12 = cpu_to_le32(cdw12),
};
NVMeCoData data = {
.ctx = bdrv_get_aio_context(bs),
.co = qemu_coroutine_self(),
.ret = -EINPROGRESS,
};
@@ -1223,9 +1270,7 @@ static coroutine_fn int nvme_co_prw_aligned(BlockDriverState *bs,
return r;
}
nvme_submit_command(ioq, req, &cmd, nvme_rw_cb, &data);
data.co = qemu_coroutine_self();
while (data.ret == -EINPROGRESS) {
if (!data.skip_yield) {
qemu_coroutine_yield();
}
@@ -1321,7 +1366,7 @@ static coroutine_fn int nvme_co_flush(BlockDriverState *bs)
.nsid = cpu_to_le32(s->nsid),
};
NVMeCoData data = {
.ctx = bdrv_get_aio_context(bs),
.co = qemu_coroutine_self(),
.ret = -EINPROGRESS,
};
@@ -1329,9 +1374,7 @@ static coroutine_fn int nvme_co_flush(BlockDriverState *bs)
req = nvme_get_free_req(ioq);
assert(req);
nvme_submit_command(ioq, req, &cmd, nvme_rw_cb, &data);
data.co = qemu_coroutine_self();
if (data.ret == -EINPROGRESS) {
if (!data.skip_yield) {
qemu_coroutine_yield();
}
@@ -1372,7 +1415,7 @@ static coroutine_fn int nvme_co_pwrite_zeroes(BlockDriverState *bs,
};
NVMeCoData data = {
.ctx = bdrv_get_aio_context(bs),
.co = qemu_coroutine_self(),
.ret = -EINPROGRESS,
};
@@ -1392,9 +1435,7 @@ static coroutine_fn int nvme_co_pwrite_zeroes(BlockDriverState *bs,
assert(req);
nvme_submit_command(ioq, req, &cmd, nvme_rw_cb, &data);
data.co = qemu_coroutine_self();
while (data.ret == -EINPROGRESS) {
if (!data.skip_yield) {
qemu_coroutine_yield();
}
@@ -1422,7 +1463,7 @@ static int coroutine_fn nvme_co_pdiscard(BlockDriverState *bs,
};
NVMeCoData data = {
.ctx = bdrv_get_aio_context(bs),
.co = qemu_coroutine_self(),
.ret = -EINPROGRESS,
};
@@ -1467,9 +1508,7 @@ static int coroutine_fn nvme_co_pdiscard(BlockDriverState *bs,
trace_nvme_dsm(s, offset, bytes);
nvme_submit_command(ioq, req, &cmd, nvme_rw_cb, &data);
data.co = qemu_coroutine_self();
while (data.ret == -EINPROGRESS) {
if (!data.skip_yield) {
qemu_coroutine_yield();
}
block/qcow2.c
+115 -28
View File
@@ -835,41 +835,113 @@ static const char *overlap_bool_option_names[QCOW2_OL_MAX_BITNR] = {
[QCOW2_OL_BITMAP_DIRECTORY_BITNR] = QCOW2_OPT_OVERLAP_BITMAP_DIRECTORY,
};
static void cache_clean_timer_cb(void *opaque)
static void coroutine_fn cache_clean_timer(void *opaque)
{
BlockDriverState *bs = opaque;
BDRVQcow2State *s = bs->opaque;
qcow2_cache_clean_unused(s->l2_table_cache);
qcow2_cache_clean_unused(s->refcount_block_cache);
timer_mod(s->cache_clean_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
(int64_t) s->cache_clean_interval * 1000);
BDRVQcow2State *s = opaque;
uint64_t wait_ns;
WITH_QEMU_LOCK_GUARD(&s->lock) {
wait_ns = s->cache_clean_interval * NANOSECONDS_PER_SECOND;
}
while (wait_ns > 0) {
qemu_co_sleep_ns_wakeable(&s->cache_clean_timer_wake,
QEMU_CLOCK_REALTIME, wait_ns);
WITH_QEMU_LOCK_GUARD(&s->lock) {
if (s->cache_clean_interval > 0) {
qcow2_cache_clean_unused(s->l2_table_cache);
qcow2_cache_clean_unused(s->refcount_block_cache);
}
wait_ns = s->cache_clean_interval * NANOSECONDS_PER_SECOND;
}
}
WITH_QEMU_LOCK_GUARD(&s->lock) {
s->cache_clean_timer_co = NULL;
qemu_co_queue_restart_all(&s->cache_clean_timer_exit);
}
}
static void cache_clean_timer_init(BlockDriverState *bs, AioContext *context)
{
BDRVQcow2State *s = bs->opaque;
if (s->cache_clean_interval > 0) {
s->cache_clean_timer =
aio_timer_new_with_attrs(context, QEMU_CLOCK_VIRTUAL,
SCALE_MS, QEMU_TIMER_ATTR_EXTERNAL,
cache_clean_timer_cb, bs);
timer_mod(s->cache_clean_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
(int64_t) s->cache_clean_interval * 1000);
assert(!s->cache_clean_timer_co);
s->cache_clean_timer_co = qemu_coroutine_create(cache_clean_timer, s);
aio_co_enter(context, s->cache_clean_timer_co);
}
}
static void cache_clean_timer_del(BlockDriverState *bs)
/**
* Delete the cache clean timer and await any yet running instance.
* Called holding s->lock.
*/
static void coroutine_fn
cache_clean_timer_co_locked_del_and_wait(BlockDriverState *bs)
{
BDRVQcow2State *s = bs->opaque;
if (s->cache_clean_timer) {
timer_free(s->cache_clean_timer);
s->cache_clean_timer = NULL;
if (s->cache_clean_timer_co) {
s->cache_clean_interval = 0;
qemu_co_sleep_wake(&s->cache_clean_timer_wake);
qemu_co_queue_wait(&s->cache_clean_timer_exit, &s->lock);
}
}
/**
* Same as cache_clean_timer_co_locked_del_and_wait(), but takes s->lock.
*/
static void coroutine_fn
cache_clean_timer_co_del_and_wait(BlockDriverState *bs)
{
BDRVQcow2State *s = bs->opaque;
WITH_QEMU_LOCK_GUARD(&s->lock) {
cache_clean_timer_co_locked_del_and_wait(bs);
}
}
struct CacheCleanTimerDelAndWaitCoParams {
BlockDriverState *bs;
bool done;
};
static void coroutine_fn cache_clean_timer_del_and_wait_co_entry(void *opaque)
{
struct CacheCleanTimerDelAndWaitCoParams *p = opaque;
cache_clean_timer_co_del_and_wait(p->bs);
p->done = true;
aio_wait_kick();
}
/**
* Delete the cache clean timer and await any yet running instance.
* Must be called from the main or BDS AioContext without s->lock held.
*/
static void coroutine_mixed_fn
cache_clean_timer_del_and_wait(BlockDriverState *bs)
{
IO_OR_GS_CODE();
if (qemu_in_coroutine()) {
cache_clean_timer_co_del_and_wait(bs);
} else {
struct CacheCleanTimerDelAndWaitCoParams p = { .bs = bs };
Coroutine *co;
co = qemu_coroutine_create(cache_clean_timer_del_and_wait_co_entry, &p);
qemu_coroutine_enter(co);
BDRV_POLL_WHILE(bs, !p.done);
}
}
static void qcow2_detach_aio_context(BlockDriverState *bs)
{
cache_clean_timer_del(bs);
cache_clean_timer_del_and_wait(bs);
}
static void qcow2_attach_aio_context(BlockDriverState *bs,
@@ -1214,12 +1286,24 @@ fail:
return ret;
}
/* s_locked specifies whether s->lock is held or not */
static void qcow2_update_options_commit(BlockDriverState *bs,
Qcow2ReopenState *r)
Qcow2ReopenState *r,
bool s_locked)
{
BDRVQcow2State *s = bs->opaque;
int i;
/*
* We need to stop the cache-clean-timer before destroying the metadata
* table caches
*/
if (s_locked) {
cache_clean_timer_co_locked_del_and_wait(bs);
} else {
cache_clean_timer_del_and_wait(bs);
}
if (s->l2_table_cache) {
qcow2_cache_destroy(s->l2_table_cache);
}
@@ -1228,6 +1312,7 @@ static void qcow2_update_options_commit(BlockDriverState *bs,
}
s->l2_table_cache = r->l2_table_cache;
s->refcount_block_cache = r->refcount_block_cache;
s->l2_slice_size = r->l2_slice_size;
s->overlap_check = r->overlap_check;
@@ -1239,11 +1324,8 @@ static void qcow2_update_options_commit(BlockDriverState *bs,
s->discard_no_unref = r->discard_no_unref;
if (s->cache_clean_interval != r->cache_clean_interval) {
cache_clean_timer_del(bs);
s->cache_clean_interval = r->cache_clean_interval;
cache_clean_timer_init(bs, bdrv_get_aio_context(bs));
}
s->cache_clean_interval = r->cache_clean_interval;
cache_clean_timer_init(bs, bdrv_get_aio_context(bs));
qapi_free_QCryptoBlockOpenOptions(s->crypto_opts);
s->crypto_opts = r->crypto_opts;
@@ -1261,6 +1343,7 @@ static void qcow2_update_options_abort(BlockDriverState *bs,
qapi_free_QCryptoBlockOpenOptions(r->crypto_opts);
}
/* Called with s->lock held */
static int coroutine_fn GRAPH_RDLOCK
qcow2_update_options(BlockDriverState *bs, QDict *options, int flags,
Error **errp)
@@ -1270,7 +1353,7 @@ qcow2_update_options(BlockDriverState *bs, QDict *options, int flags,
ret = qcow2_update_options_prepare(bs, &r, options, flags, errp);
if (ret >= 0) {
qcow2_update_options_commit(bs, &r);
qcow2_update_options_commit(bs, &r, true);
} else {
qcow2_update_options_abort(bs, &r);
}
@@ -1908,7 +1991,7 @@ qcow2_do_open(BlockDriverState *bs, QDict *options, int flags,
qemu_vfree(s->l1_table);
/* else pre-write overlap checks in cache_destroy may crash */
s->l1_table = NULL;
cache_clean_timer_del(bs);
cache_clean_timer_co_locked_del_and_wait(bs);
if (s->l2_table_cache) {
qcow2_cache_destroy(s->l2_table_cache);
}
@@ -1963,6 +2046,7 @@ static int qcow2_open(BlockDriverState *bs, QDict *options, int flags,
/* Initialise locks */
qemu_co_mutex_init(&s->lock);
qemu_co_queue_init(&s->cache_clean_timer_exit);
assert(!qemu_in_coroutine());
assert(qemu_get_current_aio_context() == qemu_get_aio_context());
@@ -2048,7 +2132,7 @@ static void qcow2_reopen_commit(BDRVReopenState *state)
GRAPH_RDLOCK_GUARD_MAINLOOP();
qcow2_update_options_commit(state->bs, state->opaque);
qcow2_update_options_commit(state->bs, state->opaque, false);
if (!s->data_file) {
/*
* If we don't have an external data file, s->data_file was cleared by
@@ -2805,7 +2889,7 @@ qcow2_do_close(BlockDriverState *bs, bool close_data_file)
qcow2_inactivate(bs);
}
cache_clean_timer_del(bs);
cache_clean_timer_del_and_wait(bs);
qcow2_cache_destroy(s->l2_table_cache);
qcow2_cache_destroy(s->refcount_block_cache);
@@ -2873,6 +2957,9 @@ qcow2_co_invalidate_cache(BlockDriverState *bs, Error **errp)
data_file = s->data_file;
memset(s, 0, sizeof(BDRVQcow2State));
s->data_file = data_file;
/* Re-initialize objects initialized in qcow2_open() */
qemu_co_mutex_init(&s->lock);
qemu_co_queue_init(&s->cache_clean_timer_exit);
options = qdict_clone_shallow(bs->options);
block/qcow2.h
+4 -1
View File
@@ -345,8 +345,11 @@ typedef struct BDRVQcow2State {
Qcow2Cache *l2_table_cache;
Qcow2Cache *refcount_block_cache;
QEMUTimer *cache_clean_timer;
/* Non-NULL while the timer is running */
Coroutine *cache_clean_timer_co;
unsigned cache_clean_interval;
QemuCoSleep cache_clean_timer_wake;
CoQueue cache_clean_timer_exit;
QLIST_HEAD(, QCowL2Meta) cluster_allocs;
block/rbd.c
+4 -8
View File
@@ -110,9 +110,7 @@ typedef struct BDRVRBDState {
} BDRVRBDState;
typedef struct RBDTask {
BlockDriverState *bs;
Coroutine *co;
bool complete;
int64_t ret;
} RBDTask;
@@ -1309,7 +1307,6 @@ static int qemu_rbd_resize(BlockDriverState *bs, uint64_t size)
static void qemu_rbd_finish_bh(void *opaque)
{
RBDTask *task = opaque;
task->complete = true;
aio_co_wake(task->co);
}
@@ -1326,7 +1323,7 @@ static void qemu_rbd_completion_cb(rbd_completion_t c, RBDTask *task)
{
task->ret = rbd_aio_get_return_value(c);
rbd_aio_release(c);
aio_bh_schedule_oneshot(bdrv_get_aio_context(task->bs),
aio_bh_schedule_oneshot(qemu_coroutine_get_aio_context(task->co),
qemu_rbd_finish_bh, task);
}
@@ -1338,7 +1335,7 @@ static int coroutine_fn qemu_rbd_start_co(BlockDriverState *bs,
RBDAIOCmd cmd)
{
BDRVRBDState *s = bs->opaque;
RBDTask task = { .bs = bs, .co = qemu_coroutine_self() };
RBDTask task = { .co = qemu_coroutine_self() };
rbd_completion_t c;
int r;
@@ -1401,9 +1398,8 @@ static int coroutine_fn qemu_rbd_start_co(BlockDriverState *bs,
return r;
}
while (!task.complete) {
qemu_coroutine_yield();
}
/* Expect exactly a single wake from qemu_rbd_finish_bh() */
qemu_coroutine_yield();
if (task.ret < 0) {
error_report("rbd request failed: cmd %d offset %" PRIu64 " bytes %"
block/ssh.c
+11 -11
View File
@@ -1010,19 +1010,18 @@ static int ssh_has_zero_init(BlockDriverState *bs)
}
typedef struct BDRVSSHRestart {
BlockDriverState *bs;
BDRVSSHState *s;
Coroutine *co;
} BDRVSSHRestart;
static void restart_coroutine(void *opaque)
{
BDRVSSHRestart *restart = opaque;
BlockDriverState *bs = restart->bs;
BDRVSSHState *s = bs->opaque;
AioContext *ctx = bdrv_get_aio_context(bs);
BDRVSSHState *s = restart->s;
trace_ssh_restart_coroutine(restart->co);
aio_set_fd_handler(ctx, s->sock, NULL, NULL, NULL, NULL, NULL);
aio_set_fd_handler(qemu_get_current_aio_context(), s->sock,
NULL, NULL, NULL, NULL, NULL);
aio_co_wake(restart->co);
}
@@ -1031,12 +1030,13 @@ static void restart_coroutine(void *opaque)
* handlers are set up so that we'll be rescheduled when there is an
* interesting event on the socket.
*/
static coroutine_fn void co_yield(BDRVSSHState *s, BlockDriverState *bs)
static coroutine_fn void co_yield(BDRVSSHState *s)
{
int r;
IOHandler *rd_handler = NULL, *wr_handler = NULL;
AioContext *ctx = qemu_get_current_aio_context();
BDRVSSHRestart restart = {
.bs = bs,
.s = s,
.co = qemu_coroutine_self()
};
@@ -1051,7 +1051,7 @@ static coroutine_fn void co_yield(BDRVSSHState *s, BlockDriverState *bs)
trace_ssh_co_yield(s->sock, rd_handler, wr_handler);
aio_set_fd_handler(bdrv_get_aio_context(bs), s->sock,
aio_set_fd_handler(ctx, s->sock,
rd_handler, wr_handler, NULL, NULL, &restart);
qemu_coroutine_yield();
trace_ssh_co_yield_back(s->sock);
@@ -1093,7 +1093,7 @@ static coroutine_fn int ssh_read(BDRVSSHState *s, BlockDriverState *bs,
trace_ssh_read_return(r, sftp_get_error(s->sftp));
if (r == SSH_AGAIN) {
co_yield(s, bs);
co_yield(s);
goto again;
}
if (r == SSH_EOF || (r == 0 && sftp_get_error(s->sftp) == SSH_FX_EOF)) {
@@ -1168,7 +1168,7 @@ static coroutine_fn int ssh_write(BDRVSSHState *s, BlockDriverState *bs,
trace_ssh_write_return(r, sftp_get_error(s->sftp));
if (r == SSH_AGAIN) {
co_yield(s, bs);
co_yield(s);
goto again;
}
if (r < 0) {
@@ -1233,7 +1233,7 @@ static coroutine_fn int ssh_flush(BDRVSSHState *s, BlockDriverState *bs)
again:
r = sftp_fsync(s->sftp_handle);
if (r == SSH_AGAIN) {
co_yield(s, bs);
co_yield(s);
goto again;
}
if (r < 0) {
block/win32-aio.c
+24 -7
View File
@@ -48,48 +48,62 @@ struct QEMUWin32AIOState {
typedef struct QEMUWin32AIOCB {
BlockAIOCB common;
struct QEMUWin32AIOState *ctx;
AioContext *req_ctx;
int nbytes;
OVERLAPPED ov;
QEMUIOVector *qiov;
void *buf;
bool is_read;
bool is_linear;
int ret;
} QEMUWin32AIOCB;
static void win32_aio_completion_cb_bh(void *opaque)
{
QEMUWin32AIOCB *waiocb = opaque;
waiocb->common.cb(waiocb->common.opaque, waiocb->ret);
aio_context_unref(waiocb->req_ctx);
qemu_aio_unref(waiocb);
}
/*
* Completes an AIO request (calls the callback and frees the ACB).
*/
static void win32_aio_process_completion(QEMUWin32AIOState *s,
QEMUWin32AIOCB *waiocb, DWORD count)
{
int ret;
s->count--;
if (waiocb->ov.Internal != 0) {
ret = -EIO;
waiocb->ret = -EIO;
} else {
ret = 0;
waiocb->ret = 0;
if (count < waiocb->nbytes) {
/* Short reads mean EOF, pad with zeros. */
if (waiocb->is_read) {
qemu_iovec_memset(waiocb->qiov, count, 0,
waiocb->qiov->size - count);
} else {
ret = -EINVAL;
waiocb->ret = -EINVAL;
}
}
}
if (!waiocb->is_linear) {
if (ret == 0 && waiocb->is_read) {
if (waiocb->ret == 0 && waiocb->is_read) {
QEMUIOVector *qiov = waiocb->qiov;
iov_from_buf(qiov->iov, qiov->niov, 0, waiocb->buf, qiov->size);
}
qemu_vfree(waiocb->buf);
}
waiocb->common.cb(waiocb->common.opaque, ret);
qemu_aio_unref(waiocb);
if (waiocb->req_ctx == s->aio_ctx) {
win32_aio_completion_cb_bh(waiocb);
} else {
aio_bh_schedule_oneshot(waiocb->req_ctx, win32_aio_completion_cb_bh,
waiocb);
}
}
static void win32_aio_completion_cb(EventNotifier *e)
@@ -120,10 +134,13 @@ BlockAIOCB *win32_aio_submit(BlockDriverState *bs,
DWORD rc;
waiocb = qemu_aio_get(&win32_aiocb_info, bs, cb, opaque);
waiocb->req_ctx = qemu_get_current_aio_context();
waiocb->nbytes = bytes;
waiocb->qiov = qiov;
waiocb->is_read = (type == QEMU_AIO_READ);
aio_context_ref(waiocb->req_ctx);
if (qiov->niov > 1) {
waiocb->buf = qemu_try_blockalign(bs, qiov->size);
if (waiocb->buf == NULL) {
include/block/aio.h
+15
View File
@@ -746,6 +746,21 @@ void coroutine_fn aio_co_reschedule_self(AioContext *new_ctx);
* aio_co_wake may be executed either in coroutine or non-coroutine
* context. The coroutine must not be entered by anyone else while
* aio_co_wake() is active.
*
* If `co`'s AioContext differs from the current AioContext, this will call
* aio_co_schedule(), which makes this safe to use even when `co` has not
* yielded yet. In such a case, it will be entered once it yields.
*
* In contrast, if `co`'s AioContext is equal to the current one, it is
* required for `co` to currently be yielding. This is generally the case
* if the caller is not in `co` (i.e. invoked by `co`), because the only
* other way for the caller to be running then is for `co` to currently be
* yielding.
*
* Therefore, if there is no way for the caller to be invoked/entered by
* `co`, it is generally safe to call this regardless of whether `co` is
* known to already be yielding or not -- it only has to yield at some
* point.
*/
void aio_co_wake(Coroutine *co);
include/block/block_int-common.h
+6 -1
View File
@@ -508,7 +508,12 @@ struct BlockDriver {
BlockDriverState *bs, BlockdevAmendOptions *opts, bool force,
Error **errp);
/* aio */
/*
* AIO
* The given completion callback will be run in the same AioContext as the
* one in which the AIO function was called.
*/
BlockAIOCB * GRAPH_RDLOCK_PTR (*bdrv_aio_preadv)(BlockDriverState *bs,
int64_t offset, int64_t bytes, QEMUIOVector *qiov,
BdrvRequestFlags flags, BlockCompletionFunc *cb, void *opaque);