==> src/fleetd.c <==1/* The fleet member's start, its bridges and its stop (main.rs's fleet thread; peer.rs `run`'s start). */
2#include "fleetd_int.h"
3
4#include "mesh/abi.h"
5#include "mesh/buf.h"
6#include "mesh/fleet_store.h"
7#include "mesh/log.h"
8
9#include <stdio.h>
10#include <stdlib.h>
11#include <string.h>
12#include <unistd.h>
13
14mesh_fleetd *mesh_fleetd_open(mesh_shared *sh, const mcc_topology *topo, const char *state_path, bool *skip_startup, char **err) {
15 *skip_startup = false;
16 if (state_path) {
17 /* Read (and written back, so its directory is known writable) before anything runs; the fleet
18 * thread opens it again when it starts. */
19 mfs_store st;
20 mfp_persist p;
21 if (!mfs_open(state_path, topo->n, &st, &p, err)) return NULL;
22 *skip_startup = p.committed.version > 0;
23 mfp_free(&p);
24 mfs_close(&st);
25 }
26 mesh_fleetd *f = mesh_xcalloc(1, sizeof *f);
27 f->sh = sh;
28 f->topo = *topo;
29 f->topo.listen = topo->listen ? mesh_xstrdup(topo->listen) : NULL;
30 for (size_t i = 0; i < topo->ndial; i++) f->topo.dial_addr[i] = mesh_xstrdup(topo->dial_addr[i]);
31 f->state_path = state_path ? mesh_xstrdup(state_path) : NULL;
32 pthread_mutex_init(&f->mu, NULL);
33 f->reason = mesh_xstrdup("");
34 atomic_init(&f->relay, NULL);
35 atomic_init(&f->starting, false);
36 atomic_init(&f->stop, false);
37 return f;
38}
39
40static void put_str(char *out, size_t cap, const char *s) {
41 snprintf(out, cap, "%s", s ? s : "");
42}
43
44mpk_caps mesh_fleetd_caps(mesh_shared *sh, const char *gpu, const char *arch) {
45 mesh_mailbox mb = sh->mb;
46 mpk_caps c;
47 memset(&c, 0, sizeof c);
48 c.cc = mesh_mailbox_get(mb, MESH_H_CC);
49 c.sms = mesh_mailbox_get(mb, MESH_H_SMS);
50 c.sms_total = mesh_mailbox_get(mb, MESH_H_SMS_TOTAL);
51 c.blocks_per_sm = mesh_mailbox_get(mb, MESH_H_BLOCKS_PER_SM);
52 c.backend = mesh_mailbox_get(mb, MESH_H_BACKEND);
53 c.max_threads = mesh_mailbox_get(mb, MESH_H_MAX_TG_THREADS);
54 c.max_shared = mesh_mailbox_get(mb, MESH_H_MAX_SHARED);
55 put_str(c.gpu, sizeof c.gpu, gpu);
56 put_str(c.arch, sizeof c.arch, arch);
57 char host[256] = "";
58 gethostname(host, sizeof host - 1);
59 put_str(c.host, sizeof c.host, host);
60 return c;
61}
62
63mpr_relay *fd_relay(mesh_fleetd *f) {
64 for (int i = 0; i < 5000; i++) {
65 mpr_relay *r = atomic_load(&f->relay);
66 if (r || !atomic_load(&f->starting) || atomic_load(&f->stop)) return r;
67 mesh_sleep_us(1000);
68 }
69 return atomic_load(&f->relay);
70}
71
72mpr_relay *mesh_fleetd_relay(mesh_fleetd *f) { return atomic_load(&f->relay); }
73
74static bool quitting(mesh_fleetd *f) { return atomic_load(&f->stop) || atomic_load(f->sh->quit); }
75
76/* The fleet thread: the links wait until the startup loads ran (the first leader's document is what
77 * it runs). A listener that cannot be bound ends it (the Rust's thread returns): this member then
78 * runs alone, and changes are refused. */
79static void *fleet_main(void *p) {
80 mesh_fleetd *f = p;
81 while (!atomic_load(&f->sh->ready) && !quitting(f)) mesh_sleep_us(10000);
82 if (quitting(f)) {
83 atomic_store(&f->starting, false);
84 return NULL;
85 }
86 char *err = NULL;
87 mpr_relay *r = mpr_start(&f->topo, &f->caps, f->state_path, &f->host, &err);
88 if (!r) {
89 mesh_error("fleet: %s", err ? err : "the fleet thread did not start");
90 free(err);
91 mesh_chan_close_rx(f->sh->peer);
92 } else {
93 atomic_store(&f->relay, r);
94 mesh_info("system %u of %u: a fleet member (a lockstep program runs once its group's members HELLO)", f->topo.me, f->topo.n);
95 }
96 atomic_store(&f->starting, false);
97 return NULL;
98}
99
100/* The `peer` channel's events from other threads: a change (control port, driver, jobs thread) and a
101 * wake (the reconciler). It drains once the relay runs; until then they wait, as in the Rust's
102 * channel. */
103static void *bridge_main(void *p) {
104 mesh_fleetd *f = p;
105 mesh_shared *sh = f->sh;
106 while (!quitting(f) && !atomic_load(&f->relay) && (atomic_load(&f->starting) || !atomic_load(&sh->ready))) mesh_sleep_us(10000);
107 mpr_relay *r = atomic_load(&f->relay);
108 if (!r) return NULL;
109 while (!quitting(f)) {
110 mesh_peer_event e;
111 int got = mesh_chan_recv(sh->peer, &e, 100);
112 if (got == MESH_CHAN_CLOSED) break;
113 if (got != MESH_CHAN_OK) continue;
114 switch (e.k) {
115 case MESH_PEV_CHANGE:
116 mpr_change(r, e.line ? e.line : "", (mpr_reply){e.reply ? fd_reply_fn : NULL, e.reply});
117 e.reply = NULL;
118 break;
119 case MESH_PEV_WAKE: mpr_wake(r); break;
120 default: mesh_error("fleet: an event of kind %d on the peer channel, which the relay receives itself; dropped", (int)e.k);
121 }
122 mesh_peer_event_free(&e);
123 }
124 return NULL;
125}
126
127typedef struct {
128 mesh_fleetd *f;
129 uint32_t k;
130} fwd_arg;
131
132/* The member's link queue to take from (retained), as the linked mask left it. */
133static mesh_chan *queue_of(mesh_shared *sh, uint32_t k, bool bulk) {
134 pthread_mutex_lock(&sh->mu);
135 mesh_chan *q = bulk ? sh->bulk_out[k] : sh->link_out[k];
136 if (q) mesh_chan_retain(q);
137 pthread_mutex_unlock(&sh->mu);
138 return q;
139}
140
141/* A frame to member k's link; a full link queue is waited on (2 s at most) while k stays linked, as the
142 * Rust's writer blocks the queue behind it. */
143static void forward(mesh_fleetd *f, mpr_relay *r, uint32_t k, const mesh_link_frame *x, bool bulk) {
144 if (!x->n) return; /* a doorbell */
145 int64_t t0 = mesh_mono_ms();
146 while (!mpr_send(r, k, x->b, x->n, bulk)) {
147 uint32_t linked = mesh_host_block_get_u32(&f->sh->host, MESH_S_LINKED);
148 if (!(linked >> k & 1) || quitting(f) || mesh_mono_ms() - t0 > 2000) return;
149 mesh_sleep_us(1000);
150 }
151}
152
153/* peer.rs writer: member k's queued frames, every control frame before each bulk frame. */
154static void *fwd_main(void *p) {
155 fwd_arg a = *(fwd_arg *)p;
156 free(p);
157 mesh_fleetd *f = a.f;
158 mesh_shared *sh = f->sh;
159 while (!quitting(f) && !atomic_load(&f->relay) && (atomic_load(&f->starting) || !atomic_load(&sh->ready))) mesh_sleep_us(10000);
160 mpr_relay *r = atomic_load(&f->relay);
161 if (!r) return NULL;
162 while (!quitting(f)) {
163 mesh_chan *q = queue_of(sh, a.k, false), *bq = queue_of(sh, a.k, true);
164 if (!q || !bq) {
165 if (q) mesh_chan_release(q);
166 if (bq) mesh_chan_release(bq);
167 mesh_sleep_us(20000);
168 continue;
169 }
170 mesh_link_frame x;
171 for (;;) {
172 while (mesh_chan_try_recv(q, &x) == MESH_CHAN_OK) {
173 forward(f, r, a.k, &x, false);
174 free(x.b);
175 }
176 if (mesh_chan_try_recv(bq, &x) == MESH_CHAN_OK) {
177 forward(f, r, a.k, &x, true);
178 free(x.b);
179 continue;
180 }
181 if (quitting(f)) break;
182 int got = mesh_chan_recv(q, &x, 100);
183 if (got == MESH_CHAN_OK) {
184 forward(f, r, a.k, &x, false);
185 free(x.b);
186 } else if (got == MESH_CHAN_CLOSED) break;
187 /* The link ended (its queues were dropped): take the next one's. */
188 pthread_mutex_lock(&sh->mu);
189 bool gone = sh->link_out[a.k] != q;
190 pthread_mutex_unlock(&sh->mu);
191 if (gone) break;
192 }
193 while (mesh_chan_try_recv(q, &x) == MESH_CHAN_OK) free(x.b);
194 while (mesh_chan_try_recv(bq, &x) == MESH_CHAN_OK) free(x.b);
195 mesh_chan_release(q);
196 mesh_chan_release(bq);
197 }
198 return NULL;
199}
200
201void mesh_fleetd_start(mesh_fleetd *f, const mpk_caps *caps) {
202 f->caps = *caps;
203 f->host = fd_host(f);
204 mrc_ops ops = fd_recon_ops(f);
205 f->rec = mrc_thread_start(&ops);
206 fd_perf_set(f);
207 atomic_store(&f->starting, true);
208 f->fleet_on = pthread_create(&f->fleet_t, NULL, fleet_main, f) == 0;
209 if (!f->fleet_on) atomic_store(&f->starting, false);
210 f->bridge_on = pthread_create(&f->bridge_t, NULL, bridge_main, f) == 0;
211 for (uint32_t k = 0; k < f->topo.n && k < MESH_MAX_SYSTEMS; k++) {
212 if (k == f->topo.me) continue;
213 fwd_arg *a = mesh_xmalloc(sizeof *a);
214 *a = (fwd_arg){f, k};
215 f->fwd_on[k] = pthread_create(&f->fwd_t[k], NULL, fwd_main, a) == 0;
216 if (!f->fwd_on[k]) free(a);
217 }
218}
219
220void mesh_fleetd_stop(mesh_fleetd *f) {
221 if (!f) return;
222 atomic_store(&f->stop, true);
223 if (f->fleet_on) pthread_join(f->fleet_t, NULL);
224 if (f->bridge_on) pthread_join(f->bridge_t, NULL);
225 for (uint32_t k = 0; k < MESH_MAX_SYSTEMS; k++)
226 if (f->fwd_on[k]) pthread_join(f->fwd_t[k], NULL);
227 f->fleet_on = f->bridge_on = false;
228 memset(f->fwd_on, 0, sizeof f->fwd_on);
229 fd_perf_set(NULL);
230 if (f->rec) mrc_thread_stop(f->rec);
231 f->rec = NULL;
232 mpr_relay *r = atomic_exchange(&f->relay, NULL);
233 if (r) mpr_stop(r);
234}
235
236void mesh_fleetd_free(mesh_fleetd *f) {
237 if (!f) return;
238 mesh_fleetd_stop(f);
239 free(f->topo.listen);
240 for (size_t i = 0; i < f->topo.ndial; i++) free(f->topo.dial_addr[i]);
241 free(f->state_path);
242 free(f->reason);
243 for (size_t i = 0; i < f->nmoves; i++) free(f->moves[i].name);
244 free(f->moves);
245 for (size_t i = 0; i < f->nfails; i++) {
246 free(f->fails[i].name);
247 free(f->fails[i].why);
248 }
249 free(f->fails);
250 pthread_mutex_destroy(&f->mu);
251 free(f);
252}