Skip to content

vllm.v1.engine.utils

STARTUP_POLL_PERIOD_MS module-attribute

STARTUP_POLL_PERIOD_MS = 10000

logger module-attribute

logger = init_logger(__name__)

CoreEngine

One per data parallel rank, used to track state during handshaking.

Source code in vllm/v1/engine/utils.py
class CoreEngine:
    """One per data parallel rank, used to track state during handshaking."""

    def __init__(self, index: int = 0, local: bool = True):
        self.local = local
        self.identity = index.to_bytes(2, "little")

        self.state = CoreEngineState.NEW

identity instance-attribute

identity = to_bytes(2, 'little')

local instance-attribute

local = local

state instance-attribute

state = NEW

__init__

__init__(index: int = 0, local: bool = True)
Source code in vllm/v1/engine/utils.py
def __init__(self, index: int = 0, local: bool = True):
    self.local = local
    self.identity = index.to_bytes(2, "little")

    self.state = CoreEngineState.NEW

CoreEngineActorManager

Utility class to handle creation, readiness, and shutdown of core engine Ray actors used by the AsyncLLM and LLMEngine.

Different from CoreEngineProcManager, this class manages core engines for both local and remote nodes.

Source code in vllm/v1/engine/utils.py
class CoreEngineActorManager:
    """
    Utility class to handle creation, readiness, and shutdown
    of core engine Ray actors used by the AsyncLLM and LLMEngine.

    Different from CoreEngineProcManager, this class manages
    core engines for both local and remote nodes.
    """

    def __init__(
        self,
        vllm_config: VllmConfig,
        addresses: EngineZmqAddresses,
        executor_class: type[Executor],
        log_stats: bool,
        placement_groups: Optional[list["PlacementGroup"]] = None,
        local_dp_ranks: Optional[list[int]] = None,
    ):
        import copy

        import ray
        from ray.util.scheduling_strategies import (
            PlacementGroupSchedulingStrategy)

        from vllm.v1.engine.core import DPEngineCoreActor

        self.local_engine_actors: list[ray.ActorHandle] = []
        self.remote_engine_actors: list[ray.ActorHandle] = []
        dp_size = vllm_config.parallel_config.data_parallel_size
        local_engine_count = \
            vllm_config.parallel_config.data_parallel_size_local
        world_size = vllm_config.parallel_config.world_size

        if ray.is_initialized():
            logger.info(
                "Ray is already initialized. Skipping Ray initialization.")
        else:
            ray.init()

        if placement_groups is not None:
            assert local_dp_ranks is not None, (
                "local_dp_ranks must be provided if "
                "placement_groups is provided")
            assert len(placement_groups) == len(local_dp_ranks), (
                "placement_groups and local_dp_ranks must "
                "have the same length")
            logger.info("Using provided placement groups")
            # TODO(rui): validate passed-in placement groups
            self.created_placement_groups = []
        else:
            placement_groups, local_dp_ranks = \
                CoreEngineActorManager.create_dp_placement_groups(vllm_config)
            self.created_placement_groups = placement_groups
        assert len(placement_groups) == dp_size, (
            "Number of placement groups must match data parallel size")

        refs = []
        for index in range(dp_size):
            local_index = local_dp_ranks[index]
            dp_vllm_config = copy.deepcopy(vllm_config)
            pg = placement_groups[index]
            dp_vllm_config.parallel_config.placement_group = pg
            local_client = index < local_engine_count
            actor = ray.remote(DPEngineCoreActor).options(
                scheduling_strategy=PlacementGroupSchedulingStrategy(
                    placement_group=pg,
                    placement_group_bundle_index=world_size,
                )).remote(vllm_config=dp_vllm_config,
                          executor_class=executor_class,
                          log_stats=log_stats,
                          local_client=local_client,
                          addresses=addresses,
                          dp_rank=index,
                          local_dp_rank=local_index)
            if local_client:
                self.local_engine_actors.append(actor)
            else:
                self.remote_engine_actors.append(actor)
            refs.append(actor.wait_for_init.remote())

        ray.get(refs)
        self.run_refs = []
        for actor in self.local_engine_actors + self.remote_engine_actors:
            self.run_refs.append(actor.run.remote())

    @staticmethod
    def create_dp_placement_groups(
            vllm_config: VllmConfig
    ) -> tuple[list["PlacementGroup"], list[int]]:

        import ray
        from ray._private.state import available_resources_per_node
        from ray.util.state import list_nodes

        logger.info("Creating placement groups for data parallel")
        dp_master_ip = \
            vllm_config.parallel_config.data_parallel_master_ip
        dp_size = vllm_config.parallel_config.data_parallel_size
        local_engine_count = \
            vllm_config.parallel_config.data_parallel_size_local

        nodes = sorted(list_nodes(),
                       key=lambda node: node.node_ip != dp_master_ip)
        assert nodes[0].node_ip == dp_master_ip, (
            "The first node must be the head node")
        assert len(nodes) == 1 or nodes[1].node_ip != dp_master_ip, (
            "There can only be one head node")

        available_resources = available_resources_per_node()
        world_size = vllm_config.parallel_config.world_size
        placement_groups: list[PlacementGroup] = []
        local_dp_ranks: list[int] = []

        for node in nodes:
            node_ip = node.node_ip
            node_resources = available_resources[node.node_id]
            # For now, each DP rank can only be assigned to one node
            # TODO(rui): support allocating a single DP rank
            # to multiple nodes
            available_engine_count = int(node_resources["GPU"]) // world_size
            if node_ip == dp_master_ip:
                assert available_engine_count >= local_engine_count, (
                    "Not enough resources to allocate DP ranks "
                    f"on DP master node {node_ip}")
                for i in range(local_engine_count):
                    bundles = [{
                        "GPU": 1.0,
                        "node:" + dp_master_ip: 0.001
                    }] * world_size + [{
                        "CPU": 1.0
                    }]
                    pg = ray.util.placement_group(
                        name=f"dp_rank_{len(placement_groups)}",
                        strategy="STRICT_PACK",
                        bundles=bundles,
                    )
                    placement_groups.append(pg)
                    local_dp_ranks.append(i)
            else:
                for i in range(available_engine_count):
                    if len(placement_groups) == dp_size:
                        break
                    bundles = [{"GPU": 1.0}] * world_size + [{"CPU": 1.0}]
                    pg = ray.util.placement_group(
                        name=f"dp_rank_{len(placement_groups)}",
                        strategy="STRICT_PACK",
                        bundles=bundles,
                    )
                    placement_groups.append(pg)
                    local_dp_ranks.append(i)
        return placement_groups, local_dp_ranks

    def get_run_refs(self):
        return self.run_refs

    def close(self):
        import ray
        for actor in self.local_engine_actors + self.remote_engine_actors:
            ray.kill(actor)
        for pg in self.created_placement_groups:
            ray.util.remove_placement_group(pg)

