qemu/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     bool rewrite = false;
 288
 289     if (ret == 0) {
 290         acb->success_count++;
 291     } else {
 292         QuorumOpType type;
 293         type = acb->is_read ? QUORUM_OP_TYPE_READ : QUORUM_OP_TYPE_WRITE;
 294         quorum_report_bad(type, acb->sector_num, acb->nb_sectors,
 295                           sacb->aiocb->bs->node_name, ret);
 296     }
 297
 298     if (acb->is_read && s->read_pattern == QUORUM_READ_PATTERN_FIFO) {
 299         /* We try to read next child in FIFO order if we fail to read */
 300         if (ret < 0 && (acb->child_iter + 1) < s->num_children) {
 301             acb->child_iter++;
 302             read_fifo_child(acb);
 303             return;
 304         }
 305
 306         if (ret == 0) {
 307             quorum_copy_qiov(acb->qiov, &acb->qcrs[acb->child_iter].qiov);
 308         }
 309         acb->vote_ret = ret;
 310         quorum_aio_finalize(acb);
 311         return;
 312     }
 313
 314     sacb->ret = ret;
 315     acb->count++;
 316     assert(acb->count <= s->num_children);
 317     assert(acb->success_count <= s->num_children);
 318     if (acb->count < s->num_children) {
 319         return;
 320     }
 321
 322     /* Do the vote on read */
 323     if (acb->is_read) {
 324         rewrite = quorum_vote(acb);
 325     } else {
 326         quorum_has_too_much_io_failed(acb);
 327     }
 328
 329     /* if no rewrite is done the code will finish right away */
 330     if (!rewrite) {
 331         quorum_aio_finalize(acb);
 332     }
 333 }
 334
 335 static void quorum_report_bad_versions(BDRVQuorumState *s,
 336                                        QuorumAIOCB *acb,
 337                                        QuorumVoteValue *value)
 338 {
 339     QuorumVoteVersion *version;
 340     QuorumVoteItem *item;
 341
 342     QLIST_FOREACH(version, &acb->votes.vote_list, next) {
 343         if (acb->votes.compare(&version->value, value)) {
 344             continue;
 345         }
 346         QLIST_FOREACH(item, &version->items, next) {
 347             quorum_report_bad(QUORUM_OP_TYPE_READ, acb->sector_num,
 348                               acb->nb_sectors,
 349                               s->children[item->index]->bs->node_name, 0);
 350         }
 351     }
 352 }
 353
 354 static bool quorum_rewrite_bad_versions(BDRVQuorumState *s, QuorumAIOCB *acb,
 355                                         QuorumVoteValue *value)
 356 {
 357     QuorumVoteVersion *version;
 358     QuorumVoteItem *item;
 359     int count = 0;
 360
 361     /* first count the number of bad versions: done first to avoid concurrency
 362      * issues.
 363      */
 364     QLIST_FOREACH(version, &acb->votes.vote_list, next) {
 365         if (acb->votes.compare(&version->value, value)) {
 366             continue;
 367         }
 368         QLIST_FOREACH(item, &version->items, next) {
 369             count++;
 370         }
 371     }
 372
 373     /* quorum_rewrite_aio_cb will count down this to zero */
 374     acb->rewrite_count = count;
 375
 376     /* now fire the correcting rewrites */
 377     QLIST_FOREACH(version, &acb->votes.vote_list, next) {
 378         if (acb->votes.compare(&version->value, value)) {
 379             continue;
 380         }
 381         QLIST_FOREACH(item, &version->items, next) {
 382             bdrv_aio_writev(s->children[item->index]->bs, acb->sector_num,
 383                             acb->qiov, acb->nb_sectors, quorum_rewrite_aio_cb,
 384                             acb);
 385         }
 386     }
 387
 388     /* return true if any rewrite is done else false */
 389     return count;
 390 }
 391
 392 static void quorum_count_vote(QuorumVotes *votes,
 393                               QuorumVoteValue *value,
 394                               int index)
 395 {
 396     QuorumVoteVersion *v = NULL, *version = NULL;
 397     QuorumVoteItem *item;
 398
 399     /* look if we have something with this hash */
 400     QLIST_FOREACH(v, &votes->vote_list, next) {
 401         if (votes->compare(&v->value, value)) {
 402             version = v;
 403             break;
 404         }
 405     }
 406
 407     /* It's a version not yet in the list add it */
 408     if (!version) {
 409         version = g_new0(QuorumVoteVersion, 1);
 410         QLIST_INIT(&version->items);
 411         memcpy(&version->value, value, sizeof(version->value));
 412         version->index = index;
 413         version->vote_count = 0;
 414         QLIST_INSERT_HEAD(&votes->vote_list, version, next);
 415     }
 416
 417     version->vote_count++;
 418
 419     item = g_new0(QuorumVoteItem, 1);
 420     item->index = index;
 421     QLIST_INSERT_HEAD(&version->items, item, next);
 422 }
 423
 424 static void quorum_free_vote_list(QuorumVotes *votes)
 425 {
 426     QuorumVoteVersion *version, *next_version;
 427     QuorumVoteItem *item, *next_item;
 428
 429     QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
 430         QLIST_REMOVE(version, next);
 431         QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
 432             QLIST_REMOVE(item, next);
 433             g_free(item);
 434         }
 435         g_free(version);
 436     }
 437 }
 438
 439 static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
 440 {
 441     QEMUIOVector *qiov = &acb->qcrs[i].qiov;
 442     size_t len = sizeof(hash->h);
 443     uint8_t *data = hash->h;
 444
 445     /* XXX - would be nice if we could pass in the Error **
 446      * and propagate that back, but this quorum code is
 447      * restricted to just errno values currently */
 448     if (qcrypto_hash_bytesv(QCRYPTO_HASH_ALG_SHA256,
 449                             qiov->iov, qiov->niov,
 450                             &data, &len,
 451                             NULL) < 0) {
 452         return -EINVAL;
 453     }
 454
 455     return 0;
 456 }
 457
 458 static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
 459 {
 460     int max = 0;
 461     QuorumVoteVersion *candidate, *winner = NULL;
 462
 463     QLIST_FOREACH(candidate, &votes->vote_list, next) {
 464         if (candidate->vote_count > max) {
 465             max = candidate->vote_count;
 466             winner = candidate;
 467         }
 468     }
 469
 470     return winner;
 471 }
 472
 473 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
 474  * differing byte location. Yet it is handcoded to compare vectors one byte
 475  * after another so it does not benefit from the libc SIMD optimizations.
 476  * quorum_iovec_compare is written for speed and should be used in the non
 477  * blkverify mode of quorum.
 478  */
 479 static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
 480 {
 481     int i;
 482     int result;
 483
 484     assert(a->niov == b->niov);
 485     for (i = 0; i < a->niov; i++) {
 486         assert(a->iov[i].iov_len == b->iov[i].iov_len);
 487         result = memcmp(a->iov[i].iov_base,
 488                         b->iov[i].iov_base,
 489                         a->iov[i].iov_len);
 490         if (result) {
 491             return false;
 492         }
 493     }
 494
 495     return true;
 496 }
 497
 498 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB *acb,
 499                                           const char *fmt, ...)
 500 {
 501     va_list ap;
 502
 503     va_start(ap, fmt);
 504     fprintf(stderr, "quorum: sector_num=%" PRId64 " nb_sectors=%d ",
 505             acb->sector_num, acb->nb_sectors);
 506     vfprintf(stderr, fmt, ap);
 507     fprintf(stderr, "\n");
 508     va_end(ap);
 509     exit(1);
 510 }
 511
 512 static bool quorum_compare(QuorumAIOCB *acb,
 513                            QEMUIOVector *a,
 514                            QEMUIOVector *b)
 515 {
 516     BDRVQuorumState *s = acb->common.bs->opaque;
 517     ssize_t offset;
 518
 519     /* This driver will replace blkverify in this particular case */
 520     if (s->is_blkverify) {
 521         offset = qemu_iovec_compare(a, b);
 522         if (offset != -1) {
 523             quorum_err(acb, "contents mismatch in sector %" PRId64,
 524                        acb->sector_num +
 525                        (uint64_t)(offset / BDRV_SECTOR_SIZE));
 526         }
 527         return true;
 528     }
 529
 530     return quorum_iovec_compare(a, b);
 531 }
 532
 533 /* Do a vote to get the error code */
 534 static int quorum_vote_error(QuorumAIOCB *acb)
 535 {
 536     BDRVQuorumState *s = acb->common.bs->opaque;
 537     QuorumVoteVersion *winner = NULL;
 538     QuorumVotes error_votes;
 539     QuorumVoteValue result_value;
 540     int i, ret = 0;
 541     bool error = false;
 542
 543     QLIST_INIT(&error_votes.vote_list);
 544     error_votes.compare = quorum_64bits_compare;
 545
 546     for (i = 0; i < s->num_children; i++) {
 547         ret = acb->qcrs[i].ret;
 548         if (ret) {
 549             error = true;
 550             result_value.l = ret;
 551             quorum_count_vote(&error_votes, &result_value, i);
 552         }
 553     }
 554
 555     if (error) {
 556         winner = quorum_get_vote_winner(&error_votes);
 557         ret = winner->value.l;
 558     }
 559
 560     quorum_free_vote_list(&error_votes);
 561
 562     return ret;
 563 }
 564
 565 static bool quorum_vote(QuorumAIOCB *acb)
 566 {
 567     bool quorum = true;
 568     bool rewrite = false;
 569     int i, j, ret;
 570     QuorumVoteValue hash;
 571     BDRVQuorumState *s = acb->common.bs->opaque;
 572     QuorumVoteVersion *winner;
 573
 574     if (quorum_has_too_much_io_failed(acb)) {
 575         return false;
 576     }
 577
 578     /* get the index of the first successful read */
 579     for (i = 0; i < s->num_children; i++) {
 580         if (!acb->qcrs[i].ret) {
 581             break;
 582         }
 583     }
 584
 585     assert(i < s->num_children);
 586
 587     /* compare this read with all other successful reads stopping at quorum
 588      * failure
 589      */
 590     for (j = i + 1; j < s->num_children; j++) {
 591         if (acb->qcrs[j].ret) {
 592             continue;
 593         }
 594         quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
 595         if (!quorum) {
 596             break;
 597        }
 598     }
 599
 600     /* Every successful read agrees */
 601     if (quorum) {
 602         quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
 603         return false;
 604     }
 605
 606     /* compute hashes for each successful read, also store indexes */
 607     for (i = 0; i < s->num_children; i++) {
 608         if (acb->qcrs[i].ret) {
 609             continue;
 610         }
 611         ret = quorum_compute_hash(acb, i, &hash);
 612         /* if ever the hash computation failed */
 613         if (ret < 0) {
 614             acb->vote_ret = ret;
 615             goto free_exit;
 616         }
 617         quorum_count_vote(&acb->votes, &hash, i);
 618     }
 619
 620     /* vote to select the most represented version */
 621     winner = quorum_get_vote_winner(&acb->votes);
 622
 623     /* if the winner count is smaller than threshold the read fails */
 624     if (winner->vote_count < s->threshold) {
 625         quorum_report_failure(acb);
 626         acb->vote_ret = -EIO;
 627         goto free_exit;
 628     }
 629
 630     /* we have a winner: copy it */
 631     quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
 632
 633     /* some versions are bad print them */
 634     quorum_report_bad_versions(s, acb, &winner->value);
 635
 636     /* corruption correction is enabled */
 637     if (s->rewrite_corrupted) {
 638         rewrite = quorum_rewrite_bad_versions(s, acb, &winner->value);
 639     }
 640
 641 free_exit:
 642     /* free lists */
 643     quorum_free_vote_list(&acb->votes);
 644     return rewrite;
 645 }
 646
 647 static BlockAIOCB *read_quorum_children(QuorumAIOCB *acb)
 648 {
 649     BDRVQuorumState *s = acb->common.bs->opaque;
 650     int i;
 651
 652     for (i = 0; i < s->num_children; i++) {
 653         acb->qcrs[i].buf = qemu_blockalign(s->children[i]->bs, acb->qiov->size);
 654         qemu_iovec_init(&acb->qcrs[i].qiov, acb->qiov->niov);
 655         qemu_iovec_clone(&acb->qcrs[i].qiov, acb->qiov, acb->qcrs[i].buf);
 656     }
 657
 658     for (i = 0; i < s->num_children; i++) {
 659         acb->qcrs[i].aiocb = bdrv_aio_readv(s->children[i]->bs, acb->sector_num,
 660                                             &acb->qcrs[i].qiov, acb->nb_sectors,
 661                                             quorum_aio_cb, &acb->qcrs[i]);
 662     }
 663
 664     return &acb->common;
 665 }
 666
 667 static BlockAIOCB *read_fifo_child(QuorumAIOCB *acb)
 668 {
 669     BDRVQuorumState *s = acb->common.bs->opaque;
 670
 671     acb->qcrs[acb->child_iter].buf =
 672         qemu_blockalign(s->children[acb->child_iter]->bs, acb->qiov->size);
 673     qemu_iovec_init(&acb->qcrs[acb->child_iter].qiov, acb->qiov->niov);
 674     qemu_iovec_clone(&acb->qcrs[acb->child_iter].qiov, acb->qiov,
 675                      acb->qcrs[acb->child_iter].buf);
 676     acb->qcrs[acb->child_iter].aiocb =
 677         bdrv_aio_readv(s->children[acb->child_iter]->bs, acb->sector_num,
 678                        &acb->qcrs[acb->child_iter].qiov, acb->nb_sectors,
 679                        quorum_aio_cb, &acb->qcrs[acb->child_iter]);
 680
 681     return &acb->common;
 682 }
 683
 684 static BlockAIOCB *quorum_aio_readv(BlockDriverState *bs,
 685                                     int64_t sector_num,
 686                                     QEMUIOVector *qiov,
 687                                     int nb_sectors,
 688                                     BlockCompletionFunc *cb,
 689                                     void *opaque)
 690 {
 691     BDRVQuorumState *s = bs->opaque;
 692     QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num,
 693                                       nb_sectors, cb, opaque);
 694     acb->is_read = true;
 695
 696     if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
 697         acb->child_iter = s->num_children - 1;
 698         return read_quorum_children(acb);
 699     }
 700
 701     acb->child_iter = 0;
 702     return read_fifo_child(acb);
 703 }
 704
 705 static BlockAIOCB *quorum_aio_writev(BlockDriverState *bs,
 706                                      int64_t sector_num,
 707                                      QEMUIOVector *qiov,
 708                                      int nb_sectors,
 709                                      BlockCompletionFunc *cb,
 710                                      void *opaque)
 711 {
 712     BDRVQuorumState *s = bs->opaque;
 713     QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num, nb_sectors,
 714                                       cb, opaque);
 715     int i;
 716
 717     for (i = 0; i < s->num_children; i++) {
 718         acb->qcrs[i].aiocb = bdrv_aio_writev(s->children[i]->bs, sector_num,
 719                                              qiov, nb_sectors, &quorum_aio_cb,
 720                                              &acb->qcrs[i]);
 721     }
 722
 723     return &acb->common;
 724 }
 725
 726 static int64_t quorum_getlength(BlockDriverState *bs)
 727 {
 728     BDRVQuorumState *s = bs->opaque;
 729     int64_t result;
 730     int i;
 731
 732     /* check that all file have the same length */
 733     result = bdrv_getlength(s->children[0]->bs);
 734     if (result < 0) {
 735         return result;
 736     }
 737     for (i = 1; i < s->num_children; i++) {
 738         int64_t value = bdrv_getlength(s->children[i]->bs);
 739         if (value < 0) {
 740             return value;
 741         }
 742         if (value != result) {
 743             return -EIO;
 744         }
 745     }
 746
 747     return result;
 748 }
 749
 750 static void quorum_invalidate_cache(BlockDriverState *bs, Error **errp)
 751 {
 752     BDRVQuorumState *s = bs->opaque;
 753     Error *local_err = NULL;
 754     int i;
 755
 756     for (i = 0; i < s->num_children; i++) {
 757         bdrv_invalidate_cache(s->children[i]->bs, &local_err);
 758         if (local_err) {
 759             error_propagate(errp, local_err);
 760             return;
 761         }
 762     }
 763 }
 764
 765 static coroutine_fn int quorum_co_flush(BlockDriverState *bs)
 766 {
 767     BDRVQuorumState *s = bs->opaque;
 768     QuorumVoteVersion *winner = NULL;
 769     QuorumVotes error_votes;
 770     QuorumVoteValue result_value;
 771     int i;
 772     int result = 0;
 773     int success_count = 0;
 774
 775     QLIST_INIT(&error_votes.vote_list);
 776     error_votes.compare = quorum_64bits_compare;
 777
 778     for (i = 0; i < s->num_children; i++) {
 779         result = bdrv_co_flush(s->children[i]->bs);
 780         if (result) {
 781             quorum_report_bad(QUORUM_OP_TYPE_FLUSH, 0,
 782                               bdrv_nb_sectors(s->children[i]->bs),
 783                               s->children[i]->bs->node_name, result);
 784             result_value.l = result;
 785             quorum_count_vote(&error_votes, &result_value, i);
 786         } else {
 787             success_count++;
 788         }
 789     }
 790
 791     if (success_count >= s->threshold) {
 792         result = 0;
 793     } else {
 794         winner = quorum_get_vote_winner(&error_votes);
 795         result = winner->value.l;
 796     }
 797     quorum_free_vote_list(&error_votes);
 798
 799     return result;
 800 }
 801
 802 static bool quorum_recurse_is_first_non_filter(BlockDriverState *bs,
 803                                                BlockDriverState *candidate)
 804 {
 805     BDRVQuorumState *s = bs->opaque;
 806     int i;
 807
 808     for (i = 0; i < s->num_children; i++) {
 809         bool perm = bdrv_recurse_is_first_non_filter(s->children[i]->bs,
 810                                                      candidate);
 811         if (perm) {
 812             return true;
 813         }
 814     }
 815
 816     return false;
 817 }
 818
 819 static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
 820 {
 821
 822     if (threshold < 1) {
 823         error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
 824                    "vote-threshold", "value >= 1");
 825         return -ERANGE;
 826     }
 827
 828     if (threshold > num_children) {
 829         error_setg(errp, "threshold may not exceed children count");
 830         return -ERANGE;
 831     }
 832
 833     return 0;
 834 }
 835
 836 static QemuOptsList quorum_runtime_opts = {
 837     .name = "quorum",
 838     .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
 839     .desc = {
 840         {
 841             .name = QUORUM_OPT_VOTE_THRESHOLD,
 842             .type = QEMU_OPT_NUMBER,
 843             .help = "The number of vote needed for reaching quorum",
 844         },
 845         {
 846             .name = QUORUM_OPT_BLKVERIFY,
 847             .type = QEMU_OPT_BOOL,
 848             .help = "Trigger block verify mode if set",
 849         },
 850         {
 851             .name = QUORUM_OPT_REWRITE,
 852             .type = QEMU_OPT_BOOL,
 853             .help = "Rewrite corrupted block on read quorum",
 854         },
 855         {
 856             .name = QUORUM_OPT_READ_PATTERN,
 857             .type = QEMU_OPT_STRING,
 858             .help = "Allowed pattern: quorum, fifo. Quorum is default",
 859         },
 860         { /* end of list */ }
 861     },
 862 };
 863
 864 static int parse_read_pattern(const char *opt)
 865 {
 866     int i;
 867
 868     if (!opt) {
 869         /* Set quorum as default */
 870         return QUORUM_READ_PATTERN_QUORUM;
 871     }
 872
 873     for (i = 0; i < QUORUM_READ_PATTERN__MAX; i++) {
 874         if (!strcmp(opt, QuorumReadPattern_lookup[i])) {
 875             return i;
 876         }
 877     }
 878
 879     return -EINVAL;
 880 }
 881
 882 static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
 883                        Error **errp)
 884 {
 885     BDRVQuorumState *s = bs->opaque;
 886     Error *local_err = NULL;
 887     QemuOpts *opts = NULL;
 888     bool *opened;
 889     int i;
 890     int ret = 0;
 891
 892     qdict_flatten(options);
 893
 894     /* count how many different children are present */
 895     s->num_children = qdict_array_entries(options, "children.");
 896     if (s->num_children < 0) {
 897         error_setg(&local_err, "Option children is not a valid array");
 898         ret = -EINVAL;
 899         goto exit;
 900     }
 901     if (s->num_children < 2) {
 902         error_setg(&local_err,
 903                    "Number of provided children must be greater than 1");
 904         ret = -EINVAL;
 905         goto exit;
 906     }
 907
 908     opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
 909     qemu_opts_absorb_qdict(opts, options, &local_err);
 910     if (local_err) {
 911         ret = -EINVAL;
 912         goto exit;
 913     }
 914
 915     s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
 916     /* and validate it against s->num_children */
 917     ret = quorum_valid_threshold(s->threshold, s->num_children, &local_err);
 918     if (ret < 0) {
 919         goto exit;
 920     }
 921
 922     ret = parse_read_pattern(qemu_opt_get(opts, QUORUM_OPT_READ_PATTERN));
 923     if (ret < 0) {
 924         error_setg(&local_err, "Please set read-pattern as fifo or quorum");
 925         goto exit;
 926     }
 927     s->read_pattern = ret;
 928
 929     if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
 930         /* is the driver in blkverify mode */
 931         if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false) &&
 932             s->num_children == 2 && s->threshold == 2) {
 933             s->is_blkverify = true;
 934         } else if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false)) {
 935             fprintf(stderr, "blkverify mode is set by setting blkverify=on "
 936                     "and using two files with vote_threshold=2\n");
 937         }
 938
 939         s->rewrite_corrupted = qemu_opt_get_bool(opts, QUORUM_OPT_REWRITE,
 940                                                  false);
 941         if (s->rewrite_corrupted && s->is_blkverify) {
 942             error_setg(&local_err,
 943                        "rewrite-corrupted=on cannot be used with blkverify=on");
 944             ret = -EINVAL;
 945             goto exit;
 946         }
 947     }
 948
 949     /* allocate the children array */
 950     s->children = g_new0(BdrvChild *, s->num_children);
 951     opened = g_new0(bool, s->num_children);
 952
 953     for (i = 0; i < s->num_children; i++) {
 954         char indexstr[32];
 955         ret = snprintf(indexstr, 32, "children.%d", i);
 956         assert(ret < 32);
 957
 958         s->children[i] = bdrv_open_child(NULL, options, indexstr, bs,
 959                                          &child_format, false, &local_err);
 960         if (local_err) {
 961             ret = -EINVAL;
 962             goto close_exit;
 963         }
 964
 965         opened[i] = true;
 966     }
 967
 968     g_free(opened);
 969     goto exit;
 970
 971 close_exit:
 972     /* cleanup on error */
 973     for (i = 0; i < s->num_children; i++) {
 974         if (!opened[i]) {
 975             continue;
 976         }
 977         bdrv_unref_child(bs, s->children[i]);
 978     }
 979     g_free(s->children);
 980     g_free(opened);
 981 exit:
 982     qemu_opts_del(opts);
 983     /* propagate error */
 984     if (local_err) {
 985         error_propagate(errp, local_err);
 986     }
 987     return ret;
 988 }
 989
 990 static void quorum_close(BlockDriverState *bs)
 991 {
 992     BDRVQuorumState *s = bs->opaque;
 993     int i;
 994
 995     for (i = 0; i < s->num_children; i++) {
 996         bdrv_unref_child(bs, s->children[i]);
 997     }
 998
 999     g_free(s->children);
1000 }
1001
1002 static void quorum_detach_aio_context(BlockDriverState *bs)
1003 {
1004     BDRVQuorumState *s = bs->opaque;
1005     int i;
1006
1007     for (i = 0; i < s->num_children; i++) {
1008         bdrv_detach_aio_context(s->children[i]->bs);
1009     }
1010 }
1011
1012 static void quorum_attach_aio_context(BlockDriverState *bs,
1013                                       AioContext *new_context)
1014 {
1015     BDRVQuorumState *s = bs->opaque;
1016     int i;
1017
1018     for (i = 0; i < s->num_children; i++) {
1019         bdrv_attach_aio_context(s->children[i]->bs, new_context);
1020     }
1021 }
1022
1023 static void quorum_refresh_filename(BlockDriverState *bs, QDict *options)
1024 {
1025     BDRVQuorumState *s = bs->opaque;
1026     QDict *opts;
1027     QList *children;
1028     int i;
1029
1030     for (i = 0; i < s->num_children; i++) {
1031         bdrv_refresh_filename(s->children[i]->bs);
1032         if (!s->children[i]->bs->full_open_options) {
1033             return;
1034         }
1035     }
1036
1037     children = qlist_new();
1038     for (i = 0; i < s->num_children; i++) {
1039         QINCREF(s->children[i]->bs->full_open_options);
1040         qlist_append_obj(children,
1041                          QOBJECT(s->children[i]->bs->full_open_options));
1042     }
1043
1044     opts = qdict_new();
1045     qdict_put_obj(opts, "driver", QOBJECT(qstring_from_str("quorum")));
1046     qdict_put_obj(opts, QUORUM_OPT_VOTE_THRESHOLD,
1047                   QOBJECT(qint_from_int(s->threshold)));
1048     qdict_put_obj(opts, QUORUM_OPT_BLKVERIFY,
1049                   QOBJECT(qbool_from_bool(s->is_blkverify)));
1050     qdict_put_obj(opts, QUORUM_OPT_REWRITE,
1051                   QOBJECT(qbool_from_bool(s->rewrite_corrupted)));
1052     qdict_put_obj(opts, "children", QOBJECT(children));
1053
1054     bs->full_open_options = opts;
1055 }
1056
1057 static BlockDriver bdrv_quorum = {
1058     .format_name                        = "quorum",
1059     .protocol_name                      = "quorum",
1060
1061     .instance_size                      = sizeof(BDRVQuorumState),
1062
1063     .bdrv_file_open                     = quorum_open,
1064     .bdrv_close                         = quorum_close,
1065     .bdrv_refresh_filename              = quorum_refresh_filename,
1066
1067     .bdrv_co_flush_to_disk              = quorum_co_flush,
1068
1069     .bdrv_getlength                     = quorum_getlength,
1070
1071     .bdrv_aio_readv                     = quorum_aio_readv,
1072     .bdrv_aio_writev                    = quorum_aio_writev,
1073     .bdrv_invalidate_cache              = quorum_invalidate_cache,
1074
1075     .bdrv_detach_aio_context            = quorum_detach_aio_context,
1076     .bdrv_attach_aio_context            = quorum_attach_aio_context,
1077
1078     .is_filter                          = true,
1079     .bdrv_recurse_is_first_non_filter   = quorum_recurse_is_first_non_filter,
1080 };
1081
1082 static void bdrv_quorum_init(void)
1083 {
1084     if (!qcrypto_hash_supports(QCRYPTO_HASH_ALG_SHA256)) {
1085         /* SHA256 hash support is required for quorum device */
1086         return;
1087     }
1088     bdrv_register(&bdrv_quorum);
1089 }
1090
1091 block_init(bdrv_quorum_init);