use super::{ctx, moe_q8, wait_fast_checked, Ctx};
use cortiq_core::CmfModel;
use metal::{Buffer, CommandBuffer, ComputeCommandEncoderRef, ComputePipelineState, MTLResourceOptions, MTLSize};
use std::collections::HashMap;
use std::ffi::c_void;
use std::sync::{Arc, Mutex, OnceLock};
const CMSL: &str = include_str!("clover_msl.metal");
struct Pipes {
conv: ComputePipelineState,
mix: ComputePipelineState,
attend: ComputePipelineState,
q8: [ComputePipelineState; 2],
}
static PIPES: OnceLock<Result<Pipes, String>> = OnceLock::new();
fn pipes(c: &Ctx) -> Result<&'static Pipes, String> {
PIPES
.get_or_init(|| {
let opts = metal::CompileOptions::new();
opts.set_language_version(metal::MTLLanguageVersion::V3_0);
let lib = c
._device
.new_library_with_source(CMSL, &opts)
.map_err(|e| format!("clover MSL compile: {e}"))?;
let pso = |name: &str| -> Result<ComputePipelineState, String> {
let f = lib.get_function(name, None).map_err(|e| format!("kernel {name}: {e}"))?;
c._device
.new_compute_pipeline_state_with_function(&f)
.map_err(|e| format!("pipeline {name}: {e}"))
};
let q8 = |mode: u32| -> Result<ComputePipelineState, String> {
let src = format!("#define Q8MM_MODE {mode}\n{}", moe_q8::QMSL);
let lib = c
._device
.new_library_with_source(&src, &opts)
.map_err(|e| format!("q8 blob GEMM (mode {mode}) compile: {e}"))?;
let f = lib.get_function("q8f_mul_mm_blob", None).map_err(|e| format!("q8f_mul_mm_blob: {e}"))?;
c._device
.new_compute_pipeline_state_with_function(&f)
.map_err(|e| format!("pipeline q8f_mul_mm_blob: {e}"))
};
Ok(Pipes {
conv: pso("bdp_conv")?,
mix: pso("bdp_mix")?,
attend: pso("bdp_attend_blk")?,
q8: [q8(2)?, q8(1)?],
})
})
.as_ref()
.map_err(|e| e.clone())
}
pub fn ready() -> Result<(), String> {
pipes(ctx().ok_or("clover metal: no device")?).map(|_| ())
}
#[derive(Clone, Copy)]
pub struct GdnGeom {
pub nv: usize,
pub nk: usize,
pub dk: usize,
pub dv: usize,
pub kk: usize,
pub eps: f32,
pub gate_sigmoid: bool,
}
impl GdnGeom {
fn c_dim(&self) -> usize {
2 * self.nk * self.dk + self.nv * self.dv
}
fn ring_len(&self) -> usize {
(self.kk - 1) * self.c_dim()
}
fn state_len(&self) -> usize {
self.ring_len() + self.nv * self.dk * self.dv
}
}
pub struct GdnLayer<'a> {
pub qkv: usize,
pub z: usize,
pub out: usize,
pub a: &'a [f32],
pub bproj: &'a [f32],
pub conv1d: &'a [f32],
pub a_log: &'a [f32],
pub dt_bias: &'a [f32],
pub norm: &'a [f32],
pub conv_right: &'a [f32],
pub logit_scale: &'a [f32],
pub history_bias: &'a [f32],
pub history_c: &'a [f32],
pub state: &'a [f32],
}
pub struct AttnLayer<'a> {
pub wq: usize,
pub wk: usize,
pub wv: usize,
pub wo: usize,
pub q_norm: Option<&'a [f32]>,
pub k_norm: Option<&'a [f32]>,
pub output_gate: bool,
pub nh: usize,
pub nkv: usize,
pub hd: usize,
pub rd: usize,
pub scale: f32,
pub rope_scale: f32,
pub inv_freq: &'a [f32],
pub k: Vec<&'a [f32]>,
pub v: Vec<&'a [f32]>,
}
pub enum Mixer<'a> {
Gdn(GdnLayer<'a>),
Attn(AttnLayer<'a>),
}
pub struct MoeLayer<'a> {
pub router: &'a [f32],
pub sgate: &'a [f32],
pub experts: Vec<(usize, usize, usize)>,
pub shared: (usize, usize, usize),
}
pub struct Layer<'a> {
pub input_norm: &'a [f32],
pub post_norm: &'a [f32],
pub mixer: Mixer<'a>,
pub moe: MoeLayer<'a>,
}
pub struct Pass<'a> {
pub model: &'a Arc<CmfModel>,
pub layers: Vec<Layer<'a>>,
pub final_norm: &'a [f32],
pub lm_head: usize,
pub lm_rows: usize,
pub hidden: usize,
pub inter: usize,
pub eps: f32,
pub gemma: bool,
pub gdn: GdnGeom,
pub epoch: (u64, usize),
}
pub type RouteFn<'r> = dyn FnMut(usize, &[f32], usize) -> Option<Vec<Vec<(usize, f32)>>> + 'r;
#[derive(Clone, Copy, Debug, Default)]
pub struct PreviewStats {
pub gpu_s: f64,
pub exec_s: f64,
pub route_s: f64,
pub fill_s: f64,
pub hits: u64,
pub misses: u64,
pub stage_s: [f64; 9],
}
struct GdnConst {
a: Buffer,
bproj: Buffer,
conv1d: Buffer,
a_log: Buffer,
dt_bias: Buffer,
norm: Buffer,
right: Buffer,
lsc: Buffer,
hb: Buffer,
hc: Buffer,
state: Buffer,
}
struct AttnConst {
nkv: usize,
hd: usize,
qn: Buffer,
kn: Buffer,
invf: Buffer,
k: Buffer,
v: Buffer,
cap: usize,
uploaded: usize,
}
struct LayerConst {
in_norm: Buffer,
post_norm: Buffer,
router: Buffer,
gdn: Option<GdnConst>,
attn: Option<AttnConst>,
}
type View = (Buffer, u64);
fn tensor_view(c: &Ctx, model: &CmfModel, idx: usize) -> Result<(View, usize), String> {
let bytes = model.primary_bytes();
let page = super::page_size();
let e = &model.tensors[idx];
let abs = model
.entry_abs_offset(e)
.ok_or_else(|| format!("clover metal: tensor {idx} is outside the primary mapping"))?;
let len = model.entry_bytes(e).len();
let (base, end) = (abs / page * page, (abs + len).div_ceil(page) * page);
if !(bytes.as_ptr() as usize).is_multiple_of(page) || end > bytes.len().div_ceil(page) * page {
return Err(format!("clover metal: tensor {idx} cannot be viewed"));
}
let buf = unsafe {
c._device.new_buffer_with_bytes_no_copy(
bytes.as_ptr().add(base) as *const c_void,
(end - base) as u64,
MTLResourceOptions::StorageModeShared,
None,
)
};
Ok(((buf, (abs - base) as u64), end - base))
}
struct ExpertView {
w: [View; 3],
bytes: usize,
used: u64,
}
struct ExpertViews {
map: HashMap<(usize, usize), ExpertView>,
bytes: usize,
cap: usize,
tick: u64,
}
impl ExpertViews {
fn get(
&mut self,
c: &Ctx,
model: &CmfModel,
layer: usize,
e: usize,
trio: (usize, usize, usize),
st: &mut PreviewStats,
) -> Result<[View; 3], String> {
if let Some(v) = self.map.get_mut(&(layer, e)) {
v.used = self.tick;
st.hits += 1;
return Ok(v.w.clone());
}
let t0 = std::time::Instant::now();
let (g, gb) = tensor_view(c, model, trio.0)?;
let (u, ub) = tensor_view(c, model, trio.1)?;
let (d, db) = tensor_view(c, model, trio.2)?;
let bytes = gb + ub + db;
while self.bytes + bytes > self.cap {
let Some(victim) = self
.map
.iter()
.filter(|(_, v)| v.used < self.tick)
.min_by_key(|(_, v)| v.used)
.map(|(k, _)| *k)
else {
break; };
let old = self.map.remove(&victim).unwrap();
self.bytes -= old.bytes;
}
let w = [g, u, d];
self.map.insert((layer, e), ExpertView { w: w.clone(), bytes, used: self.tick });
self.bytes += bytes;
st.fill_s += t0.elapsed().as_secs_f64();
st.misses += 1;
Ok(w)
}
}
struct Exec {
model_uid: u64,
views: HashMap<usize, View>,
consts: Vec<LayerConst>,
final_norm: Buffer,
ones: Buffer,
experts: ExpertViews,
epoch: Option<(u64, usize)>,
last_commit: Option<(usize, usize, usize)>,
scratch: HashMap<&'static str, Buffer>,
}
static EXEC: Mutex<Option<Exec>> = Mutex::new(None);
fn shared_buf(c: &Ctx, bytes: usize) -> Buffer {
c._device.new_buffer(bytes.max(16) as u64, MTLResourceOptions::StorageModeShared)
}
fn f32_buf(c: &Ctx, data: &[f32]) -> Buffer {
let b = shared_buf(c, data.len() * 4);
write_f32(&b, 0, data);
b
}
fn write_f32(b: &Buffer, at: usize, data: &[f32]) {
unsafe { std::ptr::copy_nonoverlapping(data.as_ptr(), (b.contents() as *mut f32).add(at), data.len()) }
}
fn set_u32(enc: &ComputeCommandEncoderRef, slot: u64, v: u32) {
enc.set_bytes(slot, 4, &v as *const u32 as *const c_void);
}
fn set_f32(enc: &ComputeCommandEncoderRef, slot: u64, v: f32) {
enc.set_bytes(slot, 4, &v as *const f32 as *const c_void);
}
fn bind(enc: &ComputeCommandEncoderRef, bufs: &[(&Buffer, u64)]) {
for (i, (b, off)) in bufs.iter().enumerate() {
enc.set_buffer(i as u64, Some(b), *off);
}
}
impl Exec {
fn new(c: &Ctx, d: &Pass) -> Result<Exec, String> {
let model = d.model.as_ref();
let mut idxs = vec![d.lm_head];
for l in &d.layers {
match &l.mixer {
Mixer::Gdn(g) => idxs.extend([g.qkv, g.z, g.out]),
Mixer::Attn(a) => idxs.extend([a.wq, a.wk, a.wv, a.wo]),
}
idxs.extend([l.moe.shared.0, l.moe.shared.1, l.moe.shared.2]);
}
let t0 = std::time::Instant::now();
let mut views = HashMap::new();
let mut view_bytes = 0usize;
for &i in &idxs {
if let std::collections::hash_map::Entry::Vacant(v) = views.entry(i) {
let (w, n) = tensor_view(c, model, i)?;
v.insert(w);
view_bytes += n;
}
}
let ne = d.layers[0].moe.experts.len();
let hs = d.hidden;
let mut consts = Vec::with_capacity(d.layers.len());
for l in &d.layers {
let mut router = Vec::with_capacity((ne + 1) * hs);
router.extend_from_slice(l.moe.router);
router.extend_from_slice(l.moe.sgate);
let gdn = match &l.mixer {
Mixer::Gdn(g) => Some(GdnConst {
a: f32_buf(c, g.a),
bproj: f32_buf(c, g.bproj),
conv1d: f32_buf(c, g.conv1d),
a_log: f32_buf(c, g.a_log),
dt_bias: f32_buf(c, g.dt_bias),
norm: f32_buf(c, g.norm),
right: f32_buf(c, g.conv_right),
lsc: f32_buf(c, g.logit_scale),
hb: f32_buf(c, g.history_bias),
hc: f32_buf(c, g.history_c),
state: shared_buf(c, d.gdn.state_len() * 4),
}),
Mixer::Attn(_) => None,
};
let attn = match &l.mixer {
Mixer::Attn(a) => Some(AttnConst {
nkv: a.nkv,
hd: a.hd,
qn: f32_buf(c, a.q_norm.unwrap_or(&[0.0])),
kn: f32_buf(c, a.k_norm.unwrap_or(&[0.0])),
invf: f32_buf(c, a.inv_freq),
k: shared_buf(c, 16),
v: shared_buf(c, 16),
cap: 0,
uploaded: 0,
}),
Mixer::Gdn(_) => None,
};
consts.push(LayerConst {
in_norm: f32_buf(c, l.input_norm),
post_norm: f32_buf(c, l.post_norm),
router: f32_buf(c, &router),
gdn,
attn,
});
}
let cap_mb = std::env::var("CMF_CLOVER_METAL_CACHE_MB")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(6144);
tracing::info!(
"clover metal: {} MB of attention / DeltaNet / shared / head weights in {} views ({:.2} s), expert views up to {} MB",
view_bytes >> 20,
views.len(),
t0.elapsed().as_secs_f64(),
cap_mb
);
Ok(Exec {
model_uid: model.uid(),
views,
consts,
final_norm: f32_buf(c, d.final_norm),
ones: f32_buf(c, &[1.0; 8]),
experts: ExpertViews { map: HashMap::new(), bytes: 0, cap: cap_mb << 20, tick: 1 },
epoch: None,
last_commit: None,
scratch: HashMap::new(),
})
}
fn scratch(&mut self, c: &Ctx, name: &'static str, bytes: usize) -> Buffer {
let e = self.scratch.entry(name).or_insert_with(|| shared_buf(c, bytes));
if (e.length() as usize) < bytes {
*e = shared_buf(c, bytes);
}
e.clone()
}
fn view(&self, idx: usize) -> Result<View, String> {
self.views.get(&idx).cloned().ok_or_else(|| format!("clover metal: tensor {idx} has no view"))
}
fn sync(&mut self, c: &Ctx, d: &Pass, b: usize) -> Result<(), String> {
let new_request = self.epoch.is_none_or(|(r, _)| r != d.epoch.0);
let changed = self.epoch != Some(d.epoch);
for (l, k) in d.layers.iter().zip(self.consts.iter_mut()) {
match (&l.mixer, k.gdn.as_ref(), k.attn.as_mut()) {
(Mixer::Gdn(g), Some(gc), _) if changed => {
if g.state.len() == d.gdn.state_len() {
write_f32(&gc.state, 0, g.state);
} else {
unsafe { std::ptr::write_bytes(gc.state.contents() as *mut u8, 0, d.gdn.state_len() * 4) };
}
}
(Mixer::Attn(a), _, Some(ac)) => {
let stored = a.k.first().map_or(0, |k| k.len() / a.hd);
if a.k.len() != a.nkv || a.v.len() != a.nkv || a.k.iter().chain(&a.v).any(|x| x.len() != stored * a.hd) {
return Err("clover metal: K/V heads of unequal length".into());
}
if new_request || stored < ac.uploaded {
ac.uploaded = 0;
}
if stored + b > ac.cap {
let cap = (stored + b).next_power_of_two().max(1024);
ac.k = shared_buf(c, a.nkv * cap * a.hd * 4);
ac.v = shared_buf(c, a.nkv * cap * a.hd * 4);
ac.cap = cap;
ac.uploaded = 0;
}
for h in 0..a.nkv {
let (lo, hi) = (ac.uploaded * a.hd, stored * a.hd);
write_f32(&ac.k, h * ac.cap * a.hd + lo, &a.k[h][lo..hi]);
write_f32(&ac.v, h * ac.cap * a.hd + lo, &a.v[h][lo..hi]);
}
ac.uploaded = stored;
}
_ => {}
}
}
self.epoch = Some(d.epoch);
Ok(())
}
}
fn skip(what: char) -> bool {
static S: OnceLock<String> = OnceLock::new();
S.get_or_init(|| std::env::var("CMF_CLOVER_SKIP").unwrap_or_default()).contains(what)
}
fn q8_mode() -> usize {
static M: OnceLock<usize> = OnceLock::new();
*M.get_or_init(|| match std::env::var("CMF_CLOVER_Q8").as_deref() {
Ok("f32") => 0,
_ => 1,
})
}
#[allow(clippy::too_many_arguments)]
fn q8mm(p: &Pipes, enc: &ComputeCommandEncoderRef, w: View, x: &Buffer, y: &Buffer, b: usize, rows: usize, cols: usize) {
enc.set_compute_pipeline_state(&p.q8[q8_mode()]);
enc.set_buffer(0, Some(&w.0), w.1);
enc.set_buffer(1, Some(x), 0);
enc.set_buffer(2, Some(y), 0);
set_u32(enc, 3, cols as u32);
set_u32(enc, 4, rows as u32);
set_u32(enc, 5, b as u32);
set_u32(enc, 6, 1);
enc.dispatch_thread_groups(
MTLSize::new((b as u64).div_ceil(32), (rows as u64).div_ceil(64), 1),
MTLSize::new(128, 1, 1),
);
}
#[allow(clippy::too_many_arguments)]
fn q4mm(
c: &Ctx,
enc: &ComputeCommandEncoderRef,
w: &Buffer,
w_off: u64,
x: &Buffer,
y: &Buffer,
row0: usize,
nb: usize,
rows: usize,
cols: usize,
) {
enc.set_compute_pipeline_state(&c.q4tpmm);
enc.set_buffer(0, Some(w), w_off);
enc.set_buffer(1, Some(x), (row0 * cols * 4) as u64);
enc.set_buffer(2, Some(y), (row0 * rows * 4) as u64);
set_u32(enc, 3, cols as u32);
set_u32(enc, 4, rows as u32);
set_u32(enc, 5, nb as u32);
set_f32(enc, 6, 1.0);
enc.dispatch_thread_groups(
MTLSize::new((nb as u64).div_ceil(32), (rows as u64).div_ceil(64), 1),
MTLSize::new(128, 1, 1),
);
}
fn n8_max() -> usize {
static N: OnceLock<usize> = OnceLock::new();
*N.get_or_init(|| std::env::var("CMF_CLOVER_N8_MAX").ok().and_then(|v| v.parse().ok()).unwrap_or(32))
}
#[allow(clippy::too_many_arguments)]
fn q4rows(
c: &Ctx,
enc: &ComputeCommandEncoderRef,
w: &View,
ones: &Buffer,
x: &Buffer,
y: &Buffer,
row0: usize,
nr: usize,
rows: usize,
cols: usize,
) {
if nr > n8_max() {
q4mm(c, enc, &w.0, w.1, x, y, row0, nr, rows, cols);
return;
}
let mut r = 0;
while r < nr {
let nb = (nr - r).min(8);
enc.set_compute_pipeline_state(&c.q4tpmmn8);
enc.set_buffer(0, Some(&w.0), w.1);
enc.set_buffer(1, Some(x), ((row0 + r) * cols * 4) as u64);
enc.set_buffer(2, Some(y), ((row0 + r) * rows * 4) as u64);
set_u32(enc, 3, cols as u32);
set_u32(enc, 4, rows as u32);
set_u32(enc, 5, nb as u32);
enc.set_buffer(6, Some(ones), 0);
enc.dispatch_thread_groups(MTLSize::new((rows as u64).div_ceil(64), 1, 1), MTLSize::new(128, 1, 1));
r += nb;
}
}
#[allow(clippy::too_many_arguments)]
fn norm_rows(
c: &Ctx,
enc: &ComputeCommandEncoderRef,
h: &Buffer,
d: Option<&Buffer>,
w: &Buffer,
o: &Buffer,
b: usize,
n: usize,
gemma: bool,
eps: f32,
) {
match d {
None => {
enc.set_compute_pipeline_state(&c.rmsrows);
bind(enc, &[(h, 0), (w, 0), (o, 0)]);
set_u32(enc, 3, n as u32);
set_u32(enc, 4, gemma as u32);
set_f32(enc, 5, eps);
}
Some(d) => {
enc.set_compute_pipeline_state(&c.addnorm);
bind(enc, &[(h, 0), (d, 0), (w, 0), (o, 0)]);
set_u32(enc, 4, n as u32);
set_u32(enc, 5, gemma as u32);
set_f32(enc, 6, eps);
set_u32(enc, 7, 1);
}
}
enc.dispatch_thread_groups(MTLSize::new(b as u64, 1, 1), MTLSize::new(256, 1, 1));
}
#[allow(clippy::too_many_arguments)]
fn f32mv(c: &Ctx, enc: &ComputeCommandEncoderRef, w: &Buffer, x: &Buffer, y: &Buffer, b: usize, rows: usize, cols: usize) {
enc.set_compute_pipeline_state(&c.f32mvr8);
bind(enc, &[(w, 0), (x, 0), (y, 0)]);
set_u32(enc, 3, cols as u32);
set_u32(enc, 4, rows as u32);
set_u32(enc, 5, b as u32);
enc.dispatch_thread_groups(
MTLSize::new((rows as u64).div_ceil(8), (b as u64).div_ceil(8), 1),
MTLSize::new(256, 1, 1),
);
}
fn commit_wait(cmd: &CommandBuffer, st: &mut PreviewStats) -> Result<(), String> {
let t = std::time::Instant::now();
cmd.commit();
let r = wait_fast_checked(cmd);
st.gpu_s += t.elapsed().as_secs_f64();
st.exec_s += super::cmd_gpu_ms(cmd) / 1e3;
r
}
fn fits(d: &Pass, b: usize, causal: bool) -> Result<(), String> {
let g = d.gdn;
let ok = (1..=256).contains(&b)
&& d.hidden.is_multiple_of(32)
&& !d.layers.is_empty()
&& g.dk <= 256
&& (causal || b * g.dk <= 4096)
&& g.dv <= 256
&& 256 % g.dv == 0
&& g.nk > 0
&& g.nv.is_multiple_of(g.nk)
&& g.kk >= 2
&& g.kk <= 8
&& (g.dk == 64 || g.dk == 128)
&& d.layers.iter().all(|l| {
l.moe.experts.len() == d.layers[0].moe.experts.len()
&& l.moe.router.len() == l.moe.experts.len() * d.hidden
&& l.moe.sgate.len() == d.hidden
&& match &l.mixer {
Mixer::Gdn(x) => x.a.len() == g.nv * d.hidden && x.bproj.len() == g.nv * d.hidden,
Mixer::Attn(a) => {
a.hd <= 256
&& a.hd % 32 == 0
&& a.rd <= a.hd
&& (a.rd / 2) % 32 == 0
&& a.nkv > 0
&& a.nh % a.nkv == 0
&& a.nh / a.nkv <= 8
}
}
});
if ok {
Ok(())
} else {
Err("clover metal: a layer's shape is outside the preview kernels".into())
}
}
pub fn preview(
d: &Pass,
h0: &[f32],
b: usize,
pos: usize,
route: &mut RouteFn<'_>,
logits: &mut Vec<f32>,
st: &mut PreviewStats,
) -> Result<(), String> {
run(d, h0, b, pos, false, route, Some(logits), st)
}
pub fn commit(d: &Pass, h0: &[f32], b: usize, pos: usize, route: &mut RouteFn<'_>, st: &mut PreviewStats) -> Result<(), String> {
run(d, h0, b, pos, true, route, None, st)
}
pub fn commit_scored(
d: &Pass,
h0: &[f32],
b: usize,
pos: usize,
route: &mut RouteFn<'_>,
logits: &mut Vec<f32>,
st: &mut PreviewStats,
) -> Result<(), String> {
run(d, h0, b, pos, true, route, Some(logits), st)
}
pub enum CommitPart<'s> {
State(&'s [f32]),
Kv(&'s [f32], &'s [f32]),
}
pub fn commit_results(out: &mut dyn FnMut(usize, CommitPart<'_>)) -> Result<(), String> {
let guard = EXEC.lock().unwrap();
let ex = guard.as_ref().ok_or("clover metal: no executor")?;
let (b, stored, state_len) = ex.last_commit.ok_or("clover metal: no commit to read")?;
for (li, k) in ex.consts.iter().enumerate() {
if let Some(gc) = &k.gdn {
out(li, CommitPart::State(unsafe { std::slice::from_raw_parts(gc.state.contents() as *const f32, state_len) }));
}
if let Some(ac) = &k.attn {
let (nkv, hd) = (ac.nkv, ac.hd);
let (mut kr, mut vr) = (vec![0f32; b * nkv * hd], vec![0f32; b * nkv * hd]);
for (src, dst) in [(&ac.k, &mut kr), (&ac.v, &mut vr)] {
let p = src.contents() as *const f32;
for r in 0..b {
for h in 0..nkv {
let row = unsafe { std::slice::from_raw_parts(p.add((h * ac.cap + stored + r) * hd), hd) };
dst[(r * nkv + h) * hd..(r * nkv + h + 1) * hd].copy_from_slice(row);
}
}
}
out(li, CommitPart::Kv(&kr, &vr));
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn run(
d: &Pass,
h0: &[f32],
b: usize,
pos: usize,
causal: bool,
route: &mut RouteFn<'_>,
logits: Option<&mut Vec<f32>>,
st: &mut PreviewStats,
) -> Result<(), String> {
let c = ctx().ok_or("clover metal: no device")?;
let p = pipes(c)?;
fits(d, b, causal)?;
let hs = d.hidden;
if h0.len() != b * hs {
return Err("clover metal: hidden rows of the wrong size".into());
}
let mut guard = EXEC.lock().unwrap();
if guard.as_ref().is_none_or(|e| e.model_uid != d.model.uid() || e.consts.len() != d.layers.len()) {
*guard = None;
*guard = Some(Exec::new(c, d)?);
}
let ex = guard.as_mut().unwrap();
ex.sync(c, d, b)?;
let r = run_on(c, p, ex, d, h0, b, pos, causal, route, logits, st);
if r.is_err() && causal {
ex.epoch = None;
for k in &mut ex.consts {
if let Some(ac) = k.attn.as_mut() {
ac.uploaded = 0;
}
}
}
r
}
#[allow(clippy::too_many_arguments)]
fn run_on(
c: &Ctx,
p: &Pipes,
ex: &mut Exec,
d: &Pass,
h0: &[f32],
b: usize,
pos: usize,
causal: bool,
route: &mut RouteFn<'_>,
logits: Option<&mut Vec<f32>>,
st: &mut PreviewStats,
) -> Result<(), String> {
let hs = d.hidden;
let g = d.gdn;
let (cdim, vd, kd) = (g.c_dim(), g.nv * g.dv, g.nk * g.dk);
let ne = d.layers[0].moe.experts.len();
let inter = d.inter;
let gemma = d.gemma;
let h = ex.scratch(c, "h", b * hs * 4);
let n = ex.scratch(c, "n", b * hs * 4);
let dl = ex.scratch(c, "d", b * hs * 4);
let lg = ex.scratch(c, "lg", b * (ne + 1) * 4);
write_f32(&h, 0, h0);
let max_rows = |pick: fn(&Mixer) -> usize| d.layers.iter().map(|l| pick(&l.mixer)).max().unwrap_or(0);
let wq_rows = max_rows(|m| match m {
Mixer::Attn(a) => a.nh * a.hd * if a.output_gate { 2 } else { 1 },
_ => 0,
});
let nhd = max_rows(|m| match m {
Mixer::Attn(a) => a.nh * a.hd,
_ => 0,
});
let kvd = max_rows(|m| match m {
Mixer::Attn(a) => a.nkv * a.hd,
_ => 0,
});
let nblk_max = d
.layers
.iter()
.filter_map(|l| match &l.mixer {
Mixer::Attn(a) => Some((a.k.first().map_or(0, |k| k.len() / a.hd) + b).div_ceil(128) * a.nh),
_ => None,
})
.max()
.unwrap_or(0);
let qkv_b = ex.scratch(c, "qkv", b * cdim * 4);
let cq_b = ex.scratch(c, "cq", b * cdim * 4);
let z_b = ex.scratch(c, "z", b * vd * 4);
let a_b = ex.scratch(c, "a", b * g.nv * 4);
let o_b = ex.scratch(c, "o", b * vd * 4);
let q_b = ex.scratch(c, "q", b * wq_rows * 4);
let k_b = ex.scratch(c, "k", b * kvd * 4);
let v_b = ex.scratch(c, "v", b * kvd * 4);
let qr_b = ex.scratch(c, "qr", b * nhd * 4);
let gt_b = ex.scratch(c, "gt", b * nhd * 4);
let ao_b = ex.scratch(c, "ao", b * nhd * 4);
let part_b = ex.scratch(c, "part", b * nblk_max * (256 + 2) * 4);
let bb_b = ex.scratch(c, "bb", b * g.nv * 4);
let gg_b = ex.scratch(c, "gg", b * g.nv * 4);
let bt_b = ex.scratch(c, "bt", b * g.nv * 4);
let iq_b = ex.scratch(c, "iq", b * g.nk * 4);
let ik_b = ex.scratch(c, "ik", b * g.nk * 4);
let gcfg = super::GdnGpuCfg {
nv: g.nv,
nk: g.nk,
dk: g.dk,
dv: g.dv,
kk: g.kk,
hidden: hs,
inter: d.inter,
c_dim: cdim,
eps: g.eps,
gemma: d.gemma,
};
let prof = std::env::var("CMF_CLOVER_METAL_PROF").is_ok_and(|v| v == "1");
let cut = |cmd: &mut CommandBuffer, st: &mut PreviewStats, slot: usize| -> Result<(), String> {
if prof {
commit_wait(cmd, st)?;
st.stage_s[slot] += super::cmd_gpu_ms(cmd) / 1e3;
*cmd = c.queue.new_command_buffer().to_owned();
}
Ok(())
};
let mut cmd = c.queue.new_command_buffer().to_owned();
for (li, l) in d.layers.iter().enumerate() {
let k = &ex.consts[li];
let mut enc = cmd.new_compute_command_encoder();
norm_rows(c, enc, &h, (li > 0).then_some(&dl), &k.in_norm, &n, b, hs, gemma, d.eps);
match (&l.mixer, k.gdn.as_ref(), k.attn.as_ref()) {
(Mixer::Gdn(gl), Some(gc), _) if causal => {
q8mm(p, enc, ex.view(gl.qkv)?, &n, &qkv_b, b, cdim, hs);
q8mm(p, enc, ex.view(gl.z)?, &n, &z_b, b, vd, hs);
f32mv(c, enc, &gc.a, &n, &a_b, b, g.nv, hs);
f32mv(c, enc, &gc.bproj, &n, &bb_b, b, g.nv, hs);
enc.set_compute_pipeline_state(&c.gdnconvb);
bind(enc, &[(&qkv_b, 0), (&gc.state, 0), (&gc.conv1d, 0), (&cq_b, 0)]);
set_u32(enc, 4, cdim as u32);
set_u32(enc, 5, g.kk as u32);
set_u32(enc, 6, b as u32);
enc.dispatch_threads(MTLSize::new(cdim as u64, b as u64, 1), MTLSize::new(256, 1, 1));
enc.set_compute_pipeline_state(&c.gdngatesb);
bind(enc, &[(&a_b, 0), (&bb_b, 0), (&gc.a_log, 0), (&gc.dt_bias, 0), (&gg_b, 0), (&bt_b, 0)]);
set_u32(enc, 6, g.nv as u32);
set_u32(enc, 7, (b * g.nv) as u32);
enc.dispatch_threads(MTLSize::new((b * g.nv) as u64, 1, 1), MTLSize::new(64, 1, 1));
enc.set_compute_pipeline_state(&c.gdnqknb);
bind(enc, &[(&cq_b, 0), (&iq_b, 0), (&ik_b, 0)]);
for (i, v) in [g.nk, g.dk, cdim, b].iter().enumerate() {
set_u32(enc, 3 + i as u64, *v as u32);
}
enc.dispatch_threads(MTLSize::new(((b * g.nk) as u64).div_ceil(8) * 256, 1, 1), MTLSize::new(256, 1, 1));
super::encode_gdn_state_b(
c,
enc,
&gc.state,
(g.ring_len() * 4) as u64,
&cq_b,
&z_b,
&gg_b,
&bt_b,
&iq_b,
&ik_b,
&gc.norm,
&o_b,
&gcfg,
b,
3,
!super::state4_on(),
);
enc.set_compute_pipeline_state(&c.gdnringcb);
bind(enc, &[(&gc.state, 0), (&qkv_b, 0)]);
set_u32(enc, 2, cdim as u32);
set_u32(enc, 3, g.kk as u32);
set_u32(enc, 4, b as u32);
enc.dispatch_threads(MTLSize::new(cdim as u64, 1, 1), MTLSize::new(256, 1, 1));
q8mm(p, enc, ex.view(gl.out)?, &o_b, &dl, b, hs, vd);
}
(Mixer::Gdn(gl), Some(gc), _) => {
if !skip('q') {
q8mm(p, enc, ex.view(gl.qkv)?, &n, &qkv_b, b, cdim, hs);
q8mm(p, enc, ex.view(gl.z)?, &n, &z_b, b, vd, hs);
}
f32mv(c, enc, &gc.a, &n, &a_b, b, g.nv, hs);
enc.set_compute_pipeline_state(&p.conv);
bind(enc, &[(&qkv_b, 0), (&gc.conv1d, 0), (&gc.right, 0), (&gc.state, 0), (&cq_b, 0)]);
set_u32(enc, 5, cdim as u32);
set_u32(enc, 6, g.kk as u32);
set_u32(enc, 7, b as u32);
enc.dispatch_threads(MTLSize::new(cdim as u64, b as u64, 1), MTLSize::new(256, 1, 1));
if !skip('m') {
enc.set_compute_pipeline_state(&p.mix);
bind(
enc,
&[
(&cq_b, 0),
(&z_b, 0),
(&a_b, 0),
(&gc.a_log, 0),
(&gc.dt_bias, 0),
(&gc.norm, 0),
(&gc.state, (g.ring_len() * 4) as u64),
(&o_b, 0),
(&gc.lsc, 0),
(&gc.hb, 0),
(&gc.hc, 0),
],
);
for (i, v) in [g.nv, g.dk, g.dv, kd, g.nv / g.nk, cdim, b, g.gate_sigmoid as usize].iter().enumerate() {
set_u32(enc, 11 + i as u64, *v as u32);
}
set_f32(enc, 19, g.eps);
set_f32(enc, 20, (b as f64 / 32.0).ln() as f32);
enc.dispatch_thread_groups(
MTLSize::new(g.nv as u64, (b as u64).div_ceil(4), 1),
MTLSize::new(256, 1, 1),
);
}
if !skip('o') {
q8mm(p, enc, ex.view(gl.out)?, &o_b, &dl, b, hs, vd);
}
}
(Mixer::Attn(al), _, Some(ac)) => {
let (nh, nkv, hd) = (al.nh, al.nkv, al.hd);
let stored = ac.uploaded;
q8mm(p, enc, ex.view(al.wq)?, &n, &q_b, b, nh * hd * if al.output_gate { 2 } else { 1 }, hs);
q8mm(p, enc, ex.view(al.wk)?, &n, &k_b, b, nkv * hd, hs);
q8mm(p, enc, ex.view(al.wv)?, &n, &v_b, b, nkv * hd, hs);
let flags = (al.output_gate as u32)
| ((al.q_norm.is_some() as u32) << 1)
| ((al.k_norm.is_some() as u32) << 2)
| ((gemma as u32) << 3);
enc.set_compute_pipeline_state(&c.rqknb);
bind(enc, &[(&q_b, 0), (&k_b, 0), (&qr_b, 0), (>_b, 0), (&ac.qn, 0), (&ac.kn, 0), (&ac.invf, 0)]);
for (i, v) in [nh, nkv, hd, al.rd, pos, flags as usize].iter().enumerate() {
set_u32(enc, 7 + i as u64, *v as u32);
}
set_f32(enc, 13, d.eps);
set_u32(enc, 14, b as u32);
set_f32(enc, 15, al.rope_scale);
enc.dispatch_threads(MTLSize::new(((nh + nkv) * 32) as u64, b as u64, 1), MTLSize::new(256, 1, 1));
enc.set_compute_pipeline_state(&c.kvappb);
bind(enc, &[(&k_b, 0), (&v_b, 0), (&ac.k, 0), (&ac.v, 0)]);
for (i, v) in [nkv, hd, ac.cap, stored, b].iter().enumerate() {
set_u32(enc, 4 + i as u64, *v as u32);
}
enc.dispatch_threads(MTLSize::new((nkv * hd) as u64, b as u64, 1), MTLSize::new(256, 1, 1));
let n_all = stored + b;
let nblk = n_all.div_ceil(128);
let hpk = nh / nkv;
if causal {
if !super::encode_gqa_attend_blk(
c, enc, &qr_b, &ac.k, &ac.v, 0, &ao_b, nh, nkv, hd, ac.cap, stored, b, al.scale, false,
) {
return Err("clover metal: the causal attend refused the layer".into());
}
} else {
enc.set_compute_pipeline_state(&p.attend);
bind(enc, &[(&qr_b, 0), (&ac.k, 0), (&ac.v, 0), (&part_b, 0)]);
for (i, v) in [nh, hpk, hd, ac.cap, n_all, b, nblk].iter().enumerate() {
set_u32(enc, 4 + i as u64, *v as u32);
}
set_f32(enc, 11, al.scale);
enc.dispatch_thread_groups(
MTLSize::new(nkv as u64, nblk as u64, b as u64),
MTLSize::new((hpk * 32) as u64, 1, 1),
);
enc.set_compute_pipeline_state(&c.gqacomb);
bind(enc, &[(&part_b, 0), (&ao_b, 0)]);
for (i, v) in [nh, hd, b, nblk].iter().enumerate() {
set_u32(enc, 2 + i as u64, *v as u32);
}
enc.dispatch_thread_groups(MTLSize::new(nh as u64, b as u64, 1), MTLSize::new(hd as u64, 1, 1));
}
if al.output_gate {
enc.set_compute_pipeline_state(&c.sgate);
bind(enc, &[(&ao_b, 0), (>_b, 0)]);
set_u32(enc, 2, (b * nh * hd) as u32);
enc.dispatch_threads(MTLSize::new((b * nh * hd) as u64, 1, 1), MTLSize::new(256, 1, 1));
}
q8mm(p, enc, ex.view(al.wo)?, &ao_b, &dl, b, hs, nh * hd);
}
_ => return Err(format!("clover metal: layer {li} has no device constants for its mixer")),
}
if causal && logits.is_none() && li + 1 == d.layers.len() {
enc.end_encoding();
break;
}
if prof {
enc.end_encoding();
commit_wait(&cmd, st)?;
st.stage_s[if matches!(l.mixer, Mixer::Gdn(_)) { 0 } else { 1 }] += super::cmd_gpu_ms(&cmd) / 1e3;
cmd = c.queue.new_command_buffer().to_owned();
enc = cmd.new_compute_command_encoder();
}
if prof {
st.stage_s[2] += super::cmd_gpu_ms(&cmd) / 1e3;
}
norm_rows(c, enc, &h, Some(&dl), &k.post_norm, &n, b, hs, gemma, d.eps);
f32mv(c, enc, &k.router, &n, &lg, b, ne + 1, hs);
enc.end_encoding();
commit_wait(&cmd, st)?;
cmd = c.queue.new_command_buffer().to_owned();
let t_route = std::time::Instant::now();
let lgx = unsafe { std::slice::from_raw_parts(lg.contents() as *const f32, b * (ne + 1)) };
let mut logits_ne = Vec::with_capacity(b * ne);
for r in 0..b {
logits_ne.extend_from_slice(&lgx[r * (ne + 1)..r * (ne + 1) + ne]);
}
let routes = route(li, &logits_ne, b).ok_or("clover metal: routing failed")?;
if routes.len() != b {
return Err("clover metal: routing returned the wrong row count".into());
}
let mut assign: Vec<Vec<(usize, f32)>> = vec![Vec::new(); ne];
for (r, row) in routes.iter().enumerate() {
for &(e, w) in row {
if e >= ne {
return Err("clover metal: routed expert out of range".into());
}
assign[e].push((r, w));
}
}
st.route_s += t_route.elapsed().as_secs_f64();
ex.experts.tick += 1;
let mut panels: Vec<([View; 3], usize, usize)> = Vec::new(); let mut gidx: Vec<u32> = Vec::new();
let mut slots: Vec<Vec<(u32, f32)>> = vec![Vec::new(); b];
for (e, list) in assign.iter().enumerate() {
if list.is_empty() {
continue;
}
let w = ex.experts.get(c, d.model, li, e, l.moe.experts[e], st)?;
let row0 = gidx.len();
for &(r, w) in list {
slots[r].push((gidx.len() as u32, w));
gidx.push(r as u32);
}
panels.push((w, row0, list.len()));
}
let row0 = gidx.len();
for (r, slot) in slots.iter_mut().enumerate() {
let sg = lgx[r * (ne + 1) + ne];
slot.push((gidx.len() as u32, 1.0 / (1.0 + (-sg).exp())));
gidx.push(r as u32);
}
let sh = l.moe.shared;
panels.push(([ex.view(sh.0)?, ex.view(sh.1)?, ex.view(sh.2)?], row0, b));
let n_as = gidx.len();
let kmax = slots.iter().map(|s| s.len()).max().unwrap_or(1);
let gidx_b = ex.scratch(c, "gidx", n_as * 4);
let cslot_b = ex.scratch(c, "cslot", b * kmax * 4);
let cw_b = ex.scratch(c, "cw", b * kmax * 4);
unsafe {
std::ptr::copy_nonoverlapping(gidx.as_ptr(), gidx_b.contents() as *mut u32, n_as);
let (ps, pw) = (cslot_b.contents() as *mut u32, cw_b.contents() as *mut f32);
for (r, row) in slots.iter().enumerate() {
for j in 0..kmax {
let (s, w) = row.get(j).copied().unwrap_or((row[0].0, 0.0));
*ps.add(r * kmax + j) = s;
*pw.add(r * kmax + j) = w;
}
}
}
let xg_b = ex.scratch(c, "xg", n_as * hs * 4);
let gb = ex.scratch(c, "gb", n_as * inter * 4);
let ub = ex.scratch(c, "ub", n_as * inter * 4);
let act_b = ex.scratch(c, "act", n_as * inter * 4);
let up_b = ex.scratch(c, "up", n_as * 4);
let eo_b = ex.scratch(c, "eo", n_as * hs * 4);
let enc = cmd.new_compute_command_encoder();
enc.set_compute_pipeline_state(&c.gatherr);
bind(enc, &[(&n, 0), (&xg_b, 0), (&gidx_b, 0)]);
set_u32(enc, 3, hs as u32);
enc.dispatch_threads(MTLSize::new(hs as u64, n_as as u64, 1), MTLSize::new(256, 1, 1));
enc.end_encoding();
cut(&mut cmd, st, 3)?;
let enc = cmd.compute_command_encoder_with_dispatch_type(metal::MTLDispatchType::Concurrent);
for (w, r0, nr) in &panels {
q4rows(c, enc, &w[0], &ex.ones, &xg_b, &gb, *r0, *nr, inter, hs);
q4rows(c, enc, &w[1], &ex.ones, &xg_b, &ub, *r0, *nr, inter, hs);
}
enc.end_encoding();
cut(&mut cmd, st, 4)?;
let enc = cmd.new_compute_command_encoder();
enc.set_compute_pipeline_state(&c.moeact);
bind(enc, &[(&gb, 0), (&ub, 0), (&act_b, 0), (&up_b, 0)]);
set_u32(enc, 4, inter as u32);
enc.dispatch_thread_groups(MTLSize::new(n_as as u64, 1, 1), MTLSize::new(256, 1, 1));
enc.end_encoding();
cut(&mut cmd, st, 5)?;
let enc = cmd.compute_command_encoder_with_dispatch_type(metal::MTLDispatchType::Concurrent);
for (w, r0, nr) in &panels {
q4rows(c, enc, &w[2], &ex.ones, &act_b, &eo_b, *r0, *nr, hs, inter);
}
enc.end_encoding();
cut(&mut cmd, st, 6)?;
let enc = cmd.new_compute_command_encoder();
enc.set_compute_pipeline_state(&c.moecomb);
bind(enc, &[(&eo_b, 0), (&dl, 0), (&cslot_b, 0), (&cw_b, 0)]);
set_u32(enc, 4, hs as u32);
set_u32(enc, 5, kmax as u32);
enc.set_buffer(6, Some(&up_b), 0);
enc.dispatch_threads(MTLSize::new(hs as u64, b as u64, 1), MTLSize::new(256, 1, 1));
enc.end_encoding();
cut(&mut cmd, st, 7)?;
}
let book = |ex: &mut Exec| {
let stored = ex.consts.iter().find_map(|k| k.attn.as_ref().map(|a| a.uploaded)).unwrap_or(0);
for k in &mut ex.consts {
if let Some(ac) = k.attn.as_mut() {
ac.uploaded += b;
}
}
ex.last_commit = Some((b, stored, g.state_len()));
ex.epoch = Some((d.epoch.0, pos + b));
};
let Some(logits) = logits else {
commit_wait(&cmd, st)?;
book(ex);
return Ok(());
};
let lm_rows = d.lm_rows;
let lo_b = ex.scratch(c, "logits", b * lm_rows * 4);
let enc = cmd.new_compute_command_encoder();
norm_rows(c, enc, &h, Some(&dl), &ex.final_norm, &n, b, hs, gemma, d.eps);
q8mm(p, enc, ex.view(d.lm_head)?, &n, &lo_b, b, lm_rows, hs);
enc.end_encoding();
commit_wait(&cmd, st)?;
if prof {
st.stage_s[8] += super::cmd_gpu_ms(&cmd) / 1e3;
}
if causal {
book(ex);
}
logits.resize(b * lm_rows, 0.0);
unsafe { std::ptr::copy_nonoverlapping(lo_b.contents() as *const f32, logits.as_mut_ptr(), b * lm_rows) };
Ok(())
}
pub fn release() {
*EXEC.lock().unwrap() = None;
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn clover_kernels_compile() {
let Some(c) = ctx() else {
return;
};
if let Err(e) = pipes(c) {
panic!("{e}");
}
}
}