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_started(Job *job)
 125 {
 126     return job->co;
 127 }
 128
 129 bool job_should_pause(Job *job)
 130 {
 131     return job->pause_count > 0;
 132 }
 133
 134 Job *job_next(Job *job)
 135 {
 136     if (!job) {
 137         return QLIST_FIRST(&jobs);
 138     }
 139     return QLIST_NEXT(job, job_list);
 140 }
 141
 142 Job *job_get(const char *id)
 143 {
 144     Job *job;
 145
 146     QLIST_FOREACH(job, &jobs, job_list) {
 147         if (job->id && !strcmp(id, job->id)) {
 148             return job;
 149         }
 150     }
 151
 152     return NULL;
 153 }
 154
 155 void *job_create(const char *job_id, const JobDriver *driver, AioContext *ctx,
 156                  Error **errp)
 157 {
 158     Job *job;
 159
 160     if (job_id) {
 161         if (!id_wellformed(job_id)) {
 162             error_setg(errp, "Invalid job ID '%s'", job_id);
 163             return NULL;
 164         }
 165         if (job_get(job_id)) {
 166             error_setg(errp, "Job ID '%s' already in use", job_id);
 167             return NULL;
 168         }
 169     }
 170
 171     job = g_malloc0(driver->instance_size);
 172     job->driver        = driver;
 173     job->id            = g_strdup(job_id);
 174     job->refcnt        = 1;
 175     job->aio_context   = ctx;
 176     job->busy          = false;
 177     job->paused        = true;
 178     job->pause_count   = 1;
 179
 180     job_state_transition(job, JOB_STATUS_CREATED);
 181
 182     QLIST_INSERT_HEAD(&jobs, job, job_list);
 183
 184     return job;
 185 }
 186
 187 void job_ref(Job *job)
 188 {
 189     ++job->refcnt;
 190 }
 191
 192 void job_unref(Job *job)
 193 {
 194     if (--job->refcnt == 0) {
 195         assert(job->status == JOB_STATUS_NULL);
 196
 197         if (job->driver->free) {
 198             job->driver->free(job);
 199         }
 200
 201         QLIST_REMOVE(job, job_list);
 202
 203         g_free(job->id);
 204         g_free(job);
 205     }
 206 }
 207
 208 void job_enter_cond(Job *job, bool(*fn)(Job *job))
 209 {
 210     if (!job_started(job)) {
 211         return;
 212     }
 213     if (job->deferred_to_main_loop) {
 214         return;
 215     }
 216
 217     job_lock();
 218     if (job->busy) {
 219         job_unlock();
 220         return;
 221     }
 222
 223     if (fn && !fn(job)) {
 224         job_unlock();
 225         return;
 226     }
 227
 228     assert(!job->deferred_to_main_loop);
 229     timer_del(&job->sleep_timer);
 230     job->busy = true;
 231     job_unlock();
 232     aio_co_wake(job->co);
 233 }
 234
 235 /* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
 236  * Reentering the job coroutine with block_job_enter() before the timer has
 237  * expired is allowed and cancels the timer.
 238  *
 239  * If @ns is (uint64_t) -1, no timer is scheduled and block_job_enter() must be
 240  * called explicitly. */
 241 void coroutine_fn job_do_yield(Job *job, uint64_t ns)
 242 {
 243     job_lock();
 244     if (ns != -1) {
 245         timer_mod(&job->sleep_timer, ns);
 246     }
 247     job->busy = false;
 248     job_unlock();
 249     qemu_coroutine_yield();
 250
 251     /* Set by job_enter_cond() before re-entering the coroutine.  */
 252     assert(job->busy);
 253 }
 254
 255 void coroutine_fn job_pause_point(Job *job)
 256 {
 257     assert(job && job_started(job));
 258
 259     if (!job_should_pause(job)) {
 260         return;
 261     }
 262     if (job_is_cancelled(job)) {
 263         return;
 264     }
 265
 266     if (job->driver->pause) {
 267         job->driver->pause(job);
 268     }
 269
 270     if (job_should_pause(job) && !job_is_cancelled(job)) {
 271         JobStatus status = job->status;
 272         job_state_transition(job, status == JOB_STATUS_READY
 273                                   ? JOB_STATUS_STANDBY
 274                                   : JOB_STATUS_PAUSED);
 275         job->paused = true;
 276         job_do_yield(job, -1);
 277         job->paused = false;
 278         job_state_transition(job, status);
 279     }
 280
 281     if (job->driver->resume) {
 282         job->driver->resume(job);
 283     }
 284 }
 285
 286 /**
 287  * All jobs must allow a pause point before entering their job proper. This
 288  * ensures that jobs can be paused prior to being started, then resumed later.
 289  */
 290 static void coroutine_fn job_co_entry(void *opaque)
 291 {
 292     Job *job = opaque;
 293
 294     assert(job && job->driver && job->driver->start);
 295     job_pause_point(job);
 296     job->driver->start(job);
 297 }
 298
 299
 300 void job_start(Job *job)
 301 {
 302     assert(job && !job_started(job) && job->paused &&
 303            job->driver && job->driver->start);
 304     job->co = qemu_coroutine_create(job_co_entry, job);
 305     job->pause_count--;
 306     job->busy = true;
 307     job->paused = false;
 308     job_state_transition(job, JOB_STATUS_RUNNING);
 309     aio_co_enter(job->aio_context, job->co);
 310 }
 311
 312 typedef struct {
 313     Job *job;
 314     JobDeferToMainLoopFn *fn;
 315     void *opaque;
 316 } JobDeferToMainLoopData;
 317
 318 static void job_defer_to_main_loop_bh(void *opaque)
 319 {
 320     JobDeferToMainLoopData *data = opaque;
 321     Job *job = data->job;
 322     AioContext *aio_context = job->aio_context;
 323
 324     aio_context_acquire(aio_context);
 325     data->fn(data->job, data->opaque);
 326     aio_context_release(aio_context);
 327
 328     g_free(data);
 329 }
 330
 331 void job_defer_to_main_loop(Job *job, JobDeferToMainLoopFn *fn, void *opaque)
 332 {
 333     JobDeferToMainLoopData *data = g_malloc(sizeof(*data));
 334     data->job = job;
 335     data->fn = fn;
 336     data->opaque = opaque;
 337     job->deferred_to_main_loop = true;
 338
 339     aio_bh_schedule_oneshot(qemu_get_aio_context(),
 340                             job_defer_to_main_loop_bh, data);
 341 }