diamond_flow/
diamond_flow.rs

1//! Example: Diamond flow pattern with branching and merging
2//!
3//! Graph structure:
4//!        A
5//!       / \
6//!      B   C
7//!       \ /
8//!        D
9//!
10//! This example demonstrates:
11//! - Branching: A sends packets to both B and C
12//! - Merging: D waits for packets from both B and C
13//! - Synchronization: D's epoch only triggers when both inputs are present
14
15use netrun_sim::graph::{
16    Edge, Graph, Node, Port, PortRef, PortSlotSpec, PortState, PortType,
17    SalvoCondition, SalvoConditionTerm,
18};
19use netrun_sim::net::{
20    Net, NetAction, NetActionResponse, NetActionResponseData, PacketLocation,
21};
22use std::collections::HashMap;
23
24fn main() {
25    // Create a diamond graph
26    let graph = create_diamond_graph();
27    println!("Created diamond graph: A -> B,C -> D");
28    println!("D requires inputs from BOTH B and C\n");
29
30    let mut net = Net::new(graph);
31
32    // Create two packets and place them on edges from A
33    let packet1 = create_packet(&mut net);
34    let packet2 = create_packet(&mut net);
35    println!("Created packets: {} and {}", packet1, packet2);
36
37    // Place packet1 on edge A -> B
38    let edge_a_b = edge_location("A", "out1", "B", "in");
39    net.do_action(&NetAction::TransportPacketToLocation(packet1.clone(), edge_a_b));
40    println!("Placed packet1 on edge A -> B");
41
42    // Place packet2 on edge A -> C
43    let edge_a_c = edge_location("A", "out2", "C", "in");
44    net.do_action(&NetAction::TransportPacketToLocation(packet2.clone(), edge_a_c));
45    println!("Placed packet2 on edge A -> C");
46
47    // Run network - packets move to B and C, triggering epochs
48    net.do_action(&NetAction::RunNetUntilBlocked);
49
50    let startable = net.get_startable_epochs();
51    println!("\nAfter first run: {} startable epochs (B and C)", startable.len());
52
53    // Process B and C, sending outputs to D
54    for epoch_id in startable {
55        let epoch = net.get_epoch(&epoch_id).unwrap();
56        let node_name = epoch.node_name.clone();
57        println!("\nProcessing node {}", node_name);
58
59        // Start the epoch
60        let started = match net.do_action(&NetAction::StartEpoch(epoch_id.clone())) {
61            NetActionResponse::Success(NetActionResponseData::StartedEpoch(e), _) => e,
62            _ => panic!("Failed to start epoch"),
63        };
64
65        // Find and consume the input packet
66        let input_packet = started.in_salvo.packets[0].1.clone();
67        net.do_action(&NetAction::ConsumePacket(input_packet));
68
69        // Create output packet
70        let output = create_packet_in_epoch(&mut net, &started.id);
71
72        // Load into output port and send
73        net.do_action(&NetAction::LoadPacketIntoOutputPort(output, "out".to_string()));
74        net.do_action(&NetAction::SendOutputSalvo(started.id.clone(), "default".to_string()));
75
76        // Finish epoch
77        net.do_action(&NetAction::FinishEpoch(started.id));
78        println!("  Finished {} - sent packet to D", node_name);
79    }
80
81    // Run network - packets move from B->D and C->D edges to D's input ports
82    net.do_action(&NetAction::RunNetUntilBlocked);
83
84    // Check D's input ports
85    let d_in1 = PacketLocation::InputPort("D".to_string(), "in1".to_string());
86    let d_in2 = PacketLocation::InputPort("D".to_string(), "in2".to_string());
87    println!("\nD's input ports:");
88    println!("  in1 (from B): {} packets", net.packet_count_at(&d_in1));
89    println!("  in2 (from C): {} packets", net.packet_count_at(&d_in2));
90
91    // D should now have a startable epoch (both inputs present)
92    let startable_d = net.get_startable_epochs();
93    println!("\nStartable epochs at D: {}", startable_d.len());
94
95    if let Some(d_epoch_id) = startable_d.first() {
96        let d_epoch = net.get_epoch(d_epoch_id).unwrap();
97        println!("D's epoch received {} packets from both branches!", d_epoch.in_salvo.packets.len());
98    }
99
100    println!("\nDiamond flow example complete!");
101}
102
103fn create_diamond_graph() -> Graph {
104    // Node A: source with two outputs
105    let node_a = Node {
106        name: "A".to_string(),
107        in_ports: HashMap::new(),
108        out_ports: [
109            ("out1".to_string(), Port { slots_spec: PortSlotSpec::Infinite }),
110            ("out2".to_string(), Port { slots_spec: PortSlotSpec::Infinite }),
111        ].into(),
112        in_salvo_conditions: HashMap::new(),
113        out_salvo_conditions: HashMap::new(),
114    };
115
116    // Node B: one input, one output
117    let node_b = create_simple_node("B");
118
119    // Node C: one input, one output
120    let node_c = create_simple_node("C");
121
122    // Node D: TWO inputs (requires both), no outputs
123    let node_d = Node {
124        name: "D".to_string(),
125        in_ports: [
126            ("in1".to_string(), Port { slots_spec: PortSlotSpec::Infinite }),
127            ("in2".to_string(), Port { slots_spec: PortSlotSpec::Infinite }),
128        ].into(),
129        out_ports: HashMap::new(),
130        in_salvo_conditions: [(
131            "default".to_string(),
132            SalvoCondition {
133                max_salvos: 1,
134                ports: vec!["in1".to_string(), "in2".to_string()],
135                // Require BOTH inputs to be non-empty
136                term: SalvoConditionTerm::And(vec![
137                    SalvoConditionTerm::Port {
138                        port_name: "in1".to_string(),
139                        state: PortState::NonEmpty,
140                    },
141                    SalvoConditionTerm::Port {
142                        port_name: "in2".to_string(),
143                        state: PortState::NonEmpty,
144                    },
145                ]),
146            },
147        )].into(),
148        out_salvo_conditions: HashMap::new(),
149    };
150
151    let edges = vec![
152        create_edge("A", "out1", "B", "in"),
153        create_edge("A", "out2", "C", "in"),
154        create_edge("B", "out", "D", "in1"),
155        create_edge("C", "out", "D", "in2"),
156    ];
157
158    let graph = Graph::new(vec![node_a, node_b, node_c, node_d], edges);
159    assert!(graph.validate().is_empty(), "Graph validation failed");
160    graph
161}
162
163fn create_simple_node(name: &str) -> Node {
164    Node {
165        name: name.to_string(),
166        in_ports: [("in".to_string(), Port { slots_spec: PortSlotSpec::Infinite })].into(),
167        out_ports: [("out".to_string(), Port { slots_spec: PortSlotSpec::Infinite })].into(),
168        in_salvo_conditions: [(
169            "default".to_string(),
170            SalvoCondition {
171                max_salvos: 1,
172                ports: vec!["in".to_string()],
173                term: SalvoConditionTerm::Port {
174                    port_name: "in".to_string(),
175                    state: PortState::NonEmpty,
176                },
177            },
178        )].into(),
179        out_salvo_conditions: [(
180            "default".to_string(),
181            SalvoCondition {
182                max_salvos: 0,
183                ports: vec!["out".to_string()],
184                term: SalvoConditionTerm::Port {
185                    port_name: "out".to_string(),
186                    state: PortState::NonEmpty,
187                },
188            },
189        )].into(),
190    }
191}
192
193fn create_edge(src_node: &str, src_port: &str, tgt_node: &str, tgt_port: &str) -> Edge {
194    Edge {
195        source: PortRef {
196            node_name: src_node.to_string(),
197            port_type: PortType::Output,
198            port_name: src_port.to_string(),
199        },
200        target: PortRef {
201            node_name: tgt_node.to_string(),
202            port_type: PortType::Input,
203            port_name: tgt_port.to_string(),
204        },
205    }
206}
207
208fn edge_location(src_node: &str, src_port: &str, tgt_node: &str, tgt_port: &str) -> PacketLocation {
209    PacketLocation::Edge(Edge {
210        source: PortRef {
211            node_name: src_node.to_string(),
212            port_type: PortType::Output,
213            port_name: src_port.to_string(),
214        },
215        target: PortRef {
216            node_name: tgt_node.to_string(),
217            port_type: PortType::Input,
218            port_name: tgt_port.to_string(),
219        },
220    })
221}
222
223fn create_packet(net: &mut Net) -> ulid::Ulid {
224    match net.do_action(&NetAction::CreatePacket(None)) {
225        NetActionResponse::Success(NetActionResponseData::Packet(id), _) => id,
226        _ => panic!("Failed to create packet"),
227    }
228}
229
230fn create_packet_in_epoch(net: &mut Net, epoch_id: &ulid::Ulid) -> ulid::Ulid {
231    match net.do_action(&NetAction::CreatePacket(Some(epoch_id.clone()))) {
232        NetActionResponse::Success(NetActionResponseData::Packet(id), _) => id,
233        _ => panic!("Failed to create packet in epoch"),
234    }
235}
diamond_flow/diamond_flow.rs

diamond_flow/
diamond_flow.rs
