Skip to main content

kime_cpu/
plan.rs

1//! The CPU backend: graphs lowered to a list of steps over one arena, run on a persistent pool.
2//!
3//! Lowering resolves every value to an arena offset, every weight to its converted tensor and every
4//! RoPE base to its table, and checks every shape, so running a step is a match and a call. The
5//! arena, the per worker attention scratch and the per batch index tables are sized for the bucket
6//! when the plan is built, and nothing on the run path allocates.
7//!
8//! Each step runs on the rows the batch has rather than the bucket's padded count. Padding only
9//! matters to a backend that captures a fixed shape, and on the CPU it would be wasted work. The
10//! kernels are the reference ones, so a plan gives the same bits as [`Compat`](crate::Compat).
11//!
12//! A backend made [`with_int8`](CpuBackend::with_int8) runs the GEMMs over token rows in INT8
13//! instead, which is every GEMM of the encoder and the decision head. The scorer and the act head
14//! run on a row per option or per question and stay in FP32, as spec/10-cpu.md has it.
15
16use std::cell::UnsafeCell;
17use std::sync::Arc;
18use std::time::Instant;
19
20use kime_tensor::plan::{Epilogue, Graph, Op, Rows, Val, layout};
21use kime_tensor::{Backend, Batch, Bucket, Caps, Error, HostTensor, Outputs, Result};
22
23use crate::attention::{self, HEAD, QB};
24use crate::gemm::{self, Gemm};
25use crate::ops::{Rope, geglu, layer_norm};
26use crate::par::{self, Shared};
27use crate::pool::Pool;
28use crate::qgemm::{self, QGemm, QMatrix};
29
30/// A weight converted to f32.
31#[derive(Debug)]
32pub struct Tensor {
33    /// Shape.
34    pub shape: Vec<usize>,
35    /// Row major values, empty for a weight only GEMMs read.
36    pub data: Vec<f32>,
37    /// The values as [`gemm::pack`] lays them out, for a weight FP32 GEMMs read, and empty
38    /// otherwise.
39    pub packed: Vec<f32>,
40    /// The values rounded to INT8, for a weight INT8 GEMMs read.
41    pub quant: Option<QMatrix>,
42}
43
44/// Every weight of a checkpoint, shared by all the plans built from it.
45#[derive(Debug, Clone)]
46pub struct Weights(Arc<[Tensor]>);
47
48/// The CPU backend, which owns its threads.
49#[derive(Debug)]
50pub struct CpuBackend {
51    pool: Pool,
52    int8: bool,
53}
54
55impl CpuBackend {
56    /// A backend on `threads` threads, the calling thread included.
57    #[must_use]
58    pub fn new(threads: usize) -> Self {
59        Self { pool: Pool::new(threads.max(1)), int8: false }
60    }
61
62    /// The same backend with the GEMMs over token rows in INT8 when `on`: weights rounded per
63    /// output channel at upload, activations per row as they are read, sums in i32. See
64    /// [`qgemm`](crate::qgemm).
65    #[must_use]
66    pub fn with_int8(mut self, on: bool) -> Self {
67        self.int8 = on;
68        self
69    }
70
71    /// Whether the GEMMs over token rows run in INT8.
72    #[must_use]
73    pub fn int8(&self) -> bool {
74        self.int8
75    }
76
77    /// Whether a GEMM reading `a` runs in INT8.
78    fn int8_rows(&self, rows: Rows) -> bool {
79        self.int8 && rows == Rows::Tokens
80    }
81
82    /// Threads.
83    #[must_use]
84    pub fn threads(&self) -> usize {
85        self.pool.threads()
86    }
87}
88
89/// A value resolved to its place in the arena.
90#[derive(Debug, Clone, Copy)]
91struct Loc {
92    off: usize,
93    rows: Rows,
94    width: usize,
95}
96
97#[derive(Debug, Clone, Copy)]
98enum Step {
99    Embed { table: usize, out: Loc },
100    LayerNorm { x: Loc, w: usize, b: Option<usize>, eps: f64, out: Loc },
101    Gemm { a: Loc, w: usize, b: Option<usize>, ep: Epilogue, out: Loc },
102    Gemm8 { a: Loc, w: usize, b: Option<usize>, ep: Epilogue, out: Loc },
103    Rope { qkv: Loc, rope: usize },
104    Attention { qkv: Loc, window: Option<usize>, out: Loc },
105    GeGlu { x: Loc, out: Loc },
106    AddType { h: Loc, table: usize },
107    Gather { h: Loc, out: Loc },
108    ActFeatures { h: Loc, logits: Loc, out: Loc },
109}
110
111impl Step {
112    fn name(&self) -> &'static str {
113        match self {
114            Step::Embed { .. } => "embed",
115            Step::LayerNorm { .. } => "layer norm",
116            Step::Gemm { .. } => "gemm",
117            Step::Gemm8 { .. } => "gemm int8",
118            Step::Rope { .. } => "rope",
119            Step::Attention { .. } => "attention",
120            Step::GeGlu { .. } => "geglu",
121            Step::AddType { .. } => "type embedding",
122            Step::Gather { .. } => "gather markers",
123            Step::ActFeatures { .. } => "act features",
124        }
125    }
126
127    /// Where the step writes.
128    fn out(&self) -> Loc {
129        match *self {
130            Step::Embed { out, .. }
131            | Step::LayerNorm { out, .. }
132            | Step::Gemm { out, .. }
133            | Step::Gemm8 { out, .. }
134            | Step::Attention { out, .. }
135            | Step::GeGlu { out, .. }
136            | Step::Gather { out, .. }
137            | Step::ActFeatures { out, .. } => out,
138            Step::Rope { qkv, .. } => qkv,
139            Step::AddType { h, .. } => h,
140        }
141    }
142}
143
144/// What one step wrote in one run, from [`CpuPlan::dumps`].
145#[derive(Debug, Clone)]
146pub struct Dump {
147    /// The kind of step.
148    pub name: &'static str,
149    /// Whether the rows are tokens, sequences or markers.
150    pub rows: Rows,
151    /// Values per row.
152    pub width: usize,
153    /// The live rows, row major.
154    pub data: Vec<f32>,
155}
156
157/// One slot per worker, each touched only by its own worker.
158struct PerWorker<T>(Vec<UnsafeCell<T>>);
159
160// SAFETY: slot w is only reached through `get(w)` from the task running on worker w, and the pool
161// never runs two tasks on one worker at once.
162unsafe impl<T: Send> Sync for PerWorker<T> {}
163
164impl<T> PerWorker<T> {
165    /// # Safety
166    ///
167    /// Only the task running on `worker` may call this, and only for its own worker.
168    #[allow(clippy::mut_from_ref)]
169    unsafe fn get(&self, worker: usize) -> &mut T {
170        // SAFETY: the caller is the only user of slot `worker` right now.
171        unsafe { &mut *self.0[worker].get() }
172    }
173}
174
175/// A graph lowered for one bucket.
176pub struct CpuPlan {
177    w: Weights,
178    bucket: Bucket,
179    steps: Vec<Step>,
180    arena: Vec<f32>,
181    ropes: Vec<Rope>,
182    logits: Loc,
183    act: Loc,
184    /// Token starts of each sequence, then marker starts, then the sequence of each token, then
185    /// the query blocks attention runs. Rebuilt per batch in place.
186    cu: Vec<usize>,
187    mcu: Vec<usize>,
188    row_seq: Vec<u32>,
189    blocks: Vec<(u32, u32)>,
190    scratch: PerWorker<Vec<f32>>,
191    /// Nanoseconds per step, summed over runs, when profiling.
192    profile: Option<Vec<u64>>,
193    /// Every step's output from the last run, when dumping.
194    dumps: Option<Vec<Dump>>,
195}
196
197impl std::fmt::Debug for CpuPlan {
198    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
199        f.debug_struct("CpuPlan")
200            .field("bucket", &self.bucket)
201            .field("steps", &self.steps.len())
202            .field("arena", &self.arena.len())
203            .finish_non_exhaustive()
204    }
205}
206
207impl CpuPlan {
208    /// The bucket it was built for.
209    #[must_use]
210    pub fn bucket(&self) -> Bucket {
211        self.bucket
212    }
213
214    /// Arena size in bytes.
215    #[must_use]
216    pub fn arena_bytes(&self) -> usize {
217        self.arena.len() * 4
218    }
219
220    /// Starts timing every step, which costs two clock reads per step.
221    pub fn profile(&mut self) {
222        self.profile = Some(vec![0; self.steps.len()]);
223    }
224
225    /// Keeps a copy of what every step writes from now on, for finding the first step where two
226    /// runs differ. It allocates on every run, so it is for tests and debugging only.
227    pub fn dump(&mut self) {
228        self.dumps = Some(Vec::new());
229    }
230
231    /// What every step wrote in the last run since [`CpuPlan::dump`], in order.
232    #[must_use]
233    pub fn dumps(&self) -> &[Dump] {
234        self.dumps.as_deref().unwrap_or_default()
235    }
236
237    /// Time per kind of step since [`CpuPlan::profile`], in nanoseconds, largest first.
238    #[must_use]
239    pub fn timings(&self) -> Vec<(&'static str, u64)> {
240        let mut by: Vec<(&'static str, u64)> = Vec::new();
241        for (s, &ns) in self.steps.iter().zip(self.profile.iter().flatten()) {
242            match by.iter_mut().find(|b| b.0 == s.name()) {
243                Some(b) => b.1 += ns,
244                None => by.push((s.name(), ns)),
245            }
246        }
247        by.sort_by_key(|b| std::cmp::Reverse(b.1));
248        by
249    }
250}
251
252/// A pointer to an arena that steps carve disjoint slices from.
253#[derive(Clone, Copy)]
254struct Arena(*mut f32);
255
256// SAFETY: the plan hands out slices of the arena only as the layout allows: values live at the
257// same time never overlap, and within a step each task writes rows no other task touches.
258unsafe impl Send for Arena {}
259// SAFETY: as above.
260unsafe impl Sync for Arena {}
261
262impl Arena {
263    /// # Safety
264    ///
265    /// The range must be in the arena and not written by anyone else while the slice lives.
266    unsafe fn slice<'a>(self, off: usize, len: usize) -> &'a [f32] {
267        // SAFETY: by the caller.
268        unsafe { std::slice::from_raw_parts(self.0.add(off), len) }
269    }
270
271    /// # Safety
272    ///
273    /// The range must be in the arena and not touched by anyone else while the slice lives.
274    #[allow(clippy::mut_from_ref)]
275    unsafe fn slice_mut<'a>(self, off: usize, len: usize) -> &'a mut [f32] {
276        // SAFETY: by the caller.
277        unsafe { std::slice::from_raw_parts_mut(self.0.add(off), len) }
278    }
279}
280
281/// Rows per task for the row wise steps.
282const ROWS: usize = 16;
283
284impl Backend for CpuBackend {
285    type Weights = Weights;
286    type Plan = CpuPlan;
287
288    fn caps(&self) -> Caps {
289        Caps { name: "cpu", threads: self.threads(), graphs: false, unified_memory: true }
290    }
291
292    fn weight_bytes(&self, w: &Weights) -> usize {
293        w.0.iter()
294            .map(|t| {
295                let q = t.quant.as_ref().map_or(0, |q| q.q.len() + 4 * q.scale.len());
296                4 * (t.data.len() + t.packed.len()) + q
297            })
298            .sum()
299    }
300
301    fn plan_bytes(&self, p: &CpuPlan) -> usize {
302        p.arena_bytes()
303    }
304
305    fn upload(&self, tensors: &[HostTensor<'_>], graph: &Graph) -> Result<Weights> {
306        // Weights the GEMMs read are packed or rounded once here, and kept row major only if
307        // something else reads them too.
308        let (mut gemm, mut other) = (vec![false; tensors.len()], vec![false; tensors.len()]);
309        let mut gemm8 = vec![false; tensors.len()];
310        let mark = |flags: &mut Vec<bool>, w: Option<usize>| {
311            if let Some(f) = w.and_then(|w| flags.get_mut(w)) {
312                *f = true;
313            }
314        };
315        for op in &graph.ops {
316            match *op {
317                Op::Gemm { a, w, b, .. } => {
318                    match self.int8_rows(graph.shape(a).rows) {
319                        true => mark(&mut gemm8, Some(w)),
320                        false => mark(&mut gemm, Some(w)),
321                    }
322                    mark(&mut other, b);
323                }
324                Op::Embed { table, .. } | Op::AddType { table, .. } => {
325                    mark(&mut other, Some(table))
326                }
327                Op::LayerNorm { w, b, .. } => {
328                    mark(&mut other, Some(w));
329                    mark(&mut other, b);
330                }
331                Op::Rope { .. }
332                | Op::Attention { .. }
333                | Op::GeGlu { .. }
334                | Op::GatherMarkers { .. }
335                | Op::ActFeatures { .. } => {}
336            }
337        }
338        let t = par::map(tensors.len(), self.threads(), |i| {
339            let h = &tensors[i];
340            let n = h.bytes.len() / h.dtype.size();
341            if n != h.shape.iter().product::<usize>() {
342                return Err(Error::Unsupported(format!(
343                    "tensor {i} has {n} values for {:?}",
344                    h.shape
345                )));
346            }
347            let data: Vec<f32> = (0..n).map(|j| h.dtype.read_f32(h.bytes, j)).collect();
348            let packed = match (gemm[i], h.shape) {
349                (true, &[rows, cols]) => gemm::pack(&data, rows, cols),
350                _ => Vec::new(),
351            };
352            let quant = match (gemm8[i], h.shape) {
353                (true, &[rows, cols]) => Some(QMatrix::quantize(&data, rows, cols)),
354                _ => None,
355            };
356            // The row major values go when every reader has its own copy.
357            let copied = (!gemm[i] || !packed.is_empty()) && (!gemm8[i] || quant.is_some());
358            let data = if (gemm[i] || gemm8[i]) && !other[i] && copied { Vec::new() } else { data };
359            Ok(Tensor { shape: h.shape.to_vec(), data, packed, quant })
360        });
361        Ok(Weights(t.into_iter().collect::<Result<Vec<_>>>()?.into()))
362    }
363
364    fn lower(&self, w: &Weights, graph: &Graph, bucket: Bucket) -> Result<CpuPlan> {
365        let lay = layout(graph, |r| bucket.rows(r));
366        let loc = |v: Val| {
367            let s = graph.shape(v);
368            Loc { off: lay.offsets[v.0 as usize], rows: s.rows, width: s.width }
369        };
370        let bad = |m: String| Err(Error::Unsupported(m));
371        let shape = |i: usize| -> Result<&[usize]> {
372            match w.0.get(i) {
373                Some(t) => Ok(&t.shape),
374                None => Err(Error::Unsupported(format!("weight {i} is not in the checkpoint"))),
375            }
376        };
377        let mut ropes: Vec<(u64, Rope)> = Vec::new();
378        let mut scratch = bucket.tokens;
379        let mut steps = Vec::with_capacity(graph.ops.len());
380        for (i, op) in graph.ops.iter().enumerate() {
381            let step = match *op {
382                Op::Embed { table, out } => {
383                    let out = loc(out);
384                    if shape(table)?.get(1) != Some(&out.width) || out.rows != Rows::Tokens {
385                        return bad(format!("op {i}: embedding table does not match its output"));
386                    }
387                    Step::Embed { table, out }
388                }
389                Op::LayerNorm { x, w: nw, b, eps, out } => {
390                    let (x, out) = (loc(x), loc(out));
391                    let ok = shape(nw)? == [x.width]
392                        && b.map_or(Ok(true), |b| shape(b).map(|s| s == [x.width]))?
393                        && x.width == out.width
394                        && x.rows == out.rows;
395                    if !ok {
396                        return bad(format!("op {i}: layer norm shapes do not match"));
397                    }
398                    Step::LayerNorm { x, w: nw, b, eps, out }
399                }
400                Op::Gemm { a, w: gw, b, epilogue, out } => {
401                    let (a, out) = (loc(a), loc(out));
402                    let ok = shape(gw)? == [out.width, a.width]
403                        && b.map_or(Ok(true), |b| shape(b).map(|s| s == [out.width]))?
404                        && a.rows == out.rows;
405                    if !ok {
406                        return bad(format!("op {i}: gemm shapes do not match"));
407                    }
408                    if self.int8_rows(a.rows) {
409                        if w.0[gw].quant.is_none() {
410                            return bad(format!("op {i}: gemm weight {gw} was not rounded"));
411                        }
412                        scratch = scratch.max(qgemm::scratch_len(a.width));
413                        Step::Gemm8 { a, w: gw, b, ep: epilogue, out }
414                    } else {
415                        if w.0[gw].packed.is_empty() && a.width * out.width > 0 {
416                            return bad(format!("op {i}: gemm weight {gw} was not packed"));
417                        }
418                        scratch = scratch.max(gemm::scratch_len(a.width, out.width));
419                        Step::Gemm { a, w: gw, b, ep: epilogue, out }
420                    }
421                }
422                Op::Rope { qkv, theta } => {
423                    let qkv = loc(qkv);
424                    if !qkv.width.is_multiple_of(3 * HEAD) || qkv.rows != Rows::Tokens {
425                        return bad(format!("op {i}: rope needs token rows of 3 heads 64"));
426                    }
427                    let at = match ropes.iter().position(|r| r.0 == theta.to_bits()) {
428                        Some(at) => at,
429                        None => {
430                            ropes.push((theta.to_bits(), Rope::new(theta, HEAD, bucket.tokens)));
431                            ropes.len() - 1
432                        }
433                    };
434                    Step::Rope { qkv, rope: at }
435                }
436                Op::Attention { qkv, window, out } => {
437                    let (qkv, out) = (loc(qkv), loc(out));
438                    let ok = qkv.width.is_multiple_of(3 * HEAD)
439                        && out.width * 3 == qkv.width
440                        && qkv.rows == Rows::Tokens
441                        && out.rows == Rows::Tokens;
442                    if !ok {
443                        return bad(format!("op {i}: attention shapes do not match"));
444                    }
445                    Step::Attention { qkv, window, out }
446                }
447                Op::GeGlu { x, out } => {
448                    let (x, out) = (loc(x), loc(out));
449                    if x.width != 2 * out.width || x.rows != out.rows {
450                        return bad(format!("op {i}: geglu input is not twice its output"));
451                    }
452                    Step::GeGlu { x, out }
453                }
454                Op::AddType { h, table } => {
455                    let h = loc(h);
456                    if shape(table)?.get(1) != Some(&h.width) || h.rows != Rows::Tokens {
457                        return bad(format!("op {i}: type table does not match"));
458                    }
459                    Step::AddType { h, table }
460                }
461                Op::GatherMarkers { h, out } => {
462                    let (h, out) = (loc(h), loc(out));
463                    if h.width != out.width || h.rows != Rows::Tokens || out.rows != Rows::Markers {
464                        return bad(format!("op {i}: gather shapes do not match"));
465                    }
466                    Step::Gather { h, out }
467                }
468                Op::ActFeatures { h, logits, out } => {
469                    let (h, logits, out) = (loc(h), loc(logits), loc(out));
470                    let ok = out.width == h.width + 4
471                        && logits.width == 1
472                        && logits.rows == Rows::Markers
473                        && out.rows == Rows::Seqs;
474                    if !ok {
475                        return bad(format!("op {i}: act feature shapes do not match"));
476                    }
477                    Step::ActFeatures { h, logits, out }
478                }
479            };
480            steps.push(step);
481        }
482        let (Some(logits), Some(act)) = (graph.logits, graph.act) else {
483            return bad("the graph has no logits or act output".into());
484        };
485        let (logits, act) = (loc(logits), loc(act));
486        if logits.width != 1
487            || logits.rows != Rows::Markers
488            || act.width != 2
489            || act.rows != Rows::Seqs
490        {
491            return bad(
492                "outputs must be one logit per marker and two act logits per sequence".into()
493            );
494        }
495        let threads = self.threads();
496        Ok(CpuPlan {
497            w: w.clone(),
498            bucket,
499            steps,
500            arena: vec![0.0; lay.len],
501            ropes: ropes.into_iter().map(|r| r.1).collect(),
502            logits,
503            act,
504            cu: Vec::with_capacity(bucket.seqs + 1),
505            mcu: Vec::with_capacity(bucket.seqs + 1),
506            row_seq: Vec::with_capacity(bucket.tokens),
507            blocks: Vec::with_capacity(bucket.tokens.div_ceil(QB) + bucket.seqs),
508            scratch: PerWorker(
509                (0..threads).map(|_| UnsafeCell::new(Vec::with_capacity(scratch))).collect(),
510            ),
511            profile: None,
512            dumps: None,
513        })
514    }
515
516    fn run(&self, plan: &mut CpuPlan, batch: &Batch<'_>, out: &mut Outputs) -> Result<()> {
517        let (t, s, m) = (batch.ids.len(), batch.seqs(), batch.markers.len());
518        if !plan.bucket.holds(t, s, m) {
519            return Err(Error::Batch(format!("batch does not fit bucket {}", plan.bucket)));
520        }
521        plan.cu.clear();
522        plan.cu.extend(batch.cu.iter().map(|&c| c as usize));
523        plan.mcu.clear();
524        plan.mcu.extend(batch.mcu.iter().map(|&c| c as usize));
525        plan.row_seq.clear();
526        plan.blocks.clear();
527        for q in 0..s {
528            let (lo, hi) = (plan.cu[q], plan.cu[q + 1]);
529            plan.row_seq.extend(std::iter::repeat_n(q as u32, hi - lo));
530            plan.blocks.extend((lo..hi).step_by(QB).map(|q0| (q as u32, q0 as u32)));
531        }
532        let ctx = Ctx {
533            pool: &self.pool,
534            w: &plan.w.0,
535            arena: Arena(plan.arena.as_mut_ptr()),
536            ropes: &plan.ropes,
537            batch,
538            cu: &plan.cu,
539            mcu: &plan.mcu,
540            row_seq: &plan.row_seq,
541            blocks: &plan.blocks,
542            scratch: &plan.scratch,
543            counts: [t, s, m],
544        };
545        if let Some(d) = plan.dumps.as_mut() {
546            d.clear();
547        }
548        for (i, step) in plan.steps.iter().enumerate() {
549            match plan.profile.as_mut() {
550                None => ctx.step(step),
551                Some(p) => {
552                    let at = Instant::now();
553                    ctx.step(step);
554                    p[i] += u64::try_from(at.elapsed().as_nanos()).unwrap_or(u64::MAX);
555                }
556            }
557            // Copied now, because a later step may reuse the same part of the arena.
558            if let Some(d) = plan.dumps.as_mut() {
559                let l = step.out();
560                // SAFETY: the step is done and the next has not started, so nothing else
561                // touches the arena.
562                let live = unsafe { ctx.arena.slice(l.off, ctx.rows(l.rows) * l.width) };
563                d.push(Dump {
564                    name: step.name(),
565                    rows: l.rows,
566                    width: l.width,
567                    data: live.to_vec(),
568                });
569            }
570        }
571
572        // SAFETY: the steps are done, so nothing else touches the arena.
573        let logits = unsafe { ctx.arena.slice(plan.logits.off, m) };
574        // SAFETY: as above.
575        let act = unsafe { ctx.arena.slice(plan.act.off, 2 * s) };
576        out.logits.clear();
577        out.logits.extend_from_slice(logits);
578        out.act.clear();
579        out.act.extend_from_slice(act.as_chunks::<2>().0);
580        Ok(())
581    }
582}
583
584/// Everything a step needs for one batch.
585struct Ctx<'a> {
586    pool: &'a Pool,
587    w: &'a [Tensor],
588    arena: Arena,
589    ropes: &'a [Rope],
590    batch: &'a Batch<'a>,
591    cu: &'a [usize],
592    mcu: &'a [usize],
593    row_seq: &'a [u32],
594    blocks: &'a [(u32, u32)],
595    scratch: &'a PerWorker<Vec<f32>>,
596    counts: [usize; 3],
597}
598
599impl Ctx<'_> {
600    fn rows(&self, r: Rows) -> usize {
601        self.counts[r as usize]
602    }
603
604    fn w(&self, i: usize) -> &[f32] {
605        &self.w[i].data
606    }
607
608    /// The live rows of `l`.
609    ///
610    /// # Safety
611    ///
612    /// Nobody may write `l` while the slice lives.
613    unsafe fn get(&self, l: Loc) -> &[f32] {
614        // SAFETY: in the arena by the layout, and by the caller.
615        unsafe { self.arena.slice(l.off, self.rows(l.rows) * l.width) }
616    }
617
618    /// Runs `f(first row, rows of out)` over the live rows of `out` in blocks of [`ROWS`].
619    ///
620    /// # Safety
621    ///
622    /// Nobody else may touch `out` meanwhile, and `f` must not reach `out` any other way.
623    unsafe fn rows_of(&self, out: Loc, f: &(dyn Fn(usize, &mut [f32]) + Sync)) {
624        let n = self.rows(out.rows);
625        let arena = self.arena;
626        self.pool.run(n.div_ceil(ROWS), &|task, _| {
627            let r0 = task * ROWS;
628            let r1 = (r0 + ROWS).min(n);
629            // SAFETY: rows r0..r1 of out belong to this task alone.
630            let rows = unsafe { arena.slice_mut(out.off + r0 * out.width, (r1 - r0) * out.width) };
631            f(r0, rows);
632        });
633    }
634
635    fn step(&self, step: &Step) {
636        // The layout gives the inputs and the output of a step disjoint arena ranges unless the
637        // step is in place, and an in place step takes one slice only. That is the argument behind
638        // every SAFETY comment below.
639        match *step {
640            Step::Embed { table, out } => {
641                let (tab, ids, d) = (self.w(table), self.batch.ids, out.width);
642                // SAFETY: the layout keeps the inputs and the output of a step apart.
643                unsafe {
644                    self.rows_of(out, &|r0, rows| {
645                        for (i, row) in rows.chunks_exact_mut(d).enumerate() {
646                            let id = ids[r0 + i] as usize;
647                            row.copy_from_slice(&tab[id * d..(id + 1) * d]);
648                        }
649                    });
650                }
651            }
652            Step::LayerNorm { x, w, b, eps, out } => {
653                // SAFETY: the layout keeps the inputs and the output of a step apart.
654                let x = unsafe { self.get(x) };
655                let (nw, nb, d) = (self.w(w), b.map(|b| self.w(b)), out.width);
656                // SAFETY: the layout keeps the inputs and the output of a step apart.
657                unsafe {
658                    self.rows_of(out, &|r0, rows| {
659                        layer_norm(&x[r0 * d..r0 * d + rows.len()], d, nw, nb, eps, rows);
660                    });
661                }
662            }
663            Step::Gemm { a, w, b, ep, out } => {
664                let rows = self.rows(a.rows);
665                // SAFETY: the layout keeps the inputs and the output of a step apart.
666                let x = unsafe { self.get(a) };
667                // SAFETY: the layout keeps the inputs and the output of a step apart.
668                let y = unsafe { self.arena.slice_mut(out.off, rows * out.width) };
669                let g = Gemm {
670                    x,
671                    m: rows,
672                    k: a.width,
673                    w: &self.w[w].packed,
674                    n: out.width,
675                    b: b.map(|b| self.w(b)),
676                    ep,
677                };
678                let scratch = self.scratch;
679                g.run(y, self.pool.threads(), |n, f| {
680                    self.pool.run(n, &|i, worker| {
681                        // SAFETY: the scratch is this worker's, and lowering reserved enough of it
682                        // for every GEMM in the plan, so this does not allocate.
683                        let s = unsafe { scratch.get(worker) };
684                        s.resize(gemm::scratch_len(g.k, g.n), 0.0);
685                        f(i, s);
686                    });
687                });
688            }
689            Step::Gemm8 { a, w, b, ep, out } => {
690                let rows = self.rows(a.rows);
691                // SAFETY: the layout keeps the inputs and the output of a step apart.
692                let x = unsafe { self.get(a) };
693                // SAFETY: the layout keeps the inputs and the output of a step apart.
694                let y = unsafe { self.arena.slice_mut(out.off, rows * out.width) };
695                let q = self.w[w]
696                    .quant
697                    .as_ref()
698                    .unwrap_or_else(|| unreachable!("checked when lowered"));
699                let g = QGemm { x, m: rows, w: q, b: b.map(|b| self.w(b)), ep };
700                let scratch = self.scratch;
701                g.run(y, self.pool.threads(), |n, f| {
702                    self.pool.run(n, &|i, worker| {
703                        // SAFETY: the scratch is this worker's, and lowering reserved enough of it
704                        // for every GEMM in the plan, so this does not allocate.
705                        let s = unsafe { scratch.get(worker) };
706                        s.resize(qgemm::scratch_len(q.k), 0.0);
707                        f(i, s);
708                    });
709                });
710            }
711            Step::Rope { qkv, rope } => {
712                let (rope, d, cu, seq) = (&self.ropes[rope], qkv.width / 3, self.cu, self.row_seq);
713                // SAFETY: the layout keeps the inputs and the output of a step apart.
714                unsafe {
715                    self.rows_of(qkv, &|r0, rows| {
716                        for (i, row) in rows.chunks_exact_mut(qkv.width).enumerate() {
717                            let r = r0 + i;
718                            let pos = r - cu[seq[r] as usize];
719                            for head in row[..2 * d].as_chunks_mut::<HEAD>().0 {
720                                rope.apply(head, pos);
721                            }
722                        }
723                    });
724                }
725            }
726            Step::Attention { qkv, window, out } => {
727                let heads = out.width / HEAD;
728                // SAFETY: the layout keeps the inputs and the output of a step apart.
729                let x = unsafe { self.get(qkv) };
730                // SAFETY: the layout keeps the inputs and the output of a step apart.
731                let y = unsafe { self.arena.slice_mut(out.off, self.rows(out.rows) * out.width) };
732                let shared = Shared::new(y);
733                let (blocks, cu, scratch) = (self.blocks, self.cu, self.scratch);
734                self.pool.run(blocks.len() * heads, &|task, worker| {
735                    let (s, q0) = blocks[task / heads];
736                    let (s, q0, h) = (s as usize, q0 as usize, task % heads);
737                    // SAFETY: the scratch is this worker's, and this task owns rows q0 to q0 + QB
738                    // of head h.
739                    unsafe {
740                        let p = scratch.get(worker);
741                        attention::block(x, heads, (cu[s], cu[s + 1]), q0, h, window, p, &shared);
742                    }
743                });
744            }
745            Step::GeGlu { x, out } => {
746                // SAFETY: the layout keeps the inputs and the output of a step apart.
747                let (x, d) = (unsafe { self.get(x) }, out.width);
748                // SAFETY: the layout keeps the inputs and the output of a step apart.
749                unsafe {
750                    self.rows_of(out, &|r0, rows| {
751                        geglu(&x[2 * r0 * d..2 * (r0 * d + rows.len())], d, rows);
752                    });
753                }
754            }
755            Step::AddType { h, table } => {
756                let (tab, d, seq, qt) = (self.w(table), h.width, self.row_seq, self.batch.qtype);
757                // SAFETY: the layout keeps the inputs and the output of a step apart.
758                unsafe {
759                    self.rows_of(h, &|r0, rows| {
760                        for (i, row) in rows.chunks_exact_mut(d).enumerate() {
761                            let q = usize::from(qt[seq[r0 + i] as usize]);
762                            row.iter_mut().zip(&tab[q * d..(q + 1) * d]).for_each(|(a, b)| *a += b);
763                        }
764                    });
765                }
766            }
767            Step::Gather { h, out } => {
768                // SAFETY: the layout keeps the inputs and the output of a step apart.
769                let (x, d) = (unsafe { self.get(h) }, h.width);
770                // SAFETY: the layout keeps the inputs and the output of a step apart.
771                let y = unsafe { self.arena.slice_mut(out.off, self.rows(out.rows) * d) };
772                let mut at = 0;
773                for s in 0..self.cu.len() - 1 {
774                    for &p in &self.batch.markers[self.mcu[s]..self.mcu[s + 1]] {
775                        let r = self.cu[s] + p as usize;
776                        y[at * d..(at + 1) * d].copy_from_slice(&x[r * d..(r + 1) * d]);
777                        at += 1;
778                    }
779                }
780            }
781            Step::ActFeatures { h, logits, out } => {
782                // SAFETY: the layout keeps the inputs and the output of a step apart.
783                let (x, d) = (unsafe { self.get(h) }, h.width);
784                // SAFETY: the layout keeps the inputs and the output of a step apart.
785                let l = unsafe { self.get(logits) };
786                // SAFETY: the layout keeps the inputs and the output of a step apart.
787                let y = unsafe { self.arena.slice_mut(out.off, self.rows(out.rows) * out.width) };
788                for (s, row) in y.chunks_exact_mut(out.width).enumerate() {
789                    let (lo, hi) = (self.cu[s], self.cu[s + 1]);
790                    if hi > lo {
791                        row[..d].copy_from_slice(&x[lo * d..(lo + 1) * d]);
792                    } else {
793                        row[..d].fill(0.0);
794                    }
795                    row[d..].copy_from_slice(&act_features(&l[self.mcu[s]..self.mcu[s + 1]]));
796                }
797            }
798        }
799    }
800}
801
802/// `[top1, top1 - top2, entropy / ln k, k / 255]` over the softmax of the logits, with `k` at
803/// least 2, the same arithmetic as [`crate::compat::act_features`] without its buffers.
804fn act_features(logits: &[f32]) -> [f32; 4] {
805    let kf = logits.len().max(2) as f32;
806    if logits.is_empty() {
807        return [0.0, 0.0, 0.0, kf / 255.0];
808    }
809    let mx = logits.iter().copied().fold(f32::NEG_INFINITY, f32::max);
810    let sum: f32 = logits.iter().map(|&l| (l - mx).exp()).sum();
811    let (mut top1, mut top2, mut ent) = (f32::NEG_INFINITY, 0f32, 0f32);
812    let mut first = true;
813    for &l in logits {
814        let q = (l - mx).exp() / sum;
815        ent += q * q.max(1e-9).ln();
816        if q > top1 || first {
817            if !first {
818                top2 = top1;
819            }
820            top1 = q;
821            first = false;
822        } else if q > top2 {
823            top2 = q;
824        }
825    }
826    let ent = -ent / kf.ln();
827    [top1, top1 - top2, ent, kf / 255.0]
828}
829
830#[cfg(test)]
831mod tests {
832    use super::*;
833
834    #[test]
835    fn act_features_match_the_reference() {
836        let cases: [&[f32]; 6] =
837            [&[], &[0.3], &[1.0, 1.0], &[2.0, -1.0, 2.0], &[5.0, 0.1, -3.0, 4.9], &[-1e3, 0.0]];
838        for l in cases {
839            assert_eq!(
840                act_features(l).map(f32::to_bits),
841                crate::compat::act_features(l).map(f32::to_bits),
842                "{l:?}"
843            );
844        }
845    }
846}