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/qjson.h"
21 #define HASH_LENGTH 32
23 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
24 #define QUORUM_OPT_BLKVERIFY "blkverify"
26 /* This union holds a vote hash value */
27 typedef union QuorumVoteValue
{
28 char h
[HASH_LENGTH
]; /* SHA-256 hash */
29 int64_t l
; /* simpler 64 bits hash */
33 typedef struct QuorumVoteItem
{
35 QLIST_ENTRY(QuorumVoteItem
) next
;
38 /* this structure is a vote version. A version is the set of votes sharing the
40 * The set of votes will be tracked with the items field and its cardinality is
43 typedef struct QuorumVoteVersion
{
44 QuorumVoteValue value
;
47 QLIST_HEAD(, QuorumVoteItem
) items
;
48 QLIST_ENTRY(QuorumVoteVersion
) next
;
51 /* this structure holds a group of vote versions together */
52 typedef struct QuorumVotes
{
53 QLIST_HEAD(, QuorumVoteVersion
) vote_list
;
54 bool (*compare
)(QuorumVoteValue
*a
, QuorumVoteValue
*b
);
57 /* the following structure holds the state of one quorum instance */
58 typedef struct BDRVQuorumState
{
59 BlockDriverState
**bs
; /* children BlockDriverStates */
60 int num_children
; /* children count */
61 int threshold
; /* if less than threshold children reads gave the
62 * same result a quorum error occurs.
64 bool is_blkverify
; /* true if the driver is in blkverify mode
65 * Writes are mirrored on two children devices.
66 * On reads the two children devices' contents are
67 * compared and if a difference is spotted its
68 * location is printed and the code aborts.
69 * It is useful to debug other block drivers by
70 * comparing them with a reference one.
74 typedef struct QuorumAIOCB QuorumAIOCB
;
76 /* Quorum will create one instance of the following structure per operation it
77 * performs on its children.
78 * So for each read/write operation coming from the upper layer there will be
79 * $children_count QuorumChildRequest.
81 typedef struct QuorumChildRequest
{
82 BlockDriverAIOCB
*aiocb
;
89 /* Quorum will use the following structure to track progress of each read/write
90 * operation received by the upper layer.
91 * This structure hold pointers to the QuorumChildRequest structures instances
92 * used to do operations on each children and track overall progress.
95 BlockDriverAIOCB common
;
97 /* Request metadata */
101 QEMUIOVector
*qiov
; /* calling IOV */
103 QuorumChildRequest
*qcrs
; /* individual child requests */
104 int count
; /* number of completed AIOCB */
105 int success_count
; /* number of successfully completed AIOCB */
113 static void quorum_vote(QuorumAIOCB
*acb
);
115 static void quorum_aio_cancel(BlockDriverAIOCB
*blockacb
)
117 QuorumAIOCB
*acb
= container_of(blockacb
, QuorumAIOCB
, common
);
118 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
121 /* cancel all callbacks */
122 for (i
= 0; i
< s
->num_children
; i
++) {
123 bdrv_aio_cancel(acb
->qcrs
[i
].aiocb
);
127 qemu_aio_release(acb
);
130 static AIOCBInfo quorum_aiocb_info
= {
131 .aiocb_size
= sizeof(QuorumAIOCB
),
132 .cancel
= quorum_aio_cancel
,
135 static void quorum_aio_finalize(QuorumAIOCB
*acb
)
137 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
144 acb
->common
.cb(acb
->common
.opaque
, ret
);
147 for (i
= 0; i
< s
->num_children
; i
++) {
148 qemu_vfree(acb
->qcrs
[i
].buf
);
149 qemu_iovec_destroy(&acb
->qcrs
[i
].qiov
);
154 qemu_aio_release(acb
);
157 static bool quorum_sha256_compare(QuorumVoteValue
*a
, QuorumVoteValue
*b
)
159 return !memcmp(a
->h
, b
->h
, HASH_LENGTH
);
162 static bool quorum_64bits_compare(QuorumVoteValue
*a
, QuorumVoteValue
*b
)
167 static QuorumAIOCB
*quorum_aio_get(BDRVQuorumState
*s
,
168 BlockDriverState
*bs
,
172 BlockDriverCompletionFunc
*cb
,
175 QuorumAIOCB
*acb
= qemu_aio_get(&quorum_aiocb_info
, bs
, cb
, opaque
);
178 acb
->common
.bs
->opaque
= s
;
179 acb
->sector_num
= sector_num
;
180 acb
->nb_sectors
= nb_sectors
;
182 acb
->qcrs
= g_new0(QuorumChildRequest
, s
->num_children
);
184 acb
->success_count
= 0;
185 acb
->votes
.compare
= quorum_sha256_compare
;
186 QLIST_INIT(&acb
->votes
.vote_list
);
187 acb
->is_read
= false;
190 for (i
= 0; i
< s
->num_children
; i
++) {
191 acb
->qcrs
[i
].buf
= NULL
;
192 acb
->qcrs
[i
].ret
= 0;
193 acb
->qcrs
[i
].parent
= acb
;
199 static void quorum_report_bad(QuorumAIOCB
*acb
, char *node_name
, int ret
)
203 data
= qobject_from_jsonf("{ 'ret': %d"
205 ", 'sector-num': %" PRId64
206 ", 'sectors-count': %d }",
207 ret
, node_name
, acb
->sector_num
, acb
->nb_sectors
);
208 monitor_protocol_event(QEVENT_QUORUM_REPORT_BAD
, data
);
209 qobject_decref(data
);
212 static void quorum_report_failure(QuorumAIOCB
*acb
)
215 const char *reference
= acb
->common
.bs
->device_name
[0] ?
216 acb
->common
.bs
->device_name
:
217 acb
->common
.bs
->node_name
;
218 data
= qobject_from_jsonf("{ 'reference': %s"
219 ", 'sector-num': %" PRId64
220 ", 'sectors-count': %d }",
221 reference
, acb
->sector_num
, acb
->nb_sectors
);
222 monitor_protocol_event(QEVENT_QUORUM_FAILURE
, data
);
223 qobject_decref(data
);
226 static int quorum_vote_error(QuorumAIOCB
*acb
);
228 static bool quorum_has_too_much_io_failed(QuorumAIOCB
*acb
)
230 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
232 if (acb
->success_count
< s
->threshold
) {
233 acb
->vote_ret
= quorum_vote_error(acb
);
234 quorum_report_failure(acb
);
241 static void quorum_aio_cb(void *opaque
, int ret
)
243 QuorumChildRequest
*sacb
= opaque
;
244 QuorumAIOCB
*acb
= sacb
->parent
;
245 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
250 acb
->success_count
++;
252 quorum_report_bad(acb
, sacb
->aiocb
->bs
->node_name
, ret
);
254 assert(acb
->count
<= s
->num_children
);
255 assert(acb
->success_count
<= s
->num_children
);
256 if (acb
->count
< s
->num_children
) {
260 /* Do the vote on read */
264 quorum_has_too_much_io_failed(acb
);
267 quorum_aio_finalize(acb
);
270 static void quorum_report_bad_versions(BDRVQuorumState
*s
,
272 QuorumVoteValue
*value
)
274 QuorumVoteVersion
*version
;
275 QuorumVoteItem
*item
;
277 QLIST_FOREACH(version
, &acb
->votes
.vote_list
, next
) {
278 if (acb
->votes
.compare(&version
->value
, value
)) {
281 QLIST_FOREACH(item
, &version
->items
, next
) {
282 quorum_report_bad(acb
, s
->bs
[item
->index
]->node_name
, 0);
287 static void quorum_copy_qiov(QEMUIOVector
*dest
, QEMUIOVector
*source
)
290 assert(dest
->niov
== source
->niov
);
291 assert(dest
->size
== source
->size
);
292 for (i
= 0; i
< source
->niov
; i
++) {
293 assert(dest
->iov
[i
].iov_len
== source
->iov
[i
].iov_len
);
294 memcpy(dest
->iov
[i
].iov_base
,
295 source
->iov
[i
].iov_base
,
296 source
->iov
[i
].iov_len
);
300 static void quorum_count_vote(QuorumVotes
*votes
,
301 QuorumVoteValue
*value
,
304 QuorumVoteVersion
*v
= NULL
, *version
= NULL
;
305 QuorumVoteItem
*item
;
307 /* look if we have something with this hash */
308 QLIST_FOREACH(v
, &votes
->vote_list
, next
) {
309 if (votes
->compare(&v
->value
, value
)) {
315 /* It's a version not yet in the list add it */
317 version
= g_new0(QuorumVoteVersion
, 1);
318 QLIST_INIT(&version
->items
);
319 memcpy(&version
->value
, value
, sizeof(version
->value
));
320 version
->index
= index
;
321 version
->vote_count
= 0;
322 QLIST_INSERT_HEAD(&votes
->vote_list
, version
, next
);
325 version
->vote_count
++;
327 item
= g_new0(QuorumVoteItem
, 1);
329 QLIST_INSERT_HEAD(&version
->items
, item
, next
);
332 static void quorum_free_vote_list(QuorumVotes
*votes
)
334 QuorumVoteVersion
*version
, *next_version
;
335 QuorumVoteItem
*item
, *next_item
;
337 QLIST_FOREACH_SAFE(version
, &votes
->vote_list
, next
, next_version
) {
338 QLIST_REMOVE(version
, next
);
339 QLIST_FOREACH_SAFE(item
, &version
->items
, next
, next_item
) {
340 QLIST_REMOVE(item
, next
);
347 static int quorum_compute_hash(QuorumAIOCB
*acb
, int i
, QuorumVoteValue
*hash
)
350 gnutls_hash_hd_t dig
;
351 QEMUIOVector
*qiov
= &acb
->qcrs
[i
].qiov
;
353 ret
= gnutls_hash_init(&dig
, GNUTLS_DIG_SHA256
);
359 for (j
= 0; j
< qiov
->niov
; j
++) {
360 ret
= gnutls_hash(dig
, qiov
->iov
[j
].iov_base
, qiov
->iov
[j
].iov_len
);
366 gnutls_hash_deinit(dig
, (void *) hash
);
370 static QuorumVoteVersion
*quorum_get_vote_winner(QuorumVotes
*votes
)
373 QuorumVoteVersion
*candidate
, *winner
= NULL
;
375 QLIST_FOREACH(candidate
, &votes
->vote_list
, next
) {
376 if (candidate
->vote_count
> max
) {
377 max
= candidate
->vote_count
;
385 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
386 * differing byte location. Yet it is handcoded to compare vectors one byte
387 * after another so it does not benefit from the libc SIMD optimizations.
388 * quorum_iovec_compare is written for speed and should be used in the non
389 * blkverify mode of quorum.
391 static bool quorum_iovec_compare(QEMUIOVector
*a
, QEMUIOVector
*b
)
396 assert(a
->niov
== b
->niov
);
397 for (i
= 0; i
< a
->niov
; i
++) {
398 assert(a
->iov
[i
].iov_len
== b
->iov
[i
].iov_len
);
399 result
= memcmp(a
->iov
[i
].iov_base
,
410 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB
*acb
,
411 const char *fmt
, ...)
416 fprintf(stderr
, "quorum: sector_num=%" PRId64
" nb_sectors=%d ",
417 acb
->sector_num
, acb
->nb_sectors
);
418 vfprintf(stderr
, fmt
, ap
);
419 fprintf(stderr
, "\n");
424 static bool quorum_compare(QuorumAIOCB
*acb
,
428 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
431 /* This driver will replace blkverify in this particular case */
432 if (s
->is_blkverify
) {
433 offset
= qemu_iovec_compare(a
, b
);
435 quorum_err(acb
, "contents mismatch in sector %" PRId64
,
437 (uint64_t)(offset
/ BDRV_SECTOR_SIZE
));
442 return quorum_iovec_compare(a
, b
);
445 /* Do a vote to get the error code */
446 static int quorum_vote_error(QuorumAIOCB
*acb
)
448 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
449 QuorumVoteVersion
*winner
= NULL
;
450 QuorumVotes error_votes
;
451 QuorumVoteValue result_value
;
455 QLIST_INIT(&error_votes
.vote_list
);
456 error_votes
.compare
= quorum_64bits_compare
;
458 for (i
= 0; i
< s
->num_children
; i
++) {
459 ret
= acb
->qcrs
[i
].ret
;
462 result_value
.l
= ret
;
463 quorum_count_vote(&error_votes
, &result_value
, i
);
468 winner
= quorum_get_vote_winner(&error_votes
);
469 ret
= winner
->value
.l
;
472 quorum_free_vote_list(&error_votes
);
477 static void quorum_vote(QuorumAIOCB
*acb
)
481 QuorumVoteValue hash
;
482 BDRVQuorumState
*s
= acb
->common
.bs
->opaque
;
483 QuorumVoteVersion
*winner
;
485 if (quorum_has_too_much_io_failed(acb
)) {
489 /* get the index of the first successful read */
490 for (i
= 0; i
< s
->num_children
; i
++) {
491 if (!acb
->qcrs
[i
].ret
) {
496 assert(i
< s
->num_children
);
498 /* compare this read with all other successful reads stopping at quorum
501 for (j
= i
+ 1; j
< s
->num_children
; j
++) {
502 if (acb
->qcrs
[j
].ret
) {
505 quorum
= quorum_compare(acb
, &acb
->qcrs
[i
].qiov
, &acb
->qcrs
[j
].qiov
);
511 /* Every successful read agrees */
513 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[i
].qiov
);
517 /* compute hashes for each successful read, also store indexes */
518 for (i
= 0; i
< s
->num_children
; i
++) {
519 if (acb
->qcrs
[i
].ret
) {
522 ret
= quorum_compute_hash(acb
, i
, &hash
);
523 /* if ever the hash computation failed */
528 quorum_count_vote(&acb
->votes
, &hash
, i
);
531 /* vote to select the most represented version */
532 winner
= quorum_get_vote_winner(&acb
->votes
);
534 /* if the winner count is smaller than threshold the read fails */
535 if (winner
->vote_count
< s
->threshold
) {
536 quorum_report_failure(acb
);
537 acb
->vote_ret
= -EIO
;
541 /* we have a winner: copy it */
542 quorum_copy_qiov(acb
->qiov
, &acb
->qcrs
[winner
->index
].qiov
);
544 /* some versions are bad print them */
545 quorum_report_bad_versions(s
, acb
, &winner
->value
);
549 quorum_free_vote_list(&acb
->votes
);
552 static BlockDriverAIOCB
*quorum_aio_readv(BlockDriverState
*bs
,
556 BlockDriverCompletionFunc
*cb
,
559 BDRVQuorumState
*s
= bs
->opaque
;
560 QuorumAIOCB
*acb
= quorum_aio_get(s
, bs
, qiov
, sector_num
,
561 nb_sectors
, cb
, opaque
);
566 for (i
= 0; i
< s
->num_children
; i
++) {
567 acb
->qcrs
[i
].buf
= qemu_blockalign(s
->bs
[i
], qiov
->size
);
568 qemu_iovec_init(&acb
->qcrs
[i
].qiov
, qiov
->niov
);
569 qemu_iovec_clone(&acb
->qcrs
[i
].qiov
, qiov
, acb
->qcrs
[i
].buf
);
572 for (i
= 0; i
< s
->num_children
; i
++) {
573 bdrv_aio_readv(s
->bs
[i
], sector_num
, &acb
->qcrs
[i
].qiov
, nb_sectors
,
574 quorum_aio_cb
, &acb
->qcrs
[i
]);
580 static BlockDriverAIOCB
*quorum_aio_writev(BlockDriverState
*bs
,
584 BlockDriverCompletionFunc
*cb
,
587 BDRVQuorumState
*s
= bs
->opaque
;
588 QuorumAIOCB
*acb
= quorum_aio_get(s
, bs
, qiov
, sector_num
, nb_sectors
,
592 for (i
= 0; i
< s
->num_children
; i
++) {
593 acb
->qcrs
[i
].aiocb
= bdrv_aio_writev(s
->bs
[i
], sector_num
, qiov
,
594 nb_sectors
, &quorum_aio_cb
,
601 static int64_t quorum_getlength(BlockDriverState
*bs
)
603 BDRVQuorumState
*s
= bs
->opaque
;
607 /* check that all file have the same length */
608 result
= bdrv_getlength(s
->bs
[0]);
612 for (i
= 1; i
< s
->num_children
; i
++) {
613 int64_t value
= bdrv_getlength(s
->bs
[i
]);
617 if (value
!= result
) {
625 static void quorum_invalidate_cache(BlockDriverState
*bs
)
627 BDRVQuorumState
*s
= bs
->opaque
;
630 for (i
= 0; i
< s
->num_children
; i
++) {
631 bdrv_invalidate_cache(s
->bs
[i
]);
635 static coroutine_fn
int quorum_co_flush(BlockDriverState
*bs
)
637 BDRVQuorumState
*s
= bs
->opaque
;
638 QuorumVoteVersion
*winner
= NULL
;
639 QuorumVotes error_votes
;
640 QuorumVoteValue result_value
;
644 QLIST_INIT(&error_votes
.vote_list
);
645 error_votes
.compare
= quorum_64bits_compare
;
647 for (i
= 0; i
< s
->num_children
; i
++) {
648 result
= bdrv_co_flush(s
->bs
[i
]);
649 result_value
.l
= result
;
650 quorum_count_vote(&error_votes
, &result_value
, i
);
653 winner
= quorum_get_vote_winner(&error_votes
);
654 result
= winner
->value
.l
;
656 quorum_free_vote_list(&error_votes
);
661 static bool quorum_recurse_is_first_non_filter(BlockDriverState
*bs
,
662 BlockDriverState
*candidate
)
664 BDRVQuorumState
*s
= bs
->opaque
;
667 for (i
= 0; i
< s
->num_children
; i
++) {
668 bool perm
= bdrv_recurse_is_first_non_filter(s
->bs
[i
],
678 static int quorum_valid_threshold(int threshold
, int num_children
, Error
**errp
)
682 error_set(errp
, QERR_INVALID_PARAMETER_VALUE
,
683 "vote-threshold", "value >= 1");
687 if (threshold
> num_children
) {
688 error_setg(errp
, "threshold may not exceed children count");
695 static QemuOptsList quorum_runtime_opts
= {
697 .head
= QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts
.head
),
700 .name
= QUORUM_OPT_VOTE_THRESHOLD
,
701 .type
= QEMU_OPT_NUMBER
,
702 .help
= "The number of vote needed for reaching quorum",
705 .name
= QUORUM_OPT_BLKVERIFY
,
706 .type
= QEMU_OPT_BOOL
,
707 .help
= "Trigger block verify mode if set",
709 { /* end of list */ }
713 static int quorum_open(BlockDriverState
*bs
, QDict
*options
, int flags
,
716 BDRVQuorumState
*s
= bs
->opaque
;
717 Error
*local_err
= NULL
;
722 const QListEntry
*lentry
;
726 qdict_flatten(options
);
727 qdict_extract_subqdict(options
, &sub
, "children.");
728 qdict_array_split(sub
, &list
);
730 if (qdict_size(sub
)) {
731 error_setg(&local_err
, "Invalid option children.%s",
732 qdict_first(sub
)->key
);
737 /* count how many different children are present */
738 s
->num_children
= qlist_size(list
);
739 if (s
->num_children
< 2) {
740 error_setg(&local_err
,
741 "Number of provided children must be greater than 1");
746 opts
= qemu_opts_create(&quorum_runtime_opts
, NULL
, 0, &error_abort
);
747 qemu_opts_absorb_qdict(opts
, options
, &local_err
);
748 if (error_is_set(&local_err
)) {
753 s
->threshold
= qemu_opt_get_number(opts
, QUORUM_OPT_VOTE_THRESHOLD
, 0);
755 /* and validate it against s->num_children */
756 ret
= quorum_valid_threshold(s
->threshold
, s
->num_children
, &local_err
);
761 /* is the driver in blkverify mode */
762 if (qemu_opt_get_bool(opts
, QUORUM_OPT_BLKVERIFY
, false) &&
763 s
->num_children
== 2 && s
->threshold
== 2) {
764 s
->is_blkverify
= true;
765 } else if (qemu_opt_get_bool(opts
, QUORUM_OPT_BLKVERIFY
, false)) {
766 fprintf(stderr
, "blkverify mode is set by setting blkverify=on "
767 "and using two files with vote_threshold=2\n");
770 /* allocate the children BlockDriverState array */
771 s
->bs
= g_new0(BlockDriverState
*, s
->num_children
);
772 opened
= g_new0(bool, s
->num_children
);
774 for (i
= 0, lentry
= qlist_first(list
); lentry
;
775 lentry
= qlist_next(lentry
), i
++) {
779 switch (qobject_type(lentry
->value
))
781 /* List of options */
783 d
= qobject_to_qdict(lentry
->value
);
785 ret
= bdrv_open(&s
->bs
[i
], NULL
, NULL
, d
, flags
, NULL
,
791 string
= qobject_to_qstring(lentry
->value
);
792 ret
= bdrv_open(&s
->bs
[i
], NULL
, qstring_get_str(string
), NULL
,
793 flags
, NULL
, &local_err
);
797 error_setg(&local_err
, "Specification of child block device %i "
812 /* cleanup on error */
813 for (i
= 0; i
< s
->num_children
; i
++) {
817 bdrv_unref(s
->bs
[i
]);
822 /* propagate error */
823 if (error_is_set(&local_err
)) {
824 error_propagate(errp
, local_err
);
831 static void quorum_close(BlockDriverState
*bs
)
833 BDRVQuorumState
*s
= bs
->opaque
;
836 for (i
= 0; i
< s
->num_children
; i
++) {
837 bdrv_unref(s
->bs
[i
]);
843 static BlockDriver bdrv_quorum
= {
844 .format_name
= "quorum",
845 .protocol_name
= "quorum",
847 .instance_size
= sizeof(BDRVQuorumState
),
849 .bdrv_file_open
= quorum_open
,
850 .bdrv_close
= quorum_close
,
852 .authorizations
= { true, true },
854 .bdrv_co_flush_to_disk
= quorum_co_flush
,
856 .bdrv_getlength
= quorum_getlength
,
858 .bdrv_aio_readv
= quorum_aio_readv
,
859 .bdrv_aio_writev
= quorum_aio_writev
,
860 .bdrv_invalidate_cache
= quorum_invalidate_cache
,
862 .bdrv_recurse_is_first_non_filter
= quorum_recurse_is_first_non_filter
,
865 static void bdrv_quorum_init(void)
867 bdrv_register(&bdrv_quorum
);
870 block_init(bdrv_quorum_init
);