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}")