TensorRT Optimization Disaster Recovery Plan

TensorRT Optimization Disaster Recovery Plan คืออะไร

TensorRT เป็น NVIDIA SDK สำหรับ optimize deep learning inference บน NVIDIA GPUs ช่วยเพิ่มความเร็ว inference 2-6x เทียบกับ frameworks ทั่วไป ด้วยเทคนิค layer fusion, kernel auto-tuning, precision calibration (FP16/INT8) และ dynamic tensor memory Disaster Recovery (DR) Plan คือแผนกู้คืนระบบเมื่อเกิดเหตุฉุกเฉิน การรวมสองแนวคิดนี้ช่วยให้ ML inference infrastructure มี resilience สูง สามารถ failover, restore optimized models และกลับมาให้บริการได้เร็วหลังเกิด outage

TensorRT Architecture

# tensorrt_arch.py — TensorRT architecture
import json

class TensorRTArch:
    OPTIMIZATION = {
        "layer_fusion": {
            "name": "Layer Fusion",
            "description": "รวม layers หลายๆ อันเป็น kernel เดียว ลด memory transfers",
            "example": "Conv + BatchNorm + ReLU → 1 fused kernel",
        },
        "precision": {
            "name": "Precision Calibration",
            "description": "แปลง FP32 → FP16/INT8 เพิ่มความเร็ว 2-4x ลด memory 50-75%",
            "modes": ["FP32 (full precision)", "FP16 (half precision)", "INT8 (quantized)", "TF32 (Ampere+)"],
        },
        "kernel_tuning": {
            "name": "Kernel Auto-Tuning",
            "description": "เลือก CUDA kernel ที่เร็วที่สุดสำหรับแต่ละ layer บน GPU ที่ใช้",
            "note": "Engine เป็น GPU-specific — ต้อง build ใหม่ถ้าเปลี่ยน GPU",
        },
        "dynamic_memory": {
            "name": "Dynamic Tensor Memory",
            "description": "จัดสรร GPU memory อย่างมีประสิทธิภาพ reuse memory ระหว่าง layers",
        },
    }

    WORKFLOW = """
    TensorRT Optimization Workflow:
    
    1. Train Model (PyTorch/TensorFlow)
         ↓
    2. Export to ONNX
         ↓
    3. TensorRT Builder — optimize + calibrate
         ↓
    4. TensorRT Engine (.engine/.plan file)
         ↓
    5. Deploy Engine — TensorRT Runtime / Triton Inference Server
    """

    def show_optimizations(self):
        print("=== TensorRT Optimizations ===\n")
        for key, opt in self.OPTIMIZATION.items():
            print(f"[{opt['name']}]")
            print(f"  {opt['description']}")
            print()

    def show_workflow(self):
        print("=== Workflow ===")
        print(self.WORKFLOW)

arch = TensorRTArch()
arch.show_optimizations()
arch.show_workflow()

TensorRT Implementation

# implementation.py — TensorRT optimization code
import json

