4 * Copyright (C) 2012-2014 Nodalink, EURL.
7 * BenoƮt Canet <benoit.canet@irqsave.net>
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).
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.
16 #include "qemu/osdep.h"
17 #include "qemu/cutils.h"
18 #include "qemu/module.h"
19 #include "qemu/option.h"
20 #include "qemu/memalign.h"
21 #include "block/block_int.h"
22 #include "block/coroutines.h"
23 #include "block/qdict.h"
24 #include "qapi/error.h"
25 #include "qapi/qapi-events-block.h"
26 #include "qapi/qmp/qdict.h"
27 #include "qapi/qmp/qerror.h"
28 #include "qapi/qmp/qlist.h"
29 #include "qapi/qmp/qstring.h"
30 #include "crypto/hash.h"
32 #define HASH_LENGTH 32
34 #define INDEXSTR_LEN 32
36 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
37 #define QUORUM_OPT_BLKVERIFY "blkverify"
38 #define QUORUM_OPT_REWRITE "rewrite-corrupted"
39 #define QUORUM_OPT_READ_PATTERN "read-pattern"
41 /* This union holds a vote hash value */
42 typedef union QuorumVoteValue {
43 uint8_t h[HASH_LENGTH]; /* SHA-256 hash */
44 int64_t l; /* simpler 64 bits hash */
48 typedef struct QuorumVoteItem {
50 QLIST_ENTRY(QuorumVoteItem) next;
53 /* this structure is a vote version. A version is the set of votes sharing the
55 * The set of votes will be tracked with the items field and its cardinality is
58 typedef struct QuorumVoteVersion {
59 QuorumVoteValue value;
62 QLIST_HEAD(, QuorumVoteItem) items;
63 QLIST_ENTRY(QuorumVoteVersion) next;
66 /* this structure holds a group of vote versions together */
67 typedef struct QuorumVotes {
68 QLIST_HEAD(, QuorumVoteVersion) vote_list;
69 bool (*compare)(QuorumVoteValue *a, QuorumVoteValue *b);
72 /* the following structure holds the state of one quorum instance */
73 typedef struct BDRVQuorumState {
74 BdrvChild **children; /* children BlockDriverStates */
75 int num_children; /* children count */
76 unsigned next_child_index; /* the index of the next child that should
79 int threshold; /* if less than threshold children reads gave the
80 * same result a quorum error occurs.
82 bool is_blkverify; /* true if the driver is in blkverify mode
83 * Writes are mirrored on two children devices.
84 * On reads the two children devices' contents are
85 * compared and if a difference is spotted its
86 * location is printed and the code aborts.
87 * It is useful to debug other block drivers by
88 * comparing them with a reference one.
90 bool rewrite_corrupted;/* true if the driver must rewrite-on-read corrupted
91 * block if Quorum is reached.
94 QuorumReadPattern read_pattern;
97 typedef struct QuorumAIOCB QuorumAIOCB;
99 /* Quorum will create one instance of the following structure per operation it
100 * performs on its children.
101 * So for each read/write operation coming from the upper layer there will be
102 * $children_count QuorumChildRequest.
104 typedef struct QuorumChildRequest {
105 BlockDriverState *bs;
110 } QuorumChildRequest;
112 /* Quorum will use the following structure to track progress of each read/write
113 * operation received by the upper layer.
114 * This structure hold pointers to the QuorumChildRequest structures instances
115 * used to do operations on each children and track overall progress.
118 BlockDriverState *bs;
121 /* Request metadata */
126 QEMUIOVector *qiov; /* calling IOV */
128 QuorumChildRequest *qcrs; /* individual child requests */
129 int count; /* number of completed AIOCB */
130 int success_count; /* number of successfully completed AIOCB */
132 int rewrite_count; /* number of replica to rewrite: count down to
133 * zero once writes are fired
140 int children_read; /* how many children have been read from */
143 typedef struct QuorumCo {
148 static void quorum_aio_finalize(QuorumAIOCB *acb)
154 static bool quorum_sha256_compare(QuorumVoteValue *a, QuorumVoteValue *b)
156 return !memcmp(a->h, b->h, HASH_LENGTH);
159 static bool quorum_64bits_compare(QuorumVoteValue *a, QuorumVoteValue *b)
164 static QuorumAIOCB *quorum_aio_get(BlockDriverState *bs,
170 BDRVQuorumState *s = bs->opaque;
171 QuorumAIOCB *acb = g_new(QuorumAIOCB, 1);
174 *acb = (QuorumAIOCB) {
175 .co = qemu_coroutine_self(),
181 .votes.compare = quorum_sha256_compare,
182 .votes.vote_list = QLIST_HEAD_INITIALIZER(acb.votes.vote_list),
185 acb->qcrs = g_new0(QuorumChildRequest, s->num_children);
186 for (i = 0; i < s->num_children; i++) {
187 acb->qcrs[i].buf = NULL;
188 acb->qcrs[i].ret = 0;
189 acb->qcrs[i].parent = acb;
195 static void quorum_report_bad(QuorumOpType type, uint64_t offset,
196 uint64_t bytes, char *node_name, int ret)
198 const char *msg = NULL;
199 int64_t start_sector = offset / BDRV_SECTOR_SIZE;
200 int64_t end_sector = DIV_ROUND_UP(offset + bytes, BDRV_SECTOR_SIZE);
203 msg = strerror(-ret);
206 qapi_event_send_quorum_report_bad(type, !!msg, msg, node_name, start_sector,
207 end_sector - start_sector);
210 static void quorum_report_failure(QuorumAIOCB *acb)
212 const char *reference = bdrv_get_device_or_node_name(acb->bs);
213 int64_t start_sector = acb->offset / BDRV_SECTOR_SIZE;
214 int64_t end_sector = DIV_ROUND_UP(acb->offset + acb->bytes,
217 qapi_event_send_quorum_failure(reference, start_sector,
218 end_sector - start_sector);
221 static int quorum_vote_error(QuorumAIOCB *acb);
223 static bool quorum_has_too_much_io_failed(QuorumAIOCB *acb)
225 BDRVQuorumState *s = acb->bs->opaque;
227 if (acb->success_count < s->threshold) {
228 acb->vote_ret = quorum_vote_error(acb);
229 quorum_report_failure(acb);
236 static int read_fifo_child(QuorumAIOCB *acb);
238 static void quorum_copy_qiov(QEMUIOVector *dest, QEMUIOVector *source)
241 assert(dest->niov == source->niov);
242 assert(dest->size == source->size);
243 for (i = 0; i < source->niov; i++) {
244 assert(dest->iov[i].iov_len == source->iov[i].iov_len);
245 memcpy(dest->iov[i].iov_base,
246 source->iov[i].iov_base,
247 source->iov[i].iov_len);
251 static void quorum_report_bad_acb(QuorumChildRequest *sacb, int ret)
253 QuorumAIOCB *acb = sacb->parent;
254 QuorumOpType type = acb->is_read ? QUORUM_OP_TYPE_READ : QUORUM_OP_TYPE_WRITE;
255 quorum_report_bad(type, acb->offset, acb->bytes, sacb->bs->node_name, ret);
258 static void quorum_report_bad_versions(BDRVQuorumState *s,
260 QuorumVoteValue *value)
262 QuorumVoteVersion *version;
263 QuorumVoteItem *item;
265 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
266 if (acb->votes.compare(&version->value, value)) {
269 QLIST_FOREACH(item, &version->items, next) {
270 quorum_report_bad(QUORUM_OP_TYPE_READ, acb->offset, acb->bytes,
271 s->children[item->index]->bs->node_name, 0);
276 static void quorum_rewrite_entry(void *opaque)
278 QuorumCo *co = opaque;
279 QuorumAIOCB *acb = co->acb;
280 BDRVQuorumState *s = acb->bs->opaque;
282 /* Ignore any errors, it's just a correction attempt for already
284 * Mask out BDRV_REQ_WRITE_UNCHANGED because this overwrites the
285 * area with different data from the other children. */
286 bdrv_co_pwritev(s->children[co->idx], acb->offset, acb->bytes,
287 acb->qiov, acb->flags & ~BDRV_REQ_WRITE_UNCHANGED);
289 /* Wake up the caller after the last rewrite */
290 acb->rewrite_count--;
291 if (!acb->rewrite_count) {
292 qemu_coroutine_enter_if_inactive(acb->co);
296 static bool quorum_rewrite_bad_versions(QuorumAIOCB *acb,
297 QuorumVoteValue *value)
299 QuorumVoteVersion *version;
300 QuorumVoteItem *item;
303 /* first count the number of bad versions: done first to avoid concurrency
306 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
307 if (acb->votes.compare(&version->value, value)) {
310 QLIST_FOREACH(item, &version->items, next) {
315 /* quorum_rewrite_entry will count down this to zero */
316 acb->rewrite_count = count;
318 /* now fire the correcting rewrites */
319 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
320 if (acb->votes.compare(&version->value, value)) {
323 QLIST_FOREACH(item, &version->items, next) {
330 co = qemu_coroutine_create(quorum_rewrite_entry, &data);
331 qemu_coroutine_enter(co);
335 /* return true if any rewrite is done else false */
339 static void quorum_count_vote(QuorumVotes *votes,
340 QuorumVoteValue *value,
343 QuorumVoteVersion *v = NULL, *version = NULL;
344 QuorumVoteItem *item;
346 /* look if we have something with this hash */
347 QLIST_FOREACH(v, &votes->vote_list, next) {
348 if (votes->compare(&v->value, value)) {
354 /* It's a version not yet in the list add it */
356 version = g_new0(QuorumVoteVersion, 1);
357 QLIST_INIT(&version->items);
358 memcpy(&version->value, value, sizeof(version->value));
359 version->index = index;
360 version->vote_count = 0;
361 QLIST_INSERT_HEAD(&votes->vote_list, version, next);
364 version->vote_count++;
366 item = g_new0(QuorumVoteItem, 1);
368 QLIST_INSERT_HEAD(&version->items, item, next);
371 static void quorum_free_vote_list(QuorumVotes *votes)
373 QuorumVoteVersion *version, *next_version;
374 QuorumVoteItem *item, *next_item;
376 QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
377 QLIST_REMOVE(version, next);
378 QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
379 QLIST_REMOVE(item, next);
386 static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
388 QEMUIOVector *qiov = &acb->qcrs[i].qiov;
389 size_t len = sizeof(hash->h);
390 uint8_t *data = hash->h;
392 /* XXX - would be nice if we could pass in the Error **
393 * and propagate that back, but this quorum code is
394 * restricted to just errno values currently */
395 if (qcrypto_hash_bytesv(QCRYPTO_HASH_ALG_SHA256,
396 qiov->iov, qiov->niov,
405 static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
408 QuorumVoteVersion *candidate, *winner = NULL;
410 QLIST_FOREACH(candidate, &votes->vote_list, next) {
411 if (candidate->vote_count > max) {
412 max = candidate->vote_count;
420 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
421 * differing byte location. Yet it is handcoded to compare vectors one byte
422 * after another so it does not benefit from the libc SIMD optimizations.
423 * quorum_iovec_compare is written for speed and should be used in the non
424 * blkverify mode of quorum.
426 static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
431 assert(a->niov == b->niov);
432 for (i = 0; i < a->niov; i++) {
433 assert(a->iov[i].iov_len == b->iov[i].iov_len);
434 result = memcmp(a->iov[i].iov_base,
445 static bool quorum_compare(QuorumAIOCB *acb, QEMUIOVector *a, QEMUIOVector *b)
447 BDRVQuorumState *s = acb->bs->opaque;
450 /* This driver will replace blkverify in this particular case */
451 if (s->is_blkverify) {
452 offset = qemu_iovec_compare(a, b);
454 fprintf(stderr, "quorum: offset=%" PRIu64 " bytes=%" PRIu64
455 " contents mismatch at offset %" PRIu64 "\n",
456 acb->offset, acb->bytes, acb->offset + offset);
462 return quorum_iovec_compare(a, b);
465 /* Do a vote to get the error code */
466 static int quorum_vote_error(QuorumAIOCB *acb)
468 BDRVQuorumState *s = acb->bs->opaque;
469 QuorumVoteVersion *winner = NULL;
470 QuorumVotes error_votes;
471 QuorumVoteValue result_value;
475 QLIST_INIT(&error_votes.vote_list);
476 error_votes.compare = quorum_64bits_compare;
478 for (i = 0; i < s->num_children; i++) {
479 ret = acb->qcrs[i].ret;
482 result_value.l = ret;
483 quorum_count_vote(&error_votes, &result_value, i);
488 winner = quorum_get_vote_winner(&error_votes);
489 ret = winner->value.l;
492 quorum_free_vote_list(&error_votes);
497 static void quorum_vote(QuorumAIOCB *acb)
501 QuorumVoteValue hash;
502 BDRVQuorumState *s = acb->bs->opaque;
503 QuorumVoteVersion *winner;
505 if (quorum_has_too_much_io_failed(acb)) {
509 /* get the index of the first successful read */
510 for (i = 0; i < s->num_children; i++) {
511 if (!acb->qcrs[i].ret) {
516 assert(i < s->num_children);
518 /* compare this read with all other successful reads stopping at quorum
521 for (j = i + 1; j < s->num_children; j++) {
522 if (acb->qcrs[j].ret) {
525 quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
531 /* Every successful read agrees */
533 quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
537 /* compute hashes for each successful read, also store indexes */
538 for (i = 0; i < s->num_children; i++) {
539 if (acb->qcrs[i].ret) {
542 ret = quorum_compute_hash(acb, i, &hash);
543 /* if ever the hash computation failed */
548 quorum_count_vote(&acb->votes, &hash, i);
551 /* vote to select the most represented version */
552 winner = quorum_get_vote_winner(&acb->votes);
554 /* if the winner count is smaller than threshold the read fails */
555 if (winner->vote_count < s->threshold) {
556 quorum_report_failure(acb);
557 acb->vote_ret = -EIO;
561 /* we have a winner: copy it */
562 quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
564 /* some versions are bad print them */
565 quorum_report_bad_versions(s, acb, &winner->value);
567 /* corruption correction is enabled */
568 if (s->rewrite_corrupted) {
569 quorum_rewrite_bad_versions(acb, &winner->value);
574 quorum_free_vote_list(&acb->votes);
577 static void read_quorum_children_entry(void *opaque)
579 QuorumCo *co = opaque;
580 QuorumAIOCB *acb = co->acb;
581 BDRVQuorumState *s = acb->bs->opaque;
583 QuorumChildRequest *sacb = &acb->qcrs[i];
585 sacb->bs = s->children[i]->bs;
586 sacb->ret = bdrv_co_preadv(s->children[i], acb->offset, acb->bytes,
587 &acb->qcrs[i].qiov, 0);
589 if (sacb->ret == 0) {
590 acb->success_count++;
592 quorum_report_bad_acb(sacb, sacb->ret);
596 assert(acb->count <= s->num_children);
597 assert(acb->success_count <= s->num_children);
599 /* Wake up the caller after the last read */
600 if (acb->count == s->num_children) {
601 qemu_coroutine_enter_if_inactive(acb->co);
605 static int read_quorum_children(QuorumAIOCB *acb)
607 BDRVQuorumState *s = acb->bs->opaque;
610 acb->children_read = s->num_children;
611 for (i = 0; i < s->num_children; i++) {
612 acb->qcrs[i].buf = qemu_blockalign(s->children[i]->bs, acb->qiov->size);
613 qemu_iovec_init(&acb->qcrs[i].qiov, acb->qiov->niov);
614 qemu_iovec_clone(&acb->qcrs[i].qiov, acb->qiov, acb->qcrs[i].buf);
617 for (i = 0; i < s->num_children; i++) {
624 co = qemu_coroutine_create(read_quorum_children_entry, &data);
625 qemu_coroutine_enter(co);
628 while (acb->count < s->num_children) {
629 qemu_coroutine_yield();
632 /* Do the vote on read */
634 for (i = 0; i < s->num_children; i++) {
635 qemu_vfree(acb->qcrs[i].buf);
636 qemu_iovec_destroy(&acb->qcrs[i].qiov);
639 while (acb->rewrite_count) {
640 qemu_coroutine_yield();
643 return acb->vote_ret;
646 static int read_fifo_child(QuorumAIOCB *acb)
648 BDRVQuorumState *s = acb->bs->opaque;
651 /* We try to read the next child in FIFO order if we failed to read */
653 n = acb->children_read++;
654 acb->qcrs[n].bs = s->children[n]->bs;
655 ret = bdrv_co_preadv(s->children[n], acb->offset, acb->bytes,
658 quorum_report_bad_acb(&acb->qcrs[n], ret);
660 } while (ret < 0 && acb->children_read < s->num_children);
662 /* FIXME: rewrite failed children if acb->children_read > 1? */
667 static int quorum_co_preadv(BlockDriverState *bs, int64_t offset, int64_t bytes,
668 QEMUIOVector *qiov, BdrvRequestFlags flags)
670 BDRVQuorumState *s = bs->opaque;
671 QuorumAIOCB *acb = quorum_aio_get(bs, qiov, offset, bytes, flags);
675 acb->children_read = 0;
677 if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
678 ret = read_quorum_children(acb);
680 ret = read_fifo_child(acb);
682 quorum_aio_finalize(acb);
687 static void write_quorum_entry(void *opaque)
689 QuorumCo *co = opaque;
690 QuorumAIOCB *acb = co->acb;
691 BDRVQuorumState *s = acb->bs->opaque;
693 QuorumChildRequest *sacb = &acb->qcrs[i];
695 sacb->bs = s->children[i]->bs;
696 if (acb->flags & BDRV_REQ_ZERO_WRITE) {
697 sacb->ret = bdrv_co_pwrite_zeroes(s->children[i], acb->offset,
698 acb->bytes, acb->flags);
700 sacb->ret = bdrv_co_pwritev(s->children[i], acb->offset, acb->bytes,
701 acb->qiov, acb->flags);
703 if (sacb->ret == 0) {
704 acb->success_count++;
706 quorum_report_bad_acb(sacb, sacb->ret);
709 assert(acb->count <= s->num_children);
710 assert(acb->success_count <= s->num_children);
712 /* Wake up the caller after the last write */
713 if (acb->count == s->num_children) {
714 qemu_coroutine_enter_if_inactive(acb->co);
718 static int quorum_co_pwritev(BlockDriverState *bs, int64_t offset,
719 int64_t bytes, QEMUIOVector *qiov,
720 BdrvRequestFlags flags)
722 BDRVQuorumState *s = bs->opaque;
723 QuorumAIOCB *acb = quorum_aio_get(bs, qiov, offset, bytes, flags);
726 for (i = 0; i < s->num_children; i++) {
733 co = qemu_coroutine_create(write_quorum_entry, &data);
734 qemu_coroutine_enter(co);
737 while (acb->count < s->num_children) {
738 qemu_coroutine_yield();
741 quorum_has_too_much_io_failed(acb);
744 quorum_aio_finalize(acb);
749 static int quorum_co_pwrite_zeroes(BlockDriverState *bs, int64_t offset,
750 int64_t bytes, BdrvRequestFlags flags)
753 return quorum_co_pwritev(bs, offset, bytes, NULL,
754 flags | BDRV_REQ_ZERO_WRITE);
757 static int64_t quorum_getlength(BlockDriverState *bs)
759 BDRVQuorumState *s = bs->opaque;
763 /* check that all file have the same length */
764 result = bdrv_getlength(s->children[0]->bs);
768 for (i = 1; i < s->num_children; i++) {
769 int64_t value = bdrv_getlength(s->children[i]->bs);
773 if (value != result) {
781 static coroutine_fn int quorum_co_flush(BlockDriverState *bs)
783 BDRVQuorumState *s = bs->opaque;
784 QuorumVoteVersion *winner = NULL;
785 QuorumVotes error_votes;
786 QuorumVoteValue result_value;
789 int success_count = 0;
791 QLIST_INIT(&error_votes.vote_list);
792 error_votes.compare = quorum_64bits_compare;
794 for (i = 0; i < s->num_children; i++) {
795 result = bdrv_co_flush(s->children[i]->bs);
797 quorum_report_bad(QUORUM_OP_TYPE_FLUSH, 0, 0,
798 s->children[i]->bs->node_name, result);
799 result_value.l = result;
800 quorum_count_vote(&error_votes, &result_value, i);
806 if (success_count >= s->threshold) {
809 winner = quorum_get_vote_winner(&error_votes);
810 result = winner->value.l;
812 quorum_free_vote_list(&error_votes);
817 static bool quorum_recurse_can_replace(BlockDriverState *bs,
818 BlockDriverState *to_replace)
820 BDRVQuorumState *s = bs->opaque;
823 for (i = 0; i < s->num_children; i++) {
825 * We have no idea whether our children show the same data as
826 * this node (@bs). It is actually highly likely that
827 * @to_replace does not, because replacing a broken child is
828 * one of the main use cases here.
830 * We do know that the new BDS will match @bs, so replacing
831 * any of our children by it will be safe. It cannot change
832 * the data this quorum node presents to its parents.
834 * However, replacing @to_replace by @bs in any of our
835 * children's chains may change visible data somewhere in
836 * there. We therefore cannot recurse down those chains with
837 * bdrv_recurse_can_replace().
838 * (More formally, bdrv_recurse_can_replace() requires that
839 * @to_replace will be replaced by something matching the @bs
840 * passed to it. We cannot guarantee that.)
842 * Thus, we can only check whether any of our immediate
843 * children matches @to_replace.
845 * (In the future, we might add a function to recurse down a
846 * chain that checks that nothing there cares about a change
847 * in data from the respective child in question. For
848 * example, most filters do not care when their child's data
849 * suddenly changes, as long as their parents do not care.)
851 if (s->children[i]->bs == to_replace) {
853 * We now have to ensure that there is no other parent
854 * that cares about replacing this child by a node with
855 * potentially different data.
856 * We do so by checking whether there are any other parents
857 * at all, which is stricter than necessary, but also very
858 * simple. (We may decide to implement something more
859 * complex and permissive when there is an actual need for
862 return QLIST_FIRST(&to_replace->parents) == s->children[i] &&
863 QLIST_NEXT(s->children[i], next_parent) == NULL;
870 static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
874 error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
875 "vote-threshold", "a value >= 1");
879 if (threshold > num_children) {
880 error_setg(errp, "threshold may not exceed children count");
887 static QemuOptsList quorum_runtime_opts = {
889 .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
892 .name = QUORUM_OPT_VOTE_THRESHOLD,
893 .type = QEMU_OPT_NUMBER,
894 .help = "The number of vote needed for reaching quorum",
897 .name = QUORUM_OPT_BLKVERIFY,
898 .type = QEMU_OPT_BOOL,
899 .help = "Trigger block verify mode if set",
902 .name = QUORUM_OPT_REWRITE,
903 .type = QEMU_OPT_BOOL,
904 .help = "Rewrite corrupted block on read quorum",
907 .name = QUORUM_OPT_READ_PATTERN,
908 .type = QEMU_OPT_STRING,
909 .help = "Allowed pattern: quorum, fifo. Quorum is default",
911 { /* end of list */ }
915 static void quorum_refresh_flags(BlockDriverState *bs)
917 BDRVQuorumState *s = bs->opaque;
920 bs->supported_zero_flags =
921 BDRV_REQ_FUA | BDRV_REQ_MAY_UNMAP | BDRV_REQ_NO_FALLBACK;
923 for (i = 0; i < s->num_children; i++) {
924 bs->supported_zero_flags &= s->children[i]->bs->supported_zero_flags;
927 bs->supported_zero_flags |= BDRV_REQ_WRITE_UNCHANGED;
930 static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
933 BDRVQuorumState *s = bs->opaque;
934 QemuOpts *opts = NULL;
935 const char *pattern_str;
940 qdict_flatten(options);
942 /* count how many different children are present */
943 s->num_children = qdict_array_entries(options, "children.");
944 if (s->num_children < 0) {
945 error_setg(errp, "Option children is not a valid array");
949 if (s->num_children < 1) {
950 error_setg(errp, "Number of provided children must be 1 or more");
955 opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
956 if (!qemu_opts_absorb_qdict(opts, options, errp)) {
961 s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
962 /* and validate it against s->num_children */
963 ret = quorum_valid_threshold(s->threshold, s->num_children, errp);
968 pattern_str = qemu_opt_get(opts, QUORUM_OPT_READ_PATTERN);
970 ret = QUORUM_READ_PATTERN_QUORUM;
972 ret = qapi_enum_parse(&QuorumReadPattern_lookup, pattern_str,
976 error_setg(errp, "Please set read-pattern as fifo or quorum");
979 s->read_pattern = ret;
981 if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
982 s->is_blkverify = qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false);
983 if (s->is_blkverify && (s->num_children != 2 || s->threshold != 2)) {
984 error_setg(errp, "blkverify=on can only be set if there are "
985 "exactly two files and vote-threshold is 2");
990 s->rewrite_corrupted = qemu_opt_get_bool(opts, QUORUM_OPT_REWRITE,
992 if (s->rewrite_corrupted && s->is_blkverify) {
994 "rewrite-corrupted=on cannot be used with blkverify=on");
1000 /* allocate the children array */
1001 s->children = g_new0(BdrvChild *, s->num_children);
1002 opened = g_new0(bool, s->num_children);
1004 for (i = 0; i < s->num_children; i++) {
1005 char indexstr[INDEXSTR_LEN];
1006 ret = snprintf(indexstr, INDEXSTR_LEN, "children.%d", i);
1007 assert(ret < INDEXSTR_LEN);
1009 s->children[i] = bdrv_open_child(NULL, options, indexstr, bs,
1010 &child_of_bds, BDRV_CHILD_DATA, false,
1012 if (!s->children[i]) {
1019 s->next_child_index = s->num_children;
1021 bs->supported_write_flags = BDRV_REQ_WRITE_UNCHANGED;
1022 quorum_refresh_flags(bs);
1028 /* cleanup on error */
1029 for (i = 0; i < s->num_children; i++) {
1033 bdrv_unref_child(bs, s->children[i]);
1035 g_free(s->children);
1038 qemu_opts_del(opts);
1042 static void quorum_close(BlockDriverState *bs)
1044 BDRVQuorumState *s = bs->opaque;
1047 for (i = 0; i < s->num_children; i++) {
1048 bdrv_unref_child(bs, s->children[i]);
1051 g_free(s->children);
1054 static void quorum_add_child(BlockDriverState *bs, BlockDriverState *child_bs,
1057 BDRVQuorumState *s = bs->opaque;
1059 char indexstr[INDEXSTR_LEN];
1062 if (s->is_blkverify) {
1063 error_setg(errp, "Cannot add a child to a quorum in blkverify mode");
1067 assert(s->num_children <= INT_MAX / sizeof(BdrvChild *));
1068 if (s->num_children == INT_MAX / sizeof(BdrvChild *) ||
1069 s->next_child_index == UINT_MAX) {
1070 error_setg(errp, "Too many children");
1074 ret = snprintf(indexstr, INDEXSTR_LEN, "children.%u", s->next_child_index);
1075 if (ret < 0 || ret >= INDEXSTR_LEN) {
1076 error_setg(errp, "cannot generate child name");
1079 s->next_child_index++;
1081 bdrv_drained_begin(bs);
1083 /* We can safely add the child now */
1086 child = bdrv_attach_child(bs, child_bs, indexstr, &child_of_bds,
1087 BDRV_CHILD_DATA, errp);
1088 if (child == NULL) {
1089 s->next_child_index--;
1092 s->children = g_renew(BdrvChild *, s->children, s->num_children + 1);
1093 s->children[s->num_children++] = child;
1094 quorum_refresh_flags(bs);
1097 bdrv_drained_end(bs);
1100 static void quorum_del_child(BlockDriverState *bs, BdrvChild *child,
1103 BDRVQuorumState *s = bs->opaque;
1104 char indexstr[INDEXSTR_LEN];
1107 for (i = 0; i < s->num_children; i++) {
1108 if (s->children[i] == child) {
1113 /* we have checked it in bdrv_del_child() */
1114 assert(i < s->num_children);
1116 if (s->num_children <= s->threshold) {
1118 "The number of children cannot be lower than the vote threshold %d",
1123 /* We know now that num_children > threshold, so blkverify must be false */
1124 assert(!s->is_blkverify);
1126 snprintf(indexstr, INDEXSTR_LEN, "children.%u", s->next_child_index - 1);
1127 if (!strncmp(child->name, indexstr, INDEXSTR_LEN)) {
1128 s->next_child_index--;
1131 bdrv_drained_begin(bs);
1133 /* We can safely remove this child now */
1134 memmove(&s->children[i], &s->children[i + 1],
1135 (s->num_children - i - 1) * sizeof(BdrvChild *));
1136 s->children = g_renew(BdrvChild *, s->children, --s->num_children);
1137 bdrv_unref_child(bs, child);
1139 quorum_refresh_flags(bs);
1140 bdrv_drained_end(bs);
1143 static void quorum_gather_child_options(BlockDriverState *bs, QDict *target,
1144 bool backing_overridden)
1146 BDRVQuorumState *s = bs->opaque;
1147 QList *children_list;
1151 * The generic implementation for gathering child options in
1152 * bdrv_refresh_filename() would use the names of the children
1153 * as specified for bdrv_open_child() or bdrv_attach_child(),
1154 * which is "children.%u" with %u being a value
1155 * (s->next_child_index) that is incremented each time a new child
1156 * is added (and never decremented). Since children can be
1157 * deleted at runtime, there may be gaps in that enumeration.
1158 * When creating a new quorum BDS and specifying the children for
1159 * it through runtime options, the enumeration used there may not
1160 * have any gaps, though.
1162 * Therefore, we have to create a new gap-less enumeration here
1163 * (which we can achieve by simply putting all of the children's
1164 * full_open_options into a QList).
1166 * XXX: Note that there are issues with the current child option
1167 * structure quorum uses (such as the fact that children do
1168 * not really have unique permanent names). Therefore, this
1169 * is going to have to change in the future and ideally we
1170 * want quorum to be covered by the generic implementation.
1173 children_list = qlist_new();
1174 qdict_put(target, "children", children_list);
1176 for (i = 0; i < s->num_children; i++) {
1177 qlist_append(children_list,
1178 qobject_ref(s->children[i]->bs->full_open_options));
1182 static char *quorum_dirname(BlockDriverState *bs, Error **errp)
1184 /* In general, there are multiple BDSs with different dirnames below this
1185 * one; so there is no unique dirname we could return (unless all are equal
1186 * by chance, or there is only one). Therefore, to be consistent, just
1187 * always return NULL. */
1188 error_setg(errp, "Cannot generate a base directory for quorum nodes");
1192 static void quorum_child_perm(BlockDriverState *bs, BdrvChild *c,
1194 BlockReopenQueue *reopen_queue,
1195 uint64_t perm, uint64_t shared,
1196 uint64_t *nperm, uint64_t *nshared)
1198 BDRVQuorumState *s = bs->opaque;
1200 *nperm = perm & DEFAULT_PERM_PASSTHROUGH;
1201 if (s->rewrite_corrupted) {
1202 *nperm |= BLK_PERM_WRITE;
1206 * We cannot share RESIZE or WRITE, as this would make the
1207 * children differ from each other.
1209 *nshared = (shared & (BLK_PERM_CONSISTENT_READ |
1210 BLK_PERM_WRITE_UNCHANGED))
1211 | DEFAULT_PERM_UNCHANGED;
1215 * Each one of the children can report different status flags even
1216 * when they contain the same data, so what this function does is
1217 * return BDRV_BLOCK_ZERO if *all* children agree that a certain
1218 * region contains zeroes, and BDRV_BLOCK_DATA otherwise.
1220 static int coroutine_fn quorum_co_block_status(BlockDriverState *bs,
1222 int64_t offset, int64_t count,
1223 int64_t *pnum, int64_t *map,
1224 BlockDriverState **file)
1226 BDRVQuorumState *s = bs->opaque;
1228 int64_t pnum_zero = count;
1229 int64_t pnum_data = 0;
1231 for (i = 0; i < s->num_children; i++) {
1233 ret = bdrv_co_common_block_status_above(s->children[i]->bs, NULL, false,
1234 want_zero, offset, count,
1235 &bytes, NULL, NULL, NULL);
1237 quorum_report_bad(QUORUM_OP_TYPE_READ, offset, count,
1238 s->children[i]->bs->node_name, ret);
1243 * Even if all children agree about whether there are zeroes
1244 * or not at @offset they might disagree on the size, so use
1245 * the smallest when reporting BDRV_BLOCK_ZERO and the largest
1246 * when reporting BDRV_BLOCK_DATA.
1248 if (ret & BDRV_BLOCK_ZERO) {
1249 pnum_zero = MIN(pnum_zero, bytes);
1251 pnum_data = MAX(pnum_data, bytes);
1257 return BDRV_BLOCK_DATA;
1260 return BDRV_BLOCK_ZERO;
1264 static const char *const quorum_strong_runtime_opts[] = {
1265 QUORUM_OPT_VOTE_THRESHOLD,
1266 QUORUM_OPT_BLKVERIFY,
1268 QUORUM_OPT_READ_PATTERN,
1273 static BlockDriver bdrv_quorum = {
1274 .format_name = "quorum",
1276 .instance_size = sizeof(BDRVQuorumState),
1278 .bdrv_open = quorum_open,
1279 .bdrv_close = quorum_close,
1280 .bdrv_gather_child_options = quorum_gather_child_options,
1281 .bdrv_dirname = quorum_dirname,
1282 .bdrv_co_block_status = quorum_co_block_status,
1284 .bdrv_co_flush = quorum_co_flush,
1286 .bdrv_getlength = quorum_getlength,
1288 .bdrv_co_preadv = quorum_co_preadv,
1289 .bdrv_co_pwritev = quorum_co_pwritev,
1290 .bdrv_co_pwrite_zeroes = quorum_co_pwrite_zeroes,
1292 .bdrv_add_child = quorum_add_child,
1293 .bdrv_del_child = quorum_del_child,
1295 .bdrv_child_perm = quorum_child_perm,
1297 .bdrv_recurse_can_replace = quorum_recurse_can_replace,
1299 .strong_runtime_opts = quorum_strong_runtime_opts,
1302 static void bdrv_quorum_init(void)
1304 if (!qcrypto_hash_supports(QCRYPTO_HASH_ALG_SHA256)) {
1305 /* SHA256 hash support is required for quorum device */
1308 bdrv_register(&bdrv_quorum);
1311 block_init(bdrv_quorum_init);