This lock is going to replace most of the AioContext locks
in the job and blockjob, so that a Job can run in an arbitrary
AioContext.
Signed-off-by: Emanuele Giuseppe Esposito <eespo...@redhat.com>
---
include/block/blockjob_int.h | 1 +
include/qemu/job.h | 2 +
block/backup.c | 4 +
block/mirror.c | 11 +-
blockdev.c | 62 ++++----
blockjob.c | 67 +++++++--
job-qmp.c | 55 +++----
job.c | 284 +++++++++++++++++++++++++++--------
qemu-img.c | 15 +-
9 files changed, 350 insertions(+), 151 deletions(-)
diff --git a/include/block/blockjob_int.h b/include/block/blockjob_int.h
index 6633d83da2..8b91126506 100644
--- a/include/block/blockjob_int.h
+++ b/include/block/blockjob_int.h
@@ -53,6 +53,7 @@ struct BlockJobDriver {
*/
void (*attached_aio_context)(BlockJob *job, AioContext *new_context);
+ /* Called with job mutex *not* held. */
void (*set_speed)(BlockJob *job, int64_t speed);
};
diff --git a/include/qemu/job.h b/include/qemu/job.h
index 4421d08d93..359f4e6b3a 100644
--- a/include/qemu/job.h
+++ b/include/qemu/job.h
@@ -49,6 +49,8 @@ typedef struct Job {
/**
* The type of this job.
* Set it in job_create and just read.
+ * All calls to the driver function must be not locked by job_mutex,
+ * to avoid deadlocks.
*/
const JobDriver *driver;
diff --git a/block/backup.c b/block/backup.c
index bd3614ce70..80ce956299 100644
--- a/block/backup.c
+++ b/block/backup.c
@@ -315,6 +315,10 @@ static void coroutine_fn backup_pause(Job *job)
}
}
+/*
+ * Called with job mutex *not* held (we don't want to call block_copy_kick
+ * with the lock held!)
+ */
static void coroutine_fn backup_set_speed(BlockJob *job, int64_t speed)
{
BackupBlockJob *s = container_of(job, BackupBlockJob, common);
diff --git a/block/mirror.c b/block/mirror.c
index 49aaaafffa..deefaa6a39 100644
--- a/block/mirror.c
+++ b/block/mirror.c
@@ -1150,9 +1150,11 @@ static void mirror_complete(Job *job, Error **errp)
s->should_complete = true;
/* If the job is paused, it will be re-entered when it is resumed */
+ job_lock();
if (!job_is_paused(job)) {
- job_enter(job);
+ job_enter_locked(job);
}
+ job_unlock();
}
static void coroutine_fn mirror_pause(Job *job)
@@ -1171,10 +1173,13 @@ static bool mirror_drained_poll(BlockJob *job)
* from one of our own drain sections, to avoid a deadlock waiting for
* ourselves.
*/
- if (!job_is_paused(&s->common.job) && !job_is_cancelled(&s->common.job) &&
- !s->in_drain) {
+ job_lock();
+ if (!job_is_paused(&s->common.job) &&
+ !job_is_cancelled_locked(&s->common.job) && !s->in_drain) {
+ job_unlock();
return true;
}
+ job_unlock();
return !!s->in_flight;
}
diff --git a/blockdev.c b/blockdev.c
index 8e2c15370e..9255aea6a2 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -150,9 +150,11 @@ void blockdev_mark_auto_del(BlockBackend *blk)
AioContext *aio_context = job_get_aiocontext(&job->job);
aio_context_acquire(aio_context);
+ job_lock();
job_cancel(&job->job, false);
aio_context_release(aio_context);
+ job_unlock();
}
}
@@ -3309,48 +3311,44 @@ out:
aio_context_release(aio_context);
}
-/* Get a block job using its ID and acquire its AioContext */
-static BlockJob *find_block_job(const char *id, AioContext **aio_context,
- Error **errp)
+/* Get a block job using its ID and acquire its job_lock */
+static BlockJob *find_block_job(const char *id, Error **errp)
{
BlockJob *job;
assert(id != NULL);
- *aio_context = NULL;
-
+ job_lock();
job = block_job_get(id);
if (!job) {
error_set(errp, ERROR_CLASS_DEVICE_NOT_ACTIVE,
"Block job '%s' not found", id);
+ job_unlock();
return NULL;
}
- *aio_context = blk_get_aio_context(job->blk);
- aio_context_acquire(*aio_context);
-
return job;
}
+/* Called with job_mutex *not* held. */
void qmp_block_job_set_speed(const char *device, int64_t speed, Error **errp)
{
- AioContext *aio_context;
- BlockJob *job = find_block_job(device, &aio_context, errp);
+ BlockJob *job = find_block_job(device, errp);
if (!job) {
return;
}
block_job_set_speed(job, speed, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
void qmp_block_job_cancel(const char *device,
bool has_force, bool force, Error **errp)
{
- AioContext *aio_context;
- BlockJob *job = find_block_job(device, &aio_context, errp);
+ BlockJob *job = find_block_job(device, errp);
if (!job) {
return;
@@ -3369,13 +3367,13 @@ void qmp_block_job_cancel(const char *device,
trace_qmp_block_job_cancel(job);
job_user_cancel(&job->job, force, errp);
out:
- aio_context_release(aio_context);
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
void qmp_block_job_pause(const char *device, Error **errp)
{
- AioContext *aio_context;
- BlockJob *job = find_block_job(device, &aio_context, errp);
+ BlockJob *job = find_block_job(device, errp);
if (!job) {
return;
@@ -3383,13 +3381,13 @@ void qmp_block_job_pause(const char *device, Error
**errp)
trace_qmp_block_job_pause(job);
job_user_pause(&job->job, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
void qmp_block_job_resume(const char *device, Error **errp)
{
- AioContext *aio_context;
- BlockJob *job = find_block_job(device, &aio_context, errp);
+ BlockJob *job = find_block_job(device, errp);
if (!job) {
return;
@@ -3397,13 +3395,13 @@ void qmp_block_job_resume(const char *device, Error
**errp)
trace_qmp_block_job_resume(job);
job_user_resume(&job->job, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
void qmp_block_job_complete(const char *device, Error **errp)
{
- AioContext *aio_context;
- BlockJob *job = find_block_job(device, &aio_context, errp);
+ BlockJob *job = find_block_job(device, errp);
if (!job) {
return;
@@ -3411,13 +3409,13 @@ void qmp_block_job_complete(const char *device, Error
**errp)
trace_qmp_block_job_complete(job);
job_complete(&job->job, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
void qmp_block_job_finalize(const char *id, Error **errp)
{
- AioContext *aio_context;
- BlockJob *job = find_block_job(id, &aio_context, errp);
+ BlockJob *job = find_block_job(id, errp);
if (!job) {
return;
@@ -3427,20 +3425,14 @@ void qmp_block_job_finalize(const char *id, Error
**errp)
job_ref(&job->job);
job_finalize(&job->job, errp);
- /*
- * Job's context might have changed via job_finalize (and job_txn_apply
- * automatically acquires the new one), so make sure we release the correct
- * one.
- */
- aio_context = blk_get_aio_context(job->blk);
job_unref(&job->job);
- aio_context_release(aio_context);
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
void qmp_block_job_dismiss(const char *id, Error **errp)
{
- AioContext *aio_context;
- BlockJob *bjob = find_block_job(id, &aio_context, errp);
+ BlockJob *bjob = find_block_job(id, errp);
Job *job;
if (!bjob) {
@@ -3450,7 +3442,7 @@ void qmp_block_job_dismiss(const char *id, Error **errp)
trace_qmp_block_job_dismiss(bjob);
job = &bjob->job;
job_dismiss(&job, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
void qmp_change_backing_file(const char *device,
diff --git a/blockjob.c b/blockjob.c
index 7f49f03ec7..e7b289089b 100644
--- a/blockjob.c
+++ b/blockjob.c
@@ -42,15 +42,16 @@
* The first includes functions used by the monitor. The monitor is
* peculiar in that it accesses the block job list with block_job_get, and
* therefore needs consistency across block_job_get and the actual operation
- * (e.g. block_job_set_speed). The consistency is achieved with
- * aio_context_acquire/release. These functions are declared in blockjob.h.
+ * (e.g. block_job_set_speed). To achieve this consistency, the caller
+ * calls block_job_lock/block_job_unlock itself around the whole operation.
+ * These functions are declared in blockjob.h.
*
* The second includes functions used by the block job drivers and sometimes
- * by the core block layer. These do not care about locking, because the
- * whole coroutine runs under the AioContext lock, and are declared in
- * blockjob_int.h.
+ * by the core block layer. These delegate the locking to the callee instead,
+ * and are declared in blockjob_int.h.
*/
+/* Does not need job_mutex. Value is never modified */
static bool is_block_job(Job *job)
{
return job_type(job) == JOB_TYPE_BACKUP ||
@@ -59,6 +60,7 @@ static bool is_block_job(Job *job)
job_type(job) == JOB_TYPE_STREAM;
}
+/* Called with job_mutex *not* held. */
BlockJob *block_job_next(BlockJob *bjob)
{
Job *job = bjob ? &bjob->job : NULL;
@@ -70,6 +72,7 @@ BlockJob *block_job_next(BlockJob *bjob)
return job ? container_of(job, BlockJob, job) : NULL;
}
+/* Called with job_mutex held. */
BlockJob *block_job_get(const char *id)
{
Job *job = job_get(id);
@@ -97,24 +100,31 @@ static char *child_job_get_parent_desc(BdrvChild *c)
return g_strdup_printf("%s job '%s'", job_type_str(&job->job),
job->job.id);
}
+/* Called with job_mutex *not* held. */
static void child_job_drained_begin(BdrvChild *c)
{
BlockJob *job = c->opaque;
+ job_lock();
job_pause(&job->job);
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
static bool child_job_drained_poll(BdrvChild *c)
{
BlockJob *bjob = c->opaque;
Job *job = &bjob->job;
const BlockJobDriver *drv = block_job_driver(bjob);
+ job_lock();
/* An inactive or completed job doesn't have any pending requests. Jobs
* with !job->busy are either already paused or have a pause point after
* being reentered, so no job driver code will run before they pause. */
- if (!job_is_busy(job) || job_is_completed(job)) {
+ if (!job_is_busy(job) || job_is_completed_locked(job)) {
+ job_unlock();
return false;
}
+ job_unlock();
/* Otherwise, assume that it isn't fully stopped yet, but allow the job to
* override this assumption. */
@@ -125,10 +135,13 @@ static bool child_job_drained_poll(BdrvChild *c)
}
}
+/* Called with job_mutex *not* held. */
static void child_job_drained_end(BdrvChild *c, int *drained_end_counter)
{
BlockJob *job = c->opaque;
+ job_lock();
job_resume(&job->job);
+ job_unlock();
}
static bool child_job_can_set_aio_ctx(BdrvChild *c, AioContext *ctx,
@@ -246,11 +259,15 @@ int block_job_add_bdrv(BlockJob *job, const char *name,
BlockDriverState *bs,
return 0;
}
+/* Called with job_mutex held. Temporarly releases the lock. */
static void block_job_on_idle(Notifier *n, void *opaque)
{
+ job_unlock();
aio_wait_kick();
+ job_lock();
}
+/* Does not need job_mutex. Value is never modified */
bool block_job_is_internal(BlockJob *job)
{
return (job->job.id == NULL);
@@ -267,6 +284,7 @@ static bool job_timer_pending(Job *job)
return timer_pending(&job->sleep_timer);
}
+/* Called with job_mutex held. May temporarly release the lock. */
bool block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
{
const BlockJobDriver *drv = block_job_driver(job);
@@ -286,7 +304,9 @@ bool block_job_set_speed(BlockJob *job, int64_t speed,
Error **errp)
job->speed = speed;
if (drv->set_speed) {
+ job_unlock();
drv->set_speed(job, speed);
+ job_lock();
}
if (speed && speed <= old_speed) {
@@ -304,6 +324,7 @@ int64_t block_job_ratelimit_get_delay(BlockJob *job,
uint64_t n)
return ratelimit_calculate_delay(&job->limit, n);
}
+/* Called with block_job_mutex *not* held. */
BlockJobInfo *block_job_query(BlockJob *blkjob, Error **errp)
{
BlockJobInfo *info;
@@ -319,6 +340,7 @@ BlockJobInfo *block_job_query(BlockJob *blkjob, Error
**errp)
progress_get_snapshot(&job->progress, &progress_current,
&progress_total);
+ job_lock();
info = g_new0(BlockJobInfo, 1);
info->type = g_strdup(job_type_str(job));
info->device = g_strdup(job->id);
@@ -328,11 +350,11 @@ BlockJobInfo *block_job_query(BlockJob *blkjob, Error
**errp)
info->len = progress_total;
info->speed = blkjob->speed;
info->io_status = blkjob->iostatus;
- info->ready = job_is_ready(job);
+ info->ready = job_is_ready_locked(job);
info->status = job_get_status(job);
info->auto_finalize = job->auto_finalize;
info->auto_dismiss = job->auto_dismiss;
- job_ret = job_get_ret(job);
+ job_ret = job_get_ret_locked(job);
if (job_ret) {
Error *job_err = job_get_err(job);
info->has_error = true;
@@ -340,9 +362,11 @@ BlockJobInfo *block_job_query(BlockJob *blkjob, Error
**errp)
g_strdup(error_get_pretty(job_err)) :
g_strdup(strerror(-job_ret));
}
+ job_unlock();
return info;
}
+/* Called with job_mutex held. */
static void block_job_iostatus_set_err(BlockJob *job, int error)
{
if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
@@ -351,6 +375,7 @@ static void block_job_iostatus_set_err(BlockJob *job, int
error)
}
}
+/* Called with job_mutex held. */
static void block_job_event_cancelled(Notifier *n, void *opaque)
{
BlockJob *job = opaque;
@@ -370,6 +395,7 @@ static void block_job_event_cancelled(Notifier *n, void
*opaque)
job->speed);
}
+/* Called with job_mutex held. */
static void block_job_event_completed(Notifier *n, void *opaque)
{
BlockJob *blkjob = opaque;
@@ -381,7 +407,7 @@ static void block_job_event_completed(Notifier *n, void
*opaque)
return;
}
- if (job_get_ret(job) < 0) {
+ if (job_get_ret_locked(job) < 0) {
msg = error_get_pretty(job_get_err(job));
}
@@ -397,6 +423,7 @@ static void block_job_event_completed(Notifier *n, void
*opaque)
msg);
}
+/* Called with job_mutex held. */
static void block_job_event_pending(Notifier *n, void *opaque)
{
BlockJob *job = opaque;
@@ -409,6 +436,7 @@ static void block_job_event_pending(Notifier *n, void
*opaque)
job->job.id);
}
+/* Called with job_mutex held. */
static void block_job_event_ready(Notifier *n, void *opaque)
{
BlockJob *job = opaque;
@@ -430,10 +458,11 @@ static void block_job_event_ready(Notifier *n, void
*opaque)
/*
- * API for block job drivers and the block layer. These functions are
- * declared in blockjob_int.h.
+ * API for block job drivers and the block layer, who do not know about
+ * job_mutex. These functions are declared in blockjob_int.h.
*/
+/* Called with block_job_mutex *not* held, but temporarly releases it. */
void *block_job_create(const char *job_id, const BlockJobDriver *driver,
JobTxn *txn, BlockDriverState *bs, uint64_t perm,
uint64_t shared_perm, int64_t speed, int flags,
@@ -472,6 +501,8 @@ void *block_job_create(const char *job_id, const
BlockJobDriver *driver,
job->ready_notifier.notify = block_job_event_ready;
job->idle_notifier.notify = block_job_on_idle;
+ job_lock();
+
notifier_list_add(&job->job.on_finalize_cancelled,
&job->finalize_cancelled_notifier);
notifier_list_add(&job->job.on_finalize_completed,
@@ -482,7 +513,11 @@ void *block_job_create(const char *job_id, const
BlockJobDriver *driver,
error_setg(&job->blocker, "block device is in use by block job: %s",
job_type_str(&job->job));
+
+ job_unlock();
+ /* calls drain and friends, that already take the lock */
block_job_add_bdrv(job, "main node", bs, 0, BLK_PERM_ALL, &error_abort);
+ job_lock();
bdrv_op_unblock(bs, BLOCK_OP_TYPE_DATAPLANE, job->blocker);
@@ -493,27 +528,35 @@ void *block_job_create(const char *job_id, const
BlockJobDriver *driver,
if (!block_job_set_speed(job, speed, errp)) {
job_early_fail(&job->job);
+ job_unlock();
return NULL;
}
+ job_unlock();
return job;
}
+/* Called with job_mutex *not* held. */
void block_job_iostatus_reset(BlockJob *job)
{
+ job_lock();
if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
+ job_unlock();
return;
}
assert(job_user_paused(&job->job) && job_should_pause(&job->job));
job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
void block_job_user_resume(Job *job)
{
BlockJob *bjob = container_of(job, BlockJob, job);
block_job_iostatus_reset(bjob);
}
+/* Called with job_mutex *not* held. */
BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
int is_read, int error)
{
@@ -544,12 +587,14 @@ BlockErrorAction block_job_error_action(BlockJob *job,
BlockdevOnError on_err,
action);
}
if (action == BLOCK_ERROR_ACTION_STOP) {
+ job_lock();
if (!job_user_paused(&job->job)) {
job_pause(&job->job);
/* make the pause user visible, which will be resumed from QMP. */
job_set_user_paused(&job->job);
}
block_job_iostatus_set_err(job, error);
+ job_unlock();
}
return action;
}
diff --git a/job-qmp.c b/job-qmp.c
index 12238a1643..03f3946490 100644
--- a/job-qmp.c
+++ b/job-qmp.c
@@ -29,29 +29,26 @@
#include "qapi/error.h"
#include "trace/trace-root.h"
-/* Get a job using its ID and acquire its AioContext */
-static Job *find_job(const char *id, AioContext **aio_context, Error **errp)
+/* Get a job using its ID and acquire its job_lock */
+static Job *find_job(const char *id, Error **errp)
{
Job *job;
- *aio_context = NULL;
+ job_lock();
job = job_get(id);
if (!job) {
error_setg(errp, "Job not found");
+ job_unlock();
return NULL;
}
- *aio_context = job_get_aiocontext(job);
- aio_context_acquire(*aio_context);
-
return job;
}
void qmp_job_cancel(const char *id, Error **errp)
{
- AioContext *aio_context;
- Job *job = find_job(id, &aio_context, errp);
+ Job *job = find_job(id, errp);
if (!job) {
return;
@@ -59,13 +56,12 @@ void qmp_job_cancel(const char *id, Error **errp)
trace_qmp_job_cancel(job);
job_user_cancel(job, true, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
void qmp_job_pause(const char *id, Error **errp)
{
- AioContext *aio_context;
- Job *job = find_job(id, &aio_context, errp);
+ Job *job = find_job(id, errp);
if (!job) {
return;
@@ -73,13 +69,12 @@ void qmp_job_pause(const char *id, Error **errp)
trace_qmp_job_pause(job);
job_user_pause(job, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
void qmp_job_resume(const char *id, Error **errp)
{
- AioContext *aio_context;
- Job *job = find_job(id, &aio_context, errp);
+ Job *job = find_job(id, errp);
if (!job) {
return;
@@ -87,13 +82,12 @@ void qmp_job_resume(const char *id, Error **errp)
trace_qmp_job_resume(job);
job_user_resume(job, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
void qmp_job_complete(const char *id, Error **errp)
{
- AioContext *aio_context;
- Job *job = find_job(id, &aio_context, errp);
+ Job *job = find_job(id, errp);
if (!job) {
return;
@@ -101,13 +95,12 @@ void qmp_job_complete(const char *id, Error **errp)
trace_qmp_job_complete(job);
job_complete(job, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
void qmp_job_finalize(const char *id, Error **errp)
{
- AioContext *aio_context;
- Job *job = find_job(id, &aio_context, errp);
+ Job *job = find_job(id, errp);
if (!job) {
return;
@@ -117,20 +110,13 @@ void qmp_job_finalize(const char *id, Error **errp)
job_ref(job);
job_finalize(job, errp);
- /*
- * Job's context might have changed via job_finalize (and job_txn_apply
- * automatically acquires the new one), so make sure we release the correct
- * one.
- */
- aio_context = job_get_aiocontext(job);
job_unref(job);
- aio_context_release(aio_context);
+ job_unlock();
}
void qmp_job_dismiss(const char *id, Error **errp)
{
- AioContext *aio_context;
- Job *job = find_job(id, &aio_context, errp);
+ Job *job = find_job(id, errp);
if (!job) {
return;
@@ -138,9 +124,10 @@ void qmp_job_dismiss(const char *id, Error **errp)
trace_qmp_job_dismiss(job);
job_dismiss(&job, errp);
- aio_context_release(aio_context);
+ job_unlock();
}
+/* Called with job_mutex held. */
static JobInfo *job_query_single(Job *job, Error **errp)
{
JobInfo *info;
@@ -175,15 +162,15 @@ JobInfoList *qmp_query_jobs(Error **errp)
for (job = job_next(NULL); job; job = job_next(job)) {
JobInfo *value;
- AioContext *aio_context;
if (job_is_internal(job)) {
continue;
}
- aio_context = job_get_aiocontext(job);
- aio_context_acquire(aio_context);
+
+ job_lock();
value = job_query_single(job, errp);
- aio_context_release(aio_context);
+ job_unlock();
+
if (!value) {
qapi_free_JobInfoList(head);
return NULL;
diff --git a/job.c b/job.c
index 48b304c3ff..e2006532b5 100644
--- a/job.c
+++ b/job.c
@@ -93,19 +93,22 @@ static void __attribute__((__constructor__)) job_init(void)
qemu_mutex_init(&job_mutex);
}
+/* Does not need job_mutex */
AioContext *job_get_aiocontext(Job *job)
{
- return job->aio_context;
+ return qatomic_read(&job->aio_context);
}
+/* Does not need job_mutex */
void job_set_aiocontext(Job *job, AioContext *aio)
{
- job->aio_context = aio;
+ qatomic_set(&job->aio_context, aio);
}
+/* Called with job_mutex held. */
bool job_is_busy(Job *job)
{
- return qatomic_read(&job->busy);
+ return job->busy;
}
/* Called with job_mutex held. */
@@ -124,59 +127,75 @@ int job_get_ret(Job *job)
return ret;
}
+/* Called with job_mutex held. */
Error *job_get_err(Job *job)
{
return job->err;
}
+/* Called with job_mutex held. */
JobStatus job_get_status(Job *job)
{
return job->status;
}
-
+/* Called with job_mutex *not* held. */
void job_set_cancelled(Job *job, bool cancel)
{
+ job_lock();
job->cancelled = cancel;
+ job_unlock();
}
+/* Called with job_mutex *not* held. */
bool job_is_force_cancel(Job *job)
{
- return job->force_cancel;
+ bool ret;
+ job_lock();
+ ret = job->force_cancel;
+ job_unlock();
+ return ret;
}
+/* Does not need job_mutex */
JobTxn *job_txn_new(void)
{
JobTxn *txn = g_new0(JobTxn, 1);
QLIST_INIT(&txn->jobs);
- txn->refcnt = 1;
+ qatomic_set(&txn->refcnt, 1);
return txn;
}
+/* Does not need job_mutex */
static void job_txn_ref(JobTxn *txn)
{
- txn->refcnt++;
+ qatomic_inc(&txn->refcnt);
}
+/* Does not need job_mutex */
void job_txn_unref(JobTxn *txn)
{
- if (txn && --txn->refcnt == 0) {
+ if (txn && qatomic_dec_fetch(&txn->refcnt) == 0) {
g_free(txn);
}
}
+/* Called with job_mutex *not* held. */
void job_txn_add_job(JobTxn *txn, Job *job)
{
if (!txn) {
return;
}
+ job_lock();
assert(!job->txn);
job->txn = txn;
QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
+ job_unlock();
job_txn_ref(txn);
}
+/* Called with job_mutex held. */
static void job_txn_del_job(Job *job)
{
if (job->txn) {
@@ -186,6 +205,7 @@ static void job_txn_del_job(Job *job)
}
}
+/* Called with job_mutex held. */
static int job_txn_apply(Job *job, int fn(Job *))
{
AioContext *inner_ctx;
@@ -221,11 +241,13 @@ static int job_txn_apply(Job *job, int fn(Job *))
return rc;
}
+/* Does not need job_mutex */
bool job_is_internal(Job *job)
{
return (job->id == NULL);
}
+/* Called with job_mutex held. */
static void job_state_transition(Job *job, JobStatus s1)
{
JobStatus s0 = job->status;
@@ -241,6 +263,7 @@ static void job_state_transition(Job *job, JobStatus s1)
}
}
+/* Called with job_mutex held. */
int job_apply_verb(Job *job, JobVerb verb, Error **errp)
{
JobStatus s0 = job->status;
@@ -255,11 +278,13 @@ int job_apply_verb(Job *job, JobVerb verb, Error **errp)
return -EPERM;
}
+/* Does not need job_mutex. Value is never modified */
JobType job_type(const Job *job)
{
return job->driver->job_type;
}
+/* Does not need job_mutex. Value is never modified */
const char *job_type_str(const Job *job)
{
return JobType_str(job_type(job));
@@ -353,24 +378,34 @@ static bool job_started(Job *job)
return job->co;
}
+/* Called with job_mutex held. */
bool job_should_pause(Job *job)
{
return job->pause_count > 0;
}
+/* Called with job_mutex held. */
bool job_is_paused(Job *job)
{
return job->paused;
}
+/* Called with job_mutex *not* held. */
Job *job_next(Job *job)
{
+ Job *ret;
+ job_lock();
if (!job) {
- return QLIST_FIRST(&jobs);
+ ret = QLIST_FIRST(&jobs);
+ job_unlock();
+ return ret;
}
- return QLIST_NEXT(job, job_list);
+ ret = QLIST_NEXT(job, job_list);
+ job_unlock();
+ return ret;
}
+/* Called with job_mutex held. */
Job *job_get(const char *id)
{
Job *job;
@@ -388,13 +423,14 @@ Job *job_get(const char *id)
return NULL;
}
+/* Called with job_mutex *not* held. */
static void job_sleep_timer_cb(void *opaque)
{
Job *job = opaque;
-
job_enter(job);
}
+/* Called with job_mutex *not* held. */
void *job_create(const char *job_id, const JobDriver *driver, JobTxn *txn,
AioContext *ctx, int flags, BlockCompletionFunc *cb,
void *opaque, Error **errp)
@@ -449,6 +485,7 @@ void *job_create(const char *job_id, const JobDriver
*driver, JobTxn *txn,
job_sleep_timer_cb, job);
QLIST_INSERT_HEAD(&jobs, job, job_list);
+ job_unlock();
/* Single jobs are modeled as single-job transactions for sake of
* consolidating the job management logic */
@@ -463,11 +500,13 @@ void *job_create(const char *job_id, const JobDriver
*driver, JobTxn *txn,
return job;
}
+/* Called with job_mutex held. */
void job_ref(Job *job)
{
++job->refcnt;
}
+/* Called with job_mutex held. Temporarly releases the lock. */
void job_unref(Job *job)
{
if (--job->refcnt == 0) {
@@ -476,7 +515,9 @@ void job_unref(Job *job)
assert(!job->txn);
if (job->driver->free) {
+ job_unlock();
job->driver->free(job);
+ job_lock();
}
QLIST_REMOVE(job, job_list);
@@ -488,46 +529,55 @@ void job_unref(Job *job)
}
}
+/* API is thread safe */
void job_progress_update(Job *job, uint64_t done)
{
progress_work_done(&job->progress, done);
}
+/* API is thread safe */
void job_progress_set_remaining(Job *job, uint64_t remaining)
{
progress_set_remaining(&job->progress, remaining);
}
+/* API is thread safe */
void job_progress_increase_remaining(Job *job, uint64_t delta)
{
progress_increase_remaining(&job->progress, delta);
}
+/* Called with job_mutex held. */
void job_event_cancelled(Job *job)
{
notifier_list_notify(&job->on_finalize_cancelled, job);
}
+/* Called with job_mutex held. */
void job_event_completed(Job *job)
{
notifier_list_notify(&job->on_finalize_completed, job);
}
+/* Called with job_mutex held. */
static void job_event_pending(Job *job)
{
notifier_list_notify(&job->on_pending, job);
}
+/* Called with job_mutex held. */
static void job_event_ready(Job *job)
{
notifier_list_notify(&job->on_ready, job);
}
+/* Called with job_mutex held. */
static void job_event_idle(Job *job)
{
notifier_list_notify(&job->on_idle, job);
}
+/* Called with job_mutex held, but releases it temporarly. */
void job_enter_cond(Job *job, bool(*fn)(Job *job))
{
if (!job_started(job)) {
@@ -537,14 +587,11 @@ void job_enter_cond(Job *job, bool(*fn)(Job *job))
return;
}
- job_lock();
if (job->busy) {
- job_unlock();
return;
}
if (fn && !fn(job)) {
- job_unlock();
return;
}
@@ -552,7 +599,8 @@ void job_enter_cond(Job *job, bool(*fn)(Job *job))
timer_del(&job->sleep_timer);
job->busy = true;
job_unlock();
- aio_co_enter(job->aio_context, job->co);
+ aio_co_enter(job_get_aiocontext(job), job->co);
+ job_lock();
}
/* Called with job_mutex held. */
@@ -565,7 +613,7 @@ void job_enter_locked(Job *job)
void job_enter(Job *job)
{
job_lock();
- job_enter_locked(job, NULL);
+ job_enter_locked(job);
job_unlock();
}
@@ -574,7 +622,11 @@ void job_enter(Job *job)
* is allowed and cancels the timer.
*
* If @ns is (uint64_t) -1, no timer is scheduled and job_enter() must be
- * called explicitly. */
+ * called explicitly.
+ *
+ * Called with job_mutex *not* held (we don't want the coroutine
+ * to yield with the lock held!).
+ */
static void coroutine_fn job_do_yield(Job *job, uint64_t ns)
{
job_lock();
@@ -587,86 +639,122 @@ static void coroutine_fn job_do_yield(Job *job, uint64_t
ns)
qemu_coroutine_yield();
/* Set by job_enter_cond() before re-entering the coroutine. */
+ job_lock();
assert(job->busy);
+ job_unlock();
}
+/*
+ * Called with job_mutex *not* held (we don't want the coroutine
+ * to yield with the lock held!).
+ */
void coroutine_fn job_pause_point(Job *job)
{
assert(job && job_started(job));
+ job_lock();
if (!job_should_pause(job)) {
+ job_unlock();
return;
}
- if (job_is_cancelled(job)) {
+ if (job_is_cancelled_locked(job)) {
+ job_unlock();
return;
}
if (job->driver->pause) {
+ job_unlock();
job->driver->pause(job);
+ job_lock();
}
- if (job_should_pause(job) && !job_is_cancelled(job)) {
+ if (job_should_pause(job) && !job_is_cancelled_locked(job)) {
JobStatus status = job->status;
job_state_transition(job, status == JOB_STATUS_READY
? JOB_STATUS_STANDBY
: JOB_STATUS_PAUSED);
job->paused = true;
+ job_unlock();
job_do_yield(job, -1);
+ job_lock();
job->paused = false;
job_state_transition(job, status);
}
+ job_unlock();
if (job->driver->resume) {
job->driver->resume(job);
}
}
+/*
+ * Called with job_mutex *not* held (we don't want the coroutine
+ * to yield with the lock held!).
+ */
void job_yield(Job *job)
{
+ bool res;
+ job_lock();
assert(job->busy);
/* Check cancellation *before* setting busy = false, too! */
- if (job_is_cancelled(job)) {
+ if (job_is_cancelled_locked(job)) {
+ job_unlock();
return;
}
- if (!job_should_pause(job)) {
+ res = job_should_pause(job);
+ job_unlock();
+
+ if (!res) {
job_do_yield(job, -1);
}
job_pause_point(job);
}
+/*
+ * Called with job_mutex *not* held (we don't want the coroutine
+ * to yield with the lock held!).
+ */
void coroutine_fn job_sleep_ns(Job *job, int64_t ns)
{
+ bool res;
+ job_lock();
assert(job->busy);
/* Check cancellation *before* setting busy = false, too! */
- if (job_is_cancelled(job)) {
+ if (job_is_cancelled_locked(job)) {
+ job_unlock();
return;
}
- if (!job_should_pause(job)) {
+ res = job_should_pause(job);
+ job_unlock();
+
+ if (!res) {
job_do_yield(job, qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + ns);
}
job_pause_point(job);
}
-/* Assumes the block_job_mutex is held */
+/* Called with job_mutex held. */
static bool job_timer_not_pending(Job *job)
{
return !timer_pending(&job->sleep_timer);
}
+/* Called with job_mutex held. */
void job_pause(Job *job)
{
job->pause_count++;
if (!job->paused) {
- job_enter(job);
+ job_enter_locked(job);
}
}
+/* Called with job_mutex held. */
void job_resume(Job *job)
{
assert(job->pause_count > 0);
@@ -679,6 +767,7 @@ void job_resume(Job *job)
job_enter_cond(job, job_timer_not_pending);
}
+/* Called with job_mutex held. */
void job_user_pause(Job *job, Error **errp)
{
if (job_apply_verb(job, JOB_VERB_PAUSE, errp)) {
@@ -692,16 +781,19 @@ void job_user_pause(Job *job, Error **errp)
job_pause(job);
}
+/* Called with job_mutex held. */
bool job_user_paused(Job *job)
{
return job->user_paused;
}
+/* Called with job_mutex held. */
void job_set_user_paused(Job *job)
{
job->user_paused = true;
}
+/* Called with job_mutex held. Temporarly releases the lock. */
void job_user_resume(Job *job, Error **errp)
{
assert(job);
@@ -713,12 +805,15 @@ void job_user_resume(Job *job, Error **errp)
return;
}
if (job->driver->user_resume) {
+ job_unlock();
job->driver->user_resume(job);
+ job_lock();
}
job->user_paused = false;
job_resume(job);
}
+/* Called with job_mutex held. */
static void job_do_dismiss(Job *job)
{
assert(job);
@@ -732,6 +827,7 @@ static void job_do_dismiss(Job *job)
job_unref(job);
}
+/* Called with job_mutex held. */
void job_dismiss(Job **jobptr, Error **errp)
{
Job *job = *jobptr;
@@ -761,9 +857,10 @@ static void job_conclude(Job *job)
}
}
+/* Called with job_mutex held. */
static void job_update_rc(Job *job)
{
- if (!job->ret && job_is_cancelled(job)) {
+ if (!job->ret && job_is_cancelled_locked(job)) {
job->ret = -ECANCELED;
}
if (job->ret) {
@@ -774,22 +871,25 @@ static void job_update_rc(Job *job)
}
}
+/* Called with job_mutex *not* held. */
static void job_commit(Job *job)
{
- assert(!job->ret);
+ assert(!job_get_ret(job));
if (job->driver->commit) {
job->driver->commit(job);
}
}
+/* Called with job_mutex *not* held. */
static void job_abort(Job *job)
{
- assert(job->ret);
+ assert(job_get_ret(job));
if (job->driver->abort) {
job->driver->abort(job);
}
}
+/* Called with job_mutex *not* held. */
static void job_clean(Job *job)
{
if (job->driver->clean) {
@@ -797,14 +897,18 @@ static void job_clean(Job *job)
}
}
+/* Called with job lock held, but it releases it temporarily */
static int job_finalize_single(Job *job)
{
- assert(job_is_completed(job));
+ int ret;
+ assert(job_is_completed_locked(job));
/* Ensure abort is called for late-transactional failures */
job_update_rc(job);
- if (!job->ret) {
+ ret = job->ret;
+ job_unlock();
+ if (!ret) {
job_commit(job);
} else {
job_abort(job);
@@ -812,12 +916,13 @@ static int job_finalize_single(Job *job)
job_clean(job);
if (job->cb) {
- job->cb(job->opaque, job->ret);
+ job->cb(job->opaque, ret);
}
+ job_lock();
/* Emit events only if we actually started */
if (job_started(job)) {
- if (job_is_cancelled(job)) {
+ if (job_is_cancelled_locked(job)) {
job_event_cancelled(job);
} else {
job_event_completed(job);
@@ -829,15 +934,20 @@ static int job_finalize_single(Job *job)
return 0;
}
+/* Called with job_mutex held. Temporarly releases the lock. */
static void job_cancel_async(Job *job, bool force)
{
if (job->driver->cancel) {
+ job_unlock();
job->driver->cancel(job, force);
+ job_lock();
}
if (job->user_paused) {
/* Do not call job_enter here, the caller will handle it. */
if (job->driver->user_resume) {
+ job_unlock();
job->driver->user_resume(job);
+ job_lock();
}
job->user_paused = false;
assert(job->pause_count > 0);
@@ -848,27 +958,21 @@ static void job_cancel_async(Job *job, bool force)
job->force_cancel |= force;
}
+/* Called with job_mutex held. */
static void job_completed_txn_abort(Job *job)
{
- AioContext *outer_ctx = job->aio_context;
AioContext *ctx;
JobTxn *txn = job->txn;
Job *other_job;
- if (txn->aborting) {
+ if (qatomic_cmpxchg(&txn->aborting, false, true)) {
/*
* We are cancelled by another job, which will handle everything.
*/
return;
}
- txn->aborting = true;
job_txn_ref(txn);
- /* We can only hold the single job's AioContext lock while calling
- * job_finalize_single() because the finalization callbacks can involve
- * calls of AIO_WAIT_WHILE(), which could deadlock otherwise. */
- aio_context_release(outer_ctx);
-
/* Other jobs are effectively cancelled by us, set the status for
* them; this job, however, may or may not be cancelled, depending
* on the caller, so leave it. */
@@ -884,33 +988,39 @@ static void job_completed_txn_abort(Job *job)
other_job = QLIST_FIRST(&txn->jobs);
ctx = other_job->aio_context;
aio_context_acquire(ctx);
- if (!job_is_completed(other_job)) {
- assert(job_is_cancelled(other_job));
+ if (!job_is_completed_locked(other_job)) {
+ assert(job_is_cancelled_locked(other_job));
job_finish_sync(other_job, NULL, NULL);
}
job_finalize_single(other_job);
aio_context_release(ctx);
}
- aio_context_acquire(outer_ctx);
-
job_txn_unref(txn);
}
+/* Called with job_mutex held. Temporarly releases the lock. */
static int job_prepare(Job *job)
{
+ int ret;
+
if (job->ret == 0 && job->driver->prepare) {
- job->ret = job->driver->prepare(job);
+ job_unlock();
+ ret = job->driver->prepare(job);
+ job_lock();
+ job->ret = ret;
job_update_rc(job);
}
return job->ret;
}
+/* Does not need job_mutex */
static int job_needs_finalize(Job *job)
{
return !job->auto_finalize;
}
+/* Called with job_mutex held. */
static void job_do_finalize(Job *job)
{
int rc;
@@ -925,6 +1035,7 @@ static void job_do_finalize(Job *job)
}
}
+/* Called with job_mutex held. */
void job_finalize(Job *job, Error **errp)
{
assert(job && job->id);
@@ -934,6 +1045,7 @@ void job_finalize(Job *job, Error **errp)
job_do_finalize(job);
}
+/* Called with job_mutex held. */
static int job_transition_to_pending(Job *job)
{
job_state_transition(job, JOB_STATUS_PENDING);
@@ -943,17 +1055,22 @@ static int job_transition_to_pending(Job *job)
return 0;
}
+/* Called with job_mutex *not* held. */
void job_transition_to_ready(Job *job)
{
+ job_lock();
job_state_transition(job, JOB_STATUS_READY);
job_event_ready(job);
+ job_unlock();
}
+/* Called with job_mutex held. */
static void job_completed_txn_success(Job *job)
{
- JobTxn *txn = job->txn;
+ JobTxn *txn;
Job *other_job;
+ txn = job->txn;
job_state_transition(job, JOB_STATUS_WAITING);
/*
@@ -961,7 +1078,7 @@ static void job_completed_txn_success(Job *job)
* txn.
*/
QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
- if (!job_is_completed(other_job)) {
+ if (!job_is_completed_locked(other_job)) {
return;
}
assert(other_job->ret == 0);
@@ -975,9 +1092,10 @@ static void job_completed_txn_success(Job *job)
}
}
+/* Called with job_mutex held. */
static void job_completed(Job *job)
{
- assert(job && job->txn && !job_is_completed(job));
+ assert(job && job->txn && !job_is_completed_locked(job));
job_update_rc(job);
trace_job_completed(job, job->ret);
@@ -988,14 +1106,16 @@ static void job_completed(Job *job)
}
}
-/** Useful only as a type shim for aio_bh_schedule_oneshot. */
+/**
+ * Useful only as a type shim for aio_bh_schedule_oneshot.
+ * Called with job_mutex *not* held.
+ */
static void job_exit(void *opaque)
{
Job *job = (Job *)opaque;
- AioContext *ctx;
+ job_lock();
job_ref(job);
- aio_context_acquire(job->aio_context);
/* This is a lie, we're not quiescent, but still doing the completion
* callbacks. However, completion callbacks tend to involve operations that
@@ -1012,29 +1132,40 @@ static void job_exit(void *opaque)
* acquiring the new lock, and we ref/unref to avoid job_completed freeing
* the job underneath us.
*/
- ctx = job->aio_context;
job_unref(job);
- aio_context_release(ctx);
+ job_unlock();
}
/**
* All jobs must allow a pause point before entering their job proper. This
* ensures that jobs can be paused prior to being started, then resumed later.
+ *
+ * Called with job_mutex *not* held.
*/
static void coroutine_fn job_co_entry(void *opaque)
{
Job *job = opaque;
+ Error *local_error = NULL;
+ int ret;
assert(job && job->driver && job->driver->run);
job_pause_point(job);
- job->ret = job->driver->run(job, &job->err);
+ ret = job->driver->run(job, &local_error);
+ job_lock();
+ if (local_error) {
+ error_propagate(&job->err, local_error);
+ }
+ job->ret = ret;
job->deferred_to_main_loop = true;
job->busy = true;
+ job_unlock();
aio_bh_schedule_oneshot(qemu_get_aio_context(), job_exit, job);
}
+/* Called with job_mutex *not* held. */
void job_start(Job *job)
{
+ job_lock();
assert(job && !job_started(job) && job->paused &&
job->driver && job->driver->run);
job->co = qemu_coroutine_create(job_co_entry, job);
@@ -1042,9 +1173,11 @@ void job_start(Job *job)
job->busy = true;
job->paused = false;
job_state_transition(job, JOB_STATUS_RUNNING);
- aio_co_enter(job->aio_context, job->co);
+ job_unlock();
+ aio_co_enter(job_get_aiocontext(job), job->co);
}
+/* Called with job_mutex held. */
void job_cancel(Job *job, bool force)
{
if (job->status == JOB_STATUS_CONCLUDED) {
@@ -1057,10 +1190,11 @@ void job_cancel(Job *job, bool force)
} else if (job->deferred_to_main_loop) {
job_completed_txn_abort(job);
} else {
- job_enter(job);
+ job_enter_locked(job);
}
}
+/* Called with job_mutex held. */
void job_user_cancel(Job *job, bool force, Error **errp)
{
if (job_apply_verb(job, JOB_VERB_CANCEL, errp)) {
@@ -1069,19 +1203,36 @@ void job_user_cancel(Job *job, bool force, Error **errp)
job_cancel(job, force);
}
-/* A wrapper around job_cancel() taking an Error ** parameter so it may be
+/*
+ * A wrapper around job_cancel() taking an Error ** parameter so it may be
* used with job_finish_sync() without the need for (rather nasty) function
- * pointer casts there. */
+ * pointer casts there.
+ *
+ * Called with job_mutex held.
+ */
static void job_cancel_err(Job *job, Error **errp)
{
job_cancel(job, false);
}
+/*
+ * Called with job_mutex *not* held, unlike most other APIs consumed
+ * by the monitor!
+ */
int job_cancel_sync(Job *job)
{
- return job_finish_sync(job, &job_cancel_err, NULL);
+ int ret;
+
+ job_lock();
+ ret = job_finish_sync(job, &job_cancel_err, NULL);
+ job_unlock();
+ return ret;
}
+/*
+ * Called with job_mutex *not* held, unlike most other APIs consumed
+ * by the monitor!
+ */
void job_cancel_sync_all(void)
{
Job *job;
@@ -1095,11 +1246,13 @@ void job_cancel_sync_all(void)
}
}
+/* Called with job_mutex held. */
int job_complete_sync(Job *job, Error **errp)
{
return job_finish_sync(job, job_complete, errp);
}
+/* Called with job_mutex held. Temporarly releases the lock. */
void job_complete(Job *job, Error **errp)
{
/* Should not be reachable via external interface for internal jobs */
@@ -1107,15 +1260,18 @@ void job_complete(Job *job, Error **errp)
if (job_apply_verb(job, JOB_VERB_COMPLETE, errp)) {
return;
}
- if (job_is_cancelled(job) || !job->driver->complete) {
+ if (job_is_cancelled_locked(job) || !job->driver->complete) {
error_setg(errp, "The active block job '%s' cannot be completed",
job->id);
return;
}
+ job_unlock();
job->driver->complete(job, errp);
+ job_lock();
}
+/* Called with job_mutex held. */
int job_finish_sync(Job *job, void (*finish)(Job *, Error **errp), Error
**errp)
{
Error *local_err = NULL;
@@ -1132,10 +1288,12 @@ int job_finish_sync(Job *job, void (*finish)(Job *,
Error **errp), Error **errp)
return -EBUSY;
}
- AIO_WAIT_WHILE(job->aio_context,
- (job_enter(job), !job_is_completed(job)));
+ job_unlock();
+ AIO_WAIT_WHILE(NULL, (job_enter(job), !job_is_completed(job)));
+ job_lock();
- ret = (job_is_cancelled(job) && job->ret == 0) ? -ECANCELED : job->ret;
+ ret = (job_is_cancelled_locked(job) && job->ret == 0) ?
+ -ECANCELED : job->ret;
job_unref(job);
return ret;
}
diff --git a/qemu-img.c b/qemu-img.c
index d16bd367d9..82debde038 100644
--- a/qemu-img.c
+++ b/qemu-img.c
@@ -898,17 +898,19 @@ static void common_block_job_cb(void *opaque, int ret)
}
}
+/* Called with job_mutex held. Releases it temporarly */
static void run_block_job(BlockJob *job, Error **errp)
{
uint64_t progress_current, progress_total;
AioContext *aio_context = blk_get_aio_context(job->blk);
int ret = 0;
- aio_context_acquire(aio_context);
job_ref(&job->job);
do {
float progress = 0.0f;
+ job_unlock();
aio_poll(aio_context, true);
+ job_lock();
progress_get_snapshot(&job->job.progress, &progress_current,
&progress_total);
@@ -916,15 +918,15 @@ static void run_block_job(BlockJob *job, Error **errp)
progress = (float)progress_current / progress_total * 100.f;
}
qemu_progress_print(progress, 0);
- } while (!job_is_ready(&job->job) && !job_is_completed(&job->job));
+ } while (!job_is_ready_locked(&job->job) &&
+ !job_is_completed_locked(&job->job));
- if (!job_is_completed(&job->job)) {
+ if (!job_is_completed_locked(&job->job)) {
ret = job_complete_sync(&job->job, errp);
} else {
- ret = job_get_ret(&job->job);
+ ret = job_get_ret_locked(&job->job);
}
job_unref(&job->job);
- aio_context_release(aio_context);
/* publish completion progress only when success */
if (!ret) {
@@ -1076,9 +1078,12 @@ static int img_commit(int argc, char **argv)
bdrv_ref(bs);
}
+ job_lock();
job = block_job_get("commit");
assert(job);
run_block_job(job, &local_err);
+ job_unlock();
+
if (local_err) {
goto unref_backing;
}
--
2.31.1
