Skip to content

vllm.v1.utils

T module-attribute 

T = TypeVar('T')

logger module-attribute 

logger = init_logger(__name__)

ConstantList

Bases: Generic[T], Sequence

Source code in vllm/v1/utils.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class ConstantList(Generic[T], Sequence):

    def __init__(self, x: list[T]) -> None:
        self._x = x

    def append(self, item):
        raise Exception("Cannot append to a constant list")

    def extend(self, item):
        raise Exception("Cannot extend a constant list")

    def insert(self, item):
        raise Exception("Cannot insert into a constant list")

    def pop(self, item):
        raise Exception("Cannot pop from a constant list")

    def remove(self, item):
        raise Exception("Cannot remove from a constant list")

    def clear(self):
        raise Exception("Cannot clear a constant list")

    def index(self,
              item: T,
              start: int = 0,
              stop: Optional[int] = None) -> int:
        return self._x.index(item, start,
                             stop if stop is not None else len(self._x))

    @overload
    def __getitem__(self, item: int) -> T:
        ...

    @overload
    def __getitem__(self, s: slice, /) -> list[T]:
        ...

    def __getitem__(self, item: Union[int, slice]) -> Union[T, list[T]]:
        return self._x[item]

    @overload
    def __setitem__(self, item: int, value: T):
        ...

    @overload
    def __setitem__(self, s: slice, value: T, /):
        ...

    def __setitem__(self, item: Union[int, slice], value: Union[T, list[T]]):
        raise Exception("Cannot set item in a constant list")

    def __delitem__(self, item):
        raise Exception("Cannot delete item from a constant list")

    def __iter__(self):
        return iter(self._x)

    def __contains__(self, item):
        return item in self._x

    def __len__(self):
        return len(self._x)

    def __repr__(self):
        return f"ConstantList({self._x})"

_x instance-attribute 

x = x

__contains__ 

__contains__(item)
Source code in vllm/v1/utils.py 
88
89
def __contains__(self, item):
    return item in self._x

__delitem__ 

__delitem__(item)
Source code in vllm/v1/utils.py 
82
83
def __delitem__(self, item):
    raise Exception("Cannot delete item from a constant list")

__getitem__ 

__getitem__(item: int) -> T

__getitem__(s: slice) -> list[T]

__getitem__(item: Union[int, slice]) -> Union[T, list[T]]
Source code in vllm/v1/utils.py 
68
69
def __getitem__(self, item: Union[int, slice]) -> Union[T, list[T]]:
    return self._x[item]

__init__ 

__init__(x: list[T]) -> None
Source code in vllm/v1/utils.py 
32
33
def __init__(self, x: list[T]) -> None:
    self._x = x

__iter__ 

__iter__()
Source code in vllm/v1/utils.py 
85
86
def __iter__(self):
    return iter(self._x)

__len__ 

__len__()
Source code in vllm/v1/utils.py 
91
92
def __len__(self):
    return len(self._x)

__repr__ 

__repr__()
Source code in vllm/v1/utils.py 
94
95
def __repr__(self):
    return f"ConstantList({self._x})"

__setitem__ 

__setitem__(item: int, value: T)

__setitem__(s: slice, value: T)

__setitem__(
    item: Union[int, slice], value: Union[T, list[T]]
)
Source code in vllm/v1/utils.py 
79
80
def __setitem__(self, item: Union[int, slice], value: Union[T, list[T]]):
    raise Exception("Cannot set item in a constant list")

append 

append(item)
Source code in vllm/v1/utils.py 
35
36
def append(self, item):
    raise Exception("Cannot append to a constant list")

clear 

clear()
Source code in vllm/v1/utils.py 
50
51
def clear(self):
    raise Exception("Cannot clear a constant list")

extend 

extend(item)
Source code in vllm/v1/utils.py 
38
39
def extend(self, item):
    raise Exception("Cannot extend a constant list")

index 

