1/* The fleet thread's state, shared by its source files (relay.c: events, links, changes, the leader;
2 * group.c: the lockstep data plane; sched.c: the scheduler glue and migrations; publish.c: the status
3 * block and the texts; board.c: the perf board). Not installed. */
4#ifndef RELAY_H
5#define RELAY_H
6
7#include "mesh/fleet_core.h"
8#include "mesh/fleet_store.h"
9#include "mesh/peer_link.h"
10#include "mesh/peer_relay.h"
11#include "mesh/peer_view.h"
12#include "mesh/sched_place.h"
13#include "mesh/sched_tracker.h"
14#include "mesh/mt.h"
15
16#include <pthread.h>
17#include <stdatomic.h>
18
19#define MAXS MESH_MAX_SYSTEMS
20
21uint64_t mpr_mono_ms(void);
22uint64_t mpr_mono_us(void);
23
24/* One link and what the data plane keeps per link. */
25typedef struct {
26    struct mpr_relay *r;
27    mpn_link *l;
28    uint32_t from;
29    uint64_t lid;
30    atomic_uint_fast64_t last_rx; /* this meshd's uptime ms when its last frame arrived */
31    pthread_mutex_t sent_mu;      /* our rounds in flight to it: (counter, sent µs) */
32    uint64_t sent_c[64], sent_t[64];
33    size_t sent_n;
34    atomic_uint_fast64_t last_round; /* its last round counter seen here */
35    atomic_int refs;                 /* the rounds sender writing to it (freeing waits for 0) */
36} linkrec;
37
38typedef enum { EV_LINK, EV_CLOSED, EV_FRAME, EV_CHANGE, EV_WAKE, EV_FETCH_IMAGE, EV_FETCH_VIEW, EV_CTL_REPLY } ev_kind;
39
40/* A blocked caller's answer slot (fetches). */
41typedef struct {
42    pthread_mutex_t mu;
43    pthread_cond_t cv;
44    bool done, ok;
45    char *err;
46    mpr_payload payload;
47    uint32_t view_abi;
48    mpw_bytes view;
49} waiter;
50
51typedef struct {
52    ev_kind kind;
53    uint32_t from, id;
54    uint64_t lid, hash;
55    int fd;
56    mpk_join join;
57    atomic_bool *alive;
58    char *text, *name, *dir;
59    mpw_bytes frame;
60    mpr_reply reply;
61    waiter *w;
62    bool ok;
63} event;
64
65/* Who waits for a change's answer: a local caller (or nobody), or a member's F_CTL id. */
66typedef struct {
67    bool remote;
68    uint32_t k, id;
69    mpr_reply local;
70} replyto;
71
72/* The change the leader is committing or waiting on. */
73typedef struct {
74    char *what;
75    replyto reply;
76    uint64_t version;
77    bool committed;
78    uint32_t up;       /* the members up when it committed ... */
79    uint64_t deadline; /* ... and when we stop waiting for them (ms) */
80    mfd_doc before, after;
81} pending_t;
82
83/* The leader's migration in flight. */
84typedef struct {
85    mst_move mv;
86    uint64_t version, t0_ms;
87} mig_t;
88
89typedef struct {
90    bool used;
91    uint32_t id;
92    waiter *w;
93    uint32_t from;
94    uint64_t deadline;
95} image_wait;
96
97typedef struct {
98    bool used;
99    uint32_t id;
100    waiter *w;
101    mpv_wait v;
102} view_wait;
103
104typedef struct {
105    bool used;
106    uint32_t id;
107    mpr_reply reply;
108    uint32_t leader;
109    uint64_t deadline;
110} fwd_t;
111
112/* A failure the scheduler avoids a while: (program, member) -> (hash, why, unsupported, when). */
113typedef struct {
114    char *name;
115    uint32_t k;
116    uint64_t hash;
117    char *why;
118    bool unsupported;
119    uint64_t at;
120} fail_mem_t;
121
122typedef struct {
123    char *name;
124    uint32_t k;
125    msm_perf perf;
126    uint64_t at;
127} perf_mem_t;
128
129/* The perf board: every member's last F_PERF (and when it came), and the link RTTs. */
130typedef struct {
131    pthread_mutex_t mu;
132    bool has[MAXS];
133    msr_report rep[MAXS];
134    uint64_t at_us[MAXS];
135    mof32 rtt_us[MAXS];
136} board_t;
137
138/* The lockstep rounds' tag: (epoch, cycle) and each system's rank (0xff: not in it). */
139typedef struct {
140    uint32_t epoch, cycle;
141    uint8_t rank_of[MAXS], me;
142} tag_t;
143
144struct mpr_relay {
145    mpr_host h;
146    uint32_t me, n;
147    mcc_tunables tun;
148    mpk_caps my_caps;
149    mfc_core *core;
150    mfs_store store;
151    bool has_store;
152    uint64_t t0_ms;
153    /* events */
154    pthread_mutex_t ev_mu;
155    pthread_cond_t ev_cv;
156    event *ev;
157    size_t ev_n, ev_cap;
158    pthread_t thread, sender;
159    atomic_bool quit, fatal;
160    /* links */
161    pthread_mutex_t links_mu;
162    linkrec *links[MAXS];
163    mpk_caps caps[MAXS];
164    bool has_caps[MAXS];
165    uint64_t next_lid;
166    mpn_acceptor *acceptor;
167    mpn_dialer *dialers[MAXS];
168    size_t ndialers;
169    /* members' beats */
170    mpk_beat_status st[MAXS];
171    uint64_t last_beat_ms;
172    /* changes */
173    char **queue_line;
174    replyto *queue_to;
175    size_t nqueue;
176    bool has_pending;
177    pending_t pending;
178    pending_t *waiting;
179    size_t nwaiting;
180    fwd_t fwd[64];
181    image_wait images[32];
182    view_wait views[32];
183    uint32_t next_id;
184    ms_map paused_since; /* group -> ms */
185    bool has_first_doc_at;
186    uint64_t first_doc_at_ms;
187    uint32_t seen_status[MAXS];
188    char *seen_reason[MAXS];
189    /* the driver's last view */
190    bool has_view;
191    uint32_t view_up;
192    bool view_has_leader;
193    uint32_t view_leader;
194    char *view_places;
195    /* the lockstep group */
196    pthread_mutex_t group_mu; /* the reconciler's wish */
197    bool want_set, want_mirror, want_member, want_changed;
198    mpk_hello want;
199    bool has_hello;
200    mpk_hello hello;
201    bool has_theirs[MAXS];
202    mpk_hello theirs[MAXS];
203    atomic_uint hello_state;
204    char *hello_why;
205    uint64_t hello_sent_ms;
206    bool has_mismatch[MAXS];
207    mpk_hello mismatch[MAXS];
208    atomic_bool mirroring;
209    pthread_mutex_t send_mu; /* held by the sender for a whole pass (a retag waits for it) */
210    pthread_mutex_t tag_mu;
211    tag_t tag;
212    atomic_uint_fast64_t last_sent;
213    pthread_mutex_t own_mu;
214    uint64_t own_c[64], own_t[64];
215    size_t own_n;
216    /* publishing */
217    uint64_t last_publish_ms, last_sched_text_ms;
218    /* the scheduler */
219    msw_weights w;
220    mst_tracker *tracker;
221    bool has_mig;
222    mig_t mig;
223    bool down[MAXS];
224    uint64_t down_since[MAXS];
225    perf_mem_t *perf_mem;
226    size_t nperf_mem;
227    uint64_t last_sched, last_rebalance;
228    uint32_t moves_done, last_gap_us;
229    mt_strs sched_log;
230    char *group_prop;
231    msp_explains explains;
232    fail_mem_t *fail_mem;
233    size_t nfail_mem;
234    bool has_load[MAXS];
235    mpv_load load_avg[MAXS];
236    board_t board;
237    /* What other threads may read without the core: set at the end of every tick. */
238    atomic_uint_fast64_t pub_version;
239    atomic_uint pub_leader, pub_role, pub_alive;
240};
241
242/* relay.c */
243uint64_t relay_now(const mpr_relay *r);
244void relay_send(mpr_relay *r, uint32_t k, const mpw_bytes *f);
245void relay_broadcast(mpr_relay *r, const mpw_bytes *f);
246void relay_log(mpr_relay *r, bool error, char *line); /* takes line */
247bool relay_propose(mpr_relay *r, mfd_doc doc, char *what, replyto to, uint64_t *version);
248uint32_t relay_linked_mask(mpr_relay *r);
249bool relay_linked(mpr_relay *r, uint32_t k);
250uint64_t relay_last_rx(mpr_relay *r, uint32_t k, bool *linked);
251void relay_beat_all(mpr_relay *r);
252void relay_my_status(mpr_relay *r, mpk_beat_status *out);
253void relay_reply(mpr_relay *r, replyto to, bool ok, const char *text);
254void waiter_done(waiter *w, bool ok, char *err);
255
256/* group.c */
257void group_want(mpr_relay *r);
258void group_hello(mpr_relay *r, uint32_t from, const mpk_hello *spec);
259void group_hello_send(mpr_relay *r);
260void group_pause(mpr_relay *r, const char *why);
261void group_set_state(mpr_relay *r, uint32_t s, const char *why);
262void group_round(mpr_relay *r, linkrec *lr, const uint8_t *body);
263void *group_sender(void *arg);
264
265/* sched.c (sched_publish: publish.c) */
266msm_member *sched_members(mpr_relay *r, uint64_t now, size_t *n);
267void sched_members_free(msm_member *ms, size_t n);
268msm_perfs sched_perf_map(mpr_relay *r, uint64_t now);
269mfd_doc sched_pass(mpr_relay *r, const mfd_doc *doc, uint64_t now, mt_strs *changes);
270mfd_doc sched_edit(mpr_relay *r, const mfd_doc *base, mfd_doc doc, uint64_t now, mt_strs *changes);
271void sched_schedule(mpr_relay *r, uint64_t now);
272void sched_serve_view(mpr_relay *r, uint32_t k, uint32_t id, const char *name, uint64_t hash);
273void sched_fetch_view(mpr_relay *r, const char *name, uint64_t hash, uint32_t from, waiter *w);
274void sched_view_chunk(mpr_relay *r, uint32_t id, const uint8_t *body, size_t n);
275void sched_view_timeouts(mpr_relay *r, bool has_lost, uint32_t lost);
276void sched_publish(mpr_relay *r, uint64_t now, uint32_t alive, uint32_t suspect, const mpr_recon *rs);
277void sched_log_decision(mpr_relay *r, const char *what);
278
279/* publish.c */
280void publish_all(mpr_relay *r, uint64_t now);
281void publish_member(mpr_relay *r, uint32_t k, uint32_t state, const mpk_caps *caps);
282char *publish_group_line(const mfd_group *g);
283
284/* board.c */
285void board_init(board_t *b);
286void board_free(board_t *b);
287/* An F_PERF from member from (its text). */
288void board_report(mpr_relay *r, uint32_t from, const char *text);
289
290#endif