[sync] vLLM 2026-07-30 — 542a8fad6..e04a30a77

.sync/vllm-sha

@@ -1 +1 @@
-9a4e5f95390fc759ada001444637fe2e96a29ad9
+e04a30a77cb17d5a9753c53c51ee30bba887053c

aphrodite/_aiter_ops.py

@@ -1380,6 +1380,61 @@ def _triton_rotary_embedding_fake(
     return
 
 
+def _rocm_aiter_fp8_attn_impl(
+    q: torch.Tensor,
+    k: torch.Tensor,
+    v: torch.Tensor,
+    q_descale: torch.Tensor,
+    k_descale: torch.Tensor,
+    v_descale: torch.Tensor,
+    batch_size: int,
+    output_dtype: torch.dtype,
+    scale: float | None = None,
+    cu_seqlens: torch.Tensor | None = None,
+    max_seqlen: torch.Tensor | None = None,
+) -> torch.Tensor:
+    """Run AITER FP8 attention for fixed or packed inputs."""
+    from aiter import flash_attn_varlen_fp8_pertensor_func
+
+    q_len = q.size(1)
+    if cu_seqlens is None:
+        cu_seqlens = torch.arange(0, (batch_size + 1) * q_len, step=q_len, dtype=torch.int32, device=q.device)
+    max_seqlen_value = q_len if max_seqlen is None else max_seqlen.item()
+
+    q, k, v = (x.flatten(0, 1) for x in (q, k, v))
+    output = flash_attn_varlen_fp8_pertensor_func(
+        q,
+        k,
+        v,
+        q_descale=q_descale,
+        k_descale=k_descale,
+        v_descale=v_descale,
+        cu_seqlens_q=cu_seqlens,
+        cu_seqlens_k=cu_seqlens,
+        max_seqlen_q=max_seqlen_value,
+        max_seqlen_k=max_seqlen_value,
+        causal=False,
+        softmax_scale=scale,
+    )
+    return output.to(output_dtype).reshape(batch_size, q_len, *output.shape[1:])
+
+
+def _rocm_aiter_fp8_attn_fake(
+    q: torch.Tensor,
+    k: torch.Tensor,
+    v: torch.Tensor,
+    q_descale: torch.Tensor,
+    k_descale: torch.Tensor,
+    v_descale: torch.Tensor,
+    batch_size: int,
+    output_dtype: torch.dtype,
+    scale: float | None = None,
+    cu_seqlens: torch.Tensor | None = None,
+    max_seqlen: torch.Tensor | None = None,
+) -> torch.Tensor:
+    return torch.empty((*q.shape[:-1], v.shape[-1]), device=q.device, dtype=output_dtype)
+
+
 # Global flag to ensure ops are registered only once
 _OPS_REGISTERED = False
 
@@ -1929,6 +1984,14 @@ def register_ops_once() -> None:
                 dispatch_key=current_platform.dispatch_key,
             )
 
+            direct_register_custom_op(
+                op_name="aiter_fp8_attn_wrapper",
+                op_func=_rocm_aiter_fp8_attn_impl,
+                mutates_args=[],
+                fake_impl=_rocm_aiter_fp8_attn_fake,
+                dispatch_key=current_platform.dispatch_key,
+            )
+
             direct_register_custom_op(
                 op_name="rocm_aiter_gemm_a8wfp4",
                 op_func=_rocm_aiter_gemm_a8wfp4_impl,
@@ -2772,6 +2835,34 @@ def flash_attn_varlen_func(
             sink_ptr=sink_ptr,
         )
 
+    @staticmethod
+    def fp8_attn_wrapper(
+        q: torch.Tensor,
+        k: torch.Tensor,
+        v: torch.Tensor,
+        q_descale: torch.Tensor,
+        k_descale: torch.Tensor,
+        v_descale: torch.Tensor,
+        batch_size: int,
+        output_dtype: torch.dtype,
+        scale: float | None = None,
+        cu_seqlens: torch.Tensor | None = None,
+        max_seqlen: torch.Tensor | None = None,
+    ) -> torch.Tensor:
+        return torch.ops.aphrodite.aiter_fp8_attn_wrapper(
+            q,
+            k,
+            v,
+            q_descale,
+            k_descale,
+            v_descale,
+            batch_size,
+            output_dtype,
+            scale,
+            cu_seqlens,
+            max_seqlen,
+        )
+
     @staticmethod
     def pa_fwd_asm(
         Q: torch.Tensor,

aphrodite/compilation/passes/fusion/allreduce_rms_fusion.py

@@ -119,6 +119,11 @@ def _norm_input_weight_dtype_match(match: pm.Match) -> bool:
         8: 2,  # 2MB
         16: 64,  # 64MB (mnnvl multi-node)
     },
+    107: {
+        2: 64,  # 64MB
+        4: 64,  # 64MB
+        8: 2,  # 2MB
+    },
 }
 
 # Max size of the input tensor per world size per device capability
