[PATCH v2 0/8] Block layer thread-safety, continued

[PATCH v2 0/8] Block layer thread-safety, continued

[Emanuele Giuseppe Esposito]

This and the following serie of patches are based on Paolo's
v1 patches sent in 2017[*]. They have been ported to the current QEMU
version, but the goal remains the same: 
- make the block layer thread-safe (patches 1-5), and
- remove aio_context_acquire/release (patches 6-8).

[*] = https://lists.gnu.org/archive/html/qemu-devel/2017-07/msg01398.html

Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
v1 (2017) -> v2 (2021):
- v1 Patch "block-backup: add reqs_lock" has been dropped, because now
  is completely different from the old version and all functions
  that were affected by it have been moved or deleted. 
  It will be replaced by another serie that aims to thread safety to 
  block/block-copy.c
- remaining v1 patches will be integrated in next serie.
- Patch "block: do not acquire AioContext in check_to_replace_node"
  moves part of the logic of check_to_replace_node to the caller,
  so that the function can be included in the aio_context_acquire/release
  block that follows.

Emanuele Giuseppe Esposito (8):
  block: prepare write threshold code for thread safety
  block: protect write threshold QMP commands from concurrent requests
  util: use RCU accessors for notifiers
  block: make before-write notifiers thread-safe
  block: add a few more notes on locking
  block: do not acquire AioContext in check_to_replace_node
  block/replication: do not acquire AioContext
  block: do not take AioContext around reopen

 block.c                   | 28 ++++++--------------
 block/block-backend.c     |  4 ---
 block/io.c                | 12 +++++++++
 block/mirror.c            |  9 -------
 block/replication.c       | 54 +++++++++------------------------------
 block/write-threshold.c   | 39 ++++++++++++++--------------
 blockdev.c                | 26 +++++++++----------
 include/block/block.h     |  1 +
 include/block/block_int.h | 42 +++++++++++++++++++++++++++++-
 util/notify.c             | 13 +++++-----
 10 files changed, 113 insertions(+), 115 deletions(-)

-- 
2.30.2

[PATCH v2 1/8] block: prepare write threshold code for thread safety

[Emanuele Giuseppe Esposito]

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Co-developed-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 block/write-threshold.c | 28 ++++++++++++++++------------
 1 file changed, 16 insertions(+), 12 deletions(-)

diff --git a/block/write-threshold.c b/block/write-threshold.c
index 85b78dc2a9..63040fa4f2 100644
--- a/block/write-threshold.c
+++ b/block/write-threshold.c
@@ -37,18 +37,22 @@ static void write_threshold_disable(BlockDriverState *bs)
     }
 }
 
