Skip to content

vllm.utils.watch_dog

Classes:

Functions:

  • get_watch_dog –

    Return the process-wide WatchDog singleton.

  • start_watch_dog –

    Start the process-wide watchdog if watchdog_config.dump_dir is set.

ThreadedWatchDog

Bases: WatchDog

Dumps the stacks of all threads when not fed within the timeout.

Methods:

  • handle_failure –

    Append a numbered dump of all thread stacks to the dump file.

Source code in vllm/utils/watch_dog.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class ThreadedWatchDog(WatchDog):
    """Dumps the stacks of all threads when not fed within the timeout."""

    def __init__(self, name: str, watchdog_config: WatchdogConfig, logger: Logger):
        self.name = name
        self.timeout = watchdog_config.timeout
        self.check_interval = watchdog_config.check_interval
        self.dump_dir = watchdog_config.dump_dir
        self.dump_file = os.path.join(
            self.dump_dir, f"VLLM_STACK_DUMP_for_{name}_{os.getpid()}.log"
        )
        self.dump_seq = 1
        self.logger = logger
        self.last_feed_time = time.monotonic()
        # 0.0 so that the first timeout is always dumped.
        self.last_timeout_log = 0.0
        self.active = True
        # Reentrant: handle_failure() may be re-entered from a signal handler.
        self.lock = threading.RLock()
        self.stop_event = threading.Event()
        self.thread: threading.Thread | None = None
        self.handler = None

    def feed(self) -> None:
        self.last_feed_time = time.monotonic()

    @contextmanager
    def disable(self) -> Iterator[None]:
        assert self.active, "WatchDog.disable() is not re-entrant"
        self.active = False
        try:
            yield
        finally:
            self.active = True

    def handle_failure(self, reason: str) -> None:
        """Append a numbered dump of all thread stacks to the dump file."""
        with self.lock:
            if self.handler is None:
                self.dump_stack(reason)
            else:
                self.handler(reason)

    def dump_stack(self, reason: str) -> None:
        try:
            os.makedirs(self.dump_dir, exist_ok=True)
            with open(self.dump_file, "a") as f:
                f.write(
                    f"\n>>> Call stack dump #{self.dump_seq} at "
                    f"{time.ctime()} due to {reason}\n\n"
                )
                faulthandler.dump_traceback(file=f, all_threads=True)
                f.write("\n" + "=" * 64 + "\n")
            self.dump_seq += 1
            self.logger.info(
                "[Watchdog]Dumped stack to %s due to %s", self.dump_file, reason
            )
        except Exception as e:
            self.logger.warning(
                "[Watchdog]Failed to dump stack trace to %s: %s",
                self.dump_file,
                e,
            )

    def set_handler(self, handler) -> None:
        self.handler = handler

    def _check_loop(self) -> None:
        while not self.stop_event.wait(self.check_interval):
            now = time.monotonic()
            # While disabled, treat the watchdog as continuously fed so the
            # timeout window restarts once it is re-enabled.
            if not self.active:
                self.last_feed_time = now
                continue
            # Dump at most once per stall.
            if (
                now - self.last_feed_time > self.timeout
                and self.last_timeout_log < self.last_feed_time
            ):
                self.last_timeout_log = now
                self.handle_failure("feed timeout")

    def start(self) -> None:
        if self.thread is not None and self.thread.is_alive():
            return
        self.stop_event.clear()
        self.thread = threading.Thread(target=self._check_loop, daemon=True)
        self.feed()
        self.thread.start()
        self.logger.info(
            "[Watchdog]Started thread for %s (pid=%d)", self.name, os.getpid()
        )

    def stop(self) -> None:
        self.stop_event.set()
        if self.thread is not None:
            self.thread.join(timeout=2)
            # Keep a still-alive thread so a later start() won't spawn another.
            if not self.thread.is_alive():
                self.thread = None

handle_failure(reason)

Append a numbered dump of all thread stacks to the dump file.

Source code in vllm/utils/watch_dog.py 
76
77
78
79
80
81
82
def handle_failure(self, reason: str) -> None:
    """Append a numbered dump of all thread stacks to the dump file."""
    with self.lock:
        if self.handler is None:
            self.dump_stack(reason)
        else:
            self.handler(reason)

WatchDog

Bases: ABC

Methods:

  • disable –

    Suspend timeout detection, e.g. while blocking on idle input.

  • feed –

    Mark that the process is making progress.

  • handle_failure –

    Record the state of all threads for debugging.

Source code in vllm/utils/watch_dog.py 
15
16
17
18
19
20
21
22
23
24
25
26
class WatchDog(ABC):
    @abstractmethod
    def feed(self) -> None:
        """Mark that the process is making progress."""

    @abstractmethod
    def disable(self) -> AbstractContextManager[None]:
        """Suspend timeout detection, e.g. while blocking on idle input."""

    @abstractmethod
    def handle_failure(self, reason: str) -> None:
        """Record the state of all threads for debugging."""

disable() abstractmethod

Suspend timeout detection, e.g. while blocking on idle input.

Source code in vllm/utils/watch_dog.py 
20
21
22
@abstractmethod
def disable(self) -> AbstractContextManager[None]:
    """Suspend timeout detection, e.g. while blocking on idle input."""

feed() abstractmethod

Mark that the process is making progress.

Source code in vllm/utils/watch_dog.py 
16
17
18
@abstractmethod
def feed(self) -> None:
    """Mark that the process is making progress."""

handle_failure(reason) abstractmethod

Record the state of all threads for debugging.

Source code in vllm/utils/watch_dog.py 
24
25
26
@abstractmethod
def handle_failure(self, reason: str) -> None:
    """Record the state of all threads for debugging."""

get_watch_dog()

Return the process-wide WatchDog singleton.

Source code in vllm/utils/watch_dog.py 
147
148
149
def get_watch_dog() -> WatchDog:
    """Return the process-wide WatchDog singleton."""
    return _watch_dog

start_watch_dog(name, watchdog_config, logger)

Start the process-wide watchdog if watchdog_config.dump_dir is set.

Returns the WatchDog singleton either way.

Source code in vllm/utils/watch_dog.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def start_watch_dog(
    name: str,
    watchdog_config: WatchdogConfig,
    logger: Logger,
) -> WatchDog:
    """Start the process-wide watchdog if ``watchdog_config.dump_dir`` is set.

    Returns the WatchDog singleton either way.
    """
    global _watch_dog
    if not watchdog_config.dump_dir:
        return _watch_dog

    logger.info(
        "[Watchdog]Start with timeout=%s check_interval=%s dump_dir=%s",
        watchdog_config.timeout,
        watchdog_config.check_interval,
        watchdog_config.dump_dir,
    )
    faulthandler.enable(all_threads=True)
    _watch_dog = ThreadedWatchDog(name, watchdog_config, logger)
    _watch_dog.start()
    return _watch_dog