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/qint.h"
22 #include "qapi/qmp/qjson.h"
23 #include "qapi/qmp/qlist.h"
24 #include "qapi/qmp/qstring.h"
25 #include "qapi-event.h"
27 #define HASH_LENGTH 32
29 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
30 #define QUORUM_OPT_BLKVERIFY "blkverify"
31 #define QUORUM_OPT_REWRITE "rewrite-corrupted"
32 #define QUORUM_OPT_READ_PATTERN "read-pattern"
34 /* This union holds a vote hash value */
35 typedef union QuorumVoteValue {
36 char h[HASH_LENGTH]; /* SHA-256 hash */
37 int64_t l; /* simpler 64 bits hash */
41 typedef struct QuorumVoteItem {
43 QLIST_ENTRY(QuorumVoteItem) next;
46 /* this structure is a vote version. A version is the set of votes sharing the
48 * The set of votes will be tracked with the items field and its cardinality is
51 typedef struct QuorumVoteVersion {
52 QuorumVoteValue value;
55 QLIST_HEAD(, QuorumVoteItem) items;
56 QLIST_ENTRY(QuorumVoteVersion) next;
59 /* this structure holds a group of vote versions together */
60 typedef struct QuorumVotes {
61 QLIST_HEAD(, QuorumVoteVersion) vote_list;
62 bool (*compare)(QuorumVoteValue *a, QuorumVoteValue *b);
65 /* the following structure holds the state of one quorum instance */
66 typedef struct BDRVQuorumState {
67 BlockDriverState **bs; /* children BlockDriverStates */
68 int num_children; /* children count */
69 int threshold; /* if less than threshold children reads gave the
70 * same result a quorum error occurs.
72 bool is_blkverify; /* true if the driver is in blkverify mode
73 * Writes are mirrored on two children devices.
74 * On reads the two children devices' contents are
75 * compared and if a difference is spotted its
76 * location is printed and the code aborts.
77 * It is useful to debug other block drivers by
78 * comparing them with a reference one.
80 bool rewrite_corrupted;/* true if the driver must rewrite-on-read corrupted
81 * block if Quorum is reached.
84 QuorumReadPattern read_pattern;
87 typedef struct QuorumAIOCB QuorumAIOCB;
89 /* Quorum will create one instance of the following structure per operation it
90 * performs on its children.
91 * So for each read/write operation coming from the upper layer there will be
92 * $children_count QuorumChildRequest.
94 typedef struct QuorumChildRequest {
100 } QuorumChildRequest;
102 /* Quorum will use the following structure to track progress of each read/write
103 * operation received by the upper layer.
104 * This structure hold pointers to the QuorumChildRequest structures instances
105 * used to do operations on each children and track overall progress.
110 /* Request metadata */
114 QEMUIOVector *qiov; /* calling IOV */
116 QuorumChildRequest *qcrs; /* individual child requests */
117 int count; /* number of completed AIOCB */
118 int success_count; /* number of successfully completed AIOCB */
120 int rewrite_count; /* number of replica to rewrite: count down to
121 * zero once writes are fired
128 int child_iter; /* which child to read in fifo pattern */
131 static bool quorum_vote(QuorumAIOCB *acb);
133 static void quorum_aio_cancel(BlockAIOCB *blockacb)
135 QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
136 BDRVQuorumState *s = acb->common.bs->opaque;
139 /* cancel all callbacks */
140 for (i = 0; i < s->num_children; i++) {
141 if (acb->qcrs[i].aiocb) {
142 bdrv_aio_cancel_async(acb->qcrs[i].aiocb);
147 static AIOCBInfo quorum_aiocb_info = {
148 .aiocb_size = sizeof(QuorumAIOCB),
149 .cancel_async = quorum_aio_cancel,
152 static void quorum_aio_finalize(QuorumAIOCB *acb)
160 acb->common.cb(acb->common.opaque, ret);
163 /* on the quorum case acb->child_iter == s->num_children - 1 */
164 for (i = 0; i <= acb->child_iter; i++) {
165 qemu_vfree(acb->qcrs[i].buf);
166 qemu_iovec_destroy(&acb->qcrs[i].qiov);
174 static bool quorum_sha256_compare(QuorumVoteValue *a, QuorumVoteValue *b)
176 return !memcmp(a->h, b->h, HASH_LENGTH);
179 static bool quorum_64bits_compare(QuorumVoteValue *a, QuorumVoteValue *b)
184 static QuorumAIOCB *quorum_aio_get(BDRVQuorumState *s,
185 BlockDriverState *bs,
189 BlockCompletionFunc *cb,
192 QuorumAIOCB *acb = qemu_aio_get(&quorum_aiocb_info, bs, cb, opaque);
195 acb->common.bs->opaque = s;
196 acb->sector_num = sector_num;
197 acb->nb_sectors = nb_sectors;
199 acb->qcrs = g_new0(QuorumChildRequest, s->num_children);
201 acb->success_count = 0;
202 acb->rewrite_count = 0;
203 acb->votes.compare = quorum_sha256_compare;
204 QLIST_INIT(&acb->votes.vote_list);
205 acb->is_read = false;
208 for (i = 0; i < s->num_children; i++) {
209 acb->qcrs[i].buf = NULL;
210 acb->qcrs[i].ret = 0;
211 acb->qcrs[i].parent = acb;
217 static void quorum_report_bad(QuorumAIOCB *acb, char *node_name, int ret)
219 const char *msg = NULL;
221 msg = strerror(-ret);
223 qapi_event_send_quorum_report_bad(!!msg, msg, node_name,
224 acb->sector_num, acb->nb_sectors, &error_abort);
227 static void quorum_report_failure(QuorumAIOCB *acb)
229 const char *reference = bdrv_get_device_name(acb->common.bs)[0] ?
230 bdrv_get_device_name(acb->common.bs) :
231 acb->common.bs->node_name;
233 qapi_event_send_quorum_failure(reference, acb->sector_num,
234 acb->nb_sectors, &error_abort);
237 static int quorum_vote_error(QuorumAIOCB *acb);
239 static bool quorum_has_too_much_io_failed(QuorumAIOCB *acb)
241 BDRVQuorumState *s = acb->common.bs->opaque;
243 if (acb->success_count < s->threshold) {
244 acb->vote_ret = quorum_vote_error(acb);
245 quorum_report_failure(acb);
252 static void quorum_rewrite_aio_cb(void *opaque, int ret)
254 QuorumAIOCB *acb = opaque;
256 /* one less rewrite to do */
257 acb->rewrite_count--;
259 /* wait until all rewrite callbacks have completed */
260 if (acb->rewrite_count) {
264 quorum_aio_finalize(acb);
267 static BlockAIOCB *read_fifo_child(QuorumAIOCB *acb);
269 static void quorum_copy_qiov(QEMUIOVector *dest, QEMUIOVector *source)
272 assert(dest->niov == source->niov);
273 assert(dest->size == source->size);
274 for (i = 0; i < source->niov; i++) {
275 assert(dest->iov[i].iov_len == source->iov[i].iov_len);
276 memcpy(dest->iov[i].iov_base,
277 source->iov[i].iov_base,
278 source->iov[i].iov_len);
282 static void quorum_aio_cb(void *opaque, int ret)
284 QuorumChildRequest *sacb = opaque;
285 QuorumAIOCB *acb = sacb->parent;
286 BDRVQuorumState *s = acb->common.bs->opaque;
287 bool rewrite = false;
289 if (acb->is_read && s->read_pattern == QUORUM_READ_PATTERN_FIFO) {
290 /* We try to read next child in FIFO order if we fail to read */
291 if (ret < 0 && ++acb->child_iter < s->num_children) {
292 read_fifo_child(acb);
297 quorum_copy_qiov(acb->qiov, &acb->qcrs[acb->child_iter].qiov);
300 quorum_aio_finalize(acb);
307 acb->success_count++;
309 quorum_report_bad(acb, sacb->aiocb->bs->node_name, ret);
311 assert(acb->count <= s->num_children);
312 assert(acb->success_count <= s->num_children);
313 if (acb->count < s->num_children) {
317 /* Do the vote on read */
319 rewrite = quorum_vote(acb);
321 quorum_has_too_much_io_failed(acb);
324 /* if no rewrite is done the code will finish right away */
326 quorum_aio_finalize(acb);
330 static void quorum_report_bad_versions(BDRVQuorumState *s,
332 QuorumVoteValue *value)
334 QuorumVoteVersion *version;
335 QuorumVoteItem *item;
337 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
338 if (acb->votes.compare(&version->value, value)) {
341 QLIST_FOREACH(item, &version->items, next) {
342 quorum_report_bad(acb, s->bs[item->index]->node_name, 0);
347 static bool quorum_rewrite_bad_versions(BDRVQuorumState *s, QuorumAIOCB *acb,
348 QuorumVoteValue *value)
350 QuorumVoteVersion *version;
351 QuorumVoteItem *item;
354 /* first count the number of bad versions: done first to avoid concurrency
357 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
358 if (acb->votes.compare(&version->value, value)) {
361 QLIST_FOREACH(item, &version->items, next) {
366 /* quorum_rewrite_aio_cb will count down this to zero */
367 acb->rewrite_count = count;
369 /* now fire the correcting rewrites */
370 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
371 if (acb->votes.compare(&version->value, value)) {
374 QLIST_FOREACH(item, &version->items, next) {
375 bdrv_aio_writev(s->bs[item->index], acb->sector_num, acb->qiov,
376 acb->nb_sectors, quorum_rewrite_aio_cb, acb);
380 /* return true if any rewrite is done else false */
384 static void quorum_count_vote(QuorumVotes *votes,
385 QuorumVoteValue *value,
388 QuorumVoteVersion *v = NULL, *version = NULL;
389 QuorumVoteItem *item;
391 /* look if we have something with this hash */
392 QLIST_FOREACH(v, &votes->vote_list, next) {
393 if (votes->compare(&v->value, value)) {
399 /* It's a version not yet in the list add it */
401 version = g_new0(QuorumVoteVersion, 1);
402 QLIST_INIT(&version->items);
403 memcpy(&version->value, value, sizeof(version->value));
404 version->index = index;
405 version->vote_count = 0;
406 QLIST_INSERT_HEAD(&votes->vote_list, version, next);
409 version->vote_count++;
411 item = g_new0(QuorumVoteItem, 1);
413 QLIST_INSERT_HEAD(&version->items, item, next);
416 static void quorum_free_vote_list(QuorumVotes *votes)
418 QuorumVoteVersion *version, *next_version;
419 QuorumVoteItem *item, *next_item;
421 QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
422 QLIST_REMOVE(version, next);
423 QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
424 QLIST_REMOVE(item, next);
431 static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
434 gnutls_hash_hd_t dig;
435 QEMUIOVector *qiov = &acb->qcrs[i].qiov;
437 ret = gnutls_hash_init(&dig, GNUTLS_DIG_SHA256);
443 for (j = 0; j < qiov->niov; j++) {
444 ret = gnutls_hash(dig, qiov->iov[j].iov_base, qiov->iov[j].iov_len);
450 gnutls_hash_deinit(dig, (void *) hash);
454 static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
457 QuorumVoteVersion *candidate, *winner = NULL;
459 QLIST_FOREACH(candidate, &votes->vote_list, next) {
460 if (candidate->vote_count > max) {
461 max = candidate->vote_count;
469 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
470 * differing byte location. Yet it is handcoded to compare vectors one byte
471 * after another so it does not benefit from the libc SIMD optimizations.
472 * quorum_iovec_compare is written for speed and should be used in the non
473 * blkverify mode of quorum.
475 static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
480 assert(a->niov == b->niov);
481 for (i = 0; i < a->niov; i++) {
482 assert(a->iov[i].iov_len == b->iov[i].iov_len);
483 result = memcmp(a->iov[i].iov_base,
494 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB *acb,
495 const char *fmt, ...)
500 fprintf(stderr, "quorum: sector_num=%" PRId64 " nb_sectors=%d ",
501 acb->sector_num, acb->nb_sectors);
502 vfprintf(stderr, fmt, ap);
503 fprintf(stderr, "\n");
508 static bool quorum_compare(QuorumAIOCB *acb,
512 BDRVQuorumState *s = acb->common.bs->opaque;
515 /* This driver will replace blkverify in this particular case */
516 if (s->is_blkverify) {
517 offset = qemu_iovec_compare(a, b);
519 quorum_err(acb, "contents mismatch in sector %" PRId64,
521 (uint64_t)(offset / BDRV_SECTOR_SIZE));
526 return quorum_iovec_compare(a, b);
529 /* Do a vote to get the error code */
530 static int quorum_vote_error(QuorumAIOCB *acb)
532 BDRVQuorumState *s = acb->common.bs->opaque;
533 QuorumVoteVersion *winner = NULL;
534 QuorumVotes error_votes;
535 QuorumVoteValue result_value;
539 QLIST_INIT(&error_votes.vote_list);
540 error_votes.compare = quorum_64bits_compare;
542 for (i = 0; i < s->num_children; i++) {
543 ret = acb->qcrs[i].ret;
546 result_value.l = ret;
547 quorum_count_vote(&error_votes, &result_value, i);
552 winner = quorum_get_vote_winner(&error_votes);
553 ret = winner->value.l;
556 quorum_free_vote_list(&error_votes);
561 static bool quorum_vote(QuorumAIOCB *acb)
564 bool rewrite = false;
566 QuorumVoteValue hash;
567 BDRVQuorumState *s = acb->common.bs->opaque;
568 QuorumVoteVersion *winner;
570 if (quorum_has_too_much_io_failed(acb)) {
574 /* get the index of the first successful read */
575 for (i = 0; i < s->num_children; i++) {
576 if (!acb->qcrs[i].ret) {
581 assert(i < s->num_children);
583 /* compare this read with all other successful reads stopping at quorum
586 for (j = i + 1; j < s->num_children; j++) {
587 if (acb->qcrs[j].ret) {
590 quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
596 /* Every successful read agrees */
598 quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
602 /* compute hashes for each successful read, also store indexes */
603 for (i = 0; i < s->num_children; i++) {
604 if (acb->qcrs[i].ret) {
607 ret = quorum_compute_hash(acb, i, &hash);
608 /* if ever the hash computation failed */
613 quorum_count_vote(&acb->votes, &hash, i);
616 /* vote to select the most represented version */
617 winner = quorum_get_vote_winner(&acb->votes);
619 /* if the winner count is smaller than threshold the read fails */
620 if (winner->vote_count < s->threshold) {
621 quorum_report_failure(acb);
622 acb->vote_ret = -EIO;
626 /* we have a winner: copy it */
627 quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
629 /* some versions are bad print them */
630 quorum_report_bad_versions(s, acb, &winner->value);
632 /* corruption correction is enabled */
633 if (s->rewrite_corrupted) {
634 rewrite = quorum_rewrite_bad_versions(s, acb, &winner->value);
639 quorum_free_vote_list(&acb->votes);
643 static BlockAIOCB *read_quorum_children(QuorumAIOCB *acb)
645 BDRVQuorumState *s = acb->common.bs->opaque;
648 for (i = 0; i < s->num_children; i++) {
649 acb->qcrs[i].buf = qemu_blockalign(s->bs[i], acb->qiov->size);
650 qemu_iovec_init(&acb->qcrs[i].qiov, acb->qiov->niov);
651 qemu_iovec_clone(&acb->qcrs[i].qiov, acb->qiov, acb->qcrs[i].buf);
654 for (i = 0; i < s->num_children; i++) {
655 bdrv_aio_readv(s->bs[i], acb->sector_num, &acb->qcrs[i].qiov,
656 acb->nb_sectors, quorum_aio_cb, &acb->qcrs[i]);
662 static BlockAIOCB *read_fifo_child(QuorumAIOCB *acb)
664 BDRVQuorumState *s = acb->common.bs->opaque;
666 acb->qcrs[acb->child_iter].buf = qemu_blockalign(s->bs[acb->child_iter],
668 qemu_iovec_init(&acb->qcrs[acb->child_iter].qiov, acb->qiov->niov);
669 qemu_iovec_clone(&acb->qcrs[acb->child_iter].qiov, acb->qiov,
670 acb->qcrs[acb->child_iter].buf);
671 bdrv_aio_readv(s->bs[acb->child_iter], acb->sector_num,
672 &acb->qcrs[acb->child_iter].qiov, acb->nb_sectors,
673 quorum_aio_cb, &acb->qcrs[acb->child_iter]);
678 static BlockAIOCB *quorum_aio_readv(BlockDriverState *bs,
682 BlockCompletionFunc *cb,
685 BDRVQuorumState *s = bs->opaque;
686 QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num,
687 nb_sectors, cb, opaque);
690 if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
691 acb->child_iter = s->num_children - 1;
692 return read_quorum_children(acb);
696 return read_fifo_child(acb);
699 static BlockAIOCB *quorum_aio_writev(BlockDriverState *bs,
703 BlockCompletionFunc *cb,
706 BDRVQuorumState *s = bs->opaque;
707 QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num, nb_sectors,
711 for (i = 0; i < s->num_children; i++) {
712 acb->qcrs[i].aiocb = bdrv_aio_writev(s->bs[i], sector_num, qiov,
713 nb_sectors, &quorum_aio_cb,
720 static int64_t quorum_getlength(BlockDriverState *bs)
722 BDRVQuorumState *s = bs->opaque;
726 /* check that all file have the same length */
727 result = bdrv_getlength(s->bs[0]);
731 for (i = 1; i < s->num_children; i++) {
732 int64_t value = bdrv_getlength(s->bs[i]);
736 if (value != result) {
744 static void quorum_invalidate_cache(BlockDriverState *bs, Error **errp)
746 BDRVQuorumState *s = bs->opaque;
747 Error *local_err = NULL;
750 for (i = 0; i < s->num_children; i++) {
751 bdrv_invalidate_cache(s->bs[i], &local_err);
753 error_propagate(errp, local_err);
759 static coroutine_fn int quorum_co_flush(BlockDriverState *bs)
761 BDRVQuorumState *s = bs->opaque;
762 QuorumVoteVersion *winner = NULL;
763 QuorumVotes error_votes;
764 QuorumVoteValue result_value;
768 QLIST_INIT(&error_votes.vote_list);
769 error_votes.compare = quorum_64bits_compare;
771 for (i = 0; i < s->num_children; i++) {
772 result = bdrv_co_flush(s->bs[i]);
773 result_value.l = result;
774 quorum_count_vote(&error_votes, &result_value, i);
777 winner = quorum_get_vote_winner(&error_votes);
778 result = winner->value.l;
780 quorum_free_vote_list(&error_votes);
785 static bool quorum_recurse_is_first_non_filter(BlockDriverState *bs,
786 BlockDriverState *candidate)
788 BDRVQuorumState *s = bs->opaque;
791 for (i = 0; i < s->num_children; i++) {
792 bool perm = bdrv_recurse_is_first_non_filter(s->bs[i],
802 static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
806 error_set(errp, QERR_INVALID_PARAMETER_VALUE,
807 "vote-threshold", "value >= 1");
811 if (threshold > num_children) {
812 error_setg(errp, "threshold may not exceed children count");
819 static QemuOptsList quorum_runtime_opts = {
821 .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
824 .name = QUORUM_OPT_VOTE_THRESHOLD,
825 .type = QEMU_OPT_NUMBER,
826 .help = "The number of vote needed for reaching quorum",
829 .name = QUORUM_OPT_BLKVERIFY,
830 .type = QEMU_OPT_BOOL,
831 .help = "Trigger block verify mode if set",
834 .name = QUORUM_OPT_REWRITE,
835 .type = QEMU_OPT_BOOL,
836 .help = "Rewrite corrupted block on read quorum",
839 .name = QUORUM_OPT_READ_PATTERN,
840 .type = QEMU_OPT_STRING,
841 .help = "Allowed pattern: quorum, fifo. Quorum is default",
843 { /* end of list */ }
847 static int parse_read_pattern(const char *opt)
852 /* Set quorum as default */
853 return QUORUM_READ_PATTERN_QUORUM;
856 for (i = 0; i < QUORUM_READ_PATTERN_MAX; i++) {
857 if (!strcmp(opt, QuorumReadPattern_lookup[i])) {
865 static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
868 BDRVQuorumState *s = bs->opaque;
869 Error *local_err = NULL;
870 QemuOpts *opts = NULL;
874 const QListEntry *lentry;
878 qdict_flatten(options);
879 qdict_extract_subqdict(options, &sub, "children.");
880 qdict_array_split(sub, &list);
882 if (qdict_size(sub)) {
883 error_setg(&local_err, "Invalid option children.%s",
884 qdict_first(sub)->key);
889 /* count how many different children are present */
890 s->num_children = qlist_size(list);
891 if (s->num_children < 2) {
892 error_setg(&local_err,
893 "Number of provided children must be greater than 1");
898 opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
899 qemu_opts_absorb_qdict(opts, options, &local_err);
905 s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
906 ret = parse_read_pattern(qemu_opt_get(opts, QUORUM_OPT_READ_PATTERN));
908 error_setg(&local_err, "Please set read-pattern as fifo or quorum");
911 s->read_pattern = ret;
913 if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
914 /* and validate it against s->num_children */
915 ret = quorum_valid_threshold(s->threshold, s->num_children, &local_err);
920 /* is the driver in blkverify mode */
921 if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false) &&
922 s->num_children == 2 && s->threshold == 2) {
923 s->is_blkverify = true;
924 } else if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false)) {
925 fprintf(stderr, "blkverify mode is set by setting blkverify=on "
926 "and using two files with vote_threshold=2\n");
929 s->rewrite_corrupted = qemu_opt_get_bool(opts, QUORUM_OPT_REWRITE,
931 if (s->rewrite_corrupted && s->is_blkverify) {
932 error_setg(&local_err,
933 "rewrite-corrupted=on cannot be used with blkverify=on");
939 /* allocate the children BlockDriverState array */
940 s->bs = g_new0(BlockDriverState *, s->num_children);
941 opened = g_new0(bool, s->num_children);
943 for (i = 0, lentry = qlist_first(list); lentry;
944 lentry = qlist_next(lentry), i++) {
948 switch (qobject_type(lentry->value))
950 /* List of options */
952 d = qobject_to_qdict(lentry->value);
954 ret = bdrv_open(&s->bs[i], NULL, NULL, d, flags, NULL,
960 string = qobject_to_qstring(lentry->value);
961 ret = bdrv_open(&s->bs[i], NULL, qstring_get_str(string), NULL,
962 flags, NULL, &local_err);
966 error_setg(&local_err, "Specification of child block device %i "
981 /* cleanup on error */
982 for (i = 0; i < s->num_children; i++) {
986 bdrv_unref(s->bs[i]);
992 /* propagate error */
994 error_propagate(errp, local_err);
1001 static void quorum_close(BlockDriverState *bs)
1003 BDRVQuorumState *s = bs->opaque;
1006 for (i = 0; i < s->num_children; i++) {
1007 bdrv_unref(s->bs[i]);
1013 static void quorum_detach_aio_context(BlockDriverState *bs)
1015 BDRVQuorumState *s = bs->opaque;
1018 for (i = 0; i < s->num_children; i++) {
1019 bdrv_detach_aio_context(s->bs[i]);
1023 static void quorum_attach_aio_context(BlockDriverState *bs,
1024 AioContext *new_context)
1026 BDRVQuorumState *s = bs->opaque;
1029 for (i = 0; i < s->num_children; i++) {
1030 bdrv_attach_aio_context(s->bs[i], new_context);
1034 static void quorum_refresh_filename(BlockDriverState *bs)
1036 BDRVQuorumState *s = bs->opaque;
1041 for (i = 0; i < s->num_children; i++) {
1042 bdrv_refresh_filename(s->bs[i]);
1043 if (!s->bs[i]->full_open_options) {
1048 children = qlist_new();
1049 for (i = 0; i < s->num_children; i++) {
1050 QINCREF(s->bs[i]->full_open_options);
1051 qlist_append_obj(children, QOBJECT(s->bs[i]->full_open_options));
1055 qdict_put_obj(opts, "driver", QOBJECT(qstring_from_str("quorum")));
1056 qdict_put_obj(opts, QUORUM_OPT_VOTE_THRESHOLD,
1057 QOBJECT(qint_from_int(s->threshold)));
1058 qdict_put_obj(opts, QUORUM_OPT_BLKVERIFY,
1059 QOBJECT(qbool_from_int(s->is_blkverify)));
1060 qdict_put_obj(opts, QUORUM_OPT_REWRITE,
1061 QOBJECT(qbool_from_int(s->rewrite_corrupted)));
1062 qdict_put_obj(opts, "children", QOBJECT(children));
1064 bs->full_open_options = opts;
1067 static BlockDriver bdrv_quorum = {
1068 .format_name = "quorum",
1069 .protocol_name = "quorum",
1071 .instance_size = sizeof(BDRVQuorumState),
1073 .bdrv_file_open = quorum_open,
1074 .bdrv_close = quorum_close,
1075 .bdrv_refresh_filename = quorum_refresh_filename,
1077 .bdrv_co_flush_to_disk = quorum_co_flush,
1079 .bdrv_getlength = quorum_getlength,
1081 .bdrv_aio_readv = quorum_aio_readv,
1082 .bdrv_aio_writev = quorum_aio_writev,
1083 .bdrv_invalidate_cache = quorum_invalidate_cache,
1085 .bdrv_detach_aio_context = quorum_detach_aio_context,
1086 .bdrv_attach_aio_context = quorum_attach_aio_context,
1089 .bdrv_recurse_is_first_non_filter = quorum_recurse_is_first_non_filter,
1092 static void bdrv_quorum_init(void)
1094 bdrv_register(&bdrv_quorum);
1097 block_init(bdrv_quorum_init);