created_placement_groups instance-attribute

created_placement_groups = []

local_engine_actors instance-attribute

local_engine_actors: list[ActorHandle] = []

remote_engine_actors instance-attribute

remote_engine_actors: list[ActorHandle] = []

run_refs instance-attribute

run_refs = []

__init__

__init__(
    vllm_config: VllmConfig,
    addresses: EngineZmqAddresses,
    executor_class: type[Executor],
    log_stats: bool,
    placement_groups: Optional[list[PlacementGroup]] = None,
    local_dp_ranks: Optional[list[int]] = None,
)
Source code in vllm/v1/engine/utils.py
def __init__(
    self,
    vllm_config: VllmConfig,
    addresses: EngineZmqAddresses,
    executor_class: type[Executor],
    log_stats: bool,
    placement_groups: Optional[list["PlacementGroup"]] = None,
    local_dp_ranks: Optional[list[int]] = None,
):
    import copy

    import ray
    from ray.util.scheduling_strategies import (
        PlacementGroupSchedulingStrategy)

    from vllm.v1.engine.core import DPEngineCoreActor

    self.local_engine_actors: list[ray.ActorHandle] = []
    self.remote_engine_actors: list[ray.ActorHandle] = []
    dp_size = vllm_config.parallel_config.data_parallel_size
    local_engine_count = \
        vllm_config.parallel_config.data_parallel_size_local
    world_size = vllm_config.parallel_config.world_size

    if ray.is_initialized():
        logger.info(
            "Ray is already initialized. Skipping Ray initialization.")
    else:
        ray.init()

    if placement_groups is not None:
        assert local_dp_ranks is not None, (
            "local_dp_ranks must be provided if "
            "placement_groups is provided")
        assert len(placement_groups) == len(local_dp_ranks), (
            "placement_groups and local_dp_ranks must "
            "have the same length")
        logger.info("Using provided placement groups")
        # TODO(rui): validate passed-in placement groups
        self.created_placement_groups = []
    else:
        placement_groups, local_dp_ranks = \
            CoreEngineActorManager.create_dp_placement_groups(vllm_config)
        self.created_placement_groups = placement_groups
    assert len(placement_groups) == dp_size, (
        "Number of placement groups must match data parallel size")

    refs = []
    for index in range(dp_size):
        local_index = local_dp_ranks[index]
        dp_vllm_config = copy.deepcopy(vllm_config)
        pg = placement_groups[index]
        dp_vllm_config.parallel_config.placement_group = pg
        local_client = index < local_engine_count
        actor = ray.remote(DPEngineCoreActor).options(
            scheduling_strategy=PlacementGroupSchedulingStrategy(
                placement_group=pg,
                placement_group_bundle_index=world_size,
            )).remote(vllm_config=dp_vllm_config,
                      executor_class=executor_class,
                      log_stats=log_stats,
                      local_client=local_client,
                      addresses=addresses,
                      dp_rank=index,
                      local_dp_rank=local_index)
        if local_client:
            self.local_engine_actors.append(actor)
        else:
            self.remote_engine_actors.append(actor)
        refs.append(actor.wait_for_init.remote())

    ray.get(refs)
    self.run_refs = []
    for actor in self.local_engine_actors + self.remote_engine_actors:
        self.run_refs.append(actor.run.remote())