+static uint64_t treshold_overage(const BdrvTrackedRequest *req,
+                                uint64_t thres)
+{
+    if (thres > 0 && req->offset + req->bytes > thres) {
+        return req->offset + req->bytes - thres;
+    } else {
+        return 0;
+    }
+}
+
 uint64_t bdrv_write_threshold_exceeded(const BlockDriverState *bs,
                                        const BdrvTrackedRequest *req)
 {
-    if (bdrv_write_threshold_is_set(bs)) {
-        if (req->offset > bs->write_threshold_offset) {
-            return (req->offset - bs->write_threshold_offset) + req->bytes;
-        }
-        if ((req->offset + req->bytes) > bs->write_threshold_offset) {
-            return (req->offset + req->bytes) - bs->write_threshold_offset;
-        }
-    }
-    return 0;
+    uint64_t thres = bdrv_write_threshold_get(bs);
+
+    return treshold_overage(req, thres);
 }
 
 static int coroutine_fn before_write_notify(NotifierWithReturn *notifier,
@@ -56,14 +60,14 @@ static int coroutine_fn before_write_notify(NotifierWithReturn *notifier,
 {
     BdrvTrackedRequest *req = opaque;
     BlockDriverState *bs = req->bs;
-    uint64_t amount = 0;
+    uint64_t thres = bdrv_write_threshold_get(bs);
+    uint64_t amount = treshold_overage(req, thres);
 
-    amount = bdrv_write_threshold_exceeded(bs, req);
     if (amount > 0) {
         qapi_event_send_block_write_threshold(
             bs->node_name,
             amount,
-            bs->write_threshold_offset);
+            thres);
 
         /* autodisable to avoid flooding the monitor */
         write_threshold_disable(bs);
-- 
2.30.2

[PATCH v2 2/8] block: protect write threshold QMP commands from concurrent requests

[Emanuele Giuseppe Esposito]

For simplicity, use bdrv_drained_begin/end to avoid concurrent
writes to the write threshold, or reading it while it is being set.
qmp_block_set_write_threshold is protected by the BQL.

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Co-developed-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 block/write-threshold.c | 9 +++------
 1 file changed, 3 insertions(+), 6 deletions(-)

diff --git a/block/write-threshold.c b/block/write-threshold.c
index 63040fa4f2..77c74bdaa7 100644
--- a/block/write-threshold.c
+++ b/block/write-threshold.c
@@ -111,7 +111,6 @@ void qmp_block_set_write_threshold(const char *node_name,
                                    Error **errp)
 {
     BlockDriverState *bs;
-    AioContext *aio_context;
 
     bs = bdrv_find_node(node_name);
     if (!bs) {
@@ -119,10 +118,8 @@ void qmp_block_set_write_threshold(const char *node_name,
         return;
     }
 
-    aio_context = bdrv_get_aio_context(bs);
-    aio_context_acquire(aio_context);
-
+    /* Avoid a concurrent write_threshold_disable.  */
+    bdrv_drained_begin(bs);
     bdrv_write_threshold_set(bs, threshold_bytes);
-
-    aio_context_release(aio_context);
+    bdrv_drained_end(bs);
 }
-- 
2.30.2

[PATCH v2 3/8] util: use RCU accessors for notifiers

[Emanuele Giuseppe Esposito]

Note that calling rcu_read_lock() is left to the caller.  In fact,
if the notifier is really only used within the BQL, it's unnecessary.

Even outside the BQL, RCU accessors can also be used with any API that has
the same contract as synchronize_rcu, i.e. it stops until all concurrent
readers complete, no matter how "readers" are defined.  In the next patch,
for example, synchronize_rcu's role is taken by bdrv_drain (which is a
superset of synchronize_rcu, since it also blocks new incoming readers).

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 util/notify.c | 13 +++++++------
 1 file changed, 7 insertions(+), 6 deletions(-)

diff --git a/util/notify.c b/util/notify.c
index 76bab212ae..529f1d121e 100644
--- a/util/notify.c
+++ b/util/notify.c
@@ -15,6 +15,7 @@
 
 #include "qemu/osdep.h"
 #include "qemu/notify.h"
+#include "qemu/rcu_queue.h"
 
 void notifier_list_init(NotifierList *list)
 {
@@ -23,19 +24,19 @@ void notifier_list_init(NotifierList *list)
 
 void notifier_list_add(NotifierList *list, Notifier *notifier)
 {
-    QLIST_INSERT_HEAD(&list->notifiers, notifier, node);
+    QLIST_INSERT_HEAD_RCU(&list->notifiers, notifier, node);
 }
 
 void notifier_remove(Notifier *notifier)
 {
-    QLIST_REMOVE(notifier, node);
+    QLIST_REMOVE_RCU(notifier, node);
 }
 
 void notifier_list_notify(NotifierList *list, void *data)
 {
     Notifier *notifier, *next;
 
-    QLIST_FOREACH_SAFE(notifier, &list->notifiers, node, next) {
+    QLIST_FOREACH_SAFE_RCU(notifier, &list->notifiers, node, next) {
         notifier->notify(notifier, data);
     }
 }
@@ -53,12 +54,12 @@ void notifier_with_return_list_init(NotifierWithReturnList *list)
 void notifier_with_return_list_add(NotifierWithReturnList *list,
                                    NotifierWithReturn *notifier)
 {
-    QLIST_INSERT_HEAD(&list->notifiers, notifier, node);
+    QLIST_INSERT_HEAD_RCU(&list->notifiers, notifier, node);
 }
 
 void notifier_with_return_remove(NotifierWithReturn *notifier)
 {
-    QLIST_REMOVE(notifier, node);
+    QLIST_REMOVE_RCU(notifier, node);
 }
 
 int notifier_with_return_list_notify(NotifierWithReturnList *list, void *data)
@@ -66,7 +67,7 @@ int notifier_with_return_list_notify(NotifierWithReturnList *list, void *data)
     NotifierWithReturn *notifier, *next;
     int ret = 0;
 
-    QLIST_FOREACH_SAFE(notifier, &list->notifiers, node, next) {
+    QLIST_FOREACH_SAFE_RCU(notifier, &list->notifiers, node, next) {
         ret = notifier->notify(notifier, data);
         if (ret != 0) {
             break;
-- 
2.30.2

[PATCH v2 4/8] block: make before-write notifiers thread-safe

[Emanuele Giuseppe Esposito]

Reads access the list in RCU style, so be careful to avoid use-after-free
scenarios in the backup block job.  Apart from this, all that's needed
is protecting updates with a mutex.

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 block.c                   |  6 +++---
 block/io.c                | 12 ++++++++++++
 block/write-threshold.c   |  2 +-
 include/block/block_int.h | 37 +++++++++++++++++++++++++++++++++++++
 4 files changed, 53 insertions(+), 4 deletions(-)

diff --git a/block.c b/block.c
index c5b887cec1..f40fb63c75 100644
--- a/block.c
+++ b/block.c
@@ -6434,7 +6434,7 @@ static void bdrv_do_remove_aio_context_notifier(BdrvAioNotifier *ban)
     g_free(ban);
 }
 
-static void bdrv_detach_aio_context(BlockDriverState *bs)
+void bdrv_detach_aio_context(BlockDriverState *bs)
 {
     BdrvAioNotifier *baf, *baf_tmp;
 
@@ -6462,8 +6462,8 @@ static void bdrv_detach_aio_context(BlockDriverState *bs)
     bs->aio_context = NULL;
 }
 
-static void bdrv_attach_aio_context(BlockDriverState *bs,
-                                    AioContext *new_context)
+void bdrv_attach_aio_context(BlockDriverState *bs,
+                             AioContext *new_context)
 {
     BdrvAioNotifier *ban, *ban_tmp;
 
diff --git a/block/io.c b/block/io.c
index ca2dca3007..8f6af6077b 100644
--- a/block/io.c
+++ b/block/io.c
@@ -3137,10 +3137,22 @@ bool bdrv_qiov_is_aligned(BlockDriverState *bs, QEMUIOVector *qiov)
     return true;
 }
 
+static QemuSpin notifiers_spin_lock;
+
 void bdrv_add_before_write_notifier(BlockDriverState *bs,
                                     NotifierWithReturn *notifier)
 {
+    qemu_spin_lock(&notifiers_spin_lock);
     notifier_with_return_list_add(&bs->before_write_notifiers, notifier);
+    qemu_spin_unlock(&notifiers_spin_lock);
+}
+
+void bdrv_remove_before_write_notifier(BlockDriverState *bs,
+                                       NotifierWithReturn *notifier)
+{
+    qemu_spin_lock(&notifiers_spin_lock);
+    notifier_with_return_remove(notifier);
+    qemu_spin_unlock(&notifiers_spin_lock);
 }
 
 void bdrv_io_plug(BlockDriverState *bs)
diff --git a/block/write-threshold.c b/block/write-threshold.c
index 77c74bdaa7..b87b749b02 100644
--- a/block/write-threshold.c
+++ b/block/write-threshold.c
@@ -32,7 +32,7 @@ bool bdrv_write_threshold_is_set(const BlockDriverState *bs)
 static void write_threshold_disable(BlockDriverState *bs)
 {
     if (bdrv_write_threshold_is_set(bs)) {
-        notifier_with_return_remove(&bs->write_threshold_notifier);
+        bdrv_remove_before_write_notifier(bs, &bs->write_threshold_notifier);
         bs->write_threshold_offset = 0;
     }
 }
diff --git a/include/block/block_int.h b/include/block/block_int.h
index 88e4111939..a1aad5ad2d 100644
--- a/include/block/block_int.h
+++ b/include/block/block_int.h
@@ -1089,6 +1089,8 @@ bool bdrv_backing_overridden(BlockDriverState *bs);
 
 /**
  * bdrv_add_before_write_notifier:
+ * @bs: The #BlockDriverState for which to register the notifier
+ * @notifier: The notifier to add
  *
  * Register a callback that is invoked before write requests are processed but
  * after any throttling or waiting for overlapping requests.
@@ -1096,6 +1098,41 @@ bool bdrv_backing_overridden(BlockDriverState *bs);
 void bdrv_add_before_write_notifier(BlockDriverState *bs,
                                     NotifierWithReturn *notifier);
 
+/**
+ * bdrv_remove_before_write_notifier:
+ * @bs: The #BlockDriverState for which to register the notifier
+ * @notifier: The notifier to add
+ *
+ * Unregister a callback that is invoked before write requests are processed but
+ * after any throttling or waiting for overlapping requests.
+ *
+ * Note that more I/O could be pending on @bs.  You need to wait for
+ * it to finish, for example with bdrv_drain(), before freeing @notifier.
+ */
+void bdrv_remove_before_write_notifier(BlockDriverState *bs,
+                                       NotifierWithReturn *notifier);
+
+/**
+ * bdrv_detach_aio_context:
+ *
+ * May be called from .bdrv_detach_aio_context() to detach children from the
+ * current #AioContext.  This is only needed by block drivers that manage their
+ * own children.  Both ->file and ->backing are automatically handled and
+ * block drivers should not call this function on them explicitly.
+ */
+void bdrv_detach_aio_context(BlockDriverState *bs);
+
+/**
+ * bdrv_attach_aio_context:
+ *
+ * May be called from .bdrv_attach_aio_context() to attach children to the new
+ * #AioContext.  This is only needed by block drivers that manage their own
+ * children.  Both ->file and ->backing are automatically handled and block
+ * drivers should not call this function on them explicitly.
+ */
+void bdrv_attach_aio_context(BlockDriverState *bs,
+                             AioContext *new_context);
+
 /**
  * bdrv_add_aio_context_notifier:
  *
-- 
2.30.2

[PATCH v2 5/8] block: add a few more notes on locking

[Emanuele Giuseppe Esposito]

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 include/block/block_int.h | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/include/block/block_int.h b/include/block/block_int.h
index a1aad5ad2d..67a0777e12 100644
--- a/include/block/block_int.h
+++ b/include/block/block_int.h
@@ -154,7 +154,9 @@ struct BlockDriver {
      */
     bool supports_backing;
 
-    /* For handling image reopen for split or non-split files */
+    /* For handling image reopen for split or non-split files.  Called
+     * with no I/O pending.
+     */
     int (*bdrv_reopen_prepare)(BDRVReopenState *reopen_state,
                                BlockReopenQueue *queue, Error **errp);
     void (*bdrv_reopen_commit)(BDRVReopenState *reopen_state);
@@ -168,6 +170,7 @@ struct BlockDriver {
     /* Protocol drivers should implement this instead of bdrv_open */
     int (*bdrv_file_open)(BlockDriverState *bs, QDict *options, int flags,
                           Error **errp);
+    /* Called from main thread.  */
     void (*bdrv_close)(BlockDriverState *bs);
 
 
-- 
2.30.2

[PATCH v2 6/8] block: do not acquire AioContext in check_to_replace_node

[Emanuele Giuseppe Esposito]

Make the (only) caller do it.

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 block.c               | 22 +++++-----------------
 blockdev.c            |  7 ++++++-
 include/block/block.h |  1 +
 3 files changed, 12 insertions(+), 18 deletions(-)

diff --git a/block.c b/block.c
index f40fb63c75..fd12f88062 100644
--- a/block.c
+++ b/block.c
@@ -6776,24 +6776,15 @@ bool bdrv_recurse_can_replace(BlockDriverState *bs,
  *
  * The result (whether the node can be replaced or not) is only valid
  * for as long as no graph or permission changes occur.
+ *
+ * Called with AioContext lock held.
  */
 BlockDriverState *check_to_replace_node(BlockDriverState *parent_bs,
+                                        BlockDriverState *to_replace_bs,
                                         const char *node_name, Error **errp)
 {
-    BlockDriverState *to_replace_bs = bdrv_find_node(node_name);
-    AioContext *aio_context;
-
-    if (!to_replace_bs) {
-        error_setg(errp, "Failed to find node with node-name='%s'", node_name);
-        return NULL;
-    }
-
-    aio_context = bdrv_get_aio_context(to_replace_bs);
-    aio_context_acquire(aio_context);
-
     if (bdrv_op_is_blocked(to_replace_bs, BLOCK_OP_TYPE_REPLACE, errp)) {
-        to_replace_bs = NULL;
-        goto out;
+        return NULL;
     }
 
     /* We don't want arbitrary node of the BDS chain to be replaced only the top
@@ -6806,12 +6797,9 @@ BlockDriverState *check_to_replace_node(BlockDriverState *parent_bs,
                    "because it cannot be guaranteed that doing so would not "
                    "lead to an abrupt change of visible data",
                    node_name, parent_bs->node_name);
-        to_replace_bs = NULL;
-        goto out;
+        return NULL;
     }
 
-out:
-    aio_context_release(aio_context);
     return to_replace_bs;
 }
 
diff --git a/blockdev.c b/blockdev.c
index a57590aae4..e901107344 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -3056,13 +3056,18 @@ static void blockdev_mirror_common(const char *job_id, BlockDriverState *bs,
             return;
         }
 
-        to_replace_bs = check_to_replace_node(bs, replaces, errp);
+        to_replace_bs = bdrv_find_node(replaces);
         if (!to_replace_bs) {
+            error_setg(errp, "Failed to find node with node-name='%s'",
+                       replaces);
             return;
         }
 
         replace_aio_context = bdrv_get_aio_context(to_replace_bs);
         aio_context_acquire(replace_aio_context);
+        if (!check_to_replace_node(bs, to_replace_bs, replaces, errp)) {
+            return;
+        }
         replace_size = bdrv_getlength(to_replace_bs);
         aio_context_release(replace_aio_context);
 
diff --git a/include/block/block.h b/include/block/block.h
index b3f6e509d4..f57c34a19a 100644
--- a/include/block/block.h
+++ b/include/block/block.h
@@ -474,6 +474,7 @@ int bdrv_amend_options(BlockDriverState *bs_new, QemuOpts *opts,
 
 /* check if a named node can be replaced when doing drive-mirror */
 BlockDriverState *check_to_replace_node(BlockDriverState *parent_bs,
+                                        BlockDriverState *to_replace_bs,
                                         const char *node_name, Error **errp);
 
 /* async block I/O */
-- 
2.30.2

[PATCH v2 7/8] block/replication: do not acquire AioContext

[Emanuele Giuseppe Esposito]

Replication functions are mostly called when the BDS is quiescent and
does not have any pending I/O.  They do not need to synchronize on
anything since BDS and BB are now thread-safe.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 block/replication.c | 54 ++++++++++-----------------------------------
 1 file changed, 12 insertions(+), 42 deletions(-)

diff --git a/block/replication.c b/block/replication.c
index 97be7ef4de..25ee37b21b 100644
--- a/block/replication.c
+++ b/block/replication.c
@@ -45,6 +45,8 @@ typedef struct BDRVReplicationState {
     Error *blocker;
     bool orig_hidden_read_only;
     bool orig_secondary_read_only;
+
+    /* This field is accessed asynchronously.  */
     int error;
 } BDRVReplicationState;
 
@@ -210,7 +212,7 @@ static int replication_return_value(BDRVReplicationState *s, int ret)
     }
 
     if (ret < 0) {
-        s->error = ret;
+        qatomic_set(&s->error, ret);
         ret = 0;
     }
 
@@ -307,6 +309,7 @@ out:
     return ret;
 }
 
+/* Called with no I/O pending.  */
 static void secondary_do_checkpoint(BDRVReplicationState *s, Error **errp)
 {
     Error *local_err = NULL;
@@ -420,7 +423,7 @@ static void backup_job_completed(void *opaque, int ret)
 
     if (s->stage != BLOCK_REPLICATION_FAILOVER) {
         /* The backup job is cancelled unexpectedly */
-        s->error = -EIO;
+        qatomic_set(&s->error, -EIO);
     }
 
     backup_job_cleanup(bs);
@@ -445,6 +448,7 @@ static bool check_top_bs(BlockDriverState *top_bs, BlockDriverState *bs)
     return false;
 }
 
+/* Called with no I/O pending.  */
 static void replication_start(ReplicationState *rs, ReplicationMode mode,
                               Error **errp)
 {
@@ -452,12 +456,9 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
     BDRVReplicationState *s;
     BlockDriverState *top_bs;
     int64_t active_length, hidden_length, disk_length;
-    AioContext *aio_context;
     Error *local_err = NULL;
     BackupPerf perf = { .use_copy_range = true, .max_workers = 1 };
 
-    aio_context = bdrv_get_aio_context(bs);
-    aio_context_acquire(aio_context);
     s = bs->opaque;
 
     if (s->stage == BLOCK_REPLICATION_DONE ||
@@ -467,20 +468,17 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
          * Ignore the request because the secondary side of replication
          * doesn't have to do anything anymore.
          */
-        aio_context_release(aio_context);
         return;
     }
 
     if (s->stage != BLOCK_REPLICATION_NONE) {
         error_setg(errp, "Block replication is running or done");
-        aio_context_release(aio_context);
         return;
     }
 
     if (s->mode != mode) {
         error_setg(errp, "The parameter mode's value is invalid, needs %d,"
                    " but got %d", s->mode, mode);
-        aio_context_release(aio_context);
         return;
     }
 
@@ -492,21 +490,18 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
         if (!s->active_disk || !s->active_disk->bs ||
                                     !s->active_disk->bs->backing) {
             error_setg(errp, "Active disk doesn't have backing file");
-            aio_context_release(aio_context);
             return;
         }
 
         s->hidden_disk = s->active_disk->bs->backing;
         if (!s->hidden_disk->bs || !s->hidden_disk->bs->backing) {
             error_setg(errp, "Hidden disk doesn't have backing file");
-            aio_context_release(aio_context);
             return;
         }
 
         s->secondary_disk = s->hidden_disk->bs->backing;
         if (!s->secondary_disk->bs || !bdrv_has_blk(s->secondary_disk->bs)) {
             error_setg(errp, "The secondary disk doesn't have block backend");
-            aio_context_release(aio_context);
             return;
         }
 
@@ -518,7 +513,6 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
             active_length != hidden_length || hidden_length != disk_length) {
             error_setg(errp, "Active disk, hidden disk, secondary disk's length"
                        " are not the same");
-            aio_context_release(aio_context);
             return;
         }
 
@@ -529,7 +523,6 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
             !s->hidden_disk->bs->drv->bdrv_make_empty) {
             error_setg(errp,
                        "Active disk or hidden disk doesn't support make_empty");
-            aio_context_release(aio_context);
             return;
         }
 
@@ -537,7 +530,6 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
         reopen_backing_file(bs, true, &local_err);
         if (local_err) {
             error_propagate(errp, local_err);
-            aio_context_release(aio_context);
             return;
         }
 
@@ -550,7 +542,6 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
             !check_top_bs(top_bs, bs)) {
             error_setg(errp, "No top_bs or it is invalid");
             reopen_backing_file(bs, false, NULL);
-            aio_context_release(aio_context);
             return;
         }
         bdrv_op_block_all(top_bs, s->blocker);
@@ -566,13 +557,11 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
         if (local_err) {
             error_propagate(errp, local_err);
             backup_job_cleanup(bs);
-            aio_context_release(aio_context);
             return;
         }
         job_start(&s->backup_job->job);
         break;
     default:
-        aio_context_release(aio_context);
         abort();
     }
 
@@ -582,18 +571,15 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
         secondary_do_checkpoint(s, errp);
     }
 
-    s->error = 0;
-    aio_context_release(aio_context);
+    qatomic_set(&s->error, 0);
 }
 
+/* Called with no I/O pending.  */
 static void replication_do_checkpoint(ReplicationState *rs, Error **errp)
 {
     BlockDriverState *bs = rs->opaque;
     BDRVReplicationState *s;
-    AioContext *aio_context;
 
-    aio_context = bdrv_get_aio_context(bs);
-    aio_context_acquire(aio_context);
     s = bs->opaque;
 
     if (s->stage == BLOCK_REPLICATION_DONE ||
@@ -603,38 +589,30 @@ static void replication_do_checkpoint(ReplicationState *rs, Error **errp)
          * Ignore the request because the secondary side of replication
          * doesn't have to do anything anymore.
          */
-        aio_context_release(aio_context);
         return;
     }
 
     if (s->mode == REPLICATION_MODE_SECONDARY) {
         secondary_do_checkpoint(s, errp);
     }
-    aio_context_release(aio_context);
 }
 
 static void replication_get_error(ReplicationState *rs, Error **errp)
 {
     BlockDriverState *bs = rs->opaque;
     BDRVReplicationState *s;
-    AioContext *aio_context;
 
-    aio_context = bdrv_get_aio_context(bs);
-    aio_context_acquire(aio_context);
     s = bs->opaque;
 
     if (s->stage == BLOCK_REPLICATION_NONE) {
         error_setg(errp, "Block replication is not running");
-        aio_context_release(aio_context);
         return;
     }
 
-    if (s->error) {
+    if (qatomic_read(&s->error)) {
         error_setg(errp, "I/O error occurred");
-        aio_context_release(aio_context);
         return;
     }
-    aio_context_release(aio_context);
 }
 
 static void replication_done(void *opaque, int ret)
@@ -648,10 +626,10 @@ static void replication_done(void *opaque, int ret)
         s->active_disk = NULL;
         s->secondary_disk = NULL;
         s->hidden_disk = NULL;
-        s->error = 0;
+        qatomic_set(&s->error, 0);
     } else {
         s->stage = BLOCK_REPLICATION_FAILOVER_FAILED;
-        s->error = -EIO;
+        qatomic_set(&s->error, -EIO);
     }
 }
 
@@ -659,10 +637,7 @@ static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
 {
     BlockDriverState *bs = rs->opaque;
     BDRVReplicationState *s;
-    AioContext *aio_context;
 
-    aio_context = bdrv_get_aio_context(bs);
-    aio_context_acquire(aio_context);
     s = bs->opaque;
 
     if (s->stage == BLOCK_REPLICATION_DONE ||
@@ -672,20 +647,18 @@ static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
          * Ignore the request because the secondary side of replication
          * doesn't have to do anything anymore.
          */
-        aio_context_release(aio_context);
         return;
     }
 
     if (s->stage != BLOCK_REPLICATION_RUNNING) {
         error_setg(errp, "Block replication is not running");
-        aio_context_release(aio_context);
         return;
     }
 
     switch (s->mode) {
     case REPLICATION_MODE_PRIMARY:
         s->stage = BLOCK_REPLICATION_DONE;
-        s->error = 0;
+        qatomic_set(&s->error, 0);
         break;
     case REPLICATION_MODE_SECONDARY:
         /*
@@ -700,7 +673,6 @@ static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
         if (!failover) {
             secondary_do_checkpoint(s, errp);
             s->stage = BLOCK_REPLICATION_DONE;
-            aio_context_release(aio_context);
             return;
         }
 
@@ -711,10 +683,8 @@ static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
                             NULL, replication_done, bs, true, errp);
         break;
     default:
-        aio_context_release(aio_context);
         abort();
     }
-    aio_context_release(aio_context);
 }
 
 static const char *const replication_strong_runtime_opts[] = {
-- 
2.30.2

[PATCH v2 8/8] block: do not take AioContext around reopen

[Emanuele Giuseppe Esposito]

Reopen needs to handle AioContext carefully due to calling
bdrv_drain_all_begin/end.  By not taking AioContext around calls to
bdrv_reopen_multiple, we can drop the function's release/acquire
pair and the AioContext argument too.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
---
 block/block-backend.c |  4 ----
 block/mirror.c        |  9 ---------
 blockdev.c            | 19 ++++++-------------
 3 files changed, 6 insertions(+), 26 deletions(-)

diff --git a/block/block-backend.c b/block/block-backend.c
index 413af51f3b..6fdc698e9e 100644
--- a/block/block-backend.c
+++ b/block/block-backend.c
@@ -2291,20 +2291,16 @@ int blk_commit_all(void)
     BlockBackend *blk = NULL;
 
     while ((blk = blk_all_next(blk)) != NULL) {
-        AioContext *aio_context = blk_get_aio_context(blk);
         BlockDriverState *unfiltered_bs = bdrv_skip_filters(blk_bs(blk));
 
-        aio_context_acquire(aio_context);
         if (blk_is_inserted(blk) && bdrv_cow_child(unfiltered_bs)) {
             int ret;
 
             ret = bdrv_commit(unfiltered_bs);
             if (ret < 0) {
-                aio_context_release(aio_context);
                 return ret;
             }
         }
-        aio_context_release(aio_context);
     }
     return 0;
 }
diff --git a/block/mirror.c b/block/mirror.c
index 5a71bd8bbc..43174bbc6b 100644
--- a/block/mirror.c
+++ b/block/mirror.c
@@ -631,7 +631,6 @@ static int mirror_exit_common(Job *job)
     MirrorBlockJob *s = container_of(job, MirrorBlockJob, common.job);
     BlockJob *bjob = &s->common;
     MirrorBDSOpaque *bs_opaque;
-    AioContext *replace_aio_context = NULL;
     BlockDriverState *src;
     BlockDriverState *target_bs;
     BlockDriverState *mirror_top_bs;
@@ -699,11 +698,6 @@ static int mirror_exit_common(Job *job)
         }
     }
 
-    if (s->to_replace) {
-        replace_aio_context = bdrv_get_aio_context(s->to_replace);
-        aio_context_acquire(replace_aio_context);
-    }
-
     if (s->should_complete && !abort) {
         BlockDriverState *to_replace = s->to_replace ?: src;
         bool ro = bdrv_is_read_only(to_replace);
@@ -740,9 +734,6 @@ static int mirror_exit_common(Job *job)
         error_free(s->replace_blocker);
         bdrv_unref(s->to_replace);
     }
-    if (replace_aio_context) {
-        aio_context_release(replace_aio_context);
-    }
     g_free(s->replaces);
     bdrv_unref(target_bs);
 
diff --git a/blockdev.c b/blockdev.c
index e901107344..1672ef756e 100644
--- a/blockdev.c
+++ b/blockdev.c
@@ -3469,7 +3469,6 @@ void qmp_change_backing_file(const char *device,
                              Error **errp)
 {
     BlockDriverState *bs = NULL;
-    AioContext *aio_context;
     BlockDriverState *image_bs = NULL;
     Error *local_err = NULL;
     bool ro;
@@ -3480,37 +3479,34 @@ void qmp_change_backing_file(const char *device,
         return;
     }
 
-    aio_context = bdrv_get_aio_context(bs);
-    aio_context_acquire(aio_context);
-
     image_bs = bdrv_lookup_bs(NULL, image_node_name, &local_err);
     if (local_err) {
         error_propagate(errp, local_err);
-        goto out;
+        return;
     }
 
     if (!image_bs) {
         error_setg(errp, "image file not found");
-        goto out;
+        return;
     }
 
     if (bdrv_find_base(image_bs) == image_bs) {
         error_setg(errp, "not allowing backing file change on an image "
                          "without a backing file");
-        goto out;
+        return;
     }
 
     /* even though we are not necessarily operating on bs, we need it to
      * determine if block ops are currently prohibited on the chain */
     if (bdrv_op_is_blocked(bs, BLOCK_OP_TYPE_CHANGE, errp)) {
-        goto out;
+        return;
     }
 
     /* final sanity check */
     if (!bdrv_chain_contains(bs, image_bs)) {
         error_setg(errp, "'%s' and image file are not in the same chain",
                    device);
-        goto out;
+        return;
     }
 
     /* if not r/w, reopen to make r/w */
@@ -3518,7 +3514,7 @@ void qmp_change_backing_file(const char *device,
 
     if (ro) {
         if (bdrv_reopen_set_read_only(image_bs, false, errp) != 0) {
-            goto out;
+            return;
         }
     }
 
@@ -3536,9 +3532,6 @@ void qmp_change_backing_file(const char *device,
     if (ro) {
         bdrv_reopen_set_read_only(image_bs, true, errp);
     }
-
-out:
-    aio_context_release(aio_context);
 }
 
 void qmp_blockdev_add(BlockdevOptions *options, Error **errp)
-- 
2.30.2

Re: [PATCH v2 8/8] block: do not take AioContext around reopen

[Paolo Bonzini]

On 19/04/21 10:55, Emanuele Giuseppe Esposito wrote:
> Reopen needs to handle AioContext carefully due to calling
> bdrv_drain_all_begin/end.  By not taking AioContext around calls to
> bdrv_reopen_multiple, we can drop the function's release/acquire
> pair and the AioContext argument too.

So... I wrote this commit message and I cannot parse it anymore---much 
less relate it to the code in the patch.  This is a problem, but it 
doesn't mean that the patch is wrong.

bdrv_reopen_multiple does not have the AioContext argument anymore. 
It's not doing release/acquire either.  The relevant commit is commit 
1a63a90750 ("block: Keep nodes drained between reopen_queue/multiple", 
2017-12-22).  You're basically cleaning up after that code in the same 
way as patch 7: reopen functions take care of keeping the BDS quiescent, 
so there's nothing to synchronize on.

For the future, the important step you missed was to check your diff 
against the one that you cherry-picked from.  Then you would have 
noticed that 1) it's much smaller 2) one thing that is mentioned in the 
commit message ("drop the function's release/acquire pair and argument") 
is not needed anymore.

Paolo

> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>   block/block-backend.c |  4 ----
>   block/mirror.c        |  9 ---------
>   blockdev.c            | 19 ++++++-------------
>   3 files changed, 6 insertions(+), 26 deletions(-)
> 
> diff --git a/block/block-backend.c b/block/block-backend.c
> index 413af51f3b..6fdc698e9e 100644
> --- a/block/block-backend.c
> +++ b/block/block-backend.c
> @@ -2291,20 +2291,16 @@ int blk_commit_all(void)
>       BlockBackend *blk = NULL;
>   
>       while ((blk = blk_all_next(blk)) != NULL) {
> -        AioContext *aio_context = blk_get_aio_context(blk);
>           BlockDriverState *unfiltered_bs = bdrv_skip_filters(blk_bs(blk));
>   
> -        aio_context_acquire(aio_context);
>           if (blk_is_inserted(blk) && bdrv_cow_child(unfiltered_bs)) {
>               int ret;
>   
>               ret = bdrv_commit(unfiltered_bs);
>               if (ret < 0) {
> -                aio_context_release(aio_context);
>                   return ret;
>               }
>           }
> -        aio_context_release(aio_context);
>       }
>       return 0;
>   }
> diff --git a/block/mirror.c b/block/mirror.c
> index 5a71bd8bbc..43174bbc6b 100644
> --- a/block/mirror.c
> +++ b/block/mirror.c
> @@ -631,7 +631,6 @@ static int mirror_exit_common(Job *job)
>       MirrorBlockJob *s = container_of(job, MirrorBlockJob, common.job);
>       BlockJob *bjob = &s->common;
>       MirrorBDSOpaque *bs_opaque;
> -    AioContext *replace_aio_context = NULL;
>       BlockDriverState *src;
>       BlockDriverState *target_bs;
>       BlockDriverState *mirror_top_bs;
> @@ -699,11 +698,6 @@ static int mirror_exit_common(Job *job)
>           }
>       }
>   
> -    if (s->to_replace) {
> -        replace_aio_context = bdrv_get_aio_context(s->to_replace);
> -        aio_context_acquire(replace_aio_context);
> -    }
> -
>       if (s->should_complete && !abort) {
>           BlockDriverState *to_replace = s->to_replace ?: src;
>           bool ro = bdrv_is_read_only(to_replace);
> @@ -740,9 +734,6 @@ static int mirror_exit_common(Job *job)
>           error_free(s->replace_blocker);
>           bdrv_unref(s->to_replace);
>       }
> -    if (replace_aio_context) {
> -        aio_context_release(replace_aio_context);
> -    }
>       g_free(s->replaces);
>       bdrv_unref(target_bs);
>   
> diff --git a/blockdev.c b/blockdev.c
> index e901107344..1672ef756e 100644
> --- a/blockdev.c
> +++ b/blockdev.c
> @@ -3469,7 +3469,6 @@ void qmp_change_backing_file(const char *device,
>                                Error **errp)
>   {
>       BlockDriverState *bs = NULL;
> -    AioContext *aio_context;
>       BlockDriverState *image_bs = NULL;
>       Error *local_err = NULL;
>       bool ro;
> @@ -3480,37 +3479,34 @@ void qmp_change_backing_file(const char *device,
>           return;
>       }
>   
> -    aio_context = bdrv_get_aio_context(bs);
> -    aio_context_acquire(aio_context);
> -
>       image_bs = bdrv_lookup_bs(NULL, image_node_name, &local_err);
>       if (local_err) {
>           error_propagate(errp, local_err);
> -        goto out;
> +        return;
>       }
>   
>       if (!image_bs) {
>           error_setg(errp, "image file not found");
> -        goto out;
> +        return;
>       }
>   
>       if (bdrv_find_base(image_bs) == image_bs) {
>           error_setg(errp, "not allowing backing file change on an image "
>                            "without a backing file");
> -        goto out;
> +        return;
>       }
>   
>       /* even though we are not necessarily operating on bs, we need it to
>        * determine if block ops are currently prohibited on the chain */
>       if (bdrv_op_is_blocked(bs, BLOCK_OP_TYPE_CHANGE, errp)) {
> -        goto out;
> +        return;
>       }
>   
>       /* final sanity check */
>       if (!bdrv_chain_contains(bs, image_bs)) {
>           error_setg(errp, "'%s' and image file are not in the same chain",
>                      device);
> -        goto out;
> +        return;
>       }
>   
>       /* if not r/w, reopen to make r/w */
> @@ -3518,7 +3514,7 @@ void qmp_change_backing_file(const char *device,
>   
>       if (ro) {
>           if (bdrv_reopen_set_read_only(image_bs, false, errp) != 0) {
> -            goto out;
> +            return;
>           }
>       }
>   
> @@ -3536,9 +3532,6 @@ void qmp_change_backing_file(const char *device,
>       if (ro) {
>           bdrv_reopen_set_read_only(image_bs, true, errp);
>       }
> -
> -out:
> -    aio_context_release(aio_context);
>   }
>   
>   void qmp_blockdev_add(BlockdevOptions *options, Error **errp)
> 

Re: [PATCH v2 0/8] Block layer thread-safety, continued

[Paolo Bonzini]

On 19/04/21 10:55, Emanuele Giuseppe Esposito wrote:
> This and the following serie of patches are based on Paolo's
> v1 patches sent in 2017[*]. They have been ported to the current QEMU
> version, but the goal remains the same:
> - make the block layer thread-safe (patches 1-5), and
> - remove aio_context_acquire/release (patches 6-8).
> 
> [*] = https://lists.gnu.org/archive/html/qemu-devel/2017-07/msg01398.html
> 
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>

This looks good to me, though the commit message of patch 8 needs to be 
rewritten.

Paolo

> ---
> v1 (2017) -> v2 (2021):
> - v1 Patch "block-backup: add reqs_lock" has been dropped, because now
>    is completely different from the old version and all functions
>    that were affected by it have been moved or deleted.
>    It will be replaced by another serie that aims to thread safety to
>    block/block-copy.c
> - remaining v1 patches will be integrated in next serie.
> - Patch "block: do not acquire AioContext in check_to_replace_node"
>    moves part of the logic of check_to_replace_node to the caller,
>    so that the function can be included in the aio_context_acquire/release
>    block that follows.
> 
> Emanuele Giuseppe Esposito (8):
>    block: prepare write threshold code for thread safety
>    block: protect write threshold QMP commands from concurrent requests
>    util: use RCU accessors for notifiers
>    block: make before-write notifiers thread-safe
>    block: add a few more notes on locking
>    block: do not acquire AioContext in check_to_replace_node
>    block/replication: do not acquire AioContext
>    block: do not take AioContext around reopen
> 
>   block.c                   | 28 ++++++--------------
>   block/block-backend.c     |  4 ---
>   block/io.c                | 12 +++++++++
>   block/mirror.c            |  9 -------
>   block/replication.c       | 54 +++++++++------------------------------
>   block/write-threshold.c   | 39 ++++++++++++++--------------
>   blockdev.c                | 26 +++++++++----------
>   include/block/block.h     |  1 +
>   include/block/block_int.h | 42 +++++++++++++++++++++++++++++-
>   util/notify.c             | 13 +++++-----
>   10 files changed, 113 insertions(+), 115 deletions(-)
> 

Re: [PATCH v2 4/8] block: make before-write notifiers thread-safe

[Vladimir Sementsov-Ogievskiy]

19.04.2021 11:55, Emanuele Giuseppe Esposito wrote:
> Reads access the list in RCU style, so be careful to avoid use-after-free
> scenarios in the backup block job.  Apart from this, all that's needed
> is protecting updates with a mutex.

Note that backup doesn't use write-notifiers now. Probably best thing to do is to update other users to use filters instead of write notifiers, and drop write notifiers at all. But it may require more work, so don't care.

> 
> Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>   block.c                   |  6 +++---
>   block/io.c                | 12 ++++++++++++
>   block/write-threshold.c   |  2 +-
>   include/block/block_int.h | 37 +++++++++++++++++++++++++++++++++++++
>   4 files changed, 53 insertions(+), 4 deletions(-)
> 
> diff --git a/block.c b/block.c
> index c5b887cec1..f40fb63c75 100644
> --- a/block.c
> +++ b/block.c
> @@ -6434,7 +6434,7 @@ static void bdrv_do_remove_aio_context_notifier(BdrvAioNotifier *ban)
>       g_free(ban);
>   }
>   
> -static void bdrv_detach_aio_context(BlockDriverState *bs)
> +void bdrv_detach_aio_context(BlockDriverState *bs)

why it become public? It's not used in this commit, and this change is unrelated to commit message..

>   {
>       BdrvAioNotifier *baf, *baf_tmp;
>   
> @@ -6462,8 +6462,8 @@ static void bdrv_detach_aio_context(BlockDriverState *bs)
>       bs->aio_context = NULL;
>   }
>   
> -static void bdrv_attach_aio_context(BlockDriverState *bs,
> -                                    AioContext *new_context)
> +void bdrv_attach_aio_context(BlockDriverState *bs,
> +                             AioContext *new_context)
>   {
>       BdrvAioNotifier *ban, *ban_tmp;
>   
> diff --git a/block/io.c b/block/io.c
> index ca2dca3007..8f6af6077b 100644
> --- a/block/io.c
> +++ b/block/io.c
> @@ -3137,10 +3137,22 @@ bool bdrv_qiov_is_aligned(BlockDriverState *bs, QEMUIOVector *qiov)
>       return true;
>   }
>   
> +static QemuSpin notifiers_spin_lock;
> +
>   void bdrv_add_before_write_notifier(BlockDriverState *bs,
>                                       NotifierWithReturn *notifier)
>   {
> +    qemu_spin_lock(&notifiers_spin_lock);
>       notifier_with_return_list_add(&bs->before_write_notifiers, notifier);
> +    qemu_spin_unlock(&notifiers_spin_lock);
> +}
> +
> +void bdrv_remove_before_write_notifier(BlockDriverState *bs,
> +                                       NotifierWithReturn *notifier)
> +{
> +    qemu_spin_lock(&notifiers_spin_lock);
> +    notifier_with_return_remove(notifier);
> +    qemu_spin_unlock(&notifiers_spin_lock);
>   }
>   
>   void bdrv_io_plug(BlockDriverState *bs)
> diff --git a/block/write-threshold.c b/block/write-threshold.c
> index 77c74bdaa7..b87b749b02 100644
> --- a/block/write-threshold.c
> +++ b/block/write-threshold.c
> @@ -32,7 +32,7 @@ bool bdrv_write_threshold_is_set(const BlockDriverState *bs)
>   static void write_threshold_disable(BlockDriverState *bs)
>   {
>       if (bdrv_write_threshold_is_set(bs)) {
> -        notifier_with_return_remove(&bs->write_threshold_notifier);
> +        bdrv_remove_before_write_notifier(bs, &bs->write_threshold_notifier);
>           bs->write_threshold_offset = 0;
>       }
>   }
> diff --git a/include/block/block_int.h b/include/block/block_int.h
> index 88e4111939..a1aad5ad2d 100644
> --- a/include/block/block_int.h
> +++ b/include/block/block_int.h
> @@ -1089,6 +1089,8 @@ bool bdrv_backing_overridden(BlockDriverState *bs);
>   
>   /**
>    * bdrv_add_before_write_notifier:
> + * @bs: The #BlockDriverState for which to register the notifier
> + * @notifier: The notifier to add
>    *
>    * Register a callback that is invoked before write requests are processed but
>    * after any throttling or waiting for overlapping requests.
> @@ -1096,6 +1098,41 @@ bool bdrv_backing_overridden(BlockDriverState *bs);
>   void bdrv_add_before_write_notifier(BlockDriverState *bs,
>                                       NotifierWithReturn *notifier);
>   
> +/**
> + * bdrv_remove_before_write_notifier:
> + * @bs: The #BlockDriverState for which to register the notifier
> + * @notifier: The notifier to add
> + *
> + * Unregister a callback that is invoked before write requests are processed but
> + * after any throttling or waiting for overlapping requests.
> + *
> + * Note that more I/O could be pending on @bs.  You need to wait for
> + * it to finish, for example with bdrv_drain(), before freeing @notifier.
> + */
> +void bdrv_remove_before_write_notifier(BlockDriverState *bs,
> +                                       NotifierWithReturn *notifier);
> +
> +/**
> + * bdrv_detach_aio_context:
> + *
> + * May be called from .bdrv_detach_aio_context() to detach children from the
> + * current #AioContext.  This is only needed by block drivers that manage their
> + * own children.  Both ->file and ->backing are automatically handled and
> + * block drivers should not call this function on them explicitly.
> + */
> +void bdrv_detach_aio_context(BlockDriverState *bs);
> +
> +/**
> + * bdrv_attach_aio_context:
> + *
> + * May be called from .bdrv_attach_aio_context() to attach children to the new
> + * #AioContext.  This is only needed by block drivers that manage their own
> + * children.  Both ->file and ->backing are automatically handled and block
> + * drivers should not call this function on them explicitly.
> + */
> +void bdrv_attach_aio_context(BlockDriverState *bs,
> +                             AioContext *new_context);
> +
>   /**
>    * bdrv_add_aio_context_notifier:
>    *
> 


-- 
Best regards,
Vladimir

Re: [PATCH v2 4/8] block: make before-write notifiers thread-safe

[Vladimir Sementsov-Ogievskiy]

22.04.2021 00:23, Vladimir Sementsov-Ogievskiy wrote:
> 19.04.2021 11:55, Emanuele Giuseppe Esposito wrote:
>> Reads access the list in RCU style, so be careful to avoid use-after-free
>> scenarios in the backup block job.  Apart from this, all that's needed
>> is protecting updates with a mutex.
> 
> Note that backup doesn't use write-notifiers now. Probably best thing to do is to update other users to use filters instead of write notifiers, and drop write notifiers at all. But it may require more work, so don't care.

But on the other hand.. Why not. We can go without filter and still drop write-notifiers. Look at my new patch "[PATCH] block: simplify write-threshold and drop write notifiers"

-- 
Best regards,
Vladimir

Re: [PATCH v2 1/8] block: prepare write threshold code for thread safety

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 2920 bytes --]

On Mon, Apr 19, 2021 at 10:55:34AM +0200, Emanuele Giuseppe Esposito wrote:

No commit description. What about write thresholds makes them thread
unsafe? Without a commit description reviewers have to reverse-engineer
the patch to figure out the author's intention, which can lead to
misunderstandings and bugs slipping through.

My guess is the point of this patch was to stop accessing fields in bs
directly?

> Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
> Co-developed-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>  block/write-threshold.c | 28 ++++++++++++++++------------
>  1 file changed, 16 insertions(+), 12 deletions(-)
> 
> diff --git a/block/write-threshold.c b/block/write-threshold.c
> index 85b78dc2a9..63040fa4f2 100644
> --- a/block/write-threshold.c
> +++ b/block/write-threshold.c
> @@ -37,18 +37,22 @@ static void write_threshold_disable(BlockDriverState *bs)
>      }
>  }
>  
> +static uint64_t treshold_overage(const BdrvTrackedRequest *req,
> +                                uint64_t thres)
> +{
> +    if (thres > 0 && req->offset + req->bytes > thres) {
> +        return req->offset + req->bytes - thres;
> +    } else {
> +        return 0;
> +    }
> +}
> +
>  uint64_t bdrv_write_threshold_exceeded(const BlockDriverState *bs,
>                                         const BdrvTrackedRequest *req)
>  {
> -    if (bdrv_write_threshold_is_set(bs)) {
> -        if (req->offset > bs->write_threshold_offset) {
> -            return (req->offset - bs->write_threshold_offset) + req->bytes;
> -        }
> -        if ((req->offset + req->bytes) > bs->write_threshold_offset) {
> -            return (req->offset + req->bytes) - bs->write_threshold_offset;
> -        }
> -    }
> -    return 0;
> +    uint64_t thres = bdrv_write_threshold_get(bs);
> +
> +    return treshold_overage(req, thres);
>  }

Hmm...this function is only used by tests now. Should the tests be
updated so that they are exercising the actual code instead of this
test-only interface?

>  
>  static int coroutine_fn before_write_notify(NotifierWithReturn *notifier,
> @@ -56,14 +60,14 @@ static int coroutine_fn before_write_notify(NotifierWithReturn *notifier,
>  {
>      BdrvTrackedRequest *req = opaque;
>      BlockDriverState *bs = req->bs;
> -    uint64_t amount = 0;
> +    uint64_t thres = bdrv_write_threshold_get(bs);
> +    uint64_t amount = treshold_overage(req, thres);
>  
> -    amount = bdrv_write_threshold_exceeded(bs, req);
>      if (amount > 0) {
>          qapi_event_send_block_write_threshold(
>              bs->node_name,
>              amount,
> -            bs->write_threshold_offset);
> +            thres);
>  
>          /* autodisable to avoid flooding the monitor */
>          write_threshold_disable(bs);
> -- 
> 2.30.2
> 

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 2/8] block: protect write threshold QMP commands from concurrent requests

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1612 bytes --]

