qemu

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 "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
/* This union holds a vote hash value */
42
typedef union QuorumVoteValue {
43
    uint8_t h[HASH_LENGTH];    /* SHA-256 hash */
44
    int64_t l;                 /* simpler 64 bits hash */
45
} QuorumVoteValue;
46

47
/* A vote item */
48
typedef struct QuorumVoteItem {
49
    int index;
50
    QLIST_ENTRY(QuorumVoteItem) next;
51
} QuorumVoteItem;
52

53
/* this structure is a vote version. A version is the set of votes sharing the
54
 * same vote value.
55
 * The set of votes will be tracked with the items field and its cardinality is
56
 * vote_count.
57
 */
58
typedef 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
/* this structure holds a group of vote versions together */
67
typedef struct QuorumVotes {
68
    QLIST_HEAD(, QuorumVoteVersion) vote_list;
69
    bool (*compare)(QuorumVoteValue *a, QuorumVoteValue *b);
70
} QuorumVotes;
71

72
/* the following structure holds the state of one quorum instance */
73
typedef struct BDRVQuorumState {
74
    BdrvChild **children;  /* children BlockDriverStates */
75
    int num_children;      /* children count */
76
    unsigned next_child_index;  /* the index of the next child that should
77
                                 * be added
78
                                 */
79
    int threshold;         /* if less than threshold children reads gave the
80
                            * same result a quorum error occurs.
81
                            */
82
    bool is_blkverify;     /* true if the driver is in blkverify mode
83
                            * Writes are mirrored on two children devices.
84
                            * On reads the two children devices' contents are
85
                            * compared and if a difference is spotted its
86
                            * location is printed and the code aborts.
87
                            * It is useful to debug other block drivers by
88
                            * comparing them with a reference one.
89
                            */
90
    bool rewrite_corrupted;/* true if the driver must rewrite-on-read corrupted
91
                            * block if Quorum is reached.
92
                            */
93

94
    QuorumReadPattern read_pattern;
95
} BDRVQuorumState;
96

97
typedef struct QuorumAIOCB QuorumAIOCB;
98

99
/* Quorum will create one instance of the following structure per operation it
100
 * performs on its children.
101
 * So for each read/write operation coming from the upper layer there will be
102
 * $children_count QuorumChildRequest.
103
 */
104
typedef struct QuorumChildRequest {
105
    BlockDriverState *bs;
106
    QEMUIOVector qiov;
107
    uint8_t *buf;
108
    int ret;
109
    QuorumAIOCB *parent;
110
} QuorumChildRequest;
111

112
/* Quorum will use the following structure to track progress of each read/write
113
 * operation received by the upper layer.
114
 * This structure hold pointers to the QuorumChildRequest structures instances
115
 * used to do operations on each children and track overall progress.
116
 */
117
struct QuorumAIOCB {
118
    BlockDriverState *bs;
119
    Coroutine *co;
120

121
    /* Request metadata */
122
    uint64_t offset;
123
    uint64_t bytes;
124
    int flags;
125

126
    QEMUIOVector *qiov;         /* calling IOV */
127

128
    QuorumChildRequest *qcrs;   /* individual child requests */
129
    int count;                  /* number of completed AIOCB */
130
    int success_count;          /* number of successfully completed AIOCB */
131

132
    int rewrite_count;          /* number of replica to rewrite: count down to
133
                                 * zero once writes are fired
134
                                 */
135

136
    QuorumVotes votes;
137

138
    bool is_read;
139
    int vote_ret;
140
    int children_read;          /* how many children have been read from */
141
};
142

143
typedef struct QuorumCo {
144
    QuorumAIOCB *acb;
145
    int idx;
146
} QuorumCo;
147

148
static void quorum_aio_finalize(QuorumAIOCB *acb)
149
{
150
    g_free(acb->qcrs);
151
    g_free(acb);
152
}
153

154
static bool quorum_sha256_compare(QuorumVoteValue *a, QuorumVoteValue *b)
155
{
156
    return !memcmp(a->h, b->h, HASH_LENGTH);
157
}
158

159
static bool quorum_64bits_compare(QuorumVoteValue *a, QuorumVoteValue *b)
160
{
161
    return a->l == b->l;
162
}
163

164
static 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

194
static 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, node_name, start_sector,
206
                                      end_sector - start_sector);
207
}
208

209
static void GRAPH_RDLOCK 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

220
static int quorum_vote_error(QuorumAIOCB *acb);
221

222
static bool GRAPH_RDLOCK 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

235
static 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

248
static 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

255
static 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

273
/*
274
 * This function can count as GRAPH_RDLOCK because read_quorum_children() holds
275
 * the graph lock and keeps it until this coroutine has terminated.
276
 */
277
static void coroutine_fn GRAPH_RDLOCK quorum_rewrite_entry(void *opaque)
278
{
279
    QuorumCo *co = opaque;
280
    QuorumAIOCB *acb = co->acb;
281
    BDRVQuorumState *s = acb->bs->opaque;
282

283
    /* Ignore any errors, it's just a correction attempt for already
284
     * corrupted data.
285
     * Mask out BDRV_REQ_WRITE_UNCHANGED because this overwrites the
286
     * area with different data from the other children. */
287
    bdrv_co_pwritev(s->children[co->idx], acb->offset, acb->bytes,
288
                    acb->qiov, acb->flags & ~BDRV_REQ_WRITE_UNCHANGED);
289

290
    /* Wake up the caller after the last rewrite */
291
    acb->rewrite_count--;
292
    if (!acb->rewrite_count) {
293
        qemu_coroutine_enter_if_inactive(acb->co);
294
    }
295
}
296

