feat: support Qwen3.8 Flash Next

docs/qwen38_adaptation.rst

@@ -0,0 +1,48 @@
+Qwen3.8-Flash-Next
+==================
+
+Qwen3.8-Flash-Next checkpoints use the ``qwen4_exp`` model type. The model
+implementation combines gated residuals, n-gram embeddings, QSA and GDN
+attention, with dedicated multimodal, MTP and cache-state integration.
+Reusable operations live in ``lightllm/common/gated_residual``,
+``lightllm/common/ngram_embedding`` and ``lightllm/common/qsa``.
+
+CPU embedding offload is disabled by default. The optional QSA TopK selector
+is enabled by the model only for its qualified H100/Torch/Triton geometry;
+other configurations use Torch. Shared-prefix admission is limited to the
+model's single-node, DP1 normal serving mode with one-page state blocks.
+
+Validation
+----------
+
+Historical H100 measurements and their configuration are recorded in
+``qwen38_h100_performance.rst``. They cover TP8 BF16, FP32 SSM states,
+fixed MTP2 and one-page linear-attention state blocks. These results do not
+qualify PD/cross-DP serving or other cache block configurations.
+
+The four-page-block pressure-recovery configuration has unresolved text
+differences. Low-load latency still trails SGLang in some metrics. Archived
+HC and sampling experiments are not part of the runtime implementation.
+
+Rebase to upstream, 2026-11-03
+------------------------------
+
+The implementation now uses AttentionStateManager. QSA keys and positions are
+owned by the model's GQA State and participate in token copies and CPU/PD/DP
+transfer layouts. Lexical histories use a separate bounded State for reset,
+prefix snapshots, CPU/PD/DP transfers and accepted MTP candidate commits.
+The removed request/cache manager classes are not restored.
+
+Model and draft registration use the upstream lazy registry. HTTP slot waiting
+preserves atomic group allocation and current PD timeouts. The GDN MTP path uses
+the upstream recurrent kernel and its explicit run configuration.
+
+The historical H100 results above predate this interface migration. They do not
+validate the rebased runtime. GPU generation, complete PD transfer and performance
+must be rechecked before deploying a new image of this revision.
+
+CPU validation after migration: 1022 checks passed and 930 GPU cases were
+skipped. The DeepSeek V4 integration module could not be collected because
+FlashMLA is absent. Two upstream State identity cases require CUDA-backed
+pinned memory; both failures reproduce on unmodified upstream/main with CUDA
+hidden. No GPU serving, PD end-to-end or performance result is claimed here.

docs/qwen38_h100_performance.rst

