gpu_worker.py 17.2 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4
5
"""A GPU worker class."""
import gc
import os
6
from typing import TYPE_CHECKING, Optional
7
8
9

import torch
import torch.distributed
10
import torch.nn as nn
11

12
import vllm.envs as envs
13
from vllm.config import VllmConfig
14
15
16
from vllm.distributed import (ensure_model_parallel_initialized,
                              init_distributed_environment,
                              set_custom_all_reduce)
17
from vllm.distributed.kv_transfer import ensure_kv_transfer_initialized
18
from vllm.distributed.parallel_state import get_pp_group, get_tp_group
19
from vllm.logger import init_logger
20
from vllm.lora.request import LoRARequest
21
22
from vllm.model_executor import set_random_seed
from vllm.platforms import current_platform
23
from vllm.sequence import IntermediateTensors
24
from vllm.utils import GiB_bytes, MemorySnapshot, memory_profiling
25
from vllm.v1.kv_cache_interface import KVCacheConfig, KVCacheSpec
26
from vllm.v1.outputs import ModelRunnerOutput
27
from vllm.v1.utils import report_usage_stats
28
from vllm.v1.worker.gpu_model_runner import GPUModelRunner
29
from vllm.v1.worker.worker_base import WorkerBase
30
31
32
33

logger = init_logger(__name__)

if TYPE_CHECKING:
34
    from vllm.model_executor.model_loader.tensorizer import TensorizerConfig
35
    from vllm.v1.core.sched.output import SchedulerOutput
36
37


38
class Worker(WorkerBase):
39
40
41

    def __init__(
        self,
42
        vllm_config: VllmConfig,
43
44
45
        local_rank: int,
        rank: int,
        distributed_init_method: str,
46
        is_driver_worker: bool = False,
47
    ):
48

49
50
51
52
53
        super().__init__(vllm_config=vllm_config,
                         local_rank=local_rank,
                         rank=rank,
                         distributed_init_method=distributed_init_method,
                         is_driver_worker=is_driver_worker)
54
55
56
57
58
59

        if self.model_config.trust_remote_code:
            # note: lazy import to avoid importing torch before initializing
            from vllm.utils import init_cached_hf_modules
            init_cached_hf_modules()

60
61
62
        # Buffers saved before sleep
        self._sleep_saved_buffers: dict[str, torch.Tensor] = {}

63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
        # Torch profiler. Enabled and configured through env vars:
        # VLLM_TORCH_PROFILER_DIR=/path/to/save/trace
        if envs.VLLM_TORCH_PROFILER_DIR:
            torch_profiler_trace_dir = envs.VLLM_TORCH_PROFILER_DIR
            logger.info("Profiling enabled. Traces will be saved to: %s",
                        torch_profiler_trace_dir)
            self.profiler = torch.profiler.profile(
                activities=[
                    torch.profiler.ProfilerActivity.CPU,
                    torch.profiler.ProfilerActivity.CUDA,
                ],
                with_stack=True,
                on_trace_ready=torch.profiler.tensorboard_trace_handler(
                    torch_profiler_trace_dir, use_gzip=True))
        else:
            self.profiler = None
79

80
    def sleep(self, level: int = 1) -> None:
81
82
        from vllm.device_allocator.cumem import CuMemAllocator

83
        free_bytes_before_sleep = torch.cuda.mem_get_info()[0]
84
85
86
87
88
89
90
91
92

        # Save the buffers before level 2 sleep
        if level == 2:
            model = self.model_runner.model
            self._sleep_saved_buffers = {
                name: buffer.cpu().clone()
                for name, buffer in model.named_buffers()
            }

93
94
95
96
97
98
99
100
101
102
103
        allocator = CuMemAllocator.get_instance()
        allocator.sleep(offload_tags=("weights", ) if level == 1 else tuple())
        free_bytes_after_sleep, total = torch.cuda.mem_get_info()
        freed_bytes = free_bytes_after_sleep - free_bytes_before_sleep
        used_bytes = total - free_bytes_after_sleep
        assert freed_bytes >= 0, "Memory usage increased after sleeping."
        logger.info(
            "Sleep mode freed %.2f GiB memory, "
            "%.2f GiB memory is still in use.", freed_bytes / GiB_bytes,
            used_bytes / GiB_bytes)

104
    def wake_up(self, tags: Optional[list[str]] = None) -> None:
105
106
        from vllm.device_allocator.cumem import CuMemAllocator

107
        allocator = CuMemAllocator.get_instance()
