Add support for distributed deployments with multiple daemons

.github/workflows/ci.yml

@@ -52,6 +52,10 @@ jobs:
         timeout-minutes: 30
         run: cargo run --example rust-dataflow
 
+      - name: "Multiple Daemons example"
+        timeout-minutes: 30
+        run: cargo run --example multiple-daemons
+
       - name: "Benchmark example"
         timeout-minutes: 30
         run: cargo run --example benchmark --release
@@ -82,7 +86,6 @@ jobs:
         run: |
           cargo install --path binaries/coordinator
           cargo install --path binaries/daemon
-          cargo install --path binaries/runtime
           cargo install --path binaries/cli
 
       - name: "Test CLI"

.github/workflows/release.yml

@@ -26,9 +26,8 @@ jobs:
 
       - name: "Build binaries"
         timeout-minutes: 60
-        run: "cargo build --release -p dora-runtime 
-         -p dora-coordinator -p dora-cli -p dora-daemon"
-
+        run: "cargo build --release -p dora-coordinator -p dora-cli -p dora-daemon"
+
       - name: "Publish packages on `crates.io`"
         if: runner.os == 'Linux'
         run: |
@@ -65,14 +64,11 @@ jobs:
             cargo publish -p dora-cli --token ${{ secrets.CARGO_REGISTRY_TOKEN }}
             cargo publish -p dora-coordinator --token ${{ secrets.CARGO_REGISTRY_TOKEN }}
             cargo publish -p dora-daemon --token ${{ secrets.CARGO_REGISTRY_TOKEN }}
-            cargo publish -p dora-runtime --token ${{ secrets.CARGO_REGISTRY_TOKEN }}
-
 
       - name: "Create Archive (Unix)"
         if: runner.os == 'Linux' || runner.os == 'macOS'
         run: |
           mkdir archive
-          cp target/release/dora-runtime archive
           cp target/release/dora-coordinator archive
           cp target/release/dora-daemon archive
           cp target/release/dora-cli archive/dora
@@ -85,7 +81,6 @@ jobs:
         shell: powershell
         run: |
           New-Item -Path archive -ItemType Directory
-          Copy-Item target/release/dora-runtime.exe -Destination archive
           Copy-Item target/release/dora-coordinator.exe -Destination archive
           Copy-Item target/release/dora-daemon.exe -Destination archive
           Copy-Item target/release/dora-cli.exe -Destination archive/dora.exe

Cargo.toml