index(
    item: T, start: int = 0, stop: Optional[int] = None
) -> int
Source code in vllm/v1/utils.py 
53
54
55
56
57
58
def index(self,
          item: T,
          start: int = 0,
          stop: Optional[int] = None) -> int:
    return self._x.index(item, start,
                         stop if stop is not None else len(self._x))

insert 

insert(item)
Source code in vllm/v1/utils.py 
41
42
def insert(self, item):
    raise Exception("Cannot insert into a constant list")

pop 

pop(item)
Source code in vllm/v1/utils.py 
44
45
def pop(self, item):
    raise Exception("Cannot pop from a constant list")

remove 

remove(item)
Source code in vllm/v1/utils.py 
47
48
def remove(self, item):
    raise Exception("Cannot remove from a constant list")

CoreEngineProcManager

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

Source code in vllm/v1/utils.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
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,
        on_head_node: bool,
        input_address: str,
        executor_class: type[Executor],
        log_stats: bool,
    ):
        context = get_mp_context()
        common_kwargs = {
            "vllm_config": vllm_config,
            "on_head_node": on_head_node,
            "input_address": input_address,
            "executor_class": executor_class,
            "log_stats": log_stats,
        }

        self.processes: list[Process] = []
        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,
                                           input_address)
        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, input_address
)

processes instance-attribute 

processes: list[Process] = []

__init__ 

__init__(
    target_fn: Callable,
    local_engine_count: int,
    start_index: int,
    local_start_index: int,
    vllm_config: VllmConfig,
    on_head_node: bool,
    input_address: str,
    executor_class: type[Executor],
    log_stats: bool,
)
Source code in vllm/v1/utils.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def __init__(
    self,
    target_fn: Callable,
    local_engine_count: int,
    start_index: int,
    local_start_index: int,
    vllm_config: VllmConfig,
    on_head_node: bool,
    input_address: str,
    executor_class: type[Executor],
    log_stats: bool,
):
    context = get_mp_context()
    common_kwargs = {
        "vllm_config": vllm_config,
        "on_head_node": on_head_node,
        "input_address": input_address,
        "executor_class": executor_class,
        "log_stats": log_stats,
    }

    self.processes: list[Process] = []
    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,
                                       input_address)
    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/utils.py 
148
149
150
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/utils.py 
159
160
161
162
163
164
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/utils.py 
152
153
154
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/utils.py 
156
157
def sentinels(self) -> list:
    return [proc.sentinel for proc in self.processes]

bind_kv_cache 

bind_kv_cache(
    kv_caches: dict[str, Tensor],
    forward_context: dict[str, Attention],
    runner_kv_caches: list[Tensor],
) -> None

Bind the allocated KV cache to both ModelRunner and forward context so that the KV cache can be used in the forward pass.

This function

1) Fills the ModelRunner's kv cache list (runner_kv_caches) with kv_caches. 2) Associates each attention layer in the forward_context with its corresponding KV cache in kv_caches.

Parameters:

Name Type Description Default
kv_caches dict[str, Tensor]

The allocated kv_caches with layer names as keys.

 required
forward_context dict[str, Attention]

The global forward context containing all Attention

required
runner_kv_caches list[Tensor]

The kv_cache declared by ModelRunner.

 required
