class ConnectionPool:
    """Manages connection pooling and threading for network scans."""
    
    def __init__(self, settings=None):
        # Thread pool configuration
        self.max_workers = self.settings.scan.max_threads
        self.executor: Optional[ThreadPoolExecutor] = None
        
        # Task management
        self.active_tasks: Dict[str, Future] = {}
        self.completed_tasks: List[ScanResult] = []
        self.failed_tasks: List[tuple] = []
        
        # Thread safety
        self._lock = threading.Lock()
        self._shutdown = False
    
    def start(self) -> None:
        """Start the connection pool."""
        self.executor = ThreadPoolExecutor(
            max_workers=self.max_workers,
            thread_name_prefix="HawkEye-Scanner"
        )

# Scanner Thread Pool
scanner_pool = ThreadPoolExecutor(
    max_workers=50,
    thread_name_prefix="HawkEye-Scanner"
)

# Batch Processing Thread Pool
batch_pool = ThreadPoolExecutor(
    max_workers=10,
    thread_name_prefix="HawkEye-Batch"
)

# Introspection Thread Pool
introspection_pool = ThreadPoolExecutor(
    max_workers=5,
    thread_name_prefix="HawkEye-MCP"
)

async def gather_with_concurrency(
    tasks: List[Awaitable[T]],
    max_concurrency: int = 10,
    return_exceptions: bool = True
) -> List[Union[T, Exception]]:
    """Execute multiple async tasks with limited concurrency."""
    semaphore = asyncio.Semaphore(max_concurrency)
    
    async def limited_task(task: Awaitable[T]) -> Union[T, Exception]:
        async with semaphore:
            try:
                return await task
            except Exception as e:
                if return_exceptions:
                    return e
                raise
    
    limited_tasks = [limited_task(task) for task in tasks]
    return await asyncio.gather(*limited_tasks, return_exceptions=return_exceptions)

class AsyncBatch:
    """Utility for batching async operations with concurrency control."""
    
    def __init__(
        self,
        max_concurrency: int = 10,
        batch_size: int = 100,
        logger: Optional[logging.Logger] = None
    ):
        self.max_concurrency = max_concurrency
        self.batch_size = batch_size
        self.logger = logger or logging.getLogger(self.__class__.__name__)
    
    async def process_batch(
        self,
        items: List[Any],
        processor: Callable[[Any], Awaitable[T]],
        error_handler: Optional[Callable[[Any, Exception], Awaitable[None]]] = None
    ) -> List[Union[T, Exception]]:
        """Process a batch of items with concurrency control."""        
        results = []
        
        for i in range(0, len(items), self.batch_size):
            batch = items[i:i + self.batch_size]
            
            # Create tasks for this batch
            tasks = []
            for item in batch:
                task = self._process_item_with_error_handling(item, processor, error_handler)
                tasks.append(task)
            
            # Execute batch with concurrency limit
            batch_results = await gather_with_concurrency(
                tasks,
                max_concurrency=self.max_concurrency,
                return_exceptions=True
            )
            
            results.extend(batch_results)
        
        return results

@asynccontextmanager
async def timeout_context(timeout_seconds: float, operation_name: str = "operation"):
    """Async context manager with timeout support."""
    try:
        async with asyncio.timeout(timeout_seconds):
            yield
    except asyncio.TimeoutError as e:
        logger.warning(f"Timeout after {timeout_seconds}s during {operation_name}")
        raise TimeoutError(
            f"Operation '{operation_name}' timed out after {timeout_seconds} seconds"
        ) from e

class MCPIntrospection:
    """Uses synchronous methods to avoid async complexity."""
    
    def introspect_server(self, server_config: MCPServerConfig) -> IntrospectionResult:
        """Synchronous wrapper for async introspection."""
        import asyncio
        try:
            # Get or create event loop
            loop = asyncio.get_event_loop()
            if loop.is_running():
                # If loop is already running, create a new one in thread
                import concurrent.futures
                with concurrent.futures.ThreadPoolExecutor() as executor:
                    future = executor.submit(
                        asyncio.run, 
                        self.mcp_client.introspect_server(server_config)
                    )
                    result = future.result(timeout=self.config.timeout)
            else:
                # Use existing loop
                result = loop.run_until_complete(
                    self.mcp_client.introspect_server(server_config)
                )
        except Exception as async_error:
            # Fall back to creating new event loop
            result = asyncio.run(self.mcp_client.introspect_server(server_config))
        
        return result

