typed action

fmt

typed action

example

typed action

TypedChannel + Chnl trait

Author: 191220029

examples/typed_action.rs

@@ -0,0 +1,122 @@
+//! Use [`TypedAction`] to rewrite compute_dag.rs
+//!
+//! Only use Dag, execute a job. The graph is as follows:
+//!
+//!    ↱----------↴
+//!    B -→ E --→ G
+//!  ↗    ↗     ↗
+//! A --→ C    /
+//!  ↘    ↘  /
+//!   D -→ F
+//!
+//! The final execution result is 272.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use dagrs::{
+    connection::{in_channel::TypedInChannels, out_channel::TypedOutChannels},
+    node::typed_action::TypedAction,
+    Content, DefaultNode, EnvVar, Graph, Node, NodeTable, Output,
+};
+
+const BASE: &str = "base";
+
+struct Compute(usize);
+
+#[async_trait]
+impl TypedAction for Compute {
+    type I = usize;
+    type O = usize;
+
+    async fn run(
+        &self,
+        mut in_channels: TypedInChannels<Self::I>,
+        out_channels: TypedOutChannels<Self::O>,
+        env: Arc<EnvVar>,
+    ) -> Output {
+        let base = env.get::<usize>(BASE).unwrap();
+        let mut sum = self.0;
+
+        // Collect all input values from input channels
+        let inputs = in_channels
+            .map(|result| {
+                if let Ok(Some(value)) = result {
+                    *value
+                } else {
+                    0
+                }
+            })
+            .await;
+
+        // Calculate the sum
+        for input in inputs {
+            sum += input * base;
+        }
+
+        // Broadcast the result to all output channels
+        out_channels.broadcast(sum).await;
+
+        Output::Out(Some(Content::new(sum)))
+    }
+}
+
+fn main() {
+    env_logger::init();
+
+    let mut node_table = NodeTable::default();
+
+    let a = DefaultNode::with_action("Compute A".to_string(), Compute(1), &mut node_table);
+    let a_id = a.id();
+
+    let b = DefaultNode::with_action("Compute B".to_string(), Compute(2), &mut node_table);
+    let b_id = b.id();
+
+    let mut c = DefaultNode::new("Compute C".to_string(), &mut node_table);
+    c.set_action(Compute(4));
+    let c_id = c.id();
+
+    let mut d = DefaultNode::new("Compute D".to_string(), &mut node_table);
+    d.set_action(Compute(8));
+    let d_id = d.id();
+
+    let e = DefaultNode::with_action("Compute E".to_string(), Compute(16), &mut node_table);
+    let e_id = e.id();
+    let f = DefaultNode::with_action("Compute F".to_string(), Compute(32), &mut node_table);
+    let f_id = f.id();
+
+    let g = DefaultNode::with_action("Compute G".to_string(), Compute(64), &mut node_table);
+    let g_id = g.id();
+
+    let mut graph = Graph::new();
+    vec![a, b, c, d, e, f, g]
+        .into_iter()
+        .for_each(|node| graph.add_node(node));
+
+    graph.add_edge(a_id, vec![b_id, c_id, d_id]);
+    graph.add_edge(b_id, vec![e_id, g_id]);
+    graph.add_edge(c_id, vec![e_id, f_id]);
+    graph.add_edge(d_id, vec![f_id]);
+    graph.add_edge(e_id, vec![g_id]);
+    graph.add_edge(f_id, vec![g_id]);
+
+    let mut env = EnvVar::new(node_table);
+    env.set("base", 2usize);
+    graph.set_env(env);
+
+    match graph.start() {
+        Ok(_) => {
+            let res = graph
+                .get_results::<usize>()
+                .get(&g_id)
+                .unwrap()
+                .clone()
+                .unwrap();
+            // 验证执行结果
+            assert_eq!(*res, 272)
+        }
+        Err(e) => {
+            panic!("图执行失败: {:?}", e);
+        }
+    }
+}

src/connection/in_channel.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, marker::PhantomData, sync::Arc};
 
 use futures::future::join_all;
 use tokio::sync::{broadcast, mpsc, Mutex};
@@ -141,7 +141,6 @@ impl InChannel {
             },
         }
     }
