vllm.v1.fault_tolerance.engine_core_sentinel

EngineCoreSentinel and fault_tolerant_wrapper for the engine core.

Classes:

• EngineCoreSentinel –

  Manages fault tolerance state for a single engine core.

Functions:

• fault_tolerant_wrapper –

  Wrap the busy loop to catch faults and delegate recovery.

EngineCoreSentinel

Manages fault tolerance state for a single engine core.

Methods:

• activate_cpu_group_timeouts –

  Give startup-created gloo groups the configured cpu timeout.

• handle_command –

  Dispatch an FT command by instruction name.

• on_executor_failed –

  Notify the client about the executor failure

• on_fault –

  Called by the wrapper when the busy loop raises an exception.

Attributes:

• coordinator_disabled (bool) –

  True once any scale_down has committed dead ranks. The DP wave

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

class EngineCoreSentinel:
    """Manages fault tolerance state for a single engine core."""

    def __init__(self, engine: "EngineCoreProc", parallel_config):
        self.engine = engine
        self.engine_index = engine.engine_index
        self.parallel_config = parallel_config
        ft_config = parallel_config.fault_tolerance_config
        self.engine_recovery_timeout_sec = ft_config.engine_recovery_timeout_sec

        self.resumed = threading.Event()
        self.resumed.set()
        self.status_type = EngineStatusType.HEALTHY
        self.fault_info: str | None = None
        self._initial_dp_size = parallel_config.data_parallel_size
        self._dead_dp_ranks: set[int] = set()
        self._recovery_store: TCPStore | None = None
        self._install_executor_failure_callback()

    @property
    def coordinator_disabled(self) -> bool:
        """True once any scale_down has committed dead ranks. The DP wave
        wake-up path depends on the DP coordinator, which has no failover;
        after a scale_down the engine never idle-pauses and keeps stepping
        dummy batches instead."""
        return bool(self._dead_dp_ranks)

    def activate_cpu_group_timeouts(self) -> None:
        """Give startup-created gloo groups the configured cpu timeout."""
        enter_steady_state()
        timeout_seconds = self.parallel_config.cpu_distributed_timeout_seconds
        if timeout_seconds is None:
            return
        if self._initial_dp_size > 1:
            set_gloo_backend_timeout(
                cast("DPEngineCoreProc", self.engine).dp_group,
                timedelta(seconds=timeout_seconds),
            )
        self.engine.model_executor.collective_rpc(
            "handle_ft_command",
            args=(
                FaultToleranceRequest(
                    instruction="activate_cpu_group_timeouts", params={}
                ),
            ),
        )

    def handle_command(self, client_idx: int, call_id: int, ft_args: dict):
        """Dispatch an FT command by instruction name."""
        ft_request = FaultToleranceRequest(**ft_args)
        if ft_request.instruction not in ALLOWED_FT_INSTRUCTIONS:
            reason = (
                f"[FT] Rejecting unknown instruction "
                f"'{ft_request.instruction}' on engine {self.engine_index}"
            )
            logger.warning(reason)
            result = FaultToleranceResult(
                request_id=ft_request.request_id,
                success=False,
                reason=reason,
            )
        elif self.status_type != EngineStatusType.UNHEALTHY:
            reason = (
                f"[FT] Rejecting {ft_request.instruction} on engine "
                f"{self.engine_index}: status is {self.status_type.name}"
            )
            logger.warning(reason)
            result = FaultToleranceResult(
                request_id=ft_request.request_id,
                success=False,
                reason=reason,
            )
        else:
            try:
                result = run_method(self, ft_request.instruction, (ft_request,), {})
            except Exception as e:
                logger.exception("[FT] Instruction '%s' failed", ft_request.instruction)
                result = FaultToleranceResult(
                    request_id=ft_request.request_id, success=False, reason=str(e)
                )

        uo = UtilityOutput(call_id)
        uo.result = UtilityResult(msgspec.structs.asdict(result))
        self.engine.output_queue.put_nowait(
            (client_idx, EngineCoreOutputs(utility_output=uo))
        )

    def on_fault(self, exc: Exception):
        """Called by the wrapper when the busy loop raises an exception."""
        self.resumed.clear()
        logger.warning(
            "[FT] Busy loop raised %s. Waiting for recovery.", type(exc).__name__
        )

        engine = self.engine
        aborted = engine.scheduler.finish_requests(None, RequestStatus.FINISHED_ABORTED)
        engine._send_abort_outputs(aborted)

        if (
            hasattr(engine.model_executor, "is_failed")
            and engine.model_executor.is_failed
        ):
            self.status_type = EngineStatusType.DEAD
            mask = None
        else:
            # Push DIAGNOSING first. Commands are rejected until UNHEALTHY.
            self.status_type = EngineStatusType.DIAGNOSING
            self._push_status()
            try:
                mask = self._query_mask()
            except Exception:
                logger.warning("[FT] Failed to query mask for status push")
                mask = None
            # on_executor_failed may have set DEAD concurrently.
            if self.status_type == EngineStatusType.DIAGNOSING:
                self.status_type = EngineStatusType.UNHEALTHY
        self.fault_info = f"{type(exc).__name__}"
        logger.info(
            "[FT] Engine %d status -> %s:",
            self.engine_index,
            self.status_type.name,
            exc_info=exc,
        )
        self._push_status(mask)

    def _drain_batch_queue(self) -> None:
        """Drain the batch queue, feeding leftover kv_connector_outputs to
        the scheduler via update_from_output() with an empty SchedulerOutput.
        """
        engine = self.engine
        if engine.batch_queue is None:
            return
        scheduler = cast(Scheduler, engine.scheduler)

        while engine.batch_queue:
            future, _, _ = engine.batch_queue.pop()
            try:
                model_output = future.result()
            except Exception as e:
                # Safe to drop: the KV output was already salvaged by the
                # executor.
                logger.warning(
                    "[FT] Dropping exception from batch queue during fault "
                    "handling: %s",
                    e,
                )
                continue
            kv_connector_output = (
                model_output.kv_connector_output if model_output is not None else None
            )
            if kv_connector_output is None or kv_connector_output.is_empty():
                continue
            scheduler.update_from_output(
                SchedulerOutput.make_empty(),
                ModelRunnerOutput.with_kv_conn_output_only(kv_connector_output),
            )
        engine.batch_queue.clear()

    def _install_executor_failure_callback(self) -> None:
        """Report executor failure to the sentinel as well.

        The busy loop cannot consume EXECUTOR_FAILED while it waits for
        recovery, so the sentinel must observe the failure directly.
        """
        executor = self.engine.model_executor
        upstream_callback = getattr(executor, "failure_callback", None)
        if upstream_callback is None:
            return

        def callback() -> None:
            upstream_callback()
            self.on_executor_failed()

        executor.register_failure_callback(callback)

    def on_executor_failed(self):
        """Notify the client about the executor failure"""
        if self.status_type != EngineStatusType.HEALTHY:
            self.status_type = EngineStatusType.DEAD
            self._push_status()

    def _push_status(self, mask: list[int] | None = None):
        """Push current health to the client so it can refresh its cache."""
        payload = {"id": self.engine_index, "status": self.status_type.name.lower()}
        if self.status_type == EngineStatusType.UNHEALTHY:
            payload["fault_info"] = self.fault_info
            if mask is not None:
                payload["mask"] = mask
        outputs = EngineCoreOutputs(
            utility_output=UtilityOutput(
                call_id=FT_STATUS_CALL_ID,
                result=UtilityResult(payload),
            )
        )
        outputs.engine_index = self.engine_index
        self.engine.output_queue.put_nowait((0, outputs))

    def _query_mask(self) -> list[int]:
        """Union of all workers' all2all masks.

        A rank is excluded if any worker suspects it.
        """
        ft_request = FaultToleranceRequest(instruction="query_mask", params={})
        results = self.engine.model_executor.collective_rpc(
            "handle_ft_command",
            args=(ft_request,),
            timeout=self.engine_recovery_timeout_sec,
        )
        return [max(bits) for bits in zip(*(r["mask"] for r in results))]

    def retry(self, ft_request: FaultToleranceRequest) -> FaultToleranceResult:
        # Workers replay masks for the cumulative dead set.
        ft_request.params["dead_dp_ranks"] = sorted(self._dead_dp_ranks)
        self._prepare_round(ft_request)
        self._recover_and_vote(ft_request, terminal=False)
        return self._mark_healthy(ft_request)

    def scale_down(self, ft_request: FaultToleranceRequest) -> FaultToleranceResult:
        if self._initial_dp_size == 1:
            raise ValueError(
                "scale_down requires data_parallel_size > 1; a dp=1 "
                "engine has no ranks to remove"
            )
        removed_set = set(ft_request.params["removed_dp_ranks"])
        my_rank = self.parallel_config.data_parallel_rank
        newly_dead = removed_set - self._dead_dp_ranks
        if (
            not removed_set
            or not removed_set <= set(range(self._initial_dp_size))
            or my_rank in removed_set
            or set(range(self._initial_dp_size)) <= self._dead_dp_ranks | newly_dead
            or not newly_dead
        ):
            raise ValueError(
                f"Invalid removed_dp_ranks {sorted(removed_set)} for engine "
                f"{self.engine_index} (dp_rank={my_rank}, "
                f"dead_dp_ranks={sorted(self._dead_dp_ranks)})"
            )

        # Removing the store master (the lowest alive rank) requires the new
        # store location in the request.
        alive_before = sorted(set(range(self._initial_dp_size)) - self._dead_dp_ranks)
        if alive_before[0] in newly_dead and (
            ft_request.params.get("dp_store_port") is None
            or ft_request.params.get("dp_master_ip") is None
        ):
            raise ValueError(
                "dp_store_port and dp_master_ip required when the store "
                f"master (rank {alive_before[0]}) is removed"
            )

        new_dead = self._dead_dp_ranks | newly_dead
        ft_request.params["dead_dp_ranks"] = sorted(new_dead)
        self._prepare_round(ft_request, dead_ranks=new_dead)
        self._recover_and_vote(ft_request, terminal=True)
        # Commit the dead set only after the full recovery succeeded.
        self._dead_dp_ranks = new_dead
        logger.info(
            "[FT] Engine %d scale_down complete: removed %s, "
            "cumulative dead_dp_ranks=%s",
            self.engine_index,
            sorted(newly_dead),
            sorted(new_dead),
        )
        return self._mark_healthy(ft_request)

    def _prepare_round(
        self, ft_request: FaultToleranceRequest, dead_ranks: set[int] | None = None
    ) -> None:
        """Fill the round's coordination params.

        The round's rendezvous keys are namespaced by request_id, so every
        engine must carry the same one.
        """
        params = ft_request.params
        dead = self._dead_dp_ranks if dead_ranks is None else dead_ranks
        # The rebuilt gloo group contains only alive members; its internal
        # rank/size are dense over sorted(alive), while parallel_config keeps
        # the frozen original values.
        alive = sorted(set(range(self._initial_dp_size)) - dead)
        master_ip = self.parallel_config.data_parallel_master_ip
        if self._initial_dp_size > 1 and dead_ranks is not None:
            # The lowest alive rank hosts the TCPStore master; rebuild the
            # store if that rank was just removed.
            alive_before = sorted(
                set(range(self._initial_dp_size)) - self._dead_dp_ranks
            )
            if alive_before[0] in dead_ranks - self._dead_dp_ranks:
                master_ip = params["dp_master_ip"]
                self._rebuild_dp_store(
                    master_ip,
                    params["dp_store_port"],
                    is_master=(self.parallel_config.data_parallel_rank == alive[0]),
                    num_clients=len(alive),
                )
        params["dp_master_ip"] = master_ip
        params["dp_group_rank"] = alive.index(self.parallel_config.data_parallel_rank)
        params["dp_group_size"] = len(alive)
        recovery_round = ft_request.request_id
        params["recovery_round"] = recovery_round
        if self._initial_dp_size == 1:
            # dp=1 has no dp_store/dp_group; the engine still hosts the
            # recovery store for its own workers' TP group reinit.
            params["recovery_store_port"] = self._coordinate_recovery_store_port(
                master_ip, recovery_round, is_master=True
            )
            return

        store = cast("DPEngineCoreProc", self.engine).dp_store
        member_prefix = f"ft_round_member_{recovery_round}"
        store.set(f"{member_prefix}_{params['dp_group_rank']}", b"1")
        try:
            store.wait(
                [f"{member_prefix}_{r}" for r in range(len(alive))],
                timedelta(seconds=_ROUND_CHECKIN_TIMEOUT_SEC),
            )
        except Exception as e:
            raise RuntimeError(
                f"[FT] Not all engines joined recovery round "
                f"'{recovery_round}' within {_ROUND_CHECKIN_TIMEOUT_SEC}s; "
                f"the same request_id must be sent to every engine."
            ) from e

    def _rebuild_dp_store(
        self,
        host: str,
        port: int,
        is_master: bool,
        num_clients: int,
    ) -> None:
        """Rebuild dp_store when the old store master was removed."""
        timeout_seconds = self.parallel_config.cpu_distributed_timeout_seconds
        if timeout_seconds is not None:
            timeout = timedelta(seconds=timeout_seconds)
        else:
            timeout = _get_default_timeout(Backend.GLOO)
        engine = cast("DPEngineCoreProc", self.engine)

        if hasattr(engine, "dp_store"):
            del engine.dp_store
        engine.dp_store = TCPStore(
            host,
            port,
            num_clients,
            is_master=is_master,
            timeout=timeout,
        )

    def _reinit_groups(self, ft_request: FaultToleranceRequest) -> None:
        """Reinit the engine DP group (dp>1) and host the recovery store."""
        if self._initial_dp_size == 1:
            return
        params = ft_request.params
        recovery_round = params["recovery_round"]
        master_ip = params["dp_master_ip"]
        params["recovery_store_port"] = self._coordinate_recovery_store_port(
            master_ip, recovery_round, is_master=params["dp_group_rank"] == 0
        )
        self._reinit_engine_groups(
            params["dp_group_rank"], params["dp_group_size"], recovery_round
        )
        # Commit the master IP only after the group reinit succeeded, so a
        # failed recovery leaves a consistent state that can be retried.
        self.parallel_config.data_parallel_master_ip = master_ip

    def _recover_and_vote(
        self,
        ft_request: FaultToleranceRequest,
        terminal: bool,
    ) -> None:
        """Reinit the engine groups, dispatch the command to workers, then
        all surviving engines vote one success bit over the rebuilt DP group
        so they commit or fail together. Any failure is terminal when
        `terminal` is set; otherwise the engine stays UNHEALTHY, retryable."""
        engine = self.engine
        if hasattr(engine, "step_counter"):
            engine.step_counter = 0

        local_error: Exception | None = None
        try:
            self._reinit_groups(ft_request)
            engine.model_executor.collective_rpc(
                "handle_ft_command", args=(ft_request,)
            )
            self._drain_batch_queue()
        except Exception as exc:
            local_error = exc

        success = torch.tensor([local_error is None], dtype=torch.int32)
        if self._initial_dp_size > 1:
            try:
                dist.all_reduce(
                    success,
                    op=dist.ReduceOp.MIN,
                    group=cast("DPEngineCoreProc", engine).dp_group,
                )
            except Exception:
                success.zero_()
        if success.item():
            return

        if terminal:
            self.status_type = EngineStatusType.DEAD
            self.fault_info = "Recovery failed; full restart required"
            self._push_status()
        if local_error is not None:
            raise local_error
        raise RuntimeError("[FT] Recovery vote failed")

    def _mark_healthy(self, ft_request: FaultToleranceRequest) -> FaultToleranceResult:
        self.status_type = EngineStatusType.HEALTHY
        logger.info("[FT] Engine %d status -> HEALTHY", self.engine_index)
        self.resumed.set()
        self._push_status()
        return FaultToleranceResult(request_id=ft_request.request_id, success=True)

    def _reinit_engine_groups(
        self, dense_rank: int, dense_size: int, recovery_round: str
    ) -> None:
        """Reinit the engine DP group (dp>1). Worker group ports are
        coordinated by the workers themselves via the recovery store."""
        engine = cast("DPEngineCoreProc", self.engine)

        prefix_store = PrefixStore(f"ft_engine_dp_{recovery_round}", engine.dp_store)
        timeout = self.parallel_config.cpu_distributed_timeout
        if timeout is None:
            timeout = _get_default_timeout(Backend.GLOO)
        new_group = init_gloo_process_group(
            prefix_store=prefix_store,
            group_rank=dense_rank,
            group_size=dense_size,
            timeout=timeout,
        )
        stateless_destroy_torch_distributed_process_group(engine.dp_group)
        engine.dp_group = new_group

    def _coordinate_recovery_store_port(
        self, master_ip: str, recovery_round: str, is_master: bool
    ) -> int:
        """Create (is_master) or look up (others) the per-round TCPStore
        that workers use to coordinate reinit ports; return its port."""
        if is_master:
            store = create_tcp_store(
                master_ip, 0, is_master=True, world_size=-1, wait_for_workers=False
            )
            self._recovery_store = store
            if self._initial_dp_size > 1:
                key = f"ft_recovery_store_port_{recovery_round}"
                cast("DPEngineCoreProc", self.engine).dp_store.set(
                    key, str(store.port).encode()
                )
            return store.port
        key = f"ft_recovery_store_port_{recovery_round}"
        return int(cast("DPEngineCoreProc", self.engine).dp_store.get(key).decode())