class BatchProcessor:
    """Processes batches of MCP servers efficiently."""
    
    def __init__(self, config: ScalingConfig, introspection: MCPIntrospection):
        self.config = config
        self.introspection = introspection
        
        # Worker management
        self._executor: Optional[ThreadPoolExecutor] = None
        self._active_futures: Set[Future] = set()
        
        # Resource monitoring
        self._resource_monitor = ResourceMonitor(config)
    
    def process_servers_in_batches(
        self,
        servers: List[MCPServerConfig],
        progress_callback: Optional[Callable[[int, int], None]] = None
    ) -> Iterator[Tuple[str, List[IntrospectionResult]]]:
        """Process servers in optimized batches."""
        
        # Initialize thread pool
        max_workers = min(self.config.max_workers, len(batches))
        self._executor = ThreadPoolExecutor(max_workers=max_workers)
        
        # Process batches
        batch_futures = {}
        
        for batch_id, batch_servers in batches:
            # Check resource constraints
            if self.config.enable_resource_monitoring and self._resource_monitor.should_throttle():
                self.logger.warning("Resource constraints detected, throttling batch submission")
                time.sleep(5.0)
            
            # Submit batch for processing
            future = self._executor.submit(self._process_batch, batch_id, batch_servers)
            batch_futures[future] = batch_id
            self._active_futures.add(future)
        
        # Collect results as batches complete
        for future in as_completed(batch_futures, timeout=self.config.batch_timeout * len(batches)):
            batch_id = batch_futures[future]
            results = future.result()
            yield batch_id, results

def submit_multiple_scans(self, tasks: List[ScanTask]) -> List[Future[ScanResult]]:
    """Submit multiple scan tasks concurrently."""
    if not tasks:
        return []
    
    futures = []
    for task in tasks:
        future = self.submit_scan(task)
        futures.append(future)
    
    self.logger.info(f"Submitted {len(tasks)} scan tasks")
    return futures

def collect_results(self, futures: List[Future], timeout: Optional[float] = None) -> List[ScanResult]:
    """Collect results from concurrent operations."""
    results = []
    completed_count = 0
    
    for future in as_completed(futures, timeout=timeout):
        try:
            result = future.result()
            results.append(result)
            completed_count += 1
        except Exception as e:
            self.logger.error(f"Task failed: {e}")
        
        # Update progress
        if self.progress_callback:
            self.progress_callback(completed_count, len(futures))
    
    return results

class ConnectionPool:
    def __init__(self):
        # Thread safety
        self._lock = threading.Lock()
        self._shutdown = False
        
        # Connection tracking
        self.active_tasks: Dict[str, Future] = {}
        self.stats = {
            'total_tasks': 0,
            'completed_tasks': 0,
            'failed_tasks': 0,
            'active_tasks': 0,
        }
    
    def submit_scan(self, task: ScanTask) -> Future[ScanResult]:
        """Thread-safe task submission."""
        with self._lock:
            self.active_tasks[task.task_id] = future
            self.stats['total_tasks'] += 1
            self.stats['active_tasks'] += 1
        
        return future
    
    def _task_completed(self, task_id: str, future: Future) -> None:
        """Thread-safe task completion handler."""
        with self._lock:
            if task_id in self.active_tasks:
                del self.active_tasks[task_id]
                self.stats['active_tasks'] -= 1
                
                if future.exception():
                    self.stats['failed_tasks'] += 1
                else:
                    self.stats['completed_tasks'] += 1