close

close()
Source code in vllm/v1/engine/utils.py
def close(self):
    import ray
    for actor in self.local_engine_actors + self.remote_engine_actors:
        ray.kill(actor)
    for pg in self.created_placement_groups:
        ray.util.remove_placement_group(pg)

create_dp_placement_groups staticmethod

create_dp_placement_groups(
    vllm_config: VllmConfig,
) -> tuple[list[PlacementGroup], list[int]]
Source code in vllm/v1/engine/utils.py
@staticmethod
def create_dp_placement_groups(
        vllm_config: VllmConfig
) -> tuple[list["PlacementGroup"], list[int]]:

    import ray
    from ray._private.state import available_resources_per_node
    from ray.util.state import list_nodes

    logger.info("Creating placement groups for data parallel")
    dp_master_ip = \
        vllm_config.parallel_config.data_parallel_master_ip
    dp_size = vllm_config.parallel_config.data_parallel_size
    local_engine_count = \
        vllm_config.parallel_config.data_parallel_size_local

    nodes = sorted(list_nodes(),
                   key=lambda node: node.node_ip != dp_master_ip)
    assert nodes[0].node_ip == dp_master_ip, (
        "The first node must be the head node")
    assert len(nodes) == 1 or nodes[1].node_ip != dp_master_ip, (
        "There can only be one head node")

    available_resources = available_resources_per_node()
    world_size = vllm_config.parallel_config.world_size
    placement_groups: list[PlacementGroup] = []
    local_dp_ranks: list[int] = []

    for node in nodes:
        node_ip = node.node_ip
        node_resources = available_resources[node.node_id]
        # For now, each DP rank can only be assigned to one node
        # TODO(rui): support allocating a single DP rank
        # to multiple nodes
        available_engine_count = int(node_resources["GPU"]) // world_size
        if node_ip == dp_master_ip:
            assert available_engine_count >= local_engine_count, (
                "Not enough resources to allocate DP ranks "
                f"on DP master node {node_ip}")
            for i in range(local_engine_count):
                bundles = [{
                    "GPU": 1.0,
                    "node:" + dp_master_ip: 0.001
                }] * world_size + [{
                    "CPU": 1.0
                }]
                pg = ray.util.placement_group(
                    name=f"dp_rank_{len(placement_groups)}",
                    strategy="STRICT_PACK",
                    bundles=bundles,
                )
                placement_groups.append(pg)
                local_dp_ranks.append(i)
        else:
            for i in range(available_engine_count):
                if len(placement_groups) == dp_size:
                    break
                bundles = [{"GPU": 1.0}] * world_size + [{"CPU": 1.0}]
                pg = ray.util.placement_group(
                    name=f"dp_rank_{len(placement_groups)}",
                    strategy="STRICT_PACK",
                    bundles=bundles,
                )
                placement_groups.append(pg)
                local_dp_ranks.append(i)
    return placement_groups, local_dp_ranks

