Worker Pool：并发执行与资源复用

传统模型（每任务一线程）：
  总成本 = N × (创建成本 + 销毁成本 + 任务成本)

Worker Pool 模型：
  总成本 = M × 创建成本 + M × 销毁成本 + N × 任务成本
  其中 M << N（M 是 Worker 数量，N 是任务数量）

class ThresholdBasedScaler:
    def __init__(self, pool, 
                 min_workers: int = 1,
                 max_workers: int = 32,
                 scale_up_threshold: int = 10,   # 队列深度 > 10 触发扩容
                 scale_down_threshold: int = 2,  # 队列深度 < 2 触发缩容
                 scale_up_ratio: float = 2.0,     # 每次扩容翻倍
                 scale_down_ratio: float = 0.5,   # 每次缩容减半
                 cooldown_seconds: float = 10.0): # 冷却时间防止震荡
        
        self.pool = pool
        self.min_workers = min_workers
        self.max_workers = max_workers
        self.scale_up_threshold = scale_up_threshold
        self.scale_down_threshold = scale_down_threshold
        self.scale_up_ratio = scale_up_ratio
        self.scale_down_ratio = scale_down_ratio
        self.cooldown_seconds = cooldown_seconds
        self.last_scale_time = 0
    
    def should_scale(self, queue_depth: int, current_time: float) -> Tuple[str, int]:
        """返回 (动作, 目标 Worker 数)"""
        if current_time - self.last_scale_time < self.cooldown_seconds:
            return "none", self.pool.worker_count
        
        if queue_depth > self.scale_up_threshold:
            new_count = min(int(self.pool.worker_count * self.scale_up_ratio),
                           self.max_workers)
            self.last_scale_time = current_time
            return "scale_up", new_count
        
        if queue_depth < self.scale_down_threshold:
            new_count = max(int(self.pool.worker_count * self.scale_down_ratio),
                           self.min_workers)
            self.last_scale_time = current_time
            return "scale_down", new_count
        
        return "none", self.pool.worker_count

import time

class PIDAutoscaler:
    """
    PID 控制器实现自适应扩缩容
    Kp: 比例系数，控制当前误差
    Ki: 积分系数，消除稳态误差  
    Kd: 微分系数，抑制超调和振荡
    """
    def __init__(self, pool,
                 target_queue_depth: int = 5,
                 Kp: float = 0.5, Ki: float = 0.1, Kd: float = 0.2,
                 min_workers: int = 1, max_workers: int = 64):
        self.pool = pool
        self.target_queue_depth = target_queue_depth
        
        # PID 参数
        self.Kp = Kp
        self.Ki = Ki
        self.Kd = Kd
        
        self.min_workers = min_workers
        self.max_workers = max_workers
        
        # 状态变量
        self.prev_error = 0.0
        self.integral = 0.0
        self.last_time = time.time()
    
    def compute_scale_delta(self, current_queue_depth: int) -> int:
        """
        计算 Worker 数量的调整量
        返回值 > 0 表示需要扩容，< 0 表示需要缩容，0 表示不变
        """
        current_time = time.time()
        dt = current_time - self.last_time
        
        if dt <= 0:
            return 0
        
        # 计算误差：正值表示队列过深（需要扩容），负值表示队列过浅（可以缩容）
        error = current_queue_depth - self.target_queue_depth
        
        # 积分项：累积误差，消除稳态误差
        self.integral += error * dt
        self.integral = max(-100, min(100, self.integral))  # 抗积分饱和
        
        # 微分项：预测误差变化趋势
        derivative = (error - self.prev_error) / dt if dt > 0 else 0
        
        # PID 输出
        output = self.Kp * error + self.Ki * self.integral + self.Kd * derivative
        
        self.prev_error = error
        self.last_time = current_time
        
        # 将连续输出转换为离散的 Worker 数量调整
        return int(round(output))
    
    def should_scale(self, queue_depth: int) -> Tuple[str, int]:
        """返回 (动作, 目标 Worker 数)"""
        delta = self.compute_scale_delta(queue_depth)
        current_count = self.pool.worker_count
        
        if delta > 0:
            new_count = min(current_count + delta, self.max_workers)
            return "scale_up", new_count
        elif delta < 0:
            new_count = max(current_count + delta, self.min_workers)
            return "scale_down", new_count
        
        return "none", current_count

import threading
import time
import logging
from enum import Enum, auto
from typing import Optional, Callable, Any
from dataclasses import dataclass, field
from collections import deque

logger = logging.getLogger(__name__)


class WorkerState(Enum):
    """Worker 生命周期状态"""
    INITIALIZING = auto()
    REGISTERED = auto()
    READY = auto()
    WORKING = auto()
    STOPPING = auto()
    STOPPED = auto()
    FAILED = auto()


@dataclass
class Task:
    """任务抽象"""
    id: str
    payload: Any
    callback: Optional[Callable] = None
    timeout: float = 30.0
    created_at: float = field(default_factory=time.time)
    
    def __hash__(self):
        return hash(self.id)


@dataclass
class WorkerMetrics:
    """Worker 性能指标"""
    tasks_completed: int = 0
    tasks_failed: int = 0
    total_execution_time: float = 0.0
    last_task_start: Optional[float] = None
    last_task_end: Optional[float] = None
    current_task: Optional[Task] = None


class StateTransitionError(Exception):
    """非法状态转换异常"""
    pass


