File size: 4,709 Bytes
a39bf55 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 |
# GPUManager.py
import torch
from typing import List, Dict
class GPUManager:
"""
Classe singleton para gerenciar e alocar recursos de GPU de forma centralizada.
Esta classe detecta as GPUs disponíveis e as aloca para diferentes
tipos de tarefas com base em uma estratégia predefinida. Ela foi projetada
para ser extensível a outros modelos e tarefas no futuro.
Estratégia Padrão:
- LTX_PRIMARY: Tarefas pesadas e sequenciais (Transformer, Text Encoder).
Alocado para a primeira GPU (cuda:0) para maximizar o desempenho.
- VAE_POOL: Tarefas mais leves e paralelizáveis (decodificação VAE).
Alocado para as GPUs restantes (cuda:1, cuda:2, ...) em um pool de workers.
Em caso de GPU única, todos os modelos compartilham o mesmo dispositivo.
Em caso de ausência de GPUs, opera em modo CPU.
"""
def __init__(self):
self.num_gpus = 0
self.devices: List[str] = []
self._allocations: Dict[str, List[str]] = {}
self._worker_indices: Dict[str, int] = {}
if torch.cuda.is_available():
self.num_gpus = torch.cuda.device_count()
self._initialize_allocations()
def _initialize_allocations(self):
"""Define a estratégia de alocação de dispositivos com base nas GPUs disponíveis."""
print("[GPUManager] Inicializando alocação de dispositivos...")
if self.num_gpus == 0:
print("[GPUManager] Nenhuma GPU CUDA detectada. Operando em modo CPU.")
self.devices = ["cpu"]
self._allocations['LTX_PRIMARY'] = ["cpu"]
self._allocations['VAE_POOL'] = ["cpu"]
elif self.num_gpus == 1:
print("[GPUManager] Detectada 1 GPU. Todos os modelos compartilharão cuda:0.")
self.devices = ["cuda:0"]
self._allocations['LTX_PRIMARY'] = ["cuda:0"]
self._allocations['VAE_POOL'] = ["cuda:0"]
else: # Múltiplas GPUs
print(f"[GPUManager] Detectadas {self.num_gpus} GPUs. Ativando modo Multi-GPU.")
self.devices = [f"cuda:{i}" for i in range(self.num_gpus)]
# Aloca a primeira GPU para os modelos principais
primary_device = self.devices[0]
self._allocations['LTX_PRIMARY'] = [primary_device]
print(f" - Tarefas LTX_PRIMARY (Transformer/TextEncoder) alocadas para: {primary_device}")
# Aloca as GPUs restantes para o pool de workers do VAE
worker_devices = self.devices[1:]
self._allocations['VAE_POOL'] = worker_devices
self._worker_indices['VAE_POOL'] = 0
print(f" - Tarefas VAE_POOL (VAE Decode) alocadas para: {worker_devices}")
print("[GPUManager] Alocação concluída.")
def get_device_for(self, task_key: str) -> str:
"""
Retorna o dispositivo principal alocado para uma chave de tarefa.
Use para modelos que residem em um único dispositivo.
Args:
task_key (str): A chave da tarefa (ex: 'LTX_PRIMARY').
Returns:
str: O nome do dispositivo (ex: 'cuda:0').
"""
if task_key not in self._allocations:
raise ValueError(f"Chave de tarefa '{task_key}' não registrada no GPUManager.")
# Retorna o primeiro (e geralmente único) dispositivo da lista de alocação
return self._allocations[task_key][0]
def get_next_worker_for(self, pool_key: str) -> str:
"""
Retorna o próximo dispositivo disponível de um pool de workers em rodízio.
Use para tarefas que podem ser paralelizadas em várias GPUs.
Args:
pool_key (str): A chave do pool de workers (ex: 'VAE_POOL').
Returns:
str: O nome do próximo dispositivo worker (ex: 'cuda:1').
"""
if pool_key not in self._allocations or pool_key not in self._worker_indices:
raise ValueError(f"Pool de workers '{pool_key}' não registrado no GPUManager.")
worker_pool = self._allocations[pool_key]
if not worker_pool:
raise RuntimeError(f"O pool de workers '{pool_key}' está vazio.")
# Obtém o índice atual, calcula o próximo e retorna o dispositivo
current_idx = self._worker_indices[pool_key]
device = worker_pool[current_idx]
# Avança o índice para o próximo worker (com wrap-around)
self._worker_indices[pool_key] = (current_idx + 1) % len(worker_pool)
return device
# --- Instância Singleton ---
# Importe esta instância em outros arquivos para garantir um único ponto de gerenciamento.
gpu_manager = GPUManager() |