108
        allocator.wake_up(tags)
109

110
111
112
113
114
115
116
117
        # Restore the buffers after level 2 sleep
        if len(self._sleep_saved_buffers):
            model = self.model_runner.model
            for name, buffer in model.named_buffers():
                if name in self._sleep_saved_buffers:
                    buffer.data.copy_(self._sleep_saved_buffers[name].data)
            self._sleep_saved_buffers = {}

118
119
120
121
122
    def initialize_cache(self, num_gpu_blocks: int,
                         num_cpu_blocks: int) -> None:
        self.cache_config.num_gpu_blocks = num_gpu_blocks
        self.cache_config.num_cpu_blocks = num_cpu_blocks

123
    def init_device(self):
124
125
126
127
128
129
130
131
132
133
134
135
        if self.device_config.device.type == "cuda":
            # torch.distributed.all_reduce does not free the input tensor until
            # the synchronization point. This causes the memory usage to grow
            # as the number of all_reduce calls increases. This env var disables
            # this behavior.
            # Related issue:
            # https://discuss.pytorch.org/t/cuda-allocation-lifetime-for-inputs-to-distributed-all-reduce/191573
            os.environ["TORCH_NCCL_AVOID_RECORD_STREAMS"] = "1"

            # This env var set by Ray causes exceptions with graph building.
            os.environ.pop("NCCL_ASYNC_ERROR_HANDLING", None)
            self.device = torch.device(f"cuda:{self.local_rank}")
136
            current_platform.set_device(self.device)
137
138
139
140

            _check_if_gpu_supports_dtype(self.model_config.dtype)
            gc.collect()
            torch.cuda.empty_cache()
141
142
143
144
145
146

            # take current memory snapshot
            self.init_snapshot = MemorySnapshot()
            self.requested_memory = (self.init_snapshot.total_memory *
                                     self.cache_config.gpu_memory_utilization)
            if self.init_snapshot.free_memory < self.requested_memory:
147
148
                GiB = lambda b: round(b / GiB_bytes, 2)
                raise ValueError(
149
150
151
152
                    f"Free memory on device "
                    f"({GiB(self.init_snapshot.free_memory)}/"
                    f"{GiB(self.init_snapshot.total_memory)} GiB) on startup "
                    f"is less than desired GPU memory utilization "
153
                    f"({self.cache_config.gpu_memory_utilization}, "
154
                    f"{GiB(self.requested_memory)} GiB). Decrease GPU memory "
155
156
                    f"utilization or reduce GPU memory used by other processes."
                )
157
158
159
160
        else:
            raise RuntimeError(
                f"Not support device type: {self.device_config.device}")
        # Initialize the distributed environment.
161
        init_worker_distributed_environment(self.vllm_config, self.rank,
162
                                            self.distributed_init_method,
163
164
                                            self.local_rank,
                                            current_platform.dist_backend)
165
166
167
        # Set random seed.
        set_random_seed(self.model_config.seed)

168
        # Construct the model runner
169
170
        self.model_runner: GPUModelRunner = GPUModelRunner(
            self.vllm_config, self.device)
171

172
173
174
175
        if self.rank == 0:
            # If usage stat is enabled, collect relevant info.
            report_usage_stats(self.vllm_config)

176
177
    # FIXME(youkaichao & ywang96): Use TorchDispatchMode instead of memory pool
    # to hijack tensor allocation.
178
    def load_model(self) -> None:
179
        if self.vllm_config.model_config.enable_sleep_mode:
180
181
            from vllm.device_allocator.cumem import CuMemAllocator

182
183
184
185
186
187
188
189
190
191
            allocator = CuMemAllocator.get_instance()
            assert allocator.get_current_usage() == 0, (
                "Sleep mode can only be "
                "used for one instance per process.")
            context = allocator.use_memory_pool(tag="weights")
        else:
            from contextlib import nullcontext
            context = nullcontext()
        with context:
            self.model_runner.load_model()
192
193

    @torch.inference_mode()
194
195
196
    def determine_available_memory(self) -> int:
        """Profiles the peak memory usage of the model to determine how much 
        memory can be used for KV cache without OOMs.
197
198

        The engine will first conduct a profiling of the existing memory usage.
199
200
        Then, it calculate the free memory that can be used for KV cache in
        bytes.
201

202
203
204
        Tip:
            You may limit the usage of GPU memory
            by adjusting the `gpu_memory_utilization` parameter.
205
206
        """
        torch.cuda.empty_cache()