On Mon, Apr 19, 2021 at 10:55:35AM +0200, Emanuele Giuseppe Esposito wrote:
> For simplicity, use bdrv_drained_begin/end to avoid concurrent
> writes to the write threshold, or reading it while it is being set.
> qmp_block_set_write_threshold is protected by the BQL.
> 
> Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
> Co-developed-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>  block/write-threshold.c | 9 +++------
>  1 file changed, 3 insertions(+), 6 deletions(-)
> 
> diff --git a/block/write-threshold.c b/block/write-threshold.c
> index 63040fa4f2..77c74bdaa7 100644
> --- a/block/write-threshold.c
> +++ b/block/write-threshold.c
> @@ -111,7 +111,6 @@ void qmp_block_set_write_threshold(const char *node_name,
>                                     Error **errp)
>  {
>      BlockDriverState *bs;
> -    AioContext *aio_context;
>  
>      bs = bdrv_find_node(node_name);
>      if (!bs) {
> @@ -119,10 +118,8 @@ void qmp_block_set_write_threshold(const char *node_name,
>          return;
>      }
>  
> -    aio_context = bdrv_get_aio_context(bs);
> -    aio_context_acquire(aio_context);
> -
> +    /* Avoid a concurrent write_threshold_disable.  */
> +    bdrv_drained_begin(bs);

Is there documentation that says it's safe to call
bdrv_drained_begin(bs) without the AioContext acquired? AIO_WAIT_WHILE()
contradicts this:

  The caller's thread must be the IOThread that owns @ctx or the main loop
  thread (with @ctx acquired exactly once).
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 3/8] util: use RCU accessors for notifiers

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1643 bytes --]

