[Qemu-block] [PATCH 03/11] block: add transactional callbacks feature

[John Snow]

The goal here is to add a new method to transactions that allows
developers to specify a callback that will get invoked only once
all jobs spawned by a transaction are completed, allowing developers
the chance to perform actions conditionally pending complete success
or complete failure.

In order to register the new callback to be invoked, a user must request
a callback pointer and closure by calling new_transaction_wrapper, which
creates a wrapper around a closure and callback that would originally have
been passed to e.g. backup_start().

The function will return a function pointer and a new closure to be passed
instead. The transaction system will effectively intercept these callbacks
and execute the desired actions upon reception of all intercepted callbacks.

This means that the registered callbacks will be called after all other
transaction actions that requested a callback have completed. The feature
has no knowledge of jobs spawned without informing the BlkTransactionList.

For an example of how to use the feature, please skip ahead to:
'block: drive_backup transaction callback support' which serves as an example
for how to hook in drive_backup (or any block job launched by transactions).


Note 1: Defining a callback method alone is not sufficient to have the new
        method invoked. You must call new_transaction_wrapper AND ensure the
        callback it returns to you is used as the callback for the job.

Note 2: You can use this feature for any system that registers completions of
        an action via a callback of the form (void *opaque, int ret), not just
        block job callbacks.

Note 3: new_blk_transaction has no users in this patch, but will in
        the next patch where it will become static and local to blockdev.c.

Signed-off-by: John Snow <address@hidden>
---
 blockdev.c | 225 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 223 insertions(+), 2 deletions(-)

diff --git a/blockdev.c b/blockdev.c
index 5120af1..3153ee7 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -1207,6 +1207,8 @@ static BdrvDirtyBitmap *block_dirty_bitmap_lookup(const 
char *node,
 /* New and old BlockDriverState structs for atomic group operations */
 
 typedef struct BlkTransactionState BlkTransactionState;
+typedef struct BlkTransactionList BlkTransactionList;
+typedef struct BlkTransactionData BlkTransactionData;
 
 /* Only prepare() may fail. In a single transaction, only one of commit() or
    abort() will be called, clean() will always be called if it present. */
@@ -1221,6 +1223,8 @@ typedef struct BdrvActionOps {
     void (*abort)(BlkTransactionState *common);
     /* Clean up resource in the end, can be NULL. */
     void (*clean)(BlkTransactionState *common);
+    /* Execute this after +all+ jobs in the transaction finish */
+    void (*cb)(BlkTransactionState *common);
 } BdrvActionOps;
 
 /*
@@ -1231,9 +1235,220 @@ typedef struct BdrvActionOps {
 struct BlkTransactionState {
     TransactionAction *action;
     const BdrvActionOps *ops;
+    BlkTransactionList *list;
+    void *opaque;
+    /* Allow external users (callbacks) to reference this obj past .clean() */
+    int refcount;
+    /* All transactions in the current group */
     QSIMPLEQ_ENTRY(BlkTransactionState) entry;
+    /* Transactions in the current group with callbacks */
+    QSIMPLEQ_ENTRY(BlkTransactionState) list_entry;
 };
 
