block/quorum.c

   1 /*
   2  * Quorum Block filter
   3  *
   4  * Copyright (C) 2012-2014 Nodalink, EURL.
   5  *
   6  * Author:
   7  *   Benoît Canet <benoit.canet@irqsave.net>
   8  *
   9  * Based on the design and code of blkverify.c (Copyright (C) 2010 IBM, Corp)
  10  * and blkmirror.c (Copyright (C) 2011 Red Hat, Inc).
  11  *
  12  * This work is licensed under the terms of the GNU GPL, version 2 or later.
  13  * See the COPYING file in the top-level directory.
  14  */
  15
  16 #include <gnutls/gnutls.h>
  17 #include <gnutls/crypto.h>
  18 #include "block/block_int.h"
  19 #include "qapi/qmp/qjson.h"
  20
  21 #define HASH_LENGTH 32
  22
  23 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
  24 #define QUORUM_OPT_BLKVERIFY      "blkverify"
  25
  26 /* This union holds a vote hash value */
  27 typedef union QuorumVoteValue {
  28     char h[HASH_LENGTH];       /* SHA-256 hash */
  29     int64_t l;                 /* simpler 64 bits hash */
  30 } QuorumVoteValue;
  31
  32 /* A vote item */
  33 typedef struct QuorumVoteItem {
  34     int index;
  35     QLIST_ENTRY(QuorumVoteItem) next;
  36 } QuorumVoteItem;
  37
  38 /* this structure is a vote version. A version is the set of votes sharing the
  39  * same vote value.
  40  * The set of votes will be tracked with the items field and its cardinality is
  41  * vote_count.
  42  */
  43 typedef struct QuorumVoteVersion {
  44     QuorumVoteValue value;
  45     int index;
  46     int vote_count;
  47     QLIST_HEAD(, QuorumVoteItem) items;
  48     QLIST_ENTRY(QuorumVoteVersion) next;
  49 } QuorumVoteVersion;
  50
  51 /* this structure holds a group of vote versions together */
  52 typedef struct QuorumVotes {
  53     QLIST_HEAD(, QuorumVoteVersion) vote_list;
  54     bool (*compare)(QuorumVoteValue *a, QuorumVoteValue *b);
  55 } QuorumVotes;
  56
  57 /* the following structure holds the state of one quorum instance */
  58 typedef struct BDRVQuorumState {
  59     BlockDriverState **bs; /* children BlockDriverStates */
  60     int num_children;      /* children count */
  61     int threshold;         /* if less than threshold children reads gave the
  62                             * same result a quorum error occurs.
  63                             */
  64     bool is_blkverify;     /* true if the driver is in blkverify mode
  65                             * Writes are mirrored on two children devices.
  66                             * On reads the two children devices' contents are
  67                             * compared and if a difference is spotted its
  68                             * location is printed and the code aborts.
  69                             * It is useful to debug other block drivers by
  70                             * comparing them with a reference one.
  71                             */
  72 } BDRVQuorumState;
  73
  74 typedef struct QuorumAIOCB QuorumAIOCB;
  75
  76 /* Quorum will create one instance of the following structure per operation it
  77  * performs on its children.
  78  * So for each read/write operation coming from the upper layer there will be
  79  * $children_count QuorumChildRequest.
  80  */
  81 typedef struct QuorumChildRequest {
  82     BlockDriverAIOCB *aiocb;
  83     QEMUIOVector qiov;
  84     uint8_t *buf;
  85     int ret;
  86     QuorumAIOCB *parent;
  87 } QuorumChildRequest;
  88
  89 /* Quorum will use the following structure to track progress of each read/write
  90  * operation received by the upper layer.
  91  * This structure hold pointers to the QuorumChildRequest structures instances
  92  * used to do operations on each children and track overall progress.
  93  */
  94 struct QuorumAIOCB {
  95     BlockDriverAIOCB common;
  96
  97     /* Request metadata */
  98     uint64_t sector_num;
  99     int nb_sectors;
 100
 101     QEMUIOVector *qiov;         /* calling IOV */
 102
 103     QuorumChildRequest *qcrs;   /* individual child requests */
 104     int count;                  /* number of completed AIOCB */
 105     int success_count;          /* number of successfully completed AIOCB */
 106
 107     QuorumVotes votes;
 108
 109     bool is_read;
 110     int vote_ret;
 111 };
 112
 113 static void quorum_vote(QuorumAIOCB *acb);
 114
 115 static void quorum_aio_cancel(BlockDriverAIOCB *blockacb)
 116 {
 117     QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
 118     BDRVQuorumState *s = acb->common.bs->opaque;
 119     int i;
 120
 121     /* cancel all callbacks */
 122     for (i = 0; i < s->num_children; i++) {
 123         bdrv_aio_cancel(acb->qcrs[i].aiocb);
 124     }
 125
 126     g_free(acb->qcrs);
 127     qemu_aio_release(acb);
 128 }
 129
 130 static AIOCBInfo quorum_aiocb_info = {
 131     .aiocb_size         = sizeof(QuorumAIOCB),
 132     .cancel             = quorum_aio_cancel,
 133 };
 134
 135 static void quorum_aio_finalize(QuorumAIOCB *acb)
 136 {
 137     BDRVQuorumState *s = acb->common.bs->opaque;
 138     int i, ret = 0;
 139
 140     if (acb->vote_ret) {
 141         ret = acb->vote_ret;
 142     }
 143
 144     acb->common.cb(acb->common.opaque, ret);
 145
 146     if (acb->is_read) {
 147         for (i = 0; i < s->num_children; i++) {
 148             qemu_vfree(acb->qcrs[i].buf);
 149             qemu_iovec_destroy(&acb->qcrs[i].qiov);
 150         }
 151     }
 152
 153     g_free(acb->qcrs);
 154     qemu_aio_release(acb);
 155 }
 156
 157 static bool quorum_sha256_compare(QuorumVoteValue *a, QuorumVoteValue *b)
 158 {
 159     return !memcmp(a->h, b->h, HASH_LENGTH);
 160 }
 161
 162 static bool quorum_64bits_compare(QuorumVoteValue *a, QuorumVoteValue *b)
 163 {
 164     return a->l == b->l;
 165 }
 166
 167 static QuorumAIOCB *quorum_aio_get(BDRVQuorumState *s,
 168                                    BlockDriverState *bs,
 169                                    QEMUIOVector *qiov,
 170                                    uint64_t sector_num,
 171                                    int nb_sectors,
 172                                    BlockDriverCompletionFunc *cb,
 173                                    void *opaque)
 174 {
 175     QuorumAIOCB *acb = qemu_aio_get(&quorum_aiocb_info, bs, cb, opaque);
 176     int i;
 177
 178     acb->common.bs->opaque = s;
 179     acb->sector_num = sector_num;
 180     acb->nb_sectors = nb_sectors;
 181     acb->qiov = qiov;
 182     acb->qcrs = g_new0(QuorumChildRequest, s->num_children);
 183     acb->count = 0;
 184     acb->success_count = 0;
 185     acb->votes.compare = quorum_sha256_compare;
 186     QLIST_INIT(&acb->votes.vote_list);
 187     acb->is_read = false;
 188     acb->vote_ret = 0;
 189
 190     for (i = 0; i < s->num_children; i++) {
 191         acb->qcrs[i].buf = NULL;
 192         acb->qcrs[i].ret = 0;
 193         acb->qcrs[i].parent = acb;
 194     }
 195
 196     return acb;
 197 }
 198
 199 static void quorum_report_bad(QuorumAIOCB *acb, char *node_name, int ret)
 200 {
 201     QObject *data;
 202     assert(node_name);
 203     data = qobject_from_jsonf("{ 'node-name': %s"
 204                               ", 'sector-num': %" PRId64
 205                               ", 'sectors-count': %d }",
 206                               node_name, acb->sector_num, acb->nb_sectors);
 207     if (ret < 0) {
 208         QDict *dict = qobject_to_qdict(data);
 209         qdict_put(dict, "error", qstring_from_str(strerror(-ret)));
 210     }
 211     monitor_protocol_event(QEVENT_QUORUM_REPORT_BAD, data);
 212     qobject_decref(data);
 213 }
 214
 215 static void quorum_report_failure(QuorumAIOCB *acb)
 216 {
 217     QObject *data;
 218     const char *reference = acb->common.bs->device_name[0] ?
 219                             acb->common.bs->device_name :
 220                             acb->common.bs->node_name;
 221     data = qobject_from_jsonf("{ 'reference': %s"
 222                               ", 'sector-num': %" PRId64
 223                               ", 'sectors-count': %d }",
 224                               reference, acb->sector_num, acb->nb_sectors);
 225     monitor_protocol_event(QEVENT_QUORUM_FAILURE, data);
 226     qobject_decref(data);
 227 }
 228
 229 static int quorum_vote_error(QuorumAIOCB *acb);
 230
 231 static bool quorum_has_too_much_io_failed(QuorumAIOCB *acb)
 232 {
 233     BDRVQuorumState *s = acb->common.bs->opaque;
 234
 235     if (acb->success_count < s->threshold) {
 236         acb->vote_ret = quorum_vote_error(acb);
 237         quorum_report_failure(acb);
 238         return true;
 239     }
 240
 241     return false;
 242 }
 243
 244 static void quorum_aio_cb(void *opaque, int ret)
 245 {
 246     QuorumChildRequest *sacb = opaque;
 247     QuorumAIOCB *acb = sacb->parent;
 248     BDRVQuorumState *s = acb->common.bs->opaque;
 249
 250     sacb->ret = ret;
 251     acb->count++;
 252     if (ret == 0) {
 253         acb->success_count++;
 254     } else {
 255         quorum_report_bad(acb, sacb->aiocb->bs->node_name, ret);
 256     }
 257     assert(acb->count <= s->num_children);
 258     assert(acb->success_count <= s->num_children);
 259     if (acb->count < s->num_children) {
 260         return;
 261     }
 262
 263     /* Do the vote on read */
 264     if (acb->is_read) {
 265         quorum_vote(acb);
 266     } else {
 267         quorum_has_too_much_io_failed(acb);
 268     }
 269
 270     quorum_aio_finalize(acb);
 271 }
 272
 273 static void quorum_report_bad_versions(BDRVQuorumState *s,
 274                                        QuorumAIOCB *acb,
 275                                        QuorumVoteValue *value)
 276 {
 277     QuorumVoteVersion *version;
 278     QuorumVoteItem *item;
 279
 280     QLIST_FOREACH(version, &acb->votes.vote_list, next) {
 281         if (acb->votes.compare(&version->value, value)) {
 282             continue;
 283         }
 284         QLIST_FOREACH(item, &version->items, next) {
 285             quorum_report_bad(acb, s->bs[item->index]->node_name, 0);
 286         }
 287     }
 288 }
 289
 290 static void quorum_copy_qiov(QEMUIOVector *dest, QEMUIOVector *source)
 291 {
 292     int i;
 293     assert(dest->niov == source->niov);
 294     assert(dest->size == source->size);
 295     for (i = 0; i < source->niov; i++) {
 296         assert(dest->iov[i].iov_len == source->iov[i].iov_len);
 297         memcpy(dest->iov[i].iov_base,
 298                source->iov[i].iov_base,
 299                source->iov[i].iov_len);
 300     }
 301 }
 302
 303 static void quorum_count_vote(QuorumVotes *votes,
 304                               QuorumVoteValue *value,
 305                               int index)
 306 {
 307     QuorumVoteVersion *v = NULL, *version = NULL;
 308     QuorumVoteItem *item;
 309
 310     /* look if we have something with this hash */
 311     QLIST_FOREACH(v, &votes->vote_list, next) {
 312         if (votes->compare(&v->value, value)) {
 313             version = v;
 314             break;
 315         }
 316     }
 317
 318     /* It's a version not yet in the list add it */
 319     if (!version) {
 320         version = g_new0(QuorumVoteVersion, 1);
 321         QLIST_INIT(&version->items);
 322         memcpy(&version->value, value, sizeof(version->value));
 323         version->index = index;
 324         version->vote_count = 0;
 325         QLIST_INSERT_HEAD(&votes->vote_list, version, next);
 326     }
 327
 328     version->vote_count++;
 329
 330     item = g_new0(QuorumVoteItem, 1);
 331     item->index = index;
 332     QLIST_INSERT_HEAD(&version->items, item, next);
 333 }
 334
 335 static void quorum_free_vote_list(QuorumVotes *votes)
 336 {
 337     QuorumVoteVersion *version, *next_version;
 338     QuorumVoteItem *item, *next_item;
 339
 340     QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
 341         QLIST_REMOVE(version, next);
 342         QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
 343             QLIST_REMOVE(item, next);
 344             g_free(item);
 345         }
 346         g_free(version);
 347     }
 348 }
 349
 350 static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
 351 {
 352     int j, ret;
 353     gnutls_hash_hd_t dig;
 354     QEMUIOVector *qiov = &acb->qcrs[i].qiov;
 355
 356     ret = gnutls_hash_init(&dig, GNUTLS_DIG_SHA256);
 357
 358     if (ret < 0) {
 359         return ret;
 360     }
 361
 362     for (j = 0; j < qiov->niov; j++) {
 363         ret = gnutls_hash(dig, qiov->iov[j].iov_base, qiov->iov[j].iov_len);
 364         if (ret < 0) {
 365             break;
 366         }
 367     }
 368
 369     gnutls_hash_deinit(dig, (void *) hash);
 370     return ret;
 371 }
 372
 373 static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
 374 {
 375     int max = 0;
 376     QuorumVoteVersion *candidate, *winner = NULL;
 377
 378     QLIST_FOREACH(candidate, &votes->vote_list, next) {
 379         if (candidate->vote_count > max) {
 380             max = candidate->vote_count;
 381             winner = candidate;
 382         }
 383     }
 384
 385     return winner;
 386 }
 387
 388 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
 389  * differing byte location. Yet it is handcoded to compare vectors one byte
 390  * after another so it does not benefit from the libc SIMD optimizations.
 391  * quorum_iovec_compare is written for speed and should be used in the non
 392  * blkverify mode of quorum.
 393  */
 394 static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
 395 {
 396     int i;
 397     int result;
 398
 399     assert(a->niov == b->niov);
 400     for (i = 0; i < a->niov; i++) {
 401         assert(a->iov[i].iov_len == b->iov[i].iov_len);
 402         result = memcmp(a->iov[i].iov_base,
 403                         b->iov[i].iov_base,
 404                         a->iov[i].iov_len);
 405         if (result) {
 406             return false;
 407         }
 408     }
 409
 410     return true;
 411 }
 412
 413 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB *acb,
 414                                           const char *fmt, ...)
 415 {
 416     va_list ap;
 417
 418     va_start(ap, fmt);
 419     fprintf(stderr, "quorum: sector_num=%" PRId64 " nb_sectors=%d ",
 420             acb->sector_num, acb->nb_sectors);
 421     vfprintf(stderr, fmt, ap);
 422     fprintf(stderr, "\n");
 423     va_end(ap);
 424     exit(1);
 425 }
 426
 427 static bool quorum_compare(QuorumAIOCB *acb,
 428                            QEMUIOVector *a,
 429                            QEMUIOVector *b)
 430 {
 431     BDRVQuorumState *s = acb->common.bs->opaque;
 432     ssize_t offset;
 433
 434     /* This driver will replace blkverify in this particular case */
 435     if (s->is_blkverify) {
 436         offset = qemu_iovec_compare(a, b);
 437         if (offset != -1) {
 438             quorum_err(acb, "contents mismatch in sector %" PRId64,
 439                        acb->sector_num +
 440                        (uint64_t)(offset / BDRV_SECTOR_SIZE));
 441         }
 442         return true;
 443     }
 444
 445     return quorum_iovec_compare(a, b);
 446 }
 447
 448 /* Do a vote to get the error code */
 449 static int quorum_vote_error(QuorumAIOCB *acb)
 450 {
 451     BDRVQuorumState *s = acb->common.bs->opaque;
 452     QuorumVoteVersion *winner = NULL;
 453     QuorumVotes error_votes;
 454     QuorumVoteValue result_value;
 455     int i, ret = 0;
 456     bool error = false;
 457
 458     QLIST_INIT(&error_votes.vote_list);
 459     error_votes.compare = quorum_64bits_compare;
 460
 461     for (i = 0; i < s->num_children; i++) {
 462         ret = acb->qcrs[i].ret;
 463         if (ret) {
 464             error = true;
 465             result_value.l = ret;
 466             quorum_count_vote(&error_votes, &result_value, i);
 467         }
 468     }
 469
 470     if (error) {
 471         winner = quorum_get_vote_winner(&error_votes);
 472         ret = winner->value.l;
 473     }
 474
 475     quorum_free_vote_list(&error_votes);
 476
 477     return ret;
 478 }
 479
 480 static void quorum_vote(QuorumAIOCB *acb)
 481 {
 482     bool quorum = true;
 483     int i, j, ret;
 484     QuorumVoteValue hash;
 485     BDRVQuorumState *s = acb->common.bs->opaque;
 486     QuorumVoteVersion *winner;
 487
 488     if (quorum_has_too_much_io_failed(acb)) {
 489         return;
 490     }
 491
 492     /* get the index of the first successful read */
 493     for (i = 0; i < s->num_children; i++) {
 494         if (!acb->qcrs[i].ret) {
 495             break;
 496         }
 497     }
 498
 499     assert(i < s->num_children);
 500
 501     /* compare this read with all other successful reads stopping at quorum
 502      * failure
 503      */
 504     for (j = i + 1; j < s->num_children; j++) {
 505         if (acb->qcrs[j].ret) {
 506             continue;
 507         }
 508         quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
 509         if (!quorum) {
 510             break;
 511        }
 512     }
 513
 514     /* Every successful read agrees */
 515     if (quorum) {
 516         quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
 517         return;
 518     }
 519
 520     /* compute hashes for each successful read, also store indexes */
 521     for (i = 0; i < s->num_children; i++) {
 522         if (acb->qcrs[i].ret) {
 523             continue;
 524         }
 525         ret = quorum_compute_hash(acb, i, &hash);
 526         /* if ever the hash computation failed */
 527         if (ret < 0) {
 528             acb->vote_ret = ret;
 529             goto free_exit;
 530         }
 531         quorum_count_vote(&acb->votes, &hash, i);
 532     }
 533
 534     /* vote to select the most represented version */
 535     winner = quorum_get_vote_winner(&acb->votes);
 536
 537     /* if the winner count is smaller than threshold the read fails */
 538     if (winner->vote_count < s->threshold) {
 539         quorum_report_failure(acb);
 540         acb->vote_ret = -EIO;
 541         goto free_exit;
 542     }
 543
 544     /* we have a winner: copy it */
 545     quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
 546
 547     /* some versions are bad print them */
 548     quorum_report_bad_versions(s, acb, &winner->value);
 549
 550 free_exit:
 551     /* free lists */
 552     quorum_free_vote_list(&acb->votes);
 553 }
 554
 555 static BlockDriverAIOCB *quorum_aio_readv(BlockDriverState *bs,
 556                                          int64_t sector_num,
 557                                          QEMUIOVector *qiov,
 558                                          int nb_sectors,
 559                                          BlockDriverCompletionFunc *cb,
 560                                          void *opaque)
 561 {
 562     BDRVQuorumState *s = bs->opaque;
 563     QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num,
 564                                       nb_sectors, cb, opaque);
 565     int i;
 566
 567     acb->is_read = true;
 568
 569     for (i = 0; i < s->num_children; i++) {
 570         acb->qcrs[i].buf = qemu_blockalign(s->bs[i], qiov->size);
 571         qemu_iovec_init(&acb->qcrs[i].qiov, qiov->niov);
 572         qemu_iovec_clone(&acb->qcrs[i].qiov, qiov, acb->qcrs[i].buf);
 573     }
 574
 575     for (i = 0; i < s->num_children; i++) {
 576         bdrv_aio_readv(s->bs[i], sector_num, &acb->qcrs[i].qiov, nb_sectors,
 577                        quorum_aio_cb, &acb->qcrs[i]);
 578     }
 579
 580     return &acb->common;
 581 }
 582
 583 static BlockDriverAIOCB *quorum_aio_writev(BlockDriverState *bs,
 584                                           int64_t sector_num,
 585                                           QEMUIOVector *qiov,
 586                                           int nb_sectors,
 587                                           BlockDriverCompletionFunc *cb,
 588                                           void *opaque)
 589 {
 590     BDRVQuorumState *s = bs->opaque;
 591     QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num, nb_sectors,
 592                                       cb, opaque);
 593     int i;
 594
 595     for (i = 0; i < s->num_children; i++) {
 596         acb->qcrs[i].aiocb = bdrv_aio_writev(s->bs[i], sector_num, qiov,
 597                                              nb_sectors, &quorum_aio_cb,
 598                                              &acb->qcrs[i]);
 599     }
 600
 601     return &acb->common;
 602 }
 603
 604 static int64_t quorum_getlength(BlockDriverState *bs)
 605 {
 606     BDRVQuorumState *s = bs->opaque;
 607     int64_t result;
 608     int i;
 609
 610     /* check that all file have the same length */
 611     result = bdrv_getlength(s->bs[0]);
 612     if (result < 0) {
 613         return result;
 614     }
 615     for (i = 1; i < s->num_children; i++) {
 616         int64_t value = bdrv_getlength(s->bs[i]);
 617         if (value < 0) {
 618             return value;
 619         }
 620         if (value != result) {
 621             return -EIO;
 622         }
 623     }
 624
 625     return result;
 626 }
 627
 628 static void quorum_invalidate_cache(BlockDriverState *bs, Error **errp)
 629 {
 630     BDRVQuorumState *s = bs->opaque;
 631     Error *local_err = NULL;
 632     int i;
 633
 634     for (i = 0; i < s->num_children; i++) {
 635         bdrv_invalidate_cache(s->bs[i], &local_err);
 636         if (local_err) {
 637             error_propagate(errp, local_err);
 638             return;
 639         }
 640     }
 641 }
 642
 643 static coroutine_fn int quorum_co_flush(BlockDriverState *bs)
 644 {
 645     BDRVQuorumState *s = bs->opaque;
 646     QuorumVoteVersion *winner = NULL;
 647     QuorumVotes error_votes;
 648     QuorumVoteValue result_value;
 649     int i;
 650     int result = 0;
 651
 652     QLIST_INIT(&error_votes.vote_list);
 653     error_votes.compare = quorum_64bits_compare;
 654
 655     for (i = 0; i < s->num_children; i++) {
 656         result = bdrv_co_flush(s->bs[i]);
 657         result_value.l = result;
 658         quorum_count_vote(&error_votes, &result_value, i);
 659     }
 660
 661     winner = quorum_get_vote_winner(&error_votes);
 662     result = winner->value.l;
 663
 664     quorum_free_vote_list(&error_votes);
 665
 666     return result;
 667 }
 668
 669 static bool quorum_recurse_is_first_non_filter(BlockDriverState *bs,
 670                                                BlockDriverState *candidate)
 671 {
 672     BDRVQuorumState *s = bs->opaque;
 673     int i;
 674
 675     for (i = 0; i < s->num_children; i++) {
 676         bool perm = bdrv_recurse_is_first_non_filter(s->bs[i],
 677                                                      candidate);
 678         if (perm) {
 679             return true;
 680         }
 681     }
 682
 683     return false;
 684 }
 685
 686 static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
 687 {
 688
 689     if (threshold < 1) {
 690         error_set(errp, QERR_INVALID_PARAMETER_VALUE,
 691                   "vote-threshold", "value >= 1");
 692         return -ERANGE;
 693     }
 694
 695     if (threshold > num_children) {
 696         error_setg(errp, "threshold may not exceed children count");
 697         return -ERANGE;
 698     }
 699
 700     return 0;
 701 }
 702
 703 static QemuOptsList quorum_runtime_opts = {
 704     .name = "quorum",
 705     .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
 706     .desc = {
 707         {
 708             .name = QUORUM_OPT_VOTE_THRESHOLD,
 709             .type = QEMU_OPT_NUMBER,
 710             .help = "The number of vote needed for reaching quorum",
 711         },
 712         {
 713             .name = QUORUM_OPT_BLKVERIFY,
 714             .type = QEMU_OPT_BOOL,
 715             .help = "Trigger block verify mode if set",
 716         },
 717         { /* end of list */ }
 718     },
 719 };
 720
 721 static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
 722                        Error **errp)
 723 {
 724     BDRVQuorumState *s = bs->opaque;
 725     Error *local_err = NULL;
 726     QemuOpts *opts;
 727     bool *opened;
 728     QDict *sub = NULL;
 729     QList *list = NULL;
 730     const QListEntry *lentry;
 731     int i;
 732     int ret = 0;
 733
 734     qdict_flatten(options);
 735     qdict_extract_subqdict(options, &sub, "children.");
 736     qdict_array_split(sub, &list);
 737
 738     if (qdict_size(sub)) {
 739         error_setg(&local_err, "Invalid option children.%s",
 740                    qdict_first(sub)->key);
 741         ret = -EINVAL;
 742         goto exit;
 743     }
 744
 745     /* count how many different children are present */
 746     s->num_children = qlist_size(list);
 747     if (s->num_children < 2) {
 748         error_setg(&local_err,
 749                    "Number of provided children must be greater than 1");
 750         ret = -EINVAL;
 751         goto exit;
 752     }
 753
 754     opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
 755     qemu_opts_absorb_qdict(opts, options, &local_err);
 756     if (local_err) {
 757         ret = -EINVAL;
 758         goto exit;
 759     }
 760
 761     s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
 762
 763     /* and validate it against s->num_children */
 764     ret = quorum_valid_threshold(s->threshold, s->num_children, &local_err);
 765     if (ret < 0) {
 766         goto exit;
 767     }
 768
 769     /* is the driver in blkverify mode */
 770     if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false) &&
 771         s->num_children == 2 && s->threshold == 2) {
 772         s->is_blkverify = true;
 773     } else if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false)) {
 774         fprintf(stderr, "blkverify mode is set by setting blkverify=on "
 775                 "and using two files with vote_threshold=2\n");
 776     }
 777
 778     /* allocate the children BlockDriverState array */
 779     s->bs = g_new0(BlockDriverState *, s->num_children);
 780     opened = g_new0(bool, s->num_children);
 781
 782     for (i = 0, lentry = qlist_first(list); lentry;
 783          lentry = qlist_next(lentry), i++) {
 784         QDict *d;
 785         QString *string;
 786
 787         switch (qobject_type(lentry->value))
 788         {
 789             /* List of options */
 790             case QTYPE_QDICT:
 791                 d = qobject_to_qdict(lentry->value);
 792                 QINCREF(d);
 793                 ret = bdrv_open(&s->bs[i], NULL, NULL, d, flags, NULL,
 794                                 &local_err);
 795                 break;
 796
 797             /* QMP reference */
 798             case QTYPE_QSTRING:
 799                 string = qobject_to_qstring(lentry->value);
 800                 ret = bdrv_open(&s->bs[i], NULL, qstring_get_str(string), NULL,
 801                                 flags, NULL, &local_err);
 802                 break;
 803
 804             default:
 805                 error_setg(&local_err, "Specification of child block device %i "
 806                            "is invalid", i);
 807                 ret = -EINVAL;
 808         }
 809
 810         if (ret < 0) {
 811             goto close_exit;
 812         }
 813         opened[i] = true;
 814     }
 815
 816     g_free(opened);
 817     goto exit;
 818
 819 close_exit:
 820     /* cleanup on error */
 821     for (i = 0; i < s->num_children; i++) {
 822         if (!opened[i]) {
 823             continue;
 824         }
 825         bdrv_unref(s->bs[i]);
 826     }
 827     g_free(s->bs);
 828     g_free(opened);
 829 exit:
 830     /* propagate error */
 831     if (local_err) {
 832         error_propagate(errp, local_err);
 833     }
 834     QDECREF(list);
 835     QDECREF(sub);
 836     return ret;
 837 }
 838
 839 static void quorum_close(BlockDriverState *bs)
 840 {
 841     BDRVQuorumState *s = bs->opaque;
 842     int i;
 843
 844     for (i = 0; i < s->num_children; i++) {
 845         bdrv_unref(s->bs[i]);
 846     }
 847
 848     g_free(s->bs);
 849 }
 850
 851 static BlockDriver bdrv_quorum = {
 852     .format_name        = "quorum",
 853     .protocol_name      = "quorum",
 854
 855     .instance_size      = sizeof(BDRVQuorumState),
 856
 857     .bdrv_file_open     = quorum_open,
 858     .bdrv_close         = quorum_close,
 859
 860     .bdrv_co_flush_to_disk = quorum_co_flush,
 861
 862     .bdrv_getlength     = quorum_getlength,
 863
 864     .bdrv_aio_readv     = quorum_aio_readv,
 865     .bdrv_aio_writev    = quorum_aio_writev,
 866     .bdrv_invalidate_cache = quorum_invalidate_cache,
 867
 868     .is_filter           = true,
 869     .bdrv_recurse_is_first_non_filter = quorum_recurse_is_first_non_filter,
 870 };
 871
 872 static void bdrv_quorum_init(void)
 873 {
 874     bdrv_register(&bdrv_quorum);
 875 }
 876
 877 block_init(bdrv_quorum_init);