class TensorRTImpl:
    PYTHON_CODE = """
# tensorrt_optimize.py — Build TensorRT engine from ONNX
import tensorrt as trt
import numpy as np

class TensorRTBuilder:
    def __init__(self, onnx_path, precision="fp16"):
        self.onnx_path = onnx_path
        self.precision = precision
        self.logger = trt.Logger(trt.Logger.WARNING)
    
    def build_engine(self, output_path, max_batch=8, workspace_gb=4):
        builder = trt.Builder(self.logger)
        network = builder.create_network(
            1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
        )
        parser = trt.OnnxParser(network, self.logger)
        
        # Parse ONNX model
        with open(self.onnx_path, "rb") as f:
            if not parser.parse(f.read()):
                for i in range(parser.num_errors):
                    print(f"ONNX Parse Error: {parser.get_error(i)}")
                return None
        
        # Build config
        config = builder.create_builder_config()
        config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, workspace_gb << 30)
        
        # Set precision
        if self.precision == "fp16":
            config.set_flag(trt.BuilderFlag.FP16)
        elif self.precision == "int8":
            config.set_flag(trt.BuilderFlag.INT8)
            # Need calibrator for INT8
        
        # Dynamic batch size
        profile = builder.create_optimization_profile()
        input_shape = network.get_input(0).shape
        profile.set_shape(
            network.get_input(0).name,
            min=(1, *input_shape[1:]),
            opt=(max_batch // 2, *input_shape[1:]),
            max=(max_batch, *input_shape[1:]),
        )
        config.add_optimization_profile(profile)
        
        # Build engine
        engine = builder.build_serialized_network(network, config)
        
        # Save engine
        with open(output_path, "wb") as f:
            f.write(engine)
        
        print(f"Engine saved: {output_path}")
        return output_path

# Usage
builder = TensorRTBuilder("model.onnx", precision="fp16")
builder.build_engine("model_fp16.engine", max_batch=16)
"""

    INFERENCE = """
# tensorrt_inference.py — Run inference with TensorRT
import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np

class TensorRTInference:
    def __init__(self, engine_path):
        self.logger = trt.Logger(trt.Logger.WARNING)
        with open(engine_path, "rb") as f:
            runtime = trt.Runtime(self.logger)
            self.engine = runtime.deserialize_cuda_engine(f.read())
        self.context = self.engine.create_execution_context()
    
    def infer(self, input_data):
        # Allocate GPU memory
        d_input = cuda.mem_alloc(input_data.nbytes)
        output_shape = self.engine.get_binding_shape(1)
        output = np.empty(output_shape, dtype=np.float32)
        d_output = cuda.mem_alloc(output.nbytes)
        
        # Transfer input to GPU
        cuda.memcpy_htod(d_input, input_data)
        
        # Run inference
        self.context.execute_v2([int(d_input), int(d_output)])
        
        # Transfer output back
        cuda.memcpy_dtoh(output, d_output)
        return output

# Usage
engine = TensorRTInference("model_fp16.engine")
input_data = np.random.randn(1, 3, 224, 224).astype(np.float32)
result = engine.infer(input_data)
print(f"Output shape: {result.shape}")
"""

    def show_build(self):
        print("=== Build TensorRT Engine ===")
        print(self.PYTHON_CODE[:600])

    def show_inference(self):
        print(f"\n=== Inference ===")
        print(self.INFERENCE[:500])

impl = TensorRTImpl()
impl.show_build()
impl.show_inference()

Disaster Recovery Plan

# dr_plan.py — DR plan for TensorRT inference
import json

class DRPlan:
    RISKS = {
        "gpu_failure": {
            "name": "GPU Hardware Failure",
            "impact": "Inference service down",
            "rto": "< 30 minutes",
            "strategy": "Redundant GPU nodes + auto-failover",
        },
        "engine_corruption": {
            "name": "TensorRT Engine File Corruption",
            "impact": "Cannot load model for inference",
            "rto": "< 15 minutes (restore from backup) or 1-4 hours (rebuild)",
            "strategy": "Engine file backup in S3 + ONNX source backup",
        },
        "model_regression": {
            "name": "Model Quality Regression (Bad Deployment)",
            "impact": "Incorrect predictions in production",
            "rto": "< 5 minutes (rollback)",
            "strategy": "Blue-green deployment + canary + auto-rollback",
        },
        "region_outage": {
            "name": "Cloud Region Outage",
            "impact": "Entire inference service unavailable",
            "rto": "< 1 hour",
            "strategy": "Multi-region deployment with DNS failover",
        },
    }

    BACKUP_STRATEGY = """
# backup_strategy.py — Model backup strategy
import boto3
import hashlib
from datetime import datetime

class ModelBackup:
    def __init__(self, bucket="model-backups"):
        self.s3 = boto3.client("s3")
        self.bucket = bucket
    
    def backup_engine(self, engine_path, model_name, version):
        key = f"engines/{model_name}/v{version}/{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.engine"
        
        # Calculate checksum
        with open(engine_path, "rb") as f:
            checksum = hashlib.sha256(f.read()).hexdigest()
        
        self.s3.upload_file(
            engine_path, self.bucket, key,
            ExtraArgs={"Metadata": {"checksum": checksum, "version": str(version)}},
        )
        print(f"Backed up: s3://{self.bucket}/{key}")
        return key
    
    def backup_onnx(self, onnx_path, model_name, version):
        key = f"onnx/{model_name}/v{version}/model.onnx"
        self.s3.upload_file(onnx_path, self.bucket, key)
        return key
    
    def restore_engine(self, model_name, version, output_path):
        prefix = f"engines/{model_name}/v{version}/"
        objects = self.s3.list_objects_v2(Bucket=self.bucket, Prefix=prefix)
        latest = sorted(objects.get("Contents", []), key=lambda x: x["LastModified"])[-1]
        self.s3.download_file(self.bucket, latest["Key"], output_path)
        print(f"Restored: {latest['Key']} → {output_path}")
    
    def list_versions(self, model_name):
        prefix = f"engines/{model_name}/"
        objects = self.s3.list_objects_v2(Bucket=self.bucket, Prefix=prefix)
        versions = set()
        for obj in objects.get("Contents", []):
            parts = obj["Key"].split("/")
            if len(parts) >= 3:
                versions.add(parts[2])
        return sorted(versions)

backup = ModelBackup()
backup.backup_engine("model_fp16.engine", "resnet50", 3)
backup.backup_onnx("model.onnx", "resnet50", 3)
"""

    def show_risks(self):
        print("=== DR Risk Assessment ===\n")
        for key, risk in self.RISKS.items():
            print(f"[{risk['name']}]")
            print(f"  Impact: {risk['impact']}")
            print(f"  RTO: {risk['rto']}")
            print(f"  Strategy: {risk['strategy']}")
            print()

    def show_backup(self):
        print("=== Backup Strategy ===")
        print(self.BACKUP_STRATEGY[:500])

