WIP: add dora dynamic nodes, wip on RTC crate

Cargo.toml

@@ -35,6 +35,7 @@ members = [
     "crates/spinlock",
     "crates/timer_list",
     "crates/tuple_for_each",
+    "crates/rtc",
 
     "modules/axalloc",
     "modules/axconfig",
@@ -75,8 +76,8 @@ members = [
     "apps/std/tokio-echo",
     "apps/std/socket2-echoserver",
     "apps/std/chrono",
-    "apps/std/dora/node",
-    "apps/std/dora/sink",
+    "apps/std/dora/sink-dynamic",
+    "apps/std/dora/node-dynamic",
 ]
 
 [workspace.dependencies]

apps/std/dora/Readme.md

@@ -1,68 +1,59 @@
 # Dora on Arceos
 
 * Dora
-	* For host: https://github.com/arceos-org/dora/tree/modify_spawn
+	* For host: https://github.com/arceos-org/dora/tree/multi_machine
 	* For ArceOS: https://github.com/arceos-org/dora/tree/arceos-porting
+* Dora-benchmark
+	* https://github.com/arceos-org/dora-benchmark/tree/rs-latency-dynamic
 * ArceOS
 	* https://github.com/arceos-org/arceos/tree/dora-wip
 
-## 1. Setup Dora coordinator and daemon on host
+## 1. Compile and setup Dora on host
 
-* In dora repo under `modify_spawn` branch
+* In dora repo under `multi_machine` branch
 
 ```bash
 # Build.
 cargo build --release --all
 
-# Run coordinator.
-./target/release/dora coordinator
+# Dora up (coordinator & daemon).
+./target/release/dora up
+```
 
-# Run daemon.
-DORA_NODE_CONFIG_PATH=`pwd` ./target/release/dora daemon
+You should see output like:
+```bash
+started dora coordinator
+started dora daemon
 ```
 
-## 2. Generate configuration file for Dora nodes.
+## 2. Start Dora based on `dataflow-dynamic.yml`.
 
-* In dora-benchmark repo:  `/dora-benchmark/dora-rs/rs-latency`
+* In dora-benchmark repo:  `/dora-benchmark/dora-rs/rs-latency-dynamic`
 ```bash
-./PATH/TO/dora/you/just/compiled start dataflow.yml
+./PATH/TO/dora/you/just/compiled start dataflow-dynamic.yml
 ```
 
-* then you can get `rust-node.yml` and `rust-sink.yml` on dora repo.
+## 3. Boot Dora nodes upon ArceOS or in host.
 
-## 3. Boot Dora nodes based on these configuration file.
+For example, we start sink in host and node upon ArceOS.
 
-### To boot Dora nodes on host
+* In dora-benchmark repo, start sink node:
 ```bash
-# Build.
-cargo build --release --all
-
-# Run nodes.
-DORA_NODE_CONFIG=`cat rust-node.yml` ./target/release/benchmark-example-node
-DORA_NODE_CONFIG=`cat rust-sink.yml` ./target/release/benchmark-example-sink
+./target/release/benchmark-example-sink-dynamic
 ```
 
-### To boot Dora nodes upon ArceOS
-
-In ArceOS dir under `dora-wip` branch.
+* In ArceOS dir:
+```bash
+make A=apps/std/dora/node-dynamic SMP=1 NET=y BLK=y LOG=debug STD=y justrun
+```
 
 Pay attention to `dora-node-api` dependency,
 this is just the `arceos-porting` branch of Dora repo,
+it's a simplified and WIP version of Dora as a dependency of ArceOS std.
 We have separated their dirs to facilitate the compilation of Dora in the host and Dora dependencies for ArceOS std.
 
 ```Toml
 [workspace.dependencies]
 # dora-node-api = "0.3.5" # { path = "../../../dora/apis/rust/node" }
 dora-node-api = { path = "../../Dora/dora-arceos/apis/rust/node" }
 ```
-
-* Run ArceOS on QEMU/KVM
-	* make `disk.img` for ArceOS, make sure it contains the configuration files just generated for the Dora nodes.
-	* change `socket_addr` to : `socket_addr: "10.0.2.2:44631"` in configuration files.
-
-```bash
-# make sure sink.yml in your disk.img
-make A=apps/std/dora/sink SMP=1 NET=y BLK=y LOG=debug STD=y build
-# make sure node.yml in your disk.img
-make A=apps/std/dora/node SMP=1 NET=y BLK=y LOG=debug STD=y build
-```

apps/std/dora/dataflow.yml

@@ -0,0 +1,17 @@
+nodes:
+  - id: rust-node
+    custom:
+      build: cargo build -p benchmark-example-node --release
+      source: ./target/release/benchmark-example-node
+      inputs:
+        tick: dora/timer/millis/20
+      outputs:
+        - latency
+        - throughput
+  - id: rust-sink
+    custom:
+      build: cargo build -p benchmark-example-sink --release
+      source: ./target/release/benchmark-example-sink
+      inputs:
+        latency: rust-node/latency
+        throughput: rust-node/throughput

apps/std/dora/node/Cargo.toml → apps/std/dora/node-dynamic/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "benchmark-example-node"
+name = "dora-arceos-example-node"
 version = "0.1.0"
 edition = "2021"
 
@@ -13,6 +13,7 @@ rand = "0.8.5"
 tokio = { version = "1.20.1", features = ["rt", "macros"] }
 uhlc = "0.5.1"
 bytemuck = "1.12"
+chrono = "0.4.38"
 
 [package.metadata.arceos]
-features = ["net", "multitask", "fs"]
+features = ["net", "multitask", "fs", "irq", "sched_rr"]

apps/std/dora/node/src/main.rs → apps/std/dora/node-dynamic/src/main.rs

@@ -1,7 +1,9 @@
+use dora_node_api::dora_core::config::{DataId, NodeId};
 use dora_node_api::Event;
-use dora_node_api::{self, arrow::array::UInt64Array, dora_core::config::DataId, DoraNode};
+use dora_node_api::{self, arrow::array::UInt64Array, DoraNode};
 use rand::Rng;
 // use std::time::Duration;
+use std::net::{IpAddr, Ipv4Addr};
 use uhlc::system_time_clock;
 
 #[no_mangle]
@@ -14,21 +16,31 @@ pub extern "C" fn sqrt() {
     println!("sqrt");
 }
 
+static REMOTE_IP: IpAddr = IpAddr::V4(Ipv4Addr::new(10, 0, 2, 2));
+
 fn main() -> eyre::Result<()> {
     ceil();
     sqrt();
+    println!(
+        "Dora node-dynamic on ArceOS booted at time {}",
+        chrono::Local::now()
+    );
+
     let latency = DataId::from("latency".to_owned());
     let _throughput = DataId::from("throughput".to_owned());
 
-    let (mut node, mut events) = DoraNode::init_from_env()?;
-    // let (mut node, mut events) = DoraNode::init_from_file("node.yml")?;
-    let sizes = [1, 10 * 512, 100 * 512, 1000 * 512, 10000 * 512];
+    let (mut node, mut events) = DoraNode::init_from_node_id(
+        NodeId::from("rust-node-dynamic".to_string()),
+        Some(REMOTE_IP),
+    )?;
+
+    let sizes = [1, 10 * 512, 100 * 512];
 
     // test latency first
     for size in sizes {
         for i in 0..100 {
             if let Some(event) = events.recv() {
-                println!("node recv event[{}] {:#?}", i, event);
+                // println!("node recv event[{}] {:#?}", i, event);
                 match event {
                     Event::Input {
                         id: _,
@@ -45,8 +57,6 @@ fn main() -> eyre::Result<()> {
 
                         let random_data: UInt64Array = random_data.into();
 
-                        println!("node send_output random_data[{}] {:#?}", i, random_data);
-
                         node.send_output(latency.clone(), metadata.parameters, random_data)?;
                     }
                     _ => {}

apps/std/dora/sink/Cargo.toml → apps/std/dora/sink-dynamic/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "benchmark-example-sink"
+name = "sink-dynamic"
 version = "0.1.0"
 edition = "2021"
 
@@ -12,6 +12,7 @@ rand = "0.8.5"
 uhlc = "0.5.1"
 bytemuck = "1.12"
 csv = "1.1.6"
+chrono = "0.4.38"
 
 [package.metadata.arceos]
-features = ["net", "multitask", "fs"]
+features = ["net", "multitask", "fs", "irq", "sched_rr"]

apps/std/dora/sink/src/main.rs → apps/std/dora/sink-dynamic/src/main.rs

@@ -1,17 +1,21 @@
 use csv::Writer;
 use dora_node_api::arrow::array::{AsArray, PrimitiveArray};
 use dora_node_api::arrow::datatypes::UInt64Type;
+use dora_node_api::dora_core::config::NodeId;
 use dora_node_api::{self, DoraNode, Event};
 use eyre::ContextCompat;
 use std::collections::HashMap;
 use std::time::{Duration, Instant};
+use std::net::{IpAddr, Ipv4Addr};
 use uhlc::system_time_clock;
 use uhlc::HLC;
 
 static LANGUAGE: &str = "Rust";
-static PLATFORM: &str = "[email protected]";
+static PLATFORM: &str = "ArceOS [email protected]";
 static NAME: &str = "dora-rs daemon Rust";
 
+static REMOTE_IP: IpAddr = IpAddr::V4(Ipv4Addr::new(10, 0, 2, 2));
+
 #[no_mangle]
 pub extern "C" fn ceil() {
     println!("ceil");
@@ -23,8 +27,15 @@ pub extern "C" fn sqrt() {
 }
 
 fn main() -> eyre::Result<()> {
-    // let (_node, mut events) = DoraNode::init_from_env()?;
-    let ( _node, mut events) = DoraNode::init_from_file("sink.yml")?;
+    println!(
+        "Dora sink-dynamic on ArceOS booted at time {}",
+        chrono::Local::now()
+    );
+
+    let (_node, mut events) = DoraNode::init_from_node_id(
+        NodeId::from("rust-sink-dynamic".to_string()),
+        Some(REMOTE_IP),
+    )?;
 
     // latency is tested first
     let latency = true;
@@ -41,14 +52,13 @@ fn main() -> eyre::Result<()> {
         .expect("Could not extract date from timestamp.");
 
     // Preallocated vector
-    let sizes = [1, 10 * 512, 100 * 512, 1000 * 512, 10000 * 512];
+    let sizes = [1, 10 * 512, 100 * 512];
     let mut root_vec = HashMap::new();
     for size in sizes {
         root_vec.insert(size, vec![0u64; size]);
     }
 
     while let Some(event) = events.recv() {
-        println!("sink recv event[{}] {:#?}", n, event);
         match event {
             Event::Input {
                 id: _,
@@ -123,15 +133,4 @@ fn record_results(
         avg_latency.as_micros().to_string(),
     ])
     .unwrap();
-    println!(
-        "{:#?}",
-        &[
-            date.to_string(),
-            LANGUAGE.to_string(),
-            PLATFORM.to_string(),
-            NAME.to_string(),
-            current_size.to_string(),
-            avg_latency.as_micros().to_string(),
-        ]
-    );
 }