-----BEGIN PGP SIGNATURE-----

iQIcBAABAgAGBQJZKIBzAAoJEL2+eyfA3jBX530P/RydmmajCUSLTSVa5CfzLR7H
 +G/wOT4h9sAZKS/Gm8Dtu55Djx5NZznSjeVf0gpCZc2iwpcp+4zmYLihBrOq87lz
 SnIrD/NI5t9/1/Uk+3aKkFlK/tcZuHBKABdbr8fYy/SmnnWB5DKMbErNvkSm4n+Q
 n3N1upIUxojO3tBqxUq/VHFGsa7eDeOMl5t698PhGIUT1rRUTK6vnD/vMpyEIdyZ
 G34UOrGkiGDMEek7bSoA1EAKeNZBDlhKsPrBOdOLCJ0W3UY83agkzsvgnWO2w/Eg
 mi05CpgVJ9Ru57lWr9EQELznIowrV0rRVYwYT35lGhIoRdVvlMWRz295QZTYzicG
 FoqJVRSUQs+JwOPLB3aMuok9HNwda1i5Iol2MksKL59juDGlBxe8tTSjNfcsqE5E
 whyMaBaucLtI9sJ/oloMDN6gUSUzOmms001rdA5LELEi/jbn1RY/tIMQWwuhae2r
 4RYN+l0049fSG89237Oa81e3DT29Wb8qoaht6VWokwmeZroOxu6fhfEzPkfFvmE7
 x6V74+T+5tvZFDI8x6hE8J5YCGaxwIilJg3o77tA7+61OKfxLmEts70Mqn+sPhkd
 DbDT2drERR8n/Rf9pi67v7Wud/fwRPmNuu6naHt+fQtCn4YhM44qNVEBRbp9QENA
 NI6ouqecl4kHvIFiVDN/
 =WLib
 -----END PGP SIGNATURE-----

Merge remote-tracking branch 'jtc/tags/block-pull-request' into staging

# gpg: Signature made Fri 26 May 2017 08:22:27 PM BST
# gpg:                using RSA key 0xBDBE7B27C0DE3057
# gpg: Good signature from "Jeffrey Cody <jcody@redhat.com>"
# gpg:                 aka "Jeffrey Cody <jeff@codyprime.org>"
# gpg:                 aka "Jeffrey Cody <codyprime@gmail.com>"
# Primary key fingerprint: 9957 4B4D 3474 90E7 9D98  D624 BDBE 7B27 C0DE 3057

* jtc/tags/block-pull-request:
  block/gluster: glfs_lseek() workaround
  blockjob: use deferred_to_main_loop to indicate the coroutine has ended
  blockjob: reorganize block_job_completed_txn_abort
  blockjob: strengthen a bit test-blockjob-txn
  blockjob: group BlockJob transaction functions together
  blockjob: introduce block_job_cancel_async, check iostatus invariants
  blockjob: move iostatus reset inside block_job_user_resume
  blockjob: separate monitor and blockjob APIs
  blockjob: introduce block_job_pause/resume_all
  blockjob: introduce block_job_early_fail
  blockjob: remove iostatus_reset callback
  blockjob: remove unnecessary check

Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>

block/backup.c

@@ -692,7 +692,7 @@ BlockJob *backup_job_create(const char *job_id, BlockDriverState *bs,
     }
     if (job) {
         backup_clean(&job->common);
-        block_job_unref(&job->common);
+        block_job_early_fail(&job->common);
     }
 
     return NULL;

block/commit.c

@@ -426,7 +426,7 @@ fail:
     if (commit_top_bs) {
         bdrv_set_backing_hd(overlay_bs, top, &error_abort);
     }
-    block_job_unref(&s->common);
+    block_job_early_fail(&s->common);
 }
 
 

block/gluster.c

