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