==> src/peer_link.c <==1/* peer.rs: connect, handshake, acceptor, dialer, reader, writer, Link. */
2#include "mesh/peer_link.h"
3#include "mesh/abi/exchange.h"
4#include "mesh/abi/host_mload.h"
5#include "mesh/abi/peer_frames.h"
6#include "mesh/abi/peer_sizes.h"
7#include "mesh/mt.h"
8
9#include <errno.h>
10#include <fcntl.h>
11#include <netdb.h>
12#include <netinet/in.h>
13#include <netinet/tcp.h>
14#include <poll.h>
15#include <pthread.h>
16#include <stdio.h>
17#include <stdlib.h>
18#include <string.h>
19#include <sys/socket.h>
20#include <time.h>
21#include <unistd.h>
22
23/* "host:port" into its parts (the last ':' splits). */
24static bool split_addr(const char *addr, char **host, char **port) {
25 const char *c = strrchr(addr, ':');
26 if (!c) return false;
27 *host = mt_strndup(addr, (size_t)(c - addr));
28 *port = mt_strdup(c + 1);
29 return true;
30}
31
32int mpn_connect(const char *addr, uint32_t timeout_ms, char **err) {
33 char *host, *port;
34 if (!split_addr(addr, &host, &port)) {
35 *err = mt_fmt("connect %s: no port", addr);
36 return -1;
37 }
38 struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_STREAM}, *res = NULL;
39 int g = getaddrinfo(host, port, &hints, &res);
40 free(host), free(port);
41 if (g != 0 || !res) {
42 *err = mt_fmt("connect %s: address resolves to nothing", addr);
43 return -1;
44 }
45 int fd = socket(res->ai_family, SOCK_STREAM, 0);
46 if (fd < 0) {
47 *err = mt_fmt("connect %s: %s", addr, strerror(errno));
48 freeaddrinfo(res);
49 return -1;
50 }
51 int fl = fcntl(fd, F_GETFL);
52 fcntl(fd, F_SETFL, fl | O_NONBLOCK);
53 int r = connect(fd, res->ai_addr, res->ai_addrlen);
54 freeaddrinfo(res);
55 if (r != 0 && errno != EINPROGRESS) goto fail;
56 if (r != 0) {
57 struct pollfd p = {.fd = fd, .events = POLLOUT};
58 int k = poll(&p, 1, (int)timeout_ms);
59 if (k == 0) {
60 errno = ETIMEDOUT;
61 goto fail;
62 }
63 int soerr = 0;
64 socklen_t sl = sizeof soerr;
65 getsockopt(fd, SOL_SOCKET, SO_ERROR, &soerr, &sl);
66 if (k < 0 || soerr) {
67 errno = soerr ? soerr : errno;
68 goto fail;
69 }
70 }
71 fcntl(fd, F_SETFL, fl);
72 return fd;
73fail:
74 *err = mt_fmt("connect %s: %s", addr, strerror(errno));
75 close(fd);
76 return -1;
77}
78
79int mpn_listen(const char *addr, char **err) {
80 char *host, *port;
81 if (!split_addr(addr, &host, &port)) {
82 *err = mt_fmt("cannot bind %s: no port", addr);
83 return -1;
84 }
85 struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_STREAM, .ai_flags = AI_PASSIVE}, *res = NULL;
86 int g = getaddrinfo(host, port, &hints, &res);
87 free(host), free(port);
88 if (g != 0 || !res) {
89 *err = mt_fmt("cannot bind %s: address resolves to nothing", addr);
90 return -1;
91 }
92 int fd = socket(res->ai_family, SOCK_STREAM, 0), one = 1;
93 setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
94 if (fd < 0 || bind(fd, res->ai_addr, res->ai_addrlen) != 0 || listen(fd, 16) != 0) {
95 *err = mt_fmt("cannot bind %s: %s", addr, strerror(errno));
96 if (fd >= 0) close(fd);
97 freeaddrinfo(res);
98 return -1;
99 }
100 freeaddrinfo(res);
101 return fd;
102}
103
104static bool write_all(int fd, const uint8_t *b, size_t n) {
105 while (n) {
106 ssize_t w = send(fd, b, n, MSG_NOSIGNAL);
107 if (w < 0 && errno == EINTR) continue;
108 if (w <= 0) return false;
109 b += w, n -= (size_t)w;
110 }
111 return true;
112}
113
114typedef struct {
115 int fd;
116 uint32_t dead_ms;
117} sockread;
118
119static bool sock_read(void *ctx, uint8_t *buf, size_t n, char **err) {
120 sockread *s = ctx;
121 while (n) {
122 ssize_t r = recv(s->fd, buf, n, 0);
123 if (r > 0) {
124 buf += r, n -= (size_t)r;
125 continue;
126 }
127 if (r < 0 && errno == EINTR) continue;
128 if (r == 0) *err = mt_strdup("closed");
129 else if (errno == EAGAIN || errno == EWOULDBLOCK || errno == ETIMEDOUT) *err = mt_fmt("silent for %u ms", s->dead_ms);
130 else *err = mt_strdup(strerror(errno));
131 return false;
132 }
133 return true;
134}
135
136static void set_timeouts(int fd, uint32_t ms) {
137 struct timeval tv = {.tv_sec = ms / 1000, .tv_usec = (ms % 1000) * 1000};
138 setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv);
139 setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof tv);
140 int one = 1;
141 setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
142}
143
144bool mpn_handshake(int fd, uint32_t me, uint32_t n, const mpk_caps *caps, bool has_expect, uint32_t expect, uint32_t dead_ms, mpk_join *out, char **err) {
145 set_timeouts(fd, dead_ms);
146 mpk_join mine = {me, MESH_PEER_VERSION, n, *caps};
147 mpw_bytes f = mpk_join_encode(&mine);
148 bool ok = write_all(fd, f.b, f.n);
149 mpw_bytes_free(&f);
150 if (!ok) {
151 *err = mt_fmt("JOIN: %s", strerror(errno));
152 return false;
153 }
154 sockread s = {fd, dead_ms};
155 uint8_t ty;
156 char *e = NULL;
157 if (!sock_read(&s, &ty, 1, &e)) {
158 *err = mt_fmt("no JOIN: %s", e);
159 free(e);
160 return false;
161 }
162 if (ty != MESH_F_JOIN) {
163 *err = mt_fmt("first frame type %u is not F_JOIN (a peer-link version below 4?)", ty);
164 return false;
165 }
166 uint8_t b[MESH_JOIN_BYTES];
167 if (!sock_read(&s, b, sizeof b, &e)) {
168 *err = mt_fmt("short JOIN: %s", e);
169 free(e);
170 return false;
171 }
172 mpk_join j = mpk_join_decode(b);
173 if (!mpk_join_check(&j, me, n, has_expect, expect, err)) return false;
174 *out = j;
175 return true;
176}
177
178/* ---- framing */
179
180static bool rd(mpn_read_fn read, void *ctx, mpw_bytes *f, size_t n, char **err) {
181 if (n == 0) return true;
182 uint8_t *b = mt_alloc(n);
183 bool ok = read(ctx, b, n, err);
184 if (ok) mpw_put(f, b, n);
185 free(b);
186 return ok;
187}
188
189bool mpn_read_frame(mpn_read_fn read, void *ctx, mpw_bytes *f, char **err) {
190 *f = (mpw_bytes){0};
191 uint8_t ty;
192 if (!read(ctx, &ty, 1, err)) return false;
193 mpw_put(f, &ty, 1);
194 size_t at = 1;
195 uint32_t n;
196#define HEAD(len) \
197 do { \
198 if (!rd(read, ctx, f, (len), err)) goto bad; \
199 } while (0)
200#define BODY(len) HEAD(len)
201 switch (ty) {
202 case MESH_F_HELLO: HEAD(MESH_HELLO_BYTES); break;
203 case MESH_F_ROUND: HEAD(MESH_ROUND_HEAD + MESH_EX_REGION_BYTES); break;
204 case MESH_F_BEAT:
205 HEAD(MESH_BEAT_HEAD);
206 n = mpw_u32_at(f->b, at + MESH_BEAT_HEAD - 4);
207 if (n > MESH_BEAT_REASON_MAX) {
208 *err = mt_fmt("F_BEAT reason of %u bytes", n);
209 goto bad;
210 }
211 BODY(n);
212 HEAD(4);
213 n = mpw_u32_at(f->b, f->n - 4);
214 if (n > MESH_BEAT_MOVES_MAX) {
215 *err = mt_fmt("F_BEAT with %u move outcomes", n);
216 goto bad;
217 }
218 BODY((size_t)n * MESH_BEAT_MOVE_BYTES);
219 break;
220 case MESH_F_VOTE_REQ: HEAD(MESH_VOTE_REQ_BYTES); break;
221 case MESH_F_VOTE: HEAD(MESH_VOTE_BYTES); break;
222 case MESH_F_STATE:
223 HEAD(MESH_STATE_HEAD);
224 n = mpw_u32_at(f->b, at + 8);
225 if (n > MESH_F_STATE_MAX) {
226 *err = mt_fmt("F_STATE of %u bytes > F_STATE_MAX", n);
227 goto bad;
228 }
229 BODY(n);
230 break;
231 case MESH_F_STATE_ACK:
232 case MESH_F_COMMIT: HEAD(MESH_ACK_BYTES); break;
233 case MESH_F_CTL:
234 HEAD(MESH_CTL_HEAD);
235 n = mpw_u32_at(f->b, at + 4);
236 if (n > MESH_F_CTL_MAX) {
237 *err = mt_fmt("F_CTL of %u bytes", n);
238 goto bad;
239 }
240 BODY(n);
241 break;
242 case MESH_F_CTL_REPLY:
243 HEAD(MESH_CTL_REPLY_HEAD);
244 n = mpw_u32_at(f->b, at + 8);
245 if (n > MESH_F_CTL_REPLY_MAX) {
246 *err = mt_fmt("F_CTL_REPLY of %u bytes", n);
247 goto bad;
248 }
249 BODY(n);
250 break;
251 case MESH_F_IMAGE_REQ:
252 HEAD(MESH_IMAGE_REQ_HEAD);
253 n = mpw_u32_at(f->b, at + MESH_IMAGE_REQ_HEAD - 4);
254 if (n > 64) {
255 *err = mt_fmt("F_IMAGE_REQ dir of %u bytes", n);
256 goto bad;
257 }
258 BODY(n);
259 break;
260 case MESH_F_IMAGE: {
261 HEAD(MESH_IMAGE_HEAD);
262 size_t total = (size_t)mpw_u32_at(f->b, at + 16) + mpw_u32_at(f->b, at + 20) + mpw_u32_at(f->b, at + 24);
263 if (total > MESH_F_LOAD_MAX) {
264 *err = mt_fmt("F_IMAGE of %zu bytes > F_LOAD_MAX", total);
265 goto bad;
266 }
267 BODY(total);
268 break;
269 }
270 case MESH_F_PLACE: {
271 HEAD(MPW_PLACE_HEAD);
272 size_t rest;
273 if (!mpw_place_rest_len(f->b + at, &rest, err)) goto bad;
274 BODY(rest);
275 break;
276 }
277 case MESH_F_FRAME: {
278 HEAD(MPW_FRAME_HEAD);
279 size_t rest;
280 if (!mpw_frame_rest_len(f->b + at, &rest, err)) goto bad;
281 BODY(rest);
282 break;
283 }
284 case MESH_F_INPUT: {
285 HEAD(MPW_INPUT_HEAD);
286 size_t rest;
287 if (!mpw_input_len(f->b + at, &rest, err)) goto bad;
288 BODY(rest);
289 break;
290 }
291 case MESH_F_PERF:
292 HEAD(4);
293 n = mpw_u32_at(f->b, at);
294 if (n > MESH_F_PERF_MAX) {
295 *err = mt_fmt("F_PERF of %u bytes", n);
296 goto bad;
297 }
298 BODY(n);
299 break;
300 case MESH_F_VIEW_REQ: HEAD(MESH_VIEW_REQ_BYTES); break;
301 case MESH_F_VIEW:
302 HEAD(MESH_VIEW_HEAD);
303 n = mpw_u32_at(f->b, at + 20);
304 if (n > MESH_F_VIEW_CHUNK) {
305 *err = mt_fmt("F_VIEW chunk of %u bytes", n);
306 goto bad;
307 }
308 BODY(n);
309 break;
310 case MESH_F_LOADREP: HEAD(MESH_LOADREP_BYTES); break;
311 default:
312 if (ty >= MESH_F_JOB_AVAIL && ty <= MESH_F_RESULT) {
313 HEAD(MESH_JOB_HEAD);
314 n = mpw_u32_at(f->b, at);
315 if (n > MESH_F_JOB_MAX) {
316 *err = mt_fmt("job frame type %u of %u bytes > F_JOB_MAX", ty, n);
317 goto bad;
318 }
319 BODY(n);
320 } else if (ty >= MESH_F_BLOB_PUT && ty <= MESH_F_BLOB) {
321 HEAD(MESH_JOB_HEAD);
322 n = mpw_u32_at(f->b, at);
323 if (n > MESH_F_BLOB_MAX) {
324 *err = mt_fmt("blob frame type %u of %u bytes > F_BLOB_MAX", ty, n);
325 goto bad;
326 }
327 BODY(n);
328 } else {
329 *err = mt_fmt("bad frame type %u", ty);
330 goto bad;
331 }
332 }
333 return true;
334bad:
335 mpw_bytes_free(f);
336 return false;
337#undef HEAD
338#undef BODY
339}
340
341/* ---- a link */
342
343typedef struct {
344 uint8_t *b;
345 size_t n;
346} qframe;
347
348typedef struct {
349 qframe *v;
350 size_t head, n, cap;
351} queue;
352
353struct mpn_link {
354 int fd;
355 uint32_t from, dead_ms;
356 uint64_t lid;
357 mpn_sink sink;
358 atomic_bool alive;
359 atomic_bool *dialer_alive;
360 pthread_mutex_t wlock; /* the writer lock: one frame at a time on the socket */
361 pthread_mutex_t qmu;
362 pthread_cond_t qcv;
363 queue ctl, bulk;
364 pthread_t reader, writer;
365 bool closed_once;
366};
367
368static bool q_push(queue *q, const uint8_t *f, size_t n) {
369 if (q->n == q->cap) return false;
370 size_t i = (q->head + q->n) % q->cap;
371 q->v[i].b = mt_alloc(n ? n : 1);
372 memcpy(q->v[i].b, f, n);
373 q->v[i].n = n;
374 q->n++;
375 return true;
376}
377
378static bool q_pop(queue *q, qframe *out) {
379 if (!q->n) return false;
380 *out = q->v[q->head];
381 q->head = (q->head + 1) % q->cap;
382 q->n--;
383 return true;
384}
385
386static void q_free(queue *q) {
387 qframe f;
388 while (q_pop(q, &f)) free(f.b);
389 free(q->v);
390}
391
392static void mark_dead(mpn_link *l) {
393 atomic_store(&l->alive, false);
394 if (l->dialer_alive) atomic_store(l->dialer_alive, false);
395 shutdown(l->fd, SHUT_RDWR);
396 pthread_mutex_lock(&l->qmu);
397 pthread_cond_broadcast(&l->qcv);
398 pthread_mutex_unlock(&l->qmu);
399}
400
401static bool link_write(mpn_link *l, const uint8_t *f, size_t n) {
402 pthread_mutex_lock(&l->wlock);
403 bool ok = write_all(l->fd, f, n);
404 pthread_mutex_unlock(&l->wlock);
405 if (ok && l->sink.count && n) l->sink.count(l->sink.ctx, l->from, true, f[0], n);
406 return ok;
407}
408
409static void *reader_main(void *arg) {
410 mpn_link *l = arg;
411 sockread s = {l->fd, l->dead_ms};
412 char *why = NULL;
413 for (;;) {
414 mpw_bytes f;
415 if (!mpn_read_frame(sock_read, &s, &f, &why)) break;
416 if (l->sink.count) l->sink.count(l->sink.ctx, l->from, false, f.b[0], f.n);
417 why = l->sink.frame(l->sink.ctx, l->from, l->lid, f);
418 if (why) break;
419 }
420 mark_dead(l);
421 l->sink.closed(l->sink.ctx, l->from, l->lid, why ? why : "closed");
422 free(why);
423 return NULL;
424}
425
426/* Writes the member's queued frames until the link dies: every control frame queued before each bulk
427 * frame, so a blob in flight delays a beat or a lease by one bulk frame at most. */
428static void *writer_main(void *arg) {
429 mpn_link *l = arg;
430 for (;;) {
431 qframe f;
432 pthread_mutex_lock(&l->qmu);
433 while (!l->ctl.n && !l->bulk.n && atomic_load(&l->alive)) {
434 struct timespec ts;
435 clock_gettime(CLOCK_REALTIME, &ts);
436 ts.tv_nsec += 100 * 1000000L;
437 if (ts.tv_nsec >= 1000000000L) ts.tv_sec++, ts.tv_nsec -= 1000000000L;
438 pthread_cond_timedwait(&l->qcv, &l->qmu, &ts);
439 }
440 if (!atomic_load(&l->alive)) {
441 pthread_mutex_unlock(&l->qmu);
442 return NULL;
443 }
444 bool got = q_pop(&l->ctl, &f) || q_pop(&l->bulk, &f);
445 pthread_mutex_unlock(&l->qmu);
446 if (!got) continue;
447 bool ok = link_write(l, f.b, f.n);
448 free(f.b);
449 if (!ok) {
450 mark_dead(l);
451 return NULL;
452 }
453 }
454}
455
456mpn_link *mpn_link_start(int fd, uint32_t from, uint64_t lid, uint32_t dead_ms, const mpn_sink *sink, atomic_bool *alive) {
457 mpn_link *l = mt_alloc(sizeof *l);
458 l->fd = fd;
459 l->from = from;
460 l->lid = lid;
461 l->dead_ms = dead_ms;
462 l->sink = *sink;
463 l->dialer_alive = alive;
464 atomic_init(&l->alive, true);
465 pthread_mutex_init(&l->wlock, NULL);
466 pthread_mutex_init(&l->qmu, NULL);
467 pthread_cond_init(&l->qcv, NULL);
468 l->ctl = (queue){mt_alloc(MPN_LINK_QUEUE * sizeof(qframe)), 0, 0, MPN_LINK_QUEUE};
469 l->bulk = (queue){mt_alloc(MPN_BULK_QUEUE * sizeof(qframe)), 0, 0, MPN_BULK_QUEUE};
470 set_timeouts(fd, dead_ms);
471 pthread_create(&l->reader, NULL, reader_main, l);
472 pthread_create(&l->writer, NULL, writer_main, l);
473 return l;
474}
475
476uint32_t mpn_link_from(const mpn_link *l) { return l->from; }
477uint64_t mpn_link_lid(const mpn_link *l) { return l->lid; }
478bool mpn_link_alive(const mpn_link *l) { return atomic_load(&l->alive); }
479
480static bool enqueue(mpn_link *l, queue *q, const uint8_t *f, size_t n) {
481 if (!atomic_load(&l->alive)) return false;
482 pthread_mutex_lock(&l->qmu);
483 bool ok = q_push(q, f, n);
484 pthread_cond_signal(&l->qcv);
485 pthread_mutex_unlock(&l->qmu);
486 return ok;
487}
488
489bool mpn_send(mpn_link *l, const uint8_t *f, size_t n) { return enqueue(l, &l->ctl, f, n); }
490bool mpn_send_bulk(mpn_link *l, const uint8_t *f, size_t n) { return enqueue(l, &l->bulk, f, n); }
491
492bool mpn_write_now(mpn_link *l, const uint8_t *f, size_t n) {
493 if (!atomic_load(&l->alive)) return false;
494 if (!link_write(l, f, n)) {
495 mark_dead(l);
496 return false;
497 }
498 return true;
499}
500
501void mpn_link_close(mpn_link *l) { mark_dead(l); }
502
503void mpn_link_free(mpn_link *l) {
504 if (!l) return;
505 mark_dead(l);
506 pthread_join(l->reader, NULL);
507 pthread_join(l->writer, NULL);
508 close(l->fd);
509 q_free(&l->ctl);
510 q_free(&l->bulk);
511 pthread_mutex_destroy(&l->wlock);
512 pthread_mutex_destroy(&l->qmu);
513 pthread_cond_destroy(&l->qcv);
514 free(l);
515}
516
517/* ---- the acceptor and dialers */
518
519struct mpn_acceptor {
520 int fd;
521 uint32_t me, n, dead_ms;
522 mpk_caps caps;
523 mpn_owner owner;
524 pthread_t th;
525 atomic_bool stop;
526};
527
528typedef struct {
529 mpn_acceptor *a;
530 int fd;
531} hs_job;
532
533static void *handshake_main(void *arg) {
534 hs_job *j = arg;
535 mpk_join join;
536 char *err = NULL;
537 if (mpn_handshake(j->fd, j->a->me, j->a->n, &j->a->caps, false, 0, j->a->dead_ms, &join, &err)) j->a->owner.linked(j->a->owner.ctx, j->fd, &join, NULL);
538 else {
539 if (j->a->owner.log) {
540 char *line = mt_fmt("peer: link from a lower system refused: %s", err);
541 j->a->owner.log(j->a->owner.ctx, line);
542 free(line);
543 }
544 free(err);
545 close(j->fd);
546 }
547 free(j);
548 return NULL;
549}
550
551static void *acceptor_main(void *arg) {
552 mpn_acceptor *a = arg;
553 while (!atomic_load(&a->stop)) {
554 struct pollfd p = {.fd = a->fd, .events = POLLIN};
555 if (poll(&p, 1, 100) <= 0) continue;
556 int c = accept(a->fd, NULL, NULL);
557 if (c < 0) continue;
558 hs_job *j = mt_alloc(sizeof *j);
559 j->a = a, j->fd = c;
560 pthread_t t;
561 pthread_create(&t, NULL, handshake_main, j);
562 pthread_detach(t);
563 }
564 return NULL;
565}
566
567mpn_acceptor *mpn_acceptor_start(int listen_fd, uint32_t me, uint32_t n, const mpk_caps *caps, uint32_t dead_ms, const mpn_owner *owner) {
568 mpn_acceptor *a = mt_alloc(sizeof *a);
569 *a = (mpn_acceptor){.fd = listen_fd, .me = me, .n = n, .dead_ms = dead_ms, .caps = *caps, .owner = *owner};
570 atomic_init(&a->stop, false);
571 pthread_create(&a->th, NULL, acceptor_main, a);
572 return a;
573}
574
575void mpn_acceptor_stop(mpn_acceptor *a) {
576 if (!a) return;
577 atomic_store(&a->stop, true);
578 pthread_join(a->th, NULL);
579 close(a->fd);
580 free(a);
581}
582
583struct mpn_dialer {
584 uint32_t j, me, n, dead_ms;
585 char *addr;
586 mpk_caps caps;
587 mpn_owner owner;
588 pthread_t th;
589 atomic_bool stop, alive;
590};
591
592static void sleep_ms(uint32_t ms) {
593 struct timespec ts = {ms / 1000, (long)(ms % 1000) * 1000000L};
594 nanosleep(&ts, NULL);
595}
596
597static void *dialer_main(void *arg) {
598 mpn_dialer *d = arg;
599 uint32_t backoff = 100;
600 char *said = mt_strdup("");
601 while (!atomic_load(&d->stop)) {
602 char *err = NULL;
603 int fd = mpn_connect(d->addr, d->dead_ms, &err);
604 mpk_join join;
605 if (fd >= 0 && !mpn_handshake(fd, d->me, d->n, &d->caps, true, d->j, d->dead_ms, &join, &err)) {
606 close(fd);
607 fd = -1;
608 }
609 if (fd >= 0) {
610 backoff = 100;
611 mt_set(&said, "");
612 atomic_store(&d->alive, true);
613 d->owner.linked(d->owner.ctx, fd, &join, &d->alive);
614 while (atomic_load(&d->alive) && !atomic_load(&d->stop)) sleep_ms(50);
615 continue;
616 }
617 /* Refusals are loud once (a version or size mismatch), not every retry. */
618 if (strcmp(err, said) != 0 && !mt_starts(err, "connect ") && d->owner.log) {
619 char *line = mt_fmt("peer: link to system %u (%s): %s", d->j, d->addr, err);
620 d->owner.log(d->owner.ctx, line);
621 free(line);
622 }
623 mt_set(&said, err);
624 free(err);
625 for (uint32_t t = 0; t < backoff && !atomic_load(&d->stop); t += 20) sleep_ms(20);
626 backoff = backoff * 2 > 1000 ? 1000 : backoff * 2;
627 }
628 free(said);
629 return NULL;
630}
631
632mpn_dialer *mpn_dialer_start(uint32_t j, const char *addr, uint32_t me, uint32_t n, const mpk_caps *caps, uint32_t dead_ms, const mpn_owner *owner) {
633 mpn_dialer *d = mt_alloc(sizeof *d);
634 *d = (mpn_dialer){.j = j, .me = me, .n = n, .dead_ms = dead_ms, .addr = mt_strdup(addr), .caps = *caps, .owner = *owner};
635 atomic_init(&d->stop, false);
636 atomic_init(&d->alive, false);
637 pthread_create(&d->th, NULL, dialer_main, d);
638 return d;
639}
640
641void mpn_dialer_stop(mpn_dialer *d) {
642 if (!d) return;
643 atomic_store(&d->stop, true);
644 pthread_join(d->th, NULL);
645 free(d->addr);
646 free(d);
647}