get_run_refs

get_run_refs()
Source code in vllm/v1/engine/utils.py
def get_run_refs(self):
    return self.run_refs

CoreEngineProcManager

Utility class to handle creation, readiness, and shutdown of background processes used by the AsyncLLM and LLMEngine.

Source code in vllm/v1/engine/utils.py
class CoreEngineProcManager:
    """
    Utility class to handle creation, readiness, and shutdown
    of background processes used by the AsyncLLM and LLMEngine.
    """

    def __init__(
        self,
        target_fn: Callable,
        local_engine_count: int,
        start_index: int,
        local_start_index: int,
        vllm_config: VllmConfig,
        local_client: bool,
        handshake_address: str,
        executor_class: type[Executor],
        log_stats: bool,
        client_handshake_address: Optional[str] = None,
    ):
        context = get_mp_context()
        common_kwargs = {
            "vllm_config": vllm_config,
            "local_client": local_client,
            "handshake_address": handshake_address,
            "executor_class": executor_class,
            "log_stats": log_stats,
        }

        if client_handshake_address:
            common_kwargs[
                "client_handshake_address"] = client_handshake_address

        self.processes: list[BaseProcess] = []
        for index in range(local_engine_count):
            local_index = local_start_index + index
            global_index = start_index + index
            # Start EngineCore in background process.
            self.processes.append(
                context.Process(target=target_fn,
                                name=f"EngineCore_{global_index}",
                                kwargs=common_kwargs | {
                                    "dp_rank": global_index,
                                    "local_dp_rank": local_index,
                                }))

        self._finalizer = weakref.finalize(self, shutdown, self.processes)
        try:
            for proc in self.processes:
                proc.start()
        finally:
            # Kill other procs if not all are running.
            if self.finished_procs():
                self.close()

    def close(self):
        """Shutdown all procs."""
        self._finalizer()

    def join_first(self):
        """Wait for any process to exit."""
        connection.wait(proc.sentinel for proc in self.processes)

    def sentinels(self) -> list:
        return [proc.sentinel for proc in self.processes]

    def finished_procs(self) -> dict[str, int]:
        """Returns dict of proc name -> exit code for any finished procs."""
        return {
            proc.name: proc.exitcode
            for proc in self.processes if proc.exitcode is not None
        }

_finalizer instance-attribute

_finalizer = finalize(self, shutdown, processes)

processes instance-attribute

processes: list[BaseProcess] = []

__init__

__init__(
    target_fn: Callable,
    local_engine_count: int,
    start_index: int,
    local_start_index: int,
    vllm_config: VllmConfig,
    local_client: bool,
    handshake_address: str,
    executor_class: type[Executor],
    log_stats: bool,
    client_handshake_address: Optional[str] = None,
)
Source code in vllm/v1/engine/utils.py
def __init__(
    self,
    target_fn: Callable,
    local_engine_count: int,
    start_index: int,
    local_start_index: int,
    vllm_config: VllmConfig,
    local_client: bool,
    handshake_address: str,
    executor_class: type[Executor],
    log_stats: bool,
    client_handshake_address: Optional[str] = None,
):
    context = get_mp_context()
    common_kwargs = {
        "vllm_config": vllm_config,
        "local_client": local_client,
        "handshake_address": handshake_address,
        "executor_class": executor_class,
        "log_stats": log_stats,
    }

    if client_handshake_address:
        common_kwargs[
            "client_handshake_address"] = client_handshake_address

    self.processes: list[BaseProcess] = []
    for index in range(local_engine_count):
        local_index = local_start_index + index
        global_index = start_index + index
        # Start EngineCore in background process.
        self.processes.append(
            context.Process(target=target_fn,
                            name=f"EngineCore_{global_index}",
                            kwargs=common_kwargs | {
                                "dp_rank": global_index,
                                "local_dp_rank": local_index,
                            }))

    self._finalizer = weakref.finalize(self, shutdown, self.processes)
    try:
        for proc in self.processes:
            proc.start()
    finally:
        # Kill other procs if not all are running.
        if self.finished_procs():
            self.close()

close

close()

Shutdown all procs.

