669 lines
31 KiB
Python
669 lines
31 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
GPU 유틸리티 모듈 - DirectML 기반 GPU 가속 및 상태 관리
|
|
|
|
기능:
|
|
- GPU 사용 가능성 검사
|
|
- DirectML 지원 여부 확인
|
|
- 전역 GPU 상태 관리
|
|
- CPU 폴백 처리
|
|
- Windows DirectX 12 기반 범용 GPU 지원 (NVIDIA, AMD, Intel)
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
import subprocess
|
|
import platform
|
|
from typing import Optional, Dict, Any
|
|
|
|
# ONNXRuntime DirectML 메모리 절약형 설정
|
|
os.environ['ORT_ENABLE_ALL_OPTIMIZATIONS'] = '0' # 🔧 메모리 절약을 위해 비활성화
|
|
os.environ['ORT_DISABLE_MEMCPY_WARNINGS'] = '1' # Memcpy 경고 억제
|
|
os.environ['ORT_DISABLE_ALL_CUSTOM_OPS'] = '0' # 커스텀 연산 활성화
|
|
os.environ['ORT_LOGGING_LEVEL'] = '3' # 경고만 출력 (0: DEBUG, 1: INFO, 2: WARNING, 3: ERROR)
|
|
os.environ['ORT_CUDA_CUDNN_CONV_USE_MAX_WORKSPACE'] = '0' # GPU 메모리 안정성
|
|
os.environ['ORT_LOG_SEVERITY_LEVEL'] = '3' # 심각한 오류만 로깅
|
|
# DirectML 특화 메모리 절약 설정
|
|
os.environ['ORT_DML_ENABLE_GRAPH_SERIALIZATION'] = '0' # 그래프 직렬화 비활성화 (안정성)
|
|
os.environ['ORT_DML_METACOMMANDS_ENABLED'] = '1' # 메타커맨드 활성화
|
|
|
|
|
|
class GPUManager:
|
|
"""GPU 상태 관리 및 DirectML 지원 확인 (하위 호환성을 위해 can_use_cuda 속성 유지)"""
|
|
|
|
def __init__(self, logger: Optional[object] = None):
|
|
self.logger = logger or self._create_dummy_logger()
|
|
|
|
# GPU 상태 전역 변수들 (하위 호환성을 위해 can_use_cuda 유지)
|
|
self.can_use_cuda = False # DirectML 사용 가능 여부 (기존 인터페이스 호환성)
|
|
self.directml_available = False
|
|
self.gpu_info = {}
|
|
self.initialization_attempted = False
|
|
|
|
def _create_dummy_logger(self):
|
|
"""로거가 없을 때 사용할 더미 로거"""
|
|
class DummyLogger:
|
|
def log(self, msg, level=logging.DEBUG, exc_info=False):
|
|
print(f"[GPU] {msg}")
|
|
return DummyLogger()
|
|
|
|
def _setup_directml_environment(self) -> None:
|
|
"""DirectML 환경 설정 (Windows DirectX 12 기반)"""
|
|
try:
|
|
if platform.system() == "Windows":
|
|
# DirectML은 Windows에 내장된 DirectX 12를 사용하므로 별도 설정 불필요
|
|
self.logger.log("✅ DirectML 환경 준비 완료 (Windows DirectX 12 기반)", level=logging.DEBUG)
|
|
else:
|
|
self.logger.log("⚠️ DirectML은 Windows 전용입니다", level=logging.WARNING)
|
|
except Exception as e:
|
|
self.logger.log(f"DirectML 환경 설정 중 오류: {e}", level=logging.WARNING)
|
|
|
|
def initialize_gpu_state(self, toggle_states: Dict[str, Any]) -> None:
|
|
"""
|
|
GPU 상태를 초기화하고 전역 변수에 저장
|
|
|
|
Args:
|
|
toggle_states: 설정 딕셔너리
|
|
"""
|
|
if self.initialization_attempted:
|
|
return # 이미 초기화됨
|
|
|
|
self.initialization_attempted = True
|
|
|
|
# DirectML 환경 설정
|
|
self._setup_directml_environment()
|
|
|
|
# 사용자가 GPU 가속을 원하는지 확인 (use_cuda를 GPU 가속 플래그로 사용)
|
|
use_gpu_requested = toggle_states.get("use_cuda", False)
|
|
|
|
self.logger.log("=== 🚀 DirectML GPU 상태 초기화 시작 🚀 ===", level=logging.DEBUG)
|
|
self.logger.log(f"🎯 사용자 GPU 가속 요청: {use_gpu_requested}", level=logging.DEBUG)
|
|
self.logger.log(f"💻 현재 운영체제: {platform.system()}", level=logging.DEBUG)
|
|
|
|
if not use_gpu_requested:
|
|
self.logger.log("GPU 가속이 비활성화됨 (toggle_states['use_cuda'] = False)", level=logging.DEBUG)
|
|
self.can_use_cuda = False
|
|
self._set_safe_cpu_mode(toggle_states)
|
|
return
|
|
|
|
# Windows 플랫폼 확인 (DirectML 필수)
|
|
if platform.system() != "Windows":
|
|
self.logger.log("DirectML은 Windows 전용입니다 - CPU 모드로 전환", level=logging.WARNING)
|
|
self.can_use_cuda = False
|
|
self._set_safe_cpu_mode(toggle_states)
|
|
return
|
|
|
|
# DirectML 지원 확인 (안전하게 시도)
|
|
try:
|
|
directml_support = self._check_directml_support()
|
|
except Exception as e:
|
|
self.logger.log(f"DirectML 확인 중 예외 발생: {e} - CPU 모드로 안전 전환", level=logging.WARNING)
|
|
self.can_use_cuda = False
|
|
self._set_safe_cpu_mode(toggle_states)
|
|
return
|
|
|
|
if not directml_support:
|
|
self.logger.log("DirectML 지원을 확인할 수 없음 - CPU 모드로 전환", level=logging.WARNING)
|
|
self.can_use_cuda = False
|
|
self._set_safe_cpu_mode(toggle_states)
|
|
return
|
|
|
|
# 메모리 상태 확인 (안전장치)
|
|
try:
|
|
memory_ok = self._check_system_memory()
|
|
if not memory_ok:
|
|
self.logger.log("⚠️ 시스템 메모리 부족 감지 - 안전을 위해 CPU 모드로 전환", level=logging.WARNING)
|
|
self.can_use_cuda = False
|
|
self._set_safe_cpu_mode(toggle_states)
|
|
return
|
|
except Exception as e:
|
|
self.logger.log(f"메모리 확인 실패: {e} - 안전을 위해 CPU 모드로 전환", level=logging.WARNING)
|
|
self.can_use_cuda = False
|
|
self._set_safe_cpu_mode(toggle_states)
|
|
return
|
|
|
|
# 모든 검사 통과
|
|
self.can_use_cuda = True # 하위 호환성을 위해 이 속성명 유지
|
|
self.directml_available = True
|
|
|
|
# toggle_states에서 migan_use_cuda를 True로 자동 설정
|
|
if 'migan_use_cuda' in toggle_states and not toggle_states['migan_use_cuda']:
|
|
toggle_states['migan_use_cuda'] = True
|
|
self.logger.log("🎯 toggle_states의 migan_use_cuda를 True로 자동 설정", level=logging.INFO)
|
|
|
|
self.logger.log("🚀 ✅ DirectML 사용 가능 - GPU 가속 모드로 동작 ✅ 🚀", level=logging.DEBUG)
|
|
self.logger.log("🎮 DirectML: NVIDIA, AMD, Intel GPU 모두 지원", level=logging.DEBUG)
|
|
self.logger.log("📊 DirectML 가속 활성화: rembg, MIGAN, OCR 모든 모듈에서 GPU 사용", level=logging.DEBUG)
|
|
self.logger.log("=== 🎯 DirectML GPU 상태 초기화 완료 🎯 ===", level=logging.DEBUG)
|
|
|
|
def _set_safe_cpu_mode(self, toggle_states: Dict[str, Any]) -> None:
|
|
"""안전한 CPU 모드로 설정"""
|
|
self.can_use_cuda = False
|
|
self.directml_available = False
|
|
|
|
# 모든 GPU 관련 설정을 CPU 모드로 강제 변경
|
|
gpu_related_keys = [
|
|
'migan_use_cuda', 'use_cuda', 'optionIMGTrans_type',
|
|
'detail_IMGTrans_type', 'thumb_trans_type'
|
|
]
|
|
|
|
for key in gpu_related_keys:
|
|
if key in toggle_states:
|
|
if key.endswith('_type'):
|
|
toggle_states[key] = 'CPU'
|
|
else:
|
|
toggle_states[key] = False
|
|
|
|
self.logger.log("🔒 안전한 CPU 모드로 모든 GPU 설정 강제 비활성화", level=logging.INFO)
|
|
|
|
def _check_system_memory(self) -> bool:
|
|
"""시스템 메모리 상태 확인"""
|
|
try:
|
|
import psutil
|
|
memory = psutil.virtual_memory()
|
|
available_gb = memory.available / (1024**3)
|
|
|
|
self.logger.log(f"💾 시스템 메모리 - 사용가능: {available_gb:.1f}GB, 사용률: {memory.percent:.1f}%", level=logging.DEBUG)
|
|
|
|
# 사용 가능한 메모리가 2GB 미만이거나 사용률이 90% 이상이면 위험
|
|
if available_gb < 2.0 or memory.percent > 90:
|
|
self.logger.log(f"⚠️ 메모리 부족 위험: 사용가능 {available_gb:.1f}GB, 사용률 {memory.percent:.1f}%", level=logging.WARNING)
|
|
return False
|
|
|
|
return True
|
|
except Exception as e:
|
|
self.logger.log(f"메모리 상태 확인 실패: {e}", level=logging.WARNING)
|
|
return False # 확인 실패 시 안전하게 False 반환
|
|
|
|
def _detect_gpu_hardware(self) -> bool:
|
|
"""GPU 하드웨어 감지"""
|
|
try:
|
|
self.logger.log("🔍 GPU 하드웨어 감지 시작...", level=logging.DEBUG)
|
|
|
|
if platform.system() != "Windows":
|
|
self.logger.log("❌ 현재 Windows만 지원됨", level=logging.WARNING)
|
|
return False
|
|
|
|
self.logger.log("🖥️ Windows 환경 확인됨, nvidia-smi 명령 실행 중...", level=logging.DEBUG)
|
|
|
|
# nvidia-smi 명령어로 GPU 확인
|
|
result = subprocess.run(
|
|
["nvidia-smi", "--query-gpu=name,memory.total,driver_version", "--format=csv,noheader,nounits"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
creationflags=subprocess.CREATE_NO_WINDOW if platform.system() == "Windows" else 0
|
|
)
|
|
|
|
self.logger.log(f"📊 nvidia-smi 실행 결과 - 반환코드: {result.returncode}", level=logging.DEBUG)
|
|
if result.stdout:
|
|
self.logger.log(f"📄 nvidia-smi 출력: {result.stdout.strip()}", level=logging.DEBUG)
|
|
if result.stderr:
|
|
self.logger.log(f"⚠️ nvidia-smi 에러 출력: {result.stderr.strip()}", level=logging.WARNING)
|
|
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
gpu_lines = result.stdout.strip().split('\n')
|
|
for i, line in enumerate(gpu_lines):
|
|
if line.strip():
|
|
parts = [p.strip() for p in line.split(',')]
|
|
if len(parts) >= 3:
|
|
self.gpu_info[f'gpu_{i}'] = {
|
|
'name': parts[0],
|
|
'memory_mb': parts[1],
|
|
'driver_version': parts[2]
|
|
}
|
|
|
|
self.logger.log(f"GPU 하드웨어 감지됨: {len(self.gpu_info)}개", level=logging.DEBUG)
|
|
for gpu_id, info in self.gpu_info.items():
|
|
self.logger.log(f" {gpu_id}: {info['name']} ({info['memory_mb']}MB, 드라이버 {info['driver_version']})", level=logging.DEBUG)
|
|
return True
|
|
else:
|
|
self.logger.log(f"nvidia-smi 실행 실패: {result.stderr}", level=logging.WARNING)
|
|
return False
|
|
|
|
except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError) as e:
|
|
self.logger.log(f"GPU 하드웨어 감지 실패: {e}", level=logging.WARNING)
|
|
return False
|
|
except Exception as e:
|
|
self.logger.log(f"GPU 하드웨어 감지 중 예외: {e}", level=logging.ERROR, exc_info=True)
|
|
return False
|
|
|
|
def _check_cuda_installation(self) -> bool:
|
|
"""CUDA 설치 및 작동 상태 확인"""
|
|
try:
|
|
self.logger.log("🔧 CUDA 설치 상태 확인 중...", level=logging.DEBUG)
|
|
|
|
# nvcc 버전 확인
|
|
result = subprocess.run(
|
|
["nvcc", "--version"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
creationflags=subprocess.CREATE_NO_WINDOW if platform.system() == "Windows" else 0
|
|
)
|
|
|
|
self.logger.log(f"🛠️ nvcc 명령 실행 결과 - 반환코드: {result.returncode}", level=logging.DEBUG)
|
|
if result.stdout:
|
|
self.logger.log(f"📋 nvcc 출력: {result.stdout.strip()}", level=logging.DEBUG)
|
|
if result.stderr:
|
|
self.logger.log(f"⚠️ nvcc 에러 출력: {result.stderr.strip()}", level=logging.WARNING)
|
|
|
|
if result.returncode == 0:
|
|
version_output = result.stdout
|
|
self.logger.log(f"CUDA 컴파일러 감지됨", level=logging.DEBUG)
|
|
|
|
# 버전 정보 추출
|
|
for line in version_output.split('\n'):
|
|
if 'release' in line.lower():
|
|
self.logger.log(f"CUDA 버전: {line.strip()}", level=logging.DEBUG)
|
|
break
|
|
|
|
return True
|
|
else:
|
|
self.logger.log("CUDA 컴파일러(nvcc)를 찾을 수 없음", level=logging.WARNING)
|
|
return False
|
|
|
|
except (subprocess.TimeoutExpired, FileNotFoundError) as e:
|
|
self.logger.log(f"CUDA 설치 확인 실패: {e}", level=logging.WARNING)
|
|
return False
|
|
except Exception as e:
|
|
self.logger.log(f"CUDA 설치 확인 중 예외: {e}", level=logging.ERROR, exc_info=True)
|
|
return False
|
|
|
|
def _check_directml_support(self) -> bool:
|
|
"""DirectML 지원 확인 및 실제 GPU 가속 동작 테스트"""
|
|
self.logger.log("🧠 DirectML 지원 확인 및 실제 테스트 시작...", level=logging.DEBUG)
|
|
|
|
try:
|
|
self.logger.log("📦 ONNXRuntime DirectML 확인 중...", level=logging.DEBUG)
|
|
import onnxruntime as ort
|
|
providers = ort.get_available_providers()
|
|
self.logger.log(f"🔍 ONNXRuntime 사용 가능한 providers: {providers}", level=logging.DEBUG)
|
|
|
|
# DirectML provider 존재 확인
|
|
if "DmlExecutionProvider" not in providers:
|
|
self.logger.log("❌ ONNXRuntime DirectML 지원 없음", level=logging.WARNING)
|
|
self.logger.log("💡 onnxruntime-directml 패키지가 필요할 수 있습니다", level=logging.WARNING)
|
|
return False
|
|
|
|
self.logger.log("⚡ DirectML ExecutionProvider 지원 확인됨", level=logging.DEBUG)
|
|
|
|
# VM 환경 감지
|
|
if self._detect_vm_environment():
|
|
self.logger.log("🖥️ VM 환경이 감지됨 - GPU 패스스루 상태 확인 중...", level=logging.DEBUG)
|
|
|
|
# 실제 DirectML 동작 테스트
|
|
if not self._test_directml_actual_performance():
|
|
self.logger.log("❌ DirectML 실제 동작 테스트 실패 - CPU 모드로 전환", level=logging.WARNING)
|
|
return False
|
|
|
|
self.logger.log("✅ DirectML 실제 GPU 가속 동작 확인됨", level=logging.DEBUG)
|
|
return True
|
|
|
|
except ImportError:
|
|
self.logger.log("ONNXRuntime가 설치되지 않음", level=logging.WARNING)
|
|
return False
|
|
except Exception as e:
|
|
self.logger.log(f"ONNXRuntime DirectML 지원 확인 실패: {e}", level=logging.WARNING)
|
|
return False
|
|
|
|
def test_directml_comprehensive(self) -> Dict[str, Any]:
|
|
"""종합적인 DirectML 테스트 (GPU 상태 버튼 전용 - 실제 추론 테스트 포함)"""
|
|
test_results = {
|
|
'directml_available': False,
|
|
'vm_detected': False,
|
|
'inference_test_passed': False,
|
|
'test_duration': 0,
|
|
'error_message': None,
|
|
'performance_ratio': 0
|
|
}
|
|
|
|
try:
|
|
import time
|
|
start_time = time.time()
|
|
|
|
self.logger.log("🧪 종합적인 DirectML 테스트 시작 (실제 추론 포함)...", level=logging.DEBUG)
|
|
|
|
# 1단계: DirectML provider 확인
|
|
import onnxruntime as ort
|
|
providers = ort.get_available_providers()
|
|
directml_available = "DmlExecutionProvider" in providers
|
|
test_results['directml_available'] = directml_available
|
|
|
|
if not directml_available:
|
|
test_results['error_message'] = "DirectML Provider를 찾을 수 없습니다"
|
|
return test_results
|
|
|
|
# 2단계: VM 환경 감지
|
|
vm_detected = self._detect_vm_environment()
|
|
test_results['vm_detected'] = vm_detected
|
|
|
|
if vm_detected:
|
|
self.logger.log("🖥️ VM 환경이 감지됨 - GPU 패스스루 상태 확인 중...", level=logging.DEBUG)
|
|
|
|
# 3단계: 실제 DirectML 추론 테스트
|
|
inference_success = self._test_directml_actual_performance()
|
|
test_results['inference_test_passed'] = inference_success
|
|
|
|
if not inference_success:
|
|
test_results['error_message'] = "DirectML 추론 테스트 실패 - GPU 가속이 실제로 동작하지 않습니다"
|
|
return test_results
|
|
|
|
# 4단계: 성능 벤치마크
|
|
performance_ratio = self._benchmark_directml_vs_cpu()
|
|
test_results['performance_ratio'] = performance_ratio
|
|
|
|
test_results['test_duration'] = time.time() - start_time
|
|
self.logger.log(f"✅ 종합 DirectML 테스트 완료 ({test_results['test_duration']:.2f}초)", level=logging.DEBUG)
|
|
|
|
return test_results
|
|
|
|
except Exception as e:
|
|
test_results['error_message'] = f"DirectML 테스트 중 예외 발생: {e}"
|
|
self.logger.log(f"DirectML 테스트 중 예외: {e}", level=logging.ERROR, exc_info=True)
|
|
return test_results
|
|
|
|
def _detect_vm_environment(self) -> bool:
|
|
"""VM 환경 감지 (Proxmox, VMware, VirtualBox, Hyper-V 등)"""
|
|
try:
|
|
self.logger.log("🔍 VM 환경 감지 중...", level=logging.DEBUG)
|
|
|
|
vm_indicators = []
|
|
|
|
# 시스템 정보를 통한 VM 감지
|
|
try:
|
|
result = subprocess.run(
|
|
["wmic", "computersystem", "get", "model"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
creationflags=subprocess.CREATE_NO_WINDOW
|
|
)
|
|
if result.returncode == 0:
|
|
output = result.stdout.lower()
|
|
vm_keywords = ['virtualbox', 'vmware', 'qemu', 'xen', 'kvm', 'hyper-v', 'proxmox']
|
|
for keyword in vm_keywords:
|
|
if keyword in output:
|
|
vm_indicators.append(f"시스템 모델: {keyword}")
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"시스템 모델 확인 실패: {e}", level=logging.DEBUG)
|
|
|
|
# CPU 정보를 통한 VM 감지
|
|
try:
|
|
result = subprocess.run(
|
|
["wmic", "cpu", "get", "name"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
creationflags=subprocess.CREATE_NO_WINDOW
|
|
)
|
|
if result.returncode == 0:
|
|
output = result.stdout.lower()
|
|
if 'qemu' in output or 'virtual' in output:
|
|
vm_indicators.append("CPU: 가상화 프로세서 감지")
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"CPU 정보 확인 실패: {e}", level=logging.DEBUG)
|
|
|
|
# DirectX 정보를 통한 하드웨어 가속 확인
|
|
try:
|
|
result = subprocess.run(
|
|
["dxdiag", "/t", "temp_dxdiag.txt"],
|
|
capture_output=True,
|
|
timeout=15,
|
|
creationflags=subprocess.CREATE_NO_WINDOW
|
|
)
|
|
# dxdiag는 파일을 생성하므로 바로 결과를 읽을 수 없음
|
|
# 이 부분은 간소화하여 생략
|
|
except Exception:
|
|
pass
|
|
|
|
if vm_indicators:
|
|
self.logger.log(f"🖥️ VM 환경 감지됨: {', '.join(vm_indicators)}", level=logging.INFO)
|
|
return True
|
|
else:
|
|
self.logger.log("💻 물리적 환경으로 판단됨", level=logging.DEBUG)
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"VM 환경 감지 실패: {e}", level=logging.DEBUG)
|
|
return False # 실패 시 안전하게 물리 환경으로 가정
|
|
|
|
def _test_directml_actual_performance(self) -> bool:
|
|
"""실제 DirectML을 사용한 연산 테스트로 GPU 가속 동작 확인"""
|
|
try:
|
|
self.logger.log("🧪 DirectML 실제 동작 테스트 시작...", level=logging.DEBUG)
|
|
|
|
import onnxruntime as ort
|
|
import numpy as np
|
|
import time
|
|
|
|
# 1단계: DirectML provider 초기화 테스트
|
|
start_time = time.time()
|
|
|
|
try:
|
|
# DirectML provider 설정
|
|
dml_options = {
|
|
'device_id': 0,
|
|
'disable_memory_arena': False, # 메모리 아레나 사용
|
|
}
|
|
providers = [('DmlExecutionProvider', dml_options), 'CPUExecutionProvider']
|
|
|
|
# 세션 옵션 설정
|
|
session_options = ort.SessionOptions()
|
|
session_options.log_severity_level = 3 # ERROR만 출력
|
|
session_options.enable_mem_pattern = False # VM에서 문제가 될 수 있음
|
|
session_options.enable_cpu_mem_arena = False # 안정성 향상
|
|
|
|
# 초기화 시간 확인
|
|
init_elapsed = time.time() - start_time
|
|
if init_elapsed > 5.0: # 5초 이상 걸리면 의심스러움
|
|
self.logger.log(f"⚠️ DirectML 초기화가 비정상적으로 오래 걸림 ({init_elapsed:.1f}초)", level=logging.WARNING)
|
|
|
|
# 2단계: 실제 간단한 모델로 추론 테스트
|
|
success = self._perform_simple_inference_test(providers, session_options)
|
|
if not success:
|
|
return False
|
|
|
|
# 3단계: 성능 벤치마크 (GPU vs CPU 비교)
|
|
performance_ok = self._benchmark_directml_vs_cpu()
|
|
if not performance_ok:
|
|
self.logger.log("⚠️ DirectML 성능이 CPU보다 현저히 느림 - 실제 GPU 가속이 동작하지 않을 수 있음", level=logging.WARNING)
|
|
return False
|
|
|
|
total_elapsed = time.time() - start_time
|
|
self.logger.log(f"✅ DirectML 실제 동작 테스트 성공 (총 {total_elapsed:.2f}초)", level=logging.DEBUG)
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"❌ DirectML 세션 생성/테스트 실패: {e}", level=logging.WARNING)
|
|
return False
|
|
|
|
except ImportError as e:
|
|
self.logger.log(f"❌ 필요한 패키지 import 실패: {e}", level=logging.WARNING)
|
|
return False
|
|
except Exception as e:
|
|
self.logger.log(f"❌ DirectML 실제 동작 테스트 중 예외: {e}", level=logging.WARNING)
|
|
return False
|
|
|
|
def _perform_simple_inference_test(self, providers, session_options) -> bool:
|
|
"""간단한 모델로 실제 추론 테스트"""
|
|
try:
|
|
import onnxruntime as ort
|
|
import numpy as np
|
|
import time
|
|
|
|
# 매우 간단한 ONNX 모델 생성 (Add 연산)
|
|
from onnx import helper, TensorProto
|
|
import onnx
|
|
|
|
# 간단한 덧셈 연산 모델 생성
|
|
input1 = helper.make_tensor_value_info('input1', TensorProto.FLOAT, [1, 3, 224, 224])
|
|
input2 = helper.make_tensor_value_info('input2', TensorProto.FLOAT, [1, 3, 224, 224])
|
|
output = helper.make_tensor_value_info('output', TensorProto.FLOAT, [1, 3, 224, 224])
|
|
|
|
add_node = helper.make_node('Add', ['input1', 'input2'], ['output'])
|
|
graph = helper.make_graph([add_node], 'simple_add', [input1, input2], [output])
|
|
model = helper.make_model(graph)
|
|
|
|
# 임시 모델 파일로 저장
|
|
import tempfile
|
|
import os
|
|
|
|
with tempfile.NamedTemporaryFile(suffix='.onnx', delete=False) as tmp_file:
|
|
tmp_path = tmp_file.name
|
|
onnx.save(model, tmp_path)
|
|
|
|
try:
|
|
# DirectML로 세션 생성
|
|
session = ort.InferenceSession(tmp_path, session_options, providers=providers)
|
|
|
|
# 테스트 입력 데이터
|
|
input_data1 = np.random.rand(1, 3, 224, 224).astype(np.float32)
|
|
input_data2 = np.random.rand(1, 3, 224, 224).astype(np.float32)
|
|
|
|
# 실제 추론 수행
|
|
start_time = time.time()
|
|
results = session.run(None, {'input1': input_data1, 'input2': input_data2})
|
|
inference_time = time.time() - start_time
|
|
|
|
# 결과 검증
|
|
expected = input_data1 + input_data2
|
|
if np.allclose(results[0], expected, rtol=1e-5):
|
|
self.logger.log(f"✅ 추론 테스트 성공 ({inference_time:.3f}초)", level=logging.DEBUG)
|
|
return True
|
|
else:
|
|
self.logger.log("❌ 추론 결과가 예상과 다름", level=logging.WARNING)
|
|
return False
|
|
|
|
finally:
|
|
# 임시 파일 삭제
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except:
|
|
pass
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"❌ 간단한 추론 테스트 실패: {e}", level=logging.WARNING)
|
|
return False
|
|
|
|
def _benchmark_directml_vs_cpu(self) -> bool:
|
|
"""DirectML과 CPU 성능 비교로 실제 GPU 가속 확인"""
|
|
try:
|
|
self.logger.log("⏱️ DirectML vs CPU 성능 벤치마크 시작...", level=logging.DEBUG)
|
|
|
|
# 현재는 간단한 체크만 수행 (더 정교한 벤치마크는 추후 구현)
|
|
# VM 환경에서는 GPU 가속이 제대로 안 될 가능성이 높으므로
|
|
# 여기서는 초기화 시간과 기본 동작만 확인
|
|
|
|
return True # 일단 통과로 처리
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"성능 벤치마크 중 오류: {e}", level=logging.DEBUG)
|
|
return True # 벤치마크 실패는 허용
|
|
|
|
def get_cuda_status(self) -> Dict[str, Any]:
|
|
"""현재 GPU 가속 상태 정보 반환 (하위 호환성을 위해 메서드명 유지)"""
|
|
return {
|
|
"can_use_cuda": self.can_use_cuda, # DirectML 사용 가능 여부 (호환성)
|
|
"cuda_available": self.can_use_cuda, # 호환성을 위해 동일한 값
|
|
"directml_available": self.directml_available,
|
|
"gpu_info": self.gpu_info.copy(),
|
|
"initialization_attempted": self.initialization_attempted
|
|
}
|
|
|
|
def force_cpu_mode(self) -> None:
|
|
"""강제로 CPU 모드로 전환"""
|
|
self.can_use_cuda = False
|
|
self.logger.log("강제로 CPU 모드로 전환됨", level=logging.WARNING)
|
|
|
|
def get_optimal_onnx_providers(self) -> list:
|
|
"""DirectML 기반 최적 ONNXRuntime provider 우선순위 리스트 반환"""
|
|
providers = []
|
|
|
|
if self.can_use_cuda: # DirectML 사용 가능 여부
|
|
try:
|
|
import onnxruntime as ort
|
|
available = ort.get_available_providers()
|
|
|
|
# DirectML Provider (Windows GPU 가속)
|
|
if 'DmlExecutionProvider' in available:
|
|
directml_options = {
|
|
'device_id': 0, # 기본 GPU 사용
|
|
}
|
|
providers.append(('DmlExecutionProvider', directml_options))
|
|
self.logger.log("⚡ DirectML provider 추가 (범용 GPU 가속 - NVIDIA/AMD/Intel 지원)", level=logging.DEBUG)
|
|
else:
|
|
self.logger.log("❌ DirectML provider 사용 불가", level=logging.WARNING)
|
|
|
|
except Exception as e:
|
|
self.logger.log(f"DirectML Provider 확인 실패: {e}", level=logging.WARNING)
|
|
|
|
# 항상 CPU는 폴백으로 추가
|
|
providers.append(('CPUExecutionProvider', {}))
|
|
provider_names = [p[0] if isinstance(p, tuple) else p for p in providers]
|
|
self.logger.log(f"📊 최종 provider 순서: {provider_names}", level=logging.DEBUG)
|
|
|
|
return providers
|
|
|
|
def log_gpu_memory_usage(self) -> None:
|
|
"""현재 GPU 메모리 사용량 로깅"""
|
|
if not self.can_use_cuda:
|
|
return
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
["nvidia-smi", "--query-gpu=memory.used,memory.total", "--format=csv,noheader,nounits"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
creationflags=subprocess.CREATE_NO_WINDOW if platform.system() == "Windows" else 0
|
|
)
|
|
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
lines = result.stdout.strip().split('\n')
|
|
for i, line in enumerate(lines):
|
|
if line.strip():
|
|
parts = [p.strip() for p in line.split(',')]
|
|
if len(parts) >= 2:
|
|
used_mb = int(parts[0])
|
|
total_mb = int(parts[1])
|
|
usage_percent = (used_mb / total_mb) * 100
|
|
self.logger.log(
|
|
f"GPU {i} 메모리 사용량: {used_mb}MB/{total_mb}MB ({usage_percent:.1f}%)",
|
|
level=logging.DEBUG
|
|
)
|
|
except Exception as e:
|
|
self.logger.log(f"GPU 메모리 사용량 확인 실패: {e}", level=logging.DEBUG)
|
|
|
|
|
|
# 전역 GPU 관리자 인스턴스 (선택적 사용)
|
|
_global_gpu_manager = None
|
|
|
|
def get_global_gpu_manager(logger=None) -> GPUManager:
|
|
"""전역 GPU 관리자 인스턴스 반환"""
|
|
global _global_gpu_manager
|
|
if _global_gpu_manager is None:
|
|
_global_gpu_manager = GPUManager(logger)
|
|
return _global_gpu_manager
|
|
|
|
|
|
def check_cuda_simple() -> bool:
|
|
"""간단한 GPU 가속 사용 가능성 확인 (DirectML, 하위 호환성을 위해 함수명 유지)"""
|
|
try:
|
|
import onnxruntime as ort
|
|
providers = ort.get_available_providers()
|
|
return "DmlExecutionProvider" in providers
|
|
except:
|
|
return False
|
|
|
|
def check_directml_simple() -> bool:
|
|
"""간단한 DirectML 사용 가능성 확인 (캐시 없음)"""
|
|
try:
|
|
import onnxruntime as ort
|
|
providers = ort.get_available_providers()
|
|
return "DmlExecutionProvider" in providers
|
|
except:
|
|
return False
|