vllm.v1.worker.sentinel.gpu_worker_sentinel

Classes:

• WorkerSentinel –

  Holds FT state for a single worker (mask tensors, DP config).

WorkerSentinel

Holds FT state for a single worker (mask tensors, DP config).

Methods are called via collective_rpc from EngineCoreSentinel.

Methods:

• handle_command –

  Dispatch an FT command by instruction name.

• query_mask –

  Query the mask on a side stream.

Source code in vllm/v1/worker/sentinel/gpu_worker_sentinel.py

class WorkerSentinel:
    """Holds FT state for a single worker (mask tensors, DP config).

    Methods are called via collective_rpc from EngineCoreSentinel.
    """

    def __init__(self, worker: "Worker"):
        self.worker = worker
        all2all_backend = worker.parallel_config.all2all_backend
        if all2all_backend not in FT_BACKEND_SET:
            raise ValueError(
                f"Fault tolerance requires an FT-capable all2all backend "
                f"(one of {sorted(FT_BACKEND_SET)}), but got '{all2all_backend}'."
            )
        # Recover registered draft experts: until draft reload is supported,
        # reject MoE drafts before serving rather than leaving stale EP mappings.
        spec_config = worker.vllm_config.speculative_config
        if (
            spec_config is not None
            and spec_config.draft_model_config is not None
            and spec_config.draft_model_config.is_moe
        ):
            raise ValueError("Fault tolerance does not support MoE draft models yet.")
        if worker.parallel_config.prefill_context_parallel_size > 1:
            raise ValueError("Fault tolerance does not support PCP yet.")
        if not worker.vllm_config.use_v2_model_runner:
            raise ValueError(
                "Fault tolerance requires the v2 model runner "
                "(VLLM_USE_V2_MODEL_RUNNER=1)."
            )

    def handle_command(self, ft_request: FaultToleranceRequest):
        """Dispatch an FT command by instruction name."""
        with set_current_vllm_config(self.worker.vllm_config):
            return run_method(self, ft_request.instruction, (ft_request,), {})

    def activate_cpu_group_timeouts(self, ft_request: FaultToleranceRequest):
        enter_steady_state()
        parallel_config = self.worker.parallel_config
        timeout_seconds = parallel_config.cpu_distributed_timeout_seconds
        if timeout_seconds is None:
            return
        timeout = timedelta(seconds=timeout_seconds)
        groups = [get_dp_group(), get_tp_group(), get_pp_group()]
        if parallel_config.enable_expert_parallel:
            groups.append(get_ep_group())
        if parallel_config.enable_eplb:
            groups.append(get_eplb_group())
        for group in groups:
            if group.world_size > 1:
                set_gloo_backend_timeout(group.cpu_group, timeout)
                # GroupCoordinator caches cpu_timeout at init (None under FT
                # startup); its cpu_all_reduce_ forwards it, so update both.
                group.cpu_timeout = timeout

    def retry(self, ft_request: FaultToleranceRequest):
        torch.accelerator.synchronize()
        params = ft_request.params
        # After a scale_down the DP master may be re-elected and the rebuilt
        # gloo group is dense over the alive slots.
        master_ip = params["dp_master_ip"]
        dp_group_rank = params["dp_group_rank"]
        dp_group_size = params["dp_group_size"]
        round_key = params["recovery_round"]
        store = get_cached_tcp_store_client(master_ip, params["recovery_store_port"])
        self._clean_worker_state()
        reset_eplb_async_state(self.worker.model_runner)
        if self.worker.parallel_config.data_parallel_size > 1:
            get_ep_all2all_manager().clean_buffers()
            world_size = self.worker.parallel_config.world_size
            dp_index = self.worker.rank % world_size
            _reinit_cpu_group(
                get_dp_group(),
                store,
                f"ft_dp_port_{round_key}_{dp_index}",
                dp_group_rank,
                dp_group_size,
                timeout=self.worker.parallel_config.cpu_distributed_timeout,
            )
            get_dp_group().dead_dp_ranks = set(params["dead_dp_ranks"])

            # clean_buffers wiped the mask; replay masks for the cumulative dead set.
            tp_size = self.worker.parallel_config.tensor_parallel_size
            pcp_size = self.worker.parallel_config.prefill_context_parallel_size
            all2all_manager = get_ep_all2all_manager()
            for ep_rank in compute_dead_ep_ranks(
                params["dead_dp_ranks"], tp_size, pcp_size
            ):
                all2all_manager.update_mask(ep_rank, masked=True)

            if (
                self.worker.parallel_config.enable_eplb
                and not self.worker.model_runner.eep_eplb_suppressed
            ):
                self._reinit_eplb_groups(store, round_key)
        if self.worker.parallel_config.tensor_parallel_size > 1:
            # The per-step TP barrier (dp_utils) leaves the group in a
            # timed-out state on fault; rebuild it for a clean slate.
            tp_group = get_tp_group()
            # Pipeline-stage recovery ports: each stage owns a distinct TP group.
            pp_rank = get_pp_group().rank_in_group
            _reinit_cpu_group(
                tp_group,
                store,
                f"ft_tp_port_{round_key}_{dp_group_rank}_{pp_rank}",
                tp_group.rank_in_group,
                tp_group.world_size,
                timeout=self.worker.parallel_config.cpu_distributed_timeout,
            )

    def scale_down(self, ft_request: FaultToleranceRequest):
        model_runner = self.worker.model_runner
        eplb_config = self.worker.parallel_config.eplb_config
        if model_runner.eplb_state is None or eplb_config.num_redundant_experts <= 0:
            raise ValueError(
                "[FT] scale_down requires EPLB with num_redundant_experts > 0 "
                "to re-host the dead rank's experts."
            )
        tp_size = self.worker.parallel_config.tensor_parallel_size
        pcp_size = self.worker.parallel_config.prefill_context_parallel_size
        dead_dp_ranks = ft_request.params["dead_dp_ranks"]
        dead_ep_ranks = compute_dead_ep_ranks(dead_dp_ranks, tp_size, pcp_size)
        # Suppress EPLB async rebalancing before retry so it skips the EPLB
        # group reinit; placement is fixed by the redistribution below.
        model_runner.eep_eplb_suppressed = True
        self.retry(ft_request)

        self._redistribute_experts(dead_ep_ranks)

        logger.info(
            "[FT] Worker scale_down complete: dp_group_size=%d, "
            "dp_group_rank=%d, dead_ep_ranks=%s, eplb_suppressed=True",
            ft_request.params["dp_group_size"],
            ft_request.params["dp_group_rank"],
            sorted(dead_ep_ranks),
        )

    def query_mask(self, ft_request: FaultToleranceRequest) -> dict:
        """Query the mask on a side stream."""
        with torch.cuda.stream(torch.cuda.Stream()):
            mask = get_ep_all2all_manager().query_active_mask()
            return {"mask": mask.tolist()}

    def _reinit_eplb_groups(self, store: TCPStore, round_key: str) -> None:
        """Reinit the EP/EPLB Gloo groups and refresh the EPLB
        communicator's cpu_group reference."""
        # Pipeline-stage recovery ports: EP/EPLB leaders must not share keys.
        pp_rank = get_pp_group().rank_in_group
        for key_suffix, get_group in [
            ("ep", get_ep_group),
            ("eplb", get_eplb_group),
        ]:
            group = get_group()
            _reinit_cpu_group(
                group,
                store,
                f"ft_{key_suffix}_port_{round_key}_{pp_rank}",
                group.rank_in_group,
                group.world_size,
                timeout=self.worker.parallel_config.cpu_distributed_timeout,
            )
            logger.info("[FT] Reinited %s Gloo group", key_suffix)

        eplb_state = self.worker.model_runner.eplb_state
        assert eplb_state is not None
        eplb_group = get_eplb_group()
        for ms in eplb_state.model_states.values():
            if hasattr(ms.communicator, "_cpu_group"):
                ms.communicator._cpu_group = eplb_group.cpu_group

    def _eplb_model_state(self):
        model_runner = self.worker.model_runner
        assert model_runner.eplb_state is not None
        return model_runner.eplb_state.model_states[
            model_runner.model_config.compute_hash()
        ]

    def _redistribute_experts(self, dead_ep_ranks: set[int]) -> None:
        """One-shot expert redistribution after scale-down."""
        model_runner = self.worker.model_runner
        eplb_model_state = self._eplb_model_state()

        p2l = eplb_model_state.physical_to_logical_map
        num_logical = eplb_model_state.logical_replica_count.shape[1]
        ep_world_size = get_ep_group().world_size
        num_local_experts = p2l.shape[1] // ep_world_size

        mark_dead_expert_slots_inplace(p2l, dead_ep_ranks, num_local_experts)
        reassignments = redistribute_expert_placement(
            p2l, num_logical, num_local_experts
        )
        # p2l was updated in place; the commit derives l2p/lrc from it.
        _commit_eplb_maps(eplb_model_state, p2l.cpu())
        # Preserve physical expert IDs: slot ownership is unchanged by EPLB;
        # only the logical expert stored in each surviving slot changes.

        if reassignments:
            reload_experts_from_disk(
                model_runner.model,
                self.worker.vllm_config,
                reassignments,
                p2l,
            )

        logger.info(
            "[FT] Expert redistribution: num_logical=%d, "
            "ep_world_size=%d, reassignments=%d",
            num_logical,
            ep_world_size,
            len(reassignments),
        )

    def _clean_worker_state(self):
        model_runner = self.worker.model_runner
        model_runner.execute_model_state = None
        runner = cast("GPUModelRunnerV2", model_runner)
        for req_id in list(runner.req_states.req_id_to_index):
            runner._remove_request(req_id)

