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); 1017