297
static bool coroutine_fn GRAPH_RDLOCK
298
quorum_rewrite_bad_versions(QuorumAIOCB *acb, QuorumVoteValue *value)
299
{
300
    QuorumVoteVersion *version;
301
    QuorumVoteItem *item;
302
    int count = 0;
303

304
    /* first count the number of bad versions: done first to avoid concurrency
305
     * issues.
306
     */
307
    QLIST_FOREACH(version, &acb->votes.vote_list, next) {
308
        if (acb->votes.compare(&version->value, value)) {
309
            continue;
310
        }
311
        QLIST_FOREACH(item, &version->items, next) {
312
            count++;
313
        }
314
    }
315

316
    /* quorum_rewrite_entry will count down this to zero */
317
    acb->rewrite_count = count;
318

319
    /* now fire the correcting rewrites */
320
    QLIST_FOREACH(version, &acb->votes.vote_list, next) {
321
        if (acb->votes.compare(&version->value, value)) {
322
            continue;
323
        }
324
        QLIST_FOREACH(item, &version->items, next) {
325
            Coroutine *co;
326
            QuorumCo data = {
327
                .acb = acb,
328
                .idx = item->index,
329
            };
330

331
            co = qemu_coroutine_create(quorum_rewrite_entry, &data);
332
            qemu_coroutine_enter(co);
333
        }
334
    }
335

336
    /* return true if any rewrite is done else false */
337
    return count;
338
}
339

340
static void quorum_count_vote(QuorumVotes *votes,
341
                              QuorumVoteValue *value,
342
                              int index)
343
{
344
    QuorumVoteVersion *v = NULL, *version = NULL;
345
    QuorumVoteItem *item;
346

347
    /* look if we have something with this hash */
348
    QLIST_FOREACH(v, &votes->vote_list, next) {
349
        if (votes->compare(&v->value, value)) {
350
            version = v;
351
            break;
352
        }
353
    }
354

355
    /* It's a version not yet in the list add it */
356
    if (!version) {
357
        version = g_new0(QuorumVoteVersion, 1);
358
        QLIST_INIT(&version->items);
359
        memcpy(&version->value, value, sizeof(version->value));
360
        version->index = index;
361
        version->vote_count = 0;
362
        QLIST_INSERT_HEAD(&votes->vote_list, version, next);
363
    }
364

365
    version->vote_count++;
366

367
    item = g_new0(QuorumVoteItem, 1);
368
    item->index = index;
369
    QLIST_INSERT_HEAD(&version->items, item, next);
370
}
371

372
static void quorum_free_vote_list(QuorumVotes *votes)
373
{
374
    QuorumVoteVersion *version, *next_version;
375
    QuorumVoteItem *item, *next_item;
376

377
    QLIST_FOREACH_SAFE(version, &votes->vote_list, next, next_version) {
378
        QLIST_REMOVE(version, next);
379
        QLIST_FOREACH_SAFE(item, &version->items, next, next_item) {
380
            QLIST_REMOVE(item, next);
381
            g_free(item);
382
        }
383
        g_free(version);
384
    }
385
}
386

387
static int quorum_compute_hash(QuorumAIOCB *acb, int i, QuorumVoteValue *hash)
388
{
389
    QEMUIOVector *qiov = &acb->qcrs[i].qiov;
390
    size_t len = sizeof(hash->h);
391
    uint8_t *data = hash->h;
392

393
    /* XXX - would be nice if we could pass in the Error **
394
     * and propagate that back, but this quorum code is
395
     * restricted to just errno values currently */
396
    if (qcrypto_hash_bytesv(QCRYPTO_HASH_ALG_SHA256,
397
                            qiov->iov, qiov->niov,
398
                            &data, &len,
399
                            NULL) < 0) {
400
        return -EINVAL;
401
    }
402

403
    return 0;
404
}
405

406
static QuorumVoteVersion *quorum_get_vote_winner(QuorumVotes *votes)
407
{
408
    int max = 0;
409
    QuorumVoteVersion *candidate, *winner = NULL;
410

411
    QLIST_FOREACH(candidate, &votes->vote_list, next) {
412
        if (candidate->vote_count > max) {
413
            max = candidate->vote_count;
414
            winner = candidate;
415
        }
416
    }
417

418
    return winner;
419
}
420

421
/* qemu_iovec_compare is handy for blkverify mode because it returns the first
422
 * differing byte location. Yet it is handcoded to compare vectors one byte
423
 * after another so it does not benefit from the libc SIMD optimizations.
424
 * quorum_iovec_compare is written for speed and should be used in the non
425
 * blkverify mode of quorum.
426
 */
427
static bool quorum_iovec_compare(QEMUIOVector *a, QEMUIOVector *b)
428
{
429
    int i;
430
    int result;
431

432
    assert(a->niov == b->niov);
433
    for (i = 0; i < a->niov; i++) {
434
        assert(a->iov[i].iov_len == b->iov[i].iov_len);
435
        result = memcmp(a->iov[i].iov_base,
436
                        b->iov[i].iov_base,
437
                        a->iov[i].iov_len);
438
        if (result) {
439
            return false;
440
        }
441
    }
442

443
    return true;
444
}
445

