Revert "[DSv4] Use cvt PTX for FP32->FP4 conversion (#41015)"

This reverts commit d9f3961abb.

csrc/persistent_topk.cuh

@@ -887,27 +887,14 @@ __global__ void __launch_bounds__(kThreadsPerBlock, 2)
   uint32_t* shared_ordered =
       reinterpret_cast<uint32_t*>(smem_raw + kFixedSmemLarge);
 
-  // RadixRowState for multi-CTA cooperative radix
+  // RadixRowState for multi-CTA cooperative radix.
+  // Zero-initialization is done host-side via cudaMemsetAsync in topk.cu
+  // before launch — that gives a stream-ordered happens-before edge for all
+  // CTAs, which the previous in-kernel init (CTA-0 only + intra-CTA
+  // __syncthreads) did not provide and which manifested as a race against
+  // CTA-1+'s first red_release on arrival_counter.
   RadixRowState* state = &params.row_states[group_id];
 
-  // -- Initialize RadixRowState (only needed if large rows exist) --
-  if (params.max_seq_len > RADIX_THRESHOLD) {
-    if (cta_in_group == 0) {
-      for (uint32_t buf = 0; buf < 3; buf++) {
-        for (uint32_t i = tx; i < RADIX; i += kThreadsPerBlock) {
-          state->histogram[buf][i] = 0;
-        }
-      }
-      if (tx == 0) {
-        state->remaining_k = 0;
-        state->prefix = 0;
-        state->arrival_counter = 0;
-        state->output_counter = 0;
-      }
-    }
-    __syncthreads();
-  }
-
   int barrier_phase = 0;
   const uint32_t total_iters = (params.num_rows + num_groups - 1) / num_groups;
 

csrc/topk.cu