Source code in vllm/v1/engine/utils.py
def close(self):
    """Shutdown all procs."""
    self._finalizer()

finished_procs

finished_procs() -> dict[str, int]

Returns dict of proc name -> exit code for any finished procs.

Source code in vllm/v1/engine/utils.py
def finished_procs(self) -> dict[str, int]:
    """Returns dict of proc name -> exit code for any finished procs."""
    return {
        proc.name: proc.exitcode
        for proc in self.processes if proc.exitcode is not None
    }

join_first

join_first()

Wait for any process to exit.

Source code in vllm/v1/engine/utils.py
def join_first(self):
    """Wait for any process to exit."""
    connection.wait(proc.sentinel for proc in self.processes)

sentinels

sentinels() -> list
Source code in vllm/v1/engine/utils.py
def sentinels(self) -> list:
    return [proc.sentinel for proc in self.processes]

CoreEngineState

Bases: Enum

Source code in vllm/v1/engine/utils.py
class CoreEngineState(Enum):
    NEW = auto()
    CONNECTED = auto()
    READY = auto()

CONNECTED class-attribute instance-attribute

CONNECTED = auto()

NEW class-attribute instance-attribute

NEW = auto()

READY class-attribute instance-attribute

READY = auto()

EngineHandshakeMetadata dataclass

Metadata sent to each engine process during startup handshake, including addresses of the front-end ZMQ queues that they should connect to.

Source code in vllm/v1/engine/utils.py
@dataclass
class EngineHandshakeMetadata:
    """Metadata sent to each engine process during startup handshake,
    including addresses of the front-end ZMQ queues that they should
    connect to.
    """
    addresses: EngineZmqAddresses
    parallel_config: dict[str, Union[int, str]]

addresses instance-attribute

addresses: EngineZmqAddresses

parallel_config instance-attribute

parallel_config: dict[str, Union[int, str]]

__init__

__init__(
    addresses: EngineZmqAddresses,
    parallel_config: dict[str, Union[int, str]],
) -> None

EngineZmqAddresses dataclass

Source code in vllm/v1/engine/utils.py
@dataclass
class EngineZmqAddresses:
    # ZMQ input socket addresses for each front-end client (requests)
    inputs: list[str]
    # ZMQ output socket addresses for each front-end client (responses)
    outputs: list[str]
    # ZMQ input socket address of DP coordinator if applicable
    coordinator_input: Optional[str] = None
    # ZMQ output socket address of DP coordinator if applicable
    coordinator_output: Optional[str] = None
    # ZMQ socket for front-end to connect to DP coordinator.
    # Not used by engine, just relayed to front-end in handshake response.
    # Only required for external DP LB case.
    frontend_stats_publish_address: Optional[str] = None

coordinator_input class-attribute instance-attribute

coordinator_input: Optional[str] = None

coordinator_output class-attribute instance-attribute

coordinator_output: Optional[str] = None

frontend_stats_publish_address class-attribute instance-attribute

frontend_stats_publish_address: Optional[str] = None

inputs instance-attribute

inputs: list[str]

outputs instance-attribute

outputs: list[str]

__init__

__init__(
    inputs: list[str],
    outputs: list[str],
    coordinator_input: Optional[str] = None,
    coordinator_output: Optional[str] = None,
    frontend_stats_publish_address: Optional[str] = None,
) -> None

launch_core_engines

launch_core_engines(
    vllm_config: VllmConfig,
    executor_class: type[Executor],
    log_stats: bool,
    num_api_servers: int = 1,
) -> Iterator[
    tuple[
        Optional[
            Union[
                CoreEngineProcManager,
                CoreEngineActorManager,
            ]
        ],
        Optional[DPCoordinator],
        EngineZmqAddresses,
    ]
]

Launch engine and DP coordinator processes as needed.

