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 /* Right now, this mutex is only needed to synchronize accesses to job->busy
  64  * and job->sleep_timer, such as concurrent calls to job_do_yield and
  65  * job_enter. */
  66 static QemuMutex job_mutex;
  67
  68 static void job_lock(void)
  69 {
  70     qemu_mutex_lock(&job_mutex);
  71 }
  72
  73 static void job_unlock(void)
  74 {
  75     qemu_mutex_unlock(&job_mutex);
  76 }
  77
  78 static void __attribute__((__constructor__)) job_init(void)
  79 {
  80     qemu_mutex_init(&job_mutex);
  81 }
  82
  83 /* TODO Make static once the whole state machine is in job.c */
  84 void job_state_transition(Job *job, JobStatus s1)
  85 {
  86     JobStatus s0 = job->status;
  87     assert(s1 >= 0 && s1 <= JOB_STATUS__MAX);
  88     trace_job_state_transition(job, job->ret,
  89                                JobSTT[s0][s1] ? "allowed" : "disallowed",
  90                                JobStatus_str(s0), JobStatus_str(s1));
  91     assert(JobSTT[s0][s1]);
  92     job->status = s1;
  93 }
  94
  95 int job_apply_verb(Job *job, JobVerb verb, Error **errp)
  96 {
  97     JobStatus s0 = job->status;
  98     assert(verb >= 0 && verb <= JOB_VERB__MAX);
  99     trace_job_apply_verb(job, JobStatus_str(s0), JobVerb_str(verb),
 100                          JobVerbTable[verb][s0] ? "allowed" : "prohibited");
 101     if (JobVerbTable[verb][s0]) {
 102         return 0;
 103     }
 104     error_setg(errp, "Job '%s' in state '%s' cannot accept command verb '%s'",
 105                job->id, JobStatus_str(s0), JobVerb_str(verb));
 106     return -EPERM;
 107 }
 108
 109 JobType job_type(const Job *job)
 110 {
 111     return job->driver->job_type;
 112 }
 113
 114 const char *job_type_str(const Job *job)
 115 {
 116     return JobType_str(job_type(job));
 117 }
 118
 119 bool job_is_cancelled(Job *job)
 120 {
 121     return job->cancelled;
 122 }
 123
 124 bool job_is_completed(Job *job)
 125 {
 126     switch (job->status) {
 127     case JOB_STATUS_UNDEFINED:
 128     case JOB_STATUS_CREATED:
 129     case JOB_STATUS_RUNNING:
 130     case JOB_STATUS_PAUSED:
 131     case JOB_STATUS_READY:
 132     case JOB_STATUS_STANDBY:
 133         return false;
 134     case JOB_STATUS_WAITING:
 135     case JOB_STATUS_PENDING:
 136     case JOB_STATUS_ABORTING:
 137     case JOB_STATUS_CONCLUDED:
 138     case JOB_STATUS_NULL:
 139         return true;
 140     default:
 141         g_assert_not_reached();
 142     }
 143     return false;
 144 }
 145
 146 bool job_started(Job *job)
 147 {
 148     return job->co;
 149 }
 150
 151 bool job_should_pause(Job *job)
 152 {
 153     return job->pause_count > 0;
 154 }
 155
 156 Job *job_next(Job *job)
 157 {
 158     if (!job) {
 159         return QLIST_FIRST(&jobs);
 160     }
 161     return QLIST_NEXT(job, job_list);
 162 }
 163
 164 Job *job_get(const char *id)
 165 {
 166     Job *job;
 167
 168     QLIST_FOREACH(job, &jobs, job_list) {
 169         if (job->id && !strcmp(id, job->id)) {
 170             return job;
 171         }
 172     }
 173
 174     return NULL;
 175 }
 176
 177 static void job_sleep_timer_cb(void *opaque)
 178 {
 179     Job *job = opaque;
 180
 181     job_enter(job);
 182 }
 183
 184 void *job_create(const char *job_id, const JobDriver *driver, AioContext *ctx,
 185                  int flags, BlockCompletionFunc *cb, void *opaque, Error **errp)
 186 {
 187     Job *job;
 188
 189     if (job_id) {
 190         if (flags & JOB_INTERNAL) {
 191             error_setg(errp, "Cannot specify job ID for internal job");
 192             return NULL;
 193         }
 194         if (!id_wellformed(job_id)) {
 195             error_setg(errp, "Invalid job ID '%s'", job_id);
 196             return NULL;
 197         }
 198         if (job_get(job_id)) {
 199             error_setg(errp, "Job ID '%s' already in use", job_id);
 200             return NULL;
 201         }
 202     } else if (!(flags & JOB_INTERNAL)) {
 203         error_setg(errp, "An explicit job ID is required");
 204         return NULL;
 205     }
 206
 207     job = g_malloc0(driver->instance_size);
 208     job->driver        = driver;
 209     job->id            = g_strdup(job_id);
 210     job->refcnt        = 1;
 211     job->aio_context   = ctx;
 212     job->busy          = false;
 213     job->paused        = true;
 214     job->pause_count   = 1;
 215     job->auto_finalize = !(flags & JOB_MANUAL_FINALIZE);
 216     job->auto_dismiss  = !(flags & JOB_MANUAL_DISMISS);
 217     job->cb            = cb;
 218     job->opaque        = opaque;
 219
 220     notifier_list_init(&job->on_finalize_cancelled);
 221     notifier_list_init(&job->on_finalize_completed);
 222     notifier_list_init(&job->on_pending);
 223
 224     job_state_transition(job, JOB_STATUS_CREATED);
 225     aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
 226                    QEMU_CLOCK_REALTIME, SCALE_NS,
 227                    job_sleep_timer_cb, job);
 228
 229     QLIST_INSERT_HEAD(&jobs, job, job_list);
 230
 231     return job;
 232 }
 233
 234 void job_ref(Job *job)
 235 {
 236     ++job->refcnt;
 237 }
 238
 239 void job_unref(Job *job)
 240 {
 241     if (--job->refcnt == 0) {
 242         assert(job->status == JOB_STATUS_NULL);
 243         assert(!timer_pending(&job->sleep_timer));
 244
 245         if (job->driver->free) {
 246             job->driver->free(job);
 247         }
 248
 249         QLIST_REMOVE(job, job_list);
 250
 251         g_free(job->id);
 252         g_free(job);
 253     }
 254 }
 255
 256 void job_event_cancelled(Job *job)
 257 {
 258     notifier_list_notify(&job->on_finalize_cancelled, job);
 259 }
 260
 261 void job_event_completed(Job *job)
 262 {
 263     notifier_list_notify(&job->on_finalize_completed, job);
 264 }
 265
 266 void job_event_pending(Job *job)
 267 {
 268     notifier_list_notify(&job->on_pending, job);
 269 }
 270
 271 void job_enter_cond(Job *job, bool(*fn)(Job *job))
 272 {
 273     if (!job_started(job)) {
 274         return;
 275     }
 276     if (job->deferred_to_main_loop) {
 277         return;
 278     }
 279
 280     job_lock();
 281     if (job->busy) {
 282         job_unlock();
 283         return;
 284     }
 285
 286     if (fn && !fn(job)) {
 287         job_unlock();
 288         return;
 289     }
 290
 291     assert(!job->deferred_to_main_loop);
 292     timer_del(&job->sleep_timer);
 293     job->busy = true;
 294     job_unlock();
 295     aio_co_wake(job->co);
 296 }
 297
 298 void job_enter(Job *job)
 299 {
 300     job_enter_cond(job, NULL);
 301 }
 302
 303 /* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
 304  * Reentering the job coroutine with block_job_enter() before the timer has
 305  * expired is allowed and cancels the timer.
 306  *
 307  * If @ns is (uint64_t) -1, no timer is scheduled and block_job_enter() must be
 308  * called explicitly. */
 309 void coroutine_fn job_do_yield(Job *job, uint64_t ns)
 310 {
 311     job_lock();
 312     if (ns != -1) {
 313         timer_mod(&job->sleep_timer, ns);
 314     }
 315     job->busy = false;
 316     job_unlock();
 317     qemu_coroutine_yield();
 318
 319     /* Set by job_enter_cond() before re-entering the coroutine.  */
 320     assert(job->busy);
 321 }
 322
 323 void coroutine_fn job_pause_point(Job *job)
 324 {
 325     assert(job && job_started(job));
 326
 327     if (!job_should_pause(job)) {
 328         return;
 329     }
 330     if (job_is_cancelled(job)) {
 331         return;
 332     }
 333
 334     if (job->driver->pause) {
 335         job->driver->pause(job);
 336     }
 337
 338     if (job_should_pause(job) && !job_is_cancelled(job)) {
 339         JobStatus status = job->status;
 340         job_state_transition(job, status == JOB_STATUS_READY
 341                                   ? JOB_STATUS_STANDBY
 342                                   : JOB_STATUS_PAUSED);
 343         job->paused = true;
 344         job_do_yield(job, -1);
 345         job->paused = false;
 346         job_state_transition(job, status);
 347     }
 348
 349     if (job->driver->resume) {
 350         job->driver->resume(job);
 351     }
 352 }
 353
 354 void coroutine_fn job_sleep_ns(Job *job, int64_t ns)
 355 {
 356     assert(job->busy);
 357
 358     /* Check cancellation *before* setting busy = false, too!  */
 359     if (job_is_cancelled(job)) {
 360         return;
 361     }
 362
 363     if (!job_should_pause(job)) {
 364         job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
 365     }
 366
 367     job_pause_point(job);
 368 }
 369
 370 void job_drain(Job *job)
 371 {
 372     /* If job is !busy this kicks it into the next pause point. */
 373     job_enter(job);
 374
 375     if (job->driver->drain) {
 376         job->driver->drain(job);
 377     }
 378 }
 379
 380
 381 /**
 382  * All jobs must allow a pause point before entering their job proper. This
 383  * ensures that jobs can be paused prior to being started, then resumed later.
 384  */
 385 static void coroutine_fn job_co_entry(void *opaque)
 386 {
 387     Job *job = opaque;
 388
 389     assert(job && job->driver && job->driver->start);
 390     job_pause_point(job);
 391     job->driver->start(job);
 392 }
 393
 394
 395 void job_start(Job *job)
 396 {
 397     assert(job && !job_started(job) && job->paused &&
 398            job->driver && job->driver->start);
 399     job->co = qemu_coroutine_create(job_co_entry, job);
 400     job->pause_count--;
 401     job->busy = true;
 402     job->paused = false;
 403     job_state_transition(job, JOB_STATUS_RUNNING);
 404     aio_co_enter(job->aio_context, job->co);
 405 }
 406
 407 /* Assumes the block_job_mutex is held */
 408 static bool job_timer_not_pending(Job *job)
 409 {
 410     return !timer_pending(&job->sleep_timer);
 411 }
 412
 413 void job_pause(Job *job)
 414 {
 415     job->pause_count++;
 416 }
 417
 418 void job_resume(Job *job)
 419 {
 420     assert(job->pause_count > 0);
 421     job->pause_count--;
 422     if (job->pause_count) {
 423         return;
 424     }
 425
 426     /* kick only if no timer is pending */
 427     job_enter_cond(job, job_timer_not_pending);
 428 }
 429
 430 void job_user_pause(Job *job, Error **errp)
 431 {
 432     if (job_apply_verb(job, JOB_VERB_PAUSE, errp)) {
 433         return;
 434     }
 435     if (job->user_paused) {
 436         error_setg(errp, "Job is already paused");
 437         return;
 438     }
 439     job->user_paused = true;
 440     job_pause(job);
 441 }
 442
 443 bool job_user_paused(Job *job)
 444 {
 445     return job->user_paused;
 446 }
 447
 448 void job_user_resume(Job *job, Error **errp)
 449 {
 450     assert(job);
 451     if (!job->user_paused || job->pause_count <= 0) {
 452         error_setg(errp, "Can't resume a job that was not paused");
 453         return;
 454     }
 455     if (job_apply_verb(job, JOB_VERB_RESUME, errp)) {
 456         return;
 457     }
 458     if (job->driver->user_resume) {
 459         job->driver->user_resume(job);
 460     }
 461     job->user_paused = false;
 462     job_resume(job);
 463 }
 464
 465 void job_do_dismiss(Job *job)
 466 {
 467     assert(job);
 468     job->busy = false;
 469     job->paused = false;
 470     job->deferred_to_main_loop = true;
 471
 472     /* TODO Don't assume it's a BlockJob */
 473     block_job_txn_del_job((BlockJob*) job);
 474
 475     job_state_transition(job, JOB_STATUS_NULL);
 476     job_unref(job);
 477 }
 478
 479 void job_early_fail(Job *job)
 480 {
 481     assert(job->status == JOB_STATUS_CREATED);
 482     job_do_dismiss(job);
 483 }
 484
 485 static void job_conclude(Job *job)
 486 {
 487     job_state_transition(job, JOB_STATUS_CONCLUDED);
 488     if (job->auto_dismiss || !job_started(job)) {
 489         job_do_dismiss(job);
 490     }
 491 }
 492
 493 void job_update_rc(Job *job)
 494 {
 495     if (!job->ret && job_is_cancelled(job)) {
 496         job->ret = -ECANCELED;
 497     }
 498     if (job->ret) {
 499         job_state_transition(job, JOB_STATUS_ABORTING);
 500     }
 501 }
 502
 503 static void job_commit(Job *job)
 504 {
 505     assert(!job->ret);
 506     if (job->driver->commit) {
 507         job->driver->commit(job);
 508     }
 509 }
 510
 511 static void job_abort(Job *job)
 512 {
 513     assert(job->ret);
 514     if (job->driver->abort) {
 515         job->driver->abort(job);
 516     }
 517 }
 518
 519 static void job_clean(Job *job)
 520 {
 521     if (job->driver->clean) {
 522         job->driver->clean(job);
 523     }
 524 }
 525
 526 int job_finalize_single(Job *job)
 527 {
 528     assert(job_is_completed(job));
 529
 530     /* Ensure abort is called for late-transactional failures */
 531     job_update_rc(job);
 532
 533     if (!job->ret) {
 534         job_commit(job);
 535     } else {
 536         job_abort(job);
 537     }
 538     job_clean(job);
 539
 540     if (job->cb) {
 541         job->cb(job->opaque, job->ret);
 542     }
 543
 544     /* Emit events only if we actually started */
 545     if (job_started(job)) {
 546         if (job_is_cancelled(job)) {
 547             job_event_cancelled(job);
 548         } else {
 549             job_event_completed(job);
 550         }
 551     }
 552
 553     /* TODO Don't assume it's a BlockJob */
 554     block_job_txn_del_job((BlockJob*) job);
 555     job_conclude(job);
 556     return 0;
 557 }
 558
 559 void job_complete(Job *job, Error **errp)
 560 {
 561     /* Should not be reachable via external interface for internal jobs */
 562     assert(job->id);
 563     if (job_apply_verb(job, JOB_VERB_COMPLETE, errp)) {
 564         return;
 565     }
 566     if (job->pause_count || job_is_cancelled(job) || !job->driver->complete) {
 567         error_setg(errp, "The active block job '%s' cannot be completed",
 568                    job->id);
 569         return;
 570     }
 571
 572     job->driver->complete(job, errp);
 573 }
 574
 575
 576 typedef struct {
 577     Job *job;
 578     JobDeferToMainLoopFn *fn;
 579     void *opaque;
 580 } JobDeferToMainLoopData;
 581
 582 static void job_defer_to_main_loop_bh(void *opaque)
 583 {
 584     JobDeferToMainLoopData *data = opaque;
 585     Job *job = data->job;
 586     AioContext *aio_context = job->aio_context;
 587
 588     aio_context_acquire(aio_context);
 589     data->fn(data->job, data->opaque);
 590     aio_context_release(aio_context);
 591
 592     g_free(data);
 593 }
 594
 595 void job_defer_to_main_loop(Job *job, JobDeferToMainLoopFn *fn, void *opaque)
 596 {
 597     JobDeferToMainLoopData *data = g_malloc(sizeof(*data));
 598     data->job = job;
 599     data->fn = fn;
 600     data->opaque = opaque;
 601     job->deferred_to_main_loop = true;
 602
 603     aio_bh_schedule_oneshot(qemu_get_aio_context(),
 604                             job_defer_to_main_loop_bh, data);
 605 }
 606
 607 int job_finish_sync(Job *job, void (*finish)(Job *, Error **errp), Error **errp)
 608 {
 609     Error *local_err = NULL;
 610     int ret;
 611
 612     job_ref(job);
 613
 614     if (finish) {
 615         finish(job, &local_err);
 616     }
 617     if (local_err) {
 618         error_propagate(errp, local_err);
 619         job_unref(job);
 620         return -EBUSY;
 621     }
 622     /* job_drain calls job_enter, and it should be enough to induce progress
 623      * until the job completes or moves to the main thread. */
 624     while (!job->deferred_to_main_loop && !job_is_completed(job)) {
 625         job_drain(job);
 626     }
 627     while (!job_is_completed(job)) {
 628         aio_poll(qemu_get_aio_context(), true);
 629     }
 630     ret = (job_is_cancelled(job) && job->ret == 0) ? -ECANCELED : job->ret;
 631     job_unref(job);
 632     return ret;
 633 }