401d7fb · hyperswarm-peer

Changed files (6) +306 -2

swarmd/.gitignore +1 -0

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

swarmd/README.md +7 -1

…	…	@@ -39,6 +39,12 @@ encodings the same way, with `schema/build.js` and the resulting
39	39	schema/ generated wire, committed 3316 lines
40	40	uds_server local socket + per-client buffering 7/7 (D2, D3)
41	41	control generated RPC dispatch on the socket 10/10
	42	daemon UdxTransport + queue + control plane runs
42	43
43		⏳ Not built: swarm attachment (the UDX thread), channels, content. See
	44	./swarmd --socket /tmp/s.sock # public DHT
	45	[swarmd] listening on /tmp/s.sock
	46	[swarmd] dht=connecting peers=0 local=0
	47	[swarmd] dht=up peers=0 local=0
	48
	49	⏳ Not built: channels (SEND / on-message over Protomux), content. See
44	50	`../docs/SWARMD-DESIGN.md` §9 for the order and §8 for the remaining gates.

swarmd/build.sh +33 -1

…	…	@@ -6,7 +6,8 @@ set -e
6	6	cd "$(dirname "$0")"
7	7
8	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"
	9	PEER=..
	10	INC="-Isrc -Ischema -I$PEER/src -I$PEER/thirdparty/libudx/include -I$KB/tier5/libcompact/include -I$KB/tier4/libutf/include -I$KB/tier4/librpc/include"
