Bunch'o things

ntw_flowgraph/src/graph.rs

@@ -1,9 +1,25 @@
-use crate::{Block, BlockWork, RunnableBlock};
+use std::thread::JoinHandle;
+
+use crate::Block;
+use crate::BlockWork;
+use crate::RunnableBlock;
+use petgraph::dot::Dot;
 use petgraph::graph::DiGraph;
 
+#[macro_export]
+macro_rules! graph {
+    ($( $b:ident ),* $(,)?) => {{
+        let mut g = Graph::new();
+        $(
+            g.add_block($b);
+        )*
+        g
+    }};
+}
+
 pub struct Graph
 {
-    blocks: Vec<Box<dyn RunnableBlock>>,
+    blocks: Vec<Box<dyn RunnableBlock + Send>>,
 }
 
 impl Graph
@@ -13,13 +29,13 @@ impl Graph
         Graph { blocks: vec![] }
     }
 
-    pub fn add_block(&mut self, block: impl Block + BlockWork + 'static)
+    pub fn add_block(&mut self, block: impl Block + BlockWork + Send + 'static)
     {
         block.set_block_index(self.blocks.len());
         self.blocks.push(Box::new(block));
     }
 
-    pub fn run(&mut self)
+    pub fn run(mut self) -> JoinHandle<()>
     {
         // Compute the topo_order
         let mut digraph = DiGraph::<(), (), usize>::with_capacity(self.blocks.len(), 1);
@@ -43,24 +59,26 @@ impl Graph
             .collect::<Vec<_>>();
 
         // Dumbass round robin
-        loop
-        {
-            let mut one_ready = false;
-            for x in topo_order.iter()
+        std::thread::spawn(move || {
+            loop
             {
-                let block = &mut self.blocks[*x];
-                if block.ready()
+                let mut one_ready = false;
+                for x in topo_order.iter()
                 {
-                    block.work();
-                    one_ready = true;
+                    let block = &mut self.blocks[*x];
+                    if block.ready()
+                    {
+                        block.work();
+                        one_ready = true;
+                    }
+                }
+
+                if !one_ready
+                {
+                    break;
                 }
             }
-
-            if !one_ready
-            {
-                break;
-            }
-        }
+        })
     }
 
     pub fn print_deps(&self)
@@ -75,6 +93,25 @@ impl Graph
             println!();
         }
     }
+
+    pub fn get_dot(&self) -> String
+    {
+        let mut digraph = DiGraph::<String, (), usize>::with_capacity(self.blocks.len(), 1);
+        let len = self.blocks.len();
+        (0..len).for_each(|i| {
+            digraph.add_node(self.blocks[i].get_block_name().to_owned());
+        });
+
+        for (i, block) in self.blocks.iter().enumerate()
+        {
+            for next in block.get_successors()
+            {
+                digraph.add_edge(i.into(), next.into(), ());
+            }
+        }
+
+        format!("{:?}", Dot::with_config(&digraph, &[]))
+    }
 }
 
 impl Default for Graph

ntw_flowgraph/src/inout.rs

@@ -7,21 +7,20 @@ use ringbuf::traits::Observer;
 use ringbuf::traits::Producer;
 use ringbuf::traits::Split;
 use ringbuf::wrap::caching::Caching;
-use std::cell::Cell;
-use std::rc::Rc;
 use std::sync::Arc;
+use std::sync::Mutex;
 
 // Represent a block input, of which data is popped
 pub struct In<T>
 {
-    block: Rc<Cell<Option<usize>>>,
+    block: Arc<Mutex<Option<usize>>>,
     pub rb: Caching<Arc<SharedRb<Heap<T>>>, false, true>,
 }
 
 // Represent a block output, in which data is pushed
 pub struct Out<T>
 {
-    to: Rc<Cell<Option<usize>>>,
+    to: Arc<Mutex<Option<usize>>>,
     pub rb: Caching<Arc<SharedRb<Heap<T>>>, true, false>,
 }
 
@@ -32,7 +31,7 @@ impl Stream
     pub fn make<T>(length: usize) -> (Out<T>, In<T>)
     {
         let (prod, cons) = HeapRb::<T>::new(length).split();
-        let to = Rc::new(Cell::new(None));
+        let to = Arc::new(Mutex::new(None));
         (
             Out {
                 to: to.clone(),
@@ -50,7 +49,8 @@ impl<T> In<T>
 {
     pub fn set_index(&self, index: usize)
     {
-        self.block.set(Some(index))
+        let mut guard = self.block.lock().unwrap();
+        *guard = Some(index);
     }
 
     pub fn try_pop(&mut self) -> Option<T>
@@ -73,7 +73,7 @@ impl<T> Out<T>
 {
     pub fn get_successor(&self) -> Option<usize>
     {
-        self.to.get()
+        *self.to.lock().unwrap()
     }
 
     pub fn try_push(&mut self, data: T) -> Result<(), T>

ntw_flowgraph/src/lib.rs

@@ -1,9 +1,16 @@
 pub mod graph;
 pub mod inout;
 
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub enum BlockResult
+{
+    Ok,
+    Finished,
+}
+
 pub trait BlockWork
 {
-    fn work(&mut self);
+    fn work(&mut self) -> BlockResult;
     fn ready(&mut self) -> bool;
 }
 
@@ -11,6 +18,9 @@ pub trait Block
 {
     fn set_block_index(&self, index: usize);
     fn get_successors(&self) -> Vec<usize>;
+
+    // Meta functions
+    fn get_block_name(&self) -> &'static str;
 }
 
 pub trait RunnableBlock: Block + BlockWork {}