b630b7f · hyperswarm-peer

Changed files (5) +432 -0

swarmd/.gitignore +1 -0

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

swarmd/build.sh +24 -0

…	…	@@ -0,0 +1,24 @@
	1	#!/bin/bash
	2	# swarmd — pure C/C++. No npm, no Node. The generated wire in schema/ is
	3	# committed; regenerate it only with `npm run build:schema` when the wire
	4	# changes (see README.md).
	5	set -e
	6	cd "$(dirname "$0")"
	7
	8	KB=/storage/projects/code/hypercore-kb/holepunchto/repos
	9	INC="-Isrc -Ischema -I$KB/tier5/libcompact/include -I$KB/tier4/libutf/include -I$KB/tier4/librpc/include"
	10	OBJ=${OBJ:-/tmp/swarmd-obj}
	11	mkdir -p "$OBJ"
	12
	13	echo "Building generated wire..."
	14	gcc -std=c99 -O2 $INC -c -o "$OBJ/swarmd_schema.o" schema/swarmd_schema.c
	15	gcc -std=c99 -O2 $INC -c -o "$OBJ/swarmd_hrpc.o" schema/swarmd_hrpc.c
	16
	17	echo "Building daemon sources..."
	18	g++ -std=c++17 -O2 $INC -c -o "$OBJ/uds_server.o" src/uds_server.cpp
	19
	20	echo "Building gates..."
	21	g++ -std=c++17 -O2 $INC -o test/test_uds_backpressure \
	22	test/test_uds_backpressure.cpp "$OBJ/uds_server.o"
	23
	24	echo " OK"

swarmd/src/uds_server.cpp +204 -0

