[Qemu-devel] [PATCH 0/2] block: introduce and use aio_bh_schedule_oneshot

[Qemu-devel] [PATCH 0/2] block: introduce and use aio_bh_schedule_oneshot

[Paolo Bonzini]

This simplifies a bit using the bottom half API in the common
case of one-shot bottom halves, that are created once per usage.
This patch comes from the multiqueue series.

Paolo

Paolo Bonzini (2):
  async: add aio_bh_schedule_oneshot
  block: use aio_bh_schedule_oneshot

 async.c               | 27 +++++++++++++++++++++++----
 block/archipelago.c   |  5 +----
 block/blkdebug.c      |  7 +------
 block/blkverify.c     |  8 ++------
 block/block-backend.c | 23 +++++++----------------
 block/curl.c          |  7 +------
 block/gluster.c       |  6 +-----
 block/io.c            | 11 +++--------
 block/iscsi.c         |  7 ++-----
 block/nfs.c           |  7 ++-----
 block/null.c          |  5 +----
 block/qed.c           |  6 ++----
 block/qed.h           |  1 -
 block/rbd.c           |  8 ++------
 blockjob.c            |  7 ++-----
 include/block/aio.h   | 10 ++++++++++
 16 files changed, 60 insertions(+), 85 deletions(-)

-- 
2.7.4

[Qemu-devel] [PATCH 1/2] async: add aio_bh_schedule_oneshot

[Paolo Bonzini]

qemu_bh_delete is already clearing bh->scheduled at the same time
as it's setting bh->deleted.  Since it's not using any memory
barriers, there is no synchronization going on for bh->deleted,
and this makes the bh->deleted checks superfluous in aio_compute_timeout,
aio_bh_poll and aio_ctx_check.

Just remove them, and put the (bh->scheduled && bh->deleted) combo
to work in a new function aio_bh_schedule_oneshot.  The new function
removes the need to save the QEMUBH pointer between the creation
and the execution of the bottom half.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 async.c             | 27 +++++++++++++++++++++++----
 include/block/aio.h | 10 ++++++++++
 2 files changed, 33 insertions(+), 4 deletions(-)

diff --git a/async.c b/async.c
index 3bca9b0..f30d011 100644
--- a/async.c
+++ b/async.c
@@ -44,6 +44,25 @@ struct QEMUBH {
     bool deleted;
 };
 
