config: retire the last process-global config field reads (#33338)
`get_server_args().<field>` reads one process's startup record. Nine sites still did that for a value that has a namespace: the attention backend (5), `skip_tokenizer_init` (2), the draft-aware `load_format`, and a chunked-prefill size in `sglang.kernels`. They now read `get_exec().kernel` / `get_serving()` / `get_model()` / `get_schedule()`, so they see the resolved value including post-publish overrides. The multimodal processor's device selection moves to the instance it was constructed with rather than to a namespace: `base_gpu_id` differs per worker (the encode-server DP workers each specialise their own copy), so no process-global value can stand in for it, and engines sharing a tokenizer process each need their own. Branch order, the NPU preprocess patches, and the case that leaves "device" unset are unchanged. What stays on `get_server_args()` is the derived API — `@property` and method members computed from several fields plus the HF config (`mamba_cache_chunk_size`, `get_model_config()`, `enable_mamba_extra_buffer*`) — plus three config-intent reads of live-shadowed sizes, each of which needs an answer the live topology property cannot give (the DSA indexer's PP gate must short-circuit before touching the PP group, `allocation`'s DCP gate asks whether DCP was configured at all, and the CUDA-IPC recycler runs where no group exists). A new AST ratchet pins both shapes it can see — the direct call and an alias bound from it in the same function — at 0 and 12 respectively, exempting the derived APIs and those three sites by name. The alias-form baseline is not zero: those reads are mostly per-runner fields in model code, and lowering them is the next slice. Two fixtures stopped faking config: `test_dllm_fdfo_kv_reuse` rebound `allocation.get_server_args` to a SimpleNamespace, which silently stops intercepting the moment a reader migrates; it publishes a real config instead.
This commit is contained in:
@@ -69,10 +69,10 @@ resolved configuration lives in the namespace bags.**
|
|||||||
- **Per-instance boundaries** — the tokenizer-manager family, everything under
|
- **Per-instance boundaries** — the tokenizer-manager family, everything under
|
||||||
`entrypoints/`, and the tokenizer-process multimodal processors read
|
`entrypoints/`, and the tokenizer-process multimodal processors read
|
||||||
`self.server_args`: several `Engine`s can share one process, and the process-global
|
`self.server_args`: several `Engine`s can share one process, and the process-global
|
||||||
bags are last-publish-wins across engines. (The mm-processor boundary is not yet
|
bags are last-publish-wins across engines. `base_gpu_id` also differs per worker
|
||||||
airtight: `BaseMultimodalProcessor.process_mm_data` still reads `base_gpu_id` /
|
(the encode-server DP workers each specialize their own copy), so no process-global
|
||||||
`rl_on_policy_target` through `get_server_args()` — a known last-publish-wins gap,
|
value can stand in for it — `BaseMultimodalProcessor._fast_image_processor_device`
|
||||||
not a pattern to copy.)
|
is the shape to copy.
|
||||||
- **Whole-object passes** (`f(server_args)` handing the instance along) keep the
|
- **Whole-object passes** (`f(server_args)` handing the instance along) keep the
|
||||||
supplied-instance contract; don't rewrite the parameter reads to bag reads unless the
|
supplied-instance contract; don't rewrite the parameter reads to bag reads unless the
|
||||||
field is runtime-mutated (see the elastic-EP `ep_size` case in
|
field is runtime-mutated (see the elastic-EP `ep_size` case in
|
||||||
|
|||||||
@@ -579,10 +579,10 @@ def prewarm_mhc_pre(
|
|||||||
the TileLang/DeepGEMM on-disk JIT cache, so this cost is paid only on a cold
|
the TileLang/DeepGEMM on-disk JIT cache, so this cost is paid only on a cold
|
||||||
cache; later server runs hit the cache. Driven once per process from load_weights.
|
cache; later server runs hit the cache. Driven once per process from load_weights.
|
||||||
"""
|
"""
|
||||||
from sglang.srt.runtime_context import get_server_args
|
from sglang.srt.runtime_context import get_schedule
|
||||||
|
|
||||||
hc_mult, hidden_size = residual.shape[-2], residual.shape[-1]
|
hc_mult, hidden_size = residual.shape[-2], residual.shape[-1]
|
||||||
max_num_tokens = get_server_args().chunked_prefill_size
|
max_num_tokens = get_schedule().chunked_prefill_size
|
||||||
buckets = get_mhc_pre_token_count_representatives(
|
buckets = get_mhc_pre_token_count_representatives(
|
||||||
max_num_tokens, hc_mult * hidden_size
|
max_num_tokens, hc_mult * hidden_size
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ from sglang.srt.model_executor.forward_batch_info import (
|
|||||||
compute_position,
|
compute_position,
|
||||||
)
|
)
|
||||||
from sglang.srt.model_executor.forward_context import get_attn_backend
|
from sglang.srt.model_executor.forward_context import get_attn_backend
|
||||||
from sglang.srt.runtime_context import get_device, get_parallel, get_server_args
|
from sglang.srt.runtime_context import get_device, get_exec, get_parallel
|
||||||
from sglang.srt.speculative.spec_info import SpecInput
|
from sglang.srt.speculative.spec_info import SpecInput
|
||||||
from sglang.srt.utils import BumpAllocator, empty_context, get_bool_env_var, is_hip
|
from sglang.srt.utils import BumpAllocator, empty_context, get_bool_env_var, is_hip
|
||||||
|
|
||||||
@@ -634,7 +634,7 @@ class TboForwardBatchPreparer:
|
|||||||
sum_field=None,
|
sum_field=None,
|
||||||
)
|
)
|
||||||
_, child_b.extend_start_loc = compute_position(
|
_, child_b.extend_start_loc = compute_position(
|
||||||
get_server_args().attention_backend,
|
get_exec().kernel.attention_backend,
|
||||||
child_b.extend_prefix_lens,
|
child_b.extend_prefix_lens,
|
||||||
child_b.extend_seq_lens,
|
child_b.extend_seq_lens,
|
||||||
child_b.extend_num_tokens,
|
child_b.extend_num_tokens,
|
||||||
|
|||||||
@@ -376,7 +376,9 @@ def attn_backend_wrapper(runner: "ModelRunner", full_attn_backend: "AttentionBac
|
|||||||
"gdn_backend.sm100_flashinfer_default",
|
"gdn_backend.sm100_flashinfer_default",
|
||||||
linear_attn_prefill_backend=prefill_default,
|
linear_attn_prefill_backend=prefill_default,
|
||||||
)
|
)
|
||||||
initialize_linear_attn_config(runner.server_args, prefill_default)
|
initialize_linear_attn_config(
|
||||||
|
runner.server_args, prefill_default=prefill_default
|
||||||
|
)
|
||||||
hybrid_backend_cls = HybridLinearAttnBackend
|
hybrid_backend_cls = HybridLinearAttnBackend
|
||||||
if hybrid_gdn_config(runner.model_config) is not None:
|
if hybrid_gdn_config(runner.model_config) is not None:
|
||||||
if is_blackwell():
|
if is_blackwell():
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ from sglang.srt.layers.rotary_embedding.yarn import (
|
|||||||
yarn_get_mscale_simple,
|
yarn_get_mscale_simple,
|
||||||
yarn_linear_ramp_mask,
|
yarn_linear_ramp_mask,
|
||||||
)
|
)
|
||||||
from sglang.srt.runtime_context import get_exec, get_server_args
|
from sglang.srt.runtime_context import get_exec
|
||||||
from sglang.srt.utils import (
|
from sglang.srt.utils import (
|
||||||
cpu_has_amx_support,
|
cpu_has_amx_support,
|
||||||
is_cuda,
|
is_cuda,
|
||||||
@@ -42,7 +42,6 @@ if _is_xpu:
|
|||||||
from sgl_kernel import multimodal_rotary_embedding
|
from sgl_kernel import multimodal_rotary_embedding
|
||||||
|
|
||||||
from sglang.kernels.ops.attention.mrope import apply_interleaved_rope_triton
|
from sglang.kernels.ops.attention.mrope import apply_interleaved_rope_triton
|
||||||
from sglang.srt.runtime_context import get_server_args
|
|
||||||
|
|
||||||
|
|
||||||
def apply_interleaved_rope(x: torch.Tensor, mrope_section: list) -> torch.Tensor:
|
def apply_interleaved_rope(x: torch.Tensor, mrope_section: list) -> torch.Tensor:
|
||||||
@@ -144,7 +143,7 @@ class MRotaryEmbedding(RotaryEmbedding):
|
|||||||
last_dim = cos_sin.size()[-1]
|
last_dim = cos_sin.size()[-1]
|
||||||
cos, sin = cos_sin.chunk(2, dim=-1)
|
cos, sin = cos_sin.chunk(2, dim=-1)
|
||||||
if self.mrope_interleaved:
|
if self.mrope_interleaved:
|
||||||
if support_triton(get_server_args().attention_backend):
|
if support_triton(get_exec().kernel.attention_backend):
|
||||||
cos = apply_interleaved_rope_triton(cos, self.mrope_section)
|
cos = apply_interleaved_rope_triton(cos, self.mrope_section)
|
||||||
sin = apply_interleaved_rope_triton(sin, self.mrope_section)
|
sin = apply_interleaved_rope_triton(sin, self.mrope_section)
|
||||||
else:
|
else:
|
||||||
|
|||||||
@@ -38,6 +38,7 @@ from sglang.srt.runtime_context import (
|
|||||||
get_parallel,
|
get_parallel,
|
||||||
get_schedule,
|
get_schedule,
|
||||||
get_server_args,
|
get_server_args,
|
||||||
|
get_serving,
|
||||||
)
|
)
|
||||||
from sglang.srt.utils import flatten_nested_list, is_hip, is_npu, print_warning_once
|
from sglang.srt.utils import flatten_nested_list, is_hip, is_npu, print_warning_once
|
||||||
from sglang.srt.utils.stale_shm_cleanup import make_shm_name
|
from sglang.srt.utils.stale_shm_cleanup import make_shm_name
|
||||||
@@ -554,7 +555,7 @@ def _acknowledge_deferred_cuda_ipc_cache_hits(
|
|||||||
return
|
return
|
||||||
# The pool's recycler counts the whole TP group, so the acknowledgement must
|
# The pool's recycler counts the whole TP group, so the acknowledgement must
|
||||||
# match that count even when an attention subgroup is smaller.
|
# match that count even when an attention subgroup is smaller.
|
||||||
consumer_count = max(get_server_args().tp_size, 1)
|
consumer_count = max(parallel.tp_size, 1)
|
||||||
for item in items:
|
for item in items:
|
||||||
item.acknowledge_deferred_cuda_ipc_feature(consumer_count)
|
item.acknowledge_deferred_cuda_ipc_feature(consumer_count)
|
||||||
|
|
||||||
@@ -2024,7 +2025,7 @@ def wrap_shm_features(obj):
|
|||||||
"""
|
"""
|
||||||
Scan the object for multimodal tensors and wrap them in SHM pointers.
|
Scan the object for multimodal tensors and wrap them in SHM pointers.
|
||||||
"""
|
"""
|
||||||
if _get_is_default_transport() or get_server_args().skip_tokenizer_init:
|
if _get_is_default_transport() or get_serving().skip_tokenizer_init:
|
||||||
return obj
|
return obj
|
||||||
|
|
||||||
if obj.mm_inputs:
|
if obj.mm_inputs:
|
||||||
@@ -2085,7 +2086,7 @@ def unwrap_shm_features(obj):
|
|||||||
Restore ShmPointerMMData wrappers back into standard torch.Tensors.
|
Restore ShmPointerMMData wrappers back into standard torch.Tensors.
|
||||||
Handles both single requests and batch requests.
|
Handles both single requests and batch requests.
|
||||||
"""
|
"""
|
||||||
if _get_is_default_transport() or get_server_args().skip_tokenizer_init:
|
if _get_is_default_transport() or get_serving().skip_tokenizer_init:
|
||||||
return obj
|
return obj
|
||||||
# Handle batch requests
|
# Handle batch requests
|
||||||
if isinstance(obj, BaseBatchReq):
|
if isinstance(obj, BaseBatchReq):
|
||||||
|
|||||||
@@ -26,7 +26,7 @@ from sglang.srt.mem_cache.common import (
|
|||||||
evict_from_tree_cache,
|
evict_from_tree_cache,
|
||||||
)
|
)
|
||||||
from sglang.srt.mem_cache.memory_pool import HybridReqToTokenPool, ReqToTokenPool
|
from sglang.srt.mem_cache.memory_pool import HybridReqToTokenPool, ReqToTokenPool
|
||||||
from sglang.srt.runtime_context import get_server_args
|
from sglang.srt.runtime_context import get_exec, get_server_args
|
||||||
from sglang.srt.utils import (
|
from sglang.srt.utils import (
|
||||||
is_cpu,
|
is_cpu,
|
||||||
is_cuda,
|
is_cuda,
|
||||||
@@ -65,7 +65,7 @@ def write_cache_indices(
|
|||||||
prefix_tensors: list[torch.Tensor],
|
prefix_tensors: list[torch.Tensor],
|
||||||
req_to_token_pool: ReqToTokenPool,
|
req_to_token_pool: ReqToTokenPool,
|
||||||
):
|
):
|
||||||
if support_triton(get_server_args().attention_backend):
|
if support_triton(get_exec().kernel.attention_backend):
|
||||||
prefix_pointers = torch.tensor(
|
prefix_pointers = torch.tensor(
|
||||||
[t.data_ptr() for t in prefix_tensors],
|
[t.data_ptr() for t in prefix_tensors],
|
||||||
dtype=torch.uint64,
|
dtype=torch.uint64,
|
||||||
@@ -106,7 +106,7 @@ def get_last_loc(
|
|||||||
req_pool_indices_tensor: torch.Tensor,
|
req_pool_indices_tensor: torch.Tensor,
|
||||||
prefix_lens_tensor: torch.Tensor,
|
prefix_lens_tensor: torch.Tensor,
|
||||||
) -> torch.Tensor:
|
) -> torch.Tensor:
|
||||||
attn_backend = get_server_args().attention_backend
|
attn_backend = get_exec().kernel.attention_backend
|
||||||
uses_triton_dispatch = attn_backend not in ("ascend", "torch_native")
|
uses_triton_dispatch = attn_backend not in ("ascend", "torch_native")
|
||||||
|
|
||||||
if _is_hip and uses_triton_dispatch:
|
if _is_hip and uses_triton_dispatch:
|
||||||
|
|||||||
@@ -72,7 +72,6 @@ from sglang.srt.runtime_context import (
|
|||||||
get_exec,
|
get_exec,
|
||||||
get_forward,
|
get_forward,
|
||||||
get_parallel,
|
get_parallel,
|
||||||
get_server_args,
|
|
||||||
)
|
)
|
||||||
from sglang.srt.utils import (
|
from sglang.srt.utils import (
|
||||||
LazyValue,
|
LazyValue,
|
||||||
@@ -421,7 +420,7 @@ class GptOssAttention(nn.Module):
|
|||||||
|
|
||||||
# Choose dtype of sinks based on attention backend: trtllm_mha requires float32,
|
# Choose dtype of sinks based on attention backend: trtllm_mha requires float32,
|
||||||
# others can use bfloat16
|
# others can use bfloat16
|
||||||
attn_backend = get_server_args().attention_backend
|
attn_backend = get_exec().kernel.attention_backend
|
||||||
sinks_dtype = torch.float32 if attn_backend == "trtllm_mha" else torch.bfloat16
|
sinks_dtype = torch.float32 if attn_backend == "trtllm_mha" else torch.bfloat16
|
||||||
self.sinks = nn.Parameter(
|
self.sinks = nn.Parameter(
|
||||||
torch.empty(self.num_heads, dtype=sinks_dtype), requires_grad=False
|
torch.empty(self.num_heads, dtype=sinks_dtype), requires_grad=False
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ from sglang.srt.models.inkling_common.util import (
|
|||||||
lora_compatible_layout_enabled,
|
lora_compatible_layout_enabled,
|
||||||
)
|
)
|
||||||
from sglang.srt.models.llama import LlamaMLP
|
from sglang.srt.models.llama import LlamaMLP
|
||||||
from sglang.srt.runtime_context import get_exec, get_server_args
|
from sglang.srt.runtime_context import get_exec, get_model
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -484,7 +484,7 @@ class InklingBatchDenseMLP(nn.Module, FusedMoELoadingMixin):
|
|||||||
# All shared experts must share one global weight scale (reshard with
|
# All shared experts must share one global weight scale (reshard with
|
||||||
# single_global_scale=True). ModelOpt's input_scale = amax / (6 * 448).
|
# single_global_scale=True). ModelOpt's input_scale = amax / (6 * 448).
|
||||||
flat2 = scale2.reshape(-1).float()
|
flat2 = scale2.reshape(-1).float()
|
||||||
if get_server_args().load_format == "dummy" and not bool(
|
if get_model().load_format == "dummy" and not bool(
|
||||||
torch.all(flat2 == flat2[0])
|
torch.all(flat2 == flat2[0])
|
||||||
):
|
):
|
||||||
# Dummy loading uses per-element noise; replace it with a valid scale.
|
# Dummy loading uses per-element noise; replace it with a valid scale.
|
||||||
|
|||||||
@@ -20,7 +20,6 @@ from sglang.srt.managers.schedule_batch import (
|
|||||||
MultimodalProcessorOutput,
|
MultimodalProcessorOutput,
|
||||||
)
|
)
|
||||||
from sglang.srt.multimodal.processors.executor import MultimodalProcessorExecutor
|
from sglang.srt.multimodal.processors.executor import MultimodalProcessorExecutor
|
||||||
from sglang.srt.runtime_context import get_server_args
|
|
||||||
from sglang.srt.utils import (
|
from sglang.srt.utils import (
|
||||||
CLIENT_MEDIA_EXCEPTIONS,
|
CLIENT_MEDIA_EXCEPTIONS,
|
||||||
envs,
|
envs,
|
||||||
@@ -485,6 +484,37 @@ class BaseMultimodalProcessor(ABC):
|
|||||||
return self._processor, self._tokenizer
|
return self._processor, self._tokenizer
|
||||||
return processor, processor.tokenizer
|
return processor, processor.tokenizer
|
||||||
|
|
||||||
|
def _fast_image_processor_device(self, processor) -> Optional[str]:
|
||||||
|
"""The device for the fast image processor, or None to leave it unset.
|
||||||
|
|
||||||
|
Resolved from this processor's own ``server_args``: engines sharing a
|
||||||
|
tokenizer process each carry their own ``base_gpu_id``.
|
||||||
|
"""
|
||||||
|
server_args = self.server_args
|
||||||
|
if _is_cpu or server_args.rl_on_policy_target is not None:
|
||||||
|
return "cpu"
|
||||||
|
if _is_xpu:
|
||||||
|
return "xpu"
|
||||||
|
if not _is_npu:
|
||||||
|
return f"cuda:{server_args.base_gpu_id}"
|
||||||
|
if processor.__class__.__name__ not in {"Glm4vProcessor", "Glm46VProcessor"}:
|
||||||
|
# For qwen-vl, the processor hits a reshape issue from the Ascend
|
||||||
|
# dims restriction.
|
||||||
|
from sglang.srt.hardware_backend.npu.modules.qwen_vl_processor import (
|
||||||
|
npu_apply_qwen_image_preprocess_patch,
|
||||||
|
)
|
||||||
|
|
||||||
|
npu_apply_qwen_image_preprocess_patch()
|
||||||
|
return "npu"
|
||||||
|
if processor.__class__.__name__ == "Glm46VProcessor":
|
||||||
|
from sglang.srt.hardware_backend.npu.modules.glm46v_processor import (
|
||||||
|
npu_apply_glm46v_image_preprocess_patch,
|
||||||
|
)
|
||||||
|
|
||||||
|
npu_apply_glm46v_image_preprocess_patch()
|
||||||
|
return "npu"
|
||||||
|
return None
|
||||||
|
|
||||||
def process_mm_data(
|
def process_mm_data(
|
||||||
self,
|
self,
|
||||||
input_text,
|
input_text,
|
||||||
@@ -537,31 +567,9 @@ class BaseMultimodalProcessor(ABC):
|
|||||||
and isinstance(processor.image_processor, BaseImageProcessor)
|
and isinstance(processor.image_processor, BaseImageProcessor)
|
||||||
and not self.disable_fast_image_processor
|
and not self.disable_fast_image_processor
|
||||||
):
|
):
|
||||||
if _is_cpu or get_server_args().rl_on_policy_target is not None:
|
device = self._fast_image_processor_device(processor)
|
||||||
kwargs["device"] = "cpu"
|
if device is not None:
|
||||||
elif _is_xpu:
|
kwargs["device"] = device
|
||||||
kwargs["device"] = "xpu"
|
|
||||||
elif not _is_npu:
|
|
||||||
base_gpu_id = get_server_args().base_gpu_id
|
|
||||||
kwargs["device"] = f"cuda:{base_gpu_id}"
|
|
||||||
elif processor.__class__.__name__ not in {
|
|
||||||
"Glm4vProcessor",
|
|
||||||
"Glm46VProcessor",
|
|
||||||
}:
|
|
||||||
# Note: for qwen-vl, processor has some reshape issue because of dims restriction on Ascend.
|
|
||||||
from sglang.srt.hardware_backend.npu.modules.qwen_vl_processor import (
|
|
||||||
npu_apply_qwen_image_preprocess_patch,
|
|
||||||
)
|
|
||||||
|
|
||||||
npu_apply_qwen_image_preprocess_patch()
|
|
||||||
kwargs["device"] = "npu"
|
|
||||||
elif processor.__class__.__name__ == "Glm46VProcessor":
|
|
||||||
from sglang.srt.hardware_backend.npu.modules.glm46v_processor import (
|
|
||||||
npu_apply_glm46v_image_preprocess_patch,
|
|
||||||
)
|
|
||||||
|
|
||||||
npu_apply_glm46v_image_preprocess_patch()
|
|
||||||
kwargs["device"] = "npu"
|
|
||||||
|
|
||||||
# Avoid double BOS when the chat template already wrote one.
|
# Avoid double BOS when the chat template already wrote one.
|
||||||
if self._tokenizer_auto_adds_specials and isinstance(input_text, str):
|
if self._tokenizer_auto_adds_specials and isinstance(input_text, str):
|
||||||
|
|||||||
@@ -7,9 +7,9 @@ from types import SimpleNamespace
|
|||||||
import torch
|
import torch
|
||||||
|
|
||||||
from sglang.srt.dllm.mixin.scheduler import DllmManager
|
from sglang.srt.dllm.mixin.scheduler import DllmManager
|
||||||
from sglang.srt.mem_cache import allocation
|
|
||||||
from sglang.srt.mem_cache.allocation import alloc_for_extend
|
from sglang.srt.mem_cache.allocation import alloc_for_extend
|
||||||
from sglang.srt.mem_cache.memory_pool import ReqToTokenPool
|
from sglang.srt.mem_cache.memory_pool import ReqToTokenPool
|
||||||
|
from sglang.srt.runtime_context import get_context
|
||||||
from sglang.test.ci.ci_register import register_cpu_ci
|
from sglang.test.ci.ci_register import register_cpu_ci
|
||||||
|
|
||||||
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
|
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
|
||||||
@@ -117,16 +117,11 @@ class TestDllmFdfoKvReuse(unittest.TestCase):
|
|||||||
self.pool = ReqToTokenPool(
|
self.pool = ReqToTokenPool(
|
||||||
size=8, max_context_len=64, device="cpu", enable_memory_saver=False
|
size=8, max_context_len=64, device="cpu", enable_memory_saver=False
|
||||||
)
|
)
|
||||||
self._old_support_triton = allocation.support_triton
|
override = get_context().override_server_args(
|
||||||
self._old_get_server_args = allocation.get_server_args
|
|
||||||
allocation.support_triton = lambda _: False
|
|
||||||
allocation.get_server_args = lambda: SimpleNamespace(
|
|
||||||
attention_backend="torch_native", dcp_size=1
|
attention_backend="torch_native", dcp_size=1
|
||||||
)
|
)
|
||||||
|
override.install()
|
||||||
def tearDown(self):
|
self.addCleanup(override.restore)
|
||||||
allocation.support_triton = self._old_support_triton
|
|
||||||
allocation.get_server_args = self._old_get_server_args
|
|
||||||
|
|
||||||
def test_alloc_for_extend_mixed_reuse_allocates_only_fresh_and_writes_rows(self):
|
def test_alloc_for_extend_mixed_reuse_allocates_only_fresh_and_writes_rows(self):
|
||||||
allocator = _FakeAllocator(base=200)
|
allocator = _FakeAllocator(base=200)
|
||||||
|
|||||||
@@ -0,0 +1,81 @@
|
|||||||
|
"""The fast-image-processor device comes from the processor's own ServerArgs.
|
||||||
|
|
||||||
|
Regression: the device decision read the published global ServerArgs, which is
|
||||||
|
last-publish-wins. Two engines in one tokenizer process then shared whichever
|
||||||
|
config published last, so one engine's images were preprocessed on the other
|
||||||
|
engine's GPU.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import unittest
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from sglang.srt.multimodal.processors.base_processor import BaseMultimodalProcessor
|
||||||
|
from sglang.srt.server_args import ServerArgs
|
||||||
|
from sglang.test.ci.ci_register import register_cpu_ci
|
||||||
|
from sglang.test.test_utils import CustomTestCase
|
||||||
|
|
||||||
|
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
|
||||||
|
|
||||||
|
BASE = "sglang.srt.multimodal.processors.base_processor"
|
||||||
|
|
||||||
|
|
||||||
|
class _Processor:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class _StubProcessor(BaseMultimodalProcessor):
|
||||||
|
async def process_mm_data_async(self, *args, **kwargs):
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
|
def _make(**fields):
|
||||||
|
processor = _StubProcessor.__new__(_StubProcessor)
|
||||||
|
processor.server_args = ServerArgs(model_path="dummy", **fields)
|
||||||
|
return processor
|
||||||
|
|
||||||
|
|
||||||
|
class TestFastImageProcessorDevice(CustomTestCase):
|
||||||
|
def _device(self, processor, **platform):
|
||||||
|
flags = {"_is_cpu": False, "_is_xpu": False, "_is_npu": False}
|
||||||
|
flags.update(platform)
|
||||||
|
with patch.multiple(BASE, **flags):
|
||||||
|
return processor._fast_image_processor_device(_Processor())
|
||||||
|
|
||||||
|
def test_device_follows_the_instance_base_gpu_id(self):
|
||||||
|
self.assertEqual(self._device(_make(base_gpu_id=3)), "cuda:3")
|
||||||
|
|
||||||
|
def test_engines_in_one_process_keep_their_own_device(self):
|
||||||
|
first, second = _make(base_gpu_id=0), _make(base_gpu_id=5)
|
||||||
|
self.assertEqual(self._device(first), "cuda:0")
|
||||||
|
self.assertEqual(self._device(second), "cuda:5")
|
||||||
|
|
||||||
|
def test_publishing_another_config_does_not_move_the_device(self):
|
||||||
|
from sglang.srt.runtime_context import get_context
|
||||||
|
|
||||||
|
processor = _make(base_gpu_id=2)
|
||||||
|
override = get_context().override_server_args(base_gpu_id=7)
|
||||||
|
override.install()
|
||||||
|
self.addCleanup(override.restore)
|
||||||
|
self.assertEqual(self._device(processor), "cuda:2")
|
||||||
|
|
||||||
|
def test_rl_on_policy_target_forces_cpu(self):
|
||||||
|
processor = _make(base_gpu_id=3, rl_on_policy_target="fsdp")
|
||||||
|
self.assertEqual(self._device(processor), "cpu")
|
||||||
|
|
||||||
|
def test_cpu_and_xpu_platforms_win_over_base_gpu_id(self):
|
||||||
|
processor = _make(base_gpu_id=3)
|
||||||
|
self.assertEqual(self._device(processor, _is_cpu=True), "cpu")
|
||||||
|
self.assertEqual(self._device(processor, _is_xpu=True), "xpu")
|
||||||
|
|
||||||
|
def test_npu_glm4v_leaves_the_device_unset(self):
|
||||||
|
class Glm4vProcessor:
|
||||||
|
pass
|
||||||
|
|
||||||
|
processor = _make(base_gpu_id=3)
|
||||||
|
with patch.multiple(BASE, _is_cpu=False, _is_xpu=False, _is_npu=True):
|
||||||
|
device = processor._fast_image_processor_device(Glm4vProcessor())
|
||||||
|
self.assertIsNone(device)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -0,0 +1,165 @@
|
|||||||
|
"""Ratchet guard: process-global config reads may only decrease.
|
||||||
|
|
||||||
|
``get_server_args()`` returns the published ``ServerArgs`` — one process's
|
||||||
|
startup record. Config decisions read the namespace accessors instead
|
||||||
|
(``get_exec()`` / ``get_memory()`` / …), which carry the resolved value
|
||||||
|
including post-publish overrides, and per-runner values come from the runner
|
||||||
|
that owns them.
|
||||||
|
|
||||||
|
Two shapes count as a read: the direct ``get_server_args().field``, and the
|
||||||
|
alias ``sa = get_server_args()`` followed by ``sa.field`` in the same function.
|
||||||
|
A whole-object pass (``def f(server_args)``) is not a global read and is not
|
||||||
|
counted — there the caller decided which instance to hand over.
|
||||||
|
|
||||||
|
What legitimately remains:
|
||||||
|
|
||||||
|
- **Derived APIs.** ``@property`` and method members of ``ServerArgs``
|
||||||
|
(``mamba_cache_chunk_size``, ``get_model_config()``,
|
||||||
|
``enable_mamba_extra_buffer()``, …) are computed from several fields plus the
|
||||||
|
HF config, so they are not namespace leaves and ``ServerArgs`` is their only
|
||||||
|
home. Exempt by name below.
|
||||||
|
- **Config-intent reads of live-shadowed sizes.** ``get_parallel()`` shadows
|
||||||
|
``tp/pp/dcp/attn_cp/moe_dp_size`` with the live topology, so a config-intent
|
||||||
|
read of one has nowhere else to go. Each exempt site needs an answer the live
|
||||||
|
property cannot give:
|
||||||
|
|
||||||
|
- ``dsa_indexer.pp_size`` gates ``pp_size > 1 and not get_pp_group()...``, and
|
||||||
|
the short circuit is the point: with PP off the group is never touched, which
|
||||||
|
is what lets the ``Indexer`` be constructed before distributed init. The live
|
||||||
|
property would demand the group either way.
|
||||||
|
- ``allocation.dcp_size`` asks whether DCP was *configured*; the live property
|
||||||
|
reads ``get_dcp_group()``, and that group is only installed when DCP is on.
|
||||||
|
- ``cuda_ipc_transport_utils.tp_size`` runs in the tokenizer process, which has
|
||||||
|
no groups at all (the call site already guards for "not published yet").
|
||||||
|
- The alias-form baseline is not zero yet. Lowering it is the next slice; the
|
||||||
|
failure message lists the sites whenever the count moves.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from sglang.test.ci.ci_register import register_cpu_ci
|
||||||
|
|
||||||
|
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
|
||||||
|
|
||||||
|
import ast
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import sglang
|
||||||
|
from sglang.test.test_utils import CustomTestCase
|
||||||
|
|
||||||
|
# srt is the migrated surface; the rest of the package has no reads today and is
|
||||||
|
# scanned so a new one cannot appear there unnoticed.
|
||||||
|
_PACKAGE_ROOT = Path(next(iter(sglang.__path__)))
|
||||||
|
|
||||||
|
_DERIVED_MEMBERS = frozenset(
|
||||||
|
{
|
||||||
|
"cutedsl_moe_max_num_tokens",
|
||||||
|
"enable_mamba_extra_buffer",
|
||||||
|
"enable_mamba_extra_buffer_lazy",
|
||||||
|
"get_attention_backends",
|
||||||
|
"get_model_config",
|
||||||
|
"mamba_cache_chunk_size",
|
||||||
|
"max_speculative_num_draft_tokens",
|
||||||
|
"model_config",
|
||||||
|
"use_mla_backend",
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
_CONFIG_INTENT_SIZES = frozenset(
|
||||||
|
{
|
||||||
|
("srt/layers/attention/dsa/dsa_indexer.py", "pp_size"),
|
||||||
|
("srt/mem_cache/allocation.py", "dcp_size"),
|
||||||
|
("srt/utils/cuda_ipc_transport_utils.py", "tp_size"),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
_DIRECT_BASELINE = 0
|
||||||
|
_ALIAS_BASELINE = 12
|
||||||
|
|
||||||
|
|
||||||
|
def _is_global_call(node) -> bool:
|
||||||
|
return (
|
||||||
|
isinstance(node, ast.Call)
|
||||||
|
and isinstance(node.func, ast.Name)
|
||||||
|
and node.func.id == "get_server_args"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _collect(rel: str, tree: ast.AST):
|
||||||
|
"""The (direct, alias) field reads in one module."""
|
||||||
|
direct, alias = [], []
|
||||||
|
|
||||||
|
def counted(attr: str) -> bool:
|
||||||
|
return attr not in _DERIVED_MEMBERS and (rel, attr) not in _CONFIG_INTENT_SIZES
|
||||||
|
|
||||||
|
for node in ast.walk(tree):
|
||||||
|
if (
|
||||||
|
isinstance(node, ast.Attribute)
|
||||||
|
and _is_global_call(node.value)
|
||||||
|
and counted(node.attr)
|
||||||
|
):
|
||||||
|
direct.append(f"{rel}:{node.lineno}: get_server_args().{node.attr}")
|
||||||
|
|
||||||
|
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||||
|
continue
|
||||||
|
params = {a.arg for a in list(node.args.args) + list(node.args.kwonlyargs)}
|
||||||
|
bound = {}
|
||||||
|
for inner in ast.walk(node):
|
||||||
|
if isinstance(inner, ast.Assign) and _is_global_call(inner.value):
|
||||||
|
for target in inner.targets:
|
||||||
|
if isinstance(target, ast.Name) and target.id not in params:
|
||||||
|
bound.setdefault(target.id, inner.lineno)
|
||||||
|
if not bound:
|
||||||
|
continue
|
||||||
|
for inner in ast.walk(node):
|
||||||
|
if (
|
||||||
|
isinstance(inner, ast.Attribute)
|
||||||
|
and isinstance(inner.value, ast.Name)
|
||||||
|
and inner.value.id in bound
|
||||||
|
and inner.lineno >= bound[inner.value.id]
|
||||||
|
and counted(inner.attr)
|
||||||
|
):
|
||||||
|
alias.append(
|
||||||
|
f"{rel}:{inner.lineno}: {inner.value.id}.{inner.attr} "
|
||||||
|
f"(bound from get_server_args() at line {bound[inner.value.id]})"
|
||||||
|
)
|
||||||
|
return direct, alias
|
||||||
|
|
||||||
|
|
||||||
|
def _field_reads():
|
||||||
|
direct, alias = [], []
|
||||||
|
for path in sorted(_PACKAGE_ROOT.rglob("*.py")):
|
||||||
|
rel = path.relative_to(_PACKAGE_ROOT).as_posix()
|
||||||
|
try:
|
||||||
|
tree = ast.parse(path.read_text())
|
||||||
|
except SyntaxError:
|
||||||
|
continue
|
||||||
|
module_direct, module_alias = _collect(rel, tree)
|
||||||
|
direct += module_direct
|
||||||
|
alias += module_alias
|
||||||
|
return direct, alias
|
||||||
|
|
||||||
|
|
||||||
|
class TestGlobalConfigReadRatchet(CustomTestCase):
|
||||||
|
def _check(self, kind, reads, baseline):
|
||||||
|
if len(reads) > baseline:
|
||||||
|
self.fail(
|
||||||
|
f"{kind} process-global config field reads grew: {len(reads)} > "
|
||||||
|
f"baseline {baseline}. Read the namespace accessor for the "
|
||||||
|
"field's namespace, or the owning runner for a per-runner "
|
||||||
|
"field:\n" + "\n".join(reads)
|
||||||
|
)
|
||||||
|
if len(reads) < baseline:
|
||||||
|
self.fail(
|
||||||
|
f"{kind} process-global config field reads shrank: {len(reads)} < "
|
||||||
|
f"baseline {baseline}. Lower the baseline in this file to lock "
|
||||||
|
"in the progress."
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_global_field_reads_match_the_baseline(self):
|
||||||
|
direct, alias = _field_reads()
|
||||||
|
self._check("direct", direct, _DIRECT_BASELINE)
|
||||||
|
self._check("alias-form", alias, _ALIAS_BASELINE)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user