class MemoryEfficientCache:
    def __init__(self):
        self._lock = threading.RLock()  # Reentrant lock
        self._cache: Dict[str, Tuple[Any, float]] = {}
        self._access_order: deque = deque()
    
    def get(self, key: str) -> Optional[Any]:
        """Thread-safe cache access with reentrant locking."""
        with self._lock:
            if key in self._cache:
                # Reentrant operation - can call other locked methods
                self._update_access_order(key)
                return value
            return None
    
    def _update_access_order(self, key: str) -> None:
        """Internal method that requires same lock."""
        # Already holding lock - reentrant access
        self._access_order.remove(key)
        self._access_order.append(key)

async def gather_with_concurrency(
    tasks: List[Awaitable[T]],
    max_concurrency: int = 10,
    return_exceptions: bool = True
) -> List[Union[T, Exception]]:
    """Semaphore-controlled concurrent execution."""
    semaphore = asyncio.Semaphore(max_concurrency)
    
    async def limited_task(task: Awaitable[T]) -> Union[T, Exception]:
        async with semaphore:  # Acquire semaphore
            try:
                return await task
            except Exception as e:
                return e if return_exceptions else raise
    
    limited_tasks = [limited_task(task) for task in tasks]
    return await asyncio.gather(*limited_tasks, return_exceptions=return_exceptions)

class OptimizedConnectionPool:
    def __init__(self, max_connections: int = 10):
        self.max_connections = max_connections
        self._connection_semaphore = asyncio.Semaphore(max_connections)
        self._connections: Dict[str, ConnectionInfo] = {}
    
    async def acquire_connection(self, server_id: str) -> ConnectionInfo:
        """Acquire a connection with semaphore control."""
        async with self._connection_semaphore:
            if server_id in self._connections:
                return self._connections[server_id]
            
            # Create new connection
            connection = await self._create_connection(server_id)
            self._connections[server_id] = connection
            return connection

class ResourceMonitor:
    """Monitors system resources during large-scale scanning."""
    
    def __init__(self, config: ScalingConfig):
        self.config = config
        self._monitoring = False
        self._monitor_thread: Optional[threading.Thread] = None
        self._resource_history: deque = deque(maxlen=100)
        self._alerts: List[str] = []
    
    def should_throttle(self) -> bool:
        """Check if processing should be throttled due to resource constraints."""
        current = self.get_current_usage()
        return (
            current['memory_mb'] > self.config.max_memory_usage_mb or
            current['cpu_percent'] > self.config.max_cpu_usage_percent
        )
    
    def _monitor_loop(self) -> None:
        """Main monitoring loop."""
        while self._monitoring:
            # Get current resource usage
            memory_usage = psutil.virtual_memory().percent
            cpu_usage = psutil.cpu_percent(interval=1.0)
            
            # Check thresholds
            if memory_usage > 90.0:
                alert = f"High memory usage: {memory_usage:.1f}%"
                self._alerts.append(alert)
                self.logger.warning(alert)
            
            if cpu_usage > self.config.max_cpu_usage_percent:
                alert = f"High CPU usage: {cpu_usage:.1f}%"
                self._alerts.append(alert)
                self.logger.warning(alert)

@dataclass
class ScalingConfig:
    # Advanced features
    enable_backpressure: bool = True
    backpressure_threshold: int = 100
    