446
static bool quorum_compare(QuorumAIOCB *acb, QEMUIOVector *a, QEMUIOVector *b)
447
{
448
    BDRVQuorumState *s = acb->bs->opaque;
449
    ssize_t offset;
450

451
    /* This driver will replace blkverify in this particular case */
452
    if (s->is_blkverify) {
453
        offset = qemu_iovec_compare(a, b);
454
        if (offset != -1) {
455
            fprintf(stderr, "quorum: offset=%" PRIu64 " bytes=%" PRIu64
456
                    " contents mismatch at offset %" PRIu64 "\n",
457
                    acb->offset, acb->bytes, acb->offset + offset);
458
            exit(1);
459
        }
460
        return true;
461
    }
462

463
    return quorum_iovec_compare(a, b);
464
}
465

466
/* Do a vote to get the error code */
467
static int quorum_vote_error(QuorumAIOCB *acb)
468
{
469
    BDRVQuorumState *s = acb->bs->opaque;
470
    QuorumVoteVersion *winner = NULL;
471
    QuorumVotes error_votes;
472
    QuorumVoteValue result_value;
473
    int i, ret = 0;
474
    bool error = false;
475

476
    QLIST_INIT(&error_votes.vote_list);
477
    error_votes.compare = quorum_64bits_compare;
478

479
    for (i = 0; i < s->num_children; i++) {
480
        ret = acb->qcrs[i].ret;
481
        if (ret) {
482
            error = true;
483
            result_value.l = ret;
484
            quorum_count_vote(&error_votes, &result_value, i);
485
        }
486
    }
487

488
    if (error) {
489
        winner = quorum_get_vote_winner(&error_votes);
490
        ret = winner->value.l;
491
    }
492

493
    quorum_free_vote_list(&error_votes);
494

495
    return ret;
496
}
497

498
static void coroutine_fn GRAPH_RDLOCK quorum_vote(QuorumAIOCB *acb)
499
{
500
    bool quorum = true;
501
    int i, j, ret;
502
    QuorumVoteValue hash;
503
    BDRVQuorumState *s = acb->bs->opaque;
504
    QuorumVoteVersion *winner;
505

506
    if (quorum_has_too_much_io_failed(acb)) {
507
        return;
508
    }
509

510
    /* get the index of the first successful read */
511
    for (i = 0; i < s->num_children; i++) {
512
        if (!acb->qcrs[i].ret) {
513
            break;
514
        }
515
    }
516

517
    assert(i < s->num_children);
518

519
    /* compare this read with all other successful reads stopping at quorum
520
     * failure
521
     */
522
    for (j = i + 1; j < s->num_children; j++) {
523
        if (acb->qcrs[j].ret) {
524
            continue;
525
        }
526
        quorum = quorum_compare(acb, &acb->qcrs[i].qiov, &acb->qcrs[j].qiov);
527
        if (!quorum) {
528
            break;
529
       }
530
    }
531

532
    /* Every successful read agrees */
533
    if (quorum) {
534
        quorum_copy_qiov(acb->qiov, &acb->qcrs[i].qiov);
535
        return;
536
    }
537

538
    /* compute hashes for each successful read, also store indexes */
539
    for (i = 0; i < s->num_children; i++) {
540
        if (acb->qcrs[i].ret) {
541
            continue;
542
        }
543
        ret = quorum_compute_hash(acb, i, &hash);
544
        /* if ever the hash computation failed */
545
        if (ret < 0) {
546
            acb->vote_ret = ret;
547
            goto free_exit;
548
        }
549
        quorum_count_vote(&acb->votes, &hash, i);
550
    }
551

552
    /* vote to select the most represented version */
553
    winner = quorum_get_vote_winner(&acb->votes);
554

555
    /* if the winner count is smaller than threshold the read fails */
556
    if (winner->vote_count < s->threshold) {
557
        quorum_report_failure(acb);
558
        acb->vote_ret = -EIO;
559
        goto free_exit;
560
    }
561

562
    /* we have a winner: copy it */
563
    quorum_copy_qiov(acb->qiov, &acb->qcrs[winner->index].qiov);
564

565
    /* some versions are bad print them */
566
    quorum_report_bad_versions(s, acb, &winner->value);
567

568
    /* corruption correction is enabled */
569
    if (s->rewrite_corrupted) {
570
        quorum_rewrite_bad_versions(acb, &winner->value);
571
    }
572

573
free_exit:
574
    /* free lists */
575
    quorum_free_vote_list(&acb->votes);
576
}
577

578
/*
579
 * This function can count as GRAPH_RDLOCK because read_quorum_children() holds
580
 * the graph lock and keeps it until this coroutine has terminated.
581
 */
582
static void coroutine_fn GRAPH_RDLOCK read_quorum_children_entry(void *opaque)
583
{
584
    QuorumCo *co = opaque;
585
    QuorumAIOCB *acb = co->acb;
586
    BDRVQuorumState *s = acb->bs->opaque;
587
    int i = co->idx;
588
    QuorumChildRequest *sacb = &acb->qcrs[i];
589

590
    sacb->bs = s->children[i]->bs;
591
    sacb->ret = bdrv_co_preadv(s->children[i], acb->offset, acb->bytes,
592
                               &acb->qcrs[i].qiov, 0);
593

594
    if (sacb->ret == 0) {
595
        acb->success_count++;
596
    } else {
597
        quorum_report_bad_acb(sacb, sacb->ret);
598
    }
599

600
    acb->count++;
601
    assert(acb->count <= s->num_children);
602
    assert(acb->success_count <= s->num_children);
603

604
    /* Wake up the caller after the last read */
605
    if (acb->count == s->num_children) {
606
        qemu_coroutine_enter_if_inactive(acb->co);
607
    }
608
}
609

