From 987a8606a847a44ebfb633ebc080947ec0e27b7e Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 13 Mar 2025 09:48:37 -0500 Subject: [PATCH] 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