class Worker:
    """
    具有完整状态机的 Worker 实现
    
    线程安全：所有状态转换都受到锁保护
    优雅停止：支持超时强制停止
    指标收集：记录执行时间、任务数等指标
    """
    
    def __init__(self, worker_id: str, task_handler: Callable[[Task], Any]):
        self.worker_id = worker_id
        self.task_handler = task_handler
        
        # 状态机
        self._state = WorkerState.INITIALIZING
        self._state_lock = threading.RLock()
        
        # 生命周期管理
        self._stop_event = threading.Event()
        self._idle_event = threading.Event()  # 用于实现空闲等待
        
        # 当前任务
        self._current_task: Optional[Task] = None
        self._task_lock = threading.Lock()
        
        # 指标
        self.metrics = WorkerMetrics()
        
        # 工作线程
        self._thread: Optional[threading.Thread] = None
    
    @property
    def state(self) -> WorkerState:
        with self._state_lock:
            return self._state
    
    def _transition_to(self, new_state: WorkerState, reason: str = ""):
        """
        安全的状态转换
        
        Args:
            new_state: 目标状态
            reason: 转换原因（用于日志）
            
        Raises:
            StateTransitionError: 如果转换非法
        """
        with self._state_lock:
            current = self._state
            
            # 定义合法转换
            valid_transitions = {
                WorkerState.INITIALIZING: {WorkerState.REGISTERED, WorkerState.FAILED},
                WorkerState.REGISTERED: {WorkerState.READY, WorkerState.STOPPING},
                WorkerState.READY: {WorkerState.WORKING, WorkerState.STOPPING},
                WorkerState.WORKING: {WorkerState.READY, WorkerState.STOPPING},
                WorkerState.STOPPING: {WorkerState.STOPPED},
                WorkerState.FAILED: {WorkerState.INITIALIZING, WorkerState.STOPPED},
            }
            
            if new_state not in valid_transitions.get(current, set()):
                raise StateTransitionError(
                    f"Worker {self.worker_id}: 非法状态转换 {current.name} -> {new_state.name}, 原因: {reason}"
                )
            
            logger.debug(f"Worker {self.worker_id}: 状态转换 {current.name} -> {new_state.name}, 原因: {reason}")
            self._state = new_state
            
            # 状态进入/退出动作
            self._on_state_entry(new_state)
    
    def _on_state_entry(self, state: WorkerState):
        """状态进入动作"""
        if state == WorkerState.READY:
            self._idle_event.set()  # 通知调度器该 Worker 已就绪
        elif state == WorkerState.STOPPED:
            self._stop_event.set()  # 通知等待者该 Worker 已停止
    
    def initialize(self) -> bool:
        """
        初始化 Worker
        
        Returns:
            True 如果初始化成功
        """
        try:
            self._transition_to(WorkerState.REGISTERED, "初始化完成")
            return True
        except Exception as e:
            logger.error(f"Worker {self.worker_id} 初始化失败: {e}")
            self._transition_to(WorkerState.FAILED, str(e))
            return False
    
    def assign_task(self, task: Task) -> bool:
        """
        分配任务给 Worker
        
        Returns:
            True 如果任务分配成功（Worker 接受任务）
            
        Raises:
            StateTransitionError: 如果 Worker 状态不允许接收任务
        """
        if self.state != WorkerState.READY:
            raise StateTransitionError(
                f"Worker {self.worker_id} 无法接收任务，当前状态: {self.state.name}"
            )
        
        with self._task_lock:
            self._current_task = task
            self.metrics.current_task = task
            self.metrics.last_task_start = time.time()
            self._idle_event.clear()  # 清除空闲标记
        
        # 启动工作线程
        self._thread = threading.Thread(target=self._execute_task, args=(task,))
        self._thread.start()
        
        return True
    
    def _execute_task(self, task: Task):
        """执行任务的工作线程"""
        try:
            self._transition_to(WorkerState.WORKING, f"开始执行任务 {task.id}")
            
            # 执行任务（带超时保护）
            result = None
            exception = None
            
            try:
                result = self.task_handler(task)
            except Exception as e:
                exception = e
                logger.error(f"Worker {self.worker_id} 执行任务 {task.id} 失败: {e}")
            
            # 更新指标
            with self._task_lock:
                self.metrics.last_task_end = time.time()
                execution_time = self.metrics.last_task_end - self.metrics.last_task_start
                self.metrics.total_execution_time += execution_time
                
                if exception:
                    self.metrics.tasks_failed += 1
                else:
                    self.metrics.tasks_completed += 1
                
                self._current_task = None
                self.metrics.current_task = None
            
            # 调用回调
            if task.callback:
                try:
                    task.callback(result, exception)
                except Exception as e:
                    logger.error(f"Worker {self.worker_id} 任务回调失败: {e}")
            
            self._transition_to(WorkerState.READY, f"任务 {task.id} 完成")
            
        except Exception as e:
            logger.error(f"Worker {self.worker_id} 任务执行异常: {e}")
            self._transition_to(WorkerState.READY, f"异常恢复")
    
    def stop(self, timeout: float = 10.0) -> bool:
        """
        优雅停止 Worker
        
        Args:
            timeout: 等待当前任务完成的最大时间
            
        Returns:
            True 如果 Worker 已停止
        """
        if self.state in {WorkerState.STOPPING, WorkerState.STOPPED}:
            return True
        
        self._transition_to(WorkerState.STOPPING, "收到停止信号")
        
        # 等待当前任务完成
        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=timeout)
        
        self._transition_to(WorkerState.STOPPED, "资源已释放")
        return True
    
    def is_idle(self) -> bool:
        """检查 Worker 是否处于空闲状态"""
        return self.state == WorkerState.READY
    
    def is_alive(self) -> bool:
        """检查 Worker 是否存活"""
        return self.state not in {WorkerState.STOPPED, WorkerState.FAILED}

import asyncio
from typing import List

class GracefulShutdown:
    """
    优雅停止协议实现
    
    停止顺序：
    1. 停止接收新任务
    2. 发送停止信号给所有 Worker
    3. 等待 Worker 完成当前任务（带超时）
    4. 强制停止未完成的 Worker
    5. 清理剩余资源
    """
    
    def __init__(self, workers: List[Worker], timeout: float = 30.0):
        self.workers = workers
        self.timeout = timeout
        self._shutdown_complete = False
    
    async def shutdown(self):
        """执行优雅关闭"""
        logger.info("开始优雅关闭...")
        
        # Phase 1: 通知所有 Worker 停止
        logger.info(f"Phase 1: 发送停止信号给 {len(self.workers)} 个 Worker")
        stop_futures = [asyncio.to_thread(w.stop, timeout=self.timeout / 2) 
                       for w in self.workers]
        
        # 等待所有 Worker 停止
        await asyncio.gather(*stop_futures, return_exceptions=True)
        
        # Phase 2: 清理共享资源
        logger.info("Phase 2: 清理共享资源")
        await self._cleanup_resources()
        
        self._shutdown_complete = True
        logger.info("优雅关闭完成")
    
    async def _cleanup_resources(self):
        """清理池级别的共享资源"""
        # 等待一小段时间让 Worker 完成清理
        await asyncio.sleep(0.1)

# 错误的实现（存在竞态）
if worker.is_idle():  # 检查
    worker.assign_task(task)  # 使用 —— 中间可能有其他线程抢走 Worker

# 正确的实现（原子操作）
success = worker.try_assign(task)  # 检查和分配是原子的

# 使用锁保护共享状态
with self._state_lock:
    if self._state == WorkerState.READY:
        self._current_task = task
        # ...

import threading
from typing import Optional, List

class RoundRobinScheduler:
    """
    轮询任务分配器
    
    优点：实现简单，每个 Worker 负载大致均衡（长期）
    缺点：不考虑 Worker 当前负载，可能将任务分配给繁忙的 Worker
    """
    
    def __init__(self, workers: List[Worker]):
        self.workers = workers
        self._index = 0
        self._lock = threading.Lock()
    
    def select(self) -> Optional[Worker]:
        """选择一个 Worker 用于下一个任务"""
        if not self.workers:
            return None
        
        with self._lock:
            # 找到下一个可用的 Worker
            attempts = 0
            max_attempts = len(self.workers)
            
            while attempts < max_attempts:
                worker = self.workers[self._index]
                self._index = (self._index + 1) % len(self.workers)
                attempts += 1
                
                if worker.is_idle():
                    return worker
            
            return None  # 没有可用的 Worker
    
    def notify_task_complete(self, worker: Worker):
        """通知任务完成（用于更新统计等）"""
        pass  # 轮询策略不需要这个信息

import threading
from typing import Optional, List, Dict

class LeastConnectionsScheduler:
    """
    最少连接任务分配器
    
    优点：动态平衡负载，考虑了 Worker 的当前工作状态
    缺点：需要维护每个 Worker 的连接计数，有轻微开销
    """
    
    def __init__(self, workers: List[Worker]):
        self.workers = workers
        self._connections: Dict[str, int] = {w.worker_id: 0 for w in workers}
        self._lock = threading.Lock()
    
    def select(self) -> Optional[Worker]:
        """选择连接数最少的 Worker"""
        if not self.workers:
            return None
        
        with self._lock:
            # 过滤出空闲的 Worker
            idle_workers = [w for w in self.workers if w.is_idle()]
            if not idle_workers:
                return None
            
            # 找到连接数最少的 Worker
            min_conn_worker = min(idle_workers, 
                                 key=lambda w: self._connections[w.worker_id])
            
            # 增加该 Worker 的连接计数
            self._connections[min_conn_worker.worker_id] += 1
            
            return min_conn_worker
    
    def notify_task_complete(self, worker: Worker):
        """通知任务完成，减少连接计数"""
        with self._lock:
            if worker.worker_id in self._connections:
                self._connections[worker.worker_id] = max(
                    0, self._connections[worker.worker_id] - 1
                )

import threading
import time
from dataclasses import dataclass
from typing import Optional, List