@@ -1275,7 +1275,14 @@ static int find_allocation(BlockDriverState *bs, off_t start,
     if (offs < 0) {
         return -errno;          /* D3 or D4 */
     }
-    assert(offs >= start);
+
+    if (offs < start) {
+        /* This is not a valid return by lseek().  We are safe to just return
+         * -EIO in this case, and we'll treat it like D4. Unfortunately some
+         *  versions of gluster server will return offs < start, so an assert
+         *  here will unnecessarily abort QEMU. */
+        return -EIO;
+    }
 
     if (offs > start) {
         /* D2: in hole, next data at offs */
@@ -1307,7 +1314,14 @@ static int find_allocation(BlockDriverState *bs, off_t start,
     if (offs < 0) {
         return -errno;          /* D1 and (H3 or H4) */
     }
-    assert(offs >= start);
+
+    if (offs < start) {
+        /* This is not a valid return by lseek().  We are safe to just return
+         * -EIO in this case, and we'll treat it like H4. Unfortunately some
+         *  versions of gluster server will return offs < start, so an assert
+         *  here will unnecessarily abort QEMU. */
+        return -EIO;
+    }
 
     if (offs > start) {
         /*

block/io.c

@@ -26,6 +26,7 @@
 #include "trace.h"
 #include "sysemu/block-backend.h"
 #include "block/blockjob.h"
+#include "block/blockjob_int.h"
 #include "block/block_int.h"
 #include "qemu/cutils.h"
 #include "qapi/error.h"
@@ -301,16 +302,9 @@ void bdrv_drain_all_begin(void)
     bool waited = true;
     BlockDriverState *bs;
     BdrvNextIterator it;
-    BlockJob *job = NULL;
     GSList *aio_ctxs = NULL, *ctx;
 
-    while ((job = block_job_next(job))) {
+    block_job_pause_all();
-        AioContext *aio_context = blk_get_aio_context(job->blk);
-
-        aio_context_acquire(aio_context);
-        block_job_pause(job);
-        aio_context_release(aio_context);
-    }
 
     for (bs = bdrv_first(&it); bs; bs = bdrv_next(&it)) {
         AioContext *aio_context = bdrv_get_aio_context(bs);
@@ -354,7 +348,6 @@ void bdrv_drain_all_end(void)
 {
     BlockDriverState *bs;
     BdrvNextIterator it;
-    BlockJob *job = NULL;
 
     for (bs = bdrv_first(&it); bs; bs = bdrv_next(&it)) {
         AioContext *aio_context = bdrv_get_aio_context(bs);
@@ -365,13 +358,7 @@ void bdrv_drain_all_end(void)
         aio_context_release(aio_context);
     }
 
-    while ((job = block_job_next(job))) {
+    block_job_resume_all();
-        AioContext *aio_context = blk_get_aio_context(job->blk);
-
-        aio_context_acquire(aio_context);
-        block_job_resume(job);
-        aio_context_release(aio_context);
-    }
 }
 
 void bdrv_drain_all(void)

block/mirror.c

@@ -1252,7 +1252,7 @@ fail:
 
         g_free(s->replaces);
         blk_unref(s->target);
-        block_job_unref(&s->common);
+        block_job_early_fail(&s->common);
     }
 
     bdrv_child_try_set_perm(mirror_top_bs->backing, 0, BLK_PERM_ALL,

blockdev.c

@@ -3715,7 +3715,6 @@ void qmp_block_job_resume(const char *device, Error **errp)
     }
 
     trace_qmp_block_job_resume(job);
-    block_job_iostatus_reset(job);
     block_job_user_resume(job);
     aio_context_release(aio_context);
 }

blockjob.c

@@ -55,35 +55,20 @@ struct BlockJobTxn {
 
 static QLIST_HEAD(, BlockJob) block_jobs = QLIST_HEAD_INITIALIZER(block_jobs);
 
-static char *child_job_get_parent_desc(BdrvChild *c)
+/*
-{
+ * The block job API is composed of two categories of functions.
-    BlockJob *job = c->opaque;
+ *
-    return g_strdup_printf("%s job '%s'",
+ * The first includes functions used by the monitor.  The monitor is
-                           BlockJobType_lookup[job->driver->job_type],
+ * peculiar in that it accesses the block job list with block_job_get, and
-                           job->id);
+ * 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.
-static const BdrvChildRole child_job = {
+ *
-    .get_parent_desc    = child_job_get_parent_desc,
+ * The second includes functions used by the block job drivers and sometimes
-    .stay_at_node       = true,
+ * 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.
-static void block_job_drained_begin(void *opaque)
+ */
-{
-    BlockJob *job = opaque;
-    block_job_pause(job);
-}
-
-static void block_job_drained_end(void *opaque)
-{
-    BlockJob *job = opaque;
-    block_job_resume(job);
-}
-
-static const BlockDevOps block_job_dev_ops = {
-    .drained_begin = block_job_drained_begin,
-    .drained_end = block_job_drained_end,
-};
 
 BlockJob *block_job_next(BlockJob *job)
 {
@@ -106,6 +91,80 @@ BlockJob *block_job_get(const char *id)
     return NULL;
 }
 
+BlockJobTxn *block_job_txn_new(void)
+{
+    BlockJobTxn *txn = g_new0(BlockJobTxn, 1);
+    QLIST_INIT(&txn->jobs);
+    txn->refcnt = 1;
+    return txn;
+}
+
+static void block_job_txn_ref(BlockJobTxn *txn)
+{
+    txn->refcnt++;
+}
+
+void block_job_txn_unref(BlockJobTxn *txn)
+{
+    if (txn && --txn->refcnt == 0) {
+        g_free(txn);
+    }
+}
+
+void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job)
+{
+    if (!txn) {
+        return;
+    }
+
+    assert(!job->txn);
+    job->txn = txn;
+
+    QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
+    block_job_txn_ref(txn);
+}
+
+static void block_job_pause(BlockJob *job)
+{
+    job->pause_count++;
+}
+
+static void block_job_resume(BlockJob *job)
+{
+    assert(job->pause_count > 0);
+    job->pause_count--;
+    if (job->pause_count) {
+        return;
+    }
+    block_job_enter(job);
+}
+
+static void block_job_ref(BlockJob *job)
+{
+    ++job->refcnt;
+}
+
+static void block_job_attached_aio_context(AioContext *new_context,
+                                           void *opaque);
+static void block_job_detach_aio_context(void *opaque);
+
+static void block_job_unref(BlockJob *job)
+{
+    if (--job->refcnt == 0) {
+        BlockDriverState *bs = blk_bs(job->blk);
+        bs->job = NULL;
+        block_job_remove_all_bdrv(job);
+        blk_remove_aio_context_notifier(job->blk,
+                                        block_job_attached_aio_context,
+                                        block_job_detach_aio_context, job);
+        blk_unref(job->blk);
+        error_free(job->blocker);
+        g_free(job->id);
+        QLIST_REMOVE(job, job_list);
+        g_free(job);
+    }
+}
+
 static void block_job_attached_aio_context(AioContext *new_context,
                                            void *opaque)
 {
@@ -145,6 +204,36 @@ static void block_job_detach_aio_context(void *opaque)
     block_job_unref(job);
 }
 
+static char *child_job_get_parent_desc(BdrvChild *c)
+{
+    BlockJob *job = c->opaque;
+    return g_strdup_printf("%s job '%s'",
+                           BlockJobType_lookup[job->driver->job_type],
+                           job->id);
+}
+
+static const BdrvChildRole child_job = {
+    .get_parent_desc    = child_job_get_parent_desc,
+    .stay_at_node       = true,
+};
+
+static void block_job_drained_begin(void *opaque)
+{
+    BlockJob *job = opaque;
+    block_job_pause(job);
+}
+
+static void block_job_drained_end(void *opaque)
+{
+    BlockJob *job = opaque;
+    block_job_resume(job);
+}
+
+static const BlockDevOps block_job_dev_ops = {
+    .drained_begin = block_job_drained_begin,
+    .drained_end = block_job_drained_end,
+};
+
 void block_job_remove_all_bdrv(BlockJob *job)
 {
     GSList *l;
@@ -175,6 +264,350 @@ int block_job_add_bdrv(BlockJob *job, const char *name, BlockDriverState *bs,
     return 0;
 }
 
+bool block_job_is_internal(BlockJob *job)
+{
+    return (job->id == NULL);
+}
+
+static bool block_job_started(BlockJob *job)
+{
+    return job->co;
+}
+
+/**
+ * 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.
+ */
+static void coroutine_fn block_job_co_entry(void *opaque)
+{
+    BlockJob *job = opaque;
+
+    assert(job && job->driver && job->driver->start);
+    block_job_pause_point(job);
+    job->driver->start(job);
+}
+
+void block_job_start(BlockJob *job)
+{
+    assert(job && !block_job_started(job) && job->paused &&
+           job->driver && job->driver->start);
+    job->co = qemu_coroutine_create(block_job_co_entry, job);
+    job->pause_count--;
+    job->busy = true;
+    job->paused = false;
+    bdrv_coroutine_enter(blk_bs(job->blk), job->co);
+}
+
+static void block_job_completed_single(BlockJob *job)
+{
+    assert(job->completed);
+
+    if (!job->ret) {
+        if (job->driver->commit) {
+            job->driver->commit(job);
+        }
+    } else {
+        if (job->driver->abort) {
+            job->driver->abort(job);
+        }
+    }
+    if (job->driver->clean) {
+        job->driver->clean(job);
+    }
+
+    if (job->cb) {
+        job->cb(job->opaque, job->ret);
+    }
+
+    /* Emit events only if we actually started */
+    if (block_job_started(job)) {
+        if (block_job_is_cancelled(job)) {
+            block_job_event_cancelled(job);
+        } else {
+            const char *msg = NULL;
+            if (job->ret < 0) {
+                msg = strerror(-job->ret);
+            }
+            block_job_event_completed(job, msg);
+        }
+    }
+
+    if (job->txn) {
+        QLIST_REMOVE(job, txn_list);
+        block_job_txn_unref(job->txn);
+    }
+    block_job_unref(job);
+}
+
+static void block_job_cancel_async(BlockJob *job)
+{
+    if (job->iostatus != BLOCK_DEVICE_IO_STATUS_OK) {
+        block_job_iostatus_reset(job);
+    }
+    if (job->user_paused) {
+        /* Do not call block_job_enter here, the caller will handle it.  */
+        job->user_paused = false;
+        job->pause_count--;
+    }
+    job->cancelled = true;
+}
+
+static int block_job_finish_sync(BlockJob *job,
+                                 void (*finish)(BlockJob *, Error **errp),
+                                 Error **errp)
+{
+    Error *local_err = NULL;
+    int ret;
+
+    assert(blk_bs(job->blk)->job == job);
+
+    block_job_ref(job);
+
+    if (finish) {
+        finish(job, &local_err);
+    }
+    if (local_err) {
+        error_propagate(errp, local_err);
+        block_job_unref(job);
+        return -EBUSY;
+    }
+    /* block_job_drain calls block_job_enter, and it should be enough to
+     * induce progress until the job completes or moves to the main thread.
+    */
+    while (!job->deferred_to_main_loop && !job->completed) {
+        block_job_drain(job);
+    }
+    while (!job->completed) {
+        aio_poll(qemu_get_aio_context(), true);
+    }
+    ret = (job->cancelled && job->ret == 0) ? -ECANCELED : job->ret;
+    block_job_unref(job);
+    return ret;
+}
+
+static void block_job_completed_txn_abort(BlockJob *job)
+{
+    AioContext *ctx;
+    BlockJobTxn *txn = job->txn;
+    BlockJob *other_job;
+
+    if (txn->aborting) {
+        /*
+         * We are cancelled by another job, which will handle everything.
+         */
+        return;
+    }
+    txn->aborting = true;
+    block_job_txn_ref(txn);
+
+    /* We are the first failed job. Cancel other jobs. */
+    QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
+        ctx = blk_get_aio_context(other_job->blk);
+        aio_context_acquire(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. */
+    QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
+        if (other_job != job) {
+            block_job_cancel_async(other_job);
+        }
+    }
+    while (!QLIST_EMPTY(&txn->jobs)) {
+        other_job = QLIST_FIRST(&txn->jobs);
+        ctx = blk_get_aio_context(other_job->blk);
+        if (!other_job->completed) {
+            assert(other_job->cancelled);
+            block_job_finish_sync(other_job, NULL, NULL);
+        }
+        block_job_completed_single(other_job);
+        aio_context_release(ctx);
+    }
+
+    block_job_txn_unref(txn);
+}
+
+static void block_job_completed_txn_success(BlockJob *job)
+{
+    AioContext *ctx;
+    BlockJobTxn *txn = job->txn;
+    BlockJob *other_job, *next;
+    /*
+     * Successful completion, see if there are other running jobs in this
+     * txn.
+     */
+    QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
+        if (!other_job->completed) {
+            return;
+        }
+    }
+    /* We are the last completed job, commit the transaction. */
+    QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
+        ctx = blk_get_aio_context(other_job->blk);
+        aio_context_acquire(ctx);
+        assert(other_job->ret == 0);
+        block_job_completed_single(other_job);
+        aio_context_release(ctx);
+    }
+}
+
+void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
+{
+    Error *local_err = NULL;
+
+    if (!job->driver->set_speed) {
+        error_setg(errp, QERR_UNSUPPORTED);
+        return;
+    }
+    job->driver->set_speed(job, speed, &local_err);
+    if (local_err) {
+        error_propagate(errp, local_err);
+        return;
+    }
+
+    job->speed = speed;
+}
+
+void block_job_complete(BlockJob *job, Error **errp)
+{
+    /* Should not be reachable via external interface for internal jobs */
+    assert(job->id);
+    if (job->pause_count || job->cancelled ||
+        !block_job_started(job) || !job->driver->complete) {
+        error_setg(errp, "The active block job '%s' cannot be completed",
+                   job->id);
+        return;
+    }
+
+    job->driver->complete(job, errp);
+}
+
+void block_job_user_pause(BlockJob *job)
+{
+    job->user_paused = true;
+    block_job_pause(job);
+}
+
+bool block_job_user_paused(BlockJob *job)
+{
+    return job->user_paused;
+}
+
+void block_job_user_resume(BlockJob *job)
+{
+    if (job && job->user_paused && job->pause_count > 0) {
+        block_job_iostatus_reset(job);
+        job->user_paused = false;
+        block_job_resume(job);
+    }
+}
+
+void block_job_cancel(BlockJob *job)
+{
+    if (block_job_started(job)) {
+        block_job_cancel_async(job);
+        block_job_enter(job);
+    } else {
+        block_job_completed(job, -ECANCELED);
+    }
+}
+
+/* A wrapper around block_job_cancel() taking an Error ** parameter so it may be
+ * used with block_job_finish_sync() without the need for (rather nasty)
+ * function pointer casts there. */
+static void block_job_cancel_err(BlockJob *job, Error **errp)
+{
+    block_job_cancel(job);
+}
+
+int block_job_cancel_sync(BlockJob *job)
+{
+    return block_job_finish_sync(job, &block_job_cancel_err, NULL);
+}
+
+void block_job_cancel_sync_all(void)
+{
+    BlockJob *job;
+    AioContext *aio_context;
+
+    while ((job = QLIST_FIRST(&block_jobs))) {
+        aio_context = blk_get_aio_context(job->blk);
+        aio_context_acquire(aio_context);
+        block_job_cancel_sync(job);
+        aio_context_release(aio_context);
+    }
+}
+
+int block_job_complete_sync(BlockJob *job, Error **errp)
+{
+    return block_job_finish_sync(job, &block_job_complete, errp);
+}
+
+BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
+{
+    BlockJobInfo *info;
+
+    if (block_job_is_internal(job)) {
+        error_setg(errp, "Cannot query QEMU internal jobs");
+        return NULL;
+    }
+    info = g_new0(BlockJobInfo, 1);
+    info->type      = g_strdup(BlockJobType_lookup[job->driver->job_type]);
+    info->device    = g_strdup(job->id);
+    info->len       = job->len;
+    info->busy      = job->busy;
+    info->paused    = job->pause_count > 0;
+    info->offset    = job->offset;
+    info->speed     = job->speed;
+    info->io_status = job->iostatus;
+    info->ready     = job->ready;
+    return info;
+}
+
+static void block_job_iostatus_set_err(BlockJob *job, int error)
+{
+    if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
+        job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
+                                          BLOCK_DEVICE_IO_STATUS_FAILED;
+    }
+}
+
+static void block_job_event_cancelled(BlockJob *job)
+{
+    if (block_job_is_internal(job)) {
+        return;
+    }
+
+    qapi_event_send_block_job_cancelled(job->driver->job_type,
+                                        job->id,
+                                        job->len,
+                                        job->offset,
+                                        job->speed,
+                                        &error_abort);
+}
+
+static void block_job_event_completed(BlockJob *job, const char *msg)
+{
+    if (block_job_is_internal(job)) {
+        return;
+    }
+
+    qapi_event_send_block_job_completed(job->driver->job_type,
+                                        job->id,
+                                        job->len,
+                                        job->offset,
+                                        job->speed,
+                                        !!msg,
+                                        msg,
+                                        &error_abort);
+}
+
+/*
+ * API for block job drivers and the block layer.  These functions are
+ * declared in blockjob_int.h.
+ */
+
 void *block_job_create(const char *job_id, const BlockJobDriver *driver,
                        BlockDriverState *bs, uint64_t perm,
                        uint64_t shared_perm, int64_t speed, int flags,
@@ -259,163 +692,23 @@ void *block_job_create(const char *job_id, const BlockJobDriver *driver,
     return job;
 }
 
-bool block_job_is_internal(BlockJob *job)
+void block_job_pause_all(void)
 {
-    return (job->id == NULL);
+    BlockJob *job = NULL;
-}
+    while ((job = block_job_next(job))) {
+        AioContext *aio_context = blk_get_aio_context(job->blk);
 
-static bool block_job_started(BlockJob *job)
+        aio_context_acquire(aio_context);
-{
+        block_job_pause(job);
-    return job->co;
+        aio_context_release(aio_context);
-}
-
-/**
- * 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.
- */
-static void coroutine_fn block_job_co_entry(void *opaque)
-{
-    BlockJob *job = opaque;
-
-    assert(job && job->driver && job->driver->start);
-    block_job_pause_point(job);
-    job->driver->start(job);
-}
-
-void block_job_start(BlockJob *job)
-{
-    assert(job && !block_job_started(job) && job->paused &&
-           job->driver && job->driver->start);
-    job->co = qemu_coroutine_create(block_job_co_entry, job);
-    job->pause_count--;
-    job->busy = true;
-    job->paused = false;
-    bdrv_coroutine_enter(blk_bs(job->blk), job->co);
-}
-
-void block_job_ref(BlockJob *job)
-{
-    ++job->refcnt;
-}
-
-void block_job_unref(BlockJob *job)
-{
-    if (--job->refcnt == 0) {
-        BlockDriverState *bs = blk_bs(job->blk);
-        bs->job = NULL;
-        block_job_remove_all_bdrv(job);
-        blk_remove_aio_context_notifier(job->blk,
-                                        block_job_attached_aio_context,
-                                        block_job_detach_aio_context, job);
-        blk_unref(job->blk);
-        error_free(job->blocker);
-        g_free(job->id);
-        QLIST_REMOVE(job, job_list);
-        g_free(job);
     }
 }
 
-static void block_job_completed_single(BlockJob *job)
+void block_job_early_fail(BlockJob *job)
 {
-    if (!job->ret) {
-        if (job->driver->commit) {
-            job->driver->commit(job);
-        }
-    } else {
-        if (job->driver->abort) {
-            job->driver->abort(job);
-        }
-    }
-    if (job->driver->clean) {
-        job->driver->clean(job);
-    }
-
-    if (job->cb) {
-        job->cb(job->opaque, job->ret);
-    }
-
-    /* Emit events only if we actually started */
-    if (block_job_started(job)) {
-        if (block_job_is_cancelled(job)) {
-            block_job_event_cancelled(job);
-        } else {
-            const char *msg = NULL;
-            if (job->ret < 0) {
-                msg = strerror(-job->ret);
-            }
-            block_job_event_completed(job, msg);
-        }
-    }
-
-    if (job->txn) {
-        QLIST_REMOVE(job, txn_list);
-        block_job_txn_unref(job->txn);
-    }
     block_job_unref(job);
 }
 
-static void block_job_completed_txn_abort(BlockJob *job)
-{
-    AioContext *ctx;
-    BlockJobTxn *txn = job->txn;
-    BlockJob *other_job, *next;
-
-    if (txn->aborting) {
-        /*
-         * We are cancelled by another job, which will handle everything.
-         */
-        return;
-    }
-    txn->aborting = true;
-    /* We are the first failed job. Cancel other jobs. */
-    QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
-        ctx = blk_get_aio_context(other_job->blk);
-        aio_context_acquire(ctx);
-    }
-    QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
-        if (other_job == job || other_job->completed) {
-            /* 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. */
-            if (other_job != job) {
-                other_job->cancelled = true;
-            }
-            continue;
-        }
-        block_job_cancel_sync(other_job);
-        assert(other_job->completed);
-    }
-    QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
-        ctx = blk_get_aio_context(other_job->blk);
-        block_job_completed_single(other_job);
-        aio_context_release(ctx);
-    }
-}
-
-static void block_job_completed_txn_success(BlockJob *job)
-{
-    AioContext *ctx;
-    BlockJobTxn *txn = job->txn;
-    BlockJob *other_job, *next;
-    /*
-     * Successful completion, see if there are other running jobs in this
-     * txn.
-     */
-    QLIST_FOREACH(other_job, &txn->jobs, txn_list) {
-        if (!other_job->completed) {
-            return;
-        }
-    }
-    /* We are the last completed job, commit the transaction. */
-    QLIST_FOREACH_SAFE(other_job, &txn->jobs, txn_list, next) {
-        ctx = blk_get_aio_context(other_job->blk);
-        aio_context_acquire(ctx);
-        assert(other_job->ret == 0);
-        block_job_completed_single(other_job);
-        aio_context_release(ctx);
-    }
-}
-
 void block_job_completed(BlockJob *job, int ret)
 {
     assert(blk_bs(job->blk)->job == job);
@@ -431,58 +724,11 @@ void block_job_completed(BlockJob *job, int ret)
     }
 }
 
