From 7dbdc5f743202a6d2b5c7616039345107f420d60 Mon Sep 17 00:00:00 2001 From: qgai Date: Thu, 16 Jul 2026 20:44:17 -0700 Subject: [PATCH 1/2] Revert "[https://nvbugs/6435642][fix] handle session reuse worker registration (#16444)" This reverts commit e15883702bb15096c193b1a86d6dcf78f559e400. --- tensorrt_llm/executor/proxy.py | 13 +------------ tests/integration/test_lists/waives.txt | 2 ++ tests/unittest/executor/test_proxy_fast_death.py | 16 ---------------- 3 files changed, 3 insertions(+), 28 deletions(-) diff --git a/tensorrt_llm/executor/proxy.py b/tensorrt_llm/executor/proxy.py index 4850bfcf2f5d..11451bad5b67 100644 --- a/tensorrt_llm/executor/proxy.py +++ b/tensorrt_llm/executor/proxy.py @@ -573,18 +573,7 @@ def mpi_done_callback(future: concurrent.futures.Future): raise RuntimeError( "Executor worker returned error") from ready_signal - self._register_worker_processes(status) - - def _register_worker_processes(self, status: tuple) -> None: - """Register identities returned by locally spawned MPI workers. - - Test session reuse replaces this module's ``MpiPoolSession`` class - reference with a factory, so identify pool-backed sessions by excluding - the external communication session types. - """ - if not isinstance( - self.mpi_session, - (MpiCommSession, RemoteMpiCommSessionClient)) and len(status) == 3: + if isinstance(self.mpi_session, MpiPoolSession) and len(status) == 3: worker_process_identities: List[WorkerProcessIdentity] = status[2] self._worker_process_monitor.register(worker_process_identities) diff --git a/tests/integration/test_lists/waives.txt b/tests/integration/test_lists/waives.txt index ab58cd72d4e4..8841ced2acb3 100644 --- a/tests/integration/test_lists/waives.txt +++ b/tests/integration/test_lists/waives.txt @@ -435,7 +435,9 @@ test_e2e.py::test_multi_nodes_eval[MiniMax-M2-tp16-mmlu] SKIP (https://nvbugs/63 test_e2e.py::test_multi_nodes_eval[MiniMax-M3-tp16-mmlu] SKIP (https://nvbugs/6373561) test_e2e.py::test_ptp_quickstart_advanced_deepseek_r1_w4afp8_8gpus[DeepSeek-R1-W4AFP8-DeepSeek-R1/DeepSeek-R1-W4AFP8] SKIP (https://nvbugs/5836830) unittest/_torch/misc/test_share_tensor.py::TestShareTensor::test_share_tensor_different_dtypes SKIP (https://nvbugs/6418021) +unittest/_torch/modeling -k "modeling_out_of_tree" SKIP (https://nvbugs/6426847) unittest/_torch/modeling -k "modeling_qwen" SKIP (https://nvbugs/6433376) +unittest/_torch/modeling/test_modeling_out_of_tree.py::TestOutOfTree::test_llm_api[True] SKIP (https://nvbugs/6426847) unittest/_torch/modeling/test_modeling_qwen3_5_vl.py::test_qwen35_dense_vl_resolves_mamba_ssm_cache_dtype SKIP (https://nvbugs/6433376) unittest/_torch/modules/moe/test_moe_backend.py::test_moe_backend[act=Relu2-e60_k4_h2048_i1408-seq=8-dtype=torch.bfloat16-backend=TRTLLM-quant=NVFP4-routing=Renormalize] SKIP (https://nvbugs/5989912) unittest/_torch/modules/moe/test_moe_module.py::test_configurable_moe_single_gpu -k "TRTLLM" SKIP (https://nvbugs/6464169) diff --git a/tests/unittest/executor/test_proxy_fast_death.py b/tests/unittest/executor/test_proxy_fast_death.py index d6bbac01544f..ed0e62ae063a 100644 --- a/tests/unittest/executor/test_proxy_fast_death.py +++ b/tests/unittest/executor/test_proxy_fast_death.py @@ -16,12 +16,10 @@ import asyncio import queue as _queue -from unittest.mock import Mock import pytest from tensorrt_llm.executor import EngineDeadError -from tensorrt_llm.executor import proxy as proxy_module from tensorrt_llm.executor.proxy import GenerationExecutorProxy from tensorrt_llm.executor.result import GenerationResult @@ -99,20 +97,6 @@ def test_handle_worker_death_broadcasts_event_driven(): assert proxy._error_queue.get_nowait() is cause -def test_register_worker_processes_with_session_reuse_factory(monkeypatch): - """Session reuse replaces proxy.MpiPoolSession with a factory function.""" - pool_session = object() - monkeypatch.setattr(proxy_module, "MpiPoolSession", lambda n_workers: pool_session) - proxy = _bare_proxy() - proxy.mpi_session = proxy_module.MpiPoolSession(1) - proxy._worker_process_monitor = Mock() - identities = [object()] - - proxy._register_worker_processes((proxy.READY_SIGNAL, None, identities)) - - proxy._worker_process_monitor.register.assert_called_once_with(identities) - - def test_result_step_raises_on_engine_dead(): res = GenerationResult.__new__(GenerationResult) res.queue = _queue.Queue() From 67465e59c5b97a8b369a883ade7e8ccc5d9d5b09 Mon Sep 17 00:00:00 2001 From: qgai Date: Thu, 16 Jul 2026 20:44:19 -0700 Subject: [PATCH 2/2] Revert "[https://nvbugs/6435642][fix] detect killed MPI executor workers (#16338)" This reverts commit e05790a1ca8a327af9121bc5cde25c240213a793. --- tensorrt_llm/executor/proxy.py | 35 +-- tensorrt_llm/executor/worker.py | 5 +- .../executor/worker_process_monitor.py | 200 ------------------ .../executor/test_fatal_error_health_check.py | 178 +--------------- 4 files changed, 11 insertions(+), 407 deletions(-) delete mode 100644 tensorrt_llm/executor/worker_process_monitor.py diff --git a/tensorrt_llm/executor/proxy.py b/tensorrt_llm/executor/proxy.py index 11451bad5b67..5599cc8df2a8 100644 --- a/tensorrt_llm/executor/proxy.py +++ b/tensorrt_llm/executor/proxy.py @@ -46,7 +46,6 @@ get_spawn_proxy_process_env, is_llm_response, print_alive_threads) from .worker import GenerationExecutorWorker, worker_main -from .worker_process_monitor import WorkerProcessIdentity, WorkerProcessMonitor __all__ = [ "GenerationExecutorProxy", @@ -177,7 +176,6 @@ def __init__( self.dispatch_result_thread: Optional[ManagedThread] = None self.rpc_client: Optional[RPCClient] = None - self._worker_process_monitor = WorkerProcessMonitor() self._start_executor_workers(worker_kwargs) # Create RPC client after workers are started (worker starts RPC server) @@ -226,19 +224,6 @@ def _check_mpi_futures(self) -> bool: return True return False - def _check_mpi_workers(self) -> bool: - """Check OS process handles and MPI futures for worker death.""" - dead_worker = self._worker_process_monitor.find_dead_worker() - if dead_worker is not None: - self._set_fatal_error( - RuntimeError("MPI worker rank " - f"{dead_worker.rank} (pid {dead_worker.pid}) " - "exited unexpectedly")) - if not self.doing_shutdown: - self.pre_shutdown() - return True - return self._check_mpi_futures() - def _drain_error_queue(self) -> bool: """Drain all queued errors, skipping per-request errors. @@ -281,7 +266,7 @@ def check_health(self) -> bool: if self._drain_error_queue(): return self._fatal_error is None and not self.doing_shutdown - if self._check_mpi_workers(): + if self._check_mpi_futures(): return False return True @@ -358,10 +343,9 @@ def _check_remote_worker_death(self) -> bool: def _error_monitor_loop(self) -> None: """Background thread that reaps a dead engine and drives pre_shutdown. - Checks local MPI worker process handles and futures, remote-session - worker-death notifications, and the error queue using the shared - ``_check_mpi_workers()``, ``_check_remote_worker_death()`` and - ``_drain_error_queue()`` helpers. + Checks MPI worker futures, remote-session worker-death notifications, + and the error queue using the shared ``_check_mpi_futures()``, + ``_check_remote_worker_death()`` and ``_drain_error_queue()`` helpers. Propagation to pending requests is event-driven via ``_handle_worker_death`` (the MPI future done-callback) where futures @@ -371,7 +355,7 @@ def _error_monitor_loop(self) -> None: """ while not self.doing_shutdown and self._fatal_error is None: try: - if self._check_mpi_workers(): + if self._check_mpi_futures(): logger.error("Error monitor: MPI worker crash detected, " "shutting down") return @@ -553,7 +537,7 @@ def mpi_done_callback(future: concurrent.futures.Future): while True: if self.worker_init_status_queue.poll(1): - status = self.worker_init_status_queue.get() + ready_signal, error_trace = self.worker_init_status_queue.get() # Send ACK to the worker self.worker_init_status_queue.put("ACK") logger.info("get signal from executor worker") @@ -563,7 +547,6 @@ def mpi_done_callback(future: concurrent.futures.Future): raise RuntimeError("Executor worker died during initialization") self._handle_background_error() - ready_signal, error_trace = status[:2] if ready_signal != GenerationExecutorProxy.READY_SIGNAL: logger.error(f"Executor worker initialization error: {error_trace}") # Only abort a session this proxy created; an externally owned @@ -573,10 +556,6 @@ def mpi_done_callback(future: concurrent.futures.Future): raise RuntimeError( "Executor worker returned error") from ready_signal - if isinstance(self.mpi_session, MpiPoolSession) and len(status) == 3: - worker_process_identities: List[WorkerProcessIdentity] = status[2] - self._worker_process_monitor.register(worker_process_identities) - def _abort_all_requests(self): # The results can be finished during this loop, so self._results may be changed. for result in list(self._results.values()): @@ -592,8 +571,6 @@ def pre_shutdown(self): else: self.doing_shutdown = True - self._worker_process_monitor.close() - # Wake the error monitor thread immediately so it exits cleanly if hasattr(self, '_shutdown_event'): self._shutdown_event.set() diff --git a/tensorrt_llm/executor/worker.py b/tensorrt_llm/executor/worker.py index 5958d296e1c4..87841c1edd8e 100644 --- a/tensorrt_llm/executor/worker.py +++ b/tensorrt_llm/executor/worker.py @@ -27,7 +27,6 @@ from .rpc_worker_mixin import RpcWorkerMixin from .utils import (ErrorResponse, IntraProcessQueue, RequestError, WorkerCommIpcAddrs) -from .worker_process_monitor import capture_worker_process_identity __all__ = [ "GenerationExecutorWorker", @@ -311,8 +310,6 @@ def notify_proxy_threads_to_quit(): # error to the error_queue in the main thread. mpi_comm().barrier() - worker_process_identities = mpi_comm().allgather( - capture_worker_process_identity(mpi_rank())) logger_debug(f"Worker {mpi_rank()} ready to setup backend...\n", "green") try: @@ -353,7 +350,7 @@ def notify_proxy_threads_to_quit(): worker.set_result_queue(result_queue) # Send ready signal with confirmation - ready_msg = (ready_signal, None, worker_process_identities) + ready_msg = (ready_signal, None) if not worker_init_status_queue.notify_with_retry(ready_msg): logger.warning( "Failed to deliver ready signal to proxy, continuing anyway" diff --git a/tensorrt_llm/executor/worker_process_monitor.py b/tensorrt_llm/executor/worker_process_monitor.py deleted file mode 100644 index 4d00980c44a9..000000000000 --- a/tensorrt_llm/executor/worker_process_monitor.py +++ /dev/null @@ -1,200 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Linux process-liveness monitoring for locally spawned executor workers.""" - -from __future__ import annotations - -import logging -import os -import select -import socket -import threading -from typing import Dict, List, NamedTuple, Optional - -logger = logging.getLogger(__name__) - - -class WorkerProcessIdentity(NamedTuple): - """Stable identity for one executor worker process.""" - - rank: int - pid: int - start_time: Optional[int] - hostname: str - pid_namespace: Optional[int] - - -def _read_process_state(pid: int) -> Optional[tuple[str, int]]: - """Read the Linux process state and start time from ``/proc``. - - Returns: - A ``(state, start_time)`` tuple, or ``None`` when the process does not - exist or procfs is unavailable. - """ - try: - with open(f"/proc/{pid}/stat", encoding="utf-8") as stat_file: - stat = stat_file.read() - except (FileNotFoundError, ProcessLookupError): - return None - - # The second field is parenthesized and may contain spaces or parentheses. - command_end = stat.rfind(")") - if command_end < 0: - return None - fields = stat[command_end + 2 :].split() - if len(fields) <= 19: - return None - return fields[0], int(fields[19]) - - -def _read_pid_namespace(pid: int) -> Optional[int]: - """Return the inode identifying a Linux PID namespace.""" - try: - return os.stat(f"/proc/{pid}/ns/pid").st_ino - except OSError: - return None - - -def capture_worker_process_identity(rank: int, pid: Optional[int] = None) -> WorkerProcessIdentity: - """Capture the calling worker's rank, PID, and Linux start time.""" - pid = pid if pid is not None else os.getpid() - try: - process_state = _read_process_state(pid) - except (OSError, ValueError): - process_state = None - start_time = process_state[1] if process_state is not None else None - return WorkerProcessIdentity( - rank=rank, - pid=pid, - start_time=start_time, - hostname=socket.gethostname(), - pid_namespace=_read_pid_namespace(pid), - ) - - -class WorkerProcessMonitor: - """Monitor locally spawned workers without relying on MPI futures. - - Linux pidfds are used when Python exposes ``os.pidfd_open``. A procfs - identity check is retained as a fallback for older Python versions. If - neither mechanism is available, the worker is left to the existing MPI - future/error-queue checks. - """ - - def __init__(self) -> None: - self._pidfd_to_identity: Dict[int, WorkerProcessIdentity] = {} - self._procfs_identities: List[WorkerProcessIdentity] = [] - self._dead_identity: Optional[WorkerProcessIdentity] = None - self._poller = select.poll() if hasattr(select, "poll") else None - self._lock = threading.Lock() - self._local_hostname = socket.gethostname() - self._local_pid_namespace = _read_pid_namespace(os.getpid()) - - def register(self, identities: List[WorkerProcessIdentity]) -> None: - """Register locally spawned worker identities for liveness checks.""" - with self._lock: - self._close_unlocked() - pidfd_open = getattr(os, "pidfd_open", None) - for identity in identities: - if identity.hostname != self._local_hostname: - continue - if identity.pid_namespace != self._local_pid_namespace: - continue - if pidfd_open is not None and self._poller is not None: - try: - pidfd = pidfd_open(identity.pid) - except ProcessLookupError: - self._dead_identity = identity - continue - except OSError as error: - logger.debug("pidfd_open(%d) failed: %s", identity.pid, error) - else: - if identity.start_time is not None: - try: - process_state = _read_process_state(identity.pid) - except (OSError, ValueError) as error: - logger.debug( - "Failed to validate process identity for pid %d: %s", - identity.pid, - error, - ) - else: - if ( - process_state is None - or process_state[0] == "Z" - or process_state[1] != identity.start_time - ): - os.close(pidfd) - self._dead_identity = identity - continue - self._pidfd_to_identity[pidfd] = identity - self._poller.register( - pidfd, select.POLLIN | select.POLLERR | select.POLLHUP - ) - continue - - if identity.start_time is not None: - self._procfs_identities.append(identity) - - def find_dead_worker(self) -> Optional[WorkerProcessIdentity]: - """Return the first worker known to have exited, if any.""" - with self._lock: - if self._dead_identity is not None: - return self._dead_identity - - if self._poller is not None: - for pidfd, _ in self._poller.poll(0): - identity = self._pidfd_to_identity.get(pidfd) - if identity is not None: - self._dead_identity = identity - return identity - - for identity in self._procfs_identities: - try: - process_state = _read_process_state(identity.pid) - except (OSError, ValueError) as error: - logger.debug( - "Failed to check process state for pid %d: %s", identity.pid, error - ) - continue - if ( - process_state is None - or process_state[0] == "Z" - or process_state[1] != identity.start_time - ): - self._dead_identity = identity - return identity - return None - - def close(self) -> None: - """Close all process handles and reset the monitor.""" - with self._lock: - self._close_unlocked() - - def _close_unlocked(self) -> None: - for pidfd in self._pidfd_to_identity: - if self._poller is not None: - try: - self._poller.unregister(pidfd) - except (KeyError, OSError): - pass - try: - os.close(pidfd) - except OSError: - pass - self._pidfd_to_identity.clear() - self._procfs_identities.clear() - self._dead_identity = None diff --git a/tests/unittest/executor/test_fatal_error_health_check.py b/tests/unittest/executor/test_fatal_error_health_check.py index 63d31c10ad23..9b510a741396 100644 --- a/tests/unittest/executor/test_fatal_error_health_check.py +++ b/tests/unittest/executor/test_fatal_error_health_check.py @@ -31,8 +31,6 @@ import logging import pathlib import signal -import subprocess -import sys import threading import time from concurrent.futures import Future @@ -56,19 +54,6 @@ classify_error = _mod.classify_error ErrorBudget = _mod.ErrorBudget -_monitor_path = ( - pathlib.Path(__file__).resolve().parents[3] - / "tensorrt_llm" - / "executor" - / "worker_process_monitor.py" -) -_monitor_spec = importlib.util.spec_from_file_location("worker_process_monitor", _monitor_path) -_monitor_mod = importlib.util.module_from_spec(_monitor_spec) -_monitor_spec.loader.exec_module(_monitor_mod) -WorkerProcessMonitor = _monitor_mod.WorkerProcessMonitor -_read_process_state = _monitor_mod._read_process_state -capture_worker_process_identity = _monitor_mod.capture_worker_process_identity - # --------------------------------------------------------------------------- # PyExecutor mock — uses the real classify_error() for classification and @@ -181,7 +166,6 @@ def __init__(self): self.request_queue = Mock() self.workers_started: bool = True self._abort_all_requests_called: bool = False - self._worker_process_monitor = WorkerProcessMonitor() def _check_mpi_futures(self) -> bool: """Return True if any MPI worker future has completed.""" @@ -213,29 +197,13 @@ def _drain_error_queue(self) -> bool: break return drained - def _check_mpi_workers(self) -> bool: - """Return True if an OS process handle or MPI future is dead.""" - dead_worker = self._worker_process_monitor.find_dead_worker() - if dead_worker is not None: - self._set_fatal_error( - RuntimeError( - "MPI worker rank " - f"{dead_worker.rank} (pid {dead_worker.pid}) " - "exited unexpectedly" - ) - ) - if not self.doing_shutdown: - self.pre_shutdown() - return True - return self._check_mpi_futures() - def check_health(self) -> bool: """Check executor health including MPI worker liveness.""" if self.doing_shutdown or self._fatal_error is not None: return False if self._drain_error_queue(): return self._fatal_error is None and not self.doing_shutdown - if self._check_mpi_workers(): + if self._check_mpi_futures(): return False return True @@ -243,7 +211,7 @@ def _error_monitor_loop(self): """Background loop using shared helpers.""" while not self.doing_shutdown and self._fatal_error is None: try: - if self._check_mpi_workers(): + if self._check_mpi_futures(): return self._drain_error_queue() if self._fatal_error is not None: @@ -267,7 +235,6 @@ def pre_shutdown(self): if self.doing_shutdown: return self.doing_shutdown = True - self._worker_process_monitor.close() self._pre_shutdown_called = True self._shutdown_event.set() self._abort_all_requests_called = True @@ -520,114 +487,10 @@ def test_check_health(self, doing_shutdown, fatal_error, queue_error, expected): # --------------------------------------------------------------------------- -# WorkerProcessMonitor: pidfd and procfs liveness detection -# --------------------------------------------------------------------------- -class TestWorkerProcessMonitor: - """Tests for the production OS process-liveness monitor.""" - - @pytest.fixture - def identity(self): - return capture_worker_process_identity(rank=3) - - def test_pidfd_exit_is_detected(self, identity): - monitor = WorkerProcessMonitor() - poller = Mock() - poller.poll.return_value = [(42, 1)] - monitor._poller = poller - - with ( - patch.object(_monitor_mod.os, "pidfd_open", return_value=42, create=True), - patch.object(_monitor_mod.os, "close"), - ): - monitor.register([identity]) - assert monitor.find_dead_worker() == identity - monitor.close() - - poller.register.assert_called_once() - poller.unregister.assert_called_once_with(42) - - def test_exit_before_pidfd_registration_is_detected(self, identity): - monitor = WorkerProcessMonitor() - with patch.object( - _monitor_mod.os, "pidfd_open", side_effect=ProcessLookupError, create=True - ): - monitor.register([identity]) - assert monitor.find_dead_worker() == identity - - def test_pid_reuse_before_pidfd_registration_is_detected(self, identity): - if identity.start_time is None: - identity = identity._replace(start_time=100) - monitor = WorkerProcessMonitor() - with ( - patch.object(_monitor_mod.os, "pidfd_open", return_value=42, create=True), - patch.object( - _monitor_mod, - "_read_process_state", - return_value=("S", identity.start_time + 1), - ), - patch.object(_monitor_mod.os, "close") as close, - ): - monitor.register([identity]) - assert monitor.find_dead_worker() == identity - - close.assert_called_once_with(42) - - def test_procfs_start_time_change_is_detected(self, identity): - if identity.start_time is None: - identity = identity._replace(start_time=100) - monitor = WorkerProcessMonitor() - with ( - patch.object(_monitor_mod.os, "pidfd_open", None, create=True), - patch.object( - _monitor_mod, "_read_process_state", return_value=("S", identity.start_time + 1) - ), - ): - monitor.register([identity]) - assert monitor.find_dead_worker() == identity - - def test_procfs_read_error_does_not_report_death(self, identity): - if identity.start_time is None: - identity = identity._replace(start_time=100) - monitor = WorkerProcessMonitor() - with ( - patch.object(_monitor_mod.os, "pidfd_open", None, create=True), - patch.object( - _monitor_mod, "_read_process_state", side_effect=PermissionError("denied") - ), - ): - monitor.register([identity]) - assert monitor.find_dead_worker() is None - - def test_remote_worker_is_not_registered(self, identity): - monitor = WorkerProcessMonitor() - remote_identity = identity._replace(hostname="remote-host") - with patch.object( - _monitor_mod.os, "pidfd_open", return_value=42, create=True - ) as pidfd_open: - monitor.register([remote_identity]) - - pidfd_open.assert_not_called() - assert monitor.find_dead_worker() is None - - def test_different_pid_namespace_is_not_registered(self, identity): - monitor = WorkerProcessMonitor() - namespace = monitor._local_pid_namespace - other_namespace = 1 if namespace is None else namespace + 1 - other_identity = identity._replace(pid_namespace=other_namespace) - with patch.object( - _monitor_mod.os, "pidfd_open", return_value=42, create=True - ) as pidfd_open: - monitor.register([other_identity]) - - pidfd_open.assert_not_called() - assert monitor.find_dead_worker() is None - - -# --------------------------------------------------------------------------- -# GenerationExecutorProxy: check_health with MPI worker liveness +# GenerationExecutorProxy: check_health with MPI futures # --------------------------------------------------------------------------- class TestProxyCheckHealth: - """Tests for GenerationExecutorProxy's MPI worker health checks.""" + """Tests for GenerationExecutorProxy's check_health with MPI futures.""" @pytest.fixture def executor(self): @@ -660,39 +523,6 @@ def test_parent_unhealthy_short_circuits(self, executor): executor.mpi_futures = [Future()] assert executor.check_health() is False - @pytest.mark.skipif( - not sys.platform.startswith("linux"), reason="pidfd/procfs worker monitoring is Linux-only" - ) - def test_killed_process_detected_while_future_remains_pending(self, executor): - """A SIGKILLed worker is fatal even when its MPI future stays pending.""" - process = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) - future = Future() - try: - process_state = _read_process_state(process.pid) - assert process_state is not None - identity = capture_worker_process_identity(rank=0, pid=process.pid) - assert identity.start_time == process_state[1] - executor._worker_process_monitor.register([identity]) - executor.mpi_futures = [future] - assert executor.check_health() is True - - process.kill() - process.wait(timeout=5) - - deadline = time.monotonic() + 5 - while executor.check_health() and time.monotonic() < deadline: - time.sleep(0.01) - - assert executor.check_health() is False - assert future.done() is False - assert "rank 0" in str(executor._fatal_error) - assert executor._pre_shutdown_called - finally: - executor._worker_process_monitor.close() - if process.poll() is None: - process.kill() - process.wait(timeout=5) - # --------------------------------------------------------------------------- # pre_shutdown sentinel send (regression: PR #12718)