@@ -140,6 +145,11 @@ def _norm_input_weight_dtype_match(match: pm.Match) -> bool:
         4: 4,  # 4MB
         8: 2,  # 2MB
     },
+    107: {
+        2: 32,  # 32MB
+        4: 4,  # 4MB
+        8: 2,  # 2MB
+    },
 }
 
 MiB = 1024 * 1024

aphrodite/config/multimodal.py

@@ -163,9 +163,10 @@ class MultiModalConfig:
     `aphrodite.v1.attention.backends.registry.AttentionBackendEnum` (e.g. `FLASH_ATTN`)."""
     mm_encoder_attn_dtype: Literal["fp8"] | None = None
     """Optional dtype override for ViT encoder attention. Set to `"fp8"` to
-    enable FP8 quantization via the FlashInfer cuDNN backend. When set to
-    `"fp8"` without a scale file, dynamic scaling is used automatically.
-    See docs/features/quantization/fp8_vit_attn.md for details."""
+    enable FP8 quantization through FlashInfer cuDNN on NVIDIA GPUs or AITER
+    on supported AMD GPUs. When set to `"fp8"` without a scale file, dynamic
+    scaling is used automatically. See
+    docs/src/content/docs/features/fp8-vit-attention.md for details."""
     mm_encoder_fp8_scale_path: str | None = None
     """Path to a JSON file containing per-layer FP8 Q/K/V scales for ViT
     encoder attention. When provided (with `mm_encoder_attn_dtype="fp8"`),

aphrodite/distributed/device_communicators/all_reduce_utils.py

@@ -47,6 +47,12 @@
         6: 8 * MiB,  # 8 MB
         8: 4 * MiB,  # 4 MB
     },
+    "10.7": {  # sm_107 (Rubin): reuse 10.3 all-reduce thresholds
+        2: 4 * MiB,  # 4 MB
+        4: 4 * MiB,  # 4 MB
+        6: 8 * MiB,  # 8 MB
+        8: 4 * MiB,  # 4 MB
+    },
 }
 
 SYMM_MEM_ALL_REDUCE_MAX_SIZES = {
@@ -68,6 +74,12 @@
         6: 32 * MiB,  # 32 MB
         8: 64 * MiB,  # 64 MB
     },
+    "10.7": {  # sm_107 (Rubin): reuse 10.3 all-reduce thresholds
+        2: 4 * MiB,  # 4 MB
+        4: 32 * MiB,  # 32 MB
+        6: 32 * MiB,  # 32 MB
+        8: 64 * MiB,  # 64 MB
+    },
 }
 
 # NCCL symmetric memory allreduce configuration based on H100 and GB200 benchmarks.

aphrodite/distributed/device_communicators/shm_broadcast.py

@@ -890,7 +890,8 @@ def dequeue(
 
     @staticmethod
     def recv(socket: zmq.Socket, timeout: float | None) -> Any:
-        timeout_ms = None if timeout is None else int(timeout * 1000)
+        # Ensure non-negative timeout passed to zmq poll.
+        timeout_ms = None if timeout is None else max(0, int(timeout * 1000))
         if not socket.poll(timeout=timeout_ms):
             raise TimeoutError
         recv, *recv_oob = socket.recv_multipart(copy=False)

aphrodite/distributed/device_communicators/symm_mem.py

@@ -27,6 +27,7 @@ class SymmMemCommunicator:
         "9.0": [4, 6, 8],
         "10.0": [6, 8],
         "10.3": [6, 8],
+        "10.7": [6, 8],  # sm_107 (Rubin): reuse 10.3 thresholds
     }
 
     def __init__(

aphrodite/distributed/kv_events.py

@@ -265,6 +265,10 @@ def publish(self, events: EventBatch) -> None:
     def shutdown(self) -> None:
         """Shutdown the publisher."""
 
+    def get_publisher_config(self) -> KVEventsConfig | None:
+        """Return the publisher's resolved runtime configuration."""
+        return None
+
 
 class NullEventPublisher(EventPublisher):
     """No-op implementation (default when disabled)."""