-void block_job_set_speed(BlockJob *job, int64_t speed, Error **errp)
-{
-    Error *local_err = NULL;
-
-    if (!job->driver->set_speed) {
-        error_setg(errp, QERR_UNSUPPORTED);
-        return;
-    }
-    job->driver->set_speed(job, speed, &local_err);
-    if (local_err) {
-        error_propagate(errp, local_err);
-        return;
-    }
-
-    job->speed = speed;
-}
-
-void block_job_complete(BlockJob *job, Error **errp)
-{
-    /* Should not be reachable via external interface for internal jobs */
-    assert(job->id);
-    if (job->pause_count || job->cancelled ||
-        !block_job_started(job) || !job->driver->complete) {
-        error_setg(errp, "The active block job '%s' cannot be completed",
-                   job->id);
-        return;
-    }
-
-    job->driver->complete(job, errp);
-}
-
-void block_job_pause(BlockJob *job)
-{
-    job->pause_count++;
-}
-
-void block_job_user_pause(BlockJob *job)
-{
-    job->user_paused = true;
-    block_job_pause(job);
-}
-
 static bool block_job_should_pause(BlockJob *job)
 {
     return job->pause_count > 0;
 }
 
-bool block_job_user_paused(BlockJob *job)
-{
-    return job ? job->user_paused : 0;
-}
-
 void coroutine_fn block_job_pause_point(BlockJob *job)
 {
     assert(job && block_job_started(job));
@@ -511,39 +757,29 @@ void coroutine_fn block_job_pause_point(BlockJob *job)
     }
 }
 