coordinator_disabled property

True once any scale_down has committed dead ranks. The DP wave wake-up path depends on the DP coordinator, which has no failover; after a scale_down the engine never idle-pauses and keeps stepping dummy batches instead.

_coordinate_recovery_store_port(master_ip, recovery_round, is_master)

Create (is_master) or look up (others) the per-round TCPStore that workers use to coordinate reinit ports; return its port.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _coordinate_recovery_store_port(
    self, master_ip: str, recovery_round: str, is_master: bool
) -> int:
    """Create (is_master) or look up (others) the per-round TCPStore
    that workers use to coordinate reinit ports; return its port."""
    if is_master:
        store = create_tcp_store(
            master_ip, 0, is_master=True, world_size=-1, wait_for_workers=False
        )
        self._recovery_store = store
        if self._initial_dp_size > 1:
            key = f"ft_recovery_store_port_{recovery_round}"
            cast("DPEngineCoreProc", self.engine).dp_store.set(
                key, str(store.port).encode()
            )
        return store.port
    key = f"ft_recovery_store_port_{recovery_round}"
    return int(cast("DPEngineCoreProc", self.engine).dp_store.get(key).decode())

_drain_batch_queue()

Drain the batch queue, feeding leftover kv_connector_outputs to the scheduler via update_from_output() with an empty SchedulerOutput.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _drain_batch_queue(self) -> None:
    """Drain the batch queue, feeding leftover kv_connector_outputs to
    the scheduler via update_from_output() with an empty SchedulerOutput.
    """
    engine = self.engine
    if engine.batch_queue is None:
        return
    scheduler = cast(Scheduler, engine.scheduler)

    while engine.batch_queue:
        future, _, _ = engine.batch_queue.pop()
        try:
            model_output = future.result()
        except Exception as e:
            # Safe to drop: the KV output was already salvaged by the
            # executor.
            logger.warning(
                "[FT] Dropping exception from batch queue during fault "
                "handling: %s",
                e,
            )
            continue
        kv_connector_output = (
            model_output.kv_connector_output if model_output is not None else None
        )
        if kv_connector_output is None or kv_connector_output.is_empty():
            continue
        scheduler.update_from_output(
            SchedulerOutput.make_empty(),
            ModelRunnerOutput.with_kv_conn_output_only(kv_connector_output),
        )
    engine.batch_queue.clear()

