]> git.proxmox.com Git - mirror_qemu.git/blob - blockjob.c
blockjobs: Allow creating internal jobs
[mirror_qemu.git] / blockjob.c
1 /*
2 * QEMU System Emulator block driver
3 *
4 * Copyright (c) 2011 IBM Corp.
5 * Copyright (c) 2012 Red Hat, Inc.
6 *
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:
13 *
14 * The above copyright notice and this permission notice shall be included in
15 * all copies or substantial portions of the Software.
16 *
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.
24 */
25
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"
40
41 /* Transactional group of block jobs */
42 struct BlockJobTxn {
43
44 /* Is this txn being cancelled? */
45 bool aborting;
46
47 /* List of jobs */
48 QLIST_HEAD(, BlockJob) jobs;
49
50 /* Reference count */
51 int refcnt;
52 };
53
54 static QLIST_HEAD(, BlockJob) block_jobs = QLIST_HEAD_INITIALIZER(block_jobs);
55
56 BlockJob *block_job_next(BlockJob *job)
57 {
58 if (!job) {
59 return QLIST_FIRST(&block_jobs);
60 }
61 return QLIST_NEXT(job, job_list);
62 }
63
64 BlockJob *block_job_get(const char *id)
65 {
66 BlockJob *job;
67
68 QLIST_FOREACH(job, &block_jobs, job_list) {
69 if (job->id && !strcmp(id, job->id)) {
70 return job;
71 }
72 }
73
74 return NULL;
75 }
76
77 static void block_job_attached_aio_context(AioContext *new_context,
78 void *opaque)
79 {
80 BlockJob *job = opaque;
81
82 if (job->driver->attached_aio_context) {
83 job->driver->attached_aio_context(job, new_context);
84 }
85
86 block_job_resume(job);
87 }
88
89 static void block_job_drain(BlockJob *job)
90 {
91 /* If job is !job->busy this kicks it into the next pause point. */
92 block_job_enter(job);
93
94 blk_drain(job->blk);
95 if (job->driver->drain) {
96 job->driver->drain(job);
97 }
98 }
99
100 static void block_job_detach_aio_context(void *opaque)
101 {
102 BlockJob *job = opaque;
103
104 /* In case the job terminates during aio_poll()... */
105 block_job_ref(job);
106
107 block_job_pause(job);
108
109 while (!job->paused && !job->completed) {
110 block_job_drain(job);
111 }
112
113 block_job_unref(job);
114 }
115
116 void block_job_add_bdrv(BlockJob *job, BlockDriverState *bs)
117 {
118 job->nodes = g_slist_prepend(job->nodes, bs);
119 bdrv_ref(bs);
120 bdrv_op_block_all(bs, job->blocker);
121 }
122
123 void *block_job_create(const char *job_id, const BlockJobDriver *driver,
124 BlockDriverState *bs, int64_t speed, int flags,
125 BlockCompletionFunc *cb, void *opaque, Error **errp)
126 {
127 BlockBackend *blk;
128 BlockJob *job;
129
130 assert(cb);
131 if (bs->job) {
132 error_setg(errp, QERR_DEVICE_IN_USE, bdrv_get_device_name(bs));
133 return NULL;
134 }
135
136 if (job_id == NULL && !(flags & BLOCK_JOB_INTERNAL)) {
137 job_id = bdrv_get_device_name(bs);
138 if (!*job_id) {
139 error_setg(errp, "An explicit job ID is required for this node");
140 return NULL;
141 }
142 }
143
144 if (job_id) {
145 if (flags & BLOCK_JOB_INTERNAL) {
146 error_setg(errp, "Cannot specify job ID for internal block job");
147 return NULL;
148 }
149
150 if (!id_wellformed(job_id)) {
151 error_setg(errp, "Invalid job ID '%s'", job_id);
152 return NULL;
153 }
154
155 if (block_job_get(job_id)) {
156 error_setg(errp, "Job ID '%s' already in use", job_id);
157 return NULL;
158 }
159 }
160
161 blk = blk_new();
162 blk_insert_bs(blk, bs);
163
164 job = g_malloc0(driver->instance_size);
165 error_setg(&job->blocker, "block device is in use by block job: %s",
166 BlockJobType_lookup[driver->job_type]);
167 block_job_add_bdrv(job, bs);
168 bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker);
169
170 job->driver = driver;
171 job->id = g_strdup(job_id);
172 job->blk = blk;
173 job->cb = cb;
174 job->opaque = opaque;
175 job->busy = true;
176 job->refcnt = 1;
177 bs->job = job;
178
179 QLIST_INSERT_HEAD(&block_jobs, job, job_list);
180
181 blk_add_aio_context_notifier(blk, block_job_attached_aio_context,
182 block_job_detach_aio_context, job);
183
184 /* Only set speed when necessary to avoid NotSupported error */
185 if (speed != 0) {
186 Error *local_err = NULL;
187
188 block_job_set_speed(job, speed, &local_err);
189 if (local_err) {
190 block_job_unref(job);
191 error_propagate(errp, local_err);
192 return NULL;
193 }
194 }
195 return job;
196 }
197
198 bool block_job_is_internal(BlockJob *job)
199 {
200 return (job->id == NULL);
201 }
202
203 void block_job_ref(BlockJob *job)
204 {
205 ++job->refcnt;
206 }
207
208 void block_job_unref(BlockJob *job)
209 {
210 if (--job->refcnt == 0) {
211 GSList *l;
212 BlockDriverState *bs = blk_bs(job->blk);
213 bs->job = NULL;
214 for (l = job->nodes; l; l = l->next) {
215 bs = l->data;
216 bdrv_op_unblock_all(bs, job->blocker);
217 bdrv_unref(bs);
218 }
219 g_slist_free(job->nodes);
220 blk_remove_aio_context_notifier(job->blk,
221 block_job_attached_aio_context,
222 block_job_detach_aio_context, job);
223 blk_unref(job->blk);
224 error_free(job->blocker);
225 g_free(job->id);
226 QLIST_REMOVE(job, job_list);
227 g_free(job);
228 }
229 }
230
231 static void block_job_completed_single(BlockJob *job)
232 {
233 if (!job->ret) {
234 if (job->driver->commit) {
235 job->driver->commit(job);
236 }
237 } else {
238 if (job->driver->abort) {
239 job->driver->abort(job);
240 }
241 }
242 job->cb(job->opaque, job->ret);
243 if (job->txn) {
244 block_job_txn_unref(job->txn);
245 }
246 block_job_unref(job);
247 }
248
249 static void block_job_completed_txn_abort(BlockJob *job)
250 {
251 AioContext *ctx;
252 BlockJobTxn *txn = job->txn;
253 BlockJob *other_job, *next;
254
255 if (txn->aborting) {
256 /*
257 * We are cancelled by another job, which will handle everything.
258 */
259 return;
260 }
261 txn->aborting = true;
262 /* We are the first failed job. Cancel other jobs. */
263 QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
264 ctx = blk_get_aio_context(other_job->blk);
265 aio_context_acquire(ctx);
266 }
267 QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
268 if (other_job == job || other_job->completed) {
269 /* Other jobs are "effectively" cancelled by us, set the status for
270 * them; this job, however, may or may not be cancelled, depending
271 * on the caller, so leave it. */
272 if (other_job != job) {
273 other_job->cancelled = true;
274 }
275 continue;
276 }
277 block_job_cancel_sync(other_job);
278 assert(other_job->completed);
279 }
280 QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
281 ctx = blk_get_aio_context(other_job->blk);
282 block_job_completed_single(other_job);
283 aio_context_release(ctx);
284 }
285 }
286
287 static void block_job_completed_txn_success(BlockJob *job)
288 {
289 AioContext *ctx;
290 BlockJobTxn *txn = job->txn;
291 BlockJob *other_job, *next;
292 /*
293 * Successful completion, see if there are other running jobs in this
294 * txn.
295 */
296 QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
297 if (!other_job->completed) {
298 return;
299 }
300 }
301 /* We are the last completed job, commit the transaction. */
302 QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
303 ctx = blk_get_aio_context(other_job->blk);
304 aio_context_acquire(ctx);
305 assert(other_job->ret == 0);
306 block_job_completed_single(other_job);
307 aio_context_release(ctx);
308 }
309 }
310
311 void block_job_completed(BlockJob *job, int ret)
312 {
313 assert(blk_bs(job->blk)->job == job);
314 assert(!job->completed);
315 job->completed = true;
316 job->ret = ret;
317 if (!job->txn) {
318 block_job_completed_single(job);
319 } else if (ret < 0 || block_job_is_cancelled(job)) {
320 block_job_completed_txn_abort(job);
321 } else {
322 block_job_completed_txn_success(job);
323 }
324 }
325
326 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
327 {
328 Error *local_err = NULL;
329
330 if (!job->driver->set_speed) {
331 error_setg(errp, QERR_UNSUPPORTED);
332 return;
333 }
334 job->driver->set_speed(job, speed, &local_err);
335 if (local_err) {
336 error_propagate(errp, local_err);
337 return;
338 }
339
340 job->speed = speed;
341 }
342
343 void block_job_complete(BlockJob *job, Error **errp)
344 {
345 /* Should not be reachable via external interface for internal jobs */
346 assert(job->id);
347 if (job->pause_count || job->cancelled || !job->driver->complete) {
348 error_setg(errp, "The active block job '%s' cannot be completed",
349 job->id);
350 return;
351 }
352
353 job->driver->complete(job, errp);
354 }
355
356 void block_job_pause(BlockJob *job)
357 {
358 job->pause_count++;
359 }
360
361 static bool block_job_should_pause(BlockJob *job)
362 {
363 return job->pause_count > 0;
364 }
365
366 void coroutine_fn block_job_pause_point(BlockJob *job)
367 {
368 if (!block_job_should_pause(job)) {
369 return;
370 }
371 if (block_job_is_cancelled(job)) {
372 return;
373 }
374
375 if (job->driver->pause) {
376 job->driver->pause(job);
377 }
378
379 if (block_job_should_pause(job) && !block_job_is_cancelled(job)) {
380 job->paused = true;
381 job->busy = false;
382 qemu_coroutine_yield(); /* wait for block_job_resume() */
383 job->busy = true;
384 job->paused = false;
385 }
386
387 if (job->driver->resume) {
388 job->driver->resume(job);
389 }
390 }
391
392 void block_job_resume(BlockJob *job)
393 {
394 assert(job->pause_count > 0);
395 job->pause_count--;
396 if (job->pause_count) {
397 return;
398 }
399 block_job_enter(job);
400 }
401
402 void block_job_enter(BlockJob *job)
403 {
404 if (job->co && !job->busy) {
405 qemu_coroutine_enter(job->co);
406 }
407 }
408
409 void block_job_cancel(BlockJob *job)
410 {
411 job->cancelled = true;
412 block_job_iostatus_reset(job);
413 block_job_enter(job);
414 }
415
416 bool block_job_is_cancelled(BlockJob *job)
417 {
418 return job->cancelled;
419 }
420
421 void block_job_iostatus_reset(BlockJob *job)
422 {
423 job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
424 if (job->driver->iostatus_reset) {
425 job->driver->iostatus_reset(job);
426 }
427 }
428
429 static int block_job_finish_sync(BlockJob *job,
430 void (*finish)(BlockJob *, Error **errp),
431 Error **errp)
432 {
433 Error *local_err = NULL;
434 int ret;
435
436 assert(blk_bs(job->blk)->job == job);
437
438 block_job_ref(job);
439
440 finish(job, &local_err);
441 if (local_err) {
442 error_propagate(errp, local_err);
443 block_job_unref(job);
444 return -EBUSY;
445 }
446 /* block_job_drain calls block_job_enter, and it should be enough to
447 * induce progress until the job completes or moves to the main thread.
448 */
449 while (!job->deferred_to_main_loop && !job->completed) {
450 block_job_drain(job);
451 }
452 while (!job->completed) {
453 aio_poll(qemu_get_aio_context(), true);
454 }
455 ret = (job->cancelled && job->ret == 0) ? -ECANCELED : job->ret;
456 block_job_unref(job);
457 return ret;
458 }
459
460 /* A wrapper around block_job_cancel() taking an Error ** parameter so it may be
461 * used with block_job_finish_sync() without the need for (rather nasty)
462 * function pointer casts there. */
463 static void block_job_cancel_err(BlockJob *job, Error **errp)
464 {
465 block_job_cancel(job);
466 }
467
468 int block_job_cancel_sync(BlockJob *job)
469 {
470 return block_job_finish_sync(job, &block_job_cancel_err, NULL);
471 }
472
473 void block_job_cancel_sync_all(void)
474 {
475 BlockJob *job;
476 AioContext *aio_context;
477
478 while ((job = QLIST_FIRST(&block_jobs))) {
479 aio_context = blk_get_aio_context(job->blk);
480 aio_context_acquire(aio_context);
481 block_job_cancel_sync(job);
482 aio_context_release(aio_context);
483 }
484 }
485
486 int block_job_complete_sync(BlockJob *job, Error **errp)
487 {
488 return block_job_finish_sync(job, &block_job_complete, errp);
489 }
490
491 void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns)
492 {
493 assert(job->busy);
494
495 /* Check cancellation *before* setting busy = false, too! */
496 if (block_job_is_cancelled(job)) {
497 return;
498 }
499
500 job->busy = false;
501 if (!block_job_should_pause(job)) {
502 co_aio_sleep_ns(blk_get_aio_context(job->blk), type, ns);
503 }
504 job->busy = true;
505
506 block_job_pause_point(job);
507 }
508
509 void block_job_yield(BlockJob *job)
510 {
511 assert(job->busy);
512
513 /* Check cancellation *before* setting busy = false, too! */
514 if (block_job_is_cancelled(job)) {
515 return;
516 }
517
518 job->busy = false;
519 if (!block_job_should_pause(job)) {
520 qemu_coroutine_yield();
521 }
522 job->busy = true;
523
524 block_job_pause_point(job);
525 }
526
527 BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
528 {
529 BlockJobInfo *info;
530
531 if (block_job_is_internal(job)) {
532 error_setg(errp, "Cannot query QEMU internal jobs");
533 return NULL;
534 }
535 info = g_new0(BlockJobInfo, 1);
536 info->type = g_strdup(BlockJobType_lookup[job->driver->job_type]);
537 info->device = g_strdup(job->id);
538 info->len = job->len;
539 info->busy = job->busy;
540 info->paused = job->pause_count > 0;
541 info->offset = job->offset;
542 info->speed = job->speed;
543 info->io_status = job->iostatus;
544 info->ready = job->ready;
545 return info;
546 }
547
548 static void block_job_iostatus_set_err(BlockJob *job, int error)
549 {
550 if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
551 job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
552 BLOCK_DEVICE_IO_STATUS_FAILED;
553 }
554 }
555
556 void block_job_event_cancelled(BlockJob *job)
557 {
558 if (block_job_is_internal(job)) {
559 return;
560 }
561
562 qapi_event_send_block_job_cancelled(job->driver->job_type,
563 job->id,
564 job->len,
565 job->offset,
566 job->speed,
567 &error_abort);
568 }
569
570 void block_job_event_completed(BlockJob *job, const char *msg)
571 {
572 if (block_job_is_internal(job)) {
573 return;
574 }
575
576 qapi_event_send_block_job_completed(job->driver->job_type,
577 job->id,
578 job->len,
579 job->offset,
580 job->speed,
581 !!msg,
582 msg,
583 &error_abort);
584 }
585
586 void block_job_event_ready(BlockJob *job)
587 {
588 job->ready = true;
589
590 if (block_job_is_internal(job)) {
591 return;
592 }
593
594 qapi_event_send_block_job_ready(job->driver->job_type,
595 job->id,
596 job->len,
597 job->offset,
598 job->speed, &error_abort);
599 }
600
601 BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
602 int is_read, int error)
603 {
604 BlockErrorAction action;
605
606 switch (on_err) {
607 case BLOCKDEV_ON_ERROR_ENOSPC:
608 case BLOCKDEV_ON_ERROR_AUTO:
609 action = (error == ENOSPC) ?
610 BLOCK_ERROR_ACTION_STOP : BLOCK_ERROR_ACTION_REPORT;
611 break;
612 case BLOCKDEV_ON_ERROR_STOP:
613 action = BLOCK_ERROR_ACTION_STOP;
614 break;
615 case BLOCKDEV_ON_ERROR_REPORT:
616 action = BLOCK_ERROR_ACTION_REPORT;
617 break;
618 case BLOCKDEV_ON_ERROR_IGNORE:
619 action = BLOCK_ERROR_ACTION_IGNORE;
620 break;
621 default:
622 abort();
623 }
624 if (!block_job_is_internal(job)) {
625 qapi_event_send_block_job_error(job->id,
626 is_read ? IO_OPERATION_TYPE_READ :
627 IO_OPERATION_TYPE_WRITE,
628 action, &error_abort);
629 }
630 if (action == BLOCK_ERROR_ACTION_STOP) {
631 /* make the pause user visible, which will be resumed from QMP. */
632 job->user_paused = true;
633 block_job_pause(job);
634 block_job_iostatus_set_err(job, error);
635 }
636 return action;
637 }
638
639 typedef struct {
640 BlockJob *job;
641 AioContext *aio_context;
642 BlockJobDeferToMainLoopFn *fn;
643 void *opaque;
644 } BlockJobDeferToMainLoopData;
645
646 static void block_job_defer_to_main_loop_bh(void *opaque)
647 {
648 BlockJobDeferToMainLoopData *data = opaque;
649 AioContext *aio_context;
650
651 /* Prevent race with block_job_defer_to_main_loop() */
652 aio_context_acquire(data->aio_context);
653
654 /* Fetch BDS AioContext again, in case it has changed */
655 aio_context = blk_get_aio_context(data->job->blk);
656 aio_context_acquire(aio_context);
657
658 data->job->deferred_to_main_loop = false;
659 data->fn(data->job, data->opaque);
660
661 aio_context_release(aio_context);
662
663 aio_context_release(data->aio_context);
664
665 g_free(data);
666 }
667
668 void block_job_defer_to_main_loop(BlockJob *job,
669 BlockJobDeferToMainLoopFn *fn,
670 void *opaque)
671 {
672 BlockJobDeferToMainLoopData *data = g_malloc(sizeof(*data));
673 data->job = job;
674 data->aio_context = blk_get_aio_context(job->blk);
675 data->fn = fn;
676 data->opaque = opaque;
677 job->deferred_to_main_loop = true;
678
679 aio_bh_schedule_oneshot(qemu_get_aio_context(),
680 block_job_defer_to_main_loop_bh, data);
681 }
682
683 BlockJobTxn *block_job_txn_new(void)
684 {
685 BlockJobTxn *txn = g_new0(BlockJobTxn, 1);
686 QLIST_INIT(&txn->jobs);
687 txn->refcnt = 1;
688 return txn;
689 }
690
691 static void block_job_txn_ref(BlockJobTxn *txn)
692 {
693 txn->refcnt++;
694 }
695
696 void block_job_txn_unref(BlockJobTxn *txn)
697 {
698 if (txn && --txn->refcnt == 0) {
699 g_free(txn);
700 }
701 }
702
703 void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job)
704 {
705 if (!txn) {
706 return;
707 }
708
709 assert(!job->txn);
710 job->txn = txn;
711
712 QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
713 block_job_txn_ref(txn);
714 }