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