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