dagrs 0.8.1

Dagrs follows the concept of Flow-based Programming and is suitable for the execution of multiple tasks with graph-like dependencies. Dagrs has the characteristics of high performance and asynchronous execution. It provides users with a convenient programming interface.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

use async_trait::async_trait;
use std::collections::HashSet;
use std::sync::Arc;
use tokio::sync::Mutex;

use crate::{
    DagrsError, DagrsResult, EnvVar, ErrorCode, InChannels, Node, NodeId, NodeName, NodeTable,
    OutChannels, Output,
};

/// A special node type that represents a subgraph of nodes in a loop structure.
///
/// The LoopSubgraph is included in the main graph as a single node, but internally contains
/// multiple nodes that will be executed repeatedly. The connection and execution of the loop is controlled
/// by the parent graph rather than the LoopSubgraph itself.
pub struct LoopSubgraph {
    id: NodeId,
    name: NodeName,
    in_channels: InChannels,
    out_channels: OutChannels,
    // Inner nodes, contains the nodes that need to be executed in a loop
    inner_nodes: Vec<Arc<Mutex<dyn Node>>>,
    inner_node_ids: HashSet<NodeId>,
}

impl LoopSubgraph {
    pub fn new(name: NodeName, node_table: &mut NodeTable) -> Self {
        Self {
            id: node_table.alloc_id_for(&name),
            name,
            in_channels: InChannels::default(),
            out_channels: OutChannels::default(),
            inner_nodes: Vec::new(),
            inner_node_ids: HashSet::new(),
        }
    }

    /// Fallible constructor kept for API symmetry with the main graph build path.
    ///
    /// Construction is currently infallible, so this simply returns `Ok(Self::new(...))`.
    /// Callers that prefer a uniform `Result`-based build style can use this variant.
    pub fn try_new(name: NodeName, node_table: &mut NodeTable) -> DagrsResult<Self> {
        Ok(Self::new(name, node_table))
    }

    /// Add a node to the subgraph
    pub fn add_node(&mut self, node: impl Node + 'static) -> DagrsResult<NodeId> {
        let node_id = node.id();
        if self.inner_node_ids.contains(&node_id) {
            return Err(DagrsError::new(
                ErrorCode::DgBld0003DuplicateNodeId,
                "duplicate node id detected while building loop subgraph",
            )
            .with_node_id(node_id.as_usize()));
        }

        self.inner_node_ids.insert(node_id);
        self.inner_nodes.push(Arc::new(Mutex::new(node)));
        Ok(node_id)
    }
}

#[async_trait]
impl Node for LoopSubgraph {
    fn id(&self) -> NodeId {
        self.id
    }

    fn name(&self) -> NodeName {
        self.name.clone()
    }

    fn input_channels(&mut self) -> &mut InChannels {
        &mut self.in_channels
    }

    fn output_channels(&mut self) -> &mut OutChannels {
        &mut self.out_channels
    }

    fn loop_structure(&self) -> Option<Vec<Arc<Mutex<dyn Node>>>> {
        Some(self.inner_nodes.clone())
    }

    async fn run(&mut self, _: Arc<EnvVar>) -> Output {
        Output::error(DagrsError::new(
            ErrorCode::DgRun0006NodeExecutionFailed,
            "loop subgraph should not be executed directly",
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::LoopSubgraph;
    use crate::node::{NodeId, NodeName};
    use crate::{EnvVar, InChannels, Node, NodeTable, OutChannels, Output};
    use async_trait::async_trait;
    use std::sync::Arc;

    struct FixedIdNode {
        id: NodeId,
        name: NodeName,
        in_channels: InChannels,
        out_channels: OutChannels,
    }

    impl FixedIdNode {
        fn new(id: NodeId, name: impl Into<NodeName>) -> Self {
            Self {
                id,
                name: name.into(),
                in_channels: InChannels::default(),
                out_channels: OutChannels::default(),
            }
        }
    }

    #[async_trait]
    impl Node for FixedIdNode {
        fn id(&self) -> NodeId {
            self.id
        }

        fn name(&self) -> NodeName {
            self.name.clone()
        }

        fn input_channels(&mut self) -> &mut InChannels {
            &mut self.in_channels
        }

        fn output_channels(&mut self) -> &mut OutChannels {
            &mut self.out_channels
        }

        async fn run(&mut self, _: Arc<EnvVar>) -> Output {
            Output::empty()
        }
    }

    #[test]
    fn add_node_rejects_duplicate_ids() {
        let mut table = NodeTable::new();
        let mut loop_subgraph = LoopSubgraph::new("loop".to_string(), &mut table);
        let duplicate_id = NodeId(42);

        loop_subgraph
            .add_node(FixedIdNode::new(duplicate_id, "first"))
            .unwrap();
        let err = loop_subgraph
            .add_node(FixedIdNode::new(duplicate_id, "second"))
            .expect_err("duplicate loop node id should fail");

        assert_eq!(err.code, crate::ErrorCode::DgBld0003DuplicateNodeId);
    }
}