mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-17 16:50:53 +00:00
fix(proxy): enhance connection management and event handling during streaming operations
- Improved database connection management by ensuring `close_old_connections()` is called in various methods to prevent connection leaks. - Updated event dispatching to run asynchronously in gevent, preventing blocking during live-proxy and streaming paths.
This commit is contained in:
SergeantPanda
parent
db40421faa
commit
c389462dde
7 changed files with 194 additions and 101 deletions
|
|
@ -650,48 +650,48 @@ class StreamManager:
|
|||
logger.debug(f"Closing existing HTTP connections before establishing transcode connection for channel {self.channel_id}")
|
||||
self._close_connection()
|
||||
|
||||
close_old_connections()
|
||||
channel = get_stream_object(self.channel_id)
|
||||
try:
|
||||
channel = get_stream_object(self.channel_id)
|
||||
|
||||
# Use FFmpeg specifically for HLS streams
|
||||
if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg:
|
||||
from core.models import StreamProfile
|
||||
try:
|
||||
stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True)
|
||||
logger.info("Using FFmpeg stream profile for unsupported proxy content (HLS/RTSP/UDP)")
|
||||
except StreamProfile.DoesNotExist:
|
||||
# Fall back to channel's profile if FFmpeg not found
|
||||
# Use FFmpeg specifically for HLS streams
|
||||
if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg:
|
||||
from core.models import StreamProfile
|
||||
try:
|
||||
stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True)
|
||||
logger.info("Using FFmpeg stream profile for unsupported proxy content (HLS/RTSP/UDP)")
|
||||
except StreamProfile.DoesNotExist:
|
||||
# Fall back to channel's profile if FFmpeg not found
|
||||
stream_profile = channel.get_stream_profile()
|
||||
logger.warning(f"FFmpeg profile not found, using channel default profile for channel: {self.channel_id}")
|
||||
else:
|
||||
stream_profile = channel.get_stream_profile()
|
||||
logger.warning(f"FFmpeg profile not found, using channel default profile for channel: {self.channel_id}")
|
||||
else:
|
||||
stream_profile = channel.get_stream_profile()
|
||||
|
||||
# Build and start transcode command
|
||||
self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent)
|
||||
# Build and start transcode command
|
||||
self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent)
|
||||
|
||||
# Store stream command for efficient log parser routing
|
||||
self.stream_command = stream_profile.command
|
||||
# Map actual commands to parser types for direct routing
|
||||
command_to_parser = {
|
||||
'ffmpeg': 'ffmpeg',
|
||||
'cvlc': 'vlc',
|
||||
'vlc': 'vlc',
|
||||
'streamlink': 'streamlink'
|
||||
}
|
||||
self.parser_type = command_to_parser.get(self.stream_command.lower())
|
||||
if self.parser_type:
|
||||
logger.debug(f"Using {self.parser_type} parser for log parsing (command: {self.stream_command})")
|
||||
else:
|
||||
logger.debug(f"Unknown stream command '{self.stream_command}', will use auto-detection for log parsing")
|
||||
# Store stream command for efficient log parser routing
|
||||
self.stream_command = stream_profile.command
|
||||
# Map actual commands to parser types for direct routing
|
||||
command_to_parser = {
|
||||
'ffmpeg': 'ffmpeg',
|
||||
'cvlc': 'vlc',
|
||||
'vlc': 'vlc',
|
||||
'streamlink': 'streamlink'
|
||||
}
|
||||
self.parser_type = command_to_parser.get(self.stream_command.lower())
|
||||
if self.parser_type:
|
||||
logger.debug(f"Using {self.parser_type} parser for log parsing (command: {self.stream_command})")
|
||||
else:
|
||||
logger.debug(f"Unknown stream command '{self.stream_command}', will use auto-detection for log parsing")
|
||||
|
||||
# For UDP streams, remove any user_agent parameters from the command
|
||||
if hasattr(self, 'stream_type') and self.stream_type == StreamType.UDP:
|
||||
# Filter out any arguments that contain the user_agent value or related headers
|
||||
self.transcode_cmd = [arg for arg in self.transcode_cmd if self.user_agent not in arg and 'user-agent' not in arg.lower() and 'user_agent' not in arg.lower()]
|
||||
logger.debug(f"Removed user_agent parameters from UDP stream command for channel: {self.channel_id}")
|
||||
|
||||
# Profile lookup is done; release the pool slot before a long-lived ffmpeg process.
|
||||
close_old_connections()
|
||||
# For UDP streams, remove any user_agent parameters from the command
|
||||
if hasattr(self, 'stream_type') and self.stream_type == StreamType.UDP:
|
||||
# Filter out any arguments that contain the user_agent value or related headers
|
||||
self.transcode_cmd = [arg for arg in self.transcode_cmd if self.user_agent not in arg and 'user-agent' not in arg.lower() and 'user_agent' not in arg.lower()]
|
||||
logger.debug(f"Removed user_agent parameters from UDP stream command for channel: {self.channel_id}")
|
||||
finally:
|
||||
# Release the pool slot before posix_spawn or before returning on profile errors.
|
||||
close_old_connections()
|
||||
|
||||
logger.debug(f"Starting transcode process: {self.transcode_cmd} for channel: {self.channel_id}")
|
||||
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ import gevent
|
|||
from apps.proxy.config import TSConfig as Config
|
||||
from apps.channels.models import Channel, Stream
|
||||
from core.utils import log_system_event
|
||||
from django.db import close_old_connections
|
||||
from ...server import ProxyServer
|
||||
from ...utils import create_ts_packet, get_logger
|
||||
from ...redis_keys import RedisKeys
|
||||
|
|
@ -591,62 +590,59 @@ class StreamGenerator:
|
|||
|
||||
def _cleanup(self):
|
||||
"""Clean up resources and report final statistics."""
|
||||
try:
|
||||
# Client cleanup
|
||||
elapsed = time.time() - self.stream_start_time
|
||||
local_clients = 0
|
||||
total_clients = 0
|
||||
proxy_server = ProxyServer.get_instance()
|
||||
# Client cleanup
|
||||
elapsed = time.time() - self.stream_start_time
|
||||
local_clients = 0
|
||||
total_clients = 0
|
||||
proxy_server = ProxyServer.get_instance()
|
||||
|
||||
# Release M3U profile stream allocation if this is the last client
|
||||
stream_released = False
|
||||
if proxy_server.redis_client:
|
||||
try:
|
||||
if self.channel_id in proxy_server.client_managers:
|
||||
client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
|
||||
# Pool slots are global; the last client on any worker must release.
|
||||
if client_count <= 1:
|
||||
# Release M3U profile stream allocation if this is the last client
|
||||
stream_released = False
|
||||
if proxy_server.redis_client:
|
||||
try:
|
||||
if self.channel_id in proxy_server.client_managers:
|
||||
client_count = proxy_server.client_managers[self.channel_id].get_total_client_count()
|
||||
# Pool slots are global; the last client on any worker must release.
|
||||
if client_count <= 1:
|
||||
try:
|
||||
try:
|
||||
try:
|
||||
obj = Channel.objects.get(uuid=self.channel_id)
|
||||
except (Channel.DoesNotExist, Exception):
|
||||
obj = Stream.objects.get(stream_hash=self.channel_id)
|
||||
stream_released = obj.release_stream()
|
||||
if stream_released:
|
||||
logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
|
||||
else:
|
||||
logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.client_id}] Error checking stream data for release: {e}")
|
||||
obj = Channel.objects.get(uuid=self.channel_id)
|
||||
except (Channel.DoesNotExist, Exception):
|
||||
obj = Stream.objects.get(stream_hash=self.channel_id)
|
||||
stream_released = obj.release_stream()
|
||||
if stream_released:
|
||||
logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}")
|
||||
else:
|
||||
logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}")
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}")
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.client_id}] Error checking stream data for release: {e}")
|
||||
|
||||
if self.channel_id in proxy_server.client_managers:
|
||||
client_manager = proxy_server.client_managers[self.channel_id]
|
||||
if self.client_id in client_manager.clients:
|
||||
local_clients = client_manager.remove_client(self.client_id)
|
||||
else:
|
||||
local_clients = client_manager.get_client_count()
|
||||
total_clients = client_manager.get_total_client_count()
|
||||
logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})")
|
||||
if self.channel_id in proxy_server.client_managers:
|
||||
client_manager = proxy_server.client_managers[self.channel_id]
|
||||
if self.client_id in client_manager.clients:
|
||||
local_clients = client_manager.remove_client(self.client_id)
|
||||
else:
|
||||
local_clients = client_manager.get_client_count()
|
||||
total_clients = client_manager.get_total_client_count()
|
||||
logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})")
|
||||
|
||||
# Log client disconnect event
|
||||
try:
|
||||
log_system_event(
|
||||
'client_disconnect',
|
||||
channel_id=self.channel_id,
|
||||
channel_name=self.channel_name,
|
||||
client_ip=self.client_ip,
|
||||
client_id=self.client_id,
|
||||
user_agent=self.client_user_agent[:100] if self.client_user_agent else None,
|
||||
duration=round(elapsed, 2),
|
||||
bytes_sent=self.bytes_sent,
|
||||
username=self.user.username if self.user else None
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Could not log client disconnect event: {e}")
|
||||
finally:
|
||||
close_old_connections()
|
||||
# Log client disconnect event
|
||||
try:
|
||||
log_system_event(
|
||||
'client_disconnect',
|
||||
channel_id=self.channel_id,
|
||||
channel_name=self.channel_name,
|
||||
client_ip=self.client_ip,
|
||||
client_id=self.client_id,
|
||||
user_agent=self.client_user_agent[:100] if self.client_user_agent else None,
|
||||
duration=round(elapsed, 2),
|
||||
bytes_sent=self.bytes_sent,
|
||||
username=self.user.username if self.user else None
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Could not log client disconnect event: {e}")
|
||||
|
||||
|
||||
def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None):
|
||||
|
|
|
|||
|
|
@ -760,7 +760,6 @@ class ProxyServer:
|
|||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Could not log channel start event: {e}")
|
||||
finally:
|
||||
close_old_connections()
|
||||
|
||||
# Create client manager with channel_id, redis_client AND worker_id (only if not already exists)
|
||||
|
|
@ -1367,11 +1366,7 @@ class ProxyServer:
|
|||
return
|
||||
|
||||
def _log_stop():
|
||||
try:
|
||||
close_old_connections()
|
||||
log_system_event('channel_stop', **stop_event_data)
|
||||
except Exception as e:
|
||||
logger.error(f"Could not log channel stop event: {e}")
|
||||
log_system_event('channel_stop', **stop_event_data)
|
||||
|
||||
gevent.spawn(_log_stop)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue