1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16#include "qemu/osdep.h"
17#include "qemu/cutils.h"
18#include "qemu/module.h"
19#include "qemu/option.h"
20#include "qemu/memalign.h"
21#include "block/block_int.h"
22#include "block/coroutines.h"
23#include "block/qdict.h"
24#include "qapi/error.h"
25#include "qapi/qapi-events-block.h"
26#include "qapi/qmp/qdict.h"
27#include "qapi/qmp/qerror.h"
28#include "qapi/qmp/qlist.h"
29#include "qapi/qmp/qstring.h"
30#include "crypto/hash.h"
31
32#define HASH_LENGTH 32
33
34#define INDEXSTR_LEN 32
35
36#define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
37#define QUORUM_OPT_BLKVERIFY "blkverify"
38#define QUORUM_OPT_REWRITE "rewrite-corrupted"
39#define QUORUM_OPT_READ_PATTERN "read-pattern"
40
41
42typedef union QuorumVoteValue {
43 uint8_t h[HASH_LENGTH];
44 int64_t l;
45} QuorumVoteValue;
46
47
48typedef struct QuorumVoteItem {
49 int index;
50 QLIST_ENTRY(QuorumVoteItem) next;
51} QuorumVoteItem;
52
53
54
55
56
57
58typedef struct QuorumVoteVersion {
59 QuorumVoteValue value;
60 int index;
61 int vote_count;
62 QLIST_HEAD(, QuorumVoteItem) items;
63 QLIST_ENTRY(QuorumVoteVersion) next;
64} QuorumVoteVersion;
65
66
67typedef struct QuorumVotes {
68 QLIST_HEAD(, QuorumVoteVersion) vote_list;
69 bool (*compare)(QuorumVoteValue *a, QuorumVoteValue *b);
70} QuorumVotes;
71
72
73typedef struct BDRVQuorumState {
74 BdrvChild **children;
75 int num_children;
76 unsigned next_child_index;
77
78
79 int threshold;
80
81
82 bool is_blkverify;
83
84
85
86
87
88
89
90 bool rewrite_corrupted;
91
92
93
94 QuorumReadPattern read_pattern;
95} BDRVQuorumState;
96
97typedef struct QuorumAIOCB QuorumAIOCB;
98
99
100
101
102
103
104typedef struct QuorumChildRequest {
105 BlockDriverState *bs;
106 QEMUIOVector qiov;
107 uint8_t *buf;
108 int ret;
109 QuorumAIOCB *parent;
110} QuorumChildRequest;
111
112
113
114
115
116
117struct QuorumAIOCB {
118 BlockDriverState *bs;
119 Coroutine *co;
120
121
122 uint64_t offset;
123 uint64_t bytes;
124 int flags;
125
126 QEMUIOVector *qiov;
127
128 QuorumChildRequest *qcrs;
129 int count;
130 int success_count;
131
132 int rewrite_count;
133
134
135
136 QuorumVotes votes;
137
138 bool is_read;
139 int vote_ret;
140 int children_read;
141};
142
143typedef struct QuorumCo {
144 QuorumAIOCB *acb;
145 int idx;
146} QuorumCo;
147
148static void quorum_aio_finalize(QuorumAIOCB *acb)
149{
150 g_free(acb->qcrs);
151 g_free(acb);
152}
153
154static bool quorum_sha256_compare(QuorumVoteValue *a, QuorumVoteValue *b)
155{
156 return !memcmp(a->h, b->h, HASH_LENGTH);
157}
158
159static bool quorum_64bits_compare(QuorumVoteValue *a, QuorumVoteValue *b)
160{
161 return a->l == b->l;
162}
163
164static QuorumAIOCB *coroutine_fn quorum_aio_get(BlockDriverState *bs,
165 QEMUIOVector *qiov,
166 uint64_t offset, uint64_t bytes,
167 int flags)
168{
169 BDRVQuorumState *s = bs->opaque;
170 QuorumAIOCB *acb = g_new(QuorumAIOCB, 1);
171 int i;
172
173 *acb = (QuorumAIOCB) {
174 .co = qemu_coroutine_self(),
175 .bs = bs,
176 .offset = offset,
177 .bytes = bytes,
178 .flags = flags,
179 .qiov = qiov,
180 .votes.compare = quorum_sha256_compare,
181 .votes.vote_list = QLIST_HEAD_INITIALIZER(acb.votes.vote_list),
182 };
183
184 acb->qcrs = g_new0(QuorumChildRequest, s->num_children);
185 for (i = 0; i < s->num_children; i++) {
186 acb->qcrs[i].buf = NULL;
187 acb->qcrs[i].ret = 0;
188 acb->qcrs[i].parent = acb;
189 }
190
191 return acb;
192}
193
194static void quorum_report_bad(QuorumOpType type, uint64_t offset,
195 uint64_t bytes, char *node_name, int ret)
196{
197 const char *msg = NULL;
198 int64_t start_sector = offset / BDRV_SECTOR_SIZE;
199 int64_t end_sector = DIV_ROUND_UP(offset + bytes, BDRV_SECTOR_SIZE);
200
201 if (ret < 0) {
202 msg = strerror(-ret);
203 }
204
205 qapi_event_send_quorum_report_bad(type, !!msg, msg, node_name, start_sector,
206 end_sector - start_sector);
207}
208
209static void quorum_report_failure(QuorumAIOCB *acb)
210{
211 const char *reference = bdrv_get_device_or_node_name(acb->bs);
212 int64_t start_sector = acb->offset / BDRV_SECTOR_SIZE;
213 int64_t end_sector = DIV_ROUND_UP(acb->offset + acb->bytes,
214 BDRV_SECTOR_SIZE);
215
216 qapi_event_send_quorum_failure(reference, start_sector,
217 end_sector - start_sector);
218}
219
220static int quorum_vote_error(QuorumAIOCB *acb);
221
222static bool quorum_has_too_much_io_failed(QuorumAIOCB *acb)
223{
224 BDRVQuorumState *s = acb->bs->opaque;
225
226 if (acb->success_count < s->threshold) {
227 acb->vote_ret = quorum_vote_error(acb);
228 quorum_report_failure(acb);
229 return true;
230 }
231
232 return false;
233}
234
235static void quorum_copy_qiov(QEMUIOVector *dest, QEMUIOVector *source)
236{
237 int i;
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);
245 }
246}
247
248static void quorum_report_bad_acb(QuorumChildRequest *sacb, int ret)
249{
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->offset, acb->bytes, sacb->bs->node_name, ret);
253}
254
255static void quorum_report_bad_versions(BDRVQuorumState *s,
256 QuorumAIOCB *acb,
257 QuorumVoteValue *value)
258{
259 QuorumVoteVersion *version;
260 QuorumVoteItem *item;
261
262 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
263 if (acb->votes.compare(&version->value, value)) {
264 continue;
265 }
266 QLIST_FOREACH(item, &version->items, next) {
267 quorum_report_bad(QUORUM_OP_TYPE_READ, acb->offset, acb->bytes,
268 s->children[item->index]->bs->node_name, 0);
269 }
270 }
271}
272
273static void coroutine_fn quorum_rewrite_entry(void *opaque)
274{
275 QuorumCo *co = opaque;
276 QuorumAIOCB *acb = co->acb;
277 BDRVQuorumState *s = acb->bs->opaque;
278
279
280
281
282
283 bdrv_co_pwritev(s->children[co->idx], acb->offset, acb->bytes,
284 acb->qiov, acb->flags & ~BDRV_REQ_WRITE_UNCHANGED);
285
286
287 acb->rewrite_count--;
288 if (!acb->rewrite_count) {
289 qemu_coroutine_enter_if_inactive(acb->co);
290 }
291}
292
293static bool quorum_rewrite_bad_versions(QuorumAIOCB *acb,
294 QuorumVoteValue *value)
295{
296 QuorumVoteVersion *version;
297 QuorumVoteItem *item;
298 int count = 0;
299
300
301
302
303 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
304 if (acb->votes.compare(&version->value, value)) {
305 continue;
306 }
307 QLIST_FOREACH(item, &version->items, next) {
308 count++;
309 }
310 }
311
312
313 acb->rewrite_count = count;
314
315
316 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
317 if (acb->votes.compare(&version->value, value)) {
318 continue;
319 }
320 QLIST_FOREACH(item, &version->items, next) {
321 Coroutine *co;
322 QuorumCo data = {
323 .acb = acb,
324 .idx = item->index,
325 };
326
327 co = qemu_coroutine_create(quorum_rewrite_entry, &data);
328 qemu_coroutine_enter(co);
329 }
330 }
331
332
333 return count;
334}
335
336static void quorum_count_vote(QuorumVotes *votes,
337 QuorumVoteValue *value,
338 int index)
339{
340 QuorumVoteVersion *v = NULL, *version = NULL;
341 QuorumVoteItem *item;
342
343
344 QLIST_FOREACH(v, &votes->vote_list, next) {
345 if (votes->compare(&v->value, value)) {
346 version = v;
347 break;
348 }
349 }
350
351
352 if (!version) {
353 version = g_new0(QuorumVoteVersion, 1);
354 QLIST_INIT(&version->items);
355 memcpy(&version->value, value, sizeof(version->value));
356 version->index = index;
357 version->vote_count = 0;
358 QLIST_INSERT_HEAD(&votes->vote_list, version, next);
359 }
360
361 version->vote_count++;
362
363 item = g_new0(QuorumVoteItem, 1);
364 item->index = index;
365 QLIST_INSERT_HEAD(&version->items, item, next);
366}
367
368static void quorum_free_vote_list(QuorumVotes *votes)
369{
370 QuorumVoteVersion *version, *next_version;
371 QuorumVoteItem *item, *next_item;
372
373 QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
374 QLIST_REMOVE(version, next);
375 QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
376 QLIST_REMOVE(item, next);
377 g_free(item);
378 }
379 g_free(version);
380 }
381}
382
383static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
384{
385 QEMUIOVector *qiov = &acb->qcrs[i].qiov;
386 size_t len = sizeof(hash->h);
387 uint8_t *data = hash->h;
388
389
390
391
392 if (qcrypto_hash_bytesv(QCRYPTO_HASH_ALG_SHA256,
393 qiov->iov, qiov->niov,
394 &data, &len,
395 NULL) < 0) {
396 return -EINVAL;
397 }
398
399 return 0;
400}
401
402static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
403{
404 int max = 0;
405 QuorumVoteVersion *candidate, *winner = NULL;
406
407 QLIST_FOREACH(candidate, &votes->vote_list, next) {
408 if (candidate->vote_count > max) {
409 max = candidate->vote_count;
410 winner = candidate;
411 }
412 }
413
414 return winner;
415}
416
417
418
419
420
421
422
423static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
424{
425 int i;
426 int result;
427
428 assert(a->niov == b->niov);
429 for (i = 0; i < a->niov; i++) {
430 assert(a->iov[i].iov_len == b->iov[i].iov_len);
431 result = memcmp(a->iov[i].iov_base,
432 b->iov[i].iov_base,
433 a->iov[i].iov_len);
434 if (result) {
435 return false;
436 }
437 }
438
439 return true;
440}
441
442static bool quorum_compare(QuorumAIOCB *acb, QEMUIOVector *a, QEMUIOVector *b)
443{
444 BDRVQuorumState *s = acb->bs->opaque;
445 ssize_t offset;
446
447
448 if (s->is_blkverify) {
449 offset = qemu_iovec_compare(a, b);
450 if (offset != -1) {
451 fprintf(stderr, "quorum: offset=%" PRIu64 " bytes=%" PRIu64
452 " contents mismatch at offset %" PRIu64 "\n",
453 acb->offset, acb->bytes, acb->offset + offset);
454 exit(1);
455 }
456 return true;
457 }
458
459 return quorum_iovec_compare(a, b);
460}
461
462
463static int quorum_vote_error(QuorumAIOCB *acb)
464{
465 BDRVQuorumState *s = acb->bs->opaque;
466 QuorumVoteVersion *winner = NULL;
467 QuorumVotes error_votes;
468 QuorumVoteValue result_value;
469 int i, ret = 0;
470 bool error = false;
471
472 QLIST_INIT(&error_votes.vote_list);
473 error_votes.compare = quorum_64bits_compare;
474
475 for (i = 0; i < s->num_children; i++) {
476 ret = acb->qcrs[i].ret;
477 if (ret) {
478 error = true;
479 result_value.l = ret;
480 quorum_count_vote(&error_votes, &result_value, i);
481 }
482 }
483
484 if (error) {
485 winner = quorum_get_vote_winner(&error_votes);
486 ret = winner->value.l;
487 }
488
489 quorum_free_vote_list(&error_votes);
490
491 return ret;
492}
493
494static void quorum_vote(QuorumAIOCB *acb)
495{
496 bool quorum = true;
497 int i, j, ret;
498 QuorumVoteValue hash;
499 BDRVQuorumState *s = acb->bs->opaque;
500 QuorumVoteVersion *winner;
501
502 if (quorum_has_too_much_io_failed(acb)) {
503 return;
504 }
505
506
507 for (i = 0; i < s->num_children; i++) {
508 if (!acb->qcrs[i].ret) {
509 break;
510 }
511 }
512
513 assert(i < s->num_children);
514
515
516
517
518 for (j = i + 1; j < s->num_children; j++) {
519 if (acb->qcrs[j].ret) {
520 continue;
521 }
522 quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
523 if (!quorum) {
524 break;
525 }
526 }
527
528
529 if (quorum) {
530 quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
531 return;
532 }
533
534
535 for (i = 0; i < s->num_children; i++) {
536 if (acb->qcrs[i].ret) {
537 continue;
538 }
539 ret = quorum_compute_hash(acb, i, &hash);
540
541 if (ret < 0) {
542 acb->vote_ret = ret;
543 goto free_exit;
544 }
545 quorum_count_vote(&acb->votes, &hash, i);
546 }
547
548
549 winner = quorum_get_vote_winner(&acb->votes);
550
551
552 if (winner->vote_count < s->threshold) {
553 quorum_report_failure(acb);
554 acb->vote_ret = -EIO;
555 goto free_exit;
556 }
557
558
559 quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
560
561
562 quorum_report_bad_versions(s, acb, &winner->value);
563
564
565 if (s->rewrite_corrupted) {
566 quorum_rewrite_bad_versions(acb, &winner->value);
567 }
568
569free_exit:
570
571 quorum_free_vote_list(&acb->votes);
572}
573
574static void coroutine_fn read_quorum_children_entry(void *opaque)
575{
576 QuorumCo *co = opaque;
577 QuorumAIOCB *acb = co->acb;
578 BDRVQuorumState *s = acb->bs->opaque;
579 int i = co->idx;
580 QuorumChildRequest *sacb = &acb->qcrs[i];
581
582 sacb->bs = s->children[i]->bs;
583 sacb->ret = bdrv_co_preadv(s->children[i], acb->offset, acb->bytes,
584 &acb->qcrs[i].qiov, 0);
585
586 if (sacb->ret == 0) {
587 acb->success_count++;
588 } else {
589 quorum_report_bad_acb(sacb, sacb->ret);
590 }
591
592 acb->count++;
593 assert(acb->count <= s->num_children);
594 assert(acb->success_count <= s->num_children);
595
596
597 if (acb->count == s->num_children) {
598 qemu_coroutine_enter_if_inactive(acb->co);
599 }
600}
601
602static int coroutine_fn read_quorum_children(QuorumAIOCB *acb)
603{
604 BDRVQuorumState *s = acb->bs->opaque;
605 int i;
606
607 acb->children_read = s->num_children;
608 for (i = 0; i < s->num_children; i++) {
609 acb->qcrs[i].buf = qemu_blockalign(s->children[i]->bs, acb->qiov->size);
610 qemu_iovec_init(&acb->qcrs[i].qiov, acb->qiov->niov);
611 qemu_iovec_clone(&acb->qcrs[i].qiov, acb->qiov, acb->qcrs[i].buf);
612 }
613
614 for (i = 0; i < s->num_children; i++) {
615 Coroutine *co;
616 QuorumCo data = {
617 .acb = acb,
618 .idx = i,
619 };
620
621 co = qemu_coroutine_create(read_quorum_children_entry, &data);
622 qemu_coroutine_enter(co);
623 }
624
625 while (acb->count < s->num_children) {
626 qemu_coroutine_yield();
627 }
628
629
630 quorum_vote(acb);
631 for (i = 0; i < s->num_children; i++) {
632 qemu_vfree(acb->qcrs[i].buf);
633 qemu_iovec_destroy(&acb->qcrs[i].qiov);
634 }
635
636 while (acb->rewrite_count) {
637 qemu_coroutine_yield();
638 }
639
640 return acb->vote_ret;
641}
642
643static int coroutine_fn read_fifo_child(QuorumAIOCB *acb)
644{
645 BDRVQuorumState *s = acb->bs->opaque;
646 int n, ret;
647
648
649 do {
650 n = acb->children_read++;
651 acb->qcrs[n].bs = s->children[n]->bs;
652 ret = bdrv_co_preadv(s->children[n], acb->offset, acb->bytes,
653 acb->qiov, 0);
654 if (ret < 0) {
655 quorum_report_bad_acb(&acb->qcrs[n], ret);
656 }
657 } while (ret < 0 && acb->children_read < s->num_children);
658
659
660
661 return ret;
662}
663
664static int coroutine_fn quorum_co_preadv(BlockDriverState *bs,
665 int64_t offset, int64_t bytes,
666 QEMUIOVector *qiov,
667 BdrvRequestFlags flags)
668{
669 BDRVQuorumState *s = bs->opaque;
670 QuorumAIOCB *acb = quorum_aio_get(bs, qiov, offset, bytes, flags);
671 int ret;
672
673 acb->is_read = true;
674 acb->children_read = 0;
675
676 if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
677 ret = read_quorum_children(acb);
678 } else {
679 ret = read_fifo_child(acb);
680 }
681 quorum_aio_finalize(acb);
682
683 return ret;
684}
685
686static void coroutine_fn write_quorum_entry(void *opaque)
687{
688 QuorumCo *co = opaque;
689 QuorumAIOCB *acb = co->acb;
690 BDRVQuorumState *s = acb->bs->opaque;
691 int i = co->idx;
692 QuorumChildRequest *sacb = &acb->qcrs[i];
693
694 sacb->bs = s->children[i]->bs;
695 if (acb->flags & BDRV_REQ_ZERO_WRITE) {
696 sacb->ret = bdrv_co_pwrite_zeroes(s->children[i], acb->offset,
697 acb->bytes, acb->flags);
698 } else {
699 sacb->ret = bdrv_co_pwritev(s->children[i], acb->offset, acb->bytes,
700 acb->qiov, acb->flags);
701 }
702 if (sacb->ret == 0) {
703 acb->success_count++;
704 } else {
705 quorum_report_bad_acb(sacb, sacb->ret);
706 }
707 acb->count++;
708 assert(acb->count <= s->num_children);
709 assert(acb->success_count <= s->num_children);
710
711
712 if (acb->count == s->num_children) {
713 qemu_coroutine_enter_if_inactive(acb->co);
714 }
715}
716
717static int coroutine_fn quorum_co_pwritev(BlockDriverState *bs, int64_t offset,
718 int64_t bytes, QEMUIOVector *qiov,
719 BdrvRequestFlags flags)
720{
721 BDRVQuorumState *s = bs->opaque;
722 QuorumAIOCB *acb = quorum_aio_get(bs, qiov, offset, bytes, flags);
723 int i, ret;
724
725 for (i = 0; i < s->num_children; i++) {
726 Coroutine *co;
727 QuorumCo data = {
728 .acb = acb,
729 .idx = i,
730 };
731
732 co = qemu_coroutine_create(write_quorum_entry, &data);
733 qemu_coroutine_enter(co);
734 }
735
736 while (acb->count < s->num_children) {
737 qemu_coroutine_yield();
738 }
739
740 quorum_has_too_much_io_failed(acb);
741
742 ret = acb->vote_ret;
743 quorum_aio_finalize(acb);
744
745 return ret;
746}
747
748static int coroutine_fn quorum_co_pwrite_zeroes(BlockDriverState *bs,
749 int64_t offset, int64_t bytes,
750 BdrvRequestFlags flags)
751
752{
753 return quorum_co_pwritev(bs, offset, bytes, NULL,
754 flags | BDRV_REQ_ZERO_WRITE);
755}
756
757static int64_t quorum_getlength(BlockDriverState *bs)
758{
759 BDRVQuorumState *s = bs->opaque;
760 int64_t result;
761 int i;
762
763
764 result = bdrv_getlength(s->children[0]->bs);
765 if (result < 0) {
766 return result;
767 }
768 for (i = 1; i < s->num_children; i++) {
769 int64_t value = bdrv_getlength(s->children[i]->bs);
770 if (value < 0) {
771 return value;
772 }
773 if (value != result) {
774 return -EIO;
775 }
776 }
777
778 return result;
779}
780
781static coroutine_fn int quorum_co_flush(BlockDriverState *bs)
782{
783 BDRVQuorumState *s = bs->opaque;
784 QuorumVoteVersion *winner = NULL;
785 QuorumVotes error_votes;
786 QuorumVoteValue result_value;
787 int i;
788 int result = 0;
789 int success_count = 0;
790
791 QLIST_INIT(&error_votes.vote_list);
792 error_votes.compare = quorum_64bits_compare;
793
794 for (i = 0; i < s->num_children; i++) {
795 result = bdrv_co_flush(s->children[i]->bs);
796 if (result) {
797 quorum_report_bad(QUORUM_OP_TYPE_FLUSH, 0, 0,
798 s->children[i]->bs->node_name, result);
799 result_value.l = result;
800 quorum_count_vote(&error_votes, &result_value, i);
801 } else {
802 success_count++;
803 }
804 }
805
806 if (success_count >= s->threshold) {
807 result = 0;
808 } else {
809 winner = quorum_get_vote_winner(&error_votes);
810 result = winner->value.l;
811 }
812 quorum_free_vote_list(&error_votes);
813
814 return result;
815}
816
817static bool quorum_recurse_can_replace(BlockDriverState *bs,
818 BlockDriverState *to_replace)
819{
820 BDRVQuorumState *s = bs->opaque;
821 int i;
822
823 for (i = 0; i < s->num_children; i++) {
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851 if (s->children[i]->bs == to_replace) {
852
853
854
855
856
857
858
859
860
861
862 return QLIST_FIRST(&to_replace->parents) == s->children[i] &&
863 QLIST_NEXT(s->children[i], next_parent) == NULL;
864 }
865 }
866
867 return false;
868}
869
870static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
871{
872
873 if (threshold < 1) {
874 error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
875 "vote-threshold", "a value >= 1");
876 return -ERANGE;
877 }
878
879 if (threshold > num_children) {
880 error_setg(errp, "threshold may not exceed children count");
881 return -ERANGE;
882 }
883
884 return 0;
885}
886
887static QemuOptsList quorum_runtime_opts = {
888 .name = "quorum",
889 .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
890 .desc = {
891 {
892 .name = QUORUM_OPT_VOTE_THRESHOLD,
893 .type = QEMU_OPT_NUMBER,
894 .help = "The number of vote needed for reaching quorum",
895 },
896 {
897 .name = QUORUM_OPT_BLKVERIFY,
898 .type = QEMU_OPT_BOOL,
899 .help = "Trigger block verify mode if set",
900 },
901 {
902 .name = QUORUM_OPT_REWRITE,
903 .type = QEMU_OPT_BOOL,
904 .help = "Rewrite corrupted block on read quorum",
905 },
906 {
907 .name = QUORUM_OPT_READ_PATTERN,
908 .type = QEMU_OPT_STRING,
909 .help = "Allowed pattern: quorum, fifo. Quorum is default",
910 },
911 { }
912 },
913};
914
915static void quorum_refresh_flags(BlockDriverState *bs)
916{
917 BDRVQuorumState *s = bs->opaque;
918 int i;
919
920 bs->supported_zero_flags =
921 BDRV_REQ_FUA | BDRV_REQ_MAY_UNMAP | BDRV_REQ_NO_FALLBACK;
922
923 for (i = 0; i < s->num_children; i++) {
924 bs->supported_zero_flags &= s->children[i]->bs->supported_zero_flags;
925 }
926
927 bs->supported_zero_flags |= BDRV_REQ_WRITE_UNCHANGED;
928}
929
930static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
931 Error **errp)
932{
933 BDRVQuorumState *s = bs->opaque;
934 QemuOpts *opts = NULL;
935 const char *pattern_str;
936 bool *opened;
937 int i;
938 int ret = 0;
939
940 qdict_flatten(options);
941
942
943 s->num_children = qdict_array_entries(options, "children.");
944 if (s->num_children < 0) {
945 error_setg(errp, "Option children is not a valid array");
946 ret = -EINVAL;
947 goto exit;
948 }
949 if (s->num_children < 1) {
950 error_setg(errp, "Number of provided children must be 1 or more");
951 ret = -EINVAL;
952 goto exit;
953 }
954
955 opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
956 if (!qemu_opts_absorb_qdict(opts, options, errp)) {
957 ret = -EINVAL;
958 goto exit;
959 }
960
961 s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
962
963 ret = quorum_valid_threshold(s->threshold, s->num_children, errp);
964 if (ret < 0) {
965 goto exit;
966 }
967
968 pattern_str = qemu_opt_get(opts, QUORUM_OPT_READ_PATTERN);
969 if (!pattern_str) {
970 ret = QUORUM_READ_PATTERN_QUORUM;
971 } else {
972 ret = qapi_enum_parse(&QuorumReadPattern_lookup, pattern_str,
973 -EINVAL, NULL);
974 }
975 if (ret < 0) {
976 error_setg(errp, "Please set read-pattern as fifo or quorum");
977 goto exit;
978 }
979 s->read_pattern = ret;
980
981 if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
982 s->is_blkverify = qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false);
983 if (s->is_blkverify && (s->num_children != 2 || s->threshold != 2)) {
984 error_setg(errp, "blkverify=on can only be set if there are "
985 "exactly two files and vote-threshold is 2");
986 ret = -EINVAL;
987 goto exit;
988 }
989
990 s->rewrite_corrupted = qemu_opt_get_bool(opts, QUORUM_OPT_REWRITE,
991 false);
992 if (s->rewrite_corrupted && s->is_blkverify) {
993 error_setg(errp,
994 "rewrite-corrupted=on cannot be used with blkverify=on");
995 ret = -EINVAL;
996 goto exit;
997 }
998 }
999
1000
1001 s->children = g_new0(BdrvChild *, s->num_children);
1002 opened = g_new0(bool, s->num_children);
1003
1004 for (i = 0; i < s->num_children; i++) {
1005 char indexstr[INDEXSTR_LEN];
1006 ret = snprintf(indexstr, INDEXSTR_LEN, "children.%d", i);
1007 assert(ret < INDEXSTR_LEN);
1008
1009 s->children[i] = bdrv_open_child(NULL, options, indexstr, bs,
1010 &child_of_bds, BDRV_CHILD_DATA, false,
1011 errp);
1012 if (!s->children[i]) {
1013 ret = -EINVAL;
1014 goto close_exit;
1015 }
1016
1017 opened[i] = true;
1018 }
1019 s->next_child_index = s->num_children;
1020
1021 bs->supported_write_flags = BDRV_REQ_WRITE_UNCHANGED;
1022 quorum_refresh_flags(bs);
1023
1024 g_free(opened);
1025 goto exit;
1026
1027close_exit:
1028
1029 for (i = 0; i < s->num_children; i++) {
1030 if (!opened[i]) {
1031 continue;
1032 }
1033 bdrv_unref_child(bs, s->children[i]);
1034 }
1035 g_free(s->children);
1036 g_free(opened);
1037exit:
1038 qemu_opts_del(opts);
1039 return ret;
1040}
1041
1042static void quorum_close(BlockDriverState *bs)
1043{
1044 BDRVQuorumState *s = bs->opaque;
1045 int i;
1046
1047 for (i = 0; i < s->num_children; i++) {
1048 bdrv_unref_child(bs, s->children[i]);
1049 }
1050
1051 g_free(s->children);
1052}
1053
1054static void quorum_add_child(BlockDriverState *bs, BlockDriverState *child_bs,
1055 Error **errp)
1056{
1057 BDRVQuorumState *s = bs->opaque;
1058 BdrvChild *child;
1059 char indexstr[INDEXSTR_LEN];
1060 int ret;
1061
1062 if (s->is_blkverify) {
1063 error_setg(errp, "Cannot add a child to a quorum in blkverify mode");
1064 return;
1065 }
1066
1067 assert(s->num_children <= INT_MAX / sizeof(BdrvChild *));
1068 if (s->num_children == INT_MAX / sizeof(BdrvChild *) ||
1069 s->next_child_index == UINT_MAX) {
1070 error_setg(errp, "Too many children");
1071 return;
1072 }
1073
1074 ret = snprintf(indexstr, INDEXSTR_LEN, "children.%u", s->next_child_index);
1075 if (ret < 0 || ret >= INDEXSTR_LEN) {
1076 error_setg(errp, "cannot generate child name");
1077 return;
1078 }
1079 s->next_child_index++;
1080
1081 bdrv_drained_begin(bs);
1082
1083
1084 bdrv_ref(child_bs);
1085
1086 child = bdrv_attach_child(bs, child_bs, indexstr, &child_of_bds,
1087 BDRV_CHILD_DATA, errp);
1088 if (child == NULL) {
1089 s->next_child_index--;
1090 goto out;
1091 }
1092 s->children = g_renew(BdrvChild *, s->children, s->num_children + 1);
1093 s->children[s->num_children++] = child;
1094 quorum_refresh_flags(bs);
1095
1096out:
1097 bdrv_drained_end(bs);
1098}
1099
1100static void quorum_del_child(BlockDriverState *bs, BdrvChild *child,
1101 Error **errp)
1102{
1103 BDRVQuorumState *s = bs->opaque;
1104 char indexstr[INDEXSTR_LEN];
1105 int i;
1106
1107 for (i = 0; i < s->num_children; i++) {
1108 if (s->children[i] == child) {
1109 break;
1110 }
1111 }
1112
1113
1114 assert(i < s->num_children);
1115
1116 if (s->num_children <= s->threshold) {
1117 error_setg(errp,
1118 "The number of children cannot be lower than the vote threshold %d",
1119 s->threshold);
1120 return;
1121 }
1122
1123
1124 assert(!s->is_blkverify);
1125
1126 snprintf(indexstr, INDEXSTR_LEN, "children.%u", s->next_child_index - 1);
1127 if (!strncmp(child->name, indexstr, INDEXSTR_LEN)) {
1128 s->next_child_index--;
1129 }
1130
1131 bdrv_drained_begin(bs);
1132
1133
1134 memmove(&s->children[i], &s->children[i + 1],
1135 (s->num_children - i - 1) * sizeof(BdrvChild *));
1136 s->children = g_renew(BdrvChild *, s->children, --s->num_children);
1137 bdrv_unref_child(bs, child);
1138
1139 quorum_refresh_flags(bs);
1140 bdrv_drained_end(bs);
1141}
1142
1143static void quorum_gather_child_options(BlockDriverState *bs, QDict *target,
1144 bool backing_overridden)
1145{
1146 BDRVQuorumState *s = bs->opaque;
1147 QList *children_list;
1148 int i;
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173 children_list = qlist_new();
1174 qdict_put(target, "children", children_list);
1175
1176 for (i = 0; i < s->num_children; i++) {
1177 qlist_append(children_list,
1178 qobject_ref(s->children[i]->bs->full_open_options));
1179 }
1180}
1181
1182static char *quorum_dirname(BlockDriverState *bs, Error **errp)
1183{
1184
1185
1186
1187
1188 error_setg(errp, "Cannot generate a base directory for quorum nodes");
1189 return NULL;
1190}
1191
1192static void quorum_child_perm(BlockDriverState *bs, BdrvChild *c,
1193 BdrvChildRole role,
1194 BlockReopenQueue *reopen_queue,
1195 uint64_t perm, uint64_t shared,
1196 uint64_t *nperm, uint64_t *nshared)
1197{
1198 BDRVQuorumState *s = bs->opaque;
1199
1200 *nperm = perm & DEFAULT_PERM_PASSTHROUGH;
1201 if (s->rewrite_corrupted) {
1202 *nperm |= BLK_PERM_WRITE;
1203 }
1204
1205
1206
1207
1208
1209 *nshared = (shared & (BLK_PERM_CONSISTENT_READ |
1210 BLK_PERM_WRITE_UNCHANGED))
1211 | DEFAULT_PERM_UNCHANGED;
1212}
1213
1214
1215
1216
1217
1218
1219
1220static int coroutine_fn quorum_co_block_status(BlockDriverState *bs,
1221 bool want_zero,
1222 int64_t offset, int64_t count,
1223 int64_t *pnum, int64_t *map,
1224 BlockDriverState **file)
1225{
1226 BDRVQuorumState *s = bs->opaque;
1227 int i, ret;
1228 int64_t pnum_zero = count;
1229 int64_t pnum_data = 0;
1230
1231 for (i = 0; i < s->num_children; i++) {
1232 int64_t bytes;
1233 ret = bdrv_co_common_block_status_above(s->children[i]->bs, NULL, false,
1234 want_zero, offset, count,
1235 &bytes, NULL, NULL, NULL);
1236 if (ret < 0) {
1237 quorum_report_bad(QUORUM_OP_TYPE_READ, offset, count,
1238 s->children[i]->bs->node_name, ret);
1239 pnum_data = count;
1240 break;
1241 }
1242
1243
1244
1245
1246
1247
1248 if (ret & BDRV_BLOCK_ZERO) {
1249 pnum_zero = MIN(pnum_zero, bytes);
1250 } else {
1251 pnum_data = MAX(pnum_data, bytes);
1252 }
1253 }
1254
1255 if (pnum_data) {
1256 *pnum = pnum_data;
1257 return BDRV_BLOCK_DATA;
1258 } else {
1259 *pnum = pnum_zero;
1260 return BDRV_BLOCK_ZERO;
1261 }
1262}
1263
1264static const char *const quorum_strong_runtime_opts[] = {
1265 QUORUM_OPT_VOTE_THRESHOLD,
1266 QUORUM_OPT_BLKVERIFY,
1267 QUORUM_OPT_REWRITE,
1268 QUORUM_OPT_READ_PATTERN,
1269
1270 NULL
1271};
1272
1273static BlockDriver bdrv_quorum = {
1274 .format_name = "quorum",
1275
1276 .instance_size = sizeof(BDRVQuorumState),
1277
1278 .bdrv_open = quorum_open,
1279 .bdrv_close = quorum_close,
1280 .bdrv_gather_child_options = quorum_gather_child_options,
1281 .bdrv_dirname = quorum_dirname,
1282 .bdrv_co_block_status = quorum_co_block_status,
1283
1284 .bdrv_co_flush = quorum_co_flush,
1285
1286 .bdrv_getlength = quorum_getlength,
1287
1288 .bdrv_co_preadv = quorum_co_preadv,
1289 .bdrv_co_pwritev = quorum_co_pwritev,
1290 .bdrv_co_pwrite_zeroes = quorum_co_pwrite_zeroes,
1291
1292 .bdrv_add_child = quorum_add_child,
1293 .bdrv_del_child = quorum_del_child,
1294
1295 .bdrv_child_perm = quorum_child_perm,
1296
1297 .bdrv_recurse_can_replace = quorum_recurse_can_replace,
1298
1299 .strong_runtime_opts = quorum_strong_runtime_opts,
1300};
1301
1302static void bdrv_quorum_init(void)
1303{
1304 if (!qcrypto_hash_supports(QCRYPTO_HASH_ALG_SHA256)) {
1305
1306 return;
1307 }
1308 bdrv_register(&bdrv_quorum);
1309}
1310
1311block_init(bdrv_quorum_init);
1312