-void block_job_resume(BlockJob *job)
+void block_job_resume_all(void)
 {
-    assert(job->pause_count > 0);
+    BlockJob *job = NULL;
-    job->pause_count--;
+    while ((job = block_job_next(job))) {
-    if (job->pause_count) {
+        AioContext *aio_context = blk_get_aio_context(job->blk);
-        return;
-    }
-    block_job_enter(job);
-}
 
-void block_job_user_resume(BlockJob *job)
+        aio_context_acquire(aio_context);
-{
-    if (job && job->user_paused && job->pause_count > 0) {
-        job->user_paused = false;
         block_job_resume(job);
+        aio_context_release(aio_context);
     }
 }
 
 void block_job_enter(BlockJob *job)
 {
-    if (job->co && !job->busy) {
+    if (!block_job_started(job)) {
-        bdrv_coroutine_enter(blk_bs(job->blk), job->co);
+        return;
+    }
+    if (job->deferred_to_main_loop) {
+        return;
     }
-}
 
-void block_job_cancel(BlockJob *job)
+    if (!job->busy) {
-{
+        bdrv_coroutine_enter(blk_bs(job->blk), job->co);
-    if (block_job_started(job)) {
-        job->cancelled = true;
-        block_job_iostatus_reset(job);
-        block_job_enter(job);
-    } else {
-        block_job_completed(job, -ECANCELED);
     }
 }
 
@@ -552,76 +788,6 @@ bool block_job_is_cancelled(BlockJob *job)
     return job->cancelled;
 }
 
-void block_job_iostatus_reset(BlockJob *job)
-{
-    job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
-    if (job->driver->iostatus_reset) {
-        job->driver->iostatus_reset(job);
-    }
-}
-
-static int block_job_finish_sync(BlockJob *job,
-                                 void (*finish)(BlockJob *, Error **errp),
-                                 Error **errp)
-{
-    Error *local_err = NULL;
-    int ret;
-
-    assert(blk_bs(job->blk)->job == job);
-
-    block_job_ref(job);
-
-    finish(job, &local_err);
-    if (local_err) {
-        error_propagate(errp, local_err);
-        block_job_unref(job);
-        return -EBUSY;
-    }
-    /* block_job_drain calls block_job_enter, and it should be enough to
-     * induce progress until the job completes or moves to the main thread.
-    */
-    while (!job->deferred_to_main_loop && !job->completed) {
-        block_job_drain(job);
-    }
-    while (!job->completed) {
-        aio_poll(qemu_get_aio_context(), true);
-    }
-    ret = (job->cancelled && job->ret == 0) ? -ECANCELED : job->ret;
-    block_job_unref(job);
-    return ret;
-}
-
-/* A wrapper around block_job_cancel() taking an Error ** parameter so it may be
- * used with block_job_finish_sync() without the need for (rather nasty)
- * function pointer casts there. */
-static void block_job_cancel_err(BlockJob *job, Error **errp)
-{
-    block_job_cancel(job);
-}
-
-int block_job_cancel_sync(BlockJob *job)
-{
-    return block_job_finish_sync(job, &block_job_cancel_err, NULL);
-}
-
-void block_job_cancel_sync_all(void)
-{
-    BlockJob *job;
-    AioContext *aio_context;
-
-    while ((job = QLIST_FIRST(&block_jobs))) {
-        aio_context = blk_get_aio_context(job->blk);
-        aio_context_acquire(aio_context);
-        block_job_cancel_sync(job);
-        aio_context_release(aio_context);
-    }
-}
-
-int block_job_complete_sync(BlockJob *job, Error **errp)
-{
-    return block_job_finish_sync(job, &block_job_complete, errp);
-}
-
 void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns)
 {
     assert(job->busy);
@@ -658,63 +824,13 @@ void block_job_yield(BlockJob *job)
     block_job_pause_point(job);
 }
 
-BlockJobInfo *block_job_query(BlockJob *job, Error **errp)
+void block_job_iostatus_reset(BlockJob *job)
-{
-    BlockJobInfo *info;
-
-    if (block_job_is_internal(job)) {
-        error_setg(errp, "Cannot query QEMU internal jobs");
-        return NULL;
-    }
-    info = g_new0(BlockJobInfo, 1);
-    info->type      = g_strdup(BlockJobType_lookup[job->driver->job_type]);
-    info->device    = g_strdup(job->id);
-    info->len       = job->len;
-    info->busy      = job->busy;
-    info->paused    = job->pause_count > 0;
-    info->offset    = job->offset;
-    info->speed     = job->speed;
-    info->io_status = job->iostatus;
-    info->ready     = job->ready;
-    return info;
-}
-
-static void block_job_iostatus_set_err(BlockJob *job, int error)
 {
     if (job->iostatus == BLOCK_DEVICE_IO_STATUS_OK) {
-        job->iostatus = error == ENOSPC ? BLOCK_DEVICE_IO_STATUS_NOSPACE :
-                                          BLOCK_DEVICE_IO_STATUS_FAILED;
-    }
-}
-
-static void block_job_event_cancelled(BlockJob *job)
-{
-    if (block_job_is_internal(job)) {
         return;
     }
-
+    assert(job->user_paused && job->pause_count > 0);
-    qapi_event_send_block_job_cancelled(job->driver->job_type,
+    job->iostatus = BLOCK_DEVICE_IO_STATUS_OK;
-                                        job->id,
-                                        job->len,
-                                        job->offset,
-                                        job->speed,
-                                        &error_abort);
-}
-
-static void block_job_event_completed(BlockJob *job, const char *msg)
-{
-    if (block_job_is_internal(job)) {
-        return;
-    }
-
-    qapi_event_send_block_job_completed(job->driver->job_type,
-                                        job->id,
-                                        job->len,
-                                        job->offset,
-                                        job->speed,
-                                        !!msg,
-                                        msg,
-                                        &error_abort);
 }
 
 void block_job_event_ready(BlockJob *job)
@@ -790,7 +906,6 @@ static void block_job_defer_to_main_loop_bh(void *opaque)
         aio_context_acquire(aio_context);
     }
 
