==> tests/blobd.c <==1/* Three blobd's over a fake network: an own payload kept and pushed (acknowledged), a fetch from a
2 * holder, a corrupt push dropped, a forget dropping references, a full store refusing, on-disk writes
3 * through the writers, and `render` of an assembled image. */
4#include "mesh/blobd/blobd.h"
5#include "mesh/abi/peer_frames.h"
6#include "mesh/blobs/durable.h"
7#include "mesh/blobs/frame.h"
8#include "pathtrace/default_scene.h"
9#include "mesh/check.h"
10#include "mesh/ref/sha256d.h"
11#include "mesh/strbuf.h"
12#include "pathtrace/scene.h"
13#include "pathtrace/view.h"
14#include "pathtrace/z.h"
15
16#include <stdio.h>
17#include <stdlib.h>
18#include <string.h>
19#include <unistd.h>
20
21#define N 3
22
23typedef struct {
24 uint32_t from, to;
25 uint8_t *b;
26 size_t n;
27} msg;
28
29static msg q[100000];
30static size_t nq;
31static mbd *node[N];
32static int64_t clock_ms = 1;
33static int stored[N], acked_ok[N], acked_bad[N], gone[N];
34static int64_t mem[N];
35static bool watched[N][8];
36static mbd_write pending[64];
37static size_t npending;
38static int errors;
39
40typedef struct {
41 uint32_t me;
42} cx;
43static cx cxs[N];
44
45static bool send(void *c, uint32_t to, const uint8_t *f, size_t n) {
46 msg *m = &q[nq++];
47 m->from = ((cx *)c)->me;
48 m->to = to;
49 m->b = mxalloc(n);
50 memcpy(m->b, f, n);
51 m->n = n;
52 return true;
53}
54static void on_stored(void *c, const mb_ref *r, size_t n) {
55 (void)r;
56 stored[((cx *)c)->me] += (int)n;
57}
58static void on_acked(void *c, uint32_t from, mj_job_id job, const mj_hash *h, bool ok) {
59 (void)from, (void)job, (void)h;
60 if (ok) acked_ok[((cx *)c)->me]++;
61 else acked_bad[((cx *)c)->me]++;
62}
63static void on_gone(void *c, const mb_ref *r, size_t n) {
64 (void)r;
65 gone[((cx *)c)->me] += (int)n;
66}
67static uint32_t alive(void *c) {
68 (void)c;
69 return 7;
70}
71static bool leading(void *c) { return ((cx *)c)->me == 0; }
72static void watching(void *c, mj_job_id job, bool on) { watched[((cx *)c)->me][job & 7] = on; }
73static void on_mem(void *c, int64_t d) { mem[((cx *)c)->me] += d; }
74static void on_write(void *c, mbd_write w) {
75 (void)c;
76 pending[npending++] = w;
77}
78static int64_t now(void *c) {
79 (void)c;
80 return clock_ms;
81}
82static void on_log(void *c, bool error, const char *line) {
83 (void)c;
84 if (error) errors++;
85 (void)line;
86}
87
88/* Delivers every queued frame (frames sent while delivering go too). */
89static void pump(void) {
90 for (size_t i = 0; i < nq; i++) {
91 msg m = q[i];
92 uint32_t len;
93 memcpy(&len, m.b + 1, 4);
94 mbd_frame(node[m.to], m.from, m.b[0], m.b + 5, len);
95 free(m.b);
96 }
97 nq = 0;
98}
99
100/* The writers: each pending write made durable, then reported to its node. */
101static void writers(uint32_t k) {
102 bool ok[64];
103 char *err[64];
104 mbd_write_batch(pending, npending, ok, err);
105 for (size_t i = 0; i < npending; i++) {
106 mbd_written(node[k], pending[i].job, pending[i].hash, pending[i].why, pending[i].from, ok[i], err[i]);
107 free(err[i]);
108 mbd_write_free(&pending[i]);
109 }
110 npending = 0;
111}
112
113static mj_hash sha(const void *b, size_t n) {
114 mj_hash h;
115 mesh_sha256(b, n, h.b);
116 return h;
117}
118
119int main(void) {
120 for (uint32_t k = 0; k < N; k++) {
121 cxs[k].me = k;
122 mbd_io io = {&cxs[k], send, send, on_stored, on_acked, on_gone, alive, leading, watching, on_mem, on_write, now, on_log};
123 node[k] = mbd_new(k, mb_open(NULL, k == 2 ? 64 : (uint64_t)1 << 30, NULL), io);
124 }
125 /* An own payload on 0: kept (in memory: written at once), counted in memory, pushed to 1. */
126 uint8_t pay[48];
127 memset(pay, 0x5a, sizeof pay);
128 mj_hash h = sha(pay, sizeof pay);
129 uint8_t *c = mxalloc(sizeof pay);
130 memcpy(c, pay, sizeof pay);
131 mbd_local(node[0], 1, h, c, sizeof pay);
132 CHECK(stored[0] == 1 && mem[0] == 48 && mb_has_ref(mbd_store(node[0]), 1, &h));
133 mbd_push(node[0], 1, 1, h);
134 mbd_push(node[0], 1, 1, h); /* queued once */
135 mbd_tick(node[0]);
136 CHECK_U(nq, 1);
137 pump(); /* 1 keeps it and acknowledges */
138 CHECK(mb_has_ref(mbd_store(node[1]), 1, &h) && stored[1] == 1);
139 pump(); /* the ack reaches 0 */
140 CHECK_I(acked_ok[0], 1);
141 mbd_done(node[0], 1, h);
142 CHECK_I(mem[0], 0);
143 /* 2's store is tiny (64 bytes): a push of a second blob is refused (acknowledged not ok). */
144 uint8_t big[100];
145 memset(big, 1, sizeof big);
146 mj_hash hb = sha(big, sizeof big);
147 c = mxalloc(sizeof big);
148 memcpy(c, big, sizeof big);
149 mbd_local(node[0], 1, hb, c, sizeof big);
150 mbd_push(node[0], 2, 1, hb);
151 mbd_tick(node[0]);
152 pump();
153 pump();
154 CHECK(acked_bad[0] == 1 && !mb_has_ref(mbd_store(node[2]), 1, &hb));
155 /* A fetch on 2 for job 1's h: asked of the holder 1, answered, kept. */
156 mbd_fetch(node[2], 1, h, 2);
157 mbd_tick(node[2]);
158 CHECK(nq == 1 && q[0].to == 1 && q[0].b[0] == MESH_F_BLOB_GET);
159 pump(); /* 1 answers */
160 pump(); /* 2 keeps it */
161 CHECK(mb_has_ref(mbd_store(node[2]), 1, &h) && stored[2] == 1);
162 /* A corrupt push is dropped. */
163 msb f = {0};
164 mbf_put(&f, &h, 5, "not the bytes", 13);
165 int before = errors;
166 mbd_frame(node[1], 0, MESH_F_BLOB_PUT, (const uint8_t *)f.p + 5, f.n - 5);
167 msb_free(&f);
168 CHECK(errors == before + 1 && !mb_known_ref(mbd_store(node[1]), 5, &h));
169 /* Job 1 forgotten on 1: its references go, jobd hears. */
170 mbd_news n = {0};
171 n.forgotten = mxalloc(sizeof *n.forgotten);
172 n.forgotten[0] = 1;
173 n.nforgotten = 1;
174 mbd_table(node[1], &n);
175 CHECK(gone[1] == 1 && !mb_has(mbd_store(node[1]), &h));
176 /* A forgotten job's push is refused. */
177 mbd_push(node[0], 1, 1, h);
178 mbd_tick(node[0]);
179 pump();
180 pump();
181 CHECK_I(acked_bad[0], 2);
182 /* On disk: a node whose store has a directory writes through the writers, then reports. */
183 char *dir = mfmt("%s/meshd-blobd-%d", getenv("TMPDIR") ? getenv("TMPDIR") : "/tmp", (int)getpid());
184 mb_remove_tree(dir);
185 cx dcx = {0};
186 mbd_io io = {&dcx, send, send, on_stored, on_acked, on_gone, alive, leading, watching, on_mem, on_write, now, on_log};
187 stored[0] = 0;
188 mbd *d = mbd_new(0, mb_open(dir, (uint64_t)1 << 30, NULL), io);
189 c = mxalloc(sizeof pay);
190 memcpy(c, pay, sizeof pay);
191 mbd_local(d, 9, h, c, sizeof pay);
192 CHECK(npending == 1 && stored[0] == 0 && mb_known_ref(mbd_store(d), 9, &h) && !mb_has_ref(mbd_store(d), 9, &h));
193 mbd *keep = node[0];
194 node[0] = d;
195 writers(0);
196 node[0] = keep;
197 CHECK(stored[0] == 1 && mb_has_ref(mbd_store(d), 9, &h));
198 size_t got_n;
199 uint8_t *got = mb_get(mbd_store(d), &h, &got_n);
200 CHECK(got && got_n == sizeof pay && memcmp(got, pay, sizeof pay) == 0);
201 free(got);
202 mbd_free(d);
203 mb_remove_tree(dir);
204 free(dir);
205 /* render: a one-pass, one-tile scene's job on 0 (the leader assembles active images): its only item's
206 * payload kept here, the result committed through news, the image complete. */
207 mj_setup setup[] = {{5, 0}, {7, pt_set_arg(PT_H_WIDTH, 32)}, {7, pt_set_arg(PT_H_HEIGHT, 24)}, {7, pt_set_arg(PT_H_PASSES, 1)}, {7, pt_set_arg(PT_H_SPP, 1)}};
208 uint32_t sc[PT_DEFAULT_SCENE_WORDS];
209 CHECK(masm_job_scene(setup, 5, sc, NULL));
210 pt_v *part = mxalloc(PT_TILE_PIX * sizeof *part);
211 uint64_t digest = pt_tile_pass_z(sc, 0, 0, part);
212 uint8_t *pb = mxalloc(PT_PAYLOAD);
213 for (size_t i = 0; i < PT_TILE_PIX; i++) {
214 memcpy(pb + 12 * i, &part[i].x, 4);
215 memcpy(pb + 12 * i + 4, &part[i].y, 4);
216 memcpy(pb + 12 * i + 8, &part[i].z, 4);
217 }
218 free(part);
219 mj_hash ph = sha(pb, PT_PAYLOAD);
220 mj_job_id job = 6;
221 mbd_local(node[0], job, ph, pb, PT_PAYLOAD);
222 mbd_news nw = {0};
223 nw.jobs = mxalloc(sizeof *nw.jobs);
224 nw.jobs[0] = (mbd_job_info){job, mstrdup("pathtrace"), PT_OP_PASS, mj_setup_clone(setup, 5), 5, 0, true, mstrdup("running")};
225 nw.njobs = 1;
226 nw.results = mxalloc(sizeof *nw.results);
227 nw.results[0] = (mbd_result){job, 0, {ph, PT_PAYLOAD, 1, 0, digest}};
228 nw.nresults = 1;
229 mbd_table(node[0], &nw); /* the leader starts assembling it, then gets its result */
230 CHECK(watched[0][job & 7]);
231 nw = (mbd_news){0};
232 nw.jobs = mxalloc(sizeof *nw.jobs);
233 nw.jobs[0] = (mbd_job_info){job, mstrdup("pathtrace"), PT_OP_PASS, mj_setup_clone(setup, 5), 5, 0, true, mstrdup("running")};
234 nw.njobs = 1;
235 nw.results = mxalloc(sizeof *nw.results);
236 nw.results[0] = (mbd_result){job, 0, {ph, PT_PAYLOAD, 1, 0, digest}};
237 nw.nresults = 1;
238 mbd_table(node[0], &nw);
239 mbd_tick(node[0]); /* read here and folded */
240 const char *w[] = {"0.6", "status"};
241 uint8_t *out;
242 size_t len;
243 char *err = NULL;
244 CHECK(mbd_render(node[0], w, 2, &out, &len, &err));
245 CHECK(strstr((char *)out, "render job 0.6 program pathtrace state running system 0 width 32 height 24 tiles 1 passes 1 spp 1 items 1 committed 1 folded 1 tiles_done 1 tone aces") != NULL);
246 CHECK(strstr((char *)out, "complete 1\n") && strstr((char *)out, "\nmember 0 items 1 tiles 1\n"));
247 free(out);
248 const char *w2[] = {"latest", "ppm"};
249 CHECK(mbd_render(node[0], w2, 2, &out, &len, &err) && len == 13 + 32 * 24 * 3 && memcmp(out, "P6\n32 24\n255\n", 13) == 0);
250 free(out);
251 const char *w3[] = {"0.7"};
252 CHECK(!mbd_render(node[0], w3, 1, &out, &len, &err));
253 CHECK_S(err, "no job 0.7 (committed here)");
254 free(err);
255 for (uint32_t k = 0; k < N; k++) mbd_free(node[k]);
256 return mesh_check_done();
257}