biondizzle/vllm
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Fix per file ruff ignores related to typing (#26254)

Browse Source 
Signed-off-by: Harry Mellor <19981378+hmellor@users.noreply.github.com>
This commit is contained in:
Harry Mellor
2025-10-05 17:37:55 +01:00
committed by GitHub
parent 5f317530ec
commit 1c0c68202c
32 changed files with 258 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
vllm/executor/executor_base.py
View File

@@ -6,7 +6,7 @@ import time
from abc import ABC, abstractmethod
from collections.abc import Awaitable
from functools import cached_property
from typing import Any, Callable, List, Optional, Set, Union
from typing import Any, Callable, Optional, Union
from typing_extensions import TypeVar
@@ -143,7 +143,7 @@ class ExecutorBase(ABC):
def execute_model(
self, execute_model_req: ExecuteModelRequest
) -> Optional[List[Union[SamplerOutput, PoolerOutput]]]:
) -> Optional[list[Union[SamplerOutput, PoolerOutput]]]:
output = self.collective_rpc("execute_model", args=(execute_model_req,))
return output[0]
@@ -163,7 +163,7 @@ class ExecutorBase(ABC):
assert lora_id > 0, "lora_id must be greater than 0."
return all(self.collective_rpc("pin_lora", args=(lora_id,)))
def list_loras(self) -> Set[int]:
def list_loras(self) -> set[int]:
sets = self.collective_rpc("list_loras")
for s in sets:
assert s == sets[0], "All workers should have the same LORAs."
@@ -238,7 +238,7 @@ class ExecutorBase(ABC):
async def execute_model_async(
self, execute_model_req: ExecuteModelRequest
) -> List[SamplerOutput]:
) -> list[SamplerOutput]:
"""Executes one model step on the given sequences."""
output = await make_async(self.execute_model)(execute_model_req)
return output
@@ -272,7 +272,7 @@ class DistributedExecutorBase(ExecutorBase):
def execute_model(
self,
execute_model_req: ExecuteModelRequest,
) -> List[SamplerOutput]:
) -> list[SamplerOutput]:
# TODO: unify into collective_rpc
if self.parallel_worker_tasks is None:
self.parallel_worker_tasks = self._run_workers(
@@ -299,7 +299,7 @@ class DistributedExecutorBase(ExecutorBase):
@abstractmethod
def _driver_execute_model(
self, execute_model_req: Optional[ExecuteModelRequest]
) -> Optional[List[SamplerOutput]]:
) -> Optional[list[SamplerOutput]]:
"""Run execute_model in the driver worker.
Passing None will cause the driver to stop the model execution loop
@@ -346,7 +346,7 @@ class DistributedExecutorBase(ExecutorBase):
async def execute_model_async(
self, execute_model_req: ExecuteModelRequest
) -> List[SamplerOutput]:
) -> list[SamplerOutput]:
if self.parallel_worker_tasks is None:
# Start model execution loop running in the parallel workers
self.parallel_worker_tasks = asyncio.create_task(
@@ -371,7 +371,7 @@ class DistributedExecutorBase(ExecutorBase):
async def _driver_execute_model_async(
self,
execute_model_req: Optional[ExecuteModelRequest] = None,
) -> List[SamplerOutput]:
) -> list[SamplerOutput]:
"""Execute the model asynchronously in the driver worker.
Passing None will cause the driver to stop the model execution

4
vllm/executor/msgspec_utils.py
View File