+struct BlkTransactionList {
+    int jobs;     /* Effectively: A refcount */
+    int status;   /* Cumulative retcode */
+    QSIMPLEQ_HEAD(actions, BlkTransactionState) actions;
+};
+
+typedef struct BlkTransactionData {
+    void *opaque; /* Data given to encapsulated callback */
+    int ret;      /* Return code given to encapsulated callback */
+} BlkTransactionData;
+
+typedef struct BlkTransactionWrapper {
+    void *opaque; /* Data to be given to encapsulated callback */
+    void (*callback)(void *opaque, int ret); /* Encapsulated callback */
+} BlkTransactionWrapper;
+
+static BlkTransactionList *new_blk_transaction_list(void)
+{
+    BlkTransactionList *btl = g_malloc0(sizeof(*btl));
+
+    /* Implicit 'job' for qmp_transaction itself */
+    btl->jobs = 1;
+    QSIMPLEQ_INIT(&btl->actions);
+    return btl;
+}
+
+static BlkTransactionState *blk_put_transaction_state(BlkTransactionState *bts)
+{
+    bts->refcount--;
+    if (bts->refcount == 0) {
+        g_free(bts);
+        return NULL;
+    }
+    return bts;
+}
+
+static void del_blk_transaction_list(BlkTransactionList *btl)
+{
+    BlkTransactionState *bts, *bts_next;
+
+    /* The list should in normal cases be empty,
+     * but in case someone really just wants to kibosh the whole deal: */
+    QSIMPLEQ_FOREACH_SAFE(bts, &btl->actions, list_entry, bts_next) {
+        g_free(bts->opaque);
+        blk_put_transaction_state(bts);
+    }
+
+    g_free(btl);
+}
+
+static void blk_run_transaction_callbacks(BlkTransactionList *btl)
+{
+    BlkTransactionState *bts, *bts_next;
+
+    QSIMPLEQ_FOREACH_SAFE(bts, &btl->actions, list_entry, bts_next) {
+        if (bts->ops->cb) {
+            bts->ops->cb(bts);
+        }
+
+        /* Free the BlkTransactionData */
+        g_free(bts->opaque);
+        bts->opaque = NULL;
+        blk_put_transaction_state(bts);
+    }
+
+    QSIMPLEQ_INIT(&btl->actions);
+}
+
+static BlkTransactionList *put_blk_transaction_list(BlkTransactionList *btl)
+{
+    btl->jobs--;
+    if (btl->jobs == 0) {
+        blk_run_transaction_callbacks(btl);
+        del_blk_transaction_list(btl);
+        return NULL;
+    }
+    return btl;
+}
+
+static void blk_transaction_complete(BlkTransactionState *common)
+{
+    BlkTransactionList *btl = common->list;
+
+    /* Add this action into the pile to be completed */
+    QSIMPLEQ_INSERT_TAIL(&btl->actions, common, list_entry);
+
+    /* Inform the list that we have a completion;
+     * possibly run all the pending actions. */
+    put_blk_transaction_list(btl);
+}
+
+/**
+ * Intercept a callback that was issued due to a transactional action.
+ */
+static void transaction_callback(void *opaque, int ret)
+{
+    BlkTransactionState *common = opaque;
+    BlkTransactionWrapper *btw = common->opaque;
+
+    /* Prepare data for ops->cb() */
+    BlkTransactionData *btd = g_malloc0(sizeof(*btd));
+    btd->opaque = btw->opaque;
+    btd->ret = ret;
+
+    /* Transaction state now tracks OUR data */
+    common->opaque = btd;
+
+    /* Keep track of the amalgamated return code */
+    common->list->status |= ret;
+
+    /* Deliver the intercepted callback FIRST */
+    btw->callback(btw->opaque, ret);
+    blk_transaction_complete(common);
+    g_free(btw);
+}
+
+typedef void (CallbackFn)(void *opaque, int ret);
+
+/* Temporary. Removed in the next patch. */
+CallbackFn *new_transaction_wrapper(BlkTransactionState *common,
+                                    void *opaque,
+                                    void (*callback)(void *, int),
+                                    void **new_opaque);
+
+void undo_transaction_wrapper(BlkTransactionState *common);
+
+/**
+ * Create a new transactional callback wrapper.
+ *
+ * Given a callback and a closure, generate a new
+ * callback and closure that will invoke the
+ * given callback with the given closure.
+ *
+ * After all wrappers in the transactional group have
+ * been processed, each action's .cb() method will be
+ * invoked.
+ *
+ * @common The transactional state to set a callback for.
+ * @opaque A closure intended for the encapsulated callback.
+ * @callback The callback we are encapsulating.
+ * @new_opaque The closure to be used instead of @opaque.
+ *
+ * @return The callback to be used instead of @callback.
+ */
+CallbackFn *new_transaction_wrapper(BlkTransactionState *common,
+                                           void *opaque,
+                                           CallbackFn *callback,
+                                           void **new_opaque)
+{
+    BlkTransactionWrapper *btw = g_malloc0(sizeof(*btw));
+    assert(new_opaque);
+
+    /* Stash the original callback information */
+    btw->opaque = opaque;
+    btw->callback = callback;
+    common->opaque = btw;
+
+    /* The BTS will serve as our new closure */
+    *new_opaque = common;
+    common->refcount++;
+
+    /* Inform the transaction BTL to expect one more return */
+    common->list->jobs++;
+
+    /* Lastly, the actual callback function to handle the interception. */
+    return transaction_callback;
+}
+
+/**
+ * Undo any actions performed by the above call.
+ */
+void undo_transaction_wrapper(BlkTransactionState *common)
+{
+    BlkTransactionList *btl = common->list;
+    BlkTransactionState *bts;
+    BlkTransactionData *btd;
+
+    /* Stage 0: Wrapper was never created: */
+    if (common->opaque == NULL && common->refcount == 1) {
+        return;
+    }
+
+    /* Stage 2: Job already completed or was canceled.
+     * Force an error in the callback data and just invoke the completion
+     * handler to perform appropriate cleanup for us.
+     */
+    QSIMPLEQ_FOREACH(bts, &btl->actions, list_entry) {
+        if (bts == common) {
+            btd = common->opaque;
+            /* Force error for callback */
+            btd->ret = -1;
+            common->ops->cb(common);
+            QSIMPLEQ_REMOVE(&btl->actions, common,
+                            BlkTransactionState, list_entry);
+            goto cleanup;
+        }
+    }
+    /* Stage 1: Callback created, but job never launched */
+    put_blk_transaction_list(common->list);
+ cleanup:
+    g_free(common->opaque);
+    blk_put_transaction_state(common);
+}
+
 /* internal snapshot private data */
 typedef struct InternalSnapshotState {
     BlkTransactionState common;
@@ -1775,7 +1990,7 @@ static const BdrvActionOps actions[] = {
         .instance_size = sizeof(DriveBackupState),
         .prepare = drive_backup_prepare,
         .abort = drive_backup_abort,
-        .clean = drive_backup_clean,
+        .clean = drive_backup_clean
     },
     [TRANSACTION_ACTION_KIND_BLOCKDEV_BACKUP] = {
         .instance_size = sizeof(BlockdevBackupState),
@@ -1815,10 +2030,12 @@ void qmp_transaction(TransactionActionList *dev_list, 
Error **errp)
 {
     TransactionActionList *dev_entry = dev_list;
     BlkTransactionState *state, *next;
+    BlkTransactionList *btl;
     Error *local_err = NULL;
 
     QSIMPLEQ_HEAD(snap_bdrv_states, BlkTransactionState) snap_bdrv_states;
     QSIMPLEQ_INIT(&snap_bdrv_states);
+    btl = new_blk_transaction_list();
 
     /* drain all i/o before any operations */
     bdrv_drain_all();
@@ -1837,8 +2054,10 @@ void qmp_transaction(TransactionActionList *dev_list, 
Error **errp)
         assert(ops->instance_size > 0);
 
         state = g_malloc0(ops->instance_size);
+        state->refcount = 1;
         state->ops = ops;
         state->action = dev_info;
+        state->list = btl;
         QSIMPLEQ_INSERT_TAIL(&snap_bdrv_states, state, entry);
 
         state->ops->prepare(state, &local_err);
@@ -1869,8 +2088,10 @@ exit:
         if (state->ops->clean) {
             state->ops->clean(state);
         }
-        g_free(state);
+        blk_put_transaction_state(state);
     }
+
+    put_blk_transaction_list(btl);
 }
 
 
-- 
1.9.3