@dataclass
class LoadMetrics:
    """Worker 负载指标"""
    cpu_usage: float = 0.0       # CPU 使用率 0-1
    memory_usage: float = 0.0    # 内存使用率 0-1
    task_queue_length: int = 0   # 任务队列长度
    avg_task_duration: float = 0.0  # 平均任务执行时间
    active_tasks: int = 0        # 正在执行的任务数
    
    @property
    def load_score(self) -> float:
        """
        计算综合负载分数
        
        负载分数越高表示越繁忙
        范围：[0, +∞)，通常 0-1 表示健康，> 1 表示过载
        """
        # 加权组合多个指标
        return (
            0.4 * self.cpu_usage +
            0.3 * self.memory_usage +
            0.2 * (self.task_queue_length / 10.0) +  # 假设 10 是正常上限
            0.1 * self.active_tasks
        )


class LoadAwareScheduler:
    """
    负载感知任务分配器
    
    综合考虑 CPU、内存、队列长度等指标，选择负载最低的 Worker
    支持实时更新 Worker 负载状态
    """
    
    def __init__(self, workers: List[Worker]):
        self.workers = workers
        self._worker_metrics: Dict[str, LoadMetrics] = {
            w.worker_id: LoadMetrics() for w in workers
        }
        self._lock = threading.Lock()
    
    def update_metrics(self, worker_id: str, metrics: LoadMetrics):
        """更新指定 Worker 的负载指标"""
        with self._lock:
            if worker_id in self._worker_metrics:
                self._worker_metrics[worker_id] = metrics
    
    def select(self) -> Optional[Worker]:
        """选择负载最低的 Worker"""
        if not self.workers:
            return None
        
        with self._lock:
            # 过滤出存活的 Worker
            alive_workers = [w for w in self.workers if w.is_alive()]
            if not alive_workers:
                return None
            
            # 找到负载最低的 Worker
            min_load_worker = min(
                alive_workers,
                key=lambda w: self._worker_metrics.get(w.worker_id, LoadMetrics()).load_score
            )
            
            return min_load_worker if min_load_worker.is_idle() else None
    
    def get_load_distribution(self) -> Dict[str, float]:
        """获取所有 Worker 的负载分布"""
        with self._lock:
            return {
                w.worker_id: self._worker_metrics[w.worker_id].load_score
                for w in self.workers
            }

import hashlib
from typing import Optional, List, Dict

class AffinityScheduler:
    """
    亲和性任务分配器
    
    根据任务的关键字（如用户 ID、会话 ID）计算哈希，
    将相关任务分配给同一个 Worker
    
    优点：提高缓存命中率，保持会话状态
    缺点：可能导致负载不均
    """
    
    def __init__(self, workers: List[Worker]):
        self.workers = workers
        self._worker_assignment: Dict[str, str] = {}  # key -> worker_id
    
    def _get_affinity_key(self, task) -> str:
        """提取任务的亲和性关键字"""
        # 尝试从 payload 中提取 user_id 或 session_id
        if hasattr(task, 'payload'):
            payload = task.payload
            if isinstance(payload, dict):
                return payload.get('user_id', '') or payload.get('session_id', '')
        return ''
    
    def select(self, task: Task) -> Optional[Worker]:
        """选择具有亲和性的 Worker"""
        if not self.workers:
            return None
        
        # 计算亲和性键
        affinity_key = self._get_affinity_key(task)
        
        if affinity_key:
            # 查找已分配的 Worker
            if affinity_key in self._worker_assignment:
                worker_id = self._worker_assignment[affinity_key]
                # 找到对应的 Worker 并检查是否可用
                for w in self.workers:
                    if w.worker_id == worker_id and w.is_idle():
                        return w
                # 如果之前的 Worker 不可用，尝试找空闲的
                for w in self.workers:
                    if w.is_idle():
                        self._worker_assignment[affinity_key] = w.worker_id
                        return w
                return None
            else:
                # 新亲和性分配：选择负载最低的 Worker
                idle_workers = [w for w in self.workers if w.is_idle()]
                if idle_workers:
                    # 选择连接数最少的
                    chosen = min(idle_workers, key=lambda w: w.metrics.tasks_completed)
                    self._worker_assignment[affinity_key] = chosen.worker_id
                    return chosen
                return None
        else:
            # 无亲和性要求的任务：使用负载最低的 Worker
            idle_workers = [w for w in self.workers if w.is_idle()]
            return idle_workers[0] if idle_workers else None

import threading
import queue
import time
from typing import Optional, Callable, Any
from contextlib import contextmanager

class ConnectionPool:
    """
    数据库连接池实现
    
    核心机制：
    1. 初始化时创建固定数量的连接
    2. 使用时"借用"连接，用完后"归还"
    3. 连接被关闭时自动重建
    """
    
    def __init__(self, 
                 factory: Callable[[], Any],  # 连接工厂
                 min_size: int = 5,
                 max_size: int = 20,
                 max_idle_time: float = 300.0,  # 最大空闲时间（秒）
                 checkout_timeout: float = 10.0): # 借用超时（秒）
        
        self.factory = factory
        self.min_size = min_size
        self.max_size = max_size
        self.max_idle_time = max_idle_time
        self.checkout_timeout = checkout_timeout
        
        # 连接队列
        self._pool: queue.Queue = queue.Queue(maxsize=max_size)
        self._total_connections = 0
        self._lock = threading.Lock()
        
        # 初始化连接
        for _ in range(min_size):
            self._add_connection()
    
    def _add_connection(self) -> bool:
        """添加一个新连接"""
        with self._lock:
            if self._total_connections >= self.max_size:
                return False
            
            try:
                conn = self.factory()
                self._pool.put({
                    'connection': conn,
                    'created_at': time.time(),
                    'last_used': time.time()
                })
                self._total_connections += 1
                return True
            except Exception as e:
                return False
    
    @contextmanager
    def checkout(self):
        """
        借用连接（上下文管理器）
        
        用法：
            with pool.checkout() as conn:
                conn.execute("SELECT * FROM users")
        """
        conn = None
        start_time = time.time()
        
        while True:
            try:
                # 尝试从池中获取连接
                item = self._pool.get(block=False)
                conn = item['connection']
                
                # 检查连接是否有效
                if self._is_connection_valid(conn, item):
                    item['last_used'] = time.time()
                    break
                else:
                    # 连接无效，丢弃并创建新的
                    self._close_connection(item)
                    with self._lock:
                        self._total_connections -= 1
                    # 尝试创建新连接
                    if not self._add_connection():
                        # 无法创建新连接，继续等待其他连接
                        pass
            except queue.Empty:
                # 池为空，尝试创建新连接
                with self._lock:
                    if self._total_connections < self.max_size:
                        if self._add_connection():
                            continue
                
                # 等待一段时间后重试
                elapsed = time.time() - start_time
                if elapsed >= self.checkout_timeout:
                    raise TimeoutError(f"获取连接超时 ({self.checkout_timeout}s)")
                time.sleep(0.01)
        
        try:
            yield conn
        finally:
            # 归还连接
            if conn is not None:
                try:
                    item = {
                        'connection': conn,
                        'created_at': time.time(),
                        'last_used': time.time()
                    }
                    self._pool.put(item, block=False)
                except queue.Full:
                    # 池已满，关闭连接
                    with self._lock:
                        self._total_connections -= 1
                    self._close_connection({'connection': conn})
    
    def _is_connection_valid(self, conn, item) -> bool:
        """检查连接是否有效"""
        # 检查空闲超时
        if time.time() - item['last_used'] > self.max_idle_time:
            return False
        
        # 执行 ping 检查连接是否存活
        try:
            # conn.ping()  # 数据库连接 ping
            return True
        except Exception:
            return False
    
    def _close_connection(self, item):
        """关闭连接"""
        try:
            if item.get('connection'):
                pass  # item['connection'].close()
        except Exception:
            pass
    
    def close(self):
        """关闭所有连接"""
        while True:
            try:
                item = self._pool.get(block=False)
                self._close_connection(item)
            except queue.Empty:
                break
        with self._lock:
            self._total_connebooks = 0

import concurrent.futures
import threading
import queue
import time
from typing import Optional, Callable, Any, List