610
static int coroutine_fn GRAPH_RDLOCK read_quorum_children(QuorumAIOCB *acb)
611
{
612
    BDRVQuorumState *s = acb->bs->opaque;
613
    int i;
614

615
    acb->children_read = s->num_children;
616
    for (i = 0; i < s->num_children; i++) {
617
        acb->qcrs[i].buf = qemu_blockalign(s->children[i]->bs, acb->qiov->size);
618
        qemu_iovec_init(&acb->qcrs[i].qiov, acb->qiov->niov);
619
        qemu_iovec_clone(&acb->qcrs[i].qiov, acb->qiov, acb->qcrs[i].buf);
620
    }
621

622
    for (i = 0; i < s->num_children; i++) {
623
        Coroutine *co;
624
        QuorumCo data = {
625
            .acb = acb,
626
            .idx = i,
627
        };
628

629
        co = qemu_coroutine_create(read_quorum_children_entry, &data);
630
        qemu_coroutine_enter(co);
631
    }
632

633
    while (acb->count < s->num_children) {
634
        qemu_coroutine_yield();
635
    }
636

637
    /* Do the vote on read */
638
    quorum_vote(acb);
639
    for (i = 0; i < s->num_children; i++) {
640
        qemu_vfree(acb->qcrs[i].buf);
641
        qemu_iovec_destroy(&acb->qcrs[i].qiov);
642
    }
643

644
    while (acb->rewrite_count) {
645
        qemu_coroutine_yield();
646
    }
647

648
    return acb->vote_ret;
649
}
650

651
static int coroutine_fn GRAPH_RDLOCK read_fifo_child(QuorumAIOCB *acb)
652
{
653
    BDRVQuorumState *s = acb->bs->opaque;
654
    int n, ret;
655

656
    /* We try to read the next child in FIFO order if we failed to read */
657
    do {
658
        n = acb->children_read++;
659
        acb->qcrs[n].bs = s->children[n]->bs;
660
        ret = bdrv_co_preadv(s->children[n], acb->offset, acb->bytes,
661
                             acb->qiov, 0);
662
        if (ret < 0) {
663
            quorum_report_bad_acb(&acb->qcrs[n], ret);
664
        }
665
    } while (ret < 0 && acb->children_read < s->num_children);
666

667
    /* FIXME: rewrite failed children if acb->children_read > 1? */
668

669
    return ret;
670
}
671

672
static int coroutine_fn GRAPH_RDLOCK
673
quorum_co_preadv(BlockDriverState *bs, int64_t offset, int64_t bytes,
674
                 QEMUIOVector *qiov, BdrvRequestFlags flags)
