From c68249c314fdf1aa3f2e7945b8fe137fae59f9ef Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Fri, 15 May 2026 16:45:09 -0500 Subject: [PATCH] fix: replace fork()-based subprocess with posix_spawn at all uWSGI spawn sites gevent registers a pthread_atfork handler that never yields. Any subprocess.Popen/run call in a uWSGI greenlet hangs indefinitely at fork() before the child process even starts. - dispatcharr/gevent_patch.py: new; monkey.patch_all() + psycogreen, loaded via uWSGI import= in all four ini files - live_proxy/input/manager.py: posix_spawn + _SpawnedProcess; removed dead forwarder code - live_proxy/input/http_streamer.py: O_NONBLOCK on relay pipe + EAGAIN retry (blocking write to a full pipe stalls the gevent hub) - live_proxy/utils.py: new posix_spawn_proc() helper with O_NONBLOCK stdin, shared by both output managers - live_proxy/output/fmp4/manager.py, output/profile/manager.py: posix_spawn_proc(); _write_all() treats EAGAIN (None) as cooperative select.select wait instead of fatal error - core/views.py (stream_view): posix_spawn; fixed pre-existing bug where return StreamingHttpResponse(...) was indented inside stream_generator, making the success path always return None - connect/handlers/script.py: _posix_run() with posix_spawn + cooperative select reads + non-blocking waitpid; fixes deadlock when a script integration fires on a uWSGI-context event (client_connect fires in the live proxy, not Celery) --- CHANGELOG.md | 6 + apps/connect/handlers/script.py | 95 ++++++- apps/proxy/config.py | 2 +- apps/proxy/live_proxy/input/http_streamer.py | 47 ++- apps/proxy/live_proxy/input/manager.py | 268 ++++++++++++------ apps/proxy/live_proxy/output/fmp4/manager.py | 60 ++-- .../live_proxy/output/profile/manager.py | 44 ++- apps/proxy/live_proxy/output/ts/generator.py | 40 +-- apps/proxy/live_proxy/utils.py | 168 ++++++++++- apps/proxy/live_proxy/views.py | 75 +---- core/views.py | 51 ++-- 11 files changed, 581 insertions(+), 275 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d257a27..60eceeae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -82,6 +82,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **~25-second transcode startup delay and worker freeze when starting ffmpeg under gevent+uWSGI.** Enabling gevent cooperative multitasking (see Changed above) exposed a deadlock: `fork()` hangs indefinitely in gevent's `_before_fork` pthread_atfork handler when called from any thread while gevent is running - including from real OS threads. `subprocess.Popen`, which all three ffmpeg spawn sites used, calls `fork()` internally, stalling the uWSGI worker for ~25 seconds or freezing it entirely and blocking all other clients on that worker. + - `input/manager.py`: replaced `subprocess.Popen` with `os.posix_spawn` + a minimal `_SpawnedProcess` wrapper. `os.posix_spawn` is POSIX-specified to skip pthread_atfork handlers entirely. + - `output/fmp4/manager.py`, `output/profile/manager.py`: replaced `subprocess.Popen` with a new shared `posix_spawn_proc()` helper in `live_proxy/utils.py`. The helper also sets `O_NONBLOCK` on the stdin pipe write-end: under gevent, `threading.Thread` is monkey-patched to greenlets, so a blocking write to a full 64 KB pipe would stall the entire hub. `_write_all()` in both output managers now treats a `None` return (EAGAIN on the non-blocking FD) as a cooperative wait via `select.select()` rather than a fatal error. + - `input/http_streamer.py`: set `O_NONBLOCK` on the HTTP-to-pipe relay write-end with an EAGAIN retry loop for the same reason. + - `core/views.py` (`stream_view`): replaced `subprocess.Popen` with `os.posix_spawn`; also fixed a pre-existing indentation bug where the `return StreamingHttpResponse(...)` was accidentally nested inside `stream_generator` (making every successful response return `None` and raise a Django error). Also corrected two `NameError` references to the undefined `stream_id` variable in log messages. + - `apps/connect/handlers/script.py` (`ScriptHandler`): replaced `subprocess.run` with a `_posix_run` helper that uses `os.posix_spawn` + cooperative `select.select` reads + non-blocking `waitpid` polling. Without this fix, any script-type connect integration configured for events fired from a uWSGI worker (e.g. `client_connect`) would deadlock the serving greenlet. Note: `cwd` is no longer set to the script's directory during execution (it inherits the worker's cwd); this was a minor convenience, not a documented guarantee. - **XC server sub-path URLs now work correctly.** When a provider serves its XC API from a sub-path (e.g. `http://server/Pluto/gb/player_api.php`), Dispatcharr was stripping the path entirely and hitting the root (`/player_api.php`) instead. `_normalize_url` now preserves sub-path components and only strips any trailing `.php` segment (covering `player_api.php`, `get.php`, `xmltv.php`, and any future endpoint without a maintained list). The same fix is applied to `get_transformed_credentials` in the M3U profile transformation path. (Fixes #1218) - **M3U filter delete confirmation showed wrong field name and had a typo.** The confirmation dialog for deleting an M3U filter read `filter.type` (always `undefined`) instead of `filter.filter_type`, leaving the "Type:" line blank, and displayed "Patter:" instead of "Pattern:". Both are corrected. — Thanks [@nick4810](https://github.com/nick4810) - **M3U form FileInput expanded the modal width on long filenames.** Uploading a local M3U file with a long name caused the `FileInput` to expand beyond the modal's layout bounds. The input now clips overflow with `textOverflow: ellipsis`. — Thanks [@nick4810](https://github.com/nick4810) diff --git a/apps/connect/handlers/script.py b/apps/connect/handlers/script.py index b6aef79c..52d861bc 100644 --- a/apps/connect/handlers/script.py +++ b/apps/connect/handlers/script.py @@ -1,11 +1,89 @@ # connect/handlers/script.py import os +import select as _select +import signal as _signal import stat -import subprocess +import time from django.conf import settings from .base import IntegrationHandler +def _waitpid_nonblocking(pid, timeout_s=2.0): + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + try: + wpid, status = os.waitpid(pid, os.WNOHANG) + except ChildProcessError: + return -1 + if wpid == pid: + return os.WEXITSTATUS(status) if os.WIFEXITED(status) else -os.WTERMSIG(status) + time.sleep(0.01) + return -1 + + +def _posix_run(path, env, timeout): + stdout_r, stdout_w = os.pipe() + stderr_r, stderr_w = os.pipe() + devnull_r = os.open(os.devnull, os.O_RDONLY) + file_actions = [ + (os.POSIX_SPAWN_DUP2, devnull_r, 0), + (os.POSIX_SPAWN_DUP2, stdout_w, 1), + (os.POSIX_SPAWN_DUP2, stderr_w, 2), + (os.POSIX_SPAWN_CLOSE, devnull_r), + (os.POSIX_SPAWN_CLOSE, stdout_w), + (os.POSIX_SPAWN_CLOSE, stderr_w), + (os.POSIX_SPAWN_CLOSE, stdout_r), + (os.POSIX_SPAWN_CLOSE, stderr_r), + ] + try: + pid = os.posix_spawn(path, [path], env, file_actions=file_actions) + except Exception: + for fd in (stdout_r, stdout_w, stderr_r, stderr_w, devnull_r): + try: + os.close(fd) + except OSError: + pass + raise + for fd in (devnull_r, stdout_w, stderr_w): + os.close(fd) + + out, err = [], [] + done = set() + deadline = time.monotonic() + timeout + timed_out = False + + try: + while len(done) < 2: + remaining = deadline - time.monotonic() + if remaining <= 0: + timed_out = True + try: + os.kill(pid, _signal.SIGKILL) + except ProcessLookupError: + pass + break + + fds = [fd for fd in (stdout_r, stderr_r) if fd not in done] + readable, _, _ = _select.select(fds, [], [], min(remaining, 0.5)) + for fd in readable: + data = os.read(fd, 8192) + if data: + (out if fd == stdout_r else err).append(data) + else: + done.add(fd) + finally: + for fd in (stdout_r, stderr_r): + try: + os.close(fd) + except OSError: + pass + + rc = _waitpid_nonblocking(pid) + if timed_out: + raise TimeoutError(f"Script exceeded {timeout}s") + return rc, b"".join(out).decode("utf-8", errors="replace"), b"".join(err).decode("utf-8", errors="replace") + + def _is_path_allowed(real_path: str) -> bool: # Ensure path is within one of the allowed directories for base in getattr(settings, "CONNECT_ALLOWED_SCRIPT_DIRS", []): @@ -56,26 +134,17 @@ class ScriptHandler(IntegrationHandler): timeout = getattr(settings, "CONNECT_SCRIPT_TIMEOUT", 10) max_out = getattr(settings, "CONNECT_SCRIPT_MAX_OUTPUT", 65536) - result = subprocess.run( - [real_path], - capture_output=True, - text=True, - env=env, - timeout=timeout, - cwd=os.path.dirname(real_path) or None, - ) + rc, stdout, stderr = _posix_run(real_path, env, timeout) # Truncate outputs to avoid excessive memory/logging - stdout = result.stdout or "" - stderr = result.stderr or "" if len(stdout) > max_out: stdout = stdout[:max_out] + "... [truncated]" if len(stderr) > max_out: stderr = stderr[:max_out] + "... [truncated]" return { - "exit_code": result.returncode, + "exit_code": rc, "stdout": stdout, "stderr": stderr, - "success": result.returncode == 0, + "success": rc == 0, } diff --git a/apps/proxy/config.py b/apps/proxy/config.py index 4de92fa0..999b7cfb 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -100,7 +100,7 @@ class TSConfig(BaseConfig): CLEANUP_CHECK_INTERVAL = 1 # How often to check for disconnected clients (seconds) CLIENT_HEARTBEAT_INTERVAL = 5 # How often to send client heartbeats (seconds) GHOST_CLIENT_MULTIPLIER = 10.0 # How many heartbeat intervals before client considered ghost (10 = 50s, must exceed STREAM_TIMEOUT + FAILOVER_GRACE_PERIOD = 40s) - CLIENT_WAIT_TIMEOUT = 30 # Seconds to wait for client to connect + CLIENT_WAIT_TIMEOUT = 60 # Seconds to wait for channel to become ready # Stream health and recovery settings MAX_HEALTH_RECOVERY_ATTEMPTS = 2 # Maximum times to attempt recovery for a single stream diff --git a/apps/proxy/live_proxy/input/http_streamer.py b/apps/proxy/live_proxy/input/http_streamer.py index 56035338..7efb5d33 100644 --- a/apps/proxy/live_proxy/input/http_streamer.py +++ b/apps/proxy/live_proxy/input/http_streamer.py @@ -28,10 +28,16 @@ class HTTPStreamReader: def start(self): """Start the HTTP stream reader thread""" - # Create a pipe (works on Windows and Unix) self.pipe_read, self.pipe_write = os.pipe() - # Start the reader thread + # Make the write end non-blocking so that os.write() raises BlockingIOError + # instead of stalling the OS thread when the pipe buffer is full. Without + # this, a full pipe blocks the entire gevent worker (all greenlets freeze) + # because gevent does not patch os.write() on pipes. + import fcntl + flags = fcntl.fcntl(self.pipe_write, fcntl.F_GETFL) + fcntl.fcntl(self.pipe_write, fcntl.F_SETFL, flags | os.O_NONBLOCK) + self.running = True self.thread = threading.Thread(target=self._read_stream, daemon=True) self.thread.start() @@ -71,6 +77,8 @@ class HTTPStreamReader: logger.info(f"HTTP reader connected successfully, streaming data...") + import select as _select + # Stream chunks to pipe chunk_count = 0 for chunk in self.response.iter_content(chunk_size=self.chunk_size): @@ -78,18 +86,33 @@ class HTTPStreamReader: break if chunk: - try: - # Write binary data to pipe - os.write(self.pipe_write, chunk) - chunk_count += 1 - - # Log progress periodically - if chunk_count % 1000 == 0: - logger.debug(f"HTTP reader streamed {chunk_count} chunks") - except OSError as e: - logger.error(f"Pipe write error: {e}") + # Write the chunk in a non-blocking loop. The pipe write end is + # set O_NONBLOCK in start(), so os.write() raises BlockingIOError + # instead of stalling the OS thread. We use select.select on the + # write fd (gevent-patched - yields to hub) to wait for space, + # then retry. Partial writes are handled by advancing the offset. + offset = 0 + write_error = False + while offset < len(chunk) and self.running: + try: + n = os.write(self.pipe_write, chunk[offset:]) + offset += n + except BlockingIOError: + _, writable, _ = _select.select([], [self.pipe_write], [], 1.0) + if not writable and not self.running: + write_error = True + break + except OSError as e: + logger.error(f"Pipe write error: {e}") + write_error = True + break + if write_error: break + chunk_count += 1 + if chunk_count % 1000 == 0: + logger.debug(f"HTTP reader streamed {chunk_count} chunks") + logger.info("HTTP stream ended") except requests.exceptions.RequestException as e: diff --git a/apps/proxy/live_proxy/input/manager.py b/apps/proxy/live_proxy/input/manager.py index 7d853742..581c6fb7 100644 --- a/apps/proxy/live_proxy/input/manager.py +++ b/apps/proxy/live_proxy/input/manager.py @@ -529,13 +529,111 @@ class StreamManager: logger.debug(f"Starting transcode process: {self.transcode_cmd} for channel: {self.channel_id}") - # Modified to capture stderr instead of discarding it - self.transcode_process = subprocess.Popen( - self.transcode_cmd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, # Capture stderr instead of discarding it - bufsize=188 * 64 # Buffer optimized for TS packets - ) + import os as _os + import shutil as _shutil + import signal as _signal + import time as _time + + relay_read, relay_write = _os.pipe() + self.socket = _os.fdopen(relay_read, 'rb', buffering=0) + stderr_read, stderr_write = _os.pipe() + _stderr_read_transferred = False + try: + _t0 = _time.monotonic() + + # os.posix_spawn does not call pthread_atfork handlers, making + # it safe to call directly from the hub's greenlet. All + # fork()-based approaches (subprocess.Popen, whether called + # from the greenlet or a threadpool thread) hang in gevent's + # _before_fork atfork handler indefinitely under gevent+uWSGI. + _executable = _shutil.which(self.transcode_cmd[0]) or self.transcode_cmd[0] + _pid = _os.posix_spawn( + _executable, + self.transcode_cmd, + _os.environ, + file_actions=[ + (_os.POSIX_SPAWN_OPEN, 0, '/dev/null', _os.O_RDONLY, 0), + (_os.POSIX_SPAWN_DUP2, relay_write, 1), + (_os.POSIX_SPAWN_DUP2, stderr_write, 2), + (_os.POSIX_SPAWN_CLOSE, relay_write), + (_os.POSIX_SPAWN_CLOSE, stderr_write), + ], + ) + logger.debug( + f"posix_spawn completed in {_time.monotonic() - _t0:.3f}s " + f"pid={_pid} for channel {self.channel_id}" + ) + + _stderr_file = _os.fdopen(stderr_read, 'rb', buffering=0) + _stderr_read_transferred = True + + class _SpawnedProcess: + """Minimal Popen-compatible wrapper for a posix_spawn'd process.""" + stdin = None + stdout = None + + def __init__(self): + self.pid = _pid + self.returncode = None + self.stderr = _stderr_file + + def _reap(self, status): + if _os.WIFEXITED(status): + self.returncode = _os.WEXITSTATUS(status) + elif _os.WIFSIGNALED(status): + self.returncode = -_os.WTERMSIG(status) + else: + self.returncode = -1 + + def poll(self): + if self.returncode is not None: + return self.returncode + try: + rpid, status = _os.waitpid(self.pid, _os.WNOHANG) + if rpid: + self._reap(status) + except ChildProcessError: + self.returncode = -1 + return self.returncode + + def wait(self, timeout=None): + if self.returncode is not None: + return self.returncode + import gevent as _gevent + deadline = _time.monotonic() + timeout if timeout is not None else None + while True: + try: + rpid, status = _os.waitpid(self.pid, _os.WNOHANG) + except ChildProcessError: + self.returncode = -1 + return self.returncode + if rpid: + self._reap(status) + return self.returncode + if deadline is not None and _time.monotonic() >= deadline: + raise subprocess.TimeoutExpired(self.pid, timeout) + _gevent.sleep(0.01) + + def kill(self): + try: + _os.kill(self.pid, _signal.SIGKILL) + except ProcessLookupError: + pass + + def terminate(self): + try: + _os.kill(self.pid, _signal.SIGTERM) + except ProcessLookupError: + pass + + self.transcode_process = _SpawnedProcess() + except Exception: + if not _stderr_read_transferred: + _os.close(stderr_read) + raise + finally: + _os.close(relay_write) + _os.close(stderr_write) # Start a thread to read stderr self._start_stderr_reader() @@ -543,7 +641,6 @@ class StreamManager: # Set flag that transcoding process is active self.transcode_process_active = True - self.socket = self.transcode_process.stdout # Read from std output self.connected = True # Set connection start time for stability tracking @@ -570,105 +667,73 @@ class StreamManager: def _read_stderr(self): """Read and log ffmpeg stderr output with real-time stats parsing""" + import os as _os + import select as _select + import gevent + try: - buffer = b"" - last_stats_line = b"" + stderr = self.transcode_process.stderr + if not stderr: + return + stderr_fd = stderr.fileno() + buf = b"" - # Read byte by byte for immediate detection - while self.transcode_process and self.transcode_process.stderr: + while self.running and self.transcode_process and self.transcode_process.stderr: try: - # Read one byte at a time for immediate processing - byte = self.transcode_process.stderr.read(1) - if not byte: - break - - buffer += byte - - # Check for frame= at the start of buffer (new stats line) - if buffer == b"frame=": - # We detected the start of a stats line, read until we get a complete line - # or hit a carriage return (which overwrites the previous stats) - while True: - next_byte = self.transcode_process.stderr.read(1) - if not next_byte: - break - - buffer += next_byte - - # Break on carriage return (stats overwrite) or newline - if next_byte in (b'\r', b'\n'): - break - - # Also break if we have enough data for a typical stats line - if len(buffer) > 200: # Typical stats line length - break - - # Process the stats line immediately - if buffer.strip(): - try: - stats_text = buffer.decode('utf-8', errors='ignore').strip() - if stats_text and "frame=" in stats_text: - self._parse_ffmpeg_stats(stats_text) - self._log_stderr_content(stats_text) - except Exception as e: - logger.debug(f"Error parsing immediate stats line: {e}") - - # Clear buffer after processing - buffer = b"" + ready, _, _ = _select.select([stderr_fd], [], [], 1.0) + if not ready: + if not self.running or not self.transcode_process: + break continue - # Handle regular line breaks for non-stats content - elif byte == b'\n': - if buffer.strip(): - line_text = buffer.decode('utf-8', errors='ignore').strip() - if line_text and not line_text.startswith("frame="): - self._log_stderr_content(line_text) - buffer = b"" + chunk = _os.read(stderr_fd, 4096) + if not chunk: + break - # Handle carriage returns (potential stats overwrite) - elif byte == b'\r': - # Check if this might be a stats line - if b"frame=" in buffer: - try: - stats_text = buffer.decode('utf-8', errors='ignore').strip() - if stats_text and "frame=" in stats_text: - self._parse_ffmpeg_stats(stats_text) - self._log_stderr_content(stats_text) - except Exception as e: - logger.debug(f"Error parsing stats on carriage return: {e}") - elif buffer.strip(): - # Regular content with carriage return - line_text = buffer.decode('utf-8', errors='ignore').strip() - if line_text: - self._log_stderr_content(line_text) - buffer = b"" + # Yield to the hub after each read so fetch_chunk and other + # greenlets can run. Without this, the byte-at-a-time loop + # monopolises the event loop during ffmpeg startup output, + # starving the data reader and preventing the buffer from filling. + gevent.sleep(0) - # Prevent buffer from growing too large for non-stats content - elif len(buffer) > 1024 and b"frame=" not in buffer: - # Process whatever we have if it's not a stats line - if buffer.strip(): - line_text = buffer.decode('utf-8', errors='ignore').strip() - if line_text: - self._log_stderr_content(line_text) - buffer = b"" + buf += chunk + + while True: + cr = buf.find(b'\r') + nl = buf.find(b'\n') + if cr == -1 and nl == -1: + if len(buf) > 1024 and b"frame=" not in buf: + line_text = buf.decode('utf-8', errors='ignore').strip() + if line_text: + self._log_stderr_content(line_text) + buf = b"" + break + if cr != -1 and (nl == -1 or cr < nl): + line, buf = buf[:cr], buf[cr + 1:] + else: + line, buf = buf[:nl], buf[nl + 1:] + line_text = line.decode('utf-8', errors='ignore').strip() + if not line_text: + continue + if "frame=" in line_text: + self._parse_ffmpeg_stats(line_text) + self._log_stderr_content(line_text) except Exception as e: - logger.error(f"Error reading stderr byte: {e}") + logger.error(f"Error reading stderr for channel {self.channel_id}: {e}") break - # Process any remaining buffer content - if buffer.strip(): + if buf.strip(): try: - remaining_text = buffer.decode('utf-8', errors='ignore').strip() + remaining_text = buf.decode('utf-8', errors='ignore').strip() if remaining_text: if "frame=" in remaining_text: self._parse_ffmpeg_stats(remaining_text) self._log_stderr_content(remaining_text) except Exception as e: - logger.debug(f"Error processing remaining buffer: {e}") + logger.debug(f"Error processing remaining stderr buffer: {e}") except Exception as e: - # Catch any other exceptions in the thread to prevent crashes try: logger.error(f"Error in stderr reader thread for channel {self.channel_id}: {e}") except: @@ -1469,16 +1534,33 @@ class StreamManager: chunk = self.socket.recv(Config.CHUNK_SIZE) self.socket.settimeout(original_timeout) # Restore original timeout else: - # SocketIO object (transcode process stdout) - use select for timeout - import select - ready, _, _ = select.select([self.socket], [], [], chunk_timeout) + # Non-socket file object (io.FileIO from os.fdopen) - use raw + # fd + os.read to stay cooperative under gevent. + import select as _select + import os as _os + + try: + fd = self.socket.fileno() + except (ValueError, OSError): + self.connected = False + return False + + try: + ready, _, _ = _select.select([fd], [], [], chunk_timeout) + except (ValueError, OSError): + self.connected = False + return False if not ready: - # Timeout occurred logger.debug(f"Chunk read timeout ({chunk_timeout}s) for channel {self.channel_id}") return False - chunk = self.socket.read(Config.CHUNK_SIZE) + try: + chunk = _os.read(fd, Config.CHUNK_SIZE) + except OSError as e: + logger.warning(f"Read error for channel {self.channel_id}: {e}") + self.connected = False + return False except socket.timeout: # Socket timeout occurred diff --git a/apps/proxy/live_proxy/output/fmp4/manager.py b/apps/proxy/live_proxy/output/fmp4/manager.py index f8a6b646..e72bad3c 100644 --- a/apps/proxy/live_proxy/output/fmp4/manager.py +++ b/apps/proxy/live_proxy/output/fmp4/manager.py @@ -9,7 +9,6 @@ One instance per channel per cluster - coordinated via Redis fmp4:owner lock. """ import select -import subprocess import threading import time import struct @@ -123,12 +122,8 @@ class FMP4RemuxManager: self.running = True self._set_state(FMP4_STATE_INITIALIZING) - self._process = subprocess.Popen( - FFMPEG_REMUX_CMD, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) + from ...utils import posix_spawn_proc + self._process = posix_spawn_proc(FFMPEG_REMUX_CMD) short_id = self.channel_id[:8] self._reader_thread = threading.Thread( @@ -195,9 +190,13 @@ class FMP4RemuxManager: if not self.running: return n = self._process.stdin.write(view[offset:]) - if n is None or n <= 0: + if n is None: + # Pipe full (EAGAIN on non-blocking FD); yield cooperatively + select.select([], [self._process.stdin], [], 1.0) + elif n <= 0: raise OSError("stdin write returned no bytes") - offset += n + else: + offset += n def _writer_loop(self): """Read TS chunks from Redis and write to FFmpeg stdin.""" @@ -347,17 +346,32 @@ class FMP4RemuxManager: def _stderr_loop(self): """Log FFmpeg stderr lines. Detect BSF codec mismatch and trigger a no-BSF retry.""" + import os as _os + import select as _select try: - for raw_line in self._process.stderr: - line = raw_line.decode(errors="replace").rstrip() - if line: - logger.warning(f"[fMP4Remux:{self.channel_id}] FFmpeg: {line}") - if "aac_adtstoasc" in line and "is not supported by the bitstream filter" in line: - threading.Thread( - target=self._handle_bsf_error, daemon=True, - name=f"fmp4-bsf-retry-{self.channel_id[:8]}" - ).start() - return + stderr_fd = self._process.stderr.fileno() + buf = b"" + while self.running: + ready, _, _ = _select.select([stderr_fd], [], [], 1.0) + if not ready: + if self._process.poll() is not None: + break + continue + chunk = _os.read(stderr_fd, 4096) + if not chunk: + break + buf += chunk + while b'\n' in buf: + line_bytes, buf = buf.split(b'\n', 1) + line = line_bytes.decode(errors="replace").rstrip() + if line: + logger.warning(f"[fMP4Remux:{self.channel_id}] FFmpeg: {line}") + if "aac_adtstoasc" in line and "is not supported by the bitstream filter" in line: + threading.Thread( + target=self._handle_bsf_error, daemon=True, + name=f"fmp4-bsf-retry-{self.channel_id[:8]}" + ).start() + return except Exception: pass @@ -380,12 +394,8 @@ class FMP4RemuxManager: pass self._set_state(FMP4_STATE_INITIALIZING) - self._process = subprocess.Popen( - FFMPEG_REMUX_CMD_NO_BSF, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) + from ...utils import posix_spawn_proc + self._process = posix_spawn_proc(FFMPEG_REMUX_CMD_NO_BSF) self.running = True if not self.running: diff --git a/apps/proxy/live_proxy/output/profile/manager.py b/apps/proxy/live_proxy/output/profile/manager.py index 26eca027..75431e94 100644 --- a/apps/proxy/live_proxy/output/profile/manager.py +++ b/apps/proxy/live_proxy/output/profile/manager.py @@ -13,7 +13,6 @@ create a read-only StreamBuffer pointing at the same Redis keys. """ import select -import subprocess import threading import time from core.utils import RedisClient @@ -82,12 +81,8 @@ class OutputProfileManager: self.output_buffer = self._make_buffer() try: - self._process = subprocess.Popen( - self.command, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) + from ...utils import posix_spawn_proc + self._process = posix_spawn_proc(self.command) except (FileNotFoundError, OSError) as e: logger.error( f"[Profile:{self.profile_id}:{self.channel_id[:8]}] " @@ -169,9 +164,13 @@ class OutputProfileManager: if not self.running: return n = self._process.stdin.write(view[offset:]) - if n is None or n <= 0: + if n is None: + # Pipe full (EAGAIN on non-blocking FD); yield cooperatively + select.select([], [self._process.stdin], [], 1.0) + elif n <= 0: raise OSError("stdin write returned no bytes") - offset += n + else: + offset += n def _writer_loop(self): """Read TS chunks from Redis and write to the transcode process stdin.""" @@ -266,13 +265,28 @@ class OutputProfileManager: def _stderr_loop(self): """Log process stderr at WARNING level.""" + import os as _os + import select as _select try: - for raw_line in self._process.stderr: - line = raw_line.decode(errors="replace").rstrip() - if line: - logger.warning( - f"[Profile:{self.profile_id}:{self.channel_id[:8]}] {line}" - ) + stderr_fd = self._process.stderr.fileno() + buf = b"" + while self.running: + ready, _, _ = _select.select([stderr_fd], [], [], 1.0) + if not ready: + if self._process.poll() is not None: + break + continue + chunk = _os.read(stderr_fd, 4096) + if not chunk: + break + buf += chunk + while b'\n' in buf: + line_bytes, buf = buf.split(b'\n', 1) + line = line_bytes.decode(errors="replace").rstrip() + if line: + logger.warning( + f"[Profile:{self.profile_id}:{self.channel_id[:8]}] {line}" + ) except Exception: pass diff --git a/apps/proxy/live_proxy/output/ts/generator.py b/apps/proxy/live_proxy/output/ts/generator.py index dcff594f..59ec8ab2 100644 --- a/apps/proxy/live_proxy/output/ts/generator.py +++ b/apps/proxy/live_proxy/output/ts/generator.py @@ -101,7 +101,7 @@ class StreamGenerator: # First handle initialization if needed if self.channel_initializing: - channel_ready = self._wait_for_initialization() + channel_ready = yield from self._wait_for_initialization() if not channel_ready: # If initialization failed or timed out, we've already sent error packets return @@ -147,9 +147,7 @@ class StreamGenerator: last_keepalive = 0 proxy_server = ProxyServer.get_instance() - # While init is happening, send keepalive packets while time.time() - initialization_start < max_init_wait: - # Check if initialization has completed if proxy_server.redis_client: metadata_key = RedisKeys.channel_metadata(self.channel_id) metadata = proxy_server.redis_client.hgetall(metadata_key) @@ -159,43 +157,29 @@ class StreamGenerator: if state in ['waiting_for_clients', 'active']: logger.info(f"[{self.client_id}] Channel {self.channel_id} now ready (state={state})") return True - elif state in ['error', 'stopped', 'stopping']: # Added 'stopping' to error states + elif state in ['error', 'stopped', 'stopping']: error_message = metadata.get('error_message', 'Unknown error') logger.error(f"[{self.client_id}] Channel {self.channel_id} in error state: {state}, message: {error_message}") - # Send error packet before giving up yield create_ts_packet('error', f"Error: {error_message}") return False - else: - # Improved logging to track initialization progress - init_time = "unknown" - if 'init_time' in metadata: - try: - init_time_float = float(metadata['init_time']) - init_duration = time.time() - init_time_float - init_time = f"{init_duration:.1f}s ago" - except: - pass - # Still initializing - send keepalive if needed - if time.time() - last_keepalive >= keepalive_interval: - status_msg = f"Initializing: {state} (started {init_time})" - keepalive_packet = create_ts_packet('keepalive', status_msg) - logger.debug(f"[{self.client_id}] Sending keepalive packet during initialization, state={state}") - yield keepalive_packet - self.bytes_sent += len(keepalive_packet) - last_keepalive = time.time() - - # Also check stopping key directly stop_key = RedisKeys.channel_stopping(self.channel_id) if proxy_server.redis_client.exists(stop_key): logger.error(f"[{self.client_id}] Channel {self.channel_id} stopping flag detected during initialization") yield create_ts_packet('error', "Error: Channel is stopping") return False - # Wait a bit before checking again - gevent.sleep(0.1) + # Send PAT+PMT+null so clients recognise a valid TS program and keep + # buffering instead of timing out from missing program info. + if time.time() - last_keepalive >= keepalive_interval: + logger.debug(f"[{self.client_id}] Sending keepalive during initialization") + keepalive_data = create_ts_packet('null') + yield keepalive_data + self.bytes_sent += len(keepalive_data) + last_keepalive = time.time() + else: + gevent.sleep(0.1) - # Timed out waiting logger.warning(f"[{self.client_id}] Timed out waiting for initialization") yield create_ts_packet('error', "Error: Initialization timeout") return False diff --git a/apps/proxy/live_proxy/utils.py b/apps/proxy/live_proxy/utils.py index e9a26093..5177f1ce 100644 --- a/apps/proxy/live_proxy/utils.py +++ b/apps/proxy/live_proxy/utils.py @@ -1,5 +1,6 @@ import logging import re +import struct from urllib.parse import urlparse import inspect @@ -57,6 +58,53 @@ def get_client_ip(request): ip = request.META.get('REMOTE_ADDR') return ip +def _mpeg_crc32(data): + crc = 0xFFFFFFFF + for b in data: + crc ^= b << 24 + for _ in range(8): + if crc & 0x80000000: + crc = ((crc << 1) ^ 0x04C11DB7) & 0xFFFFFFFF + else: + crc = (crc << 1) & 0xFFFFFFFF + return crc + + +def create_ts_pat_pmt_packets(): + """ + Return two valid TS packets: PAT (PID 0x0000) and PMT (PID 0x0100). + + Declares program 1 with an H.264 video track at PID 0x0101. + TS clients like VLC need PAT/PMT to recognise a stream as valid; without + them they time out waiting for program info even while receiving null packets. + Returns exactly 376 bytes (2 x 188-byte TS packets). + """ + # PAT section: program 1 mapped to PMT at PID 0x0100 + pat_body = bytes([ + 0x00, 0xB0, 0x0D, # table_id=PAT, section_length=13 + 0x00, 0x01, # transport_stream_id=1 + 0xC1, 0x00, 0x00, # version=0, current=1, section 0/0 + 0x00, 0x01, # program_number=1 + 0xE1, 0x00, # PMT PID=0x0100 (reserved 0b111 | PID) + ]) + pat_body += struct.pack('>I', _mpeg_crc32(pat_body)) + pat_packet = bytes([0x47, 0x40, 0x00, 0x10, 0x00]) + pat_body + bytes([0xFF] * (183 - len(pat_body))) + + # PMT section: program 1, H.264 video at PID 0x0101 + pmt_body = bytes([ + 0x02, 0xB0, 0x12, # table_id=PMT, section_length=18 + 0x00, 0x01, # program_number=1 + 0xC1, 0x00, 0x00, # version=0, current=1, section 0/0 + 0xE1, 0x01, # PCR_PID=0x0101 + 0xF0, 0x00, # program_info_length=0 + 0x1B, 0xE1, 0x01, 0xF0, 0x00, # stream_type=H.264, PID=0x0101 + ]) + pmt_body += struct.pack('>I', _mpeg_crc32(pmt_body)) + pmt_packet = bytes([0x47, 0x41, 0x00, 0x10, 0x00]) + pmt_body + bytes([0xFF] * (183 - len(pmt_body))) + + return pat_packet + pmt_packet + + def create_ts_packet(packet_type='null', message=None): """ Create a Transport Stream (TS) packet for various purposes. @@ -83,7 +131,7 @@ def create_ts_packet(packet_type='null', message=None): # Add message to payload if provided if message: - msg_bytes = message + msg_bytes = message.encode('utf-8', errors='replace') if isinstance(message, str) else message packet[4:4+min(len(msg_bytes), 180)] = msg_bytes[:180] return bytes(packet) @@ -114,3 +162,121 @@ def get_logger(component_name=None): logger_name = "live_proxy" return logging.getLogger(logger_name) + + +def posix_spawn_proc(cmd): + """ + Spawn a subprocess using os.posix_spawn() with stdin, stdout, and stderr piped. + + Returns an object compatible with subprocess.Popen(cmd, stdin=PIPE, + stdout=PIPE, stderr=PIPE). Under gevent+uWSGI, fork() hangs indefinitely + in gevent's _before_fork atfork handler regardless of whether it is called + from a hub greenlet or a threadpool thread. os.posix_spawn() is explicitly + defined by POSIX to not call pthread_atfork handlers. + """ + import os + import shutil + import signal + import subprocess + import time + + stdin_r, stdin_w = os.pipe() # child reads stdin (FD 0) from here + stdout_r, stdout_w = os.pipe() # child writes stdout (FD 1) here; parent reads + stderr_r, stderr_w = os.pipe() # child writes stderr (FD 2) here; parent reads + + stdin_w_ok = stdout_r_ok = stderr_r_ok = False + try: + executable = shutil.which(cmd[0]) or cmd[0] + child_pid = os.posix_spawn( + executable, cmd, os.environ, + file_actions=[ + (os.POSIX_SPAWN_DUP2, stdin_r, 0), + (os.POSIX_SPAWN_DUP2, stdout_w, 1), + (os.POSIX_SPAWN_DUP2, stderr_w, 2), + (os.POSIX_SPAWN_CLOSE, stdin_r), + (os.POSIX_SPAWN_CLOSE, stdout_w), + (os.POSIX_SPAWN_CLOSE, stderr_w), + ], + ) + + import fcntl + fcntl.fcntl(stdin_w, fcntl.F_SETFL, fcntl.fcntl(stdin_w, fcntl.F_GETFL) | os.O_NONBLOCK) + stdin_file = os.fdopen(stdin_w, 'wb', buffering=0) + stdin_w_ok = True + stdout_file = os.fdopen(stdout_r, 'rb', buffering=0) + stdout_r_ok = True + stderr_file = os.fdopen(stderr_r, 'rb', buffering=0) + stderr_r_ok = True + + class _Proc: + stdin = stdin_file + stdout = stdout_file + stderr = stderr_file + + def __init__(self): + self.pid = child_pid + self.returncode = None + + def _reap(self, status): + if os.WIFEXITED(status): + self.returncode = os.WEXITSTATUS(status) + elif os.WIFSIGNALED(status): + self.returncode = -os.WTERMSIG(status) + else: + self.returncode = -1 + + def poll(self): + if self.returncode is not None: + return self.returncode + try: + rpid, status = os.waitpid(self.pid, os.WNOHANG) + if rpid: + self._reap(status) + except ChildProcessError: + self.returncode = -1 + return self.returncode + + def wait(self, timeout=None): + if self.returncode is not None: + return self.returncode + import gevent as _gevent + deadline = time.monotonic() + timeout if timeout is not None else None + while True: + try: + rpid, status = os.waitpid(self.pid, os.WNOHANG) + except ChildProcessError: + self.returncode = -1 + return self.returncode + if rpid: + self._reap(status) + return self.returncode + if deadline is not None and time.monotonic() >= deadline: + raise subprocess.TimeoutExpired(self.pid, timeout) + _gevent.sleep(0.01) + + def kill(self): + try: + os.kill(self.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + def terminate(self): + try: + os.kill(self.pid, signal.SIGTERM) + except ProcessLookupError: + pass + + return _Proc() + + except Exception: + if not stdin_w_ok: + os.close(stdin_w) + if not stdout_r_ok: + os.close(stdout_r) + if not stderr_r_ok: + os.close(stderr_r) + raise + finally: + os.close(stdin_r) + os.close(stdout_w) + os.close(stderr_w) diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py index 182b202f..1369a364 100644 --- a/apps/proxy/live_proxy/views.py +++ b/apps/proxy/live_proxy/views.py @@ -25,7 +25,6 @@ from apps.accounts.permissions import ( permission_classes_by_action, ) from .constants import ChannelState, EventType, StreamType, ChannelMetadataField -from .config_helper import ConfigHelper from .services.channel_service import ChannelService from core.utils import send_websocket_update from .url_utils import ( @@ -430,11 +429,11 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): # Channel initialized - cleanup lifecycle now owns the connection release connection_allocated = False - # If we're the owner, wait for connection to establish + # If we're the owner, register the client now so the watchdog + # doesn't stop the channel during connection (which can take + # longer than the grace period). The generator handles waiting + # with keepalive packets via _wait_for_initialization(). if proxy_server.am_i_owner(channel_id): - # Register the client before waiting for the stream to connect. - # The cleanup watchdog stops channels in connecting state with 0 - # clients after the grace period - registering early prevents that. output_profile = _resolve_output_profile(request, user) output_format = _resolve_output_format(user, force_output_format, request) resolved_format = f'{output_format}:p{output_profile.id}' if output_profile else output_format @@ -449,72 +448,6 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): f"(output: {resolved_format}, profile: {output_profile.id if output_profile else None})" ) _client_pre_registered = True - manager = proxy_server.stream_managers.get(channel_id) - if manager: - wait_start = time.time() - timeout = ConfigHelper.connection_timeout() - while not manager.connected: - if time.time() - wait_start > timeout: - proxy_server.stop_channel(channel_id) - return JsonResponse( - {"error": "Connection timeout"}, status=504 - ) - - # Check if this manager should keep retrying or stop - if not manager.should_retry(): - # Check channel state in Redis to make a better decision - metadata_key = RedisKeys.channel_metadata(channel_id) - current_state = None - - if proxy_server.redis_client: - try: - state_bytes = proxy_server.redis_client.hget( - metadata_key, ChannelMetadataField.STATE - ) - if state_bytes: - current_state = state_bytes - logger.debug( - f"[{client_id}] Current state of channel {channel_id}: {current_state}" - ) - except Exception as e: - logger.warning( - f"[{client_id}] Error getting channel state: {e}" - ) - - # Allow normal transitional states to continue - if current_state in [ - ChannelState.INITIALIZING, - ChannelState.CONNECTING, - ]: - logger.info( - f"[{client_id}] Channel {channel_id} is in {current_state} state, continuing to wait" - ) - # Reset wait timer to allow the transition to complete - wait_start = time.time() - continue - - # Check if we're switching URLs - if ( - hasattr(manager, "url_switching") - and manager.url_switching - ): - logger.info( - f"[{client_id}] Stream manager is currently switching URLs for channel {channel_id}" - ) - # Reset wait timer to give the switch a chance - wait_start = time.time() - continue - - # If we reach here, we've exhausted retries and the channel isn't in a valid transitional state - logger.warning( - f"[{client_id}] Channel {channel_id} failed to connect and is not in transitional state" - ) - proxy_server.stop_channel(channel_id) - return JsonResponse( - {"error": "Failed to connect"}, status=502 - ) - - gevent.sleep(0.1) logger.info(f"[{client_id}] Successfully initialized channel {channel_id}") channel_initializing = True diff --git a/core/views.py b/core/views.py index 7321367c..aafd0f69 100644 --- a/core/views.py +++ b/core/views.py @@ -1,8 +1,8 @@ # core/views.py import os +import signal from shlex import split as shlex_split import sys -import subprocess import logging import regex import redis @@ -162,34 +162,53 @@ def stream_view(request, channel_uuid): logger.debug("Executing command: %s", cmd) try: - # Start the streaming process. - process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + stdout_r, stdout_w = os.pipe() + devnull_r = os.open(os.devnull, os.O_RDONLY) + devnull_w = os.open(os.devnull, os.O_WRONLY) + file_actions = [ + (os.POSIX_SPAWN_DUP2, devnull_r, 0), + (os.POSIX_SPAWN_DUP2, stdout_w, 1), + (os.POSIX_SPAWN_DUP2, devnull_w, 2), + (os.POSIX_SPAWN_CLOSE, devnull_r), + (os.POSIX_SPAWN_CLOSE, devnull_w), + (os.POSIX_SPAWN_CLOSE, stdout_w), + (os.POSIX_SPAWN_CLOSE, stdout_r), + ] + proc_pid = os.posix_spawn(cmd[0], cmd, dict(os.environ), file_actions=file_actions) + for fd in (devnull_r, devnull_w, stdout_w): + os.close(fd) + proc_stdout = os.fdopen(stdout_r, 'rb') except Exception as e: - persistent_lock.release() # Ensure the lock is released on error. - logger.exception("Error starting stream for channel ID=%s", stream_id) + persistent_lock.release() + logger.exception("Error starting stream for channel ID=%s", channel_uuid) return HttpResponseServerError(f"Error starting stream: {e}") except Exception as e: - logger.exception("Error preparing stream for channel ID=%s", stream_id) + logger.exception("Error preparing stream for channel ID=%s", channel_uuid) return HttpResponseServerError(f"Error preparing stream: {e}") - def stream_generator(proc, s, persistent_lock): + def stream_generator(stdout_file, pid, persistent_lock): try: while True: - chunk = proc.stdout.read(8192) + chunk = stdout_file.read(8192) if not chunk: break yield chunk finally: try: - proc.terminate() - logger.debug("Streaming process terminated for stream ID=%s", s.id) - except Exception as e: - logger.error("Error terminating process for stream ID=%s: %s", s.id, e) + os.kill(pid, signal.SIGTERM) + logger.debug("Streaming process terminated for channel ID=%s", channel.id) + except OSError: + pass + try: + os.waitpid(pid, os.WNOHANG) + except ChildProcessError: + pass + stdout_file.close() persistent_lock.release() logger.debug("Persistent lock released for channel ID=%s", channel.id) - return StreamingHttpResponse( - stream_generator(process, stream, persistent_lock), - content_type="video/MP2T" - ) + return StreamingHttpResponse( + stream_generator(proc_stdout, proc_pid, persistent_lock), + content_type="video/MP2T" + )