class BatchProcessor:
    def process_servers_in_batches(self, servers: List[MCPServerConfig]):
        """Process with backpressure management."""
        for batch_id, batch_servers in batches:
            # Apply backpressure if too many active tasks
            if self.config.enable_backpressure:
                active_count = len(self._active_futures)
                if active_count >= self.config.backpressure_threshold:
                    self.logger.info(f"Backpressure triggered: {active_count} active tasks")
                    # Wait for some tasks to complete
                    self._wait_for_available_slot()
            
            # Submit batch for processing
            future = self._executor.submit(self._process_batch, batch_id, batch_servers)
            self._active_futures.add(future)
    
    def _wait_for_available_slot(self) -> None:
        """Wait for an available processing slot."""
        while len(self._active_futures) >= self.config.backpressure_threshold:
            # Wait for any future to complete
            done, pending = concurrent.futures.wait(
                self._active_futures, 
                timeout=1.0, 
                return_when=concurrent.futures.FIRST_COMPLETED
            )
            
            # Remove completed futures
            for future in done:
                self._active_futures.discard(future)

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time: Optional[float] = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
        self._lock = threading.Lock()
    
    def call(self, func: Callable, *args, **kwargs):
        """Execute function with circuit breaker protection."""
        with self._lock:
            if self.state == "OPEN":
                if time.time() - self.last_failure_time > self.timeout:
                    self.state = "HALF_OPEN"
                else:
                    raise CircuitBreakerOpen("Circuit breaker is OPEN")
            
            try:
                result = func(*args, **kwargs)
                
                # Success - reset failure count
                if self.state == "HALF_OPEN":
                    self.state = "CLOSED"
                self.failure_count = 0
                return result
                
            except Exception as e:
                self.failure_count += 1
                self.last_failure_time = time.time()
                
                if self.failure_count >= self.failure_threshold:
                    self.state = "OPEN"
                
                raise e

class DeadlockPrevention:
    """Deadlock prevention through lock ordering."""
    
    # Global lock ordering
    LOCK_ORDER = {
        'cache_lock': 1,
        'stats_lock': 2,
        'connection_lock': 3,
        'resource_lock': 4
    }
    
    def acquire_multiple_locks(self, lock_names: List[str], timeout: float = 10.0):
        """Acquire multiple locks in order to prevent deadlock."""
        # Sort locks by order to ensure consistent acquisition sequence
        sorted_locks = sorted(lock_names, key=lambda x: self.LOCK_ORDER[x])
        acquired_locks = []
        
        try:
            for lock_name in sorted_locks:
                lock = self._get_lock(lock_name)
                if lock.acquire(timeout=timeout):
                    acquired_locks.append(lock)
                else:
                    raise TimeoutError(f"Failed to acquire {lock_name} within {timeout}s")
            
            yield acquired_locks
            
        finally:
            # Release locks in reverse order
            for lock in reversed(acquired_locks):
                lock.release()

class RateLimiter:
    def acquire(self, timeout: Optional[float] = None) -> bool:
        """Acquire with timeout to prevent indefinite blocking."""
        start_time = time.time()
        
        with self._lock:  # This could timeout
            self.stats['total_requests'] += 1
        
        # Use timeout-aware token bucket
        success = self.token_bucket.wait_for_tokens(1, timeout)
        
        if success and timeout is not None:
            elapsed = time.time() - start_time
            remaining_timeout = max(0, timeout - elapsed)
            success = self.sliding_window.wait_until_allowed(remaining_timeout)
        
        return success

class ResourceManager:
    def __init__(self):
        self._resource_locks = {}
        self._allocation_order = []
        self._allocation_lock = threading.Lock()
    
    def allocate_resources(self, resource_ids: List[str], timeout: float = 30.0):
        """Allocate multiple resources atomically."""
        # Sort resource IDs to ensure consistent ordering
        sorted_resources = sorted(resource_ids)
        
        with self._allocation_lock:
            # Check resource availability
            for resource_id in sorted_resources:
                if resource_id in self._allocation_order:
                    raise ResourceConflict(f"Resource {resource_id} already allocated")
            
            # Reserve all resources atomically
            for resource_id in sorted_resources:
                self._allocation_order.append(resource_id)
                self._resource_locks[resource_id] = threading.Lock()
        
        return ResourceAllocation(sorted_resources, self._resource_locks)

# Performance measurements for different synchronization primitives

Lock Types and Performance:
- threading.Lock(): 0.05μs acquisition time
- threading.RLock(): 0.08μs acquisition time  
- asyncio.Semaphore(): 0.12μs acquisition time
- threading.Condition(): 0.15μs acquisition time

Contention Impact:
- Low contention (1-2 threads): <5% overhead
- Medium contention (3-10 threads): 10-15% overhead
- High contention (10+ threads): 20-30% overhead