On Mon, Apr 19, 2021 at 10:55:36AM +0200, Emanuele Giuseppe Esposito wrote:

What is the goal? Making the notifier APIs usable from multiple threads
(when callers respect RCU)?

> Note that calling rcu_read_lock() is left to the caller.  In fact,
> if the notifier is really only used within the BQL, it's unnecessary.
> 
> Even outside the BQL, RCU accessors can also be used with any API that has
> the same contract as synchronize_rcu, i.e. it stops until all concurrent
> readers complete, no matter how "readers" are defined.  In the next patch,
> for example, synchronize_rcu's role is taken by bdrv_drain (which is a
> superset of synchronize_rcu, since it also blocks new incoming readers).
> 
> Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>  util/notify.c | 13 +++++++------
>  1 file changed, 7 insertions(+), 6 deletions(-)

No doc comments are added/updated so the RCU support is kind of hidden.
Please document it explicitly so that it's easy to write and review code
in the future without remembering implementation details of APIs. Using
"rcu" in function names would be most obvious and require no extra
documentation, but it probably makes more sense to have doc comments in
"qemu/notify.h" to avoid renaming everything.

> 
> diff --git a/util/notify.c b/util/notify.c
> index 76bab212ae..529f1d121e 100644
> --- a/util/notify.c
> +++ b/util/notify.c
> @@ -15,6 +15,7 @@