@@ -2,7 +2,7 @@
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from array import array
from typing import Any, Type
from typing import Any
from vllm.multimodal.inputs import MultiModalKwargs
from vllm.sequence import VLLM_TOKEN_ID_ARRAY_TYPE
@@ -23,7 +23,7 @@ def encode_hook(obj: Any) -> Any:
return dict(obj)
def decode_hook(type: Type, obj: Any) -> Any:
def decode_hook(type: type, obj: Any) -> Any:
"""Custom msgspec dec hook that supports array types and MultiModalKwargs.
See https://jcristharif.com/msgspec/api.html#msgspec.msgpack.Encoder

30
vllm/executor/ray_distributed_executor.py
View File

@@ -5,7 +5,7 @@ import asyncio
import os
from collections import defaultdict
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
from typing import TYPE_CHECKING, Any, Callable, Optional, Union
import cloudpickle
import msgspec
@@ -114,10 +114,10 @@ class RayDistributedExecutor(DistributedExecutorBase):
self._init_workers_ray(placement_group)
self.input_encoder = msgspec.msgpack.Encoder(enc_hook=encode_hook)
self.output_decoder = msgspec.msgpack.Decoder(Optional[List[SamplerOutput]])
self.output_decoder = msgspec.msgpack.Decoder(Optional[list[SamplerOutput]])
self.use_v1 = envs.VLLM_USE_V1
self.pp_locks: Optional[List[asyncio.Lock]] = None
self.pp_locks: Optional[list[asyncio.Lock]] = None
if not self.use_ray_compiled_dag:
self.driver_exec_method = make_async(self.driver_worker.execute_method)
@@ -137,7 +137,7 @@ class RayDistributedExecutor(DistributedExecutorBase):
ray.kill(worker)
self.forward_dag = None
def _configure_ray_workers_use_nsight(self, ray_remote_kwargs) -> Dict[str, Any]:
def _configure_ray_workers_use_nsight(self, ray_remote_kwargs) -> dict[str, Any]:
# If nsight profiling is enabled, we need to set the profiling
# configuration for the ray workers as runtime env.
runtime_env = ray_remote_kwargs.setdefault("runtime_env", {})
@@ -164,12 +164,12 @@ class RayDistributedExecutor(DistributedExecutorBase):
# It holds the resource for the driver worker.
self.driver_dummy_worker: Optional[RayWorkerWrapper] = None
# The remaining workers are the actual ray actors.
self.workers: List[RayWorkerWrapper] = []
self.workers: list[RayWorkerWrapper] = []
# Used in ray compiled DAG: indexed first by PP rank,
# and then TP rank. In other words, the inner list is
# the TP group of workers for a PP rank.
self.pp_tp_workers: List[List[RayWorkerWrapper]] = []
self.pp_tp_workers: list[list[RayWorkerWrapper]] = []
if self.parallel_config.ray_workers_use_nsight:
ray_remote_kwargs = self._configure_ray_workers_use_nsight(
@@ -179,7 +179,7 @@ class RayDistributedExecutor(DistributedExecutorBase):
logger.info("use_ray_spmd_worker: %s", self.use_ray_spmd_worker)
# Create the workers.
bundle_indices: List[int]
bundle_indices: list[int]
if envs.VLLM_RAY_BUNDLE_INDICES:
# Use the bundle indices specified by the user.
bundle_indices = list(map(int, envs.VLLM_RAY_BUNDLE_INDICES.split(",")))
@@ -200,7 +200,7 @@ class RayDistributedExecutor(DistributedExecutorBase):
bundle_indices.append(bundle_id)
bundle_indices = bundle_indices[: self.parallel_config.world_size]
worker_metadata: List[RayWorkerMetaData] = []
worker_metadata: list[RayWorkerMetaData] = []
driver_ip = get_ip()
for rank, bundle_id in enumerate(bundle_indices):
scheduling_strategy = PlacementGroupSchedulingStrategy(
@@ -262,7 +262,7 @@ class RayDistributedExecutor(DistributedExecutorBase):
"the driver on a GPU node."
)
ip_counts: Dict[str, int] = {}
ip_counts: dict[str, int] = {}
for ip in worker_ips:
ip_counts[ip] = ip_counts.get(ip, 0) + 1
@@ -416,11 +416,11 @@ class RayDistributedExecutor(DistributedExecutorBase):
# This is the list of workers that are rank 0 of each TP group EXCEPT
# global rank 0. These are the workers that will broadcast to the
# rest of the workers.
self.tp_driver_workers: List[RayWorkerWrapper] = []
self.tp_driver_workers: list[RayWorkerWrapper] = []
# This is the list of workers that are not drivers and not the first
# worker in a TP group. These are the workers that will be
# broadcasted to.
self.non_driver_workers: List[RayWorkerWrapper] = []
self.non_driver_workers: list[RayWorkerWrapper] = []
# Enforce rank order for correct rank to return final output.
for index, worker in enumerate(self.workers):
@@ -433,7 +433,7 @@ class RayDistributedExecutor(DistributedExecutorBase):
def _driver_execute_model(
self, execute_model_req: Optional[ExecuteModelRequest]
) -> Optional[List[SamplerOutput]]:
) -> Optional[list[SamplerOutput]]:
"""Run execute_model in the driver worker.
Passing None will cause the driver to stop the model execution
@@ -446,7 +446,7 @@ class RayDistributedExecutor(DistributedExecutorBase):
def execute_model(
self, execute_model_req: ExecuteModelRequest
) -> List[SamplerOutput]:
) -> list[SamplerOutput]:
if not self.use_ray_spmd_worker:
return super().execute_model(execute_model_req)
@@ -675,7 +675,7 @@ class RayDistributedExecutor(DistributedExecutorBase):
async def execute_model_async(
self, execute_model_req: ExecuteModelRequest
) -> List[SamplerOutput]:
) -> list[SamplerOutput]:
if not self.use_ray_spmd_worker:
return await super().execute_model_async(execute_model_req)
@@ -689,7 +689,7 @@ class RayDistributedExecutor(DistributedExecutorBase):
async def _driver_execute_model_async(
self, execute_model_req: Optional[ExecuteModelRequest] = None
) -> List[SamplerOutput]:
) -> list[SamplerOutput]:
assert not self.use_ray_spmd_worker, (
"driver_worker does not exist for VLLM_USE_RAY_SPMD_WORKER=1"
)

