block: honor BDRV_O_ALLOW_RDWR when clearing bs->read_only
[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 "qemu/osdep.h"
17 #include "qemu/cutils.h"
18 #include "block/block_int.h"
19 #include "qapi/qmp/qbool.h"
20 #include "qapi/qmp/qdict.h"
21 #include "qapi/qmp/qerror.h"
22 #include "qapi/qmp/qint.h"
23 #include "qapi/qmp/qjson.h"
24 #include "qapi/qmp/qlist.h"
25 #include "qapi/qmp/qstring.h"
26 #include "qapi-event.h"
27 #include "crypto/hash.h"
28
29 #define HASH_LENGTH 32
30
31 #define QUORUM_OPT_VOTE_THRESHOLD "vote-threshold"
32 #define QUORUM_OPT_BLKVERIFY "blkverify"
33 #define QUORUM_OPT_REWRITE "rewrite-corrupted"
34 #define QUORUM_OPT_READ_PATTERN "read-pattern"
35
36 /* This union holds a vote hash value */
37 typedef union QuorumVoteValue {
38 uint8_t h[HASH_LENGTH]; /* SHA-256 hash */
39 int64_t l; /* simpler 64 bits hash */
40 } QuorumVoteValue;
41
42 /* A vote item */
43 typedef struct QuorumVoteItem {
44 int index;
45 QLIST_ENTRY(QuorumVoteItem) next;
46 } QuorumVoteItem;
47
48 /* this structure is a vote version. A version is the set of votes sharing the
49 * same vote value.
50 * The set of votes will be tracked with the items field and its cardinality is
51 * vote_count.
52 */
53 typedef struct QuorumVoteVersion {
54 QuorumVoteValue value;
55 int index;
56 int vote_count;
57 QLIST_HEAD(, QuorumVoteItem) items;
58 QLIST_ENTRY(QuorumVoteVersion) next;
59 } QuorumVoteVersion;
60
61 /* this structure holds a group of vote versions together */
62 typedef struct QuorumVotes {
63 QLIST_HEAD(, QuorumVoteVersion) vote_list;
64 bool (*compare)(QuorumVoteValue *a, QuorumVoteValue *b);
65 } QuorumVotes;
66
67 /* the following structure holds the state of one quorum instance */
68 typedef struct BDRVQuorumState {
69 BdrvChild **children; /* children BlockDriverStates */
70 int num_children; /* children count */
71 unsigned next_child_index; /* the index of the next child that should
72 * be added
73 */
74 int threshold; /* if less than threshold children reads gave the
75 * same result a quorum error occurs.
76 */
77 bool is_blkverify; /* true if the driver is in blkverify mode
78 * Writes are mirrored on two children devices.
79 * On reads the two children devices' contents are
80 * compared and if a difference is spotted its
81 * location is printed and the code aborts.
82 * It is useful to debug other block drivers by
83 * comparing them with a reference one.
84 */
85 bool rewrite_corrupted;/* true if the driver must rewrite-on-read corrupted
86 * block if Quorum is reached.
87 */
88
89 QuorumReadPattern read_pattern;
90 } BDRVQuorumState;
91
92 typedef struct QuorumAIOCB QuorumAIOCB;
93
94 /* Quorum will create one instance of the following structure per operation it
95 * performs on its children.
96 * So for each read/write operation coming from the upper layer there will be
97 * $children_count QuorumChildRequest.
98 */
99 typedef struct QuorumChildRequest {
100 BlockDriverState *bs;
101 QEMUIOVector qiov;
102 uint8_t *buf;
103 int ret;
104 QuorumAIOCB *parent;
105 } QuorumChildRequest;
106
107 /* Quorum will use the following structure to track progress of each read/write
108 * operation received by the upper layer.
109 * This structure hold pointers to the QuorumChildRequest structures instances
110 * used to do operations on each children and track overall progress.
111 */
112 struct QuorumAIOCB {
113 BlockDriverState *bs;
114 Coroutine *co;
115
116 /* Request metadata */
117 uint64_t offset;
118 uint64_t bytes;
119
120 QEMUIOVector *qiov; /* calling IOV */
121
122 QuorumChildRequest *qcrs; /* individual child requests */
123 int count; /* number of completed AIOCB */
124 int success_count; /* number of successfully completed AIOCB */
125
126 int rewrite_count; /* number of replica to rewrite: count down to
127 * zero once writes are fired
128 */
129
130 QuorumVotes votes;
131
132 bool is_read;
133 int vote_ret;
134 int children_read; /* how many children have been read from */
135 };
136
137 typedef struct QuorumCo {
138 QuorumAIOCB *acb;
139 int idx;
140 } QuorumCo;
141
142 static void quorum_aio_finalize(QuorumAIOCB *acb)
143 {
144 g_free(acb->qcrs);
145 g_free(acb);
146 }
147
148 static bool quorum_sha256_compare(QuorumVoteValue *a, QuorumVoteValue *b)
149 {
150 return !memcmp(a->h, b->h, HASH_LENGTH);
151 }
152
153 static bool quorum_64bits_compare(QuorumVoteValue *a, QuorumVoteValue *b)
154 {
155 return a->l == b->l;
156 }
157
158 static QuorumAIOCB *quorum_aio_get(BlockDriverState *bs,
159 QEMUIOVector *qiov,
160 uint64_t offset,
161 uint64_t bytes)
162 {
163 BDRVQuorumState *s = bs->opaque;
164 QuorumAIOCB *acb = g_new(QuorumAIOCB, 1);
165 int i;
166
167 *acb = (QuorumAIOCB) {
168 .co = qemu_coroutine_self(),
169 .bs = bs,
170 .offset = offset,
171 .bytes = bytes,
172 .qiov = qiov,
173 .votes.compare = quorum_sha256_compare,
174 .votes.vote_list = QLIST_HEAD_INITIALIZER(acb.votes.vote_list),
175 };
176
177 acb->qcrs = g_new0(QuorumChildRequest, s->num_children);
178 for (i = 0; i < s->num_children; i++) {
179 acb->qcrs[i].buf = NULL;
180 acb->qcrs[i].ret = 0;
181 acb->qcrs[i].parent = acb;
182 }
183
184 return acb;
185 }
186
187 static void quorum_report_bad(QuorumOpType type, uint64_t offset,
188 uint64_t bytes, char *node_name, int ret)
189 {
190 const char *msg = NULL;
191 int64_t start_sector = offset / BDRV_SECTOR_SIZE;
192 int64_t end_sector = DIV_ROUND_UP(offset + bytes, BDRV_SECTOR_SIZE);
193
194 if (ret < 0) {
195 msg = strerror(-ret);
196 }
197
198 qapi_event_send_quorum_report_bad(type, !!msg, msg, node_name, start_sector,
199 end_sector - start_sector, &error_abort);
200 }
201
202 static void quorum_report_failure(QuorumAIOCB *acb)
203 {
204 const char *reference = bdrv_get_device_or_node_name(acb->bs);
205 int64_t start_sector = acb->offset / BDRV_SECTOR_SIZE;
206 int64_t end_sector = DIV_ROUND_UP(acb->offset + acb->bytes,
207 BDRV_SECTOR_SIZE);
208
209 qapi_event_send_quorum_failure(reference, start_sector,
210 end_sector - start_sector, &error_abort);
211 }
212
213 static int quorum_vote_error(QuorumAIOCB *acb);
214
215 static bool quorum_has_too_much_io_failed(QuorumAIOCB *acb)
216 {
217 BDRVQuorumState *s = acb->bs->opaque;
218
219 if (acb->success_count < s->threshold) {
220 acb->vote_ret = quorum_vote_error(acb);
221 quorum_report_failure(acb);
222 return true;
223 }
224
225 return false;
226 }
227
228 static int read_fifo_child(QuorumAIOCB *acb);
229
230 static void quorum_copy_qiov(QEMUIOVector *dest, QEMUIOVector *source)
231 {
232 int i;
233 assert(dest->niov == source->niov);
234 assert(dest->size == source->size);
235 for (i = 0; i < source->niov; i++) {
236 assert(dest->iov[i].iov_len == source->iov[i].iov_len);
237 memcpy(dest->iov[i].iov_base,
238 source->iov[i].iov_base,
239 source->iov[i].iov_len);
240 }
241 }
242
243 static void quorum_report_bad_acb(QuorumChildRequest *sacb, int ret)
244 {
245 QuorumAIOCB *acb = sacb->parent;
246 QuorumOpType type = acb->is_read ? QUORUM_OP_TYPE_READ : QUORUM_OP_TYPE_WRITE;
247 quorum_report_bad(type, acb->offset, acb->bytes, sacb->bs->node_name, ret);
248 }
249
250 static void quorum_report_bad_versions(BDRVQuorumState *s,
251 QuorumAIOCB *acb,
252 QuorumVoteValue *value)
253 {
254 QuorumVoteVersion *version;
255 QuorumVoteItem *item;
256
257 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
258 if (acb->votes.compare(&version->value, value)) {
259 continue;
260 }
261 QLIST_FOREACH(item, &version->items, next) {
262 quorum_report_bad(QUORUM_OP_TYPE_READ, acb->offset, acb->bytes,
263 s->children[item->index]->bs->node_name, 0);
264 }
265 }
266 }
267
268 static void quorum_rewrite_entry(void *opaque)
269 {
270 QuorumCo *co = opaque;
271 QuorumAIOCB *acb = co->acb;
272 BDRVQuorumState *s = acb->bs->opaque;
273
274 /* Ignore any errors, it's just a correction attempt for already
275 * corrupted data. */
276 bdrv_co_pwritev(s->children[co->idx], acb->offset, acb->bytes,
277 acb->qiov, 0);
278
279 /* Wake up the caller after the last rewrite */
280 acb->rewrite_count--;
281 if (!acb->rewrite_count) {
282 qemu_coroutine_enter_if_inactive(acb->co);
283 }
284 }
285
286 static bool quorum_rewrite_bad_versions(QuorumAIOCB *acb,
287 QuorumVoteValue *value)
288 {
289 QuorumVoteVersion *version;
290 QuorumVoteItem *item;
291 int count = 0;
292
293 /* first count the number of bad versions: done first to avoid concurrency
294 * issues.
295 */
296 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
297 if (acb->votes.compare(&version->value, value)) {
298 continue;
299 }
300 QLIST_FOREACH(item, &version->items, next) {
301 count++;
302 }
303 }
304
305 /* quorum_rewrite_entry will count down this to zero */
306 acb->rewrite_count = count;
307
308 /* now fire the correcting rewrites */
309 QLIST_FOREACH(version, &acb->votes.vote_list, next) {
310 if (acb->votes.compare(&version->value, value)) {
311 continue;
312 }
313 QLIST_FOREACH(item, &version->items, next) {
314 Coroutine *co;
315 QuorumCo data = {
316 .acb = acb,
317 .idx = item->index,
318 };
319
320 co = qemu_coroutine_create(quorum_rewrite_entry, &data);
321 qemu_coroutine_enter(co);
322 }
323 }
324
325 /* return true if any rewrite is done else false */
326 return count;
327 }
328
329 static void quorum_count_vote(QuorumVotes *votes,
330 QuorumVoteValue *value,
331 int index)
332 {
333 QuorumVoteVersion *v = NULL, *version = NULL;
334 QuorumVoteItem *item;
335
336 /* look if we have something with this hash */
337 QLIST_FOREACH(v, &votes->vote_list, next) {
338 if (votes->compare(&v->value, value)) {
339 version = v;
340 break;
341 }
342 }
343
344 /* It's a version not yet in the list add it */
345 if (!version) {
346 version = g_new0(QuorumVoteVersion, 1);
347 QLIST_INIT(&version->items);
348 memcpy(&version->value, value, sizeof(version->value));
349 version->index = index;
350 version->vote_count = 0;
351 QLIST_INSERT_HEAD(&votes->vote_list, version, next);
352 }
353
354 version->vote_count++;
355
356 item = g_new0(QuorumVoteItem, 1);
357 item->index = index;
358 QLIST_INSERT_HEAD(&version->items, item, next);
359 }
360
361 static void quorum_free_vote_list(QuorumVotes *votes)
362 {
363 QuorumVoteVersion *version, *next_version;
364 QuorumVoteItem *item, *next_item;
365
366 QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
367 QLIST_REMOVE(version, next);
368 QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
369 QLIST_REMOVE(item, next);
370 g_free(item);
371 }
372 g_free(version);
373 }
374 }
375
376 static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
377 {
378 QEMUIOVector *qiov = &acb->qcrs[i].qiov;
379 size_t len = sizeof(hash->h);
380 uint8_t *data = hash->h;
381
382 /* XXX - would be nice if we could pass in the Error **
383 * and propagate that back, but this quorum code is
384 * restricted to just errno values currently */
385 if (qcrypto_hash_bytesv(QCRYPTO_HASH_ALG_SHA256,
386 qiov->iov, qiov->niov,
387 &data, &len,
388 NULL) < 0) {
389 return -EINVAL;
390 }
391
392 return 0;
393 }
394
395 static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
396 {
397 int max = 0;
398 QuorumVoteVersion *candidate, *winner = NULL;
399
400 QLIST_FOREACH(candidate, &votes->vote_list, next) {
401 if (candidate->vote_count > max) {
402 max = candidate->vote_count;
403 winner = candidate;
404 }
405 }
406
407 return winner;
408 }
409
410 /* qemu_iovec_compare is handy for blkverify mode because it returns the first
411 * differing byte location. Yet it is handcoded to compare vectors one byte
412 * after another so it does not benefit from the libc SIMD optimizations.
413 * quorum_iovec_compare is written for speed and should be used in the non
414 * blkverify mode of quorum.
415 */
416 static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
417 {
418 int i;
419 int result;
420
421 assert(a->niov == b->niov);
422 for (i = 0; i < a->niov; i++) {
423 assert(a->iov[i].iov_len == b->iov[i].iov_len);
424 result = memcmp(a->iov[i].iov_base,
425 b->iov[i].iov_base,
426 a->iov[i].iov_len);
427 if (result) {
428 return false;
429 }
430 }
431
432 return true;
433 }
434
435 static void GCC_FMT_ATTR(2, 3) quorum_err(QuorumAIOCB *acb,
436 const char *fmt, ...)
437 {
438 va_list ap;
439
440 va_start(ap, fmt);
441 fprintf(stderr, "quorum: offset=%" PRIu64 " bytes=%" PRIu64 " ",
442 acb->offset, acb->bytes);
443 vfprintf(stderr, fmt, ap);
444 fprintf(stderr, "\n");
445 va_end(ap);
446 exit(1);
447 }
448
449 static bool quorum_compare(QuorumAIOCB *acb,
450 QEMUIOVector *a,
451 QEMUIOVector *b)
452 {
453 BDRVQuorumState *s = acb->bs->opaque;
454 ssize_t offset;
455
456 /* This driver will replace blkverify in this particular case */
457 if (s->is_blkverify) {
458 offset = qemu_iovec_compare(a, b);
459 if (offset != -1) {
460 quorum_err(acb, "contents mismatch at offset %" PRIu64,
461 acb->offset + offset);
462 }
463 return true;
464 }
465
466 return quorum_iovec_compare(a, b);
467 }
468
469 /* Do a vote to get the error code */
470 static int quorum_vote_error(QuorumAIOCB *acb)
471 {
472 BDRVQuorumState *s = acb->bs->opaque;
473 QuorumVoteVersion *winner = NULL;
474 QuorumVotes error_votes;
475 QuorumVoteValue result_value;
476 int i, ret = 0;
477 bool error = false;
478
479 QLIST_INIT(&error_votes.vote_list);
480 error_votes.compare = quorum_64bits_compare;
481
482 for (i = 0; i < s->num_children; i++) {
483 ret = acb->qcrs[i].ret;
484 if (ret) {
485 error = true;
486 result_value.l = ret;
487 quorum_count_vote(&error_votes, &result_value, i);
488 }
489 }
490
491 if (error) {
492 winner = quorum_get_vote_winner(&error_votes);
493 ret = winner->value.l;
494 }
495
496 quorum_free_vote_list(&error_votes);
497
498 return ret;
499 }
500
501 static void quorum_vote(QuorumAIOCB *acb)
502 {
503 bool quorum = true;
504 int i, j, ret;
505 QuorumVoteValue hash;
506 BDRVQuorumState *s = acb->bs->opaque;
507 QuorumVoteVersion *winner;
508
509 if (quorum_has_too_much_io_failed(acb)) {
510 return;
511 }
512
513 /* get the index of the first successful read */
514 for (i = 0; i < s->num_children; i++) {
515 if (!acb->qcrs[i].ret) {
516 break;
517 }
518 }
519
520 assert(i < s->num_children);
521
522 /* compare this read with all other successful reads stopping at quorum
523 * failure
524 */
525 for (j = i + 1; j < s->num_children; j++) {
526 if (acb->qcrs[j].ret) {
527 continue;
528 }
529 quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
530 if (!quorum) {
531 break;
532 }
533 }
534
535 /* Every successful read agrees */
536 if (quorum) {
537 quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
538 return;
539 }
540
541 /* compute hashes for each successful read, also store indexes */
542 for (i = 0; i < s->num_children; i++) {
543 if (acb->qcrs[i].ret) {
544 continue;
545 }
546 ret = quorum_compute_hash(acb, i, &hash);
547 /* if ever the hash computation failed */
548 if (ret < 0) {
549 acb->vote_ret = ret;
550 goto free_exit;
551 }
552 quorum_count_vote(&acb->votes, &hash, i);
553 }
554
555 /* vote to select the most represented version */
556 winner = quorum_get_vote_winner(&acb->votes);
557
558 /* if the winner count is smaller than threshold the read fails */
559 if (winner->vote_count < s->threshold) {
560 quorum_report_failure(acb);
561 acb->vote_ret = -EIO;
562 goto free_exit;
563 }
564
565 /* we have a winner: copy it */
566 quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
567
568 /* some versions are bad print them */
569 quorum_report_bad_versions(s, acb, &winner->value);
570
571 /* corruption correction is enabled */
572 if (s->rewrite_corrupted) {
573 quorum_rewrite_bad_versions(acb, &winner->value);
574 }
575
576 free_exit:
577 /* free lists */
578 quorum_free_vote_list(&acb->votes);
579 }
580
581 static void read_quorum_children_entry(void *opaque)
582 {
583 QuorumCo *co = opaque;
584 QuorumAIOCB *acb = co->acb;
585 BDRVQuorumState *s = acb->bs->opaque;
586 int i = co->idx;
587 QuorumChildRequest *sacb = &acb->qcrs[i];
588
589 sacb->bs = s->children[i]->bs;
590 sacb->ret = bdrv_co_preadv(s->children[i], acb->offset, acb->bytes,
591 &acb->qcrs[i].qiov, 0);
592
593 if (sacb->ret == 0) {
594 acb->success_count++;
595 } else {
596 quorum_report_bad_acb(sacb, sacb->ret);
597 }
598
599 acb->count++;
600 assert(acb->count <= s->num_children);
601 assert(acb->success_count <= s->num_children);
602
603 /* Wake up the caller after the last read */
604 if (acb->count == s->num_children) {
605 qemu_coroutine_enter_if_inactive(acb->co);
606 }
607 }
608
609 static int read_quorum_children(QuorumAIOCB *acb)
610 {
611 BDRVQuorumState *s = acb->bs->opaque;
612 int i, ret;
613
614 acb->children_read = s->num_children;
615 for (i = 0; i < s->num_children; i++) {
616 acb->qcrs[i].buf = qemu_blockalign(s->children[i]->bs, acb->qiov->size);
617 qemu_iovec_init(&acb->qcrs[i].qiov, acb->qiov->niov);
618 qemu_iovec_clone(&acb->qcrs[i].qiov, acb->qiov, acb->qcrs[i].buf);
619 }
620
621 for (i = 0; i < s->num_children; i++) {
622 Coroutine *co;
623 QuorumCo data = {
624 .acb = acb,
625 .idx = i,
626 };
627
628 co = qemu_coroutine_create(read_quorum_children_entry, &data);
629 qemu_coroutine_enter(co);
630 }
631
632 while (acb->count < s->num_children) {
633 qemu_coroutine_yield();
634 }
635
636 /* Do the vote on read */
637 quorum_vote(acb);
638 for (i = 0; i < s->num_children; i++) {
639 qemu_vfree(acb->qcrs[i].buf);
640 qemu_iovec_destroy(&acb->qcrs[i].qiov);
641 }
642
643 while (acb->rewrite_count) {
644 qemu_coroutine_yield();
645 }
646
647 ret = acb->vote_ret;
648
649 return ret;
650 }
651
652 static int read_fifo_child(QuorumAIOCB *acb)
653 {
654 BDRVQuorumState *s = acb->bs->opaque;
655 int n, ret;
656
657 /* We try to read the next child in FIFO order if we failed to read */
658 do {
659 n = acb->children_read++;
660 acb->qcrs[n].bs = s->children[n]->bs;
661 ret = bdrv_co_preadv(s->children[n], acb->offset, acb->bytes,
662 acb->qiov, 0);
663 if (ret < 0) {
664 quorum_report_bad_acb(&acb->qcrs[n], ret);
665 }
666 } while (ret < 0 && acb->children_read < s->num_children);
667
668 /* FIXME: rewrite failed children if acb->children_read > 1? */
669
670 return ret;
671 }
672
673 static int quorum_co_preadv(BlockDriverState *bs, uint64_t offset,
674 uint64_t bytes, QEMUIOVector *qiov, int flags)
675 {
676 BDRVQuorumState *s = bs->opaque;
677 QuorumAIOCB *acb = quorum_aio_get(bs, qiov, offset, bytes);
678 int ret;
679
680 acb->is_read = true;
681 acb->children_read = 0;
682
683 if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
684 ret = read_quorum_children(acb);
685 } else {
686 ret = read_fifo_child(acb);
687 }
688 quorum_aio_finalize(acb);
689
690 return ret;
691 }
692
693 static void write_quorum_entry(void *opaque)
694 {
695 QuorumCo *co = opaque;
696 QuorumAIOCB *acb = co->acb;
697 BDRVQuorumState *s = acb->bs->opaque;
698 int i = co->idx;
699 QuorumChildRequest *sacb = &acb->qcrs[i];
700
701 sacb->bs = s->children[i]->bs;
702 sacb->ret = bdrv_co_pwritev(s->children[i], acb->offset, acb->bytes,
703 acb->qiov, 0);
704 if (sacb->ret == 0) {
705 acb->success_count++;
706 } else {
707 quorum_report_bad_acb(sacb, sacb->ret);
708 }
709 acb->count++;
710 assert(acb->count <= s->num_children);
711 assert(acb->success_count <= s->num_children);
712
713 /* Wake up the caller after the last write */
714 if (acb->count == s->num_children) {
715 qemu_coroutine_enter_if_inactive(acb->co);
716 }
717 }
718
719 static int quorum_co_pwritev(BlockDriverState *bs, uint64_t offset,
720 uint64_t bytes, QEMUIOVector *qiov, int flags)
721 {
722 BDRVQuorumState *s = bs->opaque;
723 QuorumAIOCB *acb = quorum_aio_get(bs, qiov, offset, bytes);
724 int i, ret;
725
726 for (i = 0; i < s->num_children; i++) {
727 Coroutine *co;
728 QuorumCo data = {
729 .acb = acb,
730 .idx = i,
731 };
732
733 co = qemu_coroutine_create(write_quorum_entry, &data);
734 qemu_coroutine_enter(co);
735 }
736
737 while (acb->count < s->num_children) {
738 qemu_coroutine_yield();
739 }
740
741 quorum_has_too_much_io_failed(acb);
742
743 ret = acb->vote_ret;
744 quorum_aio_finalize(acb);
745
746 return ret;
747 }
748
749 static int64_t quorum_getlength(BlockDriverState *bs)
750 {
751 BDRVQuorumState *s = bs->opaque;
752 int64_t result;
753 int i;
754
755 /* check that all file have the same length */
756 result = bdrv_getlength(s->children[0]->bs);
757 if (result < 0) {
758 return result;
759 }
760 for (i = 1; i < s->num_children; i++) {
761 int64_t value = bdrv_getlength(s->children[i]->bs);
762 if (value < 0) {
763 return value;
764 }
765 if (value != result) {
766 return -EIO;
767 }
768 }
769
770 return result;
771 }
772
773 static coroutine_fn int quorum_co_flush(BlockDriverState *bs)
774 {
775 BDRVQuorumState *s = bs->opaque;
776 QuorumVoteVersion *winner = NULL;
777 QuorumVotes error_votes;
778 QuorumVoteValue result_value;
779 int i;
780 int result = 0;
781 int success_count = 0;
782
783 QLIST_INIT(&error_votes.vote_list);
784 error_votes.compare = quorum_64bits_compare;
785
786 for (i = 0; i < s->num_children; i++) {
787 result = bdrv_co_flush(s->children[i]->bs);
788 if (result) {
789 quorum_report_bad(QUORUM_OP_TYPE_FLUSH, 0,
790 bdrv_getlength(s->children[i]->bs),
791 s->children[i]->bs->node_name, result);
792 result_value.l = result;
793 quorum_count_vote(&error_votes, &result_value, i);
794 } else {
795 success_count++;
796 }
797 }
798
799 if (success_count >= s->threshold) {
800 result = 0;
801 } else {
802 winner = quorum_get_vote_winner(&error_votes);
803 result = winner->value.l;
804 }
805 quorum_free_vote_list(&error_votes);
806
807 return result;
808 }
809
810 static bool quorum_recurse_is_first_non_filter(BlockDriverState *bs,
811 BlockDriverState *candidate)
812 {
813 BDRVQuorumState *s = bs->opaque;
814 int i;
815
816 for (i = 0; i < s->num_children; i++) {
817 bool perm = bdrv_recurse_is_first_non_filter(s->children[i]->bs,
818 candidate);
819 if (perm) {
820 return true;
821 }
822 }
823
824 return false;
825 }
826
827 static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
828 {
829
830 if (threshold < 1) {
831 error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
832 "vote-threshold", "value >= 1");
833 return -ERANGE;
834 }
835
836 if (threshold > num_children) {
837 error_setg(errp, "threshold may not exceed children count");
838 return -ERANGE;
839 }
840
841 return 0;
842 }
843
844 static QemuOptsList quorum_runtime_opts = {
845 .name = "quorum",
846 .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
847 .desc = {
848 {
849 .name = QUORUM_OPT_VOTE_THRESHOLD,
850 .type = QEMU_OPT_NUMBER,
851 .help = "The number of vote needed for reaching quorum",
852 },
853 {
854 .name = QUORUM_OPT_BLKVERIFY,
855 .type = QEMU_OPT_BOOL,
856 .help = "Trigger block verify mode if set",
857 },
858 {
859 .name = QUORUM_OPT_REWRITE,
860 .type = QEMU_OPT_BOOL,
861 .help = "Rewrite corrupted block on read quorum",
862 },
863 {
864 .name = QUORUM_OPT_READ_PATTERN,
865 .type = QEMU_OPT_STRING,
866 .help = "Allowed pattern: quorum, fifo. Quorum is default",
867 },
868 { /* end of list */ }
869 },
870 };
871
872 static int parse_read_pattern(const char *opt)
873 {
874 int i;
875
876 if (!opt) {
877 /* Set quorum as default */
878 return QUORUM_READ_PATTERN_QUORUM;
879 }
880
881 for (i = 0; i < QUORUM_READ_PATTERN__MAX; i++) {
882 if (!strcmp(opt, QuorumReadPattern_lookup[i])) {
883 return i;
884 }
885 }
886
887 return -EINVAL;
888 }
889
890 static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
891 Error **errp)
892 {
893 BDRVQuorumState *s = bs->opaque;
894 Error *local_err = NULL;
895 QemuOpts *opts = NULL;
896 bool *opened;
897 int i;
898 int ret = 0;
899
900 qdict_flatten(options);
901
902 /* count how many different children are present */
903 s->num_children = qdict_array_entries(options, "children.");
904 if (s->num_children < 0) {
905 error_setg(&local_err, "Option children is not a valid array");
906 ret = -EINVAL;
907 goto exit;
908 }
909 if (s->num_children < 1) {
910 error_setg(&local_err,
911 "Number of provided children must be 1 or more");
912 ret = -EINVAL;
913 goto exit;
914 }
915
916 opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
917 qemu_opts_absorb_qdict(opts, options, &local_err);
918 if (local_err) {
919 ret = -EINVAL;
920 goto exit;
921 }
922
923 s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
924 /* and validate it against s->num_children */
925 ret = quorum_valid_threshold(s->threshold, s->num_children, &local_err);
926 if (ret < 0) {
927 goto exit;
928 }
929
930 ret = parse_read_pattern(qemu_opt_get(opts, QUORUM_OPT_READ_PATTERN));
931 if (ret < 0) {
932 error_setg(&local_err, "Please set read-pattern as fifo or quorum");
933 goto exit;
934 }
935 s->read_pattern = ret;
936
937 if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
938 /* is the driver in blkverify mode */
939 if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false) &&
940 s->num_children == 2 && s->threshold == 2) {
941 s->is_blkverify = true;
942 } else if (qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false)) {
943 fprintf(stderr, "blkverify mode is set by setting blkverify=on "
944 "and using two files with vote_threshold=2\n");
945 }
946
947 s->rewrite_corrupted = qemu_opt_get_bool(opts, QUORUM_OPT_REWRITE,
948 false);
949 if (s->rewrite_corrupted && s->is_blkverify) {
950 error_setg(&local_err,
951 "rewrite-corrupted=on cannot be used with blkverify=on");
952 ret = -EINVAL;
953 goto exit;
954 }
955 }
956
957 /* allocate the children array */
958 s->children = g_new0(BdrvChild *, s->num_children);
959 opened = g_new0(bool, s->num_children);
960
961 for (i = 0; i < s->num_children; i++) {
962 char indexstr[32];
963 ret = snprintf(indexstr, 32, "children.%d", i);
964 assert(ret < 32);
965
966 s->children[i] = bdrv_open_child(NULL, options, indexstr, bs,
967 &child_format, false, &local_err);
968 if (local_err) {
969 ret = -EINVAL;
970 goto close_exit;
971 }
972
973 opened[i] = true;
974 }
975 s->next_child_index = s->num_children;
976
977 g_free(opened);
978 goto exit;
979
980 close_exit:
981 /* cleanup on error */
982 for (i = 0; i < s->num_children; i++) {
983 if (!opened[i]) {
984 continue;
985 }
986 bdrv_unref_child(bs, s->children[i]);
987 }
988 g_free(s->children);
989 g_free(opened);
990 exit:
991 qemu_opts_del(opts);
992 /* propagate error */
993 error_propagate(errp, local_err);
994 return ret;
995 }
996
997 static void quorum_close(BlockDriverState *bs)
998 {
999 BDRVQuorumState *s = bs->opaque;
1000 int i;
1001
1002 for (i = 0; i < s->num_children; i++) {
1003 bdrv_unref_child(bs, s->children[i]);
1004 }
1005
1006 g_free(s->children);
1007 }
1008
1009 static void quorum_add_child(BlockDriverState *bs, BlockDriverState *child_bs,
1010 Error **errp)
1011 {
1012 BDRVQuorumState *s = bs->opaque;
1013 BdrvChild *child;
1014 char indexstr[32];
1015 int ret;
1016
1017 assert(s->num_children <= INT_MAX / sizeof(BdrvChild *));
1018 if (s->num_children == INT_MAX / sizeof(BdrvChild *) ||
1019 s->next_child_index == UINT_MAX) {
1020 error_setg(errp, "Too many children");
1021 return;
1022 }
1023
1024 ret = snprintf(indexstr, 32, "children.%u", s->next_child_index);
1025 if (ret < 0 || ret >= 32) {
1026 error_setg(errp, "cannot generate child name");
1027 return;
1028 }
1029 s->next_child_index++;
1030
1031 bdrv_drained_begin(bs);
1032
1033 /* We can safely add the child now */
1034 bdrv_ref(child_bs);
1035
1036 child = bdrv_attach_child(bs, child_bs, indexstr, &child_format, errp);
1037 if (child == NULL) {
1038 s->next_child_index--;
1039 bdrv_unref(child_bs);
1040 goto out;
1041 }
1042 s->children = g_renew(BdrvChild *, s->children, s->num_children + 1);
1043 s->children[s->num_children++] = child;
1044
1045 out:
1046 bdrv_drained_end(bs);
1047 }
1048
1049 static void quorum_del_child(BlockDriverState *bs, BdrvChild *child,
1050 Error **errp)
1051 {
1052 BDRVQuorumState *s = bs->opaque;
1053 int i;
1054
1055 for (i = 0; i < s->num_children; i++) {
1056 if (s->children[i] == child) {
1057 break;
1058 }
1059 }
1060
1061 /* we have checked it in bdrv_del_child() */
1062 assert(i < s->num_children);
1063
1064 if (s->num_children <= s->threshold) {
1065 error_setg(errp,
1066 "The number of children cannot be lower than the vote threshold %d",
1067 s->threshold);
1068 return;
1069 }
1070
1071 bdrv_drained_begin(bs);
1072
1073 /* We can safely remove this child now */
1074 memmove(&s->children[i], &s->children[i + 1],
1075 (s->num_children - i - 1) * sizeof(BdrvChild *));
1076 s->children = g_renew(BdrvChild *, s->children, --s->num_children);
1077 bdrv_unref_child(bs, child);
1078
1079 bdrv_drained_end(bs);
1080 }
1081
1082 static void quorum_refresh_filename(BlockDriverState *bs, QDict *options)
1083 {
1084 BDRVQuorumState *s = bs->opaque;
1085 QDict *opts;
1086 QList *children;
1087 int i;
1088
1089 for (i = 0; i < s->num_children; i++) {
1090 bdrv_refresh_filename(s->children[i]->bs);
1091 if (!s->children[i]->bs->full_open_options) {
1092 return;
1093 }
1094 }
1095
1096 children = qlist_new();
1097 for (i = 0; i < s->num_children; i++) {
1098 QINCREF(s->children[i]->bs->full_open_options);
1099 qlist_append_obj(children,
1100 QOBJECT(s->children[i]->bs->full_open_options));
1101 }
1102
1103 opts = qdict_new();
1104 qdict_put_obj(opts, "driver", QOBJECT(qstring_from_str("quorum")));
1105 qdict_put_obj(opts, QUORUM_OPT_VOTE_THRESHOLD,
1106 QOBJECT(qint_from_int(s->threshold)));
1107 qdict_put_obj(opts, QUORUM_OPT_BLKVERIFY,
1108 QOBJECT(qbool_from_bool(s->is_blkverify)));
1109 qdict_put_obj(opts, QUORUM_OPT_REWRITE,
1110 QOBJECT(qbool_from_bool(s->rewrite_corrupted)));
1111 qdict_put_obj(opts, "children", QOBJECT(children));
1112
1113 bs->full_open_options = opts;
1114 }
1115
1116 static BlockDriver bdrv_quorum = {
1117 .format_name = "quorum",
1118 .protocol_name = "quorum",
1119
1120 .instance_size = sizeof(BDRVQuorumState),
1121
1122 .bdrv_file_open = quorum_open,
1123 .bdrv_close = quorum_close,
1124 .bdrv_refresh_filename = quorum_refresh_filename,
1125
1126 .bdrv_co_flush_to_disk = quorum_co_flush,
1127
1128 .bdrv_getlength = quorum_getlength,
1129
1130 .bdrv_co_preadv = quorum_co_preadv,
1131 .bdrv_co_pwritev = quorum_co_pwritev,
1132
1133 .bdrv_add_child = quorum_add_child,
1134 .bdrv_del_child = quorum_del_child,
1135
1136 .bdrv_child_perm = bdrv_filter_default_perms,
1137
1138 .is_filter = true,
1139 .bdrv_recurse_is_first_non_filter = quorum_recurse_is_first_non_filter,
1140 };
1141
1142 static void bdrv_quorum_init(void)
1143 {
1144 if (!qcrypto_hash_supports(QCRYPTO_HASH_ALG_SHA256)) {
1145 /* SHA256 hash support is required for quorum device */
1146 return;
1147 }
1148 bdrv_register(&bdrv_quorum);
1149 }
1150
1151 block_init(bdrv_quorum_init);