Download code/offload/dspark/worker/exl3/kernels.py from ProprietaryLegal/dsv41-sampled-drafting: direct link, hf CLI and curl.
- Browser
- Download file 18.5 kB
-
https://huggingface.co/ProprietaryLegal/dsv41-sampled-drafting/resolve/main/code/offload/dspark/worker/exl3/kernels.py
- Command line
-
hf download hf://ProprietaryLegal/dsv41-sampled-drafting/code/offload/dspark/worker/exl3/kernels.py
-
curl -L -o kernels.py https://huggingface.co/ProprietaryLegal/dsv41-sampled-drafting/resolve/main/code/offload/dspark/worker/exl3/kernels.py
18.5 kB
| # SPDX-License-Identifier: Apache-2.0 | |
| """Thin wrappers over the exllamav3 v1.6.0 CUDA extension (exllamav3_ext, MIT, turboderp-org). | |
| Nothing from exllamav3 is copied here: we call its pybind entry points (exllamav3_ext/bindings.cpp) | |
| with tensors laid out the way its kernels expect. Citations are exllamav3 v1.6.0 @ 151539c7 paths. | |
| Entry points used: | |
| exl3_moe_coop bindings.cpp:292 fused decode MoE, 2-3 launches (rot / gate+up / down), | |
| bsz * top_k <= 256 slots, no grid barrier, deterministic, | |
| no host sync (quant/exl3_moe_coop.cu:315-364, | |
| quant/exl3_moe_coop_kernel.cuh:1-21) | |
| exl3_moe bindings.cpp:290 grouped MoE over expert-sorted rows, 16/32/64-row tiles, | |
| experts with more than R rows are skipped (quant/exl3_moe.cu:189-572) | |
| exl3_moe_max_concurrency :289 groups the exl3_moe launch can keep co-resident | |
| reconstruct bindings.cpp:170 trellis -> fp16 weight in the rotated basis | |
| had_r_128 bindings.cpp:175 128-block Hadamard with optional pre (suh) / post (svh) scale | |
| hgemm bindings.cpp:209 fp16 GEMM, fp16 or fp32 output | |
| Per-expert bitrates: every launch takes ONE (K_gate == K_up, K_down) pair | |
| (exl3_moe.cu:367-378, exl3_moe_coop.cu:193-194, modules/multilinear.py:25-26), so a layer with | |
| mixed K runs one launch per K group. Experts are stored K-sorted (plan.LayerPlan.perm) so a group | |
| is a contiguous physical range [lo, hi) and the coop kernel's expert-range filter | |
| (MoeCoopParams.min_expert/max_expert, exl3_moe_coop_kernel.cuh:211-223) masks the other groups. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| from dataclasses import dataclass | |
| import torch | |
| MOE_ACT_SILU = 0 # exl3_moe_common.cuh:7 | |
| MOE_COOP_ACT_SILU = 0 # exl3_moe_coop.cuh:13 | |
| COOP_MAX_SLOTS = 256 # exl3_moe_coop_kernel.cuh:46, exl3_moe_coop.cu:226 | |
| MAX_K_GROUPS = 8 # scratch sets for concurrent K groups (a layer has one group per distinct (k_gu, k_d)) | |
| MTILE_T1, MTILE_T2 = 16, 32 # row-tile tiers, as exllamav3 uses them (block_sparse_mlp.py:37-38) | |
| _EXT = None | |
| def ext(): | |
| """The compiled extension. Prefer the prebuilt top-level module the exllamav3 wheel installs | |
| (setup.py: extension_name = "exllamav3_ext"); fall back to exllamav3's JIT loader (needs nvcc).""" | |
| global _EXT | |
| if _EXT is None: | |
| try: | |
| import exllamav3_ext as m # type: ignore[import-not-found] | |
| except ImportError: | |
| try: | |
| from exllamav3.ext import exllamav3_ext as m # type: ignore[import-not-found,no-redef] | |
| except Exception as e: # noqa: BLE001 | |
| raise ImportError( | |
| "vllm_exl3_moe needs exllamav3 v1.6.0's compiled extension (exllamav3_ext). Build the " | |
| "wheel for sm_121 (TORCH_CUDA_ARCH_LIST=12.1) and install it with --no-deps." | |
| ) from e | |
| missing = [n for n in ("exl3_moe", "exl3_moe_coop", "exl3_moe_max_concurrency", "reconstruct", | |
| "had_r_128", "hgemm") if not hasattr(m, n)] | |
| if missing: | |
| raise ImportError(f"exllamav3_ext lacks {missing}: v1.6.0 or newer is required") | |
| _EXT = m | |
| return _EXT | |
| def have_ext() -> bool: | |
| try: | |
| ext() | |
| return True | |
| except ImportError: | |
| return False | |
| def coop_ctr_len(slots_max: int, rows_max: int, inter: int, out_dim: int) -> int: | |
| """int32 scratch of the coop kernels: completion counters + run table | |
| (exl3_moe_coop.cuh:154-158, exl3_moe_coop_ctr_len).""" | |
| return slots_max * (inter // 128) + rows_max * (out_dim // 128) + 2 + (slots_max + 1) + slots_max | |
| # --------------------------------------------------------------------------------------------- | |
| # Pointer tables | |
| class ProjTables: | |
| trellis: torch.Tensor # int64 [n] device: data_ptr of each expert's trellis | |
| suh: torch.Tensor # int64 [n] | |
| svh: torch.Tensor # int64 [n] | |
| def slice(self, lo: int, hi: int) -> "ProjTables": | |
| # 1-D slices of contiguous tensors stay contiguous (the kernels check it) | |
| return ProjTables(self.trellis[lo:hi], self.suh[lo:hi], self.svh[lo:hi]) | |
| class GroupTables: | |
| k_gu: float | |
| k_d: float | |
| lo: int | |
| hi: int | |
| g: ProjTables # gate (w1) | |
| u: ProjTables # up (w3) | |
| d: ProjTables # down (w2) | |
| def size(self) -> int: | |
| return self.hi - self.lo | |
| # --------------------------------------------------------------------------------------------- | |
| # Scratch, shared by every layer of the process (all routed layers have the same shapes and run | |
| # in stream order; the coop counters reset each other between calls, exl3_moe_coop_kernel.cuh:695-698, | |
| # 833-835, so one zeroed buffer serves every layer) | |
| class CoopScratch: | |
| rows_max: int | |
| slots_max: int | |
| had_g: torch.Tensor # fp16 [slots_max, H] | |
| had_u: torch.Tensor # fp16 [slots_max, H] | |
| gu_g: torch.Tensor # fp16 [slots_max, I] | |
| gu_u: torch.Tensor # fp16 [slots_max, I] | |
| act_out: torch.Tensor # fp16 [slots_max, I] | |
| d_out: torch.Tensor # fp32 [slots_max, H] | |
| ctr: torch.Tensor # int32 [coop_ctr_len], zero once | |
| out: tuple[torch.Tensor, torch.Tensor] # fp32 [rows_max, H] ping-pong (K-group chaining) | |
| class FusedScratch: | |
| rows: int # R: rows per expert the fused exl3_moe kernel takes | |
| concurrency: int # C | |
| tsg: torch.Tensor # fp16 [C, R, H] | |
| tsu: torch.Tensor | |
| tig: torch.Tensor # fp16 [C, R, I] | |
| tiu: torch.Tensor | |
| _COOP: dict[tuple, CoopScratch] = {} | |
| _FUSED: dict[tuple, FusedScratch] = {} | |
| _WBUF: dict[tuple, torch.Tensor] = {} | |
| def get_coop_scratch(device: torch.device, rows_max: int, top_k: int, hidden: int, inter: int) -> CoopScratch: | |
| key = (str(device), rows_max, top_k, hidden, inter) | |
| sc = _COOP.get(key) | |
| if sc is None: | |
| sc = _new_coop_scratch(device, rows_max, top_k, hidden, inter) | |
| _COOP[key] = sc | |
| return sc | |
| _PAR: dict[tuple, tuple[list[CoopScratch], list[torch.cuda.Stream]]] = {} | |
| def get_parallel_coop(device: torch.device, rows_max: int, top_k: int, hidden: int, inter: int, n: int): | |
| """n independent scratch sets + n side streams, shared by every layer of the process (layers run in stream order, | |
| and each call joins its streams back before returning), for run_decode_coop_parallel.""" | |
| key = (str(device), rows_max, top_k, hidden, inter, n) | |
| v = _PAR.get(key) | |
| if v is None: | |
| v = ([_new_coop_scratch(device, rows_max, top_k, hidden, inter) for _ in range(n)], | |
| [torch.cuda.Stream(device=device) for _ in range(n)]) | |
| _PAR[key] = v | |
| return v | |
| def _new_coop_scratch(device: torch.device, rows_max: int, top_k: int, hidden: int, inter: int) -> CoopScratch: | |
| if True: | |
| slots_max = rows_max * top_k | |
| if slots_max > COOP_MAX_SLOTS: | |
| raise ValueError(f"coop path holds at most {COOP_MAX_SLOTS} slots; rows_max {rows_max} x top_k {top_k}") | |
| f16 = dict(dtype=torch.float16, device=device) | |
| f32 = dict(dtype=torch.float32, device=device) | |
| sc = CoopScratch( | |
| rows_max=rows_max, | |
| slots_max=slots_max, | |
| had_g=torch.empty((slots_max, hidden), **f16), | |
| had_u=torch.empty((slots_max, hidden), **f16), | |
| gu_g=torch.empty((slots_max, inter), **f16), | |
| gu_u=torch.empty((slots_max, inter), **f16), | |
| act_out=torch.empty((slots_max, inter), **f16), | |
| d_out=torch.empty((slots_max, hidden), **f32), | |
| ctr=torch.zeros((coop_ctr_len(slots_max, rows_max, inter, hidden),), dtype=torch.int32, device=device), | |
| out=(torch.zeros((rows_max, hidden), **f32), torch.zeros((rows_max, hidden), **f32)), | |
| ) | |
| return sc | |
| def get_fused_scratch(device: torch.device, rows: int, hidden: int, inter: int) -> FusedScratch: | |
| key = (str(device), rows, hidden, inter) | |
| fs = _FUSED.get(key) | |
| if fs is None: | |
| c = int(ext().exl3_moe_max_concurrency(torch.device(device).index or 0)) | |
| f16 = dict(dtype=torch.float16, device=device) | |
| fs = FusedScratch( | |
| rows=rows, | |
| concurrency=c, | |
| tsg=torch.empty((c, rows, hidden), **f16), | |
| tsu=torch.empty((c, rows, hidden), **f16), | |
| tig=torch.empty((c, rows, inter), **f16), | |
| tiu=torch.empty((c, rows, inter), **f16), | |
| ) | |
| _FUSED[key] = fs | |
| return fs | |
| def _wbuf(device: torch.device, numel: int) -> torch.Tensor: | |
| key = (str(device),) | |
| b = _WBUF.get(key) | |
| if b is None or b.numel() < numel: | |
| b = torch.empty((numel,), dtype=torch.float16, device=device) | |
| _WBUF[key] = b | |
| return b[:numel] | |
| # --------------------------------------------------------------------------------------------- | |
| # Decode: exl3_moe_coop, one launch set per K group, chained through the shared-expert input | |
| def run_decode_coop( | |
| x16: torch.Tensor, # fp16 [M, H], contiguous | |
| sel: torch.Tensor, # int64 [M, top_k], physical (K-sorted) expert slots | |
| rw16: torch.Tensor, # fp16 [M, top_k], 0 = inactive slot (exl3_moe_coop_kernel.cuh:213-215) | |
| groups: list[GroupTables], | |
| sc: CoopScratch, | |
| act_limit: float, | |
| mcg: bool, | |
| mul1: bool, | |
| ) -> torch.Tensor: | |
| """Routed partial sum of this TP rank, fp32 [M, H]. | |
| Group g writes out_g = (its experts' weighted outputs) + out_{g-1}: the kernel adds its optional | |
| shared-expert input sh_out to every row it stores (exl3_moe_coop_kernel.cuh:572-610, 929-947) and | |
| rows with no active slot get sh_out alone, so chaining gives the sum over groups with no extra | |
| kernel. `out` is always passed whole so rows_max (and the counter layout derived from it, | |
| exl3_moe_coop.cu:228-238) never changes between calls. | |
| TODO(untested on sm_121): numerics vs the reconstruct reference, every K in the plan, bsz 1..16. | |
| """ | |
| m, hidden = x16.shape | |
| e = ext() | |
| prev = None | |
| for i, g in enumerate(groups): | |
| out = sc.out[i % 2] | |
| e.exl3_moe_coop( | |
| x16, sel, rw16, | |
| g.lo, g.hi, # expert-range filter: slots outside [lo, hi) are inactive | |
| hidden, # Hi (5120, a multiple of 128: no padding) | |
| g.g.trellis, g.g.suh, g.g.svh, | |
| g.u.trellis, g.u.suh, g.u.svh, | |
| g.d.trellis, g.d.suh, g.d.svh, | |
| None, None, None, # no expert biases in DeepSeek-V4.1 | |
| float(g.k_gu), float(g.k_gu), float(g.k_d), | |
| mcg, mul1, | |
| MOE_COOP_ACT_SILU, float(act_limit), | |
| True, # gated | |
| sc.had_g, sc.had_u, sc.gu_g, sc.gu_u, sc.act_out, sc.d_out, sc.ctr, | |
| out, | |
| prev, # sh_out: running sum of the previous groups | |
| None, # sh_gate_w | |
| ) | |
| prev = out | |
| assert prev is not None | |
| return prev[:m] | |
| def run_decode_coop_parallel( | |
| x16: torch.Tensor, sel: torch.Tensor, rw16: torch.Tensor, groups: list[GroupTables], | |
| scs: list[CoopScratch], streams: list[torch.cuda.Stream], act_limit: float, mcg: bool, mul1: bool, | |
| reduce: bool = True, | |
| ): | |
| """run_decode_coop with the K groups concurrent instead of chained: group i runs on side stream i with its own | |
| scratch (counters included) and writes only its own experts' sum (no sh_out), then the main stream joins every | |
| stream and adds the partial outputs in fp32. At decode sizes one layer is up to 8 small launches in sequence | |
| (2 per group); running the groups side by side lets one group's kernels fill the SMs another group's tail | |
| leaves idle. The fork / join is legal under CUDA-graph capture (side streams wait on the capturing stream and | |
| are joined before the call returns). VLLM_EXL3_MOE_PARALLEL_GROUPS=1 selects it.""" | |
| m, hidden = x16.shape | |
| e = ext() | |
| main = torch.cuda.current_stream() | |
| outs = [] | |
| for g, sc, st in zip(groups, scs, streams): | |
| st.wait_stream(main) | |
| with torch.cuda.stream(st): | |
| out = sc.out[0] | |
| e.exl3_moe_coop( | |
| x16, sel, rw16, g.lo, g.hi, hidden, | |
| g.g.trellis, g.g.suh, g.g.svh, g.u.trellis, g.u.suh, g.u.svh, g.d.trellis, g.d.suh, g.d.svh, | |
| None, None, None, | |
| float(g.k_gu), float(g.k_gu), float(g.k_d), mcg, mul1, | |
| MOE_COOP_ACT_SILU, float(act_limit), True, | |
| sc.had_g, sc.had_u, sc.gu_g, sc.gu_u, sc.act_out, sc.d_out, sc.ctr, | |
| out, None, None, | |
| ) | |
| outs.append(out) | |
| for st in streams[:len(groups)]: | |
| main.wait_stream(st) | |
| if not reduce: | |
| return outs # per-group fp32 [rows_max, hidden] scratch; the caller sums rows [:m] | |
| total = outs[0][:m].clone() | |
| for o in outs[1:]: | |
| total.add_(o[:m]) | |
| return total | |
| # --------------------------------------------------------------------------------------------- | |
| # Prefill / large batches: exl3_moe per K group (fused tier) + reconstruct for hot experts | |
| def bucket_rows(phys_flat: torch.Tensor, g: GroupTables, flat_tok: torch.Tensor, flat_w16: torch.Tensor): | |
| """Sort the batch's assignments by group-local expert id; picks outside the group go to the | |
| sentinel bin E (exl3_moe.cu:200-203: the last count is ignored). Counts come from scatter_add, | |
| not bincount: torch.bincount reads max() back to the host, which a CUDA graph cannot take.""" | |
| n = g.size | |
| loc = phys_flat - g.lo | |
| inb = (loc >= 0) & (loc < n) | |
| loc = torch.where(inb, loc, torch.full_like(loc, n)) | |
| order = torch.argsort(loc) | |
| counts = torch.zeros((n + 1,), dtype=torch.long, device=loc.device) | |
| counts.scatter_add_(0, loc, torch.ones_like(loc)) | |
| return counts, flat_tok[order].contiguous(), flat_w16[order].contiguous() | |
| def run_exl3_moe( | |
| x16: torch.Tensor, out32: torch.Tensor, g: GroupTables, counts: torch.Tensor, | |
| token_sorted: torch.Tensor, weight_sorted: torch.Tensor, fs: FusedScratch, | |
| act_limit: float, mcg: bool, mul1: bool, | |
| num_active: int = -1, count_lo: int = 1, count_hi: int | None = None, m_tile: int = 16, | |
| output_scratch: torch.Tensor | None = None, fused_base: torch.Tensor | None = None, | |
| ) -> None: | |
| """One exl3_moe launch over group g: accumulates into out32 with atomics (not bit-reproducible), or, with | |
| output_scratch [slots, H] fp32 + fused_base [n] int64, writes every assignment's weighted output into its own slot | |
| (row r of expert e -> fused_base[e] + r) for exl3_moe_gather to sum in k order (exl3_moe.cu:575-697). | |
| num_active = -1 under CUDA graphs (exl3_moe.cu:254-257).""" | |
| ext().exl3_moe( | |
| x16, out32, counts, token_sorted, weight_sorted, | |
| fs.tsg, fs.tsu, fs.tig, fs.tiu, | |
| MOE_ACT_SILU, | |
| float(g.k_gu), float(g.k_gu), float(g.k_d), | |
| g.g.trellis, g.g.suh, g.g.svh, | |
| g.u.trellis, g.u.suh, g.u.svh, | |
| g.d.trellis, g.d.suh, g.d.svh, | |
| mcg, mul1, mcg, mul1, mcg, mul1, | |
| float(act_limit), | |
| int(num_active), | |
| output_scratch, fused_base, | |
| int(count_lo), int(fs.rows if count_hi is None else count_hi), int(m_tile), | |
| ) | |
| def exl3_linear_reconstruct( | |
| x16: torch.Tensor, trellis: torch.Tensor, suh: torch.Tensor, svh: torch.Tensor, | |
| k: float, mcg: bool, mul1: bool, | |
| ) -> torch.Tensor: | |
| """y = x @ W for one EXL3 matrix, fp32 out: rotate x by suh + Hadamard, dequantize the trellis in | |
| the rotated basis (exact in fp16), GEMM, rotate back with svh. Same arithmetic as | |
| LinearEXL3.reconstruct_hgemm's unfused branch (modules/quant/exl3.py:166-223).""" | |
| e = ext() | |
| k_in, n_out = trellis.shape[0] * 16, trellis.shape[1] * 16 | |
| xh = torch.empty_like(x16) | |
| e.had_r_128(x16, xh, suh, None, 1.0) | |
| w = _wbuf(x16.device, k_in * n_out).view(k_in, n_out) | |
| e.reconstruct(w, trellis, float(k), mcg, mul1) | |
| y = torch.empty((x16.shape[0], n_out), dtype=torch.float32, device=x16.device) | |
| e.hgemm(xh, w, y) | |
| e.had_r_128(y, y, None, svh, 1.0) | |
| return y | |
| def swiglu_limit(gate: torch.Tensor, up: torch.Tensor, limit: float) -> torch.Tensor: | |
| """silu(min(gate, limit)) * clamp(up, -limit, limit): DeepSeek-V4 / vLLM semantics | |
| (vllm fused_moe/activation.py:296-310) and exllamav3's act_limit (activation_kernels.cuh:178-182).""" | |
| if limit: | |
| gate = gate.clamp(max=limit) | |
| up = up.clamp(min=-limit, max=limit) | |
| return torch.nn.functional.silu(gate) * up | |
| def prime(device: torch.device, groups: list[GroupTables], sc: CoopScratch, fs: FusedScratch | None, | |
| hidden: int, top_k: int, act_limit: float, mcg: bool, mul1: bool) -> None: | |
| """Eager warm-up before any CUDA graph capture: the first launch of each kernel instance sets its | |
| shared-memory attribute (exl3_moe_coop.cu:116-125, exl3_moe.cu:495-500) and exl3_moe allocates its | |
| lock buffer with cudaMalloc on first use (exl3_devctx.cu:89-106), which is illegal while a stream | |
| is capturing. One single-token call per group activates exactly that group's instances. | |
| Every serve since Oct 9 captures its graphs after this priming without an allocation error.""" | |
| if torch.cuda.is_current_stream_capturing(): | |
| raise RuntimeError("vllm_exl3_moe.prime() must run before CUDA graph capture") | |
| x = torch.zeros((1, hidden), dtype=torch.float16, device=device) | |
| for g in groups: | |
| # distinct picks inside the group (bsz 1 assumes distinct top-k, exl3_moe_coop_kernel.cuh:224-227); | |
| # only slot 0 carries weight, the rest are inactive (w == 0) | |
| sel = (g.lo + torch.arange(top_k, device=device) % g.size).to(torch.long).view(1, top_k) | |
| rw = torch.zeros((1, top_k), dtype=torch.float16, device=device) | |
| rw[0, 0] = 1.0 | |
| run_decode_coop(x, sel, rw, [g], sc, act_limit, mcg, mul1) | |
| if fs is not None: | |
| flat = sel.reshape(-1) | |
| counts, tok, w = bucket_rows(flat, g, torch.zeros_like(flat), rw.reshape(-1)) | |
| out = torch.zeros((1, hidden), dtype=torch.float32, device=device) | |
| run_exl3_moe(x, out, g, counts, tok, w, fs, act_limit, mcg, mul1, num_active=-1) | |
| torch.cuda.synchronize(device) | |
| def env_int(name: str, default: int) -> int: | |
| v = os.environ.get(name) | |
| return int(v) if v not in (None, "") else default | |