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, /* TODO re-enable: job->ret */ 0,
  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, 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
 218     job_state_transition(job, JOB_STATUS_CREATED);
 219     aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
 220                    QEMU_CLOCK_REALTIME, SCALE_NS,
 221                    job_sleep_timer_cb, job);
 222
 223     QLIST_INSERT_HEAD(&jobs, job, job_list);
 224
 225     return job;
 226 }
 227
 228 void job_ref(Job *job)
 229 {
 230     ++job->refcnt;
 231 }
 232
 233 void job_unref(Job *job)
 234 {
 235     if (--job->refcnt == 0) {
 236         assert(job->status == JOB_STATUS_NULL);
 237         assert(!timer_pending(&job->sleep_timer));
 238
 239         if (job->driver->free) {
 240             job->driver->free(job);
 241         }
 242
 243         QLIST_REMOVE(job, job_list);
 244
 245         g_free(job->id);
 246         g_free(job);
 247     }
 248 }
 249
 250 void job_enter_cond(Job *job, bool(*fn)(Job *job))
 251 {
 252     if (!job_started(job)) {
 253         return;
 254     }
 255     if (job->deferred_to_main_loop) {
 256         return;
 257     }
 258
 259     job_lock();
 260     if (job->busy) {
 261         job_unlock();
 262         return;
 263     }
 264
 265     if (fn && !fn(job)) {
 266         job_unlock();
 267         return;
 268     }
 269
 270     assert(!job->deferred_to_main_loop);
 271     timer_del(&job->sleep_timer);
 272     job->busy = true;
 273     job_unlock();
 274     aio_co_wake(job->co);
 275 }
 276
 277 void job_enter(Job *job)
 278 {
 279     job_enter_cond(job, NULL);
 280 }
 281
 282 /* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
 283  * Reentering the job coroutine with block_job_enter() before the timer has
 284  * expired is allowed and cancels the timer.
 285  *
 286  * If @ns is (uint64_t) -1, no timer is scheduled and block_job_enter() must be
 287  * called explicitly. */
 288 void coroutine_fn job_do_yield(Job *job, uint64_t ns)
 289 {
 290     job_lock();
 291     if (ns != -1) {
 292         timer_mod(&job->sleep_timer, ns);
 293     }
 294     job->busy = false;
 295     job_unlock();
 296     qemu_coroutine_yield();
 297
 298     /* Set by job_enter_cond() before re-entering the coroutine.  */
 299     assert(job->busy);
 300 }
 301
 302 void coroutine_fn job_pause_point(Job *job)
 303 {
 304     assert(job && job_started(job));
 305
 306     if (!job_should_pause(job)) {
 307         return;
 308     }
 309     if (job_is_cancelled(job)) {
 310         return;
 311     }
 312
 313     if (job->driver->pause) {
 314         job->driver->pause(job);
 315     }
 316
 317     if (job_should_pause(job) && !job_is_cancelled(job)) {
 318         JobStatus status = job->status;
 319         job_state_transition(job, status == JOB_STATUS_READY
 320                                   ? JOB_STATUS_STANDBY
 321                                   : JOB_STATUS_PAUSED);
 322         job->paused = true;
 323         job_do_yield(job, -1);
 324         job->paused = false;
 325         job_state_transition(job, status);
 326     }
 327
 328     if (job->driver->resume) {
 329         job->driver->resume(job);
 330     }
 331 }
 332
 333 void coroutine_fn job_sleep_ns(Job *job, int64_t ns)
 334 {
 335     assert(job->busy);
 336
 337     /* Check cancellation *before* setting busy = false, too!  */
 338     if (job_is_cancelled(job)) {
 339         return;
 340     }
 341
 342     if (!job_should_pause(job)) {
 343         job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
 344     }
 345
 346     job_pause_point(job);
 347 }
 348
 349 /**
 350  * All jobs must allow a pause point before entering their job proper. This
 351  * ensures that jobs can be paused prior to being started, then resumed later.
 352  */
 353 static void coroutine_fn job_co_entry(void *opaque)
 354 {
 355     Job *job = opaque;
 356
 357     assert(job && job->driver && job->driver->start);
 358     job_pause_point(job);
 359     job->driver->start(job);
 360 }
 361
 362
 363 void job_start(Job *job)
 364 {
 365     assert(job && !job_started(job) && job->paused &&
 366            job->driver && job->driver->start);
 367     job->co = qemu_coroutine_create(job_co_entry, job);
 368     job->pause_count--;
 369     job->busy = true;
 370     job->paused = false;
 371     job_state_transition(job, JOB_STATUS_RUNNING);
 372     aio_co_enter(job->aio_context, job->co);
 373 }
 374
 375 /* Assumes the block_job_mutex is held */
 376 static bool job_timer_not_pending(Job *job)
 377 {
 378     return !timer_pending(&job->sleep_timer);
 379 }
 380
 381 void job_pause(Job *job)
 382 {
 383     job->pause_count++;
 384 }
 385
 386 void job_resume(Job *job)
 387 {
 388     assert(job->pause_count > 0);
 389     job->pause_count--;
 390     if (job->pause_count) {
 391         return;
 392     }
 393
 394     /* kick only if no timer is pending */
 395     job_enter_cond(job, job_timer_not_pending);
 396 }
 397
 398 void job_user_pause(Job *job, Error **errp)
 399 {
 400     if (job_apply_verb(job, JOB_VERB_PAUSE, errp)) {
 401         return;
 402     }
 403     if (job->user_paused) {
 404         error_setg(errp, "Job is already paused");
 405         return;
 406     }
 407     job->user_paused = true;
 408     job_pause(job);
 409 }
 410
 411 bool job_user_paused(Job *job)
 412 {
 413     return job->user_paused;
 414 }
 415
 416 void job_user_resume(Job *job, Error **errp)
 417 {
 418     assert(job);
 419     if (!job->user_paused || job->pause_count <= 0) {
 420         error_setg(errp, "Can't resume a job that was not paused");
 421         return;
 422     }
 423     if (job_apply_verb(job, JOB_VERB_RESUME, errp)) {
 424         return;
 425     }
 426     if (job->driver->user_resume) {
 427         job->driver->user_resume(job);
 428     }
 429     job->user_paused = false;
 430     job_resume(job);
 431 }
 432
 433
 434 typedef struct {
 435     Job *job;
 436     JobDeferToMainLoopFn *fn;
 437     void *opaque;
 438 } JobDeferToMainLoopData;
 439
 440 static void job_defer_to_main_loop_bh(void *opaque)
 441 {
 442     JobDeferToMainLoopData *data = opaque;
 443     Job *job = data->job;
 444     AioContext *aio_context = job->aio_context;
 445
 446     aio_context_acquire(aio_context);
 447     data->fn(data->job, data->opaque);
 448     aio_context_release(aio_context);
 449
 450     g_free(data);
 451 }
 452
 453 void job_defer_to_main_loop(Job *job, JobDeferToMainLoopFn *fn, void *opaque)
 454 {
 455     JobDeferToMainLoopData *data = g_malloc(sizeof(*data));
 456     data->job = job;
 457     data->fn = fn;
 458     data->opaque = opaque;
 459     job->deferred_to_main_loop = true;
 460
 461     aio_bh_schedule_oneshot(qemu_get_aio_context(),
 462                             job_defer_to_main_loop_bh, data);
 463 }