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