@@ -0,0 +1,76 @@
+Qwen3.8-Flash-Next H100 性能与精度
+=================================
+
+当前结论：4× 的两轮吞吐、TTFT P95、完整 HTTP P95 和末次可见文本 P95 均优于冻结基线。
+低负载下仍有延迟差距，尚未完成“超过 vLLM 和 SGLang”的整体目标。
+
+测试口径
+--------
+
+* 8 × NVIDIA H100 80GB HBM3，TP8，BF16，SSM 状态 FP32；MTP 固定 2。
+* 模型版本 de4b8e4d43b917e7706784d8bb445c9af86a3540；4096 上下文、最多 8 个并发请求、512-token prefill chunk、16384 KV 容量。
+* 当前已验证的线性注意力状态页为 512 tokens，每块一页。
+* AIPerf 0.9 统一回放公开 Mooncake 轨迹中的 256 个请求，倍率为 1×、2×、4×。
+* 公开轨迹提供匿名长度、时间戳和 hash；请求正文由 AIPerf 合成，并非原始私有正文。
+* 每个 LightLLM 样本使用全新服务；vLLM、SGLang 只使用此前冻结的原始结果，未重新推理。
+* 原生 Torch 2.9.1+cu128 / Triton 3.5.1；vLLM 0.31；SGLang 0.5.22。
+* 1×/2× 吞吐受请求到达速率限制，不能当作最大处理容量；保留所有重复样本。
+
+已完成的性能样本
+----------------
+
+.. csv-table:: 延迟单位 ms；吞吐单位成功输出 tok/s
+   :header: "倍率", "框架", "样本", "吞吐 ↑", "TTFT P95 ↓", "完整 HTTP P95 ↓", "末次文本 P95 ↓", "空文本错误/256"
+
+   "1×", "LightLLM", "1", "276.14", "516.09", "3832.07", "3615.36", "2"
+   "1×", "LightLLM", "2", "276.16", "515.16", "3818.23", "3622.51", "2"
+   "1×", "SGLang", "冻结", "276.99", "610.40", "3864.63", "3329.87", "2"
+   "1×", "vLLM", "冻结", "277.72", "804.93", "4255.93", "3548.29", "1"
+   "2×", "LightLLM", "1", "546.33", "475.34", "4466.54", "4187.74", "2"
+   "2×", "SGLang", "冻结", "545.80", "600.51", "5302.52", "4178.47", "2"
+   "2×", "vLLM", "冻结", "513.20", "2933.67", "8695.77", "7227.73", "2"
+   "4×", "LightLLM", "1", "887.26", "376.17", "5794.25", "5188.47", "2"
+   "4×", "LightLLM", "2", "886.14", "390.13", "5770.75", "5151.58", "2"
+   "4×", "SGLang", "冻结", "835.28", "864.93", "5924.62", "5204.95", "2"
+   "4×", "vLLM", "冻结", "523.95", "3367.76", "9085.75", "7930.38", "1"
+
+完整 HTTP 统计包含全部 256 个请求；TTFT 和末次文本仅包含有可见文本的请求。
+空文本仍计为错误，其输出 token 不计入成功吞吐。AIPerf 导出的 request_latency
+在这些记录中对应末次可见文本，不能替代完整 HTTP/DONE 边界。
+
+精度与兼容性
+------------
+
+* 数学题正确 1272/1319，准确率 96.44%。
+* 全部 1319 个数学样例的生成 token 序列与原实现一致；分别保留 19 例截断、19 例无有效最终答案。
+* 480 个冷缓存请求和 48 个调度、缓存、EOS 请求核对通过；2788 个训练模型中间张量完全一致。
+* H100 共享组件/默认行为回归 134 项通过；本地 CPU 兼容性检查 124 项通过，零失败、错误、跳过。
+* 已完成的 5 轮回放共 1280 个输出文本与原实现一致，空文本错误单独保留。
+* TopK 为独立组件，共享选项默认关闭，Qwen4 在工厂启用，select 接口保持兼容。
+
+剩余工作与限制
+--------------
+
+* 1× 首轮末次文本 P95 为 3615.36 ms，高于 SGLang 3329.87 ms；中位请求延迟也仍落后。
+* 2× 末次文本 P95 为 4187.74 ms，比 SGLang 4178.47 ms 高约 9.28 ms。
+* 历史四页块压力恢复的文本差异仍未解决，不能将当前一页块精度结论推广到该配置。
+* 五轮回放及低负载剖析已完成；按用户要求收尾，优化目标已暂停，未标记为全面达成。
+
+原始结果与证据
+--------------
+
+* 五轮独立汇总：``artifacts/h100/261019/audits/topk-short-five-sample-summary.json``。
+* 性能 CSV：``artifacts/h100/261019/audits/topk-short-performance-current.csv``。
+* 完整精度：``artifacts/h100/261019/audits/topk-short-full-math-token-parity.json``。
+* 源码与本地验证：``artifacts/h100/261019/audits/topk-short-local-adoption.json``。
+* 重复回放状态：``artifacts/h100/261019/tuning/topk-short-canonical-candidate/progress.json``。
+
+收尾状态
+--------
+
+* 最终保留完整精度和五轮回放验证通过的 TopK 版本；H100 隔离源码已恢复到同一版本。
+* HC 小批量上投影候选已归档：30 项 CPU、203 项 GPU 检查通过，2788 个模型中间张量一致。
+  完整模型轨迹实际覆盖 8 行 BD8；其余选择形状只有组件检查。HC 尚未完成服务精度与公开回放验证，未纳入最终实现或上述性能结果。
+* 首次 HC 模型轨迹因测试插桩作用域断言失败而回滚；修复插桩并验证后重跑通过，失败记录一并保留。
+* H100 恢复后核对 949 项源码和 42 项测试/基准文件哈希；GPU 无遗留计算进程。未启动新的推理测试。
+* 收尾记录：``artifacts/h100/261019/audits/user-wrapup-qualified-topk-restored.json``。