Source code in vllm/v1/utils.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def bind_kv_cache(
    kv_caches: dict[str, torch.Tensor],
    forward_context: dict[str, "Attention"],
    runner_kv_caches: list[torch.Tensor],
) -> None:
    """
    Bind the allocated KV cache to both ModelRunner and forward context so
    that the KV cache can be used in the forward pass.

    This function:
      1) Fills the ModelRunner's kv cache list (`runner_kv_caches`) with
         kv_caches.
      2) Associates each attention layer in the `forward_context` with its 
         corresponding KV cache in kv_caches.

    Args:
        kv_caches: The allocated kv_caches with layer names as keys.
        forward_context: The global forward context containing all Attention 
        layers with layer names as keys.
        runner_kv_caches: The kv_cache declared by ModelRunner.
    """
    # Bind kv_caches to ModelRunner
    assert len(runner_kv_caches) == 0

    # Convert kv_caches dict to a list of tensors in the order of layer_index.
    index2name = defaultdict(list)
    for layer_name in kv_caches:
        index2name[extract_layer_index(layer_name)].append(layer_name)

    for layer_index in sorted(index2name.keys()):
        layer_names = index2name[layer_index]
        if len(layer_names) > 1:
            # One typical case is encoder-decoder model, e.g., bart.
            # The cross attention and self attention in the same decoder layer
            # has different layer_name but the same layer_index.
            raise NotImplementedError
        layer_name = layer_names[0]
        runner_kv_caches.append(kv_caches[layer_name])

    # Bind kv_caches to forward context
    for layer_name, kv_cache in kv_caches.items():
        # NOTE: Use list because of v0 PP virtual engine.
        forward_context[layer_name].kv_cache = [kv_cache]

copy_slice 

copy_slice(
    from_tensor: Tensor, to_tensor: Tensor, length: int
) -> Tensor

Copy the first length elements of a tensor into another tensor in a non-blocking manner.

Used to copy pinned CPU tensor data to pre-allocated GPU tensors.

Returns the sliced target tensor.

Source code in vllm/v1/utils.py 
240
241
242
243
244
245
246
247
248
249
250
def copy_slice(from_tensor: torch.Tensor, to_tensor: torch.Tensor,
               length: int) -> torch.Tensor:
    """
    Copy the first length elements of a tensor into another tensor in a
    non-blocking manner.

    Used to copy pinned CPU tensor data to pre-allocated GPU tensors.

    Returns the sliced target tensor.
    """
    return to_tensor[:length].copy_(from_tensor[:length], non_blocking=True)

report_usage_stats 

report_usage_stats(
    vllm_config,
    usage_context: UsageContext = ENGINE_CONTEXT,
) -> None

Report usage statistics if enabled.

Source code in vllm/v1/utils.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def report_usage_stats(
        vllm_config,
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT) -> None:
    """Report usage statistics if enabled."""

    if not is_usage_stats_enabled():
        return

    from vllm.model_executor.model_loader import get_architecture_class_name

    usage_message.report_usage(
        get_architecture_class_name(vllm_config.model_config),
        usage_context,
        extra_kvs={
            # Common configuration
            "dtype":
            str(vllm_config.model_config.dtype),
            "tensor_parallel_size":
            vllm_config.parallel_config.tensor_parallel_size,
            "block_size":
            vllm_config.cache_config.block_size,
            "gpu_memory_utilization":
            vllm_config.cache_config.gpu_memory_utilization,

            # Quantization
            "quantization":
            vllm_config.model_config.quantization,
            "kv_cache_dtype":
            str(vllm_config.cache_config.cache_dtype),

            # Feature flags
            "enable_lora":
            bool(vllm_config.lora_config),
            "enable_prompt_adapter":
            bool(vllm_config.prompt_adapter_config),
            "enable_prefix_caching":
            vllm_config.cache_config.enable_prefix_caching,
            "enforce_eager":
            vllm_config.model_config.enforce_eager,
            "disable_custom_all_reduce":
            vllm_config.parallel_config.disable_custom_all_reduce,
        })

shutdown 

shutdown(procs: list[Process], input_address: str)
Source code in vllm/v1/utils.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def shutdown(procs: list[Process], input_address: str):
    # Shutdown the process.
    for proc in procs:
        if proc.is_alive():
            proc.terminate()

    # Allow 5 seconds for remaining procs to terminate.
    deadline = time.monotonic() + 5
    for proc in procs:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            break
        if proc.is_alive():
            proc.join(remaining)

    for proc in procs:
        if proc.is_alive() and (pid := proc.pid) is not None:
            kill_process_tree(pid)

    # Remove zmq ipc socket files.
    if input_address.startswith("ipc://"):
        socket_file = input_address[len("ipc://"):]
        if os and os.path.exists(socket_file):
            os.remove(socket_file)