[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
Re: [Qemu-block] [PATCH 4/7] blockjob: centralize QMP event emissions
From: |
Jeff Cody |
Subject: |
Re: [Qemu-block] [PATCH 4/7] blockjob: centralize QMP event emissions |
Date: |
Wed, 26 Oct 2016 00:49:48 -0400 |
User-agent: |
Mutt/1.5.24 (2015-08-30) |
On Thu, Oct 13, 2016 at 06:56:59PM -0400, John Snow wrote:
> There's no reason to leave this to blockdev; we can do it in blockjobs
> directly and get rid of an extra callback for most users.
>
> All non-internal events, even those created outside of QMP, will
> consistently emit events.
>
> Signed-off-by: John Snow <address@hidden>
> ---
> block/commit.c | 8 ++++----
> block/mirror.c | 6 ++----
> block/stream.c | 7 +++----
> block/trace-events | 5 ++---
> blockdev.c | 42 ++++++++----------------------------------
> blockjob.c | 23 +++++++++++++++++++----
> include/block/block_int.h | 17 ++++-------------
> include/block/blockjob.h | 17 -----------------
> 8 files changed, 42 insertions(+), 83 deletions(-)
>
> diff --git a/block/commit.c b/block/commit.c
> index f29e341..475a375 100644
> --- a/block/commit.c
> +++ b/block/commit.c
> @@ -209,8 +209,8 @@ static const BlockJobDriver commit_job_driver = {
>
> void commit_start(const char *job_id, BlockDriverState *bs,
> BlockDriverState *base, BlockDriverState *top, int64_t
> speed,
> - BlockdevOnError on_error, BlockCompletionFunc *cb,
> - void *opaque, const char *backing_file_str, Error **errp)
> + BlockdevOnError on_error, const char *backing_file_str,
> + Error **errp)
> {
> CommitBlockJob *s;
> BlockReopenQueue *reopen_queue = NULL;
> @@ -233,7 +233,7 @@ void commit_start(const char *job_id, BlockDriverState
> *bs,
> }
>
> s = block_job_create(job_id, &commit_job_driver, bs, speed,
> - BLOCK_JOB_DEFAULT, cb, opaque, errp);
> + BLOCK_JOB_DEFAULT, NULL, NULL, errp);
> if (!s) {
> return;
> }
> @@ -276,7 +276,7 @@ void commit_start(const char *job_id, BlockDriverState
> *bs,
> s->on_error = on_error;
> s->common.co = qemu_coroutine_create(commit_run, s);
>
> - trace_commit_start(bs, base, top, s, s->common.co, opaque);
> + trace_commit_start(bs, base, top, s, s->common.co);
> qemu_coroutine_enter(s->common.co);
> }
>
> diff --git a/block/mirror.c b/block/mirror.c
> index 15d2d10..4374fb4 100644
> --- a/block/mirror.c
> +++ b/block/mirror.c
> @@ -979,9 +979,7 @@ void mirror_start(const char *job_id, BlockDriverState
> *bs,
> MirrorSyncMode mode, BlockMirrorBackingMode backing_mode,
> BlockdevOnError on_source_error,
> BlockdevOnError on_target_error,
> - bool unmap,
> - BlockCompletionFunc *cb,
> - void *opaque, Error **errp)
> + bool unmap, Error **errp)
> {
> bool is_none_mode;
> BlockDriverState *base;
> @@ -994,7 +992,7 @@ void mirror_start(const char *job_id, BlockDriverState
> *bs,
> base = mode == MIRROR_SYNC_MODE_TOP ? backing_bs(bs) : NULL;
> mirror_start_job(job_id, bs, BLOCK_JOB_DEFAULT, target, replaces,
> speed, granularity, buf_size, backing_mode,
> - on_source_error, on_target_error, unmap, cb, opaque,
> errp,
> + on_source_error, on_target_error, unmap, NULL, NULL,
> errp,
> &mirror_job_driver, is_none_mode, base, false);
> }
>
> diff --git a/block/stream.c b/block/stream.c
> index eeb6f52..7d6877d 100644
> --- a/block/stream.c
> +++ b/block/stream.c
> @@ -216,13 +216,12 @@ static const BlockJobDriver stream_job_driver = {
>
> void stream_start(const char *job_id, BlockDriverState *bs,
> BlockDriverState *base, const char *backing_file_str,
> - int64_t speed, BlockdevOnError on_error,
> - BlockCompletionFunc *cb, void *opaque, Error **errp)
> + int64_t speed, BlockdevOnError on_error, Error **errp)
> {
> StreamBlockJob *s;
>
> s = block_job_create(job_id, &stream_job_driver, bs, speed,
> - BLOCK_JOB_DEFAULT, cb, opaque, errp);
> + BLOCK_JOB_DEFAULT, NULL, NULL, errp);
> if (!s) {
> return;
> }
> @@ -232,6 +231,6 @@ void stream_start(const char *job_id, BlockDriverState
> *bs,
>
> s->on_error = on_error;
> s->common.co = qemu_coroutine_create(stream_run, s);
> - trace_stream_start(bs, base, s, s->common.co, opaque);
> + trace_stream_start(bs, base, s, s->common.co);
> qemu_coroutine_enter(s->common.co);
> }
> diff --git a/block/trace-events b/block/trace-events
> index 05fa13c..c12f91b 100644
> --- a/block/trace-events
> +++ b/block/trace-events
> @@ -20,11 +20,11 @@ bdrv_co_do_copy_on_readv(void *bs, int64_t offset,
> unsigned int bytes, int64_t c
>
> # block/stream.c
> stream_one_iteration(void *s, int64_t sector_num, int nb_sectors, int
> is_allocated) "s %p sector_num %"PRId64" nb_sectors %d is_allocated %d"
> -stream_start(void *bs, void *base, void *s, void *co, void *opaque) "bs %p
> base %p s %p co %p opaque %p"
> +stream_start(void *bs, void *base, void *s, void *co) "bs %p base %p s %p co
> %p"
>
> # block/commit.c
> commit_one_iteration(void *s, int64_t sector_num, int nb_sectors, int
> is_allocated) "s %p sector_num %"PRId64" nb_sectors %d is_allocated %d"
> -commit_start(void *bs, void *base, void *top, void *s, void *co, void
> *opaque) "bs %p base %p top %p s %p co %p opaque %p"
> +commit_start(void *bs, void *base, void *top, void *s, void *co) "bs %p base
> %p top %p s %p co %p"
>
> # block/mirror.c
> mirror_start(void *bs, void *s, void *co, void *opaque) "bs %p s %p co %p
> opaque %p"
> @@ -52,7 +52,6 @@ qmp_block_job_cancel(void *job) "job %p"
> qmp_block_job_pause(void *job) "job %p"
> qmp_block_job_resume(void *job) "job %p"
> qmp_block_job_complete(void *job) "job %p"
> -block_job_cb(void *bs, void *job, int ret) "bs %p job %p ret %d"
> qmp_block_stream(void *bs, void *job) "bs %p job %p"
>
> # block/raw-win32.c
> diff --git a/blockdev.c b/blockdev.c
> index 0ce305c..22a1280 100644
> --- a/blockdev.c
> +++ b/blockdev.c
> @@ -2905,31 +2905,6 @@ out:
> aio_context_release(aio_context);
> }
>
> -static void block_job_cb(void *opaque, int ret)
> -{
> - /* Note that this function may be executed from another AioContext
> besides
> - * the QEMU main loop. If you need to access anything that assumes the
> - * QEMU global mutex, use a BH or introduce a mutex.
> - */
> -
> - BlockDriverState *bs = opaque;
> - const char *msg = NULL;
> -
> - trace_block_job_cb(bs, bs->job, ret);
> -
> - assert(bs->job);
> -
> - if (ret < 0) {
> - msg = strerror(-ret);
> - }
> -
> - if (block_job_is_cancelled(bs->job)) {
> - block_job_event_cancelled(bs->job);
> - } else {
> - block_job_event_completed(bs->job, msg);
> - }
> -}
> -
> void qmp_block_stream(bool has_job_id, const char *job_id, const char
> *device,
> bool has_base, const char *base,
> bool has_backing_file, const char *backing_file,
> @@ -2981,7 +2956,7 @@ void qmp_block_stream(bool has_job_id, const char
> *job_id, const char *device,
> base_name = has_backing_file ? backing_file : base_name;
>
> stream_start(has_job_id ? job_id : NULL, bs, base_bs, base_name,
> - has_speed ? speed : 0, on_error, block_job_cb, bs,
> &local_err);
> + has_speed ? speed : 0, on_error, &local_err);
> if (local_err) {
> error_propagate(errp, local_err);
> goto out;
> @@ -3084,12 +3059,12 @@ void qmp_block_commit(bool has_job_id, const char
> *job_id, const char *device,
> goto out;
> }
> commit_active_start(has_job_id ? job_id : NULL, bs, base_bs,
> - BLOCK_JOB_DEFAULT, speed, on_error, block_job_cb,
> - bs, &local_err, false);
> + BLOCK_JOB_DEFAULT, speed, on_error, NULL, NULL,
> + &local_err, false);
> } else {
> commit_start(has_job_id ? job_id : NULL, bs, base_bs, top_bs, speed,
> - on_error, block_job_cb, bs,
> - has_backing_file ? backing_file : NULL, &local_err);
> + on_error, has_backing_file ? backing_file : NULL,
> + &local_err);
> }
> if (local_err != NULL) {
> error_propagate(errp, local_err);
> @@ -3210,7 +3185,7 @@ static void do_drive_backup(DriveBackup *backup,
> BlockJobTxn *txn, Error **errp)
> backup_start(backup->job_id, bs, target_bs, backup->speed, backup->sync,
> bmap, backup->compress, backup->on_source_error,
> backup->on_target_error, BLOCK_JOB_DEFAULT,
> - block_job_cb, bs, txn, &local_err);
> + NULL, NULL, txn, &local_err);
> bdrv_unref(target_bs);
> if (local_err != NULL) {
> error_propagate(errp, local_err);
> @@ -3281,7 +3256,7 @@ void do_blockdev_backup(BlockdevBackup *backup,
> BlockJobTxn *txn, Error **errp)
> backup_start(backup->job_id, bs, target_bs, backup->speed, backup->sync,
> NULL, backup->compress, backup->on_source_error,
> backup->on_target_error, BLOCK_JOB_DEFAULT,
> - block_job_cb, bs, txn, &local_err);
> + NULL, NULL, txn, &local_err);
> if (local_err != NULL) {
> error_propagate(errp, local_err);
> }
> @@ -3360,8 +3335,7 @@ static void blockdev_mirror_common(const char *job_id,
> BlockDriverState *bs,
> mirror_start(job_id, bs, target,
> has_replaces ? replaces : NULL,
> speed, granularity, buf_size, sync, backing_mode,
> - on_source_error, on_target_error, unmap,
> - block_job_cb, bs, errp);
> + on_source_error, on_target_error, unmap, errp);
> }
>
> void qmp_drive_mirror(DriveMirror *arg, Error **errp)
> diff --git a/blockjob.c b/blockjob.c
> index 017905a..e32cb78 100644
> --- a/blockjob.c
> +++ b/blockjob.c
> @@ -38,6 +38,9 @@
> #include "qemu/timer.h"
> #include "qapi-event.h"
>
> +static void block_job_event_cancelled(BlockJob *job);
> +static void block_job_event_completed(BlockJob *job, const char *msg);
> +
> /* Transactional group of block jobs */
> struct BlockJobTxn {
>
> @@ -124,7 +127,6 @@ void *block_job_create(const char *job_id, const
> BlockJobDriver *driver,
> BlockBackend *blk;
> BlockJob *job;
>
> - assert(cb);
> if (bs->job) {
> error_setg(errp, QERR_DEVICE_IN_USE, bdrv_get_device_name(bs));
> return NULL;
> @@ -230,7 +232,20 @@ static void block_job_completed_single(BlockJob *job)
> job->driver->abort(job);
> }
> }
> - job->cb(job->opaque, job->ret);
> +
> + if (job->cb) {
> + job->cb(job->opaque, job->ret);
> + }
> + if (block_job_is_cancelled(job)) {
> + block_job_event_cancelled(job);
> + } else {
> + const char *msg = NULL;
> + if (job->ret < 0) {
> + msg = strerror(-job->ret);
> + }
> + block_job_event_completed(job, msg);
> + }
> +
> if (job->txn) {
> block_job_txn_unref(job->txn);
> }
> @@ -535,7 +550,7 @@ static void block_job_iostatus_set_err(BlockJob *job, int
> error)
> }
> }
>
> -void block_job_event_cancelled(BlockJob *job)
> +static void block_job_event_cancelled(BlockJob *job)
> {
> if (block_job_is_internal(job)) {
> return;
> @@ -549,7 +564,7 @@ void block_job_event_cancelled(BlockJob *job)
> &error_abort);
> }
>
> -void block_job_event_completed(BlockJob *job, const char *msg)
> +static void block_job_event_completed(BlockJob *job, const char *msg)
> {
> if (block_job_is_internal(job)) {
> return;
> diff --git a/include/block/block_int.h b/include/block/block_int.h
> index 98f1c7f..dfbc53d 100644
> --- a/include/block/block_int.h
> +++ b/include/block/block_int.h
> @@ -647,8 +647,6 @@ int is_windows_drive(const char *filename);
> * the new backing file if the job completes. Ignored if @base is %NULL.
> * @speed: The maximum speed, in bytes per second, or 0 for unlimited.
> * @on_error: The action to take upon error.
> - * @cb: Completion function for the job.
> - * @opaque: Opaque pointer value passed to @cb.
> * @errp: Error object.
> *
> * Start a streaming operation on @bs. Clusters that are unallocated
> @@ -660,8 +658,7 @@ int is_windows_drive(const char *filename);
> */
> void stream_start(const char *job_id, BlockDriverState *bs,
> BlockDriverState *base, const char *backing_file_str,
> - int64_t speed, BlockdevOnError on_error,
> - BlockCompletionFunc *cb, void *opaque, Error **errp);
> + int64_t speed, BlockdevOnError on_error, Error **errp);
>
> /**
> * commit_start:
> @@ -672,16 +669,14 @@ void stream_start(const char *job_id, BlockDriverState
> *bs,
> * @base: Block device that will be written into, and become the new top.
> * @speed: The maximum speed, in bytes per second, or 0 for unlimited.
> * @on_error: The action to take upon error.
> - * @cb: Completion function for the job.
> - * @opaque: Opaque pointer value passed to @cb.
> * @backing_file_str: String to use as the backing file in @top's overlay
> * @errp: Error object.
> *
> */
> void commit_start(const char *job_id, BlockDriverState *bs,
> BlockDriverState *base, BlockDriverState *top, int64_t
> speed,
> - BlockdevOnError on_error, BlockCompletionFunc *cb,
> - void *opaque, const char *backing_file_str, Error **errp);
> + BlockdevOnError on_error, const char *backing_file_str,
> + Error **errp);
> /**
> * commit_active_start:
> * @job_id: The id of the newly-created job, or %NULL to use the
> @@ -719,8 +714,6 @@ void commit_active_start(const char *job_id,
> BlockDriverState *bs,
> * @on_source_error: The action to take upon error reading from the source.
> * @on_target_error: The action to take upon error writing to the target.
> * @unmap: Whether to unmap target where source sectors only contain zeroes.
> - * @cb: Completion function for the job.
> - * @opaque: Opaque pointer value passed to @cb.
> * @errp: Error object.
> *
> * Start a mirroring operation on @bs. Clusters that are allocated
> @@ -734,9 +727,7 @@ void mirror_start(const char *job_id, BlockDriverState
> *bs,
> MirrorSyncMode mode, BlockMirrorBackingMode backing_mode,
> BlockdevOnError on_source_error,
> BlockdevOnError on_target_error,
> - bool unmap,
> - BlockCompletionFunc *cb,
> - void *opaque, Error **errp);
> + bool unmap, Error **errp);
>
> /*
> * backup_start:
> diff --git a/include/block/blockjob.h b/include/block/blockjob.h
> index fdb31e0..928f0b8 100644
> --- a/include/block/blockjob.h
> +++ b/include/block/blockjob.h
> @@ -374,23 +374,6 @@ void block_job_resume(BlockJob *job);
> void block_job_enter(BlockJob *job);
>
> /**
> - * block_job_event_cancelled:
> - * @job: The job whose information is requested.
> - *
> - * Send a BLOCK_JOB_CANCELLED event for the specified job.
> - */
> -void block_job_event_cancelled(BlockJob *job);
> -
> -/**
> - * block_job_ready:
> - * @job: The job which is now ready to complete.
> - * @msg: Error message. Only present on failure.
> - *
> - * Send a BLOCK_JOB_COMPLETED event for the specified job.
> - */
> -void block_job_event_completed(BlockJob *job, const char *msg);
> -
> -/**
> * block_job_ready:
> * @job: The job which is now ready to complete.
> *
> --
> 2.7.4
>
Reviewed-by: Jeff Cody <address@hidden>
- Re: [Qemu-block] [PATCH 7/7] blockjobs: fix documentation, (continued)
- [Qemu-block] [PATCH 5/7] Blockjobs: Internalize user_pause logic, John Snow, 2016/10/13
- [Qemu-block] [PATCH 3/7] Replication/Blockjobs: Create replication jobs as internal, John Snow, 2016/10/13
- [Qemu-block] [PATCH 4/7] blockjob: centralize QMP event emissions, John Snow, 2016/10/13
- [Qemu-block] [PATCH 6/7] blockjobs: split interface into public/private, Part 1, John Snow, 2016/10/13
- Re: [Qemu-block] [Qemu-devel] [PATCH 0/7] blockjobs: preliminary refactoring work, Pt 1, no-reply, 2016/10/14
- Re: [Qemu-block] [PATCH 0/7] blockjobs: preliminary refactoring work, Pt 1, John Snow, 2016/10/14