sgl-project--sglang/python/sglang/test/ascend/output_capturer.py

import os
import select
import threading


class OutputCapturer:
    """Capture all console print information

    Class Description:
        Capture console output using low-level file descriptor redirection.
        Used to obtain print information from child processes, NPU processes,
        and underlying C/C++ modules that are not logged in sglang logs for test assertion.
        All captured output will be displayed normally in the console in real-time.
    """

    def __init__(self):
        """Initialize all member variables of the capturer"""
        self.old_stdout = None
        self.old_stderr = None
        self.pipe_out = None
        self.pipe_in = None
        self.pipe_err_out = None
        self.pipe_err_in = None
        self.captured_stdout = []
        self.captured_stderr = []
        self.stop_thread = False
        self.thread = None

    def start(self):
        """Start console output capture"""
        # Duplicate and save original stdout/stderr file descriptors
        self.old_stdout = os.dup(1)
        self.old_stderr = os.dup(2)

        # Create anonymous pipes for output redirection
        self.pipe_out, self.pipe_in = os.pipe()
        self.pipe_err_out, self.pipe_err_in = os.pipe()

        # Redirect system stdout/stderr to the write end of pipes
        os.dup2(self.pipe_in, 1)
        os.dup2(self.pipe_err_in, 2)

        # Close unused pipe write ends
        os.close(self.pipe_in)
        os.close(self.pipe_err_in)

        # Start daemon thread to read output in real time
        self.stop_thread = False
        self.thread = threading.Thread(target=self._read_loop, daemon=True)
        self.thread.start()

    def _read_loop(self):
        """The background process reads and prints pipeline data records in a loop."""
        read_fds = [self.pipe_out, self.pipe_err_out]
        while not self.stop_thread:
            try:
                # select listens to multiple file descriptors simultaneously, waiting for data in a non-blocking manner
                readable, _, exceptional = select.select(read_fds, [], read_fds, 0.01)

                # Processing file descriptors containing data
                for fd in readable:
                    if fd == self.pipe_out:
                        data = os.read(fd, 4096)
                        if data:
                            self.captured_stdout.append(data)
                            os.write(self.old_stdout, data)
                    elif fd == self.pipe_err_out:
                        err_data = os.read(fd, 4096)
                        if err_data:
                            self.captured_stderr.append(err_data)
                            os.write(self.old_stderr, err_data)

                for fd in exceptional:
                    if fd in read_fds:
                        self.stop()

            except OSError:
                self.stop()
                break

    def get_all(self):
        """Get all captured stdout and stderr as UTF-8 string

        Return: Decoded stdout and stderr string (ignore decoding errors)
        """
        return self.get_output() + self.get_error()

    def get_output(self):
        """Get all captured stdout as UTF-8 string

        Return: Decoded stdout string (ignore decoding errors)
        """
        return b"".join(self.captured_stdout).decode("utf-8", errors="ignore")

    def get_error(self):
        """Get all captured stderr as UTF-8 string

        Return: Decoded stderr string (ignore decoding errors)
        """
        return b"".join(self.captured_stderr).decode("utf-8", errors="ignore")

    def stop(self):
        """Stop capture and restore system environment"""
        self.stop_thread = True
        if self.thread:
            self.thread.join(timeout=0.5)

        # Restore original output
        os.dup2(self.old_stdout, 1)
        os.dup2(self.old_stderr, 2)

        # Close all file descriptors
        for fd in [self.pipe_out, self.pipe_err_out, self.old_stdout, self.old_stderr]:
            try:
                os.close(fd)
            except (OSError, IOError):
                pass