@@ -153,6 +153,29 @@ void launch_persistent_topk(const torch::Tensor& logits,
     TORCH_CHECK(workspace.size(0) >= static_cast<int64_t>(state_bytes),
                 "workspace too small, need ", state_bytes, " bytes");
 
+    // Zero the per-group RadixRowState region before launch — only when the
+    // radix path will actually run (max_seq_len > RADIX_THRESHOLD). The
+    // RadixRowState fields (arrival_counter, histograms) are only touched by
+    // radix_topk; the decode/medium paths inside the persistent kernel
+    // operate purely in shared memory and never read these globals, so a
+    // stale workspace is harmless for them.
+    //
+    // Why we need the memset (when needs_cooperative is true):
+    //   1. arrival_counter accumulates within a launch and is never reset,
+    //      so a prior call leaves it at a large positive value. Without this
+    //      reset, the very first wait_ge in the next call sees counter >>
+    //      target and returns instantly, breaking the barrier.
+    //   2. The previous in-kernel init only ran in CTA-0 with intra-CTA
+    //      __syncthreads(), so it had no happens-before edge to CTA-1+'s
+    //      first red_release. cudaMemsetAsync is stream-ordered: the zero
+    //      is globally visible before any CTA runs.
+    if (needs_cooperative) {
+      cudaError_t mz_err = cudaMemsetAsync(workspace.data_ptr<uint8_t>(), 0,
+                                           state_bytes, stream);
+      TORCH_CHECK(mz_err == cudaSuccess,
+                  "row_states memset failed: ", cudaGetErrorString(mz_err));
+    }
+
     P::PersistentTopKParams params;
     params.input = logits.data_ptr<float>();
     params.output = output.data_ptr<int32_t>();

docker/Dockerfile

@@ -478,6 +478,9 @@ FROM ${FINAL_BASE_IMAGE} AS vllm-base
 
 ARG CUDA_VERSION
 ARG PYTHON_VERSION
+ARG DEADSNAKES_MIRROR_URL
+ARG DEADSNAKES_GPGKEY_URL
+ARG GET_PIP_URL
 
 ENV DEBIAN_FRONTEND=noninteractive
 WORKDIR /vllm-workspace
@@ -487,12 +490,7 @@ WORKDIR /vllm-workspace
 RUN PYTHON_VERSION_STR=$(echo ${PYTHON_VERSION} | sed 's/\.//g') && \
     echo "export PYTHON_VERSION_STR=${PYTHON_VERSION_STR}" >> /etc/environment
 
-# Install Python (via uv / python-build-standalone) and system dependencies.
-# This replaces the deadsnakes PPA, removing the build-time dependency on
-# Launchpad and matching how the build-stage (`base`) installs Python.
-# python-build-standalone bundles dev headers, the venv module, and
-# python3-config, so the python3.X-dev / python3.X-venv apt packages
-# are not needed.
+# Install Python from source and system dependencies
 RUN apt-get update -y \
     && apt-get install -y --no-install-recommends \
         curl \
@@ -502,20 +500,38 @@ RUN apt-get update -y \
         libxext6 \
         libgl1 \
         libibverbs-dev \
+        build-essential \
+        libssl-dev \
+        libffi-dev \
+        zlib1g-dev \
+        libbz2-dev \
+        libreadline-dev \
+        libsqlite3-dev \
+        libncurses-dev \
+        liblzma-dev \
+        libgdbm-dev \
+        uuid-dev \
+        tk-dev \
+    && PYTHON_MAJOR_MINOR=${PYTHON_VERSION} \
+    && PYTHON_FULL_VERSION=$(curl -s https://www.python.org/ftp/python/ \
+        | grep -oE "${PYTHON_MAJOR_MINOR}\.[0-9]+" \
+        | sort -t. -k3 -n | uniq \
+        | tail -1) \
+    && echo "Building Python ${PYTHON_FULL_VERSION} from source..." \
+    && curl -fSL https://www.python.org/ftp/python/${PYTHON_FULL_VERSION}/Python-${PYTHON_FULL_VERSION}.tgz -o /tmp/python.tgz \
+    && tar -xzf /tmp/python.tgz -C /tmp \
+    && cd /tmp/Python-${PYTHON_FULL_VERSION} \
+    && ./configure --enable-optimizations --with-ensurepip=install --prefix=/usr/local \
+    && make -j$(nproc) \
+    && make install \
+    && cd / \
+    && rm -rf /tmp/python.tgz /tmp/Python-${PYTHON_FULL_VERSION} \
+    && ln -sf /usr/local/bin/python${PYTHON_MAJOR_MINOR} /usr/bin/python3 \
+    && ln -sf /usr/local/bin/python${PYTHON_MAJOR_MINOR}-config /usr/bin/python3-config \
+    && ln -sf /usr/local/bin/pip${PYTHON_MAJOR_MINOR} /usr/bin/pip \
     && rm -rf /var/lib/apt/lists/* \
-    && curl -LsSf https://astral.sh/uv/install.sh | sh \
-    && $HOME/.local/bin/uv venv /opt/venv --python ${PYTHON_VERSION} \
-    && rm -f /usr/bin/python3 /usr/bin/python3-config /usr/bin/pip \
-    && ln -s /opt/venv/bin/python3 /usr/bin/python3 \
-    && ln -s /opt/venv/bin/python${PYTHON_VERSION} /usr/bin/python${PYTHON_VERSION} \
-    && ln -s /opt/venv/bin/python3-config /usr/bin/python3-config \
-    && ln -s /opt/venv/bin/pip /usr/bin/pip \
     && python3 --version && python3 -m pip --version
 
-# Activate virtual environment and add uv to PATH
-ENV PATH="/opt/venv/bin:/root/.local/bin:$PATH"
-ENV VIRTUAL_ENV="/opt/venv"
-
 # Install CUDA development tools for runtime JIT compilation
 # (FlashInfer, DeepGEMM, EP kernels all require compilation at runtime)
 RUN CUDA_VERSION_DASH=$(echo $CUDA_VERSION | cut -d. -f1,2 | tr '.' '-') && \
@@ -540,6 +556,9 @@ RUN CUDA_VERSION_DASH=$(echo $CUDA_VERSION | cut -d. -f1,2 | tr '.' '-') && \
     apt-get install -y --no-install-recommends --allow-change-held-packages libnccl-dev=${NCCL_VER} libnccl2=${NCCL_VER} && \
     rm -rf /var/lib/apt/lists/*
 
+# Install uv for faster pip installs
+RUN python3 -m pip install uv
+
 # Environment for uv
 ENV UV_HTTP_TIMEOUT=500
 ENV UV_INDEX_STRATEGY="unsafe-best-match"
@@ -729,7 +748,7 @@ ENV HF_XET_HIGH_PERFORMANCE 1
 ENV HF_HUB_DOWNLOAD_TIMEOUT 60
 
 # Copy in the v1 package for testing (it isn't distributed yet)
-COPY vllm/v1 /opt/venv/lib/python${PYTHON_VERSION}/site-packages/vllm/v1
+COPY vllm/v1 /usr/local/lib/python${PYTHON_VERSION}/dist-packages/vllm/v1
 
 # Source code is used in the `python_only_compile.sh` test
 # We hide it inside `src/` so that this source code

tests/kernels/test_compressor_kv_cache.py

@@ -3,11 +3,12 @@
 """
 Round-trip tests for compressor → FP8 quant + KV cache insert → gather + dequant.
 
-Four test functions cover five paths:
+Two paths tested:
   A) DeepseekV4 Attention: head_dim=512 (448 FP8 nope + 64 bf16 rope), quant_block=64
   B) Indexer:       head_dim=128 (all FP8), quant_block=128
-  C) DeepseekV4 Attention magnitude range: correctness across small/large values
-  D) Indexer fused Triton kernel: compress+norm+rope+quant+insert
+
+These serve as golden references for validating the future fused
+compressor+quant+cache kernel.
 """
 
 import math
@@ -20,12 +21,6 @@ from vllm.v1.attention.ops.deepseek_v4_ops import (
     dequantize_and_gather_k_cache,
     quantize_and_insert_k_cache,
 )
-from vllm.v1.attention.ops.deepseek_v4_ops.fused_compress_quant_cache import (
-    _fused_kv_compress_norm_rope_insert_indexer_attn,
-    _fused_kv_compress_norm_rope_insert_indexer_mxfp4_attn,
-)
-
-from .test_fused_indexer_q_rope_quant import quantize_to_mxfp4
 
 
 def _ue8m0_reference(x: torch.Tensor, block_size: int, fp8_max: float):
@@ -314,222 +309,3 @@ def test_deepseek_v4_quant_magnitude_range():
                 f"Token {t}: rel_err={rel_err:.4f}, abs_diff={abs_diff:.6f}, "
                 f"magnitude={magnitude:.4f}"
             )
-
-
-# ── Test D: Indexer fused K-cache insert (Triton kernels) ────────────────────
-#
-# Both kernels share the same Triton signature; use_fp4 selects between them.
-# Full pipeline: state-cache gather → softmax-weighted compress → RMSNorm →
-#   GPT-J RoPE → quant (MXFP4 or FP8) → paged cache insert.
-
-
-def _reference_kv_compress_norm_rope(
-    state_cache: torch.Tensor,
-    block_table: torch.Tensor,
-    positions: torch.Tensor,
-    rms_weight: torch.Tensor,
-    cos_sin_cache: torch.Tensor,
-    compress_ratio: int = 1,
-    overlap: int = 0,
-    use_fp4: bool = False,
-    rms_eps: float = 1e-6,
-    fp8_max: float = 448.0,
-):
-    """Compress → RMSNorm → GPT-J RoPE → quantize.
-
-    Gathers (1+overlap)*compress_ratio state entries per output token, applies
-    per-element softmax over the scores, and computes the weighted kv sum.
-    Returns (quantized_values, scale) matching the kernel's output layout.
-    """
-    device = state_cache.device
-    head_dim = rms_weight.shape[0]
-    rope_dim = cos_sin_cache.shape[-1]
-    state_block_size = state_cache.shape[1]
-    state_width = state_cache.shape[-1] // 2
-    nope_dim = head_dim - rope_dim
-    total = (1 + overlap) * compress_ratio
-    results = []
-    for pos in positions.tolist():
-        src = torch.arange(pos - total + 1, pos + 1, dtype=torch.int64, device=device)
-        valid = src >= 0
-        idx = src.clamp(min=0)
-        pages = block_table[0, idx // state_block_size]
-        offsets = idx % state_block_size
-        raw = state_cache[pages, offsets].float()  # [total, state_dim]
-
-        # Group 0 (tokens 0..cr-1):   kv[:H],   score[SW:SW+H]
-        # Group 1 (tokens cr..2cr-1): kv[H:2H], score[SW+H:SW+2H]
-        if overlap:
-            sw = state_width
-            g0_kv = raw[:compress_ratio, :head_dim]
-            g1_kv = raw[compress_ratio:, head_dim : 2 * head_dim]
-            g0_scores = raw[:compress_ratio, sw : sw + head_dim]
-            g1_scores = raw[compress_ratio:, sw + head_dim : sw + 2 * head_dim]
-            kv = torch.cat([g0_kv, g1_kv])
-            scores = torch.cat([g0_scores, g1_scores])
-        else:
-            kv = raw[:, :head_dim]
-            scores = raw[:, state_width : state_width + head_dim]
-
-        scores[~valid] = float("-inf")
-        kv[~valid] = 0.0
-        weights = torch.softmax(scores, dim=0)
-        compressed = (kv * weights).sum(dim=0)  # [H]
-        var = (compressed * compressed).mean()
-        normed = compressed * torch.rsqrt(var + rms_eps) * rms_weight.float()
-        compressed_pos = (pos // compress_ratio) * compress_ratio
-        cos, sin = cos_sin_cache[compressed_pos].float().chunk(2)
-        nope, rope = normed.split([nope_dim, rope_dim])
-        rope = torch.stack(
-            [rope[0::2] * cos - rope[1::2] * sin, rope[1::2] * cos + rope[0::2] * sin],
-            dim=-1,
-        ).reshape(rope_dim)
-        results.append(torch.cat([nope, rope]).to(state_cache.dtype))
-    result = torch.stack(results)
-
-    if use_fp4:
-        return quantize_to_mxfp4(result)
-    else:
-        pairs = [
-            _ue8m0_reference(result[t], head_dim, fp8_max) for t in range(len(result))
-        ]
-        quants, scales = zip(*pairs)
-        return torch.stack(quants), torch.cat(scales)
-
-
-@pytest.mark.parametrize("num_tokens", [1, 7, 32])
-@pytest.mark.parametrize("kv_block_size", [16, 32])
-@pytest.mark.parametrize("use_fp4", [False, True])
-def test_fused_kv_insert_indexer(num_tokens: int, kv_block_size: int, use_fp4: bool):
-    """Fused K compress+norm+rope+quant+insert for the indexer KV cache."""
-    HEAD_DIM = 128
-    ROPE_DIM = 64
-    BLOCK_SIZE = 16
-    RMS_EPS = 1e-6
-    FP8_MAX = 448.0
-
-    device = "cuda"
-    torch.manual_seed(42)
-    compress_ratio = 4
-
-    if use_fp4:
-        TOKEN_STRIDE = HEAD_DIM // 2  # packed nibbles: 64 bytes
-        SCALE_DIM = HEAD_DIM // 32  # ue8m0 bytes: 4
-        QUANT_BLOCK = 32
-        kernel = _fused_kv_compress_norm_rope_insert_indexer_mxfp4_attn
-    else:
-        TOKEN_STRIDE = HEAD_DIM  # FP8 bytes: 128
-        SCALE_DIM = 4  # 1 float32: 4 bytes
-        QUANT_BLOCK = HEAD_DIM
-        kernel = _fused_kv_compress_norm_rope_insert_indexer_attn
-
-    # overlap=1 whenever compress_ratio==4, matching DeepseekCompressor logic.
-    overlap = 1 if compress_ratio == 4 else 0
-    coff = 1 + overlap  # multiplier for state_dim per entry
-
-    num_pages = (compress_ratio * num_tokens - 1) // BLOCK_SIZE + 2
-    state_cache = torch.randn(
-        num_pages,
-        BLOCK_SIZE,
-        2 * coff * HEAD_DIM,  # kv_state + score_state, each coff*HEAD_DIM wide
-        dtype=torch.bfloat16,
-        device=device,
-    )
-    block_table = torch.arange(num_pages, dtype=torch.int32, device=device).unsqueeze(0)
-    token_to_req = torch.zeros(num_tokens, dtype=torch.int32, device=device)
-    slot_mapping = torch.arange(num_tokens, dtype=torch.int64, device=device)
-    positions = torch.arange(
-        compress_ratio - 1,
-        compress_ratio * num_tokens,
-        compress_ratio,
-        dtype=torch.int64,
-        device=device,
-    )
-    rms_weight = torch.randn(HEAD_DIM, dtype=torch.bfloat16, device=device)
-    cos_sin_cache = torch.randn(compress_ratio * num_tokens, ROPE_DIM, device=device)
-
-    kv_n_blocks = (num_tokens + kv_block_size - 1) // kv_block_size + 1
-    kv_cache = torch.zeros(
-        kv_n_blocks,
-        kv_block_size * (TOKEN_STRIDE + SCALE_DIM),
-        dtype=torch.uint8,
-        device=device,
-    )
-
-    kernel[(num_tokens,)](
-        state_cache,
-        state_cache.stride(0),
-        state_cache.stride(1),
-        token_to_req,
-        positions,
-        slot_mapping,
-        block_table,
-        block_table.stride(0),
-        BLOCK_SIZE,
-        rms_weight,
-        RMS_EPS,
-        cos_sin_cache,
-        cos_sin_cache.stride(0),
-        kv_cache,
-        slot_mapping,
-        kv_block_size,
-        HEAD_SIZE=HEAD_DIM,
-        TRITON_BLOCK_SIZE=HEAD_DIM,
-        STATE_WIDTH=coff * HEAD_DIM,
-        COMPRESS_RATIO=compress_ratio,
-        OVERLAP=overlap,
-        ROPE_HEAD_DIM=ROPE_DIM,
-        FP8_MAX=FP8_MAX,
-        QUANT_BLOCK=QUANT_BLOCK,
-        TOKEN_STRIDE=TOKEN_STRIDE,
-        SCALE_DIM=SCALE_DIM,
-        KV_BLOCK_STRIDE=kv_cache.stride(0),
-        num_warps=1,
-    )
-
-    k_quant, scale = _reference_kv_compress_norm_rope(
-        state_cache,
-        block_table,
-        positions,
-        rms_weight,
-        cos_sin_cache,
-        compress_ratio,
-        overlap,
-        use_fp4,
-        rms_eps=RMS_EPS,
-        fp8_max=FP8_MAX,
-    )
-
-    if use_fp4:
-        for i in range(num_tokens):
-            blk, pos = i // kv_block_size, i % kv_block_size
-            val_off = pos * TOKEN_STRIDE
-            fp4_actual = kv_cache[blk, val_off : val_off + TOKEN_STRIDE]
-            assert torch.equal(k_quant[i], fp4_actual), (
-                f"token {i}: packed nibbles differ, "
-                f"{(k_quant[i] != fp4_actual).sum()} "
-                f"/ {TOKEN_STRIDE}"
-            )
-
-            scale_off = kv_block_size * TOKEN_STRIDE + pos * SCALE_DIM
-            scale_actual = kv_cache[blk, scale_off : scale_off + SCALE_DIM]
-            assert torch.equal(scale_actual, scale[i]), (
-                f"token {i}: ue8m0 {scale_actual.tolist()} != {scale[i].tolist()}"
-            )
-
-    else:
-        k_quant = k_quant.view(torch.uint8)
-        for i in range(num_tokens):
-            blk, pos = i // kv_block_size, i % kv_block_size
-            val_off = pos * TOKEN_STRIDE
-            assert torch.equal(
-                k_quant[i], kv_cache[blk, val_off : val_off + TOKEN_STRIDE]
-            ), f"token {i}: FP8 bytes differ"
-
-            scale_off = kv_block_size * TOKEN_STRIDE + pos * SCALE_DIM
-            actual_scale = kv_cache[blk, scale_off : scale_off + SCALE_DIM].view(
-                torch.float32
-            )
-            assert torch.equal(actual_scale, scale[i : i + 1]), (
-                f"token {i}: scale {actual_scale.item()} != {scale[i].item()}"
-            )

tests/kernels/test_fused_indexer_q_rope_quant.py

@@ -30,56 +30,6 @@ N_HEAD = 64
 MAX_POS = 4096
 
 
-def quantize_to_mxfp4(
-    x: torch.Tensor,
-) -> tuple[torch.Tensor, torch.Tensor]:
-    """Reference MXFP4 quantization.
-
-    Args:
-        x: [..., head_dim] where head_dim is divisible by 32
-    Returns:
-        packed: [..., head_dim//2]  uint8   2 E2M1 nibbles/byte, low nibble = even index
-        scales: [..., head_dim//32] uint8   1 ue8m0 byte
-    """
-    MXFP4_BLOCK_SIZE = 32
-    orig_shape = x.shape
-    head_dim = orig_shape[-1]
-    n_blocks = head_dim // MXFP4_BLOCK_SIZE
-
-    x_f32 = x.float().reshape(-1, n_blocks, MXFP4_BLOCK_SIZE)
-
-    # Per-block ue8m0 scale: 2^ceil(log2(amax / 6.0)), stored as byte = exp + 127
-    # 6 * 2^-126 is from https://huggingface.co/deepseek-ai/DeepSeek-V4-Pro/blob/main/inference/kernel.py#L163
-    amax = x_f32.abs().amax(dim=-1, keepdim=True).clamp(min=6 * (2**-126))
-    log2_ratio = (amax * (1.0 / 6.0)).log2().ceil().clamp(-127.0, 127.0)
-    scale = log2_ratio.exp2()
-    ue8m0 = (log2_ratio + 127.0).to(torch.uint8)  # [*, n_blocks]
-
-    # E2M1 round-to-nearest-even: midpoints round to the even code.
-    # E2M1 values: [0.00, 0.50, 1.00, 1.50, 2.00, 3.00, 4.00, 6.00]
-    # boundaries:  [   0.25, 0.75, 1.25, 1.75, 2.50, 3.50, 5.00]
-    x_scaled = (x_f32 / scale).clamp(-6.0, 6.0)
-    abs_x = x_scaled.abs()
-    code = torch.zeros_like(abs_x, dtype=torch.int32)
-    code = torch.where(abs_x > 0.25, 1, code)
-    code = torch.where(abs_x >= 0.75, 2, code)
-    code = torch.where(abs_x > 1.25, 3, code)
-    code = torch.where(abs_x >= 1.75, 4, code)
-    code = torch.where(abs_x > 2.5, 5, code)
-    code = torch.where(abs_x >= 3.5, 6, code)
-    code = torch.where(abs_x > 5.0, 7, code)
-    sign = ((x_scaled.view(torch.int32) >> 31) & 1).to(torch.uint8)
-    nibble = code.to(torch.uint8) | (sign << 3)
-
-    # Pack: even-index element → low nibble, odd-index → high nibble
-    nibble_flat = nibble.reshape(-1, head_dim)
-    packed = (nibble_flat[:, 0::2] | (nibble_flat[:, 1::2] << 4)).contiguous()
-    packed = packed.reshape(*orig_shape[:-1], head_dim // 2)
-
-    scales = ue8m0.view(*orig_shape[:-1], n_blocks)
-    return packed, scales
-
-
 def _reference(
     positions: torch.Tensor,
     q: torch.Tensor,
@@ -87,7 +37,6 @@ def _reference(
     weights: torch.Tensor,
     softmax_scale: float,
     head_scale: float,
-    use_fp4: bool = False,
 ) -> tuple[torch.Tensor, torch.Tensor]:
     q_rot = q.clone()
     ops.rotary_embedding(
@@ -100,33 +49,22 @@ def _reference(
         HEAD_DIM - ROPE_DIM,  # rope_dim_offset → rotate the tail
         False,
     )
+    q_fp8, q_scale = per_token_group_quant_fp8(
+        q_rot.view(-1, HEAD_DIM).contiguous(),
+        HEAD_DIM,
+        use_ue8m0=True,
+    )
+    q_fp8 = q_fp8.view(-1, N_HEAD, HEAD_DIM)
+    q_scale = q_scale.view(-1, N_HEAD)
 
-    if use_fp4:
-        q_packed, ue8m0 = quantize_to_mxfp4(q_rot.view(-1, N_HEAD, HEAD_DIM))
-        # Pack 4 ue8m0 bytes into 1 int32
-        q_scale = ue8m0.view(torch.int32).squeeze(-1)
-        # FP4 path: q_scale stays separate (cannot be folded into a per-token scalar)
-        weights_out = weights.to(torch.float32) * softmax_scale * head_scale
-        return (q_packed, q_scale), weights_out
-
-    else:
-        q_fp8, q_scale = per_token_group_quant_fp8(
-            q_rot.view(-1, HEAD_DIM).contiguous(),
-            HEAD_DIM,
-            use_ue8m0=True,
-        )
-        q_fp8 = q_fp8.view(-1, N_HEAD, HEAD_DIM)
-        q_scale = q_scale.view(-1, N_HEAD)
-
-        weights_out = weights.to(torch.float32) * q_scale * softmax_scale * head_scale
-        return q_fp8, weights_out
+    weights_out = weights.to(torch.float32) * q_scale * softmax_scale * head_scale
+    return q_fp8, weights_out
 
 
 @pytest.mark.parametrize("num_tokens", [1, 7, 32, 257])
 @pytest.mark.parametrize("cache_dtype", [torch.float32, torch.bfloat16])
-@pytest.mark.parametrize("use_fp4", [False, True])
 @torch.inference_mode()
-def test_fused_indexer_q_rope_quant_matches_unfused(num_tokens, cache_dtype, use_fp4):
+def test_fused_indexer_q_rope_quant_matches_unfused(num_tokens, cache_dtype):
     device = "cuda"
     torch.manual_seed(0)
 
@@ -139,32 +77,21 @@ def test_fused_indexer_q_rope_quant_matches_unfused(num_tokens, cache_dtype, use
     softmax_scale = HEAD_DIM**-0.5
     head_scale = N_HEAD**-0.5
 
-    q_quant_ref, weights_ref = _reference(
-        positions, q, cos_sin_cache, weights, softmax_scale, head_scale, use_fp4
+    q_fp8_ref, weights_ref = _reference(
+        positions, q, cos_sin_cache, weights, softmax_scale, head_scale
     )
-    q_quant_fused, weights_fused = fused_indexer_q_rope_quant(
-        positions, q.clone(), cos_sin_cache, weights, softmax_scale, head_scale, use_fp4
+    q_fp8_fused, weights_fused = fused_indexer_q_rope_quant(
+        positions, q.clone(), cos_sin_cache, weights, softmax_scale, head_scale
     )
 
-    if use_fp4:
-        q_quant_ref, q_scale_ref = q_quant_ref
-        q_quant_fused, q_scale_fused = q_quant_fused
-
-        assert torch.equal(q_scale_ref, q_scale_fused), (
-            f"q_scale mismatch: "
-            f"{(q_scale_ref != q_scale_fused).sum().item()} "
-            f"/ {q_scale_ref.numel()} bytes differ"
-        )
-
     # fp8 tensors aren't directly comparable via torch.equal — reinterpret as int8.
-    ref_bits = q_quant_ref.view(torch.int8)
-    fused_bits = q_quant_fused.view(torch.int8)
+    ref_bits = q_fp8_ref.view(torch.int8)
+    fused_bits = q_fp8_fused.view(torch.int8)
     assert torch.equal(ref_bits, fused_bits), (
-        f"q_quant_fused mismatch: "
+        f"q_fp8 mismatch: "
         f"{(ref_bits != fused_bits).sum().item()} / {ref_bits.numel()} bytes differ"
     )
 
-    assert weights_fused.dtype == torch.float32
     assert torch.equal(weights_ref, weights_fused), (
         f"weights mismatch: max abs diff "
         f"{(weights_ref - weights_fused).abs().max().item()}"

vllm/envs.py

@@ -245,7 +245,7 @@ if TYPE_CHECKING:
     VLLM_DEBUG_WORKSPACE: bool = False
     VLLM_DISABLE_SHARED_EXPERTS_STREAM: bool = False
     VLLM_SHARED_EXPERTS_STREAM_TOKEN_THRESHOLD: int = 256
-    VLLM_MULTI_STREAM_GEMM_TOKEN_THRESHOLD: int = 4096
+    VLLM_MULTI_STREAM_GEMM_TOKEN_THRESHOLD: int = 1024
     VLLM_COMPILE_CACHE_SAVE_FORMAT: Literal["binary", "unpacked"] = "binary"
     VLLM_USE_V2_MODEL_RUNNER: bool = False
     VLLM_LOG_MODEL_INSPECTION: bool = False
@@ -1669,10 +1669,10 @@ environment_variables: dict[str, Callable[[], Any]] = {
     # tokens the FP8 main GEMM has idle SMs to share with the bf16 aux GEMMs
     # and overlap is a 5-45% win; above it the FP8 GEMM saturates the device
     # and the cross-stream sync becomes pure overhead. Set to 0 to disable
-    # the multi-stream path entirely. Empirical crossover on B300 (148 SMs)
-    # is ~4096; B200 (132 SMs) is expected ~3072.
+    # the multi-stream path entirely. See #PR 41526 for the empirical result
+    # for the default value of 1024 tokens.
     "VLLM_MULTI_STREAM_GEMM_TOKEN_THRESHOLD": lambda: int(
-        os.getenv("VLLM_MULTI_STREAM_GEMM_TOKEN_THRESHOLD", "4096")
+        os.getenv("VLLM_MULTI_STREAM_GEMM_TOKEN_THRESHOLD", "1024")
     ),
     # Format for saving torch.compile cache artifacts
     # - "binary": saves as binary file

vllm/model_executor/layers/sparse_attn_indexer.py

@@ -320,7 +320,12 @@ def sparse_attn_indexer(
         num_rows = logits.shape[0]
         topk_indices = topk_indices_buffer[:num_padded_tokens, :topk_tokens]
 
-        if current_platform.is_cuda() and topk_tokens in (512, 2048):
+        allowed_topk = (
+            (512, 1024, 2048)
+            if current_platform.is_device_capability_family(100)
+            else (512, 2048)
+        )
+        if current_platform.is_cuda() and topk_tokens in allowed_topk:
             workspace_manager = current_workspace_manager()
             (topk_workspace,) = workspace_manager.get_simultaneous(
                 ((RADIX_TOPK_WORKSPACE_SIZE,), torch.uint8),

vllm/model_executor/models/deepseek_v4.py

@@ -715,12 +715,15 @@ class DeepseekV4MoE(nn.Module):
         config = vllm_config.model_config.hf_config
         quant_config = vllm_config.quant_config
         self.prefix = prefix
-        if vllm_config.parallel_config.enable_expert_parallel:
-            self.use_mega_moe = (
-                vllm_config.kernel_config.moe_backend == "deep_gemm_mega_moe"
+        self.use_mega_moe = (
+            vllm_config.kernel_config.moe_backend == "deep_gemm_mega_moe"
+        )
+        if self.use_mega_moe and not vllm_config.parallel_config.enable_expert_parallel:
+            raise NotImplementedError(
+                "DeepSeek V4 MegaMoE currently requires expert parallel. "
+                "Enable it with --enable-expert-parallel, or pick a different "
+                "moe backend."
             )
-        else:
-            self.use_mega_moe = False
 
         self.routed_scaling_factor = getattr(config, "routed_scaling_factor", 1.0)
         self.hidden_size = config.hidden_size
@@ -1224,7 +1227,15 @@ class DeepseekV4Model(nn.Module):
         config = vllm_config.model_config.hf_config
         quant_config = vllm_config.quant_config
         self.config = config
-
+        self.use_mega_moe = (
+            vllm_config.kernel_config.moe_backend == "deep_gemm_mega_moe"
+        )
+        if self.use_mega_moe and not vllm_config.parallel_config.enable_expert_parallel:
+            raise NotImplementedError(
+                "DeepSeek V4 MegaMoE currently requires expert parallel. "
+                "Enable it with --enable-expert-parallel, or pick a different "
+                "moe backend."
+            )
         self.vocab_size = config.vocab_size
         self.hc_eps = config.hc_eps
         self.hc_mult = config.hc_mult

vllm/v1/attention/ops/deepseek_v4_ops/fused_compress_quant_cache.py

@@ -21,7 +21,7 @@ and N_QUANT_BLOCKS ue8m0 bytes.
 
 from vllm.triton_utils import tl, triton
 
-from .fused_indexer_q import _fp32x2_to_fp4x2
+from .fused_indexer_q import _e2m1_nibble
 
 
 # =============================================================================
@@ -566,18 +566,18 @@ def _fused_kv_compress_norm_rope_insert_indexer_mxfp4_attn(
         tl.max(tl.abs(even_2d), axis=1),
         tl.max(tl.abs(odd_2d), axis=1),
     )
-    amax = tl.maximum(amax, 6.0 * (2**-126))
+    amax = tl.maximum(amax, 1e-4)
 
     # ue8m0 block scale: 2^ceil(log2(amax / 6.0)), stored as (exp + 127) byte.
-    log2_ratio = tl.ceil(tl.log2(amax * (1.0 / 6.0)))
+    log2_ratio = tl.ceil(tl.log2(amax / 6.0))
     log2_ratio = tl.minimum(tl.maximum(log2_ratio, -127.0), 127.0)
     inv_scale = tl.exp2(-log2_ratio)
     ue8m0 = (log2_ratio + 127.0).to(tl.uint8)  # [N_QUANT_BLOCKS]
 
     inv_scale_col = tl.reshape(inv_scale, (N_QUANT_BLOCKS, 1))
-    packed = _fp32x2_to_fp4x2(
-        even_2d * inv_scale_col, odd_2d * inv_scale_col
-    )  # (N_BLOCKS, HALF_BLOCK) uint8
+    lo_nib = _e2m1_nibble(even_2d * inv_scale_col)  # (N_BLOCKS, HALF_BLOCK) uint8
+    hi_nib = _e2m1_nibble(odd_2d * inv_scale_col)
+    packed = lo_nib | (hi_nib << 4)
     packed_flat = tl.reshape(packed, (TOKEN_STRIDE,))
 
     tl.store(val_ptr + tl.arange(0, TOKEN_STRIDE), packed_flat)

vllm/v1/attention/ops/deepseek_v4_ops/fused_indexer_q.py

@@ -24,22 +24,36 @@ def _get_cos_sin(
 
 
 @triton.jit
-def _fp32x2_to_fp4x2(x_lo, x_hi):
-    # NOTE: $1 is high nibble, $2 is low nibble
-    return tl.inline_asm_elementwise(
-        """
-        {
-            .reg .b8 tmp;
-            cvt.rn.satfinite.e2m1x2.f32 tmp, $1, $2;
-            cvt.u32.u8 $0, tmp;
-        }
-        """,
-        constraints="=r,f,f",
-        args=[x_hi, x_lo],
-        dtype=tl.uint32,
-        is_pure=True,
-        pack=1,
-    ).to(tl.uint8)
+def _e2m1_nibble(x):
+    """Quantize fp32 x (already scale-divided) to E2M1 4-bit nibble in uint8.
+    Matches torch.bucketize with boundaries
+    [0.25, 0.75, 1.25, 1.75, 2.5, 3.5, 5.0] and right=False (each boundary
+    belongs to the lower bucket), plus sign bit."""
+    abs_x = tl.minimum(tl.abs(x), 6.0)
+    code = tl.where(
+        abs_x <= 0.25,
+        0.0,
+        tl.where(
+            abs_x <= 0.75,
+            1.0,
+            tl.where(
+                abs_x <= 1.25,
+                2.0,
+                tl.where(
+                    abs_x <= 1.75,
+                    3.0,
+                    tl.where(
+                        abs_x <= 2.5,
+                        4.0,
+                        tl.where(abs_x <= 3.5, 5.0, tl.where(abs_x <= 5.0, 6.0, 7.0)),
+                    ),
+                ),
+            ),
+        ),
+    )
+    code_u8 = code.to(tl.uint8)
+    sign = ((x < 0) & (code_u8 != 0)).to(tl.uint8)
+    return code_u8 | (sign << 3)
 
 
 @triton.jit
@@ -51,16 +65,17 @@ def _quantize_mxfp4_pair(x_lo, x_hi):
         - ue8m0  : scalar uint8    (block scale = 2^(ue8m0 - 127))
     """
     amax = tl.maximum(tl.max(tl.abs(x_lo)), tl.max(tl.abs(x_hi)))
-    # 6 * 2^-126 is from https://huggingface.co/deepseek-ai/DeepSeek-V4-Pro/blob/main/inference/kernel.py#L163
-    amax = tl.maximum(amax, 6.0 * (2**-126))
+    amax = tl.maximum(amax, 1e-4)
     # ue8m0 block scale: 2^ceil(log2(amax/6.0)).
-    log2_ratio = tl.math.ceil(tl.math.log2(amax * (1.0 / 6.0)))
+    log2_ratio = tl.math.ceil(tl.math.log2(amax / 6.0))
     log2_ratio = tl.minimum(tl.maximum(log2_ratio, -127.0), 127.0)
     scale = tl.math.exp2(log2_ratio)
     ue8m0 = (log2_ratio + 127.0).to(tl.uint8)
 
     inv_scale = 1.0 / scale
-    packed = _fp32x2_to_fp4x2(x_lo * inv_scale, x_hi * inv_scale)
+    lo_nib = _e2m1_nibble(x_lo * inv_scale)
+    hi_nib = _e2m1_nibble(x_hi * inv_scale)
+    packed = lo_nib | (hi_nib << 4)
     return packed, ue8m0