block: Add error handling to bdrv_invalidate_cache()
[qemu.git] / block / quorum.c
1 /*
2 * Quorum Block filter
3 *
4 * Copyright (C) 2012-2014 Nodalink, EURL.
5 *
6 * Author:
7 * BenoƮt Canet <benoit.canet@irqsave.net>
8 *
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).
11 *
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.
14 */
15
16 #include <gnutls/gnutls.h>
17 #include <gnutls/crypto.h>
18 #include "block/block_int.h"
19 #include "qapi/qmp/qjson.h"
20
21 #define HASH_LENGTH 32
22
23 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
24 #define QUORUM_OPT_BLKVERIFY "blkverify"
25
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 */
30 } QuorumVoteValue;
31
32 /* A vote item */
33 typedef struct QuorumVoteItem {
34 int index;
35 QLIST_ENTRY(QuorumVoteItem) next;
36 } QuorumVoteItem;
37
38 /* this structure is a vote version. A version is the set of votes sharing the
39 * same vote value.
40 * The set of votes will be tracked with the items field and its cardinality is
41 * vote_count.
42 */
43 typedef struct QuorumVoteVersion {
44 QuorumVoteValue value;
45 int index;
46 int vote_count;
47 QLIST_HEAD(, QuorumVoteItem) items;
48 QLIST_ENTRY(QuorumVoteVersion) next;
49 } QuorumVoteVersion;
50
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);
55 } QuorumVotes;
56
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.
63 */
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.
71 */
72 } BDRVQuorumState;
73
74 typedef struct QuorumAIOCB QuorumAIOCB;
75
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.
80 */
81 typedef struct QuorumChildRequest {
82 BlockDriverAIOCB *aiocb;
83 QEMUIOVector qiov;
84 uint8_t *buf;
85 int ret;
86 QuorumAIOCB *parent;
87 } QuorumChildRequest;
88
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.
93 */
94 struct QuorumAIOCB {
95 BlockDriverAIOCB common;
96
97 /* Request metadata */
98 uint64_t sector_num;
99 int nb_sectors;
100
101 QEMUIOVector *qiov; /* calling IOV */
102
103 QuorumChildRequest *qcrs; /* individual child requests */
104 int count; /* number of completed AIOCB */
105 int success_count; /* number of successfully completed AIOCB */
106
107 QuorumVotes votes;
108
109 bool is_read;
110 int vote_ret;
111 };
112
113 static void quorum_vote(QuorumAIOCB *acb);
114
115 static void quorum_aio_cancel(BlockDriverAIOCB *blockacb)
116 {
117 QuorumAIOCB *acb = container_of(blockacb, QuorumAIOCB, common);
118 BDRVQuorumState *s = acb->common.bs->opaque;
119 int i;
120
121 /* cancel all callbacks */
122 for (i = 0; i < s->num_children; i++) {
123 bdrv_aio_cancel(acb->qcrs[i].aiocb);
124 }
125
126 g_free(acb->qcrs);
127 qemu_aio_release(acb);
128 }
129
130 static AIOCBInfo quorum_aiocb_info = {
131 .aiocb_size = sizeof(QuorumAIOCB),
132 .cancel = quorum_aio_cancel,
133 };
134
135 static void quorum_aio_finalize(QuorumAIOCB *acb)
136 {
137 BDRVQuorumState *s = acb->common.bs->opaque;
138 int i, ret = 0;
139
140 if (acb->vote_ret) {
141 ret = acb->vote_ret;
142 }
143
144 acb->common.cb(acb->common.opaque, ret);
145
146 if (acb->is_read) {
147 for (i = 0; i < s->num_children; i++) {
148 qemu_vfree(acb->qcrs[i].buf);
149 qemu_iovec_destroy(&acb->qcrs[i].qiov);
150 }
151 }
152
153 g_free(acb->qcrs);
154 qemu_aio_release(acb);
155 }
156
157 static bool quorum_sha256_compare(QuorumVoteValue *a, QuorumVoteValue *b)
158 {
159 return !memcmp(a->h, b->h, HASH_LENGTH);
160 }
161
162 static bool quorum_64bits_compare(QuorumVoteValue *a, QuorumVoteValue *b)
163 {
164 return a->l == b->l;
165 }
166
167 static QuorumAIOCB *quorum_aio_get(BDRVQuorumState *s,
168 BlockDriverState *bs,
169 QEMUIOVector *qiov,
170 uint64_t sector_num,
171 int nb_sectors,
172 BlockDriverCompletionFunc *cb,
173 void *opaque)
174 {
175 QuorumAIOCB *acb = qemu_aio_get(&quorum_aiocb_info, bs, cb, opaque);
176 int i;
177
178 acb->common.bs->opaque = s;
179 acb->sector_num = sector_num;
180 acb->nb_sectors = nb_sectors;
181 acb->qiov = qiov;
182 acb->qcrs = g_new0(QuorumChildRequest, s->num_children);
183 acb->count = 0;
184 acb->success_count = 0;
185 acb->votes.compare = quorum_sha256_compare;
186 QLIST_INIT(&acb->votes.vote_list);
187 acb->is_read = false;
188 acb->vote_ret = 0;
189
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;
194 }
195
196 return acb;
197 }
198
199 static void quorum_report_bad(QuorumAIOCB *acb, char *node_name, int ret)
200 {
201 QObject *data;
202 assert(node_name);
203 data = qobject_from_jsonf("{ 'node-name': %s"
204 ", 'sector-num': %" PRId64
205 ", 'sectors-count': %d }",
206 node_name, acb->sector_num, acb->nb_sectors);
207 if (ret < 0) {
208 QDict *dict = qobject_to_qdict(data);
209 qdict_put(dict, "error", qstring_from_str(strerror(-ret)));
210 }
211 monitor_protocol_event(QEVENT_QUORUM_REPORT_BAD, data);
212 qobject_decref(data);
213 }
214
215 static void quorum_report_failure(QuorumAIOCB *acb)
216 {
217 QObject *data;
218 const char *reference = acb->common.bs->device_name[0] ?
219 acb->common.bs->device_name :
220 acb->common.bs->node_name;
221 data = qobject_from_jsonf("{ 'reference': %s"
222 ", 'sector-num': %" PRId64
223 ", 'sectors-count': %d }",
224 reference, acb->sector_num, acb->nb_sectors);
225 monitor_protocol_event(QEVENT_QUORUM_FAILURE, data);
226 qobject_decref(data);
227 }
228
229 static int quorum_vote_error(QuorumAIOCB *acb);
230
231 static bool quorum_has_too_much_io_failed(QuorumAIOCB *acb)
232 {
233 BDRVQuorumState *s = acb->common.bs->opaque;
234
235 if (acb->success_count < s->threshold) {
236 acb->vote_ret = quorum_vote_error(acb);
237 quorum_report_failure(acb);
238 return true;
239 }
240
241 return false;
242 }
243
244 static void quorum_aio_cb(void *opaque, int ret)
245 {
246 QuorumChildRequest *sacb = opaque;
247 QuorumAIOCB *acb = sacb->parent;
248 BDRVQuorumState *s = acb->common.bs->opaque;
249
250 sacb->ret = ret;
251 acb->count++;
252 if (ret == 0) {
253 acb->success_count++;
254 } else {
255 quorum_report_bad(acb, sacb->aiocb->bs->node_name, ret);
256 }
257 assert(acb->count <= s->num_children);
258 assert(acb->success_count <= s->num_children);
259 if (acb->count < s->num_children) {
260 return;
261 }
262
263 /* Do the vote on read */
264 if (acb->is_read) {
265 quorum_vote(acb);
266 } else {
267 quorum_has_too_much_io_failed(acb);
268 }
269
270 quorum_aio_finalize(acb);
271 }
272
273 static void quorum_report_bad_versions(BDRVQuorumState *s,
274 QuorumAIOCB *acb,
275 QuorumVoteValue *value)
276 {
277 QuorumVoteVersion *version;
278 QuorumVoteItem *item;
279
280 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
281 if (acb->votes.compare(&version->value, value)) {
282 continue;
283 }
284 QLIST_FOREACH(item, &version->items, next) {
285 quorum_report_bad(acb, s->bs[item->index]->node_name, 0);
286 }
287 }
288 }
289
290 static void quorum_copy_qiov(QEMUIOVector *dest, QEMUIOVector *source)
291 {
292 int i;
293 assert(dest->niov == source->niov);
294 assert(dest->size == source->size);
295 for (i = 0; i < source->niov; i++) {
296 assert(dest->iov[i].iov_len == source->iov[i].iov_len);
297 memcpy(dest->iov[i].iov_base,
298 source->iov[i].iov_base,
299 source->iov[i].iov_len);
300 }
301 }
302
303 static void quorum_count_vote(QuorumVotes *votes,
304 QuorumVoteValue *value,
305 int index)
306 {
307 QuorumVoteVersion *v = NULL, *version = NULL;
308 QuorumVoteItem *item;
309
310 /* look if we have something with this hash */
311 QLIST_FOREACH(v, &votes->vote_list, next) {
312 if (votes->compare(&v->value, value)) {
313 version = v;
314 break;
315 }
316 }
317
318 /* It's a version not yet in the list add it */
319 if (!version) {
320 version = g_new0(QuorumVoteVersion, 1);
321 QLIST_INIT(&version->items);
322 memcpy(&version->value, value, sizeof(version->value));
323 version->index = index;
324 version->vote_count = 0;
325 QLIST_INSERT_HEAD(&votes->vote_list, version, next);
326 }
327
328 version->vote_count++;
329
330 item = g_new0(QuorumVoteItem, 1);
331 item->index = index;
332 QLIST_INSERT_HEAD(&version->items, item, next);
333 }
334
335 static void quorum_free_vote_list(QuorumVotes *votes)
336 {
337 QuorumVoteVersion *version, *next_version;
338 QuorumVoteItem *item, *next_item;
339
340 QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
341 QLIST_REMOVE(version, next);
342 QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
343 QLIST_REMOVE(item, next);
344 g_free(item);
345 }
346 g_free(version);
347 }
348 }
349
350 static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
351 {
352 int j, ret;
353 gnutls_hash_hd_t dig;
354 QEMUIOVector *qiov = &acb->qcrs[i].qiov;
355
356 ret = gnutls_hash_init(&dig, GNUTLS_DIG_SHA256);
357
358 if (ret < 0) {
359 return ret;
360 }
361
362 for (j = 0; j < qiov->niov; j++) {
363 ret = gnutls_hash(dig, qiov->iov[j].iov_base, qiov->iov[j].iov_len);
364 if (ret < 0) {
365 break;
366 }
367 }
368
369 gnutls_hash_deinit(dig, (void *) hash);
370 return ret;
371 }
372
373 static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
374 {
375 int max = 0;
376 QuorumVoteVersion *candidate, *winner = NULL;
377
378 QLIST_FOREACH(candidate, &votes->vote_list, next) {
379 if (candidate->vote_count > max) {
380 max = candidate->vote_count;
381 winner = candidate;
382 }
383 }
384
385 return winner;
386 }
387
388 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
389 * differing byte location. Yet it is handcoded to compare vectors one byte
390 * after another so it does not benefit from the libc SIMD optimizations.
391 * quorum_iovec_compare is written for speed and should be used in the non
392 * blkverify mode of quorum.
393 */
394 static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
395 {
396 int i;
397 int result;
398
399 assert(a->niov == b->niov);
400 for (i = 0; i < a->niov; i++) {
401 assert(a->iov[i].iov_len == b->iov[i].iov_len);
402 result = memcmp(a->iov[i].iov_base,
403 b->iov[i].iov_base,
404 a->iov[i].iov_len);
405 if (result) {
406 return false;
407 }
408 }
409
410 return true;
411 }
412
413 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB *acb,
414 const char *fmt, ...)
415 {
416 va_list ap;
417
418 va_start(ap, fmt);
419 fprintf(stderr, "quorum: sector_num=%" PRId64 " nb_sectors=%d ",
420 acb->sector_num, acb->nb_sectors);
421 vfprintf(stderr, fmt, ap);
422 fprintf(stderr, "\n");
423 va_end(ap);
424 exit(1);
425 }
426
427 static bool quorum_compare(QuorumAIOCB *acb,
428 QEMUIOVector *a,
429 QEMUIOVector *b)
430 {
431 BDRVQuorumState *s = acb->common.bs->opaque;
432 ssize_t offset;
433
434 /* This driver will replace blkverify in this particular case */
435 if (s->is_blkverify) {
436 offset = qemu_iovec_compare(a, b);
437 if (offset != -1) {
438 quorum_err(acb, "contents mismatch in sector %" PRId64,
439 acb->sector_num +
440 (uint64_t)(offset / BDRV_SECTOR_SIZE));
441 }
442 return true;
443 }
444
445 return quorum_iovec_compare(a, b);
446 }
447
448 /* Do a vote to get the error code */
449 static int quorum_vote_error(QuorumAIOCB *acb)
450 {
451 BDRVQuorumState *s = acb->common.bs->opaque;
452 QuorumVoteVersion *winner = NULL;
453 QuorumVotes error_votes;
454 QuorumVoteValue result_value;
455 int i, ret = 0;
456 bool error = false;
457
458 QLIST_INIT(&error_votes.vote_list);
459 error_votes.compare = quorum_64bits_compare;
460
461 for (i = 0; i < s->num_children; i++) {
462 ret = acb->qcrs[i].ret;
463 if (ret) {
464 error = true;
465 result_value.l = ret;
466 quorum_count_vote(&error_votes, &result_value, i);
467 }
468 }
469
470 if (error) {
471 winner = quorum_get_vote_winner(&error_votes);
472 ret = winner->value.l;
473 }
474
475 quorum_free_vote_list(&error_votes);
476
477 return ret;
478 }
479
480 static void quorum_vote(QuorumAIOCB *acb)
481 {
482 bool quorum = true;
483 int i, j, ret;
484 QuorumVoteValue hash;
485 BDRVQuorumState *s = acb->common.bs->opaque;
486 QuorumVoteVersion *winner;
487
488 if (quorum_has_too_much_io_failed(acb)) {
489 return;
490 }
491
492 /* get the index of the first successful read */
493 for (i = 0; i < s->num_children; i++) {
494 if (!acb->qcrs[i].ret) {
495 break;
496 }
497 }
498
499 assert(i < s->num_children);
500
501 /* compare this read with all other successful reads stopping at quorum
502 * failure
503 */
504 for (j = i + 1; j < s->num_children; j++) {
505 if (acb->qcrs[j].ret) {
506 continue;
507 }
508 quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
509 if (!quorum) {
510 break;
511 }
512 }
513
514 /* Every successful read agrees */
515 if (quorum) {
516 quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
517 return;
518 }
519
520 /* compute hashes for each successful read, also store indexes */
521 for (i = 0; i < s->num_children; i++) {
522 if (acb->qcrs[i].ret) {
523 continue;
524 }
525 ret = quorum_compute_hash(acb, i, &hash);
526 /* if ever the hash computation failed */
527 if (ret < 0) {
528 acb->vote_ret = ret;
529 goto free_exit;
530 }
531 quorum_count_vote(&acb->votes, &hash, i);
532 }
533
534 /* vote to select the most represented version */
535 winner = quorum_get_vote_winner(&acb->votes);
536
537 /* if the winner count is smaller than threshold the read fails */
538 if (winner->vote_count < s->threshold) {
539 quorum_report_failure(acb);
540 acb->vote_ret = -EIO;
541 goto free_exit;
542 }
543
544 /* we have a winner: copy it */
545 quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
546
547 /* some versions are bad print them */
548 quorum_report_bad_versions(s, acb, &winner->value);
549
550 free_exit:
551 /* free lists */
552 quorum_free_vote_list(&acb->votes);
553 }
554
555 static BlockDriverAIOCB *quorum_aio_readv(BlockDriverState *bs,
556 int64_t sector_num,
557 QEMUIOVector *qiov,
558 int nb_sectors,
559 BlockDriverCompletionFunc *cb,
560 void *opaque)
561 {
562 BDRVQuorumState *s = bs->opaque;
563 QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num,
564 nb_sectors, cb, opaque);
565 int i;
566
567 acb->is_read = true;
568
569 for (i = 0; i < s->num_children; i++) {
570 acb->qcrs[i].buf = qemu_blockalign(s->bs[i], qiov->size);
571 qemu_iovec_init(&acb->qcrs[i].qiov, qiov->niov);
572 qemu_iovec_clone(&acb->qcrs[i].qiov, qiov, acb->qcrs[i].buf);
573 }
574
575 for (i = 0; i < s->num_children; i++) {
576 bdrv_aio_readv(s->bs[i], sector_num, &acb->qcrs[i].qiov, nb_sectors,
577 quorum_aio_cb, &acb->qcrs[i]);
578 }
579
580 return &acb->common;
581 }
582
583 static BlockDriverAIOCB *quorum_aio_writev(BlockDriverState *bs,
584 int64_t sector_num,
585 QEMUIOVector *qiov,
586 int nb_sectors,
587 BlockDriverCompletionFunc *cb,
588 void *opaque)
589 {
590 BDRVQuorumState *s = bs->opaque;
591 QuorumAIOCB *acb = quorum_aio_get(s, bs, qiov, sector_num, nb_sectors,
592 cb, opaque);
593 int i;
594
595 for (i = 0; i < s->num_children; i++) {
596 acb->qcrs[i].aiocb = bdrv_aio_writev(s->bs[i], sector_num, qiov,
597 nb_sectors, &quorum_aio_cb,
598 &acb->qcrs[i]);
599 }
600
601 return &acb->common;
602 }
603
604 static int64_t quorum_getlength(BlockDriverState *bs)
605 {
606 BDRVQuorumState *s = bs->opaque;
607 int64_t result;
608 int i;
609
610 /* check that all file have the same length */
611 result = bdrv_getlength(s->bs[0]);
612 if (result < 0) {
613 return result;
614 }
615 for (i = 1; i < s->num_children; i++) {
616 int64_t value = bdrv_getlength(s->bs[i]);
617 if (value < 0) {
618 return value;
619 }
620 if (value != result) {
621 return -EIO;
622 }
623 }
624
625 return result;
626 }
627
628 static void quorum_invalidate_cache(BlockDriverState *bs, Error **errp)
629 {
630 BDRVQuorumState *s = bs->opaque;
631 Error *local_err = NULL;
632 int i;
633
634 for (i = 0; i < s->num_children; i++) {
635 bdrv_invalidate_cache(s->bs[i], &local_err);
636 if (local_err) {
637 error_propagate(errp, local_err);
638 return;
639 }
640 }
641 }
642
643 static coroutine_fn int quorum_co_flush(BlockDriverState *bs)
644 {
645 BDRVQuorumState *s = bs->opaque;
646 QuorumVoteVersion *winner = NULL;
647 QuorumVotes error_votes;
648 QuorumVoteValue result_value;
649 int i;
650 int result = 0;
651
652 QLIST_INIT(&error_votes.vote_list);
653 error_votes.compare = quorum_64bits_compare;
654
655 for (i = 0; i < s->num_children; i++) {
656 result = bdrv_co_flush(s->bs[i]);
657 result_value.l = result;
658 quorum_count_vote(&error_votes, &result_value, i);
659 }
660
661 winner = quorum_get_vote_winner(&error_votes);
662 result = winner->value.l;
663
664 quorum_free_vote_list(&error_votes);
665
666 return result;
667 }
668
669 static bool quorum_recurse_is_first_non_filter(BlockDriverState *bs,
670 BlockDriverState *candidate)
671 {
672 BDRVQuorumState *s = bs->opaque;
673 int i;
674
675 for (i = 0; i < s->num_children; i++) {
676 bool perm = bdrv_recurse_is_first_non_filter(s->bs[i],
677 candidate);
678 if (perm) {
679 return true;
680 }
681 }
682
683 return false;
684 }
685
686 static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
687 {
688
689 if (threshold < 1) {
690 error_set(errp, QERR_INVALID_PARAMETER_VALUE,
691 "vote-threshold", "value >= 1");
692 return -ERANGE;
693 }
694
695 if (threshold > num_children) {
696 error_setg(errp, "threshold may not exceed children count");
697 return -ERANGE;
698 }
699
700 return 0;
701 }
702
703 static QemuOptsList quorum_runtime_opts = {
704 .name = "quorum",
705 .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
706 .desc = {
707 {
708 .name = QUORUM_OPT_VOTE_THRESHOLD,
709 .type = QEMU_OPT_NUMBER,
710 .help = "The number of vote needed for reaching quorum",
711 },
712 {
713 .name = QUORUM_OPT_BLKVERIFY,
714 .type = QEMU_OPT_BOOL,
715 .help = "Trigger block verify mode if set",
716 },
717 { /* end of list */ }
718 },
719 };
720
721 static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
722 Error **errp)
723 {
724 BDRVQuorumState *s = bs->opaque;
725 Error *local_err = NULL;
726 QemuOpts *opts;
727 bool *opened;
728 QDict *sub = NULL;
729 QList *list = NULL;
730 const QListEntry *lentry;
731 int i;
732 int ret = 0;
733
734 qdict_flatten(options);
735 qdict_extract_subqdict(options, &sub, "children.");
736 qdict_array_split(sub, &list);
737
738 if (qdict_size(sub)) {
739 error_setg(&local_err, "Invalid option children.%s",
740 qdict_first(sub)->key);
741 ret = -EINVAL;
742 goto exit;
743 }
744
745 /* count how many different children are present */
746 s->num_children = qlist_size(list);
747 if (s->num_children < 2) {
748 error_setg(&local_err,
749 "Number of provided children must be greater than 1");
750 ret = -EINVAL;
751 goto exit;
752 }
753
754 opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
755 qemu_opts_absorb_qdict(opts, options, &local_err);
756 if (error_is_set(&local_err)) {
757 ret = -EINVAL;
758 goto exit;
759 }
760
761 s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
762
763 /* and validate it against s->num_children */
764 ret = quorum_valid_threshold(s->threshold, s->num_children, &local_err);
765 if (ret < 0) {
766 goto exit;
767 }
768
769 /* is the driver in blkverify mode */
770 if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false) &&
771 s->num_children == 2 && s->threshold == 2) {
772 s->is_blkverify = true;
773 } else if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false)) {
774 fprintf(stderr, "blkverify mode is set by setting blkverify=on "
775 "and using two files with vote_threshold=2\n");
776 }
777
778 /* allocate the children BlockDriverState array */
779 s->bs = g_new0(BlockDriverState *, s->num_children);
780 opened = g_new0(bool, s->num_children);
781
782 for (i = 0, lentry = qlist_first(list); lentry;
783 lentry = qlist_next(lentry), i++) {
784 QDict *d;
785 QString *string;
786
787 switch (qobject_type(lentry->value))
788 {
789 /* List of options */
790 case QTYPE_QDICT:
791 d = qobject_to_qdict(lentry->value);
792 QINCREF(d);
793 ret = bdrv_open(&s->bs[i], NULL, NULL, d, flags, NULL,
794 &local_err);
795 break;
796
797 /* QMP reference */
798 case QTYPE_QSTRING:
799 string = qobject_to_qstring(lentry->value);
800 ret = bdrv_open(&s->bs[i], NULL, qstring_get_str(string), NULL,
801 flags, NULL, &local_err);
802 break;
803
804 default:
805 error_setg(&local_err, "Specification of child block device %i "
806 "is invalid", i);
807 ret = -EINVAL;
808 }
809
810 if (ret < 0) {
811 goto close_exit;
812 }
813 opened[i] = true;
814 }
815
816 g_free(opened);
817 goto exit;
818
819 close_exit:
820 /* cleanup on error */
821 for (i = 0; i < s->num_children; i++) {
822 if (!opened[i]) {
823 continue;
824 }
825 bdrv_unref(s->bs[i]);
826 }
827 g_free(s->bs);
828 g_free(opened);
829 exit:
830 /* propagate error */
831 if (error_is_set(&local_err)) {
832 error_propagate(errp, local_err);
833 }
834 QDECREF(list);
835 QDECREF(sub);
836 return ret;
837 }
838
839 static void quorum_close(BlockDriverState *bs)
840 {
841 BDRVQuorumState *s = bs->opaque;
842 int i;
843
844 for (i = 0; i < s->num_children; i++) {
845 bdrv_unref(s->bs[i]);
846 }
847
848 g_free(s->bs);
849 }
850
851 static BlockDriver bdrv_quorum = {
852 .format_name = "quorum",
853 .protocol_name = "quorum",
854
855 .instance_size = sizeof(BDRVQuorumState),
856
857 .bdrv_file_open = quorum_open,
858 .bdrv_close = quorum_close,
859
860 .bdrv_co_flush_to_disk = quorum_co_flush,
861
862 .bdrv_getlength = quorum_getlength,
863
864 .bdrv_aio_readv = quorum_aio_readv,
865 .bdrv_aio_writev = quorum_aio_writev,
866 .bdrv_invalidate_cache = quorum_invalidate_cache,
867
868 .is_filter = true,
869 .bdrv_recurse_is_first_non_filter = quorum_recurse_is_first_non_filter,
870 };
871
872 static void bdrv_quorum_init(void)
873 {
874 bdrv_register(&bdrv_quorum);
875 }
876
877 block_init(bdrv_quorum_init);