diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 20eaca70..ea3dfd95 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -497,13 +497,12 @@ class ProxyServer: """Stop a channel with proper ownership handling""" try: logger.info(f"Stopping channel {channel_id}") - + # First set a stopping key that clients will check if self.redis_client: stop_key = f"ts_proxy:channel:{channel_id}:stopping" - # Set with 60 second TTL - enough time for clients to notice self.redis_client.setex(stop_key, 10, "true") - + # Only stop the actual stream manager if we're the owner if self.am_i_owner(channel_id): logger.info(f"This worker ({self.worker_id}) is the owner - closing provider connection") @@ -541,22 +540,41 @@ class ProxyServer: self.release_ownership(channel_id) logger.info(f"Released ownership of channel {channel_id}") - # Always clean up local resources + # Always clean up local resources - WITH SAFE CHECKS if channel_id in self.stream_managers: del self.stream_managers[channel_id] logger.info(f"Removed stream manager for channel {channel_id}") - + + # Stop buffer and ensure all its timers are cancelled - SAFE CHECK HERE if channel_id in self.stream_buffers: - del self.stream_buffers[channel_id] - logger.info(f"Removed stream buffer for channel {channel_id}") - + buffer = self.stream_buffers[channel_id] + # Call stop on buffer to properly shut it down + if hasattr(buffer, 'stop'): + try: + buffer.stop() + logger.debug(f"Buffer for channel {channel_id} properly stopped") + except Exception as e: + logger.error(f"Error stopping buffer: {e}") + + # Save reference and check again before deleting + try: + if channel_id in self.stream_buffers: # Check again to prevent race conditions + del self.stream_buffers[channel_id] + logger.info(f"Removed stream buffer for channel {channel_id}") + except KeyError: + logger.debug(f"Buffer for channel {channel_id} already removed") + + # Clean up client manager - SAFE CHECK HERE TOO if channel_id in self.client_managers: - del self.client_managers[channel_id] - logger.info(f"Removed client manager for channel {channel_id}") - + try: + del self.client_managers[channel_id] + logger.info(f"Removed client manager for channel {channel_id}") + except KeyError: + logger.debug(f"Client manager for channel {channel_id} already removed") + # Clean up Redis keys self._clean_redis_keys(channel_id) - + return True except Exception as e: logger.error(f"Error stopping channel {channel_id}: {e}", exc_info=True) diff --git a/apps/proxy/ts_proxy/stream_buffer.py b/apps/proxy/ts_proxy/stream_buffer.py index c1d11b71..4f1bcb64 100644 --- a/apps/proxy/ts_proxy/stream_buffer.py +++ b/apps/proxy/ts_proxy/stream_buffer.py @@ -39,6 +39,10 @@ class StreamBuffer: self._write_buffer = bytearray() self.target_chunk_size = getattr(Config, 'BUFFER_CHUNK_SIZE', 188 * 5644) # ~1MB default + # Track timers for proper cleanup + self.stopping = False + self.fill_timers = [] + def add_chunk(self, chunk): """Add data with optimized Redis storage and TS packet alignment""" if not chunk: @@ -215,7 +219,26 @@ class StreamBuffer: return [] def stop(self): - """Stop the buffer and flush any remaining data""" + """Stop the buffer and cancel all timers""" + # Set stopping flag first to prevent new timer creation + self.stopping = True + + # Cancel all pending timers + timers_cancelled = 0 + for timer in list(self.fill_timers): + try: + if timer and timer.is_alive(): + timer.cancel() + timers_cancelled += 1 + except Exception as e: + logger.error(f"Error canceling timer: {e}") + + if timers_cancelled: + logger.info(f"Cancelled {timers_cancelled} buffer timers for channel {self.channel_id}") + + # Clear timer list + self.fill_timers.clear() + try: # Flush any remaining data in the write buffer if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0: @@ -285,4 +308,16 @@ class StreamBuffer: chunks.extend(more_chunks) chunk_count += additional - return chunks, client_index + chunk_count \ No newline at end of file + return chunks, client_index + chunk_count + + # Add a new method to safely create timers + def schedule_timer(self, delay, callback, *args, **kwargs): + """Schedule a timer and track it for proper cleanup""" + if self.stopping: + return None + + timer = threading.Timer(delay, callback, args=args, kwargs=kwargs) + timer.daemon = True + timer.start() + self.fill_timers.append(timer) + return timer diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index 9e0840bd..48d2b8a5 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -45,6 +45,10 @@ class StreamManager: self.health_check_interval = Config.HEALTH_CHECK_INTERVAL self.chunk_size = getattr(Config, 'CHUNK_SIZE', 8192) + # Add to your __init__ method + self._buffer_check_timers = [] + self.stopping = False + logger.info(f"Initialized stream manager for channel {buffer.channel_id}") def _create_session(self): @@ -221,8 +225,23 @@ class StreamManager: break timeout = min(2 ** self.retry_count, 30) + + # When a connection fails and reconnect is needed: + self.reconnecting = True + + # Cancel all existing buffer timers during reconnect + 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 timer: {e}") + self._buffer_check_timers = [] + logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})") time.sleep(timeout) + + self.reconnecting = False # Reset flag after sleep except Exception as e: logger.error(f"Stream error: {e}", exc_info=True) @@ -250,7 +269,21 @@ class StreamManager: logger.info(f"Stream manager stopped") def stop(self): - """Stop this stream""" + """Stop the stream manager and cancel all timers""" + # Add at the beginning of your stop method + self.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: {e}") + + self._buffer_check_timers.clear() + + # Rest of your existing stop method... # Set the flag first self.stop_requested = True @@ -488,8 +521,14 @@ class StreamManager: def _check_buffer_and_set_state(self): """Check buffer size and set state to waiting_for_clients when ready""" try: - # This method will be called asynchronously to check buffer status - # and update state when enough chunks are available + # First check if we're stopping or reconnecting + if 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 + + # Clean up completed timers + self._buffer_check_timers = [t for t in self._buffer_check_timers if t.is_alive()] + if hasattr(self.buffer, 'index') and hasattr(self.buffer, 'channel_id'): current_buffer_index = self.buffer.index initial_chunks_needed = getattr(Config, 'INITIAL_BEHIND_CHUNKS', 10) @@ -502,13 +541,13 @@ class StreamManager: 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") - if current_buffer_index > 0 and current_buffer_index % 5 == 0: - # Log less frequently to avoid spamming logs - logger.info(f"Buffer filling for channel {channel_id}: {current_buffer_index}/{initial_chunks_needed} chunks") - - # Schedule another check - timer = threading.Timer(0.5, self._check_buffer_and_set_state) - timer.daemon = True - timer.start() + + # Schedule another check - NOW WITH TRACKING + if 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) except Exception as e: logger.error(f"Error in buffer check: {e}") +