10	11	OBJ=${OBJ:-/tmp/swarmd-obj}
11	12	mkdir -p "$OBJ"
12	13
…	…	@@ -40,6 +41,37 @@ echo "Building daemon sources..."
40	41	g++ -std=c++17 -O2 $INC -c -o "$OBJ/uds_server.o" src/uds_server.cpp
41	42	g++ -std=c++17 -O2 $INC -c -o "$OBJ/control.o" src/control.cpp 2>/dev/null
42	43
	44	# The transport. Reuses hyperswarm-peer's object cache when test/build_tests.sh
	45	# has populated it; otherwise builds what swarmd needs.
	46	echo "Building transport..."
	47	TOBJ=${TOBJ:-/tmp/hsp-obj}
	48	PEERSRC="noise_ik noise_handshake noise_stream dht_rpc holepuncher hyperdht dht_connector udx_transport swarm_manager protomux"
	49	TLIBS=""
	50	for n in $PEERSRC; do
	51	o="$OBJ/peer_$n.o"
	52	if [ ! -f "$o" ] \|\| [ "$PEER/src/$n.cpp" -nt "$o" ]; then
	53	g++ -std=c++17 -O2 -D_GNU_SOURCE -I$PEER/src -I$PEER/thirdparty/libudx/include -c -o "$o" "$PEER/src/$n.cpp"
	54	fi
	55	TLIBS="$TLIBS $o"
	56	done
	57	for c in "$PEER"/thirdparty/libudx/src/*.c; do
	58	b=$(basename "$c" .c)
	59	# NOT `[ ... ] && continue`: that returns 1 for every OTHER file and set -e kills the script.
	60	# Platform backends libudx picks at build time. io_uring needs headers we
	61	# do not require; io_win does not compile off Windows.
	62	case "$b" in io_uring\|io_win) continue;; esac
	63	o="$OBJ/udx_$b.o"
	64	[ -f "$o" ] \|\| gcc -O2 -D_GNU_SOURCE -I"$PEER/thirdparty/libudx/include" -c -o "$o" "$c" 2>/dev/null
	65	TLIBS="$TLIBS $o"
	66	done
	67
	68	echo "Building daemon..."
	69	g++ -std=c++17 -O2 $INC -c -o "$OBJ/daemon.o" src/daemon.cpp
	70	g++ -std=c++17 -O2 $INC -o swarmd src/main.cpp \
	71	"$OBJ/daemon.o" "$OBJ/uds_server.o" "$OBJ/control.o" \
	72	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS $TLIBS \
	73	-lsodium -luv -lpthread
	74
43	75	echo "Building gates..."
44	76	g++ -std=c++17 -O2 $INC -o test/test_uds_backpressure \
45	77	test/test_uds_backpressure.cpp "$OBJ/uds_server.o"

swarmd/src/daemon.cpp +123 -0

…	…	@@ -0,0 +1,123 @@
	1	#include "daemon.h"
	2
	3	#include <cstdio>
	4	#include <cstdlib>
	5	#include <cstring>
	6
	7	namespace swarmd {
	8
	9	bool Daemon::start(const char socket_path, const char bootstrap) {
	10	_socket_path = socket_path;
	11	if (!_uds.listen(socket_path)) return false;
	12
	13	ControlBackend backend;
	14	backend.status = [this](swarmd_status_t *out) { _fill_status(out); };
	15	backend.swarm_join = [this](const uint8_t topic[32], bool server, bool client) {
	16	_t.swarm_join(topic, server, client);
	17	return true;
	18	};
	19	backend.swarm_leave = [this](const uint8_t topic[32]) {
	20	_t.swarm_leave(topic);
	21	return true;
	22	};
	23	_control.attach(&_uds, backend);
	24
	25	// Peer lifecycle arrives on the UDX thread. Enqueue only.
	26	_t.set_mux_callbacks(
	27	[this](int32_t id, Protomux *) {
	28	Event e;
	29	e.kind = Event::PEER_UP;
	30	e.peer_id = id;
	31	memset(e.pubkey, 0, 32);
	32	_t.get_peer_public_key(id, e.pubkey);
	33	_push(e);
	34	},
	35	[this](int32_t id) {
	36	Event e;
	37	e.kind = Event::PEER_DOWN;
	38	e.peer_id = id;
	39	memset(e.pubkey, 0, 32);
	40	_push(e);
	41	});
	42
	43	Ed25519KeyPair kp = NoiseIK::generate_keypair();
	44	if (bootstrap) {
	45	const char *bs[1] = { bootstrap };
	46	_t.start_dht(kp, bs, 1);
	47	} else {
	48	// ⛔ NO BOOTSTRAP MEANS THE PUBLIC DHT, not an isolated one. Passing an
	49	// empty list used to leave the transport "active" and permanently
	50	// unreachable with no error anywhere (hyper-games #68 bug A).
	51	_t.start_dht(kp, nullptr, 0);
	52	}
	53
	54	_started = true;
	55	return true;
	56	}
	57
	58	void Daemon::stop() {
	59	if (!_started) return;
	60	_started = false;
	61	_t.set_mux_callbacks(nullptr, nullptr);
	62	_t.stop();
	63	_uds.close_all();
	64	}
	65
	66	void Daemon::_push(const Event &e) {
	67	std::lock_guard<std::mutex> lk(_m);
	68	if (_events.size() >= MAX_EVENTS) {
	69	_dropped_events++;
	70	return;
	71	}
	72	_events.push_back(e);
	73	}
	74
	75	void Daemon::_drain() {
	76	std::deque<Event> batch;
	77	{
	78	std::lock_guard<std::mutex> lk(_m);
	79	batch.swap(_events);
	80	}
	81	for (const Event &e : batch) {
	82	uint8_t *out = nullptr;
	83	size_t out_len = 0;
	84	if (e.kind == Event::PEER_UP) {
	85	swarmd_peer_up_t up;
	86	memset(&up, 0, sizeof(up));
	87	up.peer = (uint64_t)(uint32_t)e.peer_id;
	88	memcpy(up.pubkey, e.pubkey, 32);
	89	if (swarmd_encode_on_peer_up(&up, &out, &out_len) < 0) continue;
	90	} else {
	91	swarmd_peer_down_t down;
	92	memset(&down, 0, sizeof(down));
	93	down.peer = (uint64_t)(uint32_t)e.peer_id;
	94	if (swarmd_encode_on_peer_down(&down, &out, &out_len) < 0) continue;
	95	}
	96	if (out == nullptr) continue;
	97	_uds.broadcast(out, out_len);
	98	free(out); // caller frees
	99	}
	100	}
	101
	102	void Daemon::_fill_status(swarmd_status_t *out) {
	103	memset(out, 0, sizeof(*out));
	104	out->dht_state = _t.is_dht_bootstrapped() ? 2 : (_t.is_dht_active() ? 1 : 0);
	105	out->bootstrapped = _t.is_dht_bootstrapped();
	106	const SwarmStatus st = _t.get_swarm_status();
	107	out->peers_known = (uint64_t)st.peers_known;
	108	out->peers_connected = (uint64_t)st.peers_connected;
	109	out->connecting = (uint64_t)st.connecting;
	110	out->queued = (uint64_t)st.queue_size;
	111	out->local_clients = (uint64_t)_uds.client_count();
	112	}
	113
	114	int Daemon::peers() const {
	115	return const_cast<UdxTransport &>(_t).get_swarm_status().peers_connected;
	116	}
	117
	118	void Daemon::poll(int timeout_ms) {
	119	_drain();
	120	_uds.poll(timeout_ms);
	121	}
	122
	123	} // namespace swarmd

swarmd/src/daemon.h +72 -0

…	…	@@ -0,0 +1,72 @@
	1	#ifndef SWARMD_DAEMON_H
	2	#define SWARMD_DAEMON_H
	3
	4	// swarmd — the daemon proper: one UdxTransport, one local socket, one control
	5	// plane, and a queue between them.
	6	//
	7	// ⛔ THE UDX THREAD NEVER TOUCHES A CLIENT SOCKET. UdxTransport runs its own
	8	// libuv loop on its own thread; every callback it fires (peer up, peer down, a
	9	// frame off a channel) arrives THERE. Writing to a client from that thread is
	10	// what session-daemon does, and §4 of the design says why it must not: the fds
	11	// belong to the main thread, which owns the buffering. So callbacks enqueue,
	12	// and poll() drains on the main thread.
	13	//
	14	// The other half of the same rule: commands from a client (join a topic, open a
	15	// channel) must not touch udx/protomux handles from the main thread either.
	16	// UdxTransport::run_on_loop exists for that and is the only way in.
	17
	18	#include <cstdint>
	19	#include <deque>
	20	#include <mutex>
	21	#include <string>
	22	#include <vector>
	23
	24	#include "control.h"
	25	#include "uds_server.h"
	26	#include "../../src/udx_transport.h"
	27
	28	namespace swarmd {
	29
	30	class Daemon {
	31	public:
	32	bool start(const char socket_path, const char bootstrap);
	33	void stop();
	34
	35	// One pass: drain the cross-thread queue, then serve the local socket.
	36	void poll(int timeout_ms);
	37
	38	bool dht_active() const { return _t.is_dht_active(); }
	39	bool dht_bootstrapped() const { return _t.is_dht_bootstrapped(); }
	40	int local_clients() const { return _uds.client_count(); }
	41	int peers() const;
	42
	43	private:
	44	// Everything the UDX thread hands over crosses as one of these.
	45	struct Event {
	46	enum Kind { PEER_UP, PEER_DOWN } kind;
	47	int32_t peer_id;
	48	uint8_t pubkey[32];
	49	};
	50	void _push(const Event &e);
	51	void _drain();
	52
	53	void _fill_status(swarmd_status_t *out);
	54
	55	UdxTransport _t;
	56	UdsServer _uds;
	57	Control _control;
	58
	59	std::mutex _m;
	60	std::deque<Event> _events;
	61	// Bounded: a swarm that churns faster than the socket drains must drop
	62	// notifications rather than grow without limit.
	63	static constexpr size_t MAX_EVENTS = 4096;
	64	uint64_t _dropped_events = 0;
	65
	66	bool _started = false;
	67	std::string _socket_path;
	68	};
	69
	70	} // namespace swarmd
	71
	72	#endif

swarmd/src/main.cpp +70 -0

…	…	@@ -0,0 +1,70 @@
	1	// swarmd — a generic swarm daemon.
	2	//
	3	// swarmd --socket <path> [--bootstrap host:port]
	4	//
	5	// Omit --bootstrap to use the public DHT.
	6
	7	#include <csignal>
	8	#include <cstdio>
	9	#include <cstring>
	10
	11	#include <sodium.h>
	12
	13	#include "daemon.h"
	14
	15	static volatile sig_atomic_t g_stop = 0;
	16	static void on_signal(int) { g_stop = 1; }
	17
	18	int main(int argc, char **argv) {
	19	if (sodium_init() < 0) {
	20	fprintf(stderr, "sodium_init failed\n");
	21	return 1;
	22	}
	23
	24	const char *sock = nullptr;
	25	const char *bootstrap = nullptr;
	26	for (int i = 1; i < argc; i++) {
	27	if (!strcmp(argv[i], "--socket") && i + 1 < argc) sock = argv[++i];
	28	else if (!strcmp(argv[i], "--bootstrap") && i + 1 < argc) bootstrap = argv[++i];
	29	}
	30	if (!sock) {
	31	fprintf(stderr, "usage: swarmd --socket <path> [--bootstrap host:port]\n");
	32	return 1;
	33	}
	34
	35	signal(SIGINT, on_signal);
	36	signal(SIGTERM, on_signal);
	37
	38	swarmd::Daemon d;
	39	if (!d.start(sock, bootstrap)) {
	40	fprintf(stderr, "failed to start (socket %s)\n", sock);
	41	return 1;
	42	}
	43	printf("[swarmd] listening on %s\n", sock);
	44	fflush(stdout);
	45
	46	// ⛔ SAY WHETHER THE SWARM IS REACHABLE, not just whether we started.
	47	// "listening" is a LOCAL fact. Two peers once sat on one topic for ten
	48	// minutes both reporting success while neither had ever bootstrapped.
	49	int ticks = 0;
	50	int last_peers = -1;
	51	bool last_boot = false;
	52	while (!g_stop) {
	53	d.poll(200);
	54	const bool boot = d.dht_bootstrapped();
	55	const int peers = d.peers();
	56	if (boot != last_boot \|\| peers != last_peers \|\| (ticks % 25) == 0) {
	57	printf("[swarmd] dht=%s peers=%d local=%d\n",
	58	boot ? "up" : (d.dht_active() ? "connecting" : "starting"),
	59	peers, d.local_clients());
	60	fflush(stdout);
	61	last_boot = boot;
	62	last_peers = peers;
	63	}
	64	ticks++;
	65	}
	66
	67	printf("[swarmd] stopping\n");
	68	d.stop();
	69	return 0;
	70	}

hyperswarm-peer commits / 401d7fb

Changed files (6) +306 -2