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