@@ -326,6 +330,17 @@ def __init__(
 
         self._endpoint = self.offset_endpoint_port(endpoint, self._dp_rank)
         self._replay_endpoint = self.offset_endpoint_port(replay_endpoint, self._dp_rank)
+        assert self._endpoint is not None
+        self._publisher_config = KVEventsConfig(
+            enable_kv_cache_events=True,
+            publisher="zmq",
+            endpoint=self._endpoint,
+            replay_endpoint=self._replay_endpoint,
+            buffer_steps=buffer_steps,
+            hwm=hwm,
+            max_queue_size=max_queue_size,
+            topic=topic,
+        )
         self._hwm = hwm
         self._socket_setup()
 
@@ -340,6 +355,9 @@ def __init__(
         self._thread = threading.Thread(target=self._publisher_thread, daemon=True, name="zmq-publisher")
         self._thread.start()
 
+    def get_publisher_config(self) -> KVEventsConfig:
+        return self._publisher_config
+
     def publish(self, events: EventBatch) -> None:
         if not self._running:
             raise RuntimeError("Publisher is closed")

aphrodite/distributed/kv_transfer/kv_connector/v1/base.py

@@ -179,6 +179,18 @@ def prefer_cross_layer_blocks(self) -> bool:
         """
         return False
 
+    @property
+    def requires_kv_delivery(self) -> bool:
+        """Whether this connector hands off KV that must be reliably delivered.
+
+        If True, a request preempted while its hand-off is still pending is
+        recomputed rather than allowed to finish and hand off blocks that the
+        preemption already freed. Defaults to the producer role, since only a
+        producer hands KV off when a request completes. Best-effort caches
+        return False, as a dropped save is just a future cache miss.
+        """
+        return self._kv_transfer_config.is_kv_producer
+
     def __init__(
         self,
         aphrodite_config: "AphroditeConfig",

aphrodite/distributed/kv_transfer/kv_connector/v1/multi_connector.py

@@ -196,6 +196,10 @@ def prefer_cross_layer_blocks(self) -> bool:
             return False
         return all(c.prefer_cross_layer_blocks for c in self._connectors)
 
+    @property
+    def requires_kv_delivery(self) -> bool:
+        return any(c.requires_kv_delivery for c in self._connectors)
+
     @classmethod
     def _get_connector_classes_and_configs(
         cls, aphrodite_config: "AphroditeConfig"

aphrodite/distributed/kv_transfer/kv_connector/v1/nixl/base_scheduler.py

@@ -123,6 +123,16 @@ def __init__(
             cdiv(n_tokens, block_size) + 1 if n_tokens else 0 for n_tokens, block_size in sw_sizes_tokens
         ]
 
+        # Trailing scratch slots that mamba managers co-allocate per request
+        # for speculative decoding; None for non-SSM groups.
+        self._ssm_spec_blocks = [
+            g.kv_cache_spec.num_speculative_blocks if isinstance(g.kv_cache_spec, MambaSpec) else None
+            for g in kv_cache_config.kv_cache_groups
+        ]
+        # Only "all" mode keeps a state per block position; the other modes
+        # keep a single running state in the last non-speculative slot.
+        self._ssm_state_slots_are_positional = aphrodite_config.cache_config.mamba_cache_mode == "all"
+
         # Threshold to decide whether to compute kv cache locally
         # or pull from a remote node: minimum number of remote
         # tokens to amortize the xfer latencies
@@ -194,13 +204,23 @@ def _stop_heartbeat(self, req_id: ReqId) -> None:
                     # Clean up empty engines so we don't leak a key when remote dies.
                     del self._heartbeat_by_engine[engine_id]
 
-    def get_sw_clipped_blocks(self, block_ids: BlockIds) -> BlockIds:
-        """
-        Clip the number of blocks to the sliding window size for each kv cache group
-        that employs SWA.
-        This is necessary because the KV Cache manager initially allocates blocks for
-        the entire sequence length, and successively cleans up blocks that are outside
-        the window prior to the `request_finished_all_groups` hook.
+    def get_exchange_clipped_blocks(self, block_ids: BlockIds, clip_ssm: bool = True) -> BlockIds:
+        """Clip a request's block lists down to the transferable blocks.
+
+        Sliding-window groups keep only the in-window tail: the KV cache
+        manager allocates blocks for the entire sequence length and cleans up
+        out-of-window blocks only prior to the `request_finished_all_groups`
+        hook.
+
+        SSM groups keep only their state-bearing slots: the trailing
+        speculative scratch slots always go, and in single-state cache modes
+        so does everything before the running state (null placeholders and
+        the previous step's superseded state). "all" mode keeps its remaining
+        slots, which the worker pairs position-wise.
+
+        Use this at every block-id exchange point. Pass ``clip_ssm=False``
+        for per-step partial lists (host-buffer save), where the SSM strip
+        does not apply.
         """
         if len(block_ids) == 0 or not self._is_hma_required:
             # No blocks to clip eg Full prefix cache hit or not a hybrid model.
@@ -209,13 +229,18 @@ def get_sw_clipped_blocks(self, block_ids: BlockIds) -> BlockIds:
         # because offloading connectors might want to receive the whole sequence even
         # for SWA groups. We will abstract this logic once the interface is more stable
         assert len(block_ids) == len(self.blocks_per_sw), "Number of KV cache groups must match"
-        # For non-SWA groups, blocks_per_sw is 0 so we return all block_ids unchanged
-        return tuple(
-            [
-                blocks[-self.blocks_per_sw[i] :] if self.blocks_per_sw[i] > 0 else blocks
-                for i, blocks in enumerate(block_ids)
-            ]
-        )
+        clipped = []
+        for i, blocks in enumerate(block_ids):
+            if n_sw := self.blocks_per_sw[i]:
+                blocks = blocks[-n_sw:]
+            elif clip_ssm and blocks and (n_spec_blocks := self._ssm_spec_blocks[i]) is not None:
+                if n_spec := min(n_spec_blocks, len(blocks) - 1):
+                    blocks = blocks[:-n_spec]
+                if not self._ssm_state_slots_are_positional:
+                    # Never empty: downstream reads that as a full prefix hit.
+                    blocks = blocks[-1:]
+            clipped.append(blocks)
+        return tuple(clipped)
 
     def set_xfer_handshake_metadata(self, metadata: dict[int, KVConnectorHandshakeMetadata]) -> None:
         self.set_xfer_handshake_metadata_pp_aware({(0, tp_rank): meta for tp_rank, meta in metadata.items()})
@@ -357,7 +382,7 @@ def _build_save_meta(
             req = req_to_save
 
             assert req.kv_transfer_params is not None
-            clipped_block_id_groups = self.get_sw_clipped_blocks(new_block_id_groups)
+            clipped_block_id_groups = self.get_exchange_clipped_blocks(new_block_id_groups, clip_ssm=False)
             meta.add_new_req_to_save(
                 request_id=req_id,
                 local_block_ids=clipped_block_id_groups,
@@ -393,6 +418,10 @@ def build_connector_meta(
             self._build_save_meta(meta, scheduler_output)
 
         meta.reqs_to_send = self._reqs_need_send
+        # Clock reference for reqs_to_send: deadlines above are in this
+        # process's perf_counter domain; workers (possibly on other nodes,
+        # where perf_counter has a different epoch) rebase against this.
+        meta.scheduler_clock = time.perf_counter()
         meta.reqs_in_batch = self._reqs_in_batch
         meta.reqs_not_processed = self._reqs_not_processed
 

aphrodite/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py

@@ -2192,13 +2192,24 @@ def _apply_prefix_caching(
             for i, remote_group in enumerate(remote_block_ids):
                 num_local_blocks = len(local_block_ids[i])
                 num_remote_blocks = len(remote_group)
-                if _is_ssm_spec(self._group_spec_types[i]) and num_local_blocks < num_remote_blocks:
-                    # NOTE (NickLucche): With prefix caching on SSM, (remote) blocks
-                    # prior to the last one are placeholders (null blocks). Mind that
-                    # this doesn't really impact transfer, as we only still care about
-                    # the last "block", the full in-place state.
-                    assert num_local_blocks == 1, "SSM can only have one local block"
-                    remote_block_ids[i] = remote_group[-num_local_blocks:]
+                if _is_ssm_spec(self._group_spec_types[i]):
+                    if num_local_blocks == num_remote_blocks:
+                        continue
+                    # Only state-bearing slots reach here, single-state modes
+                    # just one (see get_exchange_clipped_blocks), so differing
+                    # counts mean position-indexed "all"-mode lists. A longer
+                    # remote list carries earlier positions the local side
+                    # already has (prefix hit) -> read its tail; a longer local
+                    # list holds the position D recomputes itself, which gets
+                    # no remote state.
+                    assert num_local_blocks - num_remote_blocks <= 1, (
+                        f"Group {i}: unpairable SSM state slots, local={num_local_blocks} remote={num_remote_blocks}"
+                    )
+                    num_blocks = min(num_local_blocks, num_remote_blocks)
+                    if num_local_blocks < num_remote_blocks:
+                        remote_block_ids[i] = remote_group[-num_blocks:]
+                    else:
+                        local_block_ids[i] = local_block_ids[i][:num_blocks]
                 elif (
                     self._physical_blocks_per_logical_kv_block == remote_physical_per_logical
                     and num_local_blocks < num_remote_blocks