dr = DRPlan()
dr.show_risks()
dr.show_backup()

Triton Inference Server HA

# triton_ha.py — Triton Inference Server high availability
import json
import random

class TritonHA:
    K8S_DEPLOYMENT = """
# triton-deployment.yaml — Kubernetes HA deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: triton-inference
  namespace: ml-serving
spec:
  replicas: 3  # HA: 3 replicas
  selector:
    matchLabels:
      app: triton
  template:
    metadata:
      labels:
        app: triton
    spec:
      containers:
        - name: triton
          image: nvcr.io/nvidia/tritonserver:24.01-py3
          args: ["tritonserver", "--model-repository=s3://models/repo"]
          ports:
            - containerPort: 8000  # HTTP
            - containerPort: 8001  # gRPC
            - containerPort: 8002  # Metrics
          resources:
            limits:
              nvidia.com/gpu: 1
              memory: "16Gi"
            requests:
              nvidia.com/gpu: 1
              memory: "8Gi"
          livenessProbe:
            httpGet:
              path: /v2/health/live
              port: 8000
            initialDelaySeconds: 60
          readinessProbe:
            httpGet:
              path: /v2/health/ready
              port: 8000
      tolerations:
        - key: nvidia.com/gpu
          operator: Exists
          effect: NoSchedule
---
apiVersion: v1
kind: Service
metadata:
  name: triton-inference
spec:
  selector:
    app: triton
  ports:
    - name: http
      port: 8000
    - name: grpc
      port: 8001
  type: ClusterIP
"""

    def show_deployment(self):
        print("=== Triton K8s HA Deployment ===")
        print(self.K8S_DEPLOYMENT[:600])

    def health_dashboard(self):
        print(f"\n=== Triton Health Dashboard ===")
        replicas = [
            {"name": "triton-0", "gpu": f"A100-{random.randint(0,3)}", "models": random.randint(3, 8), "rps": random.randint(50, 200), "latency": f"{random.uniform(5, 30):.1f}ms"},
            {"name": "triton-1", "gpu": f"A100-{random.randint(0,3)}", "models": random.randint(3, 8), "rps": random.randint(50, 200), "latency": f"{random.uniform(5, 30):.1f}ms"},
            {"name": "triton-2", "gpu": f"A100-{random.randint(0,3)}", "models": random.randint(3, 8), "rps": random.randint(50, 200), "latency": f"{random.uniform(5, 30):.1f}ms"},
        ]
        for r in replicas:
            print(f"  [{r['name']}] GPU: {r['gpu']} | Models: {r['models']} | RPS: {r['rps']} | P50: {r['latency']}")

triton = TritonHA()
triton.show_deployment()
triton.health_dashboard()

Monitoring & Auto-Recovery

# monitoring.py — Inference monitoring
import json
import random

class InferenceMonitoring:
    METRICS = {
        "latency_p50": "Inference latency P50 (target: < 20ms)",
        "latency_p99": "Inference latency P99 (target: < 100ms)",
        "throughput": "Requests per second (RPS)",
        "gpu_utilization": "GPU utilization % (target: 60-80%)",
        "gpu_memory": "GPU memory usage %",
        "error_rate": "Inference error rate (target: < 0.1%)",
        "model_accuracy": "Online model accuracy (drift detection)",
    }

    def show_metrics(self):
        print("=== Key Metrics ===\n")
        for name, desc in self.METRICS.items():
            print(f"  [{name}] {desc}")

    def dashboard(self):
        print(f"\n=== Inference Dashboard ===")
        models = [
            {"name": "resnet50-fp16", "rps": random.randint(100, 500), "p50": random.uniform(3, 15), "p99": random.uniform(15, 50), "gpu": random.randint(40, 80)},
            {"name": "bert-base-int8", "rps": random.randint(50, 200), "p50": random.uniform(10, 30), "p99": random.uniform(30, 100), "gpu": random.randint(50, 90)},
            {"name": "yolov8-fp16", "rps": random.randint(30, 150), "p50": random.uniform(8, 25), "p99": random.uniform(25, 80), "gpu": random.randint(40, 75)},
        ]
        for m in models:
            status = "OK" if m["p99"] < 80 else "SLOW"
            print(f"  [{status:>4}] {m['name']:<20} RPS: {m['rps']:>3} | P50: {m['p50']:>5.1f}ms | P99: {m['p99']:>5.1f}ms | GPU: {m['gpu']}%")

