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
26 #include "qemu/osdep.h"
27 #include "qapi/error.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.
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 */
89 /* Is this txn being cancelled? */
93 QLIST_HEAD(, Job
) jobs
;
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
);
122 /* Called with job_mutex held. */
123 static void job_txn_ref_locked(JobTxn
*txn
)
128 void job_txn_unref_locked(JobTxn
*txn
)
130 if (txn
&& --txn
->refcnt
== 0) {
135 void job_txn_unref(JobTxn
*txn
)
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
)
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
)
170 QLIST_REMOVE(job
, txn_list
);
171 job_txn_unref_locked(job
->txn
);
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
;
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.
191 QLIST_FOREACH_SAFE(other_job
, &txn
->jobs
, txn_list
, next
) {
198 job_unref_locked(job
);
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
]);
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
]) {
232 error_setg(errp
, "Job '%s' in state '%s' cannot accept command verb '%s'",
233 job
->id
, JobStatus_str(s0
), JobVerb_str(verb
));
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
)
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
)
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
:
285 case JOB_STATUS_READY
:
286 case JOB_STATUS_STANDBY
:
289 g_assert_not_reached();
294 bool job_is_ready(Job
*job
)
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
:
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
:
317 g_assert_not_reached();
322 static bool job_is_completed(Job
*job
)
325 return job_is_completed_locked(job
);
328 static bool job_started_locked(Job
*job
)
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
)
342 return QLIST_FIRST(&jobs
);
344 return QLIST_NEXT(job
, job_list
);
347 Job
*job_next(Job
*job
)
350 return job_next_locked(job
);
353 Job
*job_get_locked(const char *id
)
357 QLIST_FOREACH(job
, &jobs
, job_list
) {
358 if (job
->id
&& !strcmp(id
, job
->id
)) {
366 void job_set_aio_context(Job
*job
, AioContext
*ctx
)
368 /* protect against read in job_finish_sync_locked and job_start */
370 /* protect against read in job_do_yield_locked */
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
)
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
)
394 if (flags
& JOB_INTERNAL
) {
395 error_setg(errp
, "Cannot specify job ID for internal job");
398 if (!id_wellformed(job_id
)) {
399 error_setg(errp
, "Invalid job ID '%s'", job_id
);
402 if (job_get_locked(job_id
)) {
403 error_setg(errp
, "Job ID '%s' already in use", job_id
);
406 } else if (!(flags
& JOB_INTERNAL
)) {
407 error_setg(errp
, "An explicit job ID is required");
411 job
= g_malloc0(driver
->instance_size
);
412 job
->driver
= driver
;
413 job
->id
= g_strdup(job_id
);
415 job
->aio_context
= ctx
;
418 job
->pause_count
= 1;
419 job
->auto_finalize
= !(flags
& JOB_MANUAL_FINALIZE
);
420 job
->auto_dismiss
= !(flags
& JOB_MANUAL_DISMISS
);
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 */
443 job_txn_add_job_locked(txn
, job
);
444 job_txn_unref_locked(txn
);
446 job_txn_add_job_locked(txn
, job
);
452 void job_ref_locked(Job
*job
)
457 void job_unref_locked(Job
*job
)
461 if (--job
->refcnt
== 0) {
462 assert(job
->status
== JOB_STATUS_NULL
);
463 assert(!timer_pending(&job
->sleep_timer
));
466 if (job
->driver
->free
) {
467 AioContext
*aio_context
= job
->aio_context
;
469 /* FIXME: aiocontext lock is required because cb calls blk_unref */
470 aio_context_acquire(aio_context
);
471 job
->driver
->free(job
);
472 aio_context_release(aio_context
);
476 QLIST_REMOVE(job
, job_list
);
478 progress_destroy(&job
->progress
);
479 error_free(job
->err
);
485 void job_progress_update(Job
*job
, uint64_t done
)
487 progress_work_done(&job
->progress
, done
);
490 void job_progress_set_remaining(Job
*job
, uint64_t remaining
)
492 progress_set_remaining(&job
->progress
, remaining
);
495 void job_progress_increase_remaining(Job
*job
, uint64_t delta
)
497 progress_increase_remaining(&job
->progress
, delta
);
501 * To be called when a cancelled job is finalised.
502 * Called with job_mutex held.
504 static void job_event_cancelled_locked(Job
*job
)
506 notifier_list_notify(&job
->on_finalize_cancelled
, job
);
510 * To be called when a successfully completed job is finalised.
511 * Called with job_mutex held.
513 static void job_event_completed_locked(Job
*job
)
515 notifier_list_notify(&job
->on_finalize_completed
, job
);
518 /* Called with job_mutex held. */
519 static void job_event_pending_locked(Job
*job
)
521 notifier_list_notify(&job
->on_pending
, job
);
524 /* Called with job_mutex held. */
525 static void job_event_ready_locked(Job
*job
)
527 notifier_list_notify(&job
->on_ready
, job
);
530 /* Called with job_mutex held. */
531 static void job_event_idle_locked(Job
*job
)
533 notifier_list_notify(&job
->on_idle
, job
);
536 void job_enter_cond_locked(Job
*job
, bool(*fn
)(Job
*job
))
538 if (!job_started_locked(job
)) {
541 if (job
->deferred_to_main_loop
) {
549 if (fn
&& !fn(job
)) {
553 assert(!job
->deferred_to_main_loop
);
554 timer_del(&job
->sleep_timer
);
557 aio_co_wake(job
->co
);
561 void job_enter(Job
*job
)
564 job_enter_cond_locked(job
, NULL
);
567 /* Yield, and schedule a timer to reenter the coroutine after @ns nanoseconds.
568 * Reentering the job coroutine with job_enter() before the timer has expired
569 * is allowed and cancels the timer.
571 * If @ns is (uint64_t) -1, no timer is scheduled and job_enter() must be
574 * Called with job_mutex held, but releases it temporarily.
576 static void coroutine_fn
job_do_yield_locked(Job
*job
, uint64_t ns
)
578 AioContext
*next_aio_context
;
581 timer_mod(&job
->sleep_timer
, ns
);
584 job_event_idle_locked(job
);
586 qemu_coroutine_yield();
589 next_aio_context
= job
->aio_context
;
591 * Coroutine has resumed, but in the meanwhile the job AioContext
592 * might have changed via bdrv_try_change_aio_context(), so we need to move
593 * the coroutine too in the new aiocontext.
595 while (qemu_get_current_aio_context() != next_aio_context
) {
597 aio_co_reschedule_self(next_aio_context
);
599 next_aio_context
= job
->aio_context
;
602 /* Set by job_enter_cond_locked() before re-entering the coroutine. */
606 /* Called with job_mutex held, but releases it temporarily. */
607 static void coroutine_fn
job_pause_point_locked(Job
*job
)
609 assert(job
&& job_started_locked(job
));
611 if (!job_should_pause_locked(job
)) {
614 if (job_is_cancelled_locked(job
)) {
618 if (job
->driver
->pause
) {
620 job
->driver
->pause(job
);
624 if (job_should_pause_locked(job
) && !job_is_cancelled_locked(job
)) {
625 JobStatus status
= job
->status
;
626 job_state_transition_locked(job
, status
== JOB_STATUS_READY
628 : JOB_STATUS_PAUSED
);
630 job_do_yield_locked(job
, -1);
632 job_state_transition_locked(job
, status
);
635 if (job
->driver
->resume
) {
637 job
->driver
->resume(job
);
642 void coroutine_fn
job_pause_point(Job
*job
)
645 job_pause_point_locked(job
);
648 void coroutine_fn
job_yield(Job
*job
)
653 /* Check cancellation *before* setting busy = false, too! */
654 if (job_is_cancelled_locked(job
)) {
658 if (!job_should_pause_locked(job
)) {
659 job_do_yield_locked(job
, -1);
662 job_pause_point_locked(job
);
665 void coroutine_fn
job_sleep_ns(Job
*job
, int64_t ns
)
670 /* Check cancellation *before* setting busy = false, too! */
671 if (job_is_cancelled_locked(job
)) {
675 if (!job_should_pause_locked(job
)) {
676 job_do_yield_locked(job
, qemu_clock_get_ns(QEMU_CLOCK_REALTIME
) + ns
);
679 job_pause_point_locked(job
);
682 /* Assumes the job_mutex is held */
683 static bool job_timer_not_pending_locked(Job
*job
)
685 return !timer_pending(&job
->sleep_timer
);
688 void job_pause_locked(Job
*job
)
692 job_enter_cond_locked(job
, NULL
);
696 void job_pause(Job
*job
)
699 job_pause_locked(job
);
702 void job_resume_locked(Job
*job
)
704 assert(job
->pause_count
> 0);
706 if (job
->pause_count
) {
710 /* kick only if no timer is pending */
711 job_enter_cond_locked(job
, job_timer_not_pending_locked
);
714 void job_resume(Job
*job
)
717 job_resume_locked(job
);
720 void job_user_pause_locked(Job
*job
, Error
**errp
)
722 if (job_apply_verb_locked(job
, JOB_VERB_PAUSE
, errp
)) {
725 if (job
->user_paused
) {
726 error_setg(errp
, "Job is already paused");
729 job
->user_paused
= true;
730 job_pause_locked(job
);
733 bool job_user_paused_locked(Job
*job
)
735 return job
->user_paused
;
738 void job_user_resume_locked(Job
*job
, Error
**errp
)
742 if (!job
->user_paused
|| job
->pause_count
<= 0) {
743 error_setg(errp
, "Can't resume a job that was not paused");
746 if (job_apply_verb_locked(job
, JOB_VERB_RESUME
, errp
)) {
749 if (job
->driver
->user_resume
) {
751 job
->driver
->user_resume(job
);
754 job
->user_paused
= false;
755 job_resume_locked(job
);
758 /* Called with job_mutex held, but releases it temporarily. */
759 static void job_do_dismiss_locked(Job
*job
)
764 job
->deferred_to_main_loop
= true;
766 job_txn_del_job_locked(job
);
768 job_state_transition_locked(job
, JOB_STATUS_NULL
);
769 job_unref_locked(job
);
772 void job_dismiss_locked(Job
**jobptr
, Error
**errp
)
775 /* similarly to _complete, this is QMP-interface only. */
777 if (job_apply_verb_locked(job
, JOB_VERB_DISMISS
, errp
)) {
781 job_do_dismiss_locked(job
);
785 void job_early_fail(Job
*job
)
788 assert(job
->status
== JOB_STATUS_CREATED
);
789 job_do_dismiss_locked(job
);
792 /* Called with job_mutex held. */
793 static void job_conclude_locked(Job
*job
)
795 job_state_transition_locked(job
, JOB_STATUS_CONCLUDED
);
796 if (job
->auto_dismiss
|| !job_started_locked(job
)) {
797 job_do_dismiss_locked(job
);
801 /* Called with job_mutex held. */
802 static void job_update_rc_locked(Job
*job
)
804 if (!job
->ret
&& job_is_cancelled_locked(job
)) {
805 job
->ret
= -ECANCELED
;
809 error_setg(&job
->err
, "%s", strerror(-job
->ret
));
811 job_state_transition_locked(job
, JOB_STATUS_ABORTING
);
815 static void job_commit(Job
*job
)
819 if (job
->driver
->commit
) {
820 job
->driver
->commit(job
);
824 static void job_abort(Job
*job
)
828 if (job
->driver
->abort
) {
829 job
->driver
->abort(job
);
833 static void job_clean(Job
*job
)
836 if (job
->driver
->clean
) {
837 job
->driver
->clean(job
);
842 * Called with job_mutex held, but releases it temporarily.
843 * Takes AioContext lock internally to invoke a job->driver callback.
845 static int job_finalize_single_locked(Job
*job
)
848 AioContext
*ctx
= job
->aio_context
;
850 assert(job_is_completed_locked(job
));
852 /* Ensure abort is called for late-transactional failures */
853 job_update_rc_locked(job
);
857 aio_context_acquire(ctx
);
867 job
->cb(job
->opaque
, job_ret
);
870 aio_context_release(ctx
);
873 /* Emit events only if we actually started */
874 if (job_started_locked(job
)) {
875 if (job_is_cancelled_locked(job
)) {
876 job_event_cancelled_locked(job
);
878 job_event_completed_locked(job
);
882 job_txn_del_job_locked(job
);
883 job_conclude_locked(job
);
888 * Called with job_mutex held, but releases it temporarily.
889 * Takes AioContext lock internally to invoke a job->driver callback.
891 static void job_cancel_async_locked(Job
*job
, bool force
)
893 AioContext
*ctx
= job
->aio_context
;
895 if (job
->driver
->cancel
) {
897 aio_context_acquire(ctx
);
898 force
= job
->driver
->cancel(job
, force
);
899 aio_context_release(ctx
);
902 /* No .cancel() means the job will behave as if force-cancelled */
906 if (job
->user_paused
) {
907 /* Do not call job_enter here, the caller will handle it. */
908 if (job
->driver
->user_resume
) {
910 job
->driver
->user_resume(job
);
913 job
->user_paused
= false;
914 assert(job
->pause_count
> 0);
919 * Ignore soft cancel requests after the job is already done
920 * (We will still invoke job->driver->cancel() above, but if the
921 * job driver supports soft cancelling and the job is done, that
922 * should be a no-op, too. We still call it so it can override
925 if (force
|| !job
->deferred_to_main_loop
) {
926 job
->cancelled
= true;
927 /* To prevent 'force == false' overriding a previous 'force == true' */
928 job
->force_cancel
|= force
;
933 * Called with job_mutex held, but releases it temporarily.
934 * Takes AioContext lock internally to invoke a job->driver callback.
936 static void job_completed_txn_abort_locked(Job
*job
)
938 JobTxn
*txn
= job
->txn
;
943 * We are cancelled by another job, which will handle everything.
947 txn
->aborting
= true;
948 job_txn_ref_locked(txn
);
952 /* Other jobs are effectively cancelled by us, set the status for
953 * them; this job, however, may or may not be cancelled, depending
954 * on the caller, so leave it. */
955 QLIST_FOREACH(other_job
, &txn
->jobs
, txn_list
) {
956 if (other_job
!= job
) {
958 * This is a transaction: If one job failed, no result will matter.
959 * Therefore, pass force=true to terminate all other jobs as quickly
962 job_cancel_async_locked(other_job
, true);
965 while (!QLIST_EMPTY(&txn
->jobs
)) {
966 other_job
= QLIST_FIRST(&txn
->jobs
);
967 if (!job_is_completed_locked(other_job
)) {
968 assert(job_cancel_requested_locked(other_job
));
969 job_finish_sync_locked(other_job
, NULL
, NULL
);
971 job_finalize_single_locked(other_job
);
974 job_unref_locked(job
);
975 job_txn_unref_locked(txn
);
978 /* Called with job_mutex held, but releases it temporarily */
979 static int job_prepare_locked(Job
*job
)
982 AioContext
*ctx
= job
->aio_context
;
986 if (job
->ret
== 0 && job
->driver
->prepare
) {
988 aio_context_acquire(ctx
);
989 ret
= job
->driver
->prepare(job
);
990 aio_context_release(ctx
);
993 job_update_rc_locked(job
);
999 /* Called with job_mutex held */
1000 static int job_needs_finalize_locked(Job
*job
)
1002 return !job
->auto_finalize
;
1005 /* Called with job_mutex held */
1006 static void job_do_finalize_locked(Job
*job
)
1009 assert(job
&& job
->txn
);
1011 /* prepare the transaction to complete */
1012 rc
= job_txn_apply_locked(job
, job_prepare_locked
);
1014 job_completed_txn_abort_locked(job
);
1016 job_txn_apply_locked(job
, job_finalize_single_locked
);
1020 void job_finalize_locked(Job
*job
, Error
**errp
)
1022 assert(job
&& job
->id
);
1023 if (job_apply_verb_locked(job
, JOB_VERB_FINALIZE
, errp
)) {
1026 job_do_finalize_locked(job
);
1029 /* Called with job_mutex held. */
1030 static int job_transition_to_pending_locked(Job
*job
)
1032 job_state_transition_locked(job
, JOB_STATUS_PENDING
);
1033 if (!job
->auto_finalize
) {
1034 job_event_pending_locked(job
);
1039 void job_transition_to_ready(Job
*job
)
1042 job_state_transition_locked(job
, JOB_STATUS_READY
);
1043 job_event_ready_locked(job
);
1046 /* Called with job_mutex held. */
1047 static void job_completed_txn_success_locked(Job
*job
)
1049 JobTxn
*txn
= job
->txn
;
1052 job_state_transition_locked(job
, JOB_STATUS_WAITING
);
1055 * Successful completion, see if there are other running jobs in this
1058 QLIST_FOREACH(other_job
, &txn
->jobs
, txn_list
) {
1059 if (!job_is_completed_locked(other_job
)) {
1062 assert(other_job
->ret
== 0);
1065 job_txn_apply_locked(job
, job_transition_to_pending_locked
);
1067 /* If no jobs need manual finalization, automatically do so */
1068 if (job_txn_apply_locked(job
, job_needs_finalize_locked
) == 0) {
1069 job_do_finalize_locked(job
);
1073 /* Called with job_mutex held. */
1074 static void job_completed_locked(Job
*job
)
1076 assert(job
&& job
->txn
&& !job_is_completed_locked(job
));
1078 job_update_rc_locked(job
);
1079 trace_job_completed(job
, job
->ret
);
1081 job_completed_txn_abort_locked(job
);
1083 job_completed_txn_success_locked(job
);
1088 * Useful only as a type shim for aio_bh_schedule_oneshot.
1089 * Called with job_mutex *not* held.
1091 static void job_exit(void *opaque
)
1093 Job
*job
= (Job
*)opaque
;
1095 job_ref_locked(job
);
1097 /* This is a lie, we're not quiescent, but still doing the completion
1098 * callbacks. However, completion callbacks tend to involve operations that
1099 * drain block nodes, and if .drained_poll still returned true, we would
1102 job_event_idle_locked(job
);
1104 job_completed_locked(job
);
1105 job_unref_locked(job
);
1109 * All jobs must allow a pause point before entering their job proper. This
1110 * ensures that jobs can be paused prior to being started, then resumed later.
1112 static void coroutine_fn
job_co_entry(void *opaque
)
1117 assert(job
&& job
->driver
&& job
->driver
->run
);
1118 WITH_JOB_LOCK_GUARD() {
1119 assert(job
->aio_context
== qemu_get_current_aio_context());
1120 job_pause_point_locked(job
);
1122 ret
= job
->driver
->run(job
, &job
->err
);
1123 WITH_JOB_LOCK_GUARD() {
1125 job
->deferred_to_main_loop
= true;
1128 aio_bh_schedule_oneshot(qemu_get_aio_context(), job_exit
, job
);
1131 void job_start(Job
*job
)
1133 assert(qemu_in_main_thread());
1135 WITH_JOB_LOCK_GUARD() {
1136 assert(job
&& !job_started_locked(job
) && job
->paused
&&
1137 job
->driver
&& job
->driver
->run
);
1138 job
->co
= qemu_coroutine_create(job_co_entry
, job
);
1141 job
->paused
= false;
1142 job_state_transition_locked(job
, JOB_STATUS_RUNNING
);
1144 aio_co_enter(job
->aio_context
, job
->co
);
1147 void job_cancel_locked(Job
*job
, bool force
)
1149 if (job
->status
== JOB_STATUS_CONCLUDED
) {
1150 job_do_dismiss_locked(job
);
1153 job_cancel_async_locked(job
, force
);
1154 if (!job_started_locked(job
)) {
1155 job_completed_locked(job
);
1156 } else if (job
->deferred_to_main_loop
) {
1158 * job_cancel_async() ignores soft-cancel requests for jobs
1159 * that are already done (i.e. deferred to the main loop). We
1160 * have to check again whether the job is really cancelled.
1161 * (job_cancel_requested() and job_is_cancelled() are equivalent
1162 * here, because job_cancel_async() will make soft-cancel
1163 * requests no-ops when deferred_to_main_loop is true. We
1164 * choose to call job_is_cancelled() to show that we invoke
1165 * job_completed_txn_abort() only for force-cancelled jobs.)
1167 if (job_is_cancelled_locked(job
)) {
1168 job_completed_txn_abort_locked(job
);
1171 job_enter_cond_locked(job
, NULL
);
1175 void job_user_cancel_locked(Job
*job
, bool force
, Error
**errp
)
1177 if (job_apply_verb_locked(job
, JOB_VERB_CANCEL
, errp
)) {
1180 job_cancel_locked(job
, force
);
1183 /* A wrapper around job_cancel_locked() taking an Error ** parameter so it may
1184 * be used with job_finish_sync_locked() without the need for (rather nasty)
1185 * function pointer casts there.
1187 * Called with job_mutex held.
1189 static void job_cancel_err_locked(Job
*job
, Error
**errp
)
1191 job_cancel_locked(job
, false);
1195 * Same as job_cancel_err(), but force-cancel.
1196 * Called with job_mutex held.
1198 static void job_force_cancel_err_locked(Job
*job
, Error
**errp
)
1200 job_cancel_locked(job
, true);
1203 int job_cancel_sync_locked(Job
*job
, bool force
)
1206 return job_finish_sync_locked(job
, &job_force_cancel_err_locked
, NULL
);
1208 return job_finish_sync_locked(job
, &job_cancel_err_locked
, NULL
);
1212 int job_cancel_sync(Job
*job
, bool force
)
1215 return job_cancel_sync_locked(job
, force
);
1218 void job_cancel_sync_all(void)
1223 while ((job
= job_next_locked(NULL
))) {
1224 job_cancel_sync_locked(job
, true);
1228 int job_complete_sync_locked(Job
*job
, Error
**errp
)
1230 return job_finish_sync_locked(job
, job_complete_locked
, errp
);
1233 void job_complete_locked(Job
*job
, Error
**errp
)
1235 /* Should not be reachable via external interface for internal jobs */
1237 GLOBAL_STATE_CODE();
1238 if (job_apply_verb_locked(job
, JOB_VERB_COMPLETE
, errp
)) {
1241 if (job_cancel_requested_locked(job
) || !job
->driver
->complete
) {
1242 error_setg(errp
, "The active block job '%s' cannot be completed",
1248 job
->driver
->complete(job
, errp
);
1252 int job_finish_sync_locked(Job
*job
,
1253 void (*finish
)(Job
*, Error
**errp
),
1256 Error
*local_err
= NULL
;
1258 GLOBAL_STATE_CODE();
1260 job_ref_locked(job
);
1263 finish(job
, &local_err
);
1266 error_propagate(errp
, local_err
);
1267 job_unref_locked(job
);
1272 AIO_WAIT_WHILE_UNLOCKED(job
->aio_context
,
1273 (job_enter(job
), !job_is_completed(job
)));
1276 ret
= (job_is_cancelled_locked(job
) && job
->ret
== 0)
1277 ? -ECANCELED
: job
->ret
;
1278 job_unref_locked(job
);