ProprietaryLegal's picture
PLI Labs public snapshot 2026-10-28 (history squashed)
e1acc15
Raw History Blame Contribute Delete
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
@dataclass
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])
@dataclass
class GroupTables:
k_gu: float
k_d: float
lo: int
hi: int
g: ProjTables # gate (w1)
u: ProjTables # up (w3)
d: ProjTables # down (w2)
@property
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)
@dataclass
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)
@dataclass
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