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