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 "qemu/osdep.h"
  17 #include "block/block_int.h"
  18 #include "qapi/qmp/qbool.h"
  19 #include "qapi/qmp/qdict.h"
  20 #include "qapi/qmp/qerror.h"
  21 #include "qapi/qmp/qint.h"
  22 #include "qapi/qmp/qjson.h"
  23 #include "qapi/qmp/qlist.h"
  24 #include "qapi/qmp/qstring.h"
  25 #include "qapi-event.h"
  26 #include "crypto/hash.h"
  27
  28 #define HASH_LENGTH 32
  29
  30 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
  31 #define QUORUM_OPT_BLKVERIFY      "blkverify"
  32 #define QUORUM_OPT_REWRITE        "rewrite-corrupted"
  33 #define QUORUM_OPT_READ_PATTERN   "read-pattern"
  34
  35 /* This union holds a vote hash value */
  36 typedef union QuorumVoteValue {
  37     uint8_t h[HASH_LENGTH];    /* SHA-256 hash */
  38     int64_t l;                 /* simpler 64 bits hash */
  39 } QuorumVoteValue;
  40
  41 /* A vote item */
  42 typedef struct QuorumVoteItem {
  43     int index;
  44     QLIST_ENTRY(QuorumVoteItem) next;
  45 } QuorumVoteItem;
  46
  47 /* this structure is a vote version. A version is the set of votes sharing the
  48  * same vote value.
  49  * The set of votes will be tracked with the items field and its cardinality is
  50  * vote_count.
  51  */
  52 typedef struct QuorumVoteVersion {
  53     QuorumVoteValue value;
  54     int index;
  55     int vote_count;
  56     QLIST_HEAD(, QuorumVoteItem) items;
  57     QLIST_ENTRY(QuorumVoteVersion) next;
  58 } QuorumVoteVersion;
  59
  60 /* this structure holds a group of vote versions together */
  61 typedef struct QuorumVotes {
  62     QLIST_HEAD(, QuorumVoteVersion) vote_list;
  63     bool (*compare)(QuorumVoteValue *a, QuorumVoteValue *b);
  64 } QuorumVotes;
  65
  66 /* the following structure holds the state of one quorum instance */
  67 typedef struct BDRVQuorumState {
  68     BdrvChild **children;  /* children BlockDriverStates */
  69     int num_children;      /* children count */
  70     int threshold;         /* if less than threshold children reads gave the
  71                             * same result a quorum error occurs.
  72                             */
  73     bool is_blkverify;     /* true if the driver is in blkverify mode
  74                             * Writes are mirrored on two children devices.
  75                             * On reads the two children devices' contents are
  76                             * compared and if a difference is spotted its
  77                             * location is printed and the code aborts.
  78                             * It is useful to debug other block drivers by
  79                             * comparing them with a reference one.
  80                             */
  81     bool rewrite_corrupted;/* true if the driver must rewrite-on-read corrupted
  82                             * block if Quorum is reached.
  83                             */
  84
  85     QuorumReadPattern read_pattern;
  86 } BDRVQuorumState;
  87
  88 typedef struct QuorumAIOCB QuorumAIOCB;
  89
  90 /* Quorum will create one instance of the following structure per operation it
  91  * performs on its children.
  92  * So for each read/write operation coming from the upper layer there will be
  93  * $children_count QuorumChildRequest.
  94  */
  95 typedef struct QuorumChildRequest {
  96     BlockAIOCB *aiocb;
  97     QEMUIOVector qiov;
  98     uint8_t *buf;
  99     int ret;
 100     QuorumAIOCB *parent;
 101 } QuorumChildRequest;
 102
 103 /* Quorum will use the following structure to track progress of each read/write
 104  * operation received by the upper layer.
 105  * This structure hold pointers to the QuorumChildRequest structures instances
 106  * used to do operations on each children and track overall progress.
 107  */
 108 struct QuorumAIOCB {
 109     BlockAIOCB common;
 110
 111     /* Request metadata */
 112     uint64_t sector_num;
 113     int nb_sectors;
 114
 115     QEMUIOVector *qiov;         /* calling IOV */
 116
 117     QuorumChildRequest *qcrs;   /* individual child requests */
 118     int count;                  /* number of completed AIOCB */
 119     int success_count;          /* number of successfully completed AIOCB */
 120
 121     int rewrite_count;          /* number of replica to rewrite: count down to
 122                                  * zero once writes are fired
 123                                  */
 124
 125     QuorumVotes votes;
 126
 127     bool is_read;
 128     int vote_ret;
 129     int child_iter;             /* which child to read in fifo pattern */
 130 };
 131
 132 static bool quorum_vote(QuorumAIOCB *acb);
 133
 134 static void quorum_aio_cancel(BlockAIOCB *blockacb)
 135 {
 136     QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
 137     BDRVQuorumState *s = acb->common.bs->opaque;
 138     int i;
 139
 140     /* cancel all callbacks */
 141     for (i = 0; i < s->num_children; i++) {
 142         if (acb->qcrs[i].aiocb) {
 143             bdrv_aio_cancel_async(acb->qcrs[i].aiocb);
 144         }
 145     }
 146 }
 147
 148 static AIOCBInfo quorum_aiocb_info = {
 149     .aiocb_size         = sizeof(QuorumAIOCB),
 150     .cancel_async       = quorum_aio_cancel,
 151 };
 152
 153 static void quorum_aio_finalize(QuorumAIOCB *acb)
 154 {
 155     int i, ret = 0;
 156
 157     if (acb->vote_ret) {
 158         ret = acb->vote_ret;
 159     }
 160
 161     acb->common.cb(acb->common.opaque, ret);
 162
 163     if (acb->is_read) {
 164         /* on the quorum case acb->child_iter == s->num_children - 1 */
 165         for (i = 0; i <= acb->child_iter; i++) {
 166             qemu_vfree(acb->qcrs[i].buf);
 167             qemu_iovec_destroy(&acb->qcrs[i].qiov);
 168         }
 169     }
 170
 171     g_free(acb->qcrs);
 172     qemu_aio_unref(acb);
 173 }
 174
 175 static bool quorum_sha256_compare(QuorumVoteValue *a, QuorumVoteValue *b)
 176 {
 177     return !memcmp(a->h, b->h, HASH_LENGTH);
 178 }
 179
 180 static bool quorum_64bits_compare(QuorumVoteValue *a, QuorumVoteValue *b)
 181 {
 182     return a->l == b->l;
 183 }
 184
 185 static QuorumAIOCB *quorum_aio_get(BDRVQuorumState *s,
 186                                    BlockDriverState *bs,
 187                                    QEMUIOVector *qiov,
 188                                    uint64_t sector_num,
 189                                    int nb_sectors,
 190                                    BlockCompletionFunc *cb,
 191                                    void *opaque)
 192 {
 193     QuorumAIOCB *acb = qemu_aio_get(&quorum_aiocb_info, bs, cb, opaque);
 194     int i;
 195
 196     acb->common.bs->opaque = s;
 197     acb->sector_num = sector_num;
 198     acb->nb_sectors = nb_sectors;
 199     acb->qiov = qiov;
 200     acb->qcrs = g_new0(QuorumChildRequest, s->num_children);
 201     acb->count = 0;
 202     acb->success_count = 0;
 203     acb->rewrite_count = 0;
 204     acb->votes.compare = quorum_sha256_compare;
 205     QLIST_INIT(&acb->votes.vote_list);
 206     acb->is_read = false;
 207     acb->vote_ret = 0;
 208
 209     for (i = 0; i < s->num_children; i++) {
 210         acb->qcrs[i].buf = NULL;
 211         acb->qcrs[i].ret = 0;
 212         acb->qcrs[i].parent = acb;
 213     }
 214
 215     return acb;
 216 }
 217
 218 static void quorum_report_bad(QuorumOpType type, uint64_t sector_num,
 219                               int nb_sectors, char *node_name, int ret)
 220 {
 221     const char *msg = NULL;
 222     if (ret < 0) {
 223         msg = strerror(-ret);
 224     }
 225
 226     qapi_event_send_quorum_report_bad(type, !!msg, msg, node_name,
 227                                       sector_num, nb_sectors, &error_abort);
 228 }
 229
 230 static void quorum_report_failure(QuorumAIOCB *acb)
 231 {
 232     const char *reference = bdrv_get_device_or_node_name(acb->common.bs);
 233     qapi_event_send_quorum_failure(reference, acb->sector_num,
 234                                    acb->nb_sectors, &error_abort);
 235 }
 236
 237 static int quorum_vote_error(QuorumAIOCB *acb);
 238
 239 static bool quorum_has_too_much_io_failed(QuorumAIOCB *acb)
 240 {
 241     BDRVQuorumState *s = acb->common.bs->opaque;
 242
 243     if (acb->success_count < s->threshold) {
 244         acb->vote_ret = quorum_vote_error(acb);
 245         quorum_report_failure(acb);
 246         return true;
 247     }
 248
 249     return false;
 250 }
 251
 252 static void quorum_rewrite_aio_cb(void *opaque, int ret)
 253 {
 254     QuorumAIOCB *acb = opaque;
 255
 256     /* one less rewrite to do */
 257     acb->rewrite_count--;
 258
 259     /* wait until all rewrite callbacks have completed */
 260     if (acb->rewrite_count) {
 261         return;
 262     }
 263
 264     quorum_aio_finalize(acb);
 265 }
 266
 267 static BlockAIOCB *read_fifo_child(QuorumAIOCB *acb);
 268
 269 static void quorum_copy_qiov(QEMUIOVector *dest, QEMUIOVector *source)
 270 {
 271     int i;
 272     assert(dest->niov == source->niov);
 273     assert(dest->size == source->size);
 274     for (i = 0; i < source->niov; i++) {
 275         assert(dest->iov[i].iov_len == source->iov[i].iov_len);
 276         memcpy(dest->iov[i].iov_base,
 277                source->iov[i].iov_base,
 278                source->iov[i].iov_len);
 279     }
 280 }
 281
 282 static void quorum_aio_cb(void *opaque, int ret)
 283 {
 284     QuorumChildRequest *sacb = opaque;
 285     QuorumAIOCB *acb = sacb->parent;
 286     BDRVQuorumState *s = acb->common.bs->opaque;
 287     QuorumOpType type;
 288     bool rewrite = false;
 289
 290     if (acb->is_read && s->read_pattern == QUORUM_READ_PATTERN_FIFO) {
 291         /* We try to read next child in FIFO order if we fail to read */
 292         if (ret < 0 && (acb->child_iter + 1) < s->num_children) {
 293             acb->child_iter++;
 294             read_fifo_child(acb);
 295             return;
 296         }
 297
 298         if (ret == 0) {
 299             quorum_copy_qiov(acb->qiov, &acb->qcrs[acb->child_iter].qiov);
 300         }
 301         acb->vote_ret = ret;
 302         quorum_aio_finalize(acb);
 303         return;
 304     }
 305
 306     type = acb->is_read ? QUORUM_OP_TYPE_READ : QUORUM_OP_TYPE_WRITE;
 307     sacb->ret = ret;
 308     acb->count++;
 309     if (ret == 0) {
 310         acb->success_count++;
 311     } else {
 312         quorum_report_bad(type, acb->sector_num, acb->nb_sectors,
 313                           sacb->aiocb->bs->node_name, ret);
 314     }
 315     assert(acb->count <= s->num_children);
 316     assert(acb->success_count <= s->num_children);
 317     if (acb->count < s->num_children) {
 318         return;
 319     }
 320
 321     /* Do the vote on read */
 322     if (acb->is_read) {
 323         rewrite = quorum_vote(acb);
 324     } else {
 325         quorum_has_too_much_io_failed(acb);
 326     }
 327
 328     /* if no rewrite is done the code will finish right away */
 329     if (!rewrite) {
 330         quorum_aio_finalize(acb);
 331     }
 332 }
 333
 334 static void quorum_report_bad_versions(BDRVQuorumState *s,
 335                                        QuorumAIOCB *acb,
 336                                        QuorumVoteValue *value)
 337 {
 338     QuorumVoteVersion *version;
 339     QuorumVoteItem *item;
 340
 341     QLIST_FOREACH(version, &acb->votes.vote_list, next) {
 342         if (acb->votes.compare(&version->value, value)) {
 343             continue;
 344         }
 345         QLIST_FOREACH(item, &version->items, next) {
 346             quorum_report_bad(QUORUM_OP_TYPE_READ, acb->sector_num,
 347                               acb->nb_sectors,
 348                               s->children[item->index]->bs->node_name, 0);
 349         }
 350     }
 351 }
 352
 353 static bool quorum_rewrite_bad_versions(BDRVQuorumState *s, QuorumAIOCB *acb,
 354                                         QuorumVoteValue *value)
 355 {
 356     QuorumVoteVersion *version;
 357     QuorumVoteItem *item;
 358     int count = 0;
 359
 360     /* first count the number of bad versions: done first to avoid concurrency
 361      * issues.
 362      */
 363     QLIST_FOREACH(version, &acb->votes.vote_list, next) {
 364         if (acb->votes.compare(&version->value, value)) {
 365             continue;
 366         }
 367         QLIST_FOREACH(item, &version->items, next) {
 368             count++;
 369         }
 370     }
 371
 372     /* quorum_rewrite_aio_cb will count down this to zero */
 373     acb->rewrite_count = count;
 374
 375     /* now fire the correcting rewrites */
 376     QLIST_FOREACH(version, &acb->votes.vote_list, next) {
 377         if (acb->votes.compare(&version->value, value)) {
 378             continue;
 379         }
 380         QLIST_FOREACH(item, &version->items, next) {
 381             bdrv_aio_writev(s->children[item->index]->bs, acb->sector_num,
 382                             acb->qiov, acb->nb_sectors, quorum_rewrite_aio_cb,
 383                             acb);
 384         }
 385     }
 386
 387     /* return true if any rewrite is done else false */
 388     return count;
 389 }
 390
 391 static void quorum_count_vote(QuorumVotes *votes,
 392                               QuorumVoteValue *value,
 393                               int index)
 394 {
 395     QuorumVoteVersion *v = NULL, *version = NULL;
 396     QuorumVoteItem *item;
 397
 398     /* look if we have something with this hash */
 399     QLIST_FOREACH(v, &votes->vote_list, next) {
 400         if (votes->compare(&v->value, value)) {
 401             version = v;
 402             break;
 403         }
 404     }
 405
 406     /* It's a version not yet in the list add it */
 407     if (!version) {
 408         version = g_new0(QuorumVoteVersion, 1);
 409         QLIST_INIT(&version->items);
 410         memcpy(&version->value, value, sizeof(version->value));
 411         version->index = index;
 412         version->vote_count = 0;
 413         QLIST_INSERT_HEAD(&votes->vote_list, version, next);
 414     }
 415
 416     version->vote_count++;
 417
 418     item = g_new0(QuorumVoteItem, 1);
 419     item->index = index;
 420     QLIST_INSERT_HEAD(&version->items, item, next);
 421 }
 422
 423 static void quorum_free_vote_list(QuorumVotes *votes)
 424 {
 425     QuorumVoteVersion *version, *next_version;
 426     QuorumVoteItem *item, *next_item;
 427
 428     QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
 429         QLIST_REMOVE(version, next);
 430         QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
 431             QLIST_REMOVE(item, next);
 432             g_free(item);
 433         }
 434         g_free(version);
 435     }
 436 }
 437
 438 static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
 439 {
 440     QEMUIOVector *qiov = &acb->qcrs[i].qiov;
 441     size_t len = sizeof(hash->h);
 442     uint8_t *data = hash->h;
 443
 444     /* XXX - would be nice if we could pass in the Error **
 445      * and propagate that back, but this quorum code is
 446      * restricted to just errno values currently */
 447     if (qcrypto_hash_bytesv(QCRYPTO_HASH_ALG_SHA256,
 448                             qiov->iov, qiov->niov,
 449                             &data, &len,
 450                             NULL) < 0) {
 451         return -EINVAL;
 452     }
 453
 454     return 0;
 455 }
 456
 457 static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
 458 {
 459     int max = 0;
 460     QuorumVoteVersion *candidate, *winner = NULL;
 461
 462     QLIST_FOREACH(candidate, &votes->vote_list, next) {
 463         if (candidate->vote_count > max) {
 464             max = candidate->vote_count;
 465             winner = candidate;
 466         }
 467     }
 468
 469     return winner;
 470 }
 471
 472 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
 473  * differing byte location. Yet it is handcoded to compare vectors one byte
 474  * after another so it does not benefit from the libc SIMD optimizations.
 475  * quorum_iovec_compare is written for speed and should be used in the non
 476  * blkverify mode of quorum.
 477  */
 478 static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
 479 {
 480     int i;
 481     int result;
 482
 483     assert(a->niov == b->niov);
 484     for (i = 0; i < a->niov; i++) {
 485         assert(a->iov[i].iov_len == b->iov[i].iov_len);
 486         result = memcmp(a->iov[i].iov_base,
 487                         b->iov[i].iov_base,
 488                         a->iov[i].iov_len);
 489         if (result) {
 490             return false;
 491         }
 492     }
 493
 494     return true;
 495 }
 496
 497 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB *acb,
 498                                           const char *fmt, ...)
 499 {
 500     va_list ap;
 501
 502     va_start(ap, fmt);
 503     fprintf(stderr, "quorum: sector_num=%" PRId64 " nb_sectors=%d ",
 504             acb->sector_num, acb->nb_sectors);
 505     vfprintf(stderr, fmt, ap);
 506     fprintf(stderr, "\n");
 507     va_end(ap);
 508     exit(1);
 509 }
 510
 511 static bool quorum_compare(QuorumAIOCB *acb,
 512                            QEMUIOVector *a,
 513                            QEMUIOVector *b)
 514 {
 515     BDRVQuorumState *s = acb->common.bs->opaque;
 516     ssize_t offset;
 517
 518     /* This driver will replace blkverify in this particular case */
 519     if (s->is_blkverify) {
 520         offset = qemu_iovec_compare(a, b);
 521         if (offset != -1) {
 522             quorum_err(acb, "contents mismatch in sector %" PRId64,
 523                        acb->sector_num +
 524                        (uint64_t)(offset / BDRV_SECTOR_SIZE));
 525         }
 526         return true;
 527     }
 528
 529     return quorum_iovec_compare(a, b);
 530 }
 531
 532 /* Do a vote to get the error code */
 533 static int quorum_vote_error(QuorumAIOCB *acb)
 534 {
 535     BDRVQuorumState *s = acb->common.bs->opaque;
 536     QuorumVoteVersion *winner = NULL;
 537     QuorumVotes error_votes;
 538     QuorumVoteValue result_value;
 539     int i, ret = 0;
 540     bool error = false;
 541
 542     QLIST_INIT(&error_votes.vote_list);
 543     error_votes.compare = quorum_64bits_compare;
 544
 545     for (i = 0; i < s->num_children; i++) {
 546         ret = acb->qcrs[i].ret;
 547         if (ret) {
 548             error = true;
 549             result_value.l = ret;
 550             quorum_count_vote(&error_votes, &result_value, i);
 551         }
 552     }
 553
 554     if (error) {
 555         winner = quorum_get_vote_winner(&error_votes);
 556         ret = winner->value.l;
 557     }
 558
 559     quorum_free_vote_list(&error_votes);
 560
 561     return ret;
 562 }
 563
 564 static bool quorum_vote(QuorumAIOCB *acb)
 565 {
 566     bool quorum = true;
 567     bool rewrite = false;
 568     int i, j, ret;
 569     QuorumVoteValue hash;
 570     BDRVQuorumState *s = acb->common.bs->opaque;
 571     QuorumVoteVersion *winner;
 572
 573     if (quorum_has_too_much_io_failed(acb)) {
 574         return false;
 575     }
 576
 577     /* get the index of the first successful read */
 578     for (i = 0; i < s->num_children; i++) {
 579         if (!acb->qcrs[i].ret) {
 580             break;
 581         }
 582     }
 583
 584     assert(i < s->num_children);
 585
 586     /* compare this read with all other successful reads stopping at quorum
 587      * failure
 588      */
 589     for (j = i + 1; j < s->num_children; j++) {
 590         if (acb->qcrs[j].ret) {
 591             continue;
 592         }
 593         quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
 594         if (!quorum) {
 595             break;
 596        }
 597     }
 598
 599     /* Every successful read agrees */
 600     if (quorum) {
 601         quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
 602         return false;
 603     }
 604
 605     /* compute hashes for each successful read, also store indexes */
 606     for (i = 0; i < s->num_children; i++) {
 607         if (acb->qcrs[i].ret) {
 608             continue;
 609         }
 610         ret = quorum_compute_hash(acb, i, &hash);
 611         /* if ever the hash computation failed */
 612         if (ret < 0) {
 613             acb->vote_ret = ret;
 614             goto free_exit;
 615         }
 616         quorum_count_vote(&acb->votes, &hash, i);
 617     }
 618
 619     /* vote to select the most represented version */
 620     winner = quorum_get_vote_winner(&acb->votes);
 621
 622     /* if the winner count is smaller than threshold the read fails */
 623     if (winner->vote_count < s->threshold) {
 624         quorum_report_failure(acb);
 625         acb->vote_ret = -EIO;
 626         goto free_exit;
 627     }
 628
 629     /* we have a winner: copy it */
 630     quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
 631
 632     /* some versions are bad print them */
 633     quorum_report_bad_versions(s, acb, &winner->value);
 634
 635     /* corruption correction is enabled */
 636     if (s->rewrite_corrupted) {
 637         rewrite = quorum_rewrite_bad_versions(s, acb, &winner->value);
 638     }
 639
 640 free_exit:
 641     /* free lists */
 642     quorum_free_vote_list(&acb->votes);
 643     return rewrite;
 644 }
 645
 646 static BlockAIOCB *read_quorum_children(QuorumAIOCB *acb)
 647 {
 648     BDRVQuorumState *s = acb->common.bs->opaque;
 649     int i;
 650
 651     for (i = 0; i < s->num_children; i++) {
 652         acb->qcrs[i].buf = qemu_blockalign(s->children[i]->bs, acb->qiov->size);
 653         qemu_iovec_init(&acb->qcrs[i].qiov, acb->qiov->niov);
 654         qemu_iovec_clone(&acb->qcrs[i].qiov, acb->qiov, acb->qcrs[i].buf);
 655     }
 656
 657     for (i = 0; i < s->num_children; i++) {
 658         acb->qcrs[i].aiocb = bdrv_aio_readv(s->children[i]->bs, acb->sector_num,
 659                                             &acb->qcrs[i].qiov, acb->nb_sectors,
 660                                             quorum_aio_cb, &acb->qcrs[i]);
 661     }
 662
 663     return &acb->common;
 664 }
 665
 666 static BlockAIOCB *read_fifo_child(QuorumAIOCB *acb)
 667 {
 668     BDRVQuorumState *s = acb->common.bs->opaque;
 669
 670     acb->qcrs[acb->child_iter].buf =
 671         qemu_blockalign(s->children[acb->child_iter]->bs, acb->qiov->size);
 672     qemu_iovec_init(&acb->qcrs[acb->child_iter].qiov, acb->qiov->niov);
 673     qemu_iovec_clone(&acb->qcrs[acb->child_iter].qiov, acb->qiov,
 674                      acb->qcrs[acb->child_iter].buf);
 675     acb->qcrs[acb->child_iter].aiocb =
 676         bdrv_aio_readv(s->children[acb->child_iter]->bs, acb->sector_num,
 677                        &acb->qcrs[acb->child_iter].qiov, acb->nb_sectors,
 678                        quorum_aio_cb, &acb->qcrs[acb->child_iter]);
 679
 680     return &acb->common;
 681 }
 682
 683 static BlockAIOCB *quorum_aio_readv(BlockDriverState *bs,
 684                                     int64_t sector_num,
 685                                     QEMUIOVector *qiov,
 686                                     int nb_sectors,
 687                                     BlockCompletionFunc *cb,
 688                                     void *opaque)
 689 {
 690     BDRVQuorumState *s = bs->opaque;
 691     QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num,
 692                                       nb_sectors, cb, opaque);
 693     acb->is_read = true;
 694
 695     if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
 696         acb->child_iter = s->num_children - 1;
 697         return read_quorum_children(acb);
 698     }
 699
 700     acb->child_iter = 0;
 701     return read_fifo_child(acb);
 702 }
 703
 704 static BlockAIOCB *quorum_aio_writev(BlockDriverState *bs,
 705                                      int64_t sector_num,
 706                                      QEMUIOVector *qiov,
 707                                      int nb_sectors,
 708                                      BlockCompletionFunc *cb,
 709                                      void *opaque)
 710 {
 711     BDRVQuorumState *s = bs->opaque;
 712     QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num, nb_sectors,
 713                                       cb, opaque);
 714     int i;
 715
 716     for (i = 0; i < s->num_children; i++) {
 717         acb->qcrs[i].aiocb = bdrv_aio_writev(s->children[i]->bs, sector_num,
 718                                              qiov, nb_sectors, &quorum_aio_cb,
 719                                              &acb->qcrs[i]);
 720     }
 721
 722     return &acb->common;
 723 }
 724
 725 static int64_t quorum_getlength(BlockDriverState *bs)
 726 {
 727     BDRVQuorumState *s = bs->opaque;
 728     int64_t result;
 729     int i;
 730
 731     /* check that all file have the same length */
 732     result = bdrv_getlength(s->children[0]->bs);
 733     if (result < 0) {
 734         return result;
 735     }
 736     for (i = 1; i < s->num_children; i++) {
 737         int64_t value = bdrv_getlength(s->children[i]->bs);
 738         if (value < 0) {
 739             return value;
 740         }
 741         if (value != result) {
 742             return -EIO;
 743         }
 744     }
 745
 746     return result;
 747 }
 748
 749 static void quorum_invalidate_cache(BlockDriverState *bs, Error **errp)
 750 {
 751     BDRVQuorumState *s = bs->opaque;
 752     Error *local_err = NULL;
 753     int i;
 754
 755     for (i = 0; i < s->num_children; i++) {
 756         bdrv_invalidate_cache(s->children[i]->bs, &local_err);
 757         if (local_err) {
 758             error_propagate(errp, local_err);
 759             return;
 760         }
 761     }
 762 }
 763
 764 static coroutine_fn int quorum_co_flush(BlockDriverState *bs)
 765 {
 766     BDRVQuorumState *s = bs->opaque;
 767     QuorumVoteVersion *winner = NULL;
 768     QuorumVotes error_votes;
 769     QuorumVoteValue result_value;
 770     int i;
 771     int result = 0;
 772     int success_count = 0;
 773
 774     QLIST_INIT(&error_votes.vote_list);
 775     error_votes.compare = quorum_64bits_compare;
 776
 777     for (i = 0; i < s->num_children; i++) {
 778         result = bdrv_co_flush(s->children[i]->bs);
 779         if (result) {
 780             quorum_report_bad(QUORUM_OP_TYPE_FLUSH, 0,
 781                               bdrv_nb_sectors(s->children[i]->bs),
 782                               s->children[i]->bs->node_name, result);
 783             result_value.l = result;
 784             quorum_count_vote(&error_votes, &result_value, i);
 785         } else {
 786             success_count++;
 787         }
 788     }
 789
 790     if (success_count >= s->threshold) {
 791         result = 0;
 792     } else {
 793         winner = quorum_get_vote_winner(&error_votes);
 794         result = winner->value.l;
 795     }
 796     quorum_free_vote_list(&error_votes);
 797
 798     return result;
 799 }
 800
 801 static bool quorum_recurse_is_first_non_filter(BlockDriverState *bs,
 802                                                BlockDriverState *candidate)
 803 {
 804     BDRVQuorumState *s = bs->opaque;
 805     int i;
 806
 807     for (i = 0; i < s->num_children; i++) {
 808         bool perm = bdrv_recurse_is_first_non_filter(s->children[i]->bs,
 809                                                      candidate);
 810         if (perm) {
 811             return true;
 812         }
 813     }
 814
 815     return false;
 816 }
 817
 818 static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
 819 {
 820
 821     if (threshold < 1) {
 822         error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
 823                    "vote-threshold", "value >= 1");
 824         return -ERANGE;
 825     }
 826
 827     if (threshold > num_children) {
 828         error_setg(errp, "threshold may not exceed children count");
 829         return -ERANGE;
 830     }
 831
 832     return 0;
 833 }
 834
 835 static QemuOptsList quorum_runtime_opts = {
 836     .name = "quorum",
 837     .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
 838     .desc = {
 839         {
 840             .name = QUORUM_OPT_VOTE_THRESHOLD,
 841             .type = QEMU_OPT_NUMBER,
 842             .help = "The number of vote needed for reaching quorum",
 843         },
 844         {
 845             .name = QUORUM_OPT_BLKVERIFY,
 846             .type = QEMU_OPT_BOOL,
 847             .help = "Trigger block verify mode if set",
 848         },
 849         {
 850             .name = QUORUM_OPT_REWRITE,
 851             .type = QEMU_OPT_BOOL,
 852             .help = "Rewrite corrupted block on read quorum",
 853         },
 854         {
 855             .name = QUORUM_OPT_READ_PATTERN,
 856             .type = QEMU_OPT_STRING,
 857             .help = "Allowed pattern: quorum, fifo. Quorum is default",
 858         },
 859         { /* end of list */ }
 860     },
 861 };
 862
 863 static int parse_read_pattern(const char *opt)
 864 {
 865     int i;
 866
 867     if (!opt) {
 868         /* Set quorum as default */
 869         return QUORUM_READ_PATTERN_QUORUM;
 870     }
 871
 872     for (i = 0; i < QUORUM_READ_PATTERN__MAX; i++) {
 873         if (!strcmp(opt, QuorumReadPattern_lookup[i])) {
 874             return i;
 875         }
 876     }
 877
 878     return -EINVAL;
 879 }
 880
 881 static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
 882                        Error **errp)
 883 {
 884     BDRVQuorumState *s = bs->opaque;
 885     Error *local_err = NULL;
 886     QemuOpts *opts = NULL;
 887     bool *opened;
 888     int i;
 889     int ret = 0;
 890
 891     qdict_flatten(options);
 892
 893     /* count how many different children are present */
 894     s->num_children = qdict_array_entries(options, "children.");
 895     if (s->num_children < 0) {
 896         error_setg(&local_err, "Option children is not a valid array");
 897         ret = -EINVAL;
 898         goto exit;
 899     }
 900     if (s->num_children < 2) {
 901         error_setg(&local_err,
 902                    "Number of provided children must be greater than 1");
 903         ret = -EINVAL;
 904         goto exit;
 905     }
 906
 907     opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
 908     qemu_opts_absorb_qdict(opts, options, &local_err);
 909     if (local_err) {
 910         ret = -EINVAL;
 911         goto exit;
 912     }
 913
 914     s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
 915     /* and validate it against s->num_children */
 916     ret = quorum_valid_threshold(s->threshold, s->num_children, &local_err);
 917     if (ret < 0) {
 918         goto exit;
 919     }
 920
 921     ret = parse_read_pattern(qemu_opt_get(opts, QUORUM_OPT_READ_PATTERN));
 922     if (ret < 0) {
 923         error_setg(&local_err, "Please set read-pattern as fifo or quorum");
 924         goto exit;
 925     }
 926     s->read_pattern = ret;
 927
 928     if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
 929         /* is the driver in blkverify mode */
 930         if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false) &&
 931             s->num_children == 2 && s->threshold == 2) {
 932             s->is_blkverify = true;
 933         } else if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false)) {
 934             fprintf(stderr, "blkverify mode is set by setting blkverify=on "
 935                     "and using two files with vote_threshold=2\n");
 936         }
 937
 938         s->rewrite_corrupted = qemu_opt_get_bool(opts, QUORUM_OPT_REWRITE,
 939                                                  false);
 940         if (s->rewrite_corrupted && s->is_blkverify) {
 941             error_setg(&local_err,
 942                        "rewrite-corrupted=on cannot be used with blkverify=on");
 943             ret = -EINVAL;
 944             goto exit;
 945         }
 946     }
 947
 948     /* allocate the children array */
 949     s->children = g_new0(BdrvChild *, s->num_children);
 950     opened = g_new0(bool, s->num_children);
 951
 952     for (i = 0; i < s->num_children; i++) {
 953         char indexstr[32];
 954         ret = snprintf(indexstr, 32, "children.%d", i);
 955         assert(ret < 32);
 956
 957         s->children[i] = bdrv_open_child(NULL, options, indexstr, bs,
 958                                          &child_format, false, &local_err);
 959         if (local_err) {
 960             ret = -EINVAL;
 961             goto close_exit;
 962         }
 963
 964         opened[i] = true;
 965     }
 966
 967     g_free(opened);
 968     goto exit;
 969
 970 close_exit:
 971     /* cleanup on error */
 972     for (i = 0; i < s->num_children; i++) {
 973         if (!opened[i]) {
 974             continue;
 975         }
 976         bdrv_unref_child(bs, s->children[i]);
 977     }
 978     g_free(s->children);
 979     g_free(opened);
 980 exit:
 981     qemu_opts_del(opts);
 982     /* propagate error */
 983     if (local_err) {
 984         error_propagate(errp, local_err);
 985     }
 986     return ret;
 987 }
 988
 989 static void quorum_close(BlockDriverState *bs)
 990 {
 991     BDRVQuorumState *s = bs->opaque;
 992     int i;
 993
 994     for (i = 0; i < s->num_children; i++) {
 995         bdrv_unref_child(bs, s->children[i]);
 996     }
 997
 998     g_free(s->children);
 999 }