_install_executor_failure_callback()

Report executor failure to the sentinel as well.

The busy loop cannot consume EXECUTOR_FAILED while it waits for recovery, so the sentinel must observe the failure directly.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _install_executor_failure_callback(self) -> None:
    """Report executor failure to the sentinel as well.

    The busy loop cannot consume EXECUTOR_FAILED while it waits for
    recovery, so the sentinel must observe the failure directly.
    """
    executor = self.engine.model_executor
    upstream_callback = getattr(executor, "failure_callback", None)
    if upstream_callback is None:
        return

    def callback() -> None:
        upstream_callback()
        self.on_executor_failed()

    executor.register_failure_callback(callback)

_prepare_round(ft_request, dead_ranks=None)

Fill the round's coordination params.

The round's rendezvous keys are namespaced by request_id, so every engine must carry the same one.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _prepare_round(
    self, ft_request: FaultToleranceRequest, dead_ranks: set[int] | None = None
) -> None:
    """Fill the round's coordination params.

    The round's rendezvous keys are namespaced by request_id, so every
    engine must carry the same one.
    """
    params = ft_request.params
    dead = self._dead_dp_ranks if dead_ranks is None else dead_ranks
    # The rebuilt gloo group contains only alive members; its internal
    # rank/size are dense over sorted(alive), while parallel_config keeps
    # the frozen original values.
    alive = sorted(set(range(self._initial_dp_size)) - dead)
    master_ip = self.parallel_config.data_parallel_master_ip
    if self._initial_dp_size > 1 and dead_ranks is not None:
        # The lowest alive rank hosts the TCPStore master; rebuild the
        # store if that rank was just removed.
        alive_before = sorted(
            set(range(self._initial_dp_size)) - self._dead_dp_ranks
        )
        if alive_before[0] in dead_ranks - self._dead_dp_ranks:
            master_ip = params["dp_master_ip"]
            self._rebuild_dp_store(
                master_ip,
                params["dp_store_port"],
                is_master=(self.parallel_config.data_parallel_rank == alive[0]),
                num_clients=len(alive),
            )
    params["dp_master_ip"] = master_ip
    params["dp_group_rank"] = alive.index(self.parallel_config.data_parallel_rank)
    params["dp_group_size"] = len(alive)
    recovery_round = ft_request.request_id
    params["recovery_round"] = recovery_round
    if self._initial_dp_size == 1:
        # dp=1 has no dp_store/dp_group; the engine still hosts the
        # recovery store for its own workers' TP group reinit.
        params["recovery_store_port"] = self._coordinate_recovery_store_port(
            master_ip, recovery_round, is_master=True
        )
        return

    store = cast("DPEngineCoreProc", self.engine).dp_store
    member_prefix = f"ft_round_member_{recovery_round}"
    store.set(f"{member_prefix}_{params['dp_group_rank']}", b"1")
    try:
        store.wait(
            [f"{member_prefix}_{r}" for r in range(len(alive))],
            timedelta(seconds=_ROUND_CHECKIN_TIMEOUT_SEC),
        )
    except Exception as e:
        raise RuntimeError(
            f"[FT] Not all engines joined recovery round "
            f"'{recovery_round}' within {_ROUND_CHECKIN_TIMEOUT_SEC}s; "
            f"the same request_id must be sent to every engine."
        ) from e

