feat(pro6000/GLM-5.3): 部署方案入库(TP8+EAGLE 生产标准 / 场景二高并发变体 / TP4PP2+IndexCache / E7b CAR 实验补丁 + deploy profiles)
- deploy_glm53_605.sh:配置 A 生产标准(60.5 在役,全 8 台 md5 fcd9109b 一致),
支持 MEMFRAC/STEPS/TOPK/DRAFT/CHUNK/EXTRA/RESTART 调参;场景二高并发变体
仅改 mrr32 + decode 图 bs{4,8,12,16}(KV 池 16.4 驻留上限)
- deploy_glm53_optimal(_s1).sh:配置 B TP4PP2+IndexCache(场景二最优 +41~79%;
s1 形态唯一差异 radix-on)
- deploy_glm53_607_exp.sh + car_patch/:E7b custom-AR 1stage 补丁(cc1 decode
每步 -14~-16%,实验性仅 cc1-2 验证;补丁文件与 60.7:/root/patches md5 一致)
- deploy/profiles/pro6000/:两个标准 profile(sskj.deploy 可消费),关键踩坑
与场景二变体、parser 缺口均在注释中标注
This commit is contained in:
parent
f0ab17c561
commit
b3165a1d3c
@ -0,0 +1,43 @@
|
|||||||
|
# GLM-5.3-NVFP4 SGLang TP=4 PP=2 + IndexCache deployment profile (single RTX 6000D node).
|
||||||
|
# 场景二(16k 独立输入、cc8-32、prefill 主导)最优配置:吞吐 +41~79%、cc32 TTFT 减半
|
||||||
|
# (对比 TP8+EAGLE 生产配置,2026-09-07 真实语料基线)。
|
||||||
|
# 可执行部署脚本:experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/deploy_glm53_optimal.sh
|
||||||
|
#
|
||||||
|
# 关键点(实测踩坑,勿随意改):
|
||||||
|
# - index_topk_freq=4 为模型原生默认(层轴索引复用省 75% indexer,无质量损失)
|
||||||
|
# - 禁投机解码:PP2 与投机框架不兼容(已实测)
|
||||||
|
# - 本文件为场景二形态(--disable-radix-cache,独立输入无前缀复用);
|
||||||
|
# 场景一 90% 命中对比须启用 radix(唯一差异:去掉 --disable-radix-cache,
|
||||||
|
# 见 deploy_glm53_optimal_s1.sh)
|
||||||
|
# - mem 0.85:0.90 下 cuda graph 捕获余量不足会运行时 OOM(KV 池 569,600 = TP8 的 2.06 倍)
|
||||||
|
# - --disable-custom-all-reduce:TP4 over PCIe 用自定义 AR 在本栈无收益
|
||||||
|
# - 已知缺口:未带 --tool-call-parser glm47 --reasoning-parser glm45,质量门 6/7
|
||||||
|
# (tool call 失败纯属参数缺失,非模型问题);上生产必须补 parser
|
||||||
|
# - 场景一(90% 命中低并发)该配置全面劣于 TP8+EAGLE(输出吞吐 −25~−65%),勿混用选型
|
||||||
|
|
||||||
|
PLATFORM=pro6000
|
||||||
|
EXPERIMENT=glm53_nvfp4_pro6000_sglang_tp4pp2_indexcache
|
||||||
|
MODEL_NAME=GLM-5.3-NVFP4
|
||||||
|
ENGINE=sglang
|
||||||
|
RUNTIME=docker
|
||||||
|
DOCKER_IMAGE=lmsysorg/sglang:nightly-dev-20260828-daf63171
|
||||||
|
CONTAINER_NAME=glm53-nvfp4
|
||||||
|
MODEL_PATH=/data/hf_models/GLM-5.3-NVFP4
|
||||||
|
SERVED_MODEL_NAME=/data/hf_models/GLM-5.3-NVFP4
|
||||||
|
PORT=30000
|
||||||
|
HEALTH_PATH=/health
|
||||||
|
HEALTH_WAIT_S=2400
|
||||||
|
CONTAINER_PYTHON=python3
|
||||||
|
|
||||||
|
TP=4
|
||||||
|
DP=1
|
||||||
|
|
||||||
|
DEVICE_VARS="CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7"
|
||||||
|
ENGINE_ENV="PYTHONUNBUFFERED=1 HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1"
|
||||||
|
|
||||||
|
DOCKER_FLAGS="--gpus all --ipc=host --shm-size 64g --entrypoint '' -p ${PORT}:${PORT}"
|
||||||
|
VOLUMES="/data/hf_models:/data/hf_models:ro"
|
||||||
|
|
||||||
|
BOOTSTRAP="exec python3 -m sglang.launch_server ${LAUNCH_ARGS}"
|
||||||
|
|
||||||
|
LAUNCH_ARGS="--model-path ${MODEL_PATH} --tp-size ${TP} --pp-size 2 --mem-fraction-static 0.85 --max-running-requests 48 --disable-radix-cache --disable-shared-experts-fusion --moe-runner-backend flashinfer_cutlass --disable-flashinfer-autotune --disable-custom-all-reduce --chunked-prefill-size 16384 --json-model-override-args '{\"index_topk_freq\": 4}' --host 0.0.0.0 --port ${PORT}"
|
||||||
@ -0,0 +1,44 @@
|
|||||||
|
# GLM-5.3-NVFP4 SGLang TP=8 + EAGLE deployment profile (single RTX 6000D node, 8 GPUs).
|
||||||
|
# 生产标准配置(174.1.60.5 团队自用在役,2026-09-07 定稿)。
|
||||||
|
# 可执行部署脚本:experiments/pro6000/glm53_nvfp4_pro6000d_sglang_dual_scenario_bench/scripts/deploy_glm53_605.sh
|
||||||
|
#
|
||||||
|
# 关键点(实测踩坑,勿随意改):
|
||||||
|
# - SM120 必需三项:--disable-shared-experts-fusion --moe-runner-backend
|
||||||
|
# flashinfer_cutlass --disable-flashinfer-autotune
|
||||||
|
# - EAGLE 4/1/5 为 cc1-2 均衡点(3/1/4、5/1/6 在 ±10% 噪声内;树式 topk>1 在 DSA 上不可用;
|
||||||
|
# EAGLE3 本 nightly 不支持 GLM-5.3);draft 模型自动从主权重加载
|
||||||
|
# - KV fp8_e4m3 + hicache-ratio 3 是长上下文 90% 前缀命中场景的容量前提(KV 池 276,864 token)
|
||||||
|
# - chunk 8192:16384 已 A/B 证伪(MoE 工作区 OOM 风险,且 AR 带宽受限无次数红利)
|
||||||
|
# - 容器入口必须 python3 -m sglang.launch_server(镜像 entrypoint 无 shebang)
|
||||||
|
# - 重新部署前必须等显存排空(docker rm -f 异步滞留数分钟,不等会把新 KV 池压小)
|
||||||
|
# - 场景二高并发变体(16k 独立输入 cc8-32):仅改 --max-running-requests 32 +
|
||||||
|
# --cuda-graph-max-bs-decode 16 --cuda-graph-bs-decode 4 8 12 16(KV 池 276,864 ÷
|
||||||
|
# 16,896/req = 16.4 驻留上限,decode 批自然 ≤16,图覆盖到 bs16 即可,bs24/32 纯耗显存)
|
||||||
|
# - 上线前后过质量门 quality_gate_605.sh(PASS=7/7)
|
||||||
|
|
||||||
|
PLATFORM=pro6000
|
||||||
|
EXPERIMENT=glm53_nvfp4_pro6000_sglang_tp8eagle
|
||||||
|
MODEL_NAME=GLM-5.3-NVFP4
|
||||||
|
ENGINE=sglang
|
||||||
|
RUNTIME=docker
|
||||||
|
DOCKER_IMAGE=lmsysorg/sglang:nightly-dev-20260828-daf63171
|
||||||
|
CONTAINER_NAME=glm53-nvfp4
|
||||||
|
MODEL_PATH=/data/hf_models/GLM-5.3-NVFP4
|
||||||
|
SERVED_MODEL_NAME=/data/hf_models/GLM-5.3-NVFP4
|
||||||
|
PORT=30000
|
||||||
|
HEALTH_PATH=/health
|
||||||
|
HEALTH_WAIT_S=2400
|
||||||
|
CONTAINER_PYTHON=python3
|
||||||
|
|
||||||
|
TP=8
|
||||||
|
DP=1
|
||||||
|
|
||||||
|
DEVICE_VARS="CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7"
|
||||||
|
ENGINE_ENV="PYTHONUNBUFFERED=1 HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1"
|
||||||
|
|
||||||
|
DOCKER_FLAGS="--gpus all --ipc=host --shm-size 64g --entrypoint '' -p ${PORT}:${PORT}"
|
||||||
|
VOLUMES="/data/hf_models:/data/hf_models:ro"
|
||||||
|
|
||||||
|
BOOTSTRAP="exec python3 -m sglang.launch_server ${LAUNCH_ARGS}"
|
||||||
|
|
||||||
|
LAUNCH_ARGS="--model-path ${MODEL_PATH} --tp-size ${TP} --mem-fraction-static 0.90 --max-running-requests 16 --chunked-prefill-size 8192 --max-prefill-tokens 16384 --disable-shared-experts-fusion --moe-runner-backend flashinfer_cutlass --disable-flashinfer-autotune --speculative-algorithm EAGLE --speculative-num-steps 4 --speculative-eagle-topk 1 --speculative-num-draft-tokens 5 --kv-cache-dtype fp8_e4m3 --enable-hierarchical-cache --hicache-ratio 3 --cuda-graph-max-bs-decode 8 --cuda-graph-bs-decode 1 2 3 4 6 8 --cuda-graph-max-bs-prefill 8 --context-length 270336 --reasoning-parser glm45 --tool-call-parser glm47 --host 0.0.0.0 --port ${PORT}"
|
||||||
@ -0,0 +1,424 @@
|
|||||||
|
# SPDX-License-Identifier: Apache-2.0
|
||||||
|
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
||||||
|
# Adapted from https://github.com/vllm-project/vllm/blob/v0.6.4.post1/vllm/distributed/device_communicators/custom_all_reduce.py
|
||||||
|
|
||||||
|
import ctypes
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
from contextlib import contextmanager
|
||||||
|
from functools import partial
|
||||||
|
from typing import Any, List, Optional, Union
|
||||||
|
|
||||||
|
import torch
|
||||||
|
import torch.distributed as dist
|
||||||
|
from torch.distributed import ProcessGroup
|
||||||
|
|
||||||
|
import sglang.srt.distributed.device_communicators.custom_all_reduce_ops as ops
|
||||||
|
from sglang.srt.distributed.device_communicators.cuda_wrapper import CudaRTLibrary
|
||||||
|
from sglang.srt.distributed.device_communicators.custom_all_reduce_utils import (
|
||||||
|
can_use_custom_all_reduce_with_nvlink,
|
||||||
|
is_weak_contiguous,
|
||||||
|
)
|
||||||
|
from sglang.srt.environ import envs
|
||||||
|
from sglang.srt.model_executor.runner_backend_utils.tc_piecewise_cuda_graph import (
|
||||||
|
is_in_tc_piecewise_cuda_graph,
|
||||||
|
)
|
||||||
|
from sglang.srt.utils import (
|
||||||
|
get_bool_env_var,
|
||||||
|
is_cuda,
|
||||||
|
is_hip,
|
||||||
|
is_musa,
|
||||||
|
log_info_on_rank0,
|
||||||
|
)
|
||||||
|
|
||||||
|
_is_cuda = is_cuda()
|
||||||
|
_is_hip = is_hip()
|
||||||
|
_is_musa = is_musa()
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
os.environ.setdefault("SGLANG_CUSTOM_ALLREDUCE_ALGO", "1stage") # SSKJ-PATCH: C++ dispatch no-ops when full_nvlink=False; force kernel launch
|
||||||
|
|
||||||
|
|
||||||
|
class CustomAllreduce:
|
||||||
|
_SUPPORTED_WORLD_SIZES = [2, 4, 6, 8]
|
||||||
|
_MAX_CAR_SIZE = 8192 * 1024
|
||||||
|
if _is_hip:
|
||||||
|
# crossover is at 16MB buffer size for ROCm
|
||||||
|
_MAX_CAR_SIZE = 2 * 8192 * 1024
|
||||||
|
if _is_musa:
|
||||||
|
# crossover is at 128MB buffer size for MUSA
|
||||||
|
_MAX_CAR_SIZE = 16 * 8196 * 1024
|
||||||
|
|
||||||
|
# max_size: max supported allreduce size
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
group: ProcessGroup,
|
||||||
|
device: Union[int, str, torch.device],
|
||||||
|
max_size=_MAX_CAR_SIZE,
|
||||||
|
) -> None:
|
||||||
|
"""
|
||||||
|
Args:
|
||||||
|
group: the process group to work on. If None, it will use the
|
||||||
|
default process group.
|
||||||
|
device: the device to bind the CustomAllreduce to. If None,
|
||||||
|
it will be bind to f"cuda:{local_rank}".
|
||||||
|
It is the caller's responsibility to make sure each communicator
|
||||||
|
is bind to a unique device, and all communicators in this group
|
||||||
|
are in the same node.
|
||||||
|
"""
|
||||||
|
self._IS_CAPTURING = False
|
||||||
|
self.disabled = True # This can be modified in-place by context manager in piecewise cuda graph runner
|
||||||
|
self.original_disabled = True # To store the original state
|
||||||
|
self.use_amd_deterministic_impl = _use_amd_deterministic_impl()
|
||||||
|
|
||||||
|
if not ops.IS_CUSTOM_AR_AVAILABLE:
|
||||||
|
# disable because of missing custom allreduce library
|
||||||
|
# e.g. in a non-cuda environment
|
||||||
|
return
|
||||||
|
|
||||||
|
rank = dist.get_rank(group=group)
|
||||||
|
world_size = dist.get_world_size(group=group)
|
||||||
|
|
||||||
|
if isinstance(device, int):
|
||||||
|
device = torch.device(f"cuda:{device}")
|
||||||
|
elif isinstance(device, str):
|
||||||
|
device = torch.device(device)
|
||||||
|
# now `device` is a `torch.device` object
|
||||||
|
assert isinstance(device, torch.device)
|
||||||
|
self.device = device
|
||||||
|
full_nvlink = can_use_custom_all_reduce_with_nvlink(
|
||||||
|
group=group,
|
||||||
|
device=device,
|
||||||
|
supported_world_size=self._SUPPORTED_WORLD_SIZES,
|
||||||
|
cls_name="CustomAllreduce",
|
||||||
|
)
|
||||||
|
if full_nvlink is None:
|
||||||
|
return # fail to get nvlink status
|
||||||
|
|
||||||
|
self.group = group
|
||||||
|
self.max_size = max_size
|
||||||
|
self.rank = rank
|
||||||
|
self.world_size = world_size
|
||||||
|
self.full_nvlink = full_nvlink
|
||||||
|
|
||||||
|
if not _is_hip:
|
||||||
|
# Buffers memory are owned by this Python class and passed to C++.
|
||||||
|
# Meta data composes of two parts: meta data for synchronization and a
|
||||||
|
# temporary buffer for storing intermediate allreduce results.
|
||||||
|
self.meta_ptrs = self.create_shared_buffer(
|
||||||
|
ops.meta_size() + max_size, group=group
|
||||||
|
)
|
||||||
|
# This is a pre-registered IPC buffer. In eager mode, input tensors
|
||||||
|
# are first copied into this buffer before allreduce is performed
|
||||||
|
self.buffer_ptrs = self.create_shared_buffer(max_size, group=group)
|
||||||
|
# This is a buffer for storing the tuples of pointers pointing to
|
||||||
|
# IPC buffers from all ranks. Each registered tuple has size of
|
||||||
|
# 8*world_size bytes where world_size is at most 8. Allocating 8MB
|
||||||
|
# is enough for 131072 such tuples. The largest model I've seen only
|
||||||
|
# needs less than 10000 of registered tuples.
|
||||||
|
self.rank_data = torch.empty(
|
||||||
|
max_size, dtype=torch.uint8, device=self.device
|
||||||
|
)
|
||||||
|
self._ptr = ops.init_custom_ar(
|
||||||
|
self.meta_ptrs, self.rank_data, rank, self.full_nvlink
|
||||||
|
)
|
||||||
|
ops.register_buffer(self._ptr, self.buffer_ptrs)
|
||||||
|
else:
|
||||||
|
# meta data buffers need to be "uncached" for signal on MI200
|
||||||
|
self.meta = ops.allocate_meta_buffer(ops.meta_size() + max_size)
|
||||||
|
self.buffer = torch.empty(max_size, dtype=torch.uint8, device=self.device)
|
||||||
|
handle = ops.get_meta_buffer_ipc_handle(self.meta)
|
||||||
|
shard_data = (
|
||||||
|
bytes(handle), # ipc handle to base ptr
|
||||||
|
0, # offset of base ptr
|
||||||
|
)
|
||||||
|
handles, offsets = self._gather_ipc_meta(shard_data)
|
||||||
|
self.rank_data = torch.empty(
|
||||||
|
max_size, dtype=torch.uint8, device=self.device
|
||||||
|
)
|
||||||
|
self._ptr = ops.init_custom_ar(
|
||||||
|
self.meta, self.rank_data, handles, offsets, rank, self.full_nvlink
|
||||||
|
)
|
||||||
|
self.register_buffer(self.buffer)
|
||||||
|
|
||||||
|
self.disabled = False
|
||||||
|
self.original_disabled = False # Ensure original_disabled == disabled
|
||||||
|
logger.warning(f"SSKJ_CAR_PATCH_ACTIVE ws={self.world_size} full_nvlink={self.full_nvlink}")
|
||||||
|
self.tms_cudagraph = envs.SGLANG_MEMORY_SAVER_CUDA_GRAPH.get()
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def create_shared_buffer(
|
||||||
|
size_in_bytes: int, group: Optional[ProcessGroup] = None
|
||||||
|
) -> List[int]:
|
||||||
|
"""
|
||||||
|
Creates a shared buffer and returns a list of pointers
|
||||||
|
representing the buffer on all processes in the group.
|
||||||
|
"""
|
||||||
|
lib = CudaRTLibrary()
|
||||||
|
pointer = lib.cudaMalloc(size_in_bytes)
|
||||||
|
if _is_musa:
|
||||||
|
lib.cudaMemset(pointer, 0, size_in_bytes)
|
||||||
|
handle = lib.cudaIpcGetMemHandle(pointer)
|
||||||
|
world_size = dist.get_world_size(group=group)
|
||||||
|
rank = dist.get_rank(group=group)
|
||||||
|
handles = [None] * world_size
|
||||||
|
dist.all_gather_object(handles, handle, group=group)
|
||||||
|
|
||||||
|
pointers: List[int] = []
|
||||||
|
for i, h in enumerate(handles):
|
||||||
|
if i == rank:
|
||||||
|
pointers.append(pointer.value) # type: ignore
|
||||||
|
else:
|
||||||
|
pointers.append(lib.cudaIpcOpenMemHandle(h).value) # type: ignore
|
||||||
|
|
||||||
|
return pointers
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def free_shared_buffer(
|
||||||
|
pointers: List[int], group: Optional[ProcessGroup] = None
|
||||||
|
) -> None:
|
||||||
|
rank = dist.get_rank(group=group)
|
||||||
|
lib = CudaRTLibrary()
|
||||||
|
lib.cudaFree(ctypes.c_void_p(pointers[rank]))
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def capture(self):
|
||||||
|
"""
|
||||||
|
The main responsibility of this context manager is the
|
||||||
|
`register_graph_buffers` call at the end of the context.
|
||||||
|
It records all the buffer addresses used in the CUDA graph.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
self._IS_CAPTURING = True
|
||||||
|
yield
|
||||||
|
finally:
|
||||||
|
self._IS_CAPTURING = False
|
||||||
|
if not self.disabled:
|
||||||
|
self.register_graph_buffers()
|
||||||
|
|
||||||
|
def _get_ipc_meta(self, inp: torch.Tensor):
|
||||||
|
# _share_cuda_() doesn't accept meta buffer not allocated from
|
||||||
|
# PyTorch cache allocator, use direct HIP call to get IPC handle
|
||||||
|
handle = ops.get_meta_buffer_ipc_handle(inp)
|
||||||
|
shard_data = (
|
||||||
|
bytes(handle), # ipc handle to base ptr
|
||||||
|
0, # offset of base ptr
|
||||||
|
)
|
||||||
|
return self._gather_ipc_meta(shard_data)
|
||||||
|
|
||||||
|
def _gather_ipc_meta(self, shard_data):
|
||||||
|
# Note: don't use `[[None]] * self.world_size` here
|
||||||
|
# because it will create a list of the same reference
|
||||||
|
all_data: List[Optional[Any]] = [[None] for i in range(self.world_size)]
|
||||||
|
all_data[self.rank][0] = shard_data
|
||||||
|
|
||||||
|
ranks = dist.get_process_group_ranks(group=self.group)
|
||||||
|
ranks.sort()
|
||||||
|
for i, rank in enumerate(ranks):
|
||||||
|
dist.broadcast_object_list(
|
||||||
|
all_data[i], src=rank, group=self.group, device="cpu"
|
||||||
|
)
|
||||||
|
|
||||||
|
# we cannot directly use `dist.all_gather_object` here
|
||||||
|
# because it is incompatible with `gloo` backend under inference mode.
|
||||||
|
# see https://github.com/pytorch/pytorch/issues/126032 for details.
|
||||||
|
|
||||||
|
handles = []
|
||||||
|
offsets = []
|
||||||
|
for i in range(len(all_data)):
|
||||||
|
handles.append(all_data[i][0][0]) # type: ignore
|
||||||
|
offsets.append(all_data[i][0][1]) # type: ignore
|
||||||
|
return handles, offsets
|
||||||
|
|
||||||
|
def register_buffer(self, inp: torch.Tensor):
|
||||||
|
handles, offsets = self._get_ipc_meta(inp)
|
||||||
|
ops.register_buffer(self._ptr, inp, handles, offsets)
|
||||||
|
|
||||||
|
def register_graph_buffers(self):
|
||||||
|
if _is_hip:
|
||||||
|
handle, offset = ops.get_graph_buffer_ipc_meta(self._ptr)
|
||||||
|
handles, offsets = self._gather_ipc_meta((bytes(handle), offset))
|
||||||
|
log_info_on_rank0(logger, f"Registering {len(offset)} cuda graph addresses")
|
||||||
|
ops.register_graph_buffers(self._ptr, handles, offsets)
|
||||||
|
else:
|
||||||
|
handle, offset = ops.get_graph_buffer_ipc_meta(self._ptr)
|
||||||
|
log_info_on_rank0(logger, f"Registering {len(offset)} cuda graph addresses")
|
||||||
|
# We cannot directly use `dist.all_gather_object` here
|
||||||
|
# because it is incompatible with `gloo` backend under inference mode.
|
||||||
|
# see https://github.com/pytorch/pytorch/issues/126032 for details.
|
||||||
|
all_data = [
|
||||||
|
[None, None] for _ in range(dist.get_world_size(group=self.group))
|
||||||
|
]
|
||||||
|
all_data[self.rank] = [handle, offset]
|
||||||
|
ranks = sorted(dist.get_process_group_ranks(group=self.group))
|
||||||
|
for i, rank in enumerate(ranks):
|
||||||
|
dist.broadcast_object_list(
|
||||||
|
all_data[i], src=rank, group=self.group, device="cpu"
|
||||||
|
)
|
||||||
|
# Unpack list of tuples to tuple of lists.
|
||||||
|
handles = [d[0] for d in all_data] # type: ignore
|
||||||
|
offsets = [d[1] for d in all_data] # type: ignore
|
||||||
|
ops.register_graph_buffers(self._ptr, handles, offsets)
|
||||||
|
|
||||||
|
def should_custom_ar(self, inp: torch.Tensor):
|
||||||
|
if self.disabled:
|
||||||
|
return False
|
||||||
|
inp_size = inp.numel() * inp.element_size()
|
||||||
|
# custom allreduce requires input byte size to be multiples of 16
|
||||||
|
if inp_size % 16 != 0:
|
||||||
|
return False
|
||||||
|
if not is_weak_contiguous(inp):
|
||||||
|
return False
|
||||||
|
# for 4 or more non NVLink-capable GPUs, custom allreduce provides
|
||||||
|
# little performance improvement over NCCL.
|
||||||
|
if not _is_hip:
|
||||||
|
if True:
|
||||||
|
return inp_size <= self.max_size
|
||||||
|
return False
|
||||||
|
|
||||||
|
if _is_hip:
|
||||||
|
if self.use_amd_deterministic_impl:
|
||||||
|
return True
|
||||||
|
if self.full_nvlink:
|
||||||
|
return inp_size <= self.max_size
|
||||||
|
return False
|
||||||
|
|
||||||
|
return False
|
||||||
|
|
||||||
|
def _all_reduce_impl(self, inp: torch.Tensor, registered: bool):
|
||||||
|
out = torch.empty_like(inp)
|
||||||
|
if not _is_hip: # CUDA-like
|
||||||
|
if registered:
|
||||||
|
ops.all_reduce(self._ptr, inp, out, 0, 0)
|
||||||
|
else:
|
||||||
|
ops.all_reduce(
|
||||||
|
self._ptr, inp, out, self.buffer_ptrs[self.rank], self.max_size
|
||||||
|
)
|
||||||
|
elif self.use_amd_deterministic_impl:
|
||||||
|
inp_size = inp.numel() * inp.element_size()
|
||||||
|
if inp_size < self.max_size:
|
||||||
|
reg_buffer = self.buffer.view(inp.dtype)[: inp.numel()]
|
||||||
|
ops.deterministic_all_reduce_unreg(self._ptr, inp, reg_buffer, out)
|
||||||
|
else:
|
||||||
|
self.register_buffer(inp)
|
||||||
|
ops.deterministic_all_reduce_reg(self._ptr, inp, out)
|
||||||
|
else: # normal AMD ROCm path
|
||||||
|
if registered:
|
||||||
|
ops.all_reduce_reg(self._ptr, inp, out)
|
||||||
|
else:
|
||||||
|
ops.all_reduce_unreg(self._ptr, inp, self.buffer, out)
|
||||||
|
return out
|
||||||
|
|
||||||
|
def custom_all_reduce(self, input: torch.Tensor) -> Optional[torch.Tensor]:
|
||||||
|
"""The main allreduce API that provides support for cuda graph."""
|
||||||
|
# When custom allreduce is disabled, this will be None.
|
||||||
|
if self.disabled or not self.should_custom_ar(input):
|
||||||
|
return None
|
||||||
|
if self._IS_CAPTURING:
|
||||||
|
if torch.cuda.is_current_stream_capturing():
|
||||||
|
return self._all_reduce_impl(input, registered=not self.tms_cudagraph)
|
||||||
|
else:
|
||||||
|
# Could be warmup OR piecewise cuda graph split op execution.
|
||||||
|
# In piecewise cuda graph, split ops run eagerly outside the graph
|
||||||
|
# but _IS_CAPTURING is still True. We need to do real all-reduce.
|
||||||
|
if is_in_tc_piecewise_cuda_graph():
|
||||||
|
# Split op execution - do real all-reduce
|
||||||
|
return self._all_reduce_impl(input, registered=False)
|
||||||
|
else:
|
||||||
|
# True warmup - mimic the allocation pattern since custom
|
||||||
|
# allreduce is out-of-place.
|
||||||
|
return torch.zeros_like(input)
|
||||||
|
else:
|
||||||
|
return self._all_reduce_impl(input, registered=False)
|
||||||
|
|
||||||
|
def close(self):
|
||||||
|
if not self.disabled and self._ptr:
|
||||||
|
if ops is not None:
|
||||||
|
ops.dispose(self._ptr)
|
||||||
|
if _is_cuda:
|
||||||
|
self.free_shared_buffer(self.meta_ptrs)
|
||||||
|
self.free_shared_buffer(self.buffer_ptrs)
|
||||||
|
self._ptr = 0
|
||||||
|
|
||||||
|
def __del__(self):
|
||||||
|
self.close()
|
||||||
|
|
||||||
|
|
||||||
|
def dispatch_custom_allreduce(
|
||||||
|
group: ProcessGroup,
|
||||||
|
device: torch.device,
|
||||||
|
):
|
||||||
|
"""Return the CustomAllreduce class to use (aiter on ROCm if enabled).
|
||||||
|
|
||||||
|
On AMD with 1-stage AR enabled, use sglang's CustomAllreduce.
|
||||||
|
Otherwise use AiterCustomAllreduce if available.
|
||||||
|
|
||||||
|
On CUDA, the JIT-compiled v2 implementation is used by default.
|
||||||
|
Set SGLANG_OPT_USE_CUSTOM_ALL_REDUCE_V2=0 to fall back to the legacy CustomAllreduce.
|
||||||
|
Multi-node v2 is admitted only for a single NVLink clique (see
|
||||||
|
can_use_custom_all_reduce_v2); other cross-node groups fall back to NCCL.
|
||||||
|
"""
|
||||||
|
if _is_cuda and envs.SGLANG_OPT_USE_CUSTOM_ALL_REDUCE_V2.get():
|
||||||
|
from .custom_all_reduce_v2 import (
|
||||||
|
CustomAllReduceV2,
|
||||||
|
can_use_custom_all_reduce_v2,
|
||||||
|
)
|
||||||
|
|
||||||
|
if can_use_custom_all_reduce_v2(group=group, device=device):
|
||||||
|
logger.debug("[AR] Using CustomAllReduceV2 (JIT-compiled)")
|
||||||
|
return CustomAllReduceV2
|
||||||
|
|
||||||
|
if _is_cuda or _is_musa:
|
||||||
|
return CustomAllreduce
|
||||||
|
|
||||||
|
assert _is_hip
|
||||||
|
|
||||||
|
if envs.SGLANG_USE_1STAGE_ALLREDUCE.is_set():
|
||||||
|
if envs.SGLANG_USE_1STAGE_ALLREDUCE.get():
|
||||||
|
logger.debug(
|
||||||
|
"[AR] All-reduce: 1-stage kernel (SGLANG_USE_1STAGE_ALLREDUCE=1)"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
logger.debug("[AR] All-reduce: default (SGLANG_USE_1STAGE_ALLREDUCE=0)")
|
||||||
|
elif envs.SGLANG_ENABLE_DETERMINISTIC_INFERENCE.get():
|
||||||
|
logger.debug(
|
||||||
|
"[AR] All-reduce: 1-stage kernel (deterministic inference enabled)"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
logger.debug("[AR] All-reduce: default")
|
||||||
|
|
||||||
|
# On AMD with 1-stage AR, use sglang's CustomAllreduce
|
||||||
|
# (AiterCustomAllreduce doesn't have deterministic_all_reduce method)
|
||||||
|
if _use_amd_deterministic_impl():
|
||||||
|
return CustomAllreduce
|
||||||
|
|
||||||
|
if get_bool_env_var("SGLANG_USE_AITER_AR", default="true"):
|
||||||
|
try:
|
||||||
|
from aiter.dist.device_communicators.custom_all_reduce import (
|
||||||
|
CustomAllreduce as AiterCustomAllreduce,
|
||||||
|
)
|
||||||
|
|
||||||
|
logger.info("[AR] Using AiterCustomAllreduce (AMD default)")
|
||||||
|
tms_cudagraph = envs.SGLANG_MEMORY_SAVER_CUDA_GRAPH.get()
|
||||||
|
return partial(
|
||||||
|
AiterCustomAllreduce,
|
||||||
|
enable_register_for_capturing=not tms_cudagraph,
|
||||||
|
)
|
||||||
|
except ImportError as e:
|
||||||
|
logger.warning(
|
||||||
|
"[AR] Aiter custom all-reduce not available; "
|
||||||
|
"falling back to sglang CustomAllreduce. Details: %s",
|
||||||
|
e,
|
||||||
|
)
|
||||||
|
return CustomAllreduce
|
||||||
|
|
||||||
|
return CustomAllreduce
|
||||||
|
|
||||||
|
|
||||||
|
def _use_amd_deterministic_impl() -> bool:
|
||||||
|
if not _is_hip: # CUDA is always deterministic
|
||||||
|
return False
|
||||||
|
if envs.SGLANG_USE_1STAGE_ALLREDUCE.is_set():
|
||||||
|
return envs.SGLANG_USE_1STAGE_ALLREDUCE.get()
|
||||||
|
else:
|
||||||
|
return envs.SGLANG_ENABLE_DETERMINISTIC_INFERENCE.get()
|
||||||
@ -0,0 +1,519 @@
|
|||||||
|
# SPDX-License-Identifier: Apache-2.0
|
||||||
|
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
||||||
|
# Adapted from https://github.com/vllm-project/vllm/blob/v0.6.4.post1/vllm/distributed/device_communicators/custom_all_reduce_utils.py
|
||||||
|
|
||||||
|
import ctypes
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import pickle
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
from functools import wraps
|
||||||
|
from itertools import product
|
||||||
|
from typing import Callable, Dict, List, Optional, Sequence, TypeVar
|
||||||
|
|
||||||
|
import torch
|
||||||
|
import torch.distributed as dist
|
||||||
|
import torch.multiprocessing as mp
|
||||||
|
from typing_extensions import ParamSpec
|
||||||
|
|
||||||
|
from sglang.srt.distributed.device_communicators.cuda_wrapper import CudaRTLibrary
|
||||||
|
from sglang.srt.distributed.parallel_state import in_the_same_node_as
|
||||||
|
from sglang.srt.environ import envs as sglang_envs
|
||||||
|
from sglang.srt.utils import is_cuda, is_hip, is_musa
|
||||||
|
from sglang.srt.utils.cuda_vmm_utils import _gpu_fabric_clique
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
_is_cuda = is_cuda()
|
||||||
|
_is_hip = is_hip()
|
||||||
|
_is_musa = is_musa()
|
||||||
|
|
||||||
|
if _is_cuda:
|
||||||
|
try:
|
||||||
|
import pynvml
|
||||||
|
except ImportError as e:
|
||||||
|
logger.warning("Failed to import pynvml with %r", e)
|
||||||
|
|
||||||
|
if _is_musa:
|
||||||
|
try:
|
||||||
|
import pymtml as pynvml
|
||||||
|
except ImportError as e:
|
||||||
|
logger.warning("Failed to import pymtml with %r", e)
|
||||||
|
|
||||||
|
if _is_hip:
|
||||||
|
try:
|
||||||
|
from amdsmi import (
|
||||||
|
AmdSmiException,
|
||||||
|
amdsmi_get_processor_handles,
|
||||||
|
amdsmi_init,
|
||||||
|
amdsmi_shut_down,
|
||||||
|
amdsmi_topo_get_link_type,
|
||||||
|
)
|
||||||
|
except ImportError as e:
|
||||||
|
logger.warning("Failed to import amdsmi with %r", e)
|
||||||
|
|
||||||
|
_P = ParamSpec("_P")
|
||||||
|
_R = TypeVar("_R")
|
||||||
|
|
||||||
|
|
||||||
|
def update_environment_variables(envs: Dict[str, str]):
|
||||||
|
for k, v in envs.items():
|
||||||
|
if k in os.environ and os.environ[k] != v:
|
||||||
|
logger.warning(
|
||||||
|
"Overwriting environment variable %s " "from '%s' to '%s'",
|
||||||
|
k,
|
||||||
|
os.environ[k],
|
||||||
|
v,
|
||||||
|
)
|
||||||
|
os.environ[k] = v
|
||||||
|
|
||||||
|
|
||||||
|
def producer(
|
||||||
|
batch_src: Sequence[int],
|
||||||
|
producer_queue,
|
||||||
|
consumer_queue,
|
||||||
|
result_queue,
|
||||||
|
cuda_visible_devices: Optional[str] = None,
|
||||||
|
):
|
||||||
|
if cuda_visible_devices is not None:
|
||||||
|
update_environment_variables({"CUDA_VISIBLE_DEVICES": cuda_visible_devices})
|
||||||
|
|
||||||
|
lib = CudaRTLibrary()
|
||||||
|
for i in batch_src:
|
||||||
|
lib.cudaSetDevice(i)
|
||||||
|
pointer = lib.cudaMalloc(1024)
|
||||||
|
lib.cudaMemset(pointer, 1, 1024)
|
||||||
|
lib.cudaDeviceSynchronize()
|
||||||
|
handle = lib.cudaIpcGetMemHandle(pointer)
|
||||||
|
producer_queue.put(handle)
|
||||||
|
open_success = consumer_queue.get()
|
||||||
|
if open_success:
|
||||||
|
# use two queues to simulate barrier
|
||||||
|
producer_queue.put(0)
|
||||||
|
consumer_queue.get()
|
||||||
|
# check if the memory is modified
|
||||||
|
host_data = (ctypes.c_char * 1024)()
|
||||||
|
lib.cudaMemcpy(host_data, pointer, 1024) # type: ignore
|
||||||
|
for i in range(1024):
|
||||||
|
if ord(host_data[i]) != 2:
|
||||||
|
open_success = False
|
||||||
|
break
|
||||||
|
result_queue.put(open_success)
|
||||||
|
lib.cudaDeviceReset()
|
||||||
|
|
||||||
|
|
||||||
|
def consumer(
|
||||||
|
batch_tgt: Sequence[int],
|
||||||
|
producer_queue,
|
||||||
|
consumer_queue,
|
||||||
|
result_queue,
|
||||||
|
cuda_visible_devices: Optional[str] = None,
|
||||||
|
):
|
||||||
|
if cuda_visible_devices is not None:
|
||||||
|
update_environment_variables({"CUDA_VISIBLE_DEVICES": cuda_visible_devices})
|
||||||
|
|
||||||
|
lib = CudaRTLibrary()
|
||||||
|
for j in batch_tgt:
|
||||||
|
lib.cudaSetDevice(j)
|
||||||
|
handle = producer_queue.get()
|
||||||
|
open_success = False
|
||||||
|
try:
|
||||||
|
pointer = lib.cudaIpcOpenMemHandle(handle) # type: ignore
|
||||||
|
open_success = True
|
||||||
|
except RuntimeError:
|
||||||
|
# cannot error out here, because the producer process
|
||||||
|
# is still waiting for the response.
|
||||||
|
pass
|
||||||
|
consumer_queue.put(open_success)
|
||||||
|
if open_success:
|
||||||
|
# modify the memory
|
||||||
|
lib.cudaMemset(pointer, 2, 1024)
|
||||||
|
lib.cudaDeviceSynchronize()
|
||||||
|
# use two queues to simulate barrier
|
||||||
|
producer_queue.get()
|
||||||
|
consumer_queue.put(0)
|
||||||
|
# check if the memory is modified
|
||||||
|
host_data = (ctypes.c_char * 1024)()
|
||||||
|
lib.cudaMemcpy(host_data, pointer, 1024) # type: ignore
|
||||||
|
for i in range(1024):
|
||||||
|
if ord(host_data[i]) != 2:
|
||||||
|
open_success = False
|
||||||
|
break
|
||||||
|
result_queue.put(open_success)
|
||||||
|
lib.cudaDeviceReset()
|
||||||
|
|
||||||
|
|
||||||
|
def can_actually_p2p(
|
||||||
|
batch_src: Sequence[int],
|
||||||
|
batch_tgt: Sequence[int],
|
||||||
|
) -> Sequence[bool]:
|
||||||
|
"""
|
||||||
|
Usually, checking if P2P access is enabled can be done by
|
||||||
|
`torch.cuda.can_device_access_peer(src, tgt)`. However, sometimes
|
||||||
|
the driver might be broken, and `torch.cuda.can_device_access_peer(src, tgt)`
|
||||||
|
returns `True` even if P2P access is not actually possible.
|
||||||
|
See https://github.com/vllm-project/vllm/issues/2728 and
|
||||||
|
https://forums.developer.nvidia.com/t/direct-gpu-gpu-communication-does-not-seem-to-work-properly/283264/10
|
||||||
|
Therefore, we have to perform a real P2P access to check if it is actually
|
||||||
|
possible.
|
||||||
|
|
||||||
|
Note on p2p and cuda IPC:
|
||||||
|
Usually, one process uses one GPU:
|
||||||
|
GPU src --> cuda context src --> tensor src --> process src
|
||||||
|
|
||||||
|
We need to combine p2p and cuda IPC, so that:
|
||||||
|
GPU src --> cuda context src --> tensor src --> process src
|
||||||
|
|shared|
|
||||||
|
GPU tgt --> cuda context tgt --> tensor tgt --> process tgt
|
||||||
|
That is to say, process src creates a tensor in GPU src, passes IPC handle to
|
||||||
|
process tgt, and process tgt accesses the tensor in GPU tgt. Any operation on the
|
||||||
|
tensor in process tgt will be reflected in the tensor in process src, because
|
||||||
|
they are the same memory segment.
|
||||||
|
It is important to note that process tgt accesses the tensor in GPU tgt, not
|
||||||
|
GPU src. That's why we need p2p access.
|
||||||
|
|
||||||
|
The most time-consuming part is the process creation. To avoid creating
|
||||||
|
processes for every pair of GPUs, we use batched testing. We create two
|
||||||
|
processes for testing all pairs of GPUs in batch. The trick is to reset
|
||||||
|
the device after each test (which is not available in PyTorch).
|
||||||
|
""" # noqa
|
||||||
|
cuda_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES", None)
|
||||||
|
# pass the CUDA_VISIBLE_DEVICES to the child process
|
||||||
|
# to make sure they see the same set of GPUs
|
||||||
|
|
||||||
|
# make sure the processes are spawned
|
||||||
|
smp = mp.get_context("spawn")
|
||||||
|
producer_queue = smp.Queue()
|
||||||
|
consumer_queue = smp.Queue()
|
||||||
|
result_queue = smp.Queue()
|
||||||
|
p_src = smp.Process(
|
||||||
|
target=producer,
|
||||||
|
args=(
|
||||||
|
batch_src,
|
||||||
|
producer_queue,
|
||||||
|
consumer_queue,
|
||||||
|
result_queue,
|
||||||
|
cuda_visible_devices,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
p_tgt = smp.Process(
|
||||||
|
target=consumer,
|
||||||
|
args=(
|
||||||
|
batch_tgt,
|
||||||
|
producer_queue,
|
||||||
|
consumer_queue,
|
||||||
|
result_queue,
|
||||||
|
cuda_visible_devices,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
p_src.start()
|
||||||
|
p_tgt.start()
|
||||||
|
p_src.join()
|
||||||
|
p_tgt.join()
|
||||||
|
assert p_src.exitcode == 0 and p_tgt.exitcode == 0
|
||||||
|
result: List[bool] = []
|
||||||
|
for src, tgt in zip(batch_src, batch_tgt):
|
||||||
|
a = result_queue.get()
|
||||||
|
b = result_queue.get()
|
||||||
|
if a != b:
|
||||||
|
logger.warning(
|
||||||
|
"Two processes do not agree on the P2P access"
|
||||||
|
" status on %d -> %d, treat as disabled.",
|
||||||
|
src,
|
||||||
|
tgt,
|
||||||
|
)
|
||||||
|
result.append(False)
|
||||||
|
else:
|
||||||
|
result.append(a)
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
# why do we need this cache?
|
||||||
|
# we are testing peer-to-peer (p2p) access between GPUs,across processes.
|
||||||
|
# if we test it every time, it will be very slow, because we need to create
|
||||||
|
# N * N * 2 processes, where N is the world size. This is very slow.
|
||||||
|
# to reduce the time, we use a cache file to store the p2p access status.
|
||||||
|
# the cache file is generated by the master process if it does not exist.
|
||||||
|
# then all the processes can read the cache file to check the p2p access status.
|
||||||
|
# Note that the cache file is suffixed by the CUDA_VISIBLE_DEVICES, so that we
|
||||||
|
# can have different cache files for different CUDA_VISIBLE_DEVICES settings,
|
||||||
|
# e.g. used by different vllm engines. The device id in the cache file is a
|
||||||
|
# **local** device id, i.e. from 0 to num_dev-1, where num_dev is the number
|
||||||
|
# of visible devices in the vllm engine.
|
||||||
|
_gpu_p2p_access_cache: Optional[Dict[str, bool]] = None
|
||||||
|
|
||||||
|
|
||||||
|
def gpu_p2p_access_check(src: int, tgt: int) -> bool:
|
||||||
|
"""Check if GPU src can access GPU tgt."""
|
||||||
|
|
||||||
|
# if the cache variable is already calculated,
|
||||||
|
# read from the cache instead of checking it again
|
||||||
|
global _gpu_p2p_access_cache
|
||||||
|
if _gpu_p2p_access_cache is not None:
|
||||||
|
return _gpu_p2p_access_cache[f"{src}->{tgt}"]
|
||||||
|
|
||||||
|
is_distributed = dist.is_initialized()
|
||||||
|
|
||||||
|
num_dev = torch.cuda.device_count()
|
||||||
|
cuda_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES", None)
|
||||||
|
if cuda_visible_devices is None:
|
||||||
|
cuda_visible_devices = ",".join(str(i) for i in range(num_dev))
|
||||||
|
|
||||||
|
# VLLM_CACHE_ROOT -> SGLANG_CACHE_ROOT
|
||||||
|
# "~/.cache/vllm" -> envs.SGLANG_CACHE_DIR
|
||||||
|
SGLANG_CACHE_ROOT = os.path.expanduser(sglang_envs.SGLANG_CACHE_DIR.get())
|
||||||
|
path = os.path.join(
|
||||||
|
SGLANG_CACHE_ROOT, f"gpu_p2p_access_cache_for_{cuda_visible_devices}.json"
|
||||||
|
)
|
||||||
|
cache_dir = os.path.dirname(path)
|
||||||
|
try:
|
||||||
|
os.makedirs(cache_dir, exist_ok=True)
|
||||||
|
except (FileExistsError, NotADirectoryError):
|
||||||
|
if not os.path.isdir(cache_dir):
|
||||||
|
# Path exists as a file (stale cache/lock). Remove and retry.
|
||||||
|
try:
|
||||||
|
os.remove(cache_dir)
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
os.makedirs(cache_dir, exist_ok=True)
|
||||||
|
from sglang.srt.distributed.parallel_state import get_world_group
|
||||||
|
|
||||||
|
if (not is_distributed or get_world_group().local_rank == 0) and (
|
||||||
|
not os.path.exists(path)
|
||||||
|
):
|
||||||
|
# only the local master process (with local_rank == 0) can
|
||||||
|
# enter this block to calculate the cache
|
||||||
|
logger.info("generating GPU P2P access cache in %s", path)
|
||||||
|
cache: Dict[str, bool] = {}
|
||||||
|
ids = list(range(num_dev))
|
||||||
|
# batch of all pairs of GPUs
|
||||||
|
batch_src, batch_tgt = zip(*list(product(ids, ids)))
|
||||||
|
# NOTE: we use `subprocess` rather than `multiprocessing` here
|
||||||
|
# because the caller might not have `if __name__ == "__main__":`,
|
||||||
|
# in that case we cannot use spawn method in multiprocessing.
|
||||||
|
# However, `can_actually_p2p` requires spawn method.
|
||||||
|
# The fix is, we use `subprocess` to call the function,
|
||||||
|
# where we have `if __name__ == "__main__":` in this file.
|
||||||
|
|
||||||
|
# use a temporary file to store the result
|
||||||
|
# we don't use the output of the subprocess directly,
|
||||||
|
# because the subprocess might produce logging output
|
||||||
|
with tempfile.NamedTemporaryFile() as output_file:
|
||||||
|
input_bytes = pickle.dumps((batch_src, batch_tgt, output_file.name))
|
||||||
|
returned = subprocess.run(
|
||||||
|
[sys.executable, __file__], input=input_bytes, capture_output=True
|
||||||
|
)
|
||||||
|
# check if the subprocess is successful
|
||||||
|
try:
|
||||||
|
returned.check_returncode()
|
||||||
|
except Exception as e:
|
||||||
|
# wrap raised exception to provide more information
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Error happened when batch testing "
|
||||||
|
f"peer-to-peer access from {batch_src} to {batch_tgt}:\n"
|
||||||
|
f"{returned.stderr.decode()}"
|
||||||
|
) from e
|
||||||
|
with open(output_file.name, "rb") as f:
|
||||||
|
result = pickle.load(f)
|
||||||
|
for _i, _j, r in zip(batch_src, batch_tgt, result):
|
||||||
|
cache[f"{_i}->{_j}"] = r
|
||||||
|
with open(path, "w") as f:
|
||||||
|
json.dump(cache, f, indent=4)
|
||||||
|
if is_distributed:
|
||||||
|
get_world_group().barrier()
|
||||||
|
logger.info("reading GPU P2P access cache from %s", path)
|
||||||
|
with open(path) as f:
|
||||||
|
cache = json.load(f)
|
||||||
|
_gpu_p2p_access_cache = cache
|
||||||
|
return _gpu_p2p_access_cache[f"{src}->{tgt}"]
|
||||||
|
|
||||||
|
|
||||||
|
def with_nvml_context(fn: Callable[_P, _R]) -> Callable[_P, _R]:
|
||||||
|
@wraps(fn)
|
||||||
|
def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
|
||||||
|
if _is_hip:
|
||||||
|
try:
|
||||||
|
amdsmi_init()
|
||||||
|
return fn(*args, **kwargs)
|
||||||
|
finally:
|
||||||
|
amdsmi_shut_down()
|
||||||
|
else:
|
||||||
|
pynvml.nvmlInit()
|
||||||
|
try:
|
||||||
|
return fn(*args, **kwargs)
|
||||||
|
finally:
|
||||||
|
pynvml.nvmlShutdown()
|
||||||
|
|
||||||
|
return wrapper
|
||||||
|
|
||||||
|
|
||||||
|
@with_nvml_context
|
||||||
|
def is_full_nvlink(physical_device_ids: List[int], world_size: int) -> bool:
|
||||||
|
if _is_hip:
|
||||||
|
"""
|
||||||
|
query if the set of gpus are fully connected by xgmi (1 hop)
|
||||||
|
"""
|
||||||
|
handles = [amdsmi_get_processor_handles()[i] for i in physical_device_ids]
|
||||||
|
for i, handle in enumerate(handles):
|
||||||
|
for j, peer_handle in enumerate(handles):
|
||||||
|
if i < j:
|
||||||
|
try:
|
||||||
|
link_type = amdsmi_topo_get_link_type(handle, peer_handle)
|
||||||
|
# type is 2 for XGMI
|
||||||
|
if link_type["hops"] != 1 or link_type["type"] != 2:
|
||||||
|
return False
|
||||||
|
except AmdSmiException as error:
|
||||||
|
logger.error("AMD 1 hop XGMI detection failed.", exc_info=error)
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
else:
|
||||||
|
"""
|
||||||
|
query if the set of gpus are fully connected by nvlink (1 hop)
|
||||||
|
"""
|
||||||
|
handles = [pynvml.nvmlDeviceGetHandleByIndex(i) for i in physical_device_ids]
|
||||||
|
for i, handle in enumerate(handles):
|
||||||
|
for j, peer_handle in enumerate(handles):
|
||||||
|
if i < j:
|
||||||
|
try:
|
||||||
|
p2p_status = pynvml.nvmlDeviceGetP2PStatus(
|
||||||
|
handle, peer_handle, pynvml.NVML_P2P_CAPS_INDEX_NVLINK
|
||||||
|
)
|
||||||
|
if p2p_status != pynvml.NVML_P2P_STATUS_OK:
|
||||||
|
return False
|
||||||
|
except pynvml.NVMLError:
|
||||||
|
logger.exception(
|
||||||
|
"NVLink detection failed. This is normal if your"
|
||||||
|
" machine has no NVLink equipped."
|
||||||
|
)
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
@with_nvml_context
|
||||||
|
def is_one_nvlink_clique(
|
||||||
|
group: torch.distributed.ProcessGroup, device: torch.device
|
||||||
|
) -> bool:
|
||||||
|
"""True iff every rank's GPU is in the same NVLink fabric clique (one NVL72 /
|
||||||
|
MNNVL domain). Such a clique shares a single NVLink address space even across
|
||||||
|
nodes, so custom-AR v2's symm-mem storage + fabric peer VAs are valid group-wide."""
|
||||||
|
if _is_hip:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
clique = _gpu_fabric_clique(device)
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(
|
||||||
|
"GPU fabric clique query failed (%r); custom-AR stays intra-node.", e
|
||||||
|
)
|
||||||
|
clique = None
|
||||||
|
# Always all-gather (every rank calls it once) so a failed query on any rank
|
||||||
|
# resolves to a clean False rather than a collective mismatch.
|
||||||
|
world_size = dist.get_world_size(group=group)
|
||||||
|
gathered: List[object] = [None] * world_size
|
||||||
|
dist.all_gather_object(gathered, clique, group=group)
|
||||||
|
if any(c is None for c in gathered):
|
||||||
|
return False
|
||||||
|
return len(set(gathered)) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def is_weak_contiguous(inp: torch.Tensor):
|
||||||
|
return inp.is_contiguous() or (
|
||||||
|
inp.storage().nbytes() - inp.storage_offset() * inp.element_size()
|
||||||
|
== inp.numel() * inp.element_size()
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def can_p2p(rank: int, world_size: int) -> bool:
|
||||||
|
# SGLANG_SKIP_P2P_CHECK can be set to False in sglang
|
||||||
|
SGLANG_SKIP_P2P_CHECK = os.getenv("SGLANG_SKIP_P2P_CHECK", "0") == "1"
|
||||||
|
for i in range(world_size):
|
||||||
|
if i == rank:
|
||||||
|
continue
|
||||||
|
if SGLANG_SKIP_P2P_CHECK:
|
||||||
|
logger.info("Skipping P2P check and trusting the driver's P2P report.")
|
||||||
|
return torch.cuda.can_device_access_peer(rank, i)
|
||||||
|
if not gpu_p2p_access_check(rank, i):
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def can_use_custom_all_reduce_with_nvlink(
|
||||||
|
group: torch.distributed.ProcessGroup,
|
||||||
|
device: torch.device,
|
||||||
|
supported_world_size: List[int],
|
||||||
|
cls_name: str,
|
||||||
|
) -> Optional[bool]: # None if fail; otherwise return whether NVLink is available
|
||||||
|
assert (
|
||||||
|
dist.get_backend(group) != dist.Backend.NCCL
|
||||||
|
), f"{cls_name} should be attached to a non-NCCL group."
|
||||||
|
|
||||||
|
rank = dist.get_rank(group=group)
|
||||||
|
world_size = dist.get_world_size(group=group)
|
||||||
|
|
||||||
|
# No need to initialize custom allreduce for single GPU case.
|
||||||
|
if world_size == 1:
|
||||||
|
return
|
||||||
|
|
||||||
|
# No need to initialize custom allreduce for multi-node case.
|
||||||
|
if not all(in_the_same_node_as(group, source_rank=0)):
|
||||||
|
logger.warning(
|
||||||
|
f"{cls_name} is disabled because this process group" " spans across nodes."
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
|
# For not supported world size, we disable custom allreduce.
|
||||||
|
if world_size not in supported_world_size:
|
||||||
|
logger.warning(
|
||||||
|
f"{cls_name} is disabled due to an unsupported world"
|
||||||
|
f" size: {world_size}. Supported world sizes: {supported_world_size}. "
|
||||||
|
"To silence this warning, specify disable_custom_all_reduce=True explicitly.",
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
|
cuda_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES", None)
|
||||||
|
if cuda_visible_devices:
|
||||||
|
device_ids = list(map(int, cuda_visible_devices.split(",")))
|
||||||
|
else:
|
||||||
|
device_ids = list(range(torch.cuda.device_count()))
|
||||||
|
physical_device_id = device_ids[device.index]
|
||||||
|
tensor = torch.tensor([physical_device_id], dtype=torch.int, device="cpu")
|
||||||
|
gather_list = [
|
||||||
|
torch.tensor([0], dtype=torch.int, device="cpu") for _ in range(world_size)
|
||||||
|
]
|
||||||
|
dist.all_gather(gather_list, tensor, group=group)
|
||||||
|
physical_device_ids = [int(t) for t in gather_list]
|
||||||
|
full_nvlink = is_full_nvlink(physical_device_ids, world_size)
|
||||||
|
|
||||||
|
# test nvlink first, this will filter out most of the cases
|
||||||
|
# where custom allreduce is not supported
|
||||||
|
# this checks hardware and driver support for NVLink
|
||||||
|
if False:
|
||||||
|
logger.warning(
|
||||||
|
f"{cls_name} is disabled because it's not supported on"
|
||||||
|
" more than two PCIe-only GPUs. To silence this warning, "
|
||||||
|
"specify disable_custom_all_reduce=True explicitly."
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
|
# test P2P capability, this checks software/cudaruntime support
|
||||||
|
# this is expensive to compute at the first time
|
||||||
|
# then we cache the result
|
||||||
|
# On AMD GPU, p2p is always enabled between XGMI connected GPUs
|
||||||
|
if not _is_hip and not can_p2p(rank, world_size):
|
||||||
|
logger.warning(
|
||||||
|
f"{cls_name} is disabled because your platform lacks "
|
||||||
|
"GPU P2P capability or P2P test failed. To silence this "
|
||||||
|
"warning, specify disable_custom_all_reduce=True explicitly."
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
|
return full_nvlink
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
batch_src, batch_tgt, output_file = pickle.loads(sys.stdin.buffer.read())
|
||||||
|
result = can_actually_p2p(batch_src, batch_tgt)
|
||||||
|
with open(output_file, "wb") as f:
|
||||||
|
f.write(pickle.dumps(result))
|
||||||
@ -0,0 +1,27 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# prep_car_patch.sh — build patched custom-AR files at /root/patches/
|
||||||
|
# Patch A: should_custom_ar allows any CUDA AR <= max_size (8MB cap keeps
|
||||||
|
# big prefill ARs on NCCL); decode ARs (<=1MB) go custom AR.
|
||||||
|
# Patch B: utils gate no longer bails on ws>2 without NVLink (P2P probe
|
||||||
|
# verified all-True on this 8-way PCIe box).
|
||||||
|
# V2 stays excluded: can_use_custom_all_reduce_v2 requires full_nvlink is True.
|
||||||
|
set -e
|
||||||
|
D=/sgl-workspace/sglang/python/sglang/srt/distributed/device_communicators
|
||||||
|
mkdir -p /root/patches
|
||||||
|
docker cp glm53-nvfp4:$D/custom_all_reduce.py /root/patches/custom_all_reduce.py
|
||||||
|
docker cp glm53-nvfp4:$D/custom_all_reduce_utils.py /root/patches/custom_all_reduce_utils.py
|
||||||
|
|
||||||
|
ANCHOR='self.original_disabled = False # Ensure original_disabled == disabled'
|
||||||
|
n=$(grep -c "$ANCHOR" /root/patches/custom_all_reduce.py)
|
||||||
|
[ "$n" = "1" ] || { echo "ANCHOR_NOT_UNIQUE=$n"; exit 1; }
|
||||||
|
|
||||||
|
sed -i 's/if self.world_size == 2 or self.full_nvlink:/if True:/' /root/patches/custom_all_reduce.py
|
||||||
|
sed -i 's/if world_size > 2 and not full_nvlink:/if False:/' /root/patches/custom_all_reduce_utils.py
|
||||||
|
sed -i "/${ANCHOR}/a\\ logger.warning(f\"SSKJ_CAR_PATCH_ACTIVE ws={self.world_size} full_nvlink={self.full_nvlink}\")" /root/patches/custom_all_reduce.py
|
||||||
|
|
||||||
|
python3 -m py_compile /root/patches/custom_all_reduce.py /root/patches/custom_all_reduce_utils.py
|
||||||
|
echo "--- verify ---"
|
||||||
|
grep -n 'if True:' /root/patches/custom_all_reduce.py
|
||||||
|
grep -n 'if False:' /root/patches/custom_all_reduce_utils.py
|
||||||
|
grep -n 'SSKJ_CAR_PATCH_ACTIVE' /root/patches/custom_all_reduce.py
|
||||||
|
echo PREP_DONE
|
||||||
@ -0,0 +1,58 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# deploy_glm53_605.sh — GLM-5.3-NVFP4 TP8 + EAGLE for 174.1.60.5 (team service)
|
||||||
|
# Target scenario: cc1-2, 64k/128k input 90% cache hit, single request up to 256k,
|
||||||
|
# output-throughput priority. Clean deploy: rm old container, wait VRAM drain, run.
|
||||||
|
#
|
||||||
|
# Env overrides:
|
||||||
|
# MEMFRAC (0.90) STEPS (4) TOPK (1) DRAFT (5) CTXLEN (270336)
|
||||||
|
# CHUNK (8192) MAXPRE (16384) RESTART (no|yes) EXTRA ("")
|
||||||
|
# Usage:
|
||||||
|
# bash deploy_glm53_605.sh # production 4/1/5 @ 0.90
|
||||||
|
# STEPS=5 DRAFT=6 bash deploy_glm53_605.sh # tuning round
|
||||||
|
# CHUNK=16384 bash deploy_glm53_605.sh # prefill tuning round
|
||||||
|
# RESTART=yes bash deploy_glm53_605.sh # production finalize
|
||||||
|
set -e
|
||||||
|
MEMFRAC=${MEMFRAC:-0.90}
|
||||||
|
STEPS=${STEPS:-4}
|
||||||
|
TOPK=${TOPK:-1}
|
||||||
|
DRAFT=${DRAFT:-5}
|
||||||
|
CTXLEN=${CTXLEN:-270336}
|
||||||
|
CHUNK=${CHUNK:-8192}
|
||||||
|
MAXPRE=${MAXPRE:-16384}
|
||||||
|
EXTRA=${EXTRA:-}
|
||||||
|
if [ "$RESTART" = "yes" ]; then RP="--restart unless-stopped"; else RP="--restart no"; fi
|
||||||
|
|
||||||
|
echo "[deploy] removing old container (if any)"
|
||||||
|
# rm -f times out on big GPU containers on this daemon; retry until really gone
|
||||||
|
for i in $(seq 1 45); do
|
||||||
|
CID=$(docker ps -a --filter name=glm53-nvfp4 -q)
|
||||||
|
[ -z "$CID" ] && break
|
||||||
|
docker rm -f glm53-nvfp4 >/dev/null 2>&1 || true
|
||||||
|
sleep 2
|
||||||
|
done
|
||||||
|
|
||||||
|
echo "[deploy] waiting for VRAM drain"
|
||||||
|
for i in $(seq 1 45); do
|
||||||
|
used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
|
||||||
|
[ "$used" -lt 2000 ] && break
|
||||||
|
sleep 2
|
||||||
|
done
|
||||||
|
used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
|
||||||
|
echo "[deploy] VRAM now: ${used} MiB total"
|
||||||
|
|
||||||
|
echo "[deploy] starting: TP8 EAGLE ${STEPS}/${TOPK}/${DRAFT} memfrac=${MEMFRAC} extra='${EXTRA}'"
|
||||||
|
docker run -d --name glm53-nvfp4 --gpus all --shm-size 64g --ipc=host $RP \
|
||||||
|
-p 30000:30000 -v /data/hf_models:/data/hf_models \
|
||||||
|
lmsysorg/sglang:nightly-dev-20260828-daf63171 \
|
||||||
|
python3 -m sglang.launch_server \
|
||||||
|
--model-path /data/hf_models/GLM-5.3-NVFP4 --tp 8 \
|
||||||
|
--mem-fraction-static $MEMFRAC --max-running-requests 16 \
|
||||||
|
--chunked-prefill-size $CHUNK --max-prefill-tokens $MAXPRE \
|
||||||
|
--disable-shared-experts-fusion --moe-runner-backend flashinfer_cutlass --disable-flashinfer-autotune \
|
||||||
|
--speculative-algorithm EAGLE --speculative-num-steps $STEPS --speculative-eagle-topk $TOPK --speculative-num-draft-tokens $DRAFT \
|
||||||
|
--kv-cache-dtype fp8_e4m3 --enable-hierarchical-cache --hicache-ratio 3 \
|
||||||
|
--cuda-graph-max-bs-decode 8 --cuda-graph-bs-decode 1 2 3 4 6 8 --cuda-graph-max-bs-prefill 8 \
|
||||||
|
--context-length $CTXLEN --reasoning-parser glm45 --tool-call-parser glm47 \
|
||||||
|
--host 0.0.0.0 --port 30000 $EXTRA
|
||||||
|
|
||||||
|
echo "[deploy] container started; poll: docker logs -f glm53-nvfp4"
|
||||||
@ -0,0 +1,70 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# deploy_glm53_607_exp.sh — GLM-5.3-NVFP4 TP8 + EAGLE experiment deploy for 174.1.60.7
|
||||||
|
# EXACT copy of deploy_glm53_605.sh (production standard, 8-machine md5 fcd9109b)
|
||||||
|
# with only this header changed. Campaign: scenario-1 optimization (i128k/i64k,
|
||||||
|
# o512, cc1-4). Do NOT sync this file to other machines.
|
||||||
|
#
|
||||||
|
# Env overrides:
|
||||||
|
# MEMFRAC (0.90) STEPS (4) TOPK (1) DRAFT (5) CTXLEN (270336)
|
||||||
|
# CHUNK (8192) MAXPRE (16384) RESTART (no|yes) EXTRA ("")
|
||||||
|
# CAR_PATCH (1) inject custom-AR 1stage patch (E7b winner, 2026-09-07):
|
||||||
|
# small ARs <=8MB go one-shot kernel instead of NCCL RING_LL on 8-way PCIe.
|
||||||
|
# Patch files at /root/patches/ (build: bash /root/prep_car_patch.sh on a
|
||||||
|
# running stock container). Patch dies with the container; image untouched.
|
||||||
|
# Set CAR_PATCH=0 to deploy stock.
|
||||||
|
set -e
|
||||||
|
MEMFRAC=${MEMFRAC:-0.90}
|
||||||
|
STEPS=${STEPS:-4}
|
||||||
|
TOPK=${TOPK:-1}
|
||||||
|
DRAFT=${DRAFT:-5}
|
||||||
|
CTXLEN=${CTXLEN:-270336}
|
||||||
|
CHUNK=${CHUNK:-8192}
|
||||||
|
MAXPRE=${MAXPRE:-16384}
|
||||||
|
EXTRA=${EXTRA:-}
|
||||||
|
if [ "$RESTART" = "yes" ]; then RP="--restart unless-stopped"; else RP="--restart no"; fi
|
||||||
|
|
||||||
|
echo "[deploy] removing old container (if any)"
|
||||||
|
# rm -f times out on big GPU containers on this daemon; retry until really gone
|
||||||
|
for i in $(seq 1 45); do
|
||||||
|
CID=$(docker ps -a --filter name=glm53-nvfp4 -q)
|
||||||
|
[ -z "$CID" ] && break
|
||||||
|
docker rm -f glm53-nvfp4 >/dev/null 2>&1 || true
|
||||||
|
sleep 2
|
||||||
|
done
|
||||||
|
|
||||||
|
echo "[deploy] waiting for VRAM drain"
|
||||||
|
for i in $(seq 1 45); do
|
||||||
|
used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
|
||||||
|
[ "$used" -lt 2000 ] && break
|
||||||
|
sleep 2
|
||||||
|
done
|
||||||
|
used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
|
||||||
|
echo "[deploy] VRAM now: ${used} MiB total"
|
||||||
|
|
||||||
|
echo "[deploy] starting: TP8 EAGLE ${STEPS}/${TOPK}/${DRAFT} memfrac=${MEMFRAC} chunk=${CHUNK} extra='${EXTRA}'"
|
||||||
|
docker run -d --name glm53-nvfp4 --gpus all --shm-size 64g --ipc=host $RP \
|
||||||
|
-p 30000:30000 -v /data/hf_models:/data/hf_models \
|
||||||
|
lmsysorg/sglang:nightly-dev-20260828-daf63171 \
|
||||||
|
python3 -m sglang.launch_server \
|
||||||
|
--model-path /data/hf_models/GLM-5.3-NVFP4 --tp 8 \
|
||||||
|
--mem-fraction-static $MEMFRAC --max-running-requests 16 \
|
||||||
|
--chunked-prefill-size $CHUNK --max-prefill-tokens $MAXPRE \
|
||||||
|
--disable-shared-experts-fusion --moe-runner-backend flashinfer_cutlass --disable-flashinfer-autotune \
|
||||||
|
--speculative-algorithm EAGLE --speculative-num-steps $STEPS --speculative-eagle-topk $TOPK --speculative-num-draft-tokens $DRAFT \
|
||||||
|
--kv-cache-dtype fp8_e4m3 --enable-hierarchical-cache --hicache-ratio 3 \
|
||||||
|
--cuda-graph-max-bs-decode 8 --cuda-graph-bs-decode 1 2 3 4 6 8 --cuda-graph-max-bs-prefill 8 \
|
||||||
|
--context-length $CTXLEN --reasoning-parser glm45 --tool-call-parser glm47 \
|
||||||
|
--host 0.0.0.0 --port 30000 $EXTRA
|
||||||
|
|
||||||
|
echo "[deploy] container started; poll: docker logs -f glm53-nvfp4"
|
||||||
|
|
||||||
|
# --- CAR 1stage patch injection (E7b winner) ---
|
||||||
|
if [ "${CAR_PATCH:-1}" != "0" ] && [ -f /root/patches/custom_all_reduce.py ]; then
|
||||||
|
echo "[deploy] CAR_PATCH: injecting custom-AR 1stage patch"
|
||||||
|
docker stop -t 20 glm53-nvfp4 >/dev/null 2>&1 || true
|
||||||
|
DPATH=/sgl-workspace/sglang/python/sglang/srt/distributed/device_communicators
|
||||||
|
docker cp /root/patches/custom_all_reduce.py glm53-nvfp4:$DPATH/custom_all_reduce.py
|
||||||
|
docker cp /root/patches/custom_all_reduce_utils.py glm53-nvfp4:$DPATH/custom_all_reduce_utils.py
|
||||||
|
docker start glm53-nvfp4
|
||||||
|
echo "[deploy] CAR_PATCH injected, container restarted; poll health as usual"
|
||||||
|
fi
|
||||||
@ -0,0 +1,62 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# ============================================================
|
||||||
|
# GLM-5.3 最优部署方案(6000D 8卡,TP4 PP2 + IndexCache freq=4)
|
||||||
|
# 2026-09-07
|
||||||
|
#
|
||||||
|
# - 基线配置:TP4 PP2 + cps16k + mem0.85(131.6 tok/s 吞吐基线)
|
||||||
|
# - IndexCache(index_topk_freq=4):层轴索引复用,省 75% indexer
|
||||||
|
# 16K 场景无损失;128K 长上下文并发 1.35-1.47× 提速
|
||||||
|
# - 禁 radix cache;禁投机解码(PP2 与投机框架不兼容,已实测)
|
||||||
|
#
|
||||||
|
# 用法:bash deploy_glm53_optimal.sh
|
||||||
|
# (60.7 试用版:增加显存排空等待 + 就绪等待加长至 20 分钟)
|
||||||
|
# ============================================================
|
||||||
|
set -uo pipefail
|
||||||
|
|
||||||
|
CONTAINER="glm53-nvfp4"
|
||||||
|
IMAGE="lmsysorg/sglang:nightly-dev-20260828-daf63171"
|
||||||
|
MODEL="/data/hf_models/GLM-5.3-NVFP4"
|
||||||
|
PORT=30000
|
||||||
|
TP=4; PP=2; MEM=0.85; MRR=48; CPS=16384
|
||||||
|
|
||||||
|
docker rm -f ${CONTAINER} 2>/dev/null || true
|
||||||
|
|
||||||
|
# docker rm -f 后显存释放滞后数分钟,不等会把新容器 KV 池压小
|
||||||
|
for i in $(seq 1 60); do
|
||||||
|
used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
|
||||||
|
if [ "$used" -lt 1500 ]; then echo "[deploy] drained: ${used} MiB"; break; fi
|
||||||
|
echo "[deploy] drain wait ${i}: ${used} MiB"
|
||||||
|
sleep 10
|
||||||
|
done
|
||||||
|
|
||||||
|
docker run -d --name ${CONTAINER} --gpus all --shm-size 64g --ipc=host \
|
||||||
|
--restart unless-stopped \
|
||||||
|
-p ${PORT}:${PORT} \
|
||||||
|
-v /data/hf_models:/data/hf_models \
|
||||||
|
${IMAGE} \
|
||||||
|
python3 -m sglang.launch_server \
|
||||||
|
--model-path ${MODEL} \
|
||||||
|
--tp-size ${TP} --pp-size ${PP} \
|
||||||
|
--mem-fraction-static ${MEM} \
|
||||||
|
--max-running-requests ${MRR} \
|
||||||
|
--disable-radix-cache \
|
||||||
|
--disable-shared-experts-fusion \
|
||||||
|
--moe-runner-backend flashinfer_cutlass \
|
||||||
|
--disable-flashinfer-autotune \
|
||||||
|
--disable-custom-all-reduce \
|
||||||
|
--chunked-prefill-size ${CPS} \
|
||||||
|
--host 0.0.0.0 --port ${PORT} \
|
||||||
|
--json-model-override-args '{"index_topk_freq": 4}'
|
||||||
|
|
||||||
|
echo "容器已启动,等待就绪..."
|
||||||
|
for i in $(seq 1 120); do
|
||||||
|
code=$(curl -s -o /dev/null -w '%{http_code}' http://localhost:${PORT}/health 2>/dev/null)
|
||||||
|
if [ "$code" = "200" ]; then
|
||||||
|
echo "READY after ~$((i*10))s"
|
||||||
|
docker ps --filter name=${CONTAINER} --format '{{.Names}} {{.Status}}'
|
||||||
|
echo "override args: $(docker inspect ${CONTAINER} --format '{{.Config.Cmd}}' | grep -o 'index_topk_freq[^,}]*')"
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
sleep 10
|
||||||
|
done
|
||||||
|
echo "TIMEOUT"; exit 1
|
||||||
@ -0,0 +1,59 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# ============================================================
|
||||||
|
# TP4 PP2 + IndexCache(场景一轮:启用 radix cache)
|
||||||
|
# 与 deploy_glm53_optimal.sh 唯一差异:去掉 --disable-radix-cache
|
||||||
|
# ——场景一前提是 90% 前缀命中,禁 radix 则命中率为 0 无从对比
|
||||||
|
# 其余参数原样:TP4 PP2 / mem0.85 / mrr48 / cps16384 / index_topk_freq=4 / 无投机
|
||||||
|
# ============================================================
|
||||||
|
set -uo pipefail
|
||||||
|
|
||||||
|
CONTAINER="glm53-nvfp4"
|
||||||
|
IMAGE="lmsysorg/sglang:nightly-dev-20260828-daf63171"
|
||||||
|
MODEL="/data/hf_models/GLM-5.3-NVFP4"
|
||||||
|
PORT=30000
|
||||||
|
TP=4; PP=2; MEM=0.85; MRR=48; CPS=16384
|
||||||
|
|
||||||
|
docker rm -f ${CONTAINER} 2>/dev/null || true
|
||||||
|
# 补刀循环:docker rm -f 异步滞留,等容器对象彻底消失
|
||||||
|
for i in $(seq 1 40); do
|
||||||
|
n=$(docker ps -a --filter name=${CONTAINER} -q | wc -l)
|
||||||
|
if [ "$n" = "0" ]; then echo "[deploy] container object gone"; break; fi
|
||||||
|
sleep 5
|
||||||
|
done
|
||||||
|
# 排空显存
|
||||||
|
for i in $(seq 1 60); do
|
||||||
|
used=$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits | awk '{s+=$1} END {print s}')
|
||||||
|
if [ "$used" -lt 1500 ]; then echo "[deploy] drained: ${used} MiB"; break; fi
|
||||||
|
echo "[deploy] drain wait ${i}: ${used} MiB"
|
||||||
|
sleep 10
|
||||||
|
done
|
||||||
|
|
||||||
|
docker run -d --name ${CONTAINER} --gpus all --shm-size 64g --ipc=host \
|
||||||
|
--restart unless-stopped \
|
||||||
|
-p ${PORT}:${PORT} \
|
||||||
|
-v /data/hf_models:/data/hf_models \
|
||||||
|
${IMAGE} \
|
||||||
|
python3 -m sglang.launch_server \
|
||||||
|
--model-path ${MODEL} \
|
||||||
|
--tp-size ${TP} --pp-size ${PP} \
|
||||||
|
--mem-fraction-static ${MEM} \
|
||||||
|
--max-running-requests ${MRR} \
|
||||||
|
--disable-shared-experts-fusion \
|
||||||
|
--moe-runner-backend flashinfer_cutlass \
|
||||||
|
--disable-flashinfer-autotune \
|
||||||
|
--disable-custom-all-reduce \
|
||||||
|
--chunked-prefill-size ${CPS} \
|
||||||
|
--host 0.0.0.0 --port ${PORT} \
|
||||||
|
--json-model-override-args '{"index_topk_freq": 4}'
|
||||||
|
|
||||||
|
echo "容器已启动,等待就绪..."
|
||||||
|
for i in $(seq 1 120); do
|
||||||
|
code=$(curl -s -o /dev/null -w '%{http_code}' http://localhost:${PORT}/health 2>/dev/null)
|
||||||
|
if [ "$code" = "200" ]; then
|
||||||
|
echo "READY after ~$((i*10))s"
|
||||||
|
docker ps --filter name=${CONTAINER} --format '{{.Names}} {{.Status}}'
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
sleep 10
|
||||||
|
done
|
||||||
|
echo "TIMEOUT"; exit 1
|
||||||
Loading…
x
Reference in New Issue
Block a user