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 "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 #include "crypto/hash.h"
29 #define HASH_LENGTH 32
31 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
32 #define QUORUM_OPT_BLKVERIFY "blkverify"
33 #define QUORUM_OPT_REWRITE "rewrite-corrupted"
34 #define QUORUM_OPT_READ_PATTERN "read-pattern"
36 /* This union holds a vote hash value */
37 typedef union QuorumVoteValue
{
38 uint8_t h
[HASH_LENGTH
]; /* SHA-256 hash */
39 int64_t l
; /* simpler 64 bits hash */
43 typedef struct QuorumVoteItem
{
45 QLIST_ENTRY(QuorumVoteItem
) next
;
48 /* this structure is a vote version. A version is the set of votes sharing the
50 * The set of votes will be tracked with the items field and its cardinality is
53 typedef struct QuorumVoteVersion
{
54 QuorumVoteValue value
;
57 QLIST_HEAD(, QuorumVoteItem
) items
;
58 QLIST_ENTRY(QuorumVoteVersion
) next
;
61 /* this structure holds a group of vote versions together */
62 typedef struct QuorumVotes
{
63 QLIST_HEAD(, QuorumVoteVersion
) vote_list
;
64 bool (*compare
)(QuorumVoteValue
*a
, QuorumVoteValue
*b
);
67 /* the following structure holds the state of one quorum instance */
68 typedef struct BDRVQuorumState
{
69 BdrvChild
**children
; /* children BlockDriverStates */
70 int num_children
; /* children count */
71 unsigned next_child_index
; /* the index of the next child that should
74 int threshold
; /* if less than threshold children reads gave the
75 * same result a quorum error occurs.
77 bool is_blkverify
; /* true if the driver is in blkverify mode
78 * Writes are mirrored on two children devices.
79 * On reads the two children devices' contents are
80 * compared and if a difference is spotted its
81 * location is printed and the code aborts.
82 * It is useful to debug other block drivers by
83 * comparing them with a reference one.
85 bool rewrite_corrupted
;/* true if the driver must rewrite-on-read corrupted
86 * block if Quorum is reached.
89 QuorumReadPattern read_pattern
;
92 typedef struct QuorumAIOCB QuorumAIOCB
;
94 /* Quorum will create one instance of the following structure per operation it
95 * performs on its children.
96 * So for each read/write operation coming from the upper layer there will be
97 * $children_count QuorumChildRequest.
99 typedef struct QuorumChildRequest
{
100 BlockDriverState
*bs
;
105 } QuorumChildRequest
;
107 /* Quorum will use the following structure to track progress of each read/write
108 * operation received by the upper layer.
109 * This structure hold pointers to the QuorumChildRequest structures instances
110 * used to do operations on each children and track overall progress.
113 BlockDriverState
*bs
;
116 /* Request metadata */
120 QEMUIOVector
*qiov
; /* calling IOV */
122 QuorumChildRequest
*qcrs
; /* individual child requests */
123 int count
; /* number of completed AIOCB */
124 int success_count
; /* number of successfully completed AIOCB */
126 int rewrite_count
; /* number of replica to rewrite: count down to
127 * zero once writes are fired
134 int children_read
; /* how many children have been read from */
137 typedef struct QuorumCo
{
142 static void quorum_aio_finalize(QuorumAIOCB
*acb
)
148 static bool quorum_sha256_compare(QuorumVoteValue
*a
, QuorumVoteValue
*b
)
150 return !memcmp(a
->h
, b
->h
, HASH_LENGTH
);
153 static bool quorum_64bits_compare(QuorumVoteValue
*a
, QuorumVoteValue
*b
)
158 static QuorumAIOCB
*quorum_aio_get(BlockDriverState
*bs
,
163 BDRVQuorumState
*s
= bs
->opaque
;
164 QuorumAIOCB
*acb
= g_new(QuorumAIOCB
, 1);
167 acb
->co
= qemu_coroutine_self();
169 acb
->sector_num
= sector_num
;
170 acb
->nb_sectors
= nb_sectors
;
172 acb
->qcrs
= g_new0(QuorumChildRequest
, s
->num_children
);
174 acb
->success_count
= 0;
175 acb
->rewrite_count
= 0;
176 acb
->votes
.compare
= quorum_sha256_compare
;
177 QLIST_INIT(&acb
->votes
.vote_list
);
178 acb
->is_read
= false;
181 for (i
= 0; i
< s
->num_children
; i
++) {
182 acb
->qcrs
[i
].buf
= NULL
;
183 acb
->qcrs
[i
].ret
= 0;
184 acb
->qcrs
[i
].parent
= acb
;
190 static void quorum_report_bad(QuorumOpType type
, uint64_t sector_num
,
191 int nb_sectors
, char *node_name
, int ret
)
193 const char *msg
= NULL
;
195 msg
= strerror(-ret
);
198 qapi_event_send_quorum_report_bad(type
, !!msg
, msg
, node_name
,
199 sector_num
, nb_sectors
, &error_abort
);
202 static void quorum_report_failure(QuorumAIOCB
*acb
)
204 const char *reference
= bdrv_get_device_or_node_name(acb
->bs
);
205 qapi_event_send_quorum_failure(reference
, acb
->sector_num
,
206 acb
->nb_sectors
, &error_abort
);
209 static int quorum_vote_error(QuorumAIOCB
*acb
);
211 static bool quorum_has_too_much_io_failed(QuorumAIOCB
*acb
)
213 BDRVQuorumState
*s
= acb
->bs
->opaque
;
215 if (acb
->success_count
< s
->threshold
) {
216 acb
->vote_ret
= quorum_vote_error(acb
);
217 quorum_report_failure(acb
);
224 static void quorum_rewrite_aio_cb(void *opaque
, int ret
)
226 QuorumAIOCB
*acb
= opaque
;
228 /* one less rewrite to do */
229 acb
->rewrite_count
--;
230 qemu_coroutine_enter_if_inactive(acb
->co
);
233 static int read_fifo_child(QuorumAIOCB
*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
->sector_num
, acb
->nb_sectors
,
253 sacb
->bs
->node_name
, ret
);
256 static int quorum_fifo_aio_cb(void *opaque
, int ret
)
258 QuorumChildRequest
*sacb
= opaque
;
259 QuorumAIOCB
*acb
= sacb
->parent
;
260 BDRVQuorumState
*s
= acb
->bs
->opaque
;
262 assert(acb
->is_read
&& s
->read_pattern
== QUORUM_READ_PATTERN_FIFO
);
265 quorum_report_bad_acb(sacb
, ret
);
267 /* We try to read next child in FIFO order if we fail to read */
268 if (acb
->children_read
< s
->num_children
) {
269 return read_fifo_child(acb
);
275 /* FIXME: rewrite failed children if acb->children_read > 1? */
280 static void quorum_report_bad_versions(BDRVQuorumState
*s
,
282 QuorumVoteValue
*value
)
284 QuorumVoteVersion
*version
;
285 QuorumVoteItem
*item
;
287 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
288 if (acb
->votes
.compare(&version
->value
, value
)) {
291 QLIST_FOREACH(item
, &version
->items
, next
) {
292 quorum_report_bad(QUORUM_OP_TYPE_READ
, acb
->sector_num
,
294 s
->children
[item
->index
]->bs
->node_name
, 0);
299 static bool quorum_rewrite_bad_versions(BDRVQuorumState
*s
, QuorumAIOCB
*acb
,
300 QuorumVoteValue
*value
)
302 QuorumVoteVersion
*version
;
303 QuorumVoteItem
*item
;
306 /* first count the number of bad versions: done first to avoid concurrency
309 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
310 if (acb
->votes
.compare(&version
->value
, value
)) {
313 QLIST_FOREACH(item
, &version
->items
, next
) {
318 /* quorum_rewrite_aio_cb will count down this to zero */
319 acb
->rewrite_count
= count
;
321 /* now fire the correcting rewrites */
322 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
323 if (acb
->votes
.compare(&version
->value
, value
)) {
326 QLIST_FOREACH(item
, &version
->items
, next
) {
327 bdrv_aio_writev(s
->children
[item
->index
], acb
->sector_num
,
328 acb
->qiov
, acb
->nb_sectors
, quorum_rewrite_aio_cb
,
333 /* return true if any rewrite is done else false */
337 static void quorum_count_vote(QuorumVotes
*votes
,
338 QuorumVoteValue
*value
,
341 QuorumVoteVersion
*v
= NULL
, *version
= NULL
;
342 QuorumVoteItem
*item
;
344 /* look if we have something with this hash */
345 QLIST_FOREACH(v
, &votes
->vote_list
, next
) {
346 if (votes
->compare(&v
->value
, value
)) {
352 /* It's a version not yet in the list add it */
354 version
= g_new0(QuorumVoteVersion
, 1);
355 QLIST_INIT(&version
->items
);
356 memcpy(&version
->value
, value
, sizeof(version
->value
));
357 version
->index
= index
;
358 version
->vote_count
= 0;
359 QLIST_INSERT_HEAD(&votes
->vote_list
, version
, next
);
362 version
->vote_count
++;
364 item
= g_new0(QuorumVoteItem
, 1);
366 QLIST_INSERT_HEAD(&version
->items
, item
, next
);
369 static void quorum_free_vote_list(QuorumVotes
*votes
)
371 QuorumVoteVersion
*version
, *next_version
;
372 QuorumVoteItem
*item
, *next_item
;
374 QLIST_FOREACH_SAFE(version
, &votes
->vote_list
, next
, next_version
) {
375 QLIST_REMOVE(version
, next
);
376 QLIST_FOREACH_SAFE(item
, &version
->items
, next
, next_item
) {
377 QLIST_REMOVE(item
, next
);
384 static int quorum_compute_hash(QuorumAIOCB
*acb
, int i
, QuorumVoteValue
*hash
)
386 QEMUIOVector
*qiov
= &acb
->qcrs
[i
].qiov
;
387 size_t len
= sizeof(hash
->h
);
388 uint8_t *data
= hash
->h
;
390 /* XXX - would be nice if we could pass in the Error **
391 * and propagate that back, but this quorum code is
392 * restricted to just errno values currently */
393 if (qcrypto_hash_bytesv(QCRYPTO_HASH_ALG_SHA256
,
394 qiov
->iov
, qiov
->niov
,
403 static QuorumVoteVersion
*quorum_get_vote_winner(QuorumVotes
*votes
)
406 QuorumVoteVersion
*candidate
, *winner
= NULL
;
408 QLIST_FOREACH(candidate
, &votes
->vote_list
, next
) {
409 if (candidate
->vote_count
> max
) {
410 max
= candidate
->vote_count
;
418 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
419 * differing byte location. Yet it is handcoded to compare vectors one byte
420 * after another so it does not benefit from the libc SIMD optimizations.
421 * quorum_iovec_compare is written for speed and should be used in the non
422 * blkverify mode of quorum.
424 static bool quorum_iovec_compare(QEMUIOVector
*a
, QEMUIOVector
*b
)
429 assert(a
->niov
== b
->niov
);
430 for (i
= 0; i
< a
->niov
; i
++) {
431 assert(a
->iov
[i
].iov_len
== b
->iov
[i
].iov_len
);
432 result
= memcmp(a
->iov
[i
].iov_base
,
443 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB
*acb
,
444 const char *fmt
, ...)
449 fprintf(stderr
, "quorum: sector_num=%" PRId64
" nb_sectors=%d ",
450 acb
->sector_num
, acb
->nb_sectors
);
451 vfprintf(stderr
, fmt
, ap
);
452 fprintf(stderr
, "\n");
457 static bool quorum_compare(QuorumAIOCB
*acb
,
461 BDRVQuorumState
*s
= acb
->bs
->opaque
;
464 /* This driver will replace blkverify in this particular case */
465 if (s
->is_blkverify
) {
466 offset
= qemu_iovec_compare(a
, b
);
468 quorum_err(acb
, "contents mismatch in sector %" PRId64
,
470 (uint64_t)(offset
/ BDRV_SECTOR_SIZE
));
475 return quorum_iovec_compare(a
, b
);
478 /* Do a vote to get the error code */
479 static int quorum_vote_error(QuorumAIOCB
*acb
)
481 BDRVQuorumState
*s
= acb
->bs
->opaque
;
482 QuorumVoteVersion
*winner
= NULL
;
483 QuorumVotes error_votes
;
484 QuorumVoteValue result_value
;
488 QLIST_INIT(&error_votes
.vote_list
);
489 error_votes
.compare
= quorum_64bits_compare
;
491 for (i
= 0; i
< s
->num_children
; i
++) {
492 ret
= acb
->qcrs
[i
].ret
;
495 result_value
.l
= ret
;
496 quorum_count_vote(&error_votes
, &result_value
, i
);
501 winner
= quorum_get_vote_winner(&error_votes
);
502 ret
= winner
->value
.l
;
505 quorum_free_vote_list(&error_votes
);
510 static void quorum_vote(QuorumAIOCB
*acb
)
514 QuorumVoteValue hash
;
515 BDRVQuorumState
*s
= acb
->bs
->opaque
;
516 QuorumVoteVersion
*winner
;
518 if (quorum_has_too_much_io_failed(acb
)) {
522 /* get the index of the first successful read */
523 for (i
= 0; i
< s
->num_children
; i
++) {
524 if (!acb
->qcrs
[i
].ret
) {
529 assert(i
< s
->num_children
);
531 /* compare this read with all other successful reads stopping at quorum
534 for (j
= i
+ 1; j
< s
->num_children
; j
++) {
535 if (acb
->qcrs
[j
].ret
) {
538 quorum
= quorum_compare(acb
, &acb
->qcrs
[i
].qiov
, &acb
->qcrs
[j
].qiov
);
544 /* Every successful read agrees */
546 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[i
].qiov
);
550 /* compute hashes for each successful read, also store indexes */
551 for (i
= 0; i
< s
->num_children
; i
++) {
552 if (acb
->qcrs
[i
].ret
) {
555 ret
= quorum_compute_hash(acb
, i
, &hash
);
556 /* if ever the hash computation failed */
561 quorum_count_vote(&acb
->votes
, &hash
, i
);
564 /* vote to select the most represented version */
565 winner
= quorum_get_vote_winner(&acb
->votes
);
567 /* if the winner count is smaller than threshold the read fails */
568 if (winner
->vote_count
< s
->threshold
) {
569 quorum_report_failure(acb
);
570 acb
->vote_ret
= -EIO
;
574 /* we have a winner: copy it */
575 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[winner
->index
].qiov
);
577 /* some versions are bad print them */
578 quorum_report_bad_versions(s
, acb
, &winner
->value
);
580 /* corruption correction is enabled */
581 if (s
->rewrite_corrupted
) {
582 quorum_rewrite_bad_versions(s
, acb
, &winner
->value
);
587 quorum_free_vote_list(&acb
->votes
);
590 static void read_quorum_children_entry(void *opaque
)
592 QuorumCo
*co
= opaque
;
593 QuorumAIOCB
*acb
= co
->acb
;
594 BDRVQuorumState
*s
= acb
->bs
->opaque
;
596 QuorumChildRequest
*sacb
= &acb
->qcrs
[i
];
598 sacb
->bs
= s
->children
[i
]->bs
;
599 sacb
->ret
= bdrv_co_preadv(s
->children
[i
],
600 acb
->sector_num
* BDRV_SECTOR_SIZE
,
601 acb
->nb_sectors
* BDRV_SECTOR_SIZE
,
602 &acb
->qcrs
[i
].qiov
, 0);
604 if (sacb
->ret
== 0) {
605 acb
->success_count
++;
607 quorum_report_bad_acb(sacb
, sacb
->ret
);
611 assert(acb
->count
<= s
->num_children
);
612 assert(acb
->success_count
<= s
->num_children
);
614 /* Wake up the caller after the last read */
615 if (acb
->count
== s
->num_children
) {
616 qemu_coroutine_enter_if_inactive(acb
->co
);
620 static int read_quorum_children(QuorumAIOCB
*acb
)
622 BDRVQuorumState
*s
= acb
->bs
->opaque
;
625 acb
->children_read
= s
->num_children
;
626 for (i
= 0; i
< s
->num_children
; i
++) {
627 acb
->qcrs
[i
].buf
= qemu_blockalign(s
->children
[i
]->bs
, acb
->qiov
->size
);
628 qemu_iovec_init(&acb
->qcrs
[i
].qiov
, acb
->qiov
->niov
);
629 qemu_iovec_clone(&acb
->qcrs
[i
].qiov
, acb
->qiov
, acb
->qcrs
[i
].buf
);
632 for (i
= 0; i
< s
->num_children
; i
++) {
639 co
= qemu_coroutine_create(read_quorum_children_entry
, &data
);
640 qemu_coroutine_enter(co
);
643 while (acb
->count
< s
->num_children
) {
644 qemu_coroutine_yield();
647 /* Do the vote on read */
649 for (i
= 0; i
< s
->num_children
; i
++) {
650 qemu_vfree(acb
->qcrs
[i
].buf
);
651 qemu_iovec_destroy(&acb
->qcrs
[i
].qiov
);
654 while (acb
->rewrite_count
) {
655 qemu_coroutine_yield();
663 static int read_fifo_child(QuorumAIOCB
*acb
)
665 BDRVQuorumState
*s
= acb
->bs
->opaque
;
666 int n
= acb
->children_read
++;
669 acb
->qcrs
[n
].bs
= s
->children
[n
]->bs
;
670 ret
= bdrv_co_preadv(s
->children
[n
], acb
->sector_num
* BDRV_SECTOR_SIZE
,
671 acb
->nb_sectors
* BDRV_SECTOR_SIZE
, acb
->qiov
, 0);
672 ret
= quorum_fifo_aio_cb(&acb
->qcrs
[n
], ret
);
677 static int quorum_co_readv(BlockDriverState
*bs
,
678 int64_t sector_num
, int nb_sectors
,
681 BDRVQuorumState
*s
= bs
->opaque
;
682 QuorumAIOCB
*acb
= quorum_aio_get(bs
, qiov
, sector_num
, nb_sectors
);
686 acb
->children_read
= 0;
688 if (s
->read_pattern
== QUORUM_READ_PATTERN_QUORUM
) {
689 ret
= read_quorum_children(acb
);
691 ret
= read_fifo_child(acb
);
693 quorum_aio_finalize(acb
);
698 static void write_quorum_entry(void *opaque
)
700 QuorumCo
*co
= opaque
;
701 QuorumAIOCB
*acb
= co
->acb
;
702 BDRVQuorumState
*s
= acb
->bs
->opaque
;
704 QuorumChildRequest
*sacb
= &acb
->qcrs
[i
];
706 sacb
->bs
= s
->children
[i
]->bs
;
707 sacb
->ret
= bdrv_co_pwritev(s
->children
[i
],
708 acb
->sector_num
* BDRV_SECTOR_SIZE
,
709 acb
->nb_sectors
* BDRV_SECTOR_SIZE
,
711 if (sacb
->ret
== 0) {
712 acb
->success_count
++;
714 quorum_report_bad_acb(sacb
, sacb
->ret
);
717 assert(acb
->count
<= s
->num_children
);
718 assert(acb
->success_count
<= s
->num_children
);
720 /* Wake up the caller after the last write */
721 if (acb
->count
== s
->num_children
) {
722 qemu_coroutine_enter_if_inactive(acb
->co
);
726 static int quorum_co_writev(BlockDriverState
*bs
,
727 int64_t sector_num
, int nb_sectors
,
730 BDRVQuorumState
*s
= bs
->opaque
;
731 QuorumAIOCB
*acb
= quorum_aio_get(bs
, qiov
, sector_num
, nb_sectors
);
734 for (i
= 0; i
< s
->num_children
; i
++) {
741 co
= qemu_coroutine_create(write_quorum_entry
, &data
);
742 qemu_coroutine_enter(co
);
745 while (acb
->count
< s
->num_children
) {
746 qemu_coroutine_yield();
749 quorum_has_too_much_io_failed(acb
);
752 quorum_aio_finalize(acb
);
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,
798 bdrv_nb_sectors(s
->children
[i
]->bs
),
799 s
->children
[i
]->bs
->node_name
, result
);
800 result_value
.l
= result
;
801 quorum_count_vote(&error_votes
, &result_value
, i
);
807 if (success_count
>= s
->threshold
) {
810 winner
= quorum_get_vote_winner(&error_votes
);
811 result
= winner
->value
.l
;
813 quorum_free_vote_list(&error_votes
);
818 static bool quorum_recurse_is_first_non_filter(BlockDriverState
*bs
,
819 BlockDriverState
*candidate
)
821 BDRVQuorumState
*s
= bs
->opaque
;
824 for (i
= 0; i
< s
->num_children
; i
++) {
825 bool perm
= bdrv_recurse_is_first_non_filter(s
->children
[i
]->bs
,
835 static int quorum_valid_threshold(int threshold
, int num_children
, Error
**errp
)
839 error_setg(errp
, QERR_INVALID_PARAMETER_VALUE
,
840 "vote-threshold", "value >= 1");
844 if (threshold
> num_children
) {
845 error_setg(errp
, "threshold may not exceed children count");
852 static QemuOptsList quorum_runtime_opts
= {
854 .head
= QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts
.head
),
857 .name
= QUORUM_OPT_VOTE_THRESHOLD
,
858 .type
= QEMU_OPT_NUMBER
,
859 .help
= "The number of vote needed for reaching quorum",
862 .name
= QUORUM_OPT_BLKVERIFY
,
863 .type
= QEMU_OPT_BOOL
,
864 .help
= "Trigger block verify mode if set",
867 .name
= QUORUM_OPT_REWRITE
,
868 .type
= QEMU_OPT_BOOL
,
869 .help
= "Rewrite corrupted block on read quorum",
872 .name
= QUORUM_OPT_READ_PATTERN
,
873 .type
= QEMU_OPT_STRING
,
874 .help
= "Allowed pattern: quorum, fifo. Quorum is default",
876 { /* end of list */ }
880 static int parse_read_pattern(const char *opt
)
885 /* Set quorum as default */
886 return QUORUM_READ_PATTERN_QUORUM
;
889 for (i
= 0; i
< QUORUM_READ_PATTERN__MAX
; i
++) {
890 if (!strcmp(opt
, QuorumReadPattern_lookup
[i
])) {
898 static int quorum_open(BlockDriverState
*bs
, QDict
*options
, int flags
,
901 BDRVQuorumState
*s
= bs
->opaque
;
902 Error
*local_err
= NULL
;
903 QemuOpts
*opts
= NULL
;
908 qdict_flatten(options
);
910 /* count how many different children are present */
911 s
->num_children
= qdict_array_entries(options
, "children.");
912 if (s
->num_children
< 0) {
913 error_setg(&local_err
, "Option children is not a valid array");
917 if (s
->num_children
< 1) {
918 error_setg(&local_err
,
919 "Number of provided children must be 1 or more");
924 opts
= qemu_opts_create(&quorum_runtime_opts
, NULL
, 0, &error_abort
);
925 qemu_opts_absorb_qdict(opts
, options
, &local_err
);
931 s
->threshold
= qemu_opt_get_number(opts
, QUORUM_OPT_VOTE_THRESHOLD
, 0);
932 /* and validate it against s->num_children */
933 ret
= quorum_valid_threshold(s
->threshold
, s
->num_children
, &local_err
);
938 ret
= parse_read_pattern(qemu_opt_get(opts
, QUORUM_OPT_READ_PATTERN
));
940 error_setg(&local_err
, "Please set read-pattern as fifo or quorum");
943 s
->read_pattern
= ret
;
945 if (s
->read_pattern
== QUORUM_READ_PATTERN_QUORUM
) {
946 /* is the driver in blkverify mode */
947 if (qemu_opt_get_bool(opts
, QUORUM_OPT_BLKVERIFY
, false) &&
948 s
->num_children
== 2 && s
->threshold
== 2) {
949 s
->is_blkverify
= true;
950 } else if (qemu_opt_get_bool(opts
, QUORUM_OPT_BLKVERIFY
, false)) {
951 fprintf(stderr
, "blkverify mode is set by setting blkverify=on "
952 "and using two files with vote_threshold=2\n");
955 s
->rewrite_corrupted
= qemu_opt_get_bool(opts
, QUORUM_OPT_REWRITE
,
957 if (s
->rewrite_corrupted
&& s
->is_blkverify
) {
958 error_setg(&local_err
,
959 "rewrite-corrupted=on cannot be used with blkverify=on");
965 /* allocate the children array */
966 s
->children
= g_new0(BdrvChild
*, s
->num_children
);
967 opened
= g_new0(bool, s
->num_children
);
969 for (i
= 0; i
< s
->num_children
; i
++) {
971 ret
= snprintf(indexstr
, 32, "children.%d", i
);
974 s
->children
[i
] = bdrv_open_child(NULL
, options
, indexstr
, bs
,
975 &child_format
, false, &local_err
);
983 s
->next_child_index
= s
->num_children
;
989 /* cleanup on error */
990 for (i
= 0; i
< s
->num_children
; i
++) {
994 bdrv_unref_child(bs
, s
->children
[i
]);
1000 /* propagate error */
1001 error_propagate(errp
, local_err
);
1005 static void quorum_close(BlockDriverState
*bs
)
1007 BDRVQuorumState
*s
= bs
->opaque
;
1010 for (i
= 0; i
< s
->num_children
; i
++) {
1011 bdrv_unref_child(bs
, s
->children
[i
]);
1014 g_free(s
->children
);
1017 static void quorum_add_child(BlockDriverState
*bs
, BlockDriverState
*child_bs
,
1020 BDRVQuorumState
*s
= bs
->opaque
;
1025 assert(s
->num_children
<= INT_MAX
/ sizeof(BdrvChild
*));
1026 if (s
->num_children
== INT_MAX
/ sizeof(BdrvChild
*) ||
1027 s
->next_child_index
== UINT_MAX
) {
1028 error_setg(errp
, "Too many children");
1032 ret
= snprintf(indexstr
, 32, "children.%u", s
->next_child_index
);
1033 if (ret
< 0 || ret
>= 32) {
1034 error_setg(errp
, "cannot generate child name");
1037 s
->next_child_index
++;
1039 bdrv_drained_begin(bs
);
1041 /* We can safely add the child now */
1043 child
= bdrv_attach_child(bs
, child_bs
, indexstr
, &child_format
);
1044 s
->children
= g_renew(BdrvChild
*, s
->children
, s
->num_children
+ 1);
1045 s
->children
[s
->num_children
++] = child
;
1047 bdrv_drained_end(bs
);
1050 static void quorum_del_child(BlockDriverState
*bs
, BdrvChild
*child
,
1053 BDRVQuorumState
*s
= bs
->opaque
;
1056 for (i
= 0; i
< s
->num_children
; i
++) {
1057 if (s
->children
[i
] == child
) {
1062 /* we have checked it in bdrv_del_child() */
1063 assert(i
< s
->num_children
);
1065 if (s
->num_children
<= s
->threshold
) {
1067 "The number of children cannot be lower than the vote threshold %d",
1072 bdrv_drained_begin(bs
);
1074 /* We can safely remove this child now */
1075 memmove(&s
->children
[i
], &s
->children
[i
+ 1],
1076 (s
->num_children
- i
- 1) * sizeof(BdrvChild
*));
1077 s
->children
= g_renew(BdrvChild
*, s
->children
, --s
->num_children
);
1078 bdrv_unref_child(bs
, child
);
1080 bdrv_drained_end(bs
);
1083 static void quorum_refresh_filename(BlockDriverState
*bs
, QDict
*options
)
1085 BDRVQuorumState
*s
= bs
->opaque
;
1090 for (i
= 0; i
< s
->num_children
; i
++) {
1091 bdrv_refresh_filename(s
->children
[i
]->bs
);
1092 if (!s
->children
[i
]->bs
->full_open_options
) {
1097 children
= qlist_new();
1098 for (i
= 0; i
< s
->num_children
; i
++) {
1099 QINCREF(s
->children
[i
]->bs
->full_open_options
);
1100 qlist_append_obj(children
,
1101 QOBJECT(s
->children
[i
]->bs
->full_open_options
));
1105 qdict_put_obj(opts
, "driver", QOBJECT(qstring_from_str("quorum")));
1106 qdict_put_obj(opts
, QUORUM_OPT_VOTE_THRESHOLD
,
1107 QOBJECT(qint_from_int(s
->threshold
)));
1108 qdict_put_obj(opts
, QUORUM_OPT_BLKVERIFY
,
1109 QOBJECT(qbool_from_bool(s
->is_blkverify
)));
1110 qdict_put_obj(opts
, QUORUM_OPT_REWRITE
,
1111 QOBJECT(qbool_from_bool(s
->rewrite_corrupted
)));
1112 qdict_put_obj(opts
, "children", QOBJECT(children
));
1114 bs
->full_open_options
= opts
;
1117 static BlockDriver bdrv_quorum
= {
1118 .format_name
= "quorum",
1119 .protocol_name
= "quorum",
1121 .instance_size
= sizeof(BDRVQuorumState
),
1123 .bdrv_file_open
= quorum_open
,
1124 .bdrv_close
= quorum_close
,
1125 .bdrv_refresh_filename
= quorum_refresh_filename
,
1127 .bdrv_co_flush_to_disk
= quorum_co_flush
,
1129 .bdrv_getlength
= quorum_getlength
,
1131 .bdrv_co_readv
= quorum_co_readv
,
1132 .bdrv_co_writev
= quorum_co_writev
,
1134 .bdrv_add_child
= quorum_add_child
,
1135 .bdrv_del_child
= quorum_del_child
,
1138 .bdrv_recurse_is_first_non_filter
= quorum_recurse_is_first_non_filter
,
1141 static void bdrv_quorum_init(void)
1143 if (!qcrypto_hash_supports(QCRYPTO_HASH_ALG_SHA256
)) {
1144 /* SHA256 hash support is required for quorum device */
1147 bdrv_register(&bdrv_quorum
);
1150 block_init(bdrv_quorum_init
);