minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

config: a parallel leaf with no live counterpart is read bare (#36620)

Browse Source 
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Cheng Wan
2026-08-27 12:56:11 -07:00
committed by GitHub
co-authored by Claude Opus 5
parent 7c3b5a6732
commit ca1d7ed8e6
125 changed files with 343 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/sglang/srt/model_executor/cpu_graph_runner.py
+1 -1
View File
@@ -610,7 +610,7 @@ class CPUGraphRunner:
model_runner.server_args.enable_profile_cuda_graph
)
self.tp_size = get_parallel().config.tp_size
self.dp_size = get_parallel().config.dp_size
self.dp_size = get_parallel().dp_size
self.pp_size = get_parallel().config.pp_size
self.capture_forward_mode = ForwardMode.DECODE
python/sglang/srt/model_executor/model_runner.py
+11 -13
View File
@@ -486,19 +486,17 @@ class ModelRunner:
if not (get_exec().moe.elastic_ep_backend is not None and is_ep_scale_joiner()):
return
join_effective_ep_size = (
get_parallel().config.ep_join_rank_offset + self.ps.tp_size
)
join_effective_ep_size = get_parallel().ep_join_rank_offset + self.ps.tp_size
dist.barrier(group=self.tp_group.cpu_group)
if self.ps.tp_rank == 0:
register_scale_cohort(
get_parallel().config.ep_join_rank_offset,
get_parallel().ep_join_rank_offset,
join_effective_ep_size,
)
join_scale_process_group()
get_context().override("elastic_ep.scale_join", ep_size=join_effective_ep_size)
global_ep_rank = self.ps.tp_rank + get_parallel().config.ep_join_rank_offset
global_ep_rank = self.ps.tp_rank + get_parallel().ep_join_rank_offset
broadcast_global_expert_location_metadata(
model_config=self.model_config,
moe_ep_rank=global_ep_rank,
@@ -697,7 +695,7 @@ class ModelRunner:
if self.is_draft_worker:
return
expert_rank = self.ps.moe_ep_rank + (
get_parallel().config.ep_join_rank_offset if is_ep_scale_joiner() else 0
get_parallel().ep_join_rank_offset if is_ep_scale_joiner() else 0
)
set_global_expert_location_metadata(
compute_initial_expert_location_metadata(
@@ -928,7 +926,7 @@ class ModelRunner:
device=self.device,
tp_group=(
self.attention_tp_group.cpu_group
if get_parallel().config.enable_dp_attention
if get_parallel().enable_dp_attention
else self.tp_group.cpu_group
),
host_to_device_ratio=hisparse_cfg.host_to_device_ratio,
@@ -964,7 +962,7 @@ class ModelRunner:
def post_capture_elastic_ep_recover(self):
join_process_groups()
global_ep_rank = self.ps.tp_rank + get_parallel().config.ep_join_rank_offset
global_ep_rank = self.ps.tp_rank + get_parallel().ep_join_rank_offset
broadcast_global_expert_location_metadata(
model_config=self.model_config,
moe_ep_rank=global_ep_rank,
@@ -1004,7 +1002,7 @@ class ModelRunner:
self.decode_attn_backend = backends.decode_attn_backend
self.decode_attn_backend_group = backends.decode_attn_backend_group
if get_parallel().dcp_enabled and get_parallel().config.dcp_replicate_q_proj:
if get_parallel().dcp_enabled and get_parallel().dcp_replicate_q_proj:
self._prepare_replicated_q_proj()
def _prepare_replicated_q_proj(self) -> None:
@@ -1282,7 +1280,7 @@ class ModelRunner:
def maybe_init_dwdp(self):
if self.is_draft_worker:
return
if get_parallel().config.dwdp_size <= 1:
if get_parallel().dwdp_size <= 1:
return
from sglang.srt.layers.moe.dwdp import DwdpManager
@@ -1455,7 +1453,7 @@ class ModelRunner:
# rather than spawning additional processes, so dp_size must not be
# multiplied into the process count here (unlike regular DP, where
# dp_size * tp_size * pp_size is the true worker count).
dp_size = 1 if get_parallel().config.enable_dp_attention else self.ps.dp_size
dp_size = 1 if get_parallel().enable_dp_attention else self.ps.dp_size
self.local_omp_cpuid = numa_utils.init_threads_binding(
numa_index=self.gpu_id,
world_size=dp_size * self.ps.tp_size * self.ps.pp_size,
@@ -1938,7 +1936,7 @@ class ModelRunner:
if added <= 0:
return
initial_ep_size = get_parallel().config.elastic_ep_initial_size
initial_ep_size = get_parallel().elastic_ep_initial_size
assert initial_ep_size is not None
get_context().override("elastic_ep.scale", ep_size=effective_size)
@@ -1956,7 +1954,7 @@ class ModelRunner:
set_global_expert_location_metadata(new_metadata, allow_overwrite=True)
def _elastic_global_rank(self) -> int:
return self.ps.tp_rank + get_parallel().config.ep_join_rank_offset
return self.ps.tp_rank + get_parallel().ep_join_rank_offset
def _rearm_eplb_after_elastic_scale(self) -> None:
if self.eplb_manager is None:
python/sglang/srt/model_executor/model_runner_components/remote_instance_weight_transporter.py
+2 -4
View File
@@ -78,13 +78,11 @@ class RemoteInstanceWeightTransporter:
"""
import requests as http_requests
if get_parallel().config.dist_init_addr:
if get_parallel().dist_init_addr:
# Multi-node: bootstrap server is on the head node (node_rank==0).
# Derive host from dist_init_addr (shared across all nodes).
bootstrap_host = (
NetworkAddress.parse(get_parallel().config.dist_init_addr)
.resolved()
.host
NetworkAddress.parse(get_parallel().dist_init_addr).resolved().host
)
else:
bootstrap_host = "127.0.0.1"
python/sglang/srt/model_executor/model_runner_components/startup_weight_load.py
+2 -2
View File
@@ -123,8 +123,8 @@ class StartupWeightLoadOptions:
attn_cp_size=get_parallel().config.attn_cp_size,
dcp_size=get_parallel().config.dcp_size,
pp_size=get_parallel().config.pp_size,
dp_size=get_parallel().config.dp_size,
ep_size=get_parallel().config.ep_size,
dp_size=get_parallel().dp_size,
ep_size=get_parallel().ep_size,
cpu_offload_gb=get_exec().offload.cpu_offload_gb,
offload_group_size=get_exec().offload.offload_group_size,
enable_memory_saver=get_exec().features.enable_memory_saver,
python/sglang/srt/model_executor/runner/base_runner.py
+4 -4
View File
@@ -218,7 +218,7 @@ class BaseRunner(ABC):
self.device_module = torch.get_device_module(self.device)
self.tp_size = get_parallel().config.tp_size
# elastic-EP scale-up rewrites dp_size on the published config
self.dp_size = get_parallel().config.dp_size
self.dp_size = get_parallel().dp_size
self.pp_size = get_parallel().config.pp_size
self.enable_pdmux = model_runner.server_args.enable_pdmux
self.return_hidden_states_mode = (
@@ -289,7 +289,7 @@ class BaseRunner(ABC):
"""
if (
not get_parallel().dcp_enabled
or get_parallel().config.dcp_comm_backend != "fi_a2a"
or get_parallel().dcp_comm_backend != "fi_a2a"
):
return
@@ -348,7 +348,7 @@ class BaseRunner(ABC):
hidden_size=mr.model_config.hidden_size,
vocab_size=mr.model_config.vocab_size,
dtype=mr.model_config.dtype,
dp_size=get_parallel().config.dp_size,
dp_size=get_parallel().dp_size,
pp_size=get_parallel().config.pp_size,
is_encoder_decoder=mr.model_config.is_encoder_decoder,
require_mlp_tp_gather=require_mlp_tp_gather(),
@@ -541,7 +541,7 @@ class BaseRunner(ABC):
assert require_mlp_tp_gather_ or require_attn_tp_gather_
if require_mlp_tp_gather_:
global_num_tokens_cpu = [num_tokens] * get_parallel().config.dp_size
global_num_tokens_cpu = [num_tokens] * get_parallel().dp_size
elif require_attn_tp_gather_:
global_num_tokens_cpu = [num_tokens]
else:
python/sglang/srt/model_executor/runner/decode_cuda_graph_runner.py
+1 -1
View File
@@ -242,7 +242,7 @@ class DecodeCudaGraphRunner(BaseCudaGraphRunner):
self.require_mlp_tp_gather or self.require_attn_tp_gather
)
self.require_mlp_sync = (
get_parallel().config.enable_dp_attention or self.require_gathered_buffer
get_parallel().enable_dp_attention or self.require_gathered_buffer
)
self.enable_two_batch_overlap = (
model_runner.server_args.enable_two_batch_overlap
python/sglang/srt/model_executor/runner/eager_runner.py
+1 -1
View File
@@ -144,7 +144,7 @@ class EagerRunner(BaseRunner):
encoder_lens_dtype=(
torch.int64 if torch.device(mr.device).type == "cpu" else torch.int32
),
dp_size=get_parallel().config.dp_size,
dp_size=get_parallel().dp_size,
)
# Eager has no capture step, so warm up here (run-once via mr._kernel_warmed_up).
self.warmup()
python/sglang/srt/model_executor/runner/prefill_cuda_graph_runner.py
+1 -1
View File
@@ -352,7 +352,7 @@ class PrefillCudaGraphRunner(BaseCudaGraphRunner):
self.moe_fusions = self.model_runner.moe_fusions
self.dsa_indexers = getattr(self.model_runner, "dsa_indexers", None)
self.dp_size = get_parallel().config.dp_size
self.dp_size = get_parallel().dp_size
self.require_mlp_tp_gather = require_mlp_tp_gather()
self.require_attn_tp_gather = require_attn_tp_gather()