Streaming works with a single worker correctly.

apps/proxy/ts_proxy/server.py

@@ -11,11 +11,23 @@ import requests
 import threading
 import logging
 import socket 
+import random
 from collections import deque
 import time
+import sys
 from typing import Optional, Set, Deque, Dict
 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)
+
 class StreamManager:
     """Manages a connection to a TS stream with continuity tracking"""
     
@@ -72,7 +84,7 @@ class StreamManager:
             self.socket.close()
 
     def _process_complete_packets(self):
-        """Process only complete TS packets from buffer with continuity tracking"""
+        """Process TS packets with detailed logging"""
         try:
             # Find sync byte if needed
             if not self.sync_found and len(self.recv_buffer) >= 376:
@@ -102,50 +114,41 @@ class StreamManager:
             
             if packet_count == 0:
                 return False
-                
-            # Extract and process packets with continuity correction
-            processed_buffer = bytearray()
             
-            for i in range(packet_count):
-                packet_start = i * 188
-                packet = bytearray(self.recv_buffer[packet_start:packet_start + 188])
-                
-                # Check sync byte
-                if packet[0] == 0x47:
-                    # Extract PID (13 bits from bytes 1-2)
-                    pid = ((packet[1] & 0x1F) << 8) | packet[2]
-                    
-                    # Only process continuity for packets that have it
-                    # Skip adaptation-only packets and null packets
-                    if (packet[3] & 0x10) and pid != 0x1FFF:  # Has payload and not null packet
-                        # Extract continuity counter (4 bits from byte 3)
-                        cc = packet[3] & 0x0F
-                        
-                        # Correct continuity counter if we're tracking this PID
-                        if pid in self.continuity_counters:
-                            expected = (self.continuity_counters[pid] + 1) & 0x0F
-                            if cc != expected:
-                                # Rewrite continuity counter to maintain sequence
-                                packet[3] = (packet[3] & 0xF0) | expected
-                        
-                        # Update counter for next time
-                        self.continuity_counters[pid] = packet[3] & 0x0F
-                
-                # Add processed packet to buffer
-                processed_buffer.extend(packet)
+            # Log packet processing
+            logging.debug(f"Processing {packet_count} TS packets ({packet_count * 188} bytes)")
+            
+            # 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 processed packets to buffer
-            if processed_buffer:
-                self.buffer.add_chunk(bytes(processed_buffer))
-                return True
+            # 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}")
+                
+                before_index = self.buffer.index
+                success = self.buffer.add_chunk(bytes(packets))
+                after_index = self.buffer.index
+                
+                # Log successful write
+                if success:
+                    logging.debug(f"Added chunk: {packet_count} packets, buffer index {before_index} → {after_index}")
+                else:
+                    logging.warning("Failed to add chunk to buffer")
+                    
+                return success
                 
             return False
             
         except Exception as e:
-            logging.error(f"Error processing TS packets: {e}")
+            logging.error(f"Error processing TS packets: {e}", exc_info=True)
             self.sync_found = False  # Reset sync state on error
             return False
 
@@ -336,48 +339,174 @@ class StreamBuffer:
             return False
     
     def get_chunks(self, start_index=None):
