use crate::linear_core::{GdnCfg, GdnWeights, gdn_forward};
use crate::loader::{Overlay, build_ffn_at, load_f32, load_matrix};
use crate::pipeline::{FfnKind, MoeFfn, moe_ffn};
use crate::pool::Pool;
use crate::qtensor::QTensor;
use cortiq_core::{CmfError, CmfModel, LayerType, ModelArch, Qwen4ExpConfig, TensorDtype};
use std::cmp::Ordering;
use std::sync::Arc;
const PRIME_1: u64 = 10_007;
const SPLITMIX_GAMMA: u64 = 0x9E37_79B9_7F4A_7C15;
const SPLITMIX_M1: u64 = 0xBF58_476D_1CE4_E5B9;
const SPLITMIX_M2: u64 = 0x94D0_49BB_1331_11EB;
pub struct GatedResidual {
pub(crate) norm: Vec<f32>,
pub(crate) down: QTensor,
pub(crate) up: QTensor,
pub(crate) inject: Option<QTensor>,
down_idx: Option<usize>,
up_idx: Option<usize>,
inject_idx: Option<usize>,
}
struct LayerIdx {
gdn: Option<[usize; 5]>,
qsa: Option<[usize; 5]>,
ple: Option<(usize, usize)>,
router: usize,
shared_gate: Option<usize>,
}
pub struct QsaWeights {
pub(crate) q_proj: QTensor,
pub(crate) k_proj: QTensor,
pub(crate) v_proj: QTensor,
pub(crate) o_proj: QTensor,
pub(crate) q_norm: Vec<f32>,
pub(crate) k_norm: Vec<f32>,
pub(crate) index_qk: QTensor,
pub(crate) index_q_norm: Vec<f32>,
pub(crate) index_k_norm: Vec<f32>,
}
pub struct PleWeights {
shards: Vec<QTensor>,
rows_per_shard: usize,
row_dim: usize,
pub(crate) key_proj: QTensor,
pub(crate) value_proj: QTensor,
pub(crate) norm_key: Vec<f32>,
pub(crate) norm_query: Vec<f32>,
pub(crate) norm_conv: Vec<f32>,
pub(crate) conv: Vec<f32>,
multipliers: Vec<i64>,
vocab_sizes: Vec<i64>,
offsets: Vec<i64>,
}
pub enum Mixer {
Gdn(GdnWeights),
Qsa(QsaWeights),
}
pub struct Layer {
pub(crate) attn_hc: GatedResidual,
pub(crate) mlp_hc: GatedResidual,
pub(crate) mixer: Mixer,
pub(crate) moe: MoeFfn,
pub(crate) ple: Option<PleWeights>,
expert_ids: Vec<(usize, usize, usize)>,
shared_ids: Option<(usize, usize, usize)>,
idx: Option<LayerIdx>,
}
impl Layer {
pub fn expert_geometry(&self) -> (usize, usize) {
let inter = self.moe.experts.first().map_or(0, |d| d.gate_proj.rows());
(self.moe.experts.len(), inter)
}
}
pub struct Globals {
pub(crate) embed: QTensor,
pub(crate) lm_head: QTensor,
pub(crate) head_hc: GatedResidual,
lm_head_idx: Option<usize>,
embed_idx: Option<usize>,
skeleton_idxs: Vec<usize>,
}
#[derive(Clone)]
pub struct Cfg {
pub(crate) hidden: usize,
pub(crate) hc: usize,
pub(crate) eps: f64,
pub(crate) n_heads: usize,
pub(crate) n_kv_heads: usize,
pub(crate) head_dim: usize,
pub(crate) rotary_dim: usize,
pub(crate) index_heads: usize,
pub(crate) index_kv_heads: usize,
pub(crate) index_dim: usize,
pub(crate) index_budget: usize,
pub(crate) compress_ratio: usize,
pub(crate) gdn: GdnCfg,
ngram_size: usize,
heads_per_ngram: usize,
pub(crate) ple_kernel: usize,
pub(crate) ple_dilation: usize,
eos: u32,
}
#[derive(Default)]
struct QsaState {
raw_keys: Vec<f32>,
keys: Vec<f32>,
values: Vec<f32>,
}
#[derive(Default)]
struct LayerState {
gdn: Vec<f32>,
qsa: QsaState,
ple_history: Vec<f32>,
ple_history_rows: usize,
}
pub struct State {
hyper: Vec<f32>,
layers: Vec<LayerState>,
token_history: Vec<u32>,
gpu_pool: Option<QwenGpuPool>,
pub pos: usize,
#[cfg(feature = "gpu")]
dev: Option<crate::gpu_wgpu::qwen4::Dev>,
picks_prev: Vec<Vec<usize>>,
device_off: bool,
host_pos: usize,
#[cfg(feature = "cuda")]
cuda: Option<crate::gpu_cuda::qwen4::Dev>,
cuda_off: bool,
#[cfg(feature = "cuda")]
cuda_ref: Option<Box<State>>,
#[cfg(feature = "cuda")]
cuda_copy: CopyCur,
#[cfg(feature = "gpu")]
profile: Option<ExpertProfile>,
verify_window: bool,
window_rows: Option<WindowRows>,
round_end: Option<std::time::Instant>,
#[cfg(feature = "gpu")]
mtp: Option<MtpHead>,
mtp_tried: bool,
#[cfg(feature = "gpu")]
resident_all: bool,
#[cfg(feature = "gpu")]
inflight: Option<crate::gpu_wgpu::qwen4::Pending>,
#[cfg(feature = "gpu")]
route_hist: Vec<u32>,
#[cfg(feature = "gpu")]
route_tokens: usize,
#[cfg(feature = "gpu")]
phi_last: Vec<u32>,
#[cfg(feature = "gpu")]
phi_last_tokens: usize,
#[cfg(feature = "gpu")]
row_snap: Vec<u32>,
#[cfg(feature = "gpu")]
row_snap_tokens: usize,
#[cfg(feature = "gpu")]
row_gen: bool,
#[cfg(feature = "gpu")]
core_pinned: bool,
#[cfg(feature = "gpu")]
packs: Option<Option<PackSet>>,
#[cfg(feature = "gpu")]
masked: bool,
#[cfg(feature = "gpu")]
copy_frame: bool,
#[cfg(feature = "gpu")]
frame_check: bool,
#[cfg(feature = "gpu")]
frame_ids: Option<Vec<u32>>,
#[cfg(feature = "gpu")]
copy_run: usize,
#[cfg(feature = "gpu")]
draft_budget: usize,
}
pub(crate) static FILL_NS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
pub(crate) static FILL_N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
pub(crate) static FILLS_WALL_NS: std::sync::atomic::AtomicU64 =
std::sync::atomic::AtomicU64::new(0);
#[cfg(feature = "gpu")]
fn expert_slices(model: &CmfModel, t: (usize, usize, usize)) -> Option<[&[u8]; 3]> {
let bytes = model.primary_bytes();
let part = |i: usize| -> Option<&[u8]> {
let e = model.tensors.get(i)?;
let abs = model.entry_abs_offset(e)?;
bytes.get(abs..abs + e.nbytes as usize)
};
Some([part(t.0)?, part(t.1)?, part(t.2)?])
}
fn fill_trace() -> bool {
static S: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*S.get_or_init(|| std::env::var("CMF_QWEN_FILL_TRACE").as_deref() == Ok("1"))
}
pub(crate) struct QwenGpuPool {
pub(crate) hold_layers: [Option<usize>; 2],
staging_base: usize,
staging: usize,
#[cfg(feature = "gpu")]
stager: Option<crate::gpu_wgpu::qwen4::Stager>,
#[cfg(feature = "gpu")]
pub(crate) tier: Option<Arc<crate::gpu_wgpu::qwen4::HostTier>>,
admit_pool: Option<Arc<Pool>>,
pub(crate) store: Option<Arc<crate::expert_store::ExpertStore>>,
#[cfg(feature = "gpu")]
pub(crate) codec: crate::gpu_wgpu::MoeBankCodec,
pub(crate) segment_slots: usize,
floor: usize,
n_experts: usize,
fetch_quota: usize,
fetch_min_seen: u16,
fetch_max_env: &'static str,
fetch_min_env: &'static str,
owner: Vec<Option<(usize, usize)>>,
slot_for: Vec<u32>,
shared_slot: Vec<u32>,
seen: Vec<u16>,
seen_epoch: Vec<u32>,
free: Vec<usize>,
occupancy: Vec<usize>,
last: Vec<u64>,
clock: u64,
pin: Vec<bool>,
sweep_slots: Vec<usize>,
sweep_of: Vec<u32>,
}
#[cfg(feature = "gpu")]
#[must_use = "a pending admission must be finished or deliberately dropped"]
pub(crate) struct QwenGpuPendingEnsure<'pool, 'input> {
pool: Option<&'pool mut QwenGpuPool>,
model: &'input Arc<CmfModel>,
picks: &'input [usize],
triples: &'input [(usize, usize, usize)],
layer: usize,
base: usize,
shared_slot: u32,
plan: Vec<(usize, usize)>,
}
#[cfg(feature = "gpu")]
impl QwenGpuPendingEnsure<'_, '_> {
pub(crate) fn cold_ids(&self) -> Option<Vec<usize>> {
let pool = self.pool.as_deref()?;
cold_picks_from_slots(&pool.slot_for, self.base, pool.n_experts, self.picks)
}
pub(crate) fn finish(self) -> Option<(Vec<u32>, u32)> {
self.finish_inner(false)
.map(|(remap, shared_slot, _)| (remap, shared_slot))
}
pub(crate) fn finish_and_flush(self) -> Option<(Vec<u32>, u32, usize)> {
self.finish_inner(true)
}
fn finish_inner(mut self, flush_uploads: bool) -> Option<(Vec<u32>, u32, usize)> {
let pool = self.pool.take()?;
let model = self.model;
let triples = self.triples;
let layer = self.layer;
let base = self.base;
let shared_slot = self.shared_slot;
let plan = std::mem::take(&mut self.plan);
let ok = pool.run_fills(model, layer, &plan, triples, false);
let mut failed = false;
for (&(slot, e), ok) in plan.iter().zip(&ok) {
if !ok {
pool.owner[slot] = None;
pool.slot_for[base + e] = u32::MAX;
pool.occupancy[layer] = pool.occupancy[layer].saturating_sub(1);
pool.free.push(slot);
failed = true;
}
}
let result = if failed || triples.len() != pool.n_experts {
None
} else {
let remap = pool.slot_for[base..base + pool.n_experts].to_vec();
Some((remap, shared_slot, pool.segment_slots))
};
if flush_uploads {
pool.flush_uploads();
}
result
}
}
#[cfg(feature = "gpu")]
impl Drop for QwenGpuPendingEnsure<'_, '_> {
fn drop(&mut self) {
let Some(pool) = self.pool.as_deref_mut() else {
return;
};
for &(slot, expert) in &self.plan {
let Some(key) = self.base.checked_add(expert) else {
continue;
};
if pool.owner.get(slot).copied().flatten() == Some((self.layer, expert))
&& pool.slot_for.get(key) == Some(&(slot as u32))
{
pool.owner[slot] = None;
pool.slot_for[key] = u32::MAX;
if let Some(occupancy) = pool.occupancy.get_mut(self.layer) {
*occupancy = occupancy.saturating_sub(1);
}
pool.free.push(slot);
}
}
}
}
#[cfg(feature = "gpu")]
fn cold_picks_from_slots(
slot_for: &[u32],
base: usize,
n_experts: usize,
picks: &[usize],
) -> Option<Vec<usize>> {
let slots = slot_for.get(base..base.checked_add(n_experts)?)?;
let mut cold = Vec::with_capacity(picks.len());
for &expert in picks {
let slot = *slots.get(expert)?;
if slot == u32::MAX {
cold.push(expert);
}
}
Some(cold)
}
fn bounded_pool_pct(raw: Option<&str>, default: usize, min: usize, max: usize) -> usize {
raw.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(default)
.clamp(min, max)
}
fn bounded_pool_slots(raw: Option<&str>, safe: usize, cap_override: bool) -> usize {
let floor = if cap_override { safe.min(8) } else { 8 };
raw.and_then(|v| v.parse::<usize>().ok())
.map(|v| if cap_override { v.min(safe) } else { v })
.unwrap_or(safe)
.max(floor)
}
fn active_layer_floor(capacity: usize, active_layers: usize) -> usize {
capacity / active_layers.max(1)
}
impl State {
pub fn new(n_layers: usize) -> Self {
Self {
hyper: Vec::new(),
layers: (0..n_layers).map(|_| LayerState::default()).collect(),
token_history: Vec::new(),
gpu_pool: None,
pos: 0,
#[cfg(feature = "gpu")]
dev: None,
picks_prev: vec![Vec::new(); n_layers],
device_off: false,
host_pos: 0,
#[cfg(feature = "cuda")]
cuda: None,
cuda_off: false,
#[cfg(feature = "cuda")]
cuda_ref: None,
#[cfg(feature = "cuda")]
cuda_copy: CopyCur::default(),
#[cfg(feature = "gpu")]
profile: None,
verify_window: false,
window_rows: None,
round_end: None,
#[cfg(feature = "gpu")]
mtp: None,
mtp_tried: false,
#[cfg(feature = "gpu")]
resident_all: false,
#[cfg(feature = "gpu")]
inflight: None,
#[cfg(feature = "gpu")]
route_hist: Vec::new(),
#[cfg(feature = "gpu")]
route_tokens: 0,
#[cfg(feature = "gpu")]
phi_last: Vec::new(),
#[cfg(feature = "gpu")]
phi_last_tokens: 0,
#[cfg(feature = "gpu")]
row_snap: Vec::new(),
#[cfg(feature = "gpu")]
row_snap_tokens: 0,
#[cfg(feature = "gpu")]
row_gen: false,
#[cfg(feature = "gpu")]
core_pinned: false,
#[cfg(feature = "gpu")]
packs: None,
#[cfg(feature = "gpu")]
masked: false,
#[cfg(feature = "gpu")]
copy_frame: false,
#[cfg(feature = "gpu")]
frame_check: false,
#[cfg(feature = "gpu")]
frame_ids: None,
#[cfg(feature = "gpu")]
copy_run: 0,
#[cfg(feature = "gpu")]
draft_budget: usize::MAX,
}
}
#[cfg(feature = "gpu")]
pub fn set_draft_budget(&mut self, n: usize) {
self.draft_budget = n;
}
pub(crate) fn reset(&mut self) {
#[cfg(feature = "gpu")]
if let Some(p) = self.inflight.take() {
let _ = p.wait();
}
self.hyper.clear();
self.token_history.clear();
self.pos = 0;
self.host_pos = 0;
#[cfg(feature = "cuda")]
{
self.cuda_copy = CopyCur::default();
}
#[cfg(feature = "gpu")]
{
if self.row_gen {
route_row("gen", &self.route_hist, self.route_tokens, &mut self.row_snap, &mut self.row_snap_tokens);
self.row_gen = false;
}
trace_flush();
self.route_hist.iter_mut().for_each(|c| *c = 0);
self.route_tokens = 0;
self.phi_last.iter_mut().for_each(|c| *c = 0);
self.phi_last_tokens = 0;
self.row_snap.iter_mut().for_each(|c| *c = 0);
self.row_snap_tokens = 0;
}
for st in &mut self.layers {
*st = LayerState::default();
}
}
}
#[cfg(feature = "gpu")]
impl QwenGpuPool {
pub(crate) fn create(
model: &Arc<CmfModel>,
inter: usize,
hidden: usize,
n_layers: usize,
n_experts: usize,
gu_q2: bool,
) -> Option<Self> {
Self::create_with_policy(
model,
inter,
hidden,
n_layers,
n_experts,
crate::gpu_wgpu::MoeBankCodec::from_gu_q2(gu_q2),
"CMF_QWEN_POOL_PCT",
75,
25,
85,
"CMF_QWEN_EXPERT_SLOTS",
"CMF_QWEN_FETCH_MAX",
"CMF_QWEN_FETCH_MIN_SEEN",
None,
0,
)
}
pub(crate) fn create_explicit(
model: &Arc<CmfModel>,
inter: usize,
hidden: usize,
n_layers: usize,
n_experts: usize,
gu_q2: bool,
slots: usize,
staging: usize,
) -> Option<Self> {
Self::create_with_policy(
model,
inter,
hidden,
n_layers,
n_experts,
crate::gpu_wgpu::MoeBankCodec::from_gu_q2(gu_q2),
"CMF_QWEN_POOL_PCT",
75,
25,
85,
"CMF_QWEN_EXPERT_SLOTS",
"CMF_QWEN_FETCH_MAX",
"CMF_QWEN_FETCH_MIN_SEEN",
Some(slots),
staging,
)
.map(|mut pool| {
let stage_mb = std::env::var("CMF_QWEN_STAGE_MB")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(0);
pool.stager = crate::gpu_wgpu::qwen4::Stager::new(stage_mb);
let admit_threads = std::env::var("CMF_QWEN_ADMIT_THREADS")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(16);
pool.admit_pool = (admit_threads > 0).then(|| Arc::new(Pool::new(admit_threads)));
pool.fetch_min_seen = std::env::var("CMF_QWEN_FETCH_MIN_SEEN")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(1);
pool.fetch_quota = std::env::var("CMF_QWEN_FETCH_MAX")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(pool.fetch_quota.max(16));
pool
})
}
pub(crate) fn configure_minicpm_tail(&mut self, active_layers: usize) {
let layers = active_layers.clamp(1, self.occupancy.len().max(1));
self.floor = active_layer_floor(self.staging_base, layers);
if std::env::var_os("CMF_QWEN_ADMIT_THREADS").is_none() {
self.admit_pool = Some(Arc::new(Pool::new(2)));
}
}
pub(crate) fn create_exact(
model: &Arc<CmfModel>,
inter: usize,
hidden: usize,
n_experts: usize,
gu_q2: bool,
slots: usize,
) -> Option<Self> {
Self::create_with_policy(
model,
inter,
hidden,
1,
n_experts,
crate::gpu_wgpu::MoeBankCodec::from_gu_q2(gu_q2),
"CMF_QWEN_MTP_BANK",
75,
25,
85,
"CMF_QWEN_MTP_BANK_SLOTS",
"CMF_QWEN_FETCH_MAX",
"CMF_QWEN_FETCH_MIN_SEEN",
Some(slots),
0,
)
}
pub(crate) fn per_expert_bytes(inter: usize, hidden: usize, gu_q2: bool) -> Option<usize> {
let gu = cortiq_core::quant::expected_nbytes(
if gu_q2 {
TensorDtype::Q2TiledP
} else {
TensorDtype::Q4TiledP
},
&[inter, hidden],
)?;
let dn = cortiq_core::quant::expected_nbytes(TensorDtype::Q4TiledP, &[hidden, inter])?;
2usize.checked_mul(gu)?.checked_add(dn)
}
fn victim(&mut self, layer: usize, protect: &[usize]) -> Option<usize> {
if let Some(s) = self.free.pop() {
self.sweep_of[s] = 0;
return Some(s);
}
let s = self.sweep_victim(layer, protect).or_else(|| self.lru_victim(layer, protect))?;
self.sweep_of[s] = 0;
Some(s)
}
fn sweep_victim(&mut self, layer: usize, protect: &[usize]) -> Option<usize> {
let ne = self.n_experts;
let mut i = self.sweep_slots.len();
while i > 0 {
i -= 1;
let slot = self.sweep_slots[i];
let live = match self.owner[slot] {
Some((l, e)) if e != usize::MAX && self.sweep_of[slot] as usize == l * ne + e + 1 => Some((l, e)),
_ => None,
};
let Some((l, e)) = live else {
self.sweep_slots.remove(i);
continue;
};
if self.pin.get(l * ne + e).copied().unwrap_or(false) {
self.sweep_of[slot] = 0;
self.sweep_slots.remove(i);
continue;
}
if (l == layer && protect.contains(&e)) || self.hold_layers.contains(&Some(l)) {
continue;
}
self.sweep_slots.remove(i);
return Some(slot);
}
None
}
pub(crate) fn sweep_clear(&mut self) {
self.sweep_slots.clear();
self.sweep_of.fill(0);
}
fn lru_victim(&self, layer: usize, protect: &[usize]) -> Option<usize> {
let hold = self.hold_layers;
let (pin, ne) = (&self.pin, self.n_experts);
let eligible = |owner: (usize, usize)| {
owner.1 != usize::MAX
&& !pin.get(owner.0 * ne + owner.1).copied().unwrap_or(false)
&& (owner.0 != layer || !protect.contains(&owner.1))
&& !hold.contains(&Some(owner.0))
};
self.owner
.iter()
.enumerate()
.filter_map(|(slot, &o)| {
o.filter(|&x| {
eligible(x) && self.occupancy.get(x.0).copied().unwrap_or(0) > self.floor
})
.map(|_| slot)
})
.min_by_key(|&slot| self.last[slot])
.or_else(|| {
self.owner
.iter()
.enumerate()
.filter_map(|(slot, &o)| {
o.filter(|&x| eligible(x) && x.0 == layer).map(|_| slot)
})
.min_by_key(|&slot| self.last[slot])
})
.or_else(|| {
self.owner
.iter()
.enumerate()
.filter_map(|(slot, &o)| o.filter(|&x| eligible(x)).map(|_| slot))
.min_by_key(|&slot| self.last[slot])
})
}
pub(crate) fn admit_now(
&mut self,
model: &Arc<CmfModel>,
layer: usize,
expert: usize,
triples: &[(usize, usize, usize)],
protect: &[usize],
) -> Option<u32> {
if expert >= self.n_experts || triples.len() != self.n_experts {
return None;
}
let key = layer * self.n_experts + expert;
if self.slot_for[key] != u32::MAX {
return Some(self.slot_for[key]);
}
let victim = self.victim(layer, protect)?;
if !self.fill_slot(model, victim, layer, expert, triples[expert]) {
if self.owner[victim].is_none() {
self.free.push(victim);
}
return None;
}
if let Some(old) = self.owner[victim] {
self.dropped(old);
if old.1 == usize::MAX {
self.shared_slot[old.0] = u32::MAX;
} else {
self.slot_for[old.0 * self.n_experts + old.1] = u32::MAX;
self.occupancy[old.0] = self.occupancy[old.0].saturating_sub(1);
}
}
self.clock = self.clock.saturating_add(1);
self.owner[victim] = Some((layer, expert));
self.slot_for[key] = victim as u32;
self.occupancy[layer] += 1;
self.seen[key] = self.seen[key].max(1);
self.last[victim] = self.clock;
Some(victim as u32)
}
#[cfg(feature = "gpu")]
fn fill_slot(
&self,
model: &Arc<CmfModel>,
slot: usize,
layer: usize,
expert: usize,
triple: (usize, usize, usize),
) -> bool {
let t0 = std::time::Instant::now();
let key = layer * self.n_experts + expert;
let tier = self.tier.as_ref();
if tier.is_some_and(|t| t.hit(key, slot)) {
if slot < self.staging_base
&& let Some(store) = self.store.as_ref()
{
store.note_vram(layer, expert, true);
}
return true;
}
let up = |parts: [&[u8]; 3]| -> bool {
if tier.is_some_and(|t| t.fill(key, Some(slot), parts)) {
return true;
}
crate::gpu_wgpu::qwen4::upload_expert_parts(self.stager.as_ref(), model, slot, parts)
};
let stored = self
.store
.as_ref()
.and_then(|store| store.with_expert(layer, expert, true, up));
let ok = match stored {
Some(ok) => ok,
None => {
if tier.is_some()
&& let Some(parts) = expert_slices(model, triple)
{
up(parts)
} else if let Some(st) = self.stager.as_ref()
&& crate::gpu_wgpu::qwen4::stage_expert(st, model, slot, triple)
{
true
} else {
crate::gpu_wgpu::dsv4_global_slot_fill(model, slot, triple)
}
}
};
if ok && slot < self.staging_base
&& let Some(store) = self.store.as_ref()
{
store.note_vram(layer, expert, true);
}
FILL_NS.fetch_add(
t0.elapsed().as_nanos() as u64,
std::sync::atomic::Ordering::Relaxed,
);
FILL_N.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
ok
}
#[cfg(feature = "gpu")]
fn run_fills(
&self,
model: &Arc<CmfModel>,
layer: usize,
jobs: &[(usize, usize)],
triples: &[(usize, usize, usize)],
chunked: bool,
) -> Vec<bool> {
let t0 = std::time::Instant::now();
let me = self;
let device = crate::gpu::current_device();
let out: Vec<bool> = if jobs.len() <= 1 {
jobs.iter()
.map(|&(slot, e)| me.fill_slot(model, slot, layer, e, triples[e]))
.collect()
} else if let Some(pool) = self.admit_pool.as_ref() {
let done: Vec<std::sync::atomic::AtomicBool> = jobs
.iter()
.map(|_| std::sync::atomic::AtomicBool::new(false))
.collect();
let next = std::sync::atomic::AtomicUsize::new(0);
pool.run(&|_, _| {
loop {
let i = next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let Some(&(slot, e)) = jobs.get(i) else { break };
let ok = me.fill_slot(model, slot, layer, e, triples[e]);
done[i].store(ok, std::sync::atomic::Ordering::Relaxed);
}
});
done.iter()
.map(|d| d.load(std::sync::atomic::Ordering::Relaxed))
.collect()
} else if chunked {
const ADMIT_THREADS: usize = 32;
let nth = jobs.len().min(ADMIT_THREADS);
let per = jobs.len().div_ceil(nth);
std::thread::scope(|scope| {
let handles: Vec<_> = jobs
.chunks(per)
.map(|part| {
let model = model.clone();
let h = scope.spawn(move || {
crate::gpu::with_device(device, || {
part.iter()
.map(|&(slot, e)| {
me.fill_slot(&model, slot, layer, e, triples[e])
})
.collect::<Vec<bool>>()
})
});
(part.len(), h)
})
.collect();
handles
.into_iter()
.flat_map(|(n, h)| h.join().unwrap_or_else(|_| vec![false; n]))
.collect()
})
} else {
std::thread::scope(|scope| {
let hs: Vec<_> = jobs
.iter()
.map(|&(slot, e)| {
let model = model.clone();
let triple = triples[e];
scope.spawn(move || {
crate::gpu::with_device(device, || {
me.fill_slot(&model, slot, layer, e, triple)
})
})
})
.collect();
hs.into_iter().map(|h| h.join().unwrap_or(false)).collect()
})
};
FILLS_WALL_NS.fetch_add(
t0.elapsed().as_nanos() as u64,
std::sync::atomic::Ordering::Relaxed,
);
out
}
fn dropped(&self, old: (usize, usize)) {
if old.1 != usize::MAX
&& let Some(store) = self.store.as_ref()
{
store.note_vram(old.0, old.1, false);
}
}
#[cfg(feature = "gpu")]
pub(crate) fn flush_uploads(&mut self) {
if let Some(st) = self.stager.as_mut() {
st.flush();
}
if let Some(cb) = self.tier.as_ref().and_then(|t| t.take()) {
crate::gpu_wgpu::qwen4::submit_cb(cb);
}
}
pub(crate) fn take_uploads(&mut self) -> Option<wgpu::CommandBuffer> {
let staged = self.stager.as_mut().and_then(|st| st.take());
let tiered = self.tier.as_ref().and_then(|t| t.take());
match (staged, tiered) {
(Some(a), Some(b)) => {
crate::gpu_wgpu::qwen4::submit_cb(a);
Some(b)
}
(a, b) => a.or(b),
}
}
pub(crate) fn uploads_done(&self) {
if let Some(t) = self.tier.as_ref() {
t.done();
}
}
pub(crate) fn rearm_uploads(&mut self) {
if let Some(st) = self.stager.as_mut() {
st.rearm();
}
}
pub(crate) fn stage_cold(
&self,
model: &Arc<CmfModel>,
layer: usize,
tok: usize,
experts: &[usize],
triples: &[(usize, usize, usize)],
top_k: usize,
) -> Vec<Option<u32>> {
if self.staging == 0 || (tok + 1) * top_k > self.staging || triples.len() != self.n_experts
{
return vec![None; experts.len()];
}
let base = self.staging_base + tok * top_k;
if fill_trace() {
eprintln!("fill staging tok={tok} experts={experts:?}");
}
let jobs: Vec<(usize, usize)> = experts
.iter()
.enumerate()
.filter(|&(_, &e)| e < self.n_experts)
.map(|(j, &e)| (base + j, e))
.collect();
let ok = self.run_fills(model, layer, &jobs, triples, false);
let mut out = vec![None; experts.len()];
let mut ji = 0;
for (j, &e) in experts.iter().enumerate() {
if e < self.n_experts {
if ok[ji] {
out[j] = Some(jobs[ji].0 as u32);
}
ji += 1;
}
}
out
}
pub(crate) fn admit_many(
&mut self,
model: &Arc<CmfModel>,
layer: usize,
experts: &[usize],
triples: &[(usize, usize, usize)],
protect: &[usize],
) -> Vec<Option<u32>> {
if triples.len() != self.n_experts {
return vec![None; experts.len()];
}
let plan = self.reserve_many(layer, experts, protect);
let uploads: Vec<(usize, usize)> = plan
.iter()
.filter_map(|(_, p)| p.filter(|&(_, e)| e != usize::MAX))
.collect();
let failed = self.upload_reserved(model, layer, &uploads, triples, false);
Self::plan_slots(plan, &failed)
}
fn reserve_many(
&mut self,
layer: usize,
experts: &[usize],
protect: &[usize],
) -> Vec<(usize, Option<(usize, usize)>)> {
let mut plan: Vec<(usize, Option<(usize, usize)>)> = Vec::with_capacity(experts.len());
for &expert in experts {
if expert >= self.n_experts {
plan.push((expert, None));
continue;
}
let key = layer * self.n_experts + expert;
if self.slot_for[key] != u32::MAX {
plan.push((expert, Some((self.slot_for[key] as usize, usize::MAX))));
continue;
}
let Some(victim) = self.victim(layer, protect) else {
plan.push((expert, None));
continue;
};
let old = self.owner[victim].take();
if let Some(old) = old {
self.dropped(old);
if old.1 == usize::MAX {
self.shared_slot[old.0] = u32::MAX;
} else {
self.slot_for[old.0 * self.n_experts + old.1] = u32::MAX;
self.occupancy[old.0] = self.occupancy[old.0].saturating_sub(1);
}
}
self.owner[victim] = Some((layer, expert));
self.slot_for[key] = victim as u32;
self.occupancy[layer] += 1;
self.seen[key] = self.seen[key].max(1);
self.clock = self.clock.saturating_add(1);
self.last[victim] = self.clock;
if fill_trace() {
eprintln!("fill cold layer={layer} expert={expert} slot={victim}");
}
plan.push((expert, Some((victim, expert))));
}
plan
}
fn upload_reserved(
&mut self,
model: &Arc<CmfModel>,
layer: usize,
uploads: &[(usize, usize)],
triples: &[(usize, usize, usize)],
pooled: bool,
) -> Vec<usize> {
if let Some(store) = self.store.as_ref() {
let es: Vec<usize> = uploads.iter().map(|&(_, e)| e).collect();
store.prefetch(layer, &es);
}
let mut hit = vec![false; uploads.len()];
if let Some(t) = self.tier.as_ref() {
for (i, &(slot, e)) in uploads.iter().enumerate() {
if t.hit(layer * self.n_experts + e, slot) {
hit[i] = true;
if slot < self.staging_base
&& let Some(store) = self.store.as_ref()
{
store.note_vram(layer, e, true);
}
}
}
}
let all = uploads;
let rest: Vec<(usize, usize)> = all
.iter()
.zip(&hit)
.filter(|(_, h)| !**h)
.map(|(&u, _)| u)
.collect();
let results = self.run_fills(model, layer, &rest, triples, pooled);
let mut rest_ok = results.into_iter();
let results: Vec<bool> = hit
.iter()
.map(|&h| h || rest_ok.next().unwrap_or(false))
.collect();
let uploads = all;
let mut failed: Vec<usize> = Vec::new();
for (i, &(slot, e)) in uploads.iter().enumerate() {
if !results.get(i).copied().unwrap_or(false) {
self.owner[slot] = None;
self.slot_for[layer * self.n_experts + e] = u32::MAX;
self.occupancy[layer] = self.occupancy[layer].saturating_sub(1);
self.free.push(slot);
failed.push(e);
}
}
failed
}
fn plan_slots(plan: Vec<(usize, Option<(usize, usize)>)>, failed: &[usize]) -> Vec<Option<u32>> {
plan.into_iter()
.map(|(expert, p)| match p {
Some((slot, _)) if !failed.contains(&expert) => Some(slot as u32),
_ => None,
})
.collect()
}
pub(crate) fn admit_frame(
&mut self,
model: &Arc<CmfModel>,
layer: usize,
reqs: &[(Vec<usize>, Vec<usize>)],
triples: &[(usize, usize, usize)],
) -> Vec<Vec<Option<u32>>> {
if triples.len() != self.n_experts {
return reqs.iter().map(|(e, _)| vec![None; e.len()]).collect();
}
let plans: Vec<_> = reqs
.iter()
.map(|(experts, protect)| self.reserve_many(layer, experts, protect))
.collect();
let mut uploads: Vec<(usize, usize)> = Vec::new();
for plan in &plans {
for &(_, p) in plan {
if let Some((slot, e)) = p.filter(|&(_, e)| e != usize::MAX) {
uploads.retain(|&(s, _)| s != slot);
uploads.push((slot, e));
}
}
}
let failed = self.upload_reserved(model, layer, &uploads, triples, true);
plans
.into_iter()
.map(|plan| Self::plan_slots(plan, &failed))
.collect()
}
pub(crate) fn prefill(
&mut self,
model: &Arc<CmfModel>,
ranked: &[(usize, usize)],
triples: &[Vec<(usize, usize, usize)>],
) -> usize {
if let Some(store) = self.store.clone() {
return self.prefill_parallel(model, ranked, triples, &store);
}
let mut n = 0;
for &(layer, expert) in ranked {
if self.free.is_empty() {
break;
}
if layer >= triples.len() || expert >= self.n_experts {
continue;
}
let key = layer * self.n_experts + expert;
if self.slot_for[key] != u32::MAX {
continue;
}
let Some(&triple) = triples[layer].get(expert) else {
continue;
};
let Some(slot) = self.free.pop() else { break };
if !crate::gpu_wgpu::dsv4_global_slot_fill(model, slot, triple) {
self.free.push(slot);
break;
}
self.owner[slot] = Some((layer, expert));
self.slot_for[key] = slot as u32;
self.occupancy[layer] += 1;
self.seen[key] = self.seen[key].max(1);
self.clock = self.clock.saturating_add(1);
self.last[slot] = self.clock;
n += 1;
}
n
}
fn prefill_parallel(
&mut self,
model: &Arc<CmfModel>,
ranked: &[(usize, usize)],
triples: &[Vec<(usize, usize, usize)>],
store: &Arc<crate::expert_store::ExpertStore>,
) -> usize {
let mut plan: Vec<(usize, usize, usize)> = Vec::new();
for &(layer, expert) in ranked {
if layer >= triples.len() || expert >= self.n_experts {
continue;
}
let key = layer * self.n_experts + expert;
if self.slot_for[key] != u32::MAX {
continue;
}
let Some(slot) = self.free.pop() else { break };
plan.push((slot, layer, expert));
}
let threads = std::thread::available_parallelism()
.map_or(8, |n| n.get())
.clamp(4, 32);
let next = std::sync::atomic::AtomicUsize::new(0);
let done: Vec<std::sync::atomic::AtomicBool> = (0..plan.len())
.map(|_| std::sync::atomic::AtomicBool::new(false))
.collect();
std::thread::scope(|scope| {
for _ in 0..threads {
scope.spawn(|| {
loop {
let i = next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let Some(&(slot, layer, expert)) = plan.get(i) else {
break;
};
let ok = store
.with_expert(layer, expert, false, |parts| {
crate::gpu_wgpu::qwen4::upload_expert_parts(
None, model, slot, parts,
)
})
.unwrap_or(false);
done[i].store(ok, std::sync::atomic::Ordering::Relaxed);
if i % 384 == 383 {
crate::gpu_wgpu::qwen4::flush_writes();
}
}
});
}
});
crate::gpu_wgpu::qwen4::flush_writes();
let mut n = 0;
for (i, &(slot, layer, expert)) in plan.iter().enumerate() {
if !done[i].load(std::sync::atomic::Ordering::Relaxed) {
self.free.push(slot);
continue;
}
let key = layer * self.n_experts + expert;
self.owner[slot] = Some((layer, expert));
self.slot_for[key] = slot as u32;
self.occupancy[layer] += 1;
self.seen[key] = self.seen[key].max(1);
self.clock = self.clock.saturating_add(1);
self.last[slot] = self.clock;
store.note_vram(layer, expert, true);
n += 1;
}
n
}
#[cfg(feature = "gpu")]
pub(crate) fn ensure_layer(
&mut self,
model: &Arc<CmfModel>,
layer: usize,
triples: &[(usize, usize, usize)],
shared: Option<(usize, usize, usize)>,
) -> Option<(Vec<u32>, u32)> {
let ne = self.n_experts;
if triples.len() != ne || self.staging_base < ne + self.shared_slot.len() + 1 {
return None;
}
let (_, shared_slot) = self.ensure(model, layer, &[], triples, shared)?;
let all: Vec<usize> = (0..ne).collect();
self.clock = self.clock.saturating_add(1);
let now = self.clock;
let base = layer * ne;
if layer == 0 {
let mut once = vec![false; self.owner.len()];
let mut live: Vec<usize> = Vec::with_capacity(self.sweep_slots.len());
for &slot in self.sweep_slots.iter().rev() {
let tagged = matches!(self.owner[slot], Some((l, e)) if e != usize::MAX && self.sweep_of[slot] as usize == l * ne + e + 1);
if tagged && !once[slot] {
once[slot] = true;
live.push(slot);
}
}
live.reverse();
self.sweep_slots = live;
}
let mut jobs: Vec<(usize, usize)> = Vec::new();
for e in 0..ne {
let s = self.slot_for[base + e];
if s != u32::MAX {
self.last[s as usize] = now;
continue;
}
let Some(v) = self.victim(layer, &all) else {
for &(slot, e2) in &jobs {
self.owner[slot] = None;
self.slot_for[base + e2] = u32::MAX;
self.occupancy[layer] = self.occupancy[layer].saturating_sub(1);
self.free.push(slot);
}
return None;
};
if let Some(old) = self.owner[v] {
self.dropped(old);
if old.1 == usize::MAX {
self.shared_slot[old.0] = u32::MAX;
} else {
self.slot_for[old.0 * ne + old.1] = u32::MAX;
self.occupancy[old.0] = self.occupancy[old.0].saturating_sub(1);
}
}
self.owner[v] = Some((layer, e));
self.slot_for[base + e] = v as u32;
self.occupancy[layer] += 1;
self.seen[base + e] = self.seen[base + e].max(1);
self.last[v] = now;
jobs.push((v, e));
}
let ok = self.run_fills(model, layer, &jobs, triples, true);
if ok.iter().any(|&x| !x) {
for (&(slot, e), &good) in jobs.iter().zip(&ok) {
if !good {
self.owner[slot] = None;
self.slot_for[base + e] = u32::MAX;
self.occupancy[layer] = self.occupancy[layer].saturating_sub(1);
self.free.push(slot);
}
}
return None;
}
for &(slot, e) in &jobs {
self.sweep_of[slot] = (base + e + 1) as u32;
self.sweep_slots.push(slot);
}
Some((self.slot_for[base..base + ne].to_vec(), shared_slot))
}
#[cfg(feature = "gpu")]
pub(crate) fn repack(
&mut self,
model: &Arc<CmfModel>,
ranked: &[(usize, usize)],
triples: &[Vec<(usize, usize, usize)>],
) -> (usize, usize) {
let ne = self.n_experts;
let pinned = self.shared_slot.iter().filter(|&&s| s != u32::MAX).count();
let room = self.staging_base.saturating_sub(pinned);
let target: Vec<(usize, usize)> = ranked
.iter()
.copied()
.filter(|&(l, e)| l < triples.len() && e < ne)
.take(room)
.collect();
let mut want = vec![false; self.slot_for.len()];
for &(l, e) in &target {
want[l * ne + e] = true;
}
let mut evicted = 0;
for slot in 0..self.staging_base {
if let Some((l, e)) = self.owner[slot]
&& e != usize::MAX
&& !want[l * ne + e]
&& !self.pin.get(l * ne + e).copied().unwrap_or(false)
{
self.dropped((l, e));
self.slot_for[l * ne + e] = u32::MAX;
self.occupancy[l] = self.occupancy[l].saturating_sub(1);
self.owner[slot] = None;
self.free.push(slot);
evicted += 1;
}
}
let now = self.clock.saturating_add(1);
self.clock = now;
let mut by_layer: Vec<Vec<(usize, usize)>> = vec![Vec::new(); triples.len()];
for &(l, e) in &target {
let key = l * ne + e;
if self.slot_for[key] != u32::MAX {
self.last[self.slot_for[key] as usize] = now;
continue;
}
let Some(slot) = self.free.pop() else { break };
self.owner[slot] = Some((l, e));
self.slot_for[key] = slot as u32;
self.occupancy[l] += 1;
self.seen[key] = self.seen[key].max(1);
self.last[slot] = now;
by_layer[l].push((slot, e));
}
let mut admitted = 0;
for (l, jobs) in by_layer.iter().enumerate() {
if jobs.is_empty() {
continue;
}
let ok = self.run_fills(model, l, jobs, &triples[l], true);
for (&(slot, e), &good) in jobs.iter().zip(&ok) {
if good {
admitted += 1;
} else {
self.owner[slot] = None;
self.slot_for[l * ne + e] = u32::MAX;
self.occupancy[l] = self.occupancy[l].saturating_sub(1);
self.free.push(slot);
}
}
}
(evicted, admitted)
}
pub(crate) fn remap_snapshot(&self, layer: usize) -> (Vec<u32>, u32) {
let base = layer * self.n_experts;
(
self.slot_for[base..base + self.n_experts].to_vec(),
self.shared_slot.get(layer).copied().unwrap_or(u32::MAX),
)
}
pub(crate) fn create_for_dsv41(
model: &Arc<CmfModel>,
inter: usize,
hidden: usize,
n_layers: usize,
n_experts: usize,
gu_q2: bool,
) -> Option<Self> {
Self::create_with_policy(
model,
inter,
hidden,
n_layers,
n_experts,
crate::gpu_wgpu::MoeBankCodec::from_gu_q2(gu_q2),
"CMF_DSV41_POOL_PCT",
75,
25,
85,
"CMF_DSV41_EXPERT_SLOTS",
"CMF_DSV41_FETCH_MAX",
"CMF_DSV41_FETCH_MIN_SEEN",
None,
0,
)
}
pub(crate) fn create_for_dsv41_codec(
model: &Arc<CmfModel>,
inter: usize,
hidden: usize,
n_layers: usize,
n_experts: usize,
codec: crate::gpu_wgpu::MoeBankCodec,
) -> Option<Self> {
Self::create_with_policy(
model,
inter,
hidden,
n_layers,
n_experts,
codec,
"CMF_DSV41_POOL_PCT",
75,
25,
85,
"CMF_DSV41_EXPERT_SLOTS",
"CMF_DSV41_FETCH_MAX",
"CMF_DSV41_FETCH_MIN_SEEN",
None,
0,
)
}
pub(crate) fn create_for_glm(
model: &Arc<CmfModel>,
inter: usize,
hidden: usize,
n_layers: usize,
n_experts: usize,
gu_q2: bool,
) -> Option<Self> {
Self::create_with_policy(
model,
inter,
hidden,
n_layers,
n_experts,
crate::gpu_wgpu::MoeBankCodec::from_gu_q2(gu_q2),
"CMF_GLM_POOL_PCT",
40,
20,
40,
"CMF_GLM_EXPERT_SLOTS",
"CMF_GLM_FETCH_MAX",
"CMF_GLM_FETCH_MIN_SEEN",
None,
0,
)
}
fn create_with_policy(
model: &Arc<CmfModel>,
inter: usize,
hidden: usize,
n_layers: usize,
n_experts: usize,
codec: crate::gpu_wgpu::MoeBankCodec,
pct_env: &str,
default_pct: usize,
min_pct: usize,
max_pct: usize,
slots_env: &str,
fetch_max_env: &'static str,
fetch_min_env: &'static str,
explicit_slots: Option<usize>,
staging: usize,
) -> Option<Self> {
if !crate::gpu_wgpu::dsv4_global_moe_supported() {
return None;
}
let gu_q2 = codec.gu_q2();
if codec == crate::gpu_wgpu::MoeBankCodec::Mxfp4 && pct_env != "CMF_DSV41_POOL_PCT" {
return None;
}
let (gu, dn) = codec.lens(inter, hidden)?;
let per = 2usize.checked_mul(gu)?.checked_add(dn)?;
let budget = crate::gpu_wgpu::dsv4_vram_budget()? as usize;
let max_pct = if pct_env == "CMF_GLM_POOL_PCT" && budget >= 20_000_000_000 {
100
} else {
max_pct
};
let default_pct = if pct_env == "CMF_GLM_POOL_PCT" && budget >= 20_000_000_000 {
100
} else {
default_pct
};
let pool_pct = bounded_pool_pct(
std::env::var(pct_env).ok().as_deref(),
default_pct,
min_pct,
max_pct,
);
let safe_requested = budget.saturating_mul(pool_pct) / 100 / per.max(1);
let bounded_override = pct_env == "CMF_GLM_POOL_PCT" || pct_env == "CMF_DSV41_POOL_PCT";
let requested = match explicit_slots {
Some(n) => n.max(1),
None => bounded_pool_slots(
std::env::var(slots_env).ok().as_deref(),
safe_requested,
bounded_override,
),
};
let requested = if pct_env == "CMF_DSV41_POOL_PCT" {
let needed = n_layers
.saturating_mul(n_experts.saturating_add(1))
.div_ceil(16)
.saturating_mul(16);
requested.min(
needed.saturating_add(
crate::gpu_wgpu::dsv4_global_moe_workspace_slots(per).unwrap_or(0),
),
)
} else {
requested
};
let (capacity, segment_slots) = if pct_env == "CMF_DSV41_POOL_PCT" {
crate::gpu_wgpu::dsv4_global_moe_create_for_dsv41_codec(
model, requested, inter, hidden, codec,
)?
} else if explicit_slots.is_some() && pct_env == "CMF_QWEN_POOL_PCT" {
crate::gpu_wgpu::dsv4_global_moe_create_s8_exact(
model, requested, inter, hidden, gu_q2,
)?
} else if pct_env == "CMF_QWEN_MTP_BANK" {
crate::gpu_wgpu::dsv4_global_moe_create_slots(model, requested, inter, hidden, gu_q2)?
} else {
crate::gpu_wgpu::dsv4_global_moe_create(model, requested, inter, hidden, gu_q2)?
};
let (auto_quota, auto_min_seen) = crate::gpu_wgpu::dsv4_fetch_defaults();
let fetch_quota = std::env::var(fetch_max_env)
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(auto_quota);
let glm_policy = pct_env == "CMF_GLM_POOL_PCT";
let fetch_min_seen = std::env::var(fetch_min_env)
.ok()
.and_then(|v| v.parse::<u16>().ok())
.unwrap_or(if glm_policy || pct_env == "CMF_DSV41_POOL_PCT" {
1
} else {
auto_min_seen.max(2)
});
if std::env::var_os("CMF_QWEN_PROF").is_some()
|| std::env::var_os("CMF_GLM_PROF").is_some()
|| std::env::var_os("CMF_DSV41_PROF").is_some()
{
eprintln!(
"dynamic-pool capacity={capacity} segment_slots={segment_slots} requested={requested} pct={pool_pct} env={pct_env} fetch_quota={fetch_quota} min_seen={fetch_min_seen} gu={}",
codec.label()
);
}
let staging = staging.min(capacity / 2);
let staging_base = capacity - staging;
Some(Self {
hold_layers: [None, None],
staging_base,
staging,
#[cfg(feature = "gpu")]
stager: None,
#[cfg(feature = "gpu")]
tier: None,
admit_pool: None,
store: None,
codec,
segment_slots,
floor: (staging_base / n_layers.max(1)).max(2),
n_experts,
fetch_quota,
fetch_min_seen,
fetch_max_env,
fetch_min_env,
owner: vec![None; capacity],
slot_for: vec![u32::MAX; n_layers.checked_mul(n_experts)?],
shared_slot: vec![u32::MAX; n_layers],
seen: vec![0; n_layers.checked_mul(n_experts)?],
seen_epoch: vec![0; n_layers.checked_mul(n_experts)?],
free: (0..staging_base).rev().collect(),
occupancy: vec![0; n_layers.max(1)],
last: vec![0; capacity],
clock: 0,
pin: Vec::new(),
sweep_slots: Vec::new(),
sweep_of: vec![0; capacity],
})
}
pub(crate) fn set_pins(&mut self, core: &[Vec<usize>]) {
let ne = self.n_experts;
self.pin = vec![false; self.slot_for.len()];
for (l, row) in core.iter().enumerate() {
for &e in row {
if let Some(p) = self.pin.get_mut(l * ne + e) {
*p = true;
}
}
}
}
pub(crate) fn begin_ensure<'pool, 'input>(
&'pool mut self,
model: &'input Arc<CmfModel>,
layer: usize,
picks: &'input [usize],
triples: &'input [(usize, usize, usize)],
shared: Option<(usize, usize, usize)>,
) -> Option<QwenGpuPendingEnsure<'pool, 'input>> {
self.clock = self.clock.saturating_add(1);
let now = self.clock;
let shared_slot = if let Some(triple) = shared {
let slot = *self.shared_slot.get(layer)?;
if slot != u32::MAX {
self.last[slot as usize] = now;
slot
} else {
let slot = self.free.pop().or_else(|| {
self.owner
.iter()
.enumerate()
.filter(|(_, o)| o.is_some_and(|(_, e)| e != usize::MAX))
.min_by_key(|(slot, _)| self.last[*slot])
.map(|(slot, _)| slot)
})?;
if !crate::gpu_wgpu::dsv4_global_slot_fill(model, slot, triple) {
if self.owner[slot].is_none() {
self.free.push(slot);
}
return None;
}
if let Some(old) = self.owner[slot] {
if old.1 == usize::MAX {
self.shared_slot[old.0] = u32::MAX;
} else {
self.slot_for[old.0 * self.n_experts + old.1] = u32::MAX;
self.occupancy[old.0] = self.occupancy[old.0].saturating_sub(1);
}
}
self.owner[slot] = Some((layer, usize::MAX));
self.shared_slot[layer] = slot as u32;
self.occupancy[layer] += 1;
self.last[slot] = now;
slot as u32
}
} else {
0
};
let epoch = (now / 64).min(u32::MAX as u64) as u32;
let base = layer.checked_mul(self.n_experts)?;
for &expert in picks {
if expert >= self.n_experts {
return None;
}
let key = base + expert;
let delta = epoch.saturating_sub(self.seen_epoch[key]);
self.seen[key] = if delta >= u16::BITS {
0
} else {
self.seen[key] >> delta
};
self.seen_epoch[key] = epoch;
self.seen[key] = self.seen[key].saturating_add(1);
let slot = self.slot_for[key];
if slot != u32::MAX {
self.last[slot as usize] = now;
self.sweep_of[slot as usize] = 0;
}
}
let quota = std::env::var(self.fetch_max_env)
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(self.fetch_quota);
let min_seen = std::env::var(self.fetch_min_env)
.ok()
.and_then(|v| v.parse::<u16>().ok())
.unwrap_or(self.fetch_min_seen);
let mut plan: Vec<(usize, usize)> = Vec::new();
for &expert in picks {
let key = base + expert;
if self.slot_for[key] != u32::MAX || plan.len() >= quota || self.seen[key] < min_seen {
continue;
}
if triples.get(expert).is_none() {
return None;
}
let victim = self.victim(layer, picks)?;
if let Some(old) = self.owner[victim] {
self.dropped(old);
if old.1 == usize::MAX {
self.shared_slot[old.0] = u32::MAX;
} else {
self.slot_for[old.0 * self.n_experts + old.1] = u32::MAX;
self.occupancy[old.0] = self.occupancy[old.0].saturating_sub(1);
}
}
self.owner[victim] = Some((layer, expert));
self.slot_for[key] = victim as u32;
self.occupancy[layer] += 1;
self.last[victim] = now;
if fill_trace() {
eprintln!(
"fill prefetch layer={layer} expert={expert} slot={victim} seen={}",
self.seen[key]
);
}
plan.push((victim, expert));
}
Some(QwenGpuPendingEnsure {
pool: Some(self),
model,
picks,
triples,
layer,
base,
shared_slot,
plan,
})
}
pub(crate) fn ensure(
&mut self,
model: &Arc<CmfModel>,
layer: usize,
picks: &[usize],
triples: &[(usize, usize, usize)],
shared: Option<(usize, usize, usize)>,
) -> Option<(Vec<u32>, u32)> {
self.begin_ensure(model, layer, picks, triples, shared)?.finish()
}
}
fn err(s: impl Into<String>) -> CmfError {
CmfError::Parse(format!("qwen4_exp: {}", s.into()))
}
fn f(model: &CmfModel, name: &str) -> Result<Vec<f32>, CmfError> {
load_f32(model, name, &Overlay::None).map_err(err)
}
fn t(model: &Arc<CmfModel>, name: &str) -> Result<QTensor, CmfError> {
load_matrix(model, name, false, &Overlay::None)
}
fn load_hc(model: &Arc<CmfModel>, prefix: &str, inject: bool) -> Result<GatedResidual, CmfError> {
Ok(GatedResidual {
norm: f(model, &format!("{prefix}hc_norm.weight"))?,
down: t(model, &format!("{prefix}input_mix_weight_down.weight"))?,
up: t(model, &format!("{prefix}input_mix_weight_up.weight"))?,
inject: inject
.then(|| t(model, &format!("{prefix}block_inject_weight.weight")))
.transpose()?,
down_idx: model.tensor_index(&format!("{prefix}input_mix_weight_down.weight")),
up_idx: model.tensor_index(&format!("{prefix}input_mix_weight_up.weight")),
inject_idx: inject
.then(|| model.tensor_index(&format!("{prefix}block_inject_weight.weight")))
.flatten(),
})
}
impl GatedResidual {
fn idxs(&self) -> Vec<usize> {
let mut v = Vec::new();
v.extend(self.down_idx);
v.extend(self.up_idx);
v.extend(self.inject_idx);
v
}
}
fn layer_idx(model: &CmfModel, p: &str, is_gdn: bool, has_ple: bool) -> Option<LayerIdx> {
let ix = |n: String| model.tensor_index(&n);
let gdn = if is_gdn {
let q = format!("{p}linear_attn.");
Some([
ix(format!("{q}in_proj_qkv.weight"))?,
ix(format!("{q}in_proj_z.weight"))?,
ix(format!("{q}in_proj_a.weight"))?,
ix(format!("{q}in_proj_b.weight"))?,
ix(format!("{q}out_proj.weight"))?,
])
} else {
None
};
let qsa = if !is_gdn {
let q = format!("{p}self_attn.");
Some([
ix(format!("{q}q_proj.weight"))?,
ix(format!("{q}k_proj.weight"))?,
ix(format!("{q}v_proj.weight"))?,
ix(format!("{q}o_proj.weight"))?,
ix(format!("{q}indexer.index_qk_proj.weight"))?,
])
} else {
None
};
let ple = if has_ple {
Some((
ix(format!("{p}ple.key_proj.weight"))?,
ix(format!("{p}ple.value_proj.weight"))?,
))
} else {
None
};
Some(LayerIdx {
gdn,
qsa,
ple,
router: ix(format!("{p}mlp.gate.weight"))?,
shared_gate: ix(format!("{p}mlp.shared_expert_gate.weight")),
})
}
fn load_gdn(model: &Arc<CmfModel>, prefix: &str) -> Result<GdnWeights, CmfError> {
Ok(GdnWeights {
in_proj_qkv: t(model, &format!("{prefix}in_proj_qkv.weight"))?,
in_proj_z: t(model, &format!("{prefix}in_proj_z.weight"))?,
in_proj_a: t(model, &format!("{prefix}in_proj_a.weight"))?,
in_proj_b: t(model, &format!("{prefix}in_proj_b.weight"))?,
conv1d: f(model, &format!("{prefix}conv1d.weight"))?,
a_log: f(model, &format!("{prefix}A_log"))?,
dt_bias: f(model, &format!("{prefix}dt_bias"))?,
norm: f(model, &format!("{prefix}norm.weight"))?,
out_proj: t(model, &format!("{prefix}out_proj.weight"))?,
})
}
fn load_qsa(model: &Arc<CmfModel>, prefix: &str) -> Result<QsaWeights, CmfError> {
Ok(QsaWeights {
q_proj: t(model, &format!("{prefix}q_proj.weight"))?,
k_proj: t(model, &format!("{prefix}k_proj.weight"))?,
v_proj: t(model, &format!("{prefix}v_proj.weight"))?,
o_proj: t(model, &format!("{prefix}o_proj.weight"))?,
q_norm: f(model, &format!("{prefix}q_norm.weight"))?,
k_norm: f(model, &format!("{prefix}k_norm.weight"))?,
index_qk: t(model, &format!("{prefix}indexer.index_qk_proj.weight"))?,
index_q_norm: f(model, &format!("{prefix}indexer.q_norm.weight"))
.or_else(|_| f(model, &format!("{prefix}indexer.q_layernorm.weight")))?,
index_k_norm: f(model, &format!("{prefix}indexer.k_norm.weight"))
.or_else(|_| f(model, &format!("{prefix}indexer.k_layernorm.weight")))?,
})
}
fn splitmix64(mut value: u64) -> u64 {
value = value.wrapping_add(SPLITMIX_GAMMA);
value = (value ^ (value >> 30)).wrapping_mul(SPLITMIX_M1);
value = (value ^ (value >> 27)).wrapping_mul(SPLITMIX_M2);
value ^ (value >> 31)
}
fn is_prime(value: usize) -> bool {
if value < 2 {
return false;
}
if value.is_multiple_of(2) {
return value == 2;
}
let mut d = 3usize;
while d <= value / d {
if value.is_multiple_of(d) {
return false;
}
d += 2;
}
true
}
fn nth_prime_after(start: usize, count: usize) -> usize {
let mut p = start;
for _ in 0..count {
p += 1;
while !is_prime(p) {
p += 1;
}
}
p
}
fn ple_tables(
qc: &Qwen4ExpConfig,
vocab: usize,
ple_index: usize,
) -> (Vec<i64>, Vec<i64>, Vec<i64>) {
let max_mul = i64::MAX as u64 / vocab.max(1) as u64;
let half_bound = (max_mul / 2).max(1);
let base = qc.seed.wrapping_add(PRIME_1.wrapping_mul(ple_index as u64));
let multipliers = (0..qc.ngram_size)
.map(|i| {
let v = base.wrapping_add(SPLITMIX_GAMMA.wrapping_mul((i + 1) as u64));
(2 * (splitmix64(v) % half_bound) + 1) as i64
})
.collect();
let heads = (qc.ngram_size - 1) * qc.heads_per_ngram;
let mut sizes = Vec::with_capacity(heads);
let mut offsets = Vec::with_capacity(heads);
let mut off = 0i64;
for head in 0..heads {
let global = ple_index * heads + head;
let sz = nth_prime_after(qc.ngram_vocab_size_base - 1, global + 1) as i64;
sizes.push(sz);
offsets.push(off);
off += sz;
}
(multipliers, sizes, offsets)
}
fn load_ple(
model: &Arc<CmfModel>,
prefix: &str,
qc: &Qwen4ExpConfig,
vocab: usize,
ple_index: usize,
) -> Result<PleWeights, CmfError> {
let mut shards = Vec::with_capacity(qc.split_ngram_parts);
for si in 0..qc.split_ngram_parts {
shards.push(t(
model,
&format!("{prefix}ple_embedding.ngram_embedding.shard_{si}.weight"),
)?);
}
let first = shards
.first()
.ok_or_else(|| err("PLE has no embedding shards"))?;
let (rows_per_shard, row_dim) = (first.rows(), first.cols());
if shards
.iter()
.any(|s| s.rows() != rows_per_shard || s.cols() != row_dim)
{
return Err(err("PLE embedding shard shapes disagree"));
}
let (multipliers, vocab_sizes, offsets) = ple_tables(qc, vocab, ple_index);
Ok(PleWeights {
shards,
rows_per_shard,
row_dim,
key_proj: t(model, &format!("{prefix}key_proj.weight"))?,
value_proj: t(model, &format!("{prefix}value_proj.weight"))?,
norm_key: f(model, &format!("{prefix}norm_key.weight"))?,
norm_query: f(model, &format!("{prefix}norm_query.weight"))?,
norm_conv: f(model, &format!("{prefix}norm_conv.weight"))?,
conv: f(model, &format!("{prefix}conv1d.weight"))?,
multipliers,
vocab_sizes,
offsets,
})
}
pub fn load(
model: &Arc<CmfModel>,
arch: &ModelArch,
) -> Result<(Globals, Vec<Layer>, Cfg, State), CmfError> {
let qc = arch
.qwen4_exp
.as_ref()
.ok_or_else(|| err("missing qwen4_exp descriptor"))?;
if qc.hc_count == 0 || qc.indexer_compress_ratio == 0 || qc.ngram_size < 2 {
return Err(err("invalid zero geometry"));
}
let gdn = GdnCfg {
num_v_heads: arch.linear_num_value_heads.unwrap_or(48),
num_k_heads: arch.linear_num_key_heads.unwrap_or(16),
key_head_dim: arch.linear_key_head_dim.unwrap_or(128),
value_head_dim: arch.linear_value_head_dim.unwrap_or(128),
conv_kernel: arch.linear_conv_kernel_dim.unwrap_or(4),
hidden_size: arch.hidden_size,
rms_eps: arch.rms_norm_eps,
output_gate_sigmoid: true,
};
let eos = model
.header
.tokenizer_config
.as_ref()
.and_then(|tc| {
tc.pad_token_id
.or(tc.bos_token_id)
.or_else(|| tc.eos_token_ids.last().copied())
})
.unwrap_or(248_044);
let cfg = Cfg {
hidden: arch.hidden_size,
hc: qc.hc_count,
eps: arch.rms_norm_eps,
n_heads: arch.num_attention_heads,
n_kv_heads: arch.num_kv_heads,
head_dim: arch.head_dim,
rotary_dim: ((arch.head_dim as f32 * arch.partial_rotary_factor) as usize).max(2),
index_heads: qc.indexer_n_heads,
index_kv_heads: qc.indexer_kv_heads,
index_dim: qc.indexer_head_dim,
index_budget: qc.indexer_budget,
compress_ratio: qc.indexer_compress_ratio,
gdn,
ngram_size: qc.ngram_size,
heads_per_ngram: qc.heads_per_ngram,
ple_kernel: qc.ple_conv_kernel_size,
ple_dilation: qc.ngram_size,
eos,
};
if cfg.n_heads % cfg.n_kv_heads != 0 || cfg.index_kv_heads != 1 {
return Err(err("unsupported QSA head grouping"));
}
let head_hc = load_hc(model, "model.hyper_connection_mixer.", false)?;
let lm_head_idx = model.tensor_index("lm_head.weight");
let embed_idx = model.tensor_index("model.embed_tokens.weight");
let mut skeleton_idxs: Vec<usize> = head_hc.idxs();
skeleton_idxs.extend(lm_head_idx);
if std::env::var("CMF_QWEN_MTP").as_deref() != Ok("0")
&& std::env::var("CMF_QWEN_MTP_DEVICE_DRAFT").as_deref() == Ok("1")
{
let side = cortiq_core::mtp_sidecar_path(&model.path);
if mtp_embedded(model) || (side != model.path && side.exists()) {
skeleton_idxs.extend(embed_idx);
}
}
let globals_partial = (
t(model, "model.embed_tokens.weight")?,
t(model, "lm_head.weight")?,
head_hc,
);
let mut layers = Vec::with_capacity(arch.num_layers);
let mut ple_index = 0usize;
for li in 0..arch.num_layers {
let p = format!("model.layers.{li}.");
let mixer = match arch.layer_types.get(li) {
Some(LayerType::LinearAttention) => {
Mixer::Gdn(load_gdn(model, &format!("{p}linear_attn."))?)
}
_ => Mixer::Qsa(load_qsa(model, &format!("{p}self_attn."))?),
};
let ple = if qc.ple_layer_ids.contains(&(li + 1)) {
let w = load_ple(model, &format!("{p}ple."), qc, arch.vocab_size, ple_index)?;
ple_index += 1;
Some(w)
} else {
None
};
let moe = match build_ffn_at(model, arch, &p, false, &Overlay::None)? {
FfnKind::Moe(m) => m,
_ => return Err(err(format!("layer {li} is not MoE"))),
};
let expert_ids: Vec<_> = moe
.experts
.iter()
.map(|e| {
Some((
e.gate_proj.model_idx()?,
e.up_proj.model_idx()?,
e.down_proj.model_idx()?,
))
})
.collect::<Option<_>>()
.ok_or_else(|| err(format!("layer {li} experts are not mmap-backed")))?;
let shared_ids = moe
.shared
.as_ref()
.and_then(|(e, _)| {
Some((
e.gate_proj.model_idx()?,
e.up_proj.model_idx()?,
e.down_proj.model_idx()?,
))
})
.ok_or_else(|| err(format!("layer {li} shared expert is not mmap-backed")))?;
let attn_hc = load_hc(model, &format!("{p}attn_hyper_connection."), true)?;
let mlp_hc = load_hc(model, &format!("{p}mlp_hyper_connection."), true)?;
let idx = layer_idx(model, &p, matches!(mixer, Mixer::Gdn(_)), ple.is_some());
skeleton_idxs.extend(attn_hc.idxs());
skeleton_idxs.extend(mlp_hc.idxs());
if let Some(ix) = &idx {
skeleton_idxs.extend(ix.gdn.iter().flatten().copied());
skeleton_idxs.extend(ix.qsa.iter().flatten().copied());
if let Some((k, v)) = ix.ple {
skeleton_idxs.extend([k, v]);
}
skeleton_idxs.push(ix.router);
skeleton_idxs.extend(ix.shared_gate);
}
layers.push(Layer {
attn_hc,
mlp_hc,
mixer,
moe,
ple,
expert_ids,
shared_ids: Some(shared_ids),
idx,
});
}
let globals = Globals {
embed: globals_partial.0,
lm_head: globals_partial.1,
head_hc: globals_partial.2,
lm_head_idx,
embed_idx,
skeleton_idxs,
};
let state = State::new(layers.len());
Ok((globals, layers, cfg, state))
}
#[inline]
fn sigmoid(x: f32) -> f32 {
1.0 / (1.0 + (-x).exp())
}
#[inline]
fn silu(x: f32) -> f32 {
x * sigmoid(x)
}
fn group_rms_zero_into(x: &[f32], weight: &[f32], group: usize, eps: f64, out: &mut [f32]) {
debug_assert_eq!(x.len(), weight.len());
debug_assert_eq!(x.len(), out.len());
debug_assert!(group > 0 && x.len().is_multiple_of(group));
for (gi, chunk) in x.chunks(group).enumerate() {
let ss = chunk.iter().map(|&v| (v as f64) * (v as f64)).sum::<f64>();
let inv = (ss / group as f64 + eps).sqrt().recip() as f32;
let off = gi * group;
for j in 0..group {
out[off + j] = chunk[j] * inv * (1.0 + weight[off + j]);
}
}
}
fn group_rms_zero(x: &[f32], weight: &[f32], group: usize, eps: f64) -> Vec<f32> {
let mut out = vec![0.0f32; x.len()];
group_rms_zero_into(x, weight, group, eps, &mut out);
out
}
fn hc_mix(
w: &GatedResidual,
hyper: &[f32],
cfg: &Cfg,
pool: Option<&Pool>,
) -> (Vec<f32>, Option<Vec<f32>>) {
let mut normed = crate::attention::take_buf(hyper.len());
group_rms_zero_into(hyper, &w.norm, cfg.hidden, cfg.eps, &mut normed);
let mut low = crate::attention::take_buf(w.down.rows());
let mut inject = w.inject.as_ref().map(|iw| vec![0.0f32; iw.rows()]);
match (&w.inject, inject.as_mut()) {
(Some(iw), Some(inj)) => {
QTensor::matvec_many([&w.down, iw], &normed, [&mut low, inj], pool)
}
_ => w.down.matvec(&normed, &mut low, pool),
}
for v in &mut low {
*v = silu(*v / cfg.hc as f32);
}
let mut mix = crate::attention::take_buf(hyper.len());
w.up.matvec(&low, &mut mix, pool);
let mut folded = vec![0.0f32; cfg.hidden];
for stream in 0..cfg.hc {
let off = stream * cfg.hidden;
for d in 0..cfg.hidden {
folded[d] += sigmoid(mix[off + d]) * normed[off + d] / cfg.hc as f32;
}
}
let inject = inject.map(|mut v| {
for x in &mut v {
*x = 2.0 * sigmoid(*x / cfg.hc as f32);
}
v
});
crate::attention::recycle_buf(&mut mix);
crate::attention::recycle_buf(&mut low);
crate::attention::recycle_buf(&mut normed);
(folded, inject)
}
fn inject(hyper: &mut [f32], block: &[f32], weights: &[f32], cfg: &Cfg) {
for stream in 0..cfg.hc {
let off = stream * cfg.hidden;
for d in 0..cfg.hidden {
hyper[off + d] += weights[stream] * block[d];
}
}
}
fn trace_stats(label: &str, li: usize, position: usize, values: &[f32]) {
if std::env::var_os("CMF_QWEN_TRACE").is_none() {
return;
}
let mut sumsq = 0.0f64;
let mut max = 0.0f32;
let mut finite = 0usize;
for &v in values {
if v.is_finite() {
sumsq += (v as f64) * (v as f64);
max = max.max(v.abs());
finite += 1;
}
}
let rms = if finite == 0 {
f64::NAN
} else {
(sumsq / finite as f64).sqrt()
};
eprintln!(
"qwen4_exp pos={position} layer={li:02} {label}: rms={rms:.6} max={max:.6} finite={finite}/{}",
values.len()
);
}
fn dump_values(label: &str, li: usize, position: usize, values: &[f32]) {
let Some(root) = std::env::var_os("CMF_QWEN_DUMP") else {
return;
};
if let Ok(wanted) = std::env::var("CMF_QWEN_DUMP_LAYER")
&& wanted.parse::<usize>().ok() != Some(li)
{
return;
}
let root = std::path::PathBuf::from(root);
if std::fs::create_dir_all(&root).is_err() {
return;
}
let bytes = unsafe {
std::slice::from_raw_parts(values.as_ptr().cast::<u8>(), std::mem::size_of_val(values))
};
let _ = std::fs::write(
root.join(format!("p{position:06}_l{li:02}_{label}.f32")),
bytes,
);
}
static TAP: std::sync::Mutex<Vec<(String, usize, usize, Vec<f32>)>> =
std::sync::Mutex::new(Vec::new());
fn tap_layer() -> Option<usize> {
std::env::var("CMF_QWEN_DEVICE_TAP").ok()?.parse().ok()
}
#[cfg(feature = "cuda")]
thread_local! {
static CUDA_TAP: std::cell::RefCell<Option<Vec<(String, usize, Vec<f32>)>>> = const { std::cell::RefCell::new(None) };
}
fn observe(label: &str, li: usize, position: usize, values: &[f32]) {
#[cfg(feature = "cuda")]
CUDA_TAP.with(|t| {
if let Some(rows) = t.borrow_mut().as_mut() {
rows.push((label.to_string(), li, values.to_vec()));
}
});
if tap_layer() == Some(li) && position < 3 {
TAP.lock()
.unwrap()
.push((label.to_string(), li, position, values.to_vec()));
}
trace_stats(label, li, position, values);
dump_values(label, li, position, values);
}
fn rms_zero_head(v: &mut [f32], weight: &[f32], head_dim: usize, eps: f64) {
for head in v.chunks_mut(head_dim) {
let ss = head.iter().map(|&x| (x as f64) * (x as f64)).sum::<f64>();
let inv = (ss / head_dim as f64 + eps).sqrt().recip() as f32;
for (x, &w) in head.iter_mut().zip(weight) {
*x *= inv * (1.0 + w);
}
}
}
fn rope(v: &mut [f32], pos: usize, inv_freq: &[f32], rotary_dim: usize) {
let rd = rotary_dim.min(v.len()).min(inv_freq.len() * 2);
let half = rd / 2;
for i in 0..half {
let a = (pos as f32 * inv_freq[i]).cos();
let b = (pos as f32 * inv_freq[i]).sin();
let x1 = v[i];
let x2 = v[i + half];
v[i] = x1 * a - x2 * b;
v[i + half] = x2 * a + x1 * b;
}
}
fn selected_tokens(
q: &[f32],
raw_keys: &[f32],
npos: usize,
w: &QsaWeights,
cfg: &Cfg,
inv_freq: &[f32],
) -> Vec<usize> {
let cr = cfg.compress_ratio;
let complete = npos / cr;
let mut scores = Vec::with_capacity(complete);
for block in 0..complete {
let mut k = vec![0.0f32; cfg.index_dim];
for ti in block * cr..(block + 1) * cr {
let src = &raw_keys[ti * cfg.index_dim..(ti + 1) * cfg.index_dim];
for (d, &x) in src.iter().enumerate() {
k[d] += x / cr as f32;
}
}
rms_zero_head(&mut k, &w.index_k_norm, cfg.index_dim, cfg.eps);
rope(
&mut k,
block * cr,
inv_freq,
cfg.rotary_dim.min(cfg.index_dim),
);
let mut s = 0.0f32;
for h in 0..cfg.index_heads {
let qh = &q[h * cfg.index_dim..(h + 1) * cfg.index_dim];
let dot = qh.iter().zip(&k).map(|(&a, &b)| a * b).sum::<f32>();
s += dot.max(0.0);
}
scores.push((block, s / (cfg.index_dim as f32).sqrt()));
}
let keep = (cfg.index_budget / cr).min(complete);
if complete > keep {
scores.select_nth_unstable_by(keep, |a, b| {
b.1.partial_cmp(&a.1).unwrap_or(Ordering::Equal)
});
scores.truncate(keep);
}
let mut out = Vec::with_capacity(keep * cr + cr.saturating_sub(1));
for (block, _) in scores {
out.extend(block * cr..(block + 1) * cr);
}
out.extend(complete * cr..npos);
out
}
fn qsa_forward(
x: &[f32],
w: &QsaWeights,
cfg: &Cfg,
st: &mut QsaState,
position: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
) -> Vec<f32> {
let mut iqk =
crate::attention::take_buf((cfg.index_heads + cfg.index_kv_heads) * cfg.index_dim);
let mut qg = crate::attention::take_buf(cfg.n_heads * cfg.head_dim * 2);
let mut k = crate::attention::take_buf(cfg.n_kv_heads * cfg.head_dim);
let mut v = crate::attention::take_buf(cfg.n_kv_heads * cfg.head_dim);
QTensor::matvec_many(
[&w.index_qk, &w.q_proj, &w.k_proj, &w.v_proj],
x,
[&mut iqk, &mut qg, &mut k, &mut v],
pool,
);
let qlen = cfg.index_heads * cfg.index_dim;
let mut iq = crate::attention::take_buf(qlen);
iq.copy_from_slice(&iqk[..qlen]);
rms_zero_head(&mut iq, &w.index_q_norm, cfg.index_dim, cfg.eps);
for h in 0..cfg.index_heads {
rope(
&mut iq[h * cfg.index_dim..(h + 1) * cfg.index_dim],
position,
inv_freq,
cfg.rotary_dim.min(cfg.index_dim),
);
}
st.raw_keys.extend_from_slice(&iqk[qlen..]);
let mut q = crate::attention::take_buf(cfg.n_heads * cfg.head_dim);
let mut gate = crate::attention::take_buf(q.len());
for h in 0..cfg.n_heads {
let src = h * cfg.head_dim * 2;
let dst = h * cfg.head_dim;
q[dst..dst + cfg.head_dim].copy_from_slice(&qg[src..src + cfg.head_dim]);
gate[dst..dst + cfg.head_dim]
.copy_from_slice(&qg[src + cfg.head_dim..src + cfg.head_dim * 2]);
}
rms_zero_head(&mut q, &w.q_norm, cfg.head_dim, cfg.eps);
rms_zero_head(&mut k, &w.k_norm, cfg.head_dim, cfg.eps);
for h in 0..cfg.n_heads {
rope(
&mut q[h * cfg.head_dim..(h + 1) * cfg.head_dim],
position,
inv_freq,
cfg.rotary_dim,
);
}
for h in 0..cfg.n_kv_heads {
rope(
&mut k[h * cfg.head_dim..(h + 1) * cfg.head_dim],
position,
inv_freq,
cfg.rotary_dim,
);
}
st.keys.extend_from_slice(&k);
st.values.extend_from_slice(&v);
let npos = position + 1;
let selected = selected_tokens(&iq, &st.raw_keys, npos, w, cfg, inv_freq);
let groups = cfg.n_heads / cfg.n_kv_heads;
let scale = (cfg.head_dim as f32).sqrt().recip();
let mut merged = crate::attention::take_buf(cfg.n_heads * cfg.head_dim);
for qh in 0..cfg.n_heads {
let kvh = qh / groups;
let qs = &q[qh * cfg.head_dim..(qh + 1) * cfg.head_dim];
let mut scores = Vec::with_capacity(selected.len());
let mut max = f32::NEG_INFINITY;
for &ti in &selected {
let ko = (ti * cfg.n_kv_heads + kvh) * cfg.head_dim;
let s = qs
.iter()
.zip(&st.keys[ko..ko + cfg.head_dim])
.map(|(&a, &b)| a * b)
.sum::<f32>()
* scale;
max = max.max(s);
scores.push(s);
}
let z = scores.iter().map(|&s| (s - max).exp()).sum::<f32>();
let out = &mut merged[qh * cfg.head_dim..(qh + 1) * cfg.head_dim];
for (&ti, &score) in selected.iter().zip(&scores) {
let p = (score - max).exp() / z.max(f32::MIN_POSITIVE);
let vo = (ti * cfg.n_kv_heads + kvh) * cfg.head_dim;
for d in 0..cfg.head_dim {
out[d] += p * st.values[vo + d];
}
}
let go = qh * cfg.head_dim;
for d in 0..cfg.head_dim {
out[d] *= sigmoid(gate[go + d]);
}
}
let mut out = vec![0.0f32; cfg.hidden];
w.o_proj.matvec(&merged, &mut out, pool);
crate::attention::recycle_buf(&mut merged);
crate::attention::recycle_buf(&mut gate);
crate::attention::recycle_buf(&mut q);
crate::attention::recycle_buf(&mut iq);
crate::attention::recycle_buf(&mut v);
crate::attention::recycle_buf(&mut k);
crate::attention::recycle_buf(&mut qg);
crate::attention::recycle_buf(&mut iqk);
out
}
fn shifted_token(history: &[u32], current: u32, shift: usize, eos: u32) -> u32 {
if shift == 0 {
return current;
}
if history.len() < shift {
return eos;
}
let source = history.len() - shift;
if history[source + 1..].contains(&eos) || history.last() == Some(&eos) {
eos
} else {
history[source]
}
}
fn ple_embedding(w: &PleWeights, cfg: &Cfg, history: &[u32], token: u32) -> Vec<f32> {
let ids = ple_row_ids(w, cfg, history, token);
let mut out = vec![0.0f32; ids.len() * w.row_dim];
for (hi, &id) in ids.iter().enumerate() {
ple_row(w, id, &mut out[hi * w.row_dim..(hi + 1) * w.row_dim]);
}
out
}
fn ple_row(w: &PleWeights, id: i64, dst: &mut [f32]) {
let global = id as usize;
let shard = global / w.rows_per_shard;
let local = global % w.rows_per_shard;
debug_assert!(shard < w.shards.len());
w.shards[shard].row_f32(local, dst);
}
fn ple_row_ids(w: &PleWeights, cfg: &Cfg, history: &[u32], token: u32) -> Vec<i64> {
let shifted: Vec<i64> = (0..cfg.ngram_size)
.map(|s| shifted_token(history, token, s, cfg.eos) as i64)
.collect();
let mut ids = Vec::with_capacity((cfg.ngram_size - 1) * cfg.heads_per_ngram);
for ngram in 2..=cfg.ngram_size {
let mut mixed = shifted[0].wrapping_mul(w.multipliers[0]);
for p in 1..ngram {
mixed ^= shifted[p].wrapping_mul(w.multipliers[p]);
}
let h0 = (ngram - 2) * cfg.heads_per_ngram;
for hi in h0..h0 + cfg.heads_per_ngram {
ids.push(mixed.rem_euclid(w.vocab_sizes[hi]) + w.offsets[hi]);
}
}
ids
}
#[cfg(feature = "gpu")]
fn ple_prefetch_threads() -> usize {
static N: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
*N.get_or_init(|| {
std::env::var("CMF_QWEN_PLE_PREFETCH")
.ok()
.and_then(|v| v.trim().parse::<usize>().ok())
.unwrap_or(32)
.min(256)
})
}
#[cfg(feature = "gpu")]
fn ple_prefetch(layers: &[Layer], cfg: &Cfg, history: &[u32], ids: &[u32]) -> usize {
#[cfg(any(target_os = "linux", target_os = "android", target_os = "macos"))]
{
let threads = ple_prefetch_threads();
if threads == 0 || ids.len() < 64 {
return 0;
}
let t0 = std::time::Instant::now();
let mut spans: Vec<(usize, usize)> = Vec::new();
let mut model = None;
let mut hist: Vec<u32> = history.to_vec();
for l in layers {
let Some(pw) = l.ple.as_ref() else { continue };
if model.is_none() {
model = pw.shards.first().and_then(|s| s.model_arc());
}
hist.truncate(history.len());
for &id in ids {
for r in ple_row_ids(pw, cfg, &hist, id) {
let global = r as usize;
if let Some(s) = pw.shards.get(global / pw.rows_per_shard) {
s.row_spans(global % pw.rows_per_shard, &mut spans);
}
}
hist.push(id);
}
}
let Some(model) = model else { return 0 };
if spans.is_empty() {
return 0;
}
let (rows, ntok) = (spans.len() / 3, ids.len());
let prof = std::env::var_os("CMF_QWEN_PROF").is_some();
let _ = std::thread::Builder::new().name("ple-readahead".into()).spawn(move || {
let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }.max(4096) as usize;
let mut pages: Vec<usize> = Vec::with_capacity(spans.len() + spans.len() / 8);
for &(a, len) in &spans {
pages.extend(a / page..=(a + len - 1) / page);
}
pages.sort_unstable();
pages.dedup();
let mut runs: Vec<(usize, usize)> = Vec::with_capacity(pages.len());
for &p in &pages {
match runs.last_mut() {
Some((s, n)) if *s + *n == p => *n += 1,
_ => runs.push((p, 1)),
}
}
let threads = threads.min(runs.len().div_ceil(512)).max(1);
let runs = &runs;
std::thread::scope(|sc| {
for t in 0..threads {
sc.spawn(move || {
for &(p, n) in runs.iter().skip(t).step_by(threads) {
unsafe {
libc::madvise(
(p * page) as *mut libc::c_void,
n * page,
libc::MADV_WILLNEED,
);
}
}
});
}
});
if prof {
eprintln!(
"qwen4-ple: read-ahead of {} pages in {} ranges for {ntok} tokens on {threads} threads, issued in {:.1}ms",
pages.len(),
runs.len(),
t0.elapsed().as_secs_f64() * 1e3
);
}
drop(model);
});
rows
}
#[cfg(not(any(target_os = "linux", target_os = "android", target_os = "macos")))]
{
let _ = (layers, cfg, history, ids);
0
}
}
fn ple_frame_rows(
layers: &[Layer],
cfg: &Cfg,
history: &[u32],
ids: &[u32],
pool: Option<&Pool>,
) -> Vec<Vec<Vec<f32>>> {
ple_rows_of(layers, cfg, history, ids, pool, None)
}
fn ple_rows_of(
layers: &[Layer],
cfg: &Cfg,
history: &[u32],
ids: &[u32],
pool: Option<&Pool>,
only: Option<usize>,
) -> Vec<Vec<Vec<f32>>> {
let mut out: Vec<Vec<Vec<f32>>> = Vec::with_capacity(layers.len());
let mut jobs: Vec<(usize, usize, usize, i64)> = Vec::new();
let mut hist: Vec<u32> = history.to_vec();
for (li, l) in layers.iter().enumerate() {
let Some(pw) = l.ple.as_ref().filter(|_| only.is_none_or(|o| o == li)) else {
out.push(Vec::new());
continue;
};
hist.truncate(history.len());
let mut rows = Vec::with_capacity(ids.len());
for (t, &id) in ids.iter().enumerate() {
let rid = ple_row_ids(pw, cfg, &hist, id);
rows.push(vec![0.0f32; rid.len() * pw.row_dim]);
jobs.extend(rid.into_iter().enumerate().map(|(h, r)| (li, t, h, r)));
hist.push(id);
}
out.push(rows);
}
let tables: Vec<Option<&PleWeights>> = layers.iter().map(|l| l.ple.as_ref()).collect();
let dst: Vec<crate::pool::SendMut> = jobs
.iter()
.map(|&(li, t, h, _)| {
let dim = tables[li].map_or(0, |w| w.row_dim);
crate::pool::SendMut::new(out[li][t][h * dim..].as_mut_ptr())
})
.collect();
let read = |s: usize, e: usize| {
for j in s..e {
let (li, _, _, id) = jobs[j];
let Some(w) = tables[li] else {
continue;
};
let row = unsafe { std::slice::from_raw_parts_mut(dst[j].at(0), w.row_dim) };
ple_row(w, id, row);
}
};
match pool {
Some(p) if jobs.len() > 1 => p.run_rows(jobs.len(), &read),
_ => read(0, jobs.len()),
}
out
}
fn ple_forward(
hyper: &[f32],
token: u32,
history: &[u32],
w: &PleWeights,
cfg: &Cfg,
st: &mut LayerState,
pool: Option<&Pool>,
) -> Vec<f32> {
let emb = ple_embedding(w, cfg, history, token);
let mut key = vec![0.0f32; cfg.hc * cfg.hidden];
let mut value = vec![0.0f32; cfg.hidden];
w.key_proj.matvec(&emb, &mut key, pool);
w.value_proj.matvec(&emb, &mut value, pool);
let key = group_rms_zero(&key, &w.norm_key, cfg.hidden, cfg.eps);
let query = group_rms_zero(hyper, &w.norm_query, cfg.hidden, cfg.eps);
let mut gated = vec![0.0f32; cfg.hc * cfg.hidden];
for stream in 0..cfg.hc {
let off = stream * cfg.hidden;
let dot = key[off..off + cfg.hidden]
.iter()
.zip(&query[off..off + cfg.hidden])
.map(|(&a, &b)| a * b)
.sum::<f32>()
/ (cfg.hidden as f32).sqrt();
let signed_root = dot.signum() * dot.abs().max(1e-6).sqrt();
let g = sigmoid(signed_root);
for d in 0..cfg.hidden {
gated[off + d] = g * value[d];
}
}
let normed = group_rms_zero(&gated, &w.norm_conv, cfg.hidden, cfg.eps);
let hist_cap = (cfg.ple_kernel - 1) * cfg.ple_dilation;
let width = gated.len();
let mut conv = vec![0.0f32; width];
for channel in 0..width {
let mut sum = w.conv[channel * cfg.ple_kernel + cfg.ple_kernel - 1] * normed[channel];
for tap in 0..cfg.ple_kernel - 1 {
let lag = (cfg.ple_kernel - 1 - tap) * cfg.ple_dilation;
if lag <= st.ple_history_rows {
let row = st.ple_history_rows - lag;
sum +=
w.conv[channel * cfg.ple_kernel + tap] * st.ple_history[row * width + channel];
}
}
conv[channel] = silu(sum);
}
if hist_cap > 0 {
if st.ple_history_rows < hist_cap {
st.ple_history.extend_from_slice(&normed);
st.ple_history_rows += 1;
} else {
st.ple_history.copy_within(width.., 0);
let off = (hist_cap - 1) * width;
st.ple_history[off..off + width].copy_from_slice(&normed);
}
}
for (o, &g) in conv.iter_mut().zip(&gated) {
*o += g;
}
conv
}
#[cfg(feature = "gpu")]
fn dynamic_moe_gpu(
layer: &Layer,
li: usize,
x: &[f32],
state: &mut State,
pool: Option<&Pool>,
) -> Option<(Vec<f32>, Vec<usize>)> {
#[cfg(feature = "cuda")]
if crate::gpu_cuda::selected() {
return None;
}
let m = &layer.moe;
if !crate::gpu::enabled_here()
|| m.router_sigmoid
|| !m.norm_topk_prob
|| (m.routed_scaling - 1.0).abs() > 1e-9
|| m.per_expert_scale.is_some()
|| m.route_tau.is_some()
|| m.mask.is_some()
{
return None;
}
let model = m.experts.first()?.gate_proj.model_arc()?;
let gu_q2 = m
.experts
.first()
.is_some_and(|e| e.gate_proj.model_dtype() == Some(TensorDtype::Q2TiledP));
let dynamic_mode = std::env::var("CMF_QWEN_DYNAMIC_MOE").ok();
let dynamic_enabled = match dynamic_mode.as_deref() {
Some("1") => true,
Some("0") => false,
None | Some("auto") => {
gu_q2
&& crate::gpu_wgpu::dsv4_global_moe_supported()
&& crate::gpu_wgpu::dsv4_vram_budget().is_some_and(|b| b >= 14_000_000_000)
}
Some(_) => false,
};
if !dynamic_enabled {
return None;
}
if state.gpu_pool.is_none() {
state.gpu_pool = QwenGpuPool::create(
&model,
m.experts.first()?.gate_proj.rows(),
x.len(),
state.layers.len(),
m.experts.len(),
gu_q2,
);
}
let mut logits = vec![0.0f32; m.experts.len()];
m.router.matvec(x, &mut logits, pool);
let (picks, probabilities, wsum) = crate::pipeline::moe_route(&logits, m, None);
let (remap, shared_slot) =
state
.gpu_pool
.as_mut()?
.ensure(&model, li, &picks, &layer.expert_ids, layer.shared_ids)?;
let shared_weight = m.shared.as_ref().map_or(1.0, |(_, gate)| {
gate.as_ref().map_or(1.0, |gate| {
let mut y = [0.0f32; 1];
gate.matvec(x, &mut y, pool);
sigmoid(y[0])
})
});
let has_shared = m.shared.is_some();
let mut mix_weights = vec![0.0f32; m.experts.len()];
for &expert in &picks {
mix_weights[expert] =
probabilities[expert] / wsum * m.per_expert_scale.as_ref().map_or(1.0, |v| v[expert]);
}
let cold_ids: Vec<_> = picks
.iter()
.copied()
.filter(|&expert| remap[expert] == u32::MAX)
.collect();
let cold_jobs: Vec<_> = cold_ids
.iter()
.copied()
.map(|expert| (&m.experts[expert], mix_weights[expert]))
.collect();
let gp = state.gpu_pool.as_ref()?;
let weights = crate::gpu_wgpu::Dsv4MoeW {
router: &[],
experts: &layer.expert_ids,
logits: &logits,
bias: Some(&mix_weights),
mask: None,
forced: Some(&picks),
remap: Some(&remap),
global: Some(crate::gpu_wgpu::Dsv4GlobalMoe {
pool_uid: model.uid(),
shared_slot,
segment_slots: gp.segment_slots as u32,
}),
has_shared,
shared_weight,
preweighted: true,
qwen_softmax: true,
};
let geom = crate::gpu_wgpu::Dsv4MoeGeom {
hidden: x.len(),
inter: m.experts.first()?.gate_proj.rows(),
top_k: m.top_k,
route_scale: 1.0,
swiglu_limit: 0.0,
gu_q2,
bf16: false,
};
let mut out = vec![0.0f32; x.len()];
let mut cold = Vec::new();
let mut cold_x = Vec::new();
let (frame_ok, mut cold_cpu) = std::thread::scope(|scope| {
let cpu = (!cold_jobs.is_empty())
.then(|| scope.spawn(|| crate::pipeline::moe_cold_experts_cpu_with(&cold_jobs, x, pool, false)));
let ok = crate::gpu_wgpu::dsv4_moe_frame(
&model,
&weights,
geom,
x,
&mut cold,
&mut cold_x,
None,
None,
&mut out,
);
let early = cpu
.map(|job| job.join().ok())
.flatten()
.unwrap_or_else(|| vec![0.0; x.len()]);
(ok, early)
});
if !frame_ok
|| cold.len() != cold_jobs.len()
|| cold.iter().map(|&(e, _)| e).ne(cold_ids.iter().copied())
{
return None;
}
for (o, c) in out.iter_mut().zip(&cold_cpu) {
*o += c;
}
crate::attention::recycle_buf(&mut cold_cpu);
for &e in &picks {
crate::pipeline::expert_tubes_add(&m.experts[e], x, mix_weights[e], &mut out, pool);
}
if let Some((se, _)) = &m.shared {
crate::pipeline::expert_tubes_add(se, x, shared_weight, &mut out, pool);
}
Some((out, picks))
}
pub fn forward_token(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
token_id: u32,
position: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
logits: &mut Vec<f32>,
want_logits: bool,
) {
#[cfg(feature = "cuda")]
if cuda_forward(globals, layers, cfg, state, &[token_id], position, inv_freq, pool, logits, want_logits) {
return;
}
#[cfg(feature = "gpu")]
if forward_token_device(
globals,
layers,
cfg,
state,
token_id,
position,
inv_freq,
pool,
logits,
want_logits,
) {
return;
}
let prof = std::env::var_os("CMF_QWEN_PROF").is_some();
#[cfg(target_arch = "x86_64")]
if position == 0
&& std::env::var_os("CMF_POOL_SPIN").is_none()
&& let Some(workers) = pool
{
workers.set_spin_budget((workers.n_workers() * 10_000).clamp(30_000, 200_000));
}
#[cfg(feature = "gpu")]
let gpu_moe_before = if prof {
use std::sync::atomic::Ordering;
Some((
crate::gpu_wgpu::MOE_ENC_NS.load(Ordering::Relaxed),
crate::gpu_wgpu::MOE_WAIT_NS.load(Ordering::Relaxed),
crate::gpu_wgpu::MOE_GPU_NS[0].load(Ordering::Relaxed),
crate::gpu_wgpu::MOE_GPU_N.load(Ordering::Relaxed),
crate::gpu_wgpu::DSV4_FILLS.load(Ordering::Relaxed),
crate::gpu_wgpu::DSV4_FILL_BYTES.load(Ordering::Relaxed),
))
} else {
None
};
let token_t0 = std::time::Instant::now();
let mut ple_dt = std::time::Duration::ZERO;
let mut attn_hc_dt = std::time::Duration::ZERO;
let mut mixer_dt = std::time::Duration::ZERO;
let mut mlp_hc_dt = std::time::Duration::ZERO;
let mut moe_dt = std::time::Duration::ZERO;
if position == 0 || state.pos != position {
state.reset();
}
if state.host_pos != position && state.token_history.len() == position {
tracing::warn!(
"qwen4: host caches rebuilt over {position} tokens after the device path turned off"
);
if std::env::var_os("CMF_QWEN_PROF").is_some() {
eprintln!("qwen4: replaying {position} tokens on the host path");
}
let history = std::mem::take(&mut state.token_history);
state.reset();
for (p, &id) in history.iter().enumerate() {
let mut lg = Vec::new();
forward_token(
globals, layers, cfg, state, id, p, inv_freq, pool, &mut lg, false,
);
}
}
let mut emb = vec![0.0f32; cfg.hidden];
if (token_id as usize) < globals.embed.rows() {
globals.embed.row_f32(token_id as usize, &mut emb);
}
observe("embedding", 0, position, &emb);
state.hyper.clear();
state.hyper.reserve(cfg.hc * cfg.hidden);
for _ in 0..cfg.hc {
state.hyper.extend_from_slice(&emb);
}
for (li, layer) in layers.iter().enumerate() {
crate::gpu::set_layer(li as i64);
let st = &mut state.layers[li];
if let Some(ple) = &layer.ple
&& std::env::var_os("CMF_QWEN_NO_PLE").is_none()
{
let t0 = std::time::Instant::now();
let side = ple_forward(
&state.hyper,
token_id,
&state.token_history,
ple,
cfg,
st,
pool,
);
for (h, s) in state.hyper.iter_mut().zip(side) {
*h += s;
}
ple_dt += t0.elapsed();
observe("post_ple", li, position, &state.hyper);
}
let t0 = std::time::Instant::now();
let (mixed, inject_w) = hc_mix(&layer.attn_hc, &state.hyper, cfg, pool);
attn_hc_dt += t0.elapsed();
observe("attn_in", li, position, &mixed);
let t0 = std::time::Instant::now();
let block = match &layer.mixer {
Mixer::Gdn(w) => gdn_forward(&mixed, w, &cfg.gdn, &mut st.gdn, pool),
Mixer::Qsa(w) => qsa_forward(&mixed, w, cfg, &mut st.qsa, position, inv_freq, pool),
};
mixer_dt += t0.elapsed();
observe("attn_out", li, position, &block);
inject(
&mut state.hyper,
&block,
inject_w.as_deref().expect("layer HC has injection"),
cfg,
);
observe("post_attn", li, position, &state.hyper);
let t0 = std::time::Instant::now();
let (mixed, inject_w) = hc_mix(&layer.mlp_hc, &state.hyper, cfg, pool);
mlp_hc_dt += t0.elapsed();
observe("moe_in", li, position, &mixed);
let t0 = std::time::Instant::now();
#[cfg(feature = "gpu")]
let gpu_moe = dynamic_moe_gpu(layer, li, &mixed, state, pool);
#[cfg(not(feature = "gpu"))]
let gpu_moe: Option<(Vec<f32>, Vec<usize>)> = None;
let block = match gpu_moe {
Some((block, _)) => block,
None => {
moe_ffn(&layer.moe, &mixed, pool, None)
}
};
moe_dt += t0.elapsed();
observe("moe_out", li, position, &block);
inject(
&mut state.hyper,
&block,
inject_w.as_deref().expect("layer HC has injection"),
cfg,
);
observe("post_moe", li, position, &state.hyper);
}
crate::gpu::set_layer(-1);
let head_t0 = std::time::Instant::now();
if want_logits {
let (hidden, _) = hc_mix(&globals.head_hc, &state.hyper, cfg, pool);
observe("head_in", layers.len(), position, &hidden);
logits.resize(globals.lm_head.rows(), 0.0);
globals.lm_head.matvec(&hidden, logits, pool);
observe("logits", layers.len(), position, logits);
} else {
logits.clear();
}
let head_dt = head_t0.elapsed();
state.token_history.push(token_id);
state.pos = position + 1;
state.host_pos = position + 1;
if prof {
eprintln!(
"qwen-prof pos={position} total={:.3}s ple={:.3}s attn_hc={:.3}s mixer={:.3}s mlp_hc={:.3}s moe={:.3}s head={:.3}s",
token_t0.elapsed().as_secs_f64(),
ple_dt.as_secs_f64(),
attn_hc_dt.as_secs_f64(),
mixer_dt.as_secs_f64(),
mlp_hc_dt.as_secs_f64(),
moe_dt.as_secs_f64(),
head_dt.as_secs_f64(),
);
#[cfg(feature = "gpu")]
if let Some((enc0, wait0, card0, calls0, fills0, fill_bytes0)) = gpu_moe_before {
use std::sync::atomic::Ordering;
let enc = crate::gpu_wgpu::MOE_ENC_NS.load(Ordering::Relaxed) - enc0;
let wait = crate::gpu_wgpu::MOE_WAIT_NS.load(Ordering::Relaxed) - wait0;
let card = crate::gpu_wgpu::MOE_GPU_NS[0].load(Ordering::Relaxed) - card0;
let calls = crate::gpu_wgpu::MOE_GPU_N.load(Ordering::Relaxed) - calls0;
let fills = crate::gpu_wgpu::DSV4_FILLS.load(Ordering::Relaxed) - fills0;
let fill_bytes = crate::gpu_wgpu::DSV4_FILL_BYTES.load(Ordering::Relaxed) - fill_bytes0;
eprintln!(
"qwen-moe-gpu pos={position} calls={calls} fills={fills} fill={:.1}MB encode={:.1}ms wait={:.1}ms card={:.1}ms",
fill_bytes as f64 / 1e6,
enc as f64 / 1e6,
wait as f64 / 1e6,
card as f64 / 1e6,
);
}
}
}
#[cfg(feature = "gpu")]
fn device_geom(cfg: &Cfg, moe: &MoeFfn, gu_q2: bool) -> crate::gpu_wgpu::qwen4::Geom {
use crate::gpu_wgpu::qwen4::{GdnGeom, Geom};
Geom {
hidden: cfg.hidden,
hc: cfg.hc,
eps: cfg.eps as f32,
n_heads: cfg.n_heads,
n_kv_heads: cfg.n_kv_heads,
head_dim: cfg.head_dim,
rotary_dim: cfg.rotary_dim,
index_heads: cfg.index_heads,
index_dim: cfg.index_dim,
index_budget: cfg.index_budget,
compress_ratio: cfg.compress_ratio,
gdn: GdnGeom {
nv: cfg.gdn.num_v_heads,
nk: cfg.gdn.num_k_heads,
dk: cfg.gdn.key_head_dim,
dv: cfg.gdn.value_head_dim,
kk: cfg.gdn.conv_kernel,
},
ple_kernel: cfg.ple_kernel,
ple_dilation: cfg.ple_dilation,
top_k: moe.top_k,
n_experts: moe.experts.len(),
inter: moe.experts.first().map_or(0, |e| e.gate_proj.rows()),
gu_q2,
}
}
#[cfg(feature = "gpu")]
fn device_layer_w<'a>(
layer: &'a Layer,
ix: &LayerIdx,
) -> Option<crate::gpu_wgpu::qwen4::LayerW<'a>> {
use crate::gpu_wgpu::qwen4::{HcW, LayerW, MixerW, PleW};
let hc = |g: &'a GatedResidual| -> Option<HcW<'a>> {
Some(HcW {
norm: &g.norm,
down: g.down_idx?,
up: g.up_idx?,
inject: g.inject_idx,
})
};
let mixer = match &layer.mixer {
Mixer::Gdn(w) => {
let [qkv, z, a, b, out] = ix.gdn?;
MixerW::Gdn {
qkv,
z,
a,
b,
out,
conv1d: &w.conv1d,
a_log: &w.a_log,
dt_bias: &w.dt_bias,
norm: &w.norm,
}
}
Mixer::Qsa(w) => {
let [q, k, v, o, index_qk] = ix.qsa?;
MixerW::Qsa {
q,
k,
v,
o,
index_qk,
q_norm: &w.q_norm,
k_norm: &w.k_norm,
iq_norm: &w.index_q_norm,
ik_norm: &w.index_k_norm,
}
}
};
let ple = match (&layer.ple, ix.ple) {
(Some(w), Some((key_proj, value_proj))) => Some(PleW {
key_proj,
value_proj,
norm_key: &w.norm_key,
norm_query: &w.norm_query,
norm_conv: &w.norm_conv,
conv: &w.conv,
}),
(None, None) => None,
_ => return None,
};
Some(LayerW {
attn_hc: hc(&layer.attn_hc)?,
mlp_hc: hc(&layer.mlp_hc)?,
mixer,
ple,
router: ix.router,
shared_gate: ix.shared_gate,
})
}
#[cfg(feature = "gpu")]
#[allow(clippy::type_complexity)]
static DEV_TAP: std::sync::Mutex<Vec<(usize, Vec<f32>, Vec<f32>, Vec<f32>, Vec<f32>, Vec<f32>)>> =
std::sync::Mutex::new(Vec::new());
#[cfg(feature = "gpu")]
fn tap_compare(label: &str, host: &[f32], dev: &[f32]) {
let n = host.len().min(dev.len());
let mut max_abs = 0.0f32;
let mut at = 0usize;
let (mut dot, mut na, mut nb) = (0.0f64, 0.0f64, 0.0f64);
for i in 0..n {
let d = (host[i] - dev[i]).abs();
if d > max_abs {
max_abs = d;
at = i;
}
dot += host[i] as f64 * dev[i] as f64;
na += host[i] as f64 * host[i] as f64;
nb += dev[i] as f64 * dev[i] as f64;
}
eprintln!(
"qwen4-tap {label:>10}: cos={:.6} max_abs={max_abs:.4} at {at} (host {:.4} dev {:.4}) |host|={:.3} |dev|={:.3} n={n}",
dot / (na.sqrt() * nb.sqrt()).max(1e-30),
host.get(at).copied().unwrap_or(0.0),
dev.get(at).copied().unwrap_or(0.0),
na.sqrt(),
nb.sqrt()
);
}
#[cfg(feature = "gpu")]
struct ExpertProfile {
n_layers: usize,
n_experts: usize,
counts: Vec<u32>,
tokens: u64,
}
#[cfg(feature = "gpu")]
impl ExpertProfile {
const MAGIC: &'static [u8; 8] = b"CMFQ4PF\0";
fn new(n_layers: usize, n_experts: usize) -> Self {
Self {
n_layers,
n_experts,
counts: vec![0; n_layers * n_experts],
tokens: 0,
}
}
fn load(path: &str, n_layers: usize, n_experts: usize) -> Option<Self> {
let bytes = std::fs::read(path).ok()?;
if bytes.len() < 24 || &bytes[..8] != Self::MAGIC {
return None;
}
let rd = |o: usize| u64::from_le_bytes(bytes[o..o + 8].try_into().unwrap());
let (nl, ne) = (rd(8) as usize, rd(16) as usize);
if nl != n_layers || ne != n_experts || bytes.len() < 32 + nl * ne * 4 {
return None;
}
let tokens = rd(24);
let counts = bytes[32..32 + nl * ne * 4]
.chunks_exact(4)
.map(|b| u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
.collect();
Some(Self {
n_layers,
n_experts,
counts,
tokens,
})
}
fn save(&self, path: &str) {
let mut out = Vec::with_capacity(32 + self.counts.len() * 4);
out.extend_from_slice(Self::MAGIC);
out.extend_from_slice(&(self.n_layers as u64).to_le_bytes());
out.extend_from_slice(&(self.n_experts as u64).to_le_bytes());
out.extend_from_slice(&self.tokens.to_le_bytes());
for &c in &self.counts {
out.extend_from_slice(&c.to_le_bytes());
}
if let Err(e) = std::fs::write(path, out) {
tracing::warn!("expert profile not saved to {path}: {e}");
}
}
fn note(&mut self, layer: usize, picks: &[usize]) {
for &e in picks {
if let Some(c) = self.counts.get_mut(layer * self.n_experts + e) {
*c = c.saturating_add(1);
}
}
}
fn ranked(&self) -> Vec<(usize, usize)> {
let mut v: Vec<(u32, usize, usize)> = self
.counts
.iter()
.enumerate()
.filter(|&(_, &c)| c > 0)
.map(|(i, &c)| (c, i / self.n_experts, i % self.n_experts))
.collect();
v.sort_unstable_by(|a, b| b.0.cmp(&a.0).then(a.1.cmp(&b.1)).then(a.2.cmp(&b.2)));
v.into_iter().map(|(_, l, e)| (l, e)).collect()
}
}
#[cfg(feature = "gpu")]
fn route_wstat() -> bool {
static V: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*V.get_or_init(|| std::env::var("CMF_QWEN_ROUTE_WSTAT").as_deref() == Ok("1"))
}
#[cfg(feature = "gpu")]
fn note_route_weights(pairs: &[u32]) {
const CUTS: [f32; 6] = [0.01, 0.02, 0.03, 0.05, 0.08, 0.10];
static ACC: std::sync::Mutex<(u64, u64, [u64; 6], [f64; 6])> = std::sync::Mutex::new((0, 0, [0; 6], [0.0; 6]));
let mut a = ACC.lock().unwrap();
a.0 += 1;
for pr in pairs.chunks_exact(2) {
if pr[0] == u32::MAX {
continue;
}
let w = f32::from_bits(pr[1]);
a.1 += 1;
for (i, &c) in CUTS.iter().enumerate() {
if w < c {
a.2[i] += 1;
a.3[i] += f64::from(w);
}
}
}
if a.0 % 9600 == 0 {
let share: Vec<String> = a.2.iter().map(|&n| format!("{:.1}", 100.0 * n as f64 / a.1.max(1) as f64)).collect();
let mass: Vec<String> = a.3.iter().map(|&m| format!("{:.2}", 100.0 * m / a.0 as f64)).collect();
eprintln!(
"route weights: {} token-layers, {:.2} winners each; under 1/2/3/5/8/10 %: {} % of the winners, {} % of the routing",
a.0,
a.1 as f64 / a.0 as f64,
share.join("/"),
mass.join("/")
);
}
}
#[cfg(feature = "gpu")]
fn draft_vocab_rows() -> usize {
static V: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
*V.get_or_init(|| {
std::env::var("CMF_QWEN_DRAFT_VOCAB")
.ok()
.and_then(|v| v.trim().parse().ok())
.unwrap_or(0)
})
}
#[cfg(feature = "gpu")]
fn draft_gate() -> (f32, bool) {
static V: std::sync::OnceLock<(f32, bool)> = std::sync::OnceLock::new();
*V.get_or_init(|| {
let minp = std::env::var("CMF_QWEN_DRAFT_MINP")
.ok()
.and_then(|v| v.trim().parse::<f32>().ok())
.filter(|p| p.is_finite() && *p > 0.0)
.map_or(0.0, |p| p.min(1.0));
let report = minp > 0.0 || std::env::var("CMF_QWEN_DRAFT_CONF").as_deref() == Ok("1");
(minp, report)
})
}
#[cfg(feature = "gpu")]
fn draft_gate_keep(conf: &[f32], minp: f32) -> usize {
if minp <= 0.0 {
return conf.len();
}
let mut cum = 1.0f32;
let mut keep = 0;
for &p in conf {
cum *= p;
if cum.is_nan() || cum < minp {
break;
}
keep += 1;
}
keep.max(1).min(conf.len())
}
#[cfg(feature = "gpu")]
#[derive(Clone, Copy, Debug, PartialEq)]
struct SuffixCfg {
ext: usize,
strong: usize,
max: usize,
}
#[cfg(feature = "gpu")]
fn suffix_parse(raw: &str) -> Option<SuffixCfg> {
let mut it = raw.split(':').map(|v| v.trim().parse::<usize>().ok());
let ext = it.next()??;
let strong = it.next().unwrap_or(Some(0))?;
let max = it.next().unwrap_or(Some(usize::MAX))?;
let max = max.clamp(1, crate::gpu_wgpu::qwen4::TMAX - 1);
(ext > 0 || strong > 0).then_some(SuffixCfg { ext, strong, max })
}
#[cfg(feature = "gpu")]
fn suffix_cfg() -> Option<SuffixCfg> {
static V: std::sync::OnceLock<Option<SuffixCfg>> = std::sync::OnceLock::new();
*V.get_or_init(|| std::env::var("CMF_QWEN_SUFFIX").ok().and_then(|v| suffix_parse(&v)))
}
#[cfg(feature = "gpu")]
fn copy_frame_cfg() -> Option<(usize, f32)> {
static V: std::sync::OnceLock<Option<(usize, f32)>> = std::sync::OnceLock::new();
*V.get_or_init(|| copy_frame_parse(&std::env::var("CMF_QWEN_COPY_FRAME").ok()?))
}
#[cfg(feature = "gpu")]
fn copy_frame_debug() -> bool {
static V: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*V.get_or_init(|| std::env::var("CMF_QWEN_FRAME_DEBUG").as_deref() == Ok("1"))
}
#[cfg(feature = "gpu")]
fn copy_frame_parse(raw: &str) -> Option<(usize, f32)> {
let mut it = raw.split(':');
let n = it.next()?.trim().parse::<usize>().ok()?;
let f = match it.next() {
Some(v) => v.trim().parse::<f32>().ok().filter(|f| f.is_finite() && *f > 0.0)?,
None => 2.0,
};
(n >= COPY_FRAMES[COPY_FRAMES.len() - 1]).then_some((n.min(crate::gpu_wgpu::qwen4::TROWS - 1), f))
}
#[cfg(feature = "gpu")]
const COPY_FRAMES: [usize; 4] = [255, 127, 63, 31];
#[cfg(feature = "gpu")]
fn copy_frame_drafts(most: usize, factor: f32, run: usize, budget: usize) -> Option<usize> {
let lim = most.min(budget).min((run as f32 * factor) as usize);
COPY_FRAMES.into_iter().find(|&k| k <= lim)
}
#[cfg(feature = "gpu")]
fn suffix_find(ids: &[u32], min_m: usize) -> (usize, usize) {
const CAP: usize = 32;
let n = ids.len();
if min_m == 0 || n <= min_m {
return (0, 0);
}
let last = ids[n - 1];
let (mut best_m, mut best_j) = (0usize, 0usize);
for j in (min_m - 1..n - 1).rev() {
if ids[j] != last {
continue;
}
let mut m = 1;
while m < CAP && m <= j && ids[j - m] == ids[n - 1 - m] {
m += 1;
}
if m > best_m {
(best_m, best_j) = (m, j);
if m >= CAP {
break;
}
}
}
if best_m < min_m { (0, 0) } else { (best_m, best_j) }
}
#[cfg(feature = "gpu")]
fn suffix_draft(ids: &[u32], min_m: usize, max_out: usize) -> (usize, Vec<u32>) {
let n = ids.len();
if min_m == 0 || max_out == 0 || n <= min_m {
return (0, Vec::new());
}
let (best_m, best_j) = suffix_find(ids, min_m);
if best_m < min_m {
return (0, Vec::new());
}
let mut out: Vec<u32> = Vec::with_capacity(max_out);
for i in 0..max_out {
let p = best_j + 1 + i;
out.push(if p < n { ids[p] } else { out[p - n] });
}
(best_m, out)
}
#[cfg(feature = "cuda")]
const GAP_MAX: usize = 12;
#[cfg(feature = "cuda")]
#[derive(Default)]
struct CopyCur {
seen: usize,
at: Option<usize>,
run: usize,
gap: Vec<u32>,
last_gap: Vec<u32>,
rules: std::collections::HashMap<(u32, u32), (Vec<u32>, u8)>,
}
#[cfg(feature = "cuda")]
impl CopyCur {
fn key(ids: &[u32], at: usize) -> Option<(u32, u32)> {
(at >= 1 && at < ids.len()).then(|| (ids[at - 1], ids[at]))
}
fn start(&mut self, at: usize, run: usize) {
self.at = Some(at);
self.run = run;
self.gap.clear();
}
fn drop_cursor(&mut self) {
self.at = None;
self.run = 0;
self.gap.clear();
}
fn account(&mut self, ids: &[u32]) {
if self.seen > ids.len() {
*self = Self::default();
}
for j in self.seen..ids.len() {
let Some(at) = self.at else { break };
if at >= j {
self.drop_cursor();
break;
}
let key = Self::key(ids, at);
if ids[j] == ids[at] {
if !self.gap.is_empty() {
let g = std::mem::take(&mut self.gap);
if let Some(key) = key {
match self.rules.get_mut(&key) {
Some((rg, hits)) if *rg == g => *hits = hits.saturating_add(1),
_ => {
self.rules.insert(key, (g.clone(), 1));
}
}
}
self.last_gap = g;
} else if let Some(key) = key
&& let Some(r) = self.rules.get_mut(&key)
{
if r.1 <= 1 {
self.rules.remove(&key);
} else {
r.1 -= 1;
}
}
self.at = Some(at + 1);
self.run += 1;
} else {
self.gap.push(ids[j]);
let expected = key.and_then(|k| self.rules.get(&k)).is_some_and(|(rg, _)| rg.starts_with(&self.gap));
if !expected {
self.run = 0;
}
if self.gap.len() > GAP_MAX {
self.drop_cursor();
break;
}
}
}
self.seen = ids.len();
}
fn draft(&self, ids: &[u32], most: usize) -> Vec<u32> {
let Some(at) = self.at else { return Vec::new() };
let mut out: Vec<u32> = Vec::with_capacity(most);
let mut skip = None;
if !self.gap.is_empty() {
let g = &self.gap;
let rule = Self::key(ids, at).and_then(|k| self.rules.get(&k)).map(|(rg, _)| rg.as_slice());
let rest: &[u32] = match rule {
Some(rg) if rg.len() >= g.len() && rg[..g.len()] == g[..] => &rg[g.len()..],
_ if self.last_gap.len() >= g.len() && self.last_gap[..g.len()] == g[..] => &self.last_gap[g.len()..],
_ => return Vec::new(),
};
out.extend_from_slice(&rest[..rest.len().min(most)]);
skip = Some(at);
}
let mut i = at;
while out.len() < most && i < ids.len() {
if i >= 1
&& skip != Some(i)
&& let Some((rg, hits)) = self.rules.get(&(ids[i - 1], ids[i]))
&& *hits >= 2
{
for &t in rg {
if out.len() < most {
out.push(t);
}
}
if out.len() >= most {
break;
}
}
out.push(ids[i]);
i += 1;
}
out
}
}
#[cfg(feature = "cuda")]
fn cuda_copy_cfg() -> Option<SuffixCfg> {
static V: std::sync::OnceLock<Option<SuffixCfg>> = std::sync::OnceLock::new();
*V.get_or_init(|| match std::env::var("CMF_QWEN_SUFFIX") {
Ok(v) => suffix_parse(&v),
Err(_) => Some(SuffixCfg { ext: 4, strong: 8, max: 7 }),
})
}
#[cfg(feature = "cuda")]
#[allow(clippy::too_many_arguments)]
fn cuda_spec_round(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
next_pos: usize,
all_ids: &[u32],
pool: Option<&Pool>,
) -> Option<SpecRound> {
if state.cuda_off || next_pos == 0 || all_ids.len() != next_pos + 1 || state.pos != next_pos || state.token_history.len() != next_pos {
return None;
}
let sc = cuda_copy_cfg();
let tape = match state.cuda.as_ref() {
Some(d) if d.pos == next_pos => d.tape(),
_ => return None,
};
let budget = state.draft_budget;
let mut drafts: Vec<u32> = Vec::new();
if let Some(sc) = sc {
state.cuda_copy.account(all_ids);
if state.cuda_copy.at.is_none() && sc.strong > 0 {
let (m, j) = suffix_find(all_ids, sc.strong);
if m >= sc.strong {
state.cuda_copy.start(j + 1, 1);
}
}
if state.cuda_copy.at.is_some() {
let run = state.cuda_copy.run;
let mut most = (1 + 2 * run).min(sc.max);
if let Some((fmax, factor)) = copy_frame_cfg() {
let lim = fmax.min(budget).min((run as f32 * factor) as usize).min(tape.saturating_sub(1));
if lim >= 63 {
most = lim;
}
}
most = most.min(budget);
if most > 0 {
drafts = state.cuda_copy.draft(all_ids, most);
}
}
}
let copied = !drafts.is_empty();
let t_draft0 = std::time::Instant::now();
if drafts.is_empty() && budget > 0 {
if let (Some(head), Some(dev)) = (state.mtp.as_ref(), state.cuda.as_mut()) {
if dev.has_mtp() {
let embed = |id: u32, row: &mut [f32]| {
if (id as usize) < globals.embed.rows() {
globals.embed.row_f32(id as usize, row);
}
};
let kmax = head.k.min(budget).min(tape.saturating_sub(1)).min(7);
let mut out: Vec<u32> = Vec::new();
match dev.draft(&head.layer, &embed, all_ids[next_pos], next_pos, kmax, &mut out) {
Ok(true) => drafts = out,
Ok(false) => {}
Err(e) => {
eprintln!("qwen4 cuda: the MTP head failed ({e}); it stays off");
dev.mtp_off();
}
}
}
}
}
let t_draft = t_draft0.elapsed();
let k = drafts.len();
if k == 0 {
return None;
}
let ids: Vec<u32> = std::iter::once(all_ids[next_pos]).chain(drafts.iter().copied()).collect();
let mut dev = state.cuda.take()?;
let h = cfg.hidden;
let no_ple = std::env::var_os("CMF_QWEN_NO_PLE").is_some();
let prof = std::env::var_os("CMF_QWEN_PROF").is_some();
let t_round = std::time::Instant::now();
let res = (|| -> Result<(usize, Vec<f32>), String> {
let mut emb = vec![0.0f32; ids.len() * h];
for (row, &id) in emb.chunks_mut(h).zip(&ids) {
if (id as usize) < globals.embed.rows() {
globals.embed.row_f32(id as usize, row);
}
}
let ple = if no_ple { Vec::new() } else { ple_frame_rows(layers, cfg, &state.token_history, &ids, pool) };
let t_host = t_round.elapsed();
let mut row_ids: Vec<u32> = Vec::new();
dev.spec(layers, &emb, &ple, next_pos, &mut row_ids)?;
if row_ids.len() != ids.len() {
return Err("qwen4 cuda: a window came back short".into());
}
let mut a = 0;
while a < k && row_ids[a] == drafts[a] {
a += 1;
}
let mut logits: Vec<f32> = Vec::new();
if !dev.settle(a + 1, &mut logits)? {
let part: Vec<Vec<Vec<f32>>> = ple.iter().map(|rows| rows.iter().take(a + 1).cloned().collect()).collect();
dev.forward(layers, &emb[..(a + 1) * h], &part, next_pos, Some(&mut logits))?;
}
if prof {
eprintln!(
"qwen4 cuda: round of {k} drafts, {a} accepted: host rows {:.1} ms, whole {:.1} ms",
t_host.as_secs_f64() * 1e3,
t_round.elapsed().as_secs_f64() * 1e3
);
}
if std::env::var_os("CMF_QWEN_ROUND_LOG").is_some() {
eprintln!(
"qwen4 cuda: round {} k={k} a={a} draft {:.1} ms verify {:.1} ms",
if copied { "copy" } else { "mtp" },
t_draft.as_secs_f64() * 1e3,
t_round.elapsed().as_secs_f64() * 1e3
);
}
Ok((a, logits))
})();
match res {
Ok((a, logits)) => {
state.token_history.extend_from_slice(&ids[..a + 1]);
state.pos = next_pos + a + 1;
state.cuda = Some(dev);
Some(SpecRound { accepted: drafts[..a].to_vec(), logits, drafted: k })
}
Err(e) => {
eprintln!("qwen4 cuda: {e}; the sequence continues on the host path");
drop(dev);
state.cuda_off = true;
None
}
}
}
#[cfg(feature = "gpu")]
fn admit_batch_env() -> bool {
static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ON.get_or_init(|| std::env::var("CMF_QWEN_ADMIT_BATCH").as_deref() != Ok("0"))
}
#[cfg(feature = "gpu")]
#[derive(Default)]
struct DevProf {
encode: std::time::Duration,
wait: std::time::Duration,
cold_cpu: std::time::Duration,
admit: std::time::Duration,
head: std::time::Duration,
cold_experts: usize,
cold_dev: usize,
layers_with_cold: usize,
chains: usize,
aborted: usize,
finish: std::time::Duration,
spin: std::time::Duration,
post: std::time::Duration,
ple: std::time::Duration,
gpu: [f64; 17],
gpu_last_end: Option<f64>,
}
#[cfg(feature = "gpu")]
const TS_NAMES: [&str; 17] = [
"pending", "ple", "attn_hc", "gdn_proj", "gdn_rec", "gdn_out", "qsa_proj", "qsa_attn",
"qsa_out", "moe_hc", "router", "route", "gu", "dn", "tail", "head", "idle",
];
#[cfg(feature = "gpu")]
impl DevProf {
fn gpu_note(&mut self, fi: usize, n: usize, m: &[f64], gdn: Option<bool>) {
if m.len() < 13 {
return;
}
let d = |a: usize, b: usize| (m[b] - m[a]).max(0.0);
if let Some(prev) = self.gpu_last_end {
self.gpu[16] += (m[0] - prev).max(0.0);
}
self.gpu_last_end = Some(m[8]);
self.gpu[0] += d(0, 1);
if fi >= n {
self.gpu[15] += d(7, 8);
return;
}
self.gpu[1] += d(1, 2);
self.gpu[2] += d(2, 3);
let o = if gdn == Some(true) { 3 } else { 6 };
self.gpu[o] += d(3, 9);
self.gpu[o + 1] += d(9, 10);
self.gpu[o + 2] += d(10, 4);
self.gpu[9] += d(4, 5);
self.gpu[10] += d(5, 11);
self.gpu[11] += d(11, 6);
self.gpu[12] += d(6, 12);
self.gpu[13] += d(12, 7);
self.gpu[14] += d(7, 8);
}
}
#[cfg(feature = "gpu")]
pub fn prefill_chunk() -> usize {
#[cfg(feature = "cuda")]
if crate::gpu_cuda::selected() {
return std::env::var("CMF_QWEN_CUDA_CHUNK")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(if crate::gpu_cuda::qwen4::lean_card() { 1024 } else { 4096 })
.clamp(1, crate::gpu_cuda::qwen4::MAX_CHUNK);
}
let rows = crate::gpu_wgpu::qwen4::TROWS;
let dflt = if stream_on() { rows * stream_frames() } else { rows };
std::env::var("CMF_QWEN_PREFILL_CHUNK")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(dflt)
.clamp(1, rows * 64)
}
#[cfg(feature = "gpu")]
fn stream_on() -> bool {
std::env::var("CMF_QWEN_STREAM").as_deref() != Ok("0")
}
#[cfg(feature = "gpu")]
fn stream_frames() -> usize {
let rows = crate::gpu_wgpu::qwen4::TROWS;
std::env::var("CMF_QWEN_STREAM_TOKENS")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or_else(|| {
if crate::gpu_wgpu::qwen4::vram_budget().is_some_and(|b| b < 11 << 30) { 2048 } else { 4096 }
})
.div_ceil(rows)
.clamp(1, 64)
}
#[cfg(not(feature = "gpu"))]
pub fn prefill_chunk() -> usize {
1
}
#[allow(clippy::too_many_arguments)]
fn top_log(logits: &[f32], vocab: usize) {
if vocab == 0 || logits.len() < vocab {
return;
}
let row = &logits[logits.len() - vocab..];
let mut ix: Vec<usize> = (0..vocab).collect();
ix.sort_by(|&a, &b| row[b].total_cmp(&row[a]));
let top: Vec<String> = ix[..5].iter().map(|&i| format!("{i}:{:.4}", row[i])).collect();
eprintln!("[toplog] {}", top.join(" "));
}
#[cfg(feature = "cuda")]
fn cuda_check_on() -> bool {
static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ON.get_or_init(|| std::env::var("CMF_QWEN_CUDA_CHECK").as_deref() == Ok("1"))
}
#[cfg(feature = "cuda")]
#[allow(clippy::too_many_arguments)]
fn cuda_run(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
dev: &mut crate::gpu_cuda::qwen4::Dev,
history: &[u32],
ids: &[u32],
pos0: usize,
pool: Option<&Pool>,
mut logits: Option<&mut Vec<f32>>,
) -> Result<(), String> {
let h = cfg.hidden;
let chunk = dev.chunk().max(1);
let no_ple = std::env::var_os("CMF_QWEN_NO_PLE").is_some();
let prof = std::env::var_os("CMF_QWEN_PROF").is_some();
if !no_ple && ids.len() >= 64 {
ple_prefetch(layers, cfg, history, ids);
}
let mut grown: Vec<u32> = Vec::new();
let mut off = 0usize;
while off < ids.len() {
let end = (off + chunk).min(ids.len());
let part = &ids[off..end];
let t0 = std::time::Instant::now();
let mut emb = vec![0.0f32; part.len() * h];
for (row, &id) in emb.chunks_mut(h).zip(part) {
if (id as usize) < globals.embed.rows() {
globals.embed.row_f32(id as usize, row);
}
}
let ple = if no_ple {
Vec::new()
} else if off == 0 {
ple_frame_rows(layers, cfg, history, part, pool)
} else {
if grown.is_empty() {
grown.extend_from_slice(history);
}
grown.extend_from_slice(&ids[grown.len() - history.len()..off]);
ple_frame_rows(layers, cfg, &grown, part, pool)
};
if prof && part.len() > 1 {
eprintln!("qwen4 cuda: host rows of {} tokens {:.1} ms", part.len(), t0.elapsed().as_secs_f64() * 1e3);
}
let lg = if end == ids.len() { logits.take() } else { None };
dev.forward(layers, &emb, &ple, pos0 + off, lg)?;
off = end;
}
Ok(())
}
#[cfg(feature = "cuda")]
fn cuda_prompt_done(layers: &[Layer], state: &mut State) {
if std::env::var("CMF_QWEN_SELF_PACK").as_deref() != Ok("1") {
return;
}
let (n, ne) = (layers.len(), layers.first().map_or(0, |l| l.moe.experts.len()));
if n == 0 || ne == 0 || state.cuda.is_none() {
return;
}
if state.packs.is_none() {
state.packs = Some(PackSet::load(n, ne));
}
let packs = state.packs.as_ref().and_then(|p| p.as_ref());
let weighted: Vec<(usize, f32)> = match packs {
Some(p) => match p.hint_weights() {
Some(w) => w,
None => state
.cuda
.as_ref()
.and_then(|d| p.decide(d.route_hist(), ne))
.map(|(i, _)| vec![(i, 1.0)])
.unwrap_or_default(),
},
None => Vec::new(),
};
let pack: Option<Vec<f32>> = packs.filter(|_| !weighted.is_empty()).map(|p| {
let mut pk = vec![0f32; n * ne];
for &(i, w) in &weighted {
for l in 0..n {
let row = &p.profiles[i].counts[l * ne..(l + 1) * ne];
let ps = row.iter().map(|&c| c as f32).sum::<f32>().max(1.0);
for (o, &c) in pk[l * ne..(l + 1) * ne].iter_mut().zip(row) {
*o += w * c as f32 / ps;
}
}
}
pk
});
let mix: f32 = if pack.is_some() { std::env::var("CMF_QWEN_PACK_MIX").ok().and_then(|v| v.parse().ok()).unwrap_or(0.5f32).clamp(0.0, 1.0) } else { 1.0 };
let max_swap = std::env::var("CMF_QWEN_PACK_SWAP").ok().and_then(|v| v.parse::<usize>().ok()).unwrap_or(4096);
let t0 = std::time::Instant::now();
let name = weighted.first().and_then(|&(i, _)| packs.map(|p| p.names[i].clone())).unwrap_or_else(|| "prompt routing".into());
let Some(dev) = state.cuda.as_mut() else { return };
match dev.retarget(layers, pack.as_deref(), mix, max_swap) {
Ok((out, inn)) => {
if std::env::var_os("CMF_QWEN_PROF").is_some() || std::env::var_os("CMF_QWEN_PACK_LOG").is_some() {
eprintln!("qwen4 cuda: pack {name} (prompt share {mix}): {out} experts out, {inn} in, {:.0} ms", t0.elapsed().as_secs_f64() * 1e3);
}
}
Err(e) => eprintln!("qwen4 cuda: pack: {e}"),
}
}
#[cfg(feature = "cuda")]
#[allow(clippy::too_many_arguments)]
fn cuda_forward(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
ids: &[u32],
position: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
logits: &mut Vec<f32>,
want_logits: bool,
) -> bool {
if state.cuda_off || ids.is_empty() || !crate::gpu_cuda::selected() {
return false;
}
state.device_off = true;
if !crate::gpu::enabled_here() || std::env::var("CMF_QWEN_DEVICE").as_deref() == Ok("0") {
return false;
}
if !crate::gpu_cuda::enabled() {
state.cuda_off = true;
return false;
}
if state.cuda.is_none() {
if state.mtp.is_none() && !state.mtp_tried {
state.mtp_tried = true;
if let Some(model) = layers.first().and_then(|l| l.moe.experts.first()).and_then(|f| f.gate_proj.model_arc()) {
state.mtp = MtpHead::load(&model, model.arch());
}
}
let fcs = state.mtp.as_ref().and_then(|h| {
Some((t(&h.side, "model.mtp.fc_embedding.weight").ok()?, t(&h.side, "model.mtp.fc_hidden.weight").ok()?))
});
let parts = state.mtp.as_ref().zip(fcs.as_ref()).map(|(h, (fe, fh))| crate::gpu_cuda::qwen4::MtpParts {
layer: &h.layer,
mixer: &h.mixer,
enorm: &h.enorm,
hnorm: &h.hnorm,
fc_e: fe,
fc_h: fh,
});
match crate::gpu_cuda::qwen4::Dev::new(globals, layers, cfg, inv_freq, prefill_chunk(), parts.as_ref()) {
Ok(mut d) => {
if let Some(head) = state.mtp.as_ref() {
match d.warm_mtp(&head.layer) {
Ok(n) if n > 0 => {
if std::env::var_os("CMF_QWEN_PROF").is_some() {
eprintln!("qwen4 cuda: {n} experts of the MTP head resident");
}
}
Ok(_) => {}
Err(e) => {
eprintln!("qwen4 cuda: the MTP head's experts: {e}; it stays off");
d.mtp_off();
}
}
}
let ne = layers.first().map_or(0, |l| l.moe.experts.len());
if let Some(pr) = std::env::var("CMF_QWEN_PROFILE").ok().and_then(|path| ExpertProfile::load(&path, layers.len(), ne)) {
let t0 = std::time::Instant::now();
match d.prefill(layers, &pr.ranked()) {
Ok(n) => eprintln!(
"qwen4 cuda: arena prefilled with {n} of {} profiled experts ({} tokens of history) in {:.2}s",
pr.ranked().len(),
pr.tokens,
t0.elapsed().as_secs_f64()
),
Err(e) => eprintln!("qwen4 cuda: prefill: {e}"),
}
}
state.cuda = Some(d)
}
Err(e) => {
eprintln!("qwen4 cuda: {e}; staying on the host path");
state.cuda_off = true;
return false;
}
}
}
let Some(mut dev) = state.cuda.take() else {
return false;
};
let check = cuda_check_on();
dev.want_trace = check;
let res = (|| -> Result<bool, String> {
if position == 0 || state.pos != position {
state.reset();
dev.reset()?;
}
if dev.pos != position {
if state.token_history.len() != position {
return Ok(false);
}
dev.reset()?;
cuda_run(globals, layers, cfg, &mut dev, &[], &state.token_history, 0, pool, None)?;
}
cuda_run(
globals,
layers,
cfg,
&mut dev,
&state.token_history,
ids,
position,
pool,
if want_logits { Some(&mut *logits) } else { None },
)?;
Ok(true)
})();
match res {
Ok(true) => {
if check {
cuda_check(globals, layers, cfg, state, &dev, ids, position, inv_freq, pool, want_logits);
}
state.token_history.extend_from_slice(ids);
state.pos = position + ids.len();
if !want_logits {
logits.clear();
}
state.cuda = Some(dev);
if want_logits && position == 0 && ids.len() >= 32 {
cuda_prompt_done(layers, state);
}
true
}
Ok(false) => {
state.cuda = Some(dev);
false
}
Err(e) => {
eprintln!("qwen4 cuda: {e}; the sequence continues on the host path");
drop(dev);
state.cuda_off = true;
false
}
}
}
#[cfg(feature = "cuda")]
#[allow(clippy::too_many_arguments)]
fn cuda_check(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
dev: &crate::gpu_cuda::qwen4::Dev,
ids: &[u32],
position: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
want_logits: bool,
) {
if position == 0 || state.cuda_ref.is_none() {
let mut r = Box::new(State::new(layers.len()));
r.cuda_off = true;
r.device_off = true;
state.cuda_ref = Some(r);
}
let Some(host) = state.cuda_ref.as_mut() else {
return;
};
if host.pos != position {
eprintln!("qwen4 cuda check: the host reference is at {} and the chunk at {position}; not compared", host.pos);
return;
}
let mut host_logits = Vec::new();
let mut tap = Vec::new();
for (i, &id) in ids.iter().enumerate() {
let last = i + 1 == ids.len();
if last {
CUDA_TAP.with(|t| *t.borrow_mut() = Some(Vec::new()));
}
crate::qtensor::float_activations_scope(|| {
crate::gpu::cpu_scope(|| {
forward_token(
globals,
layers,
cfg,
host,
id,
position + i,
inv_freq,
pool,
&mut host_logits,
last && want_logits,
);
})
});
if last {
tap = CUDA_TAP.with(|t| t.borrow_mut().take()).unwrap_or_default();
}
}
let rel = |a: &[f32], b: &[f32]| -> f32 {
if a.len() != b.len() {
return f32::INFINITY;
}
let scale = a.iter().fold(0f32, |m, v| m.max(v.abs())).max(1e-6);
let diff = a.iter().zip(b).fold(0f32, |m, (x, y)| if (x - y).is_nan() { f32::INFINITY } else { m.max((x - y).abs()) });
diff / scale
};
let at = position + ids.len() - 1;
let mut worst = (0f32, "", 0usize);
let mut lines: Vec<String> = Vec::new();
let mut cur = usize::MAX;
for (label, li, dv) in &dev.trace {
let Some((_, _, hv)) = tap.iter().find(|(l, i, _)| l == label && i == li) else {
continue;
};
let r = rel(hv, dv);
if r > worst.0 || worst.1.is_empty() {
worst = (r, label, *li);
}
if *label == "logits" {
let arg = |v: &[f32]| v.iter().enumerate().fold((0usize, f32::MIN), |m, (i, &x)| if x > m.1 { (i, x) } else { m }).0;
let (ha, da) = (arg(hv), arg(dv));
lines.push(format!(
"qwen4 cuda check pos {at} logits: rel {r:.2e}, argmax host {ha} cuda {da} ({})",
if ha == da { "same" } else { "DIFFERENT" }
));
continue;
}
if *li != cur {
cur = *li;
lines.push(format!("qwen4 cuda check pos {at} L{li}:"));
}
if let Some(line) = lines.last_mut() {
line.push_str(&format!(" {label} {r:.1e}"));
}
}
if ids.len() > 1 || std::env::var_os("CMF_QWEN_CUDA_CHECK_ALL").is_some() {
for l in &lines {
eprintln!("{l}");
}
} else if let Some(l) = lines.iter().rev().find(|l| l.contains(" logits: ")) {
eprintln!("{l}");
}
eprintln!(
"qwen4 cuda check pos {at}: {} stages compared, worst {} at layer {} rel {:.2e}",
dev.trace.len(),
worst.1,
worst.2,
worst.0
);
}
#[allow(clippy::too_many_arguments)]
pub fn forward_tokens(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
ids: &[u32],
pos0: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
logits: &mut Vec<f32>,
want_logits: bool,
) {
#[cfg(feature = "cuda")]
if cuda_forward(globals, layers, cfg, state, ids, pos0, inv_freq, pool, logits, want_logits) {
return;
}
#[cfg(feature = "gpu")]
if ids.len() > 1 && state.masked {
state.masked = false;
if let Some(d) = state.dev.as_mut() {
d.set_masked(false);
}
}
#[cfg(feature = "gpu")]
if ids.len() > 1 && state.row_gen {
route_row("gen", &state.route_hist, state.route_tokens, &mut state.row_snap, &mut state.row_snap_tokens);
state.row_gen = false;
trace_mark(1);
trace_flush();
} else if ids.len() > 1 && pos0 == 0 {
trace_mark(1);
}
forward_tokens_stream(globals, layers, cfg, state, ids, pos0, inv_freq, pool, logits, want_logits);
#[cfg(feature = "gpu")]
if want_logits && ids.len() > 1 {
prompt_done(layers, state);
}
}
#[allow(clippy::too_many_arguments)]
fn forward_tokens_stream(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
ids: &[u32],
pos0: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
logits: &mut Vec<f32>,
want_logits: bool,
) {
#[cfg(feature = "gpu")]
{
use crate::gpu_wgpu::qwen4 as q4;
if stream_on() && !state.resident_all && !state.device_off && ids.len() > 3 * q4::TMAX {
let seen = pos0.min(state.token_history.len());
ple_prefetch(layers, cfg, &state.token_history[..seen], ids);
let head = if state.dev.is_none() || pos0 == 0 || state.pos != pos0 { q4::TMAX } else { 0 };
if head > 0 {
let mut lg = Vec::new();
forward_tokens_split(globals, layers, cfg, state, &ids[..head], pos0, inv_freq, pool, &mut lg, false);
}
let tail = if want_logits { q4::TMAX } else { 0 };
let (a, b) = (head, ids.len() - tail);
if !state.resident_all
&& !state.device_off
&& b > a + q4::TMAX
&& forward_stream_device(globals, layers, cfg, state, &ids[a..b], pos0 + a, inv_freq, pool)
{
if tail > 0 {
forward_tokens_split(globals, layers, cfg, state, &ids[b..], pos0 + b, inv_freq, pool, logits, want_logits);
} else {
logits.clear();
}
return;
}
forward_tokens_split(globals, layers, cfg, state, &ids[head..], pos0 + head, inv_freq, pool, logits, want_logits);
return;
}
}
forward_tokens_split(globals, layers, cfg, state, ids, pos0, inv_freq, pool, logits, want_logits)
}
#[allow(clippy::too_many_arguments)]
fn forward_tokens_split(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
ids: &[u32],
pos0: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
logits: &mut Vec<f32>,
want_logits: bool,
) {
#[cfg(feature = "gpu")]
{
use crate::gpu_wgpu::qwen4 as q4;
let big = state.resident_all && !state.device_off;
let cap = if big { q4::TROWS } else { q4::TMAX };
if ids.len() > q4::TMAX && (ids.len() > cap || want_logits) {
let mut at = 0usize;
while at < ids.len() {
let rem = ids.len() - at;
let big = state.resident_all && !state.device_off;
let n = if !big {
rem.min(q4::TMAX)
} else if want_logits && rem > q4::TMAX {
(rem - q4::TMAX).min(q4::TROWS)
} else {
rem.min(if want_logits { q4::TMAX } else { q4::TROWS })
};
let last = at + n == ids.len();
let mut lg = Vec::new();
forward_tokens_split(
globals,
layers,
cfg,
state,
&ids[at..at + n],
pos0 + at,
inv_freq,
pool,
&mut lg,
want_logits && last,
);
if last {
*logits = lg;
}
at += n;
}
if want_logits && std::env::var("CMF_QWEN_TOPLOG").as_deref() == Ok("1") {
top_log(logits, globals.lm_head.rows());
}
return;
}
}
#[cfg(feature = "gpu")]
if !ids.is_empty()
&& (ids.len() <= crate::gpu_wgpu::qwen4::TMAX
|| (state.resident_all && !want_logits && ids.len() <= crate::gpu_wgpu::qwen4::TROWS))
&& forward_tokens_device(
globals,
layers,
cfg,
state,
ids,
pos0,
inv_freq,
pool,
logits,
want_logits,
)
{
return;
}
for (i, &id) in ids.iter().enumerate() {
let last = i + 1 == ids.len();
let mut lg = Vec::new();
forward_token(
globals,
layers,
cfg,
state,
id,
pos0 + i,
inv_freq,
pool,
&mut lg,
want_logits && last,
);
if last {
*logits = lg;
}
}
}
#[cfg(feature = "gpu")]
#[allow(clippy::too_many_arguments)]
pub fn capture_window(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
ids: &[u32],
lo: usize,
hi: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
) -> Result<Vec<Vec<f32>>, String> {
use crate::gpu_wgpu::qwen4 as q4;
let first = layers.first().ok_or("no layers")?;
let hidden = first.moe.experts.first().ok_or("no experts")?.gate_proj.cols();
let ne = first.moe.experts.len();
let hi = hi.min(layers.len());
if lo >= hi || !q4::capture_begin(lo, hi, hidden, ne) {
return Err("capture unavailable".into());
}
let mut out = vec![Vec::with_capacity(ids.len() * (hidden + ne)); hi - lo];
let chunk = prefill_chunk();
let mut pos = 0;
let res = (|| {
while pos < ids.len() {
let end = (pos + chunk).min(ids.len());
let mut lg = Vec::new();
forward_tokens(globals, layers, cfg, state, &ids[pos..end], pos, inv_freq, pool, &mut lg, false);
let (rows, written) = q4::capture_read(end - pos).ok_or("capture readback")?;
if written < hi - lo {
return Err(format!("the device path ran {written} of {} captured layer frames (host fallback?)", hi - lo));
}
for (k, r) in rows.into_iter().enumerate() {
out[k].extend(r);
}
pos = end;
}
Ok(())
})();
q4::capture_end();
res.map(|_| out)
}
pub fn expert_dirs(layers: &[Layer], li: usize) -> Option<(Vec<(usize, usize, usize)>, Option<(usize, usize, usize)>)> {
let l = layers.get(li)?;
Some((l.expert_ids.clone(), l.shared_ids))
}
#[cfg(feature = "gpu")]
#[allow(clippy::too_many_arguments)]
fn forward_token_device(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
token_id: u32,
position: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
logits: &mut Vec<f32>,
want_logits: bool,
) -> bool {
forward_tokens_device(
globals,
layers,
cfg,
state,
&[token_id],
position,
inv_freq,
pool,
logits,
want_logits,
)
}
#[cfg(feature = "gpu")]
#[allow(clippy::too_many_arguments)]
fn forward_tokens_device(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
ids: &[u32],
pos0: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
logits: &mut Vec<f32>,
want_logits: bool,
) -> bool {
use crate::gpu_wgpu::qwen4 as q4;
let ntok = ids.len();
let frame_ok = state.copy_frame
&& (state.resident_all || state.masked)
&& state.dev.as_ref().is_some_and(|d| d.o1.is_none());
let cap = if (state.resident_all && !want_logits) || frame_ok { q4::TROWS } else { q4::TMAX };
if ntok == 0 || ntok > cap {
return false;
}
let position = pos0;
let token_id = ids[0];
if state.device_off
|| std::env::var("CMF_QWEN_DEVICE").as_deref() == Ok("0")
|| !crate::gpu::enabled_here()
{
return false;
}
fn off(state: &mut State, why: &str) -> bool {
if !state.device_off {
tracing::warn!("qwen4 device path off: {why}");
if std::env::var_os("CMF_QWEN_PROF").is_some() {
eprintln!("qwen4-device: off ({why})");
}
}
state.device_off = true;
false
}
if !q4::available() {
return off(state, "kernels or the global expert arena unavailable");
}
let Some(first) = layers.first() else {
return off(state, "no layers");
};
let Some(e0) = first.moe.experts.first() else {
return off(state, "no experts");
};
let Some(model) = e0.gate_proj.model_arc() else {
return off(state, "experts are not mmap-backed");
};
let gu_q2 = e0.gate_proj.model_dtype() == Some(TensorDtype::Q2TiledP);
if !gu_q2 && e0.gate_proj.model_dtype() != Some(TensorDtype::Q4TiledP) {
return off(state, "expert gate/up are neither q2tp nor q4tp");
}
if first.moe.router_sigmoid
|| !first.moe.norm_topk_prob
|| first.moe.route_tau.is_some()
|| first.moe.mask.is_some()
|| (first.moe.routed_scaling - 1.0).abs() > 1e-9
{
return off(state, "non-Qwen routing settings");
}
let g = device_geom(cfg, &first.moe, gu_q2);
let prof = std::env::var_os("CMF_QWEN_PROF").is_some();
let check = std::env::var("CMF_QWEN_DEVICE_CHECK").as_deref() == Ok("1");
let t_token = std::time::Instant::now();
if state.dev.is_none() {
if layers.iter().any(|l| l.idx.is_none()) || globals.lm_head_idx.is_none() {
return off(state, "skeleton tensors missing from the directory");
}
let kinds: Vec<(bool, bool)> = layers
.iter()
.map(|l| (matches!(l.mixer, Mixer::Gdn(_)), l.ple.is_some()))
.collect();
let Some(mut dev) = q4::Dev::new(model.uid(), &g, &kinds) else {
return off(state, "device state allocation refused");
};
if let Some(o1) = crate::qwen4_o1::Qwen4O1::from_model(&model) {
match o1.check(g.head_dim) {
Ok(()) => {
let (m, w, sink, seal) = (o1.o1.m, o1.o1.w, o1.o1.sink, o1.o1.seal);
let exact = o1.exact.as_ref().map_or(0, |x| x.budget);
if dev.set_o1(&g, o1) {
tracing::info!("qwen4 O(1): {m} landmarks, window {w}, {sink} sinks, {exact} exact rows, seal at {seal}");
}
}
Err(e) => tracing::warn!("{e}; the attention layers stay exact"),
}
}
let tubes: Vec<Vec<Option<(usize, usize, usize)>>> = layers
.iter()
.map(|l| {
l.moe
.experts
.iter()
.map(|e| match e.segs.as_slice() {
[s] => Some((s.gate.model_idx()?, s.up.model_idx()?, s.down.model_idx()?)),
_ => None,
})
.collect()
})
.collect();
if layers.iter().any(|l| l.moe.experts.iter().any(|e| e.segs.len() > 1)) {
return off(state, "an expert with more than one tube");
}
let shared: Vec<Option<(usize, usize, usize)>> = layers
.iter()
.map(|l| match l.moe.shared.as_ref().map(|(se, _)| se.segs.as_slice()) {
Some([s]) => Some((s.gate.model_idx()?, s.up.model_idx()?, s.down.model_idx()?)),
_ => None,
})
.collect();
if layers.iter().any(|l| l.moe.shared.as_ref().is_some_and(|(se, _)| se.segs.len() > 1)) {
return off(state, "a shared expert with more than one tube");
}
let t0 = std::time::Instant::now();
let Some(bytes) = q4::prewarm(&model, &globals.skeleton_idxs) else {
return off(state, "skeleton does not fit the VRAM budget");
};
tracing::info!(
"qwen4 device: {} skeleton tensors resident ({} MB) in {:.2}s",
globals.skeleton_idxs.len(),
bytes >> 20,
t0.elapsed().as_secs_f64()
);
if tubes.iter().flatten().any(Option::is_some) || shared.iter().any(Option::is_some) {
if !dev.set_tubes(&model, &g, &tubes, &shared) {
return off(state, "expert tubes do not load or do not fit the VRAM budget");
}
tracing::info!("qwen4 tubes: {} MB of expert tubes resident", dev.tube_bytes() >> 20);
}
if prof {
eprintln!(
"qwen4-device: skeleton {} MB resident in {:.2}s",
bytes >> 20,
t0.elapsed().as_secs_f64()
);
}
state.dev = Some(dev);
state.picks_prev = vec![Vec::new(); layers.len()];
}
if state.gpu_pool.is_none() {
if state.mtp.is_none() && !state.mtp_tried {
state.mtp_tried = true;
state.mtp = MtpHead::load(&model, model.arch());
if let Some(head) = state.mtp.as_mut()
&& let Err(why) = head.setup(&device_geom(cfg, &head.layer.moe, gu_q2), gu_q2)
{
tracing::warn!("qwen4 MTP off: {why}");
state.mtp = None;
}
}
let Some(budget) = q4::vram_budget() else {
return off(state, "no VRAM budget");
};
let resident = q4::resident_bytes();
let Some(per) = QwenGpuPool::per_expert_bytes(g.inter, g.hidden, gu_q2) else {
return off(state, "expert geometry");
};
let reserve_mb = std::env::var("CMF_QWEN_KV_RESERVE_MB")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(768);
let workspace = std::env::var("CMF_QWEN_WORKSPACE_MB")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.map(|m| m << 20)
.unwrap_or(if budget >= 24 << 30 {
0
} else {
(budget / 10).clamp(2 << 30, 4 << 30)
});
let window = state.mtp.as_ref().map_or(0, |h| h.k + 1);
let snap_extra = {
let kinds: Vec<(bool, bool)> = layers
.iter()
.map(|l| (matches!(l.mixer, Mixer::Gdn(_)), l.ple.is_some()))
.collect();
q4::Dev::snap_bytes(&g, &kinds, window.saturating_sub(MTP_MEASURED_WINDOW))
};
let free = budget
.saturating_sub(resident)
.saturating_sub(reserve_mb << 20)
.saturating_sub(workspace)
.saturating_sub(snap_extra);
let mut slots = (free / per as u64) as usize;
let n_layers = layers.len();
let whole = n_layers * g.n_experts + n_layers + q4::TMAX * g.top_k + 64;
if std::env::var("CMF_QWEN_RESIDENT").as_deref() != Ok("0") && slots >= whole {
slots = whole;
}
let staging = q4::TMAX * g.top_k;
let Some(arena) = QwenGpuPool::create_explicit(
&model,
g.inter,
g.hidden,
n_layers,
g.n_experts,
gu_q2,
slots,
staging,
) else {
return off(state, "expert arena allocation refused");
};
if prof {
eprintln!(
"qwen4-device: budget {} MB, skeleton resident {} MB, reserve {} MB, workspace {} MB, window snapshots beyond {MTP_MEASURED_WINDOW} rows {} MB, arena request {} slots ({} MB), got {} slots",
budget >> 20,
resident >> 20,
reserve_mb,
workspace >> 20,
snap_extra >> 20,
slots,
(slots as u64 * per as u64) >> 20,
arena.owner.len()
);
}
let (cap_slots, cap_staging) = (arena.staging_base, arena.staging);
state.gpu_pool = Some(arena);
let pinned = {
let arena = state.gpu_pool.as_mut().unwrap();
layers.iter().enumerate().all(|(li, l)| {
arena
.ensure(&model, li, &[], &l.expert_ids, l.shared_ids)
.is_some_and(|(_, s)| s != u32::MAX)
})
};
if !pinned {
return off(state, "shared expert could not be pinned");
}
let n_experts = g.n_experts;
let triples: Vec<Vec<(usize, usize, usize)>> =
layers.iter().map(|l| l.expert_ids.clone()).collect();
let store = (std::env::var("CMF_QWEN_STORE").as_deref() != Ok("0"))
.then(|| {
crate::expert_store::ExpertStore::new(&model, &triples, n_experts, cap_slots)
})
.flatten();
let advised =
crate::expert_store::advise_random(&model, |n| n.contains("ngram_embedding.shard_"));
if let Some(store) = store.as_ref() {
state.gpu_pool.as_mut().unwrap().store = Some(store.clone());
if prof {
eprintln!(
"qwen4-device: expert store io {:?}, RAM tier {} experts, n-gram table {} MiB advised random",
store.io_mode(),
store.tier_capacity(),
advised >> 20
);
}
}
let mut profile = std::env::var("CMF_QWEN_PROFILE")
.ok()
.and_then(|path| ExpertProfile::load(&path, layers.len(), n_experts));
let mut warm_order: Vec<(usize, usize)> = Vec::new();
if let Some(pr) = profile.as_ref() {
let ranked = pr.ranked();
warm_order = ranked.clone();
let t0 = std::time::Instant::now();
let n = state
.gpu_pool
.as_mut()
.map_or(0, |a| a.prefill(&model, &ranked, &triples));
tracing::info!(
"qwen4 device: arena prefilled with {n} of {} profiled experts ({} tokens of history) in {:.2}s",
ranked.len(),
pr.tokens,
t0.elapsed().as_secs_f64()
);
if prof {
eprintln!(
"qwen4-device: arena prefilled with {n} of {} profiled experts in {:.2}s (capacity {} slots, {} staging)",
ranked.len(),
t0.elapsed().as_secs_f64(),
cap_slots,
cap_staging
);
}
}
if std::env::var("CMF_QWEN_RESIDENT").as_deref() != Ok("0")
&& cap_slots >= layers.len() * n_experts
{
let mut seen = vec![false; layers.len() * n_experts];
let mut all: Vec<(usize, usize)> = Vec::with_capacity(seen.len());
for &(l, e) in &warm_order {
if l < layers.len() && e < n_experts && !std::mem::replace(&mut seen[l * n_experts + e], true) {
all.push((l, e));
}
}
for l in 0..layers.len() {
for e in 0..n_experts {
if !seen[l * n_experts + e] {
all.push((l, e));
}
}
}
let t0 = std::time::Instant::now();
let arena = state.gpu_pool.as_mut().unwrap();
arena.prefill(&model, &all, &triples);
let held = arena.slot_for.iter().take(layers.len() * n_experts).filter(|&&s| s != u32::MAX).count();
state.resident_all = held == layers.len() * n_experts;
tracing::info!(
"qwen4 device: {held} of {} routed experts resident in {:.2}s{}",
layers.len() * n_experts,
t0.elapsed().as_secs_f64(),
if state.resident_all { " — whole-token chained frames" } else { "" }
);
}
{
let n_keys = layers.len() * n_experts;
let want_slots = n_keys.saturating_sub(cap_slots) + cap_slots / 8;
let stride = (per as u64).div_ceil(256) * 256 + 1024;
let want = want_slots as u64 * stride;
let asked = match std::env::var("CMF_QWEN_HTIER_MB").ok().as_deref() {
None | Some("") if cap_slots < n_keys => want,
None | Some("") => 0,
Some("auto") => want,
Some(v) => v.parse::<u64>().map_or(0, |mb| mb << 20),
};
let cap_bytes = if asked == 0 {
0
} else {
crate::expert_store::host_available_bytes().map_or(0, |avail| {
asked
.min(avail / 2)
.min(avail.saturating_sub(16 << 30))
})
};
if cap_bytes > 0 {
let t0 = std::time::Instant::now();
let tier = q4::HostTier::new(&model, n_keys, cap_bytes);
if prof {
match tier.as_ref() {
Some(t) => eprintln!(
"qwen4-device: host tier {} slots ({} MiB, first GiB in {:.0} ms, the rest in the background after the first frame)",
t.capacity(),
(t.capacity() as u64 * t.stride()) >> 20,
t0.elapsed().as_secs_f64() * 1e3
),
None => eprintln!("qwen4-device: host tier unavailable"),
}
}
let fill_threads = std::env::var("CMF_QWEN_HTIER_FILL")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(8);
if let Some(t) = tier.as_ref()
&& fill_threads > 0
{
let arena = state.gpu_pool.as_ref().unwrap();
let mut order: Vec<usize> = Vec::with_capacity(n_keys);
let mut seen = vec![false; n_keys];
let ranked = profile.as_ref().map(|p| p.ranked()).unwrap_or_default();
for &(l, e) in &ranked {
if l < layers.len() && e < n_experts {
let k = l * n_experts + e;
if !seen[k] && arena.slot_for[k] == u32::MAX {
seen[k] = true;
order.push(k);
}
}
}
for e in 0..n_experts {
for l in 0..layers.len() {
let k = l * n_experts + e;
if !seen[k] && arena.slot_for[k] == u32::MAX {
seen[k] = true;
order.push(k);
}
}
}
let flat: Vec<(usize, usize, usize)> =
triples.iter().flat_map(|r| r.iter().copied()).collect();
if flat.len() == n_keys {
t.start_fill(model.clone(), Arc::new(flat), order, fill_threads);
}
}
state.gpu_pool.as_mut().unwrap().tier = tier;
}
}
if std::env::var_os("CMF_QWEN_PROFILE_SAVE").is_some() && profile.is_none() {
profile = Some(ExpertProfile::new(layers.len(), n_experts));
}
state.profile = if std::env::var_os("CMF_QWEN_PROFILE_SAVE").is_some() { profile } else { None };
if let Some(store) = store.as_ref() {
let mut seen = vec![false; layers.len() * n_experts];
for &(l, e) in &warm_order {
if l < layers.len() && e < n_experts {
seen[l * n_experts + e] = true;
}
}
for e in 0..n_experts {
for l in 0..layers.len() {
if !seen[l * n_experts + e] {
warm_order.push((l, e));
}
}
}
store.start_background(warm_order);
}
}
if position == 0 || state.pos != position {
state.reset();
let ok = state.dev.as_mut().is_some_and(|d| d.reset());
if !ok {
return off(state, "device reset failed");
}
}
let frame_mode = state.copy_frame && ntok > q4::TMAX;
let resident_all = state.resident_all || (state.masked && (ntok <= q4::TMAX || frame_mode));
let frame_verify = frame_mode && want_logits && state.frame_check;
let frame_dbg = frame_verify && copy_frame_debug();
let defer_readback = resident_all
&& !want_logits
&& ntok > 1
&& !state.verify_window
&& state.profile.is_none()
&& std::env::var("CMF_QWEN_PIPELINE").as_deref() != Ok("0")
&& std::env::var_os("CMF_QWEN_DUMP_HYPER").is_none();
if !defer_readback && let Some(p) = state.inflight.take() {
if p.wait().is_none() {
return off(state, "readback");
}
}
let dev = state.dev.as_mut().unwrap();
let arena = state.gpu_pool.as_mut().unwrap();
let mut pf = DevProf::default();
let cold_on_device = std::env::var("CMF_QWEN_COLD_DEVICE").as_deref() != Ok("0");
let chain_env = std::env::var("CMF_QWEN_CHAIN").ok().and_then(|v| v.parse::<usize>().ok());
let chain_len = if dev.o1.is_some() && !resident_all {
1
} else if resident_all {
let chains = std::env::var("CMF_QWEN_CHAINS")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(4)
.max(1);
let split = want_logits && (ntok <= crate::gpu_wgpu::qwen4::TMAX || frame_mode);
chain_env
.unwrap_or(if split { (layers.len() + 1).div_ceil(chains) } else { layers.len() + 1 })
.max(1)
} else if ntok > 1 {
1
} else {
std::env::var("CMF_QWEN_CHAIN")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(1)
.max(1)
};
let n = layers.len();
let top_k = g.top_k;
let cs = q4::cold_stride(&g);
let hidden_bytes = cfg.hidden * 4;
let cold_part = (ntok * cs).div_ceil(16) * 16;
let lean = resident_all
&& state.profile.is_none()
&& !check
&& std::env::var_os("CMF_QWEN_DUMP_HYPER").is_none()
&& std::env::var("CMF_QWEN_LEAN").as_deref() != Ok("0");
let frame_bytes = if lean { 0 } else { cold_part + (ntok * hidden_bytes).div_ceil(16) * 16 };
let snapshot = state.verify_window;
if snapshot {
dev.ensure_snaps(&g, ntok);
}
let mut cold_vec: Vec<Option<Vec<f32>>> = vec![None; ntok];
let mut cold_slots: Vec<Vec<(u32, f32)>> = vec![Vec::with_capacity(top_k); ntok];
let mut failed: Option<&'static str> = None;
let head_hc = device_hc_w(&globals.head_hc);
let lm_head_idx = globals.lm_head_idx.unwrap_or(usize::MAX);
let logit_rows = match (want_logits, snapshot) {
(false, _) => 0,
(true, false) if frame_dbg => ntok + 1,
(true, false) => 1 + usize::from(frame_verify),
(true, true) => ntok,
};
let stage_need = q4::chain_stage_bytes(
chain_len.min(n),
frame_bytes,
logit_rows,
q4::head_stride(&model, lm_head_idx).unwrap_or(0),
);
if !dev.ensure_stage(stage_need) {
failed = Some("readback stage allocation");
}
if frame_verify && !dev.checkpoint(&g) {
failed = Some("copy frame checkpoint");
}
for (t, &id) in ids.iter().enumerate() {
let mut emb = vec![0.0f32; cfg.hidden];
if (id as usize) < globals.embed.rows() {
globals.embed.row_f32(id as usize, &mut emb);
}
dev.seed(t, &emb, cfg.hc);
}
let tp = std::time::Instant::now();
let ple_rows = ple_frame_rows(layers, cfg, &state.token_history, ids, pool);
pf.ple += tp.elapsed();
let frames_total = if want_logits { n + 1 } else { n };
dev.gated = chain_len > 1 && !resident_all;
let mut fi = 0usize;
let mut vocab_out = 0usize;
let free_run = lean
&& resident_all
&& !dev.gated
&& !q4::ts_probe()
&& std::env::var("CMF_QWEN_FREE_RUN").as_deref() != Ok("0");
let mut prepared: Option<(wgpu::CommandBuffer, u64, Option<usize>, usize)> = None;
let dump_hyper = std::env::var("CMF_QWEN_DUMP_HYPER").ok();
let mut t_fence: Option<std::time::Instant> = None;
let mut lstride_out = 0usize;
while fi < frames_total && failed.is_none() {
let hi = (fi + chain_len).min(frames_total);
let t0 = std::time::Instant::now();
let encode_chain = |lo: usize,
hi: usize,
dev: &mut q4::Dev,
arena: &mut QwenGpuPool,
picks_prev: &[Vec<usize>],
pf: &mut DevProf,
vocab_out: &mut usize|
-> Result<
(wgpu::CommandEncoder, u64, Option<usize>, usize),
&'static str,
> {
let mut enc = q4::new_encoder("qwen4-chain").ok_or("encoder")?;
q4::dts_reset();
let merge = q4::merge_guard(&enc);
q4::ts_mark(&mut enc, 0);
if !q4::encode_pending(&mut enc, dev, &g, ntok) {
return Err("pending inject");
}
q4::pending_done(dev);
q4::ts_mark(&mut enc, 1);
let mut stage_off = 0u64;
let mut logits_off = None;
let mut lstride = 0usize;
for f in lo..hi {
let inject_prev = f > lo;
if f == n {
for i in 2..8 {
q4::ts_mark(&mut enc, i);
}
dev.keep_r(&mut enc, ntok - 1);
let hw = head_hc.as_ref().ok_or("head hyper-connection indices")?;
let first = if snapshot || frame_dbg { 0 } else { ntok - 1 };
let (lb, vocab, ls) = q4::encode_head(
&mut enc,
dev,
&model,
&g,
hw,
lm_head_idx,
inject_prev,
ntok,
)
.ok_or("head frame declined")?;
let bytes = ((ntok - first) * ls) as u64;
if !q4::copy_to_stage(&mut enc, dev, &lb, (first * ls) as u64, stage_off, bytes)
{
return Err("logits overrun the readback stage");
}
logits_off = Some(stage_off as usize);
*vocab_out = vocab;
lstride = ls;
stage_off += bytes.div_ceil(16) * 16;
if frame_verify {
let idb = q4::encode_argmax_rows(&mut enc, dev, &lb, vocab, ls, ntok)
.ok_or("copy frame argmax")?;
if !q4::copy_to_stage(&mut enc, dev, &idb, 0, stage_off, (ntok * 4) as u64) {
return Err("copy frame ids overrun the readback stage");
}
stage_off += ((ntok * 4) as u64).div_ceil(16) * 16;
}
continue;
}
let layer = &layers[f];
let ix = layer.idx.as_ref().ok_or("layer index")?;
let w = device_layer_w(layer, ix).ok_or("layer weights")?;
let ta = std::time::Instant::now();
let (remap, shared_slot) = match arena.ensure(
&model,
f,
&picks_prev[f],
&layer.expert_ids,
layer.shared_ids,
) {
Some(r) => r,
None => arena.remap_snapshot(f),
};
pf.admit += ta.elapsed();
if shared_slot == u32::MAX {
return Err("shared expert slot");
}
let out = q4::encode_layer(
&mut enc,
dev,
&model,
&g,
&w,
f,
pos0,
ntok,
inv_freq,
&remap,
shared_slot,
&ple_rows[f],
inject_prev,
snapshot,
)
.ok_or("layer frame declined")?;
if lean {
continue;
}
if !q4::copy_to_stage(&mut enc, dev, &out.cold, 0, stage_off, (ntok * cs) as u64)
|| !q4::copy_to_stage(
&mut enc,
dev,
&out.x2,
0,
stage_off + cold_part as u64,
(ntok * hidden_bytes) as u64,
)
{
return Err("layer frame overruns the readback stage");
}
stage_off += frame_bytes as u64;
}
q4::ts_mark(&mut enc, 8);
q4::ts_resolve(&mut enc);
dev.arm_chain(lo, hi);
drop(merge);
Ok((enc, stage_off, logits_off, lstride))
};
let (cb, stage_off, logits_off, ls) = match prepared.take() {
Some(p) => p,
None => match encode_chain(
fi,
hi,
dev,
arena,
&state.picks_prev,
&mut pf,
&mut vocab_out,
) {
Ok((e, a, b, c)) => {
let tf = std::time::Instant::now();
let cb = q4::finish_frame(e);
pf.finish += tf.elapsed();
(cb, a, b, c)
}
Err(why) => {
failed = Some(why);
break;
}
},
};
if ls != 0 {
lstride_out = ls;
}
pf.encode += t0.elapsed();
if !q4::finalize_pending(dev, &g, ntok, &cold_vec, &cold_slots) {
failed = Some("finalize");
break;
}
let t0 = std::time::Instant::now();
if free_run && hi < frames_total && stage_off == 0 && arena.tier.is_none() {
let uploads = arena.take_uploads();
if q4::submit_indexed(uploads, cb).is_none() {
failed = Some("submit");
break;
}
arena.rearm_uploads();
for f in fi..hi.min(n) {
dev.commit(f);
}
pf.wait += t0.elapsed();
pf.chains += 1;
fi = hi;
continue;
}
let uploads = arena.take_uploads();
let Some(pend) = q4::submit_frame(dev, uploads, cb, stage_off) else {
failed = Some("submit");
break;
};
arena.rearm_uploads();
if defer_readback && hi >= frames_total {
if let Some(prev) = state.inflight.replace(pend)
&& prev.wait().is_none()
{
failed = Some("readback");
break;
}
arena.uploads_done();
pf.chains += 1;
fi = hi;
continue;
}
if let Some(tf) = t_fence.take() {
pf.post += tf.elapsed();
}
arena.hold_layers = [None, None];
if (chain_len == 1 || resident_all) && hi < frames_total {
let te = std::time::Instant::now();
match encode_chain(
hi,
(hi + chain_len).min(frames_total),
dev,
arena,
&state.picks_prev,
&mut pf,
&mut vocab_out,
) {
Ok((e, a, b, c)) => {
let tf = std::time::Instant::now();
prepared = Some((q4::finish_frame(e), a, b, c));
pf.finish += tf.elapsed();
}
Err(why) => {
failed = Some(why);
break;
}
}
pf.encode += te.elapsed();
arena.hold_layers = [(hi < n).then_some(hi), (fi < n).then_some(fi)];
}
let ts = std::time::Instant::now();
let Some(bytes) = pend.wait() else {
failed = Some("readback");
break;
};
arena.uploads_done();
pf.spin += ts.elapsed();
pf.wait += t0.elapsed();
if let Some(m) = q4::ts_read() {
pf.gpu_note(fi, n, &m, layers.get(fi).map(|l| matches!(l.mixer, Mixer::Gdn(_))));
}
t_fence = Some(std::time::Instant::now());
pf.chains += 1;
let mut miss_at: Option<usize> = None;
let mut off = 0usize;
'frames: for f in fi..hi {
if f == n {
break;
}
let layer = &layers[f];
dev.commit(f);
if lean {
continue 'frames;
}
if let Some(dir) = dump_hyper.as_deref()
&& let Some(rows) = q4::read_hyper(dev, ntok, cfg.hc * cfg.hidden)
{
let _ = std::fs::create_dir_all(&dir);
for (t, row) in rows.iter().enumerate() {
let path = std::path::Path::new(dir)
.join(format!("pos{:05}_layer{f:02}.f32", pos0 + t));
let bytes: Vec<u8> = row.iter().flat_map(|v| v.to_le_bytes()).collect();
let _ = std::fs::write(path, bytes);
}
}
let mut any_cold_in_frame = false;
let frame_off = off;
off += frame_bytes;
let mut pre_slots: Vec<Option<Vec<Option<u32>>>> = vec![None; ntok];
if cold_on_device && ntok > 1 && admit_batch_env() {
let ta = std::time::Instant::now();
let mut toks = Vec::new();
let mut reqs = Vec::new();
for t in 0..ntok {
let w_off = frame_off + t * cs;
let word = |j: usize| {
let b = &bytes[w_off + 4 * j..w_off + 4 * j + 4];
u32::from_le_bytes([b[0], b[1], b[2], b[3]])
};
let eids: Vec<usize> = (0..top_k)
.map(|j| word(2 * j))
.filter(|&e| e != u32::MAX && (e as usize) < layer.moe.experts.len())
.map(|e| e as usize)
.collect();
if eids.is_empty() {
continue;
}
let picks: Vec<usize> = (0..top_k)
.map(|j| word(2 * top_k + 2 * j))
.filter(|&e| e != u32::MAX && (e as usize) < g.n_experts)
.map(|e| e as usize)
.collect();
toks.push(t);
reqs.push((eids, picks));
}
if !reqs.is_empty() {
let got = arena.admit_frame(&model, f, &reqs, &layer.expert_ids);
for (t, sl) in toks.into_iter().zip(got) {
pre_slots[t] = Some(sl);
}
}
pf.admit += ta.elapsed();
}
let mut trace_row: Option<Vec<u16>> = (!check && trace_on()).then(|| Vec::with_capacity(ntok * top_k));
for t in 0..ntok {
let w_off = frame_off + t * cs;
let words: Vec<u32> = bytes[w_off..w_off + 4 * top_k * 4]
.chunks_exact(4)
.map(|b| u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
.collect();
let x_off = frame_off + cold_part + t * hidden_bytes;
let picks: Vec<usize> = (0..top_k)
.map(|j| words[2 * top_k + 2 * j])
.filter(|&e| e != u32::MAX && (e as usize) < g.n_experts)
.map(|e| e as usize)
.collect();
if let Some(pr) = state.profile.as_mut() {
pr.note(f, &picks);
}
note_route(&mut state.route_hist, n, g.n_experts, f, &picks);
if route_wstat() {
note_route_weights(&words[2 * top_k..4 * top_k]);
}
if let Some(tr) = trace_row.as_mut() {
tr.extend((0..top_k).map(|j| picks.get(j).map_or(u16::MAX, |&e| e as u16)));
}
if check && ntok == 1 && tap_layer() == Some(f) && position < 3 && f + 1 == hi {
let bufs = q4::tap_bufs(dev, &g);
let h = cfg.hidden;
let parts: Vec<(&wgpu::Buffer, u64)> = vec![
(&bufs[0], (h * 4) as u64),
(&bufs[1], (h * 4) as u64),
(&bufs[2], (h * 4) as u64),
(&bufs[3], (h * 4) as u64),
(&bufs[4], (cfg.hc * h * 4) as u64),
];
if let Some(enc) = q4::new_encoder("qwen4-tap")
&& let Some(tb) = q4::submit_readback(enc, &parts)
{
let fl = |o: usize, cnt: usize| -> Vec<f32> {
tb[o..o + cnt * 4]
.chunks_exact(4)
.map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
.collect()
};
let al = |cnt: usize| ((cnt * 4) as u64).div_ceil(16) as usize * 16;
let mut o = 0;
let x = fl(o, h);
o += al(h);
let blk = fl(o, h);
o += al(h);
let x2v = fl(o, h);
o += al(h);
let mo = fl(o, h);
o += al(h);
let hyper = fl(o, cfg.hc * h);
DEV_TAP
.lock()
.unwrap()
.push((position, x, blk, x2v, mo, hyper));
}
}
let any_cold = (0..top_k).any(|j| words[2 * j] != u32::MAX);
if t + 1 == ntok {
state.picks_prev[f] = picks.clone();
}
cold_slots[t].clear();
cold_vec[t] = None;
if !any_cold {
continue;
}
any_cold_in_frame = true;
let ta = std::time::Instant::now();
let x2: Vec<f32> = bytes[x_off..x_off + cfg.hidden * 4]
.chunks_exact(4)
.map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
.collect();
let mut jobs: Vec<(&crate::pipeline::DenseFfn, f32)> = Vec::new();
let cold_list: Vec<(usize, f32)> = (0..top_k)
.filter_map(|j| {
let e = words[2 * j];
(e != u32::MAX && (e as usize) < layer.moe.experts.len())
.then(|| (e as usize, f32::from_bits(words[2 * j + 1])))
})
.collect();
let eids: Vec<usize> = cold_list.iter().map(|&(e, _)| e).collect();
let slots = if cold_on_device {
let mut s = match pre_slots[t].take() {
Some(s) if s.len() == eids.len() => s,
_ => arena.admit_many(&model, f, &eids, &layer.expert_ids, &picks),
};
if s.iter().any(Option::is_none) {
let rest: Vec<usize> = eids
.iter()
.zip(&s)
.filter(|(_, x)| x.is_none())
.map(|(&e, _)| e)
.collect();
let mut staged = arena
.stage_cold(&model, f, t, &rest, &layer.expert_ids, top_k)
.into_iter();
for x in s.iter_mut().filter(|x| x.is_none()) {
*x = staged.next().flatten();
}
}
s
} else {
vec![None; eids.len()]
};
for (&(e, wgt), slot) in cold_list.iter().zip(slots) {
match slot {
Some(sl) => cold_slots[t].push((sl, wgt)),
None => jobs.push((&layer.moe.experts[e], wgt)),
}
}
pf.admit += ta.elapsed();
pf.cold_experts += cold_slots[t].len() + jobs.len();
pf.cold_dev += cold_slots[t].len();
if !jobs.is_empty() {
let tc = std::time::Instant::now();
cold_vec[t] = Some(crate::pipeline::moe_cold_experts_cpu_with(&jobs, &x2, pool, false));
pf.cold_cpu += tc.elapsed();
}
if check
&& ntok == 1
&& tap_layer() == Some(f)
&& position < 3
&& let Some(cv) = cold_vec[t].as_ref()
&& let Some(last) = DEV_TAP.lock().unwrap().last_mut()
{
for (m, c) in last.4.iter_mut().zip(cv) {
*m += c;
}
}
}
if let Some(tr) = trace_row {
trace_frame(pos0, f, ntok, top_k, &tr);
}
if any_cold_in_frame {
pf.layers_with_cold += 1;
miss_at = Some(f);
break 'frames;
}
}
match miss_at {
Some(j) => {
dev.discard(j + 1, hi);
pf.aborted += hi - (j + 1);
fi = j + 1;
}
None => {
if let Some(lo) = logits_off {
let stride = lstride_out;
let read = |o: usize| -> Vec<f32> {
bytes[o..o + vocab_out * 4]
.chunks_exact(4)
.map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
.collect()
};
if snapshot {
state.window_rows =
Some(WindowRows { bytes, lo, stride, vocab: vocab_out, rows: ntok });
} else {
let back = if frame_dbg { ntok } else { 1 };
*logits = read(lo + (back - 1) * stride);
if frame_verify {
let io = lo + (back * stride).div_ceil(16) * 16;
state.frame_ids = bytes.get(io..io + ntok * 4).map(|b| {
b.chunks_exact(4).map(|w| u32::from_le_bytes([w[0], w[1], w[2], w[3]])).collect()
});
}
if frame_dbg {
let host: Vec<(u32, f32, f32)> = (0..ntok)
.map(|r| {
let row = read(lo + r * stride);
let (mut bi, mut bv, mut second) = (0usize, f32::MIN, f32::MIN);
for (i, &v) in row.iter().enumerate() {
if v >= bv {
second = bv;
bv = v;
bi = i;
} else if v > second {
second = v;
}
}
(bi as u32, bv, bv - second)
})
.collect();
eprintln!(
"qwen4-frame-debug nt={ntok} card={:?} host={:?} margin={:?}",
state.frame_ids.as_deref().unwrap_or(&[]),
host.iter().map(|h| h.0).collect::<Vec<_>>(),
host.iter().map(|h| (h.2 * 100.0).round() / 100.0).collect::<Vec<_>>(),
);
}
}
}
fi = hi;
}
}
}
arena.flush_uploads();
if let Some(why) = failed {
return off(state, why);
}
dev.clear_pending();
if !want_logits {
logits.clear();
}
if !check {
state.token_history.extend_from_slice(ids);
state.pos = pos0 + ntok;
if !lean {
state.route_tokens += ntok;
phi_dump(&state.route_hist, state.route_tokens, &mut state.phi_last, &mut state.phi_last_tokens);
}
}
if let Some(d) = state.dev.as_mut() {
d.pos = pos0 + ntok;
if !check && !snapshot && d.o1_seal_at().is_some_and(|at| pos0 + ntok >= at) {
d.seal(&g, pos0 + ntok);
}
}
if let (Some(pr), Ok(path)) = (
state.profile.as_mut(),
std::env::var("CMF_QWEN_PROFILE_SAVE"),
) {
pr.tokens += ntok as u64;
if pr.tokens % 32 < ntok as u64 {
pr.save(&path);
}
}
if prof {
let store_line = arena
.store
.as_ref()
.map(|s| s.report())
.unwrap_or_default();
let fill_ns = FILL_NS.swap(0, std::sync::atomic::Ordering::Relaxed);
let fill_n = FILL_N.swap(0, std::sync::atomic::Ordering::Relaxed);
let fills_wall = FILLS_WALL_NS.swap(0, std::sync::atomic::Ordering::Relaxed) as f64 / 1e6;
let store_line = match arena.tier.as_ref() {
Some(t) => {
use std::sync::atomic::Ordering::Relaxed;
format!(
"{store_line} htier hits={} fills={} fill_sum={:.1}ms nofit={} online={} bg_fills={}",
t.hits.swap(0, Relaxed),
t.fills.swap(0, Relaxed),
t.fill_ns.swap(0, Relaxed) as f64 / 1e6,
t.misses.swap(0, Relaxed),
t.online(),
t.bg_fills.load(Relaxed)
)
}
None => store_line,
};
eprintln!(
"qwen4-device pos={position} ntok={ntok} total={:.1}ms encode={:.1}ms finish={:.1}ms wait={:.1}ms spin={:.1}ms post={:.1}ms ple={:.1}ms cold_cpu={:.1}ms admit={:.1}ms chains={} aborted_frames={} cold={} experts ({} on device) in {} layers fills={fill_n} fill_sum={:.1}ms fill_wall={fills_wall:.1}ms {store_line}",
t_token.elapsed().as_secs_f64() * 1e3,
pf.encode.as_secs_f64() * 1e3,
pf.finish.as_secs_f64() * 1e3,
pf.wait.as_secs_f64() * 1e3,
pf.spin.as_secs_f64() * 1e3,
pf.post.as_secs_f64() * 1e3,
pf.ple.as_secs_f64() * 1e3,
pf.cold_cpu.as_secs_f64() * 1e3,
pf.admit.as_secs_f64() * 1e3,
pf.chains,
pf.aborted,
pf.cold_experts,
pf.cold_dev,
pf.layers_with_cold,
fill_ns as f64 / 1e6,
);
}
if prof && pf.gpu_last_end.is_some() {
let mut line = format!("qwen4-ts pos={position} ntok={ntok}");
for (name, v) in TS_NAMES.iter().zip(pf.gpu) {
line.push_str(&format!(" {name}={:.2}", v / 1e6));
}
line.push_str(&format!(" busy={:.2} ms", pf.gpu[..16].iter().sum::<f64>() / 1e6));
eprintln!("{line}");
}
if check && ntok == 1 {
let dev_logits = logits.clone();
let mut host = Vec::new();
state.device_off = true;
forward_token(
globals,
layers,
cfg,
state,
token_id,
position,
inv_freq,
pool,
&mut host,
want_logits,
);
state.device_off = false;
if let Some(tl) = tap_layer() {
let dev_taps = std::mem::take(&mut *DEV_TAP.lock().unwrap());
let host_taps = std::mem::take(&mut *TAP.lock().unwrap());
for (pos, x, blk, x2, mo, hyper) in &dev_taps {
let find = |lab: &str| {
host_taps
.iter()
.find(|(l, li, p, _)| l == lab && *li == tl && p == pos)
.map(|(_, _, _, v)| v.as_slice())
};
eprintln!("qwen4-tap layer {tl} pos {pos}:");
if let Some(h) = find("attn_in") {
tap_compare("attn_in", h, x);
}
if let Some(h) = find("attn_out") {
tap_compare("attn_out", h, blk);
}
if let Some(h) = find("post_attn") {
tap_compare("post_attn", h, hyper);
}
if let Some(h) = find("moe_in") {
tap_compare("moe_in", h, x2);
}
if let Some(h) = find("moe_out") {
tap_compare("moe_res", h, mo);
}
}
}
if !want_logits {
return true;
}
let n = host.len().min(dev_logits.len());
let mut max_abs = 0.0f32;
let (mut dot, mut na, mut nb) = (0.0f64, 0.0f64, 0.0f64);
for i in 0..n {
max_abs = max_abs.max((host[i] - dev_logits[i]).abs());
dot += host[i] as f64 * dev_logits[i] as f64;
na += host[i] as f64 * host[i] as f64;
nb += dev_logits[i] as f64 * dev_logits[i] as f64;
}
let argmax = |v: &[f32]| {
v.iter()
.enumerate()
.fold((0usize, f32::NEG_INFINITY), |a, (i, &x)| {
if x > a.1 { (i, x) } else { a }
})
.0
};
eprintln!(
"qwen4-check pos={position} cos={:.6} max_abs={max_abs:.4} argmax host={} dev={}",
dot / (na.sqrt() * nb.sqrt()).max(1e-30),
argmax(&host),
argmax(&dev_logits)
);
*logits = dev_logits;
}
true
}
#[cfg(feature = "gpu")]
#[allow(clippy::too_many_arguments)]
fn forward_stream_device(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
ids: &[u32],
pos0: usize,
inv_freq: &[f32],
pool: Option<&Pool>,
) -> bool {
use crate::gpu_wgpu::qwen4 as q4;
use std::time::{Duration, Instant};
let ntok = ids.len();
if ntok <= q4::TMAX
|| state.device_off
|| state.resident_all
|| state.verify_window
|| state.pos != pos0
|| state.profile.is_some()
|| state.dev.is_none()
|| state.gpu_pool.is_none()
|| q4::capture_on()
|| std::env::var_os("CMF_QWEN_DUMP_HYPER").is_some()
|| std::env::var("CMF_QWEN_DEVICE_CHECK").as_deref() == Ok("1")
{
return false;
}
let Some(first) = layers.first() else { return false };
let Some(e0) = first.moe.experts.first() else { return false };
let Some(model) = e0.gate_proj.model_arc() else { return false };
let gu_q2 = e0.gate_proj.model_dtype() == Some(TensorDtype::Q2TiledP);
let g = device_geom(cfg, &first.moe, gu_q2);
let n = layers.len();
let frames: Vec<(usize, usize)> =
(0..ntok).step_by(q4::TROWS).map(|a| (a, (ntok - a).min(q4::TROWS))).collect();
if state.gpu_pool.as_ref().is_some_and(|a| a.staging_base < g.n_experts + n + 1) {
return false;
}
let cs = q4::cold_stride(&g);
let cap_bytes = (n * frames.len() * q4::TROWS * cs) as u64;
{
let dev = state.dev.as_mut().unwrap();
if dev.o1.is_some() || !dev.stream_bank(&g, frames.len()) || !dev.route_cap_ensure(cap_bytes) {
return false;
}
dev.gated = false;
}
let prof = std::env::var_os("CMF_QWEN_PROF").is_some();
let t_all = Instant::now();
if let Some(p) = state.inflight.take()
&& p.wait().is_none()
{
state.device_off = true;
return false;
}
{
let dev = state.dev.as_ref().unwrap();
let mut emb = vec![0.0f32; cfg.hidden];
for (k, &(a, nt)) in frames.iter().enumerate() {
for t in 0..nt {
emb.fill(0.0);
let id = ids[a + t] as usize;
if id < globals.embed.rows() {
globals.embed.row_f32(id, &mut emb);
}
dev.seed_bank(k, t, &emb, cfg.hc);
}
}
}
let (mut t_admit, mut t_ple, mut t_enc, mut t_wait) = (Duration::ZERO, Duration::ZERO, Duration::ZERO, Duration::ZERO);
let (mut t_sub, mut waits, mut quick_waits) = (Duration::ZERO, 0usize, 0usize);
let lag_ms = std::env::var("CMF_QWEN_STREAM_LAG_MS").ok().and_then(|v| v.parse::<u64>().ok()).unwrap_or(0);
let stream_dts = std::env::var("CMF_QWEN_STREAM_DTS").as_deref() == Ok("1") && q4::dts_on();
if stream_dts {
q4::dts_stream_line(None);
}
let fill_n0 = FILL_N.load(std::sync::atomic::Ordering::Relaxed);
let mut prev: Option<wgpu::SubmissionIndex> = None;
let mut failed: Option<&'static str> = None;
'layers: for f in 0..n {
let layer = &layers[f];
let Some(ix) = layer.idx.as_ref() else {
failed = Some("layer index");
break;
};
let Some(w) = device_layer_w(layer, ix) else {
failed = Some("layer weights");
break;
};
let arena = state.gpu_pool.as_mut().unwrap();
arena.hold_layers = [None, None];
let ta = Instant::now();
let Some((remap, shared_slot)) = arena.ensure_layer(&model, f, &layer.expert_ids, layer.shared_ids) else {
failed = Some("the arena cannot hold a layer's experts");
break;
};
let mut uploads = arena.take_uploads();
t_admit += ta.elapsed();
let tp = Instant::now();
let rows = if layer.ple.is_some() {
ple_rows_of(layers, cfg, &state.token_history, ids, pool, Some(f)).swap_remove(f)
} else {
Vec::new()
};
t_ple += tp.elapsed();
let dev = state.dev.as_mut().unwrap();
for (k, &(a, nt)) in frames.iter().enumerate() {
let te = Instant::now();
let Some(mut enc) = q4::new_encoder("qwen4-stream") else {
failed = Some("encoder");
break 'layers;
};
let dts = stream_dts && k == 0;
if dts {
q4::dts_reset();
}
if !dev.stream_swap(&mut enc, &g, k, nt, false, f > 0) {
failed = Some("bank");
break 'layers;
}
let ple: &[Vec<f32>] = if rows.is_empty() { &[] } else { &rows[a..a + nt] };
if q4::encode_layer(&mut enc, dev, &model, &g, &w, f, pos0 + a, nt, inv_freq, &remap, shared_slot, ple, f > 0, false)
.is_none()
{
failed = Some("layer frame declined");
break 'layers;
}
dev.clear_pending();
dev.route_cap_copy(&mut enc, &g, ((f * frames.len() + k) * q4::TROWS * cs) as u64, nt);
dev.stream_swap(&mut enc, &g, k, nt, true, true);
if dts {
q4::dts_flush(&mut enc);
}
let cb = q4::finish_frame(enc);
t_enc += te.elapsed();
let ts = Instant::now();
let Some(idx) = q4::submit_indexed(uploads.take(), cb) else {
failed = Some("submit");
break 'layers;
};
dev.commit(f);
t_sub += ts.elapsed();
if dts {
if !q4::wait_submission(idx.clone()) {
failed = Some("wait");
break 'layers;
}
q4::dts_take();
}
if k + 1 == frames.len()
&& let Some(p) = prev.replace(idx)
{
if lag_ms > 0 {
std::thread::sleep(Duration::from_millis(lag_ms));
}
let tw = Instant::now();
if !q4::wait_submission(p) {
failed = Some("wait");
break 'layers;
}
let w = tw.elapsed();
waits += 1;
quick_waits += usize::from(w < Duration::from_micros(300));
t_wait += w;
}
}
state.gpu_pool.as_mut().unwrap().rearm_uploads();
}
if let Some(p) = prev.take() {
let tw = Instant::now();
if !q4::wait_submission(p) && failed.is_none() {
failed = Some("wait");
}
t_wait += tw.elapsed();
}
if let Some(a) = state.gpu_pool.as_mut() {
a.uploads_done();
a.flush_uploads();
}
if stream_dts {
q4::dts_stream_line(Some(&format!("qwen4-stream-dts pos={pos0} ntok={ntok}")));
}
if let Some(why) = failed {
tracing::warn!("qwen4 device path off: streamed prompt: {why}");
state.device_off = true;
return false;
}
let dev = state.dev.as_mut().unwrap();
dev.clear_pending();
dev.pos = pos0 + ntok;
let tr = Instant::now();
if let Some(bytes) = dev.route_cap_read(cap_bytes) {
let k_top = g.top_k;
for f in 0..n {
for (k, &(_, nt)) in frames.iter().enumerate() {
for t in 0..nt {
let base = ((f * frames.len() + k) * q4::TROWS + t) * cs;
let picks: Vec<usize> = (0..k_top)
.filter_map(|j| {
let o = base + 4 * (2 * k_top + 2 * j);
let e = u32::from_le_bytes(bytes.get(o..o + 4)?.try_into().ok()?);
(e != u32::MAX && (e as usize) < g.n_experts).then_some(e as usize)
})
.collect();
note_route(&mut state.route_hist, n, g.n_experts, f, &picks);
}
}
}
state.route_tokens += ntok;
phi_dump(&state.route_hist, state.route_tokens, &mut state.phi_last, &mut state.phi_last_tokens);
}
let t_route = tr.elapsed();
state.token_history.extend_from_slice(ids);
state.pos = pos0 + ntok;
if prof {
eprintln!(
"qwen4-stream routing read in {:.1}ms",
t_route.as_secs_f64() * 1e3
);
eprintln!(
"qwen4-stream pos={pos0} ntok={ntok} frames={} total={:.1}ms admit={:.1}ms ple={:.1}ms encode={:.1}ms wait={:.1}ms fills={} submit={:.1}ms quick_waits={quick_waits}/{waits}",
frames.len(),
t_all.elapsed().as_secs_f64() * 1e3,
t_admit.as_secs_f64() * 1e3,
t_ple.as_secs_f64() * 1e3,
t_enc.as_secs_f64() * 1e3,
t_wait.as_secs_f64() * 1e3,
FILL_N.load(std::sync::atomic::Ordering::Relaxed) - fill_n0,
t_sub.as_secs_f64() * 1e3,
);
}
true
}
#[cfg(feature = "gpu")]
fn note_route(hist: &mut Vec<u32>, n_layers: usize, ne: usize, f: usize, picks: &[usize]) {
if hist.len() != n_layers * ne {
*hist = vec![0; n_layers * ne];
}
for &e in picks {
if let Some(c) = hist.get_mut(f * ne + e) {
*c = c.saturating_add(1);
}
}
}
#[cfg(feature = "gpu")]
fn phi_dump(hist: &[u32], tokens: usize, last: &mut Vec<u32>, last_tokens: &mut usize) {
let Ok(path) = std::env::var("CMF_QWEN_PHI_DUMP") else { return };
let w = std::env::var("CMF_QWEN_PHI_WINDOW").ok().and_then(|v| v.parse::<usize>().ok()).unwrap_or(128).max(1);
if tokens < *last_tokens + w || hist.is_empty() {
return;
}
if last.len() != hist.len() {
*last = vec![0; hist.len()];
}
let mut row = Vec::with_capacity(hist.len() * 4);
for (h, l) in hist.iter().zip(last.iter()) {
row.extend_from_slice(&h.saturating_sub(*l).to_le_bytes());
}
use std::io::Write;
if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(&path) {
let _ = f.write_all(&row);
}
last.copy_from_slice(hist);
*last_tokens = tokens;
}
#[cfg(feature = "gpu")]
fn route_row(kind: &str, hist: &[u32], tokens: usize, snap: &mut Vec<u32>, snap_tokens: &mut usize) {
let Ok(prefix) = std::env::var("CMF_QWEN_ROUTE_ROWS") else { return };
if hist.is_empty() {
return;
}
if snap.len() != hist.len() {
*snap = vec![0; hist.len()];
}
let mut row = Vec::with_capacity(hist.len() * 4 + 4);
row.extend_from_slice(&(tokens.saturating_sub(*snap_tokens) as u32).to_le_bytes());
for (h, s) in hist.iter().zip(snap.iter()) {
row.extend_from_slice(&h.saturating_sub(*s).to_le_bytes());
}
use std::io::Write;
if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(format!("{prefix}.{kind}")) {
let _ = f.write_all(&row);
}
snap.copy_from_slice(hist);
*snap_tokens = tokens;
}
#[cfg(feature = "gpu")]
static ROUTE_TRACE: std::sync::Mutex<Option<std::io::BufWriter<std::fs::File>>> = std::sync::Mutex::new(None);
#[cfg(feature = "gpu")]
fn trace_on() -> bool {
static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ON.get_or_init(|| std::env::var_os("CMF_QWEN_ROUTE_TRACE").is_some())
}
#[cfg(feature = "gpu")]
fn trace_frame(pos0: usize, layer: usize, ntok: usize, top_k: usize, picks: &[u16]) {
use std::io::Write;
let mut g = ROUTE_TRACE.lock().unwrap();
if g.is_none() {
let Ok(path) = std::env::var("CMF_QWEN_ROUTE_TRACE") else { return };
match std::fs::OpenOptions::new().create(true).append(true).open(path) {
Ok(f) => *g = Some(std::io::BufWriter::with_capacity(1 << 20, f)),
Err(_) => return,
}
}
let w = g.as_mut().unwrap();
let mut rec = Vec::with_capacity(8 + ntok * top_k * 2);
rec.extend_from_slice(&(pos0 as u32).to_le_bytes());
rec.extend_from_slice(&(layer as u16).to_le_bytes());
rec.extend_from_slice(&(ntok as u16).to_le_bytes());
for &p in picks.iter().take(ntok * top_k) {
rec.extend_from_slice(&p.to_le_bytes());
}
let _ = w.write_all(&rec);
}
#[cfg(feature = "gpu")]
fn trace_mark(kind: u16) {
if trace_on() {
trace_frame(u32::MAX as usize, kind as usize, 0, 0, &[]);
}
}
#[cfg(feature = "gpu")]
fn trace_flush() {
use std::io::Write;
if let Some(w) = ROUTE_TRACE.lock().unwrap().as_mut() {
let _ = w.flush();
}
}
#[cfg(feature = "gpu")]
struct PackSet {
phi_layers: usize,
names: Vec<String>,
subs: Vec<(Vec<f32>, Vec<f32>, usize)>,
profiles: Vec<ExpertProfile>,
}
#[cfg(feature = "gpu")]
impl PackSet {
fn load(n_layers: usize, ne: usize) -> Option<Self> {
let dir = std::env::var("CMF_QWEN_PACKS").ok()?;
let path = std::path::Path::new(&dir).join("router.json");
let v: serde_json::Value = match std::fs::read(&path).ok().and_then(|b| serde_json::from_slice(&b).ok()) {
Some(v) => v,
None => {
let mut set = PackSet { phi_layers: 0, names: Vec::new(), subs: Vec::new(), profiles: Vec::new() };
let mut files: Vec<_> = std::fs::read_dir(&dir).ok()?.flatten().map(|e| e.path()).collect();
files.sort();
for p in files {
let Some(name) = p.file_name().and_then(|s| s.to_str()).and_then(|s| s.strip_suffix(".profile")) else {
continue;
};
if let Some(pr) = ExpertProfile::load(&p.to_string_lossy(), n_layers, ne) {
set.names.push(name.to_string());
set.profiles.push(pr);
}
}
if set.names.is_empty() {
tracing::warn!("qwen4 packs: {} unreadable and no profiles", path.display());
return None;
}
return Some(set);
}
};
let phi_layers = v["phi_layers"].as_u64()? as usize;
let mut set = PackSet { phi_layers, names: Vec::new(), subs: Vec::new(), profiles: Vec::new() };
for p in v["packs"].as_array()? {
let name = p["name"].as_str()?.to_string();
let d = &p["descriptor"];
let rank = d["rank"].as_u64()? as usize;
let mean = crate::router::decode_f16(d["mean"].as_str()?)?;
let basis = crate::router::decode_f16(d["basis"].as_str()?)?;
let dim = phi_layers * ne;
let prof = ExpertProfile::load(&std::path::Path::new(&dir).join(format!("{name}.profile")).to_string_lossy(), n_layers, ne);
match prof {
Some(pr) if mean.len() == dim && basis.len() == rank * dim => {
set.names.push(name);
set.subs.push((mean, basis, rank));
set.profiles.push(pr);
}
_ => tracing::warn!("qwen4 packs: pack {name} skipped (profile or topology does not fit this model)"),
}
}
(!set.names.is_empty()).then_some(set)
}
fn hint_weights(&self) -> Option<Vec<(usize, f32)>> {
let text = std::fs::read_to_string(std::env::var("CMF_QWEN_TASK_HINT").ok()?).ok()?;
let mut w: Vec<(usize, f32)> = text
.trim()
.split(',')
.filter_map(|kv| {
let (k, v) = kv.split_once('=')?;
let i = self.names.iter().position(|n| n == k.trim())?;
Some((i, v.trim().parse::<f32>().ok()?.max(0.0)))
})
.collect();
let s: f32 = w.iter().map(|x| x.1).sum();
if w.len() < 2 || s <= 0.0 {
return None;
}
w.iter_mut().for_each(|x| x.1 /= s);
Some(w)
}
fn decide(&self, hist: &[u32], ne: usize) -> Option<(usize, Vec<f32>)> {
if let Some(label) = std::env::var("CMF_QWEN_TASK_HINT").ok().and_then(|p| std::fs::read_to_string(p).ok())
&& let Some(i) = self.names.iter().position(|n| n == label.trim())
{
return Some((i, Vec::new()));
}
if let Some(w) = self.hint_weights()
&& let Some(&(i, _)) = w.iter().max_by(|a, b| a.1.total_cmp(&b.1))
{
return Some((i, Vec::new()));
}
if self.subs.is_empty() {
return None;
}
let phi = route_phi(hist, ne, self.phi_layers)?;
let errs: Vec<f32> = self.subs.iter().map(|(m, b, r)| crate::router::recon_error(&phi, m, b, *r)).collect();
let best = errs.iter().enumerate().min_by(|a, b| a.1.total_cmp(b.1))?.0;
Some((best, errs))
}
}
fn route_phi(hist: &[u32], ne: usize, layers: usize) -> Option<Vec<f32>> {
if hist.len() < layers * ne || layers == 0 {
return None;
}
let mut phi = Vec::with_capacity(layers * ne);
for l in 0..layers {
let row = &hist[l * ne..(l + 1) * ne];
let s = row.iter().map(|&c| c as f32).sum::<f32>().max(1.0);
phi.extend(row.iter().map(|&c| c as f32 / s));
}
let n = phi.iter().map(|v| v * v).sum::<f32>().sqrt();
if n <= 0.0 {
return None;
}
phi.iter_mut().for_each(|v| *v /= n);
Some(phi)
}
#[cfg(feature = "gpu")]
fn core_set(n_layers: usize, ne: usize) -> Option<Vec<Vec<usize>>> {
static CORE: std::sync::OnceLock<Option<Vec<Vec<usize>>>> = std::sync::OnceLock::new();
CORE.get_or_init(|| {
let path = std::env::var("CMF_QWEN_CORE").ok()?;
let Some(p) = ExpertProfile::load(&path, n_layers, ne) else {
tracing::warn!("qwen4 core: {path} does not fit this model");
return None;
};
let core: Vec<Vec<usize>> =
(0..n_layers).map(|l| (0..ne).filter(|&e| p.counts[l * ne + e] > 0).collect()).collect();
eprintln!("qwen4-core: {} experts pinned ({} a layer on average)", core.iter().map(Vec::len).sum::<usize>(), core.iter().map(Vec::len).sum::<usize>() / n_layers.max(1));
Some(core)
})
.clone()
}
#[cfg(feature = "gpu")]
fn prompt_done(layers: &[Layer], state: &mut State) {
route_row("prompt", &state.route_hist, state.route_tokens, &mut state.row_snap, &mut state.row_snap_tokens);
state.row_gen = true;
trace_mark(2);
if let Some(a) = state.gpu_pool.as_mut() {
a.sweep_clear();
}
if state.resident_all || state.device_off || state.gpu_pool.is_none() || state.route_tokens == 0 {
return;
}
let Some(model) = layers.first().and_then(|l| l.moe.experts.first()).and_then(|e| e.gate_proj.model_arc()) else {
return;
};
let n = layers.len();
let ne = state.gpu_pool.as_ref().unwrap().n_experts;
if state.route_hist.len() != n * ne {
return;
}
if state.packs.is_none() {
state.packs = Some(PackSet::load(n, ne));
}
let t0 = std::time::Instant::now();
let packs = state.packs.as_ref().unwrap().as_ref();
let choice = packs.and_then(|p| p.decide(&state.route_hist, ne));
if packs.is_none() && std::env::var("CMF_QWEN_SELF_PACK").as_deref() == Ok("0") {
return;
}
let mix: f32 = match choice {
None => 1.0,
Some(_) => std::env::var("CMF_QWEN_PACK_MIX").ok().and_then(|v| v.parse().ok()).unwrap_or(0.5f32).clamp(0.0, 1.0),
};
let weighted: Vec<(usize, f32)> = match packs.and_then(PackSet::hint_weights) {
Some(w) => w,
None => choice.as_ref().map(|(i, _)| vec![(*i, 1.0)]).unwrap_or_default(),
};
let mut per_layer: Vec<Vec<(f32, usize)>> = Vec::with_capacity(n);
for l in 0..n {
let h = &state.route_hist[l * ne..(l + 1) * ne];
let hs = h.iter().map(|&c| c as f32).sum::<f32>().max(1.0);
let mut pk = vec![0f32; ne];
for &(i, w) in &weighted {
let row = &packs.unwrap().profiles[i].counts[l * ne..(l + 1) * ne];
let ps = row.iter().map(|&c| c as f32).sum::<f32>().max(1.0);
for (p, &c) in pk.iter_mut().zip(row) {
*p += w * c as f32 / ps;
}
}
let mut v: Vec<(f32, usize)> = (0..ne)
.map(|e| {
let own = h[e] as f32 / hs;
(mix * own + (1.0 - mix) * pk[e], e)
})
.filter(|&(sc, _)| sc > 0.0)
.collect();
v.sort_by(|a, b| b.0.total_cmp(&a.0).then(a.1.cmp(&b.1)));
per_layer.push(v);
}
let core = core_set(n, ne);
let mut ranked: Vec<(usize, usize)> = Vec::new();
if let Some(c) = core.as_ref() {
for (l, row) in c.iter().enumerate() {
ranked.extend(row.iter().map(|&e| (l, e)));
}
}
let in_core = |l: usize, e: usize| core.as_ref().is_some_and(|c| c[l].contains(&e));
let mut cursor = vec![0usize; n];
for _ in 0..ne {
for (l, v) in per_layer.iter().enumerate() {
while let Some(&(_, e)) = v.get(cursor[l]) {
cursor[l] += 1;
if !in_core(l, e) {
ranked.push((l, e));
break;
}
}
}
}
let triples: Vec<Vec<(usize, usize, usize)>> = layers.iter().map(|l| l.expert_ids.clone()).collect();
let arena = state.gpu_pool.as_mut().unwrap();
if let Some(c) = core.as_ref()
&& !state.core_pinned
{
arena.set_pins(c);
state.core_pinned = true;
}
let (evicted, admitted) = arena.repack(&model, &ranked, &triples);
arena.flush_uploads();
let k_top = layers[0].moe.top_k;
let enough = (0..n).all(|l| arena.occupancy.get(l).copied().unwrap_or(0) > k_top);
if std::env::var("CMF_QWEN_MASK").as_deref() == Ok("1") && enough {
state.masked = true;
if let Some(d) = state.dev.as_mut() {
d.set_masked(true);
}
}
let name = choice.as_ref().map_or("prompt routing", |(i, _)| packs.unwrap().names[*i].as_str());
tracing::info!(
"qwen4 packs: {name} for this request ({} prompt tokens, prompt share {mix}), {evicted} experts out, {admitted} in, {:.0} ms",
state.route_tokens,
t0.elapsed().as_secs_f64() * 1e3
);
if std::env::var_os("CMF_QWEN_PROF").is_some() || std::env::var_os("CMF_QWEN_PACK_LOG").is_some() {
let errs = choice.as_ref().map(|(_, e)| {
packs.unwrap().names.iter().zip(e).map(|(nm, x)| format!("{nm}={x:.4}")).collect::<Vec<_>>().join(" ")
});
eprintln!(
"qwen4-pack {name} tokens={} mix={mix} out={evicted} in={admitted} {:.0}ms errors: {}",
state.route_tokens,
t0.elapsed().as_secs_f64() * 1e3,
errs.unwrap_or_default()
);
}
}
pub fn fit_packs(dir: &str, phi_layers: usize, rank: usize, blocks: &[usize]) -> Result<String, String> {
let mut names: Vec<String> = std::fs::read_dir(dir)
.map_err(|e| format!("{dir}: {e}"))?
.filter_map(|e| e.ok())
.filter_map(|e| e.file_name().to_str().and_then(|n| n.strip_suffix(".phi")).map(str::to_string))
.collect();
names.sort();
if names.is_empty() {
return Err(format!("{dir}: no <name>.phi files"));
}
let mut report = String::new();
let mut packs = Vec::new();
let mut holds: Vec<(usize, Vec<Vec<f32>>)> = Vec::new();
for (pi, name) in names.iter().enumerate() {
let prof = std::fs::read(format!("{dir}/{name}.profile")).map_err(|e| format!("{name}.profile: {e}"))?;
if prof.len() < 32 || &prof[..8] != b"CMFQ4PF\0" {
return Err(format!("{name}.profile is not a routing profile"));
}
let rd = |o: usize| u64::from_le_bytes(prof[o..o + 8].try_into().unwrap()) as usize;
let (nl, ne) = (rd(8), rd(16));
let raw = std::fs::read(format!("{dir}/{name}.phi")).map_err(|e| format!("{name}.phi: {e}"))?;
let w = nl * ne * 4;
if raw.len() < w || raw.len() % w != 0 {
return Err(format!("{name}.phi: not whole {nl}x{ne} rows"));
}
let rows: Vec<Vec<u32>> = raw
.chunks_exact(w)
.map(|r| r.chunks_exact(4).map(|b| u32::from_le_bytes([b[0], b[1], b[2], b[3]])).collect())
.collect();
let mut phis: Vec<Vec<f32>> = Vec::new();
for &b in blocks {
for grp in rows.chunks(b.max(1)) {
if grp.len() < b.max(1) {
continue;
}
let mut sum = vec![0u32; nl * ne];
for r in grp {
for (s, &v) in sum.iter_mut().zip(r) {
*s += v;
}
}
if let Some(p) = route_phi(&sum, ne, phi_layers) {
phis.push(p);
}
}
}
let d = crate::router::fit_descriptor(&phis, phi_layers, rank).ok_or(format!("{name}: too few windows ({})", phis.len()))?;
let hold = d.holdout.as_deref().and_then(crate::router::decode_f16).unwrap_or_default();
let dim = phi_layers * ne;
holds.push((pi, hold.chunks_exact(dim).map(|c| c.to_vec()).collect()));
report.push_str(&format!("{name}: {} windows ({} raw), rank {}\n", phis.len(), rows.len(), d.rank));
packs.push(serde_json::json!({ "name": name, "descriptor": d, "windows": phis.len() }));
}
let subs: Vec<(Vec<f32>, Vec<f32>, usize)> = packs
.iter()
.map(|p| {
let d = &p["descriptor"];
(
crate::router::decode_f16(d["mean"].as_str().unwrap_or("")).unwrap_or_default(),
crate::router::decode_f16(d["basis"].as_str().unwrap_or("")).unwrap_or_default(),
d["rank"].as_u64().unwrap_or(1) as usize,
)
})
.collect();
let (mut ok, mut tot) = (0usize, 0usize);
for (pi, hs) in &holds {
let mut right = 0;
let mut confused: std::collections::BTreeMap<&str, usize> = Default::default();
for h in hs {
let errs: Vec<f32> = subs.iter().map(|(m, b, r)| crate::router::recon_error(h, m, b, *r)).collect();
let best = errs.iter().enumerate().min_by(|a, b| a.1.total_cmp(b.1)).map(|x| x.0).unwrap_or(0);
if best == *pi {
right += 1;
} else {
*confused.entry(names[best].as_str()).or_default() += 1;
}
}
ok += right;
tot += hs.len();
report.push_str(&format!("held-out {}: {right}/{} {:?}\n", names[*pi], hs.len(), confused));
}
report.push_str(&format!("held-out total: {ok}/{tot}\n"));
let out = serde_json::json!({ "phi_layers": phi_layers, "rank": rank, "blocks": blocks, "packs": packs });
std::fs::write(format!("{dir}/router.json"), serde_json::to_vec_pretty(&out).map_err(|e| e.to_string())?)
.map_err(|e| format!("router.json: {e}"))?;
Ok(report)
}
#[cfg(feature = "gpu")]
fn device_hc_w(g: &GatedResidual) -> Option<crate::gpu_wgpu::qwen4::HcW<'_>> {
Some(crate::gpu_wgpu::qwen4::HcW {
norm: &g.norm,
down: g.down_idx?,
up: g.up_idx?,
inject: g.inject_idx,
})
}
#[cfg(feature = "gpu")]
const MTP_MEASURED_WINDOW: usize = 4;
#[cfg(feature = "gpu")]
fn mtp_k(raw: Option<&str>) -> (usize, bool) {
let max = crate::gpu_wgpu::qwen4::TMAX - 1;
match raw.map(|v| v.trim().parse::<usize>()) {
None => (3, false),
Some(Ok(v)) => (v.clamp(1, max), !(1..=max).contains(&v)),
Some(Err(_)) => (3, true),
}
}
struct WindowRows {
bytes: Vec<u8>,
lo: usize,
stride: usize,
vocab: usize,
rows: usize,
}
impl WindowRows {
fn row(&self, t: usize) -> &[u8] {
let o = self.lo + t * self.stride;
&self.bytes[o..o + self.vocab * 4]
}
fn floats(&self, t: usize) -> Vec<f32> {
self.row(t).chunks_exact(4).map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]])).collect()
}
}
fn argmax_le(row: &[u8]) -> usize {
const L: usize = 16;
let val = |b: &[u8]| f32::from_le_bytes([b[0], b[1], b[2], b[3]]);
let mut m = [f32::NEG_INFINITY; L];
let mut it = row.chunks_exact(4 * L);
for c in &mut it {
for (j, b) in c.chunks_exact(4).enumerate() {
let x = val(b);
m[j] = if x > m[j] { x } else { m[j] };
}
}
let mut best = f32::NEG_INFINITY;
for &x in &m {
best = if x > best { x } else { best };
}
for b in it.remainder().chunks_exact(4) {
let x = val(b);
best = if x > best { x } else { best };
}
if best == f32::NEG_INFINITY {
return 0;
}
for (ci, c) in row.chunks(4 * L).enumerate() {
let mut hit = false;
for b in c.chunks_exact(4) {
hit |= val(b) == best;
}
if hit {
return ci * L + c.chunks_exact(4).position(|b| val(b) == best).unwrap_or(0);
}
}
0
}
#[cfg(feature = "gpu")]
pub struct SpecRound {
pub accepted: Vec<u32>,
pub logits: Vec<f32>,
pub drafted: usize,
}
#[cfg(feature = "gpu")]
pub struct MtpHead {
side: Arc<CmfModel>,
layer: Layer,
enorm: Vec<f32>,
hnorm: Vec<f32>,
fc_e: usize,
fc_h: usize,
mixer: GatedResidual,
skeleton: Vec<usize>,
dev: Option<crate::gpu_wgpu::qwen4::Dev>,
arena: Option<QwenGpuPool>,
pub k: usize,
pub drafted: u64,
pub accepted: u64,
pub rounds: u64,
}
fn mtp_embedded(model: &CmfModel) -> bool {
model.tensor("model.mtp.layers.0.self_attn.q_proj.weight").is_some()
}
#[cfg(feature = "gpu")]
impl MtpHead {
pub fn load(main: &Arc<CmfModel>, arch: &ModelArch) -> Option<Self> {
if std::env::var("CMF_QWEN_MTP").as_deref() == Ok("0") {
return None;
}
let side = if mtp_embedded(main) {
match CmfModel::open(&main.path) {
Ok(m) => Arc::new(m),
Err(e) => {
tracing::warn!("qwen4 MTP head in {}: {e}", main.path.display());
return None;
}
}
} else {
let path = cortiq_core::mtp_sidecar_path(&main.path);
if path == main.path || !path.exists() {
return None;
}
match CmfModel::open(&path) {
Ok(m) => Arc::new(m),
Err(e) => {
tracing::warn!("qwen4 MTP sidecar {}: {e}", path.display());
return None;
}
}
};
let path = side.path.clone();
let sa = side.arch();
if sa.arch_name != arch.arch_name
|| sa.hidden_size != arch.hidden_size
|| sa.vocab_size != arch.vocab_size
{
tracing::warn!(
"qwen4 MTP sidecar {}: geometry differs from the backbone",
path.display()
);
return None;
}
let p = "model.mtp.layers.0.";
let build =
|| -> Result<(Layer, GatedResidual, Vec<f32>, Vec<f32>, usize, usize), CmfError> {
let mixer = Mixer::Qsa(load_qsa(&side, &format!("{p}self_attn."))?);
let moe = match build_ffn_at(&side, sa, p, false, &Overlay::None)? {
FfnKind::Moe(m) => m,
_ => return Err(err("MTP layer is not MoE")),
};
let expert_ids: Vec<_> = moe
.experts
.iter()
.map(|e| {
Some((
e.gate_proj.model_idx()?,
e.up_proj.model_idx()?,
e.down_proj.model_idx()?,
))
})
.collect::<Option<_>>()
.ok_or_else(|| err("MTP experts are not mmap-backed"))?;
let shared_ids = moe
.shared
.as_ref()
.and_then(|(e, _)| {
Some((
e.gate_proj.model_idx()?,
e.up_proj.model_idx()?,
e.down_proj.model_idx()?,
))
})
.ok_or_else(|| err("MTP shared expert is not mmap-backed"))?;
let attn_hc = load_hc(&side, &format!("{p}attn_hyper_connection."), true)?;
let mlp_hc = load_hc(&side, &format!("{p}mlp_hyper_connection."), true)?;
let idx = layer_idx(&side, p, false, false);
let layer = Layer {
attn_hc,
mlp_hc,
mixer,
moe,
ple: None,
expert_ids,
shared_ids: Some(shared_ids),
idx,
};
let mixer_hc = load_hc(&side, "model.mtp.hyper_connection_mixer.", false)?;
let enorm = f(&side, "model.mtp.enorm.weight")?;
let hnorm = f(&side, "model.mtp.hnorm.weight")?;
let fc_e = side
.tensor_index("model.mtp.fc_embedding.weight")
.ok_or_else(|| err("fc_embedding"))?;
let fc_h = side
.tensor_index("model.mtp.fc_hidden.weight")
.ok_or_else(|| err("fc_hidden"))?;
Ok((layer, mixer_hc, enorm, hnorm, fc_e, fc_h))
};
let (layer, mixer, enorm, hnorm, fc_e, fc_h) = match build() {
Ok(v) => v,
Err(e) => {
tracing::warn!("qwen4 MTP sidecar {}: {e}", path.display());
return None;
}
};
let ix = layer.idx.as_ref()?;
let mut skeleton: Vec<usize> = Vec::new();
skeleton.extend(layer.attn_hc.idxs());
skeleton.extend(layer.mlp_hc.idxs());
skeleton.extend(mixer.idxs());
skeleton.extend(ix.qsa.iter().flatten().copied());
skeleton.push(ix.router);
skeleton.extend(ix.shared_gate);
skeleton.extend([fc_e, fc_h]);
let raw = std::env::var("CMF_QWEN_MTP_K").ok();
let (k, clamped) = mtp_k(raw.as_deref());
if clamped {
tracing::warn!(
"CMF_QWEN_MTP_K={} is outside 1..={}: the verify window (k + 1 positions) is one device frame of at most {} tokens; using k = {k}",
raw.as_deref().unwrap_or(""),
crate::gpu_wgpu::qwen4::TMAX - 1,
crate::gpu_wgpu::qwen4::TMAX
);
}
tracing::info!("qwen4 MTP: draft head from {} (k = {k})", path.display());
Some(Self {
side,
layer,
enorm,
hnorm,
fc_e,
fc_h,
mixer,
skeleton,
dev: None,
arena: None,
k,
drafted: 0,
accepted: 0,
rounds: 0,
})
}
fn setup(&mut self, g: &crate::gpu_wgpu::qwen4::Geom, gu_q2: bool) -> Result<(), &'static str> {
use crate::gpu_wgpu::qwen4 as q4;
if self.dev.is_some() && self.arena.is_some() {
return Ok(());
}
let dev = q4::Dev::new(self.side.uid(), g, &[(false, false)]).ok_or("MTP device state")?;
q4::prewarm(&self.side, &self.skeleton).ok_or("MTP skeleton does not fit")?;
let n_exp = self.layer.moe.experts.len();
let mut arena =
QwenGpuPool::create_exact(&self.side, g.inter, g.hidden, n_exp, gu_q2, n_exp + 8)
.ok_or("MTP expert bank")?;
arena
.ensure(
&self.side,
0,
&[],
&self.layer.expert_ids,
self.layer.shared_ids,
)
.filter(|(_, s)| *s != u32::MAX)
.ok_or("MTP shared expert")?;
let all: Vec<(usize, usize)> = (0..n_exp).map(|e| (0, e)).collect();
let got = arena.prefill(
&self.side,
&all,
std::slice::from_ref(&self.layer.expert_ids),
);
if got < n_exp {
return Err("MTP experts do not all fit their bank");
}
self.dev = Some(dev);
self.arena = Some(arena);
Ok(())
}
}
#[cfg(feature = "gpu")]
#[allow(clippy::too_many_arguments)]
pub fn spec_round(
globals: &Globals,
layers: &[Layer],
cfg: &Cfg,
state: &mut State,
next_pos: usize,
all_ids: &[u32],
inv_freq: &[f32],
pool: Option<&Pool>,
) -> Option<SpecRound> {
use crate::gpu_wgpu::qwen4 as q4;
#[cfg(feature = "cuda")]
if crate::gpu_cuda::selected() {
return cuda_spec_round(globals, layers, cfg, state, next_pos, all_ids, pool);
}
if state.device_off || state.dev.is_none() || next_pos == 0 || all_ids.len() != next_pos + 1 {
return None;
}
if state.mtp.is_none() && !state.mtp_tried {
state.mtp_tried = true;
let model = layers.first()?.moe.experts.first()?.gate_proj.model_arc()?;
state.mtp = MtpHead::load(&model, model.arch());
}
let first = layers.first()?;
let e0 = first.moe.experts.first()?;
let gu_q2 = e0.gate_proj.model_dtype() == Some(TensorDtype::Q2TiledP);
let g = device_geom(cfg, &first.moe, gu_q2);
let prof = std::env::var_os("CMF_QWEN_PROF").is_some();
let t_round = std::time::Instant::now();
let t_gap = state.round_end.take().map(|t| t.elapsed()).unwrap_or_default();
if prof && (t_gap.is_zero() || t_gap.as_millis() > 100) {
eprintln!("qwen4-ctx pos={next_pos} ids={:?}", all_ids);
}
let sfx = suffix_cfg();
let frame_k = copy_frame_cfg()
.filter(|_| {
(state.resident_all || state.masked) && state.dev.as_ref().is_some_and(|d| d.o1.is_none())
})
.and_then(|(most, factor)| copy_frame_drafts(most, factor, state.copy_run, state.draft_budget));
let copy = sfx.filter(|s| s.strong > 0 && state.mtp.is_some()).and_then(|s| {
let (m, cont) = suffix_draft(all_ids, s.strong, frame_k.unwrap_or(s.max));
(m > 0).then_some((m, cont))
});
let copied = copy.is_some();
let framed = frame_k.is_some() && copy.as_ref().is_some_and(|(_, cont)| cont.len() + 1 > q4::TMAX);
let mut sfx_note = String::new();
let (k, drafts, conf) = if let Some((m, cont)) = copy {
sfx_note = format!(" sfx={}:{m}", if framed { "frame" } else { "copy" });
(cont.len(), cont, Vec::new())
} else {
let head = state.mtp.as_mut()?;
let g = device_geom(cfg, &head.layer.moe, gu_q2);
if let Err(why) = head.setup(&g, gu_q2) {
tracing::warn!("qwen4 MTP off: {why}");
state.mtp = None;
return None;
}
let k = head.k;
let mdev = head.dev.as_mut()?;
let arena = head.arena.as_mut()?;
let (remap, shared_slot) = arena.remap_snapshot(0);
let ix = head.layer.idx.as_ref()?;
let w = device_layer_w(&head.layer, ix)?;
let mixer_w = device_hc_w(&head.mixer)?;
let main_dev = state.dev.as_ref()?;
let main_model = e0.gate_proj.model_arc()?;
let lm_head = globals.lm_head_idx?;
if !mdev.ensure_stage(q4::chain_stage_bytes(
0,
0,
1,
q4::head_stride(&main_model, lm_head)?,
)) {
return None;
}
let mut drafts: Vec<u32> = Vec::with_capacity(k);
let mut conf: Vec<f32> = Vec::new();
let tok_in = all_ids[next_pos];
let device_chain = std::env::var("CMF_QWEN_MTP_DEVICE_DRAFT").as_deref() == Ok("1");
let draft_rows = {
let cut = draft_vocab_rows();
let guard = cut.max(131072) as u32;
let tail = &all_ids[all_ids.len().saturating_sub(16)..];
if cut > 0 && tail.iter().all(|&t| t < guard) { cut } else { 0 }
};
let mut chained = false;
let (minp, report) = draft_gate();
let conf_buf = if report { q4::draft_conf_buf() } else { None };
if device_chain
&& let Some(embed_idx) = globals.embed_idx
&& let Some(ids_buf) = q4::draft_ids(tok_in)
{
let mut run = || -> Option<(Vec<u32>, Vec<f32>)> {
let mut enc = q4::new_encoder("qwen4-mtp-draft")?;
let merge = q4::merge_guard(&enc);
for j in 0..k {
let cell = next_pos - 1 + j;
let _salt = q4::frame_salt(j);
let r = if j == 0 {
q4::whole(&main_dev.r_last)
} else {
mdev.hyper(0)
};
q4::encode_draft_gather(
&mut enc,
mdev,
&main_model,
&g,
embed_idx,
&ids_buf,
j,
)?;
if !q4::encode_mtp_input(
&mut enc,
mdev,
&head.side,
&g,
None,
&head.enorm,
&head.hnorm,
head.fc_e,
head.fc_h,
r,
) {
return None;
}
q4::encode_layer(
&mut enc,
mdev,
&head.side,
&g,
&w,
0,
cell,
1,
inv_freq,
&remap,
shared_slot,
&[],
false,
false,
)?;
if !q4::encode_pending(&mut enc, mdev, &g, 1)
|| !q4::finalize_pending(mdev, &g, 1, &[None], &[Vec::new()])
{
return None;
}
q4::pending_done(mdev);
let (lb, vocab, _) = q4::encode_head_with(
&mut enc,
mdev,
&head.side,
&main_model,
&g,
&mixer_w,
lm_head,
false,
1,
draft_rows,
)?;
q4::encode_argmax(&mut enc, mdev, &lb, vocab, &ids_buf, j + 1, draft_rows > 0)?;
if let Some(cb) = &conf_buf {
q4::encode_draft_conf(&mut enc, mdev, &lb, vocab, &ids_buf, cb, j + 1, draft_rows > 0)?;
}
}
if !q4::copy_to_stage(&mut enc, mdev, &ids_buf, 4, 0, (k * 4) as u64) {
return None;
}
let mut read = k * 4;
if let Some(cb) = &conf_buf {
if !q4::copy_to_stage(&mut enc, mdev, cb, 4, read as u64, (k * 4) as u64) {
return None;
}
read += k * 4;
}
drop(merge);
let bytes = q4::submit_chain(mdev, enc, read as u64)?.wait()?;
let word = |b: &[u8]| [b[0], b[1], b[2], b[3]];
Some((
bytes[..k * 4].chunks_exact(4).map(|b| u32::from_le_bytes(word(b))).collect(),
bytes[k * 4..read].chunks_exact(4).map(|b| f32::from_le_bytes(word(b))).collect(),
))
};
if let Some((d, c)) = run() {
drafts = d;
conf = c;
chained = true;
}
}
if !chained {
let mut tok_in = tok_in;
for j in 0..k {
let cell = next_pos - 1 + j;
let mut emb = vec![0.0f32; cfg.hidden];
if (tok_in as usize) < globals.embed.rows() {
globals.embed.row_f32(tok_in as usize, &mut emb);
}
let r = if j == 0 {
q4::whole(&main_dev.r_last)
} else {
mdev.hyper(0)
};
let mut enc = q4::new_encoder("qwen4-mtp-draft")?;
let merge = q4::merge_guard(&enc);
if !q4::encode_mtp_input(
&mut enc,
mdev,
&head.side,
&g,
Some(&emb),
&head.enorm,
&head.hnorm,
head.fc_e,
head.fc_h,
r,
) {
return None;
}
q4::encode_layer(
&mut enc,
mdev,
&head.side,
&g,
&w,
0,
cell,
1,
inv_freq,
&remap,
shared_slot,
&[],
false,
false,
)?;
if !q4::encode_pending(&mut enc, mdev, &g, 1)
|| !q4::finalize_pending(mdev, &g, 1, &[None], &[Vec::new()])
{
return None;
}
q4::pending_done(mdev);
let (lb, vocab, _) = q4::encode_head_with(
&mut enc,
mdev,
&head.side,
&main_model,
&g,
&mixer_w,
lm_head,
false,
1,
0,
)?;
if !q4::copy_to_stage(&mut enc, mdev, &lb, 0, 0, (vocab * 4) as u64) {
return None;
}
drop(merge);
let bytes = q4::submit_chain(mdev, enc, (vocab * 4) as u64)?.wait()?;
let mut best = (0usize, f32::NEG_INFINITY);
for (i, ch) in bytes[..vocab * 4].chunks_exact(4).enumerate() {
let v = f32::from_le_bytes([ch[0], ch[1], ch[2], ch[3]]);
if v > best.1 {
best = (i, v);
}
}
drafts.push(best.0 as u32);
tok_in = best.0 as u32;
}
}
let kept = if conf.len() == k { draft_gate_keep(&conf, minp) } else { k };
drafts.truncate(kept);
if let Some(s) = sfx.filter(|s| s.ext > 0 && kept == k && k < s.max) {
let mut ctx = all_ids.to_vec();
ctx.extend_from_slice(&drafts);
let (m, cont) = suffix_draft(&ctx, s.ext, s.max - k);
if m > 0 {
sfx_note = format!(" sfx=ext:{m}+{}", cont.len());
drafts.extend(cont);
}
}
(drafts.len(), drafts, conf)
};
let t_draft = t_round.elapsed();
let mut ids = Vec::with_capacity(k + 1);
ids.push(all_ids[next_pos]);
ids.extend_from_slice(&drafts);
let mut last_logits = Vec::new();
let (a, logits, t_accept) = if framed {
state.copy_frame = true;
state.frame_check = true;
state.frame_ids = None;
let ok = forward_tokens_device(
globals,
layers,
cfg,
state,
&ids,
next_pos,
inv_freq,
pool,
&mut last_logits,
true,
);
state.frame_check = false;
state.copy_frame = false;
let t_accept = std::time::Instant::now();
let rows = state.frame_ids.take().filter(|r| ok && r.len() == k + 1)?;
let mut a = 0usize;
while a < k && rows[a] == drafts[a] {
a += 1;
}
let t_accept = t_accept.elapsed();
if a < k {
let dev = state.dev.as_mut()?;
if !dev.restore_checkpoint(&g) {
return None;
}
dev.pos = next_pos;
let keep = state.token_history.len() - (k + 1);
state.token_history.truncate(keep);
state.pos = next_pos;
state.copy_frame = a + 1 > q4::TMAX;
let ok = forward_tokens_device(
globals,
layers,
cfg,
state,
&ids[..a + 1],
next_pos,
inv_freq,
pool,
&mut last_logits,
true,
);
state.copy_frame = false;
if !ok {
return None;
}
sfx_note.push_str(&format!(" redo={}", a + 1));
}
(a, last_logits, t_accept)
} else {
state.verify_window = true;
let ok = forward_tokens_device(
globals,
layers,
cfg,
state,
&ids,
next_pos,
inv_freq,
pool,
&mut last_logits,
true,
);
state.verify_window = false;
let t_accept = std::time::Instant::now();
let Some(win) = state.window_rows.take().filter(|w| ok && w.rows == k + 1) else {
return None;
};
let mut a = 0usize;
while a < k && argmax_le(win.row(a)) == drafts[a] as usize {
a += 1;
}
let logits = win.floats(a);
drop(win);
let t_accept = t_accept.elapsed();
if let Some(uid) = state.dev.as_ref().filter(|d| d.o1_sealed()).map(|d| d.uid()) {
if !crate::gpu_wgpu::o1_spec_commit(uid, next_pos, a + 1) {
return None;
}
}
if a < k {
let dev = state.dev.as_mut()?;
if !dev.restore(&g, a + 1) {
return None;
}
let keep = state.token_history.len() - (k - a);
state.token_history.truncate(keep);
state.pos = next_pos + a + 1;
dev.pos = state.pos;
if let Some(mut enc) = q4::new_encoder("qwen4-keep-r") {
dev.keep_r(&mut enc, a);
q4::submit_only(enc);
}
}
(a, logits, t_accept)
};
state.copy_run = if copied && a == k { state.copy_run + k + 1 } else { 0 };
if let Some(d) = state.dev.as_mut() {
if d.o1_seal_at().is_some_and(|at| state.pos >= at) {
d.seal(&g, state.pos);
}
}
if let Some(head) = state.mtp.as_mut() {
head.rounds += 1;
head.drafted += k as u64;
head.accepted += a as u64;
if prof {
eprintln!(
"qwen4-mtp round={} drafted={k} accepted={a} draft={:.1}ms total={:.1}ms accept={:.2}ms gap={:.2}ms (acceptance {:.1}%) ids={:?}{}{} tok={}",
head.rounds,
t_draft.as_secs_f64() * 1e3,
t_round.elapsed().as_secs_f64() * 1e3,
t_accept.as_secs_f64() * 1e3,
t_gap.as_secs_f64() * 1e3,
head.accepted as f64 * 100.0 / head.drafted.max(1) as f64,
drafts,
if conf.is_empty() {
String::new()
} else {
format!(" conf={:?}", conf.iter().map(|p| (p * 1000.0).round() / 1000.0).collect::<Vec<_>>())
},
sfx_note,
all_ids[next_pos]
);
}
}
state.round_end = Some(std::time::Instant::now());
Some(SpecRound {
accepted: drafts[..a].to_vec(),
logits,
drafted: k,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(feature = "gpu")]
#[test]
fn pending_admission_cold_picks_keep_route_order() {
let slots = [3, u32::MAX, 7, u32::MAX, 11, u32::MAX, 13, u32::MAX];
assert_eq!(
cold_picks_from_slots(&slots, 4, 4, &[3, 1, 0, 1]),
Some(vec![3, 1, 1])
);
assert_eq!(cold_picks_from_slots(&slots, 4, 4, &[4]), None);
assert_eq!(cold_picks_from_slots(&slots, 5, 4, &[0]), None);
}
#[cfg(feature = "gpu")]
#[test]
fn draft_gate_keeps_the_confident_prefix() {
assert_eq!(draft_gate_keep(&[0.1, 0.1, 0.1], 0.0), 3);
assert_eq!(draft_gate_keep(&[0.9, 0.8, 0.5], 0.5), 2);
assert_eq!(draft_gate_keep(&[0.9, 0.8, 0.5], 0.3), 3);
assert_eq!(draft_gate_keep(&[0.2, 0.99, 0.99], 0.5), 1);
assert_eq!(draft_gate_keep(&[0.9, 0.3, 0.99], 0.5), 1);
assert_eq!(draft_gate_keep(&[f32::NAN, 0.9], 0.5), 1);
assert_eq!(draft_gate_keep(&[], 0.5), 0);
}
#[cfg(feature = "gpu")]
#[test]
fn copy_frame_grows_with_the_copy_and_stops_at_the_request() {
assert_eq!(copy_frame_parse("63"), Some((63, 2.0)));
assert_eq!(copy_frame_parse("31:1"), Some((31, 1.0)));
assert_eq!(copy_frame_parse("999"), Some((255, 2.0)));
assert_eq!(copy_frame_parse("15"), None);
assert_eq!(copy_frame_parse("63:0"), None);
assert_eq!(copy_frame_parse("x"), None);
assert_eq!(copy_frame_drafts(255, 2.0, 0, usize::MAX), None);
assert_eq!(copy_frame_drafts(255, 2.0, 8, usize::MAX), None);
assert_eq!(copy_frame_drafts(255, 2.0, 16, usize::MAX), Some(31));
assert_eq!(copy_frame_drafts(255, 2.0, 48, usize::MAX), Some(63));
assert_eq!(copy_frame_drafts(255, 2.0, 112, usize::MAX), Some(127));
assert_eq!(copy_frame_drafts(255, 2.0, 240, usize::MAX), Some(255));
assert_eq!(copy_frame_drafts(255, 4.0, 8, usize::MAX), Some(31));
assert_eq!(copy_frame_drafts(63, 1.0, 16, usize::MAX), None);
assert_eq!(copy_frame_drafts(63, 1.0, 32, usize::MAX), Some(31));
assert_eq!(copy_frame_drafts(31, 2.0, 500, usize::MAX), Some(31));
assert_eq!(copy_frame_drafts(63, 2.0, 500, 40), Some(31));
assert_eq!(copy_frame_drafts(63, 2.0, 500, 30), None);
}
#[cfg(feature = "gpu")]
#[cfg(feature = "cuda")]
#[test]
fn copy_cursor_copies_and_learns_insertions() {
let mut ids: Vec<u32> = vec![5, 6, 7, 8, 9, 1, 5];
let mut cur = CopyCur::default();
cur.account(&ids);
cur.start(1, 1);
assert_eq!(cur.draft(&ids, 4), vec![6, 7, 8, 9]);
ids.extend([6, 7]);
cur.account(&ids);
assert_eq!(cur.at, Some(3));
assert_eq!(cur.draft(&ids, 3), vec![8, 9, 1]);
let p: Vec<u32> = vec![1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4];
let mut ids = p.clone();
ids.push(1);
let mut cur = CopyCur::default();
cur.account(&ids);
cur.start(1, 1);
ids.extend([2, 3, 100, 101, 4, 1, 2, 3]);
cur.account(&ids);
assert_eq!(cur.rules.get(&(3, 4)), Some(&(vec![100, 101], 1)));
assert_eq!(cur.draft(&ids, 3), vec![4, 1, 2]);
ids.push(100);
cur.account(&ids);
assert_eq!(cur.draft(&ids, 3), vec![101, 4, 1]);
ids.extend([101, 4, 1, 2, 3]);
cur.account(&ids);
assert_eq!(cur.rules.get(&(3, 4)).map(|r| r.1), Some(2));
assert_eq!(&cur.draft(&ids, 4)[..3], &[100, 101, 4]);
}
#[test]
fn suffix_draft_continues_the_latest_longest_occurrence() {
assert_eq!(suffix_draft(&[1, 2, 3, 9, 8, 7, 1, 2, 3], 3, 2), (3, vec![9, 8]));
assert_eq!(suffix_draft(&[1, 2, 3, 9, 8, 7, 1, 2, 3], 4, 2), (0, vec![]));
assert_eq!(suffix_draft(&[], 1, 2), (0, vec![]));
assert_eq!(suffix_draft(&[1, 2, 1, 2], 2, 0), (0, vec![]));
assert_eq!(suffix_draft(&[5, 1, 2, 3, 4, 0, 2, 3, 6, 0, 1, 2, 3], 2, 1), (3, vec![4]));
assert_eq!(suffix_draft(&[2, 3, 4, 0, 2, 3, 6, 0, 2, 3], 2, 1), (3, vec![6]));
assert_eq!(suffix_draft(&[7, 1, 2, 1, 2, 1, 2], 2, 5), (4, vec![1, 2, 1, 2, 1]));
assert_eq!(suffix_draft(&[4, 4, 4], 1, 3), (2, vec![4, 4, 4]));
}
#[cfg(feature = "gpu")]
#[test]
fn suffix_cfg_parses_and_clamps_to_the_frame() {
let max = crate::gpu_wgpu::qwen4::TMAX - 1;
assert_eq!(suffix_parse("6:12:5"), Some(SuffixCfg { ext: 6, strong: 12, max: 5 }));
assert_eq!(suffix_parse("6"), Some(SuffixCfg { ext: 6, strong: 0, max }));
assert_eq!(suffix_parse("0:8:99"), Some(SuffixCfg { ext: 0, strong: 8, max }));
assert_eq!(suffix_parse("0:0"), None);
assert_eq!(suffix_parse("x"), None);
assert_eq!(suffix_parse(""), None);
}
#[cfg(feature = "gpu")]
#[test]
fn mtp_k_parses_and_clamps_to_the_frame() {
assert_eq!(mtp_k(None), (3, false));
for k in 1..=7 {
assert_eq!(mtp_k(Some(&k.to_string())), (k, false));
}
assert_eq!(mtp_k(Some("0")), (1, true));
assert_eq!(mtp_k(Some("8")), (7, true));
assert_eq!(mtp_k(Some("four")), (3, true));
assert!(mtp_k(Some("99")).0 < crate::gpu_wgpu::qwen4::TMAX);
}
#[test]
fn dynamic_pool_percent_is_bounded_per_policy() {
assert_eq!(bounded_pool_pct(None, 40, 20, 60), 40);
assert_eq!(bounded_pool_pct(Some("55"), 40, 20, 40), 40);
assert_eq!(bounded_pool_pct(Some("999"), 40, 20, 40), 40);
assert_eq!(bounded_pool_pct(Some("0"), 40, 20, 40), 20);
assert_eq!(bounded_pool_pct(Some("bad"), 40, 20, 40), 40);
assert_eq!(bounded_pool_pct(Some("90"), 75, 25, 85), 85);
assert_eq!(bounded_pool_slots(Some("9999"), 752, true), 752);
assert_eq!(bounded_pool_slots(Some("9999"), 752, false), 9999);
}
#[test]
fn active_tail_floor_uses_only_reachable_layers() {
assert_eq!(active_layer_floor(512, 40), 12);
assert_eq!(active_layer_floor(512, 13), 39);
assert!(active_layer_floor(512, 13) * 13 <= 512);
assert_eq!(active_layer_floor(16, 13), 1);
assert_eq!(active_layer_floor(12, 13), 0);
}
#[test]
fn deterministic_hash_tables_match_contract() {
let q = Qwen4ExpConfig {
hc_count: 4,
hc_lowrank: 320,
indexer_n_heads: 4,
indexer_kv_heads: 1,
indexer_head_dim: 128,
indexer_budget: 2048,
indexer_compress_ratio: 4,
ple_layer_ids: vec![2],
ple_embed_dim: 2560,
ple_conv_kernel_size: 4,
ngram_size: 3,
heads_per_ngram: 8,
ngram_vocab_size_base: 20_000_000,
make_ngram_vocab_size_divisible_by: 128,
split_ngram_parts: 128,
seed: 1234,
};
let (m, sizes, offsets) = ple_tables(&q, 248_320, 0);
assert_eq!(
m,
[23_703_573_157_769, 20_109_073_645_365, 8_052_911_324_071]
);
assert!(m.iter().all(|x| x & 1 == 1));
assert_eq!(sizes.len(), 16);
assert!(sizes.iter().all(|&x| is_prime(x as usize)));
assert_eq!(offsets[0], 0);
assert_eq!(offsets[1], sizes[0]);
}
#[test]
fn eos_breaks_ngram_context_only_after_it() {
let eos = 99;
assert_eq!(shifted_token(&[1, 2], 3, 1, eos), 2);
assert_eq!(shifted_token(&[1, eos], 3, 1, eos), eos);
assert_eq!(shifted_token(&[1, 2], eos, 1, eos), 2);
assert_eq!(shifted_token(&[], 3, 2, eos), eos);
}
}