+void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
+{
+    QEMUBH *bh;
+    bh = g_new(QEMUBH, 1);
+    *bh = (QEMUBH){
+        .ctx = ctx,
+        .cb = cb,
+        .opaque = opaque,
+    };
+    qemu_mutex_lock(&ctx->bh_lock);
+    bh->next = ctx->first_bh;
+    bh->scheduled = 1;
+    bh->deleted = 1;
+    /* Make sure that the members are ready before putting bh into list */
+    smp_wmb();
+    ctx->first_bh = bh;
+    qemu_mutex_unlock(&ctx->bh_lock);
+}
+
 QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
 {
     QEMUBH *bh;
@@ -86,7 +105,7 @@ int aio_bh_poll(AioContext *ctx)
          * thread sees the zero before bh->cb has run, and thus will call
          * aio_notify again if necessary.
          */
-        if (!bh->deleted && atomic_xchg(&bh->scheduled, 0)) {
+        if (atomic_xchg(&bh->scheduled, 0)) {
             /* Idle BHs and the notify BH don't count as progress */
             if (!bh->idle && bh != ctx->notify_dummy_bh) {
                 ret = 1;
@@ -104,7 +123,7 @@ int aio_bh_poll(AioContext *ctx)
         bhp = &ctx->first_bh;
         while (*bhp) {
             bh = *bhp;
-            if (bh->deleted) {
+            if (bh->deleted && !bh->scheduled) {
                 *bhp = bh->next;
                 g_free(bh);
             } else {
@@ -168,7 +187,7 @@ aio_compute_timeout(AioContext *ctx)
     QEMUBH *bh;
 
     for (bh = ctx->first_bh; bh; bh = bh->next) {
-        if (!bh->deleted && bh->scheduled) {
+        if (bh->scheduled) {
             if (bh->idle) {
                 /* idle bottom halves will be polled at least
                  * every 10ms */
@@ -216,7 +235,7 @@ aio_ctx_check(GSource *source)
     aio_notify_accept(ctx);
 
     for (bh = ctx->first_bh; bh; bh = bh->next) {
-        if (!bh->deleted && bh->scheduled) {
+        if (bh->scheduled) {
             return true;
         }
     }
diff --git a/include/block/aio.h b/include/block/aio.h
index 173c1ed..be7dd3b 100644
--- a/include/block/aio.h
+++ b/include/block/aio.h
@@ -181,6 +181,16 @@ void aio_context_acquire(AioContext *ctx);
 void aio_context_release(AioContext *ctx);
 
 /**
+ * aio_bh_schedule_oneshot: Allocate a new bottom half structure that will run
+ * only once and as soon as possible.
+ *
+ * Bottom halves are lightweight callbacks whose invocation is guaranteed
+ * to be wait-free, thread-safe and signal-safe.  The #QEMUBH structure
+ * is opaque and must be allocated prior to its use.
+ */
+void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque);
+
+/**
  * aio_bh_new: Allocate a new bottom half structure.
  *
  * Bottom halves are lightweight callbacks whose invocation is guaranteed
-- 
2.7.4

[Qemu-devel] [PATCH 2/2] block: use aio_bh_schedule_oneshot

[Paolo Bonzini]

This simplifies bottom half handlers by removing calls to qemu_bh_delete and
thus removing the need to stash the bottom half pointer in the opaque
datum.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 block/archipelago.c   |  5 +----
 block/blkdebug.c      |  7 +------
 block/blkverify.c     |  8 ++------
 block/block-backend.c | 23 +++++++----------------
 block/curl.c          |  7 +------
 block/gluster.c       |  6 +-----
 block/io.c            | 11 +++--------
 block/iscsi.c         |  7 ++-----
 block/nfs.c           |  7 ++-----
 block/null.c          |  5 +----
 block/qed.c           |  6 ++----
 block/qed.h           |  1 -
 block/rbd.c           |  8 ++------
 blockjob.c            |  7 ++-----
 14 files changed, 27 insertions(+), 81 deletions(-)

diff --git a/block/archipelago.c b/block/archipelago.c
index 37b8aca..2449cfc 100644
--- a/block/archipelago.c
+++ b/block/archipelago.c
@@ -87,7 +87,6 @@ typedef enum {
 
 typedef struct ArchipelagoAIOCB {
     BlockAIOCB common;
-    QEMUBH *bh;
     struct BDRVArchipelagoState *s;
     QEMUIOVector *qiov;
     ARCHIPCmd cmd;
@@ -154,11 +153,10 @@ static void archipelago_finish_aiocb(AIORequestData *reqdata)
     } else if (reqdata->aio_cb->ret == reqdata->segreq->total) {
         reqdata->aio_cb->ret = 0;
     }
-    reqdata->aio_cb->bh = aio_bh_new(
+    aio_bh_schedule_oneshot(
                         bdrv_get_aio_context(reqdata->aio_cb->common.bs),
                         qemu_archipelago_complete_aio, reqdata
                         );
-    qemu_bh_schedule(reqdata->aio_cb->bh);
 }
 
 static int wait_reply(struct xseg *xseg, xport srcport, struct xseg_port *port,
@@ -313,7 +311,6 @@ static void qemu_archipelago_complete_aio(void *opaque)
     AIORequestData *reqdata = (AIORequestData *) opaque;
     ArchipelagoAIOCB *aio_cb = (ArchipelagoAIOCB *) reqdata->aio_cb;
 
-    qemu_bh_delete(aio_cb->bh);
     aio_cb->common.cb(aio_cb->common.opaque, aio_cb->ret);
     aio_cb->status = 0;
 
diff --git a/block/blkdebug.c b/block/blkdebug.c
index d5db166..4127571 100644
--- a/block/blkdebug.c
+++ b/block/blkdebug.c
@@ -49,7 +49,6 @@ typedef struct BDRVBlkdebugState {
 
 typedef struct BlkdebugAIOCB {
     BlockAIOCB common;
-    QEMUBH *bh;
     int ret;
 } BlkdebugAIOCB;
 
@@ -410,7 +409,6 @@ out:
 static void error_callback_bh(void *opaque)
 {
     struct BlkdebugAIOCB *acb = opaque;
-    qemu_bh_delete(acb->bh);
     acb->common.cb(acb->common.opaque, acb->ret);
     qemu_aio_unref(acb);
 }
@@ -421,7 +419,6 @@ static BlockAIOCB *inject_error(BlockDriverState *bs,
     BDRVBlkdebugState *s = bs->opaque;
     int error = rule->options.inject.error;
     struct BlkdebugAIOCB *acb;
-    QEMUBH *bh;
     bool immediately = rule->options.inject.immediately;
 
     if (rule->options.inject.once) {
@@ -436,9 +433,7 @@ static BlockAIOCB *inject_error(BlockDriverState *bs,
     acb = qemu_aio_get(&blkdebug_aiocb_info, bs, cb, opaque);
     acb->ret = -error;
 
-    bh = aio_bh_new(bdrv_get_aio_context(bs), error_callback_bh, acb);
-    acb->bh = bh;
-    qemu_bh_schedule(bh);
+    aio_bh_schedule_oneshot(bdrv_get_aio_context(bs), error_callback_bh, acb);
 
     return &acb->common;
 }
diff --git a/block/blkverify.c b/block/blkverify.c
index da62d75..28f9af6 100644
--- a/block/blkverify.c
+++ b/block/blkverify.c
@@ -22,7 +22,6 @@ typedef struct {
 typedef struct BlkverifyAIOCB BlkverifyAIOCB;
 struct BlkverifyAIOCB {
     BlockAIOCB common;
-    QEMUBH *bh;
 
     /* Request metadata */
     bool is_write;
@@ -175,7 +174,6 @@ static BlkverifyAIOCB *blkverify_aio_get(BlockDriverState *bs, bool is_write,
 {
     BlkverifyAIOCB *acb = qemu_aio_get(&blkverify_aiocb_info, bs, cb, opaque);
 
-    acb->bh = NULL;
     acb->is_write = is_write;
     acb->sector_num = sector_num;
     acb->nb_sectors = nb_sectors;
@@ -191,7 +189,6 @@ static void blkverify_aio_bh(void *opaque)
 {
     BlkverifyAIOCB *acb = opaque;
 
-    qemu_bh_delete(acb->bh);
     if (acb->buf) {
         qemu_iovec_destroy(&acb->raw_qiov);
         qemu_vfree(acb->buf);
@@ -218,9 +215,8 @@ static void blkverify_aio_cb(void *opaque, int ret)
             acb->verify(acb);
         }
 
-        acb->bh = aio_bh_new(bdrv_get_aio_context(acb->common.bs),
-                             blkverify_aio_bh, acb);
-        qemu_bh_schedule(acb->bh);
+        aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
+                                blkverify_aio_bh, acb);
         break;
     }
 }
diff --git a/block/block-backend.c b/block/block-backend.c
index 0bd19ab..37595d2 100644
--- a/block/block-backend.c
+++ b/block/block-backend.c
@@ -65,7 +65,6 @@ struct BlockBackend {
 
 typedef struct BlockBackendAIOCB {
     BlockAIOCB common;
-    QEMUBH *bh;
     BlockBackend *blk;
     int ret;
 } BlockBackendAIOCB;
@@ -898,7 +897,6 @@ int blk_make_zero(BlockBackend *blk, BdrvRequestFlags flags)
 static void error_callback_bh(void *opaque)
 {
     struct BlockBackendAIOCB *acb = opaque;
-    qemu_bh_delete(acb->bh);
     acb->common.cb(acb->common.opaque, acb->ret);
     qemu_aio_unref(acb);
 }
@@ -908,16 +906,12 @@ BlockAIOCB *blk_abort_aio_request(BlockBackend *blk,
                                   void *opaque, int ret)
 {
     struct BlockBackendAIOCB *acb;
-    QEMUBH *bh;
 
     acb = blk_aio_get(&block_backend_aiocb_info, blk, cb, opaque);
     acb->blk = blk;
     acb->ret = ret;
 
-    bh = aio_bh_new(blk_get_aio_context(blk), error_callback_bh, acb);
-    acb->bh = bh;
-    qemu_bh_schedule(bh);
-
+    aio_bh_schedule_oneshot(blk_get_aio_context(blk), error_callback_bh, acb);
     return &acb->common;
 }
 
@@ -926,7 +920,6 @@ typedef struct BlkAioEmAIOCB {
     BlkRwCo rwco;
     int bytes;
     bool has_returned;
-    QEMUBH* bh;
 } BlkAioEmAIOCB;
 
 static const AIOCBInfo blk_aio_em_aiocb_info = {
@@ -935,10 +928,6 @@ static const AIOCBInfo blk_aio_em_aiocb_info = {
 
 static void blk_aio_complete(BlkAioEmAIOCB *acb)
 {
-    if (acb->bh) {
-        assert(acb->has_returned);
-        qemu_bh_delete(acb->bh);
-    }
     if (acb->has_returned) {
         acb->common.cb(acb->common.opaque, acb->rwco.ret);
         qemu_aio_unref(acb);
@@ -947,7 +936,10 @@ static void blk_aio_complete(BlkAioEmAIOCB *acb)
 
 static void blk_aio_complete_bh(void *opaque)
 {
-    blk_aio_complete(opaque);
+    BlkAioEmAIOCB *acb = opaque;
+
+    assert(acb->has_returned);
+    blk_aio_complete(acb);
 }
 
 static BlockAIOCB *blk_aio_prwv(BlockBackend *blk, int64_t offset, int bytes,
@@ -967,7 +959,6 @@ static BlockAIOCB *blk_aio_prwv(BlockBackend *blk, int64_t offset, int bytes,
         .ret    = NOT_DONE,
     };
     acb->bytes = bytes;
-    acb->bh = NULL;
     acb->has_returned = false;
 
     co = qemu_coroutine_create(co_entry, acb);
@@ -975,8 +966,8 @@ static BlockAIOCB *blk_aio_prwv(BlockBackend *blk, int64_t offset, int bytes,
 
     acb->has_returned = true;
     if (acb->rwco.ret != NOT_DONE) {
-        acb->bh = aio_bh_new(blk_get_aio_context(blk), blk_aio_complete_bh, acb);
-        qemu_bh_schedule(acb->bh);
+        aio_bh_schedule_oneshot(blk_get_aio_context(blk),
+                                blk_aio_complete_bh, acb);
     }
 
     return &acb->common;
diff --git a/block/curl.c b/block/curl.c
index 571f24c..e5eaa7b 100644
--- a/block/curl.c
+++ b/block/curl.c
@@ -96,7 +96,6 @@ struct BDRVCURLState;
 
 typedef struct CURLAIOCB {
     BlockAIOCB common;
-    QEMUBH *bh;
     QEMUIOVector *qiov;
 
     int64_t sector_num;
@@ -739,9 +738,6 @@ static void curl_readv_bh_cb(void *p)
     CURLAIOCB *acb = p;
     BDRVCURLState *s = acb->common.bs->opaque;
 
-    qemu_bh_delete(acb->bh);
-    acb->bh = NULL;
-
     size_t start = acb->sector_num * SECTOR_SIZE;
     size_t end;
 
@@ -805,8 +801,7 @@ static BlockAIOCB *curl_aio_readv(BlockDriverState *bs,
     acb->sector_num = sector_num;
     acb->nb_sectors = nb_sectors;
 
-    acb->bh = aio_bh_new(bdrv_get_aio_context(bs), curl_readv_bh_cb, acb);
-    qemu_bh_schedule(acb->bh);
+    aio_bh_schedule_oneshot(bdrv_get_aio_context(bs), curl_readv_bh_cb, acb);
     return &acb->common;
 }
 
diff --git a/block/gluster.c b/block/gluster.c
index e7bd13c..af76d7d5 100644
--- a/block/gluster.c
+++ b/block/gluster.c
@@ -38,7 +38,6 @@
 typedef struct GlusterAIOCB {
     int64_t size;
     int ret;
-    QEMUBH *bh;
     Coroutine *coroutine;
     AioContext *aio_context;
 } GlusterAIOCB;
@@ -622,8 +621,6 @@ static void qemu_gluster_complete_aio(void *opaque)
 {
     GlusterAIOCB *acb = (GlusterAIOCB *)opaque;
 
-    qemu_bh_delete(acb->bh);
-    acb->bh = NULL;
     qemu_coroutine_enter(acb->coroutine);
 }
 
@@ -642,8 +639,7 @@ static void gluster_finish_aiocb(struct glfs_fd *fd, ssize_t ret, void *arg)
         acb->ret = -EIO; /* Partial read/write - fail it */
     }
 
-    acb->bh = aio_bh_new(acb->aio_context, qemu_gluster_complete_aio, acb);
-    qemu_bh_schedule(acb->bh);
+    aio_bh_schedule_oneshot(acb->aio_context, qemu_gluster_complete_aio, acb);
 }
 
 static void qemu_gluster_parse_flags(int bdrv_flags, int *open_flags)
diff --git a/block/io.c b/block/io.c
index fdf7080..c0a5410 100644
--- a/block/io.c
+++ b/block/io.c
@@ -171,7 +171,6 @@ static void bdrv_drain_recurse(BlockDriverState *bs)
 typedef struct {
     Coroutine *co;
     BlockDriverState *bs;
-    QEMUBH *bh;
     bool done;
 } BdrvCoDrainData;
 
@@ -191,7 +190,6 @@ static void bdrv_co_drain_bh_cb(void *opaque)
     BdrvCoDrainData *data = opaque;
     Coroutine *co = data->co;
 
-    qemu_bh_delete(data->bh);
     bdrv_drain_poll(data->bs);
     data->done = true;
     qemu_coroutine_enter(co);
@@ -210,9 +208,9 @@ static void coroutine_fn bdrv_co_yield_to_drain(BlockDriverState *bs)
         .co = qemu_coroutine_self(),
         .bs = bs,
         .done = false,
-        .bh = aio_bh_new(bdrv_get_aio_context(bs), bdrv_co_drain_bh_cb, &data),
     };
-    qemu_bh_schedule(data.bh);
+    aio_bh_schedule_oneshot(bdrv_get_aio_context(bs),
+                            bdrv_co_drain_bh_cb, &data);
 
     qemu_coroutine_yield();
     /* If we are resumed from some other event (such as an aio completion or a
@@ -2070,7 +2068,6 @@ typedef struct BlockAIOCBCoroutine {
     bool is_write;
     bool need_bh;
     bool *done;
-    QEMUBH* bh;
 } BlockAIOCBCoroutine;
 
 static const AIOCBInfo bdrv_em_co_aiocb_info = {
@@ -2090,7 +2087,6 @@ static void bdrv_co_em_bh(void *opaque)
     BlockAIOCBCoroutine *acb = opaque;
 
     assert(!acb->need_bh);
-    qemu_bh_delete(acb->bh);
     bdrv_co_complete(acb);
 }
 
@@ -2100,8 +2096,7 @@ static void bdrv_co_maybe_schedule_bh(BlockAIOCBCoroutine *acb)
     if (acb->req.error != -EINPROGRESS) {
         BlockDriverState *bs = acb->common.bs;
 
-        acb->bh = aio_bh_new(bdrv_get_aio_context(bs), bdrv_co_em_bh, acb);
-        qemu_bh_schedule(acb->bh);
+        aio_bh_schedule_oneshot(bdrv_get_aio_context(bs), bdrv_co_em_bh, acb);
     }
 }
 
diff --git a/block/iscsi.c b/block/iscsi.c
index dff548a..46ddc35 100644
--- a/block/iscsi.c
+++ b/block/iscsi.c
@@ -95,7 +95,6 @@ typedef struct IscsiTask {
     int do_retry;
     struct scsi_task *task;
     Coroutine *co;
-    QEMUBH *bh;
     IscsiLun *iscsilun;
     QEMUTimer retry_timer;
     int err_code;
@@ -167,7 +166,6 @@ static void iscsi_co_generic_bh_cb(void *opaque)
 {
     struct IscsiTask *iTask = opaque;
     iTask->complete = 1;
-    qemu_bh_delete(iTask->bh);
     qemu_coroutine_enter(iTask->co);
 }
 
@@ -299,9 +297,8 @@ iscsi_co_generic_cb(struct iscsi_context *iscsi, int status,
 
 out:
     if (iTask->co) {
-        iTask->bh = aio_bh_new(iTask->iscsilun->aio_context,
-                               iscsi_co_generic_bh_cb, iTask);
-        qemu_bh_schedule(iTask->bh);
+        aio_bh_schedule_oneshot(iTask->iscsilun->aio_context,
+                                 iscsi_co_generic_bh_cb, iTask);
     } else {
         iTask->complete = 1;
     }
diff --git a/block/nfs.c b/block/nfs.c
index 8602a44..c3db2ec 100644
--- a/block/nfs.c
+++ b/block/nfs.c
@@ -57,7 +57,6 @@ typedef struct NFSRPC {
     QEMUIOVector *iov;
     struct stat *st;
     Coroutine *co;
-    QEMUBH *bh;
     NFSClient *client;
 } NFSRPC;
 
@@ -103,7 +102,6 @@ static void nfs_co_generic_bh_cb(void *opaque)
 {
     NFSRPC *task = opaque;
     task->complete = 1;
-    qemu_bh_delete(task->bh);
     qemu_coroutine_enter(task->co);
 }
 
@@ -127,9 +125,8 @@ nfs_co_generic_cb(int ret, struct nfs_context *nfs, void *data,
         error_report("NFS Error: %s", nfs_get_error(nfs));
     }
     if (task->co) {
-        task->bh = aio_bh_new(task->client->aio_context,
-                              nfs_co_generic_bh_cb, task);
-        qemu_bh_schedule(task->bh);
+        aio_bh_schedule_oneshot(task->client->aio_context,
+                                nfs_co_generic_bh_cb, task);
     } else {
         task->complete = 1;
     }
diff --git a/block/null.c b/block/null.c
index b511010..b300390 100644
--- a/block/null.c
+++ b/block/null.c
@@ -124,7 +124,6 @@ static coroutine_fn int null_co_flush(BlockDriverState *bs)
 
 typedef struct {
     BlockAIOCB common;
-    QEMUBH *bh;
     QEMUTimer timer;
 } NullAIOCB;
 
@@ -136,7 +135,6 @@ static void null_bh_cb(void *opaque)
 {
     NullAIOCB *acb = opaque;
     acb->common.cb(acb->common.opaque, 0);
-    qemu_bh_delete(acb->bh);
     qemu_aio_unref(acb);
 }
 
@@ -164,8 +162,7 @@ static inline BlockAIOCB *null_aio_common(BlockDriverState *bs,
         timer_mod_ns(&acb->timer,
                      qemu_clock_get_ns(QEMU_CLOCK_REALTIME) + s->latency_ns);
     } else {
-        acb->bh = aio_bh_new(bdrv_get_aio_context(bs), null_bh_cb, acb);
-        qemu_bh_schedule(acb->bh);
+        aio_bh_schedule_oneshot(bdrv_get_aio_context(bs), null_bh_cb, acb);
     }
     return &acb->common;
 }
diff --git a/block/qed.c b/block/qed.c
index 426f3cb..3ee879b 100644
--- a/block/qed.c
+++ b/block/qed.c
@@ -909,7 +909,6 @@ static void qed_aio_complete_bh(void *opaque)
     void *user_opaque = acb->common.opaque;
     int ret = acb->bh_ret;
 
-    qemu_bh_delete(acb->bh);
     qemu_aio_unref(acb);
 
     /* Invoke callback */
@@ -934,9 +933,8 @@ static void qed_aio_complete(QEDAIOCB *acb, int ret)
 
     /* Arrange for a bh to invoke the completion function */
     acb->bh_ret = ret;
-    acb->bh = aio_bh_new(bdrv_get_aio_context(acb->common.bs),
-                         qed_aio_complete_bh, acb);
-    qemu_bh_schedule(acb->bh);
+    aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
+                            qed_aio_complete_bh, acb);
 
     /* Start next allocating write request waiting behind this one.  Note that
      * requests enqueue themselves when they first hit an unallocated cluster
diff --git a/block/qed.h b/block/qed.h
index 22b3198..9676ab9 100644
--- a/block/qed.h
+++ b/block/qed.h
@@ -130,7 +130,6 @@ enum {
 
 typedef struct QEDAIOCB {
     BlockAIOCB common;
-    QEMUBH *bh;
     int bh_ret;                     /* final return status for completion bh */
     QSIMPLEQ_ENTRY(QEDAIOCB) next;  /* next request */
     int flags;                      /* QED_AIOCB_* bits ORed together */
diff --git a/block/rbd.c b/block/rbd.c
index 0106fea..6f9eb6f 100644
--- a/block/rbd.c
+++ b/block/rbd.c
@@ -71,7 +71,6 @@ typedef enum {
 
 typedef struct RBDAIOCB {
     BlockAIOCB common;
-    QEMUBH *bh;
     int64_t ret;
     QEMUIOVector *qiov;
     char *bounce;
@@ -602,7 +601,6 @@ static const AIOCBInfo rbd_aiocb_info = {
 static void rbd_finish_bh(void *opaque)
 {
     RADOSCB *rcb = opaque;
-    qemu_bh_delete(rcb->acb->bh);
     qemu_rbd_complete_aio(rcb);
 }
 
@@ -621,9 +619,8 @@ static void rbd_finish_aiocb(rbd_completion_t c, RADOSCB *rcb)
     rcb->ret = rbd_aio_get_return_value(c);
     rbd_aio_release(c);
 
-    acb->bh = aio_bh_new(bdrv_get_aio_context(acb->common.bs),
-                         rbd_finish_bh, rcb);
-    qemu_bh_schedule(acb->bh);
+    aio_bh_schedule_oneshot(bdrv_get_aio_context(acb->common.bs),
+                            rbd_finish_bh, rcb);
 }
 
 static int rbd_aio_discard_wrapper(rbd_image_t image,
@@ -679,7 +676,6 @@ static BlockAIOCB *rbd_start_aio(BlockDriverState *bs,
     acb->ret = 0;
     acb->error = 0;
     acb->s = s;
-    acb->bh = NULL;
 
     if (cmd == RBD_AIO_WRITE) {
         qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);
diff --git a/blockjob.c b/blockjob.c
index a167f96..43fecbe 100644
--- a/blockjob.c
+++ b/blockjob.c
@@ -588,7 +588,6 @@ BlockErrorAction block_job_error_action(BlockJob *job, BlockdevOnError on_err,
 
 typedef struct {
     BlockJob *job;
-    QEMUBH *bh;
     AioContext *aio_context;
     BlockJobDeferToMainLoopFn *fn;
     void *opaque;
@@ -599,8 +598,6 @@ static void block_job_defer_to_main_loop_bh(void *opaque)
     BlockJobDeferToMainLoopData *data = opaque;
     AioContext *aio_context;
 
-    qemu_bh_delete(data->bh);
-
     /* Prevent race with block_job_defer_to_main_loop() */
     aio_context_acquire(data->aio_context);
 
@@ -624,13 +621,13 @@ void block_job_defer_to_main_loop(BlockJob *job,
 {
     BlockJobDeferToMainLoopData *data = g_malloc(sizeof(*data));
     data->job = job;
-    data->bh = qemu_bh_new(block_job_defer_to_main_loop_bh, data);
     data->aio_context = blk_get_aio_context(job->blk);
     data->fn = fn;
     data->opaque = opaque;
     job->deferred_to_main_loop = true;
 
-    qemu_bh_schedule(data->bh);
+    aio_bh_schedule_oneshot(qemu_get_aio_context(),
+                            block_job_defer_to_main_loop_bh, data);
 }
 
 BlockJobTxn *block_job_txn_new(void)
-- 
2.7.4

Re: [Qemu-devel] [PATCH 0/2] block: introduce and use aio_bh_schedule_oneshot

[Kevin Wolf]

Am 03.10.2016 um 18:14 hat Paolo Bonzini geschrieben:
> This simplifies a bit using the bottom half API in the common
> case of one-shot bottom halves, that are created once per usage.
> This patch comes from the multiqueue series.

Thanks, applied to the block branch.

Kevin

Re: [Qemu-devel] [PATCH 1/2] async: add aio_bh_schedule_oneshot

[Stefan Hajnoczi]

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

On Mon, Oct 03, 2016 at 06:14:15PM +0200, Paolo Bonzini wrote:
> qemu_bh_delete is already clearing bh->scheduled at the same time
> as it's setting bh->deleted.  Since it's not using any memory
> barriers, there is no synchronization going on for bh->deleted,
> and this makes the bh->deleted checks superfluous in aio_compute_timeout,
> aio_bh_poll and aio_ctx_check.

Yikes.  On one hand this sounds scary but in practice qemu_bh_delete()
isn't called from another thread so the next aio_bh_poll() will indeed
clean it up instead of dispatching a deleted BH.

Due to the nature of this change I suggest making it in a separate
patch.

> Just remove them, and put the (bh->scheduled && bh->deleted) combo
> to work in a new function aio_bh_schedule_oneshot.  The new function
> removes the need to save the QEMUBH pointer between the creation
> and the execution of the bottom half.
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  async.c             | 27 +++++++++++++++++++++++----
>  include/block/aio.h | 10 ++++++++++
>  2 files changed, 33 insertions(+), 4 deletions(-)
> 
> diff --git a/async.c b/async.c
> index 3bca9b0..f30d011 100644
> --- a/async.c
> +++ b/async.c
> @@ -44,6 +44,25 @@ struct QEMUBH {
>      bool deleted;
>  };
>  
> +void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
> +{
> +    QEMUBH *bh;
> +    bh = g_new(QEMUBH, 1);
> +    *bh = (QEMUBH){
> +        .ctx = ctx,
> +        .cb = cb,
> +        .opaque = opaque,
> +    };
> +    qemu_mutex_lock(&ctx->bh_lock);
> +    bh->next = ctx->first_bh;
> +    bh->scheduled = 1;
> +    bh->deleted = 1;
> +    /* Make sure that the members are ready before putting bh into list */
> +    smp_wmb();
> +    ctx->first_bh = bh;
> +    qemu_mutex_unlock(&ctx->bh_lock);
> +}
> +
>  QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
>  {
>      QEMUBH *bh;
> @@ -86,7 +105,7 @@ int aio_bh_poll(AioContext *ctx)
>           * thread sees the zero before bh->cb has run, and thus will call
>           * aio_notify again if necessary.
>           */
> -        if (!bh->deleted && atomic_xchg(&bh->scheduled, 0)) {
> +        if (atomic_xchg(&bh->scheduled, 0)) {
>              /* Idle BHs and the notify BH don't count as progress */
>              if (!bh->idle && bh != ctx->notify_dummy_bh) {
>                  ret = 1;
> @@ -104,7 +123,7 @@ int aio_bh_poll(AioContext *ctx)
>          bhp = &ctx->first_bh;
>          while (*bhp) {
>              bh = *bhp;
> -            if (bh->deleted) {
> +            if (bh->deleted && !bh->scheduled) {
>                  *bhp = bh->next;
>                  g_free(bh);
>              } else {
> @@ -168,7 +187,7 @@ aio_compute_timeout(AioContext *ctx)
>      QEMUBH *bh;
>  
>      for (bh = ctx->first_bh; bh; bh = bh->next) {
> -        if (!bh->deleted && bh->scheduled) {
> +        if (bh->scheduled) {
>              if (bh->idle) {
>                  /* idle bottom halves will be polled at least
>                   * every 10ms */
> @@ -216,7 +235,7 @@ aio_ctx_check(GSource *source)
>      aio_notify_accept(ctx);
>  
>      for (bh = ctx->first_bh; bh; bh = bh->next) {
> -        if (!bh->deleted && bh->scheduled) {
> +        if (bh->scheduled) {
>              return true;
>          }
>      }
> diff --git a/include/block/aio.h b/include/block/aio.h
> index 173c1ed..be7dd3b 100644
> --- a/include/block/aio.h
> +++ b/include/block/aio.h
> @@ -181,6 +181,16 @@ void aio_context_acquire(AioContext *ctx);
>  void aio_context_release(AioContext *ctx);
>  
>  /**
> + * aio_bh_schedule_oneshot: Allocate a new bottom half structure that will run
> + * only once and as soon as possible.
> + *
> + * Bottom halves are lightweight callbacks whose invocation is guaranteed
> + * to be wait-free, thread-safe and signal-safe.  The #QEMUBH structure
> + * is opaque and must be allocated prior to its use.

I'm confused.  There is no QEMUBH structure in this function
prototype.  Is this comment from an earlier version of this function?

> + */
> +void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque);
> +
> +/**
>   * aio_bh_new: Allocate a new bottom half structure.
>   *
>   * Bottom halves are lightweight callbacks whose invocation is guaranteed
> -- 
> 2.7.4
> 
> 

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

Re: [Qemu-devel] [PATCH 2/2] block: use aio_bh_schedule_oneshot

[Stefan Hajnoczi]

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

On Mon, Oct 03, 2016 at 06:14:16PM +0200, Paolo Bonzini wrote:
> This simplifies bottom half handlers by removing calls to qemu_bh_delete and
> thus removing the need to stash the bottom half pointer in the opaque
> datum.
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  block/archipelago.c   |  5 +----
>  block/blkdebug.c      |  7 +------
>  block/blkverify.c     |  8 ++------
>  block/block-backend.c | 23 +++++++----------------
>  block/curl.c          |  7 +------
>  block/gluster.c       |  6 +-----
>  block/io.c            | 11 +++--------
>  block/iscsi.c         |  7 ++-----
>  block/nfs.c           |  7 ++-----
>  block/null.c          |  5 +----
>  block/qed.c           |  6 ++----
>  block/qed.h           |  1 -
>  block/rbd.c           |  8 ++------
>  blockjob.c            |  7 ++-----
>  14 files changed, 27 insertions(+), 81 deletions(-)

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

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

Re: [Qemu-devel] [PATCH 1/2] async: add aio_bh_schedule_oneshot

[Paolo Bonzini]

On 05/10/2016 15:13, Stefan Hajnoczi wrote:
> > qemu_bh_delete is already clearing bh->scheduled at the same time
> > as it's setting bh->deleted.  Since it's not using any memory
> > barriers, there is no synchronization going on for bh->deleted,
> > and this makes the bh->deleted checks superfluous in aio_compute_timeout,
> > aio_bh_poll and aio_ctx_check.
> 
> Yikes.  On one hand this sounds scary but in practice qemu_bh_delete()
> isn't called from another thread so the next aio_bh_poll() will indeed
> clean it up instead of dispatching a deleted BH.
> 
> Due to the nature of this change I suggest making it in a separate
> patch.

Separate from what?  (Sorry if I'm being dense).

>>
>> + * aio_bh_schedule_oneshot: Allocate a new bottom half structure that will run
>> + * only once and as soon as possible.
>> + *
>> + * Bottom halves are lightweight callbacks whose invocation is guaranteed
>> + * to be wait-free, thread-safe and signal-safe.  The #QEMUBH structure
>> + * is opaque and must be allocated prior to its use.
> 
> I'm confused.  There is no QEMUBH structure in this function
> prototype.  Is this comment from an earlier version of this function?

No, it's from aio_bh_new.  Of course this one is neither wait-free nor
signal-safe.  Kevin, do you want me to respin?

Paolo

Re: [Qemu-devel] [PATCH 1/2] async: add aio_bh_schedule_oneshot

[Kevin Wolf]

Am 05.10.2016 um 15:55 hat Paolo Bonzini geschrieben:
> On 05/10/2016 15:13, Stefan Hajnoczi wrote:
> > > qemu_bh_delete is already clearing bh->scheduled at the same time
> > > as it's setting bh->deleted.  Since it's not using any memory
> > > barriers, there is no synchronization going on for bh->deleted,
> > > and this makes the bh->deleted checks superfluous in aio_compute_timeout,
> > > aio_bh_poll and aio_ctx_check.
> > 
> > Yikes.  On one hand this sounds scary but in practice qemu_bh_delete()
> > isn't called from another thread so the next aio_bh_poll() will indeed
> > clean it up instead of dispatching a deleted BH.
> > 
> > Due to the nature of this change I suggest making it in a separate
> > patch.
> 
> Separate from what?  (Sorry if I'm being dense).
> 
> >>
> >> + * aio_bh_schedule_oneshot: Allocate a new bottom half structure that will run
> >> + * only once and as soon as possible.
> >> + *
> >> + * Bottom halves are lightweight callbacks whose invocation is guaranteed
> >> + * to be wait-free, thread-safe and signal-safe.  The #QEMUBH structure
> >> + * is opaque and must be allocated prior to its use.
> > 
> > I'm confused.  There is no QEMUBH structure in this function
> > prototype.  Is this comment from an earlier version of this function?
> 
> No, it's from aio_bh_new.  Of course this one is neither wait-free nor
> signal-safe.  Kevin, do you want me to respin?

If the comment is wrong, either post a v2 of this patch or just reply
with a new version of the comment and I'll squash it in. Your choice, I
don't mind either way.

Kevin

Re: [Qemu-devel] [PATCH 1/2] async: add aio_bh_schedule_oneshot

[Paolo Bonzini]

On 05/10/2016 16:20, Kevin Wolf wrote:
> Am 05.10.2016 um 15:55 hat Paolo Bonzini geschrieben:
>> On 05/10/2016 15:13, Stefan Hajnoczi wrote:
>>>> qemu_bh_delete is already clearing bh->scheduled at the same time
>>>> as it's setting bh->deleted.  Since it's not using any memory
>>>> barriers, there is no synchronization going on for bh->deleted,
>>>> and this makes the bh->deleted checks superfluous in aio_compute_timeout,
>>>> aio_bh_poll and aio_ctx_check.
>>>
>>> Yikes.  On one hand this sounds scary but in practice qemu_bh_delete()
>>> isn't called from another thread so the next aio_bh_poll() will indeed
>>> clean it up instead of dispatching a deleted BH.
>>>
>>> Due to the nature of this change I suggest making it in a separate
>>> patch.
>>
>> Separate from what?  (Sorry if I'm being dense).
>>
>>>>
>>>> + * aio_bh_schedule_oneshot: Allocate a new bottom half structure that will run
>>>> + * only once and as soon as possible.
>>>> + *
>>>> + * Bottom halves are lightweight callbacks whose invocation is guaranteed
>>>> + * to be wait-free, thread-safe and signal-safe.  The #QEMUBH structure
>>>> + * is opaque and must be allocated prior to its use.
>>>
>>> I'm confused.  There is no QEMUBH structure in this function
>>> prototype.  Is this comment from an earlier version of this function?
>>
>> No, it's from aio_bh_new.  Of course this one is neither wait-free nor
>> signal-safe.  Kevin, do you want me to respin?
> 
> If the comment is wrong, either post a v2 of this patch or just reply
> with a new version of the comment and I'll squash it in. Your choice, I
> don't mind either way.

Just removing those three lines would be okay.

Paolo

Re: [Qemu-devel] [PATCH 1/2] async: add aio_bh_schedule_oneshot

[Kevin Wolf]

Am 05.10.2016 um 16:25 hat Paolo Bonzini geschrieben:
> 
> 
> On 05/10/2016 16:20, Kevin Wolf wrote:
> > Am 05.10.2016 um 15:55 hat Paolo Bonzini geschrieben:
> >> On 05/10/2016 15:13, Stefan Hajnoczi wrote:
> >>>> qemu_bh_delete is already clearing bh->scheduled at the same time
> >>>> as it's setting bh->deleted.  Since it's not using any memory
> >>>> barriers, there is no synchronization going on for bh->deleted,
> >>>> and this makes the bh->deleted checks superfluous in aio_compute_timeout,
> >>>> aio_bh_poll and aio_ctx_check.
> >>>
> >>> Yikes.  On one hand this sounds scary but in practice qemu_bh_delete()
> >>> isn't called from another thread so the next aio_bh_poll() will indeed
> >>> clean it up instead of dispatching a deleted BH.
> >>>
> >>> Due to the nature of this change I suggest making it in a separate
> >>> patch.
> >>
> >> Separate from what?  (Sorry if I'm being dense).
> >>
> >>>>
> >>>> + * aio_bh_schedule_oneshot: Allocate a new bottom half structure that will run
> >>>> + * only once and as soon as possible.
> >>>> + *
> >>>> + * Bottom halves are lightweight callbacks whose invocation is guaranteed
> >>>> + * to be wait-free, thread-safe and signal-safe.  The #QEMUBH structure
> >>>> + * is opaque and must be allocated prior to its use.
> >>>
> >>> I'm confused.  There is no QEMUBH structure in this function
> >>> prototype.  Is this comment from an earlier version of this function?
> >>
> >> No, it's from aio_bh_new.  Of course this one is neither wait-free nor
> >> signal-safe.  Kevin, do you want me to respin?
> > 
> > If the comment is wrong, either post a v2 of this patch or just reply
> > with a new version of the comment and I'll squash it in. Your choice, I
> > don't mind either way.
> 
> Just removing those three lines would be okay.

Ok, done.

Kevin