# Optimal thread counts by operation type
OPTIMAL_THREAD_COUNTS = {
    "network_scanning": 50,     # I/O bound - high thread count beneficial
    "cpu_analysis": 4,          # CPU bound - match CPU core count
    "batch_processing": 10,     # Mixed workload - moderate thread count
    "resource_monitoring": 1,   # Sequential monitoring - single thread
}

# Performance degradation points
PERFORMANCE_THRESHOLDS = {
    "context_switching_overhead": 100,  # threads
    "memory_pressure_point": 200,      # threads  
    "lock_contention_critical": 50,    # concurrent lock acquisitions
}

# Good: Composition-based thread management
class TaskProcessor:
    def __init__(self):
        self.scanner = NetworkScanner()
        self.analyzer = ThreatAnalyzer()
        self.executor = ThreadPoolExecutor(max_workers=10)
    
    def process_task(self, task):
        return self.executor.submit(self._process_single_task, task)

# Good: Automatic resource cleanup
class ConnectionPool:
    def __enter__(self):
        self.start()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.shutdown()

# Usage
with ConnectionPool() as pool:
    results = pool.process_tasks(tasks)

# Good: Comprehensive error handling in concurrent operations
async def safe_concurrent_operation(tasks):
    try:
        results = await gather_with_concurrency(
            tasks, 
            max_concurrency=10,
            return_exceptions=True
        )
        
        # Separate successful results from exceptions
        successful_results = [r for r in results if not isinstance(r, Exception)]
        exceptions = [r for r in results if isinstance(r, Exception)]
        
        return successful_results, exceptions
        
    except Exception as e:
        logger.error(f"Critical error in concurrent operation: {e}")
        raise

# Good: Immutable data structures
@dataclass(frozen=True)
class ScanConfig:
    target_host: str
    ports: List[int]
    timeout: float

# Good: Thread-local storage
import threading
thread_local_data = threading.local()

def process_request():
    thread_local_data.request_id = generate_id()
    # Use thread_local_data.request_id safely

# Good: Atomic counter operations
import threading
from collections import defaultdict

class ThreadSafeCounter:
    def __init__(self):
        self._counts = defaultdict(int)
        self._lock = threading.Lock()
    
    def increment(self, key: str) -> int:
        with self._lock:
            self._counts[key] += 1
            return self._counts[key]

# Good: Always use timeouts for blocking operations
def acquire_resource_safely(resource_id: str, timeout: float = 30.0):
    start_time = time.time()
    
    # Timeout on lock acquisition
    if not resource_lock.acquire(timeout=timeout):
        raise TimeoutError(f"Failed to acquire lock for {resource_id}")
    
    try:
        # Timeout on resource operation
        remaining_time = timeout - (time.time() - start_time)
        return perform_operation(resource_id, timeout=remaining_time)
    finally:
        resource_lock.release()

# Future: Full async/await architecture
class AsyncMCPIntrospector:
    async def introspect_servers(self, configs: List[MCPServerConfig]) -> List[IntrospectionResult]:
        async with self.connection_pool:
            tasks = [self._introspect_single_server(config) for config in configs]
            return await gather_with_concurrency(tasks, max_concurrency=self.max_concurrent)

# Future: AI-driven resource allocation
class IntelligentResourceManager:
    def __init__(self):
        self.performance_predictor = PerformancePredictor()
        self.resource_optimizer = ResourceOptimizer()
    
    async def optimize_concurrency(self, workload: Workload) -> ConcurrencyConfig:
        predicted_performance = await self.performance_predictor.predict(workload)
        return self.resource_optimizer.optimize(predicted_performance)

# Future: Multi-node concurrency coordination
class DistributedTaskCoordinator:
    async def coordinate_distributed_scan(
        self, 
        nodes: List[NodeConfig], 
        targets: List[ScanTarget]
    ) -> DistributedScanResult:
        # Distribute tasks across nodes
        task_assignments = self.load_balancer.distribute_tasks(nodes, targets)
        
        # Coordinate execution across nodes
        node_futures = []
        for node, tasks in task_assignments.items():
            future = self.execute_on_node(node, tasks)
            node_futures.append(future)
        
        return await self.aggregate_results(node_futures)