@@ -13,6 +13,7 @@ members = [
     "binaries/runtime",
     "examples/rust-dataflow/*",
     "examples/benchmark/*",
+    "examples/multiple-daemons/*",
     "libraries/communication-layer/*",
     "libraries/core",
     "libraries/message",
@@ -46,6 +47,7 @@ communication-layer-request-reply = { version = "0.2.2", path = "libraries/commu
 dora-message = { version = "0.2.2", path = "libraries/message" }
 dora-runtime = { version = "0.2.2", path = "binaries/runtime" }
 dora-daemon = { version = "0.2.2", path = "binaries/daemon" }
+dora-coordinator = { version = "0.2.2", path = "binaries/coordinator" }
 
 [package]
 name = "dora-examples"
@@ -58,12 +60,16 @@ license = "Apache-2.0"
 eyre = "0.6.8"
 tokio = "1.24.2"
 dora-daemon = { workspace = true }
+dora-coordinator = { workspace = true }
 dora-core = { workspace = true }
 dunce = "1.0.2"
 serde_yaml = "0.8.23"
 uuid = { version = "1.2.1", features = ["v4", "serde"] }
 tracing = "0.1.36"
 tracing-subscriber = "0.3.15"
+futures = "0.3.25"
+tokio-stream = "0.1.11"
+clap = { version = "4.0.3", features = ["derive"] }
 
 [[example]]
 name = "c-dataflow"
@@ -92,3 +98,7 @@ path = "examples/python-operator-dataflow/run.rs"
 [[example]]
 name = "benchmark"
 path = "examples/benchmark/run.rs"
+
+[[example]]
+name = "multiple-daemons"
+path = "examples/multiple-daemons/run.rs"

README.md

@@ -94,10 +94,6 @@ You need to add the created operators/nodes to your dataflow YAML file.
 8. You can also download already implemented operators by putting links in the dataflow. This example will launch a webcam plot stream. 
 
 ```yaml
-communication:
-  zenoh:
-    prefix: abc_project
-
 nodes:
   - id: op_1
     operator:

binaries/cli/src/attach.rs

@@ -23,7 +23,7 @@ pub fn attach_dataflow(
     // Generate path hashmap
     let mut node_path_lookup = HashMap::new();
 
-    let nodes = dataflow.resolve_aliases();
+    let nodes = dataflow.resolve_aliases_and_set_defaults();
 
     let working_dir = dataflow_path
         .canonicalize()

binaries/cli/src/main.rs

@@ -33,8 +33,6 @@ enum Command {
     Check {
         #[clap(long)]
         dataflow: Option<PathBuf>,
-        #[clap(long)]
-        runtime_path: Option<PathBuf>,
     },
     /// Generate a visualization of the given graph using mermaid.js. Use --open to open browser.
     Graph {
@@ -132,12 +130,15 @@ fn run() -> eyre::Result<()> {
     let args = Args::parse();
 
     match args.command {
-        Command::Check {
-            dataflow,
-            runtime_path,
-        } => match dataflow {
+        Command::Check { dataflow } => match dataflow {
             Some(dataflow) => {
-                Descriptor::blocking_read(&dataflow)?.check(&dataflow, runtime_path)?;
+                let working_dir = dataflow
+                    .canonicalize()
+                    .context("failed to canonicalize dataflow path")?
+                    .parent()
+                    .ok_or_else(|| eyre::eyre!("dataflow path has no parent dir"))?
+                    .to_owned();
+                Descriptor::blocking_read(&dataflow)?.check(&working_dir)?;
                 check::check_environment()?
             }
             None => check::check_environment()?,
@@ -171,18 +172,29 @@ fn run() -> eyre::Result<()> {
             attach,
             hot_reload,
         } => {
-            let dataflow_description =
+            let dataflow_descriptor =
                 Descriptor::blocking_read(&dataflow).wrap_err("Failed to read yaml dataflow")?;
-            dataflow_description
-                .check(&dataflow, None)
+            let working_dir = dataflow
+                .canonicalize()
+                .context("failed to canonicalize dataflow path")?
+                .parent()
+                .ok_or_else(|| eyre::eyre!("dataflow path has no parent dir"))?
+                .to_owned();
+            dataflow_descriptor
+                .check(&working_dir)
                 .wrap_err("Could not validate yaml")?;
             let mut session =
                 connect_to_coordinator().wrap_err("failed to connect to dora coordinator")?;
-            let dataflow_id = start_dataflow(dataflow.clone(), name, &mut *session)?;
+            let dataflow_id = start_dataflow(
+                dataflow_descriptor.clone(),
+                name,
+                working_dir,
+                &mut *session,
+            )?;
 
             if attach {
                 attach_dataflow(
-                    dataflow_description,
+                    dataflow_descriptor,
                     dataflow,
                     dataflow_id,
                     &mut *session,
@@ -212,18 +224,17 @@ fn run() -> eyre::Result<()> {
 }
 
 fn start_dataflow(
-    dataflow: PathBuf,
+    dataflow: Descriptor,
     name: Option<String>,
+    local_working_dir: PathBuf,
     session: &mut TcpRequestReplyConnection,
 ) -> Result<Uuid, eyre::ErrReport> {
-    let canonicalized = dataflow
-        .canonicalize()
-        .wrap_err("given dataflow file does not exist")?;
     let reply_raw = session
         .request(
             &serde_json::to_vec(&ControlRequest::Start {
-                dataflow_path: canonicalized,
+                dataflow,
                 name,
+                local_working_dir,
             })
             .unwrap(),
         )

binaries/cli/src/template/c/dataflow-template.yml

@@ -1,7 +1,3 @@
-communication:
-  zenoh:
-    prefix: ___name___
-
 nodes:
   - id: op_1
     operator:

binaries/cli/src/template/cxx/dataflow-template.yml

@@ -1,7 +1,3 @@
-communication:
-  zenoh:
-    prefix: ___name___
-
 nodes:
   - id: runtime-node_1
     operators:

binaries/cli/src/template/python/dataflow-template.yml

@@ -1,7 +1,3 @@
-communication:
-  zenoh:
-    prefix: ___name___
-
 nodes:
   - id: op_1
     operator:

binaries/cli/src/template/rust/dataflow-template.yml

@@ -1,24 +1,20 @@
-communication:
-  zenoh:
-    prefix: ___name___
-
 nodes:
   - id: op_1
     operator:
-        build: cargo build -p op_1
-        shared-library: target/debug/op_1
-        inputs:
-          tick: dora/timer/millis/100
-        outputs:
-          - some-output
+      build: cargo build -p op_1
+      shared-library: target/debug/op_1
+      inputs:
+        tick: dora/timer/millis/100
+      outputs:
+        - some-output
   - id: op_2
     operator:
-        build: cargo build -p op_2
-        shared-library: target/debug/op_2
-        inputs:
-          tick: dora/timer/secs/2
-        outputs:
-          - some-output
+      build: cargo build -p op_2
+      shared-library: target/debug/op_2
+      inputs:
+        tick: dora/timer/secs/2
+      outputs:
+        - some-output
 
   - id: custom-node_1
     custom:

binaries/coordinator/Cargo.toml

@@ -30,3 +30,4 @@ serde_json = "1.0.86"
 which = "4.3.0"
 thiserror = "1.0.37"
 ctrlc = "3.2.5"
+clap = { version = "3.1.8", features = ["derive"] }