ทำให้การสร้าง TensorRT Engine ใช้เวลานาน เป็นแบบสังเกตการณ์และยกเลิกได้ใน Python หรือ C++ 🛠️

# Python Example

builder_config.progress_monitor = MyProgressMonitor()



// C++ Example

builder_config->setProgressMonitor(myProgressMonitor);

import threading

from typing import Optional



from tensorrt import IProgressMonitor, BuilderConfig, Logger



class MyProgressMonitor(IProgressMonitor):

    def __init__(self):

        super().__init__()

        self._lock = threading.Lock()

        self._active_phases = {}

        self._cancelled = False



    def phase_start(self, phase_name: str, parent_phase: Optional[str], num_steps: int):

        with self._lock:

            self._active_phases[phase_name] = {'current_step': 0, 'num_steps': num_steps, 'parent': parent_phase}

            print(f"Starting phase: {phase_name} ({num_steps} steps)")



    def step_complete(self, current_step: int, num_steps: int) -> bool:

        with self._lock:

            

# Find the most recent active phase

            current_phase_name = None

            for name, phase_data in self._active_phases.items():

                if phase_data['current_step'] < num_steps: 

# Heuristic to find the active phase

                    current_phase_name = name

                    break



            if current_phase_name:

                self._active_phases[current_phase_name]['current_step'] = current_step

                print(f"  Phase '{current_phase_name}': Step {current_step}/{num_steps}")



            if self._cancelled:

                print("Cancellation requested. Stopping build.")

                return False 

# Request cancellation

        return True 

# Continue build



    def phase_finish(self, phase_name: str, num_steps: int):

        with self._lock:

            if phase_name in self._active_phases:

                print(f"Finished phase: {phase_name}")

                del self._active_phases[phase_name]

            else:

                print(f"Warning: Phase '{phase_name}' finished but was not tracked.")



    def request_cancel(self):

        self._cancelled = True



# Example usage within your build logic:

# builder_config = ...

# monitor = MyProgressMonitor()

# builder_config.progress_monitor = monitor

# ... build engine ...

# if monitor._cancelled:

#     print("Build was cancelled.")

# Example of a simple renderer using ANSI escape codes

import sys



class RichProgressMonitor(MyProgressMonitor):

    def __init__(self):

        super().__init__()

        self._lines_to_clear = 0



    def _render(self):

        with self._lock:

            

# Clear previous lines

            if self._lines_to_clear > 0:

                sys.stdout.write(f"\x1b[{self._lines_to_clear}A") 

# Move cursor up

                sys.stdout.write("\x1b[2K") 

# Clear line

            

            current_line = 0

            for phase_name, phase_data in self._active_phases.items():

                progress = f"{phase_data['current_step']}/{phase_data['num_steps']}"

                print(f"Phase: {phase_name} [{progress}]")

                current_line += 1

            

            sys.stdout.flush()

            self._lines_to_clear = current_line



    def phase_start(self, phase_name: str, parent_phase: Optional[str], num_steps: int):

        super().phase_start(phase_name, parent_phase, num_steps)

        self._render()



    def step_complete(self, current_step: int, num_steps: int) -> bool:

        result = super().step_complete(current_step, num_steps)

        self._render()

        return result



    def phase_finish(self, phase_name: str, num_steps: int):

        super().phase_finish(phase_name, num_steps)

        self._lines_to_clear -= 1 

# One less line to clear

        self._render()



# Usage:

# monitor = RichProgressMonitor()

# builder_config.progress_monitor = monitor

# ... build engine ...

import signal



def signal_handler(sig, frame):

    print('Ctrl+C detected. Requesting cancellation...')

    monitor.request_cancel() 

# Assuming 'monitor' is your globally accessible instance



# Register the handler

signal.signal(signal.SIGINT, signal_handler)



# ... then wire up your monitor and run the build ...

# build_serialized_network()

#include 

#include 

#include 

#include 



#include "NvInfer.h"



class MyProgressMonitorCpp : public nvinfer1::IProgressMonitor {

public:

    MyProgressMonitorCpp() : _cancelled(false) {}