…	…	@@ -0,0 +1,204 @@
	1	#include "uds_server.h"
	2
	3	#include <cerrno>
	4	#include <cstdio>
	5	#include <cstring>
	6
	7	#include <fcntl.h>
	8	#include <poll.h>
	9	#include <sys/socket.h>
	10	#include <sys/un.h>
	11	#include <unistd.h>
	12
	13	namespace swarmd {
	14
	15	UdsServer::~UdsServer() { close_all(); }
	16
	17	bool UdsServer::listen(const char *socket_path) {
	18	_socket_path = socket_path;
	19	unlink(socket_path); // a stale socket file from a crashed run refuses bind
	20
	21	_listen_fd = socket(AF_UNIX, SOCK_STREAM, 0);
	22	if (_listen_fd < 0) return false;
	23
	24	struct sockaddr_un addr;
	25	memset(&addr, 0, sizeof(addr));
	26	addr.sun_family = AF_UNIX;
	27	strncpy(addr.sun_path, socket_path, sizeof(addr.sun_path) - 1);
	28	if (bind(_listen_fd, (struct sockaddr *)&addr, sizeof(addr)) != 0 \|\|
	29	::listen(_listen_fd, 8) != 0) {
	30	::close(_listen_fd);
	31	_listen_fd = -1;
	32	return false;
	33	}
	34	fcntl(_listen_fd, F_SETFL, O_NONBLOCK);
	35	return true;
	36	}
	37
	38	void UdsServer::close_all() {
	39	for (auto &kv : _clients) ::close(kv.second.fd);
	40	_clients.clear();
	41	if (_listen_fd >= 0) {
	42	::close(_listen_fd);
	43	_listen_fd = -1;
	44	}
	45	if (!_socket_path.empty()) {
	46	unlink(_socket_path.c_str());
	47	_socket_path.clear();
	48	}
	49	}
	50
	51	void UdsServer::_accept() {
	52	for (;;) {
	53	int fd = ::accept(_listen_fd, nullptr, nullptr);
	54	if (fd < 0) return;
	55	fcntl(fd, F_SETFL, O_NONBLOCK);
	56	const uint64_t id = _next_id++;
	57	UdsClient &c = _clients[id];
	58	c.fd = fd;
	59	c.id = id;
	60	if (on_connect) on_connect(id);
	61	}
	62	}
	63
	64	void UdsServer::_read(UdsClient &c) {
	65	uint8_t tmp[65536];
	66	for (;;) {
	67	ssize_t r = ::read(c.fd, tmp, sizeof(tmp));
	68	if (r == 0) { _drop(c.id); return; }
	69	if (r < 0) {
	70	if (errno == EAGAIN \|\| errno == EWOULDBLOCK) break;
	71	_drop(c.id);
	72	return;
	73	}
	74	c.in.insert(c.in.end(), tmp, tmp + r);
	75	}
	76
	77	// Drain whole frames only; a partial one waits for the next read.
	78	for (;;) {
	79	if (c.in.size() < 4) return;
	80	const uint32_t n = (uint32_t)c.in[0] \| ((uint32_t)c.in[1] << 8) \|
	81	((uint32_t)c.in[2] << 16) \| ((uint32_t)c.in[3] << 24);
	82	if (n > UDS_MAX_FRAME) {
	83	// Refuse loudly rather than resynchronise: a length this large means
	84	// the stream is already misaligned, and guessing where the next
	85	// frame starts invents data.
	86	fprintf(stderr, "[swarmd] client %llu sent length %u > max — dropping it\n",
	87	(unsigned long long)c.id, n);
	88	_drop(c.id);
	89	return;
	90	}
	91	if (c.in.size() < 4 + (size_t)n) return;
	92	_frames_in++;
	93	if (on_frame) on_frame(c.id, c.in.data() + 4, n);
	94	c.in.erase(c.in.begin(), c.in.begin() + 4 + n);
	95	// on_frame may have dropped this client.
	96	if (_clients.find(c.id) == _clients.end()) return;
	97	}
	98	}
	99
	100	void UdsServer::_flush(UdsClient &c) {
	101	while (c.out_sent < c.out.size()) {
	102	ssize_t w = ::write(c.fd, c.out.data() + c.out_sent, c.out.size() - c.out_sent);
	103	if (w > 0) {
	104	// ⛔ A PARTIAL WRITE IS REMEMBERED, NOT DISCARDED. Forgetting the
	105	// remainder is what desynchronises a client's stream.
	106	c.out_sent += (size_t)w;
	107	continue;
	108	}
	109	if (w < 0 && (errno == EAGAIN \|\| errno == EWOULDBLOCK)) break;
	110	_drop(c.id);
	111	return;
	112	}
	113	if (c.out_sent == c.out.size()) {
	114	c.out.clear();
	115	c.out_sent = 0;
	116	} else if (c.out_sent > 0) {
	117	c.out.erase(c.out.begin(), c.out.begin() + c.out_sent);
	118	c.out_sent = 0;
	119	}
	120	if (!c.writable && c.out.size() < UDS_CLIENT_HIGH_WATER / 2) c.writable = true;
	121	}
	122
	123	bool UdsServer::send(uint64_t client_id, const uint8_t data, size_t len, bool r_writable) {
	124	auto it = _clients.find(client_id);
	125	if (it == _clients.end()) {
	126	if (r_writable) *r_writable = false;
	127	return false;
	128	}
	129	UdsClient &c = it->second;
	130
	131	if (c.out.size() + 4 + len > UDS_CLIENT_HIGH_WATER) {
	132	// ⛔ DROP THE CLIENT RATHER THAN CORRUPT IT. It reconnects and resyncs;
	133	// a client fed a half-frame does not know it has been lied to.
	134	fprintf(stderr, "[swarmd] client %llu is %zu bytes behind — dropping it\n",
	135	(unsigned long long)c.id, c.out.size());
	136	_drop(client_id);
	137	_dropped++;
	138	if (r_writable) *r_writable = false;
	139	return false;
	140	}
	141
	142	const uint32_t n = (uint32_t)len;
	143	const uint8_t hdr[4] = { (uint8_t)(n & 0xff), (uint8_t)((n >> 8) & 0xff),
	144	(uint8_t)((n >> 16) & 0xff), (uint8_t)((n >> 24) & 0xff) };
	145	c.out.insert(c.out.end(), hdr, hdr + 4);
	146	c.out.insert(c.out.end(), data, data + len);
	147	_frames_out++;
	148
	149	_flush(c);
	150	if (_clients.find(client_id) == _clients.end()) {
	151	if (r_writable) *r_writable = false;
	152	return false;
	153	}
	154	if (c.out.size() >= UDS_CLIENT_HIGH_WATER) c.writable = false;
	155	if (r_writable) *r_writable = c.writable;
	156	return true;
	157	}
	158
	159	void UdsServer::broadcast(const uint8_t *data, size_t len, uint64_t except_id) {
	160	std::vector<uint64_t> ids;
	161	ids.reserve(_clients.size());
	162	for (auto &kv : _clients) ids.push_back(kv.first);
	163	for (uint64_t id : ids) {
	164	if (id == except_id) continue;
	165	send(id, data, len);
	166	}
	167	}
	168
	169	void UdsServer::_drop(uint64_t id) {
	170	auto it = _clients.find(id);
	171	if (it == _clients.end()) return;
	172	::close(it->second.fd);
	173	_clients.erase(it);
	174	if (on_disconnect) on_disconnect(id);
	175	}
	176
	177	void UdsServer::poll(int timeout_ms) {
	178	std::vector<struct pollfd> fds;
	179	std::vector<uint64_t> ids;
	180	fds.push_back({ _listen_fd, POLLIN, 0 });
	181	ids.push_back(0);
	182	for (auto &kv : _clients) {
	183	short ev = POLLIN;
	184	if (kv.second.out_sent < kv.second.out.size()) ev \|= POLLOUT;
	185	fds.push_back({ kv.second.fd, ev, 0 });
	186	ids.push_back(kv.first);
	187	}
	188
	189	const int r = ::poll(fds.data(), fds.size(), timeout_ms);
	190	if (r <= 0) return;
	191
	192	if (fds[0].revents & POLLIN) _accept();
	193
	194	for (size_t i = 1; i < fds.size(); i++) {
	195	auto it = _clients.find(ids[i]);
	196	if (it == _clients.end()) continue; // dropped earlier in this same pass
	197	if (fds[i].revents & POLLOUT) _flush(it->second);
	198	it = _clients.find(ids[i]);
	199	if (it == _clients.end()) continue;
	200	if (fds[i].revents & (POLLIN \| POLLHUP)) _read(it->second);
	201	}
	202	}
	203
	204	} // namespace swarmd