_push_status(mask=None)

Push current health to the client so it can refresh its cache.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _push_status(self, mask: list[int] | None = None):
    """Push current health to the client so it can refresh its cache."""
    payload = {"id": self.engine_index, "status": self.status_type.name.lower()}
    if self.status_type == EngineStatusType.UNHEALTHY:
        payload["fault_info"] = self.fault_info
        if mask is not None:
            payload["mask"] = mask
    outputs = EngineCoreOutputs(
        utility_output=UtilityOutput(
            call_id=FT_STATUS_CALL_ID,
            result=UtilityResult(payload),
        )
    )
    outputs.engine_index = self.engine_index
    self.engine.output_queue.put_nowait((0, outputs))

_query_mask()

Union of all workers' all2all masks.

A rank is excluded if any worker suspects it.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _query_mask(self) -> list[int]:
    """Union of all workers' all2all masks.

    A rank is excluded if any worker suspects it.
    """
    ft_request = FaultToleranceRequest(instruction="query_mask", params={})
    results = self.engine.model_executor.collective_rpc(
        "handle_ft_command",
        args=(ft_request,),
        timeout=self.engine_recovery_timeout_sec,
    )
    return [max(bits) for bits in zip(*(r["mask"] for r in results))]

_rebuild_dp_store(host, port, is_master, num_clients)

