diff options
Diffstat (limited to 'packages')
23 files changed, 1109 insertions, 14 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js index 17b8c2b..d8381e1 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js @@ -985,6 +985,8 @@ export default { 'node.setting_max_uploads': 'Max. gleichzeitige Uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': 'Größter einzelner Upload', + 'node.setting_connection_processes': 'CPU-Kerne für Verbindungen', + 'node.setting_unit_of': 'von {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js index a4e07a1..07c885d 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js @@ -1135,6 +1135,8 @@ export default { 'node.setting_max_uploads': 'Max concurrent uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': 'Largest single upload', + 'node.setting_connection_processes': 'CPU cores for connections', + 'node.setting_unit_of': 'of {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js index e8ea3e7..a6fb30f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js @@ -979,6 +979,8 @@ export default { 'node.setting_max_uploads': 'Subidas simultáneas máximas', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': 'Subida individual máxima', + 'node.setting_connection_processes': 'Núcleos de CPU para las conexiones', + 'node.setting_unit_of': 'de {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js index 0d927ae..c38be75 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js @@ -982,6 +982,8 @@ export default { 'node.setting_max_uploads': 'Téléversements simultanés max', 'node.setting_transcode': 'Transcoder les vidéos incompatibles', 'node.setting_max_upload': 'Taille maximale d\'un envoi', + 'node.setting_connection_processes': 'Cœurs CPU pour les connexions', + 'node.setting_unit_of': 'sur {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'heures', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js index e2c6df3..05f903e 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js @@ -981,6 +981,8 @@ export default { 'node.setting_max_uploads': 'Caricamenti simultanei massimi', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': 'Caricamento singolo massimo', + 'node.setting_connection_processes': 'Core CPU per le connessioni', + 'node.setting_unit_of': 'su {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js index 9aa4833..bd63b4f 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js @@ -969,6 +969,8 @@ export default { 'node.setting_max_uploads': '同時アップロードの上限', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': '1ファイルあたりの上限', + 'node.setting_connection_processes': '接続に使う CPU コア数', + 'node.setting_unit_of': '/ {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js index 863f7c9..eb6c6c2 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js @@ -983,6 +983,8 @@ export default { 'node.setting_max_uploads': 'Max. gelijktijdige uploads', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': 'Grootste enkele upload', + 'node.setting_connection_processes': 'CPU-kernen voor verbindingen', + 'node.setting_unit_of': 'van {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js index 2dc2e74..1b97dae 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js @@ -1003,6 +1003,8 @@ export default { 'node.setting_max_uploads': 'Maks. równoczesnych wysyłek', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': 'Największa pojedyncza wysyłka', + 'node.setting_connection_processes': 'Rdzenie CPU dla połączeń', + 'node.setting_unit_of': 'z {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js index 6c4f03b..a680cc8 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js @@ -980,6 +980,8 @@ export default { 'node.setting_max_uploads': 'Máximo de envios simultâneos', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': 'Maior envio individual', + 'node.setting_connection_processes': 'Núcleos de CPU para as conexões', + 'node.setting_unit_of': 'de {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js index 8899ed0..ae16098 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js @@ -957,6 +957,8 @@ export default { 'node.setting_max_uploads': '最大同时上传数', 'node.setting_transcode': 'Transcode incompatible video', 'node.setting_max_upload': '单个上传的最大大小', + 'node.setting_connection_processes': '用于连接的 CPU 核心数', + 'node.setting_unit_of': '/ {n}', 'node.setting_unit_gb': 'GB', 'node.setting_unit_hours': 'hours', 'node.setting_unit_minutes': 'min', diff --git a/packages/meshbay-hub/src/meshbay_hub/static/node-page.js b/packages/meshbay-hub/src/meshbay_hub/static/node-page.js index 56166c6..88167d3 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/node-page.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/node-page.js @@ -1188,6 +1188,16 @@ export function NodePage({ groups, token, username }) { </div> </label> <label class="node-setting"> + <span class="node-setting-label">${t('node.setting_connection_processes')}</span> + <div class="node-setting-input"> + <input type="number" min="1" max=${editSettings.cpu_count || 1} + value=${editSettings.connection_processes || 1} + onInput=${e => setEditSettings(s => ({...s, connection_processes: + Math.min(Math.max(parseInt(e.target.value) || 1, 1), s.cpu_count || 1)}))} /> + <span class="node-setting-unit">${t('node.setting_unit_of', { n: editSettings.cpu_count || 1 })}</span> + </div> + </label> + <label class="node-setting"> <span class="node-setting-label">${t('node.setting_transcode')}</span> <div class="node-setting-input"> <input type="checkbox" checked=${editSettings.transcode_incompatible_video} diff --git a/packages/meshbay-hub/tests/test_node_page_connection_processes.py b/packages/meshbay-hub/tests/test_node_page_connection_processes.py new file mode 100644 index 0000000..0342e6c --- /dev/null +++ b/packages/meshbay-hub/tests/test_node_page_connection_processes.py @@ -0,0 +1,38 @@ +""" +The Node page offers `connection_processes`, bounded by the machine's cores. + +The node refuses a number above its core count, so a field that let one be +typed would be a field whose Save fails. The ceiling comes with the settings +(`cpu_count`, which the node adds and ignores when it is sent back). +""" + +import re +from pathlib import Path + +STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" + + +def _field() -> str: + src = (STATIC / "node-page.js").read_text(encoding="utf-8") + at = src.index("t('node.setting_connection_processes')") + return src[at:src.index("</label>", at)] + + +def test_the_field_cannot_hold_more_than_the_cores(): + field = _field() + assert "max=${editSettings.cpu_count || 1}" in field + assert re.search(r"Math\.min\(Math\.max\(parseInt\(e\.target\.value\) \|\| 1, 1\), " + r"s\.cpu_count \|\| 1\)", field), ( + "a typed value is not held to the ceiling, only the spinner is") + + +def test_it_says_how_many_cores_there_are(): + assert "t('node.setting_unit_of', { n: editSettings.cpu_count || 1 })" in _field() + + +def test_every_language_names_it(): + for locale in sorted((STATIC / "locales").glob("*.js")): + text = locale.read_text(encoding="utf-8") + assert "'node.setting_connection_processes':" in text, locale.name + m = re.search(r"'node\.setting_unit_of': '([^']*)'", text) + assert m and "{n}" in m.group(1), locale.name 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) + 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 + # 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: diff --git a/packages/meshbay-node/tests/test_connection_processes.py b/packages/meshbay-node/tests/test_connection_processes.py new file mode 100644 index 0000000..5eea9b9 --- /dev/null +++ b/packages/meshbay-node/tests/test_connection_processes.py @@ -0,0 +1,281 @@ +""" +Peer connections held by processes of their own (`connection_processes`). + +One process runs aiortc on one core and tops out at about 35 MB/s over all its +connections. Above 1, the node starts that many `transport/webrtc/worker.py` +processes for the stack below the data channel and keeps the sessions. + +Nothing here is stubbed below the pool: the processes are real, the connection +is a real aiortc one over loopback, and the peer on the other side is an +`RTCPeerConnection` in the test. What a stub would hide is exactly what can go +wrong, a session reading something off its connection that the stand-in does +not carry. +""" + +import asyncio +import struct +from pathlib import Path + +import msgpack +import pytest +from aiortc import RTCPeerConnection, RTCSessionDescription +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node import ops +from meshbay_node.config import load_config, max_processes, node_settings_defaults +from meshbay_node.roster import Roster +from meshbay_node.transport.webrtc.channel import _get_remote_ip, _pack +from meshbay_node.transport.webrtc.workers import WorkerPool +from meshbay_node.transport.webrtc_server import WebRTCTransport + + +async def _until(test, seconds: float = 15.0) -> None: + for _ in range(int(seconds / 0.02)): + if test(): + return + await asyncio.sleep(0.02) + raise AssertionError("did not happen in time") + + +class _Browser: + """The other side: a peer connection with one data channel, as a browser + opens it.""" + + def __init__(self) -> None: + self.pc = RTCPeerConnection() + self.channel = self.pc.createDataChannel("mnp") + self.got: list[bytes] = [] + self.channel.on("message", self.got.append) + + async def offer(self) -> str: + await self.pc.setLocalDescription(await self.pc.createOffer()) + return self.pc.localDescription.sdp + + async def accept(self, answer_sdp: str) -> None: + await self.pc.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer")) + await _until(lambda: self.channel.readyState == "open") + + +async def _connect(pool: WorkerPool): + browser = _Browser() + peer = await pool.open(await browser.offer()) + channels = [] + peer.on("datachannel", channels.append) + await browser.accept(await peer.negotiate()) + await _until(lambda: channels) + return browser, peer, channels[0] + + +@pytest.fixture +async def pool(): + p = WorkerPool(2) + await p.ensure() + yield p + await p.close() + + +@pytest.mark.asyncio +async def test_bytes_cross_in_both_directions(pool): + browser, peer, channel = await _connect(pool) + try: + heard = [] + channel.on("message", heard.append) + browser.channel.send(b"from the browser") + await _until(lambda: heard) + assert heard == [b"from the browser"] + + channel.send(b"from the node") + await _until(lambda: browser.got) + assert browser.got == [b"from the node"] + assert channel.label == "mnp" + finally: + await browser.pc.close() + + +@pytest.mark.asyncio +async def test_what_is_buffered_comes_back_to_zero(pool): + """The download path waits while `bufferedAmount` is above 2 MB (files.py). + A count that never came down would stop every transfer after its first two + chunks; one that never went up would hand the whole file to the socket.""" + browser, peer, channel = await _connect(pool) + try: + chunk = bytes(60_000) + for _ in range(40): + channel.send(chunk) + assert channel.bufferedAmount == 40 * len(chunk), ( + "nothing has been confirmed yet, so all of it is still buffered") + await _until(lambda: sum(map(len, browser.got)) == 40 * len(chunk)) + await _until(lambda: channel.bufferedAmount == 0) + finally: + await browser.pc.close() + + +@pytest.mark.asyncio +async def test_the_session_reads_what_it_needs_from_the_stand_in(pool): + browser, peer, channel = await _connect(pool) + try: + await _until(lambda: peer.connectionState == "connected") + await _until(lambda: _get_remote_ip(peer)) + assert peer.localDescription.sdp.count("a=fingerprint:") >= 1 + assert peer.remoteDescription.sdp == browser.pc.localDescription.sdp + # The SCTP summary of the transfer log line arrives once a second. + await _until(lambda: "cwnd=" in peer.sctp_line, seconds=5) + finally: + await browser.pc.close() + + +@pytest.mark.asyncio +async def test_closing_says_so_once_and_frees_the_process(pool): + browser, peer, channel = await _connect(pool) + states = [] + peer.on("connectionstatechange", lambda: states.append(peer.connectionState)) + assert sorted(pool.loads()) == [0, 1] + await peer.close() + assert states == ["closed"] + assert channel.readyState == "closed" + assert pool.loads() == [0, 0] + await asyncio.sleep(0.3) # the process confirms; that is not a second event + assert states == ["closed"] + await browser.pc.close() + + +@pytest.mark.asyncio +async def test_a_process_that_dies_takes_only_its_own_connections(pool): + first = await _connect(pool) + second = await _connect(pool) + doomed = first[1]._worker + assert second[1]._worker is not doomed, "connections go to the least loaded process" + states = [] + first[1].on("connectionstatechange", lambda: states.append(first[1].connectionState)) + doomed.process.kill() + await _until(lambda: states == ["closed"]) + assert second[1].connectionState != "closed" + + # And the node is back to its number of processes for the next connection. + third = await _connect(pool) + assert pool.processes() == 2 + for browser, _, _ in (first, second, third): + await browser.pc.close() + + +@pytest.mark.asyncio +async def test_fewer_processes_waits_for_the_connections_to_end(pool): + one = await _connect(pool) + two = await _connect(pool) + await pool.resize(1) + assert not pool.enabled + assert await pool.open("ignored") is None, ( + "at one process the node holds new connections itself") + await asyncio.sleep(0.2) + assert pool.processes() == 2, "an open connection is never cut by a setting" + assert one[1].connectionState != "closed" + await one[1].close() + await two[1].close() + await _until(lambda: pool.processes() == 0) + for browser, _, _ in (one, two): + await browser.pc.close() + + +@pytest.mark.asyncio +async def test_only_a_process_the_node_started_is_listened_to(pool): + port = pool._server.sockets[0].getsockname()[1] + reader, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write(bytes(32)) + assert await reader.read() == b"", "the socket is closed on a token nobody was given" + assert pool.processes() == 2 + writer.close() + + +@pytest.mark.asyncio +async def test_a_session_answers_through_a_connection_process(): + """The whole path: the transport gives the connection to a process, builds + the session on the stand-in, and the session's reply reaches the peer.""" + transport = WebRTCTransport( + Ed25519PrivateKey.generate(), b"", b"", None, None, + stun_servers=[], connection_processes=2) + browser = _Browser() + try: + answer, _ = await transport.handle_offer(await browser.offer(), "peer-1") + await browser.accept(answer) + assert sorted(transport._workers.loads()) == [0, 1] + browser.channel.send(_pack({"type": "ping", "token": 7})) + await _until(lambda: browser.got) + (length,) = struct.unpack(">I", browser.got[0][:4]) + reply = msgpack.unpackb(browser.got[0][4:4 + length], raw=False) + assert reply == {"type": "error", "detail": "Handshake required"} + + await transport.close_peer("peer-1") + assert transport._workers.loads() == [0, 0] + finally: + await browser.pc.close() + await transport.close_all() + assert transport._workers.processes() == 0 + + +@pytest.mark.asyncio +async def test_one_process_starts_none(): + transport = WebRTCTransport( + Ed25519PrivateKey.generate(), b"", b"", None, None, stun_servers=[]) + await transport._workers.ensure() + assert transport._workers.processes() == 0 + assert not transport._workers.enabled + await transport.close_all() + + +# ── the setting ───────────────────────────────────────────────────────────── + +def test_the_default_is_one_process(tmp_path): + assert node_settings_defaults()["connection_processes"] == 1 + conf = tmp_path / "node.toml" + conf.write_text('[hub]\nurl = "https://example.invalid"\nusername = "op"\n') + assert load_config(conf).node.connection_processes == 1 + + +def test_node_toml_cannot_ask_for_more_than_the_cores(tmp_path): + conf = tmp_path / "node.toml" + conf.write_text('[hub]\nurl = "https://example.invalid"\nusername = "op"\n\n' + '[node]\nconnection_processes = 4096\n') + assert load_config(conf).node.connection_processes == max_processes() + + +class _Webrtc: + def __init__(self) -> None: + self.asked: list[int] = [] + + async def set_connection_processes(self, n: int) -> int: + self.asked.append(n) + return n + + +@pytest.fixture +async def state(tmp_path: Path): + conf = tmp_path / "node.toml" + conf.write_text('[hub]\nurl = "https://example.invalid"\nusername = "op"\n\n' + '[node]\nui_port = 18000\n', encoding="utf-8") + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + yield {"config": load_config(conf), "config_path": str(conf), + "roster": roster, "webrtc": _Webrtc(), "node_user_id": "op"} + await roster.close() + + +@pytest.mark.asyncio +async def test_the_setting_is_applied_now_and_kept(state, monkeypatch): + monkeypatch.setattr("meshbay_node.ops.settings.max_processes", lambda: 4) + out = await ops.set_node_settings(state, {"connection_processes": 3}) + assert out == {"updated": {"connection_processes": 3}} + assert state["webrtc"].asked == [3], "written and acknowledged is not applied" + assert "connection_processes = 3" in Path(state["config_path"]).read_text() + settings = await ops.get_node_settings(state) + assert settings["connection_processes"] == 3 + assert settings["cpu_count"] == 4 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", [0, -1, 5, "many", True]) +async def test_a_number_the_machine_cannot_honour_is_refused(state, monkeypatch, value): + monkeypatch.setattr("meshbay_node.ops.settings.max_processes", lambda: 4) + with pytest.raises(ops.OpError): + await ops.set_node_settings(state, {"connection_processes": value}) + assert state["webrtc"].asked == [] + assert (await ops.get_node_settings(state))["connection_processes"] == 1 diff --git a/packages/meshbay-node/tests/test_transfer_settings.py b/packages/meshbay-node/tests/test_transfer_settings.py index a2c7fd9..1d0a341 100644 --- a/packages/meshbay-node/tests/test_transfer_settings.py +++ b/packages/meshbay-node/tests/test_transfer_settings.py @@ -142,7 +142,7 @@ async def test_the_node_wide_caps_round_trip_through_the_roster(roster): "device_request_ttl_minutes", "max_concurrent_streams", "transcode_incompatible_video", - "max_upload_gb")}, + "max_upload_gb", "connection_processes")}, "stun_servers": [], "ice_interfaces": [], } await roster.set_node_setting(roster.SETTING_MAX_DOWNLOADS, "3", "op") |