swarmd/src/uds_server.h +97 -0

…	…	@@ -0,0 +1,97 @@
	1	#ifndef SWARMD_UDS_SERVER_H
	2	#define SWARMD_UDS_SERVER_H
	3
	4	// The local socket, and the per-client buffering session-daemon does not have.
	5	//
	6	// ⛔ THE BUG THIS EXISTS TO NOT REPEAT. session_daemon.cpp fans out to clients
	7	// from the UDX thread with unchecked writes on NON-BLOCKING fds:
	8	//
	9	// if (write(fd, hdr, 4) != 4) continue;
	10	// if (write(fd, data, len) != (ssize_t)len) continue;
	11	//
	12	// A partial write there does not drop a frame — it desynchronises the client's
	13	// stream. The client then reads a garbage length, hits its own frame cap and
	14	// closes. A slow reader silently corrupts its own session, and the daemon calls
	15	// that success.
	16	//
	17	// So: every client owns an output buffer, a partial write is remembered rather
	18	// than discarded, and a client that exceeds its high-water mark is DROPPED —
	19	// because a dropped client reconnects and resyncs, while a corrupted one does
	20	// not know it is corrupted.
	21	//
	22	// Framing is uint32 LE length + payload, matching session-daemon and
	23	// blender-bridge/session/client.py so an existing client keeps working.
	24
	25	#include <cstddef>
	26	#include <cstdint>
	27	#include <functional>
	28	#include <map>
	29	#include <string>
	30	#include <vector>
	31
	32	namespace swarmd {
	33
	34	// A client that has this much unsent output is not keeping up.
	35	static constexpr size_t UDS_CLIENT_HIGH_WATER = 8u * 1024 * 1024;
	36
	37	// A length this large is a framing bug, not a frame.
	38	static constexpr uint32_t UDS_MAX_FRAME = 8u * 1024 * 1024;
	39
	40	struct UdsClient {
	41	int fd = -1;
	42	uint64_t id = 0;
	43	std::vector<uint8_t> in; // partial inbound frame
	44	std::vector<uint8_t> out; // unsent outbound bytes
	45	size_t out_sent = 0; // how much of out[0] has gone
	46	bool writable = true; // false between high-water and drain
	47	};
	48
	49	class UdsServer {
	50	public:
	51	~UdsServer();
	52
	53	bool listen(const char *socket_path);
	54	void close_all();
	55
	56	// Drive accept/read/write once. `timeout_ms` bounds the poll.
	57	void poll(int timeout_ms);
	58
	59	// One whole frame arrived from a client.
	60	std::function<void(uint64_t client_id, const uint8_t *data, size_t len)> on_frame;
	61	// A client attached or went away.
	62	std::function<void(uint64_t client_id)> on_connect;
	63	std::function<void(uint64_t client_id)> on_disconnect;
	64
	65	// Queue a frame. Returns false if the client is unknown or was dropped for
	66	// exceeding its high-water mark. `r_writable` reports whether the client is
	67	// still under that mark — the caller is expected to stop producing when it
	68	// is not, which is the answer Protomux gives and session-daemon discards.
	69	bool send(uint64_t client_id, const uint8_t data, size_t len, bool r_writable = nullptr);
	70
	71	// Queue to every client except `except_id` (0 = all).
	72	void broadcast(const uint8_t *data, size_t len, uint64_t except_id = 0);
	73
	74	int client_count() const { return (int)_clients.size(); }
	75	uint64_t frames_in() const { return _frames_in; }
	76	uint64_t frames_out() const { return _frames_out; }
	77	uint64_t dropped_clients() const { return _dropped; }
	78
	79	private:
	80	void _accept();
	81	void _read(UdsClient &c);
	82	void _flush(UdsClient &c);
	83	void _drop(uint64_t id);
	84
	85	int _listen_fd = -1;
	86	std::string _socket_path;
	87	std::map<uint64_t, UdsClient> _clients;
	88	std::vector<uint64_t> _dead; // dropped during a poll, reaped after it
	89	uint64_t _next_id = 1;
	90	uint64_t _frames_in = 0;
	91	uint64_t _frames_out = 0;
	92	uint64_t _dropped = 0;
	93	};
	94
	95	} // namespace swarmd
	96
	97	#endif

