diff --git a/CHANGELOG.md b/CHANGELOG.md index 60eceeae..c7d3d062 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -100,6 +100,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Plugin monkey-patches and module-level hooks were never applied in uWSGI workers.** Both uWSGI configs use `lazy-apps=true`, meaning each worker boots independently and never inherits state from the master. `should_skip_initialization()` correctly skipped one-shot startup tasks in workers, but also blocked `discover_plugins`, so plugin modules were never imported in any of the 4 request-serving workers. Plugins relying on patching request handling (monkey-patches, signal registrations) were silently inactive until a Connect event lazily triggered discovery in that specific worker. Discovery is now run in every process that serves requests, gated only for Celery processes (which use `worker_ready`) and management commands that don't serve requests. - **PostgreSQL connection pool exhausted under load in gevent workers.** uWSGI's gevent pool runs many greenlets concurrently on a single OS thread. With `CONN_MAX_AGE = 60`, each greenlet that touched the database retained its own open connection for the full 60 seconds (gevent thread-locals are greenlet-locals), rapidly exhausting PostgreSQL's `max_connections` limit under moderate concurrency. `DATABASE_CONN_MAX_AGE` is now `0` so connections are closed after each request. `close_old_connections()` is also called at the top of the long-running stream manager and cleanup watchdog loops so stale handles accumulated across greenlet switches are released promptly. - **Stream proxy race: cleanup watchdog could stop a channel still in the connecting phase.** When the first viewer triggered channel initialization, the client was registered only after the connect-wait loop completed. The cleanup watchdog runs concurrently and stops channels with zero connected clients after a grace period; if the grace period elapsed during the connect-wait, the watchdog killed the channel and left the viewer stuck. The client is now registered before the connect-wait loop begins so the watchdog always sees at least one client. +- **2-second "stream thread did not terminate within timeout" warning on every channel stop.** `_close_socket` in `input/manager.py` was closing the relay pipe read-end (`self.socket`) before killing the ffmpeg process. The stream OS thread blocks in `select()` on that fd; on Linux, closing an fd from another thread while a `select()` is in progress on it does not reliably interrupt the call (POSIX allows this to be undefined). The thread stayed blocked for the full chunk-timeout (5 s), and `stream_thread.join(timeout=2.0)` always expired first. Fixed by killing ffmpeg first: when ffmpeg dies its copy of the relay write-end closes, delivering EOF to `select()` immediately. `self.socket` is closed afterward as cleanup only. +- **Duplicate `stop_channel` calls on every channel shutdown.** `StreamGenerator._cleanup` triggered channel shutdown via two parallel paths: `client_manager.remove_client()` (which fires `handle_client_disconnect` on the owning worker, the correct path) and `_schedule_channel_shutdown_if_needed` (which independently spawned a delayed `stop_channel` greenlet). The duplicate call was suppressed by the `_stopping_channels` guard but produced a redundant log entry and unnecessary greenlet on every shutdown. `_schedule_channel_shutdown_if_needed` has been removed; `handle_client_disconnect` is the sole shutdown trigger. +- **Concurrent greenlets could re-enter `_close_socket` during `proc.wait()`.** `self.transcode_process` was set to `None` at the end of the `if proc:` block rather than immediately after capturing the reference. Under gevent, `proc.wait(timeout=0.5)` yields the hub, allowing a second greenlet to enter `_close_socket`, find `self.transcode_process` still set, and attempt a second kill+close. `self.transcode_process = None` is now assigned immediately after `proc = self.transcode_process` so concurrent callers see `None` and skip the block. ### Performance diff --git a/apps/proxy/live_proxy/input/manager.py b/apps/proxy/live_proxy/input/manager.py index 581c6fb7..ccc52318 100644 --- a/apps/proxy/live_proxy/input/manager.py +++ b/apps/proxy/live_proxy/input/manager.py @@ -1440,23 +1440,20 @@ class StreamManager: except Exception as e: logger.debug(f"Error stopping HTTP reader for channel {self.channel_id}: {e}") - # Otherwise handle socket and transcode resources - if self.socket: - try: - self.socket.close() - except Exception as e: - logger.debug(f"Error closing socket for channel {self.channel_id}: {e}") - pass - - # Enhanced transcode process cleanup with immediate termination - if self.transcode_process: + # Kill proc before closing self.socket. Closing relay_read while the stream + # OS thread is blocked in select() on it does not reliably wake that select() + # on Linux. Killing ffmpeg closes its relay_write, sending EOF to the stream + # thread naturally. We close self.socket afterward as cleanup only. + proc = self.transcode_process + self.transcode_process = None # claim early so concurrent greenlets skip this block + if proc: try: logger.debug(f"Killing transcode process for channel {self.channel_id}") - self.transcode_process.kill() + proc.kill() # Give it a very short time to die try: - self.transcode_process.wait(timeout=0.5) + proc.wait(timeout=0.5) except subprocess.TimeoutExpired: logger.error(f"Failed to kill transcode process even with force for channel {self.channel_id}") except Exception as e: @@ -1464,18 +1461,26 @@ class StreamManager: # Final attempt: try to kill directly try: - self.transcode_process.kill() + proc.kill() except Exception as e: logger.error(f"Final kill attempt failed for channel {self.channel_id}: {e}") + # Close relay socket after proc death; stream thread has already unblocked via EOF. + if self.socket: + try: + self.socket.close() + except Exception as e: + logger.debug(f"Error closing socket for channel {self.channel_id}: {e}") + + if proc: # Explicitly close all subprocess pipes to prevent file descriptor leaks try: - if self.transcode_process.stdin: - self.transcode_process.stdin.close() - if self.transcode_process.stdout: - self.transcode_process.stdout.close() - if self.transcode_process.stderr: - self.transcode_process.stderr.close() + if proc.stdin: + proc.stdin.close() + if proc.stdout: + proc.stdout.close() + if proc.stderr: + proc.stderr.close() logger.debug(f"Closed all subprocess pipes for channel {self.channel_id}") except Exception as e: logger.debug(f"Error closing subprocess pipes for channel {self.channel_id}: {e}") @@ -1492,8 +1497,7 @@ class StreamManager: finally: self.stderr_reader_thread = None - self.transcode_process = None - self.transcode_process_active = False # Reset the flag + self.transcode_process_active = False # Clear transcode active key in Redis if available if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: @@ -1558,6 +1562,10 @@ class StreamManager: try: chunk = _os.read(fd, Config.CHUNK_SIZE) except OSError as e: + import errno as _errno + if e.errno == _errno.EAGAIN and (self.stop_requested or not self.running): + self.connected = False + return False logger.warning(f"Read error for channel {self.channel_id}: {e}") self.connected = False return False diff --git a/apps/proxy/live_proxy/output/ts/generator.py b/apps/proxy/live_proxy/output/ts/generator.py index 59ec8ab2..68ddacca 100644 --- a/apps/proxy/live_proxy/output/ts/generator.py +++ b/apps/proxy/live_proxy/output/ts/generator.py @@ -597,43 +597,6 @@ class StreamGenerator: except Exception as e: logger.error(f"Could not log client disconnect event: {e}") - # Schedule channel shutdown if no clients left - self._schedule_channel_shutdown_if_needed(local_clients) - - def _schedule_channel_shutdown_if_needed(self, local_clients): - """ - Schedule channel shutdown if there are no clients left and we're the owner. - """ - proxy_server = ProxyServer.get_instance() - - # If no clients left and we're the owner, schedule shutdown using the config value - if local_clients == 0 and proxy_server.am_i_owner(self.channel_id): - # When output/profile managers are present, remove_client already spawned - # handle_client_disconnect which will stop them and then stop the channel. - # Spawning a second shutdown greenlet here causes a redundant concurrent - # stop_channel call that can race against the first. - if (proxy_server.output_managers.get(self.channel_id) or - proxy_server.profile_managers.get(self.channel_id)): - return - - logger.info(f"No local clients left for channel {self.channel_id}, scheduling shutdown") - - def delayed_shutdown(): - # Use the config setting instead of hardcoded value - shutdown_delay = ConfigHelper.channel_shutdown_delay() # Use ConfigHelper - logger.info(f"Waiting {shutdown_delay}s before checking if channel should be stopped") - gevent.sleep(shutdown_delay) # Replace time.sleep - - # After delay, check global client count - if self.channel_id in proxy_server.client_managers: - total = proxy_server.client_managers[self.channel_id].get_total_client_count() - if total == 0: - logger.info(f"Shutting down channel {self.channel_id} as no clients connected") - proxy_server.stop_channel(self.channel_id) - else: - logger.info(f"Not shutting down channel {self.channel_id}, {total} clients still connected") - - gevent.spawn(delayed_shutdown) def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None): """ diff --git a/apps/proxy/live_proxy/server.py b/apps/proxy/live_proxy/server.py index b4be63b2..f07ec06f 100644 --- a/apps/proxy/live_proxy/server.py +++ b/apps/proxy/live_proxy/server.py @@ -70,6 +70,7 @@ class ProxyServer: self.profile_managers = {} # {channel_id: {profile_id: OutputProfileManager}} self.profile_buffers = {} # {channel_id: {profile_id: StreamBuffer}} self._channel_names = {} + self._stopping_channels = set() # channels with an active stop_channel call in progress # Generate a unique worker ID pid = os.getpid() @@ -1267,6 +1268,10 @@ class ProxyServer: def stop_channel(self, channel_id): """Stop a channel with proper ownership handling""" + if channel_id in self._stopping_channels: + logger.debug(f"stop_channel already in progress for {channel_id}, ignoring duplicate call") + return + self._stopping_channels.add(channel_id) try: logger.info(f"Stopping channel {channel_id}") @@ -1316,7 +1321,6 @@ class ProxyServer: # Release ownership self.release_ownership(channel_id) - logger.info(f"Released ownership of channel {channel_id}") # Log channel stop event (after cleanup, before releasing ownership section ends) try: @@ -1404,6 +1408,8 @@ class ProxyServer: except Exception as e: logger.error(f"Error stopping channel {channel_id}: {e}") return False + finally: + self._stopping_channels.discard(channel_id) def check_inactive_channels(self): """Check for inactive channels (no clients) and stop them""" @@ -1615,6 +1621,10 @@ class ProxyServer: # Safety: if we have a stream_manager, we ARE the real owner # but the Redis key may have expired. Try to re-acquire. if channel_id in self.stream_managers: + # Ownership was explicitly released by an active stop_channel call - + # don't fight the shutdown by trying to re-acquire. + if channel_id in self._stopping_channels: + continue logger.warning( f"Ownership gap for {channel_id}: this worker has stream_manager " f"but am_i_owner returned False. Attempting re-acquisition."