675
{
676
    BDRVQuorumState *s = bs->opaque;
677
    QuorumAIOCB *acb = quorum_aio_get(bs, qiov, offset, bytes, flags);
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
/*
694
 * This function can count as GRAPH_RDLOCK because quorum_co_pwritev() holds the
695
 * graph lock and keeps it until this coroutine has terminated.
696
 */
697
static void coroutine_fn GRAPH_RDLOCK write_quorum_entry(void *opaque)
698
{
699
    QuorumCo *co = opaque;
700
    QuorumAIOCB *acb = co->acb;
701
    BDRVQuorumState *s = acb->bs->opaque;
702
    int i = co->idx;
703
    QuorumChildRequest *sacb = &acb->qcrs[i];
704

705
    sacb->bs = s->children[i]->bs;
706
    if (acb->flags & BDRV_REQ_ZERO_WRITE) {
707
        sacb->ret = bdrv_co_pwrite_zeroes(s->children[i], acb->offset,
708
                                          acb->bytes, acb->flags);
709
    } else {
710
        sacb->ret = bdrv_co_pwritev(s->children[i], acb->offset, acb->bytes,
711
                                    acb->qiov, acb->flags);
712
    }
713
    if (sacb->ret == 0) {
714
        acb->success_count++;
715
    } else {
716
        quorum_report_bad_acb(sacb, sacb->ret);
717
    }
718
    acb->count++;
719
    assert(acb->count <= s->num_children);
720
    assert(acb->success_count <= s->num_children);
721

722
    /* Wake up the caller after the last write */
723
    if (acb->count == s->num_children) {
724
        qemu_coroutine_enter_if_inactive(acb->co);
725
    }
726
}
727

728
static int coroutine_fn GRAPH_RDLOCK
729
quorum_co_pwritev(BlockDriverState *bs, int64_t offset, int64_t bytes,
730
                  QEMUIOVector *qiov, BdrvRequestFlags flags)
731
{
732
    BDRVQuorumState *s = bs->opaque;
733
    QuorumAIOCB *acb = quorum_aio_get(bs, qiov, offset, bytes, flags);
734
    int i, ret;
735

736
    for (i = 0; i < s->num_children; i++) {
737
        Coroutine *co;
738
        QuorumCo data = {
739
            .acb = acb,
740
            .idx = i,
741
        };
742

743
        co = qemu_coroutine_create(write_quorum_entry, &data);
744
        qemu_coroutine_enter(co);
745
    }
746

747
    while (acb->count < s->num_children) {
748
        qemu_coroutine_yield();
749
    }
750

751
    quorum_has_too_much_io_failed(acb);
752

753
    ret = acb->vote_ret;
754
    quorum_aio_finalize(acb);
755

756
    return ret;
757
}
758

759
static int coroutine_fn GRAPH_RDLOCK
760
quorum_co_pwrite_zeroes(BlockDriverState *bs, int64_t offset, int64_t bytes,
761
                        BdrvRequestFlags flags)
762
{
763
    return quorum_co_pwritev(bs, offset, bytes, NULL,
764
                             flags | BDRV_REQ_ZERO_WRITE);
765
}
766

767
static int64_t coroutine_fn GRAPH_RDLOCK
768
quorum_co_getlength(BlockDriverState *bs)
769
{
770
    BDRVQuorumState *s = bs->opaque;
771
    int64_t result;
772
    int i;
773

774
    /* check that all file have the same length */
775
    result = bdrv_co_getlength(s->children[0]->bs);
776
    if (result < 0) {
777
        return result;
778
    }
779
    for (i = 1; i < s->num_children; i++) {
780
        int64_t value = bdrv_co_getlength(s->children[i]->bs);
781
        if (value < 0) {
782
            return value;
783
        }
784
        if (value != result) {
785
            return -EIO;
786
        }
787
    }
788

789
    return result;
790
}
791

792
static coroutine_fn GRAPH_RDLOCK int quorum_co_flush(BlockDriverState *bs)
793
{
794
    BDRVQuorumState *s = bs->opaque;
795
    QuorumVoteVersion *winner = NULL;
796
    QuorumVotes error_votes;
797
    QuorumVoteValue result_value;
798
    int i;
799
    int result = 0;
800
    int success_count = 0;
801

802
    QLIST_INIT(&error_votes.vote_list);
803
    error_votes.compare = quorum_64bits_compare;
804

805
    for (i = 0; i < s->num_children; i++) {
806
        result = bdrv_co_flush(s->children[i]->bs);
807
        if (result) {
808
            quorum_report_bad(QUORUM_OP_TYPE_FLUSH, 0, 0,
809
                              s->children[i]->bs->node_name, result);
810
            result_value.l = result;
811
            quorum_count_vote(&error_votes, &result_value, i);
812
        } else {
813
            success_count++;
814
        }
815
    }
816

817
    if (success_count >= s->threshold) {
818
        result = 0;
819
    } else {
820
        winner = quorum_get_vote_winner(&error_votes);
821
        result = winner->value.l;
822
    }
823
    quorum_free_vote_list(&error_votes);
824

825
    return result;
826
}
827

828
static bool GRAPH_RDLOCK
829
quorum_recurse_can_replace(BlockDriverState *bs, BlockDriverState *to_replace)
830
{
831
    BDRVQuorumState *s = bs->opaque;
832
    int i;
833

834
    for (i = 0; i < s->num_children; i++) {
835
        /*
836
         * We have no idea whether our children show the same data as
837
         * this node (@bs).  It is actually highly likely that
838
         * @to_replace does not, because replacing a broken child is
839
         * one of the main use cases here.
840
         *
841
         * We do know that the new BDS will match @bs, so replacing
842
         * any of our children by it will be safe.  It cannot change
843
         * the data this quorum node presents to its parents.
844
         *
845
         * However, replacing @to_replace by @bs in any of our
846
         * children's chains may change visible data somewhere in
847
         * there.  We therefore cannot recurse down those chains with
848
         * bdrv_recurse_can_replace().
849
         * (More formally, bdrv_recurse_can_replace() requires that
850
         * @to_replace will be replaced by something matching the @bs
851
         * passed to it.  We cannot guarantee that.)
852
         *
853
         * Thus, we can only check whether any of our immediate
854
         * children matches @to_replace.
855
         *
856
         * (In the future, we might add a function to recurse down a
857
         * chain that checks that nothing there cares about a change
858
         * in data from the respective child in question.  For
859
         * example, most filters do not care when their child's data
860
         * suddenly changes, as long as their parents do not care.)
861
         */
862
        if (s->children[i]->bs == to_replace) {
863
            /*
864
             * We now have to ensure that there is no other parent
865
             * that cares about replacing this child by a node with
866
             * potentially different data.
867
             * We do so by checking whether there are any other parents
868
             * at all, which is stricter than necessary, but also very
869
             * simple.  (We may decide to implement something more
870
             * complex and permissive when there is an actual need for
871
             * it.)
872
             */
873
            return QLIST_FIRST(&to_replace->parents) == s->children[i] &&
874
                QLIST_NEXT(s->children[i], next_parent) == NULL;
875
        }
876
    }
877

878
    return false;
879
}
880

881
static int quorum_valid_threshold(int threshold, int num_children, Error **errp)
882
{
883

884
    if (threshold < 1) {
885
        error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
886
                   "vote-threshold", "a value >= 1");
887
        return -ERANGE;
888
    }
889

890
    if (threshold > num_children) {
891
        error_setg(errp, "threshold may not exceed children count");
892
        return -ERANGE;
893
    }
894

895
    return 0;
896
}
897

898
static QemuOptsList quorum_runtime_opts = {
899
    .name = "quorum",
900
    .head = QTAILQ_HEAD_INITIALIZER(quorum_runtime_opts.head),
901
    .desc = {
902
        {
903
            .name = QUORUM_OPT_VOTE_THRESHOLD,
904
            .type = QEMU_OPT_NUMBER,
905
            .help = "The number of vote needed for reaching quorum",
906
        },
907
        {
908
            .name = QUORUM_OPT_BLKVERIFY,
909
            .type = QEMU_OPT_BOOL,
910
            .help = "Trigger block verify mode if set",
911
        },
912
        {
913
            .name = QUORUM_OPT_REWRITE,
914
            .type = QEMU_OPT_BOOL,
915
            .help = "Rewrite corrupted block on read quorum",
916
        },
917
        {
918
            .name = QUORUM_OPT_READ_PATTERN,
919
            .type = QEMU_OPT_STRING,
920
            .help = "Allowed pattern: quorum, fifo. Quorum is default",
921
        },
922
        { /* end of list */ }
923
    },
924
};
925

