"""Stream connection management for TS proxy""" import threading import time import socket import requests import subprocess import gevent import re from django.db import connection, close_old_connections from apps.proxy.config import TSConfig as Config from apps.channels.models import Channel, Stream from core.utils import log_system_event from .buffer import StreamBuffer from ..utils import detect_stream_type, get_logger from ..redis_keys import RedisKeys from ..constants import ChannelState, EventType, StreamType, ChannelMetadataField, TS_PACKET_SIZE from ..config_helper import ConfigHelper from ..url_utils import get_alternate_streams, get_stream_info_for_switch, get_stream_object logger = get_logger() class StreamManager: """Manages a connection to a TS stream without using raw sockets""" def __init__(self, channel_id, url, buffer, user_agent=None, transcode=False, stream_id=None, worker_id=None): # Basic properties self.channel_id = channel_id # Cache channel name once to avoid repeated DB queries in hot retry/reconnect loops try: _name = Channel.objects.filter(uuid=channel_id).values_list('name', flat=True).first() self.channel_name = _name if _name else str(channel_id) except Exception: self.channel_name = str(channel_id) self.url = url self.buffer = buffer self.running = True self.connected = False self.retry_count = 0 self.max_retries = ConfigHelper.max_retries() self.current_response = None self.current_session = None self.url_switching = False self.url_switch_start_time = 0 self.url_switch_timeout = ConfigHelper.url_switch_timeout() self.buffering = False self.buffering_timeout = ConfigHelper.buffering_timeout() self.buffering_speed = ConfigHelper.buffering_speed() self.buffering_start_time = None # Store worker_id for ownership checks self.worker_id = worker_id # Sockets used for transcode jobs self.socket = None self.transcode = transcode self.transcode_process = None # User agent for connection self.user_agent = user_agent or Config.DEFAULT_USER_AGENT # Stream health monitoring self.last_data_time = time.time() self.healthy = True self.health_check_interval = ConfigHelper.get('HEALTH_CHECK_INTERVAL', 5) self.chunk_size = ConfigHelper.chunk_size() # Add to your __init__ method self._buffer_check_timers = [] self.stopping = False self.stop_requested = False # Add tracking for tried streams and current stream self.current_stream_id = stream_id self.tried_stream_ids = set() if stream_id: self.tried_stream_ids.add(stream_id) logger.info(f"Initialized stream manager for channel {buffer.channel_id} with stream ID {stream_id}") else: # Try to get stream ID from Redis metadata if available if hasattr(buffer, 'redis_client') and buffer.redis_client: try: metadata_key = RedisKeys.channel_metadata(channel_id) # Log all metadata for debugging purposes metadata = buffer.redis_client.hgetall(metadata_key) if metadata: logger.debug(f"Redis metadata for channel {channel_id}: {metadata}") # Try to get stream_id specifically stream_id_bytes = buffer.redis_client.hget(metadata_key, "stream_id") if stream_id_bytes: self.current_stream_id = int(stream_id_bytes) self.tried_stream_ids.add(self.current_stream_id) logger.info(f"Loaded stream ID {self.current_stream_id} from Redis for channel {buffer.channel_id}") else: logger.warning(f"No stream_id found in Redis for channel {channel_id}. " f"Stream switching will rely on URL comparison to avoid selecting the same stream.") except Exception as e: logger.warning(f"Error loading stream ID from Redis: {e}") else: logger.warning(f"Unable to get stream ID for channel {channel_id}. " f"Stream switching will rely on URL comparison to avoid selecting the same stream.") logger.info(f"Initialized stream manager for channel {buffer.channel_id}") self.transcode_process_active = False # Track stream command for efficient log parser routing self.stream_command = None self.parser_type = None # Will be set when transcode process starts # Add tracking for data throughput self.bytes_processed = 0 self.last_bytes_update = time.time() # Cached result of the pipelined Redis ownership audit (hot read path). self._ownership_cache_valid_until = 0.0 self._ownership_cached = True self._OWNERSHIP_CHECK_INTERVAL = 1.0 self.bytes_update_interval = 5 # Update Redis every 5 seconds # Add stderr reader thread property self.stderr_reader_thread = None self.ffmpeg_input_phase = True # Track if we're still reading input info # Add HTTP reader thread property self.http_reader = None # Output bitrate smoothing / throttled DB persistence self._smoothed_output_bitrate = None self._last_bitrate_db_save_time = 0 self._bitrate_db_save_interval = 30 # seconds between DB writes self._bitrate_warmup_samples = 10 # discard first N samples while EMA stabilizes (~5s) def _create_session(self): """Create and configure requests session with optimal settings""" session = requests.Session() # Configure session headers session.headers.update({ 'User-Agent': self.user_agent, 'Connection': 'keep-alive' }) # Set up connection pooling for better performance adapter = requests.adapters.HTTPAdapter( pool_connections=1, # Single connection for this stream pool_maxsize=1, # Max size of connection pool max_retries=3, # Auto-retry for failed requests pool_block=False # Don't block when pool is full ) # Apply adapter to both HTTP and HTTPS session.mount('http://', adapter) session.mount('https://', adapter) return session def _wait_for_existing_processes_to_close(self, timeout=5.0): """Wait for existing processes/connections to fully close before establishing new ones""" start_time = time.time() while time.time() - start_time < timeout: # Check if transcode process is still running if self.transcode_process and self.transcode_process.poll() is None: logger.debug(f"Waiting for existing transcode process to terminate for channel {self.channel_id}") gevent.sleep(0.1) continue # Check if HTTP connections are still active if self.current_response or self.current_session: logger.debug(f"Waiting for existing HTTP connections to close for channel {self.channel_id}") gevent.sleep(0.1) continue # Check if socket is still active if self.socket: logger.debug(f"Waiting for existing socket to close for channel {self.channel_id}") gevent.sleep(0.1) continue # All processes/connections are closed logger.debug(f"All existing processes closed for channel {self.channel_id}") return True # Timeout reached logger.warning(f"Timeout waiting for existing processes to close for channel {self.channel_id} after {timeout}s") return False def _invalidate_ownership_cache(self): self._ownership_cache_valid_until = 0.0 @staticmethod def _decode_redis_value(value): if value is None: return None if isinstance(value, bytes): return value.decode() return value def _disconnect_shutdown_ready(self, disconnect_value): """True when last-client disconnect has passed the configured shutdown delay.""" if not disconnect_value: return False shutdown_delay = ConfigHelper.channel_shutdown_delay() if shutdown_delay <= 0: return True disconnect_value = self._decode_redis_value(disconnect_value) try: disconnect_time = float(disconnect_value) except (ValueError, TypeError): return False return (time.time() - disconnect_time) >= shutdown_delay def _evaluate_ownership_from_redis(self, redis_client): """Single pipelined Redis round-trip for the full ownership audit.""" stop_key = RedisKeys.channel_stopping(self.channel_id) metadata_key = RedisKeys.channel_metadata(self.channel_id) clients_key = RedisKeys.clients(self.channel_id) owner_key = RedisKeys.channel_owner(self.channel_id) disconnect_key = RedisKeys.last_client_disconnect(self.channel_id) pipe = redis_client.pipeline(transaction=False) pipe.exists(stop_key) pipe.exists(metadata_key) pipe.scard(clients_key) pipe.get(owner_key) pipe.get(disconnect_key) pipe.hget(metadata_key, ChannelMetadataField.STATE) ( stop_exists, metadata_exists, client_count, current_owner, disconnect_value, state_raw, ) = pipe.execute() if stop_exists: return False if not metadata_exists: logger.warning( f"Channel {self.channel_id} metadata removed from Redis - stopping upstream" ) return False client_count = client_count or 0 state = self._decode_redis_value(state_raw) current_owner = self._decode_redis_value(current_owner) if current_owner and current_owner != self.worker_id: return False if not current_owner: if client_count == 0 and state not in ChannelState.PRE_ACTIVE: logger.warning( f"Channel {self.channel_id} has no owner and no clients - stopping upstream" ) return False return True if client_count == 0 and self._disconnect_shutdown_ready(disconnect_value): logger.info( f"Channel {self.channel_id} disconnect shutdown ready - stopping upstream" ) return False return True def _still_owner(self, *, force=False): """Return True while this worker should keep the upstream connection open.""" if self.stopping or self.stop_requested: return False if not self.worker_id: return True redis_client = getattr(self.buffer, 'redis_client', None) if not redis_client: return True now = time.time() if not force and now < self._ownership_cache_valid_until: return self._ownership_cached try: result = self._evaluate_ownership_from_redis(redis_client) self._ownership_cached = result self._ownership_cache_valid_until = now + self._OWNERSHIP_CHECK_INTERVAL return result except Exception as e: logger.debug(f"Ownership check failed for channel {self.channel_id}: {e}") return True def _upstream_may_continue(self): """ Per-chunk gate for the hot read path. Local stop flags are checked every chunk. The coordinated-teardown Redis flag is checked every chunk (one EXISTS). The full ownership audit is pipelined and cached for ~1s — enough for loop boundaries while avoiding 6+ Redis round-trips per chunk during steady streaming. """ if self.stopping or self.stop_requested or not self.running: return False if self.buffer is not None and self.buffer.stopping: return False redis_client = getattr(self.buffer, 'redis_client', None) if redis_client and self.worker_id: try: if redis_client.exists(RedisKeys.channel_stopping(self.channel_id)): self._ownership_cached = False self._ownership_cache_valid_until = 0.0 return False except Exception as e: logger.debug( f"Channel stopping check failed for {self.channel_id}: {e}" ) return self._still_owner() def _ensure_owner_or_stop(self): if self._still_owner(force=True): return True logger.warning( f"Stream manager for channel {self.channel_id} lost ownership " f"(worker {self.worker_id}) - stopping upstream" ) self.stop() return False def run(self): """Main execution loop using HTTP streaming with improved connection handling and stream switching""" # Add a stop flag to the class properties self.stop_requested = False # Add tracking for stream switching attempts stream_switch_attempts = 0 # Get max stream switches from config using the helper method max_stream_switches = ConfigHelper.max_stream_switches() # Prevent infinite switching loops try: # Start health monitor thread health_thread = threading.Thread(target=self._monitor_health, daemon=True) health_thread.start() logger.info(f"Starting stream for URL: {self.url} for channel {self.channel_id}") # Main stream switching loop - we'll try different streams if needed while self.running and stream_switch_attempts <= max_stream_switches: close_old_connections() if not self._ensure_owner_or_stop(): break # Check for stuck switching state if self.url_switching and time.time() - self.url_switch_start_time > self.url_switch_timeout: logger.warning(f"URL switching state appears stuck for channel {self.channel_id} " f"({time.time() - self.url_switch_start_time:.1f}s > {self.url_switch_timeout}s timeout). " f"Resetting switching state.") self._reset_url_switching_state() # NEW: Check for health monitor recovery requests if hasattr(self, 'needs_reconnect') and self.needs_reconnect and not self.url_switching: logger.info(f"Health monitor requested reconnect for channel {self.channel_id}") self.needs_reconnect = False # Attempt reconnect without changing streams if self._attempt_reconnect(): logger.info(f"Health-requested reconnect successful for channel {self.channel_id}") continue # Go back to main loop else: logger.warning(f"Health-requested reconnect failed, will try stream switch for channel {self.channel_id}") self.needs_stream_switch = True if hasattr(self, 'needs_stream_switch') and self.needs_stream_switch and not self.url_switching: logger.info(f"Health monitor requested stream switch for channel {self.channel_id}") self.needs_stream_switch = False if self._try_next_stream(): logger.info(f"Health-requested stream switch successful for channel {self.channel_id}") stream_switch_attempts += 1 self.retry_count = 0 # Reset retries for new stream continue # Go back to main loop with new stream else: logger.error(f"Health-requested stream switch failed for channel {self.channel_id}") # Continue with normal flow # Check stream type before connecting self.stream_type = detect_stream_type(self.url) if self.transcode == False and self.stream_type in (StreamType.HLS, StreamType.RTSP, StreamType.UDP): stream_type_name = "HLS" if self.stream_type == StreamType.HLS else ("RTSP/RTP" if self.stream_type == StreamType.RTSP else "UDP") logger.info(f"Detected {stream_type_name} stream: {self.url} for channel {self.channel_id}") logger.info(f"{stream_type_name} streams require FFmpeg for channel {self.channel_id}") # Enable transcoding for HLS, RTSP/RTP, and UDP streams self.transcode = True # We'll override the stream profile selection with ffmpeg in the transcoding section self.force_ffmpeg = True # Reset connection retry count for this specific URL self.retry_count = 0 url_failed = False if self.url_switching: logger.debug(f"Skipping connection attempt during URL switch for channel {self.channel_id}") gevent.sleep(0.1) continue # Connection retry loop for current URL while self.running and self.retry_count < self.max_retries and not url_failed and not self.needs_stream_switch: if not self._ensure_owner_or_stop(): break logger.info(f"Connection attempt {self.retry_count + 1}/{self.max_retries} for URL: {self.url} for channel {self.channel_id}") # Handle connection based on whether we transcode or not connection_result = False try: if self.transcode: connection_result = self._establish_transcode_connection() else: connection_result = self._establish_http_connection() if connection_result: # Store connection start time to measure success duration connection_start_time = time.time() # Log reconnection event if this is a retry (not first attempt) if self.retry_count > 0: try: log_system_event( 'channel_reconnect', channel_id=self.channel_id, channel_name=self.channel_name, attempt=self.retry_count + 1, max_attempts=self.max_retries ) except Exception as e: logger.error(f"Could not log reconnection event: {e}") # Successfully connected - read stream data until disconnect/error self._process_stream_data() # If we get here, the connection was closed/failed # Reset stream switch attempts if the connection lasted longer than threshold # This indicates we had a stable connection for a while before failing connection_duration = time.time() - connection_start_time stable_connection_threshold = 30 # 30 seconds threshold if self.needs_stream_switch: logger.info(f"Stream needs to switch after {connection_duration:.1f} seconds for channel: {self.channel_id}") break # Exit to switch streams if connection_duration > stable_connection_threshold: logger.info(f"Stream was stable for {connection_duration:.1f} seconds, resetting switch attempts counter for channel: {self.channel_id}") stream_switch_attempts = 0 # Connection failed or ended - decide what to do next if self.stop_requested or not self.running: # Normal shutdown requested return # Connection failed, increment retry count self.retry_count += 1 self.connected = False # If we've reached max retries, mark this URL as failed if self.retry_count >= self.max_retries: url_failed = True logger.warning(f"Maximum retry attempts ({self.max_retries}) reached for URL: {self.url} for channel: {self.channel_id}") # Log connection error event try: log_system_event( 'channel_error', channel_id=self.channel_id, channel_name=self.channel_name, error_type='connection_failed', url=self.url[:100] if self.url else None, attempts=self.max_retries ) except Exception as e: logger.error(f"Could not log connection error event: {e}") else: # Wait with exponential backoff before retrying timeout = min(.25 * self.retry_count, 3) # Cap at 3 seconds logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count}/{self.max_retries}) for channel: {self.channel_id}") gevent.sleep(timeout) except Exception as e: logger.error(f"Connection error on channel: {self.channel_id}: {e}", exc_info=True) self.retry_count += 1 self.connected = False if self.retry_count >= self.max_retries: url_failed = True # Log connection error event with exception details try: log_system_event( 'channel_error', channel_id=self.channel_id, channel_name=self.channel_name, error_type='connection_exception', error_message=str(e)[:200], url=self.url[:100] if self.url else None, attempts=self.max_retries ) except Exception as log_error: logger.error(f"Could not log connection error event: {log_error}") else: # Wait with exponential backoff before retrying timeout = min(.25 * self.retry_count, 3) # Cap at 3 seconds logger.info(f"Reconnecting in {timeout} seconds after error... (attempt {self.retry_count}/{self.max_retries}) for channel: {self.channel_id}") gevent.sleep(timeout) # If URL failed and we're still running, try switching to another stream if url_failed and self.running: logger.info(f"URL {self.url} failed after {self.retry_count} attempts, trying next stream for channel: {self.channel_id}") # Try to switch to next stream switch_result = self._try_next_stream() if switch_result: # Successfully switched to a new stream, continue with the new URL stream_switch_attempts += 1 logger.info(f"Successfully switched to new URL: {self.url} (switch attempt {stream_switch_attempts}/{max_stream_switches}) for channel: {self.channel_id}") # Reset retry count for the new stream - important for the loop to work correctly self.retry_count = 0 # Continue outer loop with new URL - DON'T add a break statement here else: # No more streams to try logger.error(f"Failed to find alternative streams after {stream_switch_attempts} attempts for channel: {self.channel_id}") break elif not self.running: # Normal shutdown was requested break except Exception as e: logger.error(f"Stream error: {e}", exc_info=True) finally: try: from ..server import ProxyServer ProxyServer.get_instance()._live_stream_managers.pop(self.channel_id, None) except Exception: pass # Enhanced cleanup in the finally block self.connected = False # Explicitly cancel all timers for timer in list(self._buffer_check_timers): try: if timer and timer.is_alive(): timer.cancel() except Exception: pass self._buffer_check_timers.clear() # Make sure transcode process is terminated if self.transcode_process_active: logger.info(f"Ensuring transcode process is terminated in finally block for channel: {self.channel_id}") self._close_socket() # Close all connections self._close_all_connections() # Transition to ERROR so clients stop waiting. Ownership may have # expired during retries, so fall back to a state guard when no # owner exists — but never clobber a new owner's active stream. if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: try: metadata_key = RedisKeys.channel_metadata(self.channel_id) owner_key = RedisKeys.channel_owner(self.channel_id) current_owner = self._decode_redis_value( self.buffer.redis_client.get(owner_key) ) is_owner = ( current_owner and self.worker_id and current_owner == self.worker_id ) no_owner = current_owner is None should_update = is_owner if not should_update and no_owner: current_state = self._decode_redis_value( self.buffer.redis_client.hget( metadata_key, ChannelMetadataField.STATE ) ) should_update = current_state in ChannelState.PRE_ACTIVE if not should_update and current_state: logger.info( f"Channel {self.channel_id} has no owner but " f"state is {current_state} — skipping ERROR update" ) if should_update: if self.tried_stream_ids and len(self.tried_stream_ids) > 0: error_message = f"All {len(self.tried_stream_ids)} stream options failed" else: error_message = f"Connection failed after {self.max_retries} attempts" update_data = { ChannelMetadataField.STATE: ChannelState.ERROR, ChannelMetadataField.STATE_CHANGED_AT: str(time.time()), ChannelMetadataField.ERROR_MESSAGE: error_message, ChannelMetadataField.ERROR_TIME: str(time.time()) } self.buffer.redis_client.hset(metadata_key, mapping=update_data) logger.info( f"Updated channel {self.channel_id} state to ERROR " f"in Redis after stream failure " f"(owner={'self' if is_owner else 'expired'})" ) # Signal clients to disconnect stop_key = RedisKeys.channel_stopping(self.channel_id) self.buffer.redis_client.setex(stop_key, 60, "true") except Exception as e: logger.error(f"Failed to update channel state in Redis: {e} for channel {self.channel_id}", exc_info=True) # Close database connection for this thread try: connection.close() except Exception: pass logger.info(f"Stream manager stopped for channel {self.channel_id}") def _establish_transcode_connection(self): """Establish a connection using transcoding""" try: logger.debug(f"Building transcode command for channel {self.channel_id}") # Check if we already have a running transcode process if self.transcode_process and self.transcode_process.poll() is None: logger.info(f"Existing transcode process found for channel {self.channel_id}, closing before establishing new connection") self._close_socket() # Wait for the process to fully terminate if not self._wait_for_existing_processes_to_close(): logger.error(f"Failed to close existing transcode process for channel {self.channel_id}") return False # Also check for any lingering HTTP connections if self.current_response or self.current_session: logger.debug(f"Closing existing HTTP connections before establishing transcode connection for channel {self.channel_id}") self._close_connection() try: channel = get_stream_object(self.channel_id) # Use FFmpeg specifically for HLS streams if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg: from core.models import StreamProfile try: stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True) logger.info("Using FFmpeg stream profile for unsupported proxy content (HLS/RTSP/UDP)") except StreamProfile.DoesNotExist: # Fall back to channel's profile if FFmpeg not found stream_profile = channel.get_stream_profile() logger.warning(f"FFmpeg profile not found, using channel default profile for channel: {self.channel_id}") else: stream_profile = channel.get_stream_profile() # Build and start transcode command self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent) # Store stream command for efficient log parser routing self.stream_command = stream_profile.command # Map actual commands to parser types for direct routing command_to_parser = { 'ffmpeg': 'ffmpeg', 'cvlc': 'vlc', 'vlc': 'vlc', 'streamlink': 'streamlink' } self.parser_type = command_to_parser.get(self.stream_command.lower()) if self.parser_type: logger.debug(f"Using {self.parser_type} parser for log parsing (command: {self.stream_command})") else: logger.debug(f"Unknown stream command '{self.stream_command}', will use auto-detection for log parsing") # For UDP streams, remove any user_agent parameters from the command if hasattr(self, 'stream_type') and self.stream_type == StreamType.UDP: # Filter out any arguments that contain the user_agent value or related headers self.transcode_cmd = [arg for arg in self.transcode_cmd if self.user_agent not in arg and 'user-agent' not in arg.lower() and 'user_agent' not in arg.lower()] logger.debug(f"Removed user_agent parameters from UDP stream command for channel: {self.channel_id}") finally: # Release the pool slot before posix_spawn or before returning on profile errors. close_old_connections() logger.debug(f"Starting transcode process: {self.transcode_cmd} for channel: {self.channel_id}") 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() # Set flag that transcoding process is active self.transcode_process_active = True self.connected = True # Set connection start time for stability tracking self.connection_start_time = time.time() # Set channel state to waiting for clients self._set_waiting_for_clients() return True except Exception as e: logger.error(f"Error establishing transcode connection for channel: {self.channel_id}: {e}", exc_info=True) self._close_socket() return False def _start_stderr_reader(self): """Start a thread to read stderr from the transcode process""" if self.transcode_process and self.transcode_process.stderr: self.stderr_reader_thread = threading.Thread( target=self._read_stderr, daemon=True # Use daemon thread so it doesn't block program exit ) self.stderr_reader_thread.start() logger.debug(f"Started stderr reader thread for channel {self.channel_id}") 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: stderr = self.transcode_process.stderr if not stderr: return stderr_fd = stderr.fileno() buf = b"" while self.running and self.transcode_process and self.transcode_process.stderr: try: ready, _, _ = _select.select([stderr_fd], [], [], 1.0) if not ready: if not self.running or not self.transcode_process: break continue chunk = _os.read(stderr_fd, 4096) if not chunk: break # 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 = 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 stderr buffer: {e}") except Exception as e: try: logger.error(f"Error in stderr reader thread for channel {self.channel_id}: {e}") except: pass def _log_stderr_content(self, content): """Log stderr content from FFmpeg with appropriate log levels""" try: content = content.strip() if not content: return # Convert to lowercase for easier matching content_lower = content.lower() # Check if we are still in the input phase if content_lower.startswith('input #') or 'decoder' in content_lower: self.ffmpeg_input_phase = True # Track FFmpeg phases - once we see output info, we're past input phase if content_lower.startswith('output #') or 'encoder' in content_lower: self.ffmpeg_input_phase = False # Route to appropriate parser based on known command type from ..services.log_parsers import LogParserFactory from ..services.channel_service import ChannelService parse_result = None # If we know the parser type, use direct routing for efficiency if self.parser_type: # Get the appropriate parser and check what it can parse parser = LogParserFactory._parsers.get(self.parser_type) if parser: stream_type = parser.can_parse(content) if stream_type: # Parser can handle this line, parse it directly parsed_data = LogParserFactory.parse(stream_type, content) if parsed_data: parse_result = (stream_type, parsed_data) else: # Unknown command type - use auto-detection as fallback parse_result = LogParserFactory.auto_parse(content) if parse_result: stream_type, parsed_data = parse_result # For FFmpeg, only parse during input phase if stream_type in ['video', 'audio', 'input']: if self.ffmpeg_input_phase: ChannelService.parse_and_store_stream_info(self.channel_id, content, stream_type, self.current_stream_id) else: # VLC and Streamlink can be parsed anytime ChannelService.parse_and_store_stream_info(self.channel_id, content, stream_type, self.current_stream_id) # Determine log level based on content if any(keyword in content_lower for keyword in ['error', 'failed', 'cannot', 'invalid', 'corrupt']): logger.error(f"Stream process error for channel {self.channel_id}: {content}") elif any(keyword in content_lower for keyword in ['warning', 'deprecated', 'ignoring']): logger.warning(f"Stream process warning for channel {self.channel_id}: {content}") elif content.startswith('frame=') or 'fps=' in content or 'speed=' in content: # Stats lines - log at trace level to avoid spam logger.trace(f"Stream stats for channel {self.channel_id}: {content}") elif any(keyword in content_lower for keyword in ['input', 'output', 'stream', 'video', 'audio']): # Stream info - log at info level logger.info(f"Stream info for channel {self.channel_id}: {content}") else: # Everything else at debug level logger.debug(f"Stream process output for channel {self.channel_id}: {content}") except Exception as e: logger.error(f"Error logging stderr content for channel {self.channel_id}: {e}") def _parse_ffmpeg_stats(self, stats_line): """Parse FFmpeg stats line and extract speed, fps, and bitrate""" try: # Example FFmpeg stats line: # frame= 1234 fps= 30 q=28.0 size= 2048kB time=00:00:41.33 bitrate= 406.1kbits/s speed=1.02x # Extract speed (e.g., "speed=1.02x") speed_match = re.search(r'speed=\s*([0-9.]+)x?', stats_line) ffmpeg_speed = float(speed_match.group(1)) if speed_match else None # Extract fps (e.g., "fps= 30") fps_match = re.search(r'fps=\s*([0-9.]+)', stats_line) ffmpeg_fps = float(fps_match.group(1)) if fps_match else None # Extract bitrate (e.g., "bitrate= 406.1kbits/s") bitrate_match = re.search(r'bitrate=\s*([0-9.]+(?:\.[0-9]+)?)\s*([kmg]?)bits/s', stats_line, re.IGNORECASE) ffmpeg_output_bitrate = None if bitrate_match: bitrate_value = float(bitrate_match.group(1)) unit = bitrate_match.group(2).lower() # Convert to kbps if unit == 'm': bitrate_value *= 1000 elif unit == 'g': bitrate_value *= 1000000 # If no unit or 'k', it's already in kbps ffmpeg_output_bitrate = bitrate_value # Calculate actual FPS actual_fps = None if ffmpeg_fps is not None and ffmpeg_speed is not None and ffmpeg_speed > 0: actual_fps = ffmpeg_fps / ffmpeg_speed # Store in Redis if we have valid data if any(x is not None for x in [ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_output_bitrate]): self._update_ffmpeg_stats_in_redis(ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_output_bitrate) # Update local EMA and periodically flush to database if ffmpeg_output_bitrate is not None and self.current_stream_id: if self._bitrate_warmup_samples > 0: # Discard early samples from the EMA self._bitrate_warmup_samples -= 1 else: if self._smoothed_output_bitrate is None: self._smoothed_output_bitrate = ffmpeg_output_bitrate else: self._smoothed_output_bitrate = 0.9 * self._smoothed_output_bitrate + 0.1 * ffmpeg_output_bitrate now = time.time() if now - self._last_bitrate_db_save_time >= self._bitrate_db_save_interval: from ..services.channel_service import ChannelService ChannelService._update_stream_stats_in_db( self.current_stream_id, ffmpeg_output_bitrate=round(self._smoothed_output_bitrate, 1) ) self._last_bitrate_db_save_time = now # Fix the f-string formatting actual_fps_str = f"{actual_fps:.1f}" if actual_fps is not None else "N/A" ffmpeg_output_bitrate_str = f"{ffmpeg_output_bitrate:.1f}" if ffmpeg_output_bitrate is not None else "N/A" # Log the stats logger.debug(f"FFmpeg stats for channel {self.channel_id}: - Speed: {ffmpeg_speed}x, FFmpeg FPS: {ffmpeg_fps}, " f"Actual FPS: {actual_fps_str}, " f"Output Bitrate: {ffmpeg_output_bitrate_str} kbps") # If we have a valid speed, check for buffering if ffmpeg_speed is not None and ffmpeg_speed < self.buffering_speed: if self.buffering: # Buffering is still ongoing, check for how long if self.buffering_start_time is None: self.buffering_start_time = time.time() else: buffering_duration = time.time() - self.buffering_start_time if buffering_duration > self.buffering_timeout: # Buffering timeout reached, log error and try next stream logger.error(f"Buffering timeout reached for channel {self.channel_id} after {buffering_duration:.1f} seconds") # Send next stream request if self._try_next_stream(): logger.info(f"Switched to next stream for channel {self.channel_id} after buffering timeout") # Reset buffering state self.buffering = False self.buffering_start_time = None # Log failover event try: log_system_event( 'channel_failover', channel_id=self.channel_id, channel_name=self.channel_name, reason='buffering_timeout', duration=buffering_duration ) except Exception as e: logger.error(f"Could not log failover event: {e}") else: logger.error(f"Failed to switch to next stream for channel {self.channel_id} after buffering timeout") else: # Buffering just started, set the flag and start timer self.buffering = True self.buffering_start_time = time.time() logger.warning(f"Buffering started for channel {self.channel_id} - speed: {ffmpeg_speed}x") # Log system event for buffering try: log_system_event( 'channel_buffering', channel_id=self.channel_id, channel_name=self.channel_name, speed=ffmpeg_speed ) except Exception as e: logger.error(f"Could not log buffering event: {e}") # Log buffering warning logger.debug(f"FFmpeg speed on channel {self.channel_id} is below {self.buffering_speed} ({ffmpeg_speed}x) - buffering detected") # Set channel state to buffering if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: metadata_key = RedisKeys.channel_metadata(self.channel_id) self.buffer.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.BUFFERING) elif ffmpeg_speed is not None and ffmpeg_speed >= self.buffering_speed: # Speed is good, check if we were buffering if self.buffering: # Reset buffering state logger.info(f"Buffering ended for channel {self.channel_id} - speed: {ffmpeg_speed}x") self.buffering = False self.buffering_start_time = None # Set channel state to active if speed is good if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: metadata_key = RedisKeys.channel_metadata(self.channel_id) self.buffer.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.ACTIVE) except Exception as e: logger.debug(f"Error parsing FFmpeg stats: {e}") def _update_ffmpeg_stats_in_redis(self, speed, fps, actual_fps, output_bitrate): """Update FFmpeg performance stats in Redis metadata""" try: if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: metadata_key = RedisKeys.channel_metadata(self.channel_id) update_data = { ChannelMetadataField.FFMPEG_STATS_UPDATED: str(time.time()) } if speed is not None: update_data[ChannelMetadataField.FFMPEG_SPEED] = str(round(speed, 3)) if fps is not None: update_data[ChannelMetadataField.FFMPEG_FPS] = str(round(fps, 1)) if actual_fps is not None: update_data[ChannelMetadataField.ACTUAL_FPS] = str(round(actual_fps, 1)) if output_bitrate is not None: update_data[ChannelMetadataField.FFMPEG_OUTPUT_BITRATE] = str(round(output_bitrate, 1)) self.buffer.redis_client.hset(metadata_key, mapping=update_data) except Exception as e: logger.error(f"Error updating FFmpeg stats in Redis: {e}") def _establish_http_connection(self): """Establish HTTP connection using thread-based reader (same as transcode path)""" try: logger.debug(f"Using HTTP streamer thread to connect to stream: {self.url}") # Check if we already have active HTTP connections if self.current_response or self.current_session: logger.info(f"Existing HTTP connection found for channel {self.channel_id}, closing before establishing new connection") self._close_connection() # Wait for connections to fully close if not self._wait_for_existing_processes_to_close(): logger.error(f"Failed to close existing HTTP connections for channel {self.channel_id}") return False # Also check for any lingering transcode processes if self.transcode_process and self.transcode_process.poll() is None: logger.debug(f"Closing existing transcode process before establishing HTTP connection for channel {self.channel_id}") self._close_socket() # Use HTTPStreamReader to fetch stream and pipe to a readable file descriptor # This allows us to use the same fetch_chunk() path as transcode from .http_streamer import HTTPStreamReader # Create and start the HTTP stream reader self.http_reader = HTTPStreamReader( url=self.url, user_agent=self.user_agent, chunk_size=self.chunk_size ) # Start the reader thread and get the read end of the pipe pipe_fd = self.http_reader.start() # Wrap the file descriptor in a file object (same as transcode stdout) import os self.socket = os.fdopen(pipe_fd, 'rb', buffering=0) self.connected = True self.healthy = True logger.info(f"Successfully started HTTP streamer thread for channel {self.channel_id}") # Store connection start time for stability tracking self.connection_start_time = time.time() # Set channel state to waiting for clients self._set_waiting_for_clients() return True except Exception as e: logger.error(f"Error establishing HTTP connection for channel {self.channel_id}: {e}", exc_info=True) self._close_socket() return False def _update_bytes_processed(self, chunk_size): """Update the total bytes processed in Redis metadata""" if not self._upstream_may_continue(): return try: # Update local counter self.bytes_processed += chunk_size # Only update Redis periodically to reduce overhead now = time.time() if now - self.last_bytes_update >= self.bytes_update_interval: if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: # Update channel metadata with total bytes metadata_key = RedisKeys.channel_metadata(self.channel_id) # Use hincrby to atomically increment the total_bytes field self.buffer.redis_client.hincrby(metadata_key, ChannelMetadataField.TOTAL_BYTES, self.bytes_processed) # Reset local counter after updating Redis self.bytes_processed = 0 self.last_bytes_update = now logger.debug(f"Updated {ChannelMetadataField.TOTAL_BYTES} in Redis for channel {self.channel_id}") except Exception as e: logger.error(f"Error updating bytes processed: {e}") def _process_stream_data(self): """Process stream data until disconnect or error - unified path for both transcode and HTTP""" try: # Both transcode and HTTP now use the same subprocess/socket approach # This gives us perfect control: check flags between chunks, timeout just returns False while self.running and self.connected and not self.stop_requested and not self.needs_stream_switch: if self.fetch_chunk(): self.last_data_time = time.time() else: # fetch_chunk() returned False - could be timeout, no data, or error if not self.running: break # Brief sleep before retry to avoid tight loop gevent.sleep(0.1) except Exception as e: logger.error(f"Error processing stream data for channel {self.channel_id}: {e}", exc_info=True) # If we exit the loop, connection is closed or failed self.connected = False def _close_all_connections(self): """Close all connection resources""" if self.socket or self.transcode_process: try: self._close_socket() except Exception as e: logger.debug(f"Error closing socket for channel {self.channel_id}: {e}") if self.current_response: try: self.current_response.close() except Exception as e: logger.debug(f"Error closing response for channel {self.channel_id}: {e}") if self.current_session: try: self.current_session.close() except Exception as e: logger.debug(f"Error closing session for channel {self.channel_id}: {e}") # Clear references self.socket = None self.current_response = None self.current_session = None self.transcode_process = None def stop(self): """Stop the stream manager and cancel all timers""" logger.info(f"Stopping stream manager for channel {self.channel_id}") self.stopping = True self._invalidate_ownership_cache() if self.buffer is not None: self.buffer.stopping = True # Cancel all buffer check timers for timer in list(self._buffer_check_timers): try: if timer and timer.is_alive(): timer.cancel() except Exception as e: logger.error(f"Error canceling buffer check timer for channel {self.channel_id}: {e}") self._buffer_check_timers.clear() # Set the flag first self.stop_requested = True # Close any active response connection if hasattr(self, 'current_response') and self.current_response: # CORRECT NAME try: self.current_response.close() # CORRECT NAME except Exception: pass # Also close the session if hasattr(self, 'current_session') and self.current_session: try: self.current_session.close() except Exception: pass # Explicitly close socket/transcode resources self._close_socket() # Set running to false to ensure thread exits self.running = False # Flush the final bitrate to DB on stop only if warmup completed and we have # a meaningful EMA. Short previews / channel hops that die during warmup do NOT # write anything, preserving any previously correct value in the database. if self._smoothed_output_bitrate is not None and self.current_stream_id: final_bitrate = self._smoothed_output_bitrate try: from ..services.channel_service import ChannelService ChannelService._update_stream_stats_in_db( self.current_stream_id, ffmpeg_output_bitrate=round(final_bitrate, 1) ) except Exception as e: logger.debug(f"Error flushing final bitrate to DB for channel {self.channel_id}: {e}") def update_url(self, new_url, stream_id=None, m3u_profile_id=None): """Update stream URL and reconnect with proper cleanup for both HTTP and transcode sessions""" if new_url == self.url: logger.info(f"URL unchanged: {new_url}") return False logger.info(f"Switching stream URL from {self.url} to {new_url} for channel {self.channel_id}") # Import both models for proper resource management from apps.channels.models import Stream, Channel from django.db import connection # Update stream profile if we're switching streams if self.current_stream_id and stream_id and self.current_stream_id != stream_id: try: # Get the channel by UUID channel = Channel.objects.get(uuid=self.channel_id) # Get stream to find its profile #new_stream = Stream.objects.get(pk=stream_id) # Use the new method to update the profile and manage connection counts if m3u_profile_id: success = channel.update_stream_profile(m3u_profile_id) if success: logger.debug(f"Updated m3u profile for channel {self.channel_id} to use profile from stream {stream_id}") else: logger.warning(f"Failed to update stream profile for channel {self.channel_id}") except Exception as e: logger.error(f"Error updating stream profile for channel {self.channel_id}: {e}") finally: # Always close database connection after profile update try: connection.close() except Exception: pass # CRITICAL: Set a flag to prevent immediate reconnection with old URL self.url_switching = True self.url_switch_start_time = time.time() try: # Check which type of connection we're using and close it properly if self.transcode or self.socket: logger.debug(f"Closing transcode process before URL change for channel {self.channel_id}") self._close_socket() else: logger.debug(f"Closing HTTP connection before URL change for channel {self.channel_id}") self._close_connection() # Update URL and reset connection state old_url = self.url self.url = new_url self.connected = False # Reset bitrate EMA on every URL change so stale data never carries over self._smoothed_output_bitrate = None self._last_bitrate_db_save_time = 0 self._bitrate_warmup_samples = 10 # Update stream ID if provided if stream_id: old_stream_id = self.current_stream_id self.current_stream_id = stream_id # Add stream ID to tried streams for proper tracking self.tried_stream_ids.add(stream_id) logger.info(f"Updated stream ID from {old_stream_id} to {stream_id} for channel {self.channel_id}") # Reset retry counter to allow immediate reconnect self.retry_count = 0 # Also reset buffer position to prevent stale data after URL change if hasattr(self.buffer, 'reset_buffer_position'): try: self.buffer.reset_buffer_position() logger.debug("Reset buffer position for clean URL switch") except Exception as e: logger.warning(f"Failed to reset buffer position: {e}") # Log stream switch event try: log_system_event( 'stream_switch', channel_id=self.channel_id, channel_name=self.channel_name, new_url=new_url[:100] if new_url else None, stream_id=stream_id ) except Exception as e: logger.error(f"Could not log stream switch event: {e}") return True except Exception as e: logger.error(f"Error during URL update for channel {self.channel_id}: {e}", exc_info=True) return False finally: # Always reset the URL switching flag when done, whether successful or not self.url_switching = False logger.info(f"Stream switch completed for channel {self.channel_id}") def should_retry(self) -> bool: """Check if connection retry is allowed""" return self.retry_count < self.max_retries def _health_inactivity_threshold(self): """How long without data before marking the stream unhealthy.""" if self.connected and getattr(self.buffer, 'index', 0) == 0: return ConfigHelper.channel_init_grace_period() return getattr(Config, 'CONNECTION_TIMEOUT', 10) def _monitor_health(self): """Monitor stream health and set flags for the main loop to handle recovery""" consecutive_unhealthy_checks = 0 max_unhealthy_checks = 3 # Add flags for the main loop to check self.needs_reconnect = False self.needs_stream_switch = False self.last_health_action_time = 0 action_cooldown = 30 # Prevent rapid recovery attempts while self.running: try: now = time.time() inactivity_duration = now - self.last_data_time timeout_threshold = self._health_inactivity_threshold() if inactivity_duration > timeout_threshold and self.connected: if self.healthy: logger.warning(f"Stream unhealthy for channel {self.channel_id} - no data for {inactivity_duration:.1f}s") self.healthy = False consecutive_unhealthy_checks += 1 # Only set flags if enough time has passed since last action if (consecutive_unhealthy_checks >= max_unhealthy_checks and now - self.last_health_action_time > action_cooldown): # Calculate stability to decide on action type connection_start_time = getattr(self, 'connection_start_time', 0) stable_time = self.last_data_time - connection_start_time if connection_start_time > 0 else 0 if stable_time >= 30: # Stream was stable, try reconnect first if not self.needs_reconnect: logger.info(f"Setting reconnect flag for stable stream (stable for {stable_time:.1f}s) for channel {self.channel_id}") self.needs_reconnect = True self.last_health_action_time = now else: # Stream wasn't stable, suggest stream switch if not self.needs_stream_switch: logger.info(f"Setting stream switch flag for unstable stream (stable for {stable_time:.1f}s) for channel {self.channel_id}") self.needs_stream_switch = True self.last_health_action_time = now consecutive_unhealthy_checks = 0 # Reset after setting flag elif self.connected and not self.healthy: # Auto-recover health when data resumes logger.info(f"Stream health restored for channel {self.channel_id} - data resumed after {inactivity_duration:.1f}s") self.healthy = True consecutive_unhealthy_checks = 0 # Clear recovery flags when healthy again self.needs_reconnect = False self.needs_stream_switch = False if self.healthy: consecutive_unhealthy_checks = 0 except Exception as e: logger.error(f"Error in health monitor: {e}") gevent.sleep(self.health_check_interval) def _attempt_reconnect(self): """Attempt to reconnect to the current stream""" try: logger.info(f"Attempting reconnect to current stream for channel {self.channel_id}") # Don't try to reconnect if we're already switching URLs if self.url_switching: logger.info(f"URL switching already in progress, skipping reconnect for channel {self.channel_id}") return False # Set a flag to prevent concurrent operations if hasattr(self, 'reconnecting') and self.reconnecting: logger.info(f"Reconnect already in progress, skipping for channel {self.channel_id}") return False self.reconnecting = True try: # Close existing connection and wait for it to fully terminate if self.transcode or self.socket: logger.debug(f"Closing transcode process before reconnect for channel {self.channel_id}") self._close_socket() else: logger.debug(f"Closing HTTP connection before reconnect for channel {self.channel_id}") self._close_connection() # Wait for all processes to fully close before attempting reconnect if not self._wait_for_existing_processes_to_close(): logger.warning(f"Some processes may still be running during reconnect for channel {self.channel_id}") self.connected = False # Attempt to establish a new connection using the same URL connection_result = False if self.transcode: connection_result = self._establish_transcode_connection() else: connection_result = self._establish_http_connection() if connection_result: self.connection_start_time = time.time() logger.info(f"Reconnect successful for channel {self.channel_id}") # Log reconnection event try: log_system_event( 'channel_reconnect', channel_id=self.channel_id, channel_name=self.channel_name, reason='health_monitor' ) except Exception as e: logger.error(f"Could not log reconnection event: {e}") return True else: logger.warning(f"Reconnect failed for channel {self.channel_id}") return False finally: self.reconnecting = False except Exception as e: logger.error(f"Error in reconnect attempt for channel {self.channel_id}: {e}", exc_info=True) self.reconnecting = False return False def _attempt_health_recovery(self): """Attempt to recover stream health by switching to another stream""" try: logger.info(f"Attempting health recovery for channel {self.channel_id}") # Don't try to switch if we're already in the process of switching URLs if self.url_switching: logger.info(f"URL switching already in progress, skipping health recovery for channel {self.channel_id}") return # Try to switch to next stream switch_result = self._try_next_stream() if switch_result: logger.info(f"Health recovery successful - switched to new stream for channel {self.channel_id}") return True else: logger.warning(f"Health recovery failed - no alternative streams available for channel {self.channel_id}") return False except Exception as e: logger.error(f"Error in health recovery attempt for channel {self.channel_id}: {e}", exc_info=True) return False def _close_connection(self): """Close HTTP connection resources""" # Close response if it exists if hasattr(self, 'current_response') and self.current_response: try: self.current_response.close() except Exception as e: logger.debug(f"Error closing response for channel {self.channel_id}: {e}") self.current_response = None # Close session if it exists if hasattr(self, 'current_session') and self.current_session: try: self.current_session.close() except Exception as e: logger.debug(f"Error closing session for channel {self.channel_id}: {e}") self.current_session = None def _close_socket(self): """Close socket and transcode resources as needed""" # First try to use _close_connection for HTTP resources if self.current_response or self.current_session: self._close_connection() # Stop HTTP reader thread if it exists if hasattr(self, 'http_reader') and self.http_reader: try: logger.debug(f"Stopping HTTP reader thread for channel {self.channel_id}") self.http_reader.stop() self.http_reader = None except Exception as e: logger.debug(f"Error stopping HTTP reader for channel {self.channel_id}: {e}") # 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}") proc.kill() # Give it a very short time to die try: 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: logger.debug(f"Error terminating transcode process for channel {self.channel_id}: {e}") # Final attempt: try to kill directly try: 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 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}") # Join stderr reader thread to ensure it's fully terminated if hasattr(self, 'stderr_reader_thread') and self.stderr_reader_thread and self.stderr_reader_thread.is_alive(): try: logger.debug(f"Waiting for stderr reader thread to terminate for channel {self.channel_id}") stderr_join_timeout = 0.25 if self.stopping else 2.0 self.stderr_reader_thread.join(timeout=stderr_join_timeout) if self.stderr_reader_thread.is_alive(): logger.warning(f"Stderr reader thread did not terminate within timeout for channel {self.channel_id}") except Exception as e: logger.debug(f"Error joining stderr reader thread for channel {self.channel_id}: {e}") finally: self.stderr_reader_thread = None self.transcode_process_active = False # Clear transcode active key in Redis if available if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: try: transcode_key = RedisKeys.transcode_active(self.channel_id) self.buffer.redis_client.delete(transcode_key) logger.debug(f"Cleared transcode active flag for channel {self.channel_id}") except Exception as e: logger.debug(f"Error clearing transcode flag for channel {self.channel_id}: {e}") self.socket = None self.connected = False # Cancel any remaining buffer check timers for timer in list(self._buffer_check_timers): try: if timer and timer.is_alive(): timer.cancel() logger.debug(f"Cancelled buffer check timer during socket close for channel {self.channel_id}") except Exception as e: logger.debug(f"Error canceling timer during socket close for channel {self.channel_id}: {e}") self._buffer_check_timers = [] def fetch_chunk(self): """Fetch data from socket with timeout handling""" if not self.connected or not self.socket: return False try: # Set timeout for chunk reads chunk_timeout = ConfigHelper.chunk_timeout() # Use centralized timeout configuration try: # Handle different socket types with timeout if hasattr(self.socket, 'recv'): # Standard socket - set timeout original_timeout = self.socket.gettimeout() self.socket.settimeout(chunk_timeout) chunk = self.socket.recv(Config.CHUNK_SIZE) self.socket.settimeout(original_timeout) # Restore original timeout else: # 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: logger.debug(f"Chunk read timeout ({chunk_timeout}s) for channel {self.channel_id}") return False 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 except socket.timeout: # Socket timeout occurred logger.debug(f"Socket timeout ({chunk_timeout}s) for channel {self.channel_id}") return False if not chunk: # Connection closed by server logger.warning(f"Server closed connection for channel {self.channel_id}") self._close_socket() self.connected = False return False if not self._upstream_may_continue(): self.stop() return False # Track chunk size before adding to buffer chunk_size = len(chunk) self._update_bytes_processed(chunk_size) # Add directly to buffer without TS-specific processing success = self.buffer.add_chunk(chunk) if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: last_data_key = RedisKeys.last_data(self.buffer.channel_id) self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) return True except (socket.timeout, socket.error) as e: # Socket error logger.error(f"Socket error: {e}") self._close_socket() self.connected = False return False except Exception as e: logger.error(f"Error in fetch_chunk: {e}") return False def _set_waiting_for_clients(self): """Set channel state to waiting for clients AFTER buffer has enough chunks""" try: if hasattr(self.buffer, 'channel_id') and hasattr(self.buffer, 'redis_client'): channel_id = self.buffer.channel_id redis_client = self.buffer.redis_client if channel_id and redis_client: current_time = str(time.time()) metadata_key = RedisKeys.channel_metadata(channel_id) # Check current state first current_state = None try: metadata = redis_client.hgetall(metadata_key) state_field = ChannelMetadataField.STATE if metadata and state_field in metadata: current_state = metadata[state_field] except Exception as e: logger.error(f"Error checking current state: {e}") # Only update if not already past connecting if not current_state or current_state in [ChannelState.INITIALIZING, ChannelState.CONNECTING]: # NEW CODE: Check if buffer has enough chunks # IMPORTANT: Read from Redis, not local buffer.index, because in multi-worker setup # each worker has its own StreamBuffer instance with potentially stale local index buffer_index_key = RedisKeys.buffer_index(channel_id) current_buffer_index = 0 try: redis_index = redis_client.get(buffer_index_key) if redis_index: current_buffer_index = int(redis_index) except Exception as e: logger.error(f"Error reading buffer index from Redis: {e}") initial_chunks_needed = ConfigHelper.initial_behind_chunks() if current_buffer_index < initial_chunks_needed: # Not enough buffer yet - set to connecting state if not already if current_state != ChannelState.CONNECTING: update_data = { ChannelMetadataField.STATE: ChannelState.CONNECTING, ChannelMetadataField.STATE_CHANGED_AT: current_time } redis_client.hset(metadata_key, mapping=update_data) logger.info(f"Channel {channel_id} connected but waiting for buffer to fill: {current_buffer_index}/{initial_chunks_needed} chunks") # Schedule a retry to check buffer status again timer = threading.Timer(0.5, self._check_buffer_and_set_state) timer.daemon = True timer.start() return False from ..services.channel_service import ChannelService ChannelService.promote_channel_when_buffer_ready(channel_id) else: logger.debug(f"Not changing state: channel {channel_id} already in {current_state} state") except Exception as e: logger.error(f"Error setting waiting for clients state for channel {channel_id}: {e}") def _check_buffer_and_set_state(self): """Check buffer size and set state to waiting_for_clients when ready""" try: # Enhanced stop detection with short-circuit return if not self.running or getattr(self, 'stopping', False) or getattr(self, 'reconnecting', False): logger.debug(f"Buffer check aborted - channel {self.buffer.channel_id} is stopping or reconnecting") return False # Return value to indicate check was aborted # Clean up completed timers self._buffer_check_timers = [t for t in self._buffer_check_timers if t.is_alive()] if hasattr(self.buffer, 'channel_id') and hasattr(self.buffer, 'redis_client'): channel_id = self.buffer.channel_id redis_client = self.buffer.redis_client # IMPORTANT: Read from Redis, not local buffer.index buffer_index_key = RedisKeys.buffer_index(channel_id) current_buffer_index = 0 try: redis_index = redis_client.get(buffer_index_key) if redis_index: current_buffer_index = int(redis_index) except Exception as e: logger.error(f"Error reading buffer index from Redis: {e}") initial_chunks_needed = ConfigHelper.initial_behind_chunks() # Use ConfigHelper for consistency if current_buffer_index >= initial_chunks_needed: # We now have enough buffer, call _set_waiting_for_clients again logger.info(f"Buffer threshold reached for channel {channel_id}: {current_buffer_index}/{initial_chunks_needed} chunks") self._set_waiting_for_clients() else: # Still waiting, log progress and schedule another check logger.debug(f"Buffer filling for channel {channel_id}: {current_buffer_index}/{initial_chunks_needed} chunks") # Schedule another check - NOW WITH STOPPING CHECK if self.running and not getattr(self, 'stopping', False): timer = threading.Timer(0.5, self._check_buffer_and_set_state) timer.daemon = True timer.start() self._buffer_check_timers.append(timer) return True # Return value to indicate check was successful except Exception as e: logger.error(f"Error in buffer check for channel {self.channel_id}: {e}") return False def _try_next_stream(self): """ Try to switch to the next available stream for this channel. Will iterate through multiple alternate streams if needed to find one with a different URL. Returns: bool: True if successfully switched to a new stream, False otherwise """ try: logger.info(f"Trying to find alternative stream for channel {self.channel_id}, current stream ID: {self.current_stream_id}") # Get alternate streams excluding the current one alternate_streams = get_alternate_streams(self.channel_id, self.current_stream_id) logger.info(f"Found {len(alternate_streams)} potential alternate streams for channel {self.channel_id}") # Filter out streams we've already tried untried_streams = [s for s in alternate_streams if s['stream_id'] not in self.tried_stream_ids] if untried_streams: ids_to_try = ', '.join([str(s['stream_id']) for s in untried_streams]) logger.info(f"Found {len(untried_streams)} untried streams for channel {self.channel_id}: [{ids_to_try}]") else: logger.warning(f"No untried streams available for channel {self.channel_id}, tried: {self.tried_stream_ids}") if not untried_streams: # Check if we have streams but they've all been tried if alternate_streams and len(self.tried_stream_ids) > 0: logger.warning(f"All {len(alternate_streams)} alternate streams have been tried for channel {self.channel_id}") return False for next_stream in untried_streams: stream_id = next_stream['stream_id'] profile_id = next_stream['profile_id'] # This is the M3U profile ID we need # Add to tried streams self.tried_stream_ids.add(stream_id) # Get stream info including URL using the profile_id we already have logger.info(f"Trying next stream ID {stream_id} with profile ID {profile_id} for channel {self.channel_id}") stream_info = get_stream_info_for_switch(self.channel_id, stream_id) if 'error' in stream_info or not stream_info.get('url'): logger.error(f"Error getting info for stream {stream_id} for channel {self.channel_id}: {stream_info.get('error', 'No URL')}") continue # Try next stream instead of giving up # Update URL and user agent new_url = stream_info['url'] new_user_agent = stream_info['user_agent'] new_transcode = stream_info['transcode'] # Check if the new URL is the same as current URL # This can happen when current_stream_id is None and we accidentally select the same stream if new_url == self.url: logger.warning(f"Stream ID {stream_id} generates the same URL as current stream ({new_url}). " f"Skipping this stream and trying next alternative.") continue # Try next stream instead of giving up logger.info(f"Switching from URL {self.url} to {new_url} for channel {self.channel_id}") # Just update the URL, don't stop the channel or release resources switch_result = self.update_url(new_url, stream_id, profile_id) if not switch_result: logger.error(f"Failed to update URL for stream ID {stream_id} for channel {self.channel_id}") continue # Try next stream # Update stream ID tracking self.current_stream_id = stream_id # Store the new user agent and transcode settings self.user_agent = new_user_agent self.transcode = new_transcode # Update stream metadata in Redis - use the profile_id we got from get_alternate_streams if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: metadata_key = RedisKeys.channel_metadata(self.channel_id) self.buffer.redis_client.hset(metadata_key, mapping={ ChannelMetadataField.URL: new_url, ChannelMetadataField.USER_AGENT: new_user_agent, ChannelMetadataField.STREAM_PROFILE: stream_info['stream_profile'], ChannelMetadataField.M3U_PROFILE: str(profile_id), # Use the profile_id from get_alternate_streams ChannelMetadataField.STREAM_ID: str(stream_id), ChannelMetadataField.STREAM_SWITCH_TIME: str(time.time()), ChannelMetadataField.STREAM_SWITCH_REASON: "max_retries_exceeded" }) # Log the switch logger.info(f"Stream metadata updated for channel {self.channel_id} to stream ID {stream_id} with M3U profile {profile_id}") logger.info(f"Successfully switched to stream ID {stream_id} with URL {new_url} for channel {self.channel_id}") return True # If we get here, we tried all streams but none worked logger.error(f"Tried {len(untried_streams)} alternate streams but none were suitable for channel {self.channel_id}") return False except Exception as e: logger.error(f"Error trying next stream for channel {self.channel_id}: {e}", exc_info=True) return False # Add a new helper method to safely reset the URL switching state def _reset_url_switching_state(self): """Safely reset the URL switching state if it gets stuck""" self.url_switching = False self.url_switch_start_time = 0 logger.info(f"Reset URL switching state for channel {self.channel_id}")