1/* @ajax/mesh-blobd: the blob thread's logic (blobd.rs), sans threads: this member's blob store
2 * (@ajax/mesh-blobs-store), the blob frames on the peer links, and the image assemblies.
3 *
4 * - Its own payloads (jobd, MBD_LOCAL): reserved in the store, written by the writers (.tmp, fsync,
5 * rename; a directory fsync per batch), kept in memory until their copies are durable (MBD_DONE);
6 * their bytes in memory are counted (blob_mem: jobd stops taking leases past MBD_BLOB_MEM_MAX).
7 * - Pushes (jobs MJ_OUT_PUSH_BLOB): F_BLOB_PUT through the link's bulk queue, at most PUT_WINDOW
8 * unacknowledged per member; F_BLOB_ACK once the receiver's copy is durable (or refused: full).
9 * - Every copy is a job's: a PUT, an own payload or a kept fetch makes that job's reference, even when
10 * the same bytes are here for another job.
11 * - Fetches (the leader's parked results; an assembly's next passes): F_BLOB_GET to a holder (else any
12 * member up), one at a time per blob, the next holder on a refusal, a timeout or a blob that does not
13 * hash to its name; at most FETCH_MAX in flight.
14 * - Assemblies: the jobs a viewer asked for (`render`) and, on the leader, every active one: jobd sends
15 * the committed results of the watched jobs (MBD_TABLE); each tile's passes are read (local) or
16 * fetched and folded in pass order (@ajax/mesh-assemble).
17 *
18 * The world is an mbd_io: the peer links, jobd's channel, the writers, the clock. */
19#ifndef MESH_BLOBD_H
20#define MESH_BLOBD_H
2122#include "mesh/blobs/store.h"
23#include "mesh/jobs/id.h"
24#include "mesh/jobs/spec.h"
25#include "mesh/assemble.h"
2627#include <stdbool.h>
28#include <stddef.h>
29#include <stdint.h>
3031#define MESH_BLOBD_PUT_WINDOW 128u /* unacknowledged pushes per member */
32#define MESH_BLOBD_PUT_TIMEOUT_MS 5000u /* a push not acknowledged in this long is forgotten here */
33#define MESH_BLOBD_FETCH_MAX 256u /* blob fetches in flight, at most */
34#define MESH_BLOBD_FETCH_TIMEOUT_MS 2000u
35#define MESH_BLOBD_LOCAL_READS 400u /* an assembly's local reads per tick, at most */
36#define MESH_BLOBD_BLOB_MEM_MAX ((uint64_t)256 << 20)
37#define MESH_BLOBD_WRITERS 3u
3839/* Why a blob is being written. */
40typedef enum { MBD_WHY_LOCAL, MBD_WHY_PUT, MBD_WHY_FETCHED } mbd_why;
4142/* A reference to make durable (for the writers). */
43typedef struct {
44 char *dir; /* its job's directory */
45 mj_hash hash;
46 char *link; /* a file of the same bytes to hard-link (another job's), or NULL */
47 uint8_t *bytes;
48 size_t n;
49 mbd_why why;
50 uint32_t from; /* MBD_WHY_PUT: the pusher */
51 mj_job_id job;
52} mbd_write;
5354void mbd_write_free(mbd_write *w);
5556/* What the jobs thread knows of a job (every 250 ms, from the committed table). */
57typedef struct {
58 mj_job_id id;
59 char *program;
60 uint32_t op;
61 mj_setup *setup;
62 size_t nsetup;
63 uint32_t tone;
64 bool active;
65 char *state;
66} mbd_job_info;
6768/* A watched image's new committed payload result. */
69typedef struct {
70 mj_job_id job;
71 uint64_t item;
72 masm_item it;
73} mbd_result;
7475/* The committed table's news for this thread. */
76typedef struct {
77 mbd_job_info *jobs;
78 size_t njobs;
79 mj_job_id *forgotten;
80 size_t nforgotten;
81 mbd_result *results;
82 size_t nresults;
83} mbd_news;
8485void mbd_news_free(mbd_news *n);
8687/* The world as blobd sees it. Frames are whole (type, u32 length, body). */
88typedef struct {
89 void *ctx;
90 bool (*send_peer)(void *ctx, uint32_t to, const uint8_t *frame, size_t n);
91 bool (*send_bulk)(void *ctx, uint32_t to, const uint8_t *frame, size_t n);
92 /* To jobd: job's copies of these hashes are durable here; member `from` acknowledged (or refused) a
93 * copy; these references are gone. */
94 void (*stored)(void *ctx, const mb_ref *refs, size_t n);
95 void (*acked)(void *ctx, uint32_t from, mj_job_id job, const mj_hash *h, bool ok);
96 void (*gone)(void *ctx, const mb_ref *refs, size_t n);
97 uint32_t (*alive)(void *ctx); /* members alive (this one included) */
98 bool (*leading)(void *ctx); /* this member leads the job table (or a single meshd) */
99 void (*watching)(void *ctx, mj_job_id job, bool on); /* the images assembled here (jobd sends their results) */
100 void (*mem)(void *ctx, int64_t delta); /* bytes of own payloads in memory */
101 void (*write)(void *ctx, mbd_write w); /* to the writers (takes w); they answer mbd_written */
102 int64_t (*now_ms)(void *ctx);
103 void (*log)(void *ctx, bool error, const char *line);
104} mbd_io;
105106typedef struct mbd mbd;
107108/* Takes store. At start the jobs thread learns what is here (io.stored). */
109mbd *mbd_new(uint32_t me, mb_store *store, mbd_io io);
110void mbd_free(mbd *b);
111mb_store *mbd_store(mbd *b);
112113/* The events (blobd.rs BlobEvent). */
114void mbd_local(mbd *b, mj_job_id job, mj_hash hash, uint8_t *bytes, size_t n); /* takes bytes */
115void mbd_push(mbd *b, uint32_t to, mj_job_id job, mj_hash hash);
116void mbd_fetch(mbd *b, mj_job_id job, mj_hash hash, uint32_t holders);
117void mbd_done(mbd *b, mj_job_id job, mj_hash hash);
118/* A blob frame from a peer link: its type and the body after the length. */
119void mbd_frame(mbd *b, uint32_t from, uint8_t ty, const uint8_t *body, size_t n);
120void mbd_table(mbd *b, mbd_news *news); /* takes news */
121/* A writer finished job's reference to hash (ok, or why not). */
122void mbd_written(mbd *b, mj_job_id job, mj_hash hash, mbd_why why, uint32_t from, bool ok, const char *err);
123/* `render [<job>|latest] [status|ppm|bmp|raw]`: the answer's bytes, or false with *err. */
124bool mbd_render(mbd *b, const char *const *words, size_t n, uint8_t **out, size_t *len, char **err);
125/* The periodic work: pushes, the assemblies' reads and fetches, fetch timeouts, stale assemblies. */
126void mbd_tick(mbd *b);
127/* References whose file turned out missing or corrupt since the last call go to jobd (io.gone). */
128void mbd_report_lost(mbd *b);
129130/* The writers' work on one batch: each reference made durable (mb_make_durable), each directory fsynced
131 * once; ok[i] and err[i] (free it) per write. */
132void mbd_write_batch(mbd_write *ws, size_t n, bool *ok, char **err);
133134#endif