1/* @ajax/mesh-jobs-node: a meshd's job node, sans IO (jobs.rs `Node`): its replica of the fleet's job
2 * table, the leader's dispatcher when it leads, and its side of the leases. jobd runs one per meshd
3 * over the peer links, the driver and a persisted log (@ajax/mesh-jobd); the tests run several over a
4 * simulated network.
5 *
6 * The node never does IO: every input (a view of the fleet, a frame, a finished item, a stored or
7 * acknowledged blob, a request, a tick) returns outputs (@ajax/mesh-jobs-out) for the caller to make
8 * durable (`mj_durable`) and act on, in order.
9 *
10 * The struct is open so jobd's views and the tests can read it; change it only through the
11 * functions. */
12#ifndef MESH_JOBS_NODE_H
13#define MESH_JOBS_NODE_H
14
15#include "mesh/jobs/frame.h"
16#include "mesh/jobs/id.h"
17#include "mesh/jobs/log.h"
18#include "mesh/jobs/merge.h"
19#include "mesh/jobs/out.h"
20#include "mesh/jobs/rate.h"
21#include "mesh/jobs/req.h"
22#include "mesh/jobs/select.h"
23#include "mesh/jobs/table.h"
24#include "mesh/omap.h"
25
26#include <stdbool.h>
27#include <stddef.h>
28#include <stdint.h>
29
30/* A member re-sends a result the table does not show yet after this long (ms). */
31#define MESH_JOBS_RESEND_MS 3000u
32/* A member's F_JOB_AVAIL every second (and on any change). */
33#define MESH_JOBS_AVAIL_MS 1000u
34/* A member whose lease expired gets no new lease of that program for this long (or 4 x the dead time). */
35#define MESH_JOBS_HOLDOFF_MS 10000u
36/* An expired lease's items each count a try only up to this many items. */
37#define MESH_JOBS_EXPIRY_TRIES_MAX 10000u
38/* `job results` lines per request, at most. */
39#define MESH_JOBS_RESULTS_MAX 100000u
40/* A submit for a program no member reports yet waits this long for one. */
41#define MESH_JOBS_SUBMIT_WAIT_MS 2000u
42
43typedef struct {
44    uint32_t me, n, quorum;
45    uint64_t dead_ms;
46} mj_cfg;
47
48/* The fleet as the jobs thread sees it: the fleet term, the leader, whom it hears (alive) and whom it
49 * is linked to. A single meshd: term 1, itself. */
50typedef struct {
51    uint64_t term;
52    bool has_leader;
53    uint32_t leader;
54    uint32_t alive, linked;
55} mj_view;
56
57static inline bool mj_view_eq(const mj_view *a, const mj_view *b) {
58    return a->term == b->term && a->has_leader == b->has_leader && (!a->has_leader || a->leader == b->leader) && a->alive == b->alive &&
59           a->linked == b->linked;
60}
61
62/* A lease, as the leader tracks it. */
63typedef struct {
64    uint64_t id;
65    mj_job_id job;
66    uint32_t member;
67    char *program;
68    mj_ranges out; /* items not reported yet */
69    uint64_t size, issued, deadline;
70} mj_lease;
71
72typedef struct {
73    mj_ranges fresh, retry;
74} mj_queue;
75
76/* A small ordered map keyed by (member, program): the leader's rates and hold-offs. */
77typedef struct {
78    uint32_t member;
79    char *program;
80    mj_rate rate;    /* rates */
81    uint64_t until;  /* holdoff */
82} mj_mp;
83
84typedef struct {
85    mj_mp *v;
86    size_t n, cap;
87} mj_mpmap;
88
89mj_mp *mj_mpmap_get(const mj_mpmap *m, uint32_t member, const char *program);
90mj_mp *mj_mpmap_put(mj_mpmap *m, uint32_t member, const char *program); /* zeroed if new */
91
92/* The leader's dispatcher (soft state: a new leader rebuilds it). */
93typedef struct {
94    momap *leases;          /* id -> mj_lease */
95    momap *queues;          /* job -> mj_queue */
96    mj_member_progs progs;  /* per member: the job programs it runs (F_JOB_AVAIL) */
97    mj_mpmap rates;         /* (member, program) -> rate */
98    mj_mpmap holdoff;       /* (member, program) -> until: no new lease of it there */
99    uint64_t next_lease;
100    momap *picks;           /* job -> mj_picks: the last select per job (for `job status`) */
101} mj_disp;
102
103typedef struct {
104    uint64_t n, token;
105    bool ok;
106    char *text;
107} mj_wait;
108
109typedef struct {
110    mj_submit s;
111    uint64_t token, until;
112} mj_held_submit;
113
114typedef struct {
115    uint32_t member;
116    mj_held h;
117} mj_adopt;
118
119typedef struct {
120    uint32_t from;
121    mj_item_res r;
122} mj_parked;
123
124typedef struct {
125    mj_parked *v;
126    size_t n, cap;
127} mj_parked_list;
128
129typedef struct {
130    uint64_t lid, term;
131    bool active;
132    /* Pulling: who answered (wholly), the answers, the leases members hold. */
133    uint32_t responded;
134    mj_answer *answers[MJ_MEMBERS];
135    mj_adopt *adopt;
136    size_t nadopt, capadopt;
137    uint64_t pull_at;
138    uint32_t heard_mask;
139    uint64_t heard[MJ_MEMBERS]; /* when a part of a member's answer last arrived */
140    mj_table working;
141    uint64_t n; /* the last entry number used */
142    uint64_t *acked; /* per member (cfg.n) */
143    uint64_t commit;
144    uint32_t snap_due;
145    mj_ops ops;  /* ops of this tick, not in a delta yet */
146    mj_wait *waits;
147    size_t nwaits, capwaits;
148    mj_disp disp;
149    momap *dupes; /* job -> uint64_t duplicates counted here */
150    mj_held_submit *held;
151    size_t nheld, capheld;
152    momap *parked;   /* (job, hash words) -> mj_parked_list: results waiting for their blob here */
153    momap *fetch_at; /* (job, hash words) -> uint64_t: when each parked blob was last asked for */
154} mj_lead;
155
156/* A payload result before its copies are durable. */
157typedef struct {
158    mj_item_res r;
159    uint32_t acked, refused;
160    uint32_t sent_mask;
161    uint64_t sent[MJ_MEMBERS]; /* when each member was last pushed to */
162} mj_staged;
163
164typedef struct {
165    mj_item_res r;
166    uint64_t at;
167} mj_outbox_entry;
168
169typedef struct {
170    mj_cfg cfg;
171    mj_view view;
172    mj_replica rep; /* this member's durable state (its log replays to it) */
173    bool has_lid_from;
174    uint32_t lid_from; /* the member that leads rep.lid */
175    uint64_t next;
176    bool has_staging;
177    uint64_t staging_n;
178    uint32_t staging_seq;
179    mj_table staging; /* a table arriving in parts */
180    uint64_t gap_at;
181    mj_lead *lead;
182    uint64_t behind;
183    /* The member side. */
184    momap *held;   /* lease -> mj_held */
185    momap *outbox; /* (job, item, lease) -> mj_outbox_entry */
186    mj_prog *progs;
187    size_t nprogs;
188    uint64_t avail_at;
189    bool avail_dirty;
190    uint64_t prune_at;
191    mj_rate rate;
192    mj_frame *lo; /* frames to this member itself (the leader is a member too) */
193    size_t nlo, lo_head, caplo;
194    mj_sched *sched; /* the fleet scheduler's model (NULL: a single meshd); owned */
195    momap *staged; /* (job, item, lease) -> mj_staged */
196    momap *have;   /* (job, hash words): the jobs' blobs this member's store holds durably */
197} mj_node;
198
199/* A node from its replayed log (takes rep). */
200mj_node *mj_node_new(mj_cfg cfg, mj_replica rep);
201void mj_node_free(mj_node *n);
202
203const mj_table *mj_node_committed(const mj_node *n);
204/* The fleet scheduler's model (every second, from the fleet thread); the node takes it. */
205void mj_node_set_sched(mj_node *n, mj_sched *cx);
206/* The leader's last member choice for a job (NULL: not leading, or nothing to lease). */
207const mj_picks *mj_node_picks(const mj_node *n, mj_job_id job);
208bool mj_node_leading(const mj_node *n);
209bool mj_node_active(const mj_node *n);
210uint64_t mj_node_lid(const mj_node *n);
211uint64_t mj_node_commit_n(const mj_node *n);
212/* The leader's working table (results not committed yet included), or NULL. */
213const mj_table *mj_node_working(const mj_node *n);
214/* The leader's leases (id -> mj_lease), or NULL. */
215const momap *mj_node_leases(const mj_node *n);
216const mj_mpmap *mj_node_rates(const mj_node *n);
217bool mj_node_queued(const mj_node *n, mj_job_id job, uint64_t *count);
218double mj_node_my_rate(const mj_node *n);
219size_t mj_node_unconfirmed(const mj_node *n);
220size_t mj_node_staged(const mj_node *n);
221size_t mj_node_parked(const mj_node *n);
222bool mj_node_has_blob(const mj_node *n, mj_job_id job, const mj_hash *h);
223
224/* The fleet's view changed (or the periodic refresh). */
225mj_outs mj_node_set_view(mj_node *n, mj_view v, uint64_t now, uint64_t unix_ms);
226/* A frame from member `from` (taken). */
227mj_outs mj_node_recv(mj_node *n, uint32_t from, mj_frame f, uint64_t now, uint64_t unix_ms);
228/* A finished item from the local driver (taken); a payload's blob already handed to blobd. */
229mj_outs mj_node_result(mj_node *n, mj_item_res r, uint64_t now);
230/* Job's blob `hash` is durable in this member's store. */
231mj_outs mj_node_blob_stored(mj_node *n, mj_job_id job, mj_hash hash, uint64_t now, uint64_t unix_ms);
232/* Job's blob `hash` left this member's store. */
233void mj_node_blob_gone(mj_node *n, mj_job_id job, const mj_hash *hash);
234/* Member `from` acknowledged job's blob durable (ok), or its store refused it. */
235void mj_node_blob_acked(mj_node *n, uint32_t from, mj_job_id job, mj_hash hash, bool ok);
236/* The job programs running here (from the driver; taken). */
237void mj_node_set_progs(mj_node *n, mj_prog *progs, size_t count);
238/* A change asked for (taken); answered by token after its commit. */
239mj_outs mj_node_request(mj_node *n, mj_req req, uint64_t token, uint64_t now, uint64_t unix_ms);
240/* The periodic work: the leader's dispatch, deltas and commits; a member's results and availability. */
241mj_outs mj_node_tick(mj_node *n, uint64_t now, uint64_t unix_ms);
242
243/* This member's answer to a pull (numbered F_JOB_HAVE parts), as frames (for tests). */
244void mj_node_answer(const mj_node *n, uint64_t lid, mj_frame **out, size_t *count);
245/* The leader's handling of results from `from` (taken; for tests: jobs.rs's tests call it directly). */
246void mj_node_on_results(mj_node *n, uint32_t from, mj_item_res *items, size_t count, uint64_t now, uint64_t unix_ms, mj_outs *out);
247
248#endif