From 987a8606a847a44ebfb633ebc080947ec0e27b7e Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 13 Mar 2025 09:48:37 -0500 Subject: [PATCH 1/7] Removed packet processing. --- apps/proxy/ts_proxy/server.py | 209 ++++++++++------------------------ apps/proxy/ts_proxy/views.py | 70 ++++++------ 2 files changed, 97 insertions(+), 182 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 12ceb045..88043b15 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -46,12 +46,6 @@ class StreamManager: # User agent for connection self.user_agent = user_agent or Config.DEFAULT_USER_AGENT - # TS packet handling - self.TS_PACKET_SIZE = 188 - self.recv_buffer = bytearray() - self.sync_found = False - self.continuity_counters = {} - # Stream health monitoring self.last_data_time = time.time() self.healthy = True @@ -95,113 +89,6 @@ class StreamManager: self._close_socket() logging.info("Stream manager resources released") - def _process_complete_packets(self): - """Process TS packets with improved resync capability""" - try: - # Enhanced sync byte detection with re-sync capability - if (not self.sync_found or - (len(self.recv_buffer) >= 188 and self.recv_buffer[0] != 0x47)): - - # Need to find sync pattern if we haven't found it yet or lost sync - if len(self.recv_buffer) >= 376: # Need at least 2 packet lengths - sync_found = False - - # Look for at least two sync bytes (0x47) at 188-byte intervals - for i in range(min(188, len(self.recv_buffer) - 188)): - if (self.recv_buffer[i] == 0x47 and - self.recv_buffer[i + 188] == 0x47): - - # If already had sync but lost it, log the issue - if self.sync_found: - logging.warning(f"Re-syncing TS stream at position {i} (lost sync)") - else: - logging.debug(f"TS sync found at position {i}") - - # Trim buffer to start at first sync byte - self.recv_buffer = self.recv_buffer[i:] - self.sync_found = True - sync_found = True - break - - # If we couldn't find sync in this buffer, discard partial data - if not sync_found: - logging.warning(f"Failed to find sync pattern - discarding {len(self.recv_buffer) - 188} bytes") - if len(self.recv_buffer) > 188: - self.recv_buffer = self.recv_buffer[-188:] # Keep last chunk for next attempt - return False - - # If we don't have a complete packet yet, wait for more data - if len(self.recv_buffer) < 188: - return False - - # Calculate how many complete packets we have - packet_count = len(self.recv_buffer) // 188 - - if packet_count == 0: - return False - - # Verify all packets have sync bytes - all_synced = True - for i in range(0, packet_count): - if self.recv_buffer[i * 188] != 0x47: - all_synced = False - break - - # If not all packets are synced, re-scan for sync - if not all_synced: - self.sync_found = False # Force re-sync on next call - return False - - # Extract complete packets - packets = self.recv_buffer[:packet_count * 188] - - # Keep remaining data in buffer - self.recv_buffer = self.recv_buffer[packet_count * 188:] - - # Send packets to buffer - if packets: - # Log first and last sync byte to validate alignment - first_sync = packets[0] if len(packets) > 0 else None - last_sync = packets[188 * (packet_count - 1)] if packet_count > 0 else None - - if first_sync != 0x47 or last_sync != 0x47: - logging.warning(f"TS packet alignment issue: first_sync=0x{first_sync:02x}, last_sync=0x{last_sync:02x}") - # Don't process misaligned packets - return False - - before_index = self.buffer.index - success = self.buffer.add_chunk(bytes(packets)) - after_index = self.buffer.index - - # Log successful write - if not success: - logging.warning("Failed to add chunk to buffer") - - # If successful, update last data timestamp in Redis - if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: - last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data" - self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) # 1 minute expiry - - return success - - return False - - except Exception as e: - logging.error(f"Error processing TS packets: {e}", exc_info=True) - self.sync_found = False # Reset sync state on error - return False - - def _process_ts_data(self, chunk): - """Process received data and add to buffer""" - if not chunk: - return False - - # Add to existing buffer - self.recv_buffer.extend(chunk) - - # Process complete packets now - return self._process_complete_packets() - def run(self): """Main execution loop with stream health monitoring""" try: @@ -348,22 +235,22 @@ class StreamManager: self.connected = False def fetch_chunk(self): - """Fetch data from socket with improved buffer management""" + """Fetch data from socket with direct pass-through to buffer""" if not self.connected or not self.socket: return False try: - # SocketIO objects use read instead of recv and don't support settimeout + # Read data chunk - no need to align with TS packet size anymore try: - # Try to read data chunk - use a multiple of TS packet size + # Try to read data chunk if hasattr(self.socket, 'recv'): - chunk = self.socket.recv(188 * 64) # Standard socket + chunk = self.socket.recv(Config.CHUNK_SIZE) # Standard socket else: - chunk = self.socket.read(188 * 64) # SocketIO object + chunk = self.socket.read(Config.CHUNK_SIZE) # SocketIO object except AttributeError: # Fall back to read() if recv() isn't available - chunk = self.socket.read(188 * 64) + chunk = self.socket.read(Config.CHUNK_SIZE) if not chunk: # Connection closed by server @@ -372,18 +259,13 @@ class StreamManager: self.connected = False return False - # Process this chunk - self._process_ts_data(chunk) + # Add directly to buffer without TS-specific processing + success = self.buffer.add_chunk(chunk) - # Memory management - clear any internal buffers periodically - current_time = time.time() - if current_time - self._last_buffer_check > 60: # Check every minute - self._last_buffer_check = current_time - if len(self.recv_buffer) > 188 * 1024: # If buffer is extremely large - logging.warning(f"Receive buffer unusually large ({len(self.recv_buffer)} bytes), trimming") - # Keep only recent data, aligned to TS packet boundary - keep_size = 188 * 128 # Keep reasonable buffer - self.recv_buffer = self.recv_buffer[-keep_size:] + # Update last data timestamp in Redis if successful + if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: + last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data" + self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) return True @@ -439,6 +321,37 @@ class StreamManager: except Exception as e: logging.error(f"Error setting waiting for clients state: {e}") + def _read_stream(self): + """Read from stream with minimal processing""" + try: + # Read up to CHUNK_SIZE bytes + chunk = self.sock.recv(self.CHUNK_SIZE) + + if not chunk: + # Connection closed + logging.debug("Connection closed by remote host") + return False + + # If we got data, just add it directly to the buffer + if chunk: + success = self.buffer.add_chunk(chunk) + + # Update last data timestamp in Redis if successful + if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: + last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data" + self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) # 1 minute expiry + + return success + return True + + except socket.timeout: + # Expected timeout - no data available + return True + except Exception as e: + # Error reading from socket + logging.error(f"Error reading from stream: {e}") + return False + class StreamBuffer: """Manages stream data buffering using Redis for persistence""" @@ -466,32 +379,34 @@ class StreamBuffer: logging.error(f"Error initializing buffer from Redis: {e}") def add_chunk(self, chunk): - """Add a chunk to the buffer""" + """Add a chunk to the buffer with minimal processing""" if not chunk: return False try: - # Ensure chunk is properly aligned with TS packets - if len(chunk) % self.TS_PACKET_SIZE != 0: - logging.warning(f"Received non-aligned chunk of size {len(chunk)}") - aligned_size = (len(chunk) // self.TS_PACKET_SIZE) * self.TS_PACKET_SIZE - if (aligned_size == 0): - return False - chunk = chunk[:aligned_size] - with self.lock: # Increment index atomically if self.redis_client: # Use Redis to store and track chunks - chunk_index = self.redis_client.incr(self.buffer_index_key) - chunk_key = f"{self.buffer_prefix}{chunk_index}" - self.redis_client.setex(chunk_key, self.chunk_ttl, chunk) - - # Update local tracking of position only - self.index = chunk_index - return True + try: + chunk_index = self.redis_client.incr(self.buffer_index_key) + chunk_key = f"{self.buffer_prefix}{chunk_index}" + setex_result = self.redis_client.setex(chunk_key, self.chunk_ttl, chunk) + + if not setex_result: + logging.error(f"Redis SETEX failed for chunk {chunk_index}") + return False + + # Update local tracking + self.index = chunk_index + + logging.debug(f"Added chunk {chunk_index} ({len(chunk)} bytes) to buffer") + return True + + except Exception as e: + logging.error(f"Redis operation failed in add_chunk: {e}") + return False else: - # No Redis - can't function in multi-worker mode logging.error("Redis not available, cannot store chunks") return False diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 23b41e1c..f1eecbfb 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -79,81 +79,82 @@ def initialize_stream(request, channel_id): @require_GET def stream_ts(request, channel_id): - """Stream TS data to client with improved waiting for initialization""" + """Stream TS data to client""" try: + # Generate a unique client ID + client_id = f"client_{int(time.time() * 1000)}_{random.randint(1000, 9999)}" + logger.info(f"[{client_id}] Requested stream for channel {channel_id}") + # Check if channel exists or initialize it if not proxy_server.check_if_channel_exists(channel_id): return JsonResponse({'error': 'Channel not found'}, status=404) - - # Get user agent from request headers - user_agent = None - for header in ['HTTP_USER_AGENT', 'User-Agent', 'user-agent']: - if header in request.META: - user_agent = request.META[header] - logger.debug(f"Found user agent in header: {header}") - break - - # Wait for channel to become ready if it's initializing + + # NEW: Wait for channel to become ready if it's initializing if proxy_server.redis_client: wait_start = time.time() max_wait = getattr(Config, 'CLIENT_WAIT_TIMEOUT', 30) # Maximum wait time in seconds # Check channel state metadata_key = f"ts_proxy:channel:{channel_id}:metadata" - while time.time() - wait_start < max_wait: + waiting = True + + while waiting and time.time() - wait_start < max_wait: metadata = proxy_server.redis_client.hgetall(metadata_key) if not metadata or b'state' not in metadata: - logger.warning(f"Channel {channel_id} metadata missing") + logger.warning(f"[{client_id}] Channel {channel_id} metadata missing") break state = metadata[b'state'].decode('utf-8') - # If channel is already active or waiting for clients, no need to wait + # If channel is ready for clients, continue if state in ['waiting_for_clients', 'active']: - logger.debug(f"Channel {channel_id} ready (state={state})") - break + logger.info(f"[{client_id}] Channel {channel_id} ready (state={state}), proceeding with connection") + waiting = False elif state in ['initializing', 'connecting']: # Channel is still initializing or connecting, wait a bit longer elapsed = time.time() - wait_start - logger.info(f"Client waiting for channel {channel_id} to become ready ({elapsed:.1f}s), current state: {state}") + logger.info(f"[{client_id}] Waiting for channel {channel_id} to become ready ({elapsed:.1f}s), current state: {state}") time.sleep(0.5) # Wait 500ms before checking again else: # Unknown or error state - logger.warning(f"Channel {channel_id} in unexpected state: {state}") + logger.warning(f"[{client_id}] Channel {channel_id} in unexpected state: {state}") break # Check if we timed out waiting - if time.time() - wait_start >= max_wait: - logger.warning(f"Timeout waiting for channel {channel_id} to become ready") + if waiting and time.time() - wait_start >= max_wait: + logger.warning(f"[{client_id}] Timeout waiting for channel {channel_id} to become ready") return JsonResponse({'error': 'Timeout waiting for channel to initialize'}, status=503) # CRITICAL FIX: Ensure local resources are properly initialized before streaming - # This handles the case where channel exists in Redis but not in this worker if channel_id not in proxy_server.stream_buffers or channel_id not in proxy_server.client_managers: - logger.warning(f"Channel {channel_id} exists in Redis but not initialized in this worker - initializing now") + logger.warning(f"[{client_id}] Channel {channel_id} exists in Redis but not initialized in this worker - initializing now") - # Get URL from Redis metadata if available + # Get URL from Redis metadata url = None - metadata_key = f"ts_proxy:channel:{channel_id}:metadata" if proxy_server.redis_client: + metadata_key = f"ts_proxy:channel:{channel_id}:metadata" url_bytes = proxy_server.redis_client.hget(metadata_key, "url") if url_bytes: url = url_bytes.decode('utf-8') - # Initialize local resources (won't recreate stream connection if another worker owns it) + # Initialize local resources success = proxy_server.initialize_channel(url, channel_id, user_agent) if not success: - logger.error(f"Failed to initialize channel {channel_id} locally") + logger.error(f"[{client_id}] Failed to initialize channel {channel_id} locally") return JsonResponse({'error': 'Failed to initialize channel locally'}, status=500) - logger.info(f"Successfully initialized channel {channel_id} locally") - - # Double-check after initialization - if channel_id not in proxy_server.client_managers: - logger.error(f"Critical error: Channel {channel_id} client manager still missing after initialization") - return JsonResponse({'error': 'Failed to create client manager'}, status=500) - - # Continue with normal streaming response + logger.info(f"[{client_id}] Successfully initialized channel {channel_id} locally") + + # Get stream buffer and client manager + buffer = proxy_server.stream_buffers[channel_id] + client_manager = proxy_server.client_managers[channel_id] + + # IMPORTANT: Register client BEFORE starting stream + # This ensures the channel knows there's a client before streaming starts + client_manager.add_client(client_id, user_agent) + logger.info(f"[{client_id}] Client registered with channel {channel_id}") + + # Start stream response def generate(): client_id = f"client_{int(time.time() * 1000)}_{random.randint(1000, 9999)}" stream_start_time = time.time() @@ -271,7 +272,6 @@ def stream_ts(request, channel_id): chunks = buffer.get_chunks_exact(local_index, Config.CHUNK_BATCH_SIZE) if chunks: - # Reset empty counters since we got data empty_reads = 0 consecutive_empty = 0 From 3483f01637937fb4801fab332ca389b303006a3b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 13 Mar 2025 11:11:31 -0500 Subject: [PATCH 2/7] Playing with chunk sizes with a more basic config. --- apps/proxy/ts_proxy/views.py | 45 ++++++++++++++---------------------- 1 file changed, 17 insertions(+), 28 deletions(-) diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index f1eecbfb..a0d70cec 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -79,17 +79,26 @@ def initialize_stream(request, channel_id): @require_GET def stream_ts(request, channel_id): - """Stream TS data to client""" + """Stream TS data to client with improved waiting for initialization""" try: # Generate a unique client ID client_id = f"client_{int(time.time() * 1000)}_{random.randint(1000, 9999)}" logger.info(f"[{client_id}] Requested stream for channel {channel_id}") + # Get user agent from request headers - MOVED UP TO ENSURE IT'S DEFINED + user_agent = None + for header in ['HTTP_USER_AGENT', 'User-Agent', 'user-agent']: + if header in request.META: + user_agent = request.META[header] + logger.debug(f"[{client_id}] Found user agent in header: {header}") + break + # Check if channel exists or initialize it if not proxy_server.check_if_channel_exists(channel_id): + logger.error(f"[{client_id}] Channel {channel_id} not found") return JsonResponse({'error': 'Channel not found'}, status=404) - - # NEW: Wait for channel to become ready if it's initializing + + # Wait for channel to become ready if it's initializing if proxy_server.redis_client: wait_start = time.time() max_wait = getattr(Config, 'CLIENT_WAIT_TIMEOUT', 30) # Maximum wait time in seconds @@ -137,7 +146,7 @@ def stream_ts(request, channel_id): if url_bytes: url = url_bytes.decode('utf-8') - # Initialize local resources + # Initialize local resources - pass the user_agent we extracted earlier success = proxy_server.initialize_channel(url, channel_id, user_agent) if not success: logger.error(f"[{client_id}] Failed to initialize channel {channel_id} locally") @@ -149,8 +158,7 @@ def stream_ts(request, channel_id): buffer = proxy_server.stream_buffers[channel_id] client_manager = proxy_server.client_managers[channel_id] - # IMPORTANT: Register client BEFORE starting stream - # This ensures the channel knows there's a client before streaming starts + # Register client before starting stream client_manager.add_client(client_id, user_agent) logger.info(f"[{client_id}] Client registered with channel {channel_id}") @@ -283,37 +291,18 @@ def stream_ts(request, channel_id): logger.debug(f"[{client_id}] Retrieved {len(chunks)} chunks ({total_size} bytes) from index {start_idx} to {end_idx}") - # Calculate total packet count for this batch to maintain timing - total_packets = sum(len(chunk) // ts_packet_size for chunk in chunks) - batch_start_time = time.time() - packets_sent_in_batch = 0 - - # Send chunks with pacing + # SIMPLIFIED: Just send chunks directly without pacing for chunk in chunks: - packets_in_chunk = len(chunk) // ts_packet_size bytes_sent += len(chunk) chunks_sent += 1 - # CRITICAL FIX: Detect client disconnection when yielding data try: yield chunk except Exception as e: logger.info(f"[{client_id}] Client disconnected while yielding data: {e}") - # Explicit client removal when we detect a broken pipe if channel_id in proxy_server.client_managers: proxy_server.client_managers[channel_id].remove_client(client_id) - break # Exit the generator - - # Pacing logic - packets_sent_in_batch += packets_in_chunk - elapsed = time.time() - batch_start_time - target_time = packets_sent_in_batch / packets_per_second - - # If we're sending too fast, add a small delay - if elapsed < target_time and packets_sent_in_batch < total_packets: - sleep_time = min(target_time - elapsed, 0.05) - if sleep_time > 0.001: - time.sleep(sleep_time) + return # Exit the generator # Log progress periodically if chunks_sent % 100 == 0: @@ -321,7 +310,7 @@ def stream_ts(request, channel_id): rate = bytes_sent / elapsed / 1024 if elapsed > 0 else 0 logger.info(f"[{client_id}] Stats: {chunks_sent} chunks, {bytes_sent/1024:.1f}KB, {rate:.1f}KB/s") - # Update local index + # Update local index and last yield time local_index = end_idx last_yield_time = time.time() else: From d00aa3444e10acc26dd9c256bb2f98972cdb7a62 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 13 Mar 2025 13:15:48 -0500 Subject: [PATCH 3/7] Switched back to http connections instead of socket. --- apps/proxy/config.py | 5 + apps/proxy/ts_proxy/server.py | 439 +++++++++++++++++++++------------- apps/proxy/ts_proxy/views.py | 88 +++++-- 3 files changed, 345 insertions(+), 187 deletions(-) diff --git a/apps/proxy/config.py b/apps/proxy/config.py index a5d39452..055e00c0 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -48,4 +48,9 @@ class TSConfig(BaseConfig): CHANNEL_INIT_GRACE_PERIOD = 5 # How long to wait for first client after initialization (seconds) CLIENT_HEARTBEAT_INTERVAL = 1 # How often to send client heartbeats (seconds) GHOST_CLIENT_MULTIPLIER = 5.0 # How many heartbeat intervals before client considered ghost (5 would mean 5 secondsif heartbeat interval is 1) + + # TS packets are 188 bytes + # Make chunk size a multiple of TS packet size for perfect alignment + # ~1MB is ideal for streaming (matches typical media buffer sizes) + BUFFER_CHUNK_SIZE = 188 * 5644 # ~1MB (exactly 1,061,072 bytes) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 88043b15..c6603ba9 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -30,18 +30,18 @@ logging.basicConfig( print("TS PROXY SERVER MODULE LOADED", file=sys.stderr) class StreamManager: - """Manages a connection to a TS stream with continuity tracking""" + """Manages a connection to a TS stream without using raw sockets""" def __init__(self, url, buffer, user_agent=None): - # Existing initialization code + # Basic properties self.url = url self.buffer = buffer self.running = True self.connected = False - self.socket = None - self.ready_event = threading.Event() self.retry_count = 0 self.max_retries = Config.MAX_RETRIES + self.current_response = None + self.current_session = None # User agent for connection self.user_agent = user_agent or Config.DEFAULT_USER_AGENT @@ -50,32 +50,187 @@ class StreamManager: self.last_data_time = time.time() self.healthy = True self.health_check_interval = Config.HEALTH_CHECK_INTERVAL + self.chunk_size = getattr(Config, 'CHUNK_SIZE', 8192) - # Buffer management - self._last_buffer_check = time.time() logging.info(f"Initialized stream manager for channel {buffer.channel_id}") - - def _create_session(self) -> requests.Session: - """Create and configure requests session""" + + 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 run(self): + """Main execution loop using HTTP streaming with improved connection handling""" + try: + # Start health monitor thread + health_thread = threading.Thread(target=self._monitor_health, daemon=True) + health_thread.start() + + logging.info(f"Starting stream for URL: {self.url}") + + while self.running: + try: + # Create new session for each connection attempt + session = self._create_session() + self.current_session = session + + # Stream the URL with proper timeout handling + response = session.get( + self.url, + stream=True, + timeout=(10, 60) # 10s connect timeout, 60s read timeout + ) + self.current_response = response + + if response.status_code == 200: + self.connected = True + self.healthy = True + logging.info("Successfully connected to stream source") + + # Set channel state to waiting for clients + self._set_waiting_for_clients() + + # Process the stream in chunks with improved error handling + try: + chunk_count = 0 + for chunk in response.iter_content(chunk_size=self.chunk_size): + if not self.running: + break + + if chunk: + # Add chunk to buffer with TS packet alignment + success = self.buffer.add_chunk(chunk) + + if success: + self.last_data_time = time.time() + chunk_count += 1 + + # Update last data timestamp in Redis + if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: + last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data" + self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) + except AttributeError as e: + # Handle the specific 'NoneType' object has no attribute 'read' error + if "'NoneType' object has no attribute 'read'" in str(e): + logging.warning(f"Connection closed by server (read {chunk_count} chunks before disconnect)") + else: + # Re-raise unexpected AttributeError + raise + else: + logging.error(f"Failed to connect to stream: HTTP {response.status_code}") + time.sleep(2) + + except requests.exceptions.ReadTimeout: + logging.warning("Read timeout - server stopped sending data") + self.connected = False + time.sleep(1) + + except requests.RequestException as e: + logging.error(f"HTTP request error: {e}") + self.connected = False + time.sleep(5) + + finally: + # Clean up response and session + if self.current_response: + try: + self.current_response.close() + except Exception as e: + logging.debug(f"Error closing response: {e}") + self.current_response = None + + if self.current_session: + try: + self.current_session.close() + except Exception as e: + logging.debug(f"Error closing session: {e}") + self.current_session = None + + # Connection retry logic + if self.running and not self.connected: + self.retry_count += 1 + if self.retry_count > self.max_retries: + logging.error(f"Maximum retry attempts ({self.max_retries}) exceeded") + break + + timeout = min(2 ** self.retry_count, 30) + logging.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})") + time.sleep(timeout) + + except Exception as e: + logging.error(f"Stream error: {e}", exc_info=True) + finally: + self.connected = False + + if self.current_response: + try: + self.current_response.close() + except: + pass + + if self.current_session: + try: + self.current_session.close() + except: + pass + + logging.info("Stream manager stopped") + + def stop(self): + """Stop the stream manager and clean up resources""" + self.running = False + + if self.current_response: + try: + self.current_response.close() + except: + pass + + if self.current_session: + try: + self.current_session.close() + except: + pass + + logging.info("Stream manager resources released") - def update_url(self, new_url: str) -> bool: - """Update stream URL and reconnect""" + def update_url(self, new_url): + """Update stream URL and reconnect with HTTP streaming approach""" if new_url == self.url: + logging.info(f"URL unchanged: {new_url}") return False logging.info(f"Switching stream URL from {self.url} to {new_url}") + + # Close existing HTTP connection resources instead of socket + self._close_connection() # Use our new method instead of _close_socket + + # Update URL and reset connection state + old_url = self.url self.url = new_url self.connected = False - self._close_socket() # Close existing connection - # Signal health monitor to reconnect immediately - self.last_data_time = 0 + # Reset retry counter to allow immediate reconnect + self.retry_count = 0 return True @@ -83,119 +238,6 @@ class StreamManager: """Check if connection retry is allowed""" return self.retry_count < self.max_retries - def stop(self) -> None: - """Stop the stream manager and close all resources""" - self.running = False - self._close_socket() - logging.info("Stream manager resources released") - - def run(self): - """Main execution loop with stream health monitoring""" - try: - # Check if buffer already has data - in which case we might not need to connect - if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: - buffer_index = self.buffer.redis_client.get(f"ts_proxy:buffer:{self.buffer.channel_id}:index") - if buffer_index and int(buffer_index) > 0: - # There's already data in Redis, check if it's recent (within last 10 seconds) - last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data" - last_data = self.buffer.redis_client.get(last_data_key) - if last_data: - last_time = float(last_data) - if time.time() - last_time < 10: - logging.info(f"Recent data found in Redis, no need to reconnect") - self.connected = True - self.healthy = True - return - - # Start health monitor thread - health_thread = threading.Thread(target=self._monitor_health, daemon=True) - health_thread.start() - - current_response = None # Track the current response object - current_session = None # Track the current session - - # Establish network connection - import socket - import requests - - logging.info(f"Starting stream for URL: {self.url}") - - while self.running: - try: - # Parse URL - if self.url.startswith("http"): - # HTTP connection - session = self._create_session() - current_session = session - - try: - # Create an initial connection to get socket - response = session.get(self.url, stream=True) - current_response = response - - if response.status_code == 200: - self.connected = True - self.socket = response.raw._fp.fp.raw - self.healthy = True - logging.info("Successfully connected to stream source") - - # Connection successful - START GRACE PERIOD HERE - self._set_waiting_for_clients() - - # Main fetch loop - while self.running and self.connected: - if self.fetch_chunk(): - self.last_data_time = time.time() - else: - if not self.running: - break - time.sleep(0.1) - else: - logging.error(f"Failed to connect to stream: HTTP {response.status_code}") - time.sleep(2) - finally: - # Properly close response before session - if current_response: - try: - # Close the response explicitly to avoid the urllib3 error - current_response.close() - except Exception as e: - logging.debug(f"Error closing response: {e}") - current_response = None - - if current_session: - try: - current_session.close() - except Exception as e: - logging.debug(f"Error closing session: {e}") - current_session = None - else: - logging.error(f"Unsupported URL scheme: {self.url}") - - # Connection retry logic - if self.running and not self.connected: - self.retry_count += 1 - if self.retry_count > self.max_retries: - logging.error(f"Maximum retry attempts ({self.max_retries}) exceeded") - break - - timeout = min(2 ** self.retry_count, 30) - logging.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})") - time.sleep(timeout) - - except Exception as e: - logging.error(f"Connection error: {e}") - self._close_socket() - time.sleep(5) - - except Exception as e: - logging.error(f"Stream error: {e}") - self._close_socket() - finally: - # Final cleanup - self._close_socket() - logging.info("Stream manager stopped") - def _monitor_health(self): """Monitor stream health and attempt recovery if needed""" while self.running: @@ -223,16 +265,28 @@ class StreamManager: time.sleep(self.health_check_interval) - def _close_socket(self): - """Close the socket connection safely""" - if self.socket: + def _close_connection(self): + """Close HTTP connection resources""" + # Close response if it exists + if hasattr(self, 'current_response') and self.current_response: try: - self.socket.close() + self.current_response.close() except Exception as e: - logging.debug(f"Error closing socket: {e}") - pass - self.socket = None - self.connected = False + logging.debug(f"Error closing response: {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: + logging.debug(f"Error closing session: {e}") + self.current_session = None + + # Keep backward compatibility - let's create an alias to the new method + def _close_socket(self): + """Backward compatibility wrapper for _close_connection""" + return self._close_connection() def fetch_chunk(self): """Fetch data from socket with direct pass-through to buffer""" @@ -355,6 +409,7 @@ class StreamManager: class StreamBuffer: """Manages stream data buffering using Redis for persistence""" + # Add a memory buffer to collect data before writing to Redis def __init__(self, channel_id=None, redis_client=None): self.channel_id = channel_id self.redis_client = redis_client @@ -377,39 +432,70 @@ class StreamBuffer: logging.info(f"Initialized buffer from Redis with index {self.index}") except Exception as e: logging.error(f"Error initializing buffer from Redis: {e}") - + + self._write_buffer = bytearray() + self.target_chunk_size = getattr(Config, 'BUFFER_CHUNK_SIZE', 188 * 5644) # ~1MB default + def add_chunk(self, chunk): - """Add a chunk to the buffer with minimal processing""" + """Add data with optimized Redis storage""" if not chunk: return False try: + # Accumulate partial packets between chunks + if not hasattr(self, '_partial_packet'): + self._partial_packet = bytearray() + + # Combine with any previous partial packet + combined_data = bytearray(self._partial_packet) + bytearray(chunk) + + # Calculate complete packets + complete_packets_size = (len(combined_data) // 188) * 188 + + if complete_packets_size == 0: + # Not enough data for a complete packet + self._partial_packet = combined_data + return True + + # Split into complete packets and remainder + complete_packets = combined_data[:complete_packets_size] + self._partial_packet = combined_data[complete_packets_size:] + + # Add completed packets to write buffer + self._write_buffer.extend(complete_packets) + + # Only write to Redis when we have enough data for an optimized chunk + writes_done = 0 with self.lock: - # Increment index atomically - if self.redis_client: - # Use Redis to store and track chunks - try: - chunk_index = self.redis_client.incr(self.buffer_index_key) - chunk_key = f"{self.buffer_prefix}{chunk_index}" - setex_result = self.redis_client.setex(chunk_key, self.chunk_ttl, chunk) - - if not setex_result: - logging.error(f"Redis SETEX failed for chunk {chunk_index}") - return False - - # Update local tracking - self.index = chunk_index - - logging.debug(f"Added chunk {chunk_index} ({len(chunk)} bytes) to buffer") - return True - - except Exception as e: - logging.error(f"Redis operation failed in add_chunk: {e}") - return False - else: - logging.error("Redis not available, cannot store chunks") - return False + while len(self._write_buffer) >= self.target_chunk_size: + # Extract a full chunk + chunk_data = self._write_buffer[:self.target_chunk_size] + self._write_buffer = self._write_buffer[self.target_chunk_size:] + # Write optimized chunk to Redis + if self.redis_client: + try: + chunk_index = self.redis_client.incr(self.buffer_index_key) + chunk_key = f"{self.buffer_prefix}{chunk_index}" + setex_result = self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(chunk_data)) + + if not setex_result: + logging.error(f"Redis SETEX failed for chunk {chunk_index}") + continue + + # Update local tracking + self.index = chunk_index + writes_done += 1 + + except Exception as e: + logging.error(f"Redis operation failed in add_chunk: {e}") + return False + + if writes_done > 0: + logging.debug(f"Added {writes_done} optimized chunks ({self.target_chunk_size} bytes each) to Redis") + + return True + except Exception as e: logging.error(f"Error adding chunk to buffer: {e}") return False @@ -534,6 +620,37 @@ class StreamBuffer: logging.error(f"Error getting exact chunks: {e}", exc_info=True) return [] + def stop(self): + """Stop the buffer and flush any remaining data""" + try: + # Flush any remaining data in the write buffer + if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0: + # Ensure remaining data is aligned to TS packets + complete_size = (len(self._write_buffer) // 188) * 188 + + if complete_size > 0: + final_chunk = self._write_buffer[:complete_size] + + # Write final chunk to Redis + with self.lock: + if self.redis_client: + try: + chunk_index = self.redis_client.incr(self.buffer_index_key) + chunk_key = f"{self.buffer_prefix}{chunk_index}" + self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(final_chunk)) + self.index = chunk_index + logging.info(f"Flushed final chunk of {len(final_chunk)} bytes to Redis") + except Exception as e: + logging.error(f"Error flushing final chunk: {e}") + + # Clear buffers + self._write_buffer = bytearray() + if hasattr(self, '_partial_packet'): + self._partial_packet = bytearray() + + except Exception as e: + logging.error(f"Error during buffer stop: {e}") + class ClientManager: """Manages connected clients for a channel with cross-worker visibility""" diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index a0d70cec..bf2d71f7 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -276,42 +276,36 @@ def stream_ts(request, channel_id): # Main streaming loop while True: - # Get chunks at client's position - chunks = buffer.get_chunks_exact(local_index, Config.CHUNK_BATCH_SIZE) + # Get chunks at client's position using improved strategy + chunks, next_index = get_client_data(buffer, local_index) if chunks: empty_reads = 0 consecutive_empty = 0 - # Track and send chunks - chunk_sizes = [len(c) for c in chunks] - total_size = sum(chunk_sizes) - start_idx = local_index + 1 - end_idx = local_index + len(chunks) + # Process and send chunks + total_size = sum(len(c) for c in chunks) + logger.debug(f"[{client_id}] Retrieved {len(chunks)} chunks ({total_size} bytes) from index {local_index+1} to {next_index}") - logger.debug(f"[{client_id}] Retrieved {len(chunks)} chunks ({total_size} bytes) from index {start_idx} to {end_idx}") - - # SIMPLIFIED: Just send chunks directly without pacing + # CRITICAL FIX: Actually send the chunks to the client for chunk in chunks: - bytes_sent += len(chunk) - chunks_sent += 1 - try: + # This is the crucial line that was likely missing yield chunk + bytes_sent += len(chunk) + chunks_sent += 1 + + # Log every 100 chunks for visibility + if chunks_sent % 100 == 0: + elapsed = time.time() - stream_start_time + rate = bytes_sent / elapsed / 1024 if elapsed > 0 else 0 + logger.info(f"[{client_id}] Stats: {chunks_sent} chunks, {bytes_sent/1024:.1f}KB, {rate:.1f}KB/s") except Exception as e: - logger.info(f"[{client_id}] Client disconnected while yielding data: {e}") - if channel_id in proxy_server.client_managers: - proxy_server.client_managers[channel_id].remove_client(client_id) - return # Exit the generator + logger.error(f"[{client_id}] Error sending chunk to client: {e}") + raise # Re-raise to exit the generator - # Log progress periodically - if chunks_sent % 100 == 0: - elapsed = time.time() - stream_start_time - rate = bytes_sent / elapsed / 1024 if elapsed > 0 else 0 - logger.info(f"[{client_id}] Stats: {chunks_sent} chunks, {bytes_sent/1024:.1f}KB, {rate:.1f}KB/s") - - # Update local index and last yield time - local_index = end_idx + # Update index after successfully sending all chunks + local_index = next_index last_yield_time = time.time() else: # No chunks available @@ -562,4 +556,46 @@ def change_stream(request, channel_id): return JsonResponse({'error': 'Invalid JSON'}, status=400) except Exception as e: logger.error(f"Failed to change stream: {e}", exc_info=True) - return JsonResponse({'error': str(e)}, status=500) \ No newline at end of file + return JsonResponse({'error': str(e)}, status=500) + +def get_client_data(buffer, local_index): + """Get optimal amount of data for client""" + # Define limits + MIN_CHUNKS = 3 # Minimum chunks to read for efficiency + MAX_CHUNKS = 20 # Safety limit to prevent memory spikes + TARGET_SIZE = 1024 * 1024 # Target ~1MB per response (typical media buffer) + MAX_SIZE = 2 * 1024 * 1024 # Hard cap at 2MB + + # Calculate how far behind we are + chunks_behind = buffer.index - local_index + + # Determine optimal chunk count + if chunks_behind <= MIN_CHUNKS: + # Not much data, retrieve what's available + chunk_count = max(1, chunks_behind) + elif chunks_behind <= MAX_CHUNKS: + # Reasonable amount behind, catch up completely + chunk_count = chunks_behind + else: + # Way behind, retrieve MAX_CHUNKS to avoid memory pressure + chunk_count = MAX_CHUNKS + + # Retrieve chunks + chunks = buffer.get_chunks_exact(local_index, chunk_count) + + # Check total size + total_size = sum(len(c) for c in chunks) + + # If we're under target and have more chunks available, get more + if total_size < TARGET_SIZE and chunks_behind > chunk_count: + # Calculate how many more chunks we can get + additional = min(MAX_CHUNKS - chunk_count, chunks_behind - chunk_count) + more_chunks = buffer.get_chunks_exact(local_index + chunk_count, additional) + + # Check if adding more would exceed MAX_SIZE + additional_size = sum(len(c) for c in more_chunks) + if total_size + additional_size <= MAX_SIZE: + chunks.extend(more_chunks) + chunk_count += additional + + return chunks, local_index + chunk_count \ No newline at end of file From 29b97731b4110c9d0f0a2c409ceee1c56348afa5 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 13 Mar 2025 13:31:47 -0500 Subject: [PATCH 4/7] Fixes duplicate clients. --- apps/proxy/ts_proxy/views.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index bf2d71f7..e2edde4f 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -163,8 +163,7 @@ def stream_ts(request, channel_id): logger.info(f"[{client_id}] Client registered with channel {channel_id}") # Start stream response - def generate(): - client_id = f"client_{int(time.time() * 1000)}_{random.randint(1000, 9999)}" + def generate(): stream_start_time = time.time() bytes_sent = 0 chunks_sent = 0 From 286a22adc4b08a7fb766bf8948a16ef326ce2d92 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 13 Mar 2025 14:17:39 -0500 Subject: [PATCH 5/7] All functionality seems to be working now with HTTP instead of sockets. --- apps/proxy/ts_proxy/server.py | 157 +++++++++++++++++----------------- apps/proxy/ts_proxy/views.py | 16 +--- 2 files changed, 83 insertions(+), 90 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index c6603ba9..e9296953 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -407,9 +407,8 @@ class StreamManager: return False class StreamBuffer: - """Manages stream data buffering using Redis for persistence""" + """Manages stream data buffering with optimized chunk storage""" - # Add a memory buffer to collect data before writing to Redis def __init__(self, channel_id=None, redis_client=None): self.channel_id = channel_id self.redis_client = redis_client @@ -437,7 +436,7 @@ class StreamBuffer: self.target_chunk_size = getattr(Config, 'BUFFER_CHUNK_SIZE', 188 * 5644) # ~1MB default def add_chunk(self, chunk): - """Add data with optimized Redis storage""" + """Add data with optimized Redis storage and TS packet alignment""" if not chunk: return False @@ -474,22 +473,13 @@ class StreamBuffer: # Write optimized chunk to Redis if self.redis_client: - try: - chunk_index = self.redis_client.incr(self.buffer_index_key) - chunk_key = f"{self.buffer_prefix}{chunk_index}" - setex_result = self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(chunk_data)) - - if not setex_result: - logging.error(f"Redis SETEX failed for chunk {chunk_index}") - continue - - # Update local tracking - self.index = chunk_index - writes_done += 1 - - except Exception as e: - logging.error(f"Redis operation failed in add_chunk: {e}") - return False + chunk_index = self.redis_client.incr(self.buffer_index_key) + chunk_key = f"{self.buffer_prefix}{chunk_index}" + self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(chunk_data)) + + # Update local tracking + self.index = chunk_index + writes_done += 1 if writes_done > 0: logging.debug(f"Added {writes_done} optimized chunks ({self.target_chunk_size} bytes each) to Redis") @@ -652,15 +642,15 @@ class StreamBuffer: logging.error(f"Error during buffer stop: {e}") class ClientManager: - """Manages connected clients for a channel with cross-worker visibility""" + """Manages client connections with no duplicates""" - def __init__(self, channel_id, redis_client=None, worker_id=None): + def __init__(self, channel_id=None, redis_client=None, heartbeat_interval=1, worker_id=None): self.channel_id = channel_id self.redis_client = redis_client - self.worker_id = worker_id self.clients = set() self.lock = threading.Lock() self.last_active_time = time.time() + self.worker_id = worker_id # Store worker ID as instance variable # STANDARDIZED KEYS: Move client set under channel namespace self.client_set_key = f"ts_proxy:channel:{channel_id}:clients" @@ -670,6 +660,7 @@ class ClientManager: # Start heartbeat thread for local clients self._start_heartbeat_thread() + self._registered_clients = set() # Track already registered client IDs def _start_heartbeat_thread(self): """Start thread to regularly refresh client presence in Redis""" @@ -781,67 +772,76 @@ class ClientManager: logging.error(f"Error notifying owner of client activity: {e}") def add_client(self, client_id, user_agent=None): - """Add a client to this channel locally and in Redis""" - with self.lock: - self.clients.add(client_id) - self.last_active_time = time.time() + """Add a client with duplicate prevention""" + if client_id in self._registered_clients: + logging.debug(f"Client {client_id} already registered, skipping") + return False - if self.redis_client: - current_time = str(time.time()) + self._registered_clients.add(client_id) + + # FIX: Consistent key naming - note the 's' in 'clients' + client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}" + + # Prepare client data + current_time = str(time.time()) + client_data = { + "user_agent": user_agent or "unknown", + "connected_at": current_time, + "last_active": current_time, + "worker_id": self.worker_id or "unknown" + } + + try: + with self.lock: + # Store client in local set + self.clients.add(client_id) - # Add to channel's client set - self.redis_client.sadd(self.client_set_key, client_id) - self.redis_client.expire(self.client_set_key, self.client_ttl) - - # STANDARDIZED KEY: Individual client under channel namespace - client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}" - - # Store client info as a hash with all info in one place - client_data = { - "last_active": current_time, - "worker_id": self.worker_id or "unknown", - "connect_time": current_time - } - - # Add user agent if provided - if user_agent: - client_data["user_agent"] = user_agent + # Store in Redis + if self.redis_client: + # FIXED: Store client data just once with proper key + self.redis_client.hset(client_key, mapping=client_data) + self.redis_client.expire(client_key, self.client_ttl) - # Use HSET to store client data as a hash - self.redis_client.hset(client_key, mapping=client_data) - self.redis_client.expire(client_key, self.client_ttl) + # Add to the client set + self.redis_client.sadd(self.client_set_key, client_id) + self.redis_client.expire(self.client_set_key, self.client_ttl) + + # Clear any initialization timer + self.redis_client.delete(f"ts_proxy:channel:{self.channel_id}:init_time") + + self._notify_owner_of_activity() + + # Publish client connected event with user agent + event_data = { + "event": "client_connected", + "channel_id": self.channel_id, + "client_id": client_id, + "worker_id": self.worker_id or "unknown", + "timestamp": time.time() + } + + if user_agent: + event_data["user_agent"] = user_agent + logging.debug(f"Storing user agent '{user_agent}' for client {client_id}") + else: + logging.debug(f"No user agent provided for client {client_id}") + + self.redis_client.publish( + f"ts_proxy:events:{self.channel_id}", + json.dumps(event_data) + ) + + # Get total clients across all workers + total_clients = self.get_total_client_count() + logging.info(f"New client connected: {client_id} (local: {len(self.clients)}, total: {total_clients})") - # Clear any initialization timer - self.redis_client.delete(f"ts_proxy:channel:{self.channel_id}:init_time") + self.last_heartbeat_time[client_id] = time.time() - self._notify_owner_of_activity() - - # Publish client connected event with user agent - event_data = { - "event": "client_connected", - "channel_id": self.channel_id, - "client_id": client_id, - "worker_id": self.worker_id or "unknown", - "timestamp": time.time() - } - - if user_agent: - event_data["user_agent"] = user_agent - logging.debug(f"Storing user agent '{user_agent}' for client {client_id}") - else: - logging.debug(f"No user agent provided for client {client_id}") - self.redis_client.publish( - f"ts_proxy:events:{self.channel_id}", - json.dumps(event_data) - ) - - # Get total clients across all workers - total_clients = self.get_total_client_count() - logging.info(f"New client connected: {client_id} (local: {len(self.clients)}, total: {total_clients})") - - self.last_heartbeat_time[client_id] = time.time() - - return len(self.clients) + return len(self.clients) + + except Exception as e: + logging.error(f"Error adding client {client_id}: {e}") + return False def remove_client(self, client_id): """Remove a client from this channel and Redis""" @@ -1729,3 +1729,4 @@ class ProxyServer: logging.error(f"Error updating channel state: {e}") return False + diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index e2edde4f..40059090 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -1,6 +1,5 @@ import json import threading -import logging import time import random import sys @@ -10,15 +9,10 @@ from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods, require_GET from apps.proxy.config import TSConfig as Config from .server import ProxyServer +from apps.proxy.ts_proxy.server import logging as server_logging # Configure logging properly to ensure visibility -logger = logging.getLogger(__name__) -handler = logging.StreamHandler(sys.stdout) -handler.setLevel(logging.DEBUG) -formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') -handler.setFormatter(formatter) -logger.addHandler(handler) -logger.setLevel(logging.DEBUG) +logger = server_logging # Print directly to output for critical messages (bypass logging system) print("TS PROXY VIEWS INITIALIZED", file=sys.stderr) @@ -79,13 +73,13 @@ def initialize_stream(request, channel_id): @require_GET def stream_ts(request, channel_id): - """Stream TS data to client with improved waiting for initialization""" + """Stream TS data to client with single client registration""" try: # Generate a unique client ID client_id = f"client_{int(time.time() * 1000)}_{random.randint(1000, 9999)}" logger.info(f"[{client_id}] Requested stream for channel {channel_id}") - # Get user agent from request headers - MOVED UP TO ENSURE IT'S DEFINED + # Extract user agent only once user_agent = None for header in ['HTTP_USER_AGENT', 'User-Agent', 'user-agent']: if header in request.META: @@ -157,8 +151,6 @@ def stream_ts(request, channel_id): # Get stream buffer and client manager buffer = proxy_server.stream_buffers[channel_id] client_manager = proxy_server.client_managers[channel_id] - - # Register client before starting stream client_manager.add_client(client_id, user_agent) logger.info(f"[{client_id}] Client registered with channel {channel_id}") From 317874b5b862a8d1d7cca98a24d0af796e223990 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 13 Mar 2025 14:33:00 -0500 Subject: [PATCH 6/7] Imporoved logging. --- apps/proxy/ts_proxy/server.py | 288 +++++++++++++++++----------------- apps/proxy/ts_proxy/views.py | 3 - 2 files changed, 140 insertions(+), 151 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index e9296953..9129f4df 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -19,15 +19,7 @@ from typing import Optional, Set, Deque, Dict import json from apps.proxy.config import TSConfig as Config -# Configure root logger for this module -logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s - TS_PROXY - %(levelname)s - %(message)s', - handlers=[logging.StreamHandler(sys.stdout)] -) - -# Force immediate output -print("TS PROXY SERVER MODULE LOADED", file=sys.stderr) +logger = logging.getLogger("ts_proxy") class StreamManager: """Manages a connection to a TS stream without using raw sockets""" @@ -52,7 +44,7 @@ class StreamManager: self.health_check_interval = Config.HEALTH_CHECK_INTERVAL self.chunk_size = getattr(Config, 'CHUNK_SIZE', 8192) - logging.info(f"Initialized stream manager for channel {buffer.channel_id}") + logger.info(f"Initialized stream manager for channel {buffer.channel_id}") def _create_session(self): """Create and configure requests session with optimal settings""" @@ -85,7 +77,7 @@ class StreamManager: health_thread = threading.Thread(target=self._monitor_health, daemon=True) health_thread.start() - logging.info(f"Starting stream for URL: {self.url}") + logger.info(f"Starting stream for URL: {self.url}") while self.running: try: @@ -104,7 +96,7 @@ class StreamManager: if response.status_code == 200: self.connected = True self.healthy = True - logging.info("Successfully connected to stream source") + logger.info("Successfully connected to stream source") # Set channel state to waiting for clients self._set_waiting_for_clients() @@ -131,21 +123,21 @@ class StreamManager: except AttributeError as e: # Handle the specific 'NoneType' object has no attribute 'read' error if "'NoneType' object has no attribute 'read'" in str(e): - logging.warning(f"Connection closed by server (read {chunk_count} chunks before disconnect)") + logger.warning(f"Connection closed by server (read {chunk_count} chunks before disconnect)") else: # Re-raise unexpected AttributeError raise else: - logging.error(f"Failed to connect to stream: HTTP {response.status_code}") + logger.error(f"Failed to connect to stream: HTTP {response.status_code}") time.sleep(2) except requests.exceptions.ReadTimeout: - logging.warning("Read timeout - server stopped sending data") + logger.warning("Read timeout - server stopped sending data") self.connected = False time.sleep(1) except requests.RequestException as e: - logging.error(f"HTTP request error: {e}") + logger.error(f"HTTP request error: {e}") self.connected = False time.sleep(5) @@ -155,29 +147,29 @@ class StreamManager: try: self.current_response.close() except Exception as e: - logging.debug(f"Error closing response: {e}") + logger.debug(f"Error closing response: {e}") self.current_response = None if self.current_session: try: self.current_session.close() except Exception as e: - logging.debug(f"Error closing session: {e}") + logger.debug(f"Error closing session: {e}") self.current_session = None # Connection retry logic if self.running and not self.connected: self.retry_count += 1 if self.retry_count > self.max_retries: - logging.error(f"Maximum retry attempts ({self.max_retries}) exceeded") + logger.error(f"Maximum retry attempts ({self.max_retries}) exceeded") break timeout = min(2 ** self.retry_count, 30) - logging.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})") + logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})") time.sleep(timeout) except Exception as e: - logging.error(f"Stream error: {e}", exc_info=True) + logger.error(f"Stream error: {e}", exc_info=True) finally: self.connected = False @@ -193,7 +185,7 @@ class StreamManager: except: pass - logging.info("Stream manager stopped") + logger.info("Stream manager stopped") def stop(self): """Stop the stream manager and clean up resources""" @@ -211,15 +203,15 @@ class StreamManager: except: pass - logging.info("Stream manager resources released") + logger.info("Stream manager resources released") def update_url(self, new_url): """Update stream URL and reconnect with HTTP streaming approach""" if new_url == self.url: - logging.info(f"URL unchanged: {new_url}") + logger.info(f"URL unchanged: {new_url}") return False - logging.info(f"Switching stream URL from {self.url} to {new_url}") + logger.info(f"Switching stream URL from {self.url} to {new_url}") # Close existing HTTP connection resources instead of socket self._close_connection() # Use our new method instead of _close_socket @@ -246,22 +238,22 @@ class StreamManager: if now - self.last_data_time > 10 and self.connected: # No data for 10 seconds, mark as unhealthy if self.healthy: - logging.warning("Stream health check: No data received for 10+ seconds") + logger.warning("Stream health check: No data received for 10+ seconds") self.healthy = False # After 30 seconds with no data, force reconnection if now - self.last_data_time > 30: - logging.warning("Stream appears dead, forcing reconnection") + logger.warning("Stream appears dead, forcing reconnection") self._close_socket() self.connected = False self.last_data_time = time.time() # Reset timer for the reconnect elif self.connected and not self.healthy: # Stream is receiving data again after being unhealthy - logging.info("Stream health restored, receiving data again") + logger.info("Stream health restored, receiving data again") self.healthy = True except Exception as e: - logging.error(f"Error in health monitor: {e}") + logger.error(f"Error in health monitor: {e}") time.sleep(self.health_check_interval) @@ -272,7 +264,7 @@ class StreamManager: try: self.current_response.close() except Exception as e: - logging.debug(f"Error closing response: {e}") + logger.debug(f"Error closing response: {e}") self.current_response = None # Close session if it exists @@ -280,7 +272,7 @@ class StreamManager: try: self.current_session.close() except Exception as e: - logging.debug(f"Error closing session: {e}") + logger.debug(f"Error closing session: {e}") self.current_session = None # Keep backward compatibility - let's create an alias to the new method @@ -308,7 +300,7 @@ class StreamManager: if not chunk: # Connection closed by server - logging.warning("Server closed connection") + logger.warning("Server closed connection") self._close_socket() self.connected = False return False @@ -325,13 +317,13 @@ class StreamManager: except (socket.timeout, socket.error) as e: # Socket error - logging.error(f"Socket error: {e}") + logger.error(f"Socket error: {e}") self._close_socket() self.connected = False return False except Exception as e: - logging.error(f"Error in fetch_chunk: {e}") + logger.error(f"Error in fetch_chunk: {e}") return False def _set_waiting_for_clients(self): @@ -354,7 +346,7 @@ class StreamManager: if metadata and b'state' in metadata: current_state = metadata[b'state'].decode('utf-8') except Exception as e: - logging.error(f"Error checking current state: {e}") + logger.error(f"Error checking current state: {e}") # Only update if not already past connecting if not current_state or current_state in ["initializing", "connecting"]: @@ -368,12 +360,12 @@ class StreamManager: # Get configured grace period or default grace_period = getattr(Config, 'CHANNEL_INIT_GRACE_PERIOD', 20) - logging.info(f"STREAM MANAGER: Updated channel {channel_id} state: {current_state or 'None'} → waiting_for_clients") - logging.info(f"Started initial connection grace period ({grace_period}s) for channel {channel_id}") + logger.info(f"STREAM MANAGER: Updated channel {channel_id} state: {current_state or 'None'} → waiting_for_clients") + logger.info(f"Started initial connection grace period ({grace_period}s) for channel {channel_id}") else: - logging.debug(f"Not changing state: channel {channel_id} already in {current_state} state") + logger.debug(f"Not changing state: channel {channel_id} already in {current_state} state") except Exception as e: - logging.error(f"Error setting waiting for clients state: {e}") + logger.error(f"Error setting waiting for clients state: {e}") def _read_stream(self): """Read from stream with minimal processing""" @@ -383,7 +375,7 @@ class StreamManager: if not chunk: # Connection closed - logging.debug("Connection closed by remote host") + logger.debug("Connection closed by remote host") return False # If we got data, just add it directly to the buffer @@ -403,7 +395,7 @@ class StreamManager: return True except Exception as e: # Error reading from socket - logging.error(f"Error reading from stream: {e}") + logger.error(f"Error reading from stream: {e}") return False class StreamBuffer: @@ -428,9 +420,9 @@ class StreamBuffer: current_index = self.redis_client.get(self.buffer_index_key) if current_index: self.index = int(current_index) - logging.info(f"Initialized buffer from Redis with index {self.index}") + logger.info(f"Initialized buffer from Redis with index {self.index}") except Exception as e: - logging.error(f"Error initializing buffer from Redis: {e}") + logger.error(f"Error initializing buffer from Redis: {e}") self._write_buffer = bytearray() self.target_chunk_size = getattr(Config, 'BUFFER_CHUNK_SIZE', 188 * 5644) # ~1MB default @@ -482,28 +474,28 @@ class StreamBuffer: writes_done += 1 if writes_done > 0: - logging.debug(f"Added {writes_done} optimized chunks ({self.target_chunk_size} bytes each) to Redis") + logger.debug(f"Added {writes_done} optimized chunks ({self.target_chunk_size} bytes each) to Redis") return True except Exception as e: - logging.error(f"Error adding chunk to buffer: {e}") + logger.error(f"Error adding chunk to buffer: {e}") return False def get_chunks(self, start_index=None): """Get chunks from the buffer with detailed logging""" try: request_id = f"req_{random.randint(1000, 9999)}" - logging.debug(f"[{request_id}] get_chunks called with start_index={start_index}") + logger.debug(f"[{request_id}] get_chunks called with start_index={start_index}") if not self.redis_client: - logging.error("Redis not available, cannot retrieve chunks") + logger.error("Redis not available, cannot retrieve chunks") return [] # If no start_index provided, use most recent chunks if start_index is None: start_index = max(0, self.index - 10) # Start closer to current position - logging.debug(f"[{request_id}] No start_index provided, using {start_index}") + logger.debug(f"[{request_id}] No start_index provided, using {start_index}") # Get current index from Redis current_index = int(self.redis_client.get(self.buffer_index_key) or 0) @@ -515,25 +507,25 @@ class StreamBuffer: # Adaptive chunk retrieval based on how far behind if chunks_behind > 100: fetch_count = 15 - logging.debug(f"[{request_id}] Client very behind ({chunks_behind} chunks), fetching {fetch_count}") + logger.debug(f"[{request_id}] Client very behind ({chunks_behind} chunks), fetching {fetch_count}") elif chunks_behind > 50: fetch_count = 10 - logging.debug(f"[{request_id}] Client moderately behind ({chunks_behind} chunks), fetching {fetch_count}") + logger.debug(f"[{request_id}] Client moderately behind ({chunks_behind} chunks), fetching {fetch_count}") elif chunks_behind > 20: fetch_count = 5 - logging.debug(f"[{request_id}] Client slightly behind ({chunks_behind} chunks), fetching {fetch_count}") + logger.debug(f"[{request_id}] Client slightly behind ({chunks_behind} chunks), fetching {fetch_count}") else: fetch_count = 3 - logging.debug(f"[{request_id}] Client up-to-date (only {chunks_behind} chunks behind), fetching {fetch_count}") + logger.debug(f"[{request_id}] Client up-to-date (only {chunks_behind} chunks behind), fetching {fetch_count}") end_id = min(current_index + 1, start_id + fetch_count) if start_id >= end_id: - logging.debug(f"[{request_id}] No new chunks to fetch (start_id={start_id}, end_id={end_id})") + logger.debug(f"[{request_id}] No new chunks to fetch (start_id={start_id}, end_id={end_id})") return [] # Log the range we're retrieving - logging.debug(f"[{request_id}] Retrieving chunks {start_id} to {end_id-1} (total: {end_id-start_id})") + logger.debug(f"[{request_id}] Retrieving chunks {start_id} to {end_id-1} (total: {end_id-start_id})") # Directly fetch from Redis using pipeline for efficiency pipe = self.redis_client.pipeline() @@ -551,7 +543,7 @@ class StreamBuffer: missing_chunks = len(results) - found_chunks if missing_chunks > 0: - logging.debug(f"[{request_id}] Missing {missing_chunks}/{len(results)} chunks in Redis") + logger.debug(f"[{request_id}] Missing {missing_chunks}/{len(results)} chunks in Redis") # Update local tracking if chunks: @@ -560,19 +552,19 @@ class StreamBuffer: # Final log message chunk_sizes = [len(c) for c in chunks] total_bytes = sum(chunk_sizes) if chunks else 0 - logging.debug(f"[{request_id}] Returning {len(chunks)} chunks ({total_bytes} bytes)") + logger.debug(f"[{request_id}] Returning {len(chunks)} chunks ({total_bytes} bytes)") return chunks except Exception as e: - logging.error(f"Error getting chunks from buffer: {e}", exc_info=True) + logger.error(f"Error getting chunks from buffer: {e}", exc_info=True) return [] def get_chunks_exact(self, start_index, count): """Get exactly the requested number of chunks from given index""" try: if not self.redis_client: - logging.error("Redis not available, cannot retrieve chunks") + logger.error("Redis not available, cannot retrieve chunks") return [] # Calculate range to retrieve @@ -607,7 +599,7 @@ class StreamBuffer: return chunks except Exception as e: - logging.error(f"Error getting exact chunks: {e}", exc_info=True) + logger.error(f"Error getting exact chunks: {e}", exc_info=True) return [] def stop(self): @@ -629,9 +621,9 @@ class StreamBuffer: chunk_key = f"{self.buffer_prefix}{chunk_index}" self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(final_chunk)) self.index = chunk_index - logging.info(f"Flushed final chunk of {len(final_chunk)} bytes to Redis") + logger.info(f"Flushed final chunk of {len(final_chunk)} bytes to Redis") except Exception as e: - logging.error(f"Error flushing final chunk: {e}") + logger.error(f"Error flushing final chunk: {e}") # Clear buffers self._write_buffer = bytearray() @@ -639,7 +631,7 @@ class StreamBuffer: self._partial_packet = bytearray() except Exception as e: - logging.error(f"Error during buffer stop: {e}") + logger.error(f"Error during buffer stop: {e}") class ClientManager: """Manages client connections with no duplicates""" @@ -687,7 +679,7 @@ class ClientManager: exists = self.redis_client.exists(client_key) if not exists: # Client entry has expired in Redis but still in our local set - logging.warning(f"Found ghost client {client_id} - expired in Redis but still in local set") + logger.warning(f"Found ghost client {client_id} - expired in Redis but still in local set") clients_to_remove.add(client_id) continue @@ -701,7 +693,7 @@ class ClientManager: # Use configurable threshold for detection ghost_threshold = getattr(Config, 'GHOST_CLIENT_MULTIPLIER', 5.0) if time_since_activity > self.heartbeat_interval * ghost_threshold: - logging.warning(f"Detected ghost client {client_id} - last active {time_since_activity:.1f}s ago") + logger.warning(f"Detected ghost client {client_id} - last active {time_since_activity:.1f}s ago") clients_to_remove.add(client_id) # Remove ghost clients in a separate step @@ -709,7 +701,7 @@ class ClientManager: self.remove_client(client_id) if clients_to_remove: - logging.info(f"Removed {len(clients_to_remove)} ghost clients from channel {self.channel_id}") + logger.info(f"Removed {len(clients_to_remove)} ghost clients from channel {self.channel_id}") # Now send heartbeats only for remaining clients pipe = self.redis_client.pipeline() @@ -746,12 +738,12 @@ class ClientManager: self._notify_owner_of_activity() except Exception as e: - logging.error(f"Error in client heartbeat thread: {e}") + logger.error(f"Error in client heartbeat thread: {e}") thread = threading.Thread(target=heartbeat_task, daemon=True) thread.name = f"client-heartbeat-{self.channel_id}" thread.start() - logging.debug(f"Started client heartbeat thread for channel {self.channel_id} (interval: {self.heartbeat_interval}s)") + logger.debug(f"Started client heartbeat thread for channel {self.channel_id} (interval: {self.heartbeat_interval}s)") def _notify_owner_of_activity(self): """Notify channel owner that clients are active on this worker""" @@ -769,12 +761,12 @@ class ClientManager: activity_key = f"ts_proxy:channel:{self.channel_id}:activity" self.redis_client.setex(activity_key, self.client_ttl, str(time.time())) except Exception as e: - logging.error(f"Error notifying owner of client activity: {e}") + logger.error(f"Error notifying owner of client activity: {e}") def add_client(self, client_id, user_agent=None): """Add a client with duplicate prevention""" if client_id in self._registered_clients: - logging.debug(f"Client {client_id} already registered, skipping") + logger.debug(f"Client {client_id} already registered, skipping") return False self._registered_clients.add(client_id) @@ -822,9 +814,9 @@ class ClientManager: if user_agent: event_data["user_agent"] = user_agent - logging.debug(f"Storing user agent '{user_agent}' for client {client_id}") + logger.debug(f"Storing user agent '{user_agent}' for client {client_id}") else: - logging.debug(f"No user agent provided for client {client_id}") + logger.debug(f"No user agent provided for client {client_id}") self.redis_client.publish( f"ts_proxy:events:{self.channel_id}", @@ -833,14 +825,14 @@ class ClientManager: # Get total clients across all workers total_clients = self.get_total_client_count() - logging.info(f"New client connected: {client_id} (local: {len(self.clients)}, total: {total_clients})") + logger.info(f"New client connected: {client_id} (local: {len(self.clients)}, total: {total_clients})") self.last_heartbeat_time[client_id] = time.time() return len(self.clients) except Exception as e: - logging.error(f"Error adding client {client_id}: {e}") + logger.error(f"Error adding client {client_id}: {e}") return False def remove_client(self, client_id): @@ -865,7 +857,7 @@ class ClientManager: # Check if this was the last client remaining = self.redis_client.scard(self.client_set_key) or 0 if remaining == 0: - logging.warning(f"Last client removed: {client_id} - channel may shut down soon") + logger.warning(f"Last client removed: {client_id} - channel may shut down soon") # Trigger disconnect time tracking even if we're not the owner disconnect_key = f"ts_proxy:channel:{self.channel_id}:last_client_disconnect_time" @@ -885,7 +877,7 @@ class ClientManager: self.redis_client.publish(f"ts_proxy:events:{self.channel_id}", event_data) total_clients = self.get_total_client_count() - logging.info(f"Client disconnected: {client_id} (local: {len(self.clients)}, total: {total_clients})") + logger.info(f"Client disconnected: {client_id} (local: {len(self.clients)}, total: {total_clients})") return len(self.clients) @@ -903,7 +895,7 @@ class ClientManager: # Count members in the client set return self.redis_client.scard(self.client_set_key) or 0 except Exception as e: - logging.error(f"Error getting total client count: {e}") + logger.error(f"Error getting total client count: {e}") return len(self.clients) # Fall back to local count def refresh_client_ttl(self): @@ -921,7 +913,7 @@ class ClientManager: # Refresh TTL on the set itself self.redis_client.expire(self.client_set_key, self.client_ttl) except Exception as e: - logging.error(f"Error refreshing client TTL: {e}") + logger.error(f"Error refreshing client TTL: {e}") class StreamFetcher: """Handles stream data fetching""" @@ -949,21 +941,21 @@ class StreamFetcher: """Handle connection state and retries""" if not self.manager.connected: if not self.manager.should_retry(): - logging.error(f"Failed to connect after {self.manager.max_retries} attempts") + logger.error(f"Failed to connect after {self.manager.max_retries} attempts") return False if not self.manager.running: return False self.manager.retry_count += 1 - logging.info(f"Connecting to stream: {self.manager.url} " + logger.info(f"Connecting to stream: {self.manager.url} " f"(attempt {self.manager.retry_count}/{self.manager.max_retries})") return True def _handle_successful_connection(self) -> None: """Handle successful stream connection""" if not self.manager.connected: - logging.info("Stream connected successfully") + logger.info("Stream connected successfully") self.manager.connected = True self.manager.retry_count = 0 @@ -971,12 +963,12 @@ class StreamFetcher: """Process incoming stream data""" for chunk in response.iter_content(chunk_size=Config.CHUNK_SIZE): if not self.manager.running: - logging.info("Stream fetch stopped - shutting down") + logger.info("Stream fetch stopped - shutting down") return if chunk: if self.manager.ready_event.is_set(): - logging.info("Stream switch in progress, closing connection") + logger.info("Stream switch in progress, closing connection") self.manager.ready_event.clear() break @@ -986,13 +978,13 @@ class StreamFetcher: def _handle_connection_error(self, error: Exception) -> None: """Handle stream connection errors""" - logging.error(f"Stream connection error: {error}") + logger.error(f"Stream connection error: {error}") self.manager.connected = False if not self.manager.running: return - logging.info(f"Attempting to reconnect in {Config.RECONNECT_DELAY} seconds...") + logger.info(f"Attempting to reconnect in {Config.RECONNECT_DELAY} seconds...") if not wait_for_running(self.manager, Config.RECONNECT_DELAY): return @@ -1029,11 +1021,11 @@ class ProxyServer: redis_url = getattr(settings, 'REDIS_URL', 'redis://localhost:6379/0') self.redis_client = redis.from_url(redis_url) - logging.info(f"Connected to Redis at {redis_url}") - logging.info(f"Worker ID: {self.worker_id}") + logger.info(f"Connected to Redis at {redis_url}") + logger.info(f"Worker ID: {self.worker_id}") except Exception as e: self.redis_client = None - logging.error(f"Failed to connect to Redis: {e}") + logger.error(f"Failed to connect to Redis: {e}") # Start cleanup thread self.cleanup_interval = getattr(Config, 'CLEANUP_INTERVAL', 60) @@ -1052,7 +1044,7 @@ class ProxyServer: pubsub = self.redis_client.pubsub() pubsub.psubscribe("ts_proxy:events:*") - logging.info("Started Redis event listener for client activity") + logger.info("Started Redis event listener for client activity") for message in pubsub.listen(): if message["type"] != "pmessage": @@ -1069,14 +1061,14 @@ class ProxyServer: # For owner, update client status immediately if self.am_i_owner(channel_id): if event_type == "client_connected": - logging.debug(f"Owner received client_connected event for channel {channel_id}") + logger.debug(f"Owner received client_connected event for channel {channel_id}") # Reset any disconnect timer # RENAMED: no_clients_since → last_client_disconnect_time disconnect_key = f"ts_proxy:channel:{channel_id}:last_client_disconnect_time" self.redis_client.delete(disconnect_key) elif event_type == "client_disconnected": - logging.debug(f"Owner received client_disconnected event for channel {channel_id}") + logger.debug(f"Owner received client_disconnected event for channel {channel_id}") # Check if any clients remain if channel_id in self.client_managers: # VERIFY REDIS CLIENT COUNT DIRECTLY @@ -1084,7 +1076,7 @@ class ProxyServer: total = self.redis_client.scard(client_set_key) or 0 if total == 0: - logging.debug(f"No clients left after disconnect event - stopping channel {channel_id}") + logger.debug(f"No clients left after disconnect event - stopping channel {channel_id}") # Set the disconnect timer for other workers to see disconnect_key = f"ts_proxy:channel:{channel_id}:last_client_disconnect_time" self.redis_client.setex(disconnect_key, 60, str(time.time())) @@ -1093,13 +1085,13 @@ class ProxyServer: shutdown_delay = getattr(Config, 'CHANNEL_SHUTDOWN_DELAY', 0) if shutdown_delay > 0: - logging.info(f"Waiting {shutdown_delay}s before stopping channel...") + logger.info(f"Waiting {shutdown_delay}s before stopping channel...") time.sleep(shutdown_delay) # Re-check client count before stopping total = self.redis_client.scard(client_set_key) or 0 if total > 0: - logging.info(f"New clients connected during shutdown delay - aborting shutdown") + logger.info(f"New clients connected during shutdown delay - aborting shutdown") self.redis_client.delete(disconnect_key) return @@ -1108,7 +1100,7 @@ class ProxyServer: elif event_type == "stream_switch": - logging.info(f"Owner received stream switch request for channel {channel_id}") + logger.info(f"Owner received stream switch request for channel {channel_id}") # Handle stream switch request new_url = data.get("url") user_agent = data.get("user_agent") @@ -1130,7 +1122,7 @@ class ProxyServer: success = stream_manager.update_url(new_url) if success: - logging.info(f"Stream switch initiated for channel {channel_id}") + logger.info(f"Stream switch initiated for channel {channel_id}") # Publish confirmation switch_result = { @@ -1149,7 +1141,7 @@ class ProxyServer: if self.redis_client: self.redis_client.set(status_key, "switched") else: - logging.error(f"Failed to switch stream for channel {channel_id}") + logger.error(f"Failed to switch stream for channel {channel_id}") # Publish failure switch_result = { @@ -1164,9 +1156,9 @@ class ProxyServer: json.dumps(switch_result) ) except Exception as e: - logging.error(f"Error processing event message: {e}") + logger.error(f"Error processing event message: {e}") except Exception as e: - logging.error(f"Error in event listener: {e}") + logger.error(f"Error in event listener: {e}") time.sleep(5) # Wait before reconnecting # Try to restart the listener self._start_event_listener() @@ -1187,7 +1179,7 @@ class ProxyServer: return owner.decode('utf-8') return None except Exception as e: - logging.error(f"Error getting channel owner: {e}") + logger.error(f"Error getting channel owner: {e}") return None def am_i_owner(self, channel_id): @@ -1210,7 +1202,7 @@ class ProxyServer: # If acquired, set expiry to prevent orphaned locks if acquired: self.redis_client.expire(lock_key, ttl) - logging.info(f"Worker {self.worker_id} acquired ownership of channel {channel_id}") + logger.info(f"Worker {self.worker_id} acquired ownership of channel {channel_id}") return True # If not acquired, check if we already own it (might be a retry) @@ -1218,14 +1210,14 @@ class ProxyServer: if current_owner and current_owner.decode('utf-8') == self.worker_id: # Refresh TTL self.redis_client.expire(lock_key, ttl) - logging.info(f"Worker {self.worker_id} refreshed ownership of channel {channel_id}") + logger.info(f"Worker {self.worker_id} refreshed ownership of channel {channel_id}") return True # Someone else owns it return False except Exception as e: - logging.error(f"Error acquiring channel ownership: {e}") + logger.error(f"Error acquiring channel ownership: {e}") return False def release_ownership(self, channel_id): @@ -1240,9 +1232,9 @@ class ProxyServer: current = self.redis_client.get(lock_key) if current and current.decode('utf-8') == self.worker_id: self.redis_client.delete(lock_key) - logging.info(f"Released ownership of channel {channel_id}") + logger.info(f"Released ownership of channel {channel_id}") except Exception as e: - logging.error(f"Error releasing channel ownership: {e}") + logger.error(f"Error releasing channel ownership: {e}") def extend_ownership(self, channel_id, ttl=30): """Extend ownership lease with grace period""" @@ -1259,7 +1251,7 @@ class ProxyServer: return True return False except Exception as e: - logging.error(f"Error extending ownership: {e}") + logger.error(f"Error extending ownership: {e}") return False def initialize_channel(self, url, channel_id, user_agent=None): @@ -1291,8 +1283,8 @@ class ProxyServer: # Exit early if another worker owns the channel if current_owner and current_owner != self.worker_id: - logging.info(f"Channel {channel_id} already owned by worker {current_owner}") - logging.info(f"This worker ({self.worker_id}) will read from Redis buffer only") + logger.info(f"Channel {channel_id} already owned by worker {current_owner}") + logger.info(f"This worker ({self.worker_id}) will read from Redis buffer only") # Create buffer but not stream manager buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client) @@ -1307,13 +1299,13 @@ class ProxyServer: # Only continue with full initialization if URL is provided # or we can get it from Redis if not channel_url: - logging.error(f"No URL available for channel {channel_id}") + logger.error(f"No URL available for channel {channel_id}") return False # Try to acquire ownership with Redis locking if not self.try_acquire_ownership(channel_id): # Another worker just acquired ownership - logging.info(f"Another worker just acquired ownership of channel {channel_id}") + logger.info(f"Another worker just acquired ownership of channel {channel_id}") # Create buffer but not stream manager buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client) @@ -1326,7 +1318,7 @@ class ProxyServer: return True # We now own the channel - ONLY NOW should we set metadata with initializing state - logging.info(f"Worker {self.worker_id} is now the owner of channel {channel_id}") + logger.info(f"Worker {self.worker_id} is now the owner of channel {channel_id}") if self.redis_client: # NOW create or update metadata with initializing state @@ -1346,12 +1338,12 @@ class ProxyServer: # Create stream buffer buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client) - logging.debug(f"Created StreamBuffer for channel {channel_id}") + logger.debug(f"Created StreamBuffer for channel {channel_id}") self.stream_buffers[channel_id] = buffer # Only the owner worker creates the actual stream manager stream_manager = StreamManager(channel_url, buffer, user_agent=channel_user_agent) - logging.debug(f"Created StreamManager for channel {channel_id}") + logger.debug(f"Created StreamManager for channel {channel_id}") self.stream_managers[channel_id] = stream_manager # Create client manager with channel_id, redis_client AND worker_id @@ -1366,7 +1358,7 @@ class ProxyServer: thread = threading.Thread(target=stream_manager.run, daemon=True) thread.name = f"stream-{channel_id}" thread.start() - logging.info(f"Started stream manager thread for channel {channel_id}") + logger.info(f"Started stream manager thread for channel {channel_id}") # If we're the owner, we need to set the channel state rather than starting a grace period immediately if self.am_i_owner(channel_id): @@ -1379,11 +1371,11 @@ class ProxyServer: attempt_key = f"ts_proxy:channel:{channel_id}:connection_attempt_time" self.redis_client.setex(attempt_key, 60, str(time.time())) - logging.info(f"Channel {channel_id} in connecting state - will start grace period after connection") + logger.info(f"Channel {channel_id} in connecting state - will start grace period after connection") return True except Exception as e: - logging.error(f"Error initializing channel {channel_id}: {e}", exc_info=True) + logger.error(f"Error initializing channel {channel_id}: {e}", exc_info=True) # Release ownership on failure self.release_ownership(channel_id) return False @@ -1419,11 +1411,11 @@ class ProxyServer: def stop_channel(self, channel_id): """Stop a channel with proper ownership handling""" try: - logging.info(f"Stopping channel {channel_id}") + logger.info(f"Stopping channel {channel_id}") # Only stop the actual stream manager if we're the owner if self.am_i_owner(channel_id): - logging.info(f"This worker ({self.worker_id}) is the owner - closing provider connection") + logger.info(f"This worker ({self.worker_id}) is the owner - closing provider connection") if channel_id in self.stream_managers: stream_manager = self.stream_managers[channel_id] @@ -1445,38 +1437,38 @@ class ProxyServer: break if stream_thread and stream_thread.is_alive(): - logging.info(f"Waiting for stream thread to terminate") + logger.info(f"Waiting for stream thread to terminate") try: # Very short timeout to prevent hanging the app stream_thread.join(timeout=2.0) if stream_thread.is_alive(): - logging.warning(f"Stream thread did not terminate within timeout") + logger.warning(f"Stream thread did not terminate within timeout") except RuntimeError: - logging.debug("Could not join stream thread (may be current thread)") + logger.debug("Could not join stream thread (may be current thread)") # Release ownership self.release_ownership(channel_id) - logging.info(f"Released ownership of channel {channel_id}") + logger.info(f"Released ownership of channel {channel_id}") # Always clean up local resources if channel_id in self.stream_managers: del self.stream_managers[channel_id] - logging.info(f"Removed stream manager for channel {channel_id}") + logger.info(f"Removed stream manager for channel {channel_id}") if channel_id in self.stream_buffers: del self.stream_buffers[channel_id] - logging.info(f"Removed stream buffer for channel {channel_id}") + logger.info(f"Removed stream buffer for channel {channel_id}") if channel_id in self.client_managers: del self.client_managers[channel_id] - logging.info(f"Removed client manager for channel {channel_id}") + logger.info(f"Removed client manager for channel {channel_id}") # Clean up Redis keys self._clean_redis_keys(channel_id) return True except Exception as e: - logging.error(f"Error stopping channel {channel_id}: {e}", exc_info=True) + logger.error(f"Error stopping channel {channel_id}: {e}", exc_info=True) return False def check_inactive_channels(self): @@ -1488,7 +1480,7 @@ class ProxyServer: channels_to_stop.append(channel_id) for channel_id in channels_to_stop: - logging.info(f"Auto-stopping inactive channel {channel_id}") + logger.info(f"Auto-stopping inactive channel {channel_id}") self.stop_channel(channel_id) def _cleanup_channel(self, channel_id: str) -> None: @@ -1529,7 +1521,7 @@ class ProxyServer: # Log client count periodically if time.time() % 30 < 1: # Every ~30 seconds - logging.info(f"Channel {channel_id} has {total_clients} clients, state: {channel_state}") + logger.info(f"Channel {channel_id} has {total_clients} clients, state: {channel_state}") # If in connecting or waiting_for_clients state, check grace period if channel_state in ["connecting", "waiting_for_clients"]: @@ -1543,7 +1535,7 @@ class ProxyServer: # If still connecting, give it more time if channel_state == "connecting": - logging.debug(f"Channel {channel_id} still connecting - not checking for clients yet") + logger.debug(f"Channel {channel_id} still connecting - not checking for clients yet") continue # If waiting for clients, check grace period @@ -1552,21 +1544,21 @@ class ProxyServer: time_since_ready = time.time() - connection_ready_time # Add this debug log - logging.debug(f"GRACE PERIOD CHECK: Channel {channel_id} in {channel_state} state, " + logger.debug(f"GRACE PERIOD CHECK: Channel {channel_id} in {channel_state} state, " f"time_since_ready={time_since_ready:.1f}s, grace_period={grace_period}s, " f"total_clients={total_clients}") if time_since_ready <= grace_period: # Still within grace period - logging.debug(f"Channel {channel_id} in grace period - {time_since_ready:.1f}s of {grace_period}s elapsed") + logger.debug(f"Channel {channel_id} in grace period - {time_since_ready:.1f}s of {grace_period}s elapsed") continue elif total_clients == 0: # Grace period expired with no clients - logging.info(f"Grace period expired ({time_since_ready:.1f}s > {grace_period}s) with no clients - stopping channel {channel_id}") + logger.info(f"Grace period expired ({time_since_ready:.1f}s > {grace_period}s) with no clients - stopping channel {channel_id}") self.stop_channel(channel_id) else: # Grace period expired but we have clients - mark channel as active - logging.info(f"Grace period expired with {total_clients} clients - marking channel {channel_id} as active") + logger.info(f"Grace period expired with {total_clients} clients - marking channel {channel_id} as active") old_state = "unknown" if metadata and b'state' in metadata: old_state = metadata[b'state'].decode('utf-8') @@ -1574,7 +1566,7 @@ class ProxyServer: "grace_period_ended_at": str(time.time()), "clients_at_activation": str(total_clients) }): - logging.info(f"Channel {channel_id} activated with {total_clients} clients after grace period") + logger.info(f"Channel {channel_id} activated with {total_clients} clients after grace period") # If active and no clients, start normal shutdown procedure elif channel_state not in ["connecting", "waiting_for_clients"] and total_clients == 0: # Check if there's a pending no-clients timeout @@ -1587,7 +1579,7 @@ class ProxyServer: try: disconnect_time = float(disconnect_value.decode('utf-8')) except (ValueError, TypeError) as e: - logging.error(f"Invalid disconnect time for channel {channel_id}: {e}") + logger.error(f"Invalid disconnect time for channel {channel_id}: {e}") current_time = time.time() @@ -1595,14 +1587,14 @@ class ProxyServer: # First time seeing zero clients, set timestamp if self.redis_client: self.redis_client.setex(disconnect_key, 60, str(current_time)) - logging.warning(f"No clients detected for channel {channel_id}, starting shutdown timer") + logger.warning(f"No clients detected for channel {channel_id}, starting shutdown timer") elif current_time - disconnect_time > getattr(Config, 'CHANNEL_SHUTDOWN_DELAY', 5): # We've had no clients for the shutdown delay period - logging.warning(f"No clients for {current_time - disconnect_time:.1f}s, stopping channel {channel_id}") + logger.warning(f"No clients for {current_time - disconnect_time:.1f}s, stopping channel {channel_id}") self.stop_channel(channel_id) else: # Still in shutdown delay period - logging.debug(f"Channel {channel_id} shutdown timer: " + logger.debug(f"Channel {channel_id} shutdown timer: " f"{current_time - disconnect_time:.1f}s of " f"{getattr(Config, 'CHANNEL_SHUTDOWN_DELAY', 5)}s elapsed") else: @@ -1611,14 +1603,14 @@ class ProxyServer: self.redis_client.delete(f"ts_proxy:channel:{channel_id}:last_client_disconnect_time") except Exception as e: - logging.error(f"Error in cleanup thread: {e}", exc_info=True) + logger.error(f"Error in cleanup thread: {e}", exc_info=True) time.sleep(getattr(Config, 'CLEANUP_CHECK_INTERVAL', 1)) thread = threading.Thread(target=cleanup_task, daemon=True) thread.name = "ts-proxy-cleanup" thread.start() - logging.info(f"Started TS proxy cleanup thread (interval: {getattr(Config, 'CLEANUP_CHECK_INTERVAL', 3)}s)") + logger.info(f"Started TS proxy cleanup thread (interval: {getattr(Config, 'CLEANUP_CHECK_INTERVAL', 3)}s)") def _check_orphaned_channels(self): """Check for orphaned channels in Redis (owner worker crashed)""" @@ -1648,16 +1640,16 @@ class ProxyServer: if client_count > 0: # Orphaned channel with clients - we could take ownership - logging.info(f"Found orphaned channel {channel_id} with {client_count} clients") + logger.info(f"Found orphaned channel {channel_id} with {client_count} clients") else: # Orphaned channel with no clients - clean it up - logging.info(f"Cleaning up orphaned channel {channel_id}") + logger.info(f"Cleaning up orphaned channel {channel_id}") self._clean_redis_keys(channel_id) except Exception as e: - logging.error(f"Error processing channel key {key}: {e}") + logger.error(f"Error processing channel key {key}: {e}") except Exception as e: - logging.error(f"Error checking orphaned channels: {e}") + logger.error(f"Error checking orphaned channels: {e}") def _clean_redis_keys(self, channel_id): """Clean up all Redis keys for a channel""" @@ -1671,10 +1663,10 @@ class ProxyServer: if all_keys: self.redis_client.delete(*all_keys) - logging.info(f"Cleaned up {len(all_keys)} Redis keys for channel {channel_id}") + logger.info(f"Cleaned up {len(all_keys)} Redis keys for channel {channel_id}") except Exception as e: - logging.error(f"Error cleaning Redis keys for channel {channel_id}: {e}") + logger.error(f"Error cleaning Redis keys for channel {channel_id}: {e}") def refresh_channel_registry(self): """Refresh TTL for active channels using standard keys""" @@ -1706,7 +1698,7 @@ class ProxyServer: # Only update if state is actually changing if current_state == new_state: - logging.debug(f"Channel {channel_id} state unchanged: {current_state}") + logger.debug(f"Channel {channel_id} state unchanged: {current_state}") return True # Prepare update data @@ -1723,10 +1715,10 @@ class ProxyServer: self.redis_client.hset(metadata_key, mapping=update_data) # Log the transition - logging.info(f"Channel {channel_id} state transition: {current_state or 'None'} → {new_state}") + logger.info(f"Channel {channel_id} state transition: {current_state or 'None'} → {new_state}") return True except Exception as e: - logging.error(f"Error updating channel state: {e}") + logger.error(f"Error updating channel state: {e}") return False diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 40059090..b95d8c8c 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -14,9 +14,6 @@ from apps.proxy.ts_proxy.server import logging as server_logging # Configure logging properly to ensure visibility logger = server_logging -# Print directly to output for critical messages (bypass logging system) -print("TS PROXY VIEWS INITIALIZED", file=sys.stderr) - # Initialize proxy server proxy_server = ProxyServer() From 69f4ceb1377be3ad373ec10dcdef7d2a637807b9 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 13 Mar 2025 15:56:04 -0500 Subject: [PATCH 7/7] Streaming is working fairly well now. --- apps/proxy/config.py | 4 ++-- apps/proxy/ts_proxy/views.py | 6 +++++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/apps/proxy/config.py b/apps/proxy/config.py index 055e00c0..acddc39d 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -29,7 +29,7 @@ class TSConfig(BaseConfig): MAX_RETRIES = 3 # maximum connection retry attempts # Buffer settings - INITIAL_BEHIND_CHUNKS = 100 # How many chunks behind to start a client + INITIAL_BEHIND_CHUNKS = 4 # How many chunks behind to start a client CHUNK_BATCH_SIZE = 5 # How many chunks to fetch in one batch KEEPALIVE_INTERVAL = 0.5 # Seconds between keepalive packets when at buffer head @@ -52,5 +52,5 @@ class TSConfig(BaseConfig): # TS packets are 188 bytes # Make chunk size a multiple of TS packet size for perfect alignment # ~1MB is ideal for streaming (matches typical media buffer sizes) - BUFFER_CHUNK_SIZE = 188 * 5644 # ~1MB (exactly 1,061,072 bytes) + BUFFER_CHUNK_SIZE = 188 * 1361 # ~256KB diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index b95d8c8c..399e93c3 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -243,7 +243,11 @@ def stream_ts(request, channel_id): return # Client state tracking - use config for initial position - local_index = max(0, buffer.index - Config.INITIAL_BEHIND_CHUNKS) + initial_behind = getattr(Config, 'INITIAL_BEHIND_CHUNKS', 10) + current_buffer_index = buffer.index + local_index = max(0, current_buffer_index - initial_behind) + logger.debug(f"[{client_id}] Buffer at {current_buffer_index}, starting {initial_behind} chunks behind at index {local_index}") + initial_position = local_index last_yield_time = time.time() empty_reads = 0