Source code in vllm/v1/engine/utils.py
@contextlib.contextmanager
def launch_core_engines(
    vllm_config: VllmConfig,
    executor_class: type[Executor],
    log_stats: bool,
    num_api_servers: int = 1,
) -> Iterator[tuple[
        Optional[Union[CoreEngineProcManager, CoreEngineActorManager]],
        Optional[DPCoordinator],
        EngineZmqAddresses,
]]:
    """Launch engine and DP coordinator processes as needed."""

    parallel_config = vllm_config.parallel_config
    dp_size = parallel_config.data_parallel_size
    local_engine_count = parallel_config.data_parallel_size_local
    local_start_index = parallel_config.data_parallel_rank_local
    dp_rank = parallel_config.data_parallel_rank
    host = parallel_config.data_parallel_master_ip
    external_dp_lb = parallel_config.data_parallel_external_lb

    # In offline mode there is an LLM instance per DP rank and
    # one core engine per LLM, see
    # examples/offline_inference/data_parallel.py.
    offline_mode = local_start_index is not None

    # client_local_only = True for cases where this front-end
    # sends requests only to colocated engines.
    client_local_only = offline_mode or external_dp_lb or (local_engine_count
                                                           == dp_size)

    # Set up input and output addresses.
    addresses = EngineZmqAddresses(
        inputs=[
            get_engine_client_zmq_addr(client_local_only, host)
            for _ in range(num_api_servers)
        ],
        outputs=[
            get_engine_client_zmq_addr(client_local_only, host)
            for _ in range(num_api_servers)
        ],
    )

    # Run the DP Coordinator process with rank 0 when in
    # online DP mode.
    run_coordinator = dp_size > 1 and not offline_mode and dp_rank == 0

    if run_coordinator:
        coordinator = DPCoordinator(parallel_config)

        addresses.coordinator_input, addresses.coordinator_output = (
            coordinator.get_engine_socket_addresses())
        addresses.frontend_stats_publish_address = (
            coordinator.get_stats_publish_address())

        logger.info("Started DP Coordinator process (PID: %d)",
                    coordinator.proc.pid)
    else:
        coordinator = None

    if parallel_config.data_parallel_backend == "ray":
        logger.info("Starting ray-based data parallel backend")

        engine_actor_manager = CoreEngineActorManager(
            vllm_config=vllm_config,
            addresses=addresses,
            executor_class=executor_class,
            log_stats=log_stats,
        )

        yield engine_actor_manager, coordinator, addresses
        return

    if offline_mode or (external_dp_lb and dp_rank > 0):
        assert local_engine_count == 1
        engines_to_handshake = [CoreEngine(index=dp_rank, local=True)]
    else:
        engines_to_handshake = [
            CoreEngine(index=i, local=(i < local_engine_count))
            for i in range(dp_size)
        ]

    # Whether the started engines will handshake only with co-located
    # front-end processes. In external_dp_lb mode, ranks > 0 handshake with
    # their co-located frontend and also the rank 0 front-end, and hence this
    # will be False.
    handshake_local_only = offline_mode or local_engine_count == dp_size

    handshake_address = get_engine_client_zmq_addr(
        handshake_local_only, host, parallel_config.data_parallel_rpc_port)

    if external_dp_lb and dp_rank > 0:
        assert not handshake_local_only
        local_handshake_address = get_open_zmq_ipc_path()
        client_handshake_address = local_handshake_address
    else:
        local_handshake_address = handshake_address
        client_handshake_address = None

    with zmq_socket_ctx(local_handshake_address, zmq.ROUTER,
                        bind=True) as handshake_socket:

        from vllm.v1.engine.core import EngineCoreProc

        # Start local engines.
        if local_engine_count:
            # In server mode, start_index and local_start_index will
            # both be 0.
            local_engine_manager = CoreEngineProcManager(
                EngineCoreProc.run_engine_core,
                vllm_config=vllm_config,
                executor_class=executor_class,
                log_stats=log_stats,
                handshake_address=handshake_address,
                client_handshake_address=client_handshake_address,
                local_client=True,
                local_engine_count=local_engine_count,
                start_index=dp_rank,
                local_start_index=local_start_index or 0)
        else:
            local_engine_manager = None

        yield local_engine_manager, coordinator, addresses

        # Now wait for engines to start.
        wait_for_engine_startup(
            handshake_socket,
            addresses,
            engines_to_handshake,
            parallel_config,
            vllm_config.cache_config,
            local_engine_manager,
            coordinator.proc if coordinator else None,
        )

wait_for_engine_startup

