diff options
Diffstat (limited to 'block')
| -rw-r--r-- | block/backup.c | 17 | ||||
| -rw-r--r-- | block/block-backend.c | 30 | ||||
| -rw-r--r-- | block/commit.c | 2 | ||||
| -rw-r--r-- | block/io.c | 137 | ||||
| -rw-r--r-- | block/mirror.c | 70 | ||||
| -rw-r--r-- | block/nfs.c | 55 | ||||
| -rw-r--r-- | block/qed-table.c | 16 | ||||
| -rw-r--r-- | block/qed.c | 16 | ||||
| -rw-r--r-- | block/replication.c | 27 | ||||
| -rw-r--r-- | block/sheepdog.c | 67 |
10 files changed, 271 insertions, 166 deletions
diff --git a/block/backup.c b/block/backup.c index 02dbe48035..81d4042ae8 100644 --- a/block/backup.c +++ b/block/backup.c @@ -300,6 +300,21 @@ void backup_cow_request_end(CowRequest *req) cow_request_end(req); } +static void backup_drain(BlockJob *job) +{ + BackupBlockJob *s = container_of(job, BackupBlockJob, common); + + /* Need to keep a reference in case blk_drain triggers execution + * of backup_complete... + */ + if (s->target) { + BlockBackend *target = s->target; + blk_ref(target); + blk_drain(target); + blk_unref(target); + } +} + static const BlockJobDriver backup_job_driver = { .instance_size = sizeof(BackupBlockJob), .job_type = BLOCK_JOB_TYPE_BACKUP, @@ -307,6 +322,7 @@ static const BlockJobDriver backup_job_driver = { .commit = backup_commit, .abort = backup_abort, .attached_aio_context = backup_attached_aio_context, + .drain = backup_drain, }; static BlockErrorAction backup_error_action(BackupBlockJob *job, @@ -331,6 +347,7 @@ static void backup_complete(BlockJob *job, void *opaque) BackupCompleteData *data = opaque; blk_unref(s->target); + s->target = NULL; block_job_completed(job, data->ret); g_free(data); diff --git a/block/block-backend.c b/block/block-backend.c index c53ca30000..27a7f6f523 100644 --- a/block/block-backend.c +++ b/block/block-backend.c @@ -799,20 +799,25 @@ int coroutine_fn blk_co_preadv(BlockBackend *blk, int64_t offset, BdrvRequestFlags flags) { int ret; + BlockDriverState *bs = blk_bs(blk); - trace_blk_co_preadv(blk, blk_bs(blk), offset, bytes, flags); + trace_blk_co_preadv(blk, bs, offset, bytes, flags); ret = blk_check_byte_request(blk, offset, bytes); if (ret < 0) { return ret; } + bdrv_inc_in_flight(bs); + /* throttling disk I/O */ if (blk->public.throttle_state) { throttle_group_co_io_limits_intercept(blk, bytes, false); } - return bdrv_co_preadv(blk->root, offset, bytes, qiov, flags); + ret = bdrv_co_preadv(blk->root, offset, bytes, qiov, flags); + bdrv_dec_in_flight(bs); + return ret; } int coroutine_fn blk_co_pwritev(BlockBackend *blk, int64_t offset, @@ -820,14 +825,17 @@ int coroutine_fn blk_co_pwritev(BlockBackend *blk, int64_t offset, BdrvRequestFlags flags) { int ret; + BlockDriverState *bs = blk_bs(blk); - trace_blk_co_pwritev(blk, blk_bs(blk), offset, bytes, flags); + trace_blk_co_pwritev(blk, bs, offset, bytes, flags); ret = blk_check_byte_request(blk, offset, bytes); if (ret < 0) { return ret; } + bdrv_inc_in_flight(bs); + /* throttling disk I/O */ if (blk->public.throttle_state) { throttle_group_co_io_limits_intercept(blk, bytes, true); @@ -837,7 +845,9 @@ int coroutine_fn blk_co_pwritev(BlockBackend *blk, int64_t offset, flags |= BDRV_REQ_FUA; } - return bdrv_co_pwritev(blk->root, offset, bytes, qiov, flags); + ret = bdrv_co_pwritev(blk->root, offset, bytes, qiov, flags); + bdrv_dec_in_flight(bs); + return ret; } typedef struct BlkRwCo { @@ -868,7 +878,6 @@ static int blk_prw(BlockBackend *blk, int64_t offset, uint8_t *buf, int64_t bytes, CoroutineEntry co_entry, BdrvRequestFlags flags) { - AioContext *aio_context; QEMUIOVector qiov; struct iovec iov; Coroutine *co; @@ -890,11 +899,7 @@ static int blk_prw(BlockBackend *blk, int64_t offset, uint8_t *buf, co = qemu_coroutine_create(co_entry, &rwco); qemu_coroutine_enter(co); - - aio_context = blk_get_aio_context(blk); - while (rwco.ret == NOT_DONE) { - aio_poll(aio_context, true); - } + BDRV_POLL_WHILE(blk_bs(blk), rwco.ret == NOT_DONE); return rwco.ret; } @@ -930,6 +935,8 @@ int blk_make_zero(BlockBackend *blk, BdrvRequestFlags flags) static void error_callback_bh(void *opaque) { struct BlockBackendAIOCB *acb = opaque; + + bdrv_dec_in_flight(acb->common.bs); acb->common.cb(acb->common.opaque, acb->ret); qemu_aio_unref(acb); } @@ -940,6 +947,7 @@ BlockAIOCB *blk_abort_aio_request(BlockBackend *blk, { struct BlockBackendAIOCB *acb; + bdrv_inc_in_flight(blk_bs(blk)); acb = blk_aio_get(&block_backend_aiocb_info, blk, cb, opaque); acb->blk = blk; acb->ret = ret; @@ -962,6 +970,7 @@ static const AIOCBInfo blk_aio_em_aiocb_info = { static void blk_aio_complete(BlkAioEmAIOCB *acb) { if (acb->has_returned) { + bdrv_dec_in_flight(acb->common.bs); acb->common.cb(acb->common.opaque, acb->rwco.ret); qemu_aio_unref(acb); } @@ -983,6 +992,7 @@ static BlockAIOCB *blk_aio_prwv(BlockBackend *blk, int64_t offset, int bytes, BlkAioEmAIOCB *acb; Coroutine *co; + bdrv_inc_in_flight(blk_bs(blk)); acb = blk_aio_get(&blk_aio_em_aiocb_info, blk, cb, opaque); acb->rwco = (BlkRwCo) { .blk = blk, diff --git a/block/commit.c b/block/commit.c index 9f67a8b121..499eccaeee 100644 --- a/block/commit.c +++ b/block/commit.c @@ -251,7 +251,7 @@ void commit_start(const char *job_id, BlockDriverState *bs, orig_overlay_flags | BDRV_O_RDWR); } if (reopen_queue) { - bdrv_reopen_multiple(reopen_queue, &local_err); + bdrv_reopen_multiple(bdrv_get_aio_context(bs), reopen_queue, &local_err); if (local_err != NULL) { error_propagate(errp, local_err); block_job_unref(&s->common); diff --git a/block/io.c b/block/io.c index 79cbbdf769..be0d862ca6 100644 --- a/block/io.c +++ b/block/io.c @@ -143,7 +143,7 @@ bool bdrv_requests_pending(BlockDriverState *bs) { BdrvChild *child; - if (!QLIST_EMPTY(&bs->tracked_requests)) { + if (atomic_read(&bs->in_flight)) { return true; } @@ -156,16 +156,22 @@ bool bdrv_requests_pending(BlockDriverState *bs) return false; } -static void bdrv_drain_recurse(BlockDriverState *bs) +static bool bdrv_drain_recurse(BlockDriverState *bs) { BdrvChild *child; + bool waited; + + waited = BDRV_POLL_WHILE(bs, atomic_read(&bs->in_flight) > 0); if (bs->drv && bs->drv->bdrv_drain) { bs->drv->bdrv_drain(bs); } + QLIST_FOREACH(child, &bs->children, next) { - bdrv_drain_recurse(child->bs); + waited |= bdrv_drain_recurse(child->bs); } + + return waited; } typedef struct { @@ -174,23 +180,14 @@ typedef struct { bool done; } BdrvCoDrainData; -static void bdrv_drain_poll(BlockDriverState *bs) -{ - bool busy = true; - - while (busy) { - /* Keep iterating */ - busy = bdrv_requests_pending(bs); - busy |= aio_poll(bdrv_get_aio_context(bs), busy); - } -} - static void bdrv_co_drain_bh_cb(void *opaque) { BdrvCoDrainData *data = opaque; Coroutine *co = data->co; + BlockDriverState *bs = data->bs; - bdrv_drain_poll(data->bs); + bdrv_dec_in_flight(bs); + bdrv_drained_begin(bs); data->done = true; qemu_coroutine_enter(co); } @@ -209,6 +206,7 @@ static void coroutine_fn bdrv_co_yield_to_drain(BlockDriverState *bs) .bs = bs, .done = false, }; + bdrv_inc_in_flight(bs); aio_bh_schedule_oneshot(bdrv_get_aio_context(bs), bdrv_co_drain_bh_cb, &data); @@ -220,6 +218,11 @@ static void coroutine_fn bdrv_co_yield_to_drain(BlockDriverState *bs) void bdrv_drained_begin(BlockDriverState *bs) { + if (qemu_in_coroutine()) { + bdrv_co_yield_to_drain(bs); + return; + } + if (!bs->quiesce_counter++) { aio_disable_external(bdrv_get_aio_context(bs)); bdrv_parent_drained_begin(bs); @@ -227,11 +230,6 @@ void bdrv_drained_begin(BlockDriverState *bs) bdrv_io_unplugged_begin(bs); bdrv_drain_recurse(bs); - if (qemu_in_coroutine()) { - bdrv_co_yield_to_drain(bs); - } else { - bdrv_drain_poll(bs); - } bdrv_io_unplugged_end(bs); } @@ -279,7 +277,7 @@ void bdrv_drain(BlockDriverState *bs) void bdrv_drain_all(void) { /* Always run first iteration so any pending completion BHs run */ - bool busy = true; + bool waited = true; BlockDriverState *bs; BdrvNextIterator it; BlockJob *job = NULL; @@ -299,7 +297,6 @@ void bdrv_drain_all(void) aio_context_acquire(aio_context); bdrv_parent_drained_begin(bs); bdrv_io_unplugged_begin(bs); - bdrv_drain_recurse(bs); aio_context_release(aio_context); if (!g_slist_find(aio_ctxs, aio_context)) { @@ -313,8 +310,8 @@ void bdrv_drain_all(void) * request completion. Therefore we must keep looping until there was no * more activity rather than simply draining each device independently. */ - while (busy) { - busy = false; + while (waited) { + waited = false; for (ctx = aio_ctxs; ctx != NULL; ctx = ctx->next) { AioContext *aio_context = ctx->data; @@ -322,13 +319,9 @@ void bdrv_drain_all(void) aio_context_acquire(aio_context); for (bs = bdrv_first(&it); bs; bs = bdrv_next(&it)) { if (aio_context == bdrv_get_aio_context(bs)) { - if (bdrv_requests_pending(bs)) { - busy = true; - aio_poll(aio_context, busy); - } + waited |= bdrv_drain_recurse(bs); } } - busy |= aio_poll(aio_context, false); aio_context_release(aio_context); } } @@ -476,6 +469,28 @@ static bool tracked_request_overlaps(BdrvTrackedRequest *req, return true; } +void bdrv_inc_in_flight(BlockDriverState *bs) +{ + atomic_inc(&bs->in_flight); +} + +static void dummy_bh_cb(void *opaque) +{ +} + +void bdrv_wakeup(BlockDriverState *bs) +{ + if (bs->wakeup) { + aio_bh_schedule_oneshot(qemu_get_aio_context(), dummy_bh_cb, NULL); + } +} + +void bdrv_dec_in_flight(BlockDriverState *bs) +{ + atomic_dec(&bs->in_flight); + bdrv_wakeup(bs); +} + static bool coroutine_fn wait_serialising_requests(BdrvTrackedRequest *self) { BlockDriverState *bs = self->bs; @@ -583,13 +598,9 @@ static int bdrv_prwv_co(BdrvChild *child, int64_t offset, /* Fast-path if already in coroutine context */ bdrv_rw_co_entry(&rwco); } else { - AioContext *aio_context = bdrv_get_aio_context(child->bs); - co = qemu_coroutine_create(bdrv_rw_co_entry, &rwco); qemu_coroutine_enter(co); - while (rwco.ret == NOT_DONE) { - aio_poll(aio_context, true); - } + BDRV_POLL_WHILE(child->bs, rwco.ret == NOT_DONE); } return rwco.ret; } @@ -1097,6 +1108,8 @@ int coroutine_fn bdrv_co_preadv(BdrvChild *child, return ret; } + bdrv_inc_in_flight(bs); + /* Don't do copy-on-read if we read data before write operation */ if (bs->copy_on_read && !(flags & BDRV_REQ_NO_SERIALISING)) { flags |= BDRV_REQ_COPY_ON_READ; @@ -1132,6 +1145,7 @@ int coroutine_fn bdrv_co_preadv(BdrvChild *child, use_local_qiov ? &local_qiov : qiov, flags); tracked_request_end(&req); + bdrv_dec_in_flight(bs); if (use_local_qiov) { qemu_iovec_destroy(&local_qiov); @@ -1480,6 +1494,7 @@ int coroutine_fn bdrv_co_pwritev(BdrvChild *child, return ret; } + bdrv_inc_in_flight(bs); /* * Align write if necessary by performing a read-modify-write cycle. * Pad qiov with the read parts and be sure to have a tracked request not @@ -1581,6 +1596,7 @@ fail: qemu_vfree(tail_buf); out: tracked_request_end(&req); + bdrv_dec_in_flight(bs); return ret; } @@ -1705,17 +1721,19 @@ static int64_t coroutine_fn bdrv_co_get_block_status(BlockDriverState *bs, } *file = NULL; + bdrv_inc_in_flight(bs); ret = bs->drv->bdrv_co_get_block_status(bs, sector_num, nb_sectors, pnum, file); if (ret < 0) { *pnum = 0; - return ret; + goto out; } if (ret & BDRV_BLOCK_RAW) { assert(ret & BDRV_BLOCK_OFFSET_VALID); - return bdrv_get_block_status(bs->file->bs, ret >> BDRV_SECTOR_BITS, - *pnum, pnum, file); + ret = bdrv_get_block_status(bs->file->bs, ret >> BDRV_SECTOR_BITS, + *pnum, pnum, file); + goto out; } if (ret & (BDRV_BLOCK_DATA | BDRV_BLOCK_ZERO)) { @@ -1757,6 +1775,8 @@ static int64_t coroutine_fn bdrv_co_get_block_status(BlockDriverState *bs, } } +out: + bdrv_dec_in_flight(bs); return ret; } @@ -1822,14 +1842,10 @@ int64_t bdrv_get_block_status_above(BlockDriverState *bs, /* Fast-path if already in coroutine context */ bdrv_get_block_status_above_co_entry(&data); } else { - AioContext *aio_context = bdrv_get_aio_context(bs); - co = qemu_coroutine_create(bdrv_get_block_status_above_co_entry, &data); qemu_coroutine_enter(co); - while (!data.done) { - aio_poll(aio_context, true); - } + BDRV_POLL_WHILE(bs, !data.done); } return data.ret; } @@ -2102,6 +2118,7 @@ static const AIOCBInfo bdrv_em_co_aiocb_info = { static void bdrv_co_complete(BlockAIOCBCoroutine *acb) { if (!acb->need_bh) { + bdrv_dec_in_flight(acb->common.bs); acb->common.cb(acb->common.opaque, acb->req.error); qemu_aio_unref(acb); } @@ -2152,6 +2169,9 @@ static BlockAIOCB *bdrv_co_aio_prw_vector(BdrvChild *child, Coroutine *co; BlockAIOCBCoroutine *acb; + /* Matched by bdrv_co_complete's bdrv_dec_in_flight. */ + bdrv_inc_in_flight(child->bs); + acb = qemu_aio_get(&bdrv_em_co_aiocb_info, child->bs, cb, opaque); acb->child = child; acb->need_bh = true; @@ -2185,6 +2205,9 @@ BlockAIOCB *bdrv_aio_flush(BlockDriverState *bs, Coroutine *co; BlockAIOCBCoroutine *acb; + /* Matched by bdrv_co_complete's bdrv_dec_in_flight. */ + bdrv_inc_in_flight(bs); + acb = qemu_aio_get(&bdrv_em_co_aiocb_info, bs, cb, opaque); acb->need_bh = true; acb->req.error = -EINPROGRESS; @@ -2244,23 +2267,22 @@ static void coroutine_fn bdrv_flush_co_entry(void *opaque) int coroutine_fn bdrv_co_flush(BlockDriverState *bs) { int ret; - BdrvTrackedRequest req; if (!bs || !bdrv_is_inserted(bs) || bdrv_is_read_only(bs) || bdrv_is_sg(bs)) { return 0; } - tracked_request_begin(&req, bs, 0, 0, BDRV_TRACKED_FLUSH); + bdrv_inc_in_flight(bs); int current_gen = bs->write_gen; /* Wait until any previous flushes are completed */ - while (bs->active_flush_req != NULL) { + while (bs->active_flush_req) { qemu_co_queue_wait(&bs->flush_queue); } - bs->active_flush_req = &req; + bs->active_flush_req = true; /* Write back all layers by calling one driver function */ if (bs->drv->bdrv_co_flush) { @@ -2330,11 +2352,11 @@ flush_parent: out: /* Notify any pending flushes that we have completed */ bs->flushed_gen = current_gen; - bs->active_flush_req = NULL; + bs->active_flush_req = false; /* Return value is ignored - it's ok if wait queue is empty */ qemu_co_queue_next(&bs->flush_queue); - tracked_request_end(&req); + bdrv_dec_in_flight(bs); return ret; } @@ -2350,13 +2372,9 @@ int bdrv_flush(BlockDriverState *bs) /* Fast-path if already in coroutine context */ bdrv_flush_co_entry(&flush_co); } else { - AioContext *aio_context = bdrv_get_aio_context(bs); - co = qemu_coroutine_create(bdrv_flush_co_entry, &flush_co); qemu_coroutine_enter(co); - while (flush_co.ret == NOT_DONE) { - aio_poll(aio_context, true); - } + BDRV_POLL_WHILE(bs, flush_co.ret == NOT_DONE); } return flush_co.ret; @@ -2417,6 +2435,7 @@ int coroutine_fn bdrv_co_pdiscard(BlockDriverState *bs, int64_t offset, return 0; } + bdrv_inc_in_flight(bs); tracked_request_begin(&req, bs, offset, count, BDRV_TRACKED_DISCARD); ret = notifier_with_return_list_notify(&bs->before_write_notifiers, &req); @@ -2463,6 +2482,7 @@ out: bdrv_set_dirty(bs, req.offset >> BDRV_SECTOR_BITS, req.bytes >> BDRV_SECTOR_BITS); tracked_request_end(&req); + bdrv_dec_in_flight(bs); return ret; } @@ -2480,13 +2500,9 @@ int bdrv_pdiscard(BlockDriverState *bs, int64_t offset, int count) /* Fast-path if already in coroutine context */ bdrv_pdiscard_co_entry(&rwco); } else { - AioContext *aio_context = bdrv_get_aio_context(bs); - co = qemu_coroutine_create(bdrv_pdiscard_co_entry, &rwco); qemu_coroutine_enter(co); - while (rwco.ret == NOT_DONE) { - aio_poll(aio_context, true); - } + BDRV_POLL_WHILE(bs, rwco.ret == NOT_DONE); } return rwco.ret; @@ -2495,13 +2511,12 @@ int bdrv_pdiscard(BlockDriverState *bs, int64_t offset, int count) int bdrv_co_ioctl(BlockDriverState *bs, int req, void *buf) { BlockDriver *drv = bs->drv; - BdrvTrackedRequest tracked_req; CoroutineIOCompletion co = { .coroutine = qemu_coroutine_self(), }; BlockAIOCB *acb; - tracked_request_begin(&tracked_req, bs, 0, 0, BDRV_TRACKED_IOCTL); + bdrv_inc_in_flight(bs); if (!drv || (!drv->bdrv_aio_ioctl && !drv->bdrv_co_ioctl)) { co.ret = -ENOTSUP; goto out; @@ -2518,7 +2533,7 @@ int bdrv_co_ioctl(BlockDriverState *bs, int req, void *buf) qemu_coroutine_yield(); } out: - tracked_request_end(&tracked_req); + bdrv_dec_in_flight(bs); return co.ret; } diff --git a/block/mirror.c b/block/mirror.c index a433e6848c..3a0788ede3 100644 --- a/block/mirror.c +++ b/block/mirror.c @@ -469,7 +469,11 @@ static void mirror_free_init(MirrorBlockJob *s) } } -static void mirror_drain(MirrorBlockJob *s) +/* This is also used for the .pause callback. There is no matching + * mirror_resume() because mirror_run() will begin iterating again + * when the job is resumed. + */ +static void mirror_wait_for_all_io(MirrorBlockJob *s) { while (s->in_flight > 0) { mirror_wait_for_io(s); @@ -528,6 +532,7 @@ static void mirror_exit(BlockJob *job, void *opaque) g_free(s->replaces); bdrv_op_unblock_all(target_bs, s->common.blocker); blk_unref(s->target); + s->target = NULL; block_job_completed(&s->common, data->ret); g_free(data); bdrv_drained_end(src); @@ -582,7 +587,7 @@ static int coroutine_fn mirror_dirty_init(MirrorBlockJob *s) sector_num += nb_sectors; } - mirror_drain(s); + mirror_wait_for_all_io(s); } /* First part, loop on the sectors and initialize the dirty bitmap. */ @@ -617,6 +622,7 @@ static void coroutine_fn mirror_run(void *opaque) MirrorExitData *data; BlockDriverState *bs = blk_bs(s->common.blk); BlockDriverState *target_bs = blk_bs(s->target); + bool need_drain = true; int64_t length; BlockDriverInfo bdi; char backing_filename[2]; /* we only need 2 characters because we are only @@ -752,11 +758,26 @@ static void coroutine_fn mirror_run(void *opaque) * source has dirty data to copy! * * Note that I/O can be submitted by the guest while - * mirror_populate runs. + * mirror_populate runs, so pause it now. Before deciding + * whether to switch to target check one last time if I/O has + * come in the meanwhile, and if not flush the data to disk. */ trace_mirror_before_drain(s, cnt); - bdrv_co_drain(bs); + + bdrv_drained_begin(bs); cnt = bdrv_get_dirty_count(s->dirty_bitmap); + if (cnt > 0) { + bdrv_drained_end(bs); + continue; + } + + /* The two disks are in sync. Exit and report successful + * completion. + */ + assert(QLIST_EMPTY(&bs->tracked_requests)); + s->common.cancelled = false; + need_drain = false; + break; } ret = 0; @@ -769,13 +790,6 @@ static void coroutine_fn mirror_run(void *opaque) } else if (!should_complete) { delay_ns = (s->in_flight == 0 && cnt == 0 ? SLICE_TIME : 0); block_job_sleep_ns(&s->common, QEMU_CLOCK_REALTIME, delay_ns); - } else if (cnt == 0) { - /* The two disks are in sync. Exit and report successful - * completion. - */ - assert(QLIST_EMPTY(&bs->tracked_requests)); - s->common.cancelled = false; - break; } s->last_pause_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME); } @@ -787,7 +801,8 @@ immediate_exit: * the target is a copy of the source. */ assert(ret < 0 || (!s->synced && block_job_is_cancelled(&s->common))); - mirror_drain(s); + assert(need_drain); + mirror_wait_for_all_io(s); } assert(s->in_flight == 0); @@ -799,9 +814,10 @@ immediate_exit: data = g_malloc(sizeof(*data)); data->ret = ret; - /* Before we switch to target in mirror_exit, make sure data doesn't - * change. */ - bdrv_drained_begin(bs); + + if (need_drain) { + bdrv_drained_begin(bs); + } block_job_defer_to_main_loop(&s->common, mirror_exit, data); } @@ -872,14 +888,11 @@ static void mirror_complete(BlockJob *job, Error **errp) block_job_enter(&s->common); } -/* There is no matching mirror_resume() because mirror_run() will begin - * iterating again when the job is resumed. - */ -static void coroutine_fn mirror_pause(BlockJob *job) +static void mirror_pause(BlockJob *job) { MirrorBlockJob *s = container_of(job, MirrorBlockJob, common); - mirror_drain(s); + mirror_wait_for_all_io(s); } static void mirror_attached_aio_context(BlockJob *job, AioContext *new_context) @@ -889,6 +902,21 @@ static void mirror_attached_aio_context(BlockJob *job, AioContext *new_context) blk_set_aio_context(s->target, new_context); } +static void mirror_drain(BlockJob *job) +{ + MirrorBlockJob *s = container_of(job, MirrorBlockJob, common); + + /* Need to keep a reference in case blk_drain triggers execution + * of mirror_complete... + */ + if (s->target) { + BlockBackend *target = s->target; + blk_ref(target); + blk_drain(target); + blk_unref(target); + } +} + static const BlockJobDriver mirror_job_driver = { .instance_size = sizeof(MirrorBlockJob), .job_type = BLOCK_JOB_TYPE_MIRROR, @@ -896,6 +924,7 @@ static const BlockJobDriver mirror_job_driver = { .complete = mirror_complete, .pause = mirror_pause, .attached_aio_context = mirror_attached_aio_context, + .drain = mirror_drain, }; static const BlockJobDriver commit_active_job_driver = { @@ -905,6 +934,7 @@ static const BlockJobDriver commit_active_job_driver = { .complete = mirror_complete, .pause = mirror_pause, .attached_aio_context = mirror_attached_aio_context, + .drain = mirror_drain, }; static void mirror_start_job(const char *job_id, BlockDriverState *bs, diff --git a/block/nfs.c b/block/nfs.c index c3db2ec58d..88c60a9118 100644 --- a/block/nfs.c +++ b/block/nfs.c @@ -52,6 +52,7 @@ typedef struct NFSClient { } NFSClient; typedef struct NFSRPC { + BlockDriverState *bs; int ret; int complete; QEMUIOVector *iov; @@ -90,11 +91,12 @@ static void nfs_process_write(void *arg) nfs_set_events(client); } -static void nfs_co_init_task(NFSClient *client, NFSRPC *task) +static void nfs_co_init_task(BlockDriverState *bs, NFSRPC *task) { *task = (NFSRPC) { .co = qemu_coroutine_self(), - .client = client, + .bs = bs, + .client = bs->opaque, }; } @@ -111,6 +113,7 @@ nfs_co_generic_cb(int ret, struct nfs_context *nfs, void *data, { NFSRPC *task = private_data; task->ret = ret; + assert(!task->st); if (task->ret > 0 && task->iov) { if (task->ret <= task->iov->size) { qemu_iovec_from_buf(task->iov, 0, data, task->ret); @@ -118,18 +121,11 @@ nfs_co_generic_cb(int ret, struct nfs_context *nfs, void *data, task->ret = -EIO; } } - if (task->ret == 0 && task->st) { - memcpy(task->st, data, sizeof(struct stat)); - } if (task->ret < 0) { error_report("NFS Error: %s", nfs_get_error(nfs)); } - if (task->co) { - aio_bh_schedule_oneshot(task->client->aio_context, - nfs_co_generic_bh_cb, task); - } else { - task->complete = 1; - } + aio_bh_schedule_oneshot(task->client->aio_context, + nfs_co_generic_bh_cb, task); } static int coroutine_fn nfs_co_readv(BlockDriverState *bs, @@ -139,7 +135,7 @@ static int coroutine_fn nfs_co_readv(BlockDriverState *bs, NFSClient *client = bs->opaque; NFSRPC task; - nfs_co_init_task(client, &task); + nfs_co_init_task(bs, &task); task.iov = iov; if (nfs_pread_async(client->context, client->fh, @@ -149,8 +145,8 @@ static int coroutine_fn nfs_co_readv(BlockDriverState *bs, return -ENOMEM; } + nfs_set_events(client); while (!task.complete) { - nfs_set_events(client); qemu_coroutine_yield(); } @@ -174,7 +170,7 @@ static int coroutine_fn nfs_co_writev(BlockDriverState *bs, NFSRPC task; char *buf = NULL; - nfs_co_init_task(client, &task); + nfs_co_init_task(bs, &task); buf = g_try_malloc(nb_sectors * BDRV_SECTOR_SIZE); if (nb_sectors && buf == NULL) { @@ -191,8 +187,8 @@ static int coroutine_fn nfs_co_writev(BlockDriverState *bs, return -ENOMEM; } + nfs_set_events(client); while (!task.complete) { - nfs_set_events(client); qemu_coroutine_yield(); } @@ -210,15 +206,15 @@ static int coroutine_fn nfs_co_flush(BlockDriverState *bs) NFSClient *client = bs->opaque; NFSRPC task; - nfs_co_init_task(client, &task); + nfs_co_init_task(bs, &task); if (nfs_fsync_async(client->context, client->fh, nfs_co_generic_cb, &task) != 0) { return -ENOMEM; } + nfs_set_events(client); while (!task.complete) { - nfs_set_events(client); qemu_coroutine_yield(); } @@ -496,6 +492,22 @@ static int nfs_has_zero_init(BlockDriverState *bs) return client->has_zero_init; } +static void +nfs_get_allocated_file_size_cb(int ret, struct nfs_context *nfs, void *data, + void *private_data) +{ + NFSRPC *task = private_data; + task->ret = ret; + if (task->ret == 0) { + memcpy(task->st, data, sizeof(struct stat)); + } + if (task->ret < 0) { + error_report("NFS Error: %s", nfs_get_error(nfs)); + } + task->complete = 1; + bdrv_wakeup(task->bs); +} + static int64_t nfs_get_allocated_file_size(BlockDriverState *bs) { NFSClient *client = bs->opaque; @@ -507,16 +519,15 @@ static int64_t nfs_get_allocated_file_size(BlockDriverState *bs) return client->st_blocks * 512; } + task.bs = bs; task.st = &st; - if (nfs_fstat_async(client->context, client->fh, nfs_co_generic_cb, + if (nfs_fstat_async(client->context, client->fh, nfs_get_allocated_file_size_cb, &task) != 0) { return -ENOMEM; } - while (!task.complete) { - nfs_set_events(client); - aio_poll(client->aio_context, true); - } + nfs_set_events(client); + BDRV_POLL_WHILE(bs, !task.complete); return (task.ret < 0 ? task.ret : st.st_blocks * 512); } diff --git a/block/qed-table.c b/block/qed-table.c index 1a731dff51..ed443e2b70 100644 --- a/block/qed-table.c +++ b/block/qed-table.c @@ -174,9 +174,7 @@ int qed_read_l1_table_sync(BDRVQEDState *s) qed_read_table(s, s->header.l1_table_offset, s->l1_table, qed_sync_cb, &ret); - while (ret == -EINPROGRESS) { - aio_poll(bdrv_get_aio_context(s->bs), true); - } + BDRV_POLL_WHILE(s->bs, ret == -EINPROGRESS); return ret; } @@ -195,9 +193,7 @@ int qed_write_l1_table_sync(BDRVQEDState *s, unsigned int index, int ret = -EINPROGRESS; qed_write_l1_table(s, index, n, qed_sync_cb, &ret); - while (ret == -EINPROGRESS) { - aio_poll(bdrv_get_aio_context(s->bs), true); - } + BDRV_POLL_WHILE(s->bs, ret == -EINPROGRESS); return ret; } @@ -268,9 +264,7 @@ int qed_read_l2_table_sync(BDRVQEDState *s, QEDRequest *request, uint64_t offset int ret = -EINPROGRESS; qed_read_l2_table(s, request, offset, qed_sync_cb, &ret); - while (ret == -EINPROGRESS) { - aio_poll(bdrv_get_aio_context(s->bs), true); - } + BDRV_POLL_WHILE(s->bs, ret == -EINPROGRESS); return ret; } @@ -290,9 +284,7 @@ int qed_write_l2_table_sync(BDRVQEDState *s, QEDRequest *request, int ret = -EINPROGRESS; qed_write_l2_table(s, request, index, n, flush, qed_sync_cb, &ret); - while (ret == -EINPROGRESS) { - aio_poll(bdrv_get_aio_context(s->bs), true); - } + BDRV_POLL_WHILE(s->bs, ret == -EINPROGRESS); return ret; } diff --git a/block/qed.c b/block/qed.c index 3ee879b52e..1a7ef0a9ce 100644 --- a/block/qed.c +++ b/block/qed.c @@ -336,7 +336,7 @@ static void qed_need_check_timer_cb(void *opaque) qed_plug_allocating_write_reqs(s); /* Ensure writes are on disk before clearing flag */ - bdrv_aio_flush(s->bs, qed_clear_need_check, s); + bdrv_aio_flush(s->bs->file->bs, qed_clear_need_check, s); } static void qed_start_need_check_timer(BDRVQEDState *s) @@ -378,6 +378,19 @@ static void bdrv_qed_attach_aio_context(BlockDriverState *bs, } } +static void bdrv_qed_drain(BlockDriverState *bs) +{ + BDRVQEDState *s = bs->opaque; + + /* Fire the timer immediately in order to start doing I/O as soon as the + * header is flushed. + */ + if (s->need_check_timer && timer_pending(s->need_check_timer)) { + qed_cancel_need_check_timer(s); + qed_need_check_timer_cb(s); + } +} + static int bdrv_qed_open(BlockDriverState *bs, QDict *options, int flags, Error **errp) { @@ -1668,6 +1681,7 @@ static BlockDriver bdrv_qed = { .bdrv_check = bdrv_qed_check, .bdrv_detach_aio_context = bdrv_qed_detach_aio_context, .bdrv_attach_aio_context = bdrv_qed_attach_aio_context, + .bdrv_drain = bdrv_qed_drain, }; static void bdrv_qed_init(void) diff --git a/block/replication.c b/block/replication.c index 8bbfc8f870..02aeaaf7d0 100644 --- a/block/replication.c +++ b/block/replication.c @@ -138,6 +138,9 @@ static void replication_close(BlockDriverState *bs) if (s->replication_state == BLOCK_REPLICATION_RUNNING) { replication_stop(s->rs, false, NULL); } + if (s->replication_state == BLOCK_REPLICATION_FAILOVER) { + block_job_cancel_sync(s->active_disk->bs->job); + } if (s->mode == REPLICATION_MODE_SECONDARY) { g_free(s->top_id); @@ -319,9 +322,10 @@ static void secondary_do_checkpoint(BDRVReplicationState *s, Error **errp) } } -static void reopen_backing_file(BDRVReplicationState *s, bool writable, +static void reopen_backing_file(BlockDriverState *bs, bool writable, Error **errp) { + BDRVReplicationState *s = bs->opaque; BlockReopenQueue *reopen_queue = NULL; int orig_hidden_flags, orig_secondary_flags; int new_hidden_flags, new_secondary_flags; @@ -356,13 +360,15 @@ static void reopen_backing_file(BDRVReplicationState *s, bool writable, } if (reopen_queue) { - bdrv_reopen_multiple(reopen_queue, &local_err); + bdrv_reopen_multiple(bdrv_get_aio_context(bs), + reopen_queue, &local_err); error_propagate(errp, local_err); } } -static void backup_job_cleanup(BDRVReplicationState *s) +static void backup_job_cleanup(BlockDriverState *bs) { + BDRVReplicationState *s = bs->opaque; BlockDriverState *top_bs; top_bs = bdrv_lookup_bs(s->top_id, s->top_id, NULL); @@ -371,19 +377,20 @@ static void backup_job_cleanup(BDRVReplicationState *s) } bdrv_op_unblock_all(top_bs, s->blocker); error_free(s->blocker); - reopen_backing_file(s, false, NULL); + reopen_backing_file(bs, false, NULL); } static void backup_job_completed(void *opaque, int ret) { - BDRVReplicationState *s = opaque; + BlockDriverState *bs = opaque; + BDRVReplicationState *s = bs->opaque; if (s->replication_state != BLOCK_REPLICATION_FAILOVER) { /* The backup job is cancelled unexpectedly */ s->error = -EIO; } - backup_job_cleanup(s); + backup_job_cleanup(bs); } static bool check_top_bs(BlockDriverState *top_bs, BlockDriverState *bs) @@ -479,7 +486,7 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode, } /* reopen the backing file in r/w mode */ - reopen_backing_file(s, true, &local_err); + reopen_backing_file(bs, true, &local_err); if (local_err) { error_propagate(errp, local_err); aio_context_release(aio_context); @@ -494,7 +501,7 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode, if (!top_bs || !bdrv_is_root_node(top_bs) || !check_top_bs(top_bs, bs)) { error_setg(errp, "No top_bs or it is invalid"); - reopen_backing_file(s, false, NULL); + reopen_backing_file(bs, false, NULL); aio_context_release(aio_context); return; } @@ -504,10 +511,10 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode, backup_start("replication-backup", s->secondary_disk->bs, s->hidden_disk->bs, 0, MIRROR_SYNC_MODE_NONE, NULL, false, BLOCKDEV_ON_ERROR_REPORT, BLOCKDEV_ON_ERROR_REPORT, - backup_job_completed, s, NULL, &local_err); + backup_job_completed, bs, NULL, &local_err); if (local_err) { error_propagate(errp, local_err); - backup_job_cleanup(s); + backup_job_cleanup(bs); aio_context_release(aio_context); return; } diff --git a/block/sheepdog.c b/block/sheepdog.c index ccbf7e1fa6..1fb917343a 100644 --- a/block/sheepdog.c +++ b/block/sheepdog.c @@ -641,6 +641,7 @@ static void restart_co_req(void *opaque) typedef struct SheepdogReqCo { int sockfd; + BlockDriverState *bs; AioContext *aio_context; SheepdogReq *hdr; void *data; @@ -701,6 +702,9 @@ out: srco->ret = ret; srco->finished = true; + if (srco->bs) { + bdrv_wakeup(srco->bs); + } } /* @@ -708,13 +712,14 @@ out: * * Return 0 on success, -errno in case of error. */ -static int do_req(int sockfd, AioContext *aio_context, SheepdogReq *hdr, +static int do_req(int sockfd, BlockDriverState *bs, SheepdogReq *hdr, void *data, unsigned int *wlen, unsigned int *rlen) { Coroutine *co; SheepdogReqCo srco = { .sockfd = sockfd, - .aio_context = aio_context, + .aio_context = bs ? bdrv_get_aio_context(bs) : qemu_get_aio_context(), + .bs = bs, .hdr = hdr, .data = data, .wlen = wlen, @@ -727,9 +732,14 @@ static int do_req(int sockfd, AioContext *aio_context, SheepdogReq *hdr, do_co_req(&srco); } else { co = qemu_coroutine_create(do_co_req, &srco); - qemu_coroutine_enter(co); - while (!srco.finished) { - aio_poll(aio_context, true); + if (bs) { + qemu_coroutine_enter(co); + BDRV_POLL_WHILE(bs, !srco.finished); + } else { + qemu_coroutine_enter(co); + while (!srco.finished) { + aio_poll(qemu_get_aio_context(), true); + } } } @@ -1125,7 +1135,7 @@ static int find_vdi_name(BDRVSheepdogState *s, const char *filename, hdr.snapid = snapid; hdr.flags = SD_FLAG_CMD_WRITE; - ret = do_req(fd, s->aio_context, (SheepdogReq *)&hdr, buf, &wlen, &rlen); + ret = do_req(fd, s->bs, (SheepdogReq *)&hdr, buf, &wlen, &rlen); if (ret) { error_setg_errno(errp, -ret, "cannot get vdi info"); goto out; @@ -1240,7 +1250,7 @@ out: qemu_co_mutex_unlock(&s->lock); } -static int read_write_object(int fd, AioContext *aio_context, char *buf, +static int read_write_object(int fd, BlockDriverState *bs, char *buf, uint64_t oid, uint8_t copies, unsigned int datalen, uint64_t offset, bool write, bool create, uint32_t cache_flags) @@ -1274,7 +1284,7 @@ static int read_write_object(int fd, AioContext *aio_context, char *buf, hdr.offset = offset; hdr.copies = copies; - ret = do_req(fd, aio_context, (SheepdogReq *)&hdr, buf, &wlen, &rlen); + ret = do_req(fd, bs, (SheepdogReq *)&hdr, buf, &wlen, &rlen); if (ret) { error_report("failed to send a request to the sheep"); return ret; @@ -1289,22 +1299,22 @@ static int read_write_object(int fd, AioContext *aio_context, char *buf, } } -static int read_object(int fd, AioContext *aio_context, char *buf, +static int read_object(int fd, BlockDriverState *bs, char *buf, uint64_t oid, uint8_t copies, unsigned int datalen, uint64_t offset, uint32_t cache_flags) { - return read_write_object(fd, aio_context, buf, oid, copies, + return read_write_object(fd, bs, buf, oid, copies, datalen, offset, false, false, cache_flags); } -static int write_object(int fd, AioContext *aio_context, char *buf, +static int write_object(int fd, BlockDriverState *bs, char *buf, uint64_t oid, uint8_t copies, unsigned int datalen, uint64_t offset, bool create, uint32_t cache_flags) { - return read_write_object(fd, aio_context, buf, oid, copies, + return read_write_object(fd, bs, buf, oid, copies, datalen, offset, true, create, cache_flags); } @@ -1331,7 +1341,7 @@ static int reload_inode(BDRVSheepdogState *s, uint32_t snapid, const char *tag) goto out; } - ret = read_object(fd, s->aio_context, (char *)inode, vid_to_vdi_oid(vid), + ret = read_object(fd, s->bs, (char *)inode, vid_to_vdi_oid(vid), s->inode.nr_copies, SD_INODE_HEADER_SIZE, 0, s->cache_flags); if (ret < 0) { @@ -1489,7 +1499,7 @@ static int sd_open(BlockDriverState *bs, QDict *options, int flags, } buf = g_malloc(SD_INODE_SIZE); - ret = read_object(fd, s->aio_context, buf, vid_to_vdi_oid(vid), + ret = read_object(fd, s->bs, buf, vid_to_vdi_oid(vid), 0, SD_INODE_SIZE, 0, s->cache_flags); closesocket(fd); @@ -1618,7 +1628,7 @@ static int do_sd_create(BDRVSheepdogState *s, uint32_t *vdi_id, int snapshot, hdr.copies = s->inode.nr_copies; hdr.block_size_shift = s->inode.block_size_shift; - ret = do_req(fd, s->aio_context, (SheepdogReq *)&hdr, buf, &wlen, &rlen); + ret = do_req(fd, NULL, (SheepdogReq *)&hdr, buf, &wlen, &rlen); closesocket(fd); @@ -1886,7 +1896,7 @@ static int sd_create(const char *filename, QemuOpts *opts, hdr.opcode = SD_OP_GET_CLUSTER_DEFAULT; hdr.proto_ver = SD_PROTO_VER; - ret = do_req(fd, s->aio_context, (SheepdogReq *)&hdr, + ret = do_req(fd, NULL, (SheepdogReq *)&hdr, NULL, &wlen, &rlen); closesocket(fd); if (ret) { @@ -1951,7 +1961,7 @@ static void sd_close(BlockDriverState *bs) hdr.data_length = wlen; hdr.flags = SD_FLAG_CMD_WRITE; - ret = do_req(fd, s->aio_context, (SheepdogReq *)&hdr, + ret = do_req(fd, s->bs, (SheepdogReq *)&hdr, s->name, &wlen, &rlen); closesocket(fd); @@ -2000,7 +2010,7 @@ static int sd_truncate(BlockDriverState *bs, int64_t offset) /* we don't need to update entire object */ datalen = SD_INODE_SIZE - sizeof(s->inode.data_vdi_id); s->inode.vdi_size = offset; - ret = write_object(fd, s->aio_context, (char *)&s->inode, + ret = write_object(fd, s->bs, (char *)&s->inode, vid_to_vdi_oid(s->inode.vdi_id), s->inode.nr_copies, datalen, 0, false, s->cache_flags); close(fd); @@ -2070,7 +2080,7 @@ static bool sd_delete(BDRVSheepdogState *s) return false; } - ret = do_req(fd, s->aio_context, (SheepdogReq *)&hdr, + ret = do_req(fd, s->bs, (SheepdogReq *)&hdr, s->name, &wlen, &rlen); closesocket(fd); if (ret) { @@ -2126,7 +2136,7 @@ static int sd_create_branch(BDRVSheepdogState *s) goto out; } - ret = read_object(fd, s->aio_context, buf, vid_to_vdi_oid(vid), + ret = read_object(fd, s->bs, buf, vid_to_vdi_oid(vid), s->inode.nr_copies, SD_INODE_SIZE, 0, s->cache_flags); closesocket(fd); @@ -2411,7 +2421,7 @@ static int sd_snapshot_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info) goto cleanup; } - ret = write_object(fd, s->aio_context, (char *)&s->inode, + ret = write_object(fd, s->bs, (char *)&s->inode, vid_to_vdi_oid(s->inode.vdi_id), s->inode.nr_copies, datalen, 0, false, s->cache_flags); if (ret < 0) { @@ -2426,7 +2436,7 @@ static int sd_snapshot_create(BlockDriverState *bs, QEMUSnapshotInfo *sn_info) goto cleanup; } - ret = read_object(fd, s->aio_context, (char *)inode, + ret = read_object(fd, s->bs, (char *)inode, vid_to_vdi_oid(new_vid), s->inode.nr_copies, datalen, 0, s->cache_flags); @@ -2528,7 +2538,7 @@ static bool remove_objects(BDRVSheepdogState *s) i++; } - ret = write_object(fd, s->aio_context, + ret = write_object(fd, s->bs, (char *)&inode->data_vdi_id[start_idx], vid_to_vdi_oid(s->inode.vdi_id), inode->nr_copies, (i - start_idx) * sizeof(uint32_t), @@ -2600,7 +2610,7 @@ static int sd_snapshot_delete(BlockDriverState *bs, return -1; } - ret = do_req(fd, s->aio_context, (SheepdogReq *)&hdr, + ret = do_req(fd, s->bs, (SheepdogReq *)&hdr, buf, &wlen, &rlen); closesocket(fd); if (ret) { @@ -2652,8 +2662,7 @@ static int sd_snapshot_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab) req.opcode = SD_OP_READ_VDIS; req.data_length = max; - ret = do_req(fd, s->aio_context, &req, - vdi_inuse, &wlen, &rlen); + ret = do_req(fd, s->bs, &req, vdi_inuse, &wlen, &rlen); closesocket(fd); if (ret) { @@ -2679,7 +2688,7 @@ static int sd_snapshot_list(BlockDriverState *bs, QEMUSnapshotInfo **psn_tab) } /* we don't need to read entire object */ - ret = read_object(fd, s->aio_context, (char *)&inode, + ret = read_object(fd, s->bs, (char *)&inode, vid_to_vdi_oid(vid), 0, SD_INODE_SIZE - sizeof(inode.data_vdi_id), 0, s->cache_flags); @@ -2745,11 +2754,11 @@ static int do_load_save_vmstate(BDRVSheepdogState *s, uint8_t *data, create = (offset == 0); if (load) { - ret = read_object(fd, s->aio_context, (char *)data, vmstate_oid, + ret = read_object(fd, s->bs, (char *)data, vmstate_oid, s->inode.nr_copies, data_len, offset, s->cache_flags); } else { - ret = write_object(fd, s->aio_context, (char *)data, vmstate_oid, + ret = write_object(fd, s->bs, (char *)data, vmstate_oid, s->inode.nr_copies, data_len, offset, create, s->cache_flags); } |