1000
1001 static void quorum_detach_aio_context(BlockDriverState *bs)
1002 {
1003     BDRVQuorumState *s = bs->opaque;
1004     int i;
1005
1006     for (i = 0; i < s->num_children; i++) {
1007         bdrv_detach_aio_context(s->children[i]->bs);
1008     }
1009 }
1010
1011 static void quorum_attach_aio_context(BlockDriverState *bs,
1012                                       AioContext *new_context)
1013 {
1014     BDRVQuorumState *s = bs->opaque;
1015     int i;
1016
1017     for (i = 0; i < s->num_children; i++) {
1018         bdrv_attach_aio_context(s->children[i]->bs, new_context);
1019     }
1020 }
1021
1022 static void quorum_refresh_filename(BlockDriverState *bs, QDict *options)
1023 {
1024     BDRVQuorumState *s = bs->opaque;
1025     QDict *opts;
1026     QList *children;
1027     int i;
1028
1029     for (i = 0; i < s->num_children; i++) {
1030         bdrv_refresh_filename(s->children[i]->bs);
1031         if (!s->children[i]->bs->full_open_options) {
1032             return;
1033         }
1034     }
1035
1036     children = qlist_new();
1037     for (i = 0; i < s->num_children; i++) {
1038         QINCREF(s->children[i]->bs->full_open_options);
1039         qlist_append_obj(children,
1040                          QOBJECT(s->children[i]->bs->full_open_options));
1041     }
1042
1043     opts = qdict_new();
1044     qdict_put_obj(opts, "driver", QOBJECT(qstring_from_str("quorum")));
1045     qdict_put_obj(opts, QUORUM_OPT_VOTE_THRESHOLD,
1046                   QOBJECT(qint_from_int(s->threshold)));
1047     qdict_put_obj(opts, QUORUM_OPT_BLKVERIFY,
1048                   QOBJECT(qbool_from_bool(s->is_blkverify)));
1049     qdict_put_obj(opts, QUORUM_OPT_REWRITE,
1050                   QOBJECT(qbool_from_bool(s->rewrite_corrupted)));
1051     qdict_put_obj(opts, "children", QOBJECT(children));
1052
1053     bs->full_open_options = opts;
1054 }
1055
1056 static BlockDriver bdrv_quorum = {
1057     .format_name                        = "quorum",
1058     .protocol_name                      = "quorum",
1059
1060     .instance_size                      = sizeof(BDRVQuorumState),
1061
1062     .bdrv_file_open                     = quorum_open,
1063     .bdrv_close                         = quorum_close,
1064     .bdrv_refresh_filename              = quorum_refresh_filename,
1065
1066     .bdrv_co_flush_to_disk              = quorum_co_flush,
1067
1068     .bdrv_getlength                     = quorum_getlength,
1069
1070     .bdrv_aio_readv                     = quorum_aio_readv,
1071     .bdrv_aio_writev                    = quorum_aio_writev,
1072     .bdrv_invalidate_cache              = quorum_invalidate_cache,
1073
1074     .bdrv_detach_aio_context            = quorum_detach_aio_context,
1075     .bdrv_attach_aio_context            = quorum_attach_aio_context,
1076
1077     .is_filter                          = true,
1078     .bdrv_recurse_is_first_non_filter   = quorum_recurse_is_first_non_filter,
1079 };
1080
1081 static void bdrv_quorum_init(void)
1082 {
1083     if (!qcrypto_hash_supports(QCRYPTO_HASH_ALG_SHA256)) {
1084         /* SHA256 hash support is required for quorum device */
1085         return;
1086     }
1087     bdrv_register(&bdrv_quorum);
1088 }
1089
1090 block_init(bdrv_quorum_init);