diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 61 |
1 files changed, 53 insertions, 8 deletions
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: |