Rebuild dp_store when the old store master was removed.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _rebuild_dp_store(
    self,
    host: str,
    port: int,
    is_master: bool,
    num_clients: int,
) -> None:
    """Rebuild dp_store when the old store master was removed."""
    timeout_seconds = self.parallel_config.cpu_distributed_timeout_seconds
    if timeout_seconds is not None:
        timeout = timedelta(seconds=timeout_seconds)
    else:
        timeout = _get_default_timeout(Backend.GLOO)
    engine = cast("DPEngineCoreProc", self.engine)

    if hasattr(engine, "dp_store"):
        del engine.dp_store
    engine.dp_store = TCPStore(
        host,
        port,
        num_clients,
        is_master=is_master,
        timeout=timeout,
    )

_recover_and_vote(ft_request, terminal)

Reinit the engine groups, dispatch the command to workers, then all surviving engines vote one success bit over the rebuilt DP group so they commit or fail together. Any failure is terminal when terminal is set; otherwise the engine stays UNHEALTHY, retryable.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _recover_and_vote(
    self,
    ft_request: FaultToleranceRequest,
    terminal: bool,
) -> None:
    """Reinit the engine groups, dispatch the command to workers, then
    all surviving engines vote one success bit over the rebuilt DP group
    so they commit or fail together. Any failure is terminal when
    `terminal` is set; otherwise the engine stays UNHEALTHY, retryable."""
    engine = self.engine
    if hasattr(engine, "step_counter"):
        engine.step_counter = 0

    local_error: Exception | None = None
    try:
        self._reinit_groups(ft_request)
        engine.model_executor.collective_rpc(
            "handle_ft_command", args=(ft_request,)
        )
        self._drain_batch_queue()
    except Exception as exc:
        local_error = exc

    success = torch.tensor([local_error is None], dtype=torch.int32)
    if self._initial_dp_size > 1:
        try:
            dist.all_reduce(
                success,
                op=dist.ReduceOp.MIN,
                group=cast("DPEngineCoreProc", engine).dp_group,
            )
        except Exception:
            success.zero_()
    if success.item():
        return

    if terminal:
        self.status_type = EngineStatusType.DEAD
        self.fault_info = "Recovery failed; full restart required"
        self._push_status()
    if local_error is not None:
        raise local_error
    raise RuntimeError("[FT] Recovery vote failed")