-        """Get chunks from the buffer starting at the given index"""
+        """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}")
+            
             if not self.redis_client:
-                # Local fallback
+                # Local fallback (unchanged)
                 chunks = []
                 with self.lock:
                     for idx in sorted(self.local_cache.keys()):
                         if start_index is None or idx > start_index:
                             chunks.append(self.local_cache[idx])
+                logging.debug(f"[{request_id}] Local fallback returned {len(chunks)} chunks")
                 return chunks
-                
+            
             # If no start_index provided, use most recent chunks
             if start_index is None:
-                start_index = max(0, self.index - 50)
+                start_index = max(0, self.index - 10)  # Start closer to current position
+                logging.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)
+            # Get current index from Redis (cached to reduce load)
+            if not hasattr(self, '_last_redis_check') or time.time() - self._last_redis_check > 0.5:
+                current_index = int(self.redis_client.get(self.buffer_index_key) or 0)
+                self._current_index = current_index
+                self._last_redis_check = time.time()
+                logging.debug(f"[{request_id}] Updated buffer index from Redis: {current_index}")
+            else:
+                current_index = self._current_index
+                logging.debug(f"[{request_id}] Using cached buffer index: {current_index}")
             
-            # Calculate range of chunks to retrieve (up to 50 at a time)
+            # Calculate range of chunks to retrieve
             start_id = start_index + 1
-            end_id = min(current_index + 1, start_id + 50)
+            chunks_behind = current_index - start_id
+            
+            # 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}")
+            elif chunks_behind > 50:
+                fetch_count = 10  
+                logging.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}")
+            else:
+                fetch_count = 3
+                logging.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})")
                 return []
             
-            # Retrieve chunks in a pipeline
-            pipe = self.redis_client.pipeline()
-            for idx in range(start_id, end_id):
-                chunk_key = f"{self.buffer_prefix}{idx}"
-                pipe.get(chunk_key)
+            # Log the range we're retrieving
+            logging.debug(f"[{request_id}] Retrieving chunks {start_id} to {end_id-1} (total: {end_id-start_id})")
             
-            results = pipe.execute()
-            chunks = [r for r in results if r]
+            # Try local cache first to reduce Redis load
+            chunks = []
+            missing_indices = []
+            
+            for idx in range(start_id, end_id):
+                if idx in self.local_cache:
+                    chunks.append(self.local_cache[idx])
+                else:
+                    missing_indices.append(idx)
+            
+            # Log cache hit rate
+            if end_id > start_id:
+                cache_hit_rate = (len(chunks) / (end_id - start_id)) * 100
+                logging.debug(f"[{request_id}] Local cache hit rate: {cache_hit_rate:.1f}% ({len(chunks)}/{end_id-start_id})")
+            
+            # Only retrieve missing chunks from Redis
+            if missing_indices:
+                logging.debug(f"[{request_id}] Fetching {len(missing_indices)} missing chunks from Redis")
+                pipe = self.redis_client.pipeline()
+                for idx in missing_indices:
+                    chunk_key = f"{self.buffer_prefix}{idx}"
+                    pipe.get(chunk_key)
+                
+                results = pipe.execute()
+                
+                # Count non-None results
+                found_chunks = sum(1 for r in results if r is not None)
+                logging.debug(f"[{request_id}] Found {found_chunks}/{len(missing_indices)} chunks in Redis")
+                
+                # Process results and update cache
+                for i, result in enumerate(results):
+                    if result:
+                        idx = missing_indices[i]
+                        chunks.append(result)
+                        # Update local cache
+                        self.local_cache[idx] = result
+                    else:
+                        # Log missing chunks
+                        logging.warning(f"[{request_id}] Chunk {missing_indices[i]} missing from Redis")
             
             # Update local tracking
             if chunks:
                 self.index = end_id - 1
+                
+            # 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)")
             
             return chunks
             
         except Exception as e:
-            logging.error(f"Error getting chunks from buffer: {e}")
+            logging.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:
+                # Local fallback
+                chunks = []
+                with self.lock:
+                    for idx in range(start_index + 1, start_index + count + 1):
+                        if idx in self.local_cache:
+                            chunks.append(self.local_cache[idx])
+                return chunks
+            
+            # Calculate range to retrieve
+            start_id = start_index + 1
+            end_id = start_id + count
+            
+            # Get current buffer position
+            current_index = int(self.redis_client.get(self.buffer_index_key) or 0)
+            
+            # If requesting beyond current buffer, return what we have
+            if start_id > current_index:
+                return []
+                
+            # Cap end at current buffer position
+            end_id = min(end_id, current_index + 1)
+            
+            # Get chunks from local cache first
+            chunks = []
+            missing_indices = []
+            
+            for idx in range(start_id, end_id):
+                if idx in self.local_cache:
+                    chunks.append(self.local_cache[idx])
+                else:
+                    missing_indices.append(idx)
+            
+            # Get missing chunks from Redis
+            if missing_indices:
+                pipe = self.redis_client.pipeline()
+                for idx in missing_indices:
+                    chunk_key = f"{self.buffer_prefix}{idx}"
+                    pipe.get(chunk_key)
+                
+                results = pipe.execute()
+                
+                # Process results
+                for i, result in enumerate(results):
+                    if result:
+                        idx = missing_indices[i]
+                        chunks.insert(i, result)  # Insert at correct position
+                        self.local_cache[idx] = result
+            
+            return chunks
+            
+        except Exception as e:
+            logging.error(f"Error getting exact chunks: {e}", exc_info=True)
             return []
 
 class ClientManager:
@@ -534,46 +663,76 @@ class ProxyServer:
             logging.error(f"Failed to connect to Redis: {e}")
     
     def initialize_channel(self, url, channel_id):
-        """Initialize a channel"""
+        """Initialize a channel with enhanced logging"""
         try:
+            logging.info(f"Initializing channel {channel_id} with URL: {url}")
+            
             # Clean up any existing Redis entries for this channel
             if self.redis_client:
-                # Delete index key
+                # Get the current state before cleanup for logging
                 index_key = f"ts_proxy:buffer:{channel_id}:index"
+                old_index = self.redis_client.get(index_key)
+                
+                # Count existing chunks for logging
+                pattern = f"ts_proxy:buffer:{channel_id}:chunk:*"
+                old_chunk_count = 0
+                cursor = 0
+                while True:
+                    cursor, keys = self.redis_client.scan(cursor, pattern, 100)
+                    old_chunk_count += len(keys)
+                    if cursor == 0:
+                        break
+                
+                logging.info(f"Found existing channel data: index={old_index}, chunks={old_chunk_count}")
+                
+                # Delete index key
                 self.redis_client.delete(index_key)
+                logging.debug(f"Deleted Redis index key: {index_key}")
                 
                 # Delete all chunks for this channel
-                pattern = f"ts_proxy:buffer:{channel_id}:chunk:*"
+                deleted_chunks = 0
                 cursor = 0
                 while True:
                     cursor, keys = self.redis_client.scan(cursor, pattern, 100)
                     if keys:
                         self.redis_client.delete(*keys)
+                        deleted_chunks += len(keys)
                     if cursor == 0:
                         break
                 
+                logging.info(f"Deleted {deleted_chunks} Redis chunks for channel {channel_id}")
+                
                 # Register this channel as active
-                self.redis_client.set(f"ts_proxy:active_channel:{channel_id}", "1", ex=60)
+                activity_key = f"ts_proxy:active_channel:{channel_id}"
+                self.redis_client.set(activity_key, "1", ex=60)
+                logging.debug(f"Set channel activity key: {activity_key}")
             
             # Create buffer and stream manager
             buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client)
+            logging.debug(f"Created StreamBuffer for channel {channel_id}")
+            
             self.stream_buffers[channel_id] = buffer
             
             stream_manager = StreamManager(url, buffer)
+            logging.debug(f"Created StreamManager for channel {channel_id}")
+            
             self.stream_managers[channel_id] = stream_manager
             
-            # FIX: Create client manager without passing channel_id
-            client_manager = ClientManager()  # No arguments here
+            # Create client manager
+            client_manager = ClientManager()
             self.client_managers[channel_id] = client_manager
-            # Set up the cleanup timer afterwards
-            client_manager.start_cleanup_timer(self, channel_id)
+            logging.debug(f"Created ClientManager for channel {channel_id}")
             
             # Start stream manager
-            threading.Thread(target=stream_manager.run, daemon=True).start()
+            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}")
+            
             return True
             
         except Exception as e:
-            logging.error(f"Error initializing channel: {e}")
+            logging.error(f"Error initializing channel {channel_id}: {e}", exc_info=True)
             return False
     
     def stop_channel(self, channel_id):
@@ -622,3 +781,4 @@ class ProxyServer:
         """Stop all channels and cleanup"""
         for channel_id in list(self.stream_managers.keys()):
             self.stop_channel(channel_id)
+

apps/proxy/ts_proxy/views.py

@@ -3,13 +3,27 @@ import threading
 import logging
 import time
 import random
+import sys
+import os
 from django.http import StreamingHttpResponse, JsonResponse
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.http import require_http_methods
-from apps.proxy.config import TSConfig as Config  # Change this line
+from apps.proxy.config import TSConfig as Config
 from .server import ProxyServer
 
+# 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)
+
+# Print directly to output for critical messages (bypass logging system)
+print("TS PROXY VIEWS INITIALIZED", file=sys.stderr)
+
+# Initialize proxy server
 proxy_server = ProxyServer()
 
 @csrf_exempt
@@ -55,53 +69,144 @@ def initialize_stream(request, channel_id):
 @csrf_exempt
 @require_http_methods(["GET"])
 def stream_ts(request, channel_id):
-    """Stream TS data to client"""
+    """Stream TS data to client with synchronized delivery timing"""
     if channel_id not in proxy_server.stream_managers:
         return JsonResponse({'error': 'Channel not found'}, status=404)
     
     def generate():
-        client_id = int(time.time() * 1000) + random.randint(1, 1000)
+        client_id = f"client_{int(time.time() * 1000)}_{random.randint(1000, 9999)}"
         
         try:
+            logger.info(f"[{client_id}] New client connected to channel {channel_id}")
+            
             # Add client to manager
             client_manager = proxy_server.client_managers[channel_id]
             client_manager.add_client(client_id)
             
-            # Get buffer
+            # Get buffer and set initial position for this client
             buffer = proxy_server.stream_buffers.get(channel_id)
             if not buffer:
-                logger.error(f"No buffer found for channel {channel_id}")
+                logger.error(f"[{client_id}] No buffer found for channel {channel_id}")
                 return
             
-            # Start at current position
-            local_index = None
+            # Start 30 chunks behind current position
+            local_index = max(0, buffer.index - 30)
+            initial_position = local_index
+            
+            # Client state tracking
+            last_yield_time = time.time()
+            empty_reads = 0
+            bytes_sent = 0
+            chunks_sent = 0
+            stream_start_time = time.time()
+            
+            # Critical timing parameters
+            ts_packet_size = 188
+            target_bitrate = 8000000  # Target ~8 Mbps (typical for HD)
+            packets_per_second = target_bitrate / (8 * ts_packet_size)
+            packets_per_batch = 200   # How many packets to send in one batch
+            target_batch_time = packets_per_batch / packets_per_second  # Time one batch should take
+
+            logger.info(f"[{client_id}] Starting stream at index {initial_position} (buffer at {buffer.index})")
+            logger.info(f"[{client_id}] Target bitrate: {target_bitrate/1000000:.1f} Mbps, packets/sec: {packets_per_second:.1f}")
+            
+            # For rate limiting
+            batch_start_time = time.time()
             
             # Main streaming loop
             while True:
-                chunks = buffer.get_chunks(local_index)
+                # Log buffer state periodically
+                if empty_reads % 50 == 0:
+                    current_buffer_index = buffer.index
+                    chunks_behind = current_buffer_index - local_index
+                    logger.debug(f"[{client_id}] Buffer state: Client={local_index}, Buffer={current_buffer_index}, Behind={chunks_behind}")
+                
+                # Get exactly 5 chunks (or whatever is available)
+                chunks = buffer.get_chunks_exact(local_index, 5)
                 
                 if chunks:
-                    # Send chunks to client
-                    for chunk in chunks:
-                        yield chunk
+                    # 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()
                     
-                    # Update local index
-                    local_index = buffer.index
+                    # Log chunk retrieval
+                    chunk_sizes = [len(c) for c in chunks]
+                    total_size = sum(chunk_sizes)
+                    start_idx = local_index + 1
+                    end_idx = local_index + len(chunks)
+                    
+                    logger.debug(f"[{client_id}] Retrieved {len(chunks)} chunks ({total_size} bytes) from index {start_idx} to {end_idx}")
+                    
+                    # Track how many packets we've sent in this batch
+                    packets_sent_in_batch = 0
+                    
+                    # Send chunks with precise timing to maintain PCR synchronization
+                    for chunk in chunks:
+                        # Count packets in this chunk
+                        packets_in_chunk = len(chunk) // ts_packet_size
+                        
+                        # Send the chunk
+                        bytes_sent += len(chunk)
+                        chunks_sent += 1
+                        yield chunk
+                        
+                        packets_sent_in_batch += packets_in_chunk
+                        
+                        # Calculate elapsed time to maintain target bitrate
+                        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 = target_time - elapsed
+                            # Limit max sleep to prevent long pauses
+                            sleep_time = min(sleep_time, 0.05)
+                            if sleep_time > 0.001:  # Only sleep for meaningful amounts
+                                time.sleep(sleep_time)
+                    
+                    # After sending a complete batch, ensure proper pacing
+                    batch_elapsed = time.time() - batch_start_time
+                    if batch_elapsed < target_batch_time:
+                        time.sleep(target_batch_time - batch_elapsed)
+                    
+                    # Update progress stats 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}] Streaming stats: {chunks_sent} chunks, {bytes_sent/1024:.1f} KB sent, {rate:.1f} KB/s")
+                    
+                    # Update local index to last chunk received
+                    local_index = end_idx
+                    last_yield_time = time.time()
+                    empty_reads = 0
                 else:
-                    # No data available, wait briefly
-                    time.sleep(0.1)
+                    # No new data yet
+                    empty_reads += 1
+                    sleep_time = 0.1
+                    time.sleep(sleep_time)
+                    
+                    # Log empty reads periodically 
+                    if empty_reads % 50 == 0:
+                        logger.debug(f"[{client_id}] Waiting for new chunks beyond index {local_index} (buffer at {buffer.index})")
+                    
+                    # Safety timeout
+                    if time.time() - last_yield_time > 10:
+                        logger.warning(f"[{client_id}] No data for 10 seconds, disconnecting")
+                        break
                     
         except Exception as e:
-            logger.error(f"Streaming error for client {client_id}, channel {channel_id}: {e}")
+            logger.error(f"[{client_id}] Streaming error: {e}", exc_info=True)
         finally:
             # Clean up client
+            elapsed = time.time() - stream_start_time
             if channel_id in proxy_server.client_managers:
                 remaining = proxy_server.client_managers[channel_id].remove_client(client_id)
-                logger.info(f"Client {client_id} disconnected from channel {channel_id} ({remaining} clients remaining)")
+                logger.info(f"[{client_id}] Disconnected after {elapsed:.2f}s, sent {bytes_sent/1024:.1f} KB in {chunks_sent} chunks ({remaining} clients left)")
     
     # Create streaming response
     response = StreamingHttpResponse(generate(), content_type='video/MP2T')
     response['Cache-Control'] = 'no-cache, no-store'
+    response['X-Accel-Buffering'] = 'no'
     return response
 
 @csrf_exempt