-
     /// Close the channel and drop the messages inside.
     fn close(&mut self) {
         match self {
@@ -152,6 +151,100 @@ impl InChannel {
     }
 }
 
+/// # Typed Input Channels
+/// A hash-table mapping `NodeId` to `InChannel`. This provides type-safe channel communication
+/// between nodes.
+#[derive(Default)]
+pub struct TypedInChannels<T: Send + Sync + 'static>(
+    pub(crate) HashMap<NodeId, Arc<Mutex<InChannel>>>,
+    // maker for type T
+    pub(crate) PhantomData<T>,
+);
+
+impl<T: Send + Sync + 'static> TypedInChannels<T> {
+    /// Perform a blocking receive on the incoming channel from `NodeId`.
+    pub fn blocking_recv_from(&mut self, id: &NodeId) -> Result<Option<Arc<T>>, RecvErr> {
+        match self.get(id) {
+            Some(channel) => {
+                let content: Content = channel.blocking_lock().blocking_recv()?;
+                Ok(content.into_inner())
+            }
+            None => Err(RecvErr::NoSuchChannel),
+        }
+    }
+
+    /// Perform a asynchronous receive on the incoming channel from `NodeId`.
+    pub async fn recv_from(&mut self, id: &NodeId) -> Result<Option<Arc<T>>, RecvErr> {
+        match self.get(id) {
+            Some(channel) => {
+                let content: Content = channel.lock().await.recv().await?;
+                Ok(content.into_inner())
+            }
+            None => Err(RecvErr::NoSuchChannel),
+        }
+    }
+
+    /// Calls `blocking_recv` for all the [`InChannel`]s, and applies transformation `f` to
+    /// the return values of the call.
+    pub fn blocking_map<F, U>(&mut self, mut f: F) -> Vec<U>
+    where
+        F: FnMut(Result<Option<Arc<T>>, RecvErr>) -> U,
+    {
+        self.keys()
+            .into_iter()
+            .map(|id| f(self.blocking_recv_from(&id)))
+            .collect()
+    }
+
+    /// Calls `recv` for all the [`InChannel`]s, and applies transformation `f` to
+    /// the return values of the call asynchronously.
+    pub async fn map<F, U>(&mut self, mut f: F) -> Vec<U>
+    where
+        F: FnMut(Result<Option<Arc<T>>, RecvErr>) -> U,
+    {
+        let futures = self.0.iter_mut().map(|(_, c)| async {
+            let content: Content = c.lock().await.recv().await?;
+            Ok(content.into_inner())
+        });
+        join_all(futures).await.into_iter().map(|x| f(x)).collect()
+    }
+
+    /// Close the channel by the given `NodeId` asynchronously, and remove the channel in this map.
+    pub async fn close_async(&mut self, id: &NodeId) {
+        if let Some(c) = self.get(id) {
+            c.lock().await.close();
+            self.0.remove(id);
+        }
+    }
+
+    /// Close the channel by the given `NodeId`, and remove the channel in this map.
+    pub fn close(&mut self, id: &NodeId) {
+        if let Some(c) = self.get(id) {
+            c.blocking_lock().close();
+            self.0.remove(id);
+        }
+    }
+
+    pub(crate) fn insert(&mut self, node_id: NodeId, channel: Arc<Mutex<InChannel>>) {
+        self.0.insert(node_id, channel);
+    }
+
+    pub(crate) fn close_all(&mut self) {
+        self.0.values_mut().for_each(|c| c.blocking_lock().close());
+    }
+
+    fn get(&self, id: &NodeId) -> Option<Arc<Mutex<InChannel>>> {
+        match self.0.get(id) {
+            Some(c) => Some(c.clone()),
+            None => None,
+        }
+    }
+
+    fn keys(&self) -> Vec<NodeId> {
+        self.0.keys().map(|x| *x).collect()
+    }
+}
+
 /// # Input Channel Error Types
 /// - NoSuchChannel: try to get a channel with an invalid `NodeId`.
 /// - Closed: the channel to receive messages from is closed and empty already.

src/connection/information_packet.rs

