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
*coroutine_fn
quorum_aio_get(BlockDriverState
*bs
,
166 uint64_t offset
, uint64_t bytes
,
169 BDRVQuorumState
*s
= bs
->opaque
;
170 QuorumAIOCB
*acb
= g_new(QuorumAIOCB
, 1);
173 *acb
= (QuorumAIOCB
) {
174 .co
= qemu_coroutine_self(),
180 .votes
.compare
= quorum_sha256_compare
,
181 .votes
.vote_list
= QLIST_HEAD_INITIALIZER(acb
.votes
.vote_list
),
184 acb
->qcrs
= g_new0(QuorumChildRequest
, s
->num_children
);
185 for (i
= 0; i
< s
->num_children
; i
++) {
186 acb
->qcrs
[i
].buf
= NULL
;
187 acb
->qcrs
[i
].ret
= 0;
188 acb
->qcrs
[i
].parent
= acb
;
194 static void quorum_report_bad(QuorumOpType type
, uint64_t offset
,
195 uint64_t bytes
, char *node_name
, int ret
)
197 const char *msg
= NULL
;
198 int64_t start_sector
= offset
/ BDRV_SECTOR_SIZE
;
199 int64_t end_sector
= DIV_ROUND_UP(offset
+ bytes
, BDRV_SECTOR_SIZE
);
202 msg
= strerror(-ret
);
205 qapi_event_send_quorum_report_bad(type
, msg
, node_name
, start_sector
,
206 end_sector
- start_sector
);
209 static void quorum_report_failure(QuorumAIOCB
*acb
)
211 const char *reference
= bdrv_get_device_or_node_name(acb
->bs
);
212 int64_t start_sector
= acb
->offset
/ BDRV_SECTOR_SIZE
;
213 int64_t end_sector
= DIV_ROUND_UP(acb
->offset
+ acb
->bytes
,
216 qapi_event_send_quorum_failure(reference
, start_sector
,
217 end_sector
- start_sector
);
220 static int quorum_vote_error(QuorumAIOCB
*acb
);
222 static bool quorum_has_too_much_io_failed(QuorumAIOCB
*acb
)
224 BDRVQuorumState
*s
= acb
->bs
->opaque
;
226 if (acb
->success_count
< s
->threshold
) {
227 acb
->vote_ret
= quorum_vote_error(acb
);
228 quorum_report_failure(acb
);
235 static void quorum_copy_qiov(QEMUIOVector
*dest
, QEMUIOVector
*source
)
238 assert(dest
->niov
== source
->niov
);
239 assert(dest
->size
== source
->size
);
240 for (i
= 0; i
< source
->niov
; i
++) {
241 assert(dest
->iov
[i
].iov_len
== source
->iov
[i
].iov_len
);
242 memcpy(dest
->iov
[i
].iov_base
,
243 source
->iov
[i
].iov_base
,
244 source
->iov
[i
].iov_len
);
248 static void quorum_report_bad_acb(QuorumChildRequest
*sacb
, int ret
)
250 QuorumAIOCB
*acb
= sacb
->parent
;
251 QuorumOpType type
= acb
->is_read
? QUORUM_OP_TYPE_READ
: QUORUM_OP_TYPE_WRITE
;
252 quorum_report_bad(type
, acb
->offset
, acb
->bytes
, sacb
->bs
->node_name
, ret
);
255 static void quorum_report_bad_versions(BDRVQuorumState
*s
,
257 QuorumVoteValue
*value
)
259 QuorumVoteVersion
*version
;
260 QuorumVoteItem
*item
;
262 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
263 if (acb
->votes
.compare(&version
->value
, value
)) {
266 QLIST_FOREACH(item
, &version
->items
, next
) {
267 quorum_report_bad(QUORUM_OP_TYPE_READ
, acb
->offset
, acb
->bytes
,
268 s
->children
[item
->index
]->bs
->node_name
, 0);
273 static void coroutine_fn
quorum_rewrite_entry(void *opaque
)
275 QuorumCo
*co
= opaque
;
276 QuorumAIOCB
*acb
= co
->acb
;
277 BDRVQuorumState
*s
= acb
->bs
->opaque
;
279 /* Ignore any errors, it's just a correction attempt for already
281 * Mask out BDRV_REQ_WRITE_UNCHANGED because this overwrites the
282 * area with different data from the other children. */
283 bdrv_co_pwritev(s
->children
[co
->idx
], acb
->offset
, acb
->bytes
,
284 acb
->qiov
, acb
->flags
& ~BDRV_REQ_WRITE_UNCHANGED
);
286 /* Wake up the caller after the last rewrite */
287 acb
->rewrite_count
--;
288 if (!acb
->rewrite_count
) {
289 qemu_coroutine_enter_if_inactive(acb
->co
);
293 static bool quorum_rewrite_bad_versions(QuorumAIOCB
*acb
,
294 QuorumVoteValue
*value
)
296 QuorumVoteVersion
*version
;
297 QuorumVoteItem
*item
;
300 /* first count the number of bad versions: done first to avoid concurrency
303 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
304 if (acb
->votes
.compare(&version
->value
, value
)) {
307 QLIST_FOREACH(item
, &version
->items
, next
) {
312 /* quorum_rewrite_entry will count down this to zero */
313 acb
->rewrite_count
= count
;
315 /* now fire the correcting rewrites */
316 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
317 if (acb
->votes
.compare(&version
->value
, value
)) {
320 QLIST_FOREACH(item
, &version
->items
, next
) {
327 co
= qemu_coroutine_create(quorum_rewrite_entry
, &data
);
328 qemu_coroutine_enter(co
);
332 /* return true if any rewrite is done else false */
336 static void quorum_count_vote(QuorumVotes
*votes
,
337 QuorumVoteValue
*value
,
340 QuorumVoteVersion
*v
= NULL
, *version
= NULL
;
341 QuorumVoteItem
*item
;
343 /* look if we have something with this hash */
344 QLIST_FOREACH(v
, &votes
->vote_list
, next
) {
345 if (votes
->compare(&v
->value
, value
)) {
351 /* It's a version not yet in the list add it */
353 version
= g_new0(QuorumVoteVersion
, 1);
354 QLIST_INIT(&version
->items
);
355 memcpy(&version
->value
, value
, sizeof(version
->value
));
356 version
->index
= index
;
357 version
->vote_count
= 0;
358 QLIST_INSERT_HEAD(&votes
->vote_list
, version
, next
);
361 version
->vote_count
++;
363 item
= g_new0(QuorumVoteItem
, 1);
365 QLIST_INSERT_HEAD(&version
->items
, item
, next
);
368 static void quorum_free_vote_list(QuorumVotes
*votes
)
370 QuorumVoteVersion
*version
, *next_version
;
371 QuorumVoteItem
*item
, *next_item
;
373 QLIST_FOREACH_SAFE(version
, &votes
->vote_list
, next
, next_version
) {
374 QLIST_REMOVE(version
, next
);
375 QLIST_FOREACH_SAFE(item
, &version
->items
, next
, next_item
) {
376 QLIST_REMOVE(item
, next
);
383 static int quorum_compute_hash(QuorumAIOCB
*acb
, int i
, QuorumVoteValue
*hash
)
385 QEMUIOVector
*qiov
= &acb
->qcrs
[i
].qiov
;
386 size_t len
= sizeof(hash
->h
);
387 uint8_t *data
= hash
->h
;
389 /* XXX - would be nice if we could pass in the Error **
390 * and propagate that back, but this quorum code is
391 * restricted to just errno values currently */
392 if (qcrypto_hash_bytesv(QCRYPTO_HASH_ALG_SHA256
,
393 qiov
->iov
, qiov
->niov
,
402 static QuorumVoteVersion
*quorum_get_vote_winner(QuorumVotes
*votes
)
405 QuorumVoteVersion
*candidate
, *winner
= NULL
;
407 QLIST_FOREACH(candidate
, &votes
->vote_list
, next
) {
408 if (candidate
->vote_count
> max
) {
409 max
= candidate
->vote_count
;
417 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
418 * differing byte location. Yet it is handcoded to compare vectors one byte
419 * after another so it does not benefit from the libc SIMD optimizations.
420 * quorum_iovec_compare is written for speed and should be used in the non
421 * blkverify mode of quorum.
423 static bool quorum_iovec_compare(QEMUIOVector
*a
, QEMUIOVector
*b
)
428 assert(a
->niov
== b
->niov
);
429 for (i
= 0; i
< a
->niov
; i
++) {
430 assert(a
->iov
[i
].iov_len
== b
->iov
[i
].iov_len
);
431 result
= memcmp(a
->iov
[i
].iov_base
,
442 static bool quorum_compare(QuorumAIOCB
*acb
, QEMUIOVector
*a
, QEMUIOVector
*b
)
444 BDRVQuorumState
*s
= acb
->bs
->opaque
;
447 /* This driver will replace blkverify in this particular case */
448 if (s
->is_blkverify
) {
449 offset
= qemu_iovec_compare(a
, b
);
451 fprintf(stderr
, "quorum: offset=%" PRIu64
" bytes=%" PRIu64
452 " contents mismatch at offset %" PRIu64
"\n",
453 acb
->offset
, acb
->bytes
, acb
->offset
+ offset
);
459 return quorum_iovec_compare(a
, b
);
462 /* Do a vote to get the error code */
463 static int quorum_vote_error(QuorumAIOCB
*acb
)
465 BDRVQuorumState
*s
= acb
->bs
->opaque
;
466 QuorumVoteVersion
*winner
= NULL
;
467 QuorumVotes error_votes
;
468 QuorumVoteValue result_value
;
472 QLIST_INIT(&error_votes
.vote_list
);
473 error_votes
.compare
= quorum_64bits_compare
;
475 for (i
= 0; i
< s
->num_children
; i
++) {
476 ret
= acb
->qcrs
[i
].ret
;
479 result_value
.l
= ret
;
480 quorum_count_vote(&error_votes
, &result_value
, i
);
485 winner
= quorum_get_vote_winner(&error_votes
);
486 ret
= winner
->value
.l
;
489 quorum_free_vote_list(&error_votes
);
494 static void quorum_vote(QuorumAIOCB
*acb
)
498 QuorumVoteValue hash
;
499 BDRVQuorumState
*s
= acb
->bs
->opaque
;
500 QuorumVoteVersion
*winner
;
502 if (quorum_has_too_much_io_failed(acb
)) {
506 /* get the index of the first successful read */
507 for (i
= 0; i
< s
->num_children
; i
++) {
508 if (!acb
->qcrs
[i
].ret
) {
513 assert(i
< s
->num_children
);
515 /* compare this read with all other successful reads stopping at quorum
518 for (j
= i
+ 1; j
< s
->num_children
; j
++) {
519 if (acb
->qcrs
[j
].ret
) {
522 quorum
= quorum_compare(acb
, &acb
->qcrs
[i
].qiov
, &acb
->qcrs
[j
].qiov
);
528 /* Every successful read agrees */
530 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[i
].qiov
);
534 /* compute hashes for each successful read, also store indexes */
535 for (i
= 0; i
< s
->num_children
; i
++) {
536 if (acb
->qcrs
[i
].ret
) {
539 ret
= quorum_compute_hash(acb
, i
, &hash
);
540 /* if ever the hash computation failed */
545 quorum_count_vote(&acb
->votes
, &hash
, i
);
548 /* vote to select the most represented version */
549 winner
= quorum_get_vote_winner(&acb
->votes
);
551 /* if the winner count is smaller than threshold the read fails */
552 if (winner
->vote_count
< s
->threshold
) {
553 quorum_report_failure(acb
);
554 acb
->vote_ret
= -EIO
;
558 /* we have a winner: copy it */
559 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[winner
->index
].qiov
);
561 /* some versions are bad print them */
562 quorum_report_bad_versions(s
, acb
, &winner
->value
);
564 /* corruption correction is enabled */
565 if (s
->rewrite_corrupted
) {
566 quorum_rewrite_bad_versions(acb
, &winner
->value
);
571 quorum_free_vote_list(&acb
->votes
);
574 static void coroutine_fn
read_quorum_children_entry(void *opaque
)
576 QuorumCo
*co
= opaque
;
577 QuorumAIOCB
*acb
= co
->acb
;
578 BDRVQuorumState
*s
= acb
->bs
->opaque
;
580 QuorumChildRequest
*sacb
= &acb
->qcrs
[i
];
582 sacb
->bs
= s
->children
[i
]->bs
;
583 sacb
->ret
= bdrv_co_preadv(s
->children
[i
], acb
->offset
, acb
->bytes
,
584 &acb
->qcrs
[i
].qiov
, 0);
586 if (sacb
->ret
== 0) {
587 acb
->success_count
++;
589 quorum_report_bad_acb(sacb
, sacb
->ret
);
593 assert(acb
->count
<= s
->num_children
);
594 assert(acb
->success_count
<= s
->num_children
);
596 /* Wake up the caller after the last read */
597 if (acb
->count
== s
->num_children
) {
598 qemu_coroutine_enter_if_inactive(acb
->co
);
602 static int coroutine_fn
read_quorum_children(QuorumAIOCB
*acb
)
604 BDRVQuorumState
*s
= acb
->bs
->opaque
;
607 acb
->children_read
= s
->num_children
;
608 for (i
= 0; i
< s
->num_children
; i
++) {
609 acb
->qcrs
[i
].buf
= qemu_blockalign(s
->children
[i
]->bs
, acb
->qiov
->size
);
610 qemu_iovec_init(&acb
->qcrs
[i
].qiov
, acb
->qiov
->niov
);
611 qemu_iovec_clone(&acb
->qcrs
[i
].qiov
, acb
->qiov
, acb
->qcrs
[i
].buf
);
614 for (i
= 0; i
< s
->num_children
; i
++) {
621 co
= qemu_coroutine_create(read_quorum_children_entry
, &data
);
622 qemu_coroutine_enter(co
);
625 while (acb
->count
< s
->num_children
) {
626 qemu_coroutine_yield();
629 /* Do the vote on read */
631 for (i
= 0; i
< s
->num_children
; i
++) {
632 qemu_vfree(acb
->qcrs
[i
].buf
);
633 qemu_iovec_destroy(&acb
->qcrs
[i
].qiov
);
636 while (acb
->rewrite_count
) {
637 qemu_coroutine_yield();
640 return acb
->vote_ret
;
643 static int coroutine_fn
read_fifo_child(QuorumAIOCB
*acb
)
645 BDRVQuorumState
*s
= acb
->bs
->opaque
;
648 /* We try to read the next child in FIFO order if we failed to read */
650 n
= acb
->children_read
++;
651 acb
->qcrs
[n
].bs
= s
->children
[n
]->bs
;
652 ret
= bdrv_co_preadv(s
->children
[n
], acb
->offset
, acb
->bytes
,
655 quorum_report_bad_acb(&acb
->qcrs
[n
], ret
);
657 } while (ret
< 0 && acb
->children_read
< s
->num_children
);
659 /* FIXME: rewrite failed children if acb->children_read > 1? */
664 static int coroutine_fn
quorum_co_preadv(BlockDriverState
*bs
,
665 int64_t offset
, int64_t bytes
,
667 BdrvRequestFlags flags
)
669 BDRVQuorumState
*s
= bs
->opaque
;
670 QuorumAIOCB
*acb
= quorum_aio_get(bs
, qiov
, offset
, bytes
, flags
);
674 acb
->children_read
= 0;
676 if (s
->read_pattern
== QUORUM_READ_PATTERN_QUORUM
) {
677 ret
= read_quorum_children(acb
);
679 ret
= read_fifo_child(acb
);
681 quorum_aio_finalize(acb
);
687 * This function can count as GRAPH_RDLOCK because quorum_co_pwritev() holds the
688 * graph lock and keeps it until this coroutine has terminated.
690 static void coroutine_fn GRAPH_RDLOCK
write_quorum_entry(void *opaque
)
692 QuorumCo
*co
= opaque
;
693 QuorumAIOCB
*acb
= co
->acb
;
694 BDRVQuorumState
*s
= acb
->bs
->opaque
;
696 QuorumChildRequest
*sacb
= &acb
->qcrs
[i
];
698 sacb
->bs
= s
->children
[i
]->bs
;
699 if (acb
->flags
& BDRV_REQ_ZERO_WRITE
) {
700 sacb
->ret
= bdrv_co_pwrite_zeroes(s
->children
[i
], acb
->offset
,
701 acb
->bytes
, acb
->flags
);
703 sacb
->ret
= bdrv_co_pwritev(s
->children
[i
], acb
->offset
, acb
->bytes
,
704 acb
->qiov
, acb
->flags
);
706 if (sacb
->ret
== 0) {
707 acb
->success_count
++;
709 quorum_report_bad_acb(sacb
, sacb
->ret
);
712 assert(acb
->count
<= s
->num_children
);
713 assert(acb
->success_count
<= s
->num_children
);
715 /* Wake up the caller after the last write */
716 if (acb
->count
== s
->num_children
) {
717 qemu_coroutine_enter_if_inactive(acb
->co
);
721 static int coroutine_fn GRAPH_RDLOCK
722 quorum_co_pwritev(BlockDriverState
*bs
, int64_t offset
, int64_t bytes
,
723 QEMUIOVector
*qiov
, BdrvRequestFlags flags
)
725 BDRVQuorumState
*s
= bs
->opaque
;
726 QuorumAIOCB
*acb
= quorum_aio_get(bs
, qiov
, offset
, bytes
, flags
);
729 for (i
= 0; i
< s
->num_children
; i
++) {
736 co
= qemu_coroutine_create(write_quorum_entry
, &data
);
737 qemu_coroutine_enter(co
);
740 while (acb
->count
< s
->num_children
) {
741 qemu_coroutine_yield();
744 quorum_has_too_much_io_failed(acb
);
747 quorum_aio_finalize(acb
);
752 static int coroutine_fn GRAPH_RDLOCK
753 quorum_co_pwrite_zeroes(BlockDriverState
*bs
, int64_t offset
, int64_t bytes
,
754 BdrvRequestFlags flags
)
756 return quorum_co_pwritev(bs
, offset
, bytes
, NULL
,
757 flags
| BDRV_REQ_ZERO_WRITE
);
760 static int64_t coroutine_fn
quorum_co_getlength(BlockDriverState
*bs
)
762 BDRVQuorumState
*s
= bs
->opaque
;
766 /* check that all file have the same length */
767 result
= bdrv_co_getlength(s
->children
[0]->bs
);
771 for (i
= 1; i
< s
->num_children
; i
++) {
772 int64_t value
= bdrv_co_getlength(s
->children
[i
]->bs
);
776 if (value
!= result
) {
784 static coroutine_fn GRAPH_RDLOCK
int quorum_co_flush(BlockDriverState
*bs
)
786 BDRVQuorumState
*s
= bs
->opaque
;
787 QuorumVoteVersion
*winner
= NULL
;
788 QuorumVotes error_votes
;
789 QuorumVoteValue result_value
;
792 int success_count
= 0;
794 QLIST_INIT(&error_votes
.vote_list
);
795 error_votes
.compare
= quorum_64bits_compare
;
797 for (i
= 0; i
< s
->num_children
; i
++) {
798 result
= bdrv_co_flush(s
->children
[i
]->bs
);
800 quorum_report_bad(QUORUM_OP_TYPE_FLUSH
, 0, 0,
801 s
->children
[i
]->bs
->node_name
, result
);
802 result_value
.l
= result
;
803 quorum_count_vote(&error_votes
, &result_value
, i
);
809 if (success_count
>= s
->threshold
) {
812 winner
= quorum_get_vote_winner(&error_votes
);
813 result
= winner
->value
.l
;
815 quorum_free_vote_list(&error_votes
);
820 static bool quorum_recurse_can_replace(BlockDriverState
*bs
,
821 BlockDriverState
*to_replace
)
823 BDRVQuorumState
*s
= bs
->opaque
;
826 for (i
= 0; i
< s
->num_children
; i
++) {
828 * We have no idea whether our children show the same data as
829 * this node (@bs). It is actually highly likely that
830 * @to_replace does not, because replacing a broken child is
831 * one of the main use cases here.
833 * We do know that the new BDS will match @bs, so replacing
834 * any of our children by it will be safe. It cannot change
835 * the data this quorum node presents to its parents.
837 * However, replacing @to_replace by @bs in any of our
838 * children's chains may change visible data somewhere in
839 * there. We therefore cannot recurse down those chains with
840 * bdrv_recurse_can_replace().
841 * (More formally, bdrv_recurse_can_replace() requires that
842 * @to_replace will be replaced by something matching the @bs
843 * passed to it. We cannot guarantee that.)
845 * Thus, we can only check whether any of our immediate
846 * children matches @to_replace.
848 * (In the future, we might add a function to recurse down a
849 * chain that checks that nothing there cares about a change
850 * in data from the respective child in question. For
851 * example, most filters do not care when their child's data
852 * suddenly changes, as long as their parents do not care.)
854 if (s
->children
[i
]->bs
== to_replace
) {
856 * We now have to ensure that there is no other parent
857 * that cares about replacing this child by a node with
858 * potentially different data.
859 * We do so by checking whether there are any other parents
860 * at all, which is stricter than necessary, but also very
861 * simple. (We may decide to implement something more
862 * complex and permissive when there is an actual need for
865 return QLIST_FIRST(&to_replace
->parents
) == s
->children
[i
] &&
866 QLIST_NEXT(s
->children
[i
], next_parent
) == NULL
;
873 static int quorum_valid_threshold(int threshold
, int num_children
, Error
**errp
)
877 error_setg(errp
, QERR_INVALID_PARAMETER_VALUE
,
878 "vote-threshold", "a value >= 1");
882 if (threshold
> num_children
) {
883 error_setg(errp
, "threshold may not exceed children count");
890 static QemuOptsList quorum_runtime_opts
= {
892 .head
= QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts
.head
),
895 .name
= QUORUM_OPT_VOTE_THRESHOLD
,
896 .type
= QEMU_OPT_NUMBER
,
897 .help
= "The number of vote needed for reaching quorum",
900 .name
= QUORUM_OPT_BLKVERIFY
,
901 .type
= QEMU_OPT_BOOL
,
902 .help
= "Trigger block verify mode if set",
905 .name
= QUORUM_OPT_REWRITE
,
906 .type
= QEMU_OPT_BOOL
,
907 .help
= "Rewrite corrupted block on read quorum",
910 .name
= QUORUM_OPT_READ_PATTERN
,
911 .type
= QEMU_OPT_STRING
,
912 .help
= "Allowed pattern: quorum, fifo. Quorum is default",
914 { /* end of list */ }
918 static void quorum_refresh_flags(BlockDriverState
*bs
)
920 BDRVQuorumState
*s
= bs
->opaque
;
923 bs
->supported_zero_flags
=
924 BDRV_REQ_FUA
| BDRV_REQ_MAY_UNMAP
| BDRV_REQ_NO_FALLBACK
;
926 for (i
= 0; i
< s
->num_children
; i
++) {
927 bs
->supported_zero_flags
&= s
->children
[i
]->bs
->supported_zero_flags
;
930 bs
->supported_zero_flags
|= BDRV_REQ_WRITE_UNCHANGED
;
933 static int quorum_open(BlockDriverState
*bs
, QDict
*options
, int flags
,
936 BDRVQuorumState
*s
= bs
->opaque
;
937 QemuOpts
*opts
= NULL
;
938 const char *pattern_str
;
943 qdict_flatten(options
);
945 /* count how many different children are present */
946 s
->num_children
= qdict_array_entries(options
, "children.");
947 if (s
->num_children
< 0) {
948 error_setg(errp
, "Option children is not a valid array");
952 if (s
->num_children
< 1) {
953 error_setg(errp
, "Number of provided children must be 1 or more");
958 opts
= qemu_opts_create(&quorum_runtime_opts
, NULL
, 0, &error_abort
);
959 if (!qemu_opts_absorb_qdict(opts
, options
, errp
)) {
964 s
->threshold
= qemu_opt_get_number(opts
, QUORUM_OPT_VOTE_THRESHOLD
, 0);
965 /* and validate it against s->num_children */
966 ret
= quorum_valid_threshold(s
->threshold
, s
->num_children
, errp
);
971 pattern_str
= qemu_opt_get(opts
, QUORUM_OPT_READ_PATTERN
);
973 ret
= QUORUM_READ_PATTERN_QUORUM
;
975 ret
= qapi_enum_parse(&QuorumReadPattern_lookup
, pattern_str
,
979 error_setg(errp
, "Please set read-pattern as fifo or quorum");
982 s
->read_pattern
= ret
;
984 if (s
->read_pattern
== QUORUM_READ_PATTERN_QUORUM
) {
985 s
->is_blkverify
= qemu_opt_get_bool(opts
, QUORUM_OPT_BLKVERIFY
, false);
986 if (s
->is_blkverify
&& (s
->num_children
!= 2 || s
->threshold
!= 2)) {
987 error_setg(errp
, "blkverify=on can only be set if there are "
988 "exactly two files and vote-threshold is 2");
993 s
->rewrite_corrupted
= qemu_opt_get_bool(opts
, QUORUM_OPT_REWRITE
,
995 if (s
->rewrite_corrupted
&& s
->is_blkverify
) {
997 "rewrite-corrupted=on cannot be used with blkverify=on");
1003 /* allocate the children array */
1004 s
->children
= g_new0(BdrvChild
*, s
->num_children
);
1005 opened
= g_new0(bool, s
->num_children
);
1007 for (i
= 0; i
< s
->num_children
; i
++) {
1008 char indexstr
[INDEXSTR_LEN
];
1009 ret
= snprintf(indexstr
, INDEXSTR_LEN
, "children.%d", i
);
1010 assert(ret
< INDEXSTR_LEN
);
1012 s
->children
[i
] = bdrv_open_child(NULL
, options
, indexstr
, bs
,
1013 &child_of_bds
, BDRV_CHILD_DATA
, false,
1015 if (!s
->children
[i
]) {
1022 s
->next_child_index
= s
->num_children
;
1024 bs
->supported_write_flags
= BDRV_REQ_WRITE_UNCHANGED
;
1025 quorum_refresh_flags(bs
);
1031 /* cleanup on error */
1032 for (i
= 0; i
< s
->num_children
; i
++) {
1036 bdrv_unref_child(bs
, s
->children
[i
]);
1038 g_free(s
->children
);
1041 qemu_opts_del(opts
);
1045 static void quorum_close(BlockDriverState
*bs
)
1047 BDRVQuorumState
*s
= bs
->opaque
;
1050 for (i
= 0; i
< s
->num_children
; i
++) {
1051 bdrv_unref_child(bs
, s
->children
[i
]);
1054 g_free(s
->children
);
1057 static void quorum_add_child(BlockDriverState
*bs
, BlockDriverState
*child_bs
,
1060 BDRVQuorumState
*s
= bs
->opaque
;
1062 char indexstr
[INDEXSTR_LEN
];
1065 if (s
->is_blkverify
) {
1066 error_setg(errp
, "Cannot add a child to a quorum in blkverify mode");
1070 assert(s
->num_children
<= INT_MAX
/ sizeof(BdrvChild
*));
1071 if (s
->num_children
== INT_MAX
/ sizeof(BdrvChild
*) ||
1072 s
->next_child_index
== UINT_MAX
) {
1073 error_setg(errp
, "Too many children");
1077 ret
= snprintf(indexstr
, INDEXSTR_LEN
, "children.%u", s
->next_child_index
);
1078 if (ret
< 0 || ret
>= INDEXSTR_LEN
) {
1079 error_setg(errp
, "cannot generate child name");
1082 s
->next_child_index
++;
1084 bdrv_drained_begin(bs
);
1086 /* We can safely add the child now */
1089 child
= bdrv_attach_child(bs
, child_bs
, indexstr
, &child_of_bds
,
1090 BDRV_CHILD_DATA
, errp
);
1091 if (child
== NULL
) {
1092 s
->next_child_index
--;
1095 s
->children
= g_renew(BdrvChild
*, s
->children
, s
->num_children
+ 1);
1096 s
->children
[s
->num_children
++] = child
;
1097 quorum_refresh_flags(bs
);
1100 bdrv_drained_end(bs
);
1103 static void quorum_del_child(BlockDriverState
*bs
, BdrvChild
*child
,
1106 BDRVQuorumState
*s
= bs
->opaque
;
1107 char indexstr
[INDEXSTR_LEN
];
1110 for (i
= 0; i
< s
->num_children
; i
++) {
1111 if (s
->children
[i
] == child
) {
1116 /* we have checked it in bdrv_del_child() */
1117 assert(i
< s
->num_children
);
1119 if (s
->num_children
<= s
->threshold
) {
1121 "The number of children cannot be lower than the vote threshold %d",
1126 /* We know now that num_children > threshold, so blkverify must be false */
1127 assert(!s
->is_blkverify
);
1129 snprintf(indexstr
, INDEXSTR_LEN
, "children.%u", s
->next_child_index
- 1);
1130 if (!strncmp(child
->name
, indexstr
, INDEXSTR_LEN
)) {
1131 s
->next_child_index
--;
1134 bdrv_drained_begin(bs
);
1136 /* We can safely remove this child now */
1137 memmove(&s
->children
[i
], &s
->children
[i
+ 1],
1138 (s
->num_children
- i
- 1) * sizeof(BdrvChild
*));
1139 s
->children
= g_renew(BdrvChild
*, s
->children
, --s
->num_children
);
1140 bdrv_unref_child(bs
, child
);
1142 quorum_refresh_flags(bs
);
1143 bdrv_drained_end(bs
);
1146 static void quorum_gather_child_options(BlockDriverState
*bs
, QDict
*target
,
1147 bool backing_overridden
)
1149 BDRVQuorumState
*s
= bs
->opaque
;
1150 QList
*children_list
;
1154 * The generic implementation for gathering child options in
1155 * bdrv_refresh_filename() would use the names of the children
1156 * as specified for bdrv_open_child() or bdrv_attach_child(),
1157 * which is "children.%u" with %u being a value
1158 * (s->next_child_index) that is incremented each time a new child
1159 * is added (and never decremented). Since children can be
1160 * deleted at runtime, there may be gaps in that enumeration.
1161 * When creating a new quorum BDS and specifying the children for
1162 * it through runtime options, the enumeration used there may not
1163 * have any gaps, though.
1165 * Therefore, we have to create a new gap-less enumeration here
1166 * (which we can achieve by simply putting all of the children's
1167 * full_open_options into a QList).
1169 * XXX: Note that there are issues with the current child option
1170 * structure quorum uses (such as the fact that children do
1171 * not really have unique permanent names). Therefore, this
1172 * is going to have to change in the future and ideally we
1173 * want quorum to be covered by the generic implementation.
1176 children_list
= qlist_new();
1177 qdict_put(target
, "children", children_list
);
1179 for (i
= 0; i
< s
->num_children
; i
++) {
1180 qlist_append(children_list
,
1181 qobject_ref(s
->children
[i
]->bs
->full_open_options
));
1185 static char *quorum_dirname(BlockDriverState
*bs
, Error
**errp
)
1187 /* In general, there are multiple BDSs with different dirnames below this
1188 * one; so there is no unique dirname we could return (unless all are equal
1189 * by chance, or there is only one). Therefore, to be consistent, just
1190 * always return NULL. */
1191 error_setg(errp
, "Cannot generate a base directory for quorum nodes");
1195 static void quorum_child_perm(BlockDriverState
*bs
, BdrvChild
*c
,
1197 BlockReopenQueue
*reopen_queue
,
1198 uint64_t perm
, uint64_t shared
,
1199 uint64_t *nperm
, uint64_t *nshared
)
1201 BDRVQuorumState
*s
= bs
->opaque
;
1203 *nperm
= perm
& DEFAULT_PERM_PASSTHROUGH
;
1204 if (s
->rewrite_corrupted
) {
1205 *nperm
|= BLK_PERM_WRITE
;
1209 * We cannot share RESIZE or WRITE, as this would make the
1210 * children differ from each other.
1212 *nshared
= (shared
& (BLK_PERM_CONSISTENT_READ
|
1213 BLK_PERM_WRITE_UNCHANGED
))
1214 | DEFAULT_PERM_UNCHANGED
;
1218 * Each one of the children can report different status flags even
1219 * when they contain the same data, so what this function does is
1220 * return BDRV_BLOCK_ZERO if *all* children agree that a certain
1221 * region contains zeroes, and BDRV_BLOCK_DATA otherwise.
1223 static int coroutine_fn GRAPH_RDLOCK
1224 quorum_co_block_status(BlockDriverState
*bs
, bool want_zero
,
1225 int64_t offset
, int64_t count
,
1226 int64_t *pnum
, int64_t *map
, BlockDriverState
**file
)
1228 BDRVQuorumState
*s
= bs
->opaque
;
1230 int64_t pnum_zero
= count
;
1231 int64_t pnum_data
= 0;
1233 for (i
= 0; i
< s
->num_children
; i
++) {
1235 ret
= bdrv_co_common_block_status_above(s
->children
[i
]->bs
, NULL
, false,
1236 want_zero
, offset
, count
,
1237 &bytes
, NULL
, NULL
, NULL
);
1239 quorum_report_bad(QUORUM_OP_TYPE_READ
, offset
, count
,
1240 s
->children
[i
]->bs
->node_name
, ret
);
1245 * Even if all children agree about whether there are zeroes
1246 * or not at @offset they might disagree on the size, so use
1247 * the smallest when reporting BDRV_BLOCK_ZERO and the largest
1248 * when reporting BDRV_BLOCK_DATA.
1250 if (ret
& BDRV_BLOCK_ZERO
) {
1251 pnum_zero
= MIN(pnum_zero
, bytes
);
1253 pnum_data
= MAX(pnum_data
, bytes
);
1259 return BDRV_BLOCK_DATA
;
1262 return BDRV_BLOCK_ZERO
;
1266 static const char *const quorum_strong_runtime_opts
[] = {
1267 QUORUM_OPT_VOTE_THRESHOLD
,
1268 QUORUM_OPT_BLKVERIFY
,
1270 QUORUM_OPT_READ_PATTERN
,
1275 static BlockDriver bdrv_quorum
= {
1276 .format_name
= "quorum",
1278 .instance_size
= sizeof(BDRVQuorumState
),
1280 .bdrv_open
= quorum_open
,
1281 .bdrv_close
= quorum_close
,
1282 .bdrv_gather_child_options
= quorum_gather_child_options
,
1283 .bdrv_dirname
= quorum_dirname
,
1284 .bdrv_co_block_status
= quorum_co_block_status
,
1286 .bdrv_co_flush
= quorum_co_flush
,
1288 .bdrv_co_getlength
= quorum_co_getlength
,
1290 .bdrv_co_preadv
= quorum_co_preadv
,
1291 .bdrv_co_pwritev
= quorum_co_pwritev
,
1292 .bdrv_co_pwrite_zeroes
= quorum_co_pwrite_zeroes
,
1294 .bdrv_add_child
= quorum_add_child
,
1295 .bdrv_del_child
= quorum_del_child
,
1297 .bdrv_child_perm
= quorum_child_perm
,
1299 .bdrv_recurse_can_replace
= quorum_recurse_can_replace
,
1301 .strong_runtime_opts
= quorum_strong_runtime_opts
,
1304 static void bdrv_quorum_init(void)
1306 if (!qcrypto_hash_supports(QCRYPTO_HASH_ALG_SHA256
)) {
1307 /* SHA256 hash support is required for quorum device */
1310 bdrv_register(&bdrv_quorum
);
1313 block_init(bdrv_quorum_init
);