class ThreadPool:
    """
    生产级线程池实现
    
    特性：
    - 固定线程数，避免线程创建销毁开销
    - 任务队列支持优先级
    - 支持任务超时和取消
    - 优雅关闭协议
    """
    
    def __init__(self, 
                 min_workers: int = 4,
                 max_workers: Optional[int] = None,
                 queue_size: int = 1000,
                 thread_name_prefix: str = "Worker"):
        
        self.min_workers = min_workers
        self.max_workers = max_workers or min_workers * 2
        self.queue_size = queue_size
        self.thread_name_prefix = thread_name_prefix
        
        # 任务队列
        self._task_queue: queue.PriorityQueue = queue.PriorityQueue(
            maxsize=queue_size
        )
        
        # Worker 线程
        self._workers: List[threading.Thread] = []
        self._worker_states = {}  # worker_id -> 'running'|'stopping'|'stopped'
        
        # 控制
        self._shutdown = False
        self._lock = threading.Lock()
        
        # 启动 Worker
        for i in range(min_workers):
            self._start_worker(i)
    
    def _start_worker(self, worker_id: int):
        """启动一个 Worker 线程"""
        worker = threading.Thread(
            target=self._worker_loop,
            name=f"{self.thread_name_prefix}-{worker_id}",
            daemon=True
        )
        worker.start()
        self._workers.append(worker)
        self._worker_states[worker.name] = 'running'
    
    def _worker_loop(self):
        """Worker 的主循环"""
        while not self._shutdown:
            try:
                # 从队列获取任务（带超时以便检查 shutdown 标志）
                try:
                    priority, task_future, args, kwargs = self._task_queue.get(
                        timeout=1.0
                    )
                except queue.Empty:
                    continue
                
                # 检查 Future 是否已取消
                if task_future.cancelled():
                    continue
                
                try:
                    # 执行任务
                    result = task_future.fn(*args, **kwargs)
                    
                    if not task_future.cancelled():
                        task_future._result = result
                        task_future._state = 'FINISHED'
                        task_future._condition.notify_all()
                except Exception as e:
                    if not task_future.cancelled():
                        task_future._exception = e
                        task_future._state = 'FINISHED'
                        task_future._condition.notify_all()
                finally:
                    self._task_queue.task_done()
                    
            except Exception as e:
                # 记录异常但继续运行
                pass
    
    def submit(self, 
               fn: Callable, 
               *args, 
               priority: int = 5,
               **kwargs) -> concurrent.futures.Future:
        """
        提交任务到线程池
        
        Args:
            fn: 要执行的函数
            priority: 优先级（数值越小优先级越高）
            *args, **kwargs: 函数参数
            
        Returns:
            Future 对象
        """
        if self._shutdown:
            raise RuntimeError("线程池已关闭")
        
        future = concurrent.futures.Future()
        
        self._task_queue.put((priority, future, args, kwargs))
        
        return future
    
    def map(self, fn: Callable, *iterables, timeout=None, chunksize=1):
        """批量提交任务（类似 concurrent.futures 的 map）"""
        futures = []
        for args in zip(*iterables):
            futures.append(self.submit(fn, *args))
        
        for future in concurrent.futures.as_completed(futures, timeout=timeout):
            yield future.result()
    
    def shutdown(self, wait: bool = True):
        """
        关闭线程池
        
        Args:
            wait: 是否等待所有任务完成
        """
        self._shutdown = True
        
        if wait:
            self._task_queue.join()
        
        # 等待所有 Worker 线程结束
        for worker in self._workers:
            worker.join(timeout=5.0)
    
    @property
    def worker_count(self) -> int:
        """当前 Worker 数量"""
        return len(self._workers)

import asyncio
import typing
from typing import Optional, Callable, Any, List

if typing.TYPE_CHECKING:
    from asyncio import Task


