uiharu_core/
graph.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Building a decode graph: sources of logic samples, decoder nodes, the connections between
4//! their typed ports and the outputs exposed to the caller. Connecting checks that the ports
5//! agree on the schema and that the input's requirements are met; building checks that every
6//! input is connected, that the graph has no cycle and that each node reads one time domain.
7
8use std::collections::{BTreeSet, VecDeque};
9use std::fmt;
10use std::marker::PhantomData;
11use std::sync::Arc;
12
13use crate::exec::Pipeline;
14use crate::node::{Node, NodeSpec};
15use crate::schema::{InputSpec, OutputSpec, Props, SchemaId};
16use crate::signal::{LogicBlock, MAX_CHANNELS};
17use crate::time::TimeBase;
18
19/// A source of logic samples in a graph.
20#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
21pub struct SourceId(pub(crate) usize);
22
23impl SourceId {
24    /// The source added `index`-th to its graph.
25    pub const fn from_index(index: usize) -> Self {
26        Self(index)
27    }
28
29    /// The source's position among the graph's sources.
30    pub const fn index(self) -> usize {
31        self.0
32    }
33}
34
35/// A node in a graph.
36#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
37pub struct NodeId(pub(crate) usize);
38
39impl NodeId {
40    /// The node added `index`-th to its graph.
41    pub const fn from_index(index: usize) -> Self {
42        Self(index)
43    }
44
45    /// The node's position among the graph's nodes.
46    pub const fn index(self) -> usize {
47        self.0
48    }
49}
50
51/// Who owns an output port.
52#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
53pub(crate) enum Owner {
54    Source(usize),
55    Node(usize),
56}
57
58/// An output port of a source or node.
59#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
60pub(crate) struct PortKey {
61    pub(crate) owner: Owner,
62    pub(crate) index: usize,
63}
64
65/// An output port carrying `T` events, as [`Graph::connect`] and [`Graph::expose`] take it.
66pub struct OutputRef<T> {
67    key: PortKey,
68    _payload: PhantomData<fn() -> T>,
69}
70
71impl<T> OutputRef<T> {
72    /// Output port `index` of `node`. The node's own helpers build these, as only the node
73    /// knows the payload types of its ports.
74    pub fn of_node(node: NodeId, index: usize) -> Self {
75        Self {
76            key: PortKey {
77                owner: Owner::Node(node.0),
78                index,
79            },
80            _payload: PhantomData,
81        }
82    }
83}
84
85impl<T> Clone for OutputRef<T> {
86    fn clone(&self) -> Self {
87        *self
88    }
89}
90
91impl<T> Copy for OutputRef<T> {}
92
93impl<T> fmt::Debug for OutputRef<T> {
94    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
95        f.debug_struct("OutputRef").field("key", &self.key).finish()
96    }
97}
98
99/// An input port reading `T` events.
100pub struct InputRef<T> {
101    node: usize,
102    index: usize,
103    _payload: PhantomData<fn() -> T>,
104}
105
106impl<T> InputRef<T> {
107    /// Input port `index` of `node`; see [`OutputRef::of_node`].
108    pub fn of_node(node: NodeId, index: usize) -> Self {
109        Self {
110            node: node.0,
111            index,
112            _payload: PhantomData,
113        }
114    }
115}
116
117impl<T> Clone for InputRef<T> {
118    fn clone(&self) -> Self {
119        *self
120    }
121}
122
123impl<T> Copy for InputRef<T> {}
124
125impl<T> fmt::Debug for InputRef<T> {
126    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
127        f.debug_struct("InputRef")
128            .field("node", &self.node)
129            .field("index", &self.index)
130            .finish()
131    }
132}
133
134/// An exposed output, read through [`Pipeline::exposed`].
135pub struct ExposedRef<T> {
136    pub(crate) index: usize,
137    _payload: PhantomData<fn() -> T>,
138}
139
140impl<T> Clone for ExposedRef<T> {
141    fn clone(&self) -> Self {
142        *self
143    }
144}
145
146impl<T> Copy for ExposedRef<T> {}
147
148impl<T> fmt::Debug for ExposedRef<T> {
149    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
150        f.debug_struct("ExposedRef")
151            .field("index", &self.index)
152            .finish()
153    }
154}
155
156/// An output port named at run time, for graphs built from a description rather than from
157/// typed code; see [`Graph::output`] and [`Graph::connect_dyn`].
158#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
159pub struct DynOutput {
160    key: PortKey,
161}
162
163/// An input port named at run time; see [`Graph::input`].
164#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
165pub struct DynInput {
166    node: usize,
167    index: usize,
168}
169
170/// An output exposed through [`Graph::expose_dyn`], read with [`Pipeline::exposed_dyn`].
171#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
172pub struct DynExposedRef {
173    pub(crate) index: usize,
174}
175
176/// A source of logic samples, as [`Graph::source`] takes it.
177#[derive(Clone, Debug, PartialEq, Eq)]
178pub struct SourceSpec {
179    /// The source's name, for diagnostics and statistics.
180    pub name: String,
181    /// Its channels, at most [`MAX_CHANNELS`].
182    pub channels: u8,
183    /// Its clock.
184    pub time_base: TimeBase,
185    /// The blocks the source holds for its readers before [`Pipeline::push`] reports it full.
186    pub capacity: usize,
187}
188
189impl SourceSpec {
190    /// A source named `name` with `channels` channels on `time_base`, holding four blocks.
191    pub fn new(name: impl Into<String>, channels: u8, time_base: TimeBase) -> Self {
192        Self {
193            name: name.into(),
194            channels,
195            time_base,
196            capacity: 4,
197        }
198    }
199
200    /// Sets how many blocks the source holds.
201    pub fn with_capacity(mut self, capacity: usize) -> Self {
202        self.capacity = capacity.max(1);
203        self
204    }
205
206    pub(crate) fn output(&self) -> OutputSpec {
207        OutputSpec::new::<Arc<LogicBlock>>(
208            "samples",
209            Props::new()
210                .with("channels", i64::from(self.channels))
211                .with(
212                    "rateHz",
213                    i64::try_from(self.time_base.rate_hz).unwrap_or(i64::MAX),
214                ),
215        )
216    }
217}
218
219/// The source as added to a graph: its id and its output.
220#[derive(Clone, Copy, Debug)]
221pub struct SourceHandle {
222    /// The source.
223    pub id: SourceId,
224    /// Its blocks of samples.
225    pub samples: OutputRef<Arc<LogicBlock>>,
226}
227
228/// Limits every node and queue of a graph shares.
229#[derive(Clone, Copy, Debug, PartialEq, Eq)]
230pub struct GraphConfig {
231    /// Items a queue between nodes holds before the writer waits.
232    pub queue_capacity: usize,
233    /// Operations a node completes in one poll before it yields to the others.
234    pub budget: u32,
235}
236
237impl Default for GraphConfig {
238    fn default() -> Self {
239        Self {
240            queue_capacity: 256,
241            budget: 1024,
242        }
243    }
244}
245
246/// Why a graph cannot be built as asked.
247#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
248pub enum GraphError {
249    /// A port reference names a node the graph does not have.
250    #[error("no node {0} in the graph")]
251    UnknownNode(usize),
252    /// A port reference names a port its owner does not declare.
253    #[error("{owner} declares no {direction} port {index}")]
254    UnknownPort {
255        /// The source or node.
256        owner: String,
257        /// `input` or `output`.
258        direction: &'static str,
259        /// The port index.
260        index: usize,
261    },
262    /// A source has more channels than a block can hold.
263    #[error("source {name} has {channels} channels; at most {MAX_CHANNELS} are supported")]
264    TooManyChannels {
265        /// The source.
266        name: String,
267        /// Its channels.
268        channels: u8,
269    },
270    /// The output and the input do not carry the same schema.
271    #[error("{from} carries {output} but {to} reads {input}")]
272    SchemaMismatch {
273        /// The output port.
274        from: String,
275        /// The input port.
276        to: String,
277        /// The output's schema.
278        output: SchemaId,
279        /// The input's schema.
280        input: SchemaId,
281    },
282    /// The output does not satisfy a requirement of the input.
283    #[error("{to} requires {key} {requirement} of {from}, which declares {found}")]
284    Requirement {
285        /// The output port.
286        from: String,
287        /// The input port.
288        to: String,
289        /// The property.
290        key: &'static str,
291        /// What the input requires.
292        requirement: String,
293        /// What the output declares, or `nothing`.
294        found: String,
295    },
296    /// The input is already connected; an input reads one output.
297    #[error("{to} is already connected")]
298    AlreadyConnected {
299        /// The input port.
300        to: String,
301    },
302    /// The input is not connected.
303    #[error("{to} is not connected")]
304    Unconnected {
305        /// The input port.
306        to: String,
307    },
308    /// The graph has a cycle through this node.
309    #[error("{node} is part of a cycle")]
310    Cycle {
311        /// The node.
312        node: String,
313    },
314    /// The node reads inputs from different sources.
315    #[error("{node} reads from more than one time domain ({first} and {second})")]
316    TimeDomain {
317        /// The node.
318        node: String,
319        /// A source it reads.
320        first: String,
321        /// Another source it reads.
322        second: String,
323    },
324    /// Two exposed outputs share a name.
325    #[error("an output named {0} is already exposed")]
326    DuplicateExposure(String),
327    /// A port named at run time does not exist.
328    #[error("{owner} has no {direction} port named {name}")]
329    UnknownPortName {
330        /// The source or node.
331        owner: String,
332        /// `input` or `output`.
333        direction: &'static str,
334        /// The name asked for.
335        name: String,
336    },
337}
338
339struct NodeEntry {
340    node: Box<dyn Node>,
341    spec: NodeSpec,
342    /// What each input reads, once connected.
343    inputs: Vec<Option<PortKey>>,
344}
345
346pub(crate) struct Exposure {
347    pub(crate) name: String,
348    pub(crate) from: PortKey,
349}
350
351/// A decode graph under construction.
352pub struct Graph {
353    config: GraphConfig,
354    sources: Vec<SourceSpec>,
355    nodes: Vec<NodeEntry>,
356    exposed: Vec<Exposure>,
357}
358
359impl fmt::Debug for Graph {
360    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
361        f.debug_struct("Graph")
362            .field("config", &self.config)
363            .field("sources", &self.sources.len())
364            .field("nodes", &self.nodes.len())
365            .field("exposed", &self.exposed.len())
366            .finish()
367    }
368}
369
370impl Default for Graph {
371    fn default() -> Self {
372        Self::new()
373    }
374}
375
376impl Graph {
377    /// An empty graph with the default limits.
378    pub fn new() -> Self {
379        Self::with_config(GraphConfig::default())
380    }
381
382    /// An empty graph with `config`'s limits.
383    pub fn with_config(config: GraphConfig) -> Self {
384        Self {
385            config,
386            sources: Vec::new(),
387            nodes: Vec::new(),
388            exposed: Vec::new(),
389        }
390    }
391
392    /// The graph's limits.
393    pub fn config(&self) -> GraphConfig {
394        self.config
395    }
396
397    /// Adds a source of logic samples.
398    pub fn source(&mut self, spec: SourceSpec) -> Result<SourceHandle, GraphError> {
399        if spec.channels > MAX_CHANNELS {
400            return Err(GraphError::TooManyChannels {
401                name: spec.name,
402                channels: spec.channels,
403            });
404        }
405        let id = SourceId(self.sources.len());
406        self.sources.push(spec);
407        Ok(SourceHandle {
408            id,
409            samples: OutputRef {
410                key: PortKey {
411                    owner: Owner::Source(id.0),
412                    index: 0,
413                },
414                _payload: PhantomData,
415            },
416        })
417    }
418
419    /// Adds a node.
420    pub fn add(&mut self, node: impl Node) -> NodeId {
421        let spec = node.spec();
422        let inputs = vec![None; spec.inputs.len()];
423        self.nodes.push(NodeEntry {
424            node: Box::new(node),
425            spec,
426            inputs,
427        });
428        NodeId(self.nodes.len() - 1)
429    }
430
431    fn owner_name(&self, owner: Owner) -> String {
432        match owner {
433            Owner::Source(index) => {
434                format!(
435                    "{}#{index}",
436                    self.sources.get(index).map_or("?", |s| s.name.as_str())
437                )
438            }
439            Owner::Node(index) => {
440                format!(
441                    "{}#{index}",
442                    self.nodes.get(index).map_or("?", |n| n.spec.name.as_str())
443                )
444            }
445        }
446    }
447
448    fn output_spec(&self, key: PortKey) -> Result<OutputSpec, GraphError> {
449        let unknown = || GraphError::UnknownPort {
450            owner: self.owner_name(key.owner),
451            direction: "output",
452            index: key.index,
453        };
454        match key.owner {
455            Owner::Source(index) => {
456                let source = self.sources.get(index).ok_or_else(unknown)?;
457                if key.index != 0 {
458                    return Err(unknown());
459                }
460                Ok(source.output())
461            }
462            Owner::Node(index) => {
463                let node = self
464                    .nodes
465                    .get(index)
466                    .ok_or(GraphError::UnknownNode(index))?;
467                node.spec
468                    .outputs
469                    .get(key.index)
470                    .cloned()
471                    .ok_or_else(unknown)
472            }
473        }
474    }
475
476    fn input_spec(&self, node: usize, index: usize) -> Result<&InputSpec, GraphError> {
477        let entry = self.nodes.get(node).ok_or(GraphError::UnknownNode(node))?;
478        entry
479            .spec
480            .inputs
481            .get(index)
482            .ok_or_else(|| GraphError::UnknownPort {
483                owner: self.owner_name(Owner::Node(node)),
484                direction: "input",
485                index,
486            })
487    }
488
489    fn port_name(&self, owner: Owner, name: &str) -> String {
490        format!("{}.{name}", self.owner_name(owner))
491    }
492
493    /// Connects an output to an input, checking that they carry the same schema and that the
494    /// output satisfies what the input requires.
495    pub fn connect<T>(&mut self, from: OutputRef<T>, to: InputRef<T>) -> Result<(), GraphError> {
496        self.connect_keys(from.key, to.node, to.index)
497    }
498
499    /// Exposes an output under `name`, to be read from the pipeline.
500    pub fn expose<T>(
501        &mut self,
502        name: impl Into<String>,
503        from: OutputRef<T>,
504    ) -> Result<ExposedRef<T>, GraphError> {
505        let index = self.expose_key(name.into(), from.key)?;
506        Ok(ExposedRef {
507            index,
508            _payload: PhantomData,
509        })
510    }
511
512    /// The samples output of `source`, for a graph built from a description.
513    pub fn source_output(&self, source: SourceId) -> Result<DynOutput, GraphError> {
514        if source.0 >= self.sources.len() {
515            return Err(GraphError::UnknownPort {
516                owner: self.owner_name(Owner::Source(source.0)),
517                direction: "output",
518                index: 0,
519            });
520        }
521        Ok(DynOutput {
522            key: PortKey {
523                owner: Owner::Source(source.0),
524                index: 0,
525            },
526        })
527    }
528
529    /// Output port `name` of `node`, for a graph built from a description.
530    pub fn output(&self, node: NodeId, name: &str) -> Result<DynOutput, GraphError> {
531        let entry = self
532            .nodes
533            .get(node.0)
534            .ok_or(GraphError::UnknownNode(node.0))?;
535        let index = entry
536            .spec
537            .outputs
538            .iter()
539            .position(|output| output.name == name)
540            .ok_or_else(|| GraphError::UnknownPortName {
541                owner: self.owner_name(Owner::Node(node.0)),
542                direction: "output",
543                name: name.to_owned(),
544            })?;
545        Ok(DynOutput {
546            key: PortKey {
547                owner: Owner::Node(node.0),
548                index,
549            },
550        })
551    }
552
553    /// Input port `name` of `node`, for a graph built from a description.
554    pub fn input(&self, node: NodeId, name: &str) -> Result<DynInput, GraphError> {
555        let entry = self
556            .nodes
557            .get(node.0)
558            .ok_or(GraphError::UnknownNode(node.0))?;
559        let index = entry
560            .spec
561            .inputs
562            .iter()
563            .position(|input| input.name == name)
564            .ok_or_else(|| GraphError::UnknownPortName {
565                owner: self.owner_name(Owner::Node(node.0)),
566                direction: "input",
567                name: name.to_owned(),
568            })?;
569        Ok(DynInput {
570            node: node.0,
571            index,
572        })
573    }
574
575    /// Connects ports named at run time, with the checks of [`Graph::connect`]; the schema
576    /// check is what keeps the event types straight here.
577    pub fn connect_dyn(&mut self, from: DynOutput, to: DynInput) -> Result<(), GraphError> {
578        self.connect_keys(from.key, to.node, to.index)
579    }
580
581    /// Exposes an output named at run time under `name`.
582    pub fn expose_dyn(
583        &mut self,
584        name: impl Into<String>,
585        from: DynOutput,
586    ) -> Result<DynExposedRef, GraphError> {
587        Ok(DynExposedRef {
588            index: self.expose_key(name.into(), from.key)?,
589        })
590    }
591
592    /// The schema of the output behind a port named at run time.
593    pub fn schema_of(&self, output: DynOutput) -> Result<SchemaId, GraphError> {
594        self.output_spec(output.key).map(|spec| spec.schema)
595    }
596
597    /// The properties the output behind a port named at run time declares.
598    pub fn props_of(&self, output: DynOutput) -> Result<Props, GraphError> {
599        self.output_spec(output.key).map(|spec| spec.props)
600    }
601
602    fn connect_keys(&mut self, from: PortKey, node: usize, index: usize) -> Result<(), GraphError> {
603        let output = self.output_spec(from)?;
604        let input = self.input_spec(node, index)?.clone();
605        let from_name = self.port_name(from.owner, output.name);
606        let to_name = self.port_name(Owner::Node(node), input.name);
607        if output.schema != input.schema {
608            return Err(GraphError::SchemaMismatch {
609                from: from_name,
610                to: to_name,
611                output: output.schema,
612                input: input.schema,
613            });
614        }
615        if let Some((key, requirement)) = input.requires.unmet(&output.props) {
616            let found = output
617                .props
618                .get(key)
619                .map_or_else(|| "nothing".to_owned(), |v| v.to_string());
620            let requirement = requirement.to_string();
621            return Err(GraphError::Requirement {
622                from: from_name,
623                to: to_name,
624                key,
625                requirement,
626                found,
627            });
628        }
629        let slot = &mut self.nodes[node].inputs[index];
630        if slot.is_some() {
631            return Err(GraphError::AlreadyConnected { to: to_name });
632        }
633        *slot = Some(from);
634        Ok(())
635    }
636
637    fn expose_key(&mut self, name: String, from: PortKey) -> Result<usize, GraphError> {
638        self.output_spec(from)?;
639        if self.exposed.iter().any(|exposure| exposure.name == name) {
640            return Err(GraphError::DuplicateExposure(name));
641        }
642        self.exposed.push(Exposure { name, from });
643        Ok(self.exposed.len() - 1)
644    }
645
646    /// The nodes in an order where every node comes after the ones it reads.
647    fn order(&self) -> Result<Vec<usize>, GraphError> {
648        let mut pending: Vec<usize> = self
649            .nodes
650            .iter()
651            .map(|entry| {
652                entry
653                    .inputs
654                    .iter()
655                    .filter(|input| {
656                        matches!(
657                            input,
658                            Some(PortKey {
659                                owner: Owner::Node(_),
660                                ..
661                            })
662                        )
663                    })
664                    .count()
665            })
666            .collect();
667        let mut ready: VecDeque<usize> = (0..self.nodes.len())
668            .filter(|&node| pending[node] == 0)
669            .collect();
670        let mut order = Vec::with_capacity(self.nodes.len());
671        while let Some(node) = ready.pop_front() {
672            order.push(node);
673            for (next, entry) in self.nodes.iter().enumerate() {
674                let reads = entry
675                    .inputs
676                    .iter()
677                    .filter(|input| matches!(input, Some(PortKey { owner: Owner::Node(n), .. }) if *n == node))
678                    .count();
679                if reads > 0 {
680                    pending[next] -= reads;
681                    if pending[next] == 0 {
682                        ready.push_back(next);
683                    }
684                }
685            }
686        }
687        if order.len() < self.nodes.len() {
688            let stuck = (0..self.nodes.len())
689                .find(|node| !order.contains(node))
690                .unwrap_or(0);
691            return Err(GraphError::Cycle {
692                node: self.owner_name(Owner::Node(stuck)),
693            });
694        }
695        Ok(order)
696    }
697
698    /// The time domain (source) of each node, in `order`.
699    fn domains(&self, order: &[usize]) -> Result<Vec<Option<usize>>, GraphError> {
700        let mut domains: Vec<Option<usize>> = vec![None; self.nodes.len()];
701        for &node in order {
702            let mut domain: Option<usize> = None;
703            for input in self.nodes[node].inputs.iter().flatten() {
704                let read = match input.owner {
705                    Owner::Source(source) => Some(source),
706                    Owner::Node(upstream) => domains[upstream],
707                };
708                match (domain, read) {
709                    (Some(first), Some(second)) if first != second => {
710                        return Err(GraphError::TimeDomain {
711                            node: self.owner_name(Owner::Node(node)),
712                            first: self.owner_name(Owner::Source(first)),
713                            second: self.owner_name(Owner::Source(second)),
714                        });
715                    }
716                    (None, Some(read)) => domain = Some(read),
717                    _ => {}
718                }
719            }
720            domains[node] = domain;
721        }
722        Ok(domains)
723    }
724
725    /// Checks the graph and starts its nodes.
726    pub fn build(self) -> Result<Pipeline, GraphError> {
727        for (index, entry) in self.nodes.iter().enumerate() {
728            for (port, input) in entry.inputs.iter().enumerate() {
729                if input.is_none() {
730                    return Err(GraphError::Unconnected {
731                        to: self.port_name(Owner::Node(index), entry.spec.inputs[port].name),
732                    });
733                }
734            }
735        }
736        let order = self.order()?;
737        let domains = self.domains(&order)?;
738        // What each input reads: the properties of the output behind it.
739        let input_props: Vec<Vec<Props>> = self
740            .nodes
741            .iter()
742            .map(|entry| {
743                entry
744                    .inputs
745                    .iter()
746                    .flatten()
747                    .map(|key| self.output_spec(*key).map(|spec| spec.props))
748                    .collect::<Result<Vec<_>, _>>()
749            })
750            .collect::<Result<_, _>>()?;
751        let mut exposed_names = BTreeSet::new();
752        for exposure in &self.exposed {
753            if !exposed_names.insert(exposure.name.as_str()) {
754                return Err(GraphError::DuplicateExposure(exposure.name.clone()));
755            }
756        }
757        let Graph {
758            config,
759            sources,
760            nodes,
761            exposed,
762        } = self;
763        let (nodes, specs, inputs): (Vec<_>, Vec<_>, Vec<_>) = nodes
764            .into_iter()
765            .map(|entry| (entry.node, entry.spec, entry.inputs))
766            .fold(
767                (Vec::new(), Vec::new(), Vec::new()),
768                |mut acc, (node, spec, input)| {
769                    acc.0.push(node);
770                    acc.1.push(spec);
771                    acc.2.push(input);
772                    acc
773                },
774            );
775        Pipeline::start(
776            config,
777            sources,
778            nodes,
779            specs,
780            inputs,
781            order,
782            domains,
783            input_props,
784            exposed,
785        )
786    }
787}
uiharu_core/graph.rs

uiharu_core/
graph.rs
