Re: [Qemu-devel] [PATCH V17 11/12] quorum: Add quorum_open() and quorum_close().

[Max Reitz <mreitz@redhat.com>]

On 12.02.2014 23:06, Benoît Canet wrote:
> From: Benoît Canet <benoit@irqsave.net>
>
> Example of command line:
> -drive if=virtio,file.driver=quorum,\
> file.children.0.file.filename=1.raw,\
> file.children.0.node-name=1.raw,\
> file.children.0.driver=raw,\
> file.children.1.file.filename=2.raw,\
> file.children.1.node-name=2.raw,\
> file.children.1.driver=raw,\
> file.children.2.file.filename=3.raw,\
> file.children.2.node-name=3.raw,\
> file.children.2.driver=raw,\
> file.vote-threshold=2
>
> file.blkverify=on with file.vote-threshold=2 and two files can be passed to
> emulated blkverify.
>
> Signed-off-by: Benoit Canet <benoit@irqsave.net>
> ---
>   block/quorum.c   | 165 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
>   monitor.c        |   2 +
>   qapi-schema.json |  21 ++++++-
>   3 files changed, 187 insertions(+), 1 deletion(-)
>
> diff --git a/block/quorum.c b/block/quorum.c
> index 8e033ef..d3a90fb 100644
> --- a/block/quorum.c
> +++ b/block/quorum.c
> @@ -20,6 +20,9 @@
>   
>   #define HASH_LENGTH 32
>   
> +#define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
> +#define QUORUM_OPT_BLKVERIFY      "blkverify"
> +
>   /* This union holds a vote hash value */
>   typedef union QuorumVoteValue {
>       char h[HASH_LENGTH];       /* SHA-256 hash */
> @@ -677,12 +680,174 @@ static bool quorum_recurse_is_first_non_filter(BlockDriverState *bs,
>       return false;
>   }
>   
> +static int quorum_valid_threshold(int threshold,
> +                                  int num_children,
> +                                  Error **errp)

This fits on a single line and I'm all for maximizing usage of the 80 
character limit.

> +{
> +
> +    if (threshold < 1) {
> +        error_set(errp, QERR_INVALID_PARAMETER_VALUE,
> +                  "vote-threshold", "value >= 1");
> +        return -ERANGE;
> +    }
> +
> +    if (threshold > num_children) {
> +        error_setg(errp, "threshold may not exceed children count");
> +        return -ERANGE;
> +    }
> +
> +    return 0;
> +}
> +
> +static QemuOptsList quorum_runtime_opts = {
> +    .name = "quorum",
> +    .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
> +    .desc = {
> +        {
> +            .name = QUORUM_OPT_VOTE_THRESHOLD,
> +            .type = QEMU_OPT_NUMBER,
> +            .help = "The number of vote needed for reaching quorum",
> +        },
> +        {
> +            .name = QUORUM_OPT_BLKVERIFY,
> +            .type = QEMU_OPT_BOOL,
> +            .help = "Trigger block verify mode if set",
> +        },
> +        { /* end of list */ }
> +    },
> +};
> +
> +static int quorum_open(BlockDriverState *bs,
> +                       QDict *options,
> +                       int flags,
> +                       Error **errp)

This takes two lines (although since you have to split at all, I can 
cope with this being split once per parameter).

> +{
> +    BDRVQuorumState *s = bs->opaque;
> +    Error *local_err = NULL;
> +    QemuOpts *opts;
> +    bool *opened;
> +    QDict *sub = NULL;
> +    QList *list = NULL;
> +    const QListEntry *lentry;
> +    const QDictEntry *dentry;
> +    int i;
> +    int ret = 0;
> +
> +    qdict_flatten(options);
> +    qdict_extract_subqdict(options, &sub, "children.");
> +    qdict_array_split(sub, &list);
> +
> +    /* count how many different children are present and validate
> +     * qdict_size(sub) address the open by reference case
> +     */
> +    s->num_children = !qlist_size(list) ? qdict_size(sub) : qlist_size(list);

Okay, so eventually we probably want to add these values in order to 
allow "mixed" parameters (some references, some "full" specifications). 
But for that to work without much hassle, qdict_array_split() still 
needs some extension (or be fixed?) to split not just QDicts but other 
objects as well. I think we (probably I) should amend 
qdict_array_split() first (perhaps add some boolean parameter to specify 
what it should do) and then fix this (that is, make mixed parameters 
work and allow detection of malformed parameters).

For this series, this is fine.

> +    if (s->num_children < 2) {
> +        error_setg(&local_err,
> +                   "Number of provided children must be greater than 1");
> +        ret = -EINVAL;
> +        goto exit;
> +    }
> +
> +    opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
> +    qemu_opts_absorb_qdict(opts, options, &local_err);
> +    if (error_is_set(&local_err)) {
> +        ret = -EINVAL;
> +        goto exit;
> +    }
> +
> +    s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
> +
> +    /* and validate it against s->num_children */
> +    ret = quorum_valid_threshold(s->threshold, s->num_children, &local_err);
> +    if (ret < 0) {
> +        goto exit;
> +    }
> +
> +    /* is the driver in blkverify mode */
> +    if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false) &&
> +        s->num_children == 2 && s->threshold == 2) {

If qemu_opt_get_bool() returns true but the other conditions aren't met, 
I'd prefer, again, a warning.

> +        s->is_blkverify = true;
> +    }
> +
> +    /* allocate the children BlockDriverState array */
> +    s->bs = g_new0(BlockDriverState *, s->num_children);
> +    opened = g_new0(bool, s->num_children);
> +
> +    /* Open by file name or options dict (command line or QMP) */
> +    if (s->num_children == qlist_size(list)) {
> +        for (i = 0, lentry = qlist_first(list); lentry;
> +            lentry = qlist_next(lentry), i++) {
> +            QDict *d = qobject_to_qdict(lentry->value);
> +            QINCREF(d);
> +            ret = bdrv_open(&s->bs[i], NULL, NULL,
> +                            d,
> +                            flags, NULL, &local_err);

This actually fits on a single line (sorry for my obsession with this).

> +            if (ret < 0) {
> +                goto close_exit;
> +            }
> +            opened[i] = true;
> +        }
> +    /* Open by QMP references */
> +    } else {
> +        for (i = 0, dentry = qdict_first(sub); dentry;
> +             dentry = qdict_next(sub, dentry), i++) {
> +            QString *string = qobject_to_qstring(dentry->value);
> +            ret = bdrv_open(&s->bs[i], NULL,
> +                            qstring_get_str(string),
> +                            NULL, flags, NULL, &local_err);

This takes two.

> +            if (ret < 0) {
> +                goto close_exit;
> +            }
> +            opened[i] = true;
> +        }
> +    }
> +
> +    g_free(opened);
> +    goto exit;
> +
> +close_exit:
> +    /* cleanup on error */
> +    for (i = 0; i < s->num_children; i++) {
> +        if (!opened[i]) {
> +            continue;
> +        }
> +        bdrv_unref(s->bs[i]);
> +    }
> +    g_free(s->bs);
> +    g_free(opened);
> +exit:
> +    /* propagate error */
> +    if (error_is_set(&local_err)) {
> +        error_propagate(errp, local_err);
> +    }
> +    QDECREF(list);
> +    QDECREF(sub);
> +    return ret;
> +}
> +
> +static void quorum_close(BlockDriverState *bs)
> +{
> +    BDRVQuorumState *s = bs->opaque;
> +    int i;
> +
> +    for (i = 0; i < s->num_children; i++) {
> +        bdrv_unref(s->bs[i]);
> +    }
> +
> +    g_free(s->bs);
> +}
> +
>   static BlockDriver bdrv_quorum = {
>       .format_name        = "quorum",
>       .protocol_name      = "quorum",
>   
>       .instance_size      = sizeof(BDRVQuorumState),
>   
> +    .bdrv_file_open     = quorum_open,
> +    .bdrv_close         = quorum_close,
> +
> +    .authorizations     = { true, true },
> +
>       .bdrv_co_flush_to_disk = quorum_co_flush,
>   
>       .bdrv_getlength     = quorum_getlength,
> diff --git a/monitor.c b/monitor.c
> index 81ffa0f..8b175fb 100644
> --- a/monitor.c
> +++ b/monitor.c
> @@ -639,6 +639,8 @@ static void monitor_protocol_event_init(void)
>       monitor_protocol_event_throttle(QEVENT_RTC_CHANGE, 1000);
>       monitor_protocol_event_throttle(QEVENT_BALLOON_CHANGE, 1000);
>       monitor_protocol_event_throttle(QEVENT_WATCHDOG, 1000);
> +    monitor_protocol_event_throttle(QEVENT_QUORUM_REPORT_BAD, 1000);
> +    monitor_protocol_event_throttle(QEVENT_QUORUM_FAILURE, 1000);

Hm, what is this for?

Max

>   }
>   
>   /**
> diff --git a/qapi-schema.json b/qapi-schema.json
> index 7cfb5e5..990d0c5 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -4352,6 +4352,24 @@
>               'raw': 'BlockdevRef' } }
>   
>   ##
> +# @BlockdevOptionsQuorum
> +#
> +# Driver specific block device options for Quorum
> +#
> +# @blkverify:      #optional true if the driver must print content mismatch
> +#
> +# @children:       the children block device to use
> +#
> +# @vote_threshold: the vote limit under which a read will fail
> +#
> +# Since: 2.0
> +##
> +{ 'type': 'BlockdevOptionsQuorum',
> +  'data': { '*blkverify': 'bool',
> +            'children': [ 'BlockdevRef' ],
> +            'vote-threshold': 'int' } }
> +
> +##
>   # @BlockdevOptions
>   #
>   # Options for creating a block device.
> @@ -4390,7 +4408,8 @@
>         'vdi':        'BlockdevOptionsGenericFormat',
>         'vhdx':       'BlockdevOptionsGenericFormat',
>         'vmdk':       'BlockdevOptionsGenericCOWFormat',
> -      'vpc':        'BlockdevOptionsGenericFormat'
> +      'vpc':        'BlockdevOptionsGenericFormat',
> +      'quorum':     'BlockdevOptionsQuorum'
>     } }
>   
>   ##