lightllm/common/basemodel/attention/linear/gdn.py

@@ -37,7 +37,6 @@ def get_prefill_kernel(self):
         pass
 
     def _init_linear_layer_metadata(self, network_config, tp_world_size):
-
         self.mtp_step = get_env_start_args().mtp_step
 
         # Linear attention specific dimensions
@@ -109,7 +108,6 @@ def create_att_decode_state(self, infer_state: "InferStateInfo") -> "LinearAttDe
 
 @dataclasses.dataclass
 class LinearAttPrefillAttState(BasePrefillAttState):
-
     b_conv_buffer_idx: torch.Tensor = None
     b_ssm_buffer_idx: torch.Tensor = None
 
@@ -207,11 +205,11 @@ def _gdn_prefill_kernel(
 
 @dataclasses.dataclass
 class LinearAttDecodeAttState(BaseDecodeAttState):
-
     b_conv_buffer_idx: torch.Tensor = None
     b_ssm_buffer_idx: torch.Tensor = None
     b1_mtp_cu_q_seq_len: torch.Tensor = None
     b_num_accepted_tokens: torch.Tensor = None
+    b1_mtp_cu_q_seq_len_int64: torch.Tensor = None
 
     def init_state(self):
         draft_step = self.backend.model.mtp_manager.get_decode_draft_step(self.backend.model.is_mtp_draft_model)
@@ -232,6 +230,7 @@ def init_state(self):
             self._init_dynamic_mtp_decode_state(draft_step + 1, req_to_mtp_state_index=state.mtp_state_index)
         else:
             self._init_fixed_mtp_decode_state(draft_step, req_to_mtp_state_index=state.mtp_state_index)
+        self.b1_mtp_cu_q_seq_len_int64 = self.b1_mtp_cu_q_seq_len.to(torch.int64)
 
     def _init_normal_decode_state(self):
         self.b_conv_buffer_idx = self.infer_state.b_req_idx
@@ -418,13 +417,18 @@ def _gdn_mtp_kernel(
             k=key,
             v=value,
             initial_state=ssm_states,
-            cu_seqlens=cu_seqlens_q.to(torch.long),
+            cu_seqlens=self.b1_mtp_cu_q_seq_len_int64,
             ssm_state_indices=self.b_ssm_buffer_idx,
             ssm_state_write_indices=self.b_ssm_buffer_idx,
             num_accepted_tokens=self.b_num_accepted_tokens,
             A_log=layer_weight.linear_A_log.weight,
             dt_bias=layer_weight.linear_dt_bias.weight,
             a_raw=a,
             b_raw=b,
+            run_config=(
+                {"BV": 32, "num_warps": 4, "num_stages": 1}
+                if getattr(backend.model, "linear_att_mtp_match_decode", False)
+                else None
+            ),
         )
         return core_attn_out

lightllm/common/basemodel/layer_weights/meta_weights/pinned_embedding_weight.py

@@ -0,0 +1,66 @@
+import torch
+
+from .row_sharded_embedding_weight import RowShardedEmbeddingWeight
+from lightllm.utils.log_utils import init_logger
+
+logger = init_logger(__name__)
+
+
+class PinnedRowShardedEmbeddingWeight(RowShardedEmbeddingWeight):
+    def __init__(
+        self, dim, vocab_size, weight_name, data_type, shard_count, *, cpu_allocator, tp_rank=None, tp_world_size=None
+    ):
+        if not callable(cpu_allocator):
+            raise ValueError("Pinned embeddings require a CPU allocation policy")
+        self.cpu_allocator = cpu_allocator
+        super().__init__(dim, vocab_size, weight_name, data_type, shard_count, tp_rank, tp_world_size)
+
+    def _create_weight(self):
+        shape = (self.tp_vocab_end_id - self.tp_vocab_start_id, self.dim)
+        with torch.cuda.device(self.device_id_):
+            self.weight = self.cpu_allocator(shape, dtype=self.data_type_, device="cpu", pin_memory=True)
+        if (
+            self.weight.device.type != "cpu"
+            or self.weight.dtype != self.data_type_
+            or tuple(self.weight.shape) != shape
+            or not self.weight.is_contiguous()
+            or (self.weight.numel() and not self.weight.is_pinned())
+        ):
+            raise ValueError(
+                "Embedding allocator must return contiguous pinned CPU storage with the requested shape/dtype"
+            )
+        self.weight.load_ok = False
+        logger.info(
+            f"Pinned embedding table: rank={self.tp_rank_} rows={shape[0]} dim={self.dim} "
+            f"bytes={self.weight.numel() * self.weight.element_size()} dtype={self.weight.dtype} residency=cpu"
+        )
+
+    def load_hf_weights(self, weights):
+        names = {self.weight_name} | {f"{self._shard_prefix}.shard_{index}.weight" for index in self._required_shards}
+        host_weights = {name: value.cpu() if name in names else value for name, value in weights.items()}
+        super().load_hf_weights(host_weights)
+
+    def _cuda_forward(self, input_ids, out=None, alloc_func=torch.empty):
+        if (
+            not input_ids.is_cuda
+            or input_ids.ndim != 1
+            or not input_ids.is_contiguous()
+            or input_ids.dtype not in (torch.int32, torch.int64)
+        ):
+            raise ValueError("Pinned embedding lookup requires a CUDA vector of integer IDs")
+        shape = (input_ids.shape[0], self.dim)
+        if out is None:
+            out = alloc_func(shape, dtype=self.weight.dtype, device=input_ids.device)
+        if (
+            tuple(out.shape) != shape
+            or out.dtype != self.weight.dtype
+            or out.device != input_ids.device
+            or out.stride(1) != 1
+        ):
+            raise ValueError("Pinned embedding output must match CUDA IDs, table dtype and lookup shape")
+        if not input_ids.numel():
+            return out
+        return super()._triton_forward(input_ids, out=out, alloc_func=alloc_func)
+
+    def _native_forward(self, *args, **kwargs):
+        raise ValueError("Pinned host embedding lookup requires the CUDA Triton implementation")

lightllm/common/basemodel/layer_weights/meta_weights/row_sharded_embedding_weight.py

@@ -0,0 +1,64 @@
+from typing import Dict, Optional
+
+import torch
+
+from .embedding_weight import EmbeddingWeight
+
+
+class RowShardedEmbeddingWeight(EmbeddingWeight):
+    def __init__(
+        self,
+        dim: int,
+        vocab_size: int,
+        weight_name: str,
+        data_type: torch.dtype,
+        shard_count: int,
+        tp_rank: Optional[int] = None,
+        tp_world_size: Optional[int] = None,
+    ):
+        if shard_count < 1 or vocab_size < 1 or vocab_size % shard_count:
+            raise ValueError("Row-sharded embeddings require a positive vocabulary divisible by shard count")
+        if not weight_name.endswith(".weight"):
+            raise ValueError("Embedding weight name must end with .weight")
+        self.shard_count = shard_count
+        self.shard_rows = vocab_size // shard_count
+        self._shard_prefix = weight_name[: -len(".weight")]
+        self._loaded_shards = set()
+        self._full_loaded = False
+        super().__init__(dim, vocab_size, weight_name, data_type, tp_rank, tp_world_size)
+        self._required_shards = {
+            index
+            for index in range(shard_count)
+            if max(index * self.shard_rows, self.tp_vocab_start_id)
+            < min((index + 1) * self.shard_rows, self.tp_vocab_end_id)
+        }
+        self.weight.load_ok = not self._required_shards
+
+    def load_hf_weights(self, weights: Dict[str, torch.Tensor]):
+        shard_keys = [(index, f"{self._shard_prefix}.shard_{index}.weight") for index in range(self.shard_count)]
+        present = [(index, key) for index, key in shard_keys if key in weights]
+        if self.weight_name in weights:
+            if present or self._loaded_shards:
+                raise ValueError("Cannot mix full and row-sharded embedding checkpoint tensors")
+            if weights[self.weight_name].shape != (self.vocab_size, self.dim):
+                raise ValueError("Full embedding checkpoint dimensions do not match")
+            super().load_hf_weights(weights)
+            self._full_loaded = True
+            return
+        if self._full_loaded:
+            if present:
+                raise ValueError("Cannot mix full and row-sharded embedding checkpoint tensors")
+            return
+        for index, key in present:
+            source = weights[key]
+            if source.shape != (self.shard_rows, self.dim):
+                raise ValueError(f"Unexpected dimensions for embedding shard {index}: {tuple(source.shape)}")
+            shard_start = index * self.shard_rows
+            begin = max(shard_start, self.tp_vocab_start_id)
+            end = min(shard_start + self.shard_rows, self.tp_vocab_end_id)
+            if begin < end:
+                self.weight[begin - self.tp_vocab_start_id : end - self.tp_vocab_start_id].copy_(
+                    source[begin - shard_start : end - shard_start].to(self.data_type_)
+                )
+                self._loaded_shards.add(index)
+        self.weight.load_ok = self._loaded_shards >= self._required_shards

lightllm/common/basemodel/multimodal_tokenizer.py

@@ -40,6 +40,9 @@ def __getattr__(self, name):
             return obj_dict[name]
         return getattr(self.tokenizer, name)
 
+    def count_prompt_tokens(self, text_tokens, image_tokens, image_count, audio_tokens, audio_count):
+        return text_tokens + image_tokens + image_count + audio_tokens + audio_count
+
     @abstractmethod
     def encode(self, prompt, multimodal_params: MultimodalParams = None, **kwargs):
         pass

lightllm/common/basemodel/tp_reduce.py

@@ -0,0 +1,58 @@
+import torch
+import triton
+import triton.language as tl
+
+from lightllm.distributed import all_gather_into_tensor
+
+
+@triton.jit
+def _sum_shards(
+    X,
+    Y,
+    N: tl.constexpr,
+    START: tl.constexpr,
+    SIZE: tl.constexpr,
+    WORLD: tl.constexpr,
+    BLOCK: tl.constexpr,
+):
+    index = tl.program_id(0) * BLOCK + tl.arange(0, BLOCK)
+    accumulator = tl.full((BLOCK,), 0, tl.float64)
+    for rank in range(WORLD):
+        value = tl.load(X + rank * N + START + index, index < SIZE, other=0)
+        accumulator += value.to(tl.float64)
+    tl.store(Y + index, accumulator, index < SIZE)
+
+
+def ordered_tp_sum(input: torch.Tensor, group, world_size: int, *, scatter_rank=None) -> torch.Tensor:
+    if world_size == 1:
+        return input
+    if scatter_rank is not None:
+        if not 0 <= scatter_rank < world_size or input.shape[0] % world_size:
+            raise ValueError("Sequence-parallel input must divide evenly across valid ranks")
+        shape = (input.shape[0] // world_size, *input.shape[1:])
+        output = torch.empty(shape, device=input.device, dtype=input.dtype)
+        start = scatter_rank * output.numel()
+    else:
+        output = input if input.is_contiguous() else torch.empty(input.shape, device=input.device, dtype=input.dtype)
+        start = 0
+    if input.numel():
+        source = input.contiguous()
+        symmetric = getattr(group, "symm_mem_reduce", None) if input.is_cuda else None
+        gathered = symmetric.ordered_all_gather(source) if symmetric is not None else None
+        if gathered is None:
+            gathered = torch.empty((world_size, *input.shape), device=input.device, dtype=input.dtype)
+            all_gather_into_tensor(gathered.view(-1, *input.shape[1:]), source, group=group)
+        if input.is_cuda:
+            _sum_shards[(triton.cdiv(output.numel(), 256),)](
+                gathered, output, input.numel(), start, output.numel(), world_size, 256, enable_fp_fusion=False
+            )
+        else:
+            accumulator = torch.zeros(output.numel(), dtype=torch.float64, device=input.device)
+            shards = gathered.reshape(world_size, -1)
+            for rank in range(world_size):
+                accumulator.add_(shards[rank, start : start + output.numel()])
+            output.copy_(accumulator.reshape(output.shape))
+    if scatter_rank is None and output is not input:
+        input.copy_(output)
+        return input
+    return output