block/quorum.c

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