_reinit_engine_groups(dense_rank, dense_size, recovery_round)

Reinit the engine DP group (dp>1). Worker group ports are coordinated by the workers themselves via the recovery store.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _reinit_engine_groups(
    self, dense_rank: int, dense_size: int, recovery_round: str
) -> None:
    """Reinit the engine DP group (dp>1). Worker group ports are
    coordinated by the workers themselves via the recovery store."""
    engine = cast("DPEngineCoreProc", self.engine)

    prefix_store = PrefixStore(f"ft_engine_dp_{recovery_round}", engine.dp_store)
    timeout = self.parallel_config.cpu_distributed_timeout
    if timeout is None:
        timeout = _get_default_timeout(Backend.GLOO)
    new_group = init_gloo_process_group(
        prefix_store=prefix_store,
        group_rank=dense_rank,
        group_size=dense_size,
        timeout=timeout,
    )
    stateless_destroy_torch_distributed_process_group(engine.dp_group)
    engine.dp_group = new_group

_reinit_groups(ft_request)

Reinit the engine DP group (dp>1) and host the recovery store.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def _reinit_groups(self, ft_request: FaultToleranceRequest) -> None:
    """Reinit the engine DP group (dp>1) and host the recovery store."""
    if self._initial_dp_size == 1:
        return
    params = ft_request.params
    recovery_round = params["recovery_round"]
    master_ip = params["dp_master_ip"]
    params["recovery_store_port"] = self._coordinate_recovery_store_port(
        master_ip, recovery_round, is_master=params["dp_group_rank"] == 0
    )
    self._reinit_engine_groups(
        params["dp_group_rank"], params["dp_group_size"], recovery_round
    )
    # Commit the master IP only after the group reinit succeeded, so a
    # failed recovery leaves a consistent state that can be retried.
    self.parallel_config.data_parallel_master_ip = master_ip