wait_for_engine_startup(
    handshake_socket: Socket,
    addresses: EngineZmqAddresses,
    core_engines: list[CoreEngine],
    parallel_config: ParallelConfig,
    cache_config: CacheConfig,
    proc_manager: Optional[CoreEngineProcManager],
    coord_process: Optional[Process],
)
Source code in vllm/v1/engine/utils.py
def wait_for_engine_startup(
    handshake_socket: zmq.Socket,
    addresses: EngineZmqAddresses,
    core_engines: list[CoreEngine],
    parallel_config: ParallelConfig,
    cache_config: CacheConfig,
    proc_manager: Optional[CoreEngineProcManager],
    coord_process: Optional[Process],
):
    # Wait for engine core process(es) to send ready messages.
    local_count = parallel_config.data_parallel_size_local
    remote_count = len(core_engines) - local_count
    # [local, remote] counts
    conn_pending, start_pending = [local_count, remote_count], [0, 0]
    poller = zmq.Poller()
    poller.register(handshake_socket, zmq.POLLIN)

    if proc_manager is not None:
        for sentinel in proc_manager.sentinels():
            poller.register(sentinel, zmq.POLLIN)
    if coord_process is not None:
        poller.register(coord_process.sentinel, zmq.POLLIN)
    while any(conn_pending) or any(start_pending):
        events = poller.poll(STARTUP_POLL_PERIOD_MS)
        if not events:
            if any(conn_pending):
                logger.debug(
                    "Waiting for %d local, %d remote core engine proc(s) "
                    "to connect.", *conn_pending)
            if any(start_pending):
                logger.debug(
                    "Waiting for %d local, %d remote core engine proc(s) "
                    "to start.", *start_pending)
            continue
        if len(events) > 1 or events[0][0] != handshake_socket:
            # One of the local core processes exited.
            finished = proc_manager.finished_procs() if proc_manager else {}
            if coord_process is not None and coord_process.exitcode is not None:
                finished[coord_process.name] = coord_process.exitcode
            raise RuntimeError("Engine core initialization failed. "
                               "See root cause above. "
                               f"Failed core proc(s): {finished}")

        # Receive HELLO and READY messages from the input socket.
        eng_identity, ready_msg_bytes = handshake_socket.recv_multipart()
        eng_index = int.from_bytes(eng_identity, "little")
        engine = next((e for e in core_engines if e.identity == eng_identity),
                      None)
        if engine is None:
            raise RuntimeError(f"Message from engine with unexpected data "
                               f"parallel rank: {eng_index}")
        msg = msgspec.msgpack.decode(ready_msg_bytes)
        status, local = msg["status"], msg["local"]
        if local != engine.local:
            raise RuntimeError(f"{status} message from "
                               f"{'local' if local else 'remote'} "
                               f"engine {eng_index}, expected it to be "
                               f"{'local' if engine.local else 'remote'}")

        if status == "HELLO" and engine.state == CoreEngineState.NEW:

            # Send init message with DP config info.
            init_message = msgspec.msgpack.encode(
                EngineHandshakeMetadata(
                    addresses=addresses,
                    parallel_config={
                        "data_parallel_master_ip":
                        parallel_config.data_parallel_master_ip,
                        "data_parallel_master_port":
                        parallel_config.data_parallel_master_port,
                        "data_parallel_size":
                        parallel_config.data_parallel_size,
                    }))
            handshake_socket.send_multipart((eng_identity, init_message),
                                            copy=False)
            conn_pending[0 if local else 1] -= 1
            start_pending[0 if local else 1] += 1
            engine.state = CoreEngineState.CONNECTED
        elif status == "READY" and engine.state == CoreEngineState.CONNECTED:
            # Setup KV cache config with initialization state from
            # engine core process. Sum values from all engines in DP case.
            num_gpu_blocks = cache_config.num_gpu_blocks or 0
            num_gpu_blocks += msg["num_gpu_blocks"]
            cache_config.num_gpu_blocks = num_gpu_blocks

            # In external DP LB mode, the coordinator address that the
            # front-end procs connect to is obtained from rank 0 via
            # one of the engine handshakes, and passed to the local
            # front-end process in the response from the other.
            if addresses.frontend_stats_publish_address is None:
                addresses.frontend_stats_publish_address = msg.get(
                    "dp_stats_address")

            start_pending[0 if local else 1] -= 1
            engine.state = CoreEngineState.READY
        else:
            raise RuntimeError(f"Unexpected {status} message for "
                               f"{'local' if local else 'remote'} engine "
                               f"{eng_index} in {engine.state} state.")

        logger.debug("%s from %s core engine process %s.", status,
                     "local" if local else "remote", eng_index)