From 176b5ed563f7022009b70751e62b7ee7901e6da0 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 11 Oct 2026 19:09:04 +0200 Subject: feat(node): carry connections on several cores connection_processes (default 1) starts that many processes for the WebRTC stack; sessions and keys stay in the node. Set from node.toml or the Node page, applied to new connections. Co-Authored-By: Claude Opus 5.5 --- packages/meshbay-node/src/meshbay_node/config.py | 20 ++ packages/meshbay-node/src/meshbay_node/daemon.py | 10 + .../meshbay-node/src/meshbay_node/ops/settings.py | 36 +- packages/meshbay-node/src/meshbay_node/roster.py | 2 + .../src/meshbay_node/transport/webrtc/channel.py | 3 + .../meshbay_node/transport/webrtc/sctp_stats.py | 3 + .../src/meshbay_node/transport/webrtc/worker.py | 245 +++++++++++++ .../src/meshbay_node/transport/webrtc/workers.py | 392 +++++++++++++++++++++ .../src/meshbay_node/transport/webrtc_server.py | 63 +++- 9 files changed, 760 insertions(+), 14 deletions(-) create mode 100644 packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py create mode 100644 packages/meshbay-node/src/meshbay_node/transport/webrtc/workers.py (limited to 'packages/meshbay-node/src/meshbay_node') diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index eb694ae..c5596f1 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -74,6 +74,14 @@ max_concurrent_uploads = 8 # the partial file is deleted. max_upload_gb = 8 +# How many processes carry the connections of members. At 1 the node does it +# all in one process, on one core: about 35 MB/s summed over every transfer in +# progress. Above 1 the node starts that many processes for the connections and +# keeps the rest for itself, so transfers use that many cores, plus part of one +# for the node. Takes effect for new connections, without a restart. Not above +# the number of cores of this machine. +connection_processes = 1 + # HEVC sources have no browser decoder on most platforms, so streaming one is # transcoded to H264 rather than the usual free copy — real CPU per viewer. # Set to false only if every viewer's client is known to decode HEVC itself. @@ -193,6 +201,9 @@ class NodeConfig: # concurrency. There is still no aggregate quota (MESHBAY_DESIGN.md # §15.3), so this is what stands between a writable root and a full disk. max_upload_gb: float = 8.0 + # How many processes hold the peer connections. One means this process + # holds them itself, as it always did. See transport/webrtc/workers.py. + connection_processes: int = 1 # HEVC (and any future codec in media_probe.py's # BROWSER_INCOMPATIBLE_VIDEO_CODECS) has no decoder in most browsers, so # streaming it needs a real re-encode to H264 rather than the usual free @@ -289,6 +300,11 @@ class Config: return self.groups[0] if self.groups else GroupConfig() +def max_processes() -> int: + """The most connection processes this machine can use: one per core.""" + return max(1, os.cpu_count() or 1) + + def _positive(value: object, default: int, name: str) -> int: """A count that must be at least one, or the default with a word about it. @@ -389,6 +405,7 @@ def node_settings_defaults(nd: NodeConfig | None = None) -> dict: "max_concurrent_downloads": nd.max_concurrent_downloads, "max_concurrent_uploads": nd.max_concurrent_uploads, "max_upload_gb": nd.max_upload_gb, + "connection_processes": nd.connection_processes, "transcode_incompatible_video": nd.transcode_incompatible_video, "stun_servers": nd.stun_servers or list(DEFAULT_STUN_SERVERS), "ice_interfaces": nd.ice_interfaces, @@ -438,6 +455,9 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.node.max_upload_gb = _positive_float( nd.get("max_upload_gb", cfg.node.max_upload_gb), cfg.node.max_upload_gb, "max_upload_gb") + cfg.node.connection_processes = min(max_processes(), _positive( + nd.get("connection_processes", cfg.node.connection_processes), + cfg.node.connection_processes, "connection_processes")) cfg.node.transcode_incompatible_video = bool( nd.get("transcode_incompatible_video", cfg.node.transcode_incompatible_video)) cfg.node.hardware_video_encode = bool( diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 6fba120..4d89f97 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -544,7 +544,12 @@ class NodeDaemon(EnrichmentMixin): max_upload_gb=self._config.node.max_upload_gb, transcode_incompatible_video=self._config.node.transcode_incompatible_video, stun_servers=self._config.node.stun_servers or None, + connection_processes=self._config.node.connection_processes, + ice_interfaces=self._config.node.ice_interfaces or None, ) + # Started now rather than by the first offer, which would wait + # for them: a process takes a second or two to come up. + await self._webrtc._workers.ensure() # No global chat_store here: each group's store lives in # groups_ctx[gid]["chat_store"] and is resolved per session via # _group_ctx(). Assigning the first group's store transport-wide @@ -1692,6 +1697,11 @@ def bind_control_port(port: int) -> socket.socket: def main() -> None: + # A connection process of a running node (transport/webrtc/workers.py). A + # frozen build has no `python -m` to start one with, so it starts itself. + if sys.argv[1:2] == ["webrtc-worker"]: + from meshbay_node.transport.webrtc.worker import main as worker_main + return worker_main() args = start() if run(args): return diff --git a/packages/meshbay-node/src/meshbay_node/ops/settings.py b/packages/meshbay-node/src/meshbay_node/ops/settings.py index 906cd03..9be9aca 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/settings.py +++ b/packages/meshbay-node/src/meshbay_node/ops/settings.py @@ -7,7 +7,7 @@ from pathlib import Path from meshbay_common.background import spawn -from meshbay_node.config import DEFAULT_CONFIG_PATH +from meshbay_node.config import DEFAULT_CONFIG_PATH, max_processes from meshbay_node.ops.core import OpError, _config, _group_ctx, _roster from meshbay_node.ops.node_toml import _update_node_toml from meshbay_node.roster import Roster @@ -60,6 +60,7 @@ NODE_SETTING_WRITERS: dict[str, tuple[str, str]] = { "max_concurrent_downloads": ("int", Roster.SETTING_MAX_DOWNLOADS), "max_concurrent_uploads": ("int", Roster.SETTING_MAX_UPLOADS), "max_upload_gb": ("size", Roster.SETTING_MAX_UPLOAD_GB), + "connection_processes": ("processes", Roster.SETTING_CONNECTION_PROCESSES), "transcode_incompatible_video": ("bool", Roster.SETTING_TRANSCODE), "stun_servers": ("stun_list", Roster.SETTING_STUN_SERVERS), "ice_interfaces": ("list", Roster.SETTING_ICE_INTERFACES), @@ -72,9 +73,10 @@ async def get_node_settings(state: dict) -> dict: roster = _roster(state) config = _config(state) defaults = node_settings_defaults(config.node) - if roster: - return await roster.node_settings(defaults) - return defaults + settings = await roster.node_settings(defaults) if roster else defaults + # Not a setting: the ceiling of `connection_processes` on this machine, so + # the Node page can say how far the operator may go. The writer ignores it. + return {**settings, "cpu_count": max_processes()} async def set_node_settings(state: dict, settings: dict) -> dict: """Update node-level daemon settings. Writes to both roster.db and node.toml.""" @@ -91,7 +93,21 @@ async def set_node_settings(state: dict, settings: dict) -> dict: if key not in allowed_keys: continue kind, setting_key = allowed_keys[key] - if kind == "int": + if kind == "processes": + # Bounded above as well as below: a process per core is all a + # machine can use, and past it they only take turns. + try: + v = int(value) + except (TypeError, ValueError): + raise OpError(f"{key} must be an integer") + if isinstance(value, bool) or not 1 <= v <= max_processes(): + raise OpError( + f"{key} must be between 1 and {max_processes()}, " + "the number of cores of this machine") + setattr(nd, key, v) + await roster.set_node_setting(setting_key, str(v), set_by) + updated[key] = v + elif kind == "int": try: v = int(value) except (TypeError, ValueError): @@ -160,15 +176,25 @@ async def set_node_settings(state: dict, settings: dict) -> dict: webrtc = state.get("webrtc") if webrtc is not None: webrtc.set_capacity(max_upload_gb=updated["max_upload_gb"]) + if "connection_processes" in updated: + webrtc = state.get("webrtc") + if webrtc is not None: + await webrtc.set_connection_processes( + updated["connection_processes"]) if "stun_servers" in updated: webrtc = state.get("webrtc") if webrtc and hasattr(webrtc, '_stun'): webrtc._stun = updated["stun_servers"] + if webrtc is not None: + webrtc.configure_ice(stun_servers=updated["stun_servers"]) from meshbay_node.transport.stun_multi import set_servers as _set_stun _set_stun(updated["stun_servers"]) if "ice_interfaces" in updated: from meshbay_node.transport.ice_filter import install as install_ice_filter install_ice_filter(updated["ice_interfaces"] or None) + webrtc = state.get("webrtc") + if webrtc is not None: + webrtc.configure_ice(ice_interfaces=updated["ice_interfaces"]) log.info("Node settings updated: %s", updated) return {"updated": updated} diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index 07b9ea6..a5d1967 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -1045,6 +1045,7 @@ class Roster: SETTING_MAX_DOWNLOADS = "max_concurrent_downloads" SETTING_MAX_UPLOADS = "max_concurrent_uploads" SETTING_MAX_UPLOAD_GB = "max_upload_gb" + SETTING_CONNECTION_PROCESSES = "connection_processes" SETTING_TRANSCODE = "transcode_incompatible_video" SETTING_STUN_SERVERS = "stun_servers" SETTING_ICE_INTERFACES = "ice_interfaces" @@ -1068,6 +1069,7 @@ class Roster: ("max_concurrent_streams", SETTING_MAX_STREAMS, "int"), ("max_concurrent_downloads", SETTING_MAX_DOWNLOADS, "int"), ("max_concurrent_uploads", SETTING_MAX_UPLOADS, "int"), + ("connection_processes", SETTING_CONNECTION_PROCESSES, "int"), ("transcode_incompatible_video", SETTING_TRANSCODE, "bool"), ("max_upload_gb", SETTING_MAX_UPLOAD_GB, "float"), ) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py index 31cc83c..4bb4e08 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py @@ -82,6 +82,9 @@ class _DataChannelBuffer: def _get_remote_ip(pc: RTCPeerConnection) -> str: """Best-effort extraction of the remote peer IP from the ICE transport.""" + reported = getattr(pc, "remote_ip", None) + if reported is not None: # held by a connection process (workers.py) + return reported try: dtls = pc.sctp and pc.sctp.transport ice = dtls and dtls.transport diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py index 338ded9..cc2eb1e 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/sctp_stats.py @@ -33,6 +33,9 @@ RTCSctpTransport._t3_expired = _counting_t3_expired def describe(pc) -> str: """The association of `pc` as a log suffix, or nothing when it has none.""" + reported = getattr(pc, "sctp_line", None) + if reported is not None: # held by a connection process (workers.py) + return reported sctp = getattr(pc, "sctp", None) if sctp is None: return "" diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py new file mode 100644 index 0000000..072c849 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py @@ -0,0 +1,245 @@ +"""A connection process: peer connections held for the node, and nothing else. + +The WebRTC stack (ICE, DTLS, SCTP) is Python and saturates one core at about +35 MB/s summed over every connection (docs/MESHBAY_DESIGN.md §5.5). This process +runs that stack and only that. It holds no node key, no group key and no index: +what crosses it is what crosses the wire, bytes of the data channel in both +directions, and the node proper decides what they mean. `workers.py` is the +other end and documents the frames. + +Started by the node with one JSON line on stdin: the loopback port to call +back, the token that proves it is the process the node started, and the ICE +settings. It ends when that socket does, so it cannot outlive its node. +""" + +import asyncio +import json +import logging +import struct +import sys + +HEADER = struct.Struct(">IBI") # length of what follows, frame type, connection +MAX_FRAME = 64 * 1024 * 1024 + +# node -> process +OFFER, SEND, CLOSE, CONFIG, QUIT = 1, 2, 3, 4, 5 +# process -> node +ANSWER, CHANNEL, MESSAGE, STATE, FLOW, STATS, LOG = 16, 17, 18, 19, 20, 21, 22 + +# How often the node learns what a channel has put on the wire. The download +# path waits for room in steps of 50 ms (files.py), so anything well under that +# is as good as reading `bufferedAmount` in place. +FLOW_INTERVAL = 0.02 +STATS_EVERY = 50 # flow ticks between two SCTP summaries + + +def pack(kind: int, conn: int, payload: bytes = b"") -> bytes: + return HEADER.pack(len(payload) + 5, kind, conn) + payload + + +async def read_frame(reader: asyncio.StreamReader) -> tuple[int, int, bytes]: + length, kind, conn = HEADER.unpack(await reader.readexactly(HEADER.size)) + if not 5 <= length <= MAX_FRAME: + raise ValueError(f"frame of {length} bytes") + return kind, conn, await reader.readexactly(length - 5) + + +class _Forward(logging.Handler): + """Hands every record to the node, which owns the log file.""" + + def __init__(self, write) -> None: + super().__init__() + self._write = write + + def emit(self, record: logging.LogRecord) -> None: + try: + body = {"name": record.name, "level": record.levelno, + "text": record.getMessage()} + self._write(pack(LOG, 0, json.dumps(body).encode())) + except Exception: + pass + + +class _Conn: + def __init__(self, pc) -> None: + self.pc = pc + self.channel = None + self.taken = 0 # bytes of SEND frames handed to the channel + self.reported = -1 + self.stats = "" + + +class Worker: + def __init__(self, reader, writer) -> None: + self._reader = reader + self._writer = writer + self._conns: dict[int, _Conn] = {} + self._stun: list[str] = [] + self._tasks: set[asyncio.Task] = set() + + def _put(self, kind: int, conn: int, payload: bytes = b"") -> None: + if not self._writer.is_closing(): + self._writer.write(pack(kind, conn, payload)) + + def _put_json(self, kind: int, conn: int, body: dict) -> None: + self._put(kind, conn, json.dumps(body).encode()) + + def _spawn(self, coro) -> None: + task = asyncio.ensure_future(coro) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + def configure(self, cfg: dict) -> None: + from meshbay_node.transport import ice_filter, stun_multi + if "ice_interfaces" in cfg: + ice_filter.install(cfg["ice_interfaces"] or None) + if "stun" in cfg: + self._stun = list(cfg["stun"] or []) + stun_multi.install(self._stun or None) + + async def run(self) -> None: + flow = asyncio.ensure_future(self._report_flow()) + try: + while True: + kind, conn, payload = await read_frame(self._reader) + if kind == SEND: + c = self._conns.get(conn) + if c is not None and c.channel is not None \ + and c.channel.readyState == "open": + c.channel.send(payload) + if c is not None: + # Counted whether or not it went out: the node's view of + # what is buffered must come back to zero on a channel + # that has closed, or its sender waits for room forever. + c.taken += len(payload) + elif kind == OFFER: + self._spawn(self._offer(conn, json.loads(payload))) + elif kind == CLOSE: + self._spawn(self._close(conn)) + elif kind == CONFIG: + self.configure(json.loads(payload)) + elif kind == QUIT: + break + except (asyncio.IncompleteReadError, ConnectionError): + pass + finally: + flow.cancel() + for conn in list(self._conns): + await self._close(conn) + + async def _offer(self, conn: int, body: dict) -> None: + from aiortc import ( + RTCConfiguration, + RTCIceServer, + RTCPeerConnection, + RTCSessionDescription, + ) + + from meshbay_node.transport.webrtc.channel import _get_remote_ip + try: + config = RTCConfiguration( + iceServers=[RTCIceServer(urls=s) for s in self._stun]) + pc = RTCPeerConnection(configuration=config) + c = self._conns[conn] = _Conn(pc) + + def state() -> None: + self._put_json(STATE, conn, { + "connection": pc.connectionState, "ice": pc.iceConnectionState, + "ip": _get_remote_ip(pc), + "channel": c.channel.readyState if c.channel else ""}) + + @pc.on("datachannel") + def on_datachannel(channel) -> None: + c.channel = channel + self._put_json(CHANNEL, conn, {"label": channel.label}) + + @channel.on("message") + def on_message(message) -> None: + if isinstance(message, str): + message = message.encode() + self._put(MESSAGE, conn, message) + + channel.on("close", state) + + @pc.on("connectionstatechange") + async def on_state() -> None: + state() + if pc.connectionState in ("failed", "closed"): + if self._conns.pop(conn, None) is not None \ + and pc.connectionState == "failed": + await pc.close() + + pc.on("iceconnectionstatechange", state) + await pc.setRemoteDescription( + RTCSessionDescription(sdp=body["sdp"], type="offer")) + await pc.setLocalDescription(await pc.createAnswer()) + self._put_json(ANSWER, conn, {"sdp": pc.localDescription.sdp}) + except Exception as e: + gone = self._conns.pop(conn, None) + self._put_json(ANSWER, conn, {"error": str(e) or type(e).__name__}) + if gone is not None: + try: + await gone.pc.close() + except Exception: + pass + + async def _close(self, conn: int) -> None: + c = self._conns.pop(conn, None) + if c is None: + return + try: + await c.pc.close() + except Exception: + pass + self._put_json(STATE, conn, {"connection": "closed", "ice": "closed", + "ip": "", "channel": "closed"}) + + async def _report_flow(self) -> None: + from meshbay_node.transport.webrtc import sctp_stats + tick = 0 + while True: + await asyncio.sleep(FLOW_INTERVAL) + tick += 1 + for conn, c in list(self._conns.items()): + if c.channel is None: + continue + sent = c.taken - c.channel.bufferedAmount \ + if c.channel.readyState == "open" else c.taken + if sent != c.reported: + c.reported = sent + self._put(FLOW, conn, struct.pack(">Q", max(0, sent))) + if tick % STATS_EVERY == 0: + line = sctp_stats.describe(c.pc) + if line != c.stats: + c.stats = line + self._put(STATS, conn, line.encode()) + + +async def _main() -> None: + hello = json.loads(sys.stdin.readline()) + reader, writer = await asyncio.open_connection("127.0.0.1", hello["port"]) + writer.write(bytes.fromhex(hello["token"])) + + root = logging.getLogger() + root.handlers[:] = [_Forward(writer.write)] + root.setLevel(hello.get("level", logging.INFO)) + + # Patches aiortc on import, as the node does for the connections it holds + # itself. + from meshbay_node.transport.webrtc import sctp_fix, sctp_stats # noqa: F401 + + worker = Worker(reader, writer) + worker.configure(hello) + await worker.run() + writer.close() + + +def main() -> None: + try: + asyncio.run(_main()) + except KeyboardInterrupt: + pass + + +if __name__ == "__main__": + main() diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/workers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/workers.py new file mode 100644 index 0000000..9a00bda --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/workers.py @@ -0,0 +1,392 @@ +"""Connection processes, seen from the node: more than one core for WebRTC. + +aiortc is Python on one core. One process moves about 35 MB/s summed over all +its connections and, at that point, answers nothing else: the event loop that +carries the transfers is the one that keeps every other connection alive +(docs/MESHBAY_DESIGN.md §5.5). With `connection_processes` above 1 the node +starts that many `worker.py` processes and gives each new peer connection to the +least loaded. The session, with everything it knows, stays here. + +What a process holds is the stack below the data channel. `RemotePeerConnection` +and `RemoteChannel` stand where aiortc's objects stand in a session and offer +the part of them a session uses, so no handler can tell which it was given. + +Frames, both ways, on one loopback socket per process: + + 4 bytes length of what follows + 1 byte type + 4 bytes connection, a number this side draws + ... payload: JSON for control frames, raw bytes for SEND and MESSAGE + +The socket is TCP on 127.0.0.1 because that is the one transport asyncio offers +identically on every platform the node runs on. A process proves it is the one +this node started by sending back 32 random bytes it was given on its stdin. +""" + +import asyncio +import json +import logging +import secrets +import struct +import sys + +from pyee.asyncio import AsyncIOEventEmitter + +from meshbay_node.config import max_processes # noqa: F401 (re-exported) +from meshbay_node.transport.webrtc.worker import ( + ANSWER, + CHANNEL, + CLOSE, + CONFIG, + FLOW, + LOG, + MESSAGE, + OFFER, + QUIT, + SEND, + STATE, + STATS, + pack, + read_frame, +) + +log = logging.getLogger("meshbay_node.transport.webrtc_server") + +START_TIMEOUT = 30 # seconds for a process to import aiortc and call back +ANSWER_TIMEOUT = 30 # an ICE gather is bounded at 5 s; this is for a dead process + + +def _argv() -> list[str]: + if getattr(sys, "frozen", False): + return [sys.executable, "webrtc-worker"] + return [sys.executable, "-m", "meshbay_node.transport.webrtc.worker"] + + +class _Description: + def __init__(self, sdp: str, kind: str) -> None: + self.sdp = sdp + self.type = kind + + +class RemoteChannel(AsyncIOEventEmitter): + """The data channel of a connection another process holds.""" + + def __init__(self, peer: "RemotePeerConnection", label: str) -> None: + super().__init__() + self._peer = peer + self.label = label + self.readyState = "open" + self._sent = 0 + self._flushed = 0 + + @property + def bufferedAmount(self) -> int: + """Bytes sent here that the process has not yet put on the wire: what + is in the socket, plus what its SCTP stack still queues.""" + return max(0, self._sent - self._flushed) + + def send(self, data: bytes) -> None: + if self.readyState != "open": + return + self._sent += len(data) + self._peer._worker.put(SEND, self._peer._id, data) + + +class RemotePeerConnection(AsyncIOEventEmitter): + """The part of an RTCPeerConnection a session uses, for a connection that + lives in a connection process.""" + + def __init__(self, worker: "_Worker", conn: int, offer_sdp: str) -> None: + super().__init__() + self._worker = worker + self._id = conn + self.connectionState = "new" + self.iceConnectionState = "new" + self.remoteDescription = _Description(offer_sdp, "offer") + self.localDescription: _Description | None = None + # Read by `_get_remote_ip` and `sctp_stats.describe` in place of the + # aiortc internals they reach into: as last reported by the process. + self.remote_ip = "" + self.sctp_line = "" + self.channel: RemoteChannel | None = None + self._answer: asyncio.Future = asyncio.get_running_loop().create_future() + + async def negotiate(self) -> str: + """Send the offer to the process; the answer SDP, candidates included.""" + self._worker.put(OFFER, self._id, json.dumps( + {"sdp": self.remoteDescription.sdp}).encode()) + try: + body = await asyncio.wait_for(self._answer, ANSWER_TIMEOUT) + except TimeoutError: + self._gone() + raise RuntimeError("Connection process did not answer") from None + if "error" in body: + self._gone() + raise RuntimeError(body["error"]) + self.localDescription = _Description(body["sdp"], "answer") + return body["sdp"] + + async def close(self) -> None: + if self.connectionState == "closed": + return + self._worker.put(CLOSE, self._id) + self._gone() + + def _gone(self) -> None: + """This connection is over, whoever ended it. Said once.""" + self._worker.forget(self._id) + if self.channel is not None: + self.channel.readyState = "closed" + if not self._answer.done(): + self._answer.set_result({"error": "Connection process stopped"}) + if self.connectionState != "closed": + self.connectionState = self.iceConnectionState = "closed" + self.emit("connectionstatechange") + + def _frame(self, kind: int, payload: bytes) -> None: + if kind == MESSAGE: + if self.channel is not None: + self.channel.emit("message", payload) + elif kind == FLOW: + if self.channel is not None: + self.channel._flushed = struct.unpack(">Q", payload)[0] + elif kind == STATS: + self.sctp_line = payload.decode() + elif kind == CHANNEL: + self.channel = RemoteChannel(self, json.loads(payload)["label"]) + self.emit("datachannel", self.channel) + elif kind == ANSWER: + if not self._answer.done(): + self._answer.set_result(json.loads(payload)) + elif kind == STATE: + body = json.loads(payload) + self.remote_ip = body.get("ip") or self.remote_ip + if self.channel is not None and body.get("channel") == "closed": + self.channel.readyState = "closed" + if body["connection"] == "closed": + self._gone() + return + ice, self.iceConnectionState = self.iceConnectionState, body["ice"] + if ice != body["ice"]: + self.emit("iceconnectionstatechange") + was, self.connectionState = self.connectionState, body["connection"] + if was != body["connection"]: + if body["connection"] == "failed": + self._worker.forget(self._id) + self.emit("connectionstatechange") + + +class _Worker: + def __init__(self, number: int) -> None: + self.number = number + self.process: asyncio.subprocess.Process | None = None + self.writer: asyncio.StreamWriter | None = None + self.conns: dict[int, RemotePeerConnection] = {} + self.draining = False + self.alive = False + self.on_empty = None + self._task: asyncio.Task | None = None + + def put(self, kind: int, conn: int, payload: bytes = b"") -> None: + if self.alive and self.writer is not None and not self.writer.is_closing(): + self.writer.write(pack(kind, conn, payload)) + + def forget(self, conn: int) -> None: + if self.conns.pop(conn, None) is not None and not self.conns \ + and self.draining and self.on_empty is not None: + self.on_empty(self) + + async def pump(self, reader: asyncio.StreamReader) -> None: + try: + while True: + kind, conn, payload = await read_frame(reader) + if kind == LOG: + body = json.loads(payload) + logging.getLogger(body["name"]).log( + body["level"], "[conn %d] %s", self.number, body["text"]) + continue + peer = self.conns.get(conn) + if peer is not None: + try: + peer._frame(kind, payload) + except Exception: + log.exception("connection process %d: frame %d", self.number, kind) + except (asyncio.IncompleteReadError, ConnectionError, ValueError): + pass + finally: + self.stopped() + # Reaped here, whichever way it went: asked to leave, killed, or + # fallen over. A process nobody waits for stays a zombie. + try: + await asyncio.wait_for(self.process.wait(), 5) + except TimeoutError: + self.process.kill() + + def stopped(self) -> None: + """The process is gone: every connection it held is gone with it.""" + was, self.alive = self.alive, False + if was and not self.draining: + log.warning("connection process %d stopped with %d connections", + self.number, len(self.conns)) + for peer in list(self.conns.values()): + peer._gone() + self.conns.clear() + if self.writer is not None: + self.writer.close() + if self.process is not None and self.process.returncode is None: + try: + self.process.terminate() + except ProcessLookupError: + pass + + +class WorkerPool: + """The connection processes of one node.""" + + def __init__(self, count: int = 1, stun: list[str] | None = None, + ice_interfaces: list[str] | None = None) -> None: + self.count = max(1, int(count)) + self._stun = list(stun or []) + self._ice = list(ice_interfaces or []) + self._workers: list[_Worker] = [] + self._leaving: list[_Worker] = [] # asked to go, not yet reaped + self._server: asyncio.base_events.Server | None = None + self._expected: dict[bytes, asyncio.Future] = {} + self._next_conn = 0 + self._next_worker = 0 + self._lock = asyncio.Lock() + + @property + def enabled(self) -> bool: + return self.count > 1 + + def processes(self) -> int: + return sum(1 for w in self._workers if w.alive) + + async def _accept(self, reader, writer) -> None: + try: + token = await asyncio.wait_for(reader.readexactly(32), 10) + except (TimeoutError, asyncio.IncompleteReadError, ConnectionError): + writer.close() + return + waiting = self._expected.pop(token, None) + if waiting is None or waiting.done(): + writer.close() + return + waiting.set_result((reader, writer)) + + async def _start_one(self) -> _Worker: + if self._server is None: + self._server = await asyncio.start_server(self._accept, "127.0.0.1", 0) + port = self._server.sockets[0].getsockname()[1] + token = secrets.token_bytes(32) + waiting = asyncio.get_running_loop().create_future() + self._expected[token] = waiting + self._next_worker += 1 + worker = _Worker(self._next_worker) + worker.on_empty = self._retire + worker.process = await asyncio.create_subprocess_exec( + *_argv(), stdin=asyncio.subprocess.PIPE) + hello = {"port": port, "token": token.hex(), "stun": self._stun, + "ice_interfaces": self._ice, + "level": logging.getLogger().getEffectiveLevel()} + worker.process.stdin.write(json.dumps(hello).encode() + b"\n") + try: + reader, worker.writer = await asyncio.wait_for(waiting, START_TIMEOUT) + except TimeoutError: + self._expected.pop(token, None) + worker.stopped() + raise RuntimeError("Connection process did not start") from None + worker.alive = True + worker._task = asyncio.ensure_future(worker.pump(reader)) + log.info("connection process %d started (pid %d)", + worker.number, worker.process.pid) + return worker + + def _retire(self, worker: _Worker) -> None: + if worker in self._workers: + self._workers.remove(worker) + self._leaving.append(worker) + if worker.alive: + log.info("connection process %d retired", worker.number) + worker.put(QUIT, 0) + worker.draining = True + + async def ensure(self) -> None: + """Bring the number of processes taking connections to `count`.""" + async with self._lock: + self._workers = [w for w in self._workers if w.alive] + self._leaving = [w for w in self._leaving + if w._task is not None and not w._task.done()] + taking = [w for w in self._workers if not w.draining] + want = self.count if self.enabled else 0 + for worker in taking[want:]: + worker.draining = True + if not worker.conns: + self._retire(worker) + missing = want - len(taking) + if missing > 0: + started = await asyncio.gather( + *(self._start_one() for _ in range(missing)), + return_exceptions=True) + for worker in started: + if isinstance(worker, _Worker): + self._workers.append(worker) + else: + log.error("connection process failed to start: %s", worker) + + async def resize(self, count: int) -> None: + """Set the number of processes. Connections already open stay where + they are: a process asked to leave does so when its last one closes.""" + self.count = max(1, int(count)) + await self.ensure() + + def configure(self, *, stun: list[str] | None = None, + ice_interfaces: list[str] | None = None) -> None: + body: dict = {} + if stun is not None: + self._stun = body["stun"] = list(stun) + if ice_interfaces is not None: + self._ice = body["ice_interfaces"] = list(ice_interfaces) + for worker in self._workers: + worker.put(CONFIG, 0, json.dumps(body).encode()) + + async def open(self, offer_sdp: str) -> RemotePeerConnection | None: + """A connection in the least loaded process, not yet negotiated. + + None when no process can take it: the caller then holds the connection + itself, which is what a node with one process does anyway. + """ + taking = [w for w in self._workers if w.alive and not w.draining] + if len(taking) < self.count: + await self.ensure() + taking = [w for w in self._workers if w.alive and not w.draining] + if not taking: + return None + worker = min(taking, key=lambda w: len(w.conns)) + self._next_conn += 1 + peer = RemotePeerConnection(worker, self._next_conn, offer_sdp) + worker.conns[peer._id] = peer + return peer + + def loads(self) -> list[int]: + return [len(w.conns) for w in self._workers if w.alive] + + async def close(self) -> None: + workers = self._workers + self._leaving + self._workers, self._leaving = [], [] + for worker in workers: + worker.draining = True + worker.put(QUIT, 0) + for worker in workers: + if worker.process is not None: + try: + await asyncio.wait_for(worker.process.wait(), 5) + except TimeoutError: + worker.process.kill() + worker.stopped() + if worker._task is not None: + await asyncio.gather(worker._task, return_exceptions=True) + if self._server is not None: + self._server.close() + self._server = None diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 0ca4b3d..0baaf19 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -58,6 +58,7 @@ from meshbay_node.transport.webrtc.handshake import HandshakeMixin from meshbay_node.transport.webrtc.node_ops import NodeOpsMixin from meshbay_node.transport.webrtc.transfer_handlers import TransferMixin from meshbay_node.transport.webrtc.upload_handlers import UploadMixin +from meshbay_node.transport.webrtc.workers import WorkerPool, max_processes log = logging.getLogger(__name__) @@ -116,6 +117,8 @@ class WebRTCTransport: max_concurrent_uploads: int | None = None, max_upload_gb: float | None = None, transcode_incompatible_video: bool = True, + connection_processes: int = 1, + ice_interfaces: list[str] | None = None, ): self._ctx: dict[str, Any] = { "sk_node": sk_node, @@ -146,6 +149,36 @@ class WebRTCTransport: self._stun = stun_servers or list(DEFAULT_STUN_SERVERS) self._sessions: dict[str, WebRTCPeerSession] = {} self._reapers: set[asyncio.Task] = set() + # Above one, the peer connections live in that many processes of their + # own and the sessions here speak to them (transport/webrtc/workers.py). + # At one, this process holds them itself and none is started. + self._workers = WorkerPool( + min(max(1, int(connection_processes or 1)), max_processes()), + self._stun, ice_interfaces) + + async def set_connection_processes(self, count: int) -> int: + """How many processes carry peer connections, from now on. + + New connections follow the new number at once. Open ones are left where + they are, in this process or in one asked to leave, until they close. + """ + n = int(count) + if not 1 <= n <= max_processes(): + raise ValueError( + f"connection_processes must be between 1 and {max_processes()}") + before = self._workers.count + await self._workers.resize(n) + if n != before: + log.info("connections: %d -> %d process(es)", before, n) + return n + + def configure_ice(self, *, stun_servers: list[str] | None = None, + ice_interfaces: list[str] | None = None) -> None: + """Pass an ICE setting on to the connection processes, which gather + candidates for the connections they hold.""" + if stun_servers is not None: + self._stun = list(stun_servers) + self._workers.configure(stun=stun_servers, ice_interfaces=ice_interfaces) def set_capacity(self, *, max_concurrent_streams: int | None = None, max_concurrent_downloads: int | None = None, @@ -289,7 +322,8 @@ class WebRTCTransport: user_id[:8], MAX_PEER_SESSIONS_PER_ACCOUNT) raise RuntimeError("Account is at its peer-connection share") - pc = RTCPeerConnection(configuration=config) + remote = await self._workers.open(offer_sdp) if self._workers.enabled else None + pc = remote or RTCPeerConnection(configuration=config) session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id) session._offer_user = user_id self._sessions[peer_id] = session @@ -328,15 +362,25 @@ class WebRTCTransport: await gone.shutdown_tasks() offer = RTCSessionDescription(sdp=offer_sdp, type="offer") - await pc.setRemoteDescription(offer) - answer = await pc.createAnswer() gather_start = time.monotonic() - await pc.setLocalDescription(answer) - - # ICE gathering runs inside setLocalDescription (non-trickle). A slow or - # unreachable STUN server shows up here as seconds of wait and zero - # srflx lines — the symptom the multi-server fan-out exists to prevent. - answer_sdp = pc.localDescription.sdp + if remote is not None: + # The process holding this connection does what the other branch + # does here, and sends back the answer. + try: + answer_sdp = await remote.negotiate() + except Exception: + self._sessions.pop(peer_id, None) + raise + else: + await pc.setRemoteDescription(offer) + answer = await pc.createAnswer() + gather_start = time.monotonic() + await pc.setLocalDescription(answer) + + # ICE gathering runs inside setLocalDescription (non-trickle). A slow or + # unreachable STUN server shows up here as seconds of wait and zero + # srflx lines — the symptom the multi-server fan-out exists to prevent. + answer_sdp = pc.localDescription.sdp srflx = answer_sdp.count(" typ srflx") # The host addresses this node put in the answer. When a peer reports # "DataChannel closed" the first question is whether the node offered @@ -399,6 +443,7 @@ class WebRTCTransport: for session in list(self._sessions.values()): await session.close() self._sessions.clear() + await self._workers.close() @property def active_peers(self) -> int: -- cgit v1.2.3