Check the topology identities where the layout is written, and build at the published widths

benchmark/hf3fs/bench_zerocopy.py

@@ -5,10 +5,10 @@
 from tqdm import tqdm
 
 from sglang.srt.distributed import (
-    get_world_group,
     init_distributed_environment,
     initialize_model_parallel,
 )
+from sglang.srt.distributed.parallel_state import get_world_group
 from sglang.srt.managers.cache_controller import (
     HiCacheController,
     PrefetchOperation,
@@ -17,6 +17,7 @@
 from sglang.srt.mem_cache.allocator import TokenToKVPoolAllocator
 from sglang.srt.mem_cache.memory_pool import MHATokenToKVPool
 from sglang.srt.mem_cache.pool_host.mha import MHATokenToKVPoolHost
+from sglang.test.test_utils import publish_build_topology
 
 init_distributed_environment(
     world_size=1,
@@ -26,10 +27,8 @@
     backend="gloo",
 )
 
-initialize_model_parallel(
-    tensor_model_parallel_size=1,
-    pipeline_model_parallel_size=1,
-)
+publish_build_topology()
+initialize_model_parallel()
 
 group = get_world_group().cpu_group
 

benchmark/kernels/all_reduce/benchmark_all_reduce.py

@@ -21,6 +21,7 @@
     init_distributed_environment,
     initialize_model_parallel,
 )
+from sglang.test.test_utils import publish_build_topology
 
 
 def parse_args():
@@ -85,7 +86,8 @@ def init_dist(backend: str):
         distributed_init_method=distributed_init_method,
         local_rank=rank,
     )
-    initialize_model_parallel(tensor_model_parallel_size=world_size)
+    publish_build_topology(world_rank=rank, tp_size=world_size)
+    initialize_model_parallel()
     return dist.group.WORLD
 
 

benchmark/kernels/all_reduce/benchmark_fused_ar_rms_amd.py

@@ -41,6 +41,7 @@
     initialize_model_parallel,
     set_custom_all_reduce,
 )
+from sglang.test.test_utils import publish_build_topology
 
 Shape = Tuple[int, int]
 
@@ -381,7 +382,8 @@ def main():
         distributed_init_method="env://",
         backend="nccl",
     )
-    initialize_model_parallel(tensor_model_parallel_size=world_size)
+    publish_build_topology(world_rank=rank, tp_size=world_size)
+    initialize_model_parallel()
 
     prefill_shapes = parse_shapes(args.prefill_shapes)
     decode_shapes = parse_shapes(args.decode_shapes)

benchmark/kernels/all_reduce/benchmark_fused_ar_rms_quant_amd.py

@@ -47,6 +47,7 @@
     initialize_model_parallel,
     set_custom_all_reduce,
 )
+from sglang.test.test_utils import publish_build_topology
 
 Shape = Tuple[int, int]
 FP8_DTYPE = torch.float8_e4m3fnuz
@@ -400,7 +401,8 @@ def main() -> None:
         distributed_init_method="env://",
         backend="nccl",
     )
-    initialize_model_parallel(tensor_model_parallel_size=world_size)
+    publish_build_topology(world_rank=rank, tp_size=world_size)
+    initialize_model_parallel()
 
     if rank == 0:
         print(

benchmark/kernels/all_reduce/benchmark_mscclpp.py

@@ -30,6 +30,7 @@
     initialize_model_parallel,
     set_mscclpp_all_reduce,
 )
+from sglang.test.test_utils import publish_build_topology
 
 
 def torch_allreduce(torch_input: torch.Tensor, group: ProcessGroup) -> torch.Tensor:
@@ -173,7 +174,8 @@ def print_markdown_table(data):
         rank=rank,
         local_rank=rank % 8,
     )
-    initialize_model_parallel(tensor_model_parallel_size=world_size)
+    publish_build_topology(world_rank=rank, tp_size=world_size)
+    initialize_model_parallel()
     group = get_tensor_model_parallel_group().device_group
     cpu_group = get_tensor_model_parallel_group().cpu_group
     pynccl_comm = get_tensor_model_parallel_group().pynccl_comm

benchmark/kernels/all_reduce/benchmark_torch_symm_mem.py

@@ -44,6 +44,7 @@
     initialize_model_parallel,
     set_torch_symm_mem_all_reduce,
 )
+from sglang.test.test_utils import publish_build_topology
 from sglang.utils import is_in_ci
 
 IS_CI = is_in_ci()
