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

@@ -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)

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,
         }

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

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,17 +86,32 @@ class HTTPStreamReader:
                     break
 
                 if chunk:
+                    # 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:
-                        # 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")
+                            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")
 

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(
+            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,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,  # Capture stderr instead of discarding it
-                bufsize=188 * 64            # Buffer optimized for TS packets
+                    _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:
+                    ready, _, _ = _select.select([stderr_fd], [], [], 1.0)
+                    if not ready:
+                        if not self.running or not self.transcode_process:
                             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""
                         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""
-
-                    # 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""
-
-                    # 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""
-
-                except Exception as e:
-                    logger.error(f"Error reading stderr byte: {e}")
+                    chunk = _os.read(stderr_fd, 4096)
+                    if not chunk:
                         break
 
-            # Process any remaining buffer content
-            if buffer.strip():
+                    # 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)
+
+                    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 for channel {self.channel_id}: {e}")
+                    break
+
+            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

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,8 +190,12 @@ 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")
+            else:
                 offset += n
 
     def _writer_loop(self):
@@ -347,9 +346,24 @@ 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()
+            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:
@@ -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:

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,8 +164,12 @@ 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")
+            else:
                 offset += n
 
     def _writer_loop(self):
@@ -266,9 +265,24 @@ 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()
+            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}"

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
+            # 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

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)

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

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),
+        stream_generator(proc_stdout, proc_pid, persistent_lock),
         content_type="video/MP2T"
     )