==> tests/link.c <==1/* peer.rs handshake_over_tcp; the framed reader's rules (every frame type, hostile lengths refused);
2 * a link's frames both ways, control before bulk, and its close. */
3#include "mesh/abi/peer_frames.h"
4#include "mesh/abi/peer_sizes.h"
5#include "mesh/abi/place_ops.h"
6#include "mesh/check.h"
7#include "mesh/peer_link.h"
8#include "mesh/mt.h"
9
10#include <netinet/in.h>
11#include <pthread.h>
12#include <stdlib.h>
13#include <sys/socket.h>
14#include <time.h>
15#include <unistd.h>
16
17static mpk_caps caps(const char *host) {
18 mpk_caps c = {.cc = 121, .sms = 48, .sms_total = 48, .blocks_per_sm = 24, .gpu = "NVIDIA GB10", .arch = "aarch64", .max_threads = 1024, .max_shared = 101376};
19 snprintf(c.host, sizeof c.host, "%s", host);
20 return c;
21}
22
23typedef struct {
24 int fd;
25 bool ok;
26 mpk_join join;
27} side;
28
29static void *accept_side(void *arg) {
30 side *s = arg;
31 int c = accept(s->fd, NULL, NULL);
32 char *err = NULL;
33 mpk_caps k = caps("dgx-00");
34 s->ok = mpn_handshake(c, 2, 3, &k, false, 0, 3000, &s->join, &err);
35 free(err);
36 close(c);
37 return NULL;
38}
39
40static void handshake(void) {
41 char *err = NULL;
42 int l = mpn_listen("127.0.0.1:0", &err);
43 CHECK(l >= 0);
44 struct sockaddr_in a;
45 socklen_t al = sizeof a;
46 getsockname(l, (struct sockaddr *)&a, &al);
47 char *addr = mt_fmt("127.0.0.1:%u", ntohs(a.sin_port));
48 side s = {.fd = l};
49 pthread_t t;
50 pthread_create(&t, NULL, accept_side, &s);
51 int c = mpn_connect(addr, 3000, &err);
52 CHECK(c >= 0);
53 mpk_caps k = caps("dialer");
54 mpk_join mine;
55 CHECK(mpn_handshake(c, 0, 3, &k, true, 2, 3000, &mine, &err));
56 pthread_join(t, NULL);
57 CHECK(s.ok && mine.system == 2 && s.join.system == 0 && !strcmp(s.join.caps.host, "dialer"));
58 close(c);
59 close(l);
60 /* Nothing listens: a bounded, clear failure. */
61 c = mpn_connect(addr, 500, &err);
62 CHECK(c < 0 && mt_starts(err, "connect 127.0.0.1:"));
63 free(err);
64 free(addr);
65}
66
67typedef struct {
68 const uint8_t *b;
69 size_t n, at;
70} memread;
71
72static bool mem_read(void *ctx, uint8_t *buf, size_t n, char **err) {
73 memread *m = ctx;
74 if (m->at + n > m->n) {
75 *err = mt_strdup("closed");
76 return false;
77 }
78 memcpy(buf, m->b + m->at, n);
79 m->at += n;
80 return true;
81}
82
83/* Reads one frame back from bytes; the frame must be exactly them. */
84static bool reads_back(const mpw_bytes *f) {
85 memread m = {f->b, f->n, 0};
86 mpw_bytes g;
87 char *err = NULL;
88 bool ok = mpn_read_frame(mem_read, &m, &g, &err) && g.n == f->n && !memcmp(g.b, f->b, f->n) && m.at == f->n;
89 if (!ok) fprintf(stderr, "frame type %u: %s\n", f->b[0], err ? err : "differs");
90 free(err);
91 mpw_bytes_free(&g);
92 return ok;
93}
94
95static bool refused(const uint8_t *b, size_t n, const char *needle) {
96 memread m = {b, n, 0};
97 mpw_bytes g;
98 char *err = NULL;
99 bool ok = !mpn_read_frame(mem_read, &m, &g, &err) && mt_contains(err, needle);
100 if (!ok) fprintf(stderr, "wanted %s, got %s\n", needle, err ? err : "a frame");
101 free(err);
102 return ok;
103}
104
105static void framing(void) {
106 mpk_beat_status st = {.reason = "ok"};
107 mfc_msg beat = {.kind = MFC_BEAT, .term = 1};
108 mpw_bytes f = mpk_beat_frame(&beat, &st);
109 CHECK(reads_back(&f));
110 /* A hostile reason length: refused before it is read. */
111 uint32_t big = MESH_BEAT_REASON_MAX + 1;
112 memcpy(f.b + 1 + MESH_BEAT_HEAD - 4, &big, 4);
113 CHECK(refused(f.b, f.n, "F_BEAT reason of 161 bytes"));
114 mpw_bytes_free(&f);
115 mfc_msg m[] = {{.kind = MFC_VOTE_REQ, .term = 2}, {.kind = MFC_VOTE, .term = 2}, {.kind = MFC_STATE, .term = 2}, {.kind = MFC_STATE_ACK}, {.kind = MFC_COMMIT}};
116 for (int i = 0; i < 5; i++) {
117 f = mpk_msg_frame(&m[i]);
118 CHECK(reads_back(&f));
119 mpw_bytes_free(&f);
120 }
121 mpk_hello h = {.group = "train", .order = {0, 1, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}};
122 f = mpk_hello_frame(1, &h);
123 CHECK(reads_back(&f));
124 mpw_bytes_free(&f);
125 mpw_place p = {.op = MESH_PL_SET, .id = 1, .name = "doom", .mode = mpm_pin(1)};
126 f = mpw_place_encode(&p);
127 CHECK(reads_back(&f));
128 mpw_bytes_free(&f);
129 mpw_frame fr = {.name = "doom", .frame = 1, .parts = 1, .data = (uint8_t *)"abcd", .ndata = 4};
130 f = mpw_frame_encode(&fr);
131 CHECK(reads_back(&f));
132 mpw_bytes_free(&f);
133 f = mpw_input_encode("doom", (const uint8_t *)"w", 1);
134 CHECK(reads_back(&f));
135 mpw_bytes_free(&f);
136 /* Length-prefixed bodies, and their limits. */
137 uint8_t perf[1 + 4 + 3] = {MESH_F_PERF, 3, 0, 0, 0, 'a', 'b', 'c'};
138 mpw_bytes pf = {perf, sizeof perf, sizeof perf};
139 CHECK(reads_back(&pf));
140 uint8_t ctl[1 + MESH_CTL_HEAD] = {MESH_F_CTL, 1, 0, 0, 0, 0xff, 0xff, 0, 0};
141 CHECK(refused(ctl, sizeof ctl, "F_CTL of 65535 bytes"));
142 uint8_t state[1 + MESH_STATE_HEAD] = {MESH_F_STATE};
143 uint32_t huge = MESH_F_STATE_MAX + 1;
144 memcpy(state + 1 + 8, &huge, 4);
145 CHECK(refused(state, sizeof state, "F_STATE of"));
146 uint8_t job[1 + 4] = {MESH_F_LEASE, 0, 0, 0, 0x7f};
147 CHECK(refused(job, sizeof job, "job frame type 26"));
148 uint8_t bad[] = {99};
149 CHECK(refused(bad, 1, "bad frame type 99"));
150 uint8_t cut[] = {MESH_F_VOTE, 1, 2};
151 CHECK(refused(cut, sizeof cut, "closed"));
152}
153
154/* ---- a link both ways over a socket pair */
155
156typedef struct {
157 pthread_mutex_t mu;
158 uint8_t types[64];
159 int n;
160 bool closed;
161 char why[64];
162} got;
163
164static char *on_frame(void *ctx, uint32_t from, uint64_t lid, mpw_bytes f) {
165 (void)from, (void)lid;
166 got *g = ctx;
167 pthread_mutex_lock(&g->mu);
168 if (g->n < 64) g->types[g->n++] = f.b[0];
169 pthread_mutex_unlock(&g->mu);
170 mpw_bytes_free(&f);
171 return NULL;
172}
173
174static void on_closed(void *ctx, uint32_t from, uint64_t lid, const char *why) {
175 (void)from, (void)lid;
176 got *g = ctx;
177 pthread_mutex_lock(&g->mu);
178 g->closed = true;
179 snprintf(g->why, sizeof g->why, "%s", why);
180 pthread_mutex_unlock(&g->mu);
181}
182
183static void wait_for(got *g, int n) {
184 for (int i = 0; i < 400; i++) {
185 pthread_mutex_lock(&g->mu);
186 bool done = g->n >= n;
187 pthread_mutex_unlock(&g->mu);
188 if (done) return;
189 struct timespec ts = {0, 5000000};
190 nanosleep(&ts, NULL);
191 }
192}
193
194static void links(void) {
195 int sv[2];
196 CHECK(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
197 got ga = {.mu = PTHREAD_MUTEX_INITIALIZER}, gb = {.mu = PTHREAD_MUTEX_INITIALIZER};
198 mpn_sink sa = {&ga, on_frame, on_closed, NULL}, sb = {&gb, on_frame, on_closed, NULL};
199 atomic_bool dialer;
200 atomic_init(&dialer, true);
201 mpn_link *a = mpn_link_start(sv[0], 1, 1, 3000, &sa, &dialer), *b = mpn_link_start(sv[1], 0, 1, 3000, &sb, NULL);
202 mfc_msg v = {.kind = MFC_VOTE, .term = 5};
203 mpw_bytes f = mpk_msg_frame(&v);
204 uint8_t perf[] = {MESH_F_PERF, 1, 0, 0, 0, 'x'};
205 CHECK(mpn_send_bulk(a, perf, sizeof perf));
206 CHECK(mpn_send(a, f.b, f.n) && mpn_send(a, f.b, f.n));
207 wait_for(&gb, 3);
208 CHECK(gb.n == 3);
209 CHECK(mpn_write_now(b, f.b, f.n));
210 wait_for(&ga, 1);
211 CHECK(ga.n == 1 && ga.types[0] == MESH_F_VOTE);
212 /* Closing one side: the other's reader sees it, and the dialer's flag goes false. */
213 mpn_link_close(b);
214 for (int i = 0; i < 400 && !ga.closed; i++) {
215 struct timespec ts = {0, 5000000};
216 nanosleep(&ts, NULL);
217 }
218 CHECK(ga.closed && !strcmp(ga.why, "closed") && !atomic_load(&dialer) && !mpn_link_alive(a));
219 CHECK(!mpn_send(a, f.b, f.n));
220 mpn_link_free(a);
221 mpn_link_free(b);
222 mpw_bytes_free(&f);
223}
224
225int main(void) {
226 handshake();
227 framing();
228 links();
229 return mesh_check_done();
230}