1818f6d · hyperswarm-peer

Changed files (6) +399 -2

swarmd/.gitignore +1 -0

…	…	@@ -1,2 +1,3 @@
1	1	node_modules/
2	2	test/test_uds_backpressure
	3	test/test_control

swarmd/README.md +12 -2

…	…	@@ -28,7 +28,17 @@ encodings the same way, with `schema/build.js` and the resulting
28	28	- libutf — string views (pulled in by libcompact)
29	29	- librpc — the C RPC codec
30	30
	31	## Build
	32
	33	./build.sh # gcc/g++ only. No npm.
	34	./test/test_uds_backpressure
	35	./test/test_control
	36
31	37	## Status
32	38
33		Wire generated and compiling. No daemon yet — see `../docs/SWARMD-DESIGN.md`,
34		§9 for the order and §8 for the gates.
	39	schema/ generated wire, committed 3316 lines
	40	uds_server local socket + per-client buffering 7/7 (D2, D3)
	41	control generated RPC dispatch on the socket 10/10
	42
	43	⏳ Not built: swarm attachment (the UDX thread), channels, content. See
	44	`../docs/SWARMD-DESIGN.md` §9 for the order and §8 for the remaining gates.

swarmd/build.sh +27 -0

…	…	@@ -10,15 +10,42 @@ INC="-Isrc -Ischema -I$KB/tier5/libcompact/include -I$KB/tier4/libutf/include -I
10	10	OBJ=${OBJ:-/tmp/swarmd-obj}
11	11	mkdir -p "$OBJ"
12	12
	13	# The holepunch C libraries. Sources compiled in directly rather than via CMake:
	14	# swarmd is one small daemon, and a cmake-fetch tree for three static libs costs
	15	# more than it explains.
	16	echo "Building libcompact / libutf / librpc..."
	17	CLIBS=""
	18	for c in "$KB"/tier5/libcompact/src/*.c; do
	19	o="$OBJ/compact_$(basename "$c" .c).o"
	20	gcc -std=c99 -O2 -I"$KB/tier5/libcompact/include" -I"$KB/tier4/libutf/include" -c -o "$o" "$c" 2>/dev/null
	21	CLIBS="$CLIBS $o"
	22	done
	23	for c in "$KB"/tier4/libutf/src/.c "$KB"/tier4/libutf/src//*.c; do
	24	[ -f "$c" ] \|\| continue
	25	o="$OBJ/utf_$(echo "$c" \| md5sum \| cut -c1-8).o"
	26	gcc -std=c99 -O2 -I"$KB/tier4/libutf/include" -c -o "$o" "$c" 2>/dev/null \|\| continue
	27	CLIBS="$CLIBS $o"
	28	done
	29	for c in "$KB"/tier4/librpc/src/*.c; do
	30	o="$OBJ/rpc_$(basename "$c" .c).o"
	31	gcc -std=c99 -O2 -I"$KB/tier4/librpc/include" -I"$KB/tier5/libcompact/include" -I"$KB/tier4/libutf/include" -c -o "$o" "$c" 2>/dev/null \|\| continue
	32	CLIBS="$CLIBS $o"
	33	done
	34
13	35	echo "Building generated wire..."
14	36	gcc -std=c99 -O2 $INC -c -o "$OBJ/swarmd_schema.o" schema/swarmd_schema.c
15	37	gcc -std=c99 -O2 $INC -c -o "$OBJ/swarmd_hrpc.o" schema/swarmd_hrpc.c
16	38
17	39	echo "Building daemon sources..."
18	40	g++ -std=c++17 -O2 $INC -c -o "$OBJ/uds_server.o" src/uds_server.cpp
	41	g++ -std=c++17 -O2 $INC -c -o "$OBJ/control.o" src/control.cpp 2>/dev/null
19	42
20	43	echo "Building gates..."
21	44	g++ -std=c++17 -O2 $INC -o test/test_uds_backpressure \
22	45	test/test_uds_backpressure.cpp "$OBJ/uds_server.o"
23	46
	47	g++ -std=c++17 -O2 $INC -o test/test_control \
	48	test/test_control.cpp "$OBJ/uds_server.o" "$OBJ/control.o" \
	49	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS
	50
24	51	echo " OK"

swarmd/src/control.cpp +121 -0

…	…	@@ -0,0 +1,121 @@
	1	#include "control.h"
	2
	3	#include <cstdio>
	4	#include <cstdlib>
	5	#include <cstring>
	6
	7	extern "C" {
	8	#include <compact.h>
	9	#include <rpc.h>
	10	}
	11
	12	namespace swarmd {
	13
	14	// The dispatcher's handlers are plain C function pointers with a void* ctx, so
	15	// the instance travels through ctx rather than through a capture.
	16	struct Ctx {
	17	Control *self;
	18	Control::Session *session;
	19	ControlBackend *backend;
	20	uint64_t *hellos;
	21	};
	22
	23	static int on_hello(void vctx, const swarmd_hello_request_t req,
	24	swarmd_hello_response_t res, hrpc_error_t ) {
	25	Ctx ctx = (Ctx )vctx;
	26	// ⛔ COPY OUT. req's views point into the inbound frame buffer, which is
	27	// reused the moment this returns.
	28	memcpy(ctx->session->app_identity, req->app_identity, 32);
	29	ctx->session->name.assign((const char *)req->name.data, req->name.len);
	30	ctx->session->hello = true;
	31	(*ctx->hellos)++;
	32
	33	res->client_id = 0; // filled by the caller, which knows it
	34	memset(res->daemon_pubkey, 0, 32); // no transport attached yet
	35	return 0;
	36	}
	37
	38	static int on_status_get(void vctx, const swarmd_none_t ,
	39	swarmd_status_t res, hrpc_error_t ) {
	40	Ctx ctx = (Ctx )vctx;
	41	memset(res, 0, sizeof(*res));
	42	if (ctx->backend->status) ctx->backend->status(res);
	43	return 0;
	44	}
	45
	46	static int on_swarm_join(void vctx, const swarmd_swarm_join_t req,
	47	swarmd_ok_t res, hrpc_error_t ) {
	48	Ctx ctx = (Ctx )vctx;
	49	res->ok = ctx->backend->swarm_join
	50	? ctx->backend->swarm_join(req->topic, req->server, req->client)
	51	: false;
	52	return 0;
	53	}
	54
	55	static int on_swarm_leave(void vctx, const swarmd_topic_ref_t req,
	56	swarmd_ok_t res, hrpc_error_t ) {
	57	Ctx ctx = (Ctx )vctx;
	58	res->ok = ctx->backend->swarm_leave ? ctx->backend->swarm_leave(req->topic) : false;
	59	return 0;
	60	}
	61
	62	void Control::attach(UdsServer *uds, ControlBackend backend) {
	63	_uds = uds;
	64	_backend = std::move(backend);
	65	_uds->on_connect = [this](uint64_t id) { _sessions[id] = Session(); };
	66	_uds->on_disconnect = [this](uint64_t id) { _sessions.erase(id); };
	67	_uds->on_frame = [this](uint64_t id, const uint8_t *d, size_t n) { _on_frame(id, d, n); };
	68	}
	69
	70	const Control::Session *Control::session(uint64_t client_id) const {
	71	auto it = _sessions.find(client_id);
	72	return it == _sessions.end() ? nullptr : &it->second;
	73	}
	74
	75	void Control::_on_frame(uint64_t client_id, const uint8_t *data, size_t len) {
	76	auto sit = _sessions.find(client_id);
	77	if (sit == _sessions.end()) return;
	78
	79	compact_state_t state = { 0, len, (uint8_t *)data };
	80	rpc_message_t msg;
	81	if (rpc_decode_message(&state, &msg) < 0) {
	82	fprintf(stderr, "[swarmd] client %llu sent an undecodable message — dropping it\n",
	83	(unsigned long long)client_id);
	84	return;
	85	}
	86
	87	Ctx ctx = { this, &sit->second, &_backend, &_hellos };
	88	swarmd_hrpc_handlers_t h;
	89	memset(&h, 0, sizeof(h));
	90	h.ctx = &ctx;
	91	h.on_hello = on_hello;
	92	h.on_status_get = on_status_get;
	93	h.on_swarm_join = on_swarm_join;
	94	h.on_swarm_leave = on_swarm_leave;
	95
	96	uint8_t *reply = nullptr;
	97	size_t reply_len = 0;
	98	_current_client = client_id;
	99	const int r = swarmd_hrpc_dispatch(&h, &msg, &reply, &reply_len);
	100	_current_client = 0;
	101
	102	if (r == hrpc_dispatch_reply && reply != nullptr) {
	103	_uds->send(client_id, reply, reply_len);
	104	// ⛔ THE CALLER FREES. Every _encode_ heap-allocates its output.
	105	free(reply);
	106	}
	107	}
	108
	109	void Control::push_status(uint64_t client_id) {
	110	swarmd_status_t st;
	111	memset(&st, 0, sizeof(st));
	112	if (_backend.status) _backend.status(&st);
	113
	114	uint8_t *out = nullptr;
	115	size_t out_len = 0;
	116	if (swarmd_encode_on_status(&st, &out, &out_len) < 0 \|\| out == nullptr) return;
	117	_uds->send(client_id, out, out_len);
	118	free(out); // caller frees
	119	}
	120
	121	} // namespace swarmd

swarmd/src/control.h +70 -0

…	…	@@ -0,0 +1,70 @@
	1	#ifndef SWARMD_CONTROL_H
	2	#define SWARMD_CONTROL_H
	3
	4	// The control plane: generated RPC on top of the local socket.
	5	//
	6	// A UDS frame is one bare-rpc message. librpc decodes it, the GENERATED
	7	// dispatcher routes it to a handler, and any reply goes back as another frame.
	8	// Nothing here parses a payload by hand — that is the entire point of
	9	// generating the wire (docs/SWARMD-DESIGN.md §3).
	10	//
	11	// ⛔ TWO OWNERSHIP RULES, both from hrpc-c's README and both the opposite of
	12	// the careless assumption:
	13	// 1. every _encode_ HEAP-ALLOCATES its output and THE CALLER FREES IT
	14	// 2. decoded views point INTO the inbound frame buffer, so anything kept
	15	// past the handler must be copied out
	16	// They are enforced at the two places they matter, marked below.
	17
	18	#include <cstdint>
	19	#include <functional>
	20	#include <map>
	21	#include <string>
	22
	23	#include "uds_server.h"
	24
	25	extern "C" {
	26	#include "swarmd_hrpc.h"
	27	#include "swarmd_schema.h"
	28	}
	29
	30	namespace swarmd {
	31
	32	// What the daemon can actually answer. Filled in by the daemon; kept behind an
	33	// interface so the control plane can be gated with no swarm at all.
	34	struct ControlBackend {
	35	std::function<void(swarmd_status_t *out)> status;
	36	std::function<bool(const uint8_t topic[32], bool server, bool client)> swarm_join;
	37	std::function<bool(const uint8_t topic[32])> swarm_leave;
	38	};
	39
	40	class Control {
	41	public:
	42	void attach(UdsServer *uds, ControlBackend backend);
	43
	44	// Per-client session state. A client that has not said HELLO has no
	45	// identity, and identity is what echo-suppression and membership need.
	46	struct Session {
	47	bool hello = false;
	48	uint8_t app_identity[32] = { 0 };
	49	std::string name;
	50	};
	51
	52	const Session *session(uint64_t client_id) const;
	53	uint64_t hellos() const { return _hellos; }
	54
	55	// Daemon -> client events. Encoded with the generated encoders.
	56	void push_status(uint64_t client_id);
	57
	58	private:
	59	void _on_frame(uint64_t client_id, const uint8_t *data, size_t len);
	60
	61	UdsServer *_uds = nullptr;
	62	ControlBackend _backend;
	63	std::map<uint64_t, Session> _sessions;
	64	uint64_t _current_client = 0; // the client being dispatched, for handlers
	65	uint64_t _hellos = 0;
	66	};
	67
	68	} // namespace swarmd
	69
	70	#endif

swarmd/test/test_control.cpp +168 -0

…	…	@@ -0,0 +1,168 @@
	1	// Gate: a real client, over the real socket, through librpc and the GENERATED
	2	// dispatcher. docs/SWARMD-DESIGN.md §5.
	3	//
	4	// This is the proof that the codegen bet pays off end to end:
	5	//
	6	// client encodes with swarmd_encode_hello (generated)
	7	// -> UDS frame (uint32 LE length + payload)
	8	// -> rpc_decode_message (librpc)
	9	// -> swarmd_hrpc_dispatch (generated)
	10	// -> handler
	11	// -> reply encoded (generated)
	12	// -> UDS frame back
	13	// -> swarmd_decode_hello_response (generated)
	14	//
	15	// Not one byte of that is parsed by hand, which was the whole argument for not
	16	// hand-rolling the protocol.
	17
	18	#include <cstdio>
	19	#include <cstdlib>
	20	#include <cstring>
	21	#include <vector>
	22
	23	#include <sys/socket.h>
	24	#include <sys/un.h>
	25	#include <unistd.h>
	26
	27	#include "control.h"
	28
	29	extern "C" {
	30	#include <compact.h>
	31	#include <rpc.h>
	32	}
	33
	34	static int checks = 0, failures = 0;
	35	static void check(bool c, const char *what) {
	36	checks++;
	37	if (c) { printf(" PASS: %s\n", what); return; }
	38	failures++;
	39	printf(" FAIL: %s\n", what);
	40	}
	41
	42	static int dial(const char *path) {
	43	int fd = socket(AF_UNIX, SOCK_STREAM, 0);
	44	struct sockaddr_un a;
	45	memset(&a, 0, sizeof(a));
	46	a.sun_family = AF_UNIX;
	47	strncpy(a.sun_path, path, sizeof(a.sun_path) - 1);
	48	if (connect(fd, (struct sockaddr *)&a, sizeof(a)) != 0) { close(fd); return -1; }
	49	return fd;
	50	}
	51
	52	static void write_frame(int fd, const uint8_t *d, size_t n) {
	53	uint8_t hdr[4] = { (uint8_t)(n & 0xff), (uint8_t)((n >> 8) & 0xff),
	54	(uint8_t)((n >> 16) & 0xff), (uint8_t)((n >> 24) & 0xff) };
	55	(void)!write(fd, hdr, 4);
	56	(void)!write(fd, d, n);
	57	}
	58
	59	static bool read_frame(int fd, std::vector<uint8_t> &out) {
	60	uint8_t hdr[4];
	61	ssize_t r = read(fd, hdr, 4);
	62	if (r != 4) return false;
	63	const uint32_t n = (uint32_t)hdr[0] \| ((uint32_t)hdr[1] << 8) \|
	64	((uint32_t)hdr[2] << 16) \| ((uint32_t)hdr[3] << 24);
	65	out.resize(n);
	66	size_t got = 0;
	67	while (got < n) {
	68	r = read(fd, out.data() + got, n - got);
	69	if (r <= 0) return false;
	70	got += (size_t)r;
	71	}
	72	return true;
	73	}
	74
	75	int main() {
	76	char path[128];
	77	snprintf(path, sizeof(path), "/tmp/swarmd-control-%d.sock", (int)getpid());
	78
	79	swarmd::UdsServer uds;
	80	if (!uds.listen(path)) { printf("listen failed\n"); return 1; }
	81
	82	bool joined = false;
	83	uint8_t joined_topic[32] = { 0 };
	84
	85	swarmd::ControlBackend backend;
	86	backend.status = [](swarmd_status_t *out) {
	87	out->dht_state = 2; // up
	88	out->bootstrapped = true;
	89	out->peers_connected = 3;
	90	out->local_clients = 1;
	91	};
	92	backend.swarm_join = [&](const uint8_t topic[32], bool, bool) {
	93	joined = true;
	94	memcpy(joined_topic, topic, 32);
	95	return true;
	96	};
	97
	98	swarmd::Control control;
	99	control.attach(&uds, backend);
	100
	101	int fd = dial(path);
	102	check(fd >= 0, "client attached");
	103	uds.poll(50);
	104
	105	// ---- HELLO, encoded by the generated client encoder ----
	106	swarmd_hello_request_t hello;
	107	memset(&hello, 0, sizeof(hello));
	108	const char *name = "gate";
	109	hello.name.data = (utf8_t *)name;
	110	hello.name.len = strlen(name);
	111	memset(hello.app_identity, 0x7e, 32);
	112	hello.version = 1;
	113
	114	uint8_t *req = nullptr;
	115	size_t req_len = 0;
	116	check(swarmd_encode_hello(1, &hello, &req, &req_len) >= 0 && req != nullptr,
	117	"generated encoder produced a request");
	118	write_frame(fd, req, req_len);
	119	free(req); // caller frees
	120
	121	for (int i = 0; i < 20; i++) uds.poll(5);
	122	check(control.hellos() == 1, "the dispatcher routed HELLO to its handler");
	123
	124	const swarmd::Control::Session *s = control.session(1);
	125	check(s && s->hello, "the session records that HELLO happened");
	126	uint8_t want[32];
	127	memset(want, 0x7e, 32);
	128	check(s && memcmp(s->app_identity, want, 32) == 0,
	129	"the app identity was COPIED OUT of the frame buffer, not aliased");
	130	check(s && s->name == "gate", "and so was the name");
	131
	132	// ---- the reply comes back, decoded by the generated decoder ----
	133	std::vector<uint8_t> frame;
	134	check(read_frame(fd, frame), "a reply frame came back");
	135	if (!frame.empty()) {
	136	compact_state_t st = { 0, frame.size(), frame.data() };
	137	rpc_message_t msg;
	138	check(rpc_decode_message(&st, &msg) >= 0, "librpc decoded the reply");
	139	swarmd_hello_response_t res;
	140	hrpc_error_t err;
	141	memset(&res, 0, sizeof(res));
	142	check(swarmd_decode_hello_response(&msg, &res, &err) >= 0,
	143	"the generated decoder read the response");
	144	}
	145
	146	// ---- a second command, to prove dispatch is by command id ----
	147	swarmd_swarm_join_t sj;
	148	memset(&sj, 0, sizeof(sj));
	149	memset(sj.topic, 0x11, 32);
	150	sj.server = true;
	151	sj.client = true;
	152	uint8_t *req2 = nullptr;
	153	size_t req2_len = 0;
	154	swarmd_encode_swarm_join(2, &sj, &req2, &req2_len);
	155	write_frame(fd, req2, req2_len);
	156	free(req2);
	157	for (int i = 0; i < 20; i++) uds.poll(5);
	158
	159	uint8_t want_topic[32];
	160	memset(want_topic, 0x11, 32);
	161	check(joined && memcmp(joined_topic, want_topic, 32) == 0,
	162	"SWARM_JOIN reached the backend with the right topic");
	163
	164	close(fd);
	165	uds.close_all();
	166	printf("\n%d/%d passed\n", checks - failures, checks);
	167	return failures ? 1 : 0;
	168	}

hyperswarm-peer commits / 1818f6d

Changed files (6) +399 -2