==> src/store.c <==1/* blobs.rs's Store. */
2#include "mesh/blobs/store.h"
3#include "mesh/blobs/durable.h"
4#include "mesh/omap.h"
5#include "mesh/ref/sha256d.h"
6#include "mesh/strbuf.h"
7
8#include <dirent.h>
9#include <stdio.h>
10#include <stdlib.h>
11#include <string.h>
12#include <sys/stat.h>
13#include <unistd.h>
14
15enum { PENDING = 1, PRESENT = 2 };
16
17typedef struct {
18 uint32_t len;
19 momap *refs; /* job -> uint8_t state */
20} entry;
21
22typedef struct {
23 uint8_t *p;
24 size_t n;
25} bytes;
26
27/* The index: blob hash -> entry, open addressing with linear probing (a sha256 is its own hash), as
28 * the Rust's HashMap; deletion shifts the run back so no tombstones are left. */
29typedef struct {
30 bool used;
31 mj_hash h;
32 entry e;
33} slot;
34
35typedef struct {
36 slot *v;
37 size_t cap, n;
38} table;
39
40struct mb_store {
41 char *dir;
42 uint64_t cap;
43 table index;
44 momap *jobs; /* job -> uint64_t references (pending included) */
45 uint64_t total;
46 momap *mem; /* hash words -> bytes (no directory) */
47 mb_refs lost;
48 void (*log)(void *ctx, const char *line);
49 void *log_ctx;
50};
51
52static size_t home(const table *t, const mj_hash *h) {
53 uint64_t x;
54 memcpy(&x, h->b, 8);
55 return (size_t)(x & (t->cap - 1));
56}
57
58static slot *tfind(const table *t, const mj_hash *h) {
59 if (!t->cap) return NULL;
60 for (size_t i = home(t, h);; i = (i + 1) & (t->cap - 1)) {
61 slot *s = &t->v[i];
62 if (!s->used) return NULL;
63 if (mj_hash_eq(&s->h, h)) return s;
64 }
65}
66
67static void tgrow(table *t);
68
69/* The entry for h, inserted (zeroed) if absent. */
70static entry *tput(table *t, const mj_hash *h, bool *fresh) {
71 slot *s = tfind(t, h);
72 if (s) {
73 *fresh = false;
74 return &s->e;
75 }
76 if ((t->n + 1) * 10 > t->cap * 7) tgrow(t);
77 size_t i = home(t, h);
78 while (t->v[i].used) i = (i + 1) & (t->cap - 1);
79 t->v[i] = (slot){true, *h, {0}};
80 t->n++;
81 *fresh = true;
82 return &t->v[i].e;
83}
84
85static void tgrow(table *t) {
86 table o = *t;
87 t->cap = o.cap ? 2 * o.cap : 64;
88 t->v = mxalloc(t->cap * sizeof *t->v);
89 t->n = 0;
90 for (size_t i = 0; i < o.cap; i++)
91 if (o.v[i].used) {
92 bool f;
93 *tput(t, &o.v[i].h, &f) = o.v[i].e;
94 }
95 free(o.v);
96}
97
98/* Removes h's slot (its entry's refs already freed), shifting the probe run back. */
99static void tdel(table *t, const mj_hash *h) {
100 slot *s = tfind(t, h);
101 if (!s) return;
102 size_t i = (size_t)(s - t->v);
103 t->v[i].used = false;
104 t->n--;
105 for (size_t j = (i + 1) & (t->cap - 1); t->v[j].used; j = (j + 1) & (t->cap - 1)) {
106 size_t k = home(t, &t->v[j].h);
107 /* Move j back to i when its home is not cyclically in (i, j]. */
108 bool between = i <= j ? (i < k && k <= j) : (i < k || k <= j);
109 if (!between) {
110 t->v[i] = t->v[j];
111 t->v[j].used = false;
112 i = j;
113 }
114 }
115}
116static void bytes_free(void *v) { free(((bytes *)v)->p); }
117
118void mb_refs_free(mb_refs *r) {
119 free(r->v);
120 *r = (mb_refs){0};
121}
122
123static void refs_push(mb_refs *r, mj_job_id job, const mj_hash *h) {
124 r->v = mxrealloc(r->v, (r->n + 1) * sizeof *r->v);
125 r->v[r->n++] = (mb_ref){job, *h};
126}
127
128void mb_reserved_free(mb_reserved *r) {
129 free(r->path);
130 mb_refs_free(&r->evicted);
131 *r = (mb_reserved){0};
132}
133
134static void add_ref(mb_store *s, mj_job_id job, const mj_hash *h, uint32_t len, uint8_t st) {
135 bool fresh;
136 entry *e = tput(&s->index, h, &fresh);
137 if (fresh) {
138 s->total += len;
139 e->len = len;
140 e->refs = momap_new(1, 1, NULL, NULL);
141 }
142 bool newref;
143 uint8_t *r = momap_put1(e->refs, job, &newref);
144 *r = st;
145 if (newref) (*(uint64_t *)momap_put1(s->jobs, job, NULL))++;
146}
147
148mb_store *mb_open(const char *dir, uint64_t cap, char **err) {
149 mb_store *s = mxalloc(sizeof *s);
150 s->cap = cap;
151 s->jobs = momap_new(1, sizeof(uint64_t), NULL, NULL);
152 s->mem = momap_new(4, sizeof(bytes), bytes_free, NULL);
153 if (!dir) return s;
154 s->dir = mstrdup(dir);
155 char *e = NULL;
156 if (!mb_mkdirs(dir, &e)) {
157 if (err) *err = mfmt("--blobs %s: %s", dir, e);
158 free(e);
159 mb_store_free(s);
160 return NULL;
161 }
162 DIR *d = opendir(dir);
163 if (!d) {
164 if (err) *err = mfmt("--blobs %s: cannot read it", dir);
165 mb_store_free(s);
166 return NULL;
167 }
168 struct dirent *j;
169 while ((j = readdir(d))) {
170 mj_job_id job;
171 if (!mj_parse_job(j->d_name, &job)) continue;
172 char *jd = mfmt("%s/%s", dir, j->d_name);
173 DIR *fd = opendir(jd);
174 if (fd) {
175 struct dirent *f;
176 while ((f = readdir(fd))) {
177 size_t nl = strlen(f->d_name);
178 char *p = mfmt("%s/%s", jd, f->d_name);
179 if (nl > 4 && strcmp(f->d_name + nl - 4, ".tmp") == 0) unlink(p);
180 else {
181 mj_hash h;
182 struct stat st;
183 if (mj_unhex(f->d_name, &h) && stat(p, &st) == 0) add_ref(s, job, &h, (uint32_t)st.st_size, PRESENT);
184 }
185 free(p);
186 }
187 closedir(fd);
188 }
189 free(jd);
190 }
191 closedir(d);
192 return s;
193}
194
195void mb_store_free(mb_store *s) {
196 if (!s) return;
197 free(s->dir);
198 for (size_t i = 0; i < s->index.cap; i++)
199 if (s->index.v[i].used) momap_free(s->index.v[i].e.refs);
200 free(s->index.v);
201 momap_free(s->jobs);
202 momap_free(s->mem);
203 mb_refs_free(&s->lost);
204 free(s);
205}
206
207void mb_set_log(mb_store *s, void (*log)(void *ctx, const char *line), void *ctx) {
208 s->log = log;
209 s->log_ctx = ctx;
210}
211
212uint64_t mb_cap(const mb_store *s) { return s->cap; }
213bool mb_persisted(const mb_store *s) { return s->dir != NULL; }
214uint64_t mb_bytes(const mb_store *s) { return s->total; }
215size_t mb_count(const mb_store *s) { return s->index.n; }
216
217uint64_t mb_refs_total(const mb_store *s) {
218 uint64_t t = 0, k;
219 for (const uint64_t *c = momap_first(s->jobs, &k); c; c = momap_next(s->jobs, &k, &k)) t += *c;
220 return t;
221}
222
223static entry *entry_of(const mb_store *s, const mj_hash *h) {
224 slot *x = tfind(&s->index, h);
225 return x ? &x->e : NULL;
226}
227
228/* The first reference (in job order) that is present. */
229static bool present(const entry *e, mj_job_id *job) {
230 uint64_t j;
231 for (const uint8_t *st = momap_first(e->refs, &j); st; st = momap_next(e->refs, &j, &j))
232 if (*st == PRESENT) {
233 *job = j;
234 return true;
235 }
236 return false;
237}
238
239bool mb_has(const mb_store *s, const mj_hash *h) {
240 const entry *e = entry_of(s, h);
241 mj_job_id j;
242 return e && present(e, &j);
243}
244
245bool mb_has_ref(const mb_store *s, mj_job_id job, const mj_hash *h) {
246 const entry *e = entry_of(s, h);
247 const uint8_t *st = e ? momap_get1(e->refs, job) : NULL;
248 return st && *st == PRESENT;
249}
250
251bool mb_known_ref(const mb_store *s, mj_job_id job, const mj_hash *h) {
252 const entry *e = entry_of(s, h);
253 return e && momap_get1(e->refs, job) != NULL;
254}
255
256mb_refs mb_all(const mb_store *s) {
257 mb_refs r = {0};
258 for (size_t i = 0; i < s->index.cap; i++) {
259 if (!s->index.v[i].used) continue;
260 const entry *e = &s->index.v[i].e;
261 mj_hash h = s->index.v[i].h;
262 uint64_t j;
263 for (const uint8_t *st = momap_first(e->refs, &j); st; st = momap_next(e->refs, &j, &j))
264 if (*st == PRESENT) refs_push(&r, j, &h);
265 }
266 return r;
267}
268
269char *mb_dir_of(const mb_store *s, mj_job_id job) {
270 if (!s->dir) return NULL;
271 char w[MJ_JOB_WORD_MAX];
272 mj_job_word(job, w);
273 return mfmt("%s/%s", s->dir, w);
274}
275
276static char *path_of(const mb_store *s, mj_job_id job, const mj_hash *h) {
277 char *d = mb_dir_of(s, job);
278 if (!d) return NULL;
279 char x[65];
280 mj_hex(h, x);
281 char *p = mfmt("%s/%s", d, x);
282 free(d);
283 return p;
284}
285
286/* Job's reference to h removed (the file under its name too); the blob's bytes go with its last reference. */
287static void unref(mb_store *s, mj_job_id job, const mj_hash *h) {
288 uint64_t k[4];
289 mj_hash_words(h, k);
290 entry *e = entry_of(s, h);
291 if (!e || !momap_del1(e->refs, job)) return;
292 char *p = path_of(s, job, h);
293 if (p) unlink(p);
294 free(p);
295 uint64_t *c = momap_get1(s->jobs, job);
296 if (c && --*c == 0) momap_del1(s->jobs, job);
297 if (momap_len(e->refs) == 0) {
298 s->total -= e->len;
299 momap_free(e->refs);
300 tdel(&s->index, h);
301 momap_del(s->mem, k);
302 }
303}
304
305mb_refs mb_drop_job(mb_store *s, mj_job_id job) {
306 mb_refs hs = {0};
307 for (size_t i = 0; i < s->index.cap; i++)
308 if (s->index.v[i].used && momap_get1(s->index.v[i].e.refs, job)) refs_push(&hs, job, &s->index.v[i].h);
309 for (size_t i = 0; i < hs.n; i++) unref(s, job, &hs.v[i].h);
310 momap_del1(s->jobs, job);
311 char *d = mb_dir_of(s, job);
312 if (d) mb_remove_tree(d);
313 free(d);
314 return hs;
315}
316
317bool mb_reserve(mb_store *s, mj_job_id job, mj_hash h, uint32_t len, mb_live_fn live, void *live_ctx, mb_reserved *out, char **err) {
318 if (mb_known_ref(s, job, &h)) {
319 char w[MJ_JOB_WORD_MAX], x[65];
320 mj_job_word(job, w);
321 mj_hex(&h, x);
322 fprintf(stderr, "reserve: job %s's reference to %s exists\n", w, x);
323 abort();
324 }
325 *out = (mb_reserved){0};
326 entry *e = entry_of(s, &h);
327 if (e) {
328 mj_job_id j;
329 if (present(e, &j)) {
330 out->link = true;
331 out->path = path_of(s, j, &h);
332 if (!out->path) out->path = mstrdup("");
333 }
334 add_ref(s, job, &h, len, PENDING);
335 return true;
336 }
337 while (s->total + len > s->cap) {
338 bool found = false;
339 uint64_t v;
340 for (const uint64_t *c = momap_first(s->jobs, &v); c; c = momap_next(s->jobs, &v, &v))
341 if (v != job && live(live_ctx, v) == MB_FORGOTTEN) {
342 found = true;
343 break;
344 }
345 if (!found) {
346 if (err)
347 *err = mfmt("blob store full: %llu of %llu MiB, every blob held for a job in the table (job forget a finished one to make room)",
348 (unsigned long long)(s->total >> 20), (unsigned long long)(s->cap >> 20));
349 mb_refs_free(&out->evicted);
350 return false;
351 }
352 mb_refs gone = mb_drop_job(s, v);
353 for (size_t i = 0; i < gone.n; i++) refs_push(&out->evicted, v, &gone.v[i].h);
354 mb_refs_free(&gone);
355 }
356 add_ref(s, job, &h, len, PENDING);
357 return true;
358}
359
360void mb_written(mb_store *s, mj_job_id job, const mj_hash *h, bool ok) {
361 if (!ok) {
362 unref(s, job, h);
363 return;
364 }
365 entry *e = entry_of(s, h);
366 uint8_t *st = e ? momap_get1(e->refs, job) : NULL;
367 if (st) *st = PRESENT;
368 else {
369 char *p = path_of(s, job, h);
370 if (p) unlink(p);
371 free(p);
372 }
373}
374
375void mb_put_mem(mb_store *s, mj_job_id job, mj_hash h, const void *b, size_t n) {
376 uint64_t k[4];
377 mj_hash_words(&h, k);
378 bool fresh;
379 bytes *x = momap_put(s->mem, k, &fresh);
380 if (fresh) {
381 x->p = mxalloc(n);
382 memcpy(x->p, b, n);
383 x->n = n;
384 }
385 mb_written(s, job, &h, true);
386}
387
388static uint8_t *read_file(const char *p, size_t *n) {
389 FILE *f = fopen(p, "rb");
390 if (!f) return NULL;
391 msb b = {0};
392 char buf[65536];
393 size_t r;
394 while ((r = fread(buf, 1, sizeof buf, f)) > 0) msb_put(&b, buf, r);
395 bool bad = ferror(f);
396 fclose(f);
397 if (bad) {
398 msb_free(&b);
399 return NULL;
400 }
401 *n = b.n;
402 if (!b.p) return mxalloc(1);
403 return (uint8_t *)msb_take(&b);
404}
405
406uint8_t *mb_get(mb_store *s, const mj_hash *h, size_t *n) {
407 entry *e = entry_of(s, h);
408 if (!e) return NULL;
409 size_t np = 0;
410 mj_job_id *pres = NULL;
411 uint64_t j;
412 for (const uint8_t *st = momap_first(e->refs, &j); st; st = momap_next(e->refs, &j, &j))
413 if (*st == PRESENT) {
414 pres = mxrealloc(pres, (np + 1) * sizeof *pres);
415 pres[np++] = j;
416 }
417 for (size_t i = 0; i < np; i++) {
418 uint8_t *b = NULL;
419 size_t bn = 0;
420 if (!s->dir) {
421 uint64_t k[4];
422 mj_hash_words(h, k);
423 const bytes *x = momap_get(s->mem, k);
424 if (x) {
425 b = mxalloc(x->n ? x->n : 1);
426 memcpy(b, x->p, x->n);
427 bn = x->n;
428 }
429 } else {
430 char *p = path_of(s, pres[i], h);
431 b = read_file(p, &bn);
432 free(p);
433 }
434 if (b) {
435 mj_hash got;
436 mesh_sha256(b, bn, got.b);
437 if (mj_hash_eq(&got, h)) {
438 free(pres);
439 *n = bn;
440 return b;
441 }
442 free(b);
443 }
444 if (s->log) {
445 char x[65], w[MJ_JOB_WORD_MAX];
446 mj_hex(h, x);
447 mj_job_word(pres[i], w);
448 char *line = mfmt("blobs: %s of job %s is missing or corrupt: dropped", x, w);
449 s->log(s->log_ctx, line);
450 free(line);
451 }
452 unref(s, pres[i], h);
453 refs_push(&s->lost, pres[i], h);
454 }
455 free(pres);
456 return NULL;
457}
458
459mb_refs mb_take_lost(mb_store *s) {
460 mb_refs r = s->lost;
461 s->lost = (mb_refs){0};
462 return r;
463}
464
465size_t mb_jobs(const mb_store *s, mj_job_id **out) {
466 size_t n = momap_len(s->jobs), i = 0;
467 *out = n ? mxalloc(n * sizeof **out) : NULL;
468 uint64_t j;
469 for (const void *c = momap_first(s->jobs, &j); c; c = momap_next(s->jobs, &j, &j)) (*out)[i++] = j;
470 return n;
471}