@@ -3,7 +3,7 @@ use std::{any::Any, sync::Arc};
 /// Container type to store task output.
 #[derive(Debug, Clone)]
 pub struct Content {
-    inner: Arc<dyn Any + Send + Sync>,
+    pub inner: Arc<dyn Any + Send + Sync>,
 }
 
 impl Content {

src/connection/out_channel.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, marker::PhantomData, sync::Arc};
 
 use futures::future::join_all;
 use tokio::sync::{broadcast, mpsc, Mutex};
@@ -124,3 +124,73 @@ pub enum SendErr {
     NoSuchChannel,
     ClosedChannel(Content),
 }
+
+/// # Typed Output Channels
+/// A hash-table mapping [`NodeId`] to [`OutChannel`]. This provides type-safe channel communication
+/// between nodes.
+#[derive(Default)]
+pub struct TypedOutChannels<T: Send + Sync + 'static>(
+    pub(crate) HashMap<NodeId, Arc<Mutex<OutChannel>>>,
+    // maker for type T
+    pub(crate) PhantomData<T>,
+);
+
+impl<T: Send + Sync + 'static> TypedOutChannels<T> {
+    /// Perform a blocking send on the outcoming channel from `NodeId`.
+    pub fn blocking_send_to(&self, id: &NodeId, content: T) -> Result<(), SendErr> {
+        match self.get(id) {
+            Some(channel) => channel.blocking_lock().blocking_send(Content::new(content)),
+            None => Err(SendErr::NoSuchChannel),
+        }
+    }
+
+    /// Perform a asynchronous send on the outcoming channel from `NodeId`.
+    pub async fn send_to(&self, id: &NodeId, content: T) -> Result<(), SendErr> {
+        match self.get(id) {
+            Some(channel) => channel.lock().await.send(Content::new(content)).await,
+            None => Err(SendErr::NoSuchChannel),
+        }
+    }
+
+    /// Broadcasts the `content` to all the [`TypedOutChannel`]s asynchronously.
+    pub async fn broadcast(&self, content: T) -> Vec<Result<(), SendErr>> {
+        let content = Content::new(content);
+        let futures = self
+            .0
+            .iter()
+            .map(|(_, c)| async { c.lock().await.send(content.clone()).await });
+
+        join_all(futures).await
+    }
+
+    /// Blocking broadcasts the `content` to all the [`TypedOutChannel`]s.
+    pub fn blocking_broadcast(&self, content: T) -> Vec<Result<(), SendErr>> {
+        let content = Content::new(content);
+        self.0
+            .iter()
+            .map(|(_, c)| c.blocking_lock().blocking_send(content.clone()))
+            .collect()
+    }
+
+    /// Close the channel by the given `NodeId`, and remove the channel in this map.
+    pub fn close(&mut self, id: &NodeId) {
+        if let Some(_) = self.get(id) {
+            self.0.remove(id);
+        }
+    }
+
+    pub(crate) fn close_all(&mut self) {
+        self.0.clear();
+    }
+
+    fn get(&self, id: &NodeId) -> Option<Arc<Mutex<OutChannel>>> {
+        match self.0.get(id) {
+            Some(c) => Some(c.clone()),
+            None => None,
+        }
+    }
+
+    pub(crate) fn insert(&mut self, node_id: NodeId, channel: Arc<Mutex<OutChannel>>) {
+        self.0.insert(node_id, channel);
+    }
+}

src/node/mod.rs

@@ -3,3 +3,4 @@ pub mod conditional_node;
 pub mod default_node;
 pub mod id_allocate;
 pub mod node;
+pub mod typed_action;

src/node/node.rs

@@ -44,6 +44,18 @@ pub trait Node: Send + Sync {
     fn loop_structure(&self) -> Option<Vec<Arc<Mutex<dyn Node>>>> {
         None
     }
+
+    /// Returns true if this node has TypedContent input.
+    /// By default, it returns false.
+    fn has_typed_input(&self) -> bool {
+        false
+    }
+
+    /// Returns true if this node has TypedContent output.
+    /// By default, it returns false.
+    fn has_typed_output(&self) -> bool {
+        false
+    }
 }
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy, Ord, PartialOrd)]