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