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 <gnutls/gnutls.h>
17 #include <gnutls/crypto.h>
18 #include "block/block_int.h"
19 #include "qapi/qmp/qbool.h"
20 #include "qapi/qmp/qdict.h"
21 #include "qapi/qmp/qerror.h"
22 #include "qapi/qmp/qint.h"
23 #include "qapi/qmp/qjson.h"
24 #include "qapi/qmp/qlist.h"
25 #include "qapi/qmp/qstring.h"
26 #include "qapi-event.h"
28 #define HASH_LENGTH 32
30 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
31 #define QUORUM_OPT_BLKVERIFY "blkverify"
32 #define QUORUM_OPT_REWRITE "rewrite-corrupted"
33 #define QUORUM_OPT_READ_PATTERN "read-pattern"
35 /* This union holds a vote hash value */
36 typedef union QuorumVoteValue
{
37 char h
[HASH_LENGTH
]; /* SHA-256 hash */
38 int64_t l
; /* simpler 64 bits hash */
42 typedef struct QuorumVoteItem
{
44 QLIST_ENTRY(QuorumVoteItem
) next
;
47 /* this structure is a vote version. A version is the set of votes sharing the
49 * The set of votes will be tracked with the items field and its cardinality is
52 typedef struct QuorumVoteVersion
{
53 QuorumVoteValue value
;
56 QLIST_HEAD(, QuorumVoteItem
) items
;
57 QLIST_ENTRY(QuorumVoteVersion
) next
;
60 /* this structure holds a group of vote versions together */
61 typedef struct QuorumVotes
{
62 QLIST_HEAD(, QuorumVoteVersion
) vote_list
;
63 bool (*compare
)(QuorumVoteValue
*a
, QuorumVoteValue
*b
);
66 /* the following structure holds the state of one quorum instance */
67 typedef struct BDRVQuorumState
{
68 BlockDriverState
**bs
; /* children BlockDriverStates */
69 int num_children
; /* children count */
70 int threshold
; /* if less than threshold children reads gave the
71 * same result a quorum error occurs.
73 bool is_blkverify
; /* true if the driver is in blkverify mode
74 * Writes are mirrored on two children devices.
75 * On reads the two children devices' contents are
76 * compared and if a difference is spotted its
77 * location is printed and the code aborts.
78 * It is useful to debug other block drivers by
79 * comparing them with a reference one.
81 bool rewrite_corrupted
;/* true if the driver must rewrite-on-read corrupted
82 * block if Quorum is reached.
85 QuorumReadPattern read_pattern
;
88 typedef struct QuorumAIOCB QuorumAIOCB
;
90 /* Quorum will create one instance of the following structure per operation it
91 * performs on its children.
92 * So for each read/write operation coming from the upper layer there will be
93 * $children_count QuorumChildRequest.
95 typedef struct QuorumChildRequest
{
101 } QuorumChildRequest
;
103 /* Quorum will use the following structure to track progress of each read/write
104 * operation received by the upper layer.
105 * This structure hold pointers to the QuorumChildRequest structures instances
106 * used to do operations on each children and track overall progress.
111 /* Request metadata */
115 QEMUIOVector
*qiov
; /* calling IOV */
117 QuorumChildRequest
*qcrs
; /* individual child requests */
118 int count
; /* number of completed AIOCB */
119 int success_count
; /* number of successfully completed AIOCB */
121 int rewrite_count
; /* number of replica to rewrite: count down to
122 * zero once writes are fired
129 int child_iter
; /* which child to read in fifo pattern */
132 static bool quorum_vote(QuorumAIOCB
*acb
);
134 static void quorum_aio_cancel(BlockAIOCB
*blockacb
)
136 QuorumAIOCB
*acb
= container_of(blockacb
, QuorumAIOCB
, common
);
137 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
140 /* cancel all callbacks */
141 for (i
= 0; i
< s
->num_children
; i
++) {
142 if (acb
->qcrs
[i
].aiocb
) {
143 bdrv_aio_cancel_async(acb
->qcrs
[i
].aiocb
);
148 static AIOCBInfo quorum_aiocb_info
= {
149 .aiocb_size
= sizeof(QuorumAIOCB
),
150 .cancel_async
= quorum_aio_cancel
,
153 static void quorum_aio_finalize(QuorumAIOCB
*acb
)
161 acb
->common
.cb(acb
->common
.opaque
, ret
);
164 /* on the quorum case acb->child_iter == s->num_children - 1 */
165 for (i
= 0; i
<= acb
->child_iter
; i
++) {
166 qemu_vfree(acb
->qcrs
[i
].buf
);
167 qemu_iovec_destroy(&acb
->qcrs
[i
].qiov
);
175 static bool quorum_sha256_compare(QuorumVoteValue
*a
, QuorumVoteValue
*b
)
177 return !memcmp(a
->h
, b
->h
, HASH_LENGTH
);
180 static bool quorum_64bits_compare(QuorumVoteValue
*a
, QuorumVoteValue
*b
)
185 static QuorumAIOCB
*quorum_aio_get(BDRVQuorumState
*s
,
186 BlockDriverState
*bs
,
190 BlockCompletionFunc
*cb
,
193 QuorumAIOCB
*acb
= qemu_aio_get(&quorum_aiocb_info
, bs
, cb
, opaque
);
196 acb
->common
.bs
->opaque
= s
;
197 acb
->sector_num
= sector_num
;
198 acb
->nb_sectors
= nb_sectors
;
200 acb
->qcrs
= g_new0(QuorumChildRequest
, s
->num_children
);
202 acb
->success_count
= 0;
203 acb
->rewrite_count
= 0;
204 acb
->votes
.compare
= quorum_sha256_compare
;
205 QLIST_INIT(&acb
->votes
.vote_list
);
206 acb
->is_read
= false;
209 for (i
= 0; i
< s
->num_children
; i
++) {
210 acb
->qcrs
[i
].buf
= NULL
;
211 acb
->qcrs
[i
].ret
= 0;
212 acb
->qcrs
[i
].parent
= acb
;
218 static void quorum_report_bad(QuorumAIOCB
*acb
, char *node_name
, int ret
)
220 const char *msg
= NULL
;
222 msg
= strerror(-ret
);
224 qapi_event_send_quorum_report_bad(!!msg
, msg
, node_name
,
225 acb
->sector_num
, acb
->nb_sectors
, &error_abort
);
228 static void quorum_report_failure(QuorumAIOCB
*acb
)
230 const char *reference
= bdrv_get_device_or_node_name(acb
->common
.bs
);
231 qapi_event_send_quorum_failure(reference
, acb
->sector_num
,
232 acb
->nb_sectors
, &error_abort
);
235 static int quorum_vote_error(QuorumAIOCB
*acb
);
237 static bool quorum_has_too_much_io_failed(QuorumAIOCB
*acb
)
239 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
241 if (acb
->success_count
< s
->threshold
) {
242 acb
->vote_ret
= quorum_vote_error(acb
);
243 quorum_report_failure(acb
);
250 static void quorum_rewrite_aio_cb(void *opaque
, int ret
)
252 QuorumAIOCB
*acb
= opaque
;
254 /* one less rewrite to do */
255 acb
->rewrite_count
--;
257 /* wait until all rewrite callbacks have completed */
258 if (acb
->rewrite_count
) {
262 quorum_aio_finalize(acb
);
265 static BlockAIOCB
*read_fifo_child(QuorumAIOCB
*acb
);
267 static void quorum_copy_qiov(QEMUIOVector
*dest
, QEMUIOVector
*source
)
270 assert(dest
->niov
== source
->niov
);
271 assert(dest
->size
== source
->size
);
272 for (i
= 0; i
< source
->niov
; i
++) {
273 assert(dest
->iov
[i
].iov_len
== source
->iov
[i
].iov_len
);
274 memcpy(dest
->iov
[i
].iov_base
,
275 source
->iov
[i
].iov_base
,
276 source
->iov
[i
].iov_len
);
280 static void quorum_aio_cb(void *opaque
, int ret
)
282 QuorumChildRequest
*sacb
= opaque
;
283 QuorumAIOCB
*acb
= sacb
->parent
;
284 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
285 bool rewrite
= false;
287 if (acb
->is_read
&& s
->read_pattern
== QUORUM_READ_PATTERN_FIFO
) {
288 /* We try to read next child in FIFO order if we fail to read */
289 if (ret
< 0 && ++acb
->child_iter
< s
->num_children
) {
290 read_fifo_child(acb
);
295 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[acb
->child_iter
].qiov
);
298 quorum_aio_finalize(acb
);
305 acb
->success_count
++;
307 quorum_report_bad(acb
, sacb
->aiocb
->bs
->node_name
, ret
);
309 assert(acb
->count
<= s
->num_children
);
310 assert(acb
->success_count
<= s
->num_children
);
311 if (acb
->count
< s
->num_children
) {
315 /* Do the vote on read */
317 rewrite
= quorum_vote(acb
);
319 quorum_has_too_much_io_failed(acb
);
322 /* if no rewrite is done the code will finish right away */
324 quorum_aio_finalize(acb
);
328 static void quorum_report_bad_versions(BDRVQuorumState
*s
,
330 QuorumVoteValue
*value
)
332 QuorumVoteVersion
*version
;
333 QuorumVoteItem
*item
;
335 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
336 if (acb
->votes
.compare(&version
->value
, value
)) {
339 QLIST_FOREACH(item
, &version
->items
, next
) {
340 quorum_report_bad(acb
, s
->bs
[item
->index
]->node_name
, 0);
345 static bool quorum_rewrite_bad_versions(BDRVQuorumState
*s
, QuorumAIOCB
*acb
,
346 QuorumVoteValue
*value
)
348 QuorumVoteVersion
*version
;
349 QuorumVoteItem
*item
;
352 /* first count the number of bad versions: done first to avoid concurrency
355 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
356 if (acb
->votes
.compare(&version
->value
, value
)) {
359 QLIST_FOREACH(item
, &version
->items
, next
) {
364 /* quorum_rewrite_aio_cb will count down this to zero */
365 acb
->rewrite_count
= count
;
367 /* now fire the correcting rewrites */
368 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
369 if (acb
->votes
.compare(&version
->value
, value
)) {
372 QLIST_FOREACH(item
, &version
->items
, next
) {
373 bdrv_aio_writev(s
->bs
[item
->index
], acb
->sector_num
, acb
->qiov
,
374 acb
->nb_sectors
, quorum_rewrite_aio_cb
, acb
);
378 /* return true if any rewrite is done else false */
382 static void quorum_count_vote(QuorumVotes
*votes
,
383 QuorumVoteValue
*value
,
386 QuorumVoteVersion
*v
= NULL
, *version
= NULL
;
387 QuorumVoteItem
*item
;
389 /* look if we have something with this hash */
390 QLIST_FOREACH(v
, &votes
->vote_list
, next
) {
391 if (votes
->compare(&v
->value
, value
)) {
397 /* It's a version not yet in the list add it */
399 version
= g_new0(QuorumVoteVersion
, 1);
400 QLIST_INIT(&version
->items
);
401 memcpy(&version
->value
, value
, sizeof(version
->value
));
402 version
->index
= index
;
403 version
->vote_count
= 0;
404 QLIST_INSERT_HEAD(&votes
->vote_list
, version
, next
);
407 version
->vote_count
++;
409 item
= g_new0(QuorumVoteItem
, 1);
411 QLIST_INSERT_HEAD(&version
->items
, item
, next
);
414 static void quorum_free_vote_list(QuorumVotes
*votes
)
416 QuorumVoteVersion
*version
, *next_version
;
417 QuorumVoteItem
*item
, *next_item
;
419 QLIST_FOREACH_SAFE(version
, &votes
->vote_list
, next
, next_version
) {
420 QLIST_REMOVE(version
, next
);
421 QLIST_FOREACH_SAFE(item
, &version
->items
, next
, next_item
) {
422 QLIST_REMOVE(item
, next
);
429 static int quorum_compute_hash(QuorumAIOCB
*acb
, int i
, QuorumVoteValue
*hash
)
432 gnutls_hash_hd_t dig
;
433 QEMUIOVector
*qiov
= &acb
->qcrs
[i
].qiov
;
435 ret
= gnutls_hash_init(&dig
, GNUTLS_DIG_SHA256
);
441 for (j
= 0; j
< qiov
->niov
; j
++) {
442 ret
= gnutls_hash(dig
, qiov
->iov
[j
].iov_base
, qiov
->iov
[j
].iov_len
);
448 gnutls_hash_deinit(dig
, (void *) hash
);
452 static QuorumVoteVersion
*quorum_get_vote_winner(QuorumVotes
*votes
)
455 QuorumVoteVersion
*candidate
, *winner
= NULL
;
457 QLIST_FOREACH(candidate
, &votes
->vote_list
, next
) {
458 if (candidate
->vote_count
> max
) {
459 max
= candidate
->vote_count
;
467 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
468 * differing byte location. Yet it is handcoded to compare vectors one byte
469 * after another so it does not benefit from the libc SIMD optimizations.
470 * quorum_iovec_compare is written for speed and should be used in the non
471 * blkverify mode of quorum.
473 static bool quorum_iovec_compare(QEMUIOVector
*a
, QEMUIOVector
*b
)
478 assert(a
->niov
== b
->niov
);
479 for (i
= 0; i
< a
->niov
; i
++) {
480 assert(a
->iov
[i
].iov_len
== b
->iov
[i
].iov_len
);
481 result
= memcmp(a
->iov
[i
].iov_base
,
492 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB
*acb
,
493 const char *fmt
, ...)
498 fprintf(stderr
, "quorum: sector_num=%" PRId64
" nb_sectors=%d ",
499 acb
->sector_num
, acb
->nb_sectors
);
500 vfprintf(stderr
, fmt
, ap
);
501 fprintf(stderr
, "\n");
506 static bool quorum_compare(QuorumAIOCB
*acb
,
510 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
513 /* This driver will replace blkverify in this particular case */
514 if (s
->is_blkverify
) {
515 offset
= qemu_iovec_compare(a
, b
);
517 quorum_err(acb
, "contents mismatch in sector %" PRId64
,
519 (uint64_t)(offset
/ BDRV_SECTOR_SIZE
));
524 return quorum_iovec_compare(a
, b
);
527 /* Do a vote to get the error code */
528 static int quorum_vote_error(QuorumAIOCB
*acb
)
530 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
531 QuorumVoteVersion
*winner
= NULL
;
532 QuorumVotes error_votes
;
533 QuorumVoteValue result_value
;
537 QLIST_INIT(&error_votes
.vote_list
);
538 error_votes
.compare
= quorum_64bits_compare
;
540 for (i
= 0; i
< s
->num_children
; i
++) {
541 ret
= acb
->qcrs
[i
].ret
;
544 result_value
.l
= ret
;
545 quorum_count_vote(&error_votes
, &result_value
, i
);
550 winner
= quorum_get_vote_winner(&error_votes
);
551 ret
= winner
->value
.l
;
554 quorum_free_vote_list(&error_votes
);
559 static bool quorum_vote(QuorumAIOCB
*acb
)
562 bool rewrite
= false;
564 QuorumVoteValue hash
;
565 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
566 QuorumVoteVersion
*winner
;
568 if (quorum_has_too_much_io_failed(acb
)) {
572 /* get the index of the first successful read */
573 for (i
= 0; i
< s
->num_children
; i
++) {
574 if (!acb
->qcrs
[i
].ret
) {
579 assert(i
< s
->num_children
);
581 /* compare this read with all other successful reads stopping at quorum
584 for (j
= i
+ 1; j
< s
->num_children
; j
++) {
585 if (acb
->qcrs
[j
].ret
) {
588 quorum
= quorum_compare(acb
, &acb
->qcrs
[i
].qiov
, &acb
->qcrs
[j
].qiov
);
594 /* Every successful read agrees */
596 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[i
].qiov
);
600 /* compute hashes for each successful read, also store indexes */
601 for (i
= 0; i
< s
->num_children
; i
++) {
602 if (acb
->qcrs
[i
].ret
) {
605 ret
= quorum_compute_hash(acb
, i
, &hash
);
606 /* if ever the hash computation failed */
611 quorum_count_vote(&acb
->votes
, &hash
, i
);
614 /* vote to select the most represented version */
615 winner
= quorum_get_vote_winner(&acb
->votes
);
617 /* if the winner count is smaller than threshold the read fails */
618 if (winner
->vote_count
< s
->threshold
) {
619 quorum_report_failure(acb
);
620 acb
->vote_ret
= -EIO
;
624 /* we have a winner: copy it */
625 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[winner
->index
].qiov
);
627 /* some versions are bad print them */
628 quorum_report_bad_versions(s
, acb
, &winner
->value
);
630 /* corruption correction is enabled */
631 if (s
->rewrite_corrupted
) {
632 rewrite
= quorum_rewrite_bad_versions(s
, acb
, &winner
->value
);
637 quorum_free_vote_list(&acb
->votes
);
641 static BlockAIOCB
*read_quorum_children(QuorumAIOCB
*acb
)
643 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
646 for (i
= 0; i
< s
->num_children
; i
++) {
647 acb
->qcrs
[i
].buf
= qemu_blockalign(s
->bs
[i
], acb
->qiov
->size
);
648 qemu_iovec_init(&acb
->qcrs
[i
].qiov
, acb
->qiov
->niov
);
649 qemu_iovec_clone(&acb
->qcrs
[i
].qiov
, acb
->qiov
, acb
->qcrs
[i
].buf
);
652 for (i
= 0; i
< s
->num_children
; i
++) {
653 bdrv_aio_readv(s
->bs
[i
], acb
->sector_num
, &acb
->qcrs
[i
].qiov
,
654 acb
->nb_sectors
, quorum_aio_cb
, &acb
->qcrs
[i
]);
660 static BlockAIOCB
*read_fifo_child(QuorumAIOCB
*acb
)
662 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
664 acb
->qcrs
[acb
->child_iter
].buf
= qemu_blockalign(s
->bs
[acb
->child_iter
],
666 qemu_iovec_init(&acb
->qcrs
[acb
->child_iter
].qiov
, acb
->qiov
->niov
);
667 qemu_iovec_clone(&acb
->qcrs
[acb
->child_iter
].qiov
, acb
->qiov
,
668 acb
->qcrs
[acb
->child_iter
].buf
);
669 bdrv_aio_readv(s
->bs
[acb
->child_iter
], acb
->sector_num
,
670 &acb
->qcrs
[acb
->child_iter
].qiov
, acb
->nb_sectors
,
671 quorum_aio_cb
, &acb
->qcrs
[acb
->child_iter
]);
676 static BlockAIOCB
*quorum_aio_readv(BlockDriverState
*bs
,
680 BlockCompletionFunc
*cb
,
683 BDRVQuorumState
*s
= bs
->opaque
;
684 QuorumAIOCB
*acb
= quorum_aio_get(s
, bs
, qiov
, sector_num
,
685 nb_sectors
, cb
, opaque
);
688 if (s
->read_pattern
== QUORUM_READ_PATTERN_QUORUM
) {
689 acb
->child_iter
= s
->num_children
- 1;
690 return read_quorum_children(acb
);
694 return read_fifo_child(acb
);
697 static BlockAIOCB
*quorum_aio_writev(BlockDriverState
*bs
,
701 BlockCompletionFunc
*cb
,
704 BDRVQuorumState
*s
= bs
->opaque
;
705 QuorumAIOCB
*acb
= quorum_aio_get(s
, bs
, qiov
, sector_num
, nb_sectors
,
709 for (i
= 0; i
< s
->num_children
; i
++) {
710 acb
->qcrs
[i
].aiocb
= bdrv_aio_writev(s
->bs
[i
], sector_num
, qiov
,
711 nb_sectors
, &quorum_aio_cb
,
718 static int64_t quorum_getlength(BlockDriverState
*bs
)
720 BDRVQuorumState
*s
= bs
->opaque
;
724 /* check that all file have the same length */
725 result
= bdrv_getlength(s
->bs
[0]);
729 for (i
= 1; i
< s
->num_children
; i
++) {
730 int64_t value
= bdrv_getlength(s
->bs
[i
]);
734 if (value
!= result
) {
742 static void quorum_invalidate_cache(BlockDriverState
*bs
, Error
**errp
)
744 BDRVQuorumState
*s
= bs
->opaque
;
745 Error
*local_err
= NULL
;
748 for (i
= 0; i
< s
->num_children
; i
++) {
749 bdrv_invalidate_cache(s
->bs
[i
], &local_err
);
751 error_propagate(errp
, local_err
);
757 static coroutine_fn
int quorum_co_flush(BlockDriverState
*bs
)
759 BDRVQuorumState
*s
= bs
->opaque
;
760 QuorumVoteVersion
*winner
= NULL
;
761 QuorumVotes error_votes
;
762 QuorumVoteValue result_value
;
766 QLIST_INIT(&error_votes
.vote_list
);
767 error_votes
.compare
= quorum_64bits_compare
;
769 for (i
= 0; i
< s
->num_children
; i
++) {
770 result
= bdrv_co_flush(s
->bs
[i
]);
771 result_value
.l
= result
;
772 quorum_count_vote(&error_votes
, &result_value
, i
);
775 winner
= quorum_get_vote_winner(&error_votes
);
776 result
= winner
->value
.l
;
778 quorum_free_vote_list(&error_votes
);
783 static bool quorum_recurse_is_first_non_filter(BlockDriverState
*bs
,
784 BlockDriverState
*candidate
)
786 BDRVQuorumState
*s
= bs
->opaque
;
789 for (i
= 0; i
< s
->num_children
; i
++) {
790 bool perm
= bdrv_recurse_is_first_non_filter(s
->bs
[i
],
800 static int quorum_valid_threshold(int threshold
, int num_children
, Error
**errp
)
804 error_setg(errp
, QERR_INVALID_PARAMETER_VALUE
,
805 "vote-threshold", "value >= 1");
809 if (threshold
> num_children
) {
810 error_setg(errp
, "threshold may not exceed children count");
817 static QemuOptsList quorum_runtime_opts
= {
819 .head
= QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts
.head
),
822 .name
= QUORUM_OPT_VOTE_THRESHOLD
,
823 .type
= QEMU_OPT_NUMBER
,
824 .help
= "The number of vote needed for reaching quorum",
827 .name
= QUORUM_OPT_BLKVERIFY
,
828 .type
= QEMU_OPT_BOOL
,
829 .help
= "Trigger block verify mode if set",
832 .name
= QUORUM_OPT_REWRITE
,
833 .type
= QEMU_OPT_BOOL
,
834 .help
= "Rewrite corrupted block on read quorum",
837 .name
= QUORUM_OPT_READ_PATTERN
,
838 .type
= QEMU_OPT_STRING
,
839 .help
= "Allowed pattern: quorum, fifo. Quorum is default",
841 { /* end of list */ }
845 static int parse_read_pattern(const char *opt
)
850 /* Set quorum as default */
851 return QUORUM_READ_PATTERN_QUORUM
;
854 for (i
= 0; i
< QUORUM_READ_PATTERN_MAX
; i
++) {
855 if (!strcmp(opt
, QuorumReadPattern_lookup
[i
])) {
863 static int quorum_open(BlockDriverState
*bs
, QDict
*options
, int flags
,
866 BDRVQuorumState
*s
= bs
->opaque
;
867 Error
*local_err
= NULL
;
868 QemuOpts
*opts
= NULL
;
873 qdict_flatten(options
);
875 /* count how many different children are present */
876 s
->num_children
= qdict_array_entries(options
, "children.");
877 if (s
->num_children
< 0) {
878 error_setg(&local_err
, "Option children is not a valid array");
882 if (s
->num_children
< 2) {
883 error_setg(&local_err
,
884 "Number of provided children must be greater than 1");
889 opts
= qemu_opts_create(&quorum_runtime_opts
, NULL
, 0, &error_abort
);
890 qemu_opts_absorb_qdict(opts
, options
, &local_err
);
896 s
->threshold
= qemu_opt_get_number(opts
, QUORUM_OPT_VOTE_THRESHOLD
, 0);
897 ret
= parse_read_pattern(qemu_opt_get(opts
, QUORUM_OPT_READ_PATTERN
));
899 error_setg(&local_err
, "Please set read-pattern as fifo or quorum");
902 s
->read_pattern
= ret
;
904 if (s
->read_pattern
== QUORUM_READ_PATTERN_QUORUM
) {
905 /* and validate it against s->num_children */
906 ret
= quorum_valid_threshold(s
->threshold
, s
->num_children
, &local_err
);
911 /* is the driver in blkverify mode */
912 if (qemu_opt_get_bool(opts
, QUORUM_OPT_BLKVERIFY
, false) &&
913 s
->num_children
== 2 && s
->threshold
== 2) {
914 s
->is_blkverify
= true;
915 } else if (qemu_opt_get_bool(opts
, QUORUM_OPT_BLKVERIFY
, false)) {
916 fprintf(stderr
, "blkverify mode is set by setting blkverify=on "
917 "and using two files with vote_threshold=2\n");
920 s
->rewrite_corrupted
= qemu_opt_get_bool(opts
, QUORUM_OPT_REWRITE
,
922 if (s
->rewrite_corrupted
&& s
->is_blkverify
) {
923 error_setg(&local_err
,
924 "rewrite-corrupted=on cannot be used with blkverify=on");
930 /* allocate the children BlockDriverState array */
931 s
->bs
= g_new0(BlockDriverState
*, s
->num_children
);
932 opened
= g_new0(bool, s
->num_children
);
934 for (i
= 0; i
< s
->num_children
; i
++) {
936 ret
= snprintf(indexstr
, 32, "children.%d", i
);
939 ret
= bdrv_open_image(&s
->bs
[i
], NULL
, options
, indexstr
, bs
,
940 &child_format
, false, &local_err
);
952 /* cleanup on error */
953 for (i
= 0; i
< s
->num_children
; i
++) {
957 bdrv_unref(s
->bs
[i
]);
963 /* propagate error */
965 error_propagate(errp
, local_err
);
970 static void quorum_close(BlockDriverState
*bs
)
972 BDRVQuorumState
*s
= bs
->opaque
;
975 for (i
= 0; i
< s
->num_children
; i
++) {
976 bdrv_unref(s
->bs
[i
]);
982 static void quorum_detach_aio_context(BlockDriverState
*bs
)
984 BDRVQuorumState
*s
= bs
->opaque
;
987 for (i
= 0; i
< s
->num_children
; i
++) {
988 bdrv_detach_aio_context(s
->bs
[i
]);
992 static void quorum_attach_aio_context(BlockDriverState
*bs
,
993 AioContext
*new_context
)
995 BDRVQuorumState
*s
= bs
->opaque
;
998 for (i
= 0; i
< s
->num_children
; i
++) {
999 bdrv_attach_aio_context(s
->bs
[i
], new_context
);
1003 static void quorum_refresh_filename(BlockDriverState
*bs
)
1005 BDRVQuorumState
*s
= bs
->opaque
;
1010 for (i
= 0; i
< s
->num_children
; i
++) {
1011 bdrv_refresh_filename(s
->bs
[i
]);
1012 if (!s
->bs
[i
]->full_open_options
) {
1017 children
= qlist_new();
1018 for (i
= 0; i
< s
->num_children
; i
++) {
1019 QINCREF(s
->bs
[i
]->full_open_options
);
1020 qlist_append_obj(children
, QOBJECT(s
->bs
[i
]->full_open_options
));
1024 qdict_put_obj(opts
, "driver", QOBJECT(qstring_from_str("quorum")));
1025 qdict_put_obj(opts
, QUORUM_OPT_VOTE_THRESHOLD
,
1026 QOBJECT(qint_from_int(s
->threshold
)));
1027 qdict_put_obj(opts
, QUORUM_OPT_BLKVERIFY
,
1028 QOBJECT(qbool_from_bool(s
->is_blkverify
)));
1029 qdict_put_obj(opts
, QUORUM_OPT_REWRITE
,
1030 QOBJECT(qbool_from_bool(s
->rewrite_corrupted
)));
1031 qdict_put_obj(opts
, "children", QOBJECT(children
));
1033 bs
->full_open_options
= opts
;
1036 static BlockDriver bdrv_quorum
= {
1037 .format_name
= "quorum",
1038 .protocol_name
= "quorum",
1040 .instance_size
= sizeof(BDRVQuorumState
),
1042 .bdrv_file_open
= quorum_open
,
1043 .bdrv_close
= quorum_close
,
1044 .bdrv_refresh_filename
= quorum_refresh_filename
,
1046 .bdrv_co_flush_to_disk
= quorum_co_flush
,
1048 .bdrv_getlength
= quorum_getlength
,
1050 .bdrv_aio_readv
= quorum_aio_readv
,
1051 .bdrv_aio_writev
= quorum_aio_writev
,
1052 .bdrv_invalidate_cache
= quorum_invalidate_cache
,
1054 .bdrv_detach_aio_context
= quorum_detach_aio_context
,
1055 .bdrv_attach_aio_context
= quorum_attach_aio_context
,
1058 .bdrv_recurse_is_first_non_filter
= quorum_recurse_is_first_non_filter
,
1061 static void bdrv_quorum_init(void)
1063 bdrv_register(&bdrv_quorum
);
1066 block_init(bdrv_quorum_init
);