diff options
26 files changed, 1166 insertions, 15 deletions
@@ -737,6 +737,18 @@ do. Read them before writing anything that touches the same mechanism. with one connection and with eight, so only a browser on another machine measures it +- **Measure which machine is the limit before adding cores to one of them.** + "Two clients will share the node's one core" was the reasoning; measured, two + browsers downloading at once left the node at 60 % of a core and saturated + the four cores of the machine receiving. The node's ceiling is real (one + aiortc process, about 35 MB/s over all its connections) but it is the third + fast client that meets it, not the second. What moved to other processes is + only the stack below the data channel (`connection_processes`, + `transport/webrtc/workers.py`, `docs/MESHBAY_DESIGN.md` §5.5), because that + is where the CPU went: with four processes the node proper used 18 % of a + core at 131 MB/s. Splitting there also means no key leaves the node's own + process + ### More of the same, from four rounds of live testing - **A protocol harness cannot test the SPA.** Any second implementation of the diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md index 81e2e33..233bcc8 100644 --- a/docs/MESHBAY_DESIGN.md +++ b/docs/MESHBAY_DESIGN.md @@ -1432,6 +1432,39 @@ there is no new message and no new field, so a node that predates this treats a sibling's `tr` as one it does not know and serves it under the leaseless ceiling, which bounds how many files a connection reads at once and not how much of one. +**Several processes for the connections.** aiortc is Python: one process sends +about 35 MB/s summed over every connection it holds, on one core, and a loop that +busy stops answering the connections that are not transferring. Two browsers +downloading at once from a server 22 ms away received 12 to 13 MB/s each where +one alone received 21, with the node at 60 % of a core and the receiving machine +saturated; a third fast client is where the node itself runs out. With +`connection_processes` above 1 (§6.8) the node starts that many processes and +gives each new peer connection to the least loaded +(`transport/webrtc/workers.py`). A process runs ICE, DTLS and SCTP and nothing +else: no node key, no group key, no index, no session. What it exchanges with the +node is what the data channel carries, framed on a loopback socket, and the +session stands on a stand-in for the peer connection that offers the few things +a session reads from one, so no handler knows which it was given. The lanes of a +download land on different processes without the client doing anything. + +| Processes | One download, same machine | CPU of the node proper | +|---|---|---| +| 1 (the node holds them) | 40 MB/s | all of it | +| 2 | 90 MB/s | 11 % of a core | +| 4 | 131 MB/s | 18 % | +| 8 | 150 MB/s, the receiving browser being the limit | 21 % | + +Measured on loopback, a 1.5 GB file through the interface in a headless browser, +byte for byte. A process costs about 85 MB of memory. The default is 1, which +starts none and leaves the node exactly as it was; the operator raises it. A +change applies to new connections, and a connection already open stays where it +is until it closes, in a process asked to leave or in the node. A process that +dies takes its own connections with it, which their clients see as a dropped +connection and reopen, and is replaced at the next offer. The loopback socket is +TCP because that is the one transport the event loop offers identically on every +platform; a process proves it is the one the node started by returning 32 random +bytes handed to it on its standard input. + The lease module is free of asyncio and of the transport: it decides, and the caller does the I/O. A queue that reveals itself only through a DataChannel is a queue nobody can prove things about. @@ -1998,7 +2031,7 @@ a real capability reduction, not an oversight. ### 6.8 Node settings -Eight `[node] `settings affect what the node does rather than how it starts, and +Nine `[node] `settings affect what the node does rather than how it starts, and their value is invisible until something goes wrong — so they are surfaced on the Node page: @@ -2011,6 +2044,7 @@ Node page: | `max_concurrent_downloads` | 8 | node-wide download leases (§5.5) | | `max_concurrent_uploads` | 8 | node-wide upload leases. A separate pool from downloads, because the two cost different things and one queue for both makes each cap meaningless | | `max_upload_gb` | 8 | the largest single upload, §6.4. Fractions are allowed, and it is read per chunk so a change reaches an upload already running | +| `connection_processes` | 1 | how many processes carry the peer connections, §5.5. One is the node itself; above it, that many cores serve transfers. Bounded by the cores of the machine, which the Node page shows beside the field | | `transcode_incompatible_video` | true | whether browser-incompatible video is transcoded during streaming. Unlike remuxing this costs real CPU per viewer | `hardware_video_encode` (default true) governs whether that transcode may use the @@ -4392,6 +4426,9 @@ process runs it — `systemctl --user` on Linux, Task Scheduler on Windows. | **"Copy stream URL" after picking a receiver casts to that receiver** | The player keeps the last device chosen, so the copy-only path reconnects it instead of only starting the relay | | **A folder renamed on the node's disk empties until the next reconciliation** | Seen live (2026-10-10, ntfs3): renaming a 2,441-file folder inside a root made the watcher drop every entry under the old path and announce none under the new one, so the group read as empty, and the loop was busy long enough for the hub's keepalive to time out. A restart (full scan) brought them back. A directory moved within a root should rescan its new subtree at once, not wait for the periodic scan | | **A photo deleted from the group comes back after a reinstall** | The backup's memory of what it sent is the phone's ledger (§9.12); a reinstall or cleared data empties it, and the folder no longer holds the deleted photo, so it is sent again. Clean fix: the node keeps a tombstone per folder on `file_delete` (name, size, SHA-256 taken before deleting), the index carries them, and the backup records a match as sent. Not done: a protocol field and a node table for a rare case | +| **`connection_processes` has no CLI verb** | Set from the Node page or `node.toml`; `meshbay-node` has no command for it, unlike the transfer caps | +| **Connection processes are untested on Windows and in a frozen build** | The frozen node starts them as `meshbay-node webrtc-worker`; written, never run there | +| **A connection process does not slow a peer the node cannot keep up with** | What a process receives is forwarded to the node unbounded. An upload is bounded by its own rate, 5 to 6 MB/s a connection, so it would take a node stalled for seconds to matter | | **The home page says "No groups yet" when the hub cannot be reached** | An unreachable hub reads as an account with no groups, rather than as an error | | **`meshbay-node init` says "Settings → Link Node"** | The control is on the Profile page, as QUICKSTART says | | **C4** for accounts with browser access | Closed against operators by the pepper; **open against an active hub**, which holds the pepper and can fetch a bundle with a token it mints — the adversary T3 already concedes for browsers (§3.7) | diff --git a/docs/USERGUIDE.md b/docs/USERGUIDE.md index bc16ba6..5649b30 100644 --- a/docs/USERGUIDE.md +++ b/docs/USERGUIDE.md @@ -745,9 +745,16 @@ and running `meshbay-node reload`: | `max_concurrent_downloads` | 8 | node-wide download slots | | `max_concurrent_uploads` | 8 | node-wide upload slots | | `max_upload_gb` | 8 | the largest single file a member may send you | +| `connection_processes` | 1 | how many CPU cores carry members' connections | | `transcode_incompatible_video` | true | re-encode video no browser can play | | `hardware_video_encode` | true | use the GPU for that, if one actually works | +`connection_processes` is the one for speed. With 1, everything a node sends +goes through one core, about 35 MB/s shared between everyone downloading at +that moment. Raise it when several people download at once and the machine has +cores to spare: each one adds a core's worth, and about 85 MB of memory. It +applies to connections opened after the change. + `max_concurrent_streams` is the one to think about: a stream costs a video process for as long as the film lasts, so this counts **viewers**, not requests. 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") |