Patchwork [V9,03/11] quorum: Add quorum_aio_writev and its dependencies.

login
register
mail settings
Submitter Benoît Canet
Date Oct. 2, 2013, 12:39 p.m.
Message ID <1380717564-11098-4-git-send-email-benoit@irqsave.net>
Download mbox | patch
Permalink /patch/279728/
State New
Headers show

Comments

Benoît Canet - Oct. 2, 2013, 12:39 p.m.
Signed-off-by: Benoit Canet <benoit@irqsave.net>
---
 block/quorum.c | 123 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 123 insertions(+)
Max Reitz - Oct. 4, 2013, 2:35 p.m.
On 2013-10-02 14:39, Benoît Canet wrote:
> Signed-off-by: Benoit Canet <benoit@irqsave.net>
> ---
>   block/quorum.c | 123 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>   1 file changed, 123 insertions(+)
>
> diff --git a/block/quorum.c b/block/quorum.c
> index 9557e61..b49e3c6 100644
> --- a/block/quorum.c
> +++ b/block/quorum.c
> @@ -64,11 +64,134 @@ struct QuorumAIOCB {
>       int vote_ret;
>   };
>   
> +static void quorum_aio_cancel(BlockDriverAIOCB *blockacb)
> +{
> +    QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
> +    bool finished = false;
> +
> +    /* Wait for the request to finish */
> +    acb->finished = &finished;
> +    while (!finished) {
> +        qemu_aio_wait();
> +    }
Hm, wouldn't it be better to pass the cancel to the children?

Max

> +}
> +
> +static AIOCBInfo quorum_aiocb_info = {
> +    .aiocb_size         = sizeof(QuorumAIOCB),
> +    .cancel             = quorum_aio_cancel,
> +};
> +
> +/* return the first error code get by each individual callbacks */
> +static int quorum_get_first_error(QuorumAIOCB *acb)
> +{
> +    BDRVQuorumState *s = acb->bqs;
> +    int i, ret = 0;
> +
> +    for (i = 0; i < s->total; i++) {
> +        ret = acb->aios[i].ret;
> +        if (ret) {
> +            return ret;
> +        }
> +    }
> +
> +    /* should not pass here */
> +    assert(false);
> +}
> +
> +static void quorum_aio_finalize(QuorumAIOCB *acb)
> +{
> +    BDRVQuorumState *s = acb->bqs;
> +    int ret;
> +
> +    ret = s->threshold <= acb->success_count ? 0 : quorum_get_first_error(acb);
> +
> +    acb->common.cb(acb->common.opaque, ret);
> +    if (acb->finished) {
> +        *acb->finished = true;
> +    }
> +    g_free(acb->aios);
> +    qemu_aio_release(acb);
> +}
> +
> +static QuorumAIOCB *quorum_aio_get(BDRVQuorumState *s,
> +                                   BlockDriverState *bs,
> +                                   QEMUIOVector *qiov,
> +                                   uint64_t sector_num,
> +                                   int nb_sectors,
> +                                   BlockDriverCompletionFunc *cb,
> +                                   void *opaque)
> +{
> +    QuorumAIOCB *acb = qemu_aio_get(&quorum_aiocb_info, bs, cb, opaque);
> +    int i;
> +
> +    acb->bqs = s;
> +    acb->sector_num = sector_num;
> +    acb->nb_sectors = nb_sectors;
> +    acb->qiov = qiov;
> +    acb->aios = g_new0(QuorumSingleAIOCB, s->total);
> +    acb->count = 0;
> +    acb->success_count = 0;
> +    acb->finished = NULL;
> +    acb->is_read = false;
> +    acb->vote_ret = 0;
> +
> +    for (i = 0; i < s->total; i++) {
> +        acb->aios[i].buf = NULL;
> +        acb->aios[i].ret = 0;
> +        acb->aios[i].parent = acb;
> +    }
> +
> +    return acb;
> +}
> +
> +static void quorum_aio_cb(void *opaque, int ret)
> +{
> +    QuorumSingleAIOCB *sacb = opaque;
> +    QuorumAIOCB *acb = sacb->parent;
> +    BDRVQuorumState *s = acb->bqs;
> +
> +    sacb->ret = ret;
> +    acb->count++;
> +    if (ret == 0) {
> +        acb->success_count++;
> +    }
> +    assert(acb->count <= s->total);
> +    assert(acb->success_count <= s->total);
> +    if (acb->count < s->total) {
> +        return;
> +    }
> +
> +    quorum_aio_finalize(acb);
> +}
> +
> +static BlockDriverAIOCB *quorum_aio_writev(BlockDriverState *bs,
> +                                          int64_t sector_num,
> +                                          QEMUIOVector *qiov,
> +                                          int nb_sectors,
> +                                          BlockDriverCompletionFunc *cb,
> +                                          void *opaque)
> +{
> +    BDRVQuorumState *s = bs->opaque;
> +    QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num, nb_sectors,
> +                                      cb, opaque);
> +    int i;
> +
> +    for (i = 0; i < s->total; i++) {
> +        acb->aios[i].aiocb = bdrv_aio_writev(&s->bs[i], sector_num, qiov,
> +                                             nb_sectors, &quorum_aio_cb,
> +                                             &acb->aios[i]);
> +    }
> +
> +    return &acb->common;
> +}
> +
>   static BlockDriver bdrv_quorum = {
>       .format_name        = "quorum",
>       .protocol_name      = "quorum",
>   
>       .instance_size      = sizeof(BDRVQuorumState),
> +
> +    .bdrv_aio_writev    = quorum_aio_writev,
>   };
>   
>   static void bdrv_quorum_init(void)
Benoît Canet - Oct. 28, 2013, 12:21 p.m.
Le Friday 04 Oct 2013 à 16:35:18 (+0200), Max Reitz a écrit :
> On 2013-10-02 14:39, Benoît Canet wrote:
> >Signed-off-by: Benoit Canet <benoit@irqsave.net>
> >---
> >  block/quorum.c | 123 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
> >  1 file changed, 123 insertions(+)
> >
> >diff --git a/block/quorum.c b/block/quorum.c
> >index 9557e61..b49e3c6 100644
> >--- a/block/quorum.c
> >+++ b/block/quorum.c
> >@@ -64,11 +64,134 @@ struct QuorumAIOCB {
> >      int vote_ret;
> >  };
> >+static void quorum_aio_cancel(BlockDriverAIOCB *blockacb)
> >+{
> >+    QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
> >+    bool finished = false;
> >+
> >+    /* Wait for the request to finish */
> >+    acb->finished = &finished;
> >+    while (!finished) {
> >+        qemu_aio_wait();
> >+    }
> Hm, wouldn't it be better to pass the cancel to the children?
> 
> Max

Hi Max,

Hi don't understand how you would do this.

Best regards

Benoît

> 
> >+}
> >+
> >+static AIOCBInfo quorum_aiocb_info = {
> >+    .aiocb_size         = sizeof(QuorumAIOCB),
> >+    .cancel             = quorum_aio_cancel,
> >+};
> >+
> >+/* return the first error code get by each individual callbacks */
> >+static int quorum_get_first_error(QuorumAIOCB *acb)
> >+{
> >+    BDRVQuorumState *s = acb->bqs;
> >+    int i, ret = 0;
> >+
> >+    for (i = 0; i < s->total; i++) {
> >+        ret = acb->aios[i].ret;
> >+        if (ret) {
> >+            return ret;
> >+        }
> >+    }
> >+
> >+    /* should not pass here */
> >+    assert(false);
> >+}
> >+
> >+static void quorum_aio_finalize(QuorumAIOCB *acb)
> >+{
> >+    BDRVQuorumState *s = acb->bqs;
> >+    int ret;
> >+
> >+    ret = s->threshold <= acb->success_count ? 0 : quorum_get_first_error(acb);
> >+
> >+    acb->common.cb(acb->common.opaque, ret);
> >+    if (acb->finished) {
> >+        *acb->finished = true;
> >+    }
> >+    g_free(acb->aios);
> >+    qemu_aio_release(acb);
> >+}
> >+
> >+static QuorumAIOCB *quorum_aio_get(BDRVQuorumState *s,
> >+                                   BlockDriverState *bs,
> >+                                   QEMUIOVector *qiov,
> >+                                   uint64_t sector_num,
> >+                                   int nb_sectors,
> >+                                   BlockDriverCompletionFunc *cb,
> >+                                   void *opaque)
> >+{
> >+    QuorumAIOCB *acb = qemu_aio_get(&quorum_aiocb_info, bs, cb, opaque);
> >+    int i;
> >+
> >+    acb->bqs = s;
> >+    acb->sector_num = sector_num;
> >+    acb->nb_sectors = nb_sectors;
> >+    acb->qiov = qiov;
> >+    acb->aios = g_new0(QuorumSingleAIOCB, s->total);
> >+    acb->count = 0;
> >+    acb->success_count = 0;
> >+    acb->finished = NULL;
> >+    acb->is_read = false;
> >+    acb->vote_ret = 0;
> >+
> >+    for (i = 0; i < s->total; i++) {
> >+        acb->aios[i].buf = NULL;
> >+        acb->aios[i].ret = 0;
> >+        acb->aios[i].parent = acb;
> >+    }
> >+
> >+    return acb;
> >+}
> >+
> >+static void quorum_aio_cb(void *opaque, int ret)
> >+{
> >+    QuorumSingleAIOCB *sacb = opaque;
> >+    QuorumAIOCB *acb = sacb->parent;
> >+    BDRVQuorumState *s = acb->bqs;
> >+
> >+    sacb->ret = ret;
> >+    acb->count++;
> >+    if (ret == 0) {
> >+        acb->success_count++;
> >+    }
> >+    assert(acb->count <= s->total);
> >+    assert(acb->success_count <= s->total);
> >+    if (acb->count < s->total) {
> >+        return;
> >+    }
> >+
> >+    quorum_aio_finalize(acb);
> >+}
> >+
> >+static BlockDriverAIOCB *quorum_aio_writev(BlockDriverState *bs,
> >+                                          int64_t sector_num,
> >+                                          QEMUIOVector *qiov,
> >+                                          int nb_sectors,
> >+                                          BlockDriverCompletionFunc *cb,
> >+                                          void *opaque)
> >+{
> >+    BDRVQuorumState *s = bs->opaque;
> >+    QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num, nb_sectors,
> >+                                      cb, opaque);
> >+    int i;
> >+
> >+    for (i = 0; i < s->total; i++) {
> >+        acb->aios[i].aiocb = bdrv_aio_writev(&s->bs[i], sector_num, qiov,
> >+                                             nb_sectors, &quorum_aio_cb,
> >+                                             &acb->aios[i]);
> >+    }
> >+
> >+    return &acb->common;
> >+}
> >+
> >  static BlockDriver bdrv_quorum = {
> >      .format_name        = "quorum",
> >      .protocol_name      = "quorum",
> >      .instance_size      = sizeof(BDRVQuorumState),
> >+
> >+    .bdrv_aio_writev    = quorum_aio_writev,
> >  };
> >  static void bdrv_quorum_init(void)
> 
>
Max Reitz - Oct. 29, 2013, 5:21 p.m.
Am 28.10.2013 13:21, schrieb Benoît Canet:
> Le Friday 04 Oct 2013 à 16:35:18 (+0200), Max Reitz a écrit :
>> On 2013-10-02 14:39, Benoît Canet wrote:
>>> Signed-off-by: Benoit Canet <benoit@irqsave.net>
>>> ---
>>>  block/quorum.c | 123 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>>>  1 file changed, 123 insertions(+)
>>>
>>> diff --git a/block/quorum.c b/block/quorum.c
>>> index 9557e61..b49e3c6 100644
>>> --- a/block/quorum.c
>>> +++ b/block/quorum.c
>>> @@ -64,11 +64,134 @@ struct QuorumAIOCB {
>>>      int vote_ret;
>>>  };
>>> +static void quorum_aio_cancel(BlockDriverAIOCB *blockacb)
>>> +{
>>> +    QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
>>> +    bool finished = false;
>>> +
>>> +    /* Wait for the request to finish */
>>> +    acb->finished = &finished;
>>> +    while (!finished) {
>>> +        qemu_aio_wait();
>>> +    }
>> Hm, wouldn't it be better to pass the cancel to the children?
>>
>> Max
> Hi Max,
>
> Hi don't understand how you would do this.
>
> Best regards
>
> Benoît

I thought of calling bdrv_aio_cancel() on every aios[i].aiocb – I don't
know if that will work but it seems reasonable to me to try to cancel
all the operations quorum has initiated when the quorum operation itself
is requested to be canceled (instead of waiting for the spawned
operations to finish normally).

Max
Benoît Canet - Oct. 29, 2013, 8:06 p.m.
L Tuesday 29 Oct 2013 à 18:21:57 (+0100), Max Reitz a écrit :
> Am 28.10.2013 13:21, schrieb Benoît Canet:
> > Le Friday 04 Oct 2013 à 16:35:18 (+0200), Max Reitz a écrit :
> >> On 2013-10-02 14:39, Benoît Canet wrote:
> >>> Signed-off-by: Benoit Canet <benoit@irqsave.net>
> >>> ---
> >>>  block/quorum.c | 123 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
> >>>  1 file changed, 123 insertions(+)
> >>>
> >>> diff --git a/block/quorum.c b/block/quorum.c
> >>> index 9557e61..b49e3c6 100644
> >>> --- a/block/quorum.c
> >>> +++ b/block/quorum.c
> >>> @@ -64,11 +64,134 @@ struct QuorumAIOCB {
> >>>      int vote_ret;
> >>>  };
> >>> +static void quorum_aio_cancel(BlockDriverAIOCB *blockacb)
> >>> +{
> >>> +    QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
> >>> +    bool finished = false;
> >>> +
> >>> +    /* Wait for the request to finish */
> >>> +    acb->finished = &finished;
> >>> +    while (!finished) {
> >>> +        qemu_aio_wait();
> >>> +    }
> >> Hm, wouldn't it be better to pass the cancel to the children?
> >>
> >> Max
> > Hi Max,
> >
> > Hi don't understand how you would do this.
> >
> > Best regards
> >
> > Benoît
> 
> I thought of calling bdrv_aio_cancel() on every aios[i].aiocb – I don't
> know if that will work but it seems reasonable to me to try to cancel
> all the operations quorum has initiated when the quorum operation itself
> is requested to be canceled (instead of waiting for the spawned
> operations to finish normally).

That makes sense. I will do this.

Best regards

Benoît

> 
> Max

Patch

diff --git a/block/quorum.c b/block/quorum.c
index 9557e61..b49e3c6 100644
--- a/block/quorum.c
+++ b/block/quorum.c
@@ -64,11 +64,134 @@  struct QuorumAIOCB {
     int vote_ret;
 };
 
+static void quorum_aio_cancel(BlockDriverAIOCB *blockacb)
+{
+    QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
+    bool finished = false;
+
+    /* Wait for the request to finish */
+    acb->finished = &finished;
+    while (!finished) {
+        qemu_aio_wait();
+    }
+}
+
+static AIOCBInfo quorum_aiocb_info = {
+    .aiocb_size         = sizeof(QuorumAIOCB),
+    .cancel             = quorum_aio_cancel,
+};
+
+/* return the first error code get by each individual callbacks */
+static int quorum_get_first_error(QuorumAIOCB *acb)
+{
+    BDRVQuorumState *s = acb->bqs;
+    int i, ret = 0;
+
+    for (i = 0; i < s->total; i++) {
+        ret = acb->aios[i].ret;
+        if (ret) {
+            return ret;
+        }
+    }
+
+    /* should not pass here */
+    assert(false);
+}
+
+static void quorum_aio_finalize(QuorumAIOCB *acb)
+{
+    BDRVQuorumState *s = acb->bqs;
+    int ret;
+
+    ret = s->threshold <= acb->success_count ? 0 : quorum_get_first_error(acb);
+
+    acb->common.cb(acb->common.opaque, ret);
+    if (acb->finished) {
+        *acb->finished = true;
+    }
+    g_free(acb->aios);
+    qemu_aio_release(acb);
+}
+
+static QuorumAIOCB *quorum_aio_get(BDRVQuorumState *s,
+                                   BlockDriverState *bs,
+                                   QEMUIOVector *qiov,
+                                   uint64_t sector_num,
+                                   int nb_sectors,
+                                   BlockDriverCompletionFunc *cb,
+                                   void *opaque)
+{
+    QuorumAIOCB *acb = qemu_aio_get(&quorum_aiocb_info, bs, cb, opaque);
+    int i;
+
+    acb->bqs = s;
+    acb->sector_num = sector_num;
+    acb->nb_sectors = nb_sectors;
+    acb->qiov = qiov;
+    acb->aios = g_new0(QuorumSingleAIOCB, s->total);
+    acb->count = 0;
+    acb->success_count = 0;
+    acb->finished = NULL;
+    acb->is_read = false;
+    acb->vote_ret = 0;
+
+    for (i = 0; i < s->total; i++) {
+        acb->aios[i].buf = NULL;
+        acb->aios[i].ret = 0;
+        acb->aios[i].parent = acb;
+    }
+
+    return acb;
+}
+
+static void quorum_aio_cb(void *opaque, int ret)
+{
+    QuorumSingleAIOCB *sacb = opaque;
+    QuorumAIOCB *acb = sacb->parent;
+    BDRVQuorumState *s = acb->bqs;
+
+    sacb->ret = ret;
+    acb->count++;
+    if (ret == 0) {
+        acb->success_count++;
+    }
+    assert(acb->count <= s->total);
+    assert(acb->success_count <= s->total);
+    if (acb->count < s->total) {
+        return;
+    }
+
+    quorum_aio_finalize(acb);
+}
+
+static BlockDriverAIOCB *quorum_aio_writev(BlockDriverState *bs,
+                                          int64_t sector_num,
+                                          QEMUIOVector *qiov,
+                                          int nb_sectors,
+                                          BlockDriverCompletionFunc *cb,
+                                          void *opaque)
+{
+    BDRVQuorumState *s = bs->opaque;
+    QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num, nb_sectors,
+                                      cb, opaque);
+    int i;
+
+    for (i = 0; i < s->total; i++) {
+        acb->aios[i].aiocb = bdrv_aio_writev(&s->bs[i], sector_num, qiov,
+                                             nb_sectors, &quorum_aio_cb,
+                                             &acb->aios[i]);
+    }
+
+    return &acb->common;
+}
+
 static BlockDriver bdrv_quorum = {
     .format_name        = "quorum",
     .protocol_name      = "quorum",
 
     .instance_size      = sizeof(BDRVQuorumState),
+
+    .bdrv_aio_writev    = quorum_aio_writev,
 };
 
 static void bdrv_quorum_init(void)