    void phase_start(const char* phase_name, const char* parent_phase, int num_steps) noexcept override {

        std::cout

ทำให้การสร้าง TensorRT Engine ใช้เวลานาน เป็นแบบสังเกตการณ์และยกเลิกได้ใน Python หรือ C++ 🛠️การสร้าง TensorRT Engine เป็นกระบวนการที่สำคัญในการปรับใช้โมเดล AI ให้มีประสิทธิภาพสูงสุดบนฮาร์ดแวร์ NVIDIA แต่บางครั้งกระบวนการนี้อาจใช้เวลานานหลายนาที หรือแม้กระทั่งหลายชั่วโมง โดยเฉพาะอย่างยิ่งกับโมเดลขนาดใหญ่ การค้นหาแทคติกที่ซับซ้อน หรือเมื่อระบบต้องทำงานกับ GPU ที่เพิ่งเปิดใช้งานใหม่ การที่ผู้พัฒนา ผู้ใช้งาน หรือเอเจนต์ AI ต้องเฝ้าดูหน้าจอที่ค้างอยู่นานโดยไม่ทราบความคืบหน้าหรือไม่สามารถยกเลิกได้ ถือเป็นปัญหาที่ทำให้เสียเวลาและทรัพยากร GPU โดยเปล่าประโยชน์ข่าวดีก็คือ NVIDIA TensorRT มี API ที่ชื่อว่า IProgressMonitor ซึ่งเป็นเครื่องมือที่ช่วยให้เราสามารถติดตามความคืบหน้าของการสร้าง Engine ได้อย่างละเอียดและปลอดภัย รวมถึงสามารถยกเลิกกระบวนการได้หากจำเป็น บทความนี้จะแนะนำวิธีการนำ IProgressMonitor มาใช้งานใน Python และ C++ เพื่อจัดการกับปัญหาดังกล่าวIProgressMonitor คืออะไร และให้ประโยชน์อะไรบ้าง? 🤔IProgressMonitor เป็นคลาสพื้นฐาน (abstract base class) ที่ TensorRT จะเรียกใช้ระหว่างการสร้าง Engine หน้าที่หลักของเราคือการสร้างคลาสย่อย (subclass) และกำหนดค่า (override) เมธอดสำคัญ 3 เมธอด ได้แก่ phasestart, stepcomplete, และ phase_finish รูปแบบของเมธอดเหล่านี้จะเหมือนกันทั้งใน Python และ C++เมธอดสำคัญของ IProgressMonitor:phasestart(phasename: str, parentphase: Optional[str], numsteps: int): เมธอดนี้จะถูกเรียกเมื่อเริ่มต้นเฟส (phase) ใหม่ของการสร้าง Engine โดยจะระบุชื่อเฟส ชื่อเฟสหลัก (หากเป็นเฟสย่อย) และจำนวนขั้นตอนทั้งหมดในเฟสนั้นstepcomplete(currentstep: int, num_steps: int) -> bool: เมธอดนี้จะถูกเรียกเมื่อแต่ละขั้นตอนย่อย (step) เสร็จสมบูรณ์ โดยจะส่งค่าจำนวนขั้นตอนปัจจุบันและจำนวนขั้นตอนทั้งหมดมาให้ และที่สำคัญคือเมธอดนี้จะคืนค่าเป็น Boolean:True: อนุญาตให้การสร้าง Engine ดำเนินต่อไปFalse: ร้องขอให้ยกเลิกกระบวนการสร้าง Enginephasefinish(phasename: str, num_steps: int): เมธอดนี้จะถูกเรียกเมื่อเฟสการสร้าง Engine สิ้นสุดลงโครงสร้างแบบ Tree ของความคืบหน้า 🌳IProgressMonitor สามารถติดตามเฟสที่ซ้อนกันได้ (nested phases) หากเมธอด phasestart ถูกเรียกพร้อมกับ parentphase ที่ไม่เป็นค่าว่าง (non-null) สิ่งนี้ทำให้เราเห็นภาพความคืบหน้าเป็นโครงสร้างแบบต้นไม้ (tree) แทนที่จะเป็นรายการแบบแบน (flat list)ความปลอดภัยของ Thread (Thread-Safety) 🧵การนำ IProgressMonitor ไปใช้งานต้องคำนึงถึงความปลอดภัยของ Thread เนื่องจาก TensorRT อาจเรียกใช้ Instance เดียวกันของ Monitor จากหลาย Thread พร้อมกัน ดังนั้น การเข้าถึงข้อมูลหรือการแสดงผลต้องทำอย่างระมัดระวัง หรือใช้ Lock เพื่อป้องกันปัญหาการแสดงผลที่ผิดพลาด (display tearing)การเชื่อมต่อ Monitor กับ Builder 🔗การเชื่อมต่อ IProgressMonitor เข้ากับกระบวนการสร้าง Engine ทำได้ง่ายๆ เพียงการตั้งค่า Monitor บน IBuilderConfig ซึ่งเป็นเพียงการเรียกเมธอดเดียวในทั้ง Python และ C++# Python Example builder_config.progress_monitor = MyProgressMonitor() // C++ Example builder_config->setProgressMonitor(myProgressMonitor);สิ่งที่บทความนี้จะสร้าง 🏗️บทความนี้จะแสดงวิธีการ:สร้าง Subclass ของ IProgressMonitor ใน Python และ C++เพิ่มความสามารถในการยกเลิก (Cancellation) ผ่านเมธอด step_complete โดยตอบสนองต่อการกด Ctrl+C หรือสัญญาณหยุดจาก Outer Event Loopแสดงผล (Route) การอัปเดตความคืบหน้า ไปยัง Terminal, IDE, Service หรือ Agent Runtimeการนำไปใช้งานจริง 🚀1. Subclass IProgressMonitor ใน Python 🐍การสร้าง Subclass ใน Python ทำได้ง่ายและมีโค้ดไม่มากนัก โดยจะเน้นการติดตามเฟสที่กำลังทำงานอยู่และจำนวนขั้นตอนของแต่ละเฟสimport threading from typing import Optional from tensorrt import IProgressMonitor, BuilderConfig, Logger class MyProgressMonitor(IProgressMonitor): def __init__(self): super().__init__() self._lock = threading.Lock() self._active_phases = {} self._cancelled = False def phase_start(self, phase_name: str, parent_phase: Optional[str], num_steps: int): with self._lock: self._active_phases[phase_name] = {'current_step': 0, 'num_steps': num_steps, 'parent': parent_phase} print(f"Starting phase: {phase_name} ({num_steps} steps)") def step_complete(self, current_step: int, num_steps: int) -> bool: with self._lock: # Find the most recent active phase current_phase_name = None for name, phase_data in self._active_phases.items(): if phase_data['current_step'] < num_steps: # Heuristic to find the active phase current_phase_name = name break if current_phase_name: self._active_phases[current_phase_name]['current_step'] = current_step print(f" Phase '{current_phase_name}': Step {current_step}/{num_steps}") if self._cancelled: print("Cancellation requested. Stopping build.") return False # Request cancellation return True # Continue build def phase_finish(self, phase_name: str, num_steps: int): with self._lock: if phase_name in self._active_phases: print(f"Finished phase: {phase_name}") del self._active_phases[phase_name] else: print(f"Warning: Phase '{phase_name}' finished but was not tracked.") def request_cancel(self): self._cancelled = True # Example usage within your build logic: # builder_config = ... # monitor = MyProgressMonitor() # builder_config.progress_monitor = monitor # ... build engine ... # if monitor._cancelled: # print("Build was cancelled.")ข้อควรทราบ:Lock: จำเป็นอย่างยิ่งเพื่อป้องกันปัญหาการแสดงผลที่ผิดพลาดเมื่อ TensorRT เรียกใช้ Monitor จากหลาย Threadstep_complete: เป็นเมธอดเดียวที่สามารถทำให้การสร้าง Engine หยุดได้ โดยการคืนค่า Falsephasestart: ไม่สามารถใช้เพื่อปฏิเสธเฟสก่อนที่จะเริ่มได้ การยกเลิกจะเกิดขึ้นได้เร็วที่สุดเมื่อ stepcomplete แรกของเฟสถูกเรียก2. การแสดงผลความคืบหน้าด้วย Virtual Terminal Escapes 📺ส่วนนี้คือส่วนที่แตกต่างกันไปตามสภาพแวดล้อม การใช้งาน Virtual Terminal Escapes ช่วยให้เราสามารถอัปเดตข้อความบนบรรทัดเดิมใน Terminal ได้ ทำให้เห็นภาพความคืบหน้าที่เคลื่อนไหว# Example of a simple renderer using ANSI escape codes import sys class RichProgressMonitor(MyProgressMonitor): def __init__(self): super().__init__() self._lines_to_clear = 0 def _render(self): with self._lock: # Clear previous lines if self._lines_to_clear > 0: sys.stdout.write(f"\x1b[{self._lines_to_clear}A") # Move cursor up sys.stdout.write("\x1b[2K") # Clear line current_line = 0 for phase_name, phase_data in self._active_phases.items(): progress = f"{phase_data['current_step']}/{phase_data['num_steps']}" print(f"Phase: {phase_name} [{progress}]") current_line += 1 sys.stdout.flush() self._lines_to_clear = current_line def phase_start(self, phase_name: str, parent_phase: Optional[str], num_steps: int): super().phase_start(phase_name, parent_phase, num_steps) self._render() def step_complete(self, current_step: int, num_steps: int) -> bool: result = super().step_complete(current_step, num_steps) self._render() return result def phase_finish(self, phase_name: str, num_steps: int): super().phase_finish(phase_name, num_steps) self._lines_to_clear -= 1 # One less line to clear self._render() # Usage: # monitor = RichProgressMonitor() # builder_config.progress_monitor = monitor # ... build engine ...ข้อควรระวัง:ห้าม Redirect stdout: หากใช้ Renderer แบบ Terminal อย่า Redirect stdout ไปยังไฟล์หรือ Pipe เพราะรหัส Escape จะถูกเขียนลงไป ทำให้ Log อ่านได้ยากสำหรับ Non-Terminal Sinks: หากต้องการส่งข้อมูลไปยังระบบอื่นที่ไม่ใช่ Terminal ให้เปลี่ยน _render() เป็นการส่งข้อมูลแบบ Structured Emission3. การเพิ่มเส้นทางการยกเลิก (Cancel Path) 🚫การเพิ่มความสามารถในการยกเลิกทำได้ง่ายๆ เพียงไม่กี่บรรทัด โดยการตั้งค่า Signal Handler เพื่อเปลี่ยน Flag และให้ step_complete ตรวจสอบ Flag นั้นimport signal def signal_handler(sig, frame): print('Ctrl+C detected. Requesting cancellation...') monitor.request_cancel() # Assuming 'monitor' is your globally accessible instance # Register the handler signal.signal(signal.SIGINT, signal_handler) # ... then wire up your monitor and run the build ... # build_serialized_network()เมื่อ monitor.requestcancel() ถูกเรียก stepcomplete จะคืนค่า False และ TensorRT จะเริ่มกระบวนการ Unwind ซึ่งอาจใช้เวลาสักครู่ โดยเฉพาะอย่างยิ่งหากอยู่ในขั้นตอน tactic search ที่ยาวนานการแสดงผล Latency:ควรแจ้งให้ผู้ใช้ทราบว่าการยกเลิกอาจใช้เวลาสักครู่ โดยแสดงข้อความ "Cancelling..." ในระหว่างที่ระบบกำลัง UnwindFlag cancelled สามารถตั้งค่าได้จากที่อื่นที่ไม่ใช่ Signal Handler เช่น ปุ่ม Stop ใน IDE, Timeout ของ Agent หรือ Webhook ของ CI/CD เพียงแค่ตั้งค่า monitor.cancelled = True การยกเลิกก็จะเกิดขึ้นที่ขอบเขตของ step_complete ถัดไป4. รูปแบบเดียวกันใน C++ ⚙️หลักการทำงานเหมือนกับ Python ทุกประการ เพียงแค่ปรับ Syntax ให้เป็น C++#include #include #include #include #include "NvInfer.h" class MyProgressMonitorCpp : public nvinfer1::IProgressMonitor { public: MyProgressMonitorCpp() : _cancelled(false) {} void phase_start(const char* phase_name, const char* parent_phase, int num_steps) noexcept override { std::couthttps://developer.nvidia.com/blog/make-long-running-nvidia-tensorrt-engine-builds-observable-and-cancelable-in-python-or-c/