diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc/apps')
3 files changed, 75 insertions, 22 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py index 857db33..a2d9f47 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py @@ -120,7 +120,7 @@ class MusicMixin: blob = await _transcode_audio_to_aac(file_path) except Exception as e: log.warning("Audio transcode failed for %s: %s", entry.id[:12], e) - self._send({"type": "error", "detail": f"Transcode failed: {e}"}) + self._send({"type": "error", "detail": "This track could not be converted"}) return transcode_hash = blake3.blake3(blob).hexdigest() diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py index 4337e24..d6a5248 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py @@ -2,6 +2,7 @@ stream a session holds, and the ffmpeg pipeline behind it.""" import asyncio +import contextlib import logging import time @@ -34,6 +35,10 @@ log = logging.getLogger("meshbay_node.transport.webrtc_server") # so the operator sets `max_concurrent_streams` under [node] in node.toml. This # value applies when they have said nothing. MAX_CONCURRENT_TRANSCODES = 8 +# Subtitle extractions one account may run at once. The player asks for one +# track at a time; two covers a quick change of track. Each holds a transcode +# slot for up to fifteen minutes on a long film. +MAX_SUBTITLE_JOBS_PER_ACCOUNT = 2 STREAM_SEGMENT_SIZE = 256 * 1024 @@ -179,6 +184,36 @@ class StreamingMixin: self._stream_task = asyncio.current_task() await self._stream_video(msg) + @contextlib.contextmanager + def _account_share(self, kind: str, limit: int): + """ + Hold one of this account's `limit` places for `kind`, or yield False. + + Counted on the node, across every session of the account: a member's + devices and tabs share one allowance. The node's own account is not + counted — it is the operator's machine. + """ + user = getattr(self, "_user_id", "") or "" + if not user or user == self._ctx.get("node_user_id"): + yield True + return + held = self._ctx.setdefault(f"_{kind}_by_account", {}) + if held.get(user, 0) >= limit: + yield False + return + held[user] = held.get(user, 0) + 1 + try: + yield True + finally: + held[user] -= 1 + if held[user] <= 0: + held.pop(user, None) + + def _streams_per_account(self) -> int: + """Half the node's viewers, rounded up: three screens in one home fit, + and no member alone takes every slot the operator set.""" + return max(1, -(-self._stream_capacity() // 2)) + def _transcode_semaphore(self) -> asyncio.Semaphore: """The node's stream budget, shared across every peer. @@ -205,22 +240,28 @@ class StreamingMixin: ctx = self._ctx log.info("stream: waiting for a slot (%d of %d in use)", ctx.get("_streams_in_flight", 0), self._stream_capacity()) - async with sem: - # Counted here rather than read back out of the semaphore's private - # `_value`: `set_capacity` needs to know how many slots are held in - # order to resize without letting the pool overshoot, and a number - # this code maintains itself is one that survives the semaphore - # object being replaced underneath it. - ctx["_streams_in_flight"] = ctx.get("_streams_in_flight", 0) + 1 - log.info("stream: slot acquired (%d of %d in use)", - ctx["_streams_in_flight"], self._stream_capacity()) - try: - await self._stream_video_inner(msg) - finally: - ctx["_streams_in_flight"] = max( - 0, ctx.get("_streams_in_flight", 1) - 1) - log.info("stream: slot released (%d of %d in use)", + with self._account_share("streams", self._streams_per_account()) as ok: + if not ok: + self._send({"type": "error", + "detail": "Too many videos playing from this account, " + "stop one and retry"}) + return + async with sem: + # Counted here rather than read back out of the semaphore's private + # `_value`: `set_capacity` needs to know how many slots are held in + # order to resize without letting the pool overshoot, and a number + # this code maintains itself is one that survives the semaphore + # object being replaced underneath it. + ctx["_streams_in_flight"] = ctx.get("_streams_in_flight", 0) + 1 + log.info("stream: slot acquired (%d of %d in use)", ctx["_streams_in_flight"], self._stream_capacity()) + try: + await self._stream_video_inner(msg) + finally: + ctx["_streams_in_flight"] = max( + 0, ctx.get("_streams_in_flight", 1) - 1) + log.info("stream: slot released (%d of %d in use)", + ctx["_streams_in_flight"], self._stream_capacity()) def _stream_capacity(self) -> int: return self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES @@ -246,7 +287,10 @@ class StreamingMixin: try: probe = await _probe_video(str(file_path)) except Exception as e: - self._send({"type": "error", "detail": f"Probe failed: {e}"}) + # The cause to the operator's log; to the member, that it failed. + # ffmpeg's own words carry the operator's paths and versions. + log.warning("stream: probe failed for %s: %s", entry.id[:12], e) + self._send({"type": "error", "detail": "This video could not be read"}) return codec_str = probe.codec duration = probe.duration diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py index 70781eb..525f2a9 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py @@ -10,6 +10,7 @@ from meshbay_common.protocol import MNP from meshbay_node.media_probe import probe_video as _probe_video from meshbay_node.roots import off_disk +from meshbay_node.transport.webrtc.apps.streaming import MAX_SUBTITLE_JOBS_PER_ACCOUNT from meshbay_node.transport.webrtc.disk import _locate from meshbay_node.transport.webrtc.media_tools import ( _extract_subtitle_to_webvtt, @@ -135,10 +136,16 @@ class SubtitlesMixin: return budget = _subtitle_timeout_for(entry.size) - async with sem: - log.info("subtitle: extracting file=%s track=%d (slot taken, up to %.0fs)", - file_id[:12], ordinal, budget) - blob = await _extract_subtitle_to_webvtt(file_path, ordinal, budget) + with self._account_share("subtitles", MAX_SUBTITLE_JOBS_PER_ACCOUNT) as ok: + if not ok: + log.info("subtitle: refused, account at its extraction share") + self._send({"type": "error", + "detail": "Subtitles are already being prepared, retry shortly"}) + return + async with sem: + log.info("subtitle: extracting file=%s track=%d (slot taken, up to %.0fs)", + file_id[:12], ordinal, budget) + blob = await _extract_subtitle_to_webvtt(file_path, ordinal, budget) subtitle_hash = blake3.blake3(blob).hexdigest() await media_cache.put_thumb(subtitle_hash, synthetic_id, blob) @@ -158,8 +165,10 @@ class SubtitlesMixin: except BaseException as e: log.warning("subtitle: extract failed file=%s track=%d after %.1fs: %r", file_id[:12], ordinal, time.monotonic() - t0, e) + # The cause is in the log line above; ffmpeg's own words carry the + # operator's paths and versions. self._send({"type": "error", - "detail": f"Subtitle extraction failed: {e}"}) + "detail": "These subtitles could not be extracted"}) if isinstance(e, asyncio.CancelledError): raise finally: |