926
static void quorum_refresh_flags(BlockDriverState *bs)
927
{
928
    BDRVQuorumState *s = bs->opaque;
929
    int i;
930

931
    bs->supported_zero_flags =
932
        BDRV_REQ_FUA | BDRV_REQ_MAY_UNMAP | BDRV_REQ_NO_FALLBACK;
933

934
    for (i = 0; i < s->num_children; i++) {
935
        bs->supported_zero_flags &= s->children[i]->bs->supported_zero_flags;
936
    }
937

938
    bs->supported_zero_flags |= BDRV_REQ_WRITE_UNCHANGED;
939
}
940

941
static int quorum_open(BlockDriverState *bs, QDict *options, int flags,
942
                       Error **errp)
943
{
944
    BDRVQuorumState *s = bs->opaque;
945
    QemuOpts *opts = NULL;
946
    const char *pattern_str;
947
    bool *opened;
948
    int i;
949
    int ret = 0;
950

951
    qdict_flatten(options);
952

953
    /* count how many different children are present */
954
    s->num_children = qdict_array_entries(options, "children.");
955
    if (s->num_children < 0) {
956
        error_setg(errp, "Option children is not a valid array");
957
        ret = -EINVAL;
958
        goto exit;
959
    }
960
    if (s->num_children < 1) {
961
        error_setg(errp, "Number of provided children must be 1 or more");
962
        ret = -EINVAL;
963
        goto exit;
964
    }
965

966
    opts = qemu_opts_create(&quorum_runtime_opts, NULL, 0, &error_abort);
967
    if (!qemu_opts_absorb_qdict(opts, options, errp)) {
968
        ret = -EINVAL;
969
        goto exit;
970
    }
971

972
    s->threshold = qemu_opt_get_number(opts, QUORUM_OPT_VOTE_THRESHOLD, 0);
973
    /* and validate it against s->num_children */
974
    ret = quorum_valid_threshold(s->threshold, s->num_children, errp);
975
    if (ret < 0) {
976
        goto exit;
977
    }
978

979
    pattern_str = qemu_opt_get(opts, QUORUM_OPT_READ_PATTERN);
980
    if (!pattern_str) {
981
        ret = QUORUM_READ_PATTERN_QUORUM;
982
    } else {
983
        ret = qapi_enum_parse(&QuorumReadPattern_lookup, pattern_str,
984
                              -EINVAL, NULL);
985
    }
986
    if (ret < 0) {
987
        error_setg(errp, "Please set read-pattern as fifo or quorum");
988
        goto exit;
989
    }
990
    s->read_pattern = ret;
991

992
    if (s->read_pattern == QUORUM_READ_PATTERN_QUORUM) {
993
        s->is_blkverify = qemu_opt_get_bool(opts, QUORUM_OPT_BLKVERIFY, false);
994
        if (s->is_blkverify && (s->num_children != 2 || s->threshold != 2)) {
995
            error_setg(errp, "blkverify=on can only be set if there are "
996
                       "exactly two files and vote-threshold is 2");
997
            ret = -EINVAL;
998
            goto exit;
999
        }
1000

1001
        s->rewrite_corrupted = qemu_opt_get_bool(opts, QUORUM_OPT_REWRITE,
1002
                                                 false);
1003
        if (s->rewrite_corrupted && s->is_blkverify) {
1004
            error_setg(errp,
1005
                       "rewrite-corrupted=on cannot be used with blkverify=on");
1006
            ret = -EINVAL;
1007
            goto exit;
1008
        }
1009
    }
1010

1011
    /* allocate the children array */
1012
    s->children = g_new0(BdrvChild *, s->num_children);
1013
    opened = g_new0(bool, s->num_children);
1014

1015
    for (i = 0; i < s->num_children; i++) {
1016
        char indexstr[INDEXSTR_LEN];
1017
        ret = snprintf(indexstr, INDEXSTR_LEN, "children.%d", i);
1018
        assert(ret < INDEXSTR_LEN);
1019

1020
        s->children[i] = bdrv_open_child(NULL, options, indexstr, bs,
1021
                                         &child_of_bds, BDRV_CHILD_DATA, false,
1022
                                         errp);
1023
        if (!s->children[i]) {
1024
            ret = -EINVAL;
1025
            goto close_exit;
1026
        }
1027

1028
        opened[i] = true;
1029
    }
1030
    s->next_child_index = s->num_children;
1031

1032
    bs->supported_write_flags = BDRV_REQ_WRITE_UNCHANGED;
1033
    quorum_refresh_flags(bs);
1034

1035
    g_free(opened);
1036
    goto exit;
1037

1038
close_exit:
1039
    /* cleanup on error */
1040
    bdrv_graph_wrlock();
1041
    for (i = 0; i < s->num_children; i++) {
1042
        if (!opened[i]) {
1043
            continue;
1044
        }
1045
        bdrv_unref_child(bs, s->children[i]);
1046
    }
1047
    bdrv_graph_wrunlock();
1048
    g_free(s->children);
1049
    g_free(opened);
1050
exit:
1051
    qemu_opts_del(opts);
1052
    return ret;
1053
}
1054