@@ -188,7 +189,8 @@ def print_markdown_table(data):
         rank=rank,
         local_rank=rank % 8,
     )
-    initialize_model_parallel(tensor_model_parallel_size=world_size)
+    publish_build_topology(world_rank=rank, tp_size=world_size)
+    initialize_model_parallel()
     group = get_tensor_model_parallel_group().device_group
     cpu_group = get_tensor_model_parallel_group().cpu_group
     pynccl_comm = get_tensor_model_parallel_group().pynccl_comm

benchmark/kernels/flashinfer_allreduce_fusion/benchmark_fused_collective.py

@@ -30,14 +30,16 @@
 
 from sglang.kernels.ops.quantization.fp8_kernel import fp8_dtype as SGLANG_FP8_DTYPE
 from sglang.kernels.ops.quantization.fp8_kernel import static_quant_fp8
-from sglang.srt.distributed import get_tp_group, tensor_model_parallel_all_reduce
+from sglang.srt.distributed import tensor_model_parallel_all_reduce
 from sglang.srt.distributed.parallel_state import (
     cleanup_dist_env_and_memory,
+    get_tp_group,
     graph_capture,
     init_distributed_environment,
     initialize_model_parallel,
 )
 from sglang.srt.layers.layernorm import RMSNorm  # noqa
+from sglang.test.test_utils import publish_build_topology
 
 try:
     from sgl_kernel import fused_add_rmsnorm as SGL_FUSED_ADD_RMS_NORM
@@ -1178,7 +1180,8 @@ def main():
         local_rank=rank,
         backend="nccl",
     )
-    initialize_model_parallel(tensor_model_parallel_size=world_size)
+    publish_build_topology(world_rank=rank, tp_size=world_size)
+    initialize_model_parallel()
 
     # Validate world size (must be > 1 for collective operations)
     if world_size <= 1:

benchmark/kernels/fused_moe_triton/benchmark_sglang_fused_moe_triton.py

@@ -26,6 +26,7 @@
     select_experts,
 )
 from sglang.srt.server_args import ServerArgs, set_global_server_args_for_scheduler
+from sglang.test.test_utils import publish_build_topology
 
 
 def fused_moe_triton_api(
@@ -227,10 +228,8 @@ def main():
             backend="nccl" if torch.cuda.is_available() else "gloo",
         )
 
-        initialize_model_parallel(
-            tensor_model_parallel_size=1,
-            expert_model_parallel_size=1,
-        )
+        publish_build_topology()
+        initialize_model_parallel()
 
         model_config = get_model_config(args.model, args.tp_size, args.ep_size)
         benchmark.run(

benchmark/kernels/fused_moe_triton/benchmark_vllm_vs_sglang_fused_moe_triton.py

@@ -15,6 +15,7 @@
 from sglang.srt.layers.moe.moe_runner.triton_utils.fused_moe import (
     fused_moe as fused_moe_sglang,
 )
+from sglang.test.test_utils import publish_build_topology
 
 from .common_utils import get_model_config
 
@@ -243,10 +244,8 @@ def main():
             backend="nccl" if torch.cuda.is_available() else "gloo",
         )
 
-        initialize_model_parallel(
-            tensor_model_parallel_size=1,
-            pipeline_model_parallel_size=1,
-        )
+        publish_build_topology()
+        initialize_model_parallel()
 
         shape_configs = get_model_config(args.model, args.tp_size, args.ep_size)
         benchmark.run(

examples/usage/modelopt_quantize_and_export.py

@@ -21,6 +21,7 @@
     initialize_model_parallel,
 )
 from sglang.srt.model_loader.loader import get_model_loader
+from sglang.test.test_utils import publish_build_topology
 
 
 def _validate_export(export_dir: str) -> bool:
@@ -113,10 +114,8 @@ def quantize_and_export_model(
             local_rank=0,
             backend="nccl" if device == "cuda" else "gloo",
         )
