1/* @ajax/mesh-peer-relay: the fleet thread (peer.rs run / Relay, peer_sched.rs). N machines run as one
2 * fleet with partial membership:
3 * - Links: a full mesh of TCP links (mesh-peer-link). Every member beats every other each ping_ms; a
4 * member is up while its link is and it was heard within dead_ms, suspect when silent for a third of
5 * that. Nothing ends when a member drops: only the lockstep group that contains it pauses.
6 * - The leader and the desired state (mesh-fleet-core): a change asked for on any member goes to the
7 * leader (F_CTL, answered with F_CTL_REPLY once the reachable members reconciled it); without a
8 * leader it is refused. The leader commits it, waits for the members to reconcile it, and reverts a
9 * change that failed everywhere. Each committed document goes to the daemon's reconciler; images and
10 * migrated views are fetched from members for it.
11 * - The leader's policies: lockstep groups (pause, resume, re-form) and the scheduler (placement,
12 * rebalancing, migrations, group proposals) from a model of every member (F_JOIN, F_LOADREP, F_PERF).
13 * - Lockstep groups: the members HELLO each epoch and cycle; rounds are mirrored point to point.
14 * - Published: the status block's fleet regions, the `fleet`, `groups` and `sched` texts and the fleet
15 * view's records.
16 * Everything daemon-specific (the status block, the mailbox, the driver, the catalog, the reconciler,
17 * jobs and blobs) comes through mpr_host. */
18#ifndef MESH_PEER_RELAY_H
19#define MESH_PEER_RELAY_H
2021#include "mesh/cluster_conf.h"
22#include "mesh/fleet_edit.h"
23#include "mesh/fleet_persist.h"
24#include "mesh/fleet_plan.h"
25#include "mesh/peer_codec.h"
26#include "mesh/place_wire.h"
27#include "mesh/sched_report.h"
28#include "mesh/sched_tracker.h"
2930#include <stdbool.h>
31#include <stdint.h>
3233/* S_PEER_STATE */
34#define MPR_PEER_WAITING 1u
35#define MPR_PEER_UP 2u
3637/* An answer to a change: ok, and the text (the change's outcome or the reason). */
38typedef struct {
39 void (*fn)(void *ctx, bool ok, const char *text);
40 void *ctx;
41} mpr_reply;
4243/* The reconciler's status as the beats say it. */
44typedef struct {
45 uint64_t applied;
46 uint32_t status;
47 char *reason;
48 mst_outcome *moves;
49 size_t nmoves;
50} mpr_recon;
5152void mpr_recon_free(mpr_recon *r);
5354/* A program image as members exchange it (F_IMAGE): its catalog dir, manifest text and PTX. */
55typedef struct {
56 char *dir, *manifest;
57 mpw_bytes ptx;
58} mpr_payload;
5960void mpr_payload_free(mpr_payload *p);
6162typedef struct {
63 void *ctx;
64 /* The status block. */
65 void (*host_u32)(void *ctx, size_t off, uint32_t v);
66 void (*host_u64)(void *ctx, size_t off, uint64_t v);
67 void (*host_str)(void *ctx, size_t off, size_t cap, const char *s);
68 void (*host_bytes)(void *ctx, size_t off, const uint8_t *b, size_t n);
69 void (*host_read)(void *ctx, size_t off, uint8_t *out, size_t n);
70 uint64_t (*uptime_ms)(void *ctx);
71 /* The mailbox: the lockstep exchange regions and the program table. */
72 uint64_t (*mb_get64)(void *ctx, size_t off);
73 uint32_t (*mb_get32)(void *ctx, size_t off);
74 void (*mb_read)(void *ctx, size_t off, uint8_t *out, size_t n);
75 void (*mb_write)(void *ctx, size_t off, const uint8_t *b, size_t n);
76 void (*mb_publish64)(void *ctx, size_t off, uint64_t v);
77 void (*lock_feed)(void *ctx, bool on); /* feed the lockstep program */
78 /* A lockstep round received from member k reached counter n: (rtt µs or < 0, wait µs). */
79 void (*round_stats)(void *ctx, uint32_t k, uint64_t n, double rtt_us, double wait_us);
80 /* The reconciler: a committed document to apply; its status. */
81 void (*committed)(void *ctx, const mfd_doc *doc);
82 void (*recon)(void *ctx, mpr_recon *out);
83 /* The driver and the catalog. */
84 bool (*catalog)(void *ctx, const char *dir, mfe_image *out, char **err);
85 mfl_actual *(*inventory)(void *ctx, size_t *n);
86 /* This member's catalog image of dir if it is name at hash (F_IMAGE_REQ). */
87 bool (*image)(void *ctx, const char *dir, const char *name, uint64_t hash, mpr_payload *out, char **err);
88 /* Stop a program and read its view (a migration's source; called on its own thread). */
89 bool (*stop_and_view)(void *ctx, const char *name, uint32_t *view_abi, mpw_bytes *view, char **err);
90 /* The fleet's view for the driver's placement: up members, leader, placement requests. */
91 void (*driver_view)(void *ctx, uint32_t up, bool has_leader, uint32_t leader, const mfd_doc *places);
92 /* Frames for others: placement (F_PLACE, F_FRAME, F_INPUT: the whole frame, taken), jobs, blobs. */
93 void (*place_frame)(void *ctx, uint32_t from, mpw_bytes frame);
94 void (*jobs_frame)(void *ctx, uint32_t from, uint8_t type, const uint8_t *body, size_t n);
95 void (*blobs_frame)(void *ctx, uint32_t from, uint8_t type, const uint8_t *body, size_t n);
96 /* A `job ...` request: local (reply) or from member from (answer with mpr_ctl_reply(id)). */
97 void (*jobs_ctl)(void *ctx, const char *line, bool remote, uint32_t from, uint32_t id, mpr_reply reply);
98 void (*jobs_view)(void *ctx, uint64_t term, bool has_leader, uint32_t leader, uint32_t alive, uint32_t linked);
99 /* The scheduler's model for the jobs dispatcher (jobs::select): the members, the programs' needs
100 * (the committed document), the measured performance and the weights. */
101 void (*jobs_sched)(void *ctx, const msm_member *members, size_t n, const mfd_doc *doc, const msm_perfs *perf, const msw_weights *w);
102 /* The texts: `fleet`, `groups`, `sched` (all, the per-program explains, the dry run) and the fleet
103 * view's records (one a line). */
104 void (*texts)(void *ctx, const char *fleet, const char *groups);
105 void (*sched_texts)(void *ctx, const char *all, const msp_explains *explains, const char *dry, const char *records);
106 void (*peer_state)(void *ctx, uint32_t state);
107 void (*net_count)(void *ctx, uint32_t k, bool tx, uint8_t type, size_t bytes);
108 void (*log)(void *ctx, bool error, const char *line);
109 /* The state file could not be written: this member stops (fail-stop). */
110 void (*fail_stop)(void *ctx, const char *why);
111} mpr_host;
112113typedef struct mpr_relay mpr_relay;
114115/* Starts the fleet thread and its links for this topology (copied), with this member's capabilities and
116 * its state file (state_path; NULL: in memory only, for tests). On error *err. */
117mpr_relay *mpr_start(const mcc_topology *topo, const mpk_caps *caps, const char *state_path, const mpr_host *host, char **err);
118/* Stops the thread and every link. */
119void mpr_stop(mpr_relay *r);
120121/* A change asked for here (a control-port line): the leader queues it, another member forwards it,
122 * without a leader it is refused. */
123void mpr_change(mpr_relay *r, const char *line, mpr_reply reply);
124/* The reconciler changed something the beats carry (status, the group's wish): beat now. */
125void mpr_wake(mpr_relay *r);
126/* The lockstep group as the reconciler wants it: ready for spec (NULL: quiesced; mirror: keep the
127 * rounds going; member: in the group's epoch). */
128void mpr_group_want(mpr_relay *r, const mpk_hello *spec, bool mirror, bool member);
129/* The data plane's state (GL_*) once it saw the wish. */
130uint32_t mpr_group_state(mpr_relay *r);
131/* For the reconciler: an image from a member (the leader if linked), and a migrated view from its source;
132 * block up to the given time. */
133bool mpr_fetch_image(mpr_relay *r, const char *name, uint64_t hash, const char *dir, uint32_t wait_ms, mpr_payload *out, char **err);
134bool mpr_fetch_view(mpr_relay *r, const char *name, uint64_t hash, uint32_t from, uint32_t wait_ms, uint32_t *view_abi, mpw_bytes *view, char **err);
135/* An answer to a forwarded `job ...` request (F_CTL_REPLY to member to). */
136void mpr_ctl_reply(mpr_relay *r, uint32_t to, uint32_t id, bool ok, const char *text);
137/* Sends a frame to member k (bulk: blobs, behind every control frame). */
138bool mpr_send(mpr_relay *r, uint32_t k, const uint8_t *f, size_t n, bool bulk);
139void mpr_broadcast(mpr_relay *r, const uint8_t *f, size_t n);
140141/* The perf board: every member's F_PERF and the link RTTs (the scheduler's inputs). The daemon puts its
142 * own report there and reads the echoes for its next F_PERF. */
143void mpr_board_own(mpr_relay *r, const msr_report *rep);
144/* The echoes for this member's next F_PERF: (member, its t_us we received last, µs held since). */
145size_t mpr_board_echoes(mpr_relay *r, uint32_t *k, uint64_t *t, uint64_t *held);
146147/* For tests and the control port: the committed document's version, the leader, the role (MFC_*). */
148uint64_t mpr_version(mpr_relay *r);
149bool mpr_leader(mpr_relay *r, uint32_t *leader);
150uint32_t mpr_role(mpr_relay *r);
151uint32_t mpr_alive(mpr_relay *r);
152153#endif