-    data->job->deferred_to_main_loop = false;
     data->fn(data->job, data->opaque);
 
     if (aio_context != data->aio_context) {
@@ -816,36 +931,3 @@ void block_job_defer_to_main_loop(BlockJob *job,
     aio_bh_schedule_oneshot(qemu_get_aio_context(),
                             block_job_defer_to_main_loop_bh, data);
 }
-
-BlockJobTxn *block_job_txn_new(void)
-{
-    BlockJobTxn *txn = g_new0(BlockJobTxn, 1);
-    QLIST_INIT(&txn->jobs);
-    txn->refcnt = 1;
-    return txn;
-}
-
-static void block_job_txn_ref(BlockJobTxn *txn)
-{
-    txn->refcnt++;
-}
-
-void block_job_txn_unref(BlockJobTxn *txn)
-{
-    if (txn && --txn->refcnt == 0) {
-        g_free(txn);
-    }
-}
-
-void block_job_txn_add_job(BlockJobTxn *txn, BlockJob *job)
-{
-    if (!txn) {
-        return;
-    }
-
-    assert(!job->txn);
-    job->txn = txn;
-
-    QLIST_INSERT_HEAD(&txn->jobs, job, txn_list);
-    block_job_txn_ref(txn);
-}

include/block/blockjob.h

@@ -234,14 +234,6 @@ void block_job_complete(BlockJob *job, Error **errp);
  */
 BlockJobInfo *block_job_query(BlockJob *job, Error **errp);
 
-/**
- * block_job_pause:
- * @job: The job to be paused.
- *
- * Asynchronously pause the specified job.
- */
-void block_job_pause(BlockJob *job);
-
 /**
  * block_job_user_pause:
  * @job: The job to be paused.
@@ -259,14 +251,6 @@ void block_job_user_pause(BlockJob *job);
  */
 bool block_job_user_paused(BlockJob *job);
 
-/**
- * block_job_resume:
- * @job: The job to be resumed.
- *
- * Resume the specified job.  Must be paired with a preceding block_job_pause.
- */
-void block_job_resume(BlockJob *job);
-
 /**
  * block_job_user_resume:
  * @job: The job to be resumed.

include/block/blockjob_int.h

@@ -44,9 +44,6 @@ struct BlockJobDriver {
     /** Optional callback for job types that support setting a speed limit */
     void (*set_speed)(BlockJob *job, int64_t speed, Error **errp);
 
-    /** Optional callback for job types that need to forward I/O status reset */
-    void (*iostatus_reset)(BlockJob *job);
-
     /** Mandatory: Entrypoint for the Coroutine. */
     CoroutineEntry *start;
 
@@ -159,21 +156,26 @@ void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns);
 void block_job_yield(BlockJob *job);
 
 /**
- * block_job_ref:
+ * block_job_pause_all:
- * @bs: The block device.
  *
- * Grab a reference to the block job. Should be paired with block_job_unref.
+ * Asynchronously pause all jobs.
  */
