From 176b5ed563f7022009b70751e62b7ee7901e6da0 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 11 Oct 2026 19:09:04 +0200 Subject: feat(node): carry connections on several cores connection_processes (default 1) starts that many processes for the WebRTC stack; sessions and keys stay in the node. Set from node.toml or the Node page, applied to new connections. Co-Authored-By: Claude Opus 5.5 --- .../meshbay-node/src/meshbay_node/ops/settings.py | 36 +++++++++++++++++++--- 1 file changed, 31 insertions(+), 5 deletions(-) (limited to 'packages/meshbay-node/src/meshbay_node/ops/settings.py') 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} -- cgit v1.2.3