MOVZXandHinata
61fbe0b865
A symbol scan over every tracked file found 123 top-level definitions in vllm/ with zero references, including references inside strings. All of them came from features this fork deleted earlier. - Delete dead attention helpers: MLADCPManager, split_prefill_chunks, the reshape helpers, compute_varlen_chunk_metadata, the LINEAR backend's unused BailingLinearAttentionBackend, and the sparse flash-attn entry points. - Delete the CPU and AMX op wrappers from _custom_ops.py: cpu_mla_decode, cpu_mla_extend, cpu_mla_get_scheduler_metadata, gather_mla_context_cache_cpu. - Delete dead quantization helpers: awq_pack, gptq_pack, gptq_quantize_weights, get_and_maybe_dequant_weights, marlin_quant_fp8_torch, the rand_marlin_weight_*_like pair, ocp_mx_weight_dtype_and_rows, get_fp8_block_weight_scale, input_to_float8. - Delete dead model helpers: resolve_visual_encoder_outputs, run_dp_sharded_vision_model, fast_topk, get_layer_index, is_pp_missing_parameter, skip_spec_layers, KimiK3MergedQKVGateLinear, selective_scan_fn, causal_conv1d_update_torch. - Delete dead utils: format_kib, format_mib, chunk_list, prev_power_of_2, split_host_port, join_host_port, make_zmq_path, make_async_with_semaphore, is_meta_module, is_non_overlapping_and_dense, supports_xpu_graph, create_kv_caches_with_random*, and the unused has_* capability probes. - Delete the unreachable deep-gemm mega-MoE warmup path, which imported vllm.models.deepseek_v4.nvidia.model and vllm.models.kimi_k3.nvidia.model, neither of which exists anymore. - Drop the imports and the two now-empty guards left behind. Verified statically: every changed file compiles, no unused import remains, and no deleted name is referenced in vllm, tests or csrc. The runtime gates still need the Python environment rebuilt. Co-authored-by: Hinata <hinata@idihore.id> Signed-off-by: MOVZX <movzx@yahoo.com>
118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
|
"""Contains helpers related to asynchronous code.
|
|
|
|
This is similar in concept to the `asyncio` module.
|
|
"""
|
|
|
|
import asyncio
|
|
import contextlib
|
|
from asyncio import FIRST_COMPLETED, AbstractEventLoop, Future, Task
|
|
from collections.abc import AsyncGenerator, Awaitable, Callable
|
|
from concurrent.futures import Executor
|
|
from functools import partial
|
|
from typing import TYPE_CHECKING, TypeVar
|
|
|
|
from typing_extensions import ParamSpec
|
|
|
|
P = ParamSpec("P")
|
|
T = TypeVar("T")
|
|
|
|
|
|
def cancel_task_threadsafe(task: Task):
|
|
if task and not task.done():
|
|
run_in_loop(task.get_loop(), task.cancel)
|
|
|
|
|
|
def make_async(
|
|
func: Callable[P, T],
|
|
executor: Executor | None = None,
|
|
) -> Callable[P, Awaitable[T]]:
|
|
"""Take a blocking function, and run it on in an executor thread.
|
|
|
|
This function prevents the blocking function from blocking the
|
|
asyncio event loop.
|
|
The code in this function needs to be thread safe.
|
|
"""
|
|
|
|
def _async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Future[T]:
|
|
loop = asyncio.get_event_loop()
|
|
p_func = partial(func, *args, **kwargs)
|
|
return loop.run_in_executor(executor=executor, func=p_func)
|
|
|
|
return _async_wrapper
|
|
|
|
|
|
def run_in_loop(loop: AbstractEventLoop, function: Callable, *args):
|
|
if in_loop(loop):
|
|
function(*args)
|
|
elif not loop.is_closed():
|
|
loop.call_soon_threadsafe(function, *args)
|
|
|
|
|
|
def in_loop(event_loop: AbstractEventLoop) -> bool:
|
|
try:
|
|
return asyncio.get_running_loop() == event_loop
|
|
except RuntimeError:
|
|
return False
|
|
|
|
|
|
# A hack to pass mypy
|
|
if TYPE_CHECKING:
|
|
|
|
def anext(it: AsyncGenerator[T, None]):
|
|
return it.__anext__()
|
|
|
|
|
|
async def merge_async_iterators(
|
|
*iterators: AsyncGenerator[T, None],
|
|
) -> AsyncGenerator[tuple[int, T], None]:
|
|
"""Merge multiple asynchronous iterators into a single iterator.
|
|
|
|
This method handle the case where some iterators finish before others.
|
|
When it yields, it yields a tuple (i, item) where i is the index of the
|
|
iterator that yields the item.
|
|
"""
|
|
if len(iterators) == 1:
|
|
# Fast-path single iterator case.
|
|
iterator: AsyncGenerator[T, None] | None = iterators[0]
|
|
try:
|
|
async for item in iterator: # type: ignore[union-attr]
|
|
yield 0, item
|
|
iterator = None
|
|
finally:
|
|
if iterator is not None:
|
|
with contextlib.suppress(BaseException):
|
|
await iterator.aclose()
|
|
return
|
|
|
|
loop = asyncio.get_running_loop()
|
|
|
|
awaits = {loop.create_task(anext(it)): (i, it) for i, it in enumerate(iterators)}
|
|
try:
|
|
while awaits:
|
|
done, _ = await asyncio.wait(awaits.keys(), return_when=FIRST_COMPLETED)
|
|
for d in done:
|
|
pair = awaits.pop(d)
|
|
try:
|
|
item = await d
|
|
i, it = pair
|
|
awaits[loop.create_task(anext(it))] = pair
|
|
yield i, item
|
|
except StopAsyncIteration:
|
|
pass
|
|
finally:
|
|
# Cancel any remaining iterators
|
|
for f, (_, it) in awaits.items():
|
|
with contextlib.suppress(BaseException):
|
|
f.cancel()
|
|
await it.aclose()
|
|
|
|
|
|
async def collect_from_async_generator(iterator: AsyncGenerator[T, None]) -> list[T]:
|
|
"""Collect all items from an async generator into a list."""
|
|
items = []
|
|
async for item in iterator:
|
|
items.append(item)
|
|
return items
|