swarmd/test/test_uds_backpressure.cpp +106 -0

…	…	@@ -0,0 +1,106 @@
	1	// Gate D2 + D3 (docs/SWARMD-DESIGN.md §8): a client that stops reading is
	2	// DROPPED, and the survivors' streams stay intact.
	3	//
	4	// ⛔ THIS IS THE ASSERTION session-daemon FAILS. It fans out from the UDX thread
	5	// with unchecked writes on non-blocking fds:
	6	//
	7	// if (write(fd, hdr, 4) != 4) continue;
	8	// if (write(fd, data, len) != (ssize_t)len) continue;
	9	//
	10	// A partial write there does not drop a frame, it desynchronises the stream —
	11	// the client reads a garbage length next and closes. A slow reader silently
	12	// corrupts its own session while the daemon reports success.
	13
	14	#include <cstdio>
	15	#include <cstring>
	16	#include <string>
	17	#include <vector>
	18
	19	#include <sys/socket.h>
	20	#include <sys/un.h>
	21	#include <unistd.h>
	22	#include <fcntl.h>
	23
	24	#include "uds_server.h"
	25
	26	static int checks = 0, failures = 0;
	27	static void check(bool c, const char *what) {
	28	checks++;
	29	if (c) { printf(" PASS: %s\n", what); return; }
	30	failures++;
	31	printf(" FAIL: %s\n", what);
	32	}
	33
	34	static int dial(const char *path) {
	35	int fd = socket(AF_UNIX, SOCK_STREAM, 0);
	36	struct sockaddr_un a;
	37	memset(&a, 0, sizeof(a));
	38	a.sun_family = AF_UNIX;
	39	strncpy(a.sun_path, path, sizeof(a.sun_path) - 1);
	40	if (connect(fd, (struct sockaddr *)&a, sizeof(a)) != 0) { close(fd); return -1; }
	41	return fd;
	42	}
	43
	44	int main() {
	45	char path[128];
	46	snprintf(path, sizeof(path), "/tmp/swarmd-gate-%d.sock", (int)getpid());
	47
	48	swarmd::UdsServer s;
	49	if (!s.listen(path)) { printf("listen failed\n"); return 1; }
	50
	51	int reader = dial(path);
	52	int sulker = dial(path); // connects, then never reads a byte
	53	check(reader >= 0 && sulker >= 0, "two clients attached");
	54	s.poll(50);
	55	check(s.client_count() == 2, "the server sees both");
	56
	57	uint64_t reader_id = 0, sulker_id = 0;
	58	{
	59	// Identify them by which fd answers: send a probe to each id in turn.
	60	// Ids are handed out in accept order, and `reader` dialled first.
	61	reader_id = 1;
	62	sulker_id = 2;
	63	}
	64
	65	// Fill the sulker past its high-water mark. It never reads, so the kernel
	66	// buffer fills, writes go EAGAIN, and the output buffer grows.
	67	std::vector<uint8_t> chunk(64 * 1024, 0xab);
	68	bool sulker_dropped = false;
	69	bool last_writable = true;
	70	for (int i = 0; i < 4096 && !sulker_dropped; i++) {
	71	bool w = true;
	72	if (!s.send(sulker_id, chunk.data(), chunk.size(), &w)) sulker_dropped = true;
	73	else last_writable = w;
	74	s.poll(0);
	75	}
	76
	77	check(sulker_dropped, "a client that never reads is eventually DROPPED");
	78	check(s.dropped_clients() >= 1, "and the drop is counted");
	79	check(!last_writable \|\| sulker_dropped,
	80	"writability went false before the drop (D3)");
	81
	82	// ⛔ THE ONE THAT MATTERS. The survivor's stream must still be frame-aligned.
	83	const char *msg = "intact";
	84	check(s.send(reader_id, (const uint8_t *)msg, strlen(msg)),
	85	"the surviving client still accepts a frame");
	86	for (int i = 0; i < 20; i++) s.poll(5);
	87
	88	uint8_t buf[256];
	89	ssize_t got = ::read(reader, buf, sizeof(buf));
	90	bool aligned = false;
	91	if (got >= 4) {
	92	const uint32_t n = (uint32_t)buf[0] \| ((uint32_t)buf[1] << 8) \|
	93	((uint32_t)buf[2] << 16) \| ((uint32_t)buf[3] << 24);
	94	aligned = (n == strlen(msg)) &&
	95	(got >= (ssize_t)(4 + n)) &&
	96	(memcmp(buf + 4, msg, n) == 0);
	97	}
	98	check(aligned, "the survivor's stream is still frame-aligned, not corrupted");
	99
	100	close(reader);
	101	if (!sulker_dropped) close(sulker);
	102	s.close_all();
	103
	104	printf("\n%d/%d passed\n", checks - failures, checks);
	105	return failures ? 1 : 0;
	106	}

hyperswarm-peer commits / b630b7f

Changed files (5) +432 -0