job.c

   1 /*
   2  * Background jobs (long-running operations)
   3  *
   4  * Copyright (c) 2011 IBM Corp.
   5  * Copyright (c) 2012, 2018 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 "qapi/error.h"
  29 #include "qemu/job.h"
  30 #include "qemu/id.h"
  31 #include "qemu/main-loop.h"
  32 #include "trace-root.h"
  33
  34 static QLIST_HEAD(, Job) jobs = QLIST_HEAD_INITIALIZER(jobs);
  35
  36 /* Job State Transition Table */
  37 bool JobSTT[JOB_STATUS__MAX][JOB_STATUS__MAX] = {
  38                                     /* U, C, R, P, Y, S, W, D, X, E, N */
  39     /* U: */ [JOB_STATUS_UNDEFINED] = {0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0},
  40     /* C: */ [JOB_STATUS_CREATED]   = {0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1},
  41     /* R: */ [JOB_STATUS_RUNNING]   = {0, 0, 0, 1, 1, 0, 1, 0, 1, 0, 0},
  42     /* P: */ [JOB_STATUS_PAUSED]    = {0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0},
  43     /* Y: */ [JOB_STATUS_READY]     = {0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0},
  44     /* S: */ [JOB_STATUS_STANDBY]   = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0},
  45     /* W: */ [JOB_STATUS_WAITING]   = {0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0},
  46     /* D: */ [JOB_STATUS_PENDING]   = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
  47     /* X: */ [JOB_STATUS_ABORTING]  = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
  48     /* E: */ [JOB_STATUS_CONCLUDED] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1},
  49     /* N: */ [JOB_STATUS_NULL]      = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
  50 };
  51
  52 bool JobVerbTable[JOB_VERB__MAX][JOB_STATUS__MAX] = {
  53                                     /* U, C, R, P, Y, S, W, D, X, E, N */
  54     [JOB_VERB_CANCEL]               = {0, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0},
  55     [JOB_VERB_PAUSE]                = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  56     [JOB_VERB_RESUME]               = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  57     [JOB_VERB_SET_SPEED]            = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
  58     [JOB_VERB_COMPLETE]             = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0},
  59     [JOB_VERB_FINALIZE]             = {0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0},
  60     [JOB_VERB_DISMISS]              = {0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0},
  61 };
  62
  63 /* Transactional group of jobs */
  64 struct JobTxn {
  65
  66     /* Is this txn being cancelled? */
  67     bool aborting;
  68
  69     /* List of jobs */
  70     QLIST_HEAD(, Job) jobs;
  71
  72     /* Reference count */
  73     int refcnt;
  74 };
  75
  76 /* Right now, this mutex is only needed to synchronize accesses to job->busy
  77  * and job->sleep_timer, such as concurrent calls to job_do_yield and
  78  * job_enter. */
  79 static QemuMutex job_mutex;
  80
  81 static void job_lock(void)
  82 {
  83     qemu_mutex_lock(&job_mutex);
  84 }
  85
  86 static void job_unlock(void)
  87 {
  88     qemu_mutex_unlock(&job_mutex);
  89 }
  90
  91 static void __attribute__((__constructor__)) job_init(void)
  92 {
  93     qemu_mutex_init(&job_mutex);
  94 }
  95
  96 JobTxn *job_txn_new(void)
  97 {
  98     JobTxn *txn = g_new0(JobTxn, 1);
  99     QLIST_INIT(&txn->jobs);
 100     txn->refcnt = 1;
 101     return txn;
 102 }
 103
 104 static void job_txn_ref(JobTxn *txn)
 105 {
 106     txn->refcnt++;
 107 }
 108
 109 void job_txn_unref(JobTxn *txn)
 110 {
 111     if (txn && --txn->refcnt == 0) {
 112         g_free(txn);
 113     }
 114 }
 115
 116 void job_txn_add_job(JobTxn *txn, Job *job)
 117 {
 118     if (!txn) {
 119         return;
 120     }
 121
 122     assert(!job->txn);
 123     job->txn = txn;
 124
 125     QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
 126     job_txn_ref(txn);
 127 }
 128
 129 static void job_txn_del_job(Job *job)
 130 {
 131     if (job->txn) {
 132         QLIST_REMOVE(job, txn_list);
 133         job_txn_unref(job->txn);
 134         job->txn = NULL;
 135     }
 136 }
 137
 138 static int job_txn_apply(JobTxn *txn, int fn(Job *), bool lock)
 139 {
 140     AioContext *ctx;
 141     Job *job, *next;
 142     int rc = 0;
 143
 144     QLIST_FOREACH_SAFE(job, &txn->jobs, txn_list, next) {
 145         if (lock) {
 146             ctx = job->aio_context;
 147             aio_context_acquire(ctx);
 148         }
 149         rc = fn(job);
 150         if (lock) {
 151             aio_context_release(ctx);
 152         }
 153         if (rc) {
 154             break;
 155         }
 156     }
 157     return rc;
 158 }
 159
 160
 161 /* TODO Make static once the whole state machine is in job.c */
 162 void job_state_transition(Job *job, JobStatus s1)
 163 {
 164     JobStatus s0 = job->status;
 165     assert(s1 >= 0 && s1 <= JOB_STATUS__MAX);
 166     trace_job_state_transition(job, job->ret,
 167                                JobSTT[s0][s1] ? "allowed" : "disallowed",
 168                                JobStatus_str(s0), JobStatus_str(s1));
 169     assert(JobSTT[s0][s1]);
 170     job->status = s1;
 171 }
 172
 173 int job_apply_verb(Job *job, JobVerb verb, Error **errp)
 174 {
 175     JobStatus s0 = job->status;
 176     assert(verb >= 0 && verb <= JOB_VERB__MAX);
 177     trace_job_apply_verb(job, JobStatus_str(s0), JobVerb_str(verb),
 178                          JobVerbTable[verb][s0] ? "allowed" : "prohibited");
 179     if (JobVerbTable[verb][s0]) {
 180         return 0;
 181     }
 182     error_setg(errp, "Job '%s' in state '%s' cannot accept command verb '%s'",
 183                job->id, JobStatus_str(s0), JobVerb_str(verb));
 184     return -EPERM;
 185 }
 186
 187 JobType job_type(const Job *job)
 188 {
 189     return job->driver->job_type;
 190 }
 191
 192 const char *job_type_str(const Job *job)
 193 {
 194     return JobType_str(job_type(job));
 195 }
 196
 197 bool job_is_cancelled(Job *job)
 198 {
 199     return job->cancelled;
 200 }
 201
 202 bool job_is_ready(Job *job)
 203 {
 204     switch (job->status) {
 205     case JOB_STATUS_UNDEFINED:
 206     case JOB_STATUS_CREATED:
 207     case JOB_STATUS_RUNNING:
 208     case JOB_STATUS_PAUSED:
 209     case JOB_STATUS_WAITING:
 210     case JOB_STATUS_PENDING:
 211     case JOB_STATUS_ABORTING:
 212     case JOB_STATUS_CONCLUDED:
 213     case JOB_STATUS_NULL:
 214         return false;
 215     case JOB_STATUS_READY:
 216     case JOB_STATUS_STANDBY:
 217         return true;
 218     default:
 219         g_assert_not_reached();
 220     }
 221     return false;
 222 }
 223
 224 bool job_is_completed(Job *job)
 225 {
 226     switch (job->status) {
 227     case JOB_STATUS_UNDEFINED:
 228     case JOB_STATUS_CREATED:
 229     case JOB_STATUS_RUNNING:
 230     case JOB_STATUS_PAUSED:
 231     case JOB_STATUS_READY:
 232     case JOB_STATUS_STANDBY:
 233         return false;
 234     case JOB_STATUS_WAITING:
 235     case JOB_STATUS_PENDING:
 236     case JOB_STATUS_ABORTING:
 237     case JOB_STATUS_CONCLUDED:
 238     case JOB_STATUS_NULL:
 239         return true;
 240     default:
 241         g_assert_not_reached();
 242     }
 243     return false;
 244 }
 245
 246 static bool job_started(Job *job)
 247 {
 248     return job->co;
 249 }
 250
 251 static bool job_should_pause(Job *job)
 252 {
 253     return job->pause_count > 0;
 254 }
 255
 256 Job *job_next(Job *job)
 257 {
 258     if (!job) {
 259         return QLIST_FIRST(&jobs);
 260     }
 261     return QLIST_NEXT(job, job_list);
 262 }
 263
 264 Job *job_get(const char *id)
 265 {
 266     Job *job;
 267
 268     QLIST_FOREACH(job, &jobs, job_list) {
 269         if (job->id && !strcmp(id, job->id)) {
 270             return job;
 271         }
 272     }
 273
 274     return NULL;
 275 }
 276
 277 static void job_sleep_timer_cb(void *opaque)
 278 {
 279     Job *job = opaque;
 280
 281     job_enter(job);
 282 }
 283
 284 void *job_create(const char *job_id, const JobDriver *driver, JobTxn *txn,
 285                  AioContext *ctx, int flags, BlockCompletionFunc *cb,
 286                  void *opaque, Error **errp)
 287 {
 288     Job *job;
 289
 290     if (job_id) {
 291         if (flags & JOB_INTERNAL) {
 292             error_setg(errp, "Cannot specify job ID for internal job");
 293             return NULL;
 294         }
 295         if (!id_wellformed(job_id)) {
 296             error_setg(errp, "Invalid job ID '%s'", job_id);
 297             return NULL;
 298         }
 299         if (job_get(job_id)) {
 300             error_setg(errp, "Job ID '%s' already in use", job_id);
 301             return NULL;
 302         }
 303     } else if (!(flags & JOB_INTERNAL)) {
 304         error_setg(errp, "An explicit job ID is required");
 305         return NULL;
 306     }
 307
 308     job = g_malloc0(driver->instance_size);
 309     job->driver        = driver;
 310     job->id            = g_strdup(job_id);
 311     job->refcnt        = 1;
 312     job->aio_context   = ctx;
 313     job->busy          = false;
 314     job->paused        = true;
 315     job->pause_count   = 1;
 316     job->auto_finalize = !(flags & JOB_MANUAL_FINALIZE);
 317     job->auto_dismiss  = !(flags & JOB_MANUAL_DISMISS);
 318     job->cb            = cb;
 319     job->opaque        = opaque;
 320
 321     notifier_list_init(&job->on_finalize_cancelled);
 322     notifier_list_init(&job->on_finalize_completed);
 323     notifier_list_init(&job->on_pending);
 324
 325     job_state_transition(job, JOB_STATUS_CREATED);
 326     aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
 327                    QEMU_CLOCK_REALTIME, SCALE_NS,
 328                    job_sleep_timer_cb, job);
 329
 330     QLIST_INSERT_HEAD(&jobs, job, job_list);
 331
 332     /* Single jobs are modeled as single-job transactions for sake of
 333      * consolidating the job management logic */
 334     if (!txn) {
 335         txn = job_txn_new();
 336         job_txn_add_job(txn, job);
 337         job_txn_unref(txn);
 338     } else {
 339         job_txn_add_job(txn, job);
 340     }
 341
 342     return job;
 343 }
 344
 345 void job_ref(Job *job)
 346 {
 347     ++job->refcnt;
 348 }
 349
 350 void job_unref(Job *job)
 351 {
 352     if (--job->refcnt == 0) {
 353         assert(job->status == JOB_STATUS_NULL);
 354         assert(!timer_pending(&job->sleep_timer));
 355         assert(!job->txn);
 356
 357         if (job->driver->free) {
 358             job->driver->free(job);
 359         }
 360
 361         QLIST_REMOVE(job, job_list);
 362
 363         g_free(job->id);
 364         g_free(job);
 365     }
 366 }
 367
 368 void job_event_cancelled(Job *job)
 369 {
 370     notifier_list_notify(&job->on_finalize_cancelled, job);
 371 }
 372
 373 void job_event_completed(Job *job)
 374 {
 375     notifier_list_notify(&job->on_finalize_completed, job);
 376 }
 377
 378 static void job_event_pending(Job *job)
 379 {
 380     notifier_list_notify(&job->on_pending, job);
 381 }
 382
 383 void job_enter_cond(Job *job, bool(*fn)(Job *job))
 384 {
 385     if (!job_started(job)) {
 386         return;
 387     }
 388     if (job->deferred_to_main_loop) {
 389         return;
 390     }
 391
 392     job_lock();
 393     if (job->busy) {
 394         job_unlock();
 395         return;
 396     }
 397
 398     if (fn && !fn(job)) {
 399         job_unlock();
 400         return;
 401     }
 402
 403     assert(!job->deferred_to_main_loop);
 404     timer_del(&job->sleep_timer);
 405     job->busy = true;
 406     job_unlock();
 407     aio_co_wake(job->co);
 408 }
 409
 410 void job_enter(Job *job)
 411 {
 412     job_enter_cond(job, NULL);
 413 }
 414
 415 /* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
 416  * Reentering the job coroutine with job_enter() before the timer has expired
 417  * is allowed and cancels the timer.
 418  *
 419  * If @ns is (uint64_t) -1, no timer is scheduled and job_enter() must be
 420  * called explicitly. */
 421 static void coroutine_fn job_do_yield(Job *job, uint64_t ns)
 422 {
 423     job_lock();
 424     if (ns != -1) {
 425         timer_mod(&job->sleep_timer, ns);
 426     }
 427     job->busy = false;
 428     job_unlock();
 429     qemu_coroutine_yield();
 430
 431     /* Set by job_enter_cond() before re-entering the coroutine.  */
 432     assert(job->busy);
 433 }
 434
 435 void coroutine_fn job_pause_point(Job *job)
 436 {
 437     assert(job && job_started(job));
 438
 439     if (!job_should_pause(job)) {
 440         return;
 441     }
 442     if (job_is_cancelled(job)) {
 443         return;
 444     }
 445
 446     if (job->driver->pause) {
 447         job->driver->pause(job);
 448     }
 449
 450     if (job_should_pause(job) && !job_is_cancelled(job)) {
 451         JobStatus status = job->status;
 452         job_state_transition(job, status == JOB_STATUS_READY
 453                                   ? JOB_STATUS_STANDBY
 454                                   : JOB_STATUS_PAUSED);
 455         job->paused = true;
 456         job_do_yield(job, -1);
 457         job->paused = false;
 458         job_state_transition(job, status);
 459     }
 460
 461     if (job->driver->resume) {
 462         job->driver->resume(job);
 463     }
 464 }
 465
 466 void job_yield(Job *job)
 467 {
 468     assert(job->busy);
 469
 470     /* Check cancellation *before* setting busy = false, too!  */
 471     if (job_is_cancelled(job)) {
 472         return;
 473     }
 474
 475     if (!job_should_pause(job)) {
 476         job_do_yield(job, -1);
 477     }
 478
 479     job_pause_point(job);
 480 }
 481
 482 void coroutine_fn job_sleep_ns(Job *job, int64_t ns)
 483 {
 484     assert(job->busy);
 485
 486     /* Check cancellation *before* setting busy = false, too!  */
 487     if (job_is_cancelled(job)) {
 488         return;
 489     }
 490
 491     if (!job_should_pause(job)) {
 492         job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
 493     }
 494
 495     job_pause_point(job);
 496 }
 497
 498 void job_drain(Job *job)
 499 {
 500     /* If job is !busy this kicks it into the next pause point. */
 501     job_enter(job);
 502
 503     if (job->driver->drain) {
 504         job->driver->drain(job);
 505     }
 506 }
 507
 508
 509 /**
 510  * All jobs must allow a pause point before entering their job proper. This
 511  * ensures that jobs can be paused prior to being started, then resumed later.
 512  */
 513 static void coroutine_fn job_co_entry(void *opaque)
 514 {
 515     Job *job = opaque;
 516
 517     assert(job && job->driver && job->driver->start);
 518     job_pause_point(job);
 519     job->driver->start(job);
 520 }
 521
 522
 523 void job_start(Job *job)
 524 {
 525     assert(job && !job_started(job) && job->paused &&
 526            job->driver && job->driver->start);
 527     job->co = qemu_coroutine_create(job_co_entry, job);
 528     job->pause_count--;
 529     job->busy = true;
 530     job->paused = false;
 531     job_state_transition(job, JOB_STATUS_RUNNING);
 532     aio_co_enter(job->aio_context, job->co);
 533 }
 534
 535 /* Assumes the block_job_mutex is held */
 536 static bool job_timer_not_pending(Job *job)
 537 {
 538     return !timer_pending(&job->sleep_timer);
 539 }
 540
 541 void job_pause(Job *job)
 542 {
 543     job->pause_count++;
 544 }
 545
 546 void job_resume(Job *job)
 547 {
 548     assert(job->pause_count > 0);
 549     job->pause_count--;
 550     if (job->pause_count) {
 551         return;
 552     }
 553
 554     /* kick only if no timer is pending */
 555     job_enter_cond(job, job_timer_not_pending);
 556 }
 557
 558 void job_user_pause(Job *job, Error **errp)
 559 {
 560     if (job_apply_verb(job, JOB_VERB_PAUSE, errp)) {
 561         return;
 562     }
 563     if (job->user_paused) {
 564         error_setg(errp, "Job is already paused");
 565         return;
 566     }
 567     job->user_paused = true;
 568     job_pause(job);
 569 }
 570
 571 bool job_user_paused(Job *job)
 572 {
 573     return job->user_paused;
 574 }
 575
 576 void job_user_resume(Job *job, Error **errp)
 577 {
 578     assert(job);
 579     if (!job->user_paused || job->pause_count <= 0) {
 580         error_setg(errp, "Can't resume a job that was not paused");
 581         return;
 582     }
 583     if (job_apply_verb(job, JOB_VERB_RESUME, errp)) {
 584         return;
 585     }
 586     if (job->driver->user_resume) {
 587         job->driver->user_resume(job);
 588     }
 589     job->user_paused = false;
 590     job_resume(job);
 591 }
 592
 593 static void job_do_dismiss(Job *job)
 594 {
 595     assert(job);
 596     job->busy = false;
 597     job->paused = false;
 598     job->deferred_to_main_loop = true;
 599
 600     job_txn_del_job(job);
 601
 602     job_state_transition(job, JOB_STATUS_NULL);
 603     job_unref(job);
 604 }
 605
 606 void job_dismiss(Job **jobptr, Error **errp)
 607 {
 608     Job *job = *jobptr;
 609     /* similarly to _complete, this is QMP-interface only. */
 610     assert(job->id);
 611     if (job_apply_verb(job, JOB_VERB_DISMISS, errp)) {
 612         return;
 613     }
 614
 615     job_do_dismiss(job);
 616     *jobptr = NULL;
 617 }
 618
 619 void job_early_fail(Job *job)
 620 {
 621     assert(job->status == JOB_STATUS_CREATED);
 622     job_do_dismiss(job);
 623 }
 624
 625 static void job_conclude(Job *job)
 626 {
 627     job_state_transition(job, JOB_STATUS_CONCLUDED);
 628     if (job->auto_dismiss || !job_started(job)) {
 629         job_do_dismiss(job);
 630     }
 631 }
 632
 633 static void job_update_rc(Job *job)
 634 {
 635     if (!job->ret && job_is_cancelled(job)) {
 636         job->ret = -ECANCELED;
 637     }
 638     if (job->ret) {
 639         job_state_transition(job, JOB_STATUS_ABORTING);
 640     }
 641 }
 642
 643 static void job_commit(Job *job)
 644 {
 645     assert(!job->ret);
 646     if (job->driver->commit) {
 647         job->driver->commit(job);
 648     }
 649 }
 650
 651 static void job_abort(Job *job)
 652 {
 653     assert(job->ret);
 654     if (job->driver->abort) {
 655         job->driver->abort(job);
 656     }
 657 }
 658
 659 static void job_clean(Job *job)
 660 {
 661     if (job->driver->clean) {
 662         job->driver->clean(job);
 663     }
 664 }
 665
 666 static int job_finalize_single(Job *job)
 667 {
 668     assert(job_is_completed(job));
 669
 670     /* Ensure abort is called for late-transactional failures */
 671     job_update_rc(job);
 672
 673     if (!job->ret) {
 674         job_commit(job);
 675     } else {
 676         job_abort(job);
 677     }
 678     job_clean(job);
 679
 680     if (job->cb) {
 681         job->cb(job->opaque, job->ret);
 682     }
 683
 684     /* Emit events only if we actually started */
 685     if (job_started(job)) {
 686         if (job_is_cancelled(job)) {
 687             job_event_cancelled(job);
 688         } else {
 689             job_event_completed(job);
 690         }
 691     }
 692
 693     job_txn_del_job(job);
 694     job_conclude(job);
 695     return 0;
 696 }
 697
 698 static void job_cancel_async(Job *job, bool force)
 699 {
 700     if (job->user_paused) {
 701         /* Do not call job_enter here, the caller will handle it.  */
 702         job->user_paused = false;
 703         if (job->driver->user_resume) {
 704             job->driver->user_resume(job);
 705         }
 706         assert(job->pause_count > 0);
 707         job->pause_count--;
 708     }
 709     job->cancelled = true;
 710     /* To prevent 'force == false' overriding a previous 'force == true' */
 711     job->force_cancel |= force;
 712 }
 713
 714 static void job_completed_txn_abort(Job *job)
 715 {
 716     AioContext *ctx;
 717     JobTxn *txn = job->txn;
 718     Job *other_job;
 719
 720     if (txn->aborting) {
 721         /*
 722          * We are cancelled by another job, which will handle everything.
 723          */
 724         return;
 725     }
 726     txn->aborting = true;
 727     job_txn_ref(txn);
 728
 729     /* We are the first failed job. Cancel other jobs. */
 730     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 731         ctx = other_job->aio_context;
 732         aio_context_acquire(ctx);
 733     }
 734
 735     /* Other jobs are effectively cancelled by us, set the status for
 736      * them; this job, however, may or may not be cancelled, depending
 737      * on the caller, so leave it. */
 738     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 739         if (other_job != job) {
 740             job_cancel_async(other_job, false);
 741         }
 742     }
 743     while (!QLIST_EMPTY(&txn->jobs)) {
 744         other_job = QLIST_FIRST(&txn->jobs);
 745         ctx = other_job->aio_context;
 746         if (!job_is_completed(other_job)) {
 747             assert(job_is_cancelled(other_job));
 748             job_finish_sync(other_job, NULL, NULL);
 749         }
 750         job_finalize_single(other_job);
 751         aio_context_release(ctx);
 752     }
 753
 754     job_txn_unref(txn);
 755 }
 756
 757 static int job_prepare(Job *job)
 758 {
 759     if (job->ret == 0 && job->driver->prepare) {
 760         job->ret = job->driver->prepare(job);
 761     }
 762     return job->ret;
 763 }
 764
 765 static int job_needs_finalize(Job *job)
 766 {
 767     return !job->auto_finalize;
 768 }
 769
 770 static void job_do_finalize(Job *job)
 771 {
 772     int rc;
 773     assert(job && job->txn);
 774
 775     /* prepare the transaction to complete */
 776     rc = job_txn_apply(job->txn, job_prepare, true);
 777     if (rc) {
 778         job_completed_txn_abort(job);
 779     } else {
 780         job_txn_apply(job->txn, job_finalize_single, true);
 781     }
 782 }
 783
 784 void job_finalize(Job *job, Error **errp)
 785 {
 786     assert(job && job->id);
 787     if (job_apply_verb(job, JOB_VERB_FINALIZE, errp)) {
 788         return;
 789     }
 790     job_do_finalize(job);
 791 }
 792
 793 static int job_transition_to_pending(Job *job)
 794 {
 795     job_state_transition(job, JOB_STATUS_PENDING);
 796     if (!job->auto_finalize) {
 797         job_event_pending(job);
 798     }
 799     return 0;
 800 }
 801
 802 static void job_completed_txn_success(Job *job)
 803 {
 804     JobTxn *txn = job->txn;
 805     Job *other_job;
 806
 807     job_state_transition(job, JOB_STATUS_WAITING);
 808
 809     /*
 810      * Successful completion, see if there are other running jobs in this
 811      * txn.
 812      */
 813     QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
 814         if (!job_is_completed(other_job)) {
 815             return;
 816         }
 817         assert(other_job->ret == 0);
 818     }
 819
 820     job_txn_apply(txn, job_transition_to_pending, false);
 821
 822     /* If no jobs need manual finalization, automatically do so */
 823     if (job_txn_apply(txn, job_needs_finalize, false) == 0) {
 824         job_do_finalize(job);
 825     }
 826 }
 827
 828 void job_completed(Job *job, int ret)
 829 {
 830     assert(job && job->txn && !job_is_completed(job));
 831     job->ret = ret;
 832     job_update_rc(job);
 833     trace_job_completed(job, ret, job->ret);
 834     if (job->ret) {
 835         job_completed_txn_abort(job);
 836     } else {
 837         job_completed_txn_success(job);
 838     }
 839 }
 840
 841 void job_cancel(Job *job, bool force)
 842 {
 843     if (job->status == JOB_STATUS_CONCLUDED) {
 844         job_do_dismiss(job);
 845         return;
 846     }
 847     job_cancel_async(job, force);
 848     if (!job_started(job)) {
 849         job_completed(job, -ECANCELED);
 850     } else if (job->deferred_to_main_loop) {
 851         job_completed_txn_abort(job);
 852     } else {
 853         job_enter(job);
 854     }
 855 }
 856
 857 void job_user_cancel(Job *job, bool force, Error **errp)
 858 {
 859     if (job_apply_verb(job, JOB_VERB_CANCEL, errp)) {
 860         return;
 861     }
 862     job_cancel(job, force);
 863 }
 864
 865 /* A wrapper around job_cancel() taking an Error ** parameter so it may be
 866  * used with job_finish_sync() without the need for (rather nasty) function
 867  * pointer casts there. */
 868 static void job_cancel_err(Job *job, Error **errp)
 869 {
 870     job_cancel(job, false);
 871 }
 872
 873 int job_cancel_sync(Job *job)
 874 {
 875     return job_finish_sync(job, &job_cancel_err, NULL);
 876 }
 877
 878 void job_cancel_sync_all(void)
 879 {
 880     Job *job;
 881     AioContext *aio_context;
 882
 883     while ((job = job_next(NULL))) {
 884         aio_context = job->aio_context;
 885         aio_context_acquire(aio_context);
 886         job_cancel_sync(job);
 887         aio_context_release(aio_context);
 888     }
 889 }
 890
 891 int job_complete_sync(Job *job, Error **errp)
 892 {
 893     return job_finish_sync(job, job_complete, errp);
 894 }
 895
 896 void job_complete(Job *job, Error **errp)
 897 {
 898     /* Should not be reachable via external interface for internal jobs */
 899     assert(job->id);
 900     if (job_apply_verb(job, JOB_VERB_COMPLETE, errp)) {
 901         return;
 902     }
 903     if (job->pause_count || job_is_cancelled(job) || !job->driver->complete) {
 904         error_setg(errp, "The active block job '%s' cannot be completed",
 905                    job->id);
 906         return;
 907     }
 908
 909     job->driver->complete(job, errp);
 910 }
 911
 912
 913 typedef struct {
 914     Job *job;
 915     JobDeferToMainLoopFn *fn;
 916     void *opaque;
 917 } JobDeferToMainLoopData;
 918
 919 static void job_defer_to_main_loop_bh(void *opaque)
 920 {
 921     JobDeferToMainLoopData *data = opaque;
 922     Job *job = data->job;
 923     AioContext *aio_context = job->aio_context;
 924
 925     aio_context_acquire(aio_context);
 926     data->fn(data->job, data->opaque);
 927     aio_context_release(aio_context);
 928
 929     g_free(data);
 930 }
 931
 932 void job_defer_to_main_loop(Job *job, JobDeferToMainLoopFn *fn, void *opaque)
 933 {
 934     JobDeferToMainLoopData *data = g_malloc(sizeof(*data));
 935     data->job = job;
 936     data->fn = fn;
 937     data->opaque = opaque;
 938     job->deferred_to_main_loop = true;
 939
 940     aio_bh_schedule_oneshot(qemu_get_aio_context(),
 941                             job_defer_to_main_loop_bh, data);
 942 }
 943
 944 int job_finish_sync(Job *job, void (*finish)(Job *, Error **errp), Error **errp)
 945 {
 946     Error *local_err = NULL;
 947     int ret;
 948
 949     job_ref(job);
 950
 951     if (finish) {
 952         finish(job, &local_err);
 953     }
 954     if (local_err) {
 955         error_propagate(errp, local_err);
 956         job_unref(job);
 957         return -EBUSY;
 958     }
 959     /* job_drain calls job_enter, and it should be enough to induce progress
 960      * until the job completes or moves to the main thread. */
 961     while (!job->deferred_to_main_loop && !job_is_completed(job)) {
 962         job_drain(job);
 963     }
 964     while (!job_is_completed(job)) {
 965         aio_poll(qemu_get_aio_context(), true);
 966     }
 967     ret = (job_is_cancelled(job) && job->ret == 0) ? -ECANCELED : job->ret;
 968     job_unref(job);
 969     return ret;
 970 }