qapi: Fix dangling references to docs/devel/qapi-code-gen.txt
[qemu/armbru.git] / job.c
blob660ce22c56b6115155451f9b60c4f237ff99b8eb
1 /*
2 * Background jobs (long-running operations)
4 * Copyright (c) 2011 IBM Corp.
5 * Copyright (c) 2012, 2018 Red Hat, Inc.
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:
14 * The above copyright notice and this permission notice shall be included in
15 * all copies or substantial portions of the Software.
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.
26 #include "qemu/osdep.h"
27 #include "qapi/error.h"
28 #include "qemu/job.h"
29 #include "qemu/id.h"
30 #include "qemu/main-loop.h"
31 #include "block/aio-wait.h"
32 #include "trace/trace-root.h"
33 #include "qapi/qapi-events-job.h"
36 * The job API is composed of two categories of functions.
38 * The first includes functions used by the monitor. The monitor is
39 * peculiar in that it accesses the job list with job_get, and
40 * therefore needs consistency across job_get and the actual operation
41 * (e.g. job_user_cancel). To achieve this consistency, the caller
42 * calls job_lock/job_unlock itself around the whole operation.
45 * The second includes functions used by the job drivers and sometimes
46 * by the core block layer. These delegate the locking to the callee instead.
50 * job_mutex protects the jobs list, but also makes the
51 * struct job fields thread-safe.
53 QemuMutex job_mutex;
55 /* Protected by job_mutex */
56 static QLIST_HEAD(, Job) jobs = QLIST_HEAD_INITIALIZER(jobs);
58 /* Job State Transition Table */
59 bool JobSTT[JOB_STATUS__MAX][JOB_STATUS__MAX] = {
60 /* U, C, R, P, Y, S, W, D, X, E, N */
61 /* U: */ [JOB_STATUS_UNDEFINED] = {0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0},
62 /* C: */ [JOB_STATUS_CREATED] = {0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1},
63 /* R: */ [JOB_STATUS_RUNNING] = {0, 0, 0, 1, 1, 0, 1, 0, 1, 0, 0},
64 /* P: */ [JOB_STATUS_PAUSED] = {0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0},
65 /* Y: */ [JOB_STATUS_READY] = {0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0},
66 /* S: */ [JOB_STATUS_STANDBY] = {0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0},
67 /* W: */ [JOB_STATUS_WAITING] = {0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0},
68 /* D: */ [JOB_STATUS_PENDING] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
69 /* X: */ [JOB_STATUS_ABORTING] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 0},
70 /* E: */ [JOB_STATUS_CONCLUDED] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1},
71 /* N: */ [JOB_STATUS_NULL] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
74 bool JobVerbTable[JOB_VERB__MAX][JOB_STATUS__MAX] = {
75 /* U, C, R, P, Y, S, W, D, X, E, N */
76 [JOB_VERB_CANCEL] = {0, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0},
77 [JOB_VERB_PAUSE] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
78 [JOB_VERB_RESUME] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
79 [JOB_VERB_SET_SPEED] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
80 [JOB_VERB_COMPLETE] = {0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 0},
81 [JOB_VERB_FINALIZE] = {0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0},
82 [JOB_VERB_DISMISS] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0},
83 [JOB_VERB_CHANGE] = {0, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0},
86 /* Transactional group of jobs */
87 struct JobTxn {
89 /* Is this txn being cancelled? */
90 bool aborting;
92 /* List of jobs */
93 QLIST_HEAD(, Job) jobs;
95 /* Reference count */
96 int refcnt;
99 void job_lock(void)
101 qemu_mutex_lock(&job_mutex);
104 void job_unlock(void)
106 qemu_mutex_unlock(&job_mutex);
109 static void __attribute__((__constructor__)) job_init(void)
111 qemu_mutex_init(&job_mutex);
114 JobTxn *job_txn_new(void)
116 JobTxn *txn = g_new0(JobTxn, 1);
117 QLIST_INIT(&txn->jobs);
118 txn->refcnt = 1;
119 return txn;
122 /* Called with job_mutex held. */
123 static void job_txn_ref_locked(JobTxn *txn)
125 txn->refcnt++;
128 void job_txn_unref_locked(JobTxn *txn)
130 if (txn && --txn->refcnt == 0) {
131 g_free(txn);
135 void job_txn_unref(JobTxn *txn)
137 JOB_LOCK_GUARD();
138 job_txn_unref_locked(txn);
142 * @txn: The transaction (may be NULL)
143 * @job: Job to add to the transaction
145 * Add @job to the transaction. The @job must not already be in a transaction.
146 * The caller must call either job_txn_unref() or job_completed() to release
147 * the reference that is automatically grabbed here.
149 * If @txn is NULL, the function does nothing.
151 * Called with job_mutex held.
153 static void job_txn_add_job_locked(JobTxn *txn, Job *job)
155 if (!txn) {
156 return;
159 assert(!job->txn);
160 job->txn = txn;
162 QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
163 job_txn_ref_locked(txn);
166 /* Called with job_mutex held. */
167 static void job_txn_del_job_locked(Job *job)
169 if (job->txn) {
170 QLIST_REMOVE(job, txn_list);
171 job_txn_unref_locked(job->txn);
172 job->txn = NULL;
176 /* Called with job_mutex held, but releases it temporarily. */
177 static int job_txn_apply_locked(Job *job, int fn(Job *))
179 Job *other_job, *next;
180 JobTxn *txn = job->txn;
181 int rc = 0;
184 * Similar to job_completed_txn_abort, we take each job's lock before
185 * applying fn, but since we assume that outer_ctx is held by the caller,
186 * we need to release it here to avoid holding the lock twice - which would
187 * break AIO_WAIT_WHILE from within fn.
189 job_ref_locked(job);
191 QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
192 rc = fn(other_job);
193 if (rc) {
194 break;
198 job_unref_locked(job);
199 return rc;
202 bool job_is_internal(Job *job)
204 return (job->id == NULL);
207 /* Called with job_mutex held. */
208 static void job_state_transition_locked(Job *job, JobStatus s1)
210 JobStatus s0 = job->status;
211 assert(s1 >= 0 && s1 < JOB_STATUS__MAX);
212 trace_job_state_transition(job, job->ret,
213 JobSTT[s0][s1] ? "allowed" : "disallowed",
214 JobStatus_str(s0), JobStatus_str(s1));
215 assert(JobSTT[s0][s1]);
216 job->status = s1;
218 if (!job_is_internal(job) && s1 != s0) {
219 qapi_event_send_job_status_change(job->id, job->status);
223 int job_apply_verb_locked(Job *job, JobVerb verb, Error **errp)
225 JobStatus s0 = job->status;
226 assert(verb >= 0 && verb < JOB_VERB__MAX);
227 trace_job_apply_verb(job, JobStatus_str(s0), JobVerb_str(verb),
228 JobVerbTable[verb][s0] ? "allowed" : "prohibited");
229 if (JobVerbTable[verb][s0]) {
230 return 0;
232 error_setg(errp, "Job '%s' in state '%s' cannot accept command verb '%s'",
233 job->id, JobStatus_str(s0), JobVerb_str(verb));
234 return -EPERM;
237 JobType job_type(const Job *job)
239 return job->driver->job_type;
242 const char *job_type_str(const Job *job)
244 return JobType_str(job_type(job));
247 bool job_is_cancelled_locked(Job *job)
249 /* force_cancel may be true only if cancelled is true, too */
250 assert(job->cancelled || !job->force_cancel);
251 return job->force_cancel;
254 bool job_is_cancelled(Job *job)
256 JOB_LOCK_GUARD();
257 return job_is_cancelled_locked(job);
260 /* Called with job_mutex held. */
261 static bool job_cancel_requested_locked(Job *job)
263 return job->cancelled;
266 bool job_cancel_requested(Job *job)
268 JOB_LOCK_GUARD();
269 return job_cancel_requested_locked(job);
272 bool job_is_ready_locked(Job *job)
274 switch (job->status) {
275 case JOB_STATUS_UNDEFINED:
276 case JOB_STATUS_CREATED:
277 case JOB_STATUS_RUNNING:
278 case JOB_STATUS_PAUSED:
279 case JOB_STATUS_WAITING:
280 case JOB_STATUS_PENDING:
281 case JOB_STATUS_ABORTING:
282 case JOB_STATUS_CONCLUDED:
283 case JOB_STATUS_NULL:
284 return false;
285 case JOB_STATUS_READY:
286 case JOB_STATUS_STANDBY:
287 return true;
288 default:
289 g_assert_not_reached();
291 return false;
294 bool job_is_ready(Job *job)
296 JOB_LOCK_GUARD();
297 return job_is_ready_locked(job);
300 bool job_is_completed_locked(Job *job)
302 switch (job->status) {
303 case JOB_STATUS_UNDEFINED:
304 case JOB_STATUS_CREATED:
305 case JOB_STATUS_RUNNING:
306 case JOB_STATUS_PAUSED:
307 case JOB_STATUS_READY:
308 case JOB_STATUS_STANDBY:
309 return false;
310 case JOB_STATUS_WAITING:
311 case JOB_STATUS_PENDING:
312 case JOB_STATUS_ABORTING:
313 case JOB_STATUS_CONCLUDED:
314 case JOB_STATUS_NULL:
315 return true;
316 default:
317 g_assert_not_reached();
319 return false;
322 static bool job_is_completed(Job *job)
324 JOB_LOCK_GUARD();
325 return job_is_completed_locked(job);
328 static bool job_started_locked(Job *job)
330 return job->co;
333 /* Called with job_mutex held. */
334 static bool job_should_pause_locked(Job *job)
336 return job->pause_count > 0;
339 Job *job_next_locked(Job *job)
341 if (!job) {
342 return QLIST_FIRST(&jobs);
344 return QLIST_NEXT(job, job_list);
347 Job *job_next(Job *job)
349 JOB_LOCK_GUARD();
350 return job_next_locked(job);
353 Job *job_get_locked(const char *id)
355 Job *job;
357 QLIST_FOREACH(job, &jobs, job_list) {
358 if (job->id && !strcmp(id, job->id)) {
359 return job;
363 return NULL;
366 void job_set_aio_context(Job *job, AioContext *ctx)
368 /* protect against read in job_finish_sync_locked and job_start */
369 GLOBAL_STATE_CODE();
370 /* protect against read in job_do_yield_locked */
371 JOB_LOCK_GUARD();
372 /* ensure the job is quiescent while the AioContext is changed */
373 assert(job->paused || job_is_completed_locked(job));
374 job->aio_context = ctx;
377 /* Called with job_mutex *not* held. */
378 static void job_sleep_timer_cb(void *opaque)
380 Job *job = opaque;
382 job_enter(job);
385 void *job_create(const char *job_id, const JobDriver *driver, JobTxn *txn,
386 AioContext *ctx, int flags, BlockCompletionFunc *cb,
387 void *opaque, Error **errp)
389 Job *job;
391 JOB_LOCK_GUARD();
393 if (job_id) {
394 if (flags & JOB_INTERNAL) {
395 error_setg(errp, "Cannot specify job ID for internal job");
396 return NULL;
398 if (!id_wellformed(job_id)) {
399 error_setg(errp, "Invalid job ID '%s'", job_id);
400 return NULL;
402 if (job_get_locked(job_id)) {
403 error_setg(errp, "Job ID '%s' already in use", job_id);
404 return NULL;
406 } else if (!(flags & JOB_INTERNAL)) {
407 error_setg(errp, "An explicit job ID is required");
408 return NULL;
411 job = g_malloc0(driver->instance_size);
412 job->driver = driver;
413 job->id = g_strdup(job_id);
414 job->refcnt = 1;
415 job->aio_context = ctx;
416 job->busy = false;
417 job->paused = true;
418 job->pause_count = 1;
419 job->auto_finalize = !(flags & JOB_MANUAL_FINALIZE);
420 job->auto_dismiss = !(flags & JOB_MANUAL_DISMISS);
421 job->cb = cb;
422 job->opaque = opaque;
424 progress_init(&job->progress);
426 notifier_list_init(&job->on_finalize_cancelled);
427 notifier_list_init(&job->on_finalize_completed);
428 notifier_list_init(&job->on_pending);
429 notifier_list_init(&job->on_ready);
430 notifier_list_init(&job->on_idle);
432 job_state_transition_locked(job, JOB_STATUS_CREATED);
433 aio_timer_init(qemu_get_aio_context(), &job->sleep_timer,
434 QEMU_CLOCK_REALTIME, SCALE_NS,
435 job_sleep_timer_cb, job);
437 QLIST_INSERT_HEAD(&jobs, job, job_list);
439 /* Single jobs are modeled as single-job transactions for sake of
440 * consolidating the job management logic */
441 if (!txn) {
442 txn = job_txn_new();
443 job_txn_add_job_locked(txn, job);
444 job_txn_unref_locked(txn);
445 } else {
446 job_txn_add_job_locked(txn, job);
449 return job;
452 void job_ref_locked(Job *job)
454 ++job->refcnt;
457 void job_unref_locked(Job *job)
459 GLOBAL_STATE_CODE();
461 if (--job->refcnt == 0) {
462 assert(job->status == JOB_STATUS_NULL);
463 assert(!timer_pending(&job->sleep_timer));
464 assert(!job->txn);
466 if (job->driver->free) {
467 job_unlock();
468 job->driver->free(job);
469 job_lock();
472 QLIST_REMOVE(job, job_list);
474 progress_destroy(&job->progress);
475 error_free(job->err);
476 g_free(job->id);
477 g_free(job);
481 void job_progress_update(Job *job, uint64_t done)
483 progress_work_done(&job->progress, done);
486 void job_progress_set_remaining(Job *job, uint64_t remaining)
488 progress_set_remaining(&job->progress, remaining);
491 void job_progress_increase_remaining(Job *job, uint64_t delta)
493 progress_increase_remaining(&job->progress, delta);
497 * To be called when a cancelled job is finalised.
498 * Called with job_mutex held.
500 static void job_event_cancelled_locked(Job *job)
502 notifier_list_notify(&job->on_finalize_cancelled, job);
506 * To be called when a successfully completed job is finalised.
507 * Called with job_mutex held.
509 static void job_event_completed_locked(Job *job)
511 notifier_list_notify(&job->on_finalize_completed, job);
514 /* Called with job_mutex held. */
515 static void job_event_pending_locked(Job *job)
517 notifier_list_notify(&job->on_pending, job);
520 /* Called with job_mutex held. */
521 static void job_event_ready_locked(Job *job)
523 notifier_list_notify(&job->on_ready, job);
526 /* Called with job_mutex held. */
527 static void job_event_idle_locked(Job *job)
529 notifier_list_notify(&job->on_idle, job);
532 void job_enter_cond_locked(Job *job, bool(*fn)(Job *job))
534 if (!job_started_locked(job)) {
535 return;
537 if (job->deferred_to_main_loop) {
538 return;
541 if (job->busy) {
542 return;
545 if (fn && !fn(job)) {
546 return;
549 assert(!job->deferred_to_main_loop);
550 timer_del(&job->sleep_timer);
551 job->busy = true;
552 job_unlock();
553 aio_co_wake(job->co);
554 job_lock();
557 void job_enter(Job *job)
559 JOB_LOCK_GUARD();
560 job_enter_cond_locked(job, NULL);
563 /* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
564 * Reentering the job coroutine with job_enter() before the timer has expired
565 * is allowed and cancels the timer.
567 * If @ns is (uint64_t) -1, no timer is scheduled and job_enter() must be
568 * called explicitly.
570 * Called with job_mutex held, but releases it temporarily.
572 static void coroutine_fn job_do_yield_locked(Job *job, uint64_t ns)
574 AioContext *next_aio_context;
576 if (ns != -1) {
577 timer_mod(&job->sleep_timer, ns);
579 job->busy = false;
580 job_event_idle_locked(job);
581 job_unlock();
582 qemu_coroutine_yield();
583 job_lock();
585 next_aio_context = job->aio_context;
587 * Coroutine has resumed, but in the meanwhile the job AioContext
588 * might have changed via bdrv_try_change_aio_context(), so we need to move
589 * the coroutine too in the new aiocontext.
591 while (qemu_get_current_aio_context() != next_aio_context) {
592 job_unlock();
593 aio_co_reschedule_self(next_aio_context);
594 job_lock();
595 next_aio_context = job->aio_context;
598 /* Set by job_enter_cond_locked() before re-entering the coroutine. */
599 assert(job->busy);
602 /* Called with job_mutex held, but releases it temporarily. */
603 static void coroutine_fn job_pause_point_locked(Job *job)
605 assert(job && job_started_locked(job));
607 if (!job_should_pause_locked(job)) {
608 return;
610 if (job_is_cancelled_locked(job)) {
611 return;
614 if (job->driver->pause) {
615 job_unlock();
616 job->driver->pause(job);
617 job_lock();
620 if (job_should_pause_locked(job) && !job_is_cancelled_locked(job)) {
621 JobStatus status = job->status;
622 job_state_transition_locked(job, status == JOB_STATUS_READY
623 ? JOB_STATUS_STANDBY
624 : JOB_STATUS_PAUSED);
625 job->paused = true;
626 job_do_yield_locked(job, -1);
627 job->paused = false;
628 job_state_transition_locked(job, status);
631 if (job->driver->resume) {
632 job_unlock();
633 job->driver->resume(job);
634 job_lock();
638 void coroutine_fn job_pause_point(Job *job)
640 JOB_LOCK_GUARD();
641 job_pause_point_locked(job);
644 void coroutine_fn job_yield(Job *job)
646 JOB_LOCK_GUARD();
647 assert(job->busy);
649 /* Check cancellation *before* setting busy = false, too! */
650 if (job_is_cancelled_locked(job)) {
651 return;
654 if (!job_should_pause_locked(job)) {
655 job_do_yield_locked(job, -1);
658 job_pause_point_locked(job);
661 void coroutine_fn job_sleep_ns(Job *job, int64_t ns)
663 JOB_LOCK_GUARD();
664 assert(job->busy);
666 /* Check cancellation *before* setting busy = false, too! */
667 if (job_is_cancelled_locked(job)) {
668 return;
671 if (!job_should_pause_locked(job)) {
672 job_do_yield_locked(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
675 job_pause_point_locked(job);
678 /* Assumes the job_mutex is held */
679 static bool job_timer_not_pending_locked(Job *job)
681 return !timer_pending(&job->sleep_timer);
684 void job_pause_locked(Job *job)
686 job->pause_count++;
687 if (!job->paused) {
688 job_enter_cond_locked(job, NULL);
692 void job_pause(Job *job)
694 JOB_LOCK_GUARD();
695 job_pause_locked(job);
698 void job_resume_locked(Job *job)
700 assert(job->pause_count > 0);
701 job->pause_count--;
702 if (job->pause_count) {
703 return;
706 /* kick only if no timer is pending */
707 job_enter_cond_locked(job, job_timer_not_pending_locked);
710 void job_resume(Job *job)
712 JOB_LOCK_GUARD();
713 job_resume_locked(job);
716 void job_user_pause_locked(Job *job, Error **errp)
718 if (job_apply_verb_locked(job, JOB_VERB_PAUSE, errp)) {
719 return;
721 if (job->user_paused) {
722 error_setg(errp, "Job is already paused");
723 return;
725 job->user_paused = true;
726 job_pause_locked(job);
729 bool job_user_paused_locked(Job *job)
731 return job->user_paused;
734 void job_user_resume_locked(Job *job, Error **errp)
736 assert(job);
737 GLOBAL_STATE_CODE();
738 if (!job->user_paused || job->pause_count <= 0) {
739 error_setg(errp, "Can't resume a job that was not paused");
740 return;
742 if (job_apply_verb_locked(job, JOB_VERB_RESUME, errp)) {
743 return;
745 if (job->driver->user_resume) {
746 job_unlock();
747 job->driver->user_resume(job);
748 job_lock();
750 job->user_paused = false;
751 job_resume_locked(job);
754 /* Called with job_mutex held, but releases it temporarily. */
755 static void job_do_dismiss_locked(Job *job)
757 assert(job);
758 job->busy = false;
759 job->paused = false;
760 job->deferred_to_main_loop = true;
762 job_txn_del_job_locked(job);
764 job_state_transition_locked(job, JOB_STATUS_NULL);
765 job_unref_locked(job);
768 void job_dismiss_locked(Job **jobptr, Error **errp)
770 Job *job = *jobptr;
771 /* similarly to _complete, this is QMP-interface only. */
772 assert(job->id);
773 if (job_apply_verb_locked(job, JOB_VERB_DISMISS, errp)) {
774 return;
777 job_do_dismiss_locked(job);
778 *jobptr = NULL;
781 void job_early_fail(Job *job)
783 JOB_LOCK_GUARD();
784 assert(job->status == JOB_STATUS_CREATED);
785 job_do_dismiss_locked(job);
788 /* Called with job_mutex held. */
789 static void job_conclude_locked(Job *job)
791 job_state_transition_locked(job, JOB_STATUS_CONCLUDED);
792 if (job->auto_dismiss || !job_started_locked(job)) {
793 job_do_dismiss_locked(job);
797 /* Called with job_mutex held. */
798 static void job_update_rc_locked(Job *job)
800 if (!job->ret && job_is_cancelled_locked(job)) {
801 job->ret = -ECANCELED;
803 if (job->ret) {
804 if (!job->err) {
805 error_setg(&job->err, "%s", strerror(-job->ret));
807 job_state_transition_locked(job, JOB_STATUS_ABORTING);
811 static void job_commit(Job *job)
813 assert(!job->ret);
814 GLOBAL_STATE_CODE();
815 if (job->driver->commit) {
816 job->driver->commit(job);
820 static void job_abort(Job *job)
822 assert(job->ret);
823 GLOBAL_STATE_CODE();
824 if (job->driver->abort) {
825 job->driver->abort(job);
829 static void job_clean(Job *job)
831 GLOBAL_STATE_CODE();
832 if (job->driver->clean) {
833 job->driver->clean(job);
838 * Called with job_mutex held, but releases it temporarily.
840 static int job_finalize_single_locked(Job *job)
842 int job_ret;
844 assert(job_is_completed_locked(job));
846 /* Ensure abort is called for late-transactional failures */
847 job_update_rc_locked(job);
849 job_ret = job->ret;
850 job_unlock();
852 if (!job_ret) {
853 job_commit(job);
854 } else {
855 job_abort(job);
857 job_clean(job);
859 if (job->cb) {
860 job->cb(job->opaque, job_ret);
863 job_lock();
865 /* Emit events only if we actually started */
866 if (job_started_locked(job)) {
867 if (job_is_cancelled_locked(job)) {
868 job_event_cancelled_locked(job);
869 } else {
870 job_event_completed_locked(job);
874 job_txn_del_job_locked(job);
875 job_conclude_locked(job);
876 return 0;
880 * Called with job_mutex held, but releases it temporarily.
882 static void job_cancel_async_locked(Job *job, bool force)
884 GLOBAL_STATE_CODE();
885 if (job->driver->cancel) {
886 job_unlock();
887 force = job->driver->cancel(job, force);
888 job_lock();
889 } else {
890 /* No .cancel() means the job will behave as if force-cancelled */
891 force = true;
894 if (job->user_paused) {
895 /* Do not call job_enter here, the caller will handle it. */
896 if (job->driver->user_resume) {
897 job_unlock();
898 job->driver->user_resume(job);
899 job_lock();
901 job->user_paused = false;
902 assert(job->pause_count > 0);
903 job->pause_count--;
907 * Ignore soft cancel requests after the job is already done
908 * (We will still invoke job->driver->cancel() above, but if the
909 * job driver supports soft cancelling and the job is done, that
910 * should be a no-op, too. We still call it so it can override
911 * @force.)
913 if (force || !job->deferred_to_main_loop) {
914 job->cancelled = true;
915 /* To prevent 'force == false' overriding a previous 'force == true' */
916 job->force_cancel |= force;
921 * Called with job_mutex held, but releases it temporarily.
923 static void job_completed_txn_abort_locked(Job *job)
925 JobTxn *txn = job->txn;
926 Job *other_job;
928 if (txn->aborting) {
930 * We are cancelled by another job, which will handle everything.
932 return;
934 txn->aborting = true;
935 job_txn_ref_locked(txn);
937 job_ref_locked(job);
939 /* Other jobs are effectively cancelled by us, set the status for
940 * them; this job, however, may or may not be cancelled, depending
941 * on the caller, so leave it. */
942 QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
943 if (other_job != job) {
945 * This is a transaction: If one job failed, no result will matter.
946 * Therefore, pass force=true to terminate all other jobs as quickly
947 * as possible.
949 job_cancel_async_locked(other_job, true);
952 while (!QLIST_EMPTY(&txn->jobs)) {
953 other_job = QLIST_FIRST(&txn->jobs);
954 if (!job_is_completed_locked(other_job)) {
955 assert(job_cancel_requested_locked(other_job));
956 job_finish_sync_locked(other_job, NULL, NULL);
958 job_finalize_single_locked(other_job);
961 job_unref_locked(job);
962 job_txn_unref_locked(txn);
965 /* Called with job_mutex held, but releases it temporarily */
966 static int job_prepare_locked(Job *job)
968 int ret;
970 GLOBAL_STATE_CODE();
972 if (job->ret == 0 && job->driver->prepare) {
973 job_unlock();
974 ret = job->driver->prepare(job);
975 job_lock();
976 job->ret = ret;
977 job_update_rc_locked(job);
980 return job->ret;
983 /* Called with job_mutex held */
984 static int job_needs_finalize_locked(Job *job)
986 return !job->auto_finalize;
989 /* Called with job_mutex held */
990 static void job_do_finalize_locked(Job *job)
992 int rc;
993 assert(job && job->txn);
995 /* prepare the transaction to complete */
996 rc = job_txn_apply_locked(job, job_prepare_locked);
997 if (rc) {
998 job_completed_txn_abort_locked(job);
999 } else {
1000 job_txn_apply_locked(job, job_finalize_single_locked);
1004 void job_finalize_locked(Job *job, Error **errp)
1006 assert(job && job->id);
1007 if (job_apply_verb_locked(job, JOB_VERB_FINALIZE, errp)) {
1008 return;
1010 job_do_finalize_locked(job);
1013 /* Called with job_mutex held. */
1014 static int job_transition_to_pending_locked(Job *job)
1016 job_state_transition_locked(job, JOB_STATUS_PENDING);
1017 if (!job->auto_finalize) {
1018 job_event_pending_locked(job);
1020 return 0;
1023 void job_transition_to_ready(Job *job)
1025 JOB_LOCK_GUARD();
1026 job_state_transition_locked(job, JOB_STATUS_READY);
1027 job_event_ready_locked(job);
1030 /* Called with job_mutex held. */
1031 static void job_completed_txn_success_locked(Job *job)
1033 JobTxn *txn = job->txn;
1034 Job *other_job;
1036 job_state_transition_locked(job, JOB_STATUS_WAITING);
1039 * Successful completion, see if there are other running jobs in this
1040 * txn.
1042 QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
1043 if (!job_is_completed_locked(other_job)) {
1044 return;
1046 assert(other_job->ret == 0);
1049 job_txn_apply_locked(job, job_transition_to_pending_locked);
1051 /* If no jobs need manual finalization, automatically do so */
1052 if (job_txn_apply_locked(job, job_needs_finalize_locked) == 0) {
1053 job_do_finalize_locked(job);
1057 /* Called with job_mutex held. */
1058 static void job_completed_locked(Job *job)
1060 assert(job && job->txn && !job_is_completed_locked(job));
1062 job_update_rc_locked(job);
1063 trace_job_completed(job, job->ret);
1064 if (job->ret) {
1065 job_completed_txn_abort_locked(job);
1066 } else {
1067 job_completed_txn_success_locked(job);
1072 * Useful only as a type shim for aio_bh_schedule_oneshot.
1073 * Called with job_mutex *not* held.
1075 static void job_exit(void *opaque)
1077 Job *job = (Job *)opaque;
1078 JOB_LOCK_GUARD();
1079 job_ref_locked(job);
1081 /* This is a lie, we're not quiescent, but still doing the completion
1082 * callbacks. However, completion callbacks tend to involve operations that
1083 * drain block nodes, and if .drained_poll still returned true, we would
1084 * deadlock. */
1085 job->busy = false;
1086 job_event_idle_locked(job);
1088 job_completed_locked(job);
1089 job_unref_locked(job);
1093 * All jobs must allow a pause point before entering their job proper. This
1094 * ensures that jobs can be paused prior to being started, then resumed later.
1096 static void coroutine_fn job_co_entry(void *opaque)
1098 Job *job = opaque;
1099 int ret;
1101 assert(job && job->driver && job->driver->run);
1102 WITH_JOB_LOCK_GUARD() {
1103 assert(job->aio_context == qemu_get_current_aio_context());
1104 job_pause_point_locked(job);
1106 ret = job->driver->run(job, &job->err);
1107 WITH_JOB_LOCK_GUARD() {
1108 job->ret = ret;
1109 job->deferred_to_main_loop = true;
1110 job->busy = true;
1112 aio_bh_schedule_oneshot(qemu_get_aio_context(), job_exit, job);
1115 void job_start(Job *job)
1117 assert(qemu_in_main_thread());
1119 WITH_JOB_LOCK_GUARD() {
1120 assert(job && !job_started_locked(job) && job->paused &&
1121 job->driver && job->driver->run);
1122 job->co = qemu_coroutine_create(job_co_entry, job);
1123 job->pause_count--;
1124 job->busy = true;
1125 job->paused = false;
1126 job_state_transition_locked(job, JOB_STATUS_RUNNING);
1128 aio_co_enter(job->aio_context, job->co);
1131 void job_cancel_locked(Job *job, bool force)
1133 if (job->status == JOB_STATUS_CONCLUDED) {
1134 job_do_dismiss_locked(job);
1135 return;
1137 job_cancel_async_locked(job, force);
1138 if (!job_started_locked(job)) {
1139 job_completed_locked(job);
1140 } else if (job->deferred_to_main_loop) {
1142 * job_cancel_async() ignores soft-cancel requests for jobs
1143 * that are already done (i.e. deferred to the main loop). We
1144 * have to check again whether the job is really cancelled.
1145 * (job_cancel_requested() and job_is_cancelled() are equivalent
1146 * here, because job_cancel_async() will make soft-cancel
1147 * requests no-ops when deferred_to_main_loop is true. We
1148 * choose to call job_is_cancelled() to show that we invoke
1149 * job_completed_txn_abort() only for force-cancelled jobs.)
1151 if (job_is_cancelled_locked(job)) {
1152 job_completed_txn_abort_locked(job);
1154 } else {
1155 job_enter_cond_locked(job, NULL);
1159 void job_user_cancel_locked(Job *job, bool force, Error **errp)
1161 if (job_apply_verb_locked(job, JOB_VERB_CANCEL, errp)) {
1162 return;
1164 job_cancel_locked(job, force);
1167 /* A wrapper around job_cancel_locked() taking an Error ** parameter so it may
1168 * be used with job_finish_sync_locked() without the need for (rather nasty)
1169 * function pointer casts there.
1171 * Called with job_mutex held.
1173 static void job_cancel_err_locked(Job *job, Error **errp)
1175 job_cancel_locked(job, false);
1179 * Same as job_cancel_err(), but force-cancel.
1180 * Called with job_mutex held.
1182 static void job_force_cancel_err_locked(Job *job, Error **errp)
1184 job_cancel_locked(job, true);
1187 int job_cancel_sync_locked(Job *job, bool force)
1189 if (force) {
1190 return job_finish_sync_locked(job, &job_force_cancel_err_locked, NULL);
1191 } else {
1192 return job_finish_sync_locked(job, &job_cancel_err_locked, NULL);
1196 int job_cancel_sync(Job *job, bool force)
1198 JOB_LOCK_GUARD();
1199 return job_cancel_sync_locked(job, force);
1202 void job_cancel_sync_all(void)
1204 Job *job;
1205 JOB_LOCK_GUARD();
1207 while ((job = job_next_locked(NULL))) {
1208 job_cancel_sync_locked(job, true);
1212 int job_complete_sync_locked(Job *job, Error **errp)
1214 return job_finish_sync_locked(job, job_complete_locked, errp);
1217 void job_complete_locked(Job *job, Error **errp)
1219 /* Should not be reachable via external interface for internal jobs */
1220 assert(job->id);
1221 GLOBAL_STATE_CODE();
1222 if (job_apply_verb_locked(job, JOB_VERB_COMPLETE, errp)) {
1223 return;
1225 if (job_cancel_requested_locked(job) || !job->driver->complete) {
1226 error_setg(errp, "The active block job '%s' cannot be completed",
1227 job->id);
1228 return;
1231 job_unlock();
1232 job->driver->complete(job, errp);
1233 job_lock();
1236 int job_finish_sync_locked(Job *job,
1237 void (*finish)(Job *, Error **errp),
1238 Error **errp)
1240 Error *local_err = NULL;
1241 int ret;
1242 GLOBAL_STATE_CODE();
1244 job_ref_locked(job);
1246 if (finish) {
1247 finish(job, &local_err);
1249 if (local_err) {
1250 error_propagate(errp, local_err);
1251 job_unref_locked(job);
1252 return -EBUSY;
1255 job_unlock();
1256 AIO_WAIT_WHILE_UNLOCKED(job->aio_context,
1257 (job_enter(job), !job_is_completed(job)));
1258 job_lock();
1260 ret = (job_is_cancelled_locked(job) && job->ret == 0)
1261 ? -ECANCELED : job->ret;
1262 job_unref_locked(job);
1263 return ret;