207
        torch.cuda.reset_peak_memory_stats()
208
        GiB = lambda b: b / GiB_bytes
209
210
211

        # Execute a forward pass with dummy inputs to profile the memory usage
        # of the model.
212
213
214
215
216
        with memory_profiling(
                self.init_snapshot,
                weights_memory=int(
                    self.model_runner.model_memory_usage)) as profile_result:
            self.model_runner.profile_run()
217

218
        free_gpu_memory = profile_result.after_profile.free_memory
219
220
        # NOTE(woosuk): Here we assume that the other processes using the same
        # GPU did not change their memory usage during the profiling.
221
        assert self.init_snapshot.free_memory > free_gpu_memory, (
222
            "Error in memory profiling. "
223
224
225
226
227
228
229
230
            f"Initial free memory {GiB(self.init_snapshot.free_memory)} GiB, "
            f"current free memory {GiB(free_gpu_memory)} GiB. "
            "This happens when other processes sharing the same container "
            "release GPU memory while vLLM is profiling during initialization. "
            "To fix this, ensure consistent GPU memory allocation or "
            "isolate vLLM in its own container.")
        available_kv_cache_memory = self.requested_memory \
            - profile_result.non_kv_cache_memory
231
232
233

        logger.debug(
            "Initial free memory: %.2f GiB, free memory: %.2f GiB, "
234
235
236
237
238
239
240
            "requested GPU memory: %.2f GiB",
            GiB(self.init_snapshot.free_memory), GiB(free_gpu_memory),
            GiB(self.requested_memory))
        logger.debug(profile_result)
        logger.info("Available KV cache memory: %.2f GiB",
                    GiB(available_kv_cache_memory))
        gc.collect()
241

242
243
        return int(available_kv_cache_memory)

244
    def get_kv_cache_spec(self) -> dict[str, KVCacheSpec]:
245
246
        return self.model_runner.get_kv_cache_spec()

247
    def initialize_from_config(self, kv_cache_config: KVCacheConfig) -> None:
248
        """Allocate GPU KV cache with the specified kv_cache_config."""
249

250
        if self.vllm_config.model_config.enable_sleep_mode:
251
252
            from vllm.device_allocator.cumem import CuMemAllocator

253
254
255
256
257
258
259
            allocator = CuMemAllocator.get_instance()
            context = allocator.use_memory_pool(tag="kv_cache")
        else:
            from contextlib import nullcontext
            context = nullcontext()
        with context:
            self.model_runner.initialize_kv_cache(kv_cache_config)
260
261

    def compile_or_warm_up_model(self) -> None:
262
263
264
265
266
267
268
269
270
        # warm up sizes that are not in cudagraph capture sizes,
        # but users still want to compile for better performance,
        # e.g. for the max-num-batched token size in chunked prefill.
        warmup_sizes = self.vllm_config.compilation_config.compile_sizes.copy()
        if not self.model_config.enforce_eager:
            warmup_sizes = [
                x for x in warmup_sizes if x not in
                self.vllm_config.compilation_config.cudagraph_capture_sizes
            ]
271
        # We skip EPLB here since we don't want to record dummy metrics
272
273
        for size in sorted(warmup_sizes, reverse=True):
            logger.info("Compile and warming up model for size %d", size)
274
            self.model_runner._dummy_run(size, skip_eplb=True)
275
276
        if not self.model_config.enforce_eager:
            self.model_runner.capture_model()
277
278
279
280
281
282

        # Warm up sampler and preallocate memory buffer for logits and other
        # sampling related tensors of max possible shape to avoid memory
        # fragmentation issue.
        # NOTE: This is called after `capture_model` on purpose to prevent
        # memory buffers from being cleared by `torch.cuda.empty_cache`.
283
        if get_pp_group().is_last_rank:
284
285
            max_num_reqs = min(self.scheduler_config.max_num_seqs,
                               self.scheduler_config.max_num_batched_tokens)
286

287
            # We skip EPLB here since we don't want to record dummy metrics
288
            hidden_states, last_hidden_states = \
289
290
291
292
                self.model_runner._dummy_run(
                    num_tokens=max_num_reqs,
                    skip_eplb=True,
                )
293
294
295
296
297
            if self.model_runner.is_pooling_model:
                self.model_runner._dummy_pooler_run(hidden_states)
            else:
                self.model_runner._dummy_sampler_run(
                    hidden_states=last_hidden_states)
298

299
300
301
302
        # Reset the seed to ensure that the random state is not affected by
        # the model initialization and profiling.
        set_random_seed(self.model_config.seed)

303
304
305
    def get_model(self) -> nn.Module:
        return self.model_runner.get_model()

306
307
308
309
    @torch.inference_mode()
    def execute_model(
        self,
        scheduler_output: "SchedulerOutput",
310
    ) -> Optional[ModelRunnerOutput]:
311
312
313
314
315
316
317
318
        intermediate_tensors = None
        if not get_pp_group().is_first_rank:
            intermediate_tensors = IntermediateTensors(
                get_pp_group().recv_tensor_dict(
                    all_gather_group=get_tp_group()))

        output = self.model_runner.execute_model(scheduler_output,
                                                 intermediate_tensors)
319
320
321
        parallel_config = self.vllm_config.parallel_config
        if parallel_config.distributed_executor_backend != "external_launcher" \
            and not get_pp_group().is_last_rank:
322
323
324
325
326
            assert isinstance(output, IntermediateTensors)
            get_pp_group().send_tensor_dict(output.tensors,
                                            all_gather_group=get_tp_group())
            return None
        assert isinstance(output, ModelRunnerOutput)
327
        return output if self.is_driver_worker else None
328

329
    def profile(self, is_start: bool = True):
330
331
332
333
334
335
        if self.profiler is None:
            raise RuntimeError("Profiler is not enabled.")
        if is_start:
            self.profiler.start()
        else:
            self.profiler.stop()
336
337
            print(self.profiler.key_averages().table(
                sort_by="self_cuda_time_total"))
338

339
340
341
    def execute_dummy_batch(self) -> None:
        self.model_runner._dummy_run(1)

342
343
344
    def add_lora(self, lora_request: LoRARequest) -> bool:
        return self.model_runner.add_lora(lora_request)

345
346
347
    def remove_lora(self, lora_id: int) -> bool:
        return self.model_runner.remove_lora(lora_id)

348
    def list_loras(self) -> set[int]:
349
350
351
352
353
        return self.model_runner.list_loras()

    def pin_lora(self, lora_id: int) -> bool:
        return self.model_runner.pin_lora(lora_id)

354
355
356
357
    def check_health(self) -> None:
        # worker will always be healthy as long as it's running.
        return

358
359
360
361
362
363
    def save_sharded_state(
        self,
        path: str,
        pattern: Optional[str] = None,
        max_size: Optional[int] = None,
    ) -> None:
364
        from vllm.model_executor.model_loader import ShardedStateLoader
365
366
367
368
369
370
371
        ShardedStateLoader.save_model(
            self.model_runner.model,
            path,
            pattern=pattern,
            max_size=max_size,
        )

372
373
374
375
376
377
378
    def save_tensorized_model(
        self,
        tensorizer_config: "TensorizerConfig",
    ) -> None:
        self.model_runner.save_tensorized_model(
            tensorizer_config=tensorizer_config, )

379
380

def init_worker_distributed_environment(
381
    vllm_config: VllmConfig,
382
383
384
    rank: int,
    distributed_init_method: Optional[str] = None,
    local_rank: int = -1,
385
    backend: str = "nccl",
386
387
) -> None:
    """Initialize the distributed environment."""
388
    parallel_config = vllm_config.parallel_config
389
390
391
    set_custom_all_reduce(not parallel_config.disable_custom_all_reduce)

    init_distributed_environment(parallel_config.world_size, rank,
392
                                 distributed_init_method, local_rank, backend)
393
394

    ensure_model_parallel_initialized(parallel_config.tensor_parallel_size,
395
                                      parallel_config.pipeline_parallel_size)
396

397
398
    ensure_kv_transfer_initialized(vllm_config)

399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415

def _check_if_gpu_supports_dtype(torch_dtype: torch.dtype):
    # Check if the GPU supports the dtype.
    if torch_dtype == torch.bfloat16:  # noqa: SIM102
        if not current_platform.has_device_capability(80):
            capability = current_platform.get_device_capability()
            gpu_name = current_platform.get_device_name()

            if capability is None:
                compute_str = "does not have a compute capability"
            else:
                version_str = capability.as_version_str()
                compute_str = f"has compute capability {version_str}"

            raise ValueError(
                "Bfloat16 is only supported on GPUs with compute capability "
                f"of at least 8.0. Your {gpu_name} GPU {compute_str}. "
416
                "You can use float16 instead by explicitly setting the "
417
                "`dtype` flag in CLI, for example: --dtype=half.")