16
vllm/executor/ray_utils.py
View File

@@ -4,7 +4,7 @@
import os
import time
from collections import defaultdict
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
from typing import TYPE_CHECKING, Optional, Union
import msgspec
@@ -59,7 +59,7 @@ try:
def get_node_ip(self) -> str:
return get_ip()
def get_node_and_gpu_ids(self) -> Tuple[str, List[int]]:
def get_node_and_gpu_ids(self) -> tuple[str, list[int]]:
node_id = ray.get_runtime_context().get_node_id()
device_key = vllm.platforms.current_platform.ray_device_key
if not device_key:
@@ -72,7 +72,7 @@ try:
def execute_model_spmd(
self,
req_or_tuple: Union[bytes, Tuple[bytes, Optional[IntermediateTensors]]],
req_or_tuple: Union[bytes, tuple[bytes, Optional[IntermediateTensors]]],
) -> bytes:
"""Execute model in SPMD fashion: used only when SPMD worker and
compiled DAG are both enabled.
@@ -126,10 +126,10 @@ try:
def execute_model_ray(
self,
scheduler_output: Union[
"SchedulerOutput", Tuple["SchedulerOutput", "IntermediateTensors"]
"SchedulerOutput", tuple["SchedulerOutput", "IntermediateTensors"]
],
) -> Union[
"ModelRunnerOutput", Tuple["SchedulerOutput", "IntermediateTensors"]
"ModelRunnerOutput", tuple["SchedulerOutput", "IntermediateTensors"]
]:
# This method is used by Ray Compiled Graph to execute the model,
# and it needs a special logic of self.setup_device_if_necessary()
@@ -156,7 +156,7 @@ try:
output = output.get_output()
return output
def override_env_vars(self, vars: Dict[str, str]):
def override_env_vars(self, vars: dict[str, str]):
os.environ.update(vars)
ray_import_err = None
@@ -201,7 +201,7 @@ def _verify_bundles(
# bundle_idx -> bundle (e.g., {"GPU": 1})
bundles = pg_data["bundles"]
# node_id -> List of bundle (e.g., {"GPU": 1})
node_id_to_bundle: Dict[str, List[Dict[str, float]]] = defaultdict(list)
node_id_to_bundle: dict[str, list[dict[str, float]]] = defaultdict(list)
for bundle_idx, node_id in bundle_to_node_ids.items():
node_id_to_bundle[node_id].append(bundles[bundle_idx])
@@ -383,7 +383,7 @@ def initialize_ray_cluster(
device_str,
)
# Create a new placement group
placement_group_specs: List[Dict[str, float]] = [
placement_group_specs: list[dict[str, float]] = [
{device_str: 1.0} for _ in range(parallel_config.world_size)
]

10
vllm/executor/uniproc_executor.py
View File

@@ -4,7 +4,7 @@ import os
from concurrent.futures import Future, ThreadPoolExecutor
from functools import cached_property
from multiprocessing import Lock
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
from typing import Any, Callable, Optional, Union
import torch
import torch.distributed as dist
@@ -68,10 +68,10 @@ class UniProcExecutor(ExecutorBase):
self,
method: Union[str, Callable],
timeout: Optional[float] = None,
args: Tuple = (),
kwargs: Optional[Dict] = None,
args: tuple = (),
kwargs: Optional[dict] = None,
non_block: bool = False,
) -> List[Any]:
) -> list[Any]:
if kwargs is None:
kwargs = {}
if self.mm_receiver_cache is not None and method == "execute_model":
@@ -158,7 +158,7 @@ class ExecutorWithExternalLauncher(UniProcExecutor):
local_rank = int(os.environ["LOCAL_RANK"])
return distributed_init_method, rank, local_rank
def determine_num_available_blocks(self) -> Tuple[int, int]:
def determine_num_available_blocks(self) -> tuple[int, int]:
"""
Determine the number of available KV blocks.
Add an additional all_reduce to get the min across all ranks.