#![forbid(unsafe_code)]
use std::cell::Cell;
use std::collections::{BinaryHeap, HashMap, HashSet};
use std::sync::{Arc, Mutex, PoisonError};
use nuvec_core::graph::{self, Fused, FusedRef, Primitive};
use nuvec_core::tree::{Tree, TreeDef};
use nuvec_core::{
Array, Complex64, DType, DeviceKind, Error, IntoResult, Result, Stream, bf16, default_stream,
f16, random,
};
fn dims(a: &Array) -> Vec<usize> {
a.shape().iter().map(|&d| d as usize).collect()
}
pub const VERSION: &str = env!("CARGO_PKG_VERSION");
const MAX_FUSED_STEPS: usize = 32;
const MAX_FUSED_ARRAYS: usize = 24;
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct Options {
pub cache_capacity: usize,
}
impl Default for Options {
fn default() -> Self {
Self { cache_capacity: 64 }
}
}
impl Options {
pub fn cache_capacity(mut self, n: usize) -> Self {
self.cache_capacity = n;
self
}
pub fn compile<'f, I, O, F, R>(self, f: F) -> impl Fn(&I) -> Result<O> + 'f
where
I: Tree,
O: Tree,
F: Fn(&I) -> R + 'f,
R: IntoResult<Ok = O>,
{
let cache: Mutex<Cache<(Tape, O)>> = Mutex::new(Cache::new(self.cache_capacity));
move |inputs: &I| {
let leaves = inputs.leaves();
if leaves.iter().any(|a| graph::is_tracer(a) || a.is_failed()) {
return f(inputs).into_result();
}
let key = Key::new(vec![inputs.def()], &leaves);
let cached = lock(&cache).get(&key);
let entry = match cached {
Some(entry) => entry,
None => {
let mut out = None;
let mut body = |xs: &[Array]| -> Result<Vec<Array>> {
let o = f(&inputs.unflatten(xs.iter().cloned())?).into_result()?;
let leaves = o.leaves();
out = Some(o.map(|_, a| light(a)));
Ok(leaves)
};
let tape = Tape::trace(&mut body, &leaves, key.stream)?;
let out =
out.ok_or_else(|| Error::InvalidArgument("compile: no output".into()))?;
let entry = Arc::new((tape, out));
if !entry.0.captures_tracer() {
lock(&cache).insert(key, entry.clone());
}
entry
}
};
let (tape, template) = &*entry;
template.unflatten(tape.replay(&leaves)?)
}
}
pub fn compile_mut<'f, S, I, O, F, R>(self, f: F) -> impl Fn(&mut S, &I) -> Result<O> + 'f
where
S: Tree,
I: Tree,
O: Tree,
F: Fn(&mut S, &I) -> R + 'f,
R: IntoResult<Ok = O>,
{
let cache: Mutex<Cache<MutTape<S, O>>> = Mutex::new(Cache::new(self.cache_capacity));
let changed_last = Cell::new(false);
move |state: &mut S, inputs: &I| {
let state_leaves = state.leaves();
let input_leaves = inputs.leaves();
let direct = state_leaves
.iter()
.chain(&input_leaves)
.any(|a| graph::is_tracer(a) || a.is_failed());
if direct {
return f(state, inputs).into_result();
}
let key0 = random::sequence_key();
let pre_def = state.def();
let mut all = Vec::with_capacity(1 + state_leaves.len() + input_leaves.len());
all.push(key0.clone());
all.extend(state_leaves.iter().cloned());
all.extend(input_leaves.iter().cloned());
let key = Key::new(vec![pre_def.clone(), inputs.def()], &all);
let cached = lock(&cache).get(&key);
let entry = match cached {
Some(entry) => entry,
None => {
let entry = Arc::new(MutTape::trace(&f, state, inputs, &all, key.stream)?);
if !entry.tape.captures_tracer() {
lock(&cache).insert(key, entry.clone());
}
entry
}
};
let changed = entry.post_def != pre_def;
if changed && changed_last.get() {
return Err(Error::InvalidArgument(
"compile_mut: state structure keeps changing; call opt.init(&model) before \
the loop, or make changing host fields arrays (or `static_unchecked`)"
.into(),
));
}
let mut outputs = entry.tape.replay(&all)?;
let next_key = outputs.pop().expect("the key is the last output");
let state_out = outputs.split_off(entry.outputs);
let out = entry.out.unflatten(outputs)?;
let next = entry.post.unflatten(state_out)?;
changed_last.set(changed);
*state = next;
random::set_sequence_key(next_key);
Ok(out)
}
}
}
pub fn compile<'f, I, O, F, R>(f: F) -> impl Fn(&I) -> Result<O> + 'f
where
I: Tree,
O: Tree,
F: Fn(&I) -> R + 'f,
R: IntoResult<Ok = O>,
{
Options::default().compile(f)
}
pub fn compile_mut<'f, S, I, O, F, R>(f: F) -> impl Fn(&mut S, &I) -> Result<O> + 'f
where
S: Tree,
I: Tree,
O: Tree,
F: Fn(&mut S, &I) -> R + 'f,
R: IntoResult<Ok = O>,
{
Options::default().compile_mut(f)
}
pub fn compile_flat<F>(f: F) -> impl Fn(&[Array]) -> Result<Vec<Array>>
where
F: Fn(&[Array]) -> Result<Vec<Array>>,
{
let cache: Mutex<Cache<Tape>> = Mutex::new(Cache::new(Options::default().cache_capacity));
move |inputs| {
if inputs.iter().any(|a| graph::is_tracer(a) || a.is_failed()) {
return f(inputs);
}
let key = Key::new(Vec::new(), inputs);
let cached = lock(&cache).get(&key);
let tape = match cached {
Some(tape) => tape,
None => {
let tape = Arc::new(Tape::trace(&mut |xs: &[Array]| f(xs), inputs, key.stream)?);
if !tape.captures_tracer() {
lock(&cache).insert(key, tape.clone());
}
tape
}
};
tape.replay(inputs)
}
}
fn lock<T>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
m.lock().unwrap_or_else(PoisonError::into_inner)
}
fn light(a: &Array) -> Array {
graph::placeholder(a.dtype(), a.shape()).unwrap_or_else(|_| a.clone())
}
struct MutTape<S, O> {
tape: Tape,
outputs: usize,
out: O,
post: S,
post_def: TreeDef,
}
impl<S: Tree, O: Tree> MutTape<S, O> {
fn trace<I: Tree, R: IntoResult<Ok = O>>(
f: &dyn Fn(&mut S, &I) -> R,
state: &S,
inputs: &I,
all: &[Array],
stream: Stream,
) -> Result<Self> {
let ns = state.len();
let key0 = all[0].clone();
let mut traced: Option<(O, S)> = None;
let mut body = |xs: &[Array]| -> Result<Vec<Array>> {
random::set_sequence_key(xs[0].clone());
let mut s = state.unflatten(xs[1..1 + ns].iter().cloned())?;
let i = inputs.unflatten(xs[1 + ns..].iter().cloned())?;
let out = f(&mut s, &i).into_result();
let key = random::sequence_key();
random::set_sequence_key(key0.clone());
let out = out?;
let mut leaves = out.leaves();
leaves.extend(s.leaves());
leaves.push(key);
traced = Some((out, s));
Ok(leaves)
};
let result = Tape::trace(&mut body, all, stream);
random::set_sequence_key(key0.clone());
let tape = result?;
let (out, post) =
traced.ok_or_else(|| Error::InvalidArgument("compile_mut: no output".into()))?;
Ok(Self {
tape,
outputs: out.len(),
out: out.map(|_, a| light(a)),
post_def: post.def(),
post: post.map(|_, a| light(a)),
})
}
}
struct Cache<T> {
tapes: HashMap<Key, (Arc<T>, u64)>,
clock: u64,
capacity: usize,
}
impl<T> Cache<T> {
fn new(capacity: usize) -> Self {
Self {
tapes: HashMap::new(),
clock: 0,
capacity,
}
}
fn get(&mut self, key: &Key) -> Option<Arc<T>> {
self.clock += 1;
let (tape, used) = self.tapes.get_mut(key)?;
*used = self.clock;
Some(tape.clone())
}
fn insert(&mut self, key: Key, tape: Arc<T>) {
if self.capacity == 0 {
return;
}
self.clock += 1;
if self.tapes.len() >= self.capacity
&& !self.tapes.contains_key(&key)
&& let Some(oldest) = self
.tapes
.iter()
.min_by_key(|(_, (_, used))| *used)
.map(|(k, _)| k.clone())
{
self.tapes.remove(&oldest);
}
self.tapes.insert(key, (tape, self.clock));
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct Key {
stream: Stream,
defs: Vec<TreeDef>,
inputs: Vec<(DType, Vec<usize>)>,
}
impl Key {
fn new(defs: Vec<TreeDef>, inputs: &[Array]) -> Self {
Self {
stream: default_stream(),
defs,
inputs: inputs.iter().map(|a| (a.dtype(), dims(a))).collect(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum Ref {
Input(usize),
Const(usize),
Step(usize, usize),
}
#[derive(Debug, Clone)]
struct Step {
primitive: Primitive,
inputs: Vec<Ref>,
outputs: Vec<(DType, Vec<usize>)>,
}
#[derive(Debug)]
struct Tape {
consts: Vec<Array>,
steps: Vec<Step>,
outputs: Vec<Ref>,
}
impl Tape {
fn trace(
f: &mut dyn FnMut(&[Array]) -> Result<Vec<Array>>,
inputs: &[Array],
stream: Stream,
) -> Result<Self> {
let placeholders = inputs
.iter()
.map(|a| graph::placeholder(a.dtype(), a.shape()))
.collect::<Result<Vec<_>>>()?;
let outputs = f(&placeholders)?;
let mut tape = Builder::new(&placeholders).build(&outputs);
if stream.device.kind == DeviceKind::Gpu {
tape.fuse(inputs);
}
Ok(tape)
}
fn captures_tracer(&self) -> bool {
self.consts.iter().any(graph::is_tracer)
}
fn replay(&self, inputs: &[Array]) -> Result<Vec<Array>> {
let mut values: Vec<Vec<Array>> = Vec::with_capacity(self.steps.len());
let get = |r: &Ref, values: &[Vec<Array>]| match *r {
Ref::Input(i) => inputs[i].clone(),
Ref::Const(c) => self.consts[c].clone(),
Ref::Step(s, o) => values[s][o].clone(),
};
for step in &self.steps {
let args = step.inputs.iter().map(|r| get(r, &values)).collect();
values.push(graph::from_primitive(step.primitive.clone(), args)?);
}
Ok(self.outputs.iter().map(|r| get(r, &values)).collect())
}
fn dtype(&self, r: Ref, inputs: &[Array]) -> DType {
match r {
Ref::Input(i) => inputs[i].dtype(),
Ref::Const(c) => self.consts[c].dtype(),
Ref::Step(s, o) => self.steps[s].outputs[o].0,
}
}
fn fuse(&mut self, inputs: &[Array]) {
let n = self.steps.len();
let fusable: Vec<bool> = self
.steps
.iter()
.map(|s| {
let dtypes: Vec<DType> = s.inputs.iter().map(|&r| self.dtype(r, inputs)).collect();
s.outputs.len() == 1 && Fused::supports(&s.primitive, &dtypes)
})
.collect();
let mut consumers: Vec<Vec<usize>> = vec![Vec::new(); n];
for (k, s) in self.steps.iter().enumerate() {
for r in &s.inputs {
if let Ref::Step(j, _) = *r {
consumers[j].push(k);
}
}
}
let mut is_output = vec![false; n];
for r in &self.outputs {
if let Ref::Step(j, _) = *r {
is_output[j] = true;
}
}
let step_inputs = |k: usize| {
self.steps[k].inputs.iter().filter_map(|r| match *r {
Ref::Step(j, _) => Some(j),
_ => None,
})
};
let mut group: Vec<Option<usize>> = vec![None; n];
let mut groups: HashMap<usize, (Vec<usize>, Vec<Ref>)> = HashMap::new();
for root in (0..n).rev() {
if group[root].is_some() || !fusable[root] {
continue;
}
let mut members = HashSet::from([root]);
let mut external: Vec<Ref> = Vec::new();
let mut outputs = 1;
let add_inputs = |k: usize, external: &mut Vec<Ref>| {
for &r in &self.steps[k].inputs {
if !external.contains(&r) {
external.push(r);
}
}
};
add_inputs(root, &mut external);
let mut candidates: BinaryHeap<usize> = step_inputs(root).collect();
let mut seen = HashSet::new();
while let Some(j) = candidates.pop() {
if !seen.insert(j)
|| group[j].is_some()
|| !fusable[j]
|| members.len() >= MAX_FUSED_STEPS
|| !consumers[j].iter().all(|c| members.contains(c))
{
continue;
}
let new = self.steps[j]
.inputs
.iter()
.filter(|r| !external.contains(r))
.collect::<HashSet<_>>()
.len();
let arrays = external.len() - 1 + new + outputs + usize::from(is_output[j]);
if arrays > MAX_FUSED_ARRAYS {
continue;
}
members.insert(j);
outputs += usize::from(is_output[j]);
external.retain(|&r| r != Ref::Step(j, 0));
add_inputs(j, &mut external);
candidates.extend(step_inputs(j));
}
if members.len() < 2 || external.len() + outputs > MAX_FUSED_ARRAYS {
continue;
}
let mut members: Vec<usize> = members.into_iter().collect();
members.sort_unstable();
for &m in &members {
group[m] = Some(root);
}
groups.insert(root, (members, external));
}
if groups.is_empty() {
return;
}
let mut index: Vec<Option<(usize, usize)>> = vec![None; n];
let remap = |r: Ref, index: &[Option<(usize, usize)>]| match r {
Ref::Step(j, o) => {
let (k, offset) = index[j].expect("defined before use");
Ref::Step(k, offset + o)
}
r => r,
};
let mut steps = Vec::with_capacity(n);
for k in 0..n {
let step = match (group[k], groups.get(&k)) {
(Some(_), None) => continue, (Some(_), Some((members, external))) => {
let position: HashMap<usize, usize> =
members.iter().enumerate().map(|(p, &m)| (m, p)).collect();
let graph = members
.iter()
.map(|&m| {
let args = self.steps[m]
.inputs
.iter()
.map(|r| match *r {
Ref::Step(j, _) if position.contains_key(&j) => {
FusedRef::Step(position[&j])
}
r => FusedRef::Input(
external.iter().position(|&e| e == r).expect("external"),
),
})
.collect();
(self.steps[m].primitive.clone(), args)
})
.collect();
let outs: Vec<usize> = members
.iter()
.copied()
.filter(|&m| m != k && is_output[m])
.chain([k])
.collect();
for (o, &m) in outs.iter().enumerate() {
index[m] = Some((steps.len(), o));
}
let dtypes: Vec<DType> =
external.iter().map(|&r| self.dtype(r, inputs)).collect();
let positions: Vec<usize> = outs.iter().map(|m| position[m]).collect();
let Ok(f) = Fused::new(&dtypes, graph, &positions) else {
unreachable!("every member is fusable")
};
steps.push(Step {
primitive: Primitive::Compiled(f),
inputs: external.iter().map(|&r| remap(r, &index)).collect(),
outputs: outs
.iter()
.map(|&m| self.steps[m].outputs[0].clone())
.collect(),
});
continue;
}
(None, _) => {
let s = &self.steps[k];
Step {
primitive: s.primitive.clone(),
inputs: s.inputs.iter().map(|&r| remap(r, &index)).collect(),
outputs: s.outputs.clone(),
}
}
};
index[k] = Some((steps.len(), 0));
steps.push(step);
}
self.outputs = self.outputs.iter().map(|&r| remap(r, &index)).collect();
self.steps = steps;
}
}
struct Builder {
refs: HashMap<(usize, usize), Ref>,
consts: Vec<Array>,
scalars: HashMap<(DType, Vec<usize>, u64), usize>,
steps: Vec<Step>,
seen: HashMap<(Primitive, Vec<Ref>), usize>,
}
impl Builder {
fn new(placeholders: &[Array]) -> Self {
let refs = placeholders
.iter()
.enumerate()
.map(|(i, p)| ((graph::id(p), graph::output_index(p)), Ref::Input(i)))
.collect();
Self {
refs,
consts: Vec::new(),
scalars: HashMap::new(),
steps: Vec::new(),
seen: HashMap::new(),
}
}
fn key(a: &Array) -> (usize, usize) {
(graph::id(a), graph::output_index(a))
}
fn build(mut self, outputs: &[Array]) -> Tape {
let mut stack: Vec<(Array, bool)> = outputs.iter().map(|a| (a.clone(), false)).collect();
while let Some((a, expanded)) = stack.pop() {
if self.refs.contains_key(&Self::key(&a)) {
continue;
}
let inputs = graph::inputs(&a);
if expanded || inputs.is_empty() {
self.resolve(&a, &inputs);
continue;
}
stack.push((a, true));
stack.extend(
inputs
.into_iter()
.filter(|i| !self.refs.contains_key(&Self::key(i)))
.map(|i| (i, false)),
);
}
let outputs = outputs.iter().map(|a| self.refs[&Self::key(a)]).collect();
Tape {
consts: self.consts,
steps: self.steps,
outputs,
}
}
fn resolve(&mut self, a: &Array, inputs: &[Array]) {
let siblings = graph::siblings(a);
let args: Vec<Ref> = inputs.iter().map(|i| self.refs[&Self::key(i)]).collect();
let primitive = graph::primitive(a);
let Some(primitive) =
primitive.filter(|_| !args.iter().all(|r| matches!(r, Ref::Const(_))))
else {
for s in &siblings {
let c = self.constant(s);
self.refs.insert(Self::key(s), Ref::Const(c));
}
return;
};
if siblings.len() == 1 && is_identity(&primitive, &inputs[0], a) {
self.refs.insert(Self::key(a), args[0]);
return;
}
let k = match self.seen.get(&(primitive.clone(), args.clone())) {
Some(&k) => k,
None => {
self.steps.push(Step {
primitive: primitive.clone(),
inputs: args.clone(),
outputs: siblings.iter().map(|s| (s.dtype(), dims(s))).collect(),
});
let k = self.steps.len() - 1;
self.seen.insert((primitive, args), k);
k
}
};
for (o, s) in siblings.iter().enumerate() {
self.refs.insert(Self::key(s), Ref::Step(k, o));
}
}
fn constant(&mut self, a: &Array) -> usize {
if graph::is_tracer(a) {
self.consts.push(a.clone());
return self.consts.len() - 1;
}
let value = scalar_bits(a).or_else(|| {
match (graph::primitive(a), graph::inputs(a).as_slice()) {
(Some(Primitive::Broadcast(_)), [x]) => scalar_bits(x),
_ => None,
}
});
let key = value.map(|v| (a.dtype(), dims(a), v));
if let Some(&c) = key.as_ref().and_then(|k| self.scalars.get(k)) {
return c;
}
self.consts.push(a.clone());
let c = self.consts.len() - 1;
if let Some(key) = key {
self.scalars.insert(key, c);
}
c
}
}
fn is_identity(primitive: &Primitive, input: &Array, output: &Array) -> bool {
let same = input.shape() == output.shape() && graph::strides(input) == graph::strides(output);
match primitive {
Primitive::Broadcast(_) | Primitive::Reshape(_) | Primitive::StopGradient => same,
Primitive::Transpose(axes) => axes.iter().enumerate().all(|(i, &a)| i == a),
Primitive::AsType(d) => *d == input.dtype(),
_ => false,
}
}
fn scalar_bits(a: &Array) -> Option<u64> {
if a.size() != 1 || !a.is_evaluated() {
return None;
}
Some(match a.dtype() {
DType::Bool => u64::from(a.item::<bool>().ok()?),
DType::U8 => u64::from(a.item::<u8>().ok()?),
DType::U16 => u64::from(a.item::<u16>().ok()?),
DType::U32 => u64::from(a.item::<u32>().ok()?),
DType::U64 => a.item::<u64>().ok()?,
DType::I8 => a.item::<i8>().ok()? as u64,
DType::I16 => a.item::<i16>().ok()? as u64,
DType::I32 => a.item::<i32>().ok()? as u64,
DType::I64 => a.item::<i64>().ok()? as u64,
DType::F16 => u64::from(a.item::<f16>().ok()?.to_bits()),
DType::BF16 => u64::from(a.item::<bf16>().ok()?.to_bits()),
DType::F32 => u64::from(a.item::<f32>().ok()?.to_bits()),
DType::F64 => a.item::<f64>().ok()?.to_bits(),
DType::Complex64 => {
let z = a.item::<Complex64>().ok()?;
u64::from(z.re.to_bits()) << 32 | u64::from(z.im.to_bits())
}
_ => return None,
})
}
#[cfg(test)]
mod tests {
use nuvec_core::{Device, array, set_default_stream};
use super::*;
fn trace(f: &dyn Fn(&[Array]) -> Result<Vec<Array>>, inputs: &[Array], device: Device) -> Tape {
Tape::trace(
&mut |xs: &[Array]| f(xs),
inputs,
Stream::default_for(device),
)
.unwrap()
}
fn names(tape: &Tape) -> Vec<&'static str> {
tape.steps.iter().map(|s| s.primitive.name()).collect()
}
#[test]
fn version_is_set() {
assert!(!VERSION.is_empty());
}
#[test]
fn simplifies_identities_constants_and_common_subexpressions() {
set_default_stream(Stream::default_for(Device::cpu()));
let x = array![1.0, 2.0];
let f = |xs: &[Array]| {
let x = xs[0].astype(DType::F32).reshape([2]);
let a = x.multiply(2.0); let b = x.multiply(2.0); Ok(vec![a.add(&b), x.exp(), x.exp()])
};
let tape = trace(&f, std::slice::from_ref(&x), Device::cpu());
let mut ops = names(&tape);
ops.sort_unstable();
assert_eq!(ops, ["add", "exp", "multiply"]);
assert_eq!(tape.outputs[1], tape.outputs[2]);
let outs = tape.replay(&[x]).unwrap();
assert_eq!(outs[0].to_vec::<f32>().unwrap(), [4.0, 8.0]);
}
#[test]
fn fuses_element_wise_chains_on_the_gpu_only() {
set_default_stream(Stream::default_for(Device::cpu()));
let x = array![1.0, 2.0];
let w = array![[1.0, 0.0], [0.0, 1.0]];
let f = |xs: &[Array]| {
let h = xs[0].reshape([1, 2]).matmul(&w);
let a = h.multiply(2.0).exp();
let b = a.multiply(&a).add(1.0);
Ok(vec![a, b.sigmoid()])
};
assert_eq!(
names(&trace(&f, std::slice::from_ref(&x), Device::cpu())),
[
"reshape", "matmul", "multiply", "exp", "multiply", "add", "sigmoid"
]
);
let tape = trace(&f, &[x], Device::gpu(0));
assert_eq!(names(&tape), ["reshape", "matmul", "compiled"]);
let Primitive::Compiled(fused) = &tape.steps[2].primitive else {
unreachable!()
};
assert_eq!((fused.len(), fused.dtypes().len()), (5, 2));
assert_eq!(tape.outputs, [Ref::Step(2, 0), Ref::Step(2, 1)]);
}
#[test]
fn copies_and_dependencies_are_kept() {
set_default_stream(Stream::default_for(Device::cpu()));
let x = array![1.0, 2.0];
let y = array![3.0, 4.0];
let f = |xs: &[Array]| {
let copy = graph::from_primitive(Primitive::Copy, vec![xs[0].clone()])?.remove(0);
let ordered = graph::from_primitive(
Primitive::Depends { n_deps: 1 },
vec![xs[0].clone(), xs[1].clone()],
)?
.remove(0);
Ok(vec![copy.add(ordered.astype(DType::F32))])
};
let tape = trace(&f, &[x.clone(), y.clone()], Device::cpu());
let mut ops = names(&tape);
ops.sort_unstable();
assert_eq!(ops, ["add", "copy", "depends"]);
let out = tape.replay(&[x, y]).unwrap().remove(0);
assert_eq!(out.to_vec::<f32>().unwrap(), [2.0, 4.0]);
}
#[test]
fn a_shared_value_used_outside_a_group_stays_outside() {
let x = array![1.0, 2.0];
let f = |xs: &[Array]| {
let t = xs[0].exp();
let s = t.sum_keepdims(..);
Ok(vec![t.multiply(3.0).add(&s)])
};
let tape = trace(&f, &[x], Device::gpu(0));
assert_eq!(names(&tape), ["exp", "sum", "broadcast_to", "compiled"]);
}
}