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/timer.h"
  38
  39 /* Transactional group of block jobs */
  40 struct BlockJobTxn {
  41
  42     /* Is this txn being cancelled? */
  43     bool aborting;
  44
  45     /* List of jobs */
  46     QLIST_HEAD(, BlockJob) jobs;
  47
  48     /* Reference count */
  49     int refcnt;
  50 };
  51
  52 /*
  53  * The block job API is composed of two categories of functions.
  54  *
  55  * The first includes functions used by the monitor.  The monitor is
  56  * peculiar in that it accesses the block job list with block_job_get, and
  57  * therefore needs consistency across block_job_get and the actual operation
  58  * (e.g. block_job_set_speed).  The consistency is achieved with
  59  * aio_context_acquire/release.  These functions are declared in blockjob.h.
  60  *
  61  * The second includes functions used by the block job drivers and sometimes
  62  * by the core block layer.  These do not care about locking, because the
  63  * whole coroutine runs under the AioContext lock, and are declared in
  64  * blockjob_int.h.
  65  */
  66
  67 static bool is_block_job(Job *job)
  68 {
  69     return job_type(job) == JOB_TYPE_BACKUP ||
  70            job_type(job) == JOB_TYPE_COMMIT ||
  71            job_type(job) == JOB_TYPE_MIRROR ||
  72            job_type(job) == JOB_TYPE_STREAM;
  73 }
  74
  75 BlockJob *block_job_next(BlockJob *bjob)
  76 {
  77     Job *job = bjob ? &bjob->job : NULL;
  78
  79     do {
  80         job = job_next(job);
  81     } while (job && !is_block_job(job));
  82
  83     return job ? container_of(job, BlockJob, job) : NULL;
  84 }
  85
  86 BlockJob *block_job_get(const char *id)
  87 {
  88     Job *job = job_get(id);
  89
  90     if (job && is_block_job(job)) {
  91         return container_of(job, BlockJob, job);
  92     } else {
  93         return NULL;
  94     }
  95 }
  96
  97 BlockJobTxn *block_job_txn_new(void)
  98 {
  99     BlockJobTxn *txn = g_new0(BlockJobTxn, 1);
 100     QLIST_INIT(&txn->jobs);
 101     txn->refcnt = 1;
 102     return txn;
 103 }
 104
 105 static void block_job_txn_ref(BlockJobTxn *txn)
 106 {
 107     txn->refcnt++;
 108 }
 109
 110 void block_job_txn_unref(BlockJobTxn *txn)
 111 {
 112     if (txn && --txn->refcnt == 0) {
 113         g_free(txn);
 114     }
 115 }
 116
 117 void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job)
 118 {
 119     if (!txn) {
 120         return;
 121     }
 122
 123     assert(!job->txn);
 124     job->txn = txn;
 125
 126     QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
 127     block_job_txn_ref(txn);
 128 }
 129
 130 void block_job_txn_del_job(BlockJob *job)
 131 {
 132     if (job->txn) {
 133         QLIST_REMOVE(job, txn_list);
 134         block_job_txn_unref(job->txn);
 135         job->txn = NULL;
 136     }
 137 }
 138
 139 static void block_job_attached_aio_context(AioContext *new_context,
 140                                            void *opaque);
 141 static void block_job_detach_aio_context(void *opaque);
 142
 143 void block_job_free(Job *job)
 144 {
 145     BlockJob *bjob = container_of(job, BlockJob, job);
 146     BlockDriverState *bs = blk_bs(bjob->blk);
 147
 148     assert(!bjob->txn);
 149
 150     bs->job = NULL;
 151     block_job_remove_all_bdrv(bjob);
 152     blk_remove_aio_context_notifier(bjob->blk,
 153                                     block_job_attached_aio_context,
 154                                     block_job_detach_aio_context, bjob);
 155     blk_unref(bjob->blk);
 156     error_free(bjob->blocker);
 157 }
 158
 159 static void block_job_attached_aio_context(AioContext *new_context,
 160                                            void *opaque)
 161 {
 162     BlockJob *job = opaque;
 163
 164     job->job.aio_context = new_context;
 165     if (job->driver->attached_aio_context) {
 166         job->driver->attached_aio_context(job, new_context);
 167     }
 168
 169     job_resume(&job->job);
 170 }
 171
 172 void block_job_drain(Job *job)
 173 {
 174     BlockJob *bjob = container_of(job, BlockJob, job);
 175
 176     blk_drain(bjob->blk);
 177     if (bjob->driver->drain) {
 178         bjob->driver->drain(bjob);
 179     }
 180 }
 181
 182 static void block_job_detach_aio_context(void *opaque)
 183 {
 184     BlockJob *job = opaque;
 185
 186     /* In case the job terminates during aio_poll()... */
 187     job_ref(&job->job);
 188
 189     job_pause(&job->job);
 190
 191     while (!job->job.paused && !job_is_completed(&job->job)) {
 192         job_drain(&job->job);
 193     }
 194
 195     job->job.aio_context = NULL;
 196     job_unref(&job->job);
 197 }
 198
 199 static char *child_job_get_parent_desc(BdrvChild *c)
 200 {
 201     BlockJob *job = c->opaque;
 202     return g_strdup_printf("%s job '%s'", job_type_str(&job->job), job->job.id);
 203 }
 204
 205 static void child_job_drained_begin(BdrvChild *c)
 206 {
 207     BlockJob *job = c->opaque;
 208     job_pause(&job->job);
 209 }
 210
 211 static void child_job_drained_end(BdrvChild *c)
 212 {
 213     BlockJob *job = c->opaque;
 214     job_resume(&job->job);
 215 }
 216
 217 static const BdrvChildRole child_job = {
 218     .get_parent_desc    = child_job_get_parent_desc,
 219     .drained_begin      = child_job_drained_begin,
 220     .drained_end        = child_job_drained_end,
 221     .stay_at_node       = true,
 222 };
 223
 224 void block_job_remove_all_bdrv(BlockJob *job)
 225 {
 226     GSList *l;
 227     for (l = job->nodes; l; l = l->next) {
 228         BdrvChild *c = l->data;
 229         bdrv_op_unblock_all(c->bs, job->blocker);
 230         bdrv_root_unref_child(c);
 231     }
 232     g_slist_free(job->nodes);
 233     job->nodes = NULL;
 234 }
 235
 236 int block_job_add_bdrv(BlockJob *job, const char *name, BlockDriverState *bs,
 237                        uint64_t perm, uint64_t shared_perm, Error **errp)
 238 {
 239     BdrvChild *c;
 240
 241     c = bdrv_root_attach_child(bs, name, &child_job, perm, shared_perm,
 242                                job, errp);
 243     if (c == NULL) {
 244         return -EPERM;
 245     }
 246
 247     job->nodes = g_slist_prepend(job->nodes, c);
 248     bdrv_ref(bs);
 249     bdrv_op_block_all(bs, job->blocker);
 250
 251     return 0;
 252 }
 253
 254 bool block_job_is_internal(BlockJob *job)
 255 {
 256     return (job->job.id == NULL);
 257 }
 258
 259 const BlockJobDriver *block_job_driver(BlockJob *job)
 260 {
 261     return job->driver;
 262 }
 263
 264 static int block_job_prepare(BlockJob *job)
 265 {
 266     if (job->job.ret == 0 && job->driver->prepare) {
 267         job->job.ret = job->driver->prepare(job);
 268     }
 269     return job->job.ret;
 270 }
 271
 272 static void job_cancel_async(Job *job, bool force)
 273 {
 274     if (job->user_paused) {
 275         /* Do not call job_enter here, the caller will handle it.  */
 276         job->user_paused = false;
 277         if (job->driver->user_resume) {
 278             job->driver->user_resume(job);
 279         }
 280         assert(job->pause_count > 0);
 281         job->pause_count--;
 282     }
 283     job->cancelled = true;
 284     /* To prevent 'force == false' overriding a previous 'force == true' */
 285     job->force_cancel |= force;
 286 }
 287
 288 static int block_job_txn_apply(BlockJobTxn *txn, int fn(BlockJob *), bool lock)
 289 {
 290     AioContext *ctx;
 291     BlockJob *job, *next;
 292     int rc = 0;
 293
 294     QLIST_FOREACH_SAFE(job, &txn->jobs, txn_list, next) {
 295         if (lock) {
 296             ctx = blk_get_aio_context(job->blk);
 297             aio_context_acquire(ctx);
 298         }
 299         rc = fn(job);
 300         if (lock) {
 301             aio_context_release(ctx);
 302         }
 303         if (rc) {
 304             break;
 305         }
 306     }
 307     return rc;
 308 }
 309
 310 static int block_job_finish_sync(BlockJob *job,
 311                                  void (*finish)(BlockJob *, Error **errp),
 312                                  Error **errp)
 313 {
 314     Error *local_err = NULL;
 315     int ret;
 316
 317     assert(blk_bs(job->blk)->job == job);
 318
 319     job_ref(&job->job);
 320
 321     if (finish) {
 322         finish(job, &local_err);
 323     }
 324     if (local_err) {
 325         error_propagate(errp, local_err);
 326         job_unref(&job->job);
 327         return -EBUSY;
 328     }
 329     /* job_drain calls job_enter, and it should be enough to induce progress
 330      * until the job completes or moves to the main thread.
 331     */
 332     while (!job->job.deferred_to_main_loop && !job_is_completed(&job->job)) {
 333         job_drain(&job->job);
 334     }
 335     while (!job_is_completed(&job->job)) {
 336         aio_poll(qemu_get_aio_context(), true);
 337     }
 338     ret = (job_is_cancelled(&job->job) && job->job.ret == 0)
 339           ? -ECANCELED : job->job.ret;
 340     job_unref(&job->job);
 341     return ret;
 342 }
 343
 344 static void block_job_completed_txn_abort(BlockJob *job)
 345 {
 346     AioContext *ctx;
 347     BlockJobTxn *txn = job->txn;
 348     BlockJob *other_job;
 349
 350     if (txn->aborting) {
 351         /*
 352          * We are cancelled by another job, which will handle everything.
 353          */
 354         return;
 355     }
 356     txn->aborting = true;
 357     block_job_txn_ref(txn);
 358
 359     /* We are the first failed job. Cancel other jobs. */
 360     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 361         ctx = blk_get_aio_context(other_job->blk);
 362         aio_context_acquire(ctx);
 363     }
 364
 365     /* Other jobs are effectively cancelled by us, set the status for
 366      * them; this job, however, may or may not be cancelled, depending
 367      * on the caller, so leave it. */
 368     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 369         if (other_job != job) {
 370             job_cancel_async(&other_job->job, false);
 371         }
 372     }
 373     while (!QLIST_EMPTY(&txn->jobs)) {
 374         other_job = QLIST_FIRST(&txn->jobs);
 375         ctx = blk_get_aio_context(other_job->blk);
 376         if (!job_is_completed(&other_job->job)) {
 377             assert(job_is_cancelled(&other_job->job));
 378             block_job_finish_sync(other_job, NULL, NULL);
 379         }
 380         job_finalize_single(&other_job->job);
 381         aio_context_release(ctx);
 382     }
 383
 384     block_job_txn_unref(txn);
 385 }
 386
 387 static int block_job_needs_finalize(BlockJob *job)
 388 {
 389     return !job->job.auto_finalize;
 390 }
 391
 392 static int block_job_finalize_single(BlockJob *job)
 393 {
 394     return job_finalize_single(&job->job);
 395 }
 396
 397 static void block_job_do_finalize(BlockJob *job)
 398 {
 399     int rc;
 400     assert(job && job->txn);
 401
 402     /* prepare the transaction to complete */
 403     rc = block_job_txn_apply(job->txn, block_job_prepare, true);
 404     if (rc) {
 405         block_job_completed_txn_abort(job);
 406     } else {
 407         block_job_txn_apply(job->txn, block_job_finalize_single, true);
 408     }
 409 }
 410
 411 static int block_job_transition_to_pending(BlockJob *job)
 412 {
 413     job_state_transition(&job->job, JOB_STATUS_PENDING);
 414     if (!job->job.auto_finalize) {
 415         job_event_pending(&job->job);
 416     }
 417     return 0;
 418 }
 419
 420 static void block_job_completed_txn_success(BlockJob *job)
 421 {
 422     BlockJobTxn *txn = job->txn;
 423     BlockJob *other_job;
 424
 425     job_state_transition(&job->job, JOB_STATUS_WAITING);
 426
 427     /*
 428      * Successful completion, see if there are other running jobs in this
 429      * txn.
 430      */
 431     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 432         if (!job_is_completed(&other_job->job)) {
 433             return;
 434         }
 435         assert(other_job->job.ret == 0);
 436     }
 437
 438     block_job_txn_apply(txn, block_job_transition_to_pending, false);
 439
 440     /* If no jobs need manual finalization, automatically do so */
 441     if (block_job_txn_apply(txn, block_job_needs_finalize, false) == 0) {
 442         block_job_do_finalize(job);
 443     }
 444 }
 445
 446 /* Assumes the job_mutex is held */
 447 static bool job_timer_pending(Job *job)
 448 {
 449     return timer_pending(&job->sleep_timer);
 450 }
 451
 452 void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
 453 {
 454     int64_t old_speed = job->speed;
 455
 456     if (job_apply_verb(&job->job, JOB_VERB_SET_SPEED, errp)) {
 457         return;
 458     }
 459     if (speed < 0) {
 460         error_setg(errp, QERR_INVALID_PARAMETER, "speed");
 461         return;
 462     }
 463
 464     ratelimit_set_speed(&job->limit, speed, BLOCK_JOB_SLICE_TIME);
 465
 466     job->speed = speed;
 467     if (speed && speed <= old_speed) {
 468         return;
 469     }
 470
 471     /* kick only if a timer is pending */
 472     job_enter_cond(&job->job, job_timer_pending);
 473 }
 474
 475 int64_t block_job_ratelimit_get_delay(BlockJob *job, uint64_t n)
 476 {
 477     if (!job->speed) {
 478         return 0;
 479     }
 480
 481     return ratelimit_calculate_delay(&job->limit, n);
 482 }
 483
 484 void block_job_finalize(BlockJob *job, Error **errp)
 485 {
 486     assert(job && job->job.id);
 487     if (job_apply_verb(&job->job, JOB_VERB_FINALIZE, errp)) {
 488         return;
 489     }
 490     block_job_do_finalize(job);
 491 }
 492
 493 void block_job_dismiss(BlockJob **jobptr, Error **errp)
 494 {
 495     BlockJob *job = *jobptr;
 496     /* similarly to _complete, this is QMP-interface only. */
 497     assert(job->job.id);
 498     if (job_apply_verb(&job->job, JOB_VERB_DISMISS, errp)) {
 499         return;
 500     }
 501
 502     job_do_dismiss(&job->job);
 503     *jobptr = NULL;
 504 }
 505
 506 void block_job_cancel(BlockJob *job, bool force)
 507 {
 508     if (job->job.status == JOB_STATUS_CONCLUDED) {
 509         job_do_dismiss(&job->job);
 510         return;
 511     }
 512     job_cancel_async(&job->job, force);
 513     if (!job_started(&job->job)) {
 514         block_job_completed(job, -ECANCELED);
 515     } else if (job->job.deferred_to_main_loop) {
 516         block_job_completed_txn_abort(job);
 517     } else {
 518         block_job_enter(job);
 519     }
 520 }
 521
 522 void block_job_user_cancel(BlockJob *job, bool force, Error **errp)
 523 {
 524     if (job_apply_verb(&job->job, JOB_VERB_CANCEL, errp)) {
 525         return;
 526     }
 527     block_job_cancel(job, force);
 528 }
 529
 530 /* A wrapper around block_job_cancel() taking an Error ** parameter so it may be
 531  * used with block_job_finish_sync() without the need for (rather nasty)
 532  * function pointer casts there. */
 533 static void block_job_cancel_err(BlockJob *job, Error **errp)
 534 {
 535     block_job_cancel(job, false);
 536 }
 537
 538 int block_job_cancel_sync(BlockJob *job)
 539 {
 540     return block_job_finish_sync(job, &block_job_cancel_err, NULL);
 541 }
 542
 543 void block_job_cancel_sync_all(void)
 544 {
 545     BlockJob *job;
 546     AioContext *aio_context;
 547
 548     while ((job = block_job_next(NULL))) {
 549         aio_context = blk_get_aio_context(job->blk);
 550         aio_context_acquire(aio_context);
 551         block_job_cancel_sync(job);
 552         aio_context_release(aio_context);
 553     }
 554 }
 555
 556 static void block_job_complete(BlockJob *job, Error **errp)
 557 {
 558     job_complete(&job->job, errp);
 559 }
 560
 561 int block_job_complete_sync(BlockJob *job, Error **errp)
 562 {
 563     return block_job_finish_sync(job, &block_job_complete, errp);
 564 }
 565
 566 void block_job_progress_update(BlockJob *job, uint64_t done)
 567 {
 568     job->offset += done;
 569 }
 570
 571 void block_job_progress_set_remaining(BlockJob *job, uint64_t remaining)
 572 {
 573     job->len = job->offset + remaining;
 574 }
 575
 576 BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
 577 {
 578     BlockJobInfo *info;
 579
 580     if (block_job_is_internal(job)) {
 581         error_setg(errp, "Cannot query QEMU internal jobs");
 582         return NULL;
 583     }
 584     info = g_new0(BlockJobInfo, 1);
 585     info->type      = g_strdup(job_type_str(&job->job));
 586     info->device    = g_strdup(job->job.id);
 587     info->len       = job->len;
 588     info->busy      = atomic_read(&job->job.busy);
 589     info->paused    = job->job.pause_count > 0;
 590     info->offset    = job->offset;
 591     info->speed     = job->speed;
 592     info->io_status = job->iostatus;
 593     info->ready     = job->ready;
 594     info->status    = job->job.status;
 595     info->auto_finalize = job->job.auto_finalize;
 596     info->auto_dismiss  = job->job.auto_dismiss;
 597     info->has_error = job->job.ret != 0;
 598     info->error     = job->job.ret ? g_strdup(strerror(-job->job.ret)) : NULL;
 599     return info;
 600 }
 601
 602 static void block_job_iostatus_set_err(BlockJob *job, int error)
 603 {
 604     if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
 605         job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
 606                                           BLOCK_DEVICE_IO_STATUS_FAILED;
 607     }
 608 }
 609
 610 static void block_job_event_cancelled(Notifier *n, void *opaque)
 611 {
 612     BlockJob *job = opaque;
 613
 614     if (block_job_is_internal(job)) {
 615         return;
 616     }
 617
 618     qapi_event_send_block_job_cancelled(job_type(&job->job),
 619                                         job->job.id,
 620                                         job->len,
 621                                         job->offset,
 622                                         job->speed,
 623                                         &error_abort);
 624 }
 625
 626 static void block_job_event_completed(Notifier *n, void *opaque)
 627 {
 628     BlockJob *job = opaque;
 629     const char *msg = NULL;
 630
 631     if (block_job_is_internal(job)) {
 632         return;
 633     }
 634
 635     if (job->job.ret < 0) {
 636         msg = strerror(-job->job.ret);
 637     }
 638
 639     qapi_event_send_block_job_completed(job_type(&job->job),
 640                                         job->job.id,
 641                                         job->len,
 642                                         job->offset,
 643                                         job->speed,
 644                                         !!msg,
 645                                         msg,
 646                                         &error_abort);
 647 }
 648
 649 static void block_job_event_pending(Notifier *n, void *opaque)
 650 {
 651     BlockJob *job = opaque;
 652
 653     if (block_job_is_internal(job)) {
 654         return;
 655     }
 656
 657     qapi_event_send_block_job_pending(job_type(&job->job),
 658                                       job->job.id,
 659                                       &error_abort);
 660 }
 661
 662 /*
 663  * API for block job drivers and the block layer.  These functions are
 664  * declared in blockjob_int.h.
 665  */
 666
 667 void *block_job_create(const char *job_id, const BlockJobDriver *driver,
 668                        BlockJobTxn *txn, BlockDriverState *bs, uint64_t perm,
 669                        uint64_t shared_perm, int64_t speed, int flags,
 670                        BlockCompletionFunc *cb, void *opaque, Error **errp)
 671 {
 672     BlockBackend *blk;
 673     BlockJob *job;
 674     int ret;
 675
 676     if (bs->job) {
 677         error_setg(errp, QERR_DEVICE_IN_USE, bdrv_get_device_name(bs));
 678         return NULL;
 679     }
 680
 681     if (job_id == NULL && !(flags & JOB_INTERNAL)) {
 682         job_id = bdrv_get_device_name(bs);
 683     }
 684
 685     blk = blk_new(perm, shared_perm);
 686     ret = blk_insert_bs(blk, bs, errp);
 687     if (ret < 0) {
 688         blk_unref(blk);
 689         return NULL;
 690     }
 691
 692     job = job_create(job_id, &driver->job_driver, blk_get_aio_context(blk),
 693                      flags, cb, opaque, errp);
 694     if (job == NULL) {
 695         blk_unref(blk);
 696         return NULL;
 697     }
 698
 699     assert(is_block_job(&job->job));
 700     assert(job->job.driver->free == &block_job_free);
 701     assert(job->job.driver->user_resume == &block_job_user_resume);
 702     assert(job->job.driver->drain == &block_job_drain);
 703
 704     job->driver        = driver;
 705     job->blk           = blk;
 706
 707     job->finalize_cancelled_notifier.notify = block_job_event_cancelled;
 708     job->finalize_completed_notifier.notify = block_job_event_completed;
 709     job->pending_notifier.notify = block_job_event_pending;
 710
 711     notifier_list_add(&job->job.on_finalize_cancelled,
 712                       &job->finalize_cancelled_notifier);
 713     notifier_list_add(&job->job.on_finalize_completed,
 714                       &job->finalize_completed_notifier);
 715     notifier_list_add(&job->job.on_pending, &job->pending_notifier);
 716
 717     error_setg(&job->blocker, "block device is in use by block job: %s",
 718                job_type_str(&job->job));
 719     block_job_add_bdrv(job, "main node", bs, 0, BLK_PERM_ALL, &error_abort);
 720     bs->job = job;
 721
 722     bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker);
 723
 724     blk_add_aio_context_notifier(blk, block_job_attached_aio_context,
 725                                  block_job_detach_aio_context, job);
 726
 727     /* Only set speed when necessary to avoid NotSupported error */
 728     if (speed != 0) {
 729         Error *local_err = NULL;
 730
 731         block_job_set_speed(job, speed, &local_err);
 732         if (local_err) {
 733             job_early_fail(&job->job);
 734             error_propagate(errp, local_err);
 735             return NULL;
 736         }
 737     }
 738
 739     /* Single jobs are modeled as single-job transactions for sake of
 740      * consolidating the job management logic */
 741     if (!txn) {
 742         txn = block_job_txn_new();
 743         block_job_txn_add_job(txn, job);
 744         block_job_txn_unref(txn);
 745     } else {
 746         block_job_txn_add_job(txn, job);
 747     }
 748
 749     return job;
 750 }
 751
 752 void block_job_completed(BlockJob *job, int ret)
 753 {
 754     assert(job && job->txn && !job_is_completed(&job->job));
 755     assert(blk_bs(job->blk)->job == job);
 756     job->job.ret = ret;
 757     job_update_rc(&job->job);
 758     trace_block_job_completed(job, ret, job->job.ret);
 759     if (job->job.ret) {
 760         block_job_completed_txn_abort(job);
 761     } else {
 762         block_job_completed_txn_success(job);
 763     }
 764 }
 765
 766 void block_job_enter(BlockJob *job)
 767 {
 768     job_enter_cond(&job->job, NULL);
 769 }
 770
 771 void block_job_yield(BlockJob *job)
 772 {
 773     assert(job->job.busy);
 774
 775     /* Check cancellation *before* setting busy = false, too!  */
 776     if (job_is_cancelled(&job->job)) {
 777         return;
 778     }
 779
 780     if (!job_should_pause(&job->job)) {
 781         job_do_yield(&job->job, -1);
 782     }
 783
 784     job_pause_point(&job->job);
 785 }
 786
 787 void block_job_iostatus_reset(BlockJob *job)
 788 {
 789     if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
 790         return;
 791     }
 792     assert(job->job.user_paused && job->job.pause_count > 0);
 793     job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
 794 }
 795
 796 void block_job_user_resume(Job *job)
 797 {
 798     BlockJob *bjob = container_of(job, BlockJob, job);
 799     block_job_iostatus_reset(bjob);
 800 }
 801
 802 void block_job_event_ready(BlockJob *job)
 803 {
 804     job_state_transition(&job->job, JOB_STATUS_READY);
 805     job->ready = true;
 806
 807     if (block_job_is_internal(job)) {
 808         return;
 809     }
 810
 811     qapi_event_send_block_job_ready(job_type(&job->job),
 812                                     job->job.id,
 813                                     job->len,
 814                                     job->offset,
 815                                     job->speed, &error_abort);
 816 }
 817
 818 BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
 819                                         int is_read, int error)
 820 {
 821     BlockErrorAction action;
 822
 823     switch (on_err) {
 824     case BLOCKDEV_ON_ERROR_ENOSPC:
 825     case BLOCKDEV_ON_ERROR_AUTO:
 826         action = (error == ENOSPC) ?
 827                  BLOCK_ERROR_ACTION_STOP : BLOCK_ERROR_ACTION_REPORT;
 828         break;
 829     case BLOCKDEV_ON_ERROR_STOP:
 830         action = BLOCK_ERROR_ACTION_STOP;
 831         break;
 832     case BLOCKDEV_ON_ERROR_REPORT:
 833         action = BLOCK_ERROR_ACTION_REPORT;
 834         break;
 835     case BLOCKDEV_ON_ERROR_IGNORE:
 836         action = BLOCK_ERROR_ACTION_IGNORE;
 837         break;
 838     default:
 839         abort();
 840     }
 841     if (!block_job_is_internal(job)) {
 842         qapi_event_send_block_job_error(job->job.id,
 843                                         is_read ? IO_OPERATION_TYPE_READ :
 844                                         IO_OPERATION_TYPE_WRITE,
 845                                         action, &error_abort);
 846     }
 847     if (action == BLOCK_ERROR_ACTION_STOP) {
 848         job_pause(&job->job);
 849         /* make the pause user visible, which will be resumed from QMP. */
 850         job->job.user_paused = true;
 851         block_job_iostatus_set_err(job, error);
 852     }
 853     return action;
 854 }