# ============================================================ # 🔄 ASYNC MANAGER - Execução Segura sem Bloqueio # ============================================================ """ Gerenciador de tarefas assíncronas para garantir que: 1. Respostas NUNCA travam (resposta retornada IMEDIATAMENTE) 2. Treinamento executa em background ISOLADO 3. Database access é sincronizado 4. Múltiplas requisições não competem por recursos """ import asyncio import threading import time import queue from typing import Callable, Any, Optional, Dict, List from dataclasses import dataclass, field from datetime import datetime, timedelta from loguru import logger # ============================================================ # 📋 ESTRUTURAS DE DADOS # ============================================================ @dataclass class AsyncTask: """Representa uma tarefa assíncrona""" task_id: str name: str func: Callable args: tuple = field(default_factory=tuple) kwargs: dict = field(default_factory=dict) priority: int = 0 # 0=low, 1=medium, 2=high created_at: datetime = field(default_factory=datetime.now) started_at: Optional[datetime] = None completed_at: Optional[datetime] = None status: str = "queued" # queued, running, completed, failed error: Optional[str] = None result: Any = None timeout: float = 30.0 # segundos class ResponseBlockManager: """Gerencia execução de tarefas sem bloquear respostas API""" def __init__(self, max_workers: int = 3, queue_size: int = 100): self.max_workers = max_workers self.queue_size = queue_size self._task_queue = queue.PriorityQueue(maxsize=queue_size) self._running_tasks: Dict[str, AsyncTask] = {} self._completed_tasks: List[AsyncTask] = [] self._workers: List[threading.Thread] = [] self._running = False self._lock = threading.Lock() # Stats self._stats = { "total_tasks": 0, "completed_tasks": 0, "failed_tasks": 0, "avg_task_time": 0.0, "queue_depth": 0 } logger.info(f"🔄 ResponseBlockManager inicializado ({max_workers} workers, queue_size={queue_size})") def start(self): """Inicia workers""" if self._running: return self._running = True for i in range(self.max_workers): worker = threading.Thread( target=self._worker_loop, name=f"AsyncWorker-{i}", daemon=True ) worker.start() self._workers.append(worker) logger.success(f"✅ {self.max_workers} workers iniciados") def stop(self, timeout: float = 10.0): """Para workers com timeout""" self._running = False # Aguarda workers terminarem for worker in self._workers: worker.join(timeout=timeout / len(self._workers)) logger.info("⏹️ ResponseBlockManager parado") def queue_task( self, name: str, func: Callable, args: tuple = (), kwargs: Optional[Dict] = None, priority: int = 0, timeout: float = 30.0, task_id: Optional[str] = None ) -> str: """ Enfileira tarefa para execução em background Returns: task_id para rastreamento """ if not self._running: logger.warning("⚠️ ResponseBlockManager não está rodando") return "" # Gera ID se não fornecido if not task_id: task_id = f"{name}_{int(time.time() * 1000)}" task = AsyncTask( task_id=task_id, name=name, func=func, args=args, kwargs=kwargs or {}, priority=priority, timeout=timeout ) try: # Enfileira com prioridade (inverte para usar max heap) self._task_queue.put((-priority, time.time(), task), timeout=1.0) with self._lock: self._stats["total_tasks"] += 1 self._stats["queue_depth"] = self._task_queue.qsize() logger.debug(f"📥 Task enfileirada: {name} (id={task_id}, priority={priority})") return task_id except queue.Full: logger.error(f"❌ Fila de tarefas cheia! Ignorando: {name}") return "" def _worker_loop(self): """Loop de execução de tarefas""" while self._running: try: # Espera por tarefa com timeout try: _, _, task = self._task_queue.get(timeout=1.0) except queue.Empty: continue # Marca como em execução with self._lock: self._running_tasks[task.task_id] = task task.status = "running" task.started_at = datetime.now() logger.debug(f"⚙️ Executando: {task.name} (id={task.task_id})") # Executa com timeout try: result = self._execute_with_timeout(task) task.result = result task.status = "completed" with self._lock: self._stats["completed_tasks"] += 1 self._completed_tasks.append(task) logger.success(f"✅ Task completada: {task.name}") except TimeoutError: task.status = "failed" task.error = f"Timeout ({task.timeout}s)" with self._lock: self._stats["failed_tasks"] += 1 logger.error(f"❌ Task timeout: {task.name}") except Exception as e: task.status = "failed" task.error = str(e) with self._lock: self._stats["failed_tasks"] += 1 logger.error(f"❌ Task failed: {task.name} - {e}") finally: # Remove de running with self._lock: task.completed_at = datetime.now() if task.task_id in self._running_tasks: del self._running_tasks[task.task_id] # Calcula tempo médio if task.completed_at and task.started_at: elapsed = (task.completed_at - task.started_at).total_seconds() self._stats["avg_task_time"] = ( self._stats["avg_task_time"] * 0.8 + elapsed * 0.2 ) except Exception as e: logger.exception(f"Erro fatal em worker: {e}") def _execute_with_timeout(self, task: AsyncTask) -> Any: """Executa função com timeout""" result_container = {} exception_container = {} def _wrapper(): try: result = task.func(*task.args, **task.kwargs) result_container['result'] = result except Exception as e: exception_container['error'] = e thread = threading.Thread(target=_wrapper, daemon=False) thread.start() thread.join(timeout=task.timeout) if thread.is_alive(): raise TimeoutError(f"Execução não completou em {task.timeout}s") if 'error' in exception_container: raise exception_container['error'] return result_container.get('result') def get_task_status(self, task_id: str) -> Dict[str, Any]: """Retorna status de uma tarefa""" # Procura em running if task_id in self._running_tasks: task = self._running_tasks[task_id] return { "id": task_id, "name": task.name, "status": task.status, "started_at": task.started_at.isoformat() if task.started_at else None, "completed_at": None # ainda em execução } # Procura em completed for task in reversed(self._completed_tasks[-100:]): # últimas 100 if task.task_id == task_id: return { "id": task_id, "name": task.name, "status": task.status, "started_at": task.started_at.isoformat() if task.started_at else None, "completed_at": task.completed_at.isoformat() if task.completed_at else None, "error": task.error, "result": task.result } return {"id": task_id, "status": "not_found"} def get_stats(self) -> Dict[str, Any]: """Retorna estatísticas""" with self._lock: return { **self._stats, "running_tasks": len(self._running_tasks), "completed_history": len(self._completed_tasks), "workers_active": len([w for w in self._workers if w.is_alive()]) } def clear_history(self): """Limpa histórico de tarefas completadas""" with self._lock: self._completed_tasks.clear() logger.info("📊 Histórico de tarefas limpo") # ============================================================ # 🌐 INSTÂNCIA GLOBAL # ============================================================ # Inicializa com 3 workers (não-bloqueante) async_manager = ResponseBlockManager(max_workers=3, queue_size=100) # Inicia automaticamente ao importar try: async_manager.start() except Exception as e: logger.warning(f"⚠️ Erro ao iniciar async_manager: {e}") # ============================================================ # 🎯 HELPERS # ============================================================ def queue_training_task( func: Callable, args: tuple = (), kwargs: Optional[Dict] = None, task_id: Optional[str] = None ) -> str: """Enfileira tarefa de treinamento (baixa prioridade)""" return async_manager.queue_task( name="training", func=func, args=args, kwargs=kwargs, priority=0, # baixa prioridade timeout=60.0, # timeout maior para treinamento task_id=task_id ) def queue_data_task( func: Callable, args: tuple = (), kwargs: Optional[Dict] = None, task_id: Optional[str] = None ) -> str: """Enfileira tarefa de armazenamento de dados (média prioridade)""" return async_manager.queue_task( name="data_storage", func=func, args=args, kwargs=kwargs, priority=1, # média prioridade timeout=10.0, task_id=task_id ) def queue_urgent_task( func: Callable, args: tuple = (), kwargs: Optional[Dict] = None, task_id: Optional[str] = None ) -> str: """Enfileira tarefa urgente (alta prioridade)""" return async_manager.queue_task( name="urgent", func=func, args=args, kwargs=kwargs, priority=2, # alta prioridade timeout=5.0, task_id=task_id )