==> src/log.c <==1/* jobs.rs's Log (encode, decode), Entry, Replica (apply, last, log) and replay. */
2#include "mesh/jobs/log.h"
3
4#include <stdlib.h>
5#include <string.h>
6
7void mj_log_free(mj_log *l) {
8 mj_ops_free(&l->ops);
9 memset(l, 0, sizeof *l);
10}
11
12mj_log mj_log_clone(const mj_log *l) {
13 mj_log c = *l;
14 c.ops = mj_ops_clone(&l->ops);
15 return c;
16}
17
18void mj_log_encode(const mj_log *l, msb *out) {
19 msb w = {0};
20 switch (l->kind) {
21 case MJ_LOG_LID: mj_w_u64(&w, l->lid); break;
22 case MJ_LOG_SNAPSHOT:
23 mj_w_u64(&w, l->lid);
24 mj_w_u64(&w, l->n);
25 mj_encode_ops(l->ops.v, l->ops.n, &w);
26 break;
27 case MJ_LOG_ENTRY:
28 mj_w_u64(&w, l->lid);
29 mj_w_u64(&w, l->n);
30 mj_w_u8(&w, l->replace ? 1 : 0);
31 mj_encode_ops(l->ops.v, l->ops.n, &w);
32 break;
33 case MJ_LOG_COMMIT:
34 mj_w_u64(&w, l->lid);
35 mj_w_u64(&w, l->n);
36 break;
37 }
38 mj_w_u8(out, (uint8_t)l->kind);
39 mj_w_u64(out, w.n);
40 msb_put(out, w.p, w.n);
41 msb_free(&w);
42}
43
44bool mj_log_decode(uint8_t kind, const void *body, size_t n, mj_log *out, char **err) {
45 mj_rd r = mj_rd_new(body, n);
46 mj_log l = {.kind = (mj_log_kind)kind};
47 bool ok;
48 switch (kind) {
49 case MJ_LOG_LID: ok = mj_rd_u64(&r, &l.lid); break;
50 case MJ_LOG_SNAPSHOT: ok = mj_rd_u64(&r, &l.lid) && mj_rd_u64(&r, &l.n) && mj_rd_ops(&r, &l.ops); break;
51 case MJ_LOG_ENTRY: {
52 uint8_t rep;
53 ok = mj_rd_u64(&r, &l.lid) && mj_rd_u64(&r, &l.n) && mj_rd_u8(&r, &rep);
54 if (ok) l.replace = rep != 0;
55 ok = ok && mj_rd_ops(&r, &l.ops);
56 break;
57 }
58 case MJ_LOG_COMMIT: ok = mj_rd_u64(&r, &l.lid) && mj_rd_u64(&r, &l.n); break;
59 default: ok = mj_rd_fail(&r, mfmt("job log: record kind %u?", kind));
60 }
61 if (ok && !mj_rd_done(&r)) ok = mj_rd_fail(&r, mfmt("job log: record kind %u: %zu trailing bytes", kind, n - r.at));
62 if (!ok) {
63 mj_log_free(&l);
64 if (err) *err = mj_rd_take_err(&r);
65 mj_rd_free(&r);
66 return false;
67 }
68 *out = l;
69 return true;
70}
71
72void mj_entry_free(mj_entry *e) {
73 mj_ops_free(&e->ops);
74 if (e->replace) mj_table_free(&e->table);
75 memset(e, 0, sizeof *e);
76}
77
78mj_entry mj_entry_clone(const mj_entry *e) {
79 mj_entry c = *e;
80 c.ops = mj_ops_clone(&e->ops);
81 if (e->replace) c.table = mj_table_clone(&e->table);
82 return c;
83}
84
85bool mj_entry_eq(const mj_entry *a, const mj_entry *b) {
86 if (a->lid != b->lid || a->n != b->n || a->replace != b->replace) return false;
87 return a->replace ? mj_table_eq(&a->table, &b->table) : mj_ops_eq(&a->ops, &b->ops);
88}
89
90void mj_replica_init(mj_replica *r) {
91 memset(r, 0, sizeof *r);
92 mj_table_init(&r->committed);
93}
94
95void mj_replica_free(mj_replica *r) {
96 mj_table_free(&r->committed);
97 for (size_t i = 0; i < r->npending; i++) mj_entry_free(&r->pending[i]);
98 free(r->pending);
99 memset(r, 0, sizeof *r);
100}
101
102mj_replica mj_replica_clone(const mj_replica *r) {
103 mj_replica c = *r;
104 c.committed = mj_table_clone(&r->committed);
105 c.pending = r->npending ? mxalloc(r->npending * sizeof *c.pending) : NULL;
106 for (size_t i = 0; i < r->npending; i++) c.pending[i] = mj_entry_clone(&r->pending[i]);
107 c.cappending = r->npending;
108 return c;
109}
110
111bool mj_replica_eq(const mj_replica *a, const mj_replica *b) {
112 if (a->lid != b->lid || a->commit_lid != b->commit_lid || a->commit_n != b->commit_n || !mj_table_eq(&a->committed, &b->committed) ||
113 a->npending != b->npending)
114 return false;
115 for (size_t i = 0; i < a->npending; i++)
116 if (!mj_entry_eq(&a->pending[i], &b->pending[i])) return false;
117 return true;
118}
119
120static void push_pending(mj_replica *r, mj_entry e) {
121 if (r->npending == r->cappending) {
122 r->cappending = r->cappending ? 2 * r->cappending : 8;
123 r->pending = mxrealloc(r->pending, r->cappending * sizeof *r->pending);
124 }
125 r->pending[r->npending++] = e;
126}
127
128static mj_table table_of(const mj_ops *ops) {
129 mj_table t;
130 mj_table_init(&t);
131 for (size_t i = 0; i < ops->n; i++) mj_table_apply(&t, &ops->v[i]);
132 return t;
133}
134
135void mj_replica_apply(mj_replica *r, const mj_log *l) {
136 switch (l->kind) {
137 case MJ_LOG_LID:
138 if (l->lid > r->lid) r->lid = l->lid;
139 break;
140 case MJ_LOG_SNAPSHOT: {
141 mj_table t = table_of(&l->ops);
142 mj_table_free(&r->committed);
143 r->committed = t;
144 r->commit_lid = l->lid;
145 r->commit_n = l->n;
146 /* What the snapshot does not cover stays (review finding 6): that leadership's entries past it,
147 * in sequence; later leaderships'; and earlier ones' while that leadership has committed
148 * nothing (n = 0: its first entry, the merge, would cover them). */
149 uint64_t next = l->n + 1;
150 size_t k = 0;
151 for (size_t i = 0; i < r->npending; i++) {
152 mj_entry *e = &r->pending[i];
153 bool keep;
154 if (e->lid < l->lid) keep = l->n == 0;
155 else if (e->lid > l->lid) keep = true;
156 else if (e->n == next) {
157 next++;
158 keep = true;
159 } else keep = false;
160 if (keep) r->pending[k++] = *e;
161 else mj_entry_free(e);
162 }
163 r->npending = k;
164 break;
165 }
166 case MJ_LOG_ENTRY: {
167 mj_entry e = {.lid = l->lid, .n = l->n, .replace = l->replace};
168 if (l->replace) {
169 /* A new leadership's merge: it holds every entry of the leaderships before it a majority had. */
170 size_t k = 0;
171 for (size_t i = 0; i < r->npending; i++) {
172 if (r->pending[i].lid == l->lid) r->pending[k++] = r->pending[i];
173 else mj_entry_free(&r->pending[i]);
174 }
175 r->npending = k;
176 e.table = table_of(&l->ops);
177 } else e.ops = mj_ops_clone(&l->ops);
178 push_pending(r, e);
179 break;
180 }
181 case MJ_LOG_COMMIT: {
182 size_t i = 0;
183 while (i < r->npending && r->pending[i].lid == l->lid && r->pending[i].n <= l->n) {
184 mj_entry *e = &r->pending[i];
185 if (e->replace) {
186 mj_table_free(&r->committed);
187 r->committed = e->table;
188 e->replace = false; /* moved */
189 } else
190 for (size_t k = 0; k < e->ops.n; k++) mj_table_apply(&r->committed, &e->ops.v[k]);
191 r->commit_lid = l->lid;
192 r->commit_n = e->n;
193 mj_entry_free(e);
194 i++;
195 }
196 if (i) {
197 memmove(r->pending, r->pending + i, (r->npending - i) * sizeof *r->pending);
198 r->npending -= i;
199 }
200 break;
201 }
202 }
203}
204
205uint64_t mj_replica_last(const mj_replica *r, uint64_t lid) {
206 for (size_t i = r->npending; i-- > 0;)
207 if (r->pending[i].lid == lid) return r->pending[i].n;
208 return 0;
209}
210
211void mj_logs_push(mj_logs *v, mj_log l) {
212 if (v->n == v->cap) {
213 v->cap = v->cap ? 2 * v->cap : 8;
214 v->v = mxrealloc(v->v, v->cap * sizeof *v->v);
215 }
216 v->v[v->n++] = l;
217}
218
219void mj_logs_free(mj_logs *v) {
220 for (size_t i = 0; i < v->n; i++) mj_log_free(&v->v[i]);
221 free(v->v);
222 *v = (mj_logs){0};
223}
224
225mj_logs mj_replica_log(const mj_replica *r) {
226 mj_logs v = {0};
227 mj_logs_push(&v, mj_log_lid(r->lid));
228 mj_ops snap = {0};
229 mj_table_ops(&r->committed, &snap);
230 mj_logs_push(&v, mj_log_snapshot(r->commit_lid, r->commit_n, snap));
231 for (size_t i = 0; i < r->npending; i++) {
232 const mj_entry *e = &r->pending[i];
233 mj_ops ops = {0};
234 if (e->replace) mj_table_ops(&e->table, &ops);
235 else ops = mj_ops_clone(&e->ops);
236 mj_logs_push(&v, mj_log_entry(e->lid, e->n, e->replace, ops));
237 }
238 return v;
239}
240
241bool mj_replay(const void *bytes, size_t n, mj_replica *out, size_t *records, size_t *whole, char **err) {
242 const uint8_t *b = bytes;
243 mj_replica rep;
244 mj_replica_init(&rep);
245 size_t at = 0, k = 0;
246 while (n - at >= 9) {
247 uint64_t len = 0;
248 for (int i = 8; i >= 1; i--) len = len << 8 | b[at + (size_t)i];
249 if ((uint64_t)(n - at - 9) < len) break;
250 mj_log l;
251 char *e = NULL;
252 if (!mj_log_decode(b[at], b + at + 9, (size_t)len, &l, &e)) {
253 if (err) *err = mfmt("%s (record %zu at byte %zu)", e, k, at);
254 free(e);
255 mj_replica_free(&rep);
256 return false;
257 }
258 mj_replica_apply(&rep, &l);
259 mj_log_free(&l);
260 at += 9 + (size_t)len;
261 k++;
262 }
263 *out = rep;
264 if (records) *records = k;
265 if (whole) *whole = at;
266 return true;
267}