activate_cpu_group_timeouts()

Give startup-created gloo groups the configured cpu timeout.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def activate_cpu_group_timeouts(self) -> None:
    """Give startup-created gloo groups the configured cpu timeout."""
    enter_steady_state()
    timeout_seconds = self.parallel_config.cpu_distributed_timeout_seconds
    if timeout_seconds is None:
        return
    if self._initial_dp_size > 1:
        set_gloo_backend_timeout(
            cast("DPEngineCoreProc", self.engine).dp_group,
            timedelta(seconds=timeout_seconds),
        )
    self.engine.model_executor.collective_rpc(
        "handle_ft_command",
        args=(
            FaultToleranceRequest(
                instruction="activate_cpu_group_timeouts", params={}
            ),
        ),
    )

handle_command(client_idx, call_id, ft_args)

Dispatch an FT command by instruction name.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def handle_command(self, client_idx: int, call_id: int, ft_args: dict):
    """Dispatch an FT command by instruction name."""
    ft_request = FaultToleranceRequest(**ft_args)
    if ft_request.instruction not in ALLOWED_FT_INSTRUCTIONS:
        reason = (
            f"[FT] Rejecting unknown instruction "
            f"'{ft_request.instruction}' on engine {self.engine_index}"
        )
        logger.warning(reason)
        result = FaultToleranceResult(
            request_id=ft_request.request_id,
            success=False,
            reason=reason,
        )
    elif self.status_type != EngineStatusType.UNHEALTHY:
        reason = (
            f"[FT] Rejecting {ft_request.instruction} on engine "
            f"{self.engine_index}: status is {self.status_type.name}"
        )
        logger.warning(reason)
        result = FaultToleranceResult(
            request_id=ft_request.request_id,
            success=False,
            reason=reason,
        )
    else:
        try:
            result = run_method(self, ft_request.instruction, (ft_request,), {})
        except Exception as e:
            logger.exception("[FT] Instruction '%s' failed", ft_request.instruction)
            result = FaultToleranceResult(
                request_id=ft_request.request_id, success=False, reason=str(e)
            )

    uo = UtilityOutput(call_id)
    uo.result = UtilityResult(msgspec.structs.asdict(result))
    self.engine.output_queue.put_nowait(
        (client_idx, EngineCoreOutputs(utility_output=uo))
    )