-        initialize_model_parallel(
-            tensor_model_parallel_size=1,
-            pipeline_model_parallel_size=1,
-        )
+        publish_build_topology()
+        initialize_model_parallel()
 
     # Configure model loading with ModelOpt quantization and export
     model_config = ModelConfig(

python/sglang/srt/arg_groups/arg_utils.py

@@ -110,7 +110,7 @@ class Derived(msgspec.Struct, frozen=True):
     Every declaration carries ``fn`` today, the parallel quotients included:
     they are a function of the configured leaves, so they are computed at
     publish like the rest. What is special about them is not how they are
-    computed but that a stamp can move one afterwards -- an elastic scale-up
+    computed but that a stamp can move one afterwards -- ``initialize_dp_attention``
     restamps ``attn_dp_size`` -- which ``ParallelContext`` answers above the
     published leaf.
     """

python/sglang/srt/disaggregation/encoder/server.py

@@ -589,10 +589,32 @@ def __init__(
             distributed_init_method=dist_init_method,
             local_rank=rank,
         )
-        initialize_model_parallel(
-            tensor_model_parallel_size=get_parallel().tp_size,
-            attention_context_model_parallel_size=get_parallel().attn_cp_size,
+        # The encoder serves the vision tower on a world of its own: `tp_size`
+        # ranks wide, with no pipeline, no expert or MoE-DP dimension and no
+        # decode context parallelism, whatever the generation side published.
+        # That has always been the layout it builds; stating it is what stops
+        # the context from answering with the other side's topology while these
+        # groups answer with this one.
+        parallel = get_parallel()
+        attn_cp_size = parallel.attn_cp_size
+        attn_tp_size = parallel.tp_size // attn_cp_size
+        attn_cp_rank, attn_tp_rank = divmod(rank, attn_tp_size)
+        parallel.override_permanently(
+            tp_rank=rank,
+            pp_size=1,
+            pp_rank=0,
+            attn_dp_size=1,
+            attn_dp_rank=0,
+            attn_tp_size=attn_tp_size,
+            attn_tp_rank=attn_tp_rank,
+            attn_cp_rank=attn_cp_rank,
+            attn_dcp_size=1,
+            moe_ep_size=1,
+            moe_ep_rank=0,
+            moe_dp_size=1,
+            moe_tp_size=parallel.tp_size,
         )
+        initialize_model_parallel()
         initialize_dp_attention(server_args, self.model_config)
 
         self.model = load_model(

python/sglang/srt/distributed/bootstrap.py

@@ -106,15 +106,6 @@ def init_torch_distributed(
             server_args=server_args,
             model_config=model_config,
             gpu_id=ps.gpu_id,
-            tp_rank=ps.tp_rank,
-            tp_size=ps.tp_size,
-            pp_rank=ps.pp_rank,
-            pp_size=ps.pp_size,
-            attn_dp_size=ps.attn_dp_size,
-            attn_cp_size=ps.attn_cp_size,
-            moe_ep_size=ps.moe_ep_size,
-            moe_dp_size=ps.moe_dp_size,
-            dcp_size=ps.attn_dcp_size,
         )
 
         # Pre-warm NCCL/RCCL/HCCL to eliminate cold-start latency in first request
@@ -255,19 +246,13 @@ def _init_parallel_groups(
     server_args: ServerArgs,
     model_config: ModelConfig,
     gpu_id: int,
-    tp_rank: int,
-    tp_size: int,
-    pp_rank: int,
-    pp_size: int,
-    attn_dp_size: int,
-    attn_cp_size: int,
-    moe_ep_size: int,
-    moe_dp_size: int,
-    dcp_size: int,
 ) -> None:
+    parallel = get_parallel()
+    tp_size, pp_size = parallel.tp_size, parallel.pp_size
+    tp_rank, pp_rank = parallel.tp_rank, parallel.pp_rank
     is_ep_joiner = get_exec().moe.is_ep_joiner
     is_scale_joiner = get_exec().moe.is_ep_scale_joiner
-    rank_offset = get_parallel().ep_join_rank_offset if is_scale_joiner else 0
+    rank_offset = parallel.ep_join_rank_offset if is_scale_joiner else 0
     world_size = (
         rank_offset + tp_size * pp_size if is_scale_joiner else tp_size * pp_size
     )
@@ -285,14 +270,6 @@ def _init_parallel_groups(
         max_world_size=get_parallel().max_ep_size,
     )
     initialize_model_parallel(
-        tensor_model_parallel_size=tp_size,
-        attention_data_parallel_size=attn_dp_size,
-        pipeline_model_parallel_size=pp_size,
-        expert_model_parallel_size=moe_ep_size,
-        attention_context_model_parallel_size=attn_cp_size,
-        moe_data_model_parallel_size=moe_dp_size,
-        decode_context_parallel_size=dcp_size,
-        shared_experts_tensor_parallel_size=get_parallel().shared_experts_tp_size,
         duplicate_tp_group=get_disagg().enable_pdmux,
         enable_symm_mem=get_exec().comm.enable_symm_mem,
         # Only WORLD is extended during scale-up. The joiner's model-parallel