notifier_list_empty(NotifierList *list) should be updated to use
QLIST_EMPTY_RCU().

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 5/8] block: add a few more notes on locking

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 1459 bytes --]

On Mon, Apr 19, 2021 at 10:55:38AM +0200, Emanuele Giuseppe Esposito wrote:
> Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
> ---
>  include/block/block_int.h | 5 ++++-
>  1 file changed, 4 insertions(+), 1 deletion(-)
> 
> diff --git a/include/block/block_int.h b/include/block/block_int.h
> index a1aad5ad2d..67a0777e12 100644
> --- a/include/block/block_int.h
> +++ b/include/block/block_int.h
> @@ -154,7 +154,9 @@ struct BlockDriver {
>       */
>      bool supports_backing;
>  
> -    /* For handling image reopen for split or non-split files */
> +    /* For handling image reopen for split or non-split files.  Called
> +     * with no I/O pending.

Does "Called with no I/O pending" mean "Called in a drained section"?

> +     */
>      int (*bdrv_reopen_prepare)(BDRVReopenState *reopen_state,
>                                 BlockReopenQueue *queue, Error **errp);
>      void (*bdrv_reopen_commit)(BDRVReopenState *reopen_state);
> @@ -168,6 +170,7 @@ struct BlockDriver {
>      /* Protocol drivers should implement this instead of bdrv_open */
>      int (*bdrv_file_open)(BlockDriverState *bs, QDict *options, int flags,
>                            Error **errp);
> +    /* Called from main thread.  */
>      void (*bdrv_close)(BlockDriverState *bs);
>  
>  
> -- 
> 2.30.2
> 

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 1/8] block: prepare write threshold code for thread safety

[Vladimir Sementsov-Ogievskiy]

Hi Stefan!

Note my "[PATCH v2 0/9] block: refactor write threshold", it's a kind of counter-proposal for first half of this series.

05.05.2021 11:50, Stefan Hajnoczi wrote:
> On Mon, Apr 19, 2021 at 10:55:34AM +0200, Emanuele Giuseppe Esposito wrote:
> 
> No commit description. What about write thresholds makes them thread
> unsafe? Without a commit description reviewers have to reverse-engineer
> the patch to figure out the author's intention, which can lead to
> misunderstandings and bugs slipping through.
> 
> My guess is the point of this patch was to stop accessing fields in bs
> directly?
> 
>> Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>
>> Co-developed-by: Paolo Bonzini <pbonzini@redhat.com>
>> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>
>> ---
>>   block/write-threshold.c | 28 ++++++++++++++++------------
>>   1 file changed, 16 insertions(+), 12 deletions(-)
>>
>> diff --git a/block/write-threshold.c b/block/write-threshold.c
>> index 85b78dc2a9..63040fa4f2 100644
>> --- a/block/write-threshold.c
>> +++ b/block/write-threshold.c
>> @@ -37,18 +37,22 @@ static void write_threshold_disable(BlockDriverState *bs)
>>       }
>>   }
>>   
>> +static uint64_t treshold_overage(const BdrvTrackedRequest *req,
>> +                                uint64_t thres)
>> +{
>> +    if (thres > 0 && req->offset + req->bytes > thres) {
>> +        return req->offset + req->bytes - thres;
>> +    } else {
>> +        return 0;
>> +    }
>> +}
>> +
>>   uint64_t bdrv_write_threshold_exceeded(const BlockDriverState *bs,
>>                                          const BdrvTrackedRequest *req)
>>   {
>> -    if (bdrv_write_threshold_is_set(bs)) {
>> -        if (req->offset > bs->write_threshold_offset) {
>> -            return (req->offset - bs->write_threshold_offset) + req->bytes;
>> -        }
>> -        if ((req->offset + req->bytes) > bs->write_threshold_offset) {
>> -            return (req->offset + req->bytes) - bs->write_threshold_offset;
>> -        }
>> -    }
>> -    return 0;
>> +    uint64_t thres = bdrv_write_threshold_get(bs);
>> +
>> +    return treshold_overage(req, thres);
>>   }
> 
> Hmm...this function is only used by tests now. Should the tests be
> updated so that they are exercising the actual code instead of this
> test-only interface?
> 
>>   
>>   static int coroutine_fn before_write_notify(NotifierWithReturn *notifier,
>> @@ -56,14 +60,14 @@ static int coroutine_fn before_write_notify(NotifierWithReturn *notifier,
>>   {
>>       BdrvTrackedRequest *req = opaque;
>>       BlockDriverState *bs = req->bs;
>> -    uint64_t amount = 0;
>> +    uint64_t thres = bdrv_write_threshold_get(bs);
>> +    uint64_t amount = treshold_overage(req, thres);
>>   
>> -    amount = bdrv_write_threshold_exceeded(bs, req);
>>       if (amount > 0) {
>>           qapi_event_send_block_write_threshold(
>>               bs->node_name,
>>               amount,
>> -            bs->write_threshold_offset);
>> +            thres);
>>   
>>           /* autodisable to avoid flooding the monitor */
>>           write_threshold_disable(bs);
>> -- 
>> 2.30.2
>>


-- 
Best regards,
Vladimir

Re: [PATCH v2 7/8] block/replication: do not acquire AioContext

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 578 bytes --]

On Mon, Apr 19, 2021 at 10:55:40AM +0200, Emanuele Giuseppe Esposito wrote:
> @@ -210,7 +212,7 @@ static int replication_return_value(BDRVReplicationState *s, int ret)
>      }
>  
>      if (ret < 0) {
> -        s->error = ret;
> +        qatomic_set(&s->error, ret);
>          ret = 0;
>      }
>  
> @@ -307,6 +309,7 @@ out:
>      return ret;
>  }
>  
> +/* Called with no I/O pending.  */

