1/* jobs.rs's Answer and merge. */
2#include "mesh/jobs/merge.h"
3
4#include <stdlib.h>
5#include <string.h>
6
7static void ops_free(void *v) { mj_ops_free(v); }
8static void ops_copy(void *d, const void *s) { *(mj_ops *)d = mj_ops_clone(s); }
9
10void mj_answer_init(mj_answer *a) {
11    memset(a, 0, sizeof *a);
12    mj_table_init(&a->committed);
13    a->entries = momap_new(2, sizeof(mj_ops), ops_free, ops_copy);
14}
15
16void mj_answer_free(mj_answer *a) {
17    mj_table_free(&a->committed);
18    momap_free(a->entries);
19    memset(a, 0, sizeof *a);
20}
21
22mj_ops *mj_answer_entry(mj_answer *a, uint64_t lid, uint64_t n) {
23    uint64_t k[2] = {lid, n};
24    return momap_put(a->entries, k, NULL);
25}
26
27mj_answer mj_answer_of(const mj_replica *r) {
28    mj_answer a;
29    mj_answer_init(&a);
30    a.commit_lid = r->commit_lid;
31    a.commit_n = r->commit_n;
32    mj_table_free(&a.committed);
33    a.committed = mj_table_clone(&r->committed);
34    for (size_t i = 0; i < r->npending; i++) {
35        const mj_entry *e = &r->pending[i];
36        mj_ops ops = {0};
37        if (e->replace) mj_table_ops(&e->table, &ops);
38        else ops = mj_ops_clone(&e->ops);
39        mj_ops *slot = mj_answer_entry(&a, e->lid, e->n);
40        mj_ops_free(slot); /* a later entry under the same key replaces it, as BTreeMap::insert does */
41        *slot = ops;
42    }
43    return a;
44}
45
46mj_table mj_merge(mj_answer *const *answers, size_t n) {
47    mj_table t;
48    mj_table_init(&t);
49    /* The committed tables, the latest commit point first (a stable sort: ties keep member order). */
50    mj_answer **by = n ? mxalloc(n * sizeof *by) : NULL;
51    memcpy(by, answers, n * sizeof *by);
52    for (size_t i = 1; i < n; i++) {
53        mj_answer *x = by[i];
54        size_t j = i;
55        while (j > 0 && (by[j - 1]->commit_lid < x->commit_lid || (by[j - 1]->commit_lid == x->commit_lid && by[j - 1]->commit_n < x->commit_n))) {
56            by[j] = by[j - 1];
57            j--;
58        }
59        by[j] = x;
60    }
61    for (size_t i = 0; i < n; i++) {
62        mj_ops ops = {0};
63        mj_table_ops(&by[i]->committed, &ops);
64        for (size_t k = 0; k < ops.n; k++) mj_table_apply(&t, &ops.v[k]);
65        mj_ops_free(&ops);
66    }
67    free(by);
68    /* The entries not committed yet, keyed (latest leadership first, then number); the first answer's wins. */
69    momap *entries = momap_new(2, sizeof(const mj_ops *), NULL, NULL);
70    for (size_t i = 0; i < n; i++) {
71        uint64_t k[2];
72        for (const mj_ops *ops = momap_first(answers[i]->entries, k); ops; ops = momap_next(answers[i]->entries, k, k)) {
73            uint64_t key[2] = {UINT64_MAX - k[0], k[1]};
74            bool fresh;
75            const mj_ops **slot = momap_put(entries, key, &fresh);
76            if (fresh) *slot = ops;
77        }
78    }
79    uint64_t k[2];
80    for (const mj_ops **ops = momap_first(entries, k); ops; ops = momap_next(entries, k, k))
81        for (size_t j = 0; j < (*ops)->n; j++) mj_table_apply(&t, &(*ops)->v[j]);
82    momap_free(entries);
83    return t;
84}