# """ # FIXED Call Manager for Handling Multiple Concurrent Calls # Manages call sessions with proper concurrency limits and isolation # """ # import asyncio # from typing import Dict, Optional # from fastapi import WebSocket # from services.log_utils import Log # from services.call_session import CallSession # from config import Config # class CallManager: # """ # Manages multiple concurrent call sessions with proper isolation. # Each call gets its own completely isolated session. # """ # def __init__(self, max_concurrent_calls: int = 50): # self.max_concurrent_calls = max_concurrent_calls # self.active_sessions: Dict[str, CallSession] = {} # self.call_semaphore = asyncio.Semaphore(max_concurrent_calls) # self.session_lock = asyncio.Lock() # Log.info(f"Call Manager initialized - Max concurrent calls: {max_concurrent_calls}") # async def create_call_session(self, session_id: str, websocket: WebSocket) -> Optional[CallSession]: # """ # Create a new isolated call session. # Returns None if max concurrent calls reached. # """ # try: # # Check if we can accept more calls # if len(self.active_sessions) >= self.max_concurrent_calls: # Log.warning(f"โŒ Max concurrent calls reached ({self.max_concurrent_calls}). Rejecting session {session_id}") # await websocket.close(code=1013, reason="Service overloaded - too many concurrent calls") # return None # # Create isolated session using fixed version # session = CallSession(session_id, websocket) # # Initialize session # if not await session.initialize(): # Log.error(f"โŒ Failed to initialize session {session_id}") # return None # # Register session # async with self.session_lock: # self.active_sessions[session_id] = session # Log.info(f"โœ… Created call session {session_id} (Active: {len(self.active_sessions)}/{self.max_concurrent_calls})") # return session # except Exception as e: # Log.error(f"โŒ Error creating call session {session_id}: {e}") # return None # async def remove_call_session(self, session_id: str) -> None: # """Remove and cleanup a call session.""" # try: # Log.info(f"๐Ÿ”„ Attempting to remove call session {session_id}") # async with self.session_lock: # if session_id in self.active_sessions: # session = self.active_sessions.pop(session_id) # Log.info(f"๐Ÿงน Calling cleanup for session {session_id}") # # Run cleanup outside lock to avoid blocking other operations # try: # await session.cleanup() # except Exception as cleanup_error: # Log.error(f"โŒ Error during cleanup for session {session_id}: {cleanup_error}") # Log.info(f"๐Ÿ—‘๏ธ Removed call session {session_id} (Active: {len(self.active_sessions)}/{self.max_concurrent_calls})") # else: # # Session already removed (race condition - this is OK) # Log.debug(f"Session {session_id} already removed (likely race condition)") # except Exception as e: # Log.error(f"โŒ Error removing call session {session_id}: {e}") # import traceback # traceback.print_exc() # def get_session(self, session_id: str) -> Optional[CallSession]: # """Get an active call session.""" # return self.active_sessions.get(session_id) # def get_active_sessions_count(self) -> int: # """Get number of active sessions.""" # return len(self.active_sessions) # def get_system_status(self) -> Dict: # """Get system-wide status for monitoring.""" # return { # "active_sessions": len(self.active_sessions), # "max_concurrent_calls": self.max_concurrent_calls, # "utilization_percent": (len(self.active_sessions) / self.max_concurrent_calls) * 100, # "available_slots": self.max_concurrent_calls - len(self.active_sessions), # "session_ids": list(self.active_sessions.keys()) # } # def get_session_statuses(self) -> Dict[str, Dict]: # """Get status of all active sessions.""" # return { # session_id: session.get_status() # for session_id, session in self.active_sessions.items() # } # async def cleanup_all_sessions(self) -> None: # """Cleanup all active sessions (for shutdown).""" # try: # Log.info(f"๐Ÿงน Cleaning up {len(self.active_sessions)} active sessions") # # Cleanup all sessions concurrently # cleanup_tasks = [] # for session_id, session in self.active_sessions.items(): # cleanup_tasks.append(session.cleanup()) # if cleanup_tasks: # await asyncio.gather(*cleanup_tasks, return_exceptions=True) # self.active_sessions.clear() # Log.info("โœ… All sessions cleaned up") # except Exception as e: # Log.error(f"โŒ Error cleaning up sessions: {e}") # # Global call manager instance # # Updated to support 50 concurrent calls # call_manager = CallManager(max_concurrent_calls=50) """ FIXED Call Manager for Handling Multiple Concurrent Calls Manages call sessions with proper concurrency limits and isolation """ import asyncio from typing import Dict, Optional from fastapi import WebSocket from services.log_utils import Log from services.call_session import CallSession from config import Config class CallManager: """ Manages multiple concurrent call sessions with proper isolation. Each call gets its own completely isolated session. """ def __init__(self, max_concurrent_calls: int = 50): self.max_concurrent_calls = max_concurrent_calls self.active_sessions: Dict[str, CallSession] = {} self.call_semaphore = asyncio.Semaphore(max_concurrent_calls) self.session_lock = asyncio.Lock() Log.info(f"Call Manager initialized - Max concurrent calls: {max_concurrent_calls}") async def create_call_session(self, session_id: str, websocket: WebSocket) -> Optional[CallSession]: """ Create a new isolated call session with proper error handling. Returns None if max concurrent calls reached or if creation fails. """ try: # Check if we can accept more calls if len(self.active_sessions) >= self.max_concurrent_calls: Log.warning(f"โŒ Max concurrent calls reached ({self.max_concurrent_calls}). Rejecting session {session_id}") await websocket.close(code=1013, reason="Service overloaded - too many concurrent calls") return None # Create isolated session using fixed version session = CallSession(session_id, websocket) # Initialize session if not await session.initialize(): Log.error(f"โŒ Failed to initialize session {session_id}") return None # Register session async with self.session_lock: self.active_sessions[session_id] = session Log.info(f"โœ… Created call session {session_id} (Active: {len(self.active_sessions)}/{self.max_concurrent_calls})") return session except RuntimeError as e: # Handle event loop mismatch errors gracefully error_msg = str(e) if "different loop" in error_msg or "attached to a different loop" in error_msg: Log.warning(f"โš ๏ธ Event loop mismatch during session creation for {session_id} - retrying...") # Let the caller retry (FastAPI will handle retry) return None else: Log.error(f"โŒ Runtime error creating call session {session_id}: {e}") return None except Exception as e: Log.error(f"โŒ Error creating call session {session_id}: {e}") return None async def remove_call_session(self, session_id: str) -> None: """Remove and cleanup a call session with proper event loop handling.""" session = None try: Log.info(f"๐Ÿ”„ Attempting to remove call session {session_id}") async with self.session_lock: if session_id in self.active_sessions: session = self.active_sessions.pop(session_id) Log.info(f"๐Ÿงน Calling cleanup for session {session_id}") else: # Session already removed (race condition - this is OK) Log.debug(f"Session {session_id} already removed (likely race condition)") return # Run cleanup outside lock to avoid blocking other operations if session: try: await session.cleanup() Log.info(f"โœ… Cleanup successful for session {session_id}") except RuntimeError as cleanup_error: # Handle event loop mismatch errors gracefully error_msg = str(cleanup_error) if "different loop" in error_msg or "attached to a different loop" in error_msg: Log.warning(f"โš ๏ธ Event loop mismatch during cleanup for {session_id} - resources will be garbage collected") # Don't re-raise - let Python's garbage collector handle it else: Log.error(f"โŒ Runtime error during cleanup for session {session_id}: {cleanup_error}") raise except Exception as cleanup_error: Log.error(f"โŒ Error during cleanup for session {session_id}: {cleanup_error}") # Don't re-raise - session is already removed from active_sessions Log.info(f"๐Ÿ—‘๏ธ Removed call session {session_id} (Active: {len(self.active_sessions)}/{self.max_concurrent_calls})") except Exception as e: # Catch any other unexpected errors error_msg = str(e) if "different loop" in error_msg or "attached to a different loop" in error_msg: Log.warning(f"โš ๏ธ Event loop mismatch when removing session {session_id} - session removed from tracking") else: Log.error(f"โŒ Error removing call session {session_id}: {e}") import traceback traceback.print_exc() def get_session(self, session_id: str) -> Optional[CallSession]: """Get an active call session.""" return self.active_sessions.get(session_id) def get_active_sessions_count(self) -> int: """Get number of active sessions.""" return len(self.active_sessions) def get_system_status(self) -> Dict: """Get system-wide status for monitoring.""" return { "active_sessions": len(self.active_sessions), "max_concurrent_calls": self.max_concurrent_calls, "utilization_percent": (len(self.active_sessions) / self.max_concurrent_calls) * 100, "available_slots": self.max_concurrent_calls - len(self.active_sessions), "session_ids": list(self.active_sessions.keys()) } def get_session_statuses(self) -> Dict[str, Dict]: """Get status of all active sessions.""" return { session_id: session.get_status() for session_id, session in self.active_sessions.items() } async def cleanup_all_sessions(self) -> None: """Cleanup all active sessions (for shutdown) with proper error handling.""" try: session_count = len(self.active_sessions) Log.info(f"๐Ÿงน Cleaning up {session_count} active sessions") # Cleanup all sessions concurrently with error handling cleanup_tasks = [] for session_id, session in self.active_sessions.items(): cleanup_tasks.append(self._safe_cleanup_session(session, session_id)) if cleanup_tasks: results = await asyncio.gather(*cleanup_tasks, return_exceptions=True) # Count successes and failures errors = [r for r in results if isinstance(r, Exception)] successes = len(results) - len(errors) if errors: Log.warning(f"โš ๏ธ Cleanup completed: {successes} succeeded, {len(errors)} failed") else: Log.info(f"โœ… All {successes} sessions cleaned up successfully") self.active_sessions.clear() except Exception as e: Log.error(f"โŒ Error cleaning up sessions: {e}") # Clear sessions anyway to prevent orphaned references self.active_sessions.clear() async def _safe_cleanup_session(self, session, session_id: str) -> None: """Safely cleanup a session with event loop error handling.""" try: await session.cleanup() except RuntimeError as e: error_msg = str(e) if "different loop" in error_msg or "attached to a different loop" in error_msg: Log.warning(f"โš ๏ธ Event loop mismatch during cleanup for {session_id} - allowing garbage collection") # Don't raise - let it be handled gracefully else: raise except Exception as e: Log.error(f"โŒ Error cleaning up session {session_id}: {e}") raise # Global call manager instance # Updated to support 50 concurrent calls call_manager = CallManager(max_concurrent_calls=50)