-void block_job_ref(BlockJob *job);
+void block_job_pause_all(void);
 
 /**
- * block_job_unref:
+ * block_job_resume_all:
+ *
+ * Resume all block jobs.  Must be paired with a preceding block_job_pause_all.
+ */
+void block_job_resume_all(void);
+
+/**
+ * block_job_early_fail:
  * @bs: The block device.
  *
- * Release reference to the block job and release resources if it is the last
+ * The block job could not be started, free it.
- * reference.
  */
-void block_job_unref(BlockJob *job);
+void block_job_early_fail(BlockJob *job);
 
 /**
  * block_job_completed:
@@ -239,7 +241,8 @@ typedef void BlockJobDeferToMainLoopFn(BlockJob *job, void *opaque);
  * @fn: The function to run in the main loop
  * @opaque: The opaque value that is passed to @fn
  *
- * Execute a given function in the main loop with the BlockDriverState
+ * This function must be called by the main job coroutine just before it
+ * returns.  @fn is executed in the main loop with the BlockDriverState
  * AioContext acquired.  Block jobs must call bdrv_unref(), bdrv_close(), and
  * anything that uses bdrv_drain_all() in the main loop.
  *

tests/test-blockjob-txn.c

@@ -167,6 +167,11 @@ static void test_pair_jobs(int expected1, int expected2)
     block_job_start(job1);
     block_job_start(job2);
 
+    /* Release our reference now to trigger as many nice
+     * use-after-free bugs as possible.
+     */
+    block_job_txn_unref(txn);
+
     if (expected1 == -ECANCELED) {
         block_job_cancel(job1);
     }