1055
static void quorum_close(BlockDriverState *bs)
1056
{
1057
    BDRVQuorumState *s = bs->opaque;
1058
    int i;
1059

1060
    bdrv_graph_wrlock();
1061
    for (i = 0; i < s->num_children; i++) {
1062
        bdrv_unref_child(bs, s->children[i]);
1063
    }
1064
    bdrv_graph_wrunlock();
1065

1066
    g_free(s->children);
1067
}
1068

1069
static void GRAPH_WRLOCK
1070
quorum_add_child(BlockDriverState *bs, BlockDriverState *child_bs, Error **errp)
1071
{
1072
    BDRVQuorumState *s = bs->opaque;
1073
    BdrvChild *child;
1074
    char indexstr[INDEXSTR_LEN];
1075
    int ret;
1076

1077
    if (s->is_blkverify) {
1078
        error_setg(errp, "Cannot add a child to a quorum in blkverify mode");
1079
        return;
1080
    }
1081

1082
    assert(s->num_children <= INT_MAX / sizeof(BdrvChild *));
1083
    if (s->num_children == INT_MAX / sizeof(BdrvChild *) ||
1084
        s->next_child_index == UINT_MAX) {
1085
        error_setg(errp, "Too many children");
1086
        return;
1087
    }
1088

1089
    ret = snprintf(indexstr, INDEXSTR_LEN, "children.%u", s->next_child_index);
1090
    if (ret < 0 || ret >= INDEXSTR_LEN) {
1091
        error_setg(errp, "cannot generate child name");
1092
        return;
1093
    }
1094
    s->next_child_index++;
1095

1096
    /* We can safely add the child now */
1097
    bdrv_ref(child_bs);
1098

1099
    child = bdrv_attach_child(bs, child_bs, indexstr, &child_of_bds,
1100
                              BDRV_CHILD_DATA, errp);
1101
    if (child == NULL) {
1102
        s->next_child_index--;
1103
        return;
1104
    }
1105
    s->children = g_renew(BdrvChild *, s->children, s->num_children + 1);
1106
    s->children[s->num_children++] = child;
1107
    quorum_refresh_flags(bs);
1108
}
1109

1110
static void GRAPH_WRLOCK
1111
quorum_del_child(BlockDriverState *bs, BdrvChild *child, Error **errp)
1112
{
1113
    BDRVQuorumState *s = bs->opaque;
1114
    char indexstr[INDEXSTR_LEN];
1115
    int i;
1116

1117
    for (i = 0; i < s->num_children; i++) {
1118
        if (s->children[i] == child) {
1119
            break;
1120
        }
1121
    }
1122

1123
    /* we have checked it in bdrv_del_child() */
1124
    assert(i < s->num_children);
1125

1126
    if (s->num_children <= s->threshold) {
1127
        error_setg(errp,
1128
            "The number of children cannot be lower than the vote threshold %d",
1129
            s->threshold);
1130
        return;
1131
    }
1132

1133
    /* We know now that num_children > threshold, so blkverify must be false */
1134
    assert(!s->is_blkverify);
1135

1136
    snprintf(indexstr, INDEXSTR_LEN, "children.%u", s->next_child_index - 1);
1137
    if (!strncmp(child->name, indexstr, INDEXSTR_LEN)) {
1138
        s->next_child_index--;
1139
    }
1140

1141
    /* We can safely remove this child now */
1142
    memmove(&s->children[i], &s->children[i + 1],
1143
            (s->num_children - i - 1) * sizeof(BdrvChild *));
1144
    s->children = g_renew(BdrvChild *, s->children, --s->num_children);
1145

1146
    bdrv_unref_child(bs, child);
1147

1148
    quorum_refresh_flags(bs);
1149
}
1150

1151
static void quorum_gather_child_options(BlockDriverState *bs, QDict *target,
1152
                                        bool backing_overridden)
1153
{
1154
    BDRVQuorumState *s = bs->opaque;
1155
    QList *children_list;
1156
    int i;
1157

1158
    /*
1159
     * The generic implementation for gathering child options in
1160
     * bdrv_refresh_filename() would use the names of the children
1161
     * as specified for bdrv_open_child() or bdrv_attach_child(),
1162
     * which is "children.%u" with %u being a value
1163
     * (s->next_child_index) that is incremented each time a new child
1164
     * is added (and never decremented).  Since children can be
1165
     * deleted at runtime, there may be gaps in that enumeration.
1166
     * When creating a new quorum BDS and specifying the children for
1167
     * it through runtime options, the enumeration used there may not
1168
     * have any gaps, though.
1169
     *
1170
     * Therefore, we have to create a new gap-less enumeration here
1171
     * (which we can achieve by simply putting all of the children's
1172
     * full_open_options into a QList).
1173
     *
1174
     * XXX: Note that there are issues with the current child option
1175
     *      structure quorum uses (such as the fact that children do
1176
     *      not really have unique permanent names).  Therefore, this
1177
     *      is going to have to change in the future and ideally we
1178
     *      want quorum to be covered by the generic implementation.
1179
     */
1180

1181
    children_list = qlist_new();
1182
    qdict_put(target, "children", children_list);
1183

1184
    for (i = 0; i < s->num_children; i++) {
1185
        qlist_append(children_list,
1186
                     qobject_ref(s->children[i]->bs->full_open_options));
1187
    }
1188
}
1189

