932528c · hyperswarm-peer

Changed files (4) +281 -4

swarmd/build.sh +6 -0

…	…	@@ -80,6 +80,12 @@ g++ -std=c++17 -O2 $INC -o test/test_control \
80	80	test/test_control.cpp "$OBJ/uds_server.o" "$OBJ/control.o" \
81	81	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS
82	82
	83	# Drives the REAL Daemon (transport and all), so it links what swarmd links.
	84	g++ -std=c++17 -O2 $INC -o test/test_send_truth \
	85	test/test_send_truth.cpp "$OBJ/daemon.o" "$OBJ/uds_server.o" "$OBJ/control.o" \
	86	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS $TLIBS \
	87	-lsodium -luv -lpthread
	88
83	89	echo "Building tools..."
84	90	g++ -std=c++17 -O2 $INC -o tools/swarmd-probe tools/swarmd_probe.cpp \
85	91	"$OBJ/swarmd_schema.o" "$OBJ/swarmd_hrpc.o" $CLIBS

swarmd/src/daemon.cpp +79 -4

…	…	@@ -24,23 +24,39 @@ bool Daemon::start(const char socket_path, const char bootstrap) {
24	24	const std::vector<uint8_t> &id,
25	25	uint32_t slots, uint64_t owner) -> uint64_t {
26	26	const uint64_t handle = _next_channel++;
	27
	28	// Registered HERE, not on the loop: a client that sends the instant
	29	// channel-open returns must get "open on nobody yet", not "no such
	30	// handle". Those are different answers and only one of them means retry.
	31	auto open_peers = std::make_shared<std::atomic<int>>(0);
	32	{
	33	std::lock_guard<std::mutex> lk(_chan_m);
	34	_chan_open[handle] = open_peers;
	35	}
	36
27	37	// ⛔ PROTOMUX BELONGS TO THE UDX THREAD. Creating a channel from here
28	38	// would touch its handles from the wrong thread; run_on_loop is the only
29	39	// way in, and it is why this returns the handle immediately rather than
30	40	// waiting for the channel to exist.
31		_t.run_on_loop([this, handle, protocol, id, slots, owner]() {
	41	_t.run_on_loop([this, handle, protocol, id, slots, owner, open_peers]() {
32	42	ChannelSpec &spec = _channels[handle];
33	43	spec.handle = handle;
34	44	spec.protocol = protocol;
35	45	spec.id = id;
36	46	spec.slots = slots ? slots : 1;
37	47	spec.owner = owner;
	48	spec.open_peers = open_peers;
38	49	for (auto &kv : _muxes) _open_on_peer(spec, kv.first, kv.second);
	50	_publish_open(handle);
39	51	});
40	52	return handle;
41	53	};
42	54
43	55	backend.channel_close = [this](uint64_t handle) {
	56	{
	57	std::lock_guard<std::mutex> lk(_chan_m);
	58	_chan_open.erase(handle);
	59	}
44	60	_t.run_on_loop([this, handle]() {
45	61	auto it = _channels.find(handle);
46	62	if (it == _channels.end()) return;
…	…	@@ -54,14 +70,45 @@ bool Daemon::start(const char socket_path, const char bootstrap) {
54	70
55	71	backend.send = [this](uint64_t handle, uint32_t slot, uint64_t target,
56	72	const uint8_t data, size_t len, bool writable) {
	73	// ⛔ DO NOT ANSWER ABOUT THE QUEUE. This used to return true and set
	74	// writable=true unconditionally, which meant "I posted a lambda" — while
	75	// ProtomuxChannel::send drops the frame outright when the channel is not
	76	// yet open, and returns false to nobody. A caller told "accepted" stops
	77	// looking. That is #72's wall in miniature: the failure path was silent.
	78	//
	79	// What is knowable here is whether the handle exists and whether the
	80	// channel is open on any peer at all. A TARGETED send can still miss
	81	// (the aggregate does not say which peer), and that case is reported by
	82	// the drop counter below rather than pretended away.
	83	int open = -1;
	84	{
	85	std::lock_guard<std::mutex> lk(_chan_m);
	86	auto it = _chan_open.find(handle);
	87	if (it != _chan_open.end()) open = it->second->load(std::memory_order_relaxed);
	88	}
	89	if (open <= 0) {
	90	*writable = false;
	91	return false;
	92	}
	93
57	94	std::vector<uint8_t> copy(data, data + len); // the frame buffer is reused
58		*writable = true; // Protomux answers on the loop; see the note below
	95	*writable = true;
59	96	_t.run_on_loop([this, handle, slot, target, copy]() {
60	97	auto it = _channels.find(handle);
61	98	if (it == _channels.end()) return;
62	99	for (auto &kv : it->second.per_peer) {
63	100	if (target != 0 && (uint64_t)(uint32_t)kv.first != target) continue;
64		if (kv.second) kv.second->send(slot, copy.data(), copy.size());
	101	if (kv.second == nullptr) continue;
	102	if (!kv.second->send(slot, copy.data(), copy.size())) {
	103	// The channel closed between the answer and the write. Say so
	104	// once per drop rather than let the frame disappear.
	105	_dropped_sends++;
	106	fprintf(stderr,
	107	"[swarmd] channel %llu: protomux refused a %zu-byte frame "
	108	"to peer %d (dropped=%llu)\n",
	109	(unsigned long long)handle, copy.size(), kv.first,
	110	(unsigned long long)_dropped_sends);
	111	}
65	112	}
66	113	});
67	114	return true;
…	…	@@ -89,7 +136,10 @@ bool Daemon::start(const char socket_path, const char bootstrap) {
89	136	// UDX THREAD. The mux is going away; drop our references to its
90	137	// channels rather than let them dangle.
91	138	_muxes.erase(id);
92		for (auto &kv : _channels) kv.second.per_peer.erase(id);
	139	for (auto &kv : _channels) {
	140	kv.second.per_peer.erase(id);
	141	_publish_open(kv.first);
	142	}
93	143
94	144	Event e;
95	145	e.kind = Event::PEER_DOWN;
…	…	@@ -130,6 +180,18 @@ void Daemon::_open_on_peer(ChannelSpec &spec, int32_t peer_id, Protomux *mux) {
130	180	if (ch == nullptr) return; // already open on this mux
131	181
132	182	const uint64_t handle = spec.handle;
	183
	184	// ⛔ `open()` IS NOT OPEN. ProtomuxChannel::opened turns true in
	185	// _fully_open(), i.e. only once the REMOTE has opened the same channel —
	186	// and until then ProtomuxChannel::send returns false and discards. So the
	187	// only trustworthy source for "can I write" is protomux's own callbacks.
	188	ch->on_open = [this, handle](const uint8_t *, size_t) { _publish_open(handle); };
	189	ch->on_close = [this, handle, peer_id](bool) {
	190	auto it = _channels.find(handle);
	191	if (it != _channels.end()) it->second.per_peer.erase(peer_id);
	192	_publish_open(handle);
	193	};
	194
133	195	for (uint32_t slot = 0; slot < spec.slots; slot++) {
134	196	ch->add_message([this, handle, slot, peer_id](const uint8_t *data, size_t len,
135	197	ProtomuxChannel *) {
…	…	@@ -146,6 +208,19 @@ void Daemon::_open_on_peer(ChannelSpec &spec, int32_t peer_id, Protomux *mux) {
146	208	}
147	209	spec.per_peer[peer_id] = ch;
148	210	ch->open();
	211	_publish_open(handle);
	212	}
	213
	214	void Daemon::_publish_open(uint64_t handle) {
	215	// UDX THREAD ONLY.
	216	auto it = _channels.find(handle);
	217	if (it == _channels.end()) return;
	218	if (!it->second.open_peers) return;
	219	int n = 0;
	220	for (auto &kv : it->second.per_peer) {
	221	if (kv.second && kv.second->opened && !kv.second->closed) n++;
	222	}
	223	it->second.open_peers->store(n, std::memory_order_relaxed);
149	224	}
150	225
151	226	void Daemon::_push(const Event &e) {

swarmd/src/daemon.h +21 -0

…	…	@@ -15,8 +15,11 @@
15	15	// channel) must not touch udx/protomux handles from the main thread either.
16	16	// UdxTransport::run_on_loop exists for that and is the only way in.
17	17
	18	#include <atomic>
18	19	#include <cstdint>
19	20	#include <deque>
	21	#include <map>
	22	#include <memory>
20	23	#include <mutex>
21	24	#include <string>
22	25	#include <vector>
…	…	@@ -60,10 +63,21 @@ private:
60	63	uint32_t slots = 1;
61	64	uint64_t owner = 0; // the client that opened it
62	65	std::map<int32_t, ProtomuxChannel *> per_peer;
	66
	67	// How many peers currently have this channel FULLY open — written on the
	68	// UDX thread, read by SEND on the main thread. Shared rather than owned
	69	// because the map entry it lives in is UDX-thread-only.
	70	std::shared_ptr<std::atomic<int>> open_peers;
63	71	};
64	72
65	73	// UDX-THREAD ONLY. Creates this spec's channel on one peer's mux.
66	74	void _open_on_peer(ChannelSpec &spec, int32_t peer_id, Protomux *mux);
	75
	76	// UDX-THREAD ONLY. Recomputes how many peers have `handle` open and
	77	// publishes it. Recomputed rather than incremented: a channel can go away
	78	// through open, close, or the whole mux vanishing, and three call sites
	79	// each adjusting a counter is three chances to drift.
	80	void _publish_open(uint64_t handle);
67	81	void _push(const Event &e);
68	82	void _drain();
69	83
…	…	@@ -79,6 +93,13 @@ private:
79	93	std::map<uint64_t, ChannelSpec> _channels;
80	94	uint64_t _next_channel = 1;
81	95
	96	// The only part of a channel the main thread may look at: does this handle
	97	// exist, and is it open on anyone? SEND needs a truthful answer to that
	98	// BEFORE it queues, because everything past the queue is out of earshot.
	99	std::mutex _chan_m;
	100	std::map<uint64_t, std::shared_ptr<std::atomic<int>>> _chan_open;
	101	uint64_t _dropped_sends = 0;
	102
82	103	std::mutex _m;
83	104	std::deque<Event> _events;
84	105	// Bounded: a swarm that churns faster than the socket drains must drop

swarmd/test/test_send_truth.cpp +175 -0

…	…	@@ -0,0 +1,175 @@
	1	// Gate: SEND must not claim a frame was accepted when there is nowhere to put it.
	2	//
	3	// The daemon's send backend crosses to the UDX thread with run_on_loop, so it
	4	// cannot know synchronously that protomux took the bytes. What it CAN know is
	5	// whether the channel is open on any peer at all — and if it is open on none,
	6	// `accepted` and `writable` are answers about a frame that will be dropped by
	7	// ProtomuxChannel::send (which returns false when !opened) with nobody told.
	8	//
	9	// ⛔ THIS IS THE SHAPE OF #72's WALL. A path that fails with a bare return and
	10	// reports success is not a smaller bug because the failure is rarer; it is a
	11	// bigger one, because the caller has been told to stop looking.
	12	//
	13	// No network: the daemon is pointed at a bootstrap that does not exist, so it
	14	// never has a peer and the answer is unambiguous.
	15
	16	#include <cstdio>
	17	#include <cstdlib>
	18	#include <cstring>
	19	#include <vector>
	20
	21	#include <sys/socket.h>
	22	#include <sys/un.h>
	23	#include <unistd.h>
	24
	25	#include "daemon.h"
	26
	27	extern "C" {
	28	#include <compact.h>
	29	#include <rpc.h>
	30	}
	31
	32	static int checks = 0, failures = 0;
	33	static void check(bool c, const char *what) {
	34	checks++;
	35	if (c) { printf(" PASS: %s\n", what); return; }
	36	failures++;
	37	printf(" FAIL: %s\n", what);
	38	}
	39
	40	static int dial(const char *path) {
	41	int fd = socket(AF_UNIX, SOCK_STREAM, 0);
	42	struct sockaddr_un a;
	43	memset(&a, 0, sizeof(a));
	44	a.sun_family = AF_UNIX;
	45	strncpy(a.sun_path, path, sizeof(a.sun_path) - 1);
	46	if (connect(fd, (struct sockaddr *)&a, sizeof(a)) != 0) { close(fd); return -1; }
	47	return fd;
	48	}
	49
	50	static void write_frame(int fd, uint8_t *d, size_t n) {
	51	uint8_t hdr[4] = { (uint8_t)(n & 0xff), (uint8_t)((n >> 8) & 0xff),
	52	(uint8_t)((n >> 16) & 0xff), (uint8_t)((n >> 24) & 0xff) };
	53	(void)!write(fd, hdr, 4);
	54	(void)!write(fd, d, n);
	55	free(d); // the caller frees what _encode_ allocated
	56	}
	57
	58	// Pump the daemon until a frame carrying a response to `want_id` arrives.
	59	static bool await_reply(swarmd::Daemon &d, int fd, uint64_t want_id, rpc_message_t *out,
	60	std::vector<uint8_t> &store) {
	61	for (int i = 0; i < 400; i++) {
	62	d.poll(5);
	63	uint8_t hdr[4];
	64	ssize_t r = recv(fd, hdr, 4, MSG_DONTWAIT \| MSG_PEEK);
	65	if (r != 4) continue;
	66	const uint32_t n = (uint32_t)hdr[0] \| ((uint32_t)hdr[1] << 8) \|
	67	((uint32_t)hdr[2] << 16) \| ((uint32_t)hdr[3] << 24);
	68	std::vector<uint8_t> buf(4 + n);
	69	size_t got = 0;
	70	while (got < buf.size()) {
	71	r = read(fd, buf.data() + got, buf.size() - got);
	72	if (r <= 0) break;
	73	got += (size_t)r;
	74	}
	75	if (got != buf.size()) continue;
	76	store.assign(buf.begin() + 4, buf.end());
	77	compact_state_t st = { 0, store.size(), store.data() };
	78	if (rpc_decode_message(&st, out) < 0) continue;
	79	if (out->type == rpc_response && out->id == want_id) return true;
	80	}
	81	return false;
	82	}
	83
	84	int main() {
	85	char path[128];
	86	snprintf(path, sizeof(path), "/tmp/swarmd-sendtruth-%d.sock", (int)getpid());
	87
	88	swarmd::Daemon d;
	89	// A bootstrap that answers nothing. The daemon runs; it never gets a peer.
	90	if (!d.start(path, "127.0.0.1:1")) { printf("start failed\n"); return 1; }
	91
	92	int fd = dial(path);
	93	check(fd >= 0, "client attached");
	94	d.poll(20);
	95
	96	uint8_t *buf = nullptr;
	97	size_t len = 0;
	98
	99	swarmd_hello_request_t hello;
	100	memset(&hello, 0, sizeof(hello));
	101	hello.name.data = (utf8_t *)"gate";
	102	hello.name.len = 4;
	103	hello.version = 1;
	104	if (swarmd_encode_hello(1, &hello, &buf, &len) >= 0) write_frame(fd, buf, len);
	105
	106	rpc_message_t msg;
	107	std::vector<uint8_t> store;
	108	check(await_reply(d, fd, 1, &msg, store), "HELLO was answered");
	109
	110	// ---- open a channel while no peer exists ----
	111	swarmd_channel_open_t co;
	112	memset(&co, 0, sizeof(co));
	113	co.protocol.data = (utf8_t *)"sendtruth/v1";
	114	co.protocol.len = 12;
	115	co.id = nullptr; co.id_len = 0;
	116	co.slots = 1;
	117	if (swarmd_encode_channel_open(2, &co, &buf, &len) >= 0) write_frame(fd, buf, len);
	118
	119	uint64_t handle = 0;
	120	if (await_reply(d, fd, 2, &msg, store)) {
	121	swarmd_channel_ref_t ref;
	122	memset(&ref, 0, sizeof(ref));
	123	compact_state_t in = { 0, msg.len, msg.data };
	124	if (swarmd_channel_ref_decode(&in, &ref) >= 0) handle = ref.channel;
	125	}
	126	check(handle != 0, "the channel was opened and has a handle");
	127
	128	// Let the run_on_loop that creates the spec actually run.
	129	for (int i = 0; i < 60; i++) d.poll(5);
	130
	131	// ---- send into it ----
	132	const char *payload = "into-the-void";
	133	swarmd_send_request_t sr;
	134	memset(&sr, 0, sizeof(sr));
	135	sr.channel = handle;
	136	sr.slot = 0;
	137	sr.target = 0;
	138	sr.bytes = (uint8_t *)payload;
	139	sr.bytes_len = strlen(payload);
	140	if (swarmd_encode_send(3, &sr, &buf, &len) >= 0) write_frame(fd, buf, len);
	141
	142	swarmd_send_result_t res;
	143	memset(&res, 0, sizeof(res));
	144	bool got_reply = false;
	145	if (await_reply(d, fd, 3, &msg, store)) {
	146	compact_state_t in = { 0, msg.len, msg.data };
	147	got_reply = swarmd_send_result_decode(&in, &res) >= 0;
	148	}
	149	check(got_reply, "SEND was answered");
	150
	151	check(!res.accepted,
	152	"a send with no peer channel open is NOT reported as accepted");
	153	check(!res.writable,
	154	"and the channel is NOT reported writable");
	155
	156	// ---- and an unknown handle is refused just as plainly ----
	157	memset(&sr, 0, sizeof(sr));
	158	sr.channel = handle + 999;
	159	sr.bytes = (uint8_t *)payload;
	160	sr.bytes_len = strlen(payload);
	161	if (swarmd_encode_send(4, &sr, &buf, &len) >= 0) write_frame(fd, buf, len);
	162
	163	memset(&res, 0, sizeof(res));
	164	res.accepted = true;
	165	if (await_reply(d, fd, 4, &msg, store)) {
	166	compact_state_t in = { 0, msg.len, msg.data };
	167	swarmd_send_result_decode(&in, &res);
	168	}
	169	check(!res.accepted, "a send on a handle that does not exist is not accepted");
	170
	171	close(fd);
	172	d.stop();
	173	printf("\n%d/%d passed\n", checks - failures, checks);
	174	return failures ? 1 : 0;
	175	}

hyperswarm-peer commits / 932528c

Changed files (4) +281 -4