on_executor_failed()

Notify the client about the executor failure

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def on_executor_failed(self):
    """Notify the client about the executor failure"""
    if self.status_type != EngineStatusType.HEALTHY:
        self.status_type = EngineStatusType.DEAD
        self._push_status()

on_fault(exc)

Called by the wrapper when the busy loop raises an exception.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def on_fault(self, exc: Exception):
    """Called by the wrapper when the busy loop raises an exception."""
    self.resumed.clear()
    logger.warning(
        "[FT] Busy loop raised %s. Waiting for recovery.", type(exc).__name__
    )

    engine = self.engine
    aborted = engine.scheduler.finish_requests(None, RequestStatus.FINISHED_ABORTED)
    engine._send_abort_outputs(aborted)

    if (
        hasattr(engine.model_executor, "is_failed")
        and engine.model_executor.is_failed
    ):
        self.status_type = EngineStatusType.DEAD
        mask = None
    else:
        # Push DIAGNOSING first. Commands are rejected until UNHEALTHY.
        self.status_type = EngineStatusType.DIAGNOSING
        self._push_status()
        try:
            mask = self._query_mask()
        except Exception:
            logger.warning("[FT] Failed to query mask for status push")
            mask = None
        # on_executor_failed may have set DEAD concurrently.
        if self.status_type == EngineStatusType.DIAGNOSING:
            self.status_type = EngineStatusType.UNHEALTHY
    self.fault_info = f"{type(exc).__name__}"
    logger.info(
        "[FT] Engine %d status -> %s:",
        self.engine_index,
        self.status_type.name,
        exc_info=exc,
    )
    self._push_status(mask)

fault_tolerant_wrapper(busy_loop_func)

Wrap the busy loop to catch faults and delegate recovery.

Source code in vllm/v1/fault_tolerance/engine_core_sentinel.py

def fault_tolerant_wrapper(busy_loop_func: Callable):
    """Wrap the busy loop to catch faults and delegate recovery."""

    def run_with_fault_tolerance(self: "EngineCoreProc"):
        if self.enable_fault_tolerance:
            self.ft_sentinel.activate_cpu_group_timeouts()
        while True:
            try:
                busy_loop_func(self)
            except SystemExit:
                raise
            except Exception as exc:
                if not self.enable_fault_tolerance:
                    raise
                self.ft_sentinel.on_fault(exc)
                recovered = self.ft_sentinel.resumed.wait(
                    timeout=self.ft_sentinel.engine_recovery_timeout_sec
                )
                if recovered:
                    continue
                logger.error(
                    "[FT] No recovery within %ds timeout.",
                    self.ft_sentinel.engine_recovery_timeout_sec,
                )
                raise

    return run_with_fault_tolerance