diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/config.py | 43 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 1 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 132 |
3 files changed, 159 insertions, 17 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index e4444d3..3de2473 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -37,6 +37,13 @@ ui_port = 18000 # local admin UI (127.0.0.1 only) invite_ttl_hours = 168 # 7 days pair_ttl_hours = 24 +# How many people may watch a video at once. One ffmpeg runs per viewer for as +# long as they watch — it remuxes rather than re-encodes, so it costs little CPU +# and around 50 MB of memory, and spends most of the film idle. Past this, a +# viewer is told the server is busy. Raise it on a machine with memory to spare; +# lower it on a Pi. +max_concurrent_streams = 8 + # Browser and native clients reach this node over WebRTC DataChannel via hub # signaling — no inbound port to open. QUIC is the optional direct path. @@ -82,6 +89,11 @@ class NodeConfig: # the SSH session that printed it. invite_ttl_hours: int = 168 # 7 days pair_ttl_hours: int = 24 + # How many people may watch a video at the same time. One ffmpeg runs per + # viewer for as long as they watch, so this is the knob that decides when + # the node answers "server busy" — see MAX_CONCURRENT_TRANSCODES in + # transport/webrtc_server.py for what one costs. + max_concurrent_streams: int = 8 @dataclass @@ -121,6 +133,30 @@ class Config: return self.groups[0] if self.groups else GroupConfig() +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. + + Zero is the dangerous one: `asyncio.Semaphore(0)` is not "no limit", it is + a node where no video ever plays and nothing in the log says why. + """ + # bool before int: TOML `true` is a bool, and `int(True)` is 1 — a node + # where exactly one person may watch, arrived at by a typo and announced + # nowhere. + if isinstance(value, bool) or not isinstance(value, (int, str)): + log.warning("%s = %r is not a count — using %d", name, value, default) + return default + try: + n = int(value) + except (TypeError, ValueError): + log.warning("%s = %r is not a number — using %d", name, value, default) + return default + if n < 1: + log.warning("%s = %d would stop the feature entirely — using %d", + name, n, default) + return default + return n + + def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: """ Load config from TOML file. Supports both single [group] and @@ -144,6 +180,9 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: nd.get("invite_ttl_hours", cfg.node.invite_ttl_hours)) cfg.node.pair_ttl_hours = int( nd.get("pair_ttl_hours", cfg.node.pair_ttl_hours)) + cfg.node.max_concurrent_streams = _positive( + nd.get("max_concurrent_streams", cfg.node.max_concurrent_streams), + cfg.node.max_concurrent_streams, "max_concurrent_streams") # Multi-group: [[groups]] array if "groups" in raw: @@ -189,6 +228,10 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.hub.username = user if port := os.environ.get("MESHBAY_QUIC_PORT"): cfg.node.quic_port = int(port) + if streams := os.environ.get("MESHBAY_MAX_CONCURRENT_STREAMS"): + cfg.node.max_concurrent_streams = _positive( + streams, cfg.node.max_concurrent_streams, + "MESHBAY_MAX_CONCURRENT_STREAMS") return cfg diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 6fafc74..8b4a1d7 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -274,6 +274,7 @@ class NodeDaemon: index=first["index"], groups=groups_ctx, denylist=denylist, + max_concurrent_streams=self._config.node.max_concurrent_streams, ) # No global chat_store here: each group's store lives in # groups_ctx[gid]["chat_store"] and is resolved per session via 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 91d801c..5f5c9ff 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -90,7 +90,20 @@ MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024 # 4 GB per file PRE_HANDSHAKE_MAX_MSG = 64 * 1024 # ffmpeg is spawned per stream request; without a cap any member can fork-bomb # the node by requesting many streams at once (H6). -MAX_CONCURRENT_TRANSCODES = 2 +# +# Two was sized when a stream was a burst: the client took segments as fast as +# it could append them, so a slot was held for the minute it took to push the +# file and then came back. Now that the client only pulls ninety seconds ahead +# of the playhead, a slot is held for as long as the film runs — so two slots +# means two people can watch anything at all, and the third is refused for the +# next hour and a half. The work behind a slot has not changed and is small: +# ffmpeg runs `-c copy`, a remux with no encoding in it, and spends most of the +# film blocked on a pipe nobody is reading. +# +# This is the default, not the policy: the right number depends on the machine, +# so the operator sets `max_concurrent_streams` under [node] in node.toml. This +# value applies when they have said nothing. +MAX_CONCURRENT_TRANSCODES = 8 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited # until the native client removes remote keypair bundles entirely. MAX_PRE_PROOF_FETCHES = 4 @@ -326,6 +339,12 @@ class WebRTCPeerSession: self._stream_credit = 0 self._stream_credit_evt = asyncio.Event() self._stream_stopped = False + # When the peer last said anything about this stream. See + # _await_stream_credit: silence is what ends a stream, not stinginess. + self._stream_heard_at = 0.0 + # Diagnostics: how many `stream_more n=0` the peer sent. See + # _grant_stream_credit — it tells a paced client from an unpaced one. + self._stream_keepalives = 0 # The stream this session currently owns. One viewer plays one film at # a time, so a second request means the first is over — see # _replace_stream for why waiting for it to time out is not an option. @@ -434,6 +453,28 @@ class WebRTCPeerSession: self._spawn(self._replace_stream(msg)) elif mtype == MNP.STREAM_MORE: self._grant_stream_credit(msg) + elif mtype == "client_diag": + # Diagnostics only. The node acts on none of it — it writes it + # next to its own view of the same stream, which is the only + # place the two halves can be compared when the client is a + # phone with no console. + # Every field is peer-controlled, so each is stringified and + # cut short: this is a log line, not a channel for writing + # whatever one likes into the operator's file. + def _f(key: str, n: int = 24) -> str: + return str(msg.get(key))[:n].replace("\n", " ") + # Debug: one line every five seconds per viewer. Run the daemon + # with --log-level debug to see inside a player that is + # misbehaving — it is the only view of the browser there is + # when the browser is a phone. + log.debug( + "stream: client t=%ss ahead=%ss ready=%s paused=%s " + "stalled=%s q=%s inflight=%s appending=%s updating=%s " + "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)", + _f("t"), _f("ahead"), _f("ready"), _f("paused"), + _f("stalled"), _f("q"), _f("inflight"), _f("appending"), + _f("updating"), _f("quota"), _f("ms"), _f("err", 80), + _f("ranges", 120), self._stream_segments) elif mtype == MNP.STREAM_STOP: age = (time.monotonic() - self._stream_started_at if self._stream_started_at else -1) @@ -1325,10 +1366,7 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not on disk"}) return - sem = self._ctx.get("_transcode_sem") - if sem is None: - sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES) - self._ctx["_transcode_sem"] = sem + sem = self._transcode_semaphore() try: async with sem: @@ -1830,14 +1868,30 @@ class WebRTCPeerSession: }) def _grant_stream_credit(self, msg: dict) -> None: + """ + The client has room for more segments. + + `n` of zero is a keepalive, not a no-op: a viewer whose buffer is + already a minute and a half ahead of the playhead deliberately grants + nothing, and must still be able to say it is there. Without that, the + stall timeout below cannot tell a paused film from a closed tab. + """ log.debug("stream credit +%s (had %d, sent %d)", msg.get("n"), self._stream_credit, self._stream_segments) - """The client has room for more segments.""" try: n = int(msg.get("n", 1)) except (TypeError, ValueError): n = 1 + if n == 0: + # The fingerprint of a client that bounds its read-ahead. A client + # that never sends one is granting credit per append — which is + # what fills the browser's buffer ceiling and wedges the player. + self._stream_keepalives += 1 + if self._stream_keepalives == 1: + log.info("stream: peer is pacing itself (first keepalive at " + "%d segments)", self._stream_segments) self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT)) + self._stream_heard_at = time.monotonic() self._stream_credit_evt.set() def _stop_stream(self) -> None: @@ -1860,8 +1914,22 @@ class WebRTCPeerSession: fast as it is produced, and the browser holds a four gigabyte film in a JavaScript array while MediaSource consumes it a segment at a time. """ - waited = 0.0 + # Measured from the last thing the peer said, not from the start of the + # wait: a viewer that is buffered well ahead sends keepalives and grants + # nothing for minutes at a time, and that is a watched film, not a + # stalled one. + self._stream_heard_at = time.monotonic() + waiting_since = 0.0 while self._stream_credit <= 0: + if waiting_since == 0.0: + waiting_since = time.monotonic() + # Debug: a paced viewer runs out of credit between every + # window, so this is one line per eight segments — hundreds + # per film. It is worth having, but not by default. + log.debug("stream: out of credit at %d segments (%.0f MB) — " + "waiting for the peer", + self._stream_segments, + self._stream_segments * STREAM_SEGMENT_SIZE / 1048576) if self._stream_stopped: return False # Checked before the wait as well as after it: a peer that vanishes @@ -1877,16 +1945,23 @@ class WebRTCPeerSession: await asyncio.wait_for(self._stream_credit_evt.wait(), timeout=STREAM_CREDIT_POLL) except asyncio.TimeoutError: - waited += STREAM_CREDIT_POLL - if waited >= STREAM_CREDIT_TIMEOUT: - log.info("Stream stalled: no credit from peer=%s", - (self._user_id or "?")[:8]) + silent = time.monotonic() - self._stream_heard_at + if silent >= STREAM_CREDIT_TIMEOUT: + log.info("Stream stalled: nothing from peer=%s for %.0fs", + (self._user_id or "?")[:8], silent) return False continue if self._stream_stopped: return False if self._channel is None or self._channel.readyState != "open": return False + if waiting_since: + waited_for = time.monotonic() - waiting_since + # Only a wait long enough to be a symptom. Normal pacing puts a + # gap of a few seconds between windows; a minute means the viewer + # is buffered right up and playing, or has stopped watching. + level = log.info if waited_for >= 10 else log.debug + level("stream: credit arrived after %.1fs", waited_for) self._stream_credit -= 1 return True @@ -1924,15 +1999,26 @@ class WebRTCPeerSession: self._stream_task = asyncio.current_task() await self._stream_video(msg) - async def _stream_video(self, msg: dict) -> None: - """Stream a video file as fMP4 segments via MSE-compatible output.""" - # One ffmpeg per request with no cap lets any member exhaust the node's - # CPU and process table (H6). The semaphore lives on the transport context - # so it is shared across all peers, not per-session. + def _transcode_semaphore(self) -> asyncio.Semaphore: + """The node's stream budget, shared across every peer. + + One ffmpeg per request with no cap lets any member exhaust the node's + CPU and process table (H6). The semaphore lives on the transport + context rather than the session so that it counts the node's viewers + and not one browser's, and it is created once: rebuilding it per call + would hand every caller its own budget and cap nothing at all. + """ sem = self._ctx.get("_transcode_sem") if sem is None: - sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES) + n = self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES + sem = asyncio.Semaphore(n) self._ctx["_transcode_sem"] = sem + log.info("stream: %d concurrent viewers allowed", n) + return sem + + async def _stream_video(self, msg: dict) -> None: + """Stream a video file as fMP4 segments via MSE-compatible output.""" + sem = self._transcode_semaphore() if sem.locked() and sem._value <= 0: self._send({"type": "error", "detail": "Server busy, retry shortly"}) return @@ -2029,6 +2115,14 @@ class WebRTCPeerSession: }) index += 1 self._stream_segments = index + if index % 100 == 0: + # A stream that stops shows up here as a last line, and the + # numbers on it say which side stopped it. + log.info("stream: %d segments (%.0f MB), credit=%d, " + "keepalives=%d, %.0fs in", + index, index * STREAM_SEGMENT_SIZE / 1048576, + self._stream_credit, self._stream_keepalives, + time.monotonic() - self._stream_started_at) await asyncio.sleep(0) except Exception as e: log.error("Stream error: %s", e) @@ -2150,6 +2244,7 @@ class WebRTCTransport: groups: dict[str, dict] | None = None, denylist: Any | None = None, stun_servers: list[str] | None = None, + max_concurrent_streams: int | None = None, ): self._ctx: dict[str, Any] = { "sk_node": sk_node, @@ -2158,6 +2253,9 @@ class WebRTCTransport: "shared_root": shared_root, "index": index, "_peers": {}, + # None means "the operator said nothing" — the default applies. It + # is read once, when the first stream builds the semaphore. + "max_concurrent_streams": max_concurrent_streams, } if groups: self._ctx["groups"] = groups |