@@ -187,8 +192,6 @@ static void test_pair_jobs(int expected1, int expected2)
 
     g_assert_cmpint(result1, ==, expected1);
     g_assert_cmpint(result2, ==, expected2);
-
-    block_job_txn_unref(txn);
 }
 
 static void test_pair_jobs_success(void)

tests/test-blockjob.c

@@ -116,11 +116,11 @@ static void test_job_ids(void)
     job[1] = do_test_id(blk[1], "id0", false);
 
     /* But once job[0] finishes we can reuse its ID */
-    block_job_unref(job[0]);
+    block_job_early_fail(job[0]);
     job[1] = do_test_id(blk[1], "id0", true);
 
     /* No job ID specified, defaults to the backend name ('drive1') */
-    block_job_unref(job[1]);
+    block_job_early_fail(job[1]);
     job[1] = do_test_id(blk[1], NULL, true);
 
     /* Duplicate job ID */
@@ -133,9 +133,9 @@ static void test_job_ids(void)
     /* This one is valid */
     job[2] = do_test_id(blk[2], "id_2", true);
 
-    block_job_unref(job[0]);
+    block_job_early_fail(job[0]);
-    block_job_unref(job[1]);
+    block_job_early_fail(job[1]);
-    block_job_unref(job[2]);
+    block_job_early_fail(job[2]);
 
     destroy_blk(blk[0]);
     destroy_blk(blk[1]);