netrun_sim/
graph.rs

1//! Graph topology types for flow-based development networks.
2//!
3//! This module defines the static structure of a network: nodes, ports, edges,
4//! and the conditions that govern packet flow (salvo conditions).
5
6use std::collections::{HashMap, HashSet};
7
8/// The name of a port on a node.
9pub type PortName = String;
10
11/// Specifies the capacity of a port (how many packets it can hold).
12#[derive(Debug, Clone)]
13pub enum PortSlotSpec {
14    /// Port can hold unlimited packets.
15    Infinite,
16    /// Port can hold at most this many packets.
17    Finite(u64),
18}
19
20/// A port on a node where packets can enter or exit.
21#[derive(Debug, Clone)]
22pub struct Port {
23    /// The capacity specification for this port.
24    pub slots_spec: PortSlotSpec,
25}
26
27/// A predicate on the state of a port, used in salvo conditions.
28///
29/// These predicates test the current packet count at a port against
30/// various conditions like empty, full, or numeric comparisons.
31#[derive(Debug, Clone)]
32pub enum PortState {
33    /// Port has zero packets.
34    Empty,
35    /// Port is at capacity (always false for infinite ports).
36    Full,
37    /// Port has at least one packet.
38    NonEmpty,
39    /// Port is below capacity (always true for infinite ports).
40    NonFull,
41    /// Port has exactly this many packets.
42    Equals(u64),
43    /// Port has fewer than this many packets.
44    LessThan(u64),
45    /// Port has more than this many packets.
46    GreaterThan(u64),
47    /// Port has at most this many packets.
48    EqualsOrLessThan(u64),
49    /// Port has at least this many packets.
50    EqualsOrGreaterThan(u64),
51}
52
53/// Specifies how many packets to take from a port in a salvo.
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub enum PacketCount {
56    /// Take all packets from the port.
57    All,
58    /// Take at most this many packets (takes fewer if port has fewer).
59    Count(u64),
60}
61
62/// Specifies the maximum number of times a salvo condition can trigger.
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub enum MaxSalvos {
65    /// No limit on how many times the condition can trigger.
66    Infinite,
67    /// Can trigger at most this many times.
68    Finite(u64),
69}
70
71/// A boolean expression over port states, used to define when salvos can trigger.
72///
73/// This forms a simple expression tree that can combine port state checks
74/// with logical operators (And, Or, Not).
75#[derive(Debug, Clone)]
76pub enum SalvoConditionTerm {
77    /// Always true. Useful for source nodes with no input ports.
78    True,
79    /// Always false. Useful as a placeholder or with Not.
80    False,
81    /// Check if a specific port matches a state predicate.
82    Port { port_name: String, state: PortState },
83    /// All sub-terms must be true.
84    And(Vec<Self>),
85    /// At least one sub-term must be true.
86    Or(Vec<Self>),
87    /// The sub-term must be false.
88    Not(Box<Self>),
89}
90
91/// Evaluates a salvo condition term against the current port packet counts.
92///
93/// # Arguments
94/// * `term` - The condition term to evaluate
95/// * `port_packet_counts` - Map of port names to their current packet counts
96/// * `ports` - Map of port names to their definitions (needed for capacity checks)
97///
98/// # Returns
99/// `true` if the condition is satisfied, `false` otherwise.
100pub fn evaluate_salvo_condition(
101    term: &SalvoConditionTerm,
102    port_packet_counts: &HashMap<PortName, u64>,
103    ports: &HashMap<PortName, Port>,
104) -> bool {
105    match term {
106        SalvoConditionTerm::True => true,
107        SalvoConditionTerm::False => false,
108        SalvoConditionTerm::Port { port_name, state } => {
109            let count = *port_packet_counts.get(port_name).unwrap_or(&0);
110            let port = ports.get(port_name);
111
112            match state {
113                PortState::Empty => count == 0,
114                PortState::Full => match port {
115                    Some(p) => match p.slots_spec {
116                        PortSlotSpec::Infinite => false, // Infinite port can never be full
117                        PortSlotSpec::Finite(max) => count >= max,
118                    },
119                    None => false,
120                },
121                PortState::NonEmpty => count > 0,
122                PortState::NonFull => match port {
123                    Some(p) => match p.slots_spec {
124                        PortSlotSpec::Infinite => true, // Infinite port is always non-full
125                        PortSlotSpec::Finite(max) => count < max,
126                    },
127                    None => true,
128                },
129                PortState::Equals(n) => count == *n,
130                PortState::LessThan(n) => count < *n,
131                PortState::GreaterThan(n) => count > *n,
132                PortState::EqualsOrLessThan(n) => count <= *n,
133                PortState::EqualsOrGreaterThan(n) => count >= *n,
134            }
135        }
136        SalvoConditionTerm::And(terms) => terms
137            .iter()
138            .all(|t| evaluate_salvo_condition(t, port_packet_counts, ports)),
139        SalvoConditionTerm::Or(terms) => terms
140            .iter()
141            .any(|t| evaluate_salvo_condition(t, port_packet_counts, ports)),
142        SalvoConditionTerm::Not(inner) => {
143            !evaluate_salvo_condition(inner, port_packet_counts, ports)
144        }
145    }
146}
147
148/// The name of a salvo condition.
149pub type SalvoConditionName = String;
150
151/// A condition that defines when packets can trigger an epoch or be sent.
152///
153/// Salvo conditions are attached to nodes and control the flow of packets:
154/// - **Input salvo conditions**: Define when packets at input ports can trigger a new epoch
155/// - **Output salvo conditions**: Define when packets at output ports can be sent out
156#[derive(Debug, Clone)]
157pub struct SalvoCondition {
158    /// Maximum number of times this condition can trigger per epoch.
159    /// For input salvo conditions, this must be `Finite(1)`.
160    pub max_salvos: MaxSalvos,
161    /// The ports whose packets are included when this condition triggers,
162    /// and how many packets to take from each port.
163    pub ports: HashMap<PortName, PacketCount>,
164    /// The boolean condition that must be satisfied for this salvo to trigger.
165    pub term: SalvoConditionTerm,
166}
167
168/// Extracts all port names referenced in a SalvoConditionTerm.
169fn collect_ports_from_term(term: &SalvoConditionTerm, ports: &mut HashSet<PortName>) {
170    match term {
171        SalvoConditionTerm::True | SalvoConditionTerm::False => {
172            // No ports referenced
173        }
174        SalvoConditionTerm::Port { port_name, .. } => {
175            ports.insert(port_name.clone());
176        }
177        SalvoConditionTerm::And(terms) | SalvoConditionTerm::Or(terms) => {
178            for t in terms {
179                collect_ports_from_term(t, ports);
180            }
181        }
182        SalvoConditionTerm::Not(inner) => {
183            collect_ports_from_term(inner, ports);
184        }
185    }
186}
187
188/// Errors that can occur during graph validation
189#[derive(Debug, Clone, PartialEq, thiserror::Error)]
190pub enum GraphValidationError {
191    /// Edge references a node that doesn't exist
192    #[error("edge {edge_source} -> {edge_target} references non-existent node '{missing_node}'")]
193    EdgeReferencesNonexistentNode {
194        edge_source: PortRef,
195        edge_target: PortRef,
196        missing_node: NodeName,
197    },
198    /// Edge references a port that doesn't exist on the node
199    #[error("edge {edge_source} -> {edge_target} references non-existent port {missing_port}")]
200    EdgeReferencesNonexistentPort {
201        edge_source: PortRef,
202        edge_target: PortRef,
203        missing_port: PortRef,
204    },
205    /// Edge source is not an output port
206    #[error("edge source {edge_source} must be an output port")]
207    EdgeSourceNotOutputPort {
208        edge_source: PortRef,
209        edge_target: PortRef,
210    },
211    /// Edge target is not an input port
212    #[error("edge target {edge_target} must be an input port")]
213    EdgeTargetNotInputPort {
214        edge_source: PortRef,
215        edge_target: PortRef,
216    },
217    /// SalvoCondition.ports references a port that doesn't exist
218    #[error("{condition_type} salvo condition '{condition_name}' on node '{node_name}' references non-existent port '{missing_port}'", condition_type = if *is_input_condition { "input" } else { "output" })]
219    SalvoConditionReferencesNonexistentPort {
220        node_name: NodeName,
221        condition_name: SalvoConditionName,
222        is_input_condition: bool,
223        missing_port: PortName,
224    },
225    /// SalvoCondition.term references a port that doesn't exist
226    #[error("{condition_type} salvo condition '{condition_name}' on node '{node_name}' has term referencing non-existent port '{missing_port}'", condition_type = if *is_input_condition { "input" } else { "output" })]
227    SalvoConditionTermReferencesNonexistentPort {
228        node_name: NodeName,
229        condition_name: SalvoConditionName,
230        is_input_condition: bool,
231        missing_port: PortName,
232    },
233    /// Input salvo condition has max_salvos != Finite(1)
234    #[error(
235        "input salvo condition '{condition_name}' on node '{node_name}' has max_salvos={max_salvos:?}, but must be Finite(1). Input salvos must have exactly one packet to trigger an epoch."
236    )]
237    InputSalvoConditionInvalidMaxSalvos {
238        node_name: NodeName,
239        condition_name: SalvoConditionName,
240        max_salvos: MaxSalvos,
241    },
242    /// Duplicate edge (same source and target)
243    #[error("duplicate edge: {edge_source} -> {edge_target}")]
244    DuplicateEdge {
245        edge_source: PortRef,
246        edge_target: PortRef,
247    },
248}
249
250/// The name of a node in the graph.
251pub type NodeName = String;
252
253/// A processing unit in the graph with input and output ports.
254///
255/// Nodes are the fundamental building blocks of a flow-based network.
256/// They have:
257/// - Input ports where packets arrive
258/// - Output ports where packets are sent
259/// - Input salvo conditions that define when arriving packets trigger an epoch
260/// - Output salvo conditions that define when output packets can be sent
261#[derive(Debug, Clone)]
262pub struct Node {
263    /// The unique name of this node.
264    pub name: NodeName,
265    /// Input ports where packets can arrive.
266    pub in_ports: HashMap<PortName, Port>,
267    /// Output ports where packets can be sent.
268    pub out_ports: HashMap<PortName, Port>,
269    /// Conditions that trigger new epochs when satisfied by packets at input ports.
270    pub in_salvo_conditions: HashMap<SalvoConditionName, SalvoCondition>,
271    /// Conditions that must be satisfied to send packets from output ports.
272    pub out_salvo_conditions: HashMap<SalvoConditionName, SalvoCondition>,
273}
274
275/// Whether a port is for input or output.
276#[derive(Debug, Clone, PartialEq, Eq, Hash)]
277#[cfg_attr(feature = "python", pyo3::pyclass(eq, eq_int, frozen, hash))]
278pub enum PortType {
279    /// An input port (packets flow into the node).
280    Input,
281    /// An output port (packets flow out of the node).
282    Output,
283}
284
285#[cfg(feature = "python")]
286#[pyo3::pymethods]
287impl PortType {
288    fn __repr__(&self) -> String {
289        match self {
290            PortType::Input => "PortType.Input".to_string(),
291            PortType::Output => "PortType.Output".to_string(),
292        }
293    }
294}
295
296/// A reference to a specific port on a specific node.
297#[derive(Debug, Clone, PartialEq, Eq, Hash)]
298#[cfg_attr(feature = "python", pyo3::pyclass(eq, frozen, hash, get_all))]
299pub struct PortRef {
300    /// The name of the node containing this port.
301    pub node_name: NodeName,
302    /// Whether this is an input or output port.
303    pub port_type: PortType,
304    /// The name of the port on the node.
305    pub port_name: PortName,
306}
307
308#[cfg(feature = "python")]
309#[pyo3::pymethods]
310impl PortRef {
311    #[new]
312    fn py_new(node_name: String, port_type: PortType, port_name: String) -> Self {
313        PortRef {
314            node_name,
315            port_type,
316            port_name,
317        }
318    }
319
320    fn __repr__(&self) -> String {
321        format!(
322            "PortRef('{}', {:?}, '{}')",
323            self.node_name, self.port_type, self.port_name
324        )
325    }
326
327    fn __str__(&self) -> String {
328        self.to_string()
329    }
330}
331
332impl std::fmt::Display for PortRef {
333    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
334        let port_type_str = match self.port_type {
335            PortType::Input => "in",
336            PortType::Output => "out",
337        };
338        write!(f, "{}.{}.{}", self.node_name, port_type_str, self.port_name)
339    }
340}
341
342/// A connection between two ports in the graph.
343///
344/// Edges connect output ports to input ports, allowing packets to flow
345/// between nodes.
346#[derive(Debug, Clone, PartialEq, Eq, Hash)]
347#[cfg_attr(feature = "python", pyo3::pyclass(eq, frozen, hash, get_all))]
348pub struct Edge {
349    /// The output port where this edge originates.
350    pub source: PortRef,
351    /// The input port where this edge terminates.
352    pub target: PortRef,
353}
354
355#[cfg(feature = "python")]
356#[pyo3::pymethods]
357impl Edge {
358    #[new]
359    fn py_new(source: PortRef, target: PortRef) -> Self {
360        Edge { source, target }
361    }
362
363    fn __repr__(&self) -> String {
364        format!("Edge({}, {})", self.source, self.target)
365    }
366
367    fn __str__(&self) -> String {
368        self.to_string()
369    }
370}
371
372impl std::fmt::Display for Edge {
373    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
374        write!(f, "{} -> {}", self.source, self.target)
375    }
376}
377
378/// The static topology of a flow-based network.
379///
380/// A Graph defines the structure of a network: which nodes exist, how they're
381/// connected, and what conditions govern packet flow. The graph is immutable
382/// after creation.
383///
384/// # Example
385///
386/// ```
387/// use netrun_sim::graph::{Graph, Node, Edge, PortRef, PortType, Port, PortSlotSpec};
388/// use std::collections::HashMap;
389///
390/// // Create a simple A -> B graph
391/// let node_a = Node {
392///     name: "A".to_string(),
393///     in_ports: HashMap::new(),
394///     out_ports: [("out".to_string(), Port { slots_spec: PortSlotSpec::Infinite })].into(),
395///     in_salvo_conditions: HashMap::new(),
396///     out_salvo_conditions: HashMap::new(),
397/// };
398/// let node_b = Node {
399///     name: "B".to_string(),
400///     in_ports: [("in".to_string(), Port { slots_spec: PortSlotSpec::Infinite })].into(),
401///     out_ports: HashMap::new(),
402///     in_salvo_conditions: HashMap::new(),
403///     out_salvo_conditions: HashMap::new(),
404/// };
405///
406/// let edge = Edge {
407///     source: PortRef { node_name: "A".to_string(), port_type: PortType::Output, port_name: "out".to_string() },
408///     target: PortRef { node_name: "B".to_string(), port_type: PortType::Input, port_name: "in".to_string() },
409/// };
410///
411/// let graph = Graph::new(vec![node_a, node_b], vec![edge]);
412/// assert!(graph.validate().is_empty());
413/// ```
414#[derive(Debug, Clone)]
415pub struct Graph {
416    nodes: HashMap<NodeName, Node>,
417    edges: HashSet<Edge>,
418    edges_by_tail: HashMap<PortRef, Edge>,
419    edges_by_head: HashMap<PortRef, Edge>,
420}
421
422impl Graph {
423    /// Creates a new Graph from a list of nodes and edges.
424    ///
425    /// Builds internal indexes for efficient edge lookups by source (tail) and target (head) ports.
426    pub fn new(nodes: Vec<Node>, edges: Vec<Edge>) -> Self {
427        let nodes_map: HashMap<NodeName, Node> = nodes
428            .into_iter()
429            .map(|node| (node.name.clone(), node))
430            .collect();
431
432        let mut edges_set: HashSet<Edge> = HashSet::new();
433        let mut edges_by_tail: HashMap<PortRef, Edge> = HashMap::new();
434        let mut edges_by_head: HashMap<PortRef, Edge> = HashMap::new();
435
436        for edge in edges {
437            edges_by_tail.insert(edge.source.clone(), edge.clone());
438            edges_by_head.insert(edge.target.clone(), edge.clone());
439            edges_set.insert(edge);
440        }
441
442        Graph {
443            nodes: nodes_map,
444            edges: edges_set,
445            edges_by_tail,
446            edges_by_head,
447        }
448    }
449
450    /// Returns a reference to all nodes in the graph, keyed by name.
451    pub fn nodes(&self) -> &HashMap<NodeName, Node> {
452        &self.nodes
453    }
454
455    /// Returns a reference to all edges in the graph.
456    pub fn edges(&self) -> &HashSet<Edge> {
457        &self.edges
458    }
459
460    /// Returns the edge that has the given output port as its source (tail).
461    pub fn get_edge_by_tail(&self, output_port_ref: &PortRef) -> Option<&Edge> {
462        self.edges_by_tail.get(output_port_ref)
463    }
464
465    /// Returns the edge that has the given input port as its target (head).
466    pub fn get_edge_by_head(&self, input_port_ref: &PortRef) -> Option<&Edge> {
467        self.edges_by_head.get(input_port_ref)
468    }
469
470    /// Validates the graph structure.
471    ///
472    /// Returns a list of all validation errors found. An empty list means the graph is valid.
473    pub fn validate(&self) -> Vec<GraphValidationError> {
474        let mut errors = Vec::new();
475
476        // Track seen edges to detect duplicates
477        let mut seen_edges: HashSet<(&PortRef, &PortRef)> = HashSet::new();
478
479        // Validate edges
480        for edge in &self.edges {
481            let source = &edge.source;
482            let target = &edge.target;
483
484            // Check for duplicate edges
485            if !seen_edges.insert((source, target)) {
486                errors.push(GraphValidationError::DuplicateEdge {
487                    edge_source: source.clone(),
488                    edge_target: target.clone(),
489                });
490                continue;
491            }
492
493            // Validate source node exists
494            let source_node = match self.nodes.get(&source.node_name) {
495                Some(node) => node,
496                None => {
497                    errors.push(GraphValidationError::EdgeReferencesNonexistentNode {
498                        edge_source: source.clone(),
499                        edge_target: target.clone(),
500                        missing_node: source.node_name.clone(),
501                    });
502                    continue;
503                }
504            };
505
506            // Validate target node exists
507            let target_node = match self.nodes.get(&target.node_name) {
508                Some(node) => node,
509                None => {
510                    errors.push(GraphValidationError::EdgeReferencesNonexistentNode {
511                        edge_source: source.clone(),
512                        edge_target: target.clone(),
513                        missing_node: target.node_name.clone(),
514                    });
515                    continue;
516                }
517            };
518
519            // Validate source is an output port
520            if source.port_type != PortType::Output {
521                errors.push(GraphValidationError::EdgeSourceNotOutputPort {
522                    edge_source: source.clone(),
523                    edge_target: target.clone(),
524                });
525            } else if !source_node.out_ports.contains_key(&source.port_name) {
526                errors.push(GraphValidationError::EdgeReferencesNonexistentPort {
527                    edge_source: source.clone(),
528                    edge_target: target.clone(),
529                    missing_port: source.clone(),
530                });
531            }
532
533            // Validate target is an input port
534            if target.port_type != PortType::Input {
535                errors.push(GraphValidationError::EdgeTargetNotInputPort {
536                    edge_source: source.clone(),
537                    edge_target: target.clone(),
538                });
539            } else if !target_node.in_ports.contains_key(&target.port_name) {
540                errors.push(GraphValidationError::EdgeReferencesNonexistentPort {
541                    edge_source: source.clone(),
542                    edge_target: target.clone(),
543                    missing_port: target.clone(),
544                });
545            }
546        }
547
548        // Validate nodes and their salvo conditions
549        for (node_name, node) in &self.nodes {
550            // Validate input salvo conditions
551            for (cond_name, condition) in &node.in_salvo_conditions {
552                // Input salvo conditions must have max_salvos == Finite(1)
553                if condition.max_salvos != MaxSalvos::Finite(1) {
554                    errors.push(GraphValidationError::InputSalvoConditionInvalidMaxSalvos {
555                        node_name: node_name.clone(),
556                        condition_name: cond_name.clone(),
557                        max_salvos: condition.max_salvos.clone(),
558                    });
559                }
560
561                // Validate ports in condition.ports exist as input ports
562                for port_name in condition.ports.keys() {
563                    if !node.in_ports.contains_key(port_name) {
564                        errors.push(
565                            GraphValidationError::SalvoConditionReferencesNonexistentPort {
566                                node_name: node_name.clone(),
567                                condition_name: cond_name.clone(),
568                                is_input_condition: true,
569                                missing_port: port_name.clone(),
570                            },
571                        );
572                    }
573                }
574
575                // Validate ports in condition.term exist as input ports
576                let mut term_ports = HashSet::new();
577                collect_ports_from_term(&condition.term, &mut term_ports);
578                for port_name in term_ports {
579                    if !node.in_ports.contains_key(&port_name) {
580                        errors.push(
581                            GraphValidationError::SalvoConditionTermReferencesNonexistentPort {
582                                node_name: node_name.clone(),
583                                condition_name: cond_name.clone(),
584                                is_input_condition: true,
585                                missing_port: port_name,
586                            },
587                        );
588                    }
589                }
590            }
591
592            // Validate output salvo conditions
593            for (cond_name, condition) in &node.out_salvo_conditions {
594                // Validate ports in condition.ports exist as output ports
595                for port_name in condition.ports.keys() {
596                    if !node.out_ports.contains_key(port_name) {
597                        errors.push(
598                            GraphValidationError::SalvoConditionReferencesNonexistentPort {
599                                node_name: node_name.clone(),
600                                condition_name: cond_name.clone(),
601                                is_input_condition: false,
602                                missing_port: port_name.clone(),
603                            },
604                        );
605                    }
606                }
607
608                // Validate ports in condition.term exist as output ports
609                let mut term_ports = HashSet::new();
610                collect_ports_from_term(&condition.term, &mut term_ports);
611                for port_name in term_ports {
612                    if !node.out_ports.contains_key(&port_name) {
613                        errors.push(
614                            GraphValidationError::SalvoConditionTermReferencesNonexistentPort {
615                                node_name: node_name.clone(),
616                                condition_name: cond_name.clone(),
617                                is_input_condition: false,
618                                missing_port: port_name,
619                            },
620                        );
621                    }
622                }
623            }
624        }
625
626        errors
627    }
628}
629
630#[cfg(test)]
631#[path = "graph_tests.rs"]
632mod tests;
netrun_sim/graph.rs

netrun_sim/
graph.rs
