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