Initial commit

ntw_flowgraph/src/graph.rs

@@ -0,0 +1,86 @@
+use petgraph::graph::DiGraph;
+use petgraph::graph::NodeIndex;
+
+use crate::Block;
+
+pub struct Graph
+{
+    blocks: Vec<Box<dyn Block>>,
+}
+
+impl Graph
+{
+    pub fn new() -> Graph
+    {
+        Graph { blocks: vec![] }
+    }
+
+    pub fn add_block(&mut self, block: impl Block + 'static)
+    {
+        block.set_block_index(self.blocks.len());
+        self.blocks.push(Box::new(block));
+    }
+
+    pub fn run(&mut self)
+    {
+        // Compute the topo_order
+        let mut digraph = DiGraph::<(), (), usize>::with_capacity(self.blocks.len(), 1);
+
+        for block in self.blocks.iter()
+        {
+            let node = digraph.add_node(());
+
+            for next in block.get_successors()
+            {
+                digraph.add_edge(node, next.into(), ());
+            }
+        }
+
+        let topo_order = petgraph::algo::toposort(&digraph, None)
+            .expect("Graph cannot be cylic")
+            .into_iter()
+            .map(|x| x.index())
+            .collect::<Vec<_>>();
+
+        // Dumbass round robin
+        loop
+        {
+            let mut one_ready = false;
+            for x in topo_order.iter()
+            {
+                let block = &mut self.blocks[*x];
+                if block.ready()
+                {
+                    block.work();
+                    one_ready = true;
+                }
+            }
+
+            if !one_ready
+            {
+                break;
+            }
+        }
+    }
+
+    pub fn print_deps(&self)
+    {
+        for (i, e) in self.blocks.iter().enumerate()
+        {
+            print!("{}: ", i);
+            for x in e.get_successors()
+            {
+                print!("{}, ", x);
+            }
+            println!();
+        }
+    }
+}
+
+impl Default for Graph
+{
+    fn default() -> Self
+    {
+        Self::new()
+    }
+}

ntw_flowgraph/src/inout.rs

@@ -0,0 +1,65 @@
+use ringbuf::HeapRb;
+use ringbuf::SharedRb;
+use ringbuf::storage::Heap;
+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;
+
+// Represent a block input, of which data is popped
+pub struct In<T>
+{
+    block: Rc<Cell<Option<usize>>>,
+    pub rb: Caching<Arc<SharedRb<Heap<T>>>, false, true>,
+}
+
+// Represent a block input, in which data is pushed
+pub struct Out<T>
+{
+    to: Rc<Cell<Option<usize>>>,
+    pub rb: Caching<Arc<SharedRb<Heap<T>>>, true, false>,
+}
+
+pub struct Stream;
+
+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));
+        (
+            Out {
+                to: to.clone(),
+                rb: prod,
+            },
+            In {
+                block: to.clone(),
+                rb: cons,
+            },
+        )
+    }
+}
+
+impl<T> In<T>
+{
+    pub fn set_index(&self, index: usize)
+    {
+        self.block.set(Some(index))
+    }
+}
+
+impl<T> Out<T>
+{
+    pub fn get_successor(&self) -> Option<usize>
+    {
+        self.to.get()
+    }
+
+    pub fn try_push(&mut self, data: T) -> Result<(), T>
+    {
+        self.rb.try_push(data)
+    }
+}

ntw_flowgraph/src/lib.rs

@@ -0,0 +1,14 @@
+pub mod graph;
+pub mod inout;
+
+pub trait BlockWork
+{
+    fn work(&mut self);
+    fn ready(&self) -> bool;
+}
+
+pub trait Block: BlockWork
+{
+    fn set_block_index(&self, index: usize);
+    fn get_successors(&self) -> Vec<usize>;
+}