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