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