class CoroutinePool:
    """
    协程池实现
    
    用于管理异步 I/O 操作的并发数量
    适用于：
    - HTTP 请求
    - 数据库查询（async）
    - 文件 I/O（aiofiles）
    - WebSocket 连接
    
    注意：协程池本身不创建新协程，而是限制并发数量
    """
    
    def __init__(self, max_concurrency: int = 100):
        self.max_concurrency = max_concurrency
        self._semaphore = asyncio.Semaphore(max_concurrency)
        self._active_count = 0
        self._lock = asyncio.Lock()
        self._waiting_tasks: List['asyncio.Task'] = []
    
    async def run(self, coro: Callable, *args, **kwargs) -> Any:
        """
        在协程池中运行一个协程
        
        使用信号量限制并发数量
        """
        async with self._semaphore:
            async with self._lock:
                self._active_count += 1
            
            try:
                result = await coro(*args, **kwargs)
                return result
            finally:
                async with self._lock:
                    self._active_count -= 1
    
    async def run_with_timeout(self, 
                               coro: Callable, 
                               timeout: float,
                               *args, **kwargs) -> Any:
        """带超时的协程执行"""
        try:
            return await asyncio.wait_for(
                self.run(coro, *args, **kwargs),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            raise TimeoutError(f"协程执行超时 ({timeout}s)")
    
    @property
    def active_count(self) -> int:
        """当前活跃的协程数量"""
        return self._active_count
    
    @property
    def available_capacity(self) -> int:
        """可用容量"""
        return self.max_concurrency - self._active_count

import time
import threading
import asyncio
from dataclasses import dataclass, field
from typing import Callable, List, Optional
from enum import Enum, auto

class ScaleDirection(Enum):
    """扩缩容方向"""
    UP = auto()
    DOWN = auto()
    NONE = auto()


@dataclass
class ScalingMetrics:
    """扩缩容决策所需的指标"""
    queue_depth: int = 0                    # 任务队列深度
    queue_wait_time_avg: float = 0.0        # 平均等待时间（秒）
    queue_wait_time_p99: float = 0.0        # P99 等待时间（秒）
    worker_utilization: float = 0.0         # Worker 利用率 0-1
    cpu_usage: float = 0.0                   # CPU 使用率 0-1
    memory_usage: float = 0.0                # 内存使用率 0-1
    active_workers: int = 0                  # 当前活跃 Worker 数
    timestamp: float = field(default_factory=time.time)


@dataclass
class ScalingConfig:
    """扩缩容配置"""
    min_workers: int = 1
    max_workers: int = 64
    
    # 扩容阈值
    scale_up_queue_depth: int = 10
    scale_up_utilization: float = 0.8
    scale_up_wait_time: float = 1.0  # 秒
    
    # 缩容阈值
    scale_down_queue_depth: int = 2
    scale_down_utilization: float = 0.3
    scale_down_wait_time: float = 0.1  # 秒
    
    # 扩缩容参数
    scale_up_ratio: float = 2.0
    scale_down_ratio: float = 0.5
    scale_step_max: int = 8  # 单次最多扩/缩容数
    
    # 冷却时间
    scale_up_cooldown: float = 10.0   # 秒
    scale_down_cooldown: float = 60.0  # 秒（缩容冷却更长）
    
    # 预测性扩缩容
    enable_predictive: bool = True
    prediction_window: float = 30.0  # 预测时间窗口（秒）


class ScalingPolicy:
    """
    扩缩容策略引擎
    
    支持：
    - 阈值触发
    - PID 控制
    - 预测性扩缩容
    """
    
    def __init__(self, config: ScalingConfig):
        self.config = config
        self._last_scale_up_time = 0.0
        self._last_scale_down_time = 0.0
        self._pending_scale_direction = ScaleDirection.NONE
    
    def evaluate(self, metrics: ScalingMetrics) -> tuple[ScaleDirection, int]:
        """
        评估是否需要扩缩容
        
        Returns:
            (方向, 目标 Worker 数)
        """
        current_time = time.time()
        
        # 扩容评估
        should_scale_up = (
            metrics.queue_depth > self.config.scale_up_queue_depth or
            metrics.worker_utilization > self.config.scale_up_utilization or
            metrics.queue_wait_time_avg > self.config.scale_up_wait_time
        )
        
        # 缩容评估
        should_scale_down = (
            metrics.queue_depth < self.config.scale_down_queue_depth and
            metrics.worker_utilization < self.config.scale_down_utilization and
            metrics.queue_wait_time_avg < self.config.scale_down_wait_time
        )
        
        # 冷却检查
        if should_scale_up:
            if current_time - self._last_scale_up_time >= self.config.scale_up_cooldown:
                new_count = min(
                    int(metrics.active_workers * self.config.scale_up_ratio),
                    metrics.active_workers + self.config.scale_step_max,
                    self.config.max_workers
                )
                self._last_scale_up_time = current_time
                return ScaleDirection.UP, new_count
        
        if should_scale_down:
            if current_time - self._last_scale_down_time >= self.config.scale_down_cooldown:
                new_count = max(
                    int(metrics.active_workers * self.config.scale_down_ratio),
                    metrics.active_workers - self.config.scale_step_max,
                    self.config.min_workers
                )
                self._last_scale_down_time = current_time
                return ScaleDirection.DOWN, new_count
        
        return ScaleDirection.NONE, metrics.active_workers

import numpy as np
from collections import deque

class PredictiveScaler:
    """
    预测性扩缩容器
    
    使用滑动窗口和线性回归预测负载趋势，
    在负载高峰到来之前提前扩容
    """
    
    def __init__(self, config: ScalingConfig, history_window: int = 60):
        self.config = config
        self.history_window = history_window
        
        # 历史数据
        self._queue_depth_history = deque(maxlen=history_window)
        self._arrival_rate_history = deque(maxlen=history_window)
        self._timestamp_history = deque(maxlen=history_window)
    
    def record(self, queue_depth: int, arrival_rate: float):
        """记录当前指标"""
        now = time.time()
        self._queue_depth_history.append(queue_depth)
        self._arrival_rate_history.append(arrival_rate)
        self._timestamp_history.append(now)
    
    def predict(self, horizon: float = 10.0) -> float:
        """
        预测指定时间后的队列深度
        
        Args:
            horizon: 预测时间范围（秒）
            
        Returns:
            预测的队列深度
        """
        if len(self._timestamp_history) < 10:
            # 数据不足，返回当前值
            return self._queue_depth_history[-1] if self._queue_depth_history else 0
        
        # 计算队列深度变化率
        times = np.array(list(self._timestamp_history))
        depths = np.array(list(self._queue_depth_history))
        
        # 线性回归：depth = slope * time + intercept
        slope, intercept = np.polyfit(times, depths, 1)
        
        # 预测
        predicted_time = time.time() + horizon
        predicted_depth = slope * predicted_time + intercept
        
        return max(0, predicted_depth)
    
    def should_preemptively_scale(self) -> tuple[bool, ScaleDirection, int]:
        """
        判断是否需要预测性扩缩容
        
        Returns:
            (是否需要, 方向, 目标 Worker 数)
        """
        # 预测 10 秒后的队列深度
        predicted_depth = self.predict(horizon=10.0)
        current_depth = self._queue_depth_history[-1] if self._queue_depth_history else 0
        
        # 如果预测值比当前值大 50%，考虑提前扩容
        if predicted_depth > current_depth * 1.5 and predicted_depth > self.config.scale_up_queue_depth:
            # 计算需要的 Worker 数
            # 假设每个 Worker 每秒处理 1 个任务
            needed_workers = int(predicted_depth / self.config.scale_up_wait_time) + 1
            return True, ScaleDirection.UP, min(needed_workers, self.config.max_workers)
        
        # 如果预测值持续走低，考虑提前缩容
        if predicted_depth < current_depth * 0.5 and predicted_depth < self.config.scale_down_queue_depth:
            needed_workers = max(1, int(predicted_depth / self.config.scale_down_wait_time))
            return True, ScaleDirection.DOWN, max(needed_workers, self.config.min_workers)
        
        return False, ScaleDirection.NONE, 0

import asyncio
from typing import Protocol

class WorkerFactory(Protocol):
    """Worker 工厂接口"""
    def create_worker(self, worker_id: int) -> Worker: ...
    def destroy_worker(self, worker: Worker) -> None: ...


class ScalingExecutor:
    """
    扩缩容执行器
    
    负责执行扩缩容决策，包括：
    - Worker 的创建和销毁
    - 迁移进行中的任务
    - 发送通知
    """
    
    def __init__(self, pool, factory: WorkerFactory, config: ScalingConfig):
        self.pool = pool
        self.factory = factory
        self.config = config
        self._is_scaling = False
        self._scale_lock = asyncio.Lock()
    
    async def scale_to(self, target_count: int) -> bool:
        """
        将池扩展到目标 Worker 数
        
        Returns:
            True 如果扩缩容成功
        """
        async with self._scale_lock:
            if self._is_scaling:
                return False
            
            self._is_scaling = True
            current_count = self.pool.worker_count
            
            try:
                if target_count > current_count:
                    # 扩容
                    await self._scale_up(target_count - current_count)
                elif target_count < current_count:
                    # 缩容
                    await self._scale_down(current_count - target_count)
                
                return True
                
            finally:
                self._is_scaling = False
    
    async def _scale_up(self, count: int):
        """扩容指定数量的 Worker"""
        for i in range(count):
            worker_id = self.pool.next_worker_id()
            worker = self.factory.create_worker(worker_id)
            
            # 初始化 Worker
            if not worker.initialize():
                self.factory.destroy_worker(worker)
                continue
            
            # 添加到池
            self.pool.add_worker(worker)
            
            # 小延迟避免瞬时压力
            await asyncio.sleep(0.01)
    
    async def _scale_down(self, count: int):
        """缩容指定数量的 Worker"""
        workers_to_remove = self.pool.get_idle_workers(count)
        
        for worker in workers_to_remove:
            # 优雅停止 Worker
            await asyncio.to_thread(worker.stop, timeout=5.0)
            
            # 从池中移除并销毁
            self.pool.remove_worker(worker)
            self.factory.destroy_worker(worker)
            
            await asyncio.sleep(0.01)

import threading
import queue
import time
import logging
import asyncio
from typing import Optional, Callable, Any, Dict, List
from dataclasses import dataclass, field
from enum import Enum, auto
from contextlib import contextmanager

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


# ============ 数据结构定义 ============

class PoolState(Enum):
    """池状态"""
    INITIALIZING = auto()
    RUNNING = auto()
    DRAINING = auto()
    STOPPED = auto()


@dataclass
class PoolMetrics:
    """池级别指标"""
    tasks_submitted: int = 0
    tasks_completed: int = 0
    tasks_failed: int = 0
    total_wait_time: float = 0.0
    total_execution_time: float = 0.0
    peak_queue_depth: int = 0
    peak_worker_count: int = 0
    scale_events: int = 0
    
    @property
    def avg_wait_time(self) -> float:
        if self.tasks_completed == 0:
            return 0.0
        return self.total_wait_time / self.tasks_completed
    
    @property
    def avg_execution_time(self) -> float:
        if self.tasks_completed == 0:
            return 0.0
        return self.total_execution_time / self.tasks_completed


class Task:
    """任务对象"""
    def __init__(self, 
                 task_id: str, 
                 payload: Any, 
                 callback: Optional[Callable] = None,
                 priority: int = 5,
                 timeout: float = 30.0):
        self.id = task_id
        self.payload = payload
        self.callback = callback
        self.priority = priority
        self.timeout = timeout
        self.created_at = time.time()
        self.started_at: Optional[float] = None
        self.completed_at: Optional[float] = None
        
    def __lt__(self, other):
        # 支持优先队列
        return self.priority < other.priority


# ============ Worker 实现 ============

class Worker:
    """Worker 实现"""
    
    def __init__(self, worker_id: str, task_handler: Callable[[Task], Any]):
        self.worker_id = worker_id
        self.task_handler = task_handler
        self._state = "idle"  # idle, busy, stopping
        self._lock = threading.Lock()
        self._current_task: Optional[Task] = None
        self._thread: Optional[threading.Thread] = None
        self._stop_event = threading.Event()
        
        # 指标
        self.tasks_completed = 0
        self.total_execution_time = 0.0
        
    @property
    def state(self) -> str:
        with self._lock:
            return self._state
    
    @property
    def is_idle(self) -> bool:
        return self.state == "idle"
    
    def start(self, task_queue: queue.PriorityQueue, idle_workers: List['Worker']):
        """启动 Worker 线程"""
        self._thread = threading.Thread(
            target=self._run,
            args=(task_queue, idle_workers),
            name=f"Worker-{self.worker_id}",
            daemon=True
        )
        self._thread.start()
    
    def _run(self, task_queue: queue.PriorityQueue, idle_workers: List['Worker']):
        """Worker 主循环"""
        while not self._stop_event.is_set():
            try:
                # 尝试获取任务
                try:
                    task = task_queue.get(timeout=0.1)
                except queue.Empty:
                    # 队列为空，将自己加入空闲列表
                    with self._lock:
                        if self._state == "idle":
                            # 在锁外操作 idle_workers 以避免死锁
                            pass
                    continue
                
                # 执行任务
                self._execute_task(task, task_queue)
                
            except Exception as e:
                logger.error(f"Worker {self.worker_id} 错误: {e}")
    
    def _execute_task(self, task: Task, task_queue: queue.PriorityQueue):
        """执行单个任务"""
        with self._lock:
            self._state = "busy"
            self._current_task = task
            task.started_at = time.time()
        
        exception = None
        result = None
        
        try:
            # 执行任务（带超时）
            result = self._execute_with_timeout(task)
        except Exception as e:
            exception = e
            logger.error(f"Worker {self.worker_id} 执行任务 {task.id} 失败: {e}")
        
        with self._lock:
            task.completed_at = time.time()
            self._current_task = None
            self._state = "idle"
            
            execution_time = task.completed_at - task.started_at
            self.total_execution_time += execution_time
            self.tasks_completed += 1
        
        # 标记任务完成
        task_queue.task_done()
        
        # 调用回调
        if task.callback:
            try:
                task.callback(result, exception)
            except Exception as e:
                logger.error(f"Worker {self.worker_id} 回调失败: {e}")
    
    def _execute_with_timeout(self, task: Task) -> Any:
        """带超时执行任务"""
        result = [None]
        exception = [None]
        finished = threading.Event()
        
        def worker_func():
            try:
                result[0] = self.task_handler(task)
            except Exception as e:
                exception[0] = e
            finally:
                finished.set()
        
        thread = threading.Thread(target=worker_func)
        thread.start()
        
        # 等待完成或超时
        if finished.wait(task.timeout):
            thread.join()
            if exception[0]:
                raise exception[0]
            return result[0]
        else:
            # 超时
            thread.join(timeout=1.0)
            raise TimeoutError(f"任务 {task.id} 执行超时 ({task.timeout}s)")
    
    def stop(self, timeout: float = 5.0):
        """停止 Worker"""
        self._stop_event.set()
        if self._thread:
            self._thread.join(timeout=timeout)


# ============ Worker Pool 主实现 ============

class WorkerPool:
    """
    支持动态扩缩的 Worker Pool
    
    特性：
    - 固定核心池 + 弹性扩展池
    - 基于阈值的自动扩缩容
    - 优先级任务队列
    - 完整的指标收集
    - 优雅关闭协议
    """
    
    def __init__(self,
                 task_handler: Callable[[Task], Any],
                 min_workers: int = 4,
                 max_workers: int = 32,
                 queue_size: int = 1000,
                 scale_up_threshold: int = 10,
                 scale_down_threshold: int = 2,
                 scale_cooldown: float = 10.0,
                 idle_timeout: float = 60.0):
        
        self.task_handler = task_handler
        self.min_workers = min_workers
        self.max_workers = max_workers
        self.scale_up_threshold = scale_up_threshold
        self.scale_down_threshold = scale_down_threshold
        self.scale_cooldown = scale_cooldown
        self.idle_timeout = idle_timeout
        
        # 状态
        self._state = PoolState.INITIALIZING
        self._state_lock = threading.RLock()
        
        # 任务队列
        self._task_queue: queue.PriorityQueue = queue.PriorityQueue(maxsize=queue_size)
        
        # Workers
        self._workers: Dict[str, Worker] = {}
        self._worker_lock = threading.RLock()
        self._next_worker_id = 0
        
        # 空闲 Worker 列表
        self._idle_workers: List[Worker] = []
        self._idle_lock = threading.Lock()
        
        # 指标
        self._metrics = PoolMetrics()
        self._metrics_lock = threading.Lock()
        
        # 扩缩容
        self._last_scale_time = 0.0
        self._scaling_lock = threading.Lock()
        
        # 后台线程
        self._monitor_thread: Optional[threading.Thread] = None
        self._monitor_stop = threading.Event()
        
        # 初始化
        self._initialize()
    
    def _initialize(self):
        """初始化池"""
        with self._state_lock:
            self._state = PoolState.INITIALIZING
            
            # 创建核心 Worker
            for _ in range(self.min_workers):
                self._add_worker()
            
            # 启动监控线程
            self._monitor_stop.clear()
            self._monitor_thread = threading.Thread(
                target=self._monitor_loop,
                name="PoolMonitor",
                daemon=True
            )
            self._monitor_thread.start()
            
            self._state = PoolState.RUNNING
            
        logger.info(f"WorkerPool 初始化完成，核心 Worker 数: {self.min_workers}")
    
    def _add_worker(self) -> Worker:
        """添加一个 Worker"""
        with self._worker_lock:
            worker_id = f"w{self._next_worker_id}"
            self._next_worker_id += 1
            
            worker = Worker(worker_id, self.task_handler)
            self._workers[worker_id] = worker
            worker.start(self._task_queue, self._idle_workers)
            
            with self._metrics_lock:
                self._metrics.peak_worker_count = max(
                    self._metrics.peak_worker_count,
                    len(self._workers)
                )
            
            return worker
    
    def _remove_worker(self, worker: Worker) -> bool:
        """移除一个 Worker"""
        with self._worker_lock:
            if worker.worker_id in self._workers:
                worker.stop()
                del self._workers[worker.worker_id]
                return True
            return False
    
    @property
    def worker_count(self) -> int:
        with self._worker_lock:
            return len(self._workers)
    
    @property
    def metrics(self) -> PoolMetrics:
        with self._metrics_lock:
            return PoolMetrics(
                tasks_submitted=self._metrics.tasks_submitted,
                tasks_completed=self._metrics.tasks_completed,
                tasks_failed=self._metrics.tasks_failed,
                total_wait_time=self._metrics.total_wait_time,
                total_execution_time=self._metrics.total_execution_time,
                peak_queue_depth=self._metrics.peak_queue_depth,
                peak_worker_count=self._metrics.peak_worker_count,
                scale_events=self._metrics.scale_events
            )
    
    def submit(self, 
               payload: Any,
               callback: Optional[Callable] = None,
               priority: int = 5,
               timeout: float = 30.0) -> Task:
        """
        提交任务到池
        
        Args:
            payload: 任务载荷
            callback: 完成回调 (result, exception) -> None
            priority: 优先级（数值越小越高）
            timeout: 超时时间（秒）
            
        Returns:
            Task 对象
        """
        with self._state_lock:
            if self._state != PoolState.RUNNING:
                raise RuntimeError(f"池状态错误: {self._state}")
        
        task = Task(
            task_id=f"t{int(time.time() * 1000000)}",
            payload=payload,
            callback=callback,
            priority=priority,
            timeout=timeout
        )
        
        self._task_queue.put(task)
        
        with self._metrics_lock:
            self._metrics.tasks_submitted += 1
            self._metrics.peak_queue_depth = max(
                self._metrics.peak_queue_depth,
                self._task_queue.qsize()
            )
        
        return task
    
    def _monitor_loop(self):
        """监控和扩缩容循环"""
        while not self._monitor_stop.is_set():
            try:
                self._check_scaling()
            except Exception as e:
                logger.error(f"扩缩容检查失败: {e}")
            
            time.sleep(1.0)
    
    def _check_scaling(self):
        """检查是否需要扩缩容"""
        current_time = time.time()
        
        # 冷却检查
        if current_time - self._last_scale_time < self.scale_cooldown:
            return
        
        queue_depth = self._task_queue.qsize()
        worker_count = self.worker_count
        
        with self._metrics_lock:
            metrics = self._metrics
        
        # 扩容决策
        if queue_depth > self.scale_up_threshold and worker_count < self.max_workers:
            # 需要扩容
            new_count = min(
                worker_count * 2,
                self.max_workers
            )
            scale_delta = new_count - worker_count
            
            if scale_delta > 0:
                logger.info(f"扩容: {worker_count} -> {new_count} (队列深度: {queue_depth})")
                
                for _ in range(scale_delta):
                    self._add_worker()
                
                with self._metrics_lock:
                    self._metrics.scale_events += 1
                
                self._last_scale_time = current_time
        
        # 缩容决策
        elif queue_depth < self.scale_down_threshold and worker_count > self.min_workers:
            # 需要缩容
            new_count = max(
                worker_count // 2,
                self.min_workers
            )
            scale_delta = worker_count - new_count
            
            if scale_delta > 0:
                # 获取空闲的 Worker
                idle_workers = [w for w in self._workers.values() if w.is_idle]
                
                if len(idle_workers) >= scale_delta:
                    logger.info(f"缩容: {worker_count} -> {new_count} (队列深度: {queue_depth})")
                    
                    for worker in idle_workers[:scale_delta]:
                        self._remove_worker(worker)
                    
                    with self._metrics_lock:
                        self._metrics.scale_events += 1
                    
                    self._last_scale_time = current_time
    
    @contextmanager
    def submit_and_wait(self, payload: Any, timeout: float = 30.0):
        """
        提交任务并等待结果（同步接口）
        
        用法：
            with pool.submit_and_wait(data) as result:
                print(result)
        """
        result = [None]
        exception = [None]
        event = threading.Event()
        
        def callback(res, exc):
            result[0] = res
            exception[0] = exc
            event.set()
        
        task = self.submit(payload, callback=callback, timeout=timeout)
        
        if not event.wait(timeout):
            raise TimeoutError(f"任务 {task.id} 执行超时")
        
        if exception[0]:
            raise exception[0]
        
        yield result[0]
    
    def shutdown(self, wait: bool = True, timeout: float = 30.0):
        """
        关闭池
        
        Args:
            wait: 是否等待所有任务完成
            timeout: 等待超时
        """
        with self._state_lock:
            self._state = PoolState.DRAINING
        
        # 停止监控
        self._monitor_stop.set()
        if self._monitor_thread:
            self._monitor_thread.join(timeout=5.0)
        
        if wait:
            # 等待队列清空
            start_time = time.time()
            while self._task_queue.qsize() > 0:
                if time.time() - start_time > timeout:
                    logger.warning("关闭超时，强制停止")
                    break
                time.sleep(0.1)
        
        # 停止所有 Worker
        with self._worker_lock:
            for worker in list(self._workers.values()):
                worker.stop(timeout=timeout / len(self._workers))
            self._workers.clear()
        
        with self._state_lock:
            self._state = PoolState.STOPPED
        
        logger.info(f"WorkerPool 已关闭，指标: {self.metrics}")

import time
import random

def example_task_handler(task: Task) -> str:
    """示例任务处理器"""
    # 模拟处理时间
    process_time = random.uniform(0.1, 0.5)
    time.sleep(process_time)
    
    # 模拟处理
    return f"任务 {task.id} 处理完成，耗时 {process_time:.3f}s"


def example_callback(result, exception):
    """示例回调函数"""
    if exception:
        print(f"任务失败: {exception}")
    else:
        print(f"任务成功: {result}")


def main():
    # 创建 Worker Pool
    pool = WorkerPool(
        task_handler=example_task_handler,
        min_workers=4,
        max_workers=16,
        scale_up_threshold=10,
        scale_down_threshold=2
    )
    
    print(f"Pool 创建完成，Worker 数: {pool.worker_count}")
    
    # 提交任务
    tasks = []
    for i in range(20):
        task = pool.submit(
            payload={"data": i},
            callback=example_callback,
            priority=5,
            timeout=30.0
        )
        tasks.append(task)
        time.sleep(0.05)  # 模拟任务到达间隔
    
    # 同步等待任务
    print("\n同步等待模式:")
    for i in range(5):
        try:
            with pool.submit_and_wait({"sync": i}, timeout=10.0) as result:
                print(f"  同步任务 {i}: {result}")
        except TimeoutError as e:
            print(f"  同步任务 {i} 超时: {e}")
    
    # 等待一段时间观察自动扩缩容
    print("\n观察自动扩缩容:")
    time.sleep(15)
    
    # 输出指标
    metrics = pool.metrics
    print(f"\n=== 池指标 ===")
    print(f"提交任务数: {metrics.tasks_submitted}")
    print(f"完成任务数: {metrics.tasks_completed}")
    print(f"失败任务数: {metrics.tasks_failed}")
    print(f"峰值队列深度: {metrics.peak_queue_depth}")
    print(f"峰值 Worker 数: {metrics.peak_worker_count}")
    print(f"扩缩容次数: {metrics.scale_events}")
    print(f"当前 Worker 数: {pool.worker_count}")
    
    # 关闭池
    pool.shutdown(wait=True, timeout=30.0)


if __name__ == "__main__":
    main()

"""
Worker Pool 完整实现
支持：动态扩缩容、优先级队列、完整指标收集、优雅关闭
"""

import threading
import queue
import time
import logging
import uuid
from typing import Any, Callable, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum, auto
from contextlib import contextmanager

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class PoolState(Enum):
    """池状态枚举"""
    INITIALIZING = auto()
    RUNNING = auto()
    DRAINING = auto()
    STOPPED = auto()


@dataclass
class PoolMetrics:
    """池级别性能指标"""
    tasks_submitted: int = 0
    tasks_completed: int = 0
    tasks_failed: int = 0
    total_wait_time: float = 0.0
    total_execution_time: float = 0.0
    peak_queue_depth: int = 0
    peak_worker_count: int = 0
    scale_events: int = 0
    last_scale_time: float = 0.0


class Task:
    """任务封装"""
    __slots__ = ['id', 'payload', 'callback', 'priority', 'timeout',
                 'created_at', 'started_at', 'completed_at', 'future']
    
    def __init__(self, payload: Any, callback: Optional[Callable] = None,
                 priority: int = 5, timeout: float = 30.0):
        self.id = str(uuid.uuid4())
        self.payload = payload
        self.callback = callback
        self.priority = priority
        self.timeout = timeout
        self.created_at = time.time()
        self.started_at: Optional[float] = None
        self.completed_at: Optional[float] = None
        self.future: Optional[threading.Event] = None


class Worker:
    """Worker 实现"""
    
    def __init__(self, worker_id: str, task_handler: Callable):
        self.worker_id = worker_id
        self.task_handler = task_handler
        self._state = "idle"
        self._lock = threading.Lock()
        self._current_task: Optional[Task] = None
        self._stop_event = threading.Event()
        self._thread: Optional[threading.Thread] = None
        self.tasks_completed = 0
        self.total_execution_time = 0.0
    
    @property
    def is_idle(self) -> bool:
        return self._state == "idle"
    
    def start(self, task_queue: queue.PriorityQueue):
        self._thread = threading.Thread(
            target=self._run,
            args=(task_queue,),
            name=f"Worker-{self.worker_id}",
            daemon=True
        )
        self._thread.start()
    
    def _run(self, task_queue: queue.PriorityQueue):
        while not self._stop_event.is_set():
            try:
                task = task_queue.get(timeout=0.1)
                self._execute_task(task)
                task_queue.task_done()
            except queue.Empty:
                continue
            except Exception as e:
                logger.error(f"Worker {self.worker_id} 错误: {e}")
    
    def _execute_task(self, task: Task):
        with self._lock:
            self._state = "busy"
            self._current_task = task
            task.started_at = time.time()
        
        try:
            result = self.task_handler(task)
            if task.callback:
                task.callback(result, None)
            if task.future:
                task.future.set()
        except Exception as e:
            if task.callback:
                task.callback(None, e)
            if task.future:
                task.future.set()
        finally:
            with self._lock:
                task.completed_at = time.time()
                self.total_execution_time += task.completed_at - task.started_at
                self.tasks_completed += 1
                self._current_task = None
                self._state = "idle"
    
    def stop(self, timeout: float = 5.0):
        self._stop_event.set()
        if self._thread:
            self._thread.join(timeout=timeout)


class WorkerPool:
    """
    生产级 Worker Pool 实现
    
    特性：
    - 核心池 + 弹性池混合架构
    - 优先级任务队列
    - 基于阈值的自动扩缩容
    - 完整的指标收集
    - 优雅关闭协议
    """
    
    def __init__(self,
                 task_handler: Callable,
                 min_workers: int = 4,
                 max_workers: int = 32,
                 queue_size: int = 1000,
                 scale_up_threshold: int = 10,
                 scale_down_threshold: int = 2,
                 scale_cooldown: float = 10.0):
        
        self.task_handler = task_handler
        self.min_workers = min_workers
        self.max_workers = max_workers
        self.scale_up_threshold = scale_up_threshold
        self.scale_down_threshold = scale_down_threshold
        self.scale_cooldown = scale_cooldown
        
        self._state = PoolState.INITIALIZING
        self._state_lock = threading.RLock()
        
        self._task_queue: queue.PriorityQueue = queue.PriorityQueue(maxsize=queue_size)
        self._workers: Dict[str, Worker] = {}
        self._worker_lock = threading.RLock()
        self._next_worker_id = 0
        
        self._metrics = PoolMetrics()
        self._metrics_lock = threading.Lock()
        
        self._monitor_thread: Optional[threading.Thread] = None
        self._monitor_stop = threading.Event()
        
        self._initialize()
    
    def _initialize(self):
        self._state = PoolState.INITIALIZING
        
        for _ in range(self.min_workers):
            self._add_worker()
        
        self._monitor_stop.clear()
        self._monitor_thread = threading.Thread(
            target=self._monitor_loop,
            name="PoolMonitor",
            daemon=True
        )
        self._monitor_thread.start()
        
        self._state = PoolState.RUNNING
        logger.info(f"WorkerPool 初始化完成，核心 Worker 数: {self.min_workers}")
    
    def _add_worker(self) -> Worker:
        with self._worker_lock:
            worker_id = f"w{self._next_worker_id}"
            self._next_worker_id += 1
            
            worker = Worker(worker_id, self.task_handler)
            self._workers[worker_id] = worker
            worker.start(self._task_queue)
            
            with self._metrics_lock:
                self._metrics.peak_worker_count = max(
                    self._metrics.peak_worker_count,
                    len(self._workers)
                )
            
            return worker
    
    def _remove_worker(self, worker: Worker):
        with self._worker_lock:
            if worker.worker_id in self._workers:
                worker.stop()
                del self._workers[worker.worker_id]
    
    @property
    def worker_count(self) -> int:
        with self._worker_lock:
            return len(self._workers)
    
    @property
    def metrics(self) -> PoolMetrics:
        with self._metrics_lock:
            return PoolMetrics(
                tasks_submitted=self._metrics.tasks_submitted,
                tasks_completed=self._metrics.tasks_completed,
                tasks_failed=self._metrics.tasks_failed,
                total_wait_time=self._metrics.total_wait_time,
                total_execution_time=self._metrics.total_execution_time,
                peak_queue_depth=self._metrics.peak_queue_depth,
                peak_worker_count=self._metrics.peak_worker_count,
                scale_events=self._metrics.scale_events
            )
    
    def submit(self, payload: Any, callback: Optional[Callable] = None,
              priority: int = 5, timeout: float = 30.0) -> Task:
        
        with self._state_lock:
            if self._state != PoolState.RUNNING:
                raise RuntimeError(f"池状态错误: {self._state}")
        
        task = Task(payload, callback, priority, timeout)
        self._task_queue.put(task)
        
        with self._metrics_lock:
            self._metrics.tasks_submitted += 1
            self._metrics.peak_queue_depth = max(
                self._metrics.peak_queue_depth,
                self._task_queue.qsize()
            )
        
        return task
    
    def _monitor_loop(self):
        while not self._monitor_stop.is_set():
            try:
                self._check_scaling()
            except Exception as e:
                logger.error(f"扩缩容检查失败: {e}")
            time.sleep(1.0)
    
    def _check_scaling(self):
        current_time = time.time()
        
        if current_time - self._metrics.last_scale_time < self.scale_cooldown:
            return
        
        queue_depth = self._task_queue.qsize()
        worker_count = self.worker_count
        
        if queue_depth > self.scale_up_threshold and worker_count < self.max_workers:
            new_count = min(worker_count * 2, self.max_workers)
            scale_delta = new_count - worker_count
            
            if scale_delta > 0:
                logger.info(f"扩容: {worker_count} -> {new_count}")
                for _ in range(scale_delta):
                    self._add_worker()
                with self._metrics_lock:
                    self._metrics.scale_events += 1
                    self._metrics.last_scale_time = current_time
        
        elif queue_depth < self.scale_down_threshold and worker_count > self.min_workers:
            new_count = max(worker_count // 2, self.min_workers)
            scale_delta = worker_count - new_count
            
            if scale_delta > 0:
                idle_workers = [w for w in self._workers.values() if w.is_idle]
                if len(idle_workers) >= scale_delta:
                    logger.info(f"缩容: {worker_count} -> {new_count}")
                    for worker in idle_workers[:scale_delta]:
                        self._remove_worker(worker)
                    with self._metrics_lock:
                        self._metrics.scale_events += 1
                        self._metrics.last_scale_time = current_time
    
    @contextmanager
    def submit_and_wait(self, payload: Any, timeout: float = 30.0):
        future = threading.Event()
        task = self.submit(payload, callback=lambda r, e: future.set(), timeout=timeout)
        task.future = future
        
        if not future.wait(timeout):
            raise TimeoutError(f"任务 {task.id} 执行超时")
        
        if hasattr(task, '_exception') and task._exception:
            raise task._exception
        
        yield task._result if hasattr(task, '_result') else None
    
    def shutdown(self, wait: bool = True, timeout: float = 30.0):
        with self._state_lock:
            self._state = PoolState.DRAINING
        
        self._monitor_stop.set()
        if self._monitor_thread:
            self._monitor_thread.join(timeout=5.0)
        
        if wait:
            start = time.time()
            while self._task_queue.qsize() > 0:
                if time.time() - start > timeout:
                    break
                time.sleep(0.1)
        
        with self._worker_lock:
            for worker in list(self._workers.values()):
                worker.stop()
            self._workers.clear()
        
        with self._state_lock:
            self._state = PoolState.STOPPED
        
        logger.info(f"WorkerPool 已关闭，指标: {self.metrics}")