mon = InferenceMonitoring()
mon.show_metrics()
mon.dashboard()

การนำไปใช้งานจริงในองค์กร

สำหรับองค์กรขนาดกลางถึงใหญ่ แนะนำให้ใช้หลัก Three-Tier Architecture คือ Core Layer ที่เป็นแกนกลางของระบบ Distribution Layer ที่ทำหน้าที่กระจาย Traffic และ Access Layer ที่เชื่อมต่อกับผู้ใช้โดยตรง การแบ่ง Layer ชัดเจนช่วยให้การ Troubleshoot ง่ายขึ้นและสามารถ Scale ระบบได้ตามความต้องการ

เรื่อง Network Security ก็สำคัญไม่แพ้กัน ควรติดตั้ง Next-Generation Firewall ที่สามารถ Deep Packet Inspection ได้ ใช้ Network Segmentation แยก VLAN สำหรับแต่ละแผนก ติดตั้ง IDS/IPS เพื่อตรวจจับการโจมตี และทำ Regular Security Audit อย่างน้อยปีละ 2 ครั้ง

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง Semgrep SAST สำหรับมือใหม่ Step by Step

เปรียบเทียบข้อดีและข้อเสีย

จากตารางเปรียบเทียบจะเห็นว่าข้อดีมีมากกว่าข้อเสียอย่างชัดเจน โดยเฉพาะในแง่ของประสิทธิภาพและความสามารถในการ Scale สำหรับข้อเสียส่วนใหญ่สามารถแก้ไขได้ด้วยการเรียนรู้อย่างเป็นระบบและวางแผนทรัพยากรให้เหมาะสม

FAQ - คำถามที่พบบ่อย

Q: TensorRT เร็วกว่า PyTorch เท่าไหร่?

แนะนำเพิ่มเติม — บทวิเคราะห์จาก XM Signal

A: โดยทั่วไป 2-6x เร็วกว่า PyTorch native inference FP16: เร็วขึ้น 2-3x จาก FP32 + ลด memory 50% INT8: เร็วขึ้น 3-6x จาก FP32 + ลด memory 75% ขึ้นอยู่กับ model architecture และ GPU — Transformer models ได้ผลดีมาก

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน Tailscale Mesh Incident Management — ทุกสิ่งที่ต้องรู้ในปี 2026

Q: TensorRT Engine ย้าย GPU ได้ไหม?

A: ไม่ได้โดยตรง — Engine เป็น GPU-specific (build สำหรับ GPU รุ่นนั้น) ถ้าเปลี่ยน GPU: ต้อง rebuild engine จาก ONNX ดังนั้น DR plan ต้องเก็บทั้ง ONNX source + Engine file ONNX เป็น portable format — rebuild ได้บนทุก GPU

แนะนำเพิ่มเติม — อ่านเพิ่มเติมที่ SiamCafeBook

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน A C Th คืออะไร — ข้อมูลครบถ้วน 2026

Q: DR สำหรับ ML inference ต้องทำอะไร?

A: 1) Backup: เก็บ ONNX + Engine files ใน S3/GCS 2) Versioning: track ทุก model version + calibration data 3) Multi-replica: Triton 3+ replicas สำหรับ HA 4) Auto-rollback: ถ้า latency/error สูง → rollback model version 5) Multi-region: deploy ใน 2+ regions สำหรับ critical services

Q: INT8 quantization ทำให้ accuracy ลดลงไหม?

เนื้อหาเกี่ยวข้อง — อ่านต่อ: Regular Expression (Regex) คืออะไร? สอน Regex ตั้งแต่เริ่มต้นสำหรับ Developer…

A: ลดลงเล็กน้อย (0.1-1%) ถ้า calibrate ดี วิธี calibrate: ใช้ representative dataset 1,000-10,000 samples ตรวจสอบ: เปรียบเทียบ FP32 vs INT8 accuracy บน validation set ถ้า accuracy drop > 1%: ใช้ FP16 แทน หรือ mixed precision (บาง layers FP16 บาง INT8)