_redistribute_experts(dead_ep_ranks)

One-shot expert redistribution after scale-down.

Source code in vllm/v1/worker/sentinel/gpu_worker_sentinel.py

def _redistribute_experts(self, dead_ep_ranks: set[int]) -> None:
    """One-shot expert redistribution after scale-down."""
    model_runner = self.worker.model_runner
    eplb_model_state = self._eplb_model_state()

    p2l = eplb_model_state.physical_to_logical_map
    num_logical = eplb_model_state.logical_replica_count.shape[1]
    ep_world_size = get_ep_group().world_size
    num_local_experts = p2l.shape[1] // ep_world_size

    mark_dead_expert_slots_inplace(p2l, dead_ep_ranks, num_local_experts)
    reassignments = redistribute_expert_placement(
        p2l, num_logical, num_local_experts
    )
    # p2l was updated in place; the commit derives l2p/lrc from it.
    _commit_eplb_maps(eplb_model_state, p2l.cpu())
    # Preserve physical expert IDs: slot ownership is unchanged by EPLB;
    # only the logical expert stored in each surviving slot changes.

    if reassignments:
        reload_experts_from_disk(
            model_runner.model,
            self.worker.vllm_config,
            reassignments,
            p2l,
        )

    logger.info(
        "[FT] Expert redistribution: num_logical=%d, "
        "ep_world_size=%d, reassignments=%d",
        num_logical,
        ep_world_size,
        len(reassignments),
    )