It would be clearer to refer to the specific guarantee that no I/O is
pending, like "Called from a drained section". There are more comments
like this one below.

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 8/8] block: do not take AioContext around reopen

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 531 bytes --]

On Mon, Apr 19, 2021 at 10:55:41AM +0200, Emanuele Giuseppe Esposito wrote:
>  block/block-backend.c |  4 ----
>  block/mirror.c        |  9 ---------
>  blockdev.c            | 19 ++++++-------------
>  3 files changed, 6 insertions(+), 26 deletions(-)

There are still many aio_context_acquire/release() calls in QEMU. I'm
not sure why these are safe to remove. The patch series needs to
communicate and document the final state of thread safety and AioContext
lock usage so that it's clear for developers going forward.

Stefan

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 6/8] block: do not acquire AioContext in check_to_replace_node

[Paolo Bonzini]

On 19/04/21 10:55, Emanuele Giuseppe Esposito wrote:
> + *
> + * Called with AioContext lock held.


... for @to_replace_bs.

>           aio_context_acquire(replace_aio_context);
> +        if (!check_to_replace_node(bs, to_replace_bs, replaces, errp)) {

A release is missing here.

Paolo

> +            return;
> +        }
>           replace_size = bdrv_getlength(to_replace_bs);

Re: [PATCH v2 7/8] block/replication: do not acquire AioContext

