1/* @ajax/mesh-peer-link: the fleet's links (peer.rs): a full mesh of TCP links, system k listening
2 * (k > 0) and dialing every j > k. Each link opens with F_JOIN both ways and then lives on its own:
3 * a reader thread reads whole frames (refusing a hostile length before reading it) and hands each to
4 * the owner, and a writer thread sends what is queued, every control frame before each bulk frame. A
5 * link that received nothing for dead_ms closes; its dialer redials (backoff 100 ms .. 1 s). */
6#ifndef MESH_PEER_LINK_H
7#define MESH_PEER_LINK_H
8
9#include "mesh/peer_codec.h"
10#include "mesh/place_wire.h"
11
12#include <stdatomic.h>
13#include <stdbool.h>
14#include <stdint.h>
15
16#define MPN_LINK_QUEUE 1024 /* control frames queued per link before new ones are dropped */
17#define MPN_BULK_QUEUE 64   /* bulk frames (blobs) queued per link */
18
19/* A TCP connect bounded in time; -1 with *err ("connect host:port: why"). */
20int mpn_connect(const char *addr, uint32_t timeout_ms, char **err);
21/* JOIN both ways on a fresh socket (has_expect: the system dialed); sets TCP_NODELAY and the dead time
22 * as the socket's read and write timeouts. */
23bool 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);
24/* Listens on addr ("host:port"); -1 with *err. */
25int mpn_listen(const char *addr, char **err);
26
27/* The total length of a frame given its first bytes: reads through read(ctx, buf, n) (false: the
28 * socket's error in *err) and returns the whole frame (type byte first). Exactly the reader's rules. */
29typedef bool (*mpn_read_fn)(void *ctx, uint8_t *buf, size_t n, char **err);
30bool mpn_read_frame(mpn_read_fn read, void *ctx, mpw_bytes *frame, char **err);
31
32typedef struct mpn_link mpn_link;
33
34/* What a link reports to its owner (called on the link's reader thread). */
35typedef struct {
36    void *ctx;
37    /* A whole frame from member from on link lid (type byte first; takes it). A non-NULL return (a
38     * malloc'd reason) closes the link with it, as the reader's own errors do. */
39    char *(*frame)(void *ctx, uint32_t from, uint64_t lid, mpw_bytes frame);
40    /* The link closed (why: "closed", "silent for N ms", or the read or decode error). */
41    void (*closed)(void *ctx, uint32_t from, uint64_t lid, const char *why);
42    /* Bytes counted (net counters): tx when written, rx when a frame was read whole. */
43    void (*count)(void *ctx, uint32_t k, bool tx, uint8_t type, size_t bytes);
44} mpn_sink;
45
46/* Starts a JOINed link's reader and writer threads (takes fd). alive: the dialer's flag (NULL for an
47 * accepted link), cleared when the link dies so the dialer redials. */
48mpn_link *mpn_link_start(int fd, uint32_t from, uint64_t lid, uint32_t dead_ms, const mpn_sink *sink, atomic_bool *alive);
49uint32_t mpn_link_from(const mpn_link *l);
50uint64_t mpn_link_lid(const mpn_link *l);
51bool mpn_link_alive(const mpn_link *l);
52/* Queues a frame; false when the queue is full (it is dropped) or the link is dead. Copies f. */
53bool mpn_send(mpn_link *l, const uint8_t *f, size_t n);
54bool mpn_send_bulk(mpn_link *l, const uint8_t *f, size_t n);
55/* Writes a frame now, under the writer lock (the lockstep rounds' path). */
56bool mpn_write_now(mpn_link *l, const uint8_t *f, size_t n);
57/* Closes the socket (a blocked send returns); the threads end. */
58void mpn_link_close(mpn_link *l);
59/* Closes, joins the threads and frees. */
60void mpn_link_free(mpn_link *l);
61
62/* Accepts the lower systems' links on a listening socket, handshakes each on its own thread, and hands
63 * the JOINed socket over (refusals are logged). */
64typedef struct {
65    void *ctx;
66    void (*linked)(void *ctx, int fd, const mpk_join *join, atomic_bool *alive);
67    void (*log)(void *ctx, const char *line);
68} mpn_owner;
69
70typedef struct mpn_acceptor mpn_acceptor;
71mpn_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);
72void mpn_acceptor_stop(mpn_acceptor *a);
73
74/* Keeps a link to system j (> me) up: dial, JOIN, hand over, redial once *alive goes false (the owner
75 * clears it when the link dies). */
76typedef struct mpn_dialer mpn_dialer;
77mpn_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);
78void mpn_dialer_stop(mpn_dialer *d);
79
80#endif