_reinit_eplb_groups(store, round_key)

Reinit the EP/EPLB Gloo groups and refresh the EPLB communicator's cpu_group reference.

Source code in vllm/v1/worker/sentinel/gpu_worker_sentinel.py

def _reinit_eplb_groups(self, store: TCPStore, round_key: str) -> None:
    """Reinit the EP/EPLB Gloo groups and refresh the EPLB
    communicator's cpu_group reference."""
    # Pipeline-stage recovery ports: EP/EPLB leaders must not share keys.
    pp_rank = get_pp_group().rank_in_group
    for key_suffix, get_group in [
        ("ep", get_ep_group),
        ("eplb", get_eplb_group),
    ]:
        group = get_group()
        _reinit_cpu_group(
            group,
            store,
            f"ft_{key_suffix}_port_{round_key}_{pp_rank}",
            group.rank_in_group,
            group.world_size,
            timeout=self.worker.parallel_config.cpu_distributed_timeout,
        )
        logger.info("[FT] Reinited %s Gloo group", key_suffix)

    eplb_state = self.worker.model_runner.eplb_state
    assert eplb_state is not None
    eplb_group = get_eplb_group()
    for ms in eplb_state.model_states.values():
        if hasattr(ms.communicator, "_cpu_group"):
            ms.communicator._cpu_group = eplb_group.cpu_group

handle_command(ft_request)

Dispatch an FT command by instruction name.

Source code in vllm/v1/worker/sentinel/gpu_worker_sentinel.py

def handle_command(self, ft_request: FaultToleranceRequest):
    """Dispatch an FT command by instruction name."""
    with set_current_vllm_config(self.worker.vllm_config):
        return run_method(self, ft_request.instruction, (ft_request,), {})

query_mask(ft_request)

Query the mask on a side stream.

Source code in vllm/v1/worker/sentinel/gpu_worker_sentinel.py

def query_mask(self, ft_request: FaultToleranceRequest) -> dict:
    """Query the mask on a side stream."""
    with torch.cuda.stream(torch.cuda.Stream()):
        mask = get_ep_all2all_manager().query_active_mask()
        return {"mask": mask.tolist()}

_reinit_cpu_group(group, store, key, rank, size, timeout=None)

Rebuild a group's Gloo cpu_group over a port coordinated via store

Source code in vllm/v1/worker/sentinel/gpu_worker_sentinel.py

def _reinit_cpu_group(
    group: "GroupCoordinator",
    store: TCPStore,
    key: str,
    rank: int,
    size: int,
    timeout: timedelta | None = None,
) -> None:
    """Rebuild a group's Gloo cpu_group over a port coordinated via ``store``"""
    # Rendezvous recovery groups on their actual leader hosts: DP replicas and
    # pipeline stages can span nodes independently of the deployment master.
    if rank == 0:
        store.set(f"{key}_host", get_ip())
    host = store.get(f"{key}_host").decode()
    if rank == 0:
        ports, socks = allocate_group_ports(store, key, host, 1)
        port, listen_socket = ports[0], socks[0]
    else:
        port, listen_socket = fetch_group_ports(store, key, 1)[0], None
    new_group = stateless_init_torch_distributed_process_group(
        host,
        port,
        rank,
        size,
        backend="gloo",
        listen_socket=listen_socket,
        timeout=timeout,
    )
    stateless_destroy_torch_distributed_process_group(group.cpu_group)
    group.cpu_group = new_group