==> include/mesh/shared.h <==1/* @ajax/mesh-shared: state shared by meshd's threads, and the messages they
2 * send the driver (commands), the fleet thread (events) and the jobs thread
3 * (shared.rs).
4 *
5 * The driver thread is the only ring publisher and program-table writer:
6 * every other thread asks it through `cmd` and waits on a one-shot reply. In
7 * a fleet every lifecycle change comes from the reconciler (the committed
8 * desired state); the control port goes through the leader (`change`).
9 *
10 * Payloads that belong to the fleet and jobs areas (placement and fleet
11 * messages, links, beats, blob lists) travel as opaque pointers with their
12 * free function: this package only carries them. */
13#ifndef MESH_SHARED_H
14#define MESH_SHARED_H
15
16#include "mesh/abi.h"
17#include "mesh/chan.h"
18#include "mesh/host_block.h"
19#include "mesh/infer_svc.h"
20#include "mesh/jobs/frame.h"
21#include "mesh/loader.h"
22#include "mesh/mailbox.h"
23#include "mesh/reply.h"
24#include "mesh/views.h"
25
26#include <pthread.h>
27#include <stdatomic.h>
28#include <stdbool.h>
29#include <stddef.h>
30#include <stdint.h>
31
32/* ---- where a load's or an upgrade's image comes from (Src) ---------------------------- */
33
34typedef struct {
35 bool is_bytes; /* false: a catalog dir under --programs */
36 char *dir; /* the catalog dir */
37 mesh_payload payload; /* bytes (the local catalog's, or a member's F_IMAGE) ... */
38 uint64_t hash; /* ... and the image hash they must have */
39} mesh_src;
40
41mesh_src mesh_src_catalog(const char *dir);
42mesh_src mesh_src_bytes(const mesh_payload *p, uint64_t hash); /* copies p */
43mesh_src mesh_src_copy(const mesh_src *s);
44void mesh_src_free(mesh_src *s);
45/* "dir" or "dir (from the peer)", as the driver's log lines name a source. */
46char *mesh_src_what(const mesh_src *s);
47
48/* ---- driver commands (Cmd) ------------------------------------------------------------ */
49
50typedef enum {
51 MESH_CMD_LOAD, /* src, run (-1: the manifest's autostart), reply (may be NULL) */
52 MESH_CMD_UPGRADE, /* name, src (has_src false: the slot's current catalog dir), reply */
53 MESH_CMD_ROLLBACK, /* name, reply */
54 MESH_CMD_UNLOAD, /* name, now, reply */
55 MESH_CMD_PROG_RUN, /* name, on, reply */
56 MESH_CMD_TASK, /* name, op, arg, reply (Ok value) */
57 MESH_CMD_REQ, /* name, bytes, reply (Ok bytes) */
58 MESH_CMD_INPUT, /* name, bytes, reply */
59 MESH_CMD_RESET, /* name, reply */
60 MESH_CMD_PLACE, /* name, words (has_words false: report), reply */
61 MESH_CMD_PEER_PLACE, /* msg (a placement message), from */
62 MESH_CMD_PEER_FRAME, /* msg (a frame message) */
63 MESH_CMD_PEER_INPUT, /* name, bytes */
64 MESH_CMD_FLEET_VIEW, /* up, has_leader/leader, places */
65 MESH_CMD_INVENTORY, /* reply (Ok ptr: mesh_actuals) */
66 MESH_CMD_VIEW_GET, /* name, reply (Ok ptr: mesh_view_bytes) */
67 MESH_CMD_VIEW_PUT, /* name, view_abi, bytes, reply */
68 MESH_CMD_JOB_LEASE, /* lease */
69 MESH_CMD_JOB_CUT, /* lease_id, cut */
70 MESH_CMD_JOB_PROGS, /* reply (Ok ptr: mesh_job_progs) */
71} mesh_cmd_kind;
72
73/* A committed placement request (FleetView). */
74typedef struct {
75 char *name;
76 char **words;
77 size_t nwords;
78} mesh_place_req;
79
80typedef struct {
81 mesh_cmd_kind k;
82 char *name;
83 bool has_src;
84 mesh_src src;
85 int run;
86 bool now, on;
87 uint32_t op, arg;
88 uint8_t *bytes;
89 size_t len;
90 bool has_words;
91 char **words;
92 size_t nwords;
93 void *msg;
94 void (*msg_free)(void *msg);
95 uint32_t from;
96 uint32_t up;
97 bool has_leader;
98 uint32_t leader;
99 mesh_place_req *places;
100 size_t nplaces;
101 uint32_t view_abi;
102 mj_lease_msg lease;
103 uint64_t lease_id, cut;
104 mesh_reply *reply;
105} mesh_cmd;
106
107mesh_cmd mesh_cmd_new(mesh_cmd_kind k);
108/* Frees what the command owns; an unanswered reply is dropped (its waiter sees that at once). */
109void mesh_cmd_free(mesh_cmd *c);
110
111/* Every loaded program as it is now (Inventory: the reconciler's input; fleet.rs Actual). */
112typedef struct {
113 char *name, *dir;
114 uint64_t hash;
115 bool has_prev;
116 uint64_t prev;
117 bool run, failed, lockstep;
118} mesh_actual;
119
120typedef struct {
121 mesh_actual *v;
122 size_t n;
123} mesh_actuals;
124
125void mesh_actuals_free(void *p); /* a heap mesh_actuals */
126
127/* A program's view in flight between members (a migration). */
128typedef struct {
129 uint32_t view_abi;
130 uint8_t *bytes;
131 size_t len;
132} mesh_view_bytes;
133
134void mesh_view_bytes_free(void *p); /* a heap mesh_view_bytes */
135
136/* The job programs running here (jobs.rs Prog). */
137typedef struct {
138 mj_prog *v;
139 size_t n;
140} mesh_job_progs;
141
142void mesh_job_progs_free(void *p);
143
144/* ---- fleet thread events (PeerEvent) -------------------------------------------------- */
145
146typedef enum {
147 MESH_PEV_LINK, /* msg: a JOINed link */
148 MESH_PEV_CLOSED, /* from, lid, text (why) */
149 MESH_PEV_FLEET, /* from, lid, msg */
150 MESH_PEV_BEAT, /* from, lid, msg */
151 MESH_PEV_HELLO, /* from, lid, msg */
152 MESH_PEV_CTL, /* from, lid, id, line */
153 MESH_PEV_CTL_REPLY, /* from, lid, id, ok, text */
154 MESH_PEV_IMAGE_REQ, /* from, lid, id, name, hash, dir */
155 MESH_PEV_IMAGE, /* from, lid, id, msg (or ok false and text) */
156 MESH_PEV_CHANGE, /* line, reply (NULL: nobody waits) */
157 MESH_PEV_FETCH_IMAGE,/* name, hash, dir, reply */
158 MESH_PEV_WAKE,
159 MESH_PEV_VIEW_REQ, /* from, lid, id, name, hash */
160 MESH_PEV_VIEW, /* from, lid, id, msg */
161 MESH_PEV_FETCH_VIEW, /* name, hash, from, reply */
162} mesh_peer_event_kind;
163
164typedef struct {
165 mesh_peer_event_kind k;
166 uint32_t from, id;
167 uint64_t lid, hash;
168 bool ok;
169 char *line, *text, *name, *dir;
170 void *msg;
171 void (*msg_free)(void *msg);
172 mesh_reply *reply;
173} mesh_peer_event;
174
175mesh_peer_event mesh_peer_event_new(mesh_peer_event_kind k);
176void mesh_peer_event_free(mesh_peer_event *e);
177
178/* ---- jobs thread events (JobEvent) ---------------------------------------------------- */
179
180typedef enum {
181 MESH_JEV_FRAME, /* from, ty, body */
182 MESH_JEV_CTL, /* line, reply_to (LOCAL: reply; REMOTE: remote_from, remote_id) */
183 MESH_JEV_DONE, /* res, has_payload/payload */
184 MESH_JEV_BLOBS_STORED, /* msg */
185 MESH_JEV_BLOB_ACKED, /* from, msg, ok */
186 MESH_JEV_BLOB_GONE, /* msg */
187} mesh_job_event_kind;
188
189typedef enum { MESH_JOB_REPLY_NONE, MESH_JOB_REPLY_LOCAL, MESH_JOB_REPLY_REMOTE } mesh_job_reply;
190
191typedef struct {
192 mesh_job_event_kind k;
193 uint32_t from;
194 uint8_t ty;
195 uint8_t *body;
196 size_t len;
197 char *line;
198 mesh_job_reply reply_to;
199 mesh_reply *reply;
200 uint32_t remote_from, remote_id;
201 mj_item_res res;
202 bool has_payload;
203 uint8_t *payload;
204 size_t payload_len;
205 bool ok;
206 void *msg;
207 void (*msg_free)(void *msg);
208} mesh_job_event;
209
210mesh_job_event mesh_job_event_new(mesh_job_event_kind k);
211void mesh_job_event_free(mesh_job_event *e);
212
213/* ---- link frames ------------------------------------------------------------------------ */
214
215/* A frame for a peer link (type byte first); an empty one is a doorbell. */
216typedef struct {
217 uint8_t *b;
218 size_t n;
219} mesh_link_frame;
220
221/* A bounded queue of link frames (a link's writer drains it). */
222mesh_chan *mesh_link_queue_new(size_t bound);
223
224/* ---- the shared state ------------------------------------------------------------------ */
225
226#define MESH_PEER_SINGLE 0u
227#define MESH_PEER_WAITING 1u
228#define MESH_PEER_UP 2u
229
230typedef struct {
231 double sum;
232 uint32_t n;
233} mesh_rtt_acc;
234
235/* A sched explain text per program. */
236typedef struct {
237 char *name, *text;
238} mesh_explain;
239
240typedef struct mesh_shared {
241 mesh_mailbox mb;
242 mesh_host_block host;
243 char *programs; /* --programs DIR: the only place `load` reads from */
244 mesh_chan *cmd; /* mesh_cmd, to the driver */
245 mesh_chan *peer; /* mesh_peer_event, to the fleet thread; NULL single */
246 uint32_t me, n; /* this system and the cluster's size (0 and 1 single) */
247 _Atomic uint32_t peer_state;
248 _Atomic bool lock_feed; /* the lockstep program may be fed */
249 /* Per member, over the current second: exchange round trips and waits for its rounds (under mu). */
250 mesh_rtt_acc rtt[MESH_MAX_SYSTEMS], wait[MESH_MAX_SYSTEMS];
251 _Atomic uint64_t round_of[MESH_MAX_SYSTEMS];
252 _Atomic bool ready; /* the startup loads are done */
253 _Atomic bool *quit; /* set by SIGINT/SIGTERM */
254 _Atomic bool fatal; /* set by fail_stop */
255 /* Each linked member's frame queue and bulk queue (mesh_link_frame), under mu; NULL: no link. */
256 mesh_chan *link_out[MESH_MAX_SYSTEMS], *bulk_out[MESH_MAX_SYSTEMS];
257 mesh_chan *blobs; /* the blob thread's events (jobs area); its messages are that area's */
258 _Atomic uint64_t blob_mem;
259 mesh_chan *jobs; /* mesh_job_event, to the jobs thread */
260 /* Texts and views other threads refresh (under mu): `fleet` and `groups`, the scheduler's, the views. */
261 char *fleet_text, *groups_text;
262 char *sched_all, *sched_dry;
263 mesh_explain *sched_explain;
264 size_t nexplain;
265 mesh_recs fleet_view, jobs_view_recs;
266 /* The pipes view's last sample per inbox (control.rs pipes_recs), under mu. */
267 struct mesh_pipe_sample *pipe_hist;
268 size_t npipe_hist;
269 /* Where the members' inference services listen (set once at start). */
270 bool has_svc;
271 mesh_infer_svc svc;
272 /* The fleet's and the jobs' own state (their areas own these). */
273 void *fleet_ext, *jobs_ext;
274 pthread_mutex_t mu;
275} mesh_shared;
276
277/* The pipes view's sample of an inbox: when, head, tail, full, rates in/out/refused per s. */
278typedef struct mesh_pipe_sample {
279 char name[MESH_NAME_BYTES + 1];
280 int64_t at_ns;
281 uint32_t head, tail, full;
282 float rates[3];
283} mesh_pipe_sample;
284
285/* A Shared for meshd, tests and tools: mailbox mb, member me of n (n = 1: single; peered: a fleet thread's channel). */
286mesh_shared *mesh_shared_new(mesh_mailbox mb, const char *programs, bool peered, uint32_t me, uint32_t n, _Atomic bool *quit);
287void mesh_shared_free(mesh_shared *sh);
288
289/* Microseconds since this process started (a monotonic clock for link RTTs). */
290uint64_t mesh_shared_mono_us(void);
291
292/* Queues a frame for system to (copied); false when it is not linked, its queue is full, or after fail_stop. */
293bool mesh_shared_send_peer(mesh_shared *sh, uint32_t to, const uint8_t *frame, size_t n);
294/* Waits up to within_ms for room in to's queue. */
295bool mesh_shared_send_peer_wait(mesh_shared *sh, uint32_t to, const uint8_t *frame, size_t n, int64_t within_ms);
296/* A bulk frame (a blob), written when the control frames are out; a doorbell wakes the writer. */
297bool mesh_shared_send_bulk(mesh_shared *sh, uint32_t to, const uint8_t *frame, size_t n);
298/* Every linked member in mask; false if none is or a queue refused it. */
299bool mesh_shared_send_mask(mesh_shared *sh, uint32_t mask, const uint8_t *frame, size_t n);
300bool mesh_shared_broadcast(mesh_shared *sh, const uint8_t *frame, size_t n);
301/* Fail-stop: this member's state could not be made durable; nothing more leaves it, and meshd shuts down. */
302void mesh_shared_fail_stop(mesh_shared *sh, const char *why);
303
304/* A driver command with a one-line reply, bounded in time. cmd's reply is set
305 * here (any given one is replaced). Returns the Ok text or the Err text in
306 * *text (free it) and whether it was Ok. "driver gone", "no answer from the
307 * driver in 60s" as the Rust. */
308bool mesh_shared_ask(mesh_shared *sh, mesh_cmd cmd, int64_t within_ms, char **text);
309void mesh_shared_set_peer_state(mesh_shared *sh, uint32_t s);
310/* A change to the desired state through the fleet (the leader commits it). */
311bool mesh_shared_change(mesh_shared *sh, const char *line, int64_t within_ms, char **text);
312/* An HTTP request head through the GPU's http program (`req http`); counted in
313 * the host block. Ok: *resp (free it). Err: *err. */
314bool mesh_shared_http(mesh_shared *sh, const uint8_t *head, size_t n, uint8_t **resp, size_t *rn, char **err);
315
316/* Locked accessors for the texts other threads refresh: a copy (free it). */
317char *mesh_shared_text(mesh_shared *sh, char *const *field);
318void mesh_shared_set_text(mesh_shared *sh, char **field, const char *text);
319
320#endif