blockjob: centralize QMP event emissions
[qemu/kevin.git] / blockjob.c
blob309ef9a35c6a77933aeb9e97f9f7cf81a7df427d
1 /*
2 * QEMU System Emulator block driver
4 * Copyright (c) 2011 IBM Corp.
5 * Copyright (c) 2012 Red Hat, Inc.
7 * Permission is hereby granted, free of charge, to any person obtaining a copy
8 * of this software and associated documentation files (the "Software"), to deal
9 * in the Software without restriction, including without limitation the rights
10 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11 * copies of the Software, and to permit persons to whom the Software is
12 * furnished to do so, subject to the following conditions:
14 * The above copyright notice and this permission notice shall be included in
15 * all copies or substantial portions of the Software.
17 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
20 * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
23 * THE SOFTWARE.
26 #include "qemu/osdep.h"
27 #include "qemu-common.h"
28 #include "trace.h"
29 #include "block/block.h"
30 #include "block/blockjob.h"
31 #include "block/block_int.h"
32 #include "sysemu/block-backend.h"
33 #include "qapi/qmp/qerror.h"
34 #include "qapi/qmp/qjson.h"
35 #include "qemu/coroutine.h"
36 #include "qemu/id.h"
37 #include "qmp-commands.h"
38 #include "qemu/timer.h"
39 #include "qapi-event.h"
41 static void block_job_event_cancelled(BlockJob *job);
42 static void block_job_event_completed(BlockJob *job, const char *msg);
44 /* Transactional group of block jobs */
45 struct BlockJobTxn {
47 /* Is this txn being cancelled? */
48 bool aborting;
50 /* List of jobs */
51 QLIST_HEAD(, BlockJob) jobs;
53 /* Reference count */
54 int refcnt;
57 static QLIST_HEAD(, BlockJob) block_jobs = QLIST_HEAD_INITIALIZER(block_jobs);
59 BlockJob *block_job_next(BlockJob *job)
61 if (!job) {
62 return QLIST_FIRST(&block_jobs);
64 return QLIST_NEXT(job, job_list);
67 BlockJob *block_job_get(const char *id)
69 BlockJob *job;
71 QLIST_FOREACH(job, &block_jobs, job_list) {
72 if (job->id && !strcmp(id, job->id)) {
73 return job;
77 return NULL;
80 static void block_job_attached_aio_context(AioContext *new_context,
81 void *opaque)
83 BlockJob *job = opaque;
85 if (job->driver->attached_aio_context) {
86 job->driver->attached_aio_context(job, new_context);
89 block_job_resume(job);
92 static void block_job_drain(BlockJob *job)
94 /* If job is !job->busy this kicks it into the next pause point. */
95 block_job_enter(job);
97 blk_drain(job->blk);
98 if (job->driver->drain) {
99 job->driver->drain(job);
103 static void block_job_detach_aio_context(void *opaque)
105 BlockJob *job = opaque;
107 /* In case the job terminates during aio_poll()... */
108 block_job_ref(job);
110 block_job_pause(job);
112 while (!job->paused && !job->completed) {
113 block_job_drain(job);
116 block_job_unref(job);
119 void block_job_add_bdrv(BlockJob *job, BlockDriverState *bs)
121 job->nodes = g_slist_prepend(job->nodes, bs);
122 bdrv_ref(bs);
123 bdrv_op_block_all(bs, job->blocker);
126 void *block_job_create(const char *job_id, const BlockJobDriver *driver,
127 BlockDriverState *bs, int64_t speed, int flags,
128 BlockCompletionFunc *cb, void *opaque, Error **errp)
130 BlockBackend *blk;
131 BlockJob *job;
133 if (bs->job) {
134 error_setg(errp, QERR_DEVICE_IN_USE, bdrv_get_device_name(bs));
135 return NULL;
138 if (job_id == NULL && !(flags & BLOCK_JOB_INTERNAL)) {
139 job_id = bdrv_get_device_name(bs);
140 if (!*job_id) {
141 error_setg(errp, "An explicit job ID is required for this node");
142 return NULL;
146 if (job_id) {
147 if (flags & BLOCK_JOB_INTERNAL) {
148 error_setg(errp, "Cannot specify job ID for internal block job");
149 return NULL;
152 if (!id_wellformed(job_id)) {
153 error_setg(errp, "Invalid job ID '%s'", job_id);
154 return NULL;
157 if (block_job_get(job_id)) {
158 error_setg(errp, "Job ID '%s' already in use", job_id);
159 return NULL;
163 blk = blk_new();
164 blk_insert_bs(blk, bs);
166 job = g_malloc0(driver->instance_size);
167 error_setg(&job->blocker, "block device is in use by block job: %s",
168 BlockJobType_lookup[driver->job_type]);
169 block_job_add_bdrv(job, bs);
170 bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker);
172 job->driver = driver;
173 job->id = g_strdup(job_id);
174 job->blk = blk;
175 job->cb = cb;
176 job->opaque = opaque;
177 job->busy = true;
178 job->refcnt = 1;
179 bs->job = job;
181 QLIST_INSERT_HEAD(&block_jobs, job, job_list);
183 blk_add_aio_context_notifier(blk, block_job_attached_aio_context,
184 block_job_detach_aio_context, job);
186 /* Only set speed when necessary to avoid NotSupported error */
187 if (speed != 0) {
188 Error *local_err = NULL;
190 block_job_set_speed(job, speed, &local_err);
191 if (local_err) {
192 block_job_unref(job);
193 error_propagate(errp, local_err);
194 return NULL;
197 return job;
200 bool block_job_is_internal(BlockJob *job)
202 return (job->id == NULL);
205 void block_job_ref(BlockJob *job)
207 ++job->refcnt;
210 void block_job_unref(BlockJob *job)
212 if (--job->refcnt == 0) {
213 GSList *l;
214 BlockDriverState *bs = blk_bs(job->blk);
215 bs->job = NULL;
216 for (l = job->nodes; l; l = l->next) {
217 bs = l->data;
218 bdrv_op_unblock_all(bs, job->blocker);
219 bdrv_unref(bs);
221 g_slist_free(job->nodes);
222 blk_remove_aio_context_notifier(job->blk,
223 block_job_attached_aio_context,
224 block_job_detach_aio_context, job);
225 blk_unref(job->blk);
226 error_free(job->blocker);
227 g_free(job->id);
228 QLIST_REMOVE(job, job_list);
229 g_free(job);
233 static void block_job_completed_single(BlockJob *job)
235 if (!job->ret) {
236 if (job->driver->commit) {
237 job->driver->commit(job);
239 } else {
240 if (job->driver->abort) {
241 job->driver->abort(job);
245 if (job->cb) {
246 job->cb(job->opaque, job->ret);
248 if (block_job_is_cancelled(job)) {
249 block_job_event_cancelled(job);
250 } else {
251 const char *msg = NULL;
252 if (job->ret < 0) {
253 msg = strerror(-job->ret);
255 block_job_event_completed(job, msg);
258 if (job->txn) {
259 block_job_txn_unref(job->txn);
261 block_job_unref(job);
264 static void block_job_completed_txn_abort(BlockJob *job)
266 AioContext *ctx;
267 BlockJobTxn *txn = job->txn;
268 BlockJob *other_job, *next;
270 if (txn->aborting) {
272 * We are cancelled by another job, which will handle everything.
274 return;
276 txn->aborting = true;
277 /* We are the first failed job. Cancel other jobs. */
278 QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
279 ctx = blk_get_aio_context(other_job->blk);
280 aio_context_acquire(ctx);
282 QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
283 if (other_job == job || other_job->completed) {
284 /* Other jobs are "effectively" cancelled by us, set the status for
285 * them; this job, however, may or may not be cancelled, depending
286 * on the caller, so leave it. */
287 if (other_job != job) {
288 other_job->cancelled = true;
290 continue;
292 block_job_cancel_sync(other_job);
293 assert(other_job->completed);
295 QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
296 ctx = blk_get_aio_context(other_job->blk);
297 block_job_completed_single(other_job);
298 aio_context_release(ctx);
302 static void block_job_completed_txn_success(BlockJob *job)
304 AioContext *ctx;
305 BlockJobTxn *txn = job->txn;
306 BlockJob *other_job, *next;
308 * Successful completion, see if there are other running jobs in this
309 * txn.
311 QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
312 if (!other_job->completed) {
313 return;
316 /* We are the last completed job, commit the transaction. */
317 QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
318 ctx = blk_get_aio_context(other_job->blk);
319 aio_context_acquire(ctx);
320 assert(other_job->ret == 0);
321 block_job_completed_single(other_job);
322 aio_context_release(ctx);
326 void block_job_completed(BlockJob *job, int ret)
328 assert(blk_bs(job->blk)->job == job);
329 assert(!job->completed);
330 job->completed = true;
331 job->ret = ret;
332 if (!job->txn) {
333 block_job_completed_single(job);
334 } else if (ret < 0 || block_job_is_cancelled(job)) {
335 block_job_completed_txn_abort(job);
336 } else {
337 block_job_completed_txn_success(job);
341 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
343 Error *local_err = NULL;
345 if (!job->driver->set_speed) {
346 error_setg(errp, QERR_UNSUPPORTED);
347 return;
349 job->driver->set_speed(job, speed, &local_err);
350 if (local_err) {
351 error_propagate(errp, local_err);
352 return;
355 job->speed = speed;
358 void block_job_complete(BlockJob *job, Error **errp)
360 /* Should not be reachable via external interface for internal jobs */
361 assert(job->id);
362 if (job->pause_count || job->cancelled || !job->driver->complete) {
363 error_setg(errp, "The active block job '%s' cannot be completed",
364 job->id);
365 return;
368 job->driver->complete(job, errp);
371 void block_job_pause(BlockJob *job)
373 job->pause_count++;
376 static bool block_job_should_pause(BlockJob *job)
378 return job->pause_count > 0;
381 void coroutine_fn block_job_pause_point(BlockJob *job)
383 if (!block_job_should_pause(job)) {
384 return;
386 if (block_job_is_cancelled(job)) {
387 return;
390 if (job->driver->pause) {
391 job->driver->pause(job);
394 if (block_job_should_pause(job) && !block_job_is_cancelled(job)) {
395 job->paused = true;
396 job->busy = false;
397 qemu_coroutine_yield(); /* wait for block_job_resume() */
398 job->busy = true;
399 job->paused = false;
402 if (job->driver->resume) {
403 job->driver->resume(job);
407 void block_job_resume(BlockJob *job)
409 assert(job->pause_count > 0);
410 job->pause_count--;
411 if (job->pause_count) {
412 return;
414 block_job_enter(job);
417 void block_job_enter(BlockJob *job)
419 if (job->co && !job->busy) {
420 qemu_coroutine_enter(job->co);
424 void block_job_cancel(BlockJob *job)
426 job->cancelled = true;
427 block_job_iostatus_reset(job);
428 block_job_enter(job);
431 bool block_job_is_cancelled(BlockJob *job)
433 return job->cancelled;
436 void block_job_iostatus_reset(BlockJob *job)
438 job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
439 if (job->driver->iostatus_reset) {
440 job->driver->iostatus_reset(job);
444 static int block_job_finish_sync(BlockJob *job,
445 void (*finish)(BlockJob *, Error **errp),
446 Error **errp)
448 Error *local_err = NULL;
449 int ret;
451 assert(blk_bs(job->blk)->job == job);
453 block_job_ref(job);
455 finish(job, &local_err);
456 if (local_err) {
457 error_propagate(errp, local_err);
458 block_job_unref(job);
459 return -EBUSY;
461 /* block_job_drain calls block_job_enter, and it should be enough to
462 * induce progress until the job completes or moves to the main thread.
464 while (!job->deferred_to_main_loop && !job->completed) {
465 block_job_drain(job);
467 while (!job->completed) {
468 aio_poll(qemu_get_aio_context(), true);
470 ret = (job->cancelled && job->ret == 0) ? -ECANCELED : job->ret;
471 block_job_unref(job);
472 return ret;
475 /* A wrapper around block_job_cancel() taking an Error ** parameter so it may be
476 * used with block_job_finish_sync() without the need for (rather nasty)
477 * function pointer casts there. */
478 static void block_job_cancel_err(BlockJob *job, Error **errp)
480 block_job_cancel(job);
483 int block_job_cancel_sync(BlockJob *job)
485 return block_job_finish_sync(job, &block_job_cancel_err, NULL);
488 void block_job_cancel_sync_all(void)
490 BlockJob *job;
491 AioContext *aio_context;
493 while ((job = QLIST_FIRST(&block_jobs))) {
494 aio_context = blk_get_aio_context(job->blk);
495 aio_context_acquire(aio_context);
496 block_job_cancel_sync(job);
497 aio_context_release(aio_context);
501 int block_job_complete_sync(BlockJob *job, Error **errp)
503 return block_job_finish_sync(job, &block_job_complete, errp);
506 void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns)
508 assert(job->busy);
510 /* Check cancellation *before* setting busy = false, too! */
511 if (block_job_is_cancelled(job)) {
512 return;
515 job->busy = false;
516 if (!block_job_should_pause(job)) {
517 co_aio_sleep_ns(blk_get_aio_context(job->blk), type, ns);
519 job->busy = true;
521 block_job_pause_point(job);
524 void block_job_yield(BlockJob *job)
526 assert(job->busy);
528 /* Check cancellation *before* setting busy = false, too! */
529 if (block_job_is_cancelled(job)) {
530 return;
533 job->busy = false;
534 if (!block_job_should_pause(job)) {
535 qemu_coroutine_yield();
537 job->busy = true;
539 block_job_pause_point(job);
542 BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
544 BlockJobInfo *info;
546 if (block_job_is_internal(job)) {
547 error_setg(errp, "Cannot query QEMU internal jobs");
548 return NULL;
550 info = g_new0(BlockJobInfo, 1);
551 info->type = g_strdup(BlockJobType_lookup[job->driver->job_type]);
552 info->device = g_strdup(job->id);
553 info->len = job->len;
554 info->busy = job->busy;
555 info->paused = job->pause_count > 0;
556 info->offset = job->offset;
557 info->speed = job->speed;
558 info->io_status = job->iostatus;
559 info->ready = job->ready;
560 return info;
563 static void block_job_iostatus_set_err(BlockJob *job, int error)
565 if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
566 job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
567 BLOCK_DEVICE_IO_STATUS_FAILED;
571 static void block_job_event_cancelled(BlockJob *job)
573 if (block_job_is_internal(job)) {
574 return;
577 qapi_event_send_block_job_cancelled(job->driver->job_type,
578 job->id,
579 job->len,
580 job->offset,
581 job->speed,
582 &error_abort);
585 static void block_job_event_completed(BlockJob *job, const char *msg)
587 if (block_job_is_internal(job)) {
588 return;
591 qapi_event_send_block_job_completed(job->driver->job_type,
592 job->id,
593 job->len,
594 job->offset,
595 job->speed,
596 !!msg,
597 msg,
598 &error_abort);
601 void block_job_event_ready(BlockJob *job)
603 job->ready = true;
605 if (block_job_is_internal(job)) {
606 return;
609 qapi_event_send_block_job_ready(job->driver->job_type,
610 job->id,
611 job->len,
612 job->offset,
613 job->speed, &error_abort);
616 BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
617 int is_read, int error)
619 BlockErrorAction action;
621 switch (on_err) {
622 case BLOCKDEV_ON_ERROR_ENOSPC:
623 case BLOCKDEV_ON_ERROR_AUTO:
624 action = (error == ENOSPC) ?
625 BLOCK_ERROR_ACTION_STOP : BLOCK_ERROR_ACTION_REPORT;
626 break;
627 case BLOCKDEV_ON_ERROR_STOP:
628 action = BLOCK_ERROR_ACTION_STOP;
629 break;
630 case BLOCKDEV_ON_ERROR_REPORT:
631 action = BLOCK_ERROR_ACTION_REPORT;
632 break;
633 case BLOCKDEV_ON_ERROR_IGNORE:
634 action = BLOCK_ERROR_ACTION_IGNORE;
635 break;
636 default:
637 abort();
639 if (!block_job_is_internal(job)) {
640 qapi_event_send_block_job_error(job->id,
641 is_read ? IO_OPERATION_TYPE_READ :
642 IO_OPERATION_TYPE_WRITE,
643 action, &error_abort);
645 if (action == BLOCK_ERROR_ACTION_STOP) {
646 /* make the pause user visible, which will be resumed from QMP. */
647 job->user_paused = true;
648 block_job_pause(job);
649 block_job_iostatus_set_err(job, error);
651 return action;
654 typedef struct {
655 BlockJob *job;
656 AioContext *aio_context;
657 BlockJobDeferToMainLoopFn *fn;
658 void *opaque;
659 } BlockJobDeferToMainLoopData;
661 static void block_job_defer_to_main_loop_bh(void *opaque)
663 BlockJobDeferToMainLoopData *data = opaque;
664 AioContext *aio_context;
666 /* Prevent race with block_job_defer_to_main_loop() */
667 aio_context_acquire(data->aio_context);
669 /* Fetch BDS AioContext again, in case it has changed */
670 aio_context = blk_get_aio_context(data->job->blk);
671 aio_context_acquire(aio_context);
673 data->job->deferred_to_main_loop = false;
674 data->fn(data->job, data->opaque);
676 aio_context_release(aio_context);
678 aio_context_release(data->aio_context);
680 g_free(data);
683 void block_job_defer_to_main_loop(BlockJob *job,
684 BlockJobDeferToMainLoopFn *fn,
685 void *opaque)
687 BlockJobDeferToMainLoopData *data = g_malloc(sizeof(*data));
688 data->job = job;
689 data->aio_context = blk_get_aio_context(job->blk);
690 data->fn = fn;
691 data->opaque = opaque;
692 job->deferred_to_main_loop = true;
694 aio_bh_schedule_oneshot(qemu_get_aio_context(),
695 block_job_defer_to_main_loop_bh, data);
698 BlockJobTxn *block_job_txn_new(void)
700 BlockJobTxn *txn = g_new0(BlockJobTxn, 1);
701 QLIST_INIT(&txn->jobs);
702 txn->refcnt = 1;
703 return txn;
706 static void block_job_txn_ref(BlockJobTxn *txn)
708 txn->refcnt++;
711 void block_job_txn_unref(BlockJobTxn *txn)
713 if (txn && --txn->refcnt == 0) {
714 g_free(txn);
718 void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job)
720 if (!txn) {
721 return;
724 assert(!job->txn);
725 job->txn = txn;
727 QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
728 block_job_txn_ref(txn);