use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use crate::diagnostics::Diagnostic;
pub const DEFAULT_PLAN_VERSION: &str = "0.1";
pub use daedalus_core::compute::ComputeAffinity;
#[allow(unused_imports)]
pub use daedalus_core::sync::{SyncGroup, SyncPolicy};
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct StableHash(pub u64);
impl StableHash {
pub fn from_bytes(bytes: &[u8]) -> Self {
const FNV_OFFSET: u64 = 0xcbf29ce484222325;
const FNV_PRIME: u64 = 0x100000001b3;
let mut hash = FNV_OFFSET;
for b in bytes {
hash ^= *b as u64;
hash = hash.wrapping_mul(FNV_PRIME);
}
StableHash(hash)
}
}
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct NodeRef(pub usize);
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct PortRef {
pub node: NodeRef,
pub port: String,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Edge {
pub from: PortRef,
pub to: PortRef,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub metadata: BTreeMap<String, daedalus_data::model::Value>,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct NodeInstance {
pub id: daedalus_registry::ids::NodeId,
pub bundle: Option<String>,
pub label: Option<String>,
pub inputs: Vec<String>,
pub outputs: Vec<String>,
#[serde(default)]
pub compute: ComputeAffinity,
#[serde(default)]
pub const_inputs: Vec<(String, daedalus_data::model::Value)>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub sync_groups: Vec<SyncGroup>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub metadata: BTreeMap<String, daedalus_data::model::Value>,
}
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct Graph {
pub nodes: Vec<NodeInstance>,
pub edges: Vec<Edge>,
pub metadata: BTreeMap<String, String>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub metadata_values: BTreeMap<String, daedalus_data::model::Value>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct GpuSegment {
pub buffer_id: usize,
pub nodes: Vec<NodeRef>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct EdgeBufferInfo {
pub edge_index: usize,
pub gpu_fast_path: bool,
pub buffer_id: Option<usize>,
}
impl Graph {
pub fn gpu_buffers(&self) -> (Vec<GpuSegment>, Vec<EdgeBufferInfo>) {
#[derive(Clone)]
struct Dsu {
parent: Vec<usize>,
}
impl Dsu {
fn new(n: usize) -> Self {
Self {
parent: (0..n).collect(),
}
}
fn find(&mut self, x: usize) -> usize {
if self.parent[x] != x {
let p = self.parent[x];
self.parent[x] = self.find(p);
}
self.parent[x]
}
fn union(&mut self, a: usize, b: usize) {
let ra = self.find(a);
let rb = self.find(b);
if ra != rb {
self.parent[rb] = ra;
}
}
}
let mut dsu = Dsu::new(self.nodes.len());
for e in &self.edges {
let from = &self.nodes[e.from.node.0];
let to = &self.nodes[e.to.node.0];
let gpu_gpu = matches!(
from.compute,
ComputeAffinity::GpuPreferred | ComputeAffinity::GpuRequired
) && matches!(
to.compute,
ComputeAffinity::GpuPreferred | ComputeAffinity::GpuRequired
);
if gpu_gpu {
dsu.union(e.from.node.0, e.to.node.0);
}
}
let mut root_to_buf = BTreeMap::new();
let mut node_buf: Vec<Option<usize>> = vec![None; self.nodes.len()];
for (idx, n) in self.nodes.iter().enumerate() {
if matches!(
n.compute,
ComputeAffinity::GpuPreferred | ComputeAffinity::GpuRequired
) {
let root = dsu.find(idx);
let buf_id = match root_to_buf.get(&root) {
Some(id) => *id,
None => {
let next = root_to_buf.len();
root_to_buf.insert(root, next);
next
}
};
node_buf[idx] = Some(buf_id);
}
}
let mut segments = Vec::new();
for (root, buf_id) in root_to_buf {
let mut members: Vec<NodeRef> = self
.nodes
.iter()
.enumerate()
.filter(|(i, _)| dsu.find(*i) == root)
.map(|(i, _)| NodeRef(i))
.collect();
members.sort_by_key(|nr| nr.0);
segments.push(GpuSegment {
buffer_id: buf_id,
nodes: members,
});
}
segments.sort_by_key(|s| s.buffer_id);
let mut edges = Vec::new();
for (i, e) in self.edges.iter().enumerate() {
let from = &self.nodes[e.from.node.0];
let to = &self.nodes[e.to.node.0];
let gpu_gpu = matches!(
from.compute,
ComputeAffinity::GpuPreferred | ComputeAffinity::GpuRequired
) && matches!(
to.compute,
ComputeAffinity::GpuPreferred | ComputeAffinity::GpuRequired
);
let buffer_id = if gpu_gpu {
node_buf[e.from.node.0]
} else {
None
};
edges.push(EdgeBufferInfo {
edge_index: i,
gpu_fast_path: gpu_gpu,
buffer_id,
});
}
(segments, edges)
}
}
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct ExecutionPlan {
pub version: String,
pub graph: Graph,
pub diagnostics: Vec<Diagnostic>,
pub hash: StableHash,
}
impl ExecutionPlan {
pub fn new(graph: Graph, diagnostics: Vec<Diagnostic>) -> Self {
let mut bytes = Vec::new();
bytes.extend_from_slice(&(graph.nodes.len() as u64).to_le_bytes());
bytes.extend_from_slice(&(graph.edges.len() as u64).to_le_bytes());
for n in &graph.nodes {
bytes.extend_from_slice(n.id.0.as_bytes());
bytes.push(match n.compute {
ComputeAffinity::CpuOnly => 0,
ComputeAffinity::GpuPreferred => 1,
ComputeAffinity::GpuRequired => 2,
});
}
for e in &graph.edges {
bytes.extend_from_slice(&(e.from.node.0 as u64).to_le_bytes());
bytes.extend_from_slice(&(e.to.node.0 as u64).to_le_bytes());
bytes.extend_from_slice(e.from.port.as_bytes());
bytes.extend_from_slice(e.to.port.as_bytes());
}
let hash = StableHash::from_bytes(&bytes);
Self {
version: DEFAULT_PLAN_VERSION.to_string(),
graph,
diagnostics,
hash,
}
}
}