From a2c7fc3046204a4fc067b72edb821616824166f8 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 10 Jun 2025 17:43:37 -0500 Subject: [PATCH 1/4] [New feature] Switch streams when buffering is detected. --- apps/proxy/config.py | 2 ++ apps/proxy/ts_proxy/config_helper.py | 8 +++++ apps/proxy/ts_proxy/constants.py | 1 + apps/proxy/ts_proxy/stream_manager.py | 48 +++++++++++++++++++++++++-- 4 files changed, 57 insertions(+), 2 deletions(-) diff --git a/apps/proxy/config.py b/apps/proxy/config.py index b00bd224..b369a92f 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -11,6 +11,8 @@ class BaseConfig: BUFFER_CHUNK_SIZE = 188 * 1361 # ~256KB # Redis settings REDIS_CHUNK_TTL = 60 # Number in seconds - Chunks expire after 1 minute + BUFFERING_TIMEOUT = 15 # Seconds to wait for buffering before switching streams + BUFFER_SPEED = 1 # What speed to condsider the stream buffering, 1x is normal speed, 2x is double speed, etc. class HLSConfig(BaseConfig): MIN_SEGMENTS = 12 diff --git a/apps/proxy/ts_proxy/config_helper.py b/apps/proxy/ts_proxy/config_helper.py index 773ab378..4057a2d5 100644 --- a/apps/proxy/ts_proxy/config_helper.py +++ b/apps/proxy/ts_proxy/config_helper.py @@ -85,3 +85,11 @@ class ConfigHelper: def failover_grace_period(): """Get extra time (in seconds) to allow for stream switching before disconnecting clients""" return ConfigHelper.get('FAILOVER_GRACE_PERIOD', 20) # Default to 20 seconds + @staticmethod + def buffering_timeout(): + """Get buffering timeout in seconds""" + return ConfigHelper.get('BUFFERING_TIMEOUT', 15) # Default to 15 seconds + @staticmethod + def buffering_speed(): + """Get buffering speed in bytes per second""" + return ConfigHelper.get('BUFFERING_SPEED',1) # Default to 1x diff --git a/apps/proxy/ts_proxy/constants.py b/apps/proxy/ts_proxy/constants.py index 385d17c1..55d6e006 100644 --- a/apps/proxy/ts_proxy/constants.py +++ b/apps/proxy/ts_proxy/constants.py @@ -18,6 +18,7 @@ class ChannelState: ERROR = "error" STOPPING = "stopping" STOPPED = "stopped" + BUFFERING = "buffering" # Event types class EventType: diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index 6e3c9e73..f8a7323b 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -40,6 +40,10 @@ class StreamManager: 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 @@ -545,7 +549,6 @@ class StreamManager: 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_bitrate]): self._update_ffmpeg_stats_in_redis(ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_bitrate) @@ -553,10 +556,51 @@ class StreamManager: # Fix the f-string formatting actual_fps_str = f"{actual_fps:.1f}" if actual_fps is not None else "N/A" ffmpeg_bitrate_str = f"{ffmpeg_bitrate:.1f}" if ffmpeg_bitrate is not None else "N/A" - + # Log the stats logger.debug(f"FFmpeg stats - Speed: {ffmpeg_speed}x, FFmpeg FPS: {ffmpeg_fps}, " f"Actual FPS: {actual_fps_str}, " f"Bitrate: {ffmpeg_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 + 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 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}") From e753d9b9f810f9b4933257cb9941512be2c9bd55 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 10 Jun 2025 19:16:52 -0500 Subject: [PATCH 2/4] Fixes a bug where stream profile name wouldn't update in stats. (Was outputting name string instead of ID --- apps/proxy/ts_proxy/url_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py index e3b1c264..dbd3c5dd 100644 --- a/apps/proxy/ts_proxy/url_utils.py +++ b/apps/proxy/ts_proxy/url_utils.py @@ -172,7 +172,7 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] # Get transcode info from the channel's stream profile stream_profile = channel.get_stream_profile() transcode = not (stream_profile.is_proxy() or stream_profile is None) - profile_value = str(stream_profile) + profile_value = stream_profile.id return { 'url': stream_url, From 743cf4e297566cef9ef4cf59e4cdd10baf572fd2 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Wed, 11 Jun 2025 16:55:14 -0500 Subject: [PATCH 3/4] Smarter parsing of ffmpeg stats output. --- apps/proxy/ts_proxy/stream_manager.py | 143 ++++++++++++++------------ 1 file changed, 76 insertions(+), 67 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index f8a7323b..c413a9f5 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -386,90 +386,99 @@ class StreamManager: buffer = b"" last_stats_line = b"" - # Read in small chunks + # Read byte by byte for immediate detection while self.transcode_process and self.transcode_process.stderr: try: - chunk = self.transcode_process.stderr.read(256) # Smaller chunks for real-time processing - if not chunk: + # Read one byte at a time for immediate processing + byte = self.transcode_process.stderr.read(1) + if not byte: break - buffer += chunk + buffer += byte - # Look for stats updates (overwrite previous stats with \r) - if b'\r' in buffer and b"frame=" in buffer: - # Split on \r to handle overwriting stats - parts = buffer.split(b'\r') + # 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 - # Process all parts except the last (which might be incomplete) - for i, part in enumerate(parts[:-1]): - if part.strip(): - if part.startswith(b"frame=") or b"frame=" in part: - # This is a stats line - keep it intact - try: - stats_text = part.decode('utf-8', errors='ignore').strip() - if stats_text and "frame=" in stats_text: - # Extract just the stats portion if there's other content - if "frame=" in stats_text: - frame_start = stats_text.find("frame=") - stats_text = stats_text[frame_start:] + buffer += next_byte - self._parse_ffmpeg_stats(stats_text) - self._log_stderr_content(stats_text) - last_stats_line = part - except Exception as e: - logger.debug(f"Error parsing stats line: {e}") - else: - # Regular content - process line by line - line_content = part - while b'\n' in line_content: - line, line_content = line_content.split(b'\n', 1) - if line.strip(): - self._log_stderr_content(line.decode('utf-8', errors='ignore')) + # Break on carriage return (stats overwrite) or newline + if next_byte in (b'\r', b'\n'): + break - # Handle remaining content without newline - if line_content.strip(): - self._log_stderr_content(line_content.decode('utf-8', errors='ignore')) + # Also break if we have enough data for a typical stats line + if len(buffer) > 200: # Typical stats line length + break - # Keep the last part as it might be incomplete - buffer = parts[-1] + # 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 b'\n' in buffer: - while b'\n' in buffer: - line, buffer = buffer.split(b'\n', 1) - if line.strip(): - line_text = line.decode('utf-8', errors='ignore').strip() - if line_text and not line_text.startswith("frame="): - self._log_stderr_content(line_text) + 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"" - # If we have a potential stats line in buffer without line breaks - elif b"frame=" in buffer and (b"speed=" in buffer or len(buffer) > 200): - # We likely have a complete or substantial stats line - try: - stats_text = buffer.decode('utf-8', errors='ignore').strip() - if "frame=" in stats_text: - # Extract just the stats portion - frame_start = stats_text.find("frame=") - stats_text = stats_text[frame_start:] + # 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"" - self._parse_ffmpeg_stats(stats_text) - self._log_stderr_content(stats_text) - buffer = b"" # Clear buffer after processing - except Exception as e: - logger.debug(f"Error parsing buffered stats: {e}") - - # Prevent buffer from growing too large - if len(buffer) > 4096: - # Try to preserve any potential stats line at the end - if b"frame=" in buffer[-1024:]: - buffer = buffer[-1024:] - else: - buffer = buffer[-512:] + # 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: {e}") + logger.error(f"Error reading stderr byte: {e}") break + # Process any remaining buffer content + if buffer.strip(): + try: + remaining_text = buffer.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}") + except Exception as e: # Catch any other exceptions in the thread to prevent crashes try: From bd53837f804736bd9eca6ac4c4a51da32786e634 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Wed, 11 Jun 2025 17:20:27 -0500 Subject: [PATCH 4/4] Better detection of input vs output stream information. --- apps/proxy/ts_proxy/stream_manager.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index c413a9f5..a57f1384 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -497,13 +497,18 @@ class StreamManager: content_lower = content.lower() # Check for stream info lines first and delegate to ChannelService + # Only parse INPUT streams (which have hex identifiers like [0x100]) not output streams if "stream #" in content_lower and ("video:" in content_lower or "audio:" in content_lower): - from .services.channel_service import ChannelService - if "video:" in content_lower: - ChannelService.parse_and_store_stream_info(self.channel_id, content, "video") - elif "audio:" in content_lower: - ChannelService.parse_and_store_stream_info(self.channel_id, content, "audio") - + # Check if this is an input stream by looking for the hex identifier pattern [0x...] + if "stream #0:" in content_lower and "[0x" in content_lower: + from .services.channel_service import ChannelService + if "video:" in content_lower: + ChannelService.parse_and_store_stream_info(self.channel_id, content, "video") + elif "audio:" in content_lower: + ChannelService.parse_and_store_stream_info(self.channel_id, content, "audio") + else: + # This is likely an output stream (no hex identifier), don't parse it + logger.debug(f"Skipping output stream info: {content}") # Determine log level based on content if any(keyword in content_lower for keyword in ['error', 'failed', 'cannot', 'invalid', 'corrupt']): logger.error(f"FFmpeg stderr: {content}")