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 "block/block.h"
  29 #include "block/blockjob_int.h"
  30 #include "block/block_int.h"
  31 #include "block/trace.h"
  32 #include "sysemu/block-backend.h"
  33 #include "qapi/error.h"
  34 #include "qapi/qapi-events-block-core.h"
  35 #include "qapi/qmp/qerror.h"
  36 #include "qemu/coroutine.h"
  37 #include "qemu/id.h"
  38 #include "qemu/timer.h"
  39
  40 /* Right now, this mutex is only needed to synchronize accesses to job->busy
  41  * and job->sleep_timer, such as concurrent calls to block_job_do_yield and
  42  * block_job_enter. */
  43 static QemuMutex block_job_mutex;
  44
  45 /* BlockJob State Transition Table */
  46 bool BlockJobSTT[BLOCK_JOB_STATUS__MAX][BLOCK_JOB_STATUS__MAX] = {
  47                                           /* U, C, R, P, Y, S, W, D, X, E, N */
  48     /* U: */ [BLOCK_JOB_STATUS_UNDEFINED] = {0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0},
  49     /* C: */ [BLOCK_JOB_STATUS_CREATED]   = {0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1},
  50     /* R: */ [BLOCK_JOB_STATUS_RUNNING]   = {0, 0, 0, 1, 1, 0, 1, 0, 1, 0, 0},
  51     /* P: */ [BLOCK_JOB_STATUS_PAUSED]    = {0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0},
  52     /* Y: */ [BLOCK_JOB_STATUS_READY]     = {0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0},
  53     /* S: */ [BLOCK_JOB_STATUS_STANDBY]   = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0},
  54     /* W: */ [BLOCK_JOB_STATUS_WAITING]   = {0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0},
  55     /* D: */ [BLOCK_JOB_STATUS_PENDING]   = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
  56     /* X: */ [BLOCK_JOB_STATUS_ABORTING]  = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
  57     /* E: */ [BLOCK_JOB_STATUS_CONCLUDED] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1},
  58     /* N: */ [BLOCK_JOB_STATUS_NULL]      = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
  59 };
  60
  61 bool BlockJobVerbTable[BLOCK_JOB_VERB__MAX][BLOCK_JOB_STATUS__MAX] = {
  62                                           /* U, C, R, P, Y, S, W, D, X, E, N */
  63     [BLOCK_JOB_VERB_CANCEL]               = {0, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0},
  64     [BLOCK_JOB_VERB_PAUSE]                = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  65     [BLOCK_JOB_VERB_RESUME]               = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  66     [BLOCK_JOB_VERB_SET_SPEED]            = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  67     [BLOCK_JOB_VERB_COMPLETE]             = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0},
  68     [BLOCK_JOB_VERB_FINALIZE]             = {0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0},
  69     [BLOCK_JOB_VERB_DISMISS]              = {0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0},
  70 };
  71
  72 static void block_job_state_transition(BlockJob *job, BlockJobStatus s1)
  73 {
  74     BlockJobStatus s0 = job->status;
  75     assert(s1 >= 0 && s1 <= BLOCK_JOB_STATUS__MAX);
  76     trace_block_job_state_transition(job, job->ret, BlockJobSTT[s0][s1] ?
  77                                      "allowed" : "disallowed",
  78                                      BlockJobStatus_str(s0),
  79                                      BlockJobStatus_str(s1));
  80     assert(BlockJobSTT[s0][s1]);
  81     job->status = s1;
  82 }
  83
  84 static int block_job_apply_verb(BlockJob *job, BlockJobVerb bv, Error **errp)
  85 {
  86     assert(bv >= 0 && bv <= BLOCK_JOB_VERB__MAX);
  87     trace_block_job_apply_verb(job, BlockJobStatus_str(job->status),
  88                                BlockJobVerb_str(bv),
  89                                BlockJobVerbTable[bv][job->status] ?
  90                                "allowed" : "prohibited");
  91     if (BlockJobVerbTable[bv][job->status]) {
  92         return 0;
  93     }
  94     error_setg(errp, "Job '%s' in state '%s' cannot accept command verb '%s'",
  95                job->id, BlockJobStatus_str(job->status), BlockJobVerb_str(bv));
  96     return -EPERM;
  97 }
  98
  99 static void block_job_lock(void)
 100 {
 101     qemu_mutex_lock(&block_job_mutex);
 102 }
 103
 104 static void block_job_unlock(void)
 105 {
 106     qemu_mutex_unlock(&block_job_mutex);
 107 }
 108
 109 static void __attribute__((__constructor__)) block_job_init(void)
 110 {
 111     qemu_mutex_init(&block_job_mutex);
 112 }
 113
 114 static void block_job_event_cancelled(BlockJob *job);
 115 static void block_job_event_completed(BlockJob *job, const char *msg);
 116 static int block_job_event_pending(BlockJob *job);
 117 static void block_job_enter_cond(BlockJob *job, bool(*fn)(BlockJob *job));
 118
 119 /* Transactional group of block jobs */
 120 struct BlockJobTxn {
 121
 122     /* Is this txn being cancelled? */
 123     bool aborting;
 124
 125     /* List of jobs */
 126     QLIST_HEAD(, BlockJob) jobs;
 127
 128     /* Reference count */
 129     int refcnt;
 130 };
 131
 132 static QLIST_HEAD(, BlockJob) block_jobs = QLIST_HEAD_INITIALIZER(block_jobs);
 133
 134 /*
 135  * The block job API is composed of two categories of functions.
 136  *
 137  * The first includes functions used by the monitor.  The monitor is
 138  * peculiar in that it accesses the block job list with block_job_get, and
 139  * therefore needs consistency across block_job_get and the actual operation
 140  * (e.g. block_job_set_speed).  The consistency is achieved with
 141  * aio_context_acquire/release.  These functions are declared in blockjob.h.
 142  *
 143  * The second includes functions used by the block job drivers and sometimes
 144  * by the core block layer.  These do not care about locking, because the
 145  * whole coroutine runs under the AioContext lock, and are declared in
 146  * blockjob_int.h.
 147  */
 148
 149 BlockJob *block_job_next(BlockJob *job)
 150 {
 151     if (!job) {
 152         return QLIST_FIRST(&block_jobs);
 153     }
 154     return QLIST_NEXT(job, job_list);
 155 }
 156
 157 BlockJob *block_job_get(const char *id)
 158 {
 159     BlockJob *job;
 160
 161     QLIST_FOREACH(job, &block_jobs, job_list) {
 162         if (job->id && !strcmp(id, job->id)) {
 163             return job;
 164         }
 165     }
 166
 167     return NULL;
 168 }
 169
 170 BlockJobTxn *block_job_txn_new(void)
 171 {
 172     BlockJobTxn *txn = g_new0(BlockJobTxn, 1);
 173     QLIST_INIT(&txn->jobs);
 174     txn->refcnt = 1;
 175     return txn;
 176 }
 177
 178 static void block_job_txn_ref(BlockJobTxn *txn)
 179 {
 180     txn->refcnt++;
 181 }
 182
 183 void block_job_txn_unref(BlockJobTxn *txn)
 184 {
 185     if (txn && --txn->refcnt == 0) {
 186         g_free(txn);
 187     }
 188 }
 189
 190 void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job)
 191 {
 192     if (!txn) {
 193         return;
 194     }
 195
 196     assert(!job->txn);
 197     job->txn = txn;
 198
 199     QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
 200     block_job_txn_ref(txn);
 201 }
 202
 203 static void block_job_txn_del_job(BlockJob *job)
 204 {
 205     if (job->txn) {
 206         QLIST_REMOVE(job, txn_list);
 207         block_job_txn_unref(job->txn);
 208         job->txn = NULL;
 209     }
 210 }
 211
 212 /* Assumes the block_job_mutex is held */
 213 static bool block_job_timer_pending(BlockJob *job)
 214 {
 215     return timer_pending(&job->sleep_timer);
 216 }
 217
 218 /* Assumes the block_job_mutex is held */
 219 static bool block_job_timer_not_pending(BlockJob *job)
 220 {
 221     return !block_job_timer_pending(job);
 222 }
 223
 224 static void block_job_pause(BlockJob *job)
 225 {
 226     job->pause_count++;
 227 }
 228
 229 static void block_job_resume(BlockJob *job)
 230 {
 231     assert(job->pause_count > 0);
 232     job->pause_count--;
 233     if (job->pause_count) {
 234         return;
 235     }
 236
 237     /* kick only if no timer is pending */
 238     block_job_enter_cond(job, block_job_timer_not_pending);
 239 }
 240
 241 void block_job_ref(BlockJob *job)
 242 {
 243     ++job->refcnt;
 244 }
 245
 246 static void block_job_attached_aio_context(AioContext *new_context,
 247                                            void *opaque);
 248 static void block_job_detach_aio_context(void *opaque);
 249
 250 void block_job_unref(BlockJob *job)
 251 {
 252     if (--job->refcnt == 0) {
 253         assert(job->status == BLOCK_JOB_STATUS_NULL);
 254         assert(!job->txn);
 255         BlockDriverState *bs = blk_bs(job->blk);
 256         QLIST_REMOVE(job, job_list);
 257         bs->job = NULL;
 258         block_job_remove_all_bdrv(job);
 259         blk_remove_aio_context_notifier(job->blk,
 260                                         block_job_attached_aio_context,
 261                                         block_job_detach_aio_context, job);
 262         blk_unref(job->blk);
 263         error_free(job->blocker);
 264         g_free(job->id);
 265         assert(!timer_pending(&job->sleep_timer));
 266         g_free(job);
 267     }
 268 }
 269
 270 static void block_job_attached_aio_context(AioContext *new_context,
 271                                            void *opaque)
 272 {
 273     BlockJob *job = opaque;
 274
 275     if (job->driver->attached_aio_context) {
 276         job->driver->attached_aio_context(job, new_context);
 277     }
 278
 279     block_job_resume(job);
 280 }
 281
 282 static void block_job_drain(BlockJob *job)
 283 {
 284     /* If job is !job->busy this kicks it into the next pause point. */
 285     block_job_enter(job);
 286
 287     blk_drain(job->blk);
 288     if (job->driver->drain) {
 289         job->driver->drain(job);
 290     }
 291 }
 292
 293 static void block_job_detach_aio_context(void *opaque)
 294 {
 295     BlockJob *job = opaque;
 296
 297     /* In case the job terminates during aio_poll()... */
 298     block_job_ref(job);
 299
 300     block_job_pause(job);
 301
 302     while (!job->paused && !job->completed) {
 303         block_job_drain(job);
 304     }
 305
 306     block_job_unref(job);
 307 }
 308
 309 static char *child_job_get_parent_desc(BdrvChild *c)
 310 {
 311     BlockJob *job = c->opaque;
 312     return g_strdup_printf("%s job '%s'",
 313                            BlockJobType_str(job->driver->job_type),
 314                            job->id);
 315 }
 316
 317 static void child_job_drained_begin(BdrvChild *c)
 318 {
 319     BlockJob *job = c->opaque;
 320     block_job_pause(job);
 321 }
 322
 323 static void child_job_drained_end(BdrvChild *c)
 324 {
 325     BlockJob *job = c->opaque;
 326     block_job_resume(job);
 327 }
 328
 329 static const BdrvChildRole child_job = {
 330     .get_parent_desc    = child_job_get_parent_desc,
 331     .drained_begin      = child_job_drained_begin,
 332     .drained_end        = child_job_drained_end,
 333     .stay_at_node       = true,
 334 };
 335
 336 void block_job_remove_all_bdrv(BlockJob *job)
 337 {
 338     GSList *l;
 339     for (l = job->nodes; l; l = l->next) {
 340         BdrvChild *c = l->data;
 341         bdrv_op_unblock_all(c->bs, job->blocker);
 342         bdrv_root_unref_child(c);
 343     }
 344     g_slist_free(job->nodes);
 345     job->nodes = NULL;
 346 }
 347
 348 int block_job_add_bdrv(BlockJob *job, const char *name, BlockDriverState *bs,
 349                        uint64_t perm, uint64_t shared_perm, Error **errp)
 350 {
 351     BdrvChild *c;
 352
 353     c = bdrv_root_attach_child(bs, name, &child_job, perm, shared_perm,
 354                                job, errp);
 355     if (c == NULL) {
 356         return -EPERM;
 357     }
 358
 359     job->nodes = g_slist_prepend(job->nodes, c);
 360     bdrv_ref(bs);
 361     bdrv_op_block_all(bs, job->blocker);
 362
 363     return 0;
 364 }
 365
 366 bool block_job_is_internal(BlockJob *job)
 367 {
 368     return (job->id == NULL);
 369 }
 370
 371 static bool block_job_started(BlockJob *job)
 372 {
 373     return job->co;
 374 }
 375
 376 const BlockJobDriver *block_job_driver(BlockJob *job)
 377 {
 378     return job->driver;
 379 }
 380
 381 /**
 382  * All jobs must allow a pause point before entering their job proper. This
 383  * ensures that jobs can be paused prior to being started, then resumed later.
 384  */
 385 static void coroutine_fn block_job_co_entry(void *opaque)
 386 {
 387     BlockJob *job = opaque;
 388
 389     assert(job && job->driver && job->driver->start);
 390     block_job_pause_point(job);
 391     job->driver->start(job);
 392 }
 393
 394 static void block_job_sleep_timer_cb(void *opaque)
 395 {
 396     BlockJob *job = opaque;
 397
 398     block_job_enter(job);
 399 }
 400
 401 void block_job_start(BlockJob *job)
 402 {
 403     assert(job && !block_job_started(job) && job->paused &&
 404            job->driver && job->driver->start);
 405     job->co = qemu_coroutine_create(block_job_co_entry, job);
 406     job->pause_count--;
 407     job->busy = true;
 408     job->paused = false;
 409     block_job_state_transition(job, BLOCK_JOB_STATUS_RUNNING);
 410     bdrv_coroutine_enter(blk_bs(job->blk), job->co);
 411 }
 412
 413 static void block_job_decommission(BlockJob *job)
 414 {
 415     assert(job);
 416     job->completed = true;
 417     job->busy = false;
 418     job->paused = false;
 419     job->deferred_to_main_loop = true;
 420     block_job_txn_del_job(job);
 421     block_job_state_transition(job, BLOCK_JOB_STATUS_NULL);
 422     block_job_unref(job);
 423 }
 424
 425 static void block_job_do_dismiss(BlockJob *job)
 426 {
 427     block_job_decommission(job);
 428 }
 429
 430 static void block_job_conclude(BlockJob *job)
 431 {
 432     block_job_state_transition(job, BLOCK_JOB_STATUS_CONCLUDED);
 433     if (job->auto_dismiss || !block_job_started(job)) {
 434         block_job_do_dismiss(job);
 435     }
 436 }
 437
 438 static void block_job_update_rc(BlockJob *job)
 439 {
 440     if (!job->ret && block_job_is_cancelled(job)) {
 441         job->ret = -ECANCELED;
 442     }
 443     if (job->ret) {
 444         block_job_state_transition(job, BLOCK_JOB_STATUS_ABORTING);
 445     }
 446 }
 447
 448 static int block_job_prepare(BlockJob *job)
 449 {
 450     if (job->ret == 0 && job->driver->prepare) {
 451         job->ret = job->driver->prepare(job);
 452     }
 453     return job->ret;
 454 }
 455
 456 static void block_job_commit(BlockJob *job)
 457 {
 458     assert(!job->ret);
 459     if (job->driver->commit) {
 460         job->driver->commit(job);
 461     }
 462 }
 463
 464 static void block_job_abort(BlockJob *job)
 465 {
 466     assert(job->ret);
 467     if (job->driver->abort) {
 468         job->driver->abort(job);
 469     }
 470 }
 471
 472 static void block_job_clean(BlockJob *job)
 473 {
 474     if (job->driver->clean) {
 475         job->driver->clean(job);
 476     }
 477 }
 478
 479 static int block_job_finalize_single(BlockJob *job)
 480 {
 481     assert(job->completed);
 482
 483     /* Ensure abort is called for late-transactional failures */
 484     block_job_update_rc(job);
 485
 486     if (!job->ret) {
 487         block_job_commit(job);
 488     } else {
 489         block_job_abort(job);
 490     }
 491     block_job_clean(job);
 492
 493     if (job->cb) {
 494         job->cb(job->opaque, job->ret);
 495     }
 496
 497     /* Emit events only if we actually started */
 498     if (block_job_started(job)) {
 499         if (block_job_is_cancelled(job)) {
 500             block_job_event_cancelled(job);
 501         } else {
 502             const char *msg = NULL;
 503             if (job->ret < 0) {
 504                 msg = strerror(-job->ret);
 505             }
 506             block_job_event_completed(job, msg);
 507         }
 508     }
 509
 510     block_job_txn_del_job(job);
 511     block_job_conclude(job);
 512     return 0;
 513 }
 514
 515 static void block_job_cancel_async(BlockJob *job, bool force)
 516 {
 517     if (job->iostatus != BLOCK_DEVICE_IO_STATUS_OK) {
 518         block_job_iostatus_reset(job);
 519     }
 520     if (job->user_paused) {
 521         /* Do not call block_job_enter here, the caller will handle it.  */
 522         job->user_paused = false;
 523         job->pause_count--;
 524     }
 525     job->cancelled = true;
 526     /* To prevent 'force == false' overriding a previous 'force == true' */
 527     job->force |= force;
 528 }
 529
 530 static int block_job_txn_apply(BlockJobTxn *txn, int fn(BlockJob *), bool lock)
 531 {
 532     AioContext *ctx;
 533     BlockJob *job, *next;
 534     int rc = 0;
 535
 536     QLIST_FOREACH_SAFE(job, &txn->jobs, txn_list, next) {
 537         if (lock) {
 538             ctx = blk_get_aio_context(job->blk);
 539             aio_context_acquire(ctx);
 540         }
 541         rc = fn(job);
 542         if (lock) {
 543             aio_context_release(ctx);
 544         }
 545         if (rc) {
 546             break;
 547         }
 548     }
 549     return rc;
 550 }
 551
 552 static int block_job_finish_sync(BlockJob *job,
 553                                  void (*finish)(BlockJob *, Error **errp),
 554                                  Error **errp)
 555 {
 556     Error *local_err = NULL;
 557     int ret;
 558
 559     assert(blk_bs(job->blk)->job == job);
 560
 561     block_job_ref(job);
 562
 563     if (finish) {
 564         finish(job, &local_err);
 565     }
 566     if (local_err) {
 567         error_propagate(errp, local_err);
 568         block_job_unref(job);
 569         return -EBUSY;
 570     }
 571     /* block_job_drain calls block_job_enter, and it should be enough to
 572      * induce progress until the job completes or moves to the main thread.
 573     */
 574     while (!job->deferred_to_main_loop && !job->completed) {
 575         block_job_drain(job);
 576     }
 577     while (!job->completed) {
 578         aio_poll(qemu_get_aio_context(), true);
 579     }
 580     ret = (job->cancelled && job->ret == 0) ? -ECANCELED : job->ret;
 581     block_job_unref(job);
 582     return ret;
 583 }
 584
 585 static void block_job_completed_txn_abort(BlockJob *job)
 586 {
 587     AioContext *ctx;
 588     BlockJobTxn *txn = job->txn;
 589     BlockJob *other_job;
 590
 591     if (txn->aborting) {
 592         /*
 593          * We are cancelled by another job, which will handle everything.
 594          */
 595         return;
 596     }
 597     txn->aborting = true;
 598     block_job_txn_ref(txn);
 599
 600     /* We are the first failed job. Cancel other jobs. */
 601     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 602         ctx = blk_get_aio_context(other_job->blk);
 603         aio_context_acquire(ctx);
 604     }
 605
 606     /* Other jobs are effectively cancelled by us, set the status for
 607      * them; this job, however, may or may not be cancelled, depending
 608      * on the caller, so leave it. */
 609     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 610         if (other_job != job) {
 611             block_job_cancel_async(other_job, false);
 612         }
 613     }
 614     while (!QLIST_EMPTY(&txn->jobs)) {
 615         other_job = QLIST_FIRST(&txn->jobs);
 616         ctx = blk_get_aio_context(other_job->blk);
 617         if (!other_job->completed) {
 618             assert(other_job->cancelled);
 619             block_job_finish_sync(other_job, NULL, NULL);
 620         }
 621         block_job_finalize_single(other_job);
 622         aio_context_release(ctx);
 623     }
 624
 625     block_job_txn_unref(txn);
 626 }
 627
 628 static int block_job_needs_finalize(BlockJob *job)
 629 {
 630     return !job->auto_finalize;
 631 }
 632
 633 static void block_job_do_finalize(BlockJob *job)
 634 {
 635     int rc;
 636     assert(job && job->txn);
 637
 638     /* prepare the transaction to complete */
 639     rc = block_job_txn_apply(job->txn, block_job_prepare, true);
 640     if (rc) {
 641         block_job_completed_txn_abort(job);
 642     } else {
 643         block_job_txn_apply(job->txn, block_job_finalize_single, true);
 644     }
 645 }
 646
 647 static void block_job_completed_txn_success(BlockJob *job)
 648 {
 649     BlockJobTxn *txn = job->txn;
 650     BlockJob *other_job;
 651
 652     block_job_state_transition(job, BLOCK_JOB_STATUS_WAITING);
 653
 654     /*
 655      * Successful completion, see if there are other running jobs in this
 656      * txn.
 657      */
 658     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 659         if (!other_job->completed) {
 660             return;
 661         }
 662         assert(other_job->ret == 0);
 663     }
 664
 665     block_job_txn_apply(txn, block_job_event_pending, false);
 666
 667     /* If no jobs need manual finalization, automatically do so */
 668     if (block_job_txn_apply(txn, block_job_needs_finalize, false) == 0) {
 669         block_job_do_finalize(job);
 670     }
 671 }
 672
 673 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
 674 {
 675     int64_t old_speed = job->speed;
 676
 677     if (block_job_apply_verb(job, BLOCK_JOB_VERB_SET_SPEED, errp)) {
 678         return;
 679     }
 680     if (speed < 0) {
 681         error_setg(errp, QERR_INVALID_PARAMETER, "speed");
 682         return;
 683     }
 684
 685     ratelimit_set_speed(&job->limit, speed, BLOCK_JOB_SLICE_TIME);
 686
 687     job->speed = speed;
 688     if (speed && speed <= old_speed) {
 689         return;
 690     }
 691
 692     /* kick only if a timer is pending */
 693     block_job_enter_cond(job, block_job_timer_pending);
 694 }
 695
 696 int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
 697 {
 698     if (!job->speed) {
 699         return 0;
 700     }
 701
 702     return ratelimit_calculate_delay(&job->limit, n);
 703 }
 704
 705 void block_job_complete(BlockJob *job, Error **errp)
 706 {
 707     /* Should not be reachable via external interface for internal jobs */
 708     assert(job->id);
 709     if (block_job_apply_verb(job, BLOCK_JOB_VERB_COMPLETE, errp)) {
 710         return;
 711     }
 712     if (job->pause_count || job->cancelled || !job->driver->complete) {
 713         error_setg(errp, "The active block job '%s' cannot be completed",
 714                    job->id);
 715         return;
 716     }
 717
 718     job->driver->complete(job, errp);
 719 }
 720
 721 void block_job_finalize(BlockJob *job, Error **errp)
 722 {
 723     assert(job && job->id);
 724     if (block_job_apply_verb(job, BLOCK_JOB_VERB_FINALIZE, errp)) {
 725         return;
 726     }
 727     block_job_do_finalize(job);
 728 }
 729
 730 void block_job_dismiss(BlockJob **jobptr, Error **errp)
 731 {
 732     BlockJob *job = *jobptr;
 733     /* similarly to _complete, this is QMP-interface only. */
 734     assert(job->id);
 735     if (block_job_apply_verb(job, BLOCK_JOB_VERB_DISMISS, errp)) {
 736         return;
 737     }
 738
 739     block_job_do_dismiss(job);
 740     *jobptr = NULL;
 741 }
 742
 743 void block_job_user_pause(BlockJob *job, Error **errp)
 744 {
 745     if (block_job_apply_verb(job, BLOCK_JOB_VERB_PAUSE, errp)) {
 746         return;
 747     }
 748     if (job->user_paused) {
 749         error_setg(errp, "Job is already paused");
 750         return;
 751     }
 752     job->user_paused = true;
 753     block_job_pause(job);
 754 }
 755
 756 bool block_job_user_paused(BlockJob *job)
 757 {
 758     return job->user_paused;
 759 }
 760
 761 void block_job_user_resume(BlockJob *job, Error **errp)
 762 {
 763     assert(job);
 764     if (!job->user_paused || job->pause_count <= 0) {
 765         error_setg(errp, "Can't resume a job that was not paused");
 766         return;
 767     }
 768     if (block_job_apply_verb(job, BLOCK_JOB_VERB_RESUME, errp)) {
 769         return;
 770     }
 771     block_job_iostatus_reset(job);
 772     job->user_paused = false;
 773     block_job_resume(job);
 774 }
 775
 776 void block_job_cancel(BlockJob *job, bool force)
 777 {
 778     if (job->status == BLOCK_JOB_STATUS_CONCLUDED) {
 779         block_job_do_dismiss(job);
 780         return;
 781     }
 782     block_job_cancel_async(job, force);
 783     if (!block_job_started(job)) {
 784         block_job_completed(job, -ECANCELED);
 785     } else if (job->deferred_to_main_loop) {
 786         block_job_completed_txn_abort(job);
 787     } else {
 788         block_job_enter(job);
 789     }
 790 }
 791
 792 void block_job_user_cancel(BlockJob *job, bool force, Error **errp)
 793 {
 794     if (block_job_apply_verb(job, BLOCK_JOB_VERB_CANCEL, errp)) {
 795         return;
 796     }
 797     block_job_cancel(job, force);
 798 }
 799
 800 /* A wrapper around block_job_cancel() taking an Error ** parameter so it may be
 801  * used with block_job_finish_sync() without the need for (rather nasty)
 802  * function pointer casts there. */
 803 static void block_job_cancel_err(BlockJob *job, Error **errp)
 804 {
 805     block_job_cancel(job, false);
 806 }
 807
 808 int block_job_cancel_sync(BlockJob *job)
 809 {
 810     return block_job_finish_sync(job, &block_job_cancel_err, NULL);
 811 }
 812
 813 void block_job_cancel_sync_all(void)
 814 {
 815     BlockJob *job;
 816     AioContext *aio_context;
 817
 818     while ((job = QLIST_FIRST(&block_jobs))) {
 819         aio_context = blk_get_aio_context(job->blk);
 820         aio_context_acquire(aio_context);
 821         block_job_cancel_sync(job);
 822         aio_context_release(aio_context);
 823     }
 824 }
 825
 826 int block_job_complete_sync(BlockJob *job, Error **errp)
 827 {
 828     return block_job_finish_sync(job, &block_job_complete, errp);
 829 }
 830
 831 void block_job_progress_update(BlockJob *job, uint64_t done)
 832 {
 833     job->offset += done;
 834 }
 835
 836 void block_job_progress_set_remaining(BlockJob *job, uint64_t remaining)
 837 {
 838     job->len = job->offset + remaining;
 839 }
 840
 841 BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
 842 {
 843     BlockJobInfo *info;
 844
 845     if (block_job_is_internal(job)) {
 846         error_setg(errp, "Cannot query QEMU internal jobs");
 847         return NULL;
 848     }
 849     info = g_new0(BlockJobInfo, 1);
 850     info->type      = g_strdup(BlockJobType_str(job->driver->job_type));
 851     info->device    = g_strdup(job->id);
 852     info->len       = job->len;
 853     info->busy      = atomic_read(&job->busy);
 854     info->paused    = job->pause_count > 0;
 855     info->offset    = job->offset;
 856     info->speed     = job->speed;
 857     info->io_status = job->iostatus;
 858     info->ready     = job->ready;
 859     info->status    = job->status;
 860     info->auto_finalize = job->auto_finalize;
 861     info->auto_dismiss  = job->auto_dismiss;
 862     info->has_error = job->ret != 0;
 863     info->error     = job->ret ? g_strdup(strerror(-job->ret)) : NULL;
 864     return info;
 865 }
 866
 867 static void block_job_iostatus_set_err(BlockJob *job, int error)
 868 {
 869     if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
 870         job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
 871                                           BLOCK_DEVICE_IO_STATUS_FAILED;
 872     }
 873 }
 874
 875 static void block_job_event_cancelled(BlockJob *job)
 876 {
 877     if (block_job_is_internal(job)) {
 878         return;
 879     }
 880
 881     qapi_event_send_block_job_cancelled(job->driver->job_type,
 882                                         job->id,
 883                                         job->len,
 884                                         job->offset,
 885                                         job->speed,
 886                                         &error_abort);
 887 }
 888
 889 static void block_job_event_completed(BlockJob *job, const char *msg)
 890 {
 891     if (block_job_is_internal(job)) {
 892         return;
 893     }
 894
 895     qapi_event_send_block_job_completed(job->driver->job_type,
 896                                         job->id,
 897                                         job->len,
 898                                         job->offset,
 899                                         job->speed,
 900                                         !!msg,
 901                                         msg,
 902                                         &error_abort);
 903 }
 904
 905 static int block_job_event_pending(BlockJob *job)
 906 {
 907     block_job_state_transition(job, BLOCK_JOB_STATUS_PENDING);
 908     if (!job->auto_finalize && !block_job_is_internal(job)) {
 909         qapi_event_send_block_job_pending(job->driver->job_type,
 910                                           job->id,
 911                                           &error_abort);
 912     }
 913     return 0;
 914 }
 915
 916 /*
 917  * API for block job drivers and the block layer.  These functions are
 918  * declared in blockjob_int.h.
 919  */
 920
 921 void *block_job_create(const char *job_id, const BlockJobDriver *driver,
 922                        BlockJobTxn *txn, BlockDriverState *bs, uint64_t perm,
 923                        uint64_t shared_perm, int64_t speed, int flags,
 924                        BlockCompletionFunc *cb, void *opaque, Error **errp)
 925 {
 926     BlockBackend *blk;
 927     BlockJob *job;
 928     int ret;
 929
 930     if (bs->job) {
 931         error_setg(errp, QERR_DEVICE_IN_USE, bdrv_get_device_name(bs));
 932         return NULL;
 933     }
 934
 935     if (job_id == NULL && !(flags & BLOCK_JOB_INTERNAL)) {
 936         job_id = bdrv_get_device_name(bs);
 937         if (!*job_id) {
 938             error_setg(errp, "An explicit job ID is required for this node");
 939             return NULL;
 940         }
 941     }
 942
 943     if (job_id) {
 944         if (flags & BLOCK_JOB_INTERNAL) {
 945             error_setg(errp, "Cannot specify job ID for internal block job");
 946             return NULL;
 947         }
 948
 949         if (!id_wellformed(job_id)) {
 950             error_setg(errp, "Invalid job ID '%s'", job_id);
 951             return NULL;
 952         }
 953
 954         if (block_job_get(job_id)) {
 955             error_setg(errp, "Job ID '%s' already in use", job_id);
 956             return NULL;
 957         }
 958     }
 959
 960     blk = blk_new(perm, shared_perm);
 961     ret = blk_insert_bs(blk, bs, errp);
 962     if (ret < 0) {
 963         blk_unref(blk);
 964         return NULL;
 965     }
 966
 967     job = g_malloc0(driver->instance_size);
 968     job->driver        = driver;
 969     job->id            = g_strdup(job_id);
 970     job->blk           = blk;
 971     job->cb            = cb;
 972     job->opaque        = opaque;
 973     job->busy          = false;
 974     job->paused        = true;
 975     job->pause_count   = 1;
 976     job->refcnt        = 1;
 977     job->auto_finalize = !(flags & BLOCK_JOB_MANUAL_FINALIZE);
 978     job->auto_dismiss  = !(flags & BLOCK_JOB_MANUAL_DISMISS);
 979     block_job_state_transition(job, BLOCK_JOB_STATUS_CREATED);
 980     aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
 981                    QEMU_CLOCK_REALTIME, SCALE_NS,
 982                    block_job_sleep_timer_cb, job);
 983
 984     error_setg(&job->blocker, "block device is in use by block job: %s",
 985                BlockJobType_str(driver->job_type));
 986     block_job_add_bdrv(job, "main node", bs, 0, BLK_PERM_ALL, &error_abort);
 987     bs->job = job;
 988
 989     bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker);
 990
 991     QLIST_INSERT_HEAD(&block_jobs, job, job_list);
 992
 993     blk_add_aio_context_notifier(blk, block_job_attached_aio_context,
 994                                  block_job_detach_aio_context, job);
 995
 996     /* Only set speed when necessary to avoid NotSupported error */
 997     if (speed != 0) {
 998         Error *local_err = NULL;
 999
1000         block_job_set_speed(job, speed, &local_err);
1001         if (local_err) {
1002             block_job_early_fail(job);
1003             error_propagate(errp, local_err);
1004             return NULL;
1005         }
1006     }
1007
1008     /* Single jobs are modeled as single-job transactions for sake of
1009      * consolidating the job management logic */
1010     if (!txn) {
1011         txn = block_job_txn_new();
1012         block_job_txn_add_job(txn, job);
1013         block_job_txn_unref(txn);
1014     } else {
1015         block_job_txn_add_job(txn, job);
1016     }
1017
1018     return job;
1019 }
1020
1021 void block_job_early_fail(BlockJob *job)
1022 {
1023     assert(job->status == BLOCK_JOB_STATUS_CREATED);
1024     block_job_decommission(job);
1025 }
1026
1027 void block_job_completed(BlockJob *job, int ret)
1028 {
1029     assert(job && job->txn && !job->completed);
1030     assert(blk_bs(job->blk)->job == job);
1031     job->completed = true;
1032     job->ret = ret;
1033     block_job_update_rc(job);
1034     trace_block_job_completed(job, ret, job->ret);
1035     if (job->ret) {
1036         block_job_completed_txn_abort(job);
1037     } else {
1038         block_job_completed_txn_success(job);
1039     }
1040 }
1041
1042 static bool block_job_should_pause(BlockJob *job)
1043 {
1044     return job->pause_count > 0;
1045 }
1046
1047 /* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
1048  * Reentering the job coroutine with block_job_enter() before the timer has
1049  * expired is allowed and cancels the timer.
1050  *
1051  * If @ns is (uint64_t) -1, no timer is scheduled and block_job_enter() must be
1052  * called explicitly. */
1053 static void block_job_do_yield(BlockJob *job, uint64_t ns)
1054 {
1055     block_job_lock();
1056     if (ns != -1) {
1057         timer_mod(&job->sleep_timer, ns);
1058     }
1059     job->busy = false;
1060     block_job_unlock();
1061     qemu_coroutine_yield();
1062
1063     /* Set by block_job_enter before re-entering the coroutine.  */
1064     assert(job->busy);
1065 }
1066
1067 void coroutine_fn block_job_pause_point(BlockJob *job)
1068 {
1069     assert(job && block_job_started(job));
1070
1071     if (!block_job_should_pause(job)) {
1072         return;
1073     }
1074     if (block_job_is_cancelled(job)) {
1075         return;
1076     }
1077
1078     if (job->driver->pause) {
1079         job->driver->pause(job);
1080     }
1081
1082     if (block_job_should_pause(job) && !block_job_is_cancelled(job)) {
1083         BlockJobStatus status = job->status;
1084         block_job_state_transition(job, status == BLOCK_JOB_STATUS_READY ? \
1085                                    BLOCK_JOB_STATUS_STANDBY :           \
1086                                    BLOCK_JOB_STATUS_PAUSED);
1087         job->paused = true;
1088         block_job_do_yield(job, -1);
1089         job->paused = false;
1090         block_job_state_transition(job, status);
1091     }
1092
1093     if (job->driver->resume) {
1094         job->driver->resume(job);
1095     }
1096 }
1097
1098 /*
1099  * Conditionally enter a block_job pending a call to fn() while
1100  * under the block_job_lock critical section.
1101  */
1102 static void block_job_enter_cond(BlockJob *job, bool(*fn)(BlockJob *job))
1103 {
1104     if (!block_job_started(job)) {
1105         return;
1106     }
1107     if (job->deferred_to_main_loop) {
1108         return;
1109     }
1110
1111     block_job_lock();
1112     if (job->busy) {
1113         block_job_unlock();
1114         return;
1115     }
1116
1117     if (fn && !fn(job)) {
1118         block_job_unlock();
1119         return;
1120     }
1121
1122     assert(!job->deferred_to_main_loop);
1123     timer_del(&job->sleep_timer);
1124     job->busy = true;
1125     block_job_unlock();
1126     aio_co_wake(job->co);
1127 }
1128
1129 void block_job_enter(BlockJob *job)
1130 {
1131     block_job_enter_cond(job, NULL);
1132 }
1133
1134 bool block_job_is_cancelled(BlockJob *job)
1135 {
1136     return job->cancelled;
1137 }
1138
1139 void block_job_sleep_ns(BlockJob *job, int64_t ns)
1140 {
1141     assert(job->busy);
1142
1143     /* Check cancellation *before* setting busy = false, too!  */
1144     if (block_job_is_cancelled(job)) {
1145         return;
1146     }
1147
1148     if (!block_job_should_pause(job)) {
1149         block_job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
1150     }
1151
1152     block_job_pause_point(job);
1153 }
1154
1155 void block_job_yield(BlockJob *job)
1156 {
1157     assert(job->busy);
1158
1159     /* Check cancellation *before* setting busy = false, too!  */
1160     if (block_job_is_cancelled(job)) {
1161         return;
1162     }
1163
1164     if (!block_job_should_pause(job)) {
1165         block_job_do_yield(job, -1);
1166     }
1167
1168     block_job_pause_point(job);
1169 }
1170
1171 void block_job_iostatus_reset(BlockJob *job)
1172 {
1173     if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
1174         return;
1175     }
1176     assert(job->user_paused && job->pause_count > 0);
1177     job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
1178 }
1179
1180 void block_job_event_ready(BlockJob *job)
1181 {
1182     block_job_state_transition(job, BLOCK_JOB_STATUS_READY);
1183     job->ready = true;
1184
1185     if (block_job_is_internal(job)) {
1186         return;
1187     }
1188
1189     qapi_event_send_block_job_ready(job->driver->job_type,
1190                                     job->id,
1191                                     job->len,
1192                                     job->offset,
1193                                     job->speed, &error_abort);
1194 }
1195
1196 BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
1197                                         int is_read, int error)
1198 {
1199     BlockErrorAction action;
1200
1201     switch (on_err) {
1202     case BLOCKDEV_ON_ERROR_ENOSPC:
1203     case BLOCKDEV_ON_ERROR_AUTO:
1204         action = (error == ENOSPC) ?
1205                  BLOCK_ERROR_ACTION_STOP : BLOCK_ERROR_ACTION_REPORT;
1206         break;
1207     case BLOCKDEV_ON_ERROR_STOP:
1208         action = BLOCK_ERROR_ACTION_STOP;
1209         break;
1210     case BLOCKDEV_ON_ERROR_REPORT:
1211         action = BLOCK_ERROR_ACTION_REPORT;
1212         break;
1213     case BLOCKDEV_ON_ERROR_IGNORE:
1214         action = BLOCK_ERROR_ACTION_IGNORE;
1215         break;
1216     default:
1217         abort();
1218     }
1219     if (!block_job_is_internal(job)) {
1220         qapi_event_send_block_job_error(job->id,
1221                                         is_read ? IO_OPERATION_TYPE_READ :
1222                                         IO_OPERATION_TYPE_WRITE,
1223                                         action, &error_abort);
1224     }
1225     if (action == BLOCK_ERROR_ACTION_STOP) {
1226         block_job_pause(job);
1227         /* make the pause user visible, which will be resumed from QMP. */
1228         job->user_paused = true;
1229         block_job_iostatus_set_err(job, error);
1230     }
1231     return action;
1232 }
1233
1234 typedef struct {
1235     BlockJob *job;
1236     AioContext *aio_context;
1237     BlockJobDeferToMainLoopFn *fn;
1238     void *opaque;
1239 } BlockJobDeferToMainLoopData;
1240
1241 static void block_job_defer_to_main_loop_bh(void *opaque)
1242 {
1243     BlockJobDeferToMainLoopData *data = opaque;
1244     AioContext *aio_context;
1245
1246     /* Prevent race with block_job_defer_to_main_loop() */
1247     aio_context_acquire(data->aio_context);
1248
1249     /* Fetch BDS AioContext again, in case it has changed */
1250     aio_context = blk_get_aio_context(data->job->blk);
1251     if (aio_context != data->aio_context) {
1252         aio_context_acquire(aio_context);
1253     }
1254
1255     data->fn(data->job, data->opaque);
1256
1257     if (aio_context != data->aio_context) {
1258         aio_context_release(aio_context);
1259     }
1260
1261     aio_context_release(data->aio_context);
1262
1263     g_free(data);
1264 }
1265
1266 void block_job_defer_to_main_loop(BlockJob *job,
1267                                   BlockJobDeferToMainLoopFn *fn,
1268                                   void *opaque)
1269 {
1270     BlockJobDeferToMainLoopData *data = g_malloc(sizeof(*data));
1271     data->job = job;
1272     data->aio_context = blk_get_aio_context(job->blk);
1273     data->fn = fn;
1274     data->opaque = opaque;
1275     job->deferred_to_main_loop = true;
1276
1277     aio_bh_schedule_oneshot(qemu_get_aio_context(),
1278                             block_job_defer_to_main_loop_bh, data);
1279 }