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