#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 동적 세션풀 관리 기반 Worker 테스트 스크립트 - GPU 메모리 자동 감지 및 최적 세션풀 구성 - LAMA:MIGAN 비율 커스터마이징 - 실시간 메모리 모니터링 """ import os import sys import asyncio import time import pathlib import logging from typing import Dict, Any, Optional, Tuple # Add parent directory to sys.path to import worker modules CURRENT_DIR = pathlib.Path(__file__).parent.absolute() ROOT_DIR = CURRENT_DIR.parent sys.path.insert(0, str(ROOT_DIR)) from worker.dynamic_session_pool_manager import ( get_optimal_session_config, session_pool_manager, print_config_summary, SessionPoolConfig ) # 로깅 설정 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def test_gpu_detection(): """GPU 감지 테스트""" print("🔍 GPU 자동 감지 테스트...") try: memory_info = session_pool_manager.detect_gpu_info() print(f"✅ GPU 감지 성공!") print(f" 이름: {memory_info.gpu_name}") print(f" 등급: {memory_info.gpu_tier.value}") print(f" VRAM: {memory_info.total_vram_mb}MB (사용가능: {memory_info.available_vram_mb}MB)") print(f" CUDA: {memory_info.cuda_capability}") print(f" FP16: {memory_info.supports_fp16}") print(f" TensorCores: {memory_info.supports_tensor_cores}") return memory_info except Exception as e: print(f"❌ GPU 감지 실패: {e}") return None def test_optimal_configs(): """다양한 비율로 최적 구성 테스트""" print("\n🎯 최적 세션풀 구성 테스트...") test_configs = [ { "name": "기본 구성 (GPU별 최적화)", "lama_migan_ratio": None, "worker_session_ratio": None }, { "name": "4:6 비율 (RTX 4080 권장)", "lama_migan_ratio": (4, 6), "worker_session_ratio": 3.0 }, { "name": "3:7 비율 (RTX 3090 권장)", "lama_migan_ratio": (3, 7), "worker_session_ratio": 3.0 }, { "name": "5:5 균형 구성", "lama_migan_ratio": (5, 5), "worker_session_ratio": 3.5 }, { "name": "보수적 구성 (LAMA 중심)", "lama_migan_ratio": (7, 3), "worker_session_ratio": 4.0 } ] results = [] for test_config in test_configs: try: print(f"\n📋 {test_config['name']} 테스트...") config = get_optimal_session_config( lama_migan_ratio=test_config['lama_migan_ratio'], worker_session_ratio=test_config['worker_session_ratio'] ) results.append((test_config['name'], config)) print(f" ✅ 성공!") print(f" 세션풀: MIGAN {config.migan_sessions}개, LAMA {config.lama_sessions}개, OCR {config.ocr_sessions}개") print(f" 워커: {config.workers}개") print(f" 비율: {config.lama_migan_ratio}, 워커:세션 = {config.worker_session_ratio:.1f}:1") print(f" 메모리: {config.total_vram_usage_mb}MB, 여유: {config.safety_margin_mb}MB") print(f" 예상 클라이언트: {config.expected_concurrent_clients}명") except Exception as e: print(f" ❌ 실패: {e}") return results def test_memory_monitoring(): """메모리 모니터링 테스트""" print("\n📊 메모리 모니터링 테스트...") try: # 현재 메모리 상태 memory_status = session_pool_manager.get_current_memory_status() if "error" in memory_status: print(f"❌ 메모리 상태 확인 실패: {memory_status['error']}") return print(f"✅ 현재 메모리 상태:") print(f" 할당된 메모리: {memory_status['allocated_mb']}MB") print(f" 예약된 메모리: {memory_status['reserved_mb']}MB") print(f" 전체 메모리: {memory_status['total_mb']}MB") print(f" 사용 가능: {memory_status['free_mb']}MB") print(f" 사용률: {memory_status['utilization_pct']}%") # 모니터링 및 조정 테스트 adjusted = session_pool_manager.monitor_and_adjust() if adjusted: print("⚠️ 메모리 조정이 수행되었습니다") else: print("✅ 메모리 상태 정상") except Exception as e: print(f"❌ 메모리 모니터링 실패: {e}") def generate_production_config(lama_migan_ratio: Tuple[int, int] = (4, 6), worker_session_ratio: float = 3.0) -> Dict[str, Any]: """프로덕션용 설정 생성""" try: config = get_optimal_session_config( lama_migan_ratio=lama_migan_ratio, worker_session_ratio=worker_session_ratio ) memory_info = session_pool_manager._memory_info production_config = { # GPU 정보 "gpu_info": { "name": memory_info.gpu_name, "tier": memory_info.gpu_tier.value, "total_vram_mb": memory_info.total_vram_mb, "available_vram_mb": memory_info.available_vram_mb, "supports_fp16": memory_info.supports_fp16, "supports_tensor_cores": memory_info.supports_tensor_cores, }, # 세션풀 설정 "session_pools": { "migan": { "count": config.migan_sessions, "memory_per_session_mb": 1200, "total_memory_mb": config.migan_sessions * 1200 }, "lama": { "count": config.lama_sessions, "memory_per_session_mb": 500, "total_memory_mb": config.lama_sessions * 500 }, "ocr": { "count": config.ocr_sessions, "memory_per_session_mb": 400, "total_memory_mb": config.ocr_sessions * 400 } }, # 워커 설정 "workers": { "count": config.workers, "worker_session_ratio": config.worker_session_ratio }, # 성능 예측 "performance": { "lama_migan_ratio": config.lama_migan_ratio, "expected_concurrent_clients": config.expected_concurrent_clients, "estimated_response_time_seconds": "1.2-2.5", "peak_throughput_req_per_sec": "20-35" }, # 메모리 사용량 "memory_usage": { "total_usage_mb": config.total_vram_usage_mb, "safety_margin_mb": config.safety_margin_mb, "utilization_percent": round(config.total_vram_usage_mb / memory_info.total_vram_mb * 100, 1) }, # 라우팅 전략 "routing_strategy": { "simple_images": "migan", "complex_images": "lama", "adjacent_textbox_threshold": 3, "text_density_threshold": 0.3 } } return production_config except Exception as e: logger.error(f"프로덕션 설정 생성 실패: {e}") return {} def save_production_config(config: Dict[str, Any], filename: str = "production_config.json"): """프로덕션 설정을 JSON 파일로 저장""" import json try: config_path = CURRENT_DIR / filename with open(config_path, 'w', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False) print(f"✅ 프로덕션 설정 저장 완료: {config_path}") except Exception as e: print(f"❌ 설정 저장 실패: {e}") def main(): """메인 테스트 실행""" print("🚀 동적 세션풀 관리 테스트 시작") print("=" * 50) # 1. GPU 감지 memory_info = test_gpu_detection() if not memory_info: print("❌ GPU 감지 실패로 테스트 중단") return # 2. 다양한 구성 테스트 configs = test_optimal_configs() # 3. 메모리 모니터링 test_memory_monitoring() # 4. 구성 요약 출력 print("\n📋 최종 구성 요약:") print_config_summary() # 5. 프로덕션 설정 생성 및 저장 print("\n🏭 프로덕션 설정 생성...") # 4:6 비율로 프로덕션 설정 prod_config_46 = generate_production_config(lama_migan_ratio=(4, 6), worker_session_ratio=3.0) if prod_config_46: save_production_config(prod_config_46, "production_config_4_6.json") # 3:7 비율로 프로덕션 설정 prod_config_37 = generate_production_config(lama_migan_ratio=(3, 7), worker_session_ratio=3.0) if prod_config_37: save_production_config(prod_config_37, "production_config_3_7.json") print("\n🎯 권장사항:") if memory_info.total_vram_mb >= 15000: # 15GB+ print(" 💎 대용량 VRAM 감지 - 3:7 비율 (MIGAN 중심) 권장") print(" 📈 예상 성능: 60-70명 동시 처리, 0.8-1.5초 응답시간") elif memory_info.total_vram_mb >= 12000: # 12GB+ print(" ⚡ 중간급 VRAM 감지 - 4:6 비율 (균형) 권장") print(" 📈 예상 성능: 40-50명 동시 처리, 1.2-2.5초 응답시간") else: print(" 💼 보급형 VRAM 감지 - 5:5 또는 6:4 비율 (LAMA 중심) 권장") print(" 📈 예상 성능: 20-30명 동시 처리, 2.0-3.5초 응답시간") print("\n✅ 동적 세션풀 관리 테스트 완료!") if __name__ == "__main__": main()