block/quorum.c

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