1190
static char *quorum_dirname(BlockDriverState *bs, Error **errp)
1191
{
1192
    /* In general, there are multiple BDSs with different dirnames below this
1193
     * one; so there is no unique dirname we could return (unless all are equal
1194
     * by chance, or there is only one). Therefore, to be consistent, just
1195
     * always return NULL. */
1196
    error_setg(errp, "Cannot generate a base directory for quorum nodes");
1197
    return NULL;
1198
}
1199

1200
static void quorum_child_perm(BlockDriverState *bs, BdrvChild *c,
1201
                              BdrvChildRole role,
1202
                              BlockReopenQueue *reopen_queue,
1203
                              uint64_t perm, uint64_t shared,
1204
                              uint64_t *nperm, uint64_t *nshared)
1205
{
1206
    BDRVQuorumState *s = bs->opaque;
1207

1208
    *nperm = perm & DEFAULT_PERM_PASSTHROUGH;
1209
    if (s->rewrite_corrupted) {
1210
        *nperm |= BLK_PERM_WRITE;
1211
    }
1212

1213
    /*
1214
     * We cannot share RESIZE or WRITE, as this would make the
1215
     * children differ from each other.
1216
     */
1217
    *nshared = (shared & (BLK_PERM_CONSISTENT_READ |
1218
                          BLK_PERM_WRITE_UNCHANGED))
1219
             | DEFAULT_PERM_UNCHANGED;
1220
}
1221

1222
/*
1223
 * Each one of the children can report different status flags even
1224
 * when they contain the same data, so what this function does is
1225
 * return BDRV_BLOCK_ZERO if *all* children agree that a certain
1226
 * region contains zeroes, and BDRV_BLOCK_DATA otherwise.
1227
 */
1228
static int coroutine_fn GRAPH_RDLOCK
1229
quorum_co_block_status(BlockDriverState *bs, bool want_zero,
1230
                       int64_t offset, int64_t count,
1231
                       int64_t *pnum, int64_t *map, BlockDriverState **file)
1232
{
1233
    BDRVQuorumState *s = bs->opaque;
1234
    int i, ret;
1235
    int64_t pnum_zero = count;
1236
    int64_t pnum_data = 0;
1237

1238
    for (i = 0; i < s->num_children; i++) {
1239
        int64_t bytes;
1240
        ret = bdrv_co_common_block_status_above(s->children[i]->bs, NULL, false,
1241
                                                want_zero, offset, count,
1242
                                                &bytes, NULL, NULL, NULL);
1243
        if (ret < 0) {
1244
            quorum_report_bad(QUORUM_OP_TYPE_READ, offset, count,
1245
                              s->children[i]->bs->node_name, ret);
1246
            pnum_data = count;
1247
            break;
1248
        }
1249
        /*
1250
         * Even if all children agree about whether there are zeroes
1251
         * or not at @offset they might disagree on the size, so use
1252
         * the smallest when reporting BDRV_BLOCK_ZERO and the largest
1253
         * when reporting BDRV_BLOCK_DATA.
1254
         */
1255
        if (ret & BDRV_BLOCK_ZERO) {
1256
            pnum_zero = MIN(pnum_zero, bytes);
1257
        } else {
1258
            pnum_data = MAX(pnum_data, bytes);
1259
        }
1260
    }
1261

1262
    if (pnum_data) {
1263
        *pnum = pnum_data;
1264
        return BDRV_BLOCK_DATA;
1265
    } else {
1266
        *pnum = pnum_zero;
1267
        return BDRV_BLOCK_ZERO;
1268
    }
1269
}
1270

1271
static const char *const quorum_strong_runtime_opts[] = {
1272
    QUORUM_OPT_VOTE_THRESHOLD,
1273
    QUORUM_OPT_BLKVERIFY,
1274
    QUORUM_OPT_REWRITE,
1275
    QUORUM_OPT_READ_PATTERN,
1276

1277
    NULL
1278
};
1279

1280
static BlockDriver bdrv_quorum = {
1281
    .format_name                        = "quorum",
1282

1283
    .instance_size                      = sizeof(BDRVQuorumState),
1284

1285
    .bdrv_open                          = quorum_open,
1286
    .bdrv_close                         = quorum_close,
1287
    .bdrv_gather_child_options          = quorum_gather_child_options,
1288
    .bdrv_dirname                       = quorum_dirname,
1289
    .bdrv_co_block_status               = quorum_co_block_status,
1290

1291
    .bdrv_co_flush                      = quorum_co_flush,
1292

1293
    .bdrv_co_getlength                  = quorum_co_getlength,
1294

1295
    .bdrv_co_preadv                     = quorum_co_preadv,
1296
    .bdrv_co_pwritev                    = quorum_co_pwritev,
1297
    .bdrv_co_pwrite_zeroes              = quorum_co_pwrite_zeroes,
1298

1299
    .bdrv_add_child                     = quorum_add_child,
1300
    .bdrv_del_child                     = quorum_del_child,
1301

1302
    .bdrv_child_perm                    = quorum_child_perm,
1303

1304
    .bdrv_recurse_can_replace           = quorum_recurse_can_replace,
1305

1306
    .strong_runtime_opts                = quorum_strong_runtime_opts,
1307
};
1308

1309
static void bdrv_quorum_init(void)
1310
{
1311
    if (!qcrypto_hash_supports(QCRYPTO_HASH_ALG_SHA256)) {
1312
        /* SHA256 hash support is required for quorum device */
1313
        return;
1314
    }
1315
    bdrv_register(&bdrv_quorum);
1316
}
1317

1318
block_init(bdrv_quorum_init);
1319