[Paolo Bonzini]

On 19/04/21 10:55, Emanuele Giuseppe Esposito wrote:
> Replication functions are mostly called when the BDS is quiescent and
> does not have any pending I/O.  They do not need to synchronize on
> anything since BDS and BB are now thread-safe.
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>

This patch has the same issue that Stefan mentioned in patch 2.  In 
particular, the following chain leads to the bdrv*drain family of 
functions and thus AIO_WAIT_WHILE:

replication_start -> reopen_backing_file -> bdrv_subtree_drained_begin

and also

replication_stop -> commit_active_start -> bdrv_reopen_set_read_only

The same is true of patch 8, where you have call sequences like

bdrv_commit -> blk_flush -> bdrv_flush (in the generated file 
block/block-gen.c) -> bdrv_poll_co

So patches 7 and 8 need to be shelved for now, as they can only go in 
with the overall removal of AioContext lock.

Paolo

> ---
>   block/replication.c | 54 ++++++++++-----------------------------------
>   1 file changed, 12 insertions(+), 42 deletions(-)
> 
> diff --git a/block/replication.c b/block/replication.c
> index 97be7ef4de..25ee37b21b 100644
> --- a/block/replication.c
> +++ b/block/replication.c
> @@ -45,6 +45,8 @@ typedef struct BDRVReplicationState {
>       Error *blocker;
>       bool orig_hidden_read_only;
>       bool orig_secondary_read_only;
> +
> +    /* This field is accessed asynchronously.  */
>       int error;
>   } BDRVReplicationState;
>   
> @@ -210,7 +212,7 @@ static int replication_return_value(BDRVReplicationState *s, int ret)
>       }
>   
>       if (ret < 0) {
> -        s->error = ret;
> +        qatomic_set(&s->error, ret);
>           ret = 0;
>       }
>   
> @@ -307,6 +309,7 @@ out:
>       return ret;
>   }
>   
> +/* Called with no I/O pending.  */
>   static void secondary_do_checkpoint(BDRVReplicationState *s, Error **errp)
>   {
>       Error *local_err = NULL;
> @@ -420,7 +423,7 @@ static void backup_job_completed(void *opaque, int ret)
>   
>       if (s->stage != BLOCK_REPLICATION_FAILOVER) {
>           /* The backup job is cancelled unexpectedly */
> -        s->error = -EIO;
> +        qatomic_set(&s->error, -EIO);
>       }
>   
>       backup_job_cleanup(bs);
> @@ -445,6 +448,7 @@ static bool check_top_bs(BlockDriverState *top_bs, BlockDriverState *bs)
>       return false;
>   }
>   
> +/* Called with no I/O pending.  */
>   static void replication_start(ReplicationState *rs, ReplicationMode mode,
>                                 Error **errp)
>   {
> @@ -452,12 +456,9 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>       BDRVReplicationState *s;
>       BlockDriverState *top_bs;
>       int64_t active_length, hidden_length, disk_length;
> -    AioContext *aio_context;
>       Error *local_err = NULL;
>       BackupPerf perf = { .use_copy_range = true, .max_workers = 1 };
>   
> -    aio_context = bdrv_get_aio_context(bs);
> -    aio_context_acquire(aio_context);
>       s = bs->opaque;
>   
>       if (s->stage == BLOCK_REPLICATION_DONE ||
> @@ -467,20 +468,17 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>            * Ignore the request because the secondary side of replication
>            * doesn't have to do anything anymore.
>            */
> -        aio_context_release(aio_context);
>           return;
>       }
>   
>       if (s->stage != BLOCK_REPLICATION_NONE) {
>           error_setg(errp, "Block replication is running or done");
> -        aio_context_release(aio_context);
>           return;
>       }
>   
>       if (s->mode != mode) {
>           error_setg(errp, "The parameter mode's value is invalid, needs %d,"
>                      " but got %d", s->mode, mode);
> -        aio_context_release(aio_context);
>           return;
>       }
>   
> @@ -492,21 +490,18 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>           if (!s->active_disk || !s->active_disk->bs ||
>                                       !s->active_disk->bs->backing) {
>               error_setg(errp, "Active disk doesn't have backing file");
> -            aio_context_release(aio_context);
>               return;
>           }
>   
>           s->hidden_disk = s->active_disk->bs->backing;
>           if (!s->hidden_disk->bs || !s->hidden_disk->bs->backing) {
>               error_setg(errp, "Hidden disk doesn't have backing file");
> -            aio_context_release(aio_context);
>               return;
>           }
>   
>           s->secondary_disk = s->hidden_disk->bs->backing;
>           if (!s->secondary_disk->bs || !bdrv_has_blk(s->secondary_disk->bs)) {
>               error_setg(errp, "The secondary disk doesn't have block backend");
> -            aio_context_release(aio_context);
>               return;
>           }
>   
> @@ -518,7 +513,6 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>               active_length != hidden_length || hidden_length != disk_length) {
>               error_setg(errp, "Active disk, hidden disk, secondary disk's length"
>                          " are not the same");
> -            aio_context_release(aio_context);
>               return;
>           }
>   
> @@ -529,7 +523,6 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>               !s->hidden_disk->bs->drv->bdrv_make_empty) {
>               error_setg(errp,
>                          "Active disk or hidden disk doesn't support make_empty");
> -            aio_context_release(aio_context);
>               return;
>           }
>   
> @@ -537,7 +530,6 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>           reopen_backing_file(bs, true, &local_err);
>           if (local_err) {
>               error_propagate(errp, local_err);
> -            aio_context_release(aio_context);
>               return;
>           }
>   
> @@ -550,7 +542,6 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>               !check_top_bs(top_bs, bs)) {
>               error_setg(errp, "No top_bs or it is invalid");
>               reopen_backing_file(bs, false, NULL);
> -            aio_context_release(aio_context);
>               return;
>           }
>           bdrv_op_block_all(top_bs, s->blocker);
> @@ -566,13 +557,11 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>           if (local_err) {
>               error_propagate(errp, local_err);
>               backup_job_cleanup(bs);
> -            aio_context_release(aio_context);
>               return;
>           }
>           job_start(&s->backup_job->job);
>           break;
>       default:
> -        aio_context_release(aio_context);
>           abort();
>       }
>   
> @@ -582,18 +571,15 @@ static void replication_start(ReplicationState *rs, ReplicationMode mode,
>           secondary_do_checkpoint(s, errp);
>       }
>   
> -    s->error = 0;
> -    aio_context_release(aio_context);
> +    qatomic_set(&s->error, 0);
>   }
>   
> +/* Called with no I/O pending.  */
>   static void replication_do_checkpoint(ReplicationState *rs, Error **errp)
>   {
>       BlockDriverState *bs = rs->opaque;
>       BDRVReplicationState *s;
> -    AioContext *aio_context;
>   
> -    aio_context = bdrv_get_aio_context(bs);
> -    aio_context_acquire(aio_context);
>       s = bs->opaque;
>   
>       if (s->stage == BLOCK_REPLICATION_DONE ||
> @@ -603,38 +589,30 @@ static void replication_do_checkpoint(ReplicationState *rs, Error **errp)
>            * Ignore the request because the secondary side of replication
>            * doesn't have to do anything anymore.
>            */
> -        aio_context_release(aio_context);
>           return;
>       }
>   
>       if (s->mode == REPLICATION_MODE_SECONDARY) {
>           secondary_do_checkpoint(s, errp);
>       }
> -    aio_context_release(aio_context);
>   }
>   
>   static void replication_get_error(ReplicationState *rs, Error **errp)
>   {
>       BlockDriverState *bs = rs->opaque;
>       BDRVReplicationState *s;
> -    AioContext *aio_context;
>   
> -    aio_context = bdrv_get_aio_context(bs);
> -    aio_context_acquire(aio_context);
>       s = bs->opaque;
>   
>       if (s->stage == BLOCK_REPLICATION_NONE) {
>           error_setg(errp, "Block replication is not running");
> -        aio_context_release(aio_context);
>           return;
>       }
>   
> -    if (s->error) {
> +    if (qatomic_read(&s->error)) {
>           error_setg(errp, "I/O error occurred");
> -        aio_context_release(aio_context);
>           return;
>       }
> -    aio_context_release(aio_context);
>   }
>   
>   static void replication_done(void *opaque, int ret)
> @@ -648,10 +626,10 @@ static void replication_done(void *opaque, int ret)
>           s->active_disk = NULL;
>           s->secondary_disk = NULL;
>           s->hidden_disk = NULL;
> -        s->error = 0;
> +        qatomic_set(&s->error, 0);
>       } else {
>           s->stage = BLOCK_REPLICATION_FAILOVER_FAILED;
> -        s->error = -EIO;
> +        qatomic_set(&s->error, -EIO);
>       }
>   }
>   
> @@ -659,10 +637,7 @@ static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
>   {
>       BlockDriverState *bs = rs->opaque;
>       BDRVReplicationState *s;
> -    AioContext *aio_context;
>   
> -    aio_context = bdrv_get_aio_context(bs);
> -    aio_context_acquire(aio_context);
>       s = bs->opaque;
>   
>       if (s->stage == BLOCK_REPLICATION_DONE ||
> @@ -672,20 +647,18 @@ static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
>            * Ignore the request because the secondary side of replication
>            * doesn't have to do anything anymore.
>            */
> -        aio_context_release(aio_context);
>           return;
>       }
>   
>       if (s->stage != BLOCK_REPLICATION_RUNNING) {
>           error_setg(errp, "Block replication is not running");
> -        aio_context_release(aio_context);
>           return;
>       }
>   
>       switch (s->mode) {
>       case REPLICATION_MODE_PRIMARY:
>           s->stage = BLOCK_REPLICATION_DONE;
> -        s->error = 0;
> +        qatomic_set(&s->error, 0);
>           break;
>       case REPLICATION_MODE_SECONDARY:
>           /*
> @@ -700,7 +673,6 @@ static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
>           if (!failover) {
>               secondary_do_checkpoint(s, errp);
>               s->stage = BLOCK_REPLICATION_DONE;
> -            aio_context_release(aio_context);
>               return;
>           }
>   
> @@ -711,10 +683,8 @@ static void replication_stop(ReplicationState *rs, bool failover, Error **errp)
>                               NULL, replication_done, bs, true, errp);
>           break;
>       default:
> -        aio_context_release(aio_context);
>           abort();
>       }
> -    aio_context_release(aio_context);
>   }
>   
>   static const char *const replication_strong_runtime_opts[] = {
> 

Re: [PATCH v2 2/8] block: protect write threshold QMP commands from concurrent requests

[Paolo Bonzini]

On 05/05/21 10:55, Stefan Hajnoczi wrote:
>> @@ -119,10 +118,8 @@ void qmp_block_set_write_threshold(const char *node_name,
>>           return;
>>       }
>>   
>> -    aio_context = bdrv_get_aio_context(bs);
>> -    aio_context_acquire(aio_context);
>> -
>> +    /* Avoid a concurrent write_threshold_disable.  */
>> +    bdrv_drained_begin(bs);
> 
> Is there documentation that says it's safe to call
> bdrv_drained_begin(bs) without the AioContext acquired? AIO_WAIT_WHILE()
> contradicts this:
> 
>    The caller's thread must be the IOThread that owns @ctx or the main loop
>    thread (with @ctx acquired exactly once).
>            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
> 

Well, I cannot remember why this patch was correct at the time I was
working on them.  Patches 7/8 on the other hand were okay because back
then bdrv_reopen_multiple() called bdrv_drain_all_begin().

Overall, what survives of this series is patches 5 and 6, plus Vladimir's
take on the write threshold code.

Anyway, things have gotten _more_ broken in the meanwhile, and this is
probably what causes the deadlocks that Emanuele has seen with the
remainder of the branch.  Since this patch:

     commit aa1361d54aac43094b98024b8b6c804eb6e41661
     Author: Kevin Wolf <kwolf@redhat.com>
     Date:   Fri Aug 17 18:54:18 2018 +0200

     block: Add missing locking in bdrv_co_drain_bh_cb()
     
     bdrv_do_drained_begin/end() assume that they are called with the
     AioContext lock of bs held. If we call drain functions from a coroutine
     with the AioContext lock held, we yield and schedule a BH to move out of
     coroutine context. This means that the lock for the home context of the
     coroutine is released and must be re-acquired in the bottom half.
     
     Signed-off-by: Kevin Wolf <kwolf@redhat.com>
     Reviewed-by: Max Reitz <mreitz@redhat.com>

AIO_WAIT_WHILE  is going down a path that does not do the release/acquire of
the AioContext, which can and will cause deadlocks when the main thread
tries to acquire the AioContext and the I/O thread is in bdrv_co_drain.

The message that introduced it does not help very much
(https://mail.gnu.org/archive/html/qemu-devel/2018-09/msg01687.html)
but I think this must be reverted.

The next steps however should have less problems with bitrot, in particular
the snapshots have changed very little.  Block job code is very different
but it is all in the main thread.

Paolo

Re: [PATCH v2 1/8] block: prepare write threshold code for thread safety

[Stefan Hajnoczi]

[-- Attachment #1: Type: text/plain, Size: 355 bytes --]

On Wed, May 05, 2021 at 12:54:09PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> Hi Stefan!
> 
> Note my "[PATCH v2 0/9] block: refactor write threshold", it's a kind of counter-proposal for first half of this series.

Thanks! Emanuele mentioned he will drop his patches.

I took a look at your series and that approach looks good to me.

Stefan

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [PATCH v2 1/8] block: prepare write threshold code for thread safety

[Vladimir Sementsov-Ogievskiy]

06.05.2021 12:04, Stefan Hajnoczi wrote:
> On Wed, May 05, 2021 at 12:54:09PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>> Hi Stefan!
>>
>> Note my "[PATCH v2 0/9] block: refactor write threshold", it's a kind of counter-proposal for first half of this series.
> 
> Thanks! Emanuele mentioned he will drop his patches.
> 
> I took a look at your series and that approach looks good to me.
> 

Note, I've sent "[PATCH v3 0/8] block: refactor write threshold" a moment ago.


-- 
Best regards,
Vladimir