| # Adapted from https://github.com/vllm-project/vllm/blob/v0.10.0/vllm/compilation/cuda_piecewise_backend.py | |
| import dataclasses | |
| import logging | |
| from contextlib import ExitStack | |
| from typing import Any, Callable, Optional, Union | |
| from unittest.mock import patch | |
| import torch | |
| import torch.fx as fx | |
| import sglang.srt.compilation.weak_ref_tensor_jit # noqa: F401 | |
| from sglang.srt.compilation.compilation_config import CompilationConfig | |
| from sglang.srt.compilation.compilation_counter import compilation_counter | |
| logger = logging.getLogger(__name__) | |
| def weak_ref_tensor(tensor: Any) -> Any: | |
| """ | |
| Create a weak reference to a tensor. | |
| The new tensor will share the same data as the original tensor, | |
| but will not keep the original tensor alive. | |
| """ | |
| if isinstance(tensor, torch.Tensor): | |
| # TODO(yuwei): introduce weak_ref_tensor from sgl_kernel | |
| return torch.ops.jit_weak_ref_tensor.weak_ref_tensor(tensor) | |
| return tensor | |
| def weak_ref_tensors( | |
| tensors: Union[torch.Tensor, list[torch.Tensor], tuple[torch.Tensor]] | |
| ) -> Union[torch.Tensor, list[Any], tuple[Any], Any]: | |
| """ | |
| Convenience function to create weak references to tensors, | |
| for single tensor, list of tensors or tuple of tensors. | |
| """ | |
| if isinstance(tensors, torch.Tensor): | |
| return weak_ref_tensor(tensors) | |
| if isinstance(tensors, list): | |
| return [weak_ref_tensor(t) for t in tensors] | |
| if isinstance(tensors, tuple): | |
| return tuple(weak_ref_tensor(t) for t in tensors) | |
| raise ValueError("Invalid type for tensors") | |
| class ConcreteSizeEntry: | |
| runtime_shape: int | |
| need_to_compile: bool # the size is in compile_sizes | |
| use_cudagraph: bool # the size is in cudagraph_capture_sizes | |
| compiled: bool = False | |
| runnable: Callable = None # type: ignore | |
| num_finished_warmup: int = 0 | |
| cudagraph: Optional[torch.cuda.CUDAGraph] = None | |
| output: Optional[Any] = None | |
| # for cudagraph debugging, track the input addresses | |
| # during capture, and check if they are the same during replay | |
| input_addresses: Optional[list[int]] = None | |
| class CUDAPiecewiseBackend: | |
| def __init__( | |
| self, | |
| graph: fx.GraphModule, | |
| compile_config: CompilationConfig, | |
| inductor_config: dict[str, Any], | |
| graph_pool: Any, | |
| piecewise_compile_index: int, | |
| total_piecewise_compiles: int, | |
| sym_shape_indices: list[int], | |
| compiled_graph_for_general_shape: Callable, | |
| sglang_backend, | |
| ): | |
| """ | |
| The backend for piecewise compilation. | |
| It mainly handles the compilation and cudagraph capturing. | |
| We will compile `self.graph` once for the general shape, | |
| and then compile for different shapes specified in | |
| `compilation_config.compile_sizes`. | |
| Independently, we will capture cudagraph for different shapes. | |
| If a shape needs both compilation and cudagraph, we will | |
| compile it first, and then capture cudagraph. | |
| """ | |
| self.graph = graph | |
| self.inductor_config = inductor_config | |
| self.graph_pool = graph_pool | |
| self.piecewise_compile_index = piecewise_compile_index | |
| self.total_piecewise_compiles = total_piecewise_compiles | |
| self.sglang_backend = sglang_backend | |
| self.is_first_graph = piecewise_compile_index == 0 | |
| self.is_last_graph = piecewise_compile_index == total_piecewise_compiles - 1 | |
| self.compile_sizes: set[int] = set([]) | |
| self.compile_config = compile_config | |
| self.cudagraph_capture_sizes: set[int] = set(compile_config.get_capture_sizes()) | |
| self.first_run_finished = False | |
| self.compiled_graph_for_general_shape = compiled_graph_for_general_shape # noqa | |
| self.sym_shape_indices = sym_shape_indices | |
| self.is_debugging_mode = True | |
| # the entries for different shapes that we need to either | |
| # compile or capture cudagraph | |
| self.concrete_size_entries: dict[int, ConcreteSizeEntry] = {} | |
| # to_be_compiled_sizes tracks the remaining sizes to compile, | |
| # and updates during the compilation process, so we need to copy it | |
| self.to_be_compiled_sizes: set[int] = self.compile_sizes.copy() | |
| for shape in self.compile_sizes.union(self.cudagraph_capture_sizes): | |
| self.concrete_size_entries[shape] = ConcreteSizeEntry( | |
| runtime_shape=shape, | |
| need_to_compile=shape in self.compile_sizes, | |
| use_cudagraph=shape in self.cudagraph_capture_sizes, | |
| ) | |
| def check_for_ending_compilation(self): | |
| if self.is_last_graph and not self.to_be_compiled_sizes: | |
| # no specific sizes to compile | |
| # save the hash of the inductor graph for the next run | |
| self.sglang_backend.compiler_manager.save_to_file() | |
| def __call__(self, *args) -> Any: | |
| if not self.first_run_finished: | |
| self.first_run_finished = True | |
| self.check_for_ending_compilation() | |
| return self.compiled_graph_for_general_shape(*args) | |
| runtime_shape = args[self.sym_shape_indices[0]] | |
| if runtime_shape not in self.concrete_size_entries: | |
| # we don't need to do anything for this shape | |
| return self.compiled_graph_for_general_shape(*args) | |
| entry = self.concrete_size_entries[runtime_shape] | |
| if entry.runnable is None: | |
| entry.runnable = self.compiled_graph_for_general_shape | |
| if entry.need_to_compile and not entry.compiled: | |
| entry.compiled = True | |
| self.to_be_compiled_sizes.remove(runtime_shape) | |
| # args are real arguments | |
| entry.runnable = self.sglang_backend.compiler_manager.compile( | |
| self.graph, | |
| args, | |
| self.inductor_config, | |
| graph_index=self.piecewise_compile_index, | |
| num_graphs=self.total_piecewise_compiles, | |
| runtime_shape=runtime_shape, | |
| ) | |
| # finished compilations for all required shapes | |
| if self.is_last_graph and not self.to_be_compiled_sizes: | |
| self.check_for_ending_compilation() | |
| # Skip CUDA graphs if this entry doesn't use them OR | |
| # if we're supposed to skip them globally | |
| # skip_cuda_graphs = get_forward_context().skip_cuda_graphs | |
| # if not entry.use_cudagraph or skip_cuda_graphs: | |
| # return entry.runnable(*args) | |
| if entry.cudagraph is None: | |
| if entry.num_finished_warmup < 1: # noqa | |
| entry.num_finished_warmup += 1 | |
| return entry.runnable(*args) | |
| input_addresses = [ | |
| x.data_ptr() for x in args if isinstance(x, torch.Tensor) | |
| ] | |
| entry.input_addresses = input_addresses | |
| cudagraph = torch.cuda.CUDAGraph() | |
| with ExitStack() as stack: | |
| if not self.is_first_graph: | |
| # during every model forward, we will capture | |
| # many pieces of cudagraphs (roughly one per layer). | |
| # running gc again and again across layers will | |
| # make the cudagraph capture very slow. | |
| # therefore, we only run gc for the first graph, | |
| # and disable gc for the rest of the graphs. | |
| stack.enter_context(patch("gc.collect", lambda: None)) | |
| stack.enter_context(patch("torch.cuda.empty_cache", lambda: None)) | |
| # mind-exploding: carefully manage the reference and memory. | |
| with torch.cuda.graph(cudagraph, pool=self.graph_pool): | |
| # `output` is managed by pytorch's cudagraph pool | |
| output = entry.runnable(*args) | |
| if self.is_last_graph: | |
| # by converting it to weak ref, | |
| # the original `output` will immediately be released | |
| # to save memory. It is only safe to do this for | |
| # the last graph, because the output of the last graph | |
| # will not be used by any other cuda graph. | |
| output = weak_ref_tensors(output) | |
| # here we always use weak ref for the output | |
| # to save memory | |
| entry.output = weak_ref_tensors(output) | |
| entry.cudagraph = cudagraph | |
| compilation_counter.num_cudagraph_captured += 1 | |
| # important: we need to return the output, rather than | |
| # the weak ref of the output, so that pytorch can correctly | |
| # manage the memory during cuda graph capture | |
| return output | |
| if self.is_debugging_mode: | |
| # check if the input addresses are the same | |
| new_input_addresses = [ | |
| x.data_ptr() for x in args if isinstance(x, torch.Tensor) | |
| ] | |
| assert new_input_addresses == entry.input_addresses, ( | |
| "Input addresses for cudagraphs are different during replay." | |
| f" Expected {entry.input_addresses}, got {new_input_addresses}" | |
| ) | |
| entry.cudagraph.replay() | |
| return entry.output | |
Xet Storage Details
- Size:
- 9.24 kB
- Xet hash:
- c6637ada317c38046e66e71ec88f02e6a578fa070b8a16609704b12cfdb3ba43
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.