Use cli to spawn daemon run-dataflow to avoid confused exe path

examples/benchmark/run.rs

@@ -13,7 +13,7 @@ async fn main() -> eyre::Result<()> {
     let dataflow = Path::new("dataflow.yml");
     build_dataflow(dataflow).await?;
 
-    dora_daemon::Daemon::run_dataflow(dataflow).await?;
+    run_dataflow(dataflow).await?;
 
     Ok(())
 }
@@ -29,3 +29,18 @@ async fn build_dataflow(dataflow: &Path) -> eyre::Result<()> {
     };
     Ok(())
 }
+
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}

examples/c++-dataflow/run.rs

@@ -112,7 +112,7 @@ async fn main() -> eyre::Result<()> {
 
     let dataflow = Path::new("dataflow.yml").to_owned();
     build_package("dora-runtime").await?;
-    dora_daemon::Daemon::run_dataflow(&dataflow).await?;
+    run_dataflow(&dataflow).await?;
 
     Ok(())
 }
@@ -128,6 +128,21 @@ async fn build_package(package: &str) -> eyre::Result<()> {
     Ok(())
 }
 
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}
+
 async fn build_cxx_node(
     root: &Path,
     paths: &[&Path],

examples/c-dataflow/run.rs

@@ -23,7 +23,7 @@ async fn main() -> eyre::Result<()> {
     build_c_operator(root).await?;
 
     let dataflow = Path::new("dataflow.yml").to_owned();
-    dora_daemon::Daemon::run_dataflow(&dataflow).await?;
+    run_dataflow(&dataflow).await?;
 
     Ok(())
 }
@@ -39,6 +39,21 @@ async fn build_package(package: &str) -> eyre::Result<()> {
     Ok(())
 }
 
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}
+
 async fn build_c_node(root: &Path, name: &str, out_name: &str) -> eyre::Result<()> {
     let mut clang = tokio::process::Command::new("clang");
     clang.arg(name);

examples/cmake-dataflow/run.rs

@@ -40,7 +40,7 @@ async fn main() -> eyre::Result<()> {
 
     let dataflow = Path::new("dataflow.yml").to_owned();
     build_package("dora-runtime").await?;
-    dora_daemon::Daemon::run_dataflow(&dataflow).await?;
+    run_dataflow(&dataflow).await?;
 
     Ok(())
 }
@@ -55,3 +55,18 @@ async fn build_package(package: &str) -> eyre::Result<()> {
     }
     Ok(())
 }
+
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}

examples/multiple-daemons/run.rs

@@ -37,8 +37,8 @@ async fn main() -> eyre::Result<()> {
     let (coordinator_port, coordinator) =
         dora_coordinator::start(Some(0), ReceiverStream::new(coordinator_events_rx)).await?;
     let coordinator_addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), coordinator_port);
-    let daemon_a = dora_daemon::Daemon::run(coordinator_addr, "A".into());
-    let daemon_b = dora_daemon::Daemon::run(coordinator_addr, "B".into());
+    let daemon_a = run_dataflow(coordinator_addr.to_string(), "A".into(), dataflow);
+    let daemon_b = run_dataflow(coordinator_addr.to_string(), "B".into(), dataflow);
 
     tracing::info!("Spawning coordinator and daemons");
     let mut tasks = JoinSet::new();
@@ -196,3 +196,22 @@ async fn build_dataflow(dataflow: &Path) -> eyre::Result<()> {
     };
     Ok(())
 }
+
+async fn run_dataflow(coordinator: String, machine_id: &str, dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow)
+        .arg("--machine-id")
+        .arg(machine_id)
+        .arg("--coordinator-addr")
+        .arg(coordinator);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}

examples/python-dataflow/run.rs

@@ -1,7 +1,7 @@
 use dora_core::{get_pip_path, get_python_path, run};
 use dora_download::download_file;
 use dora_tracing::set_up_tracing;
-use eyre::{ContextCompat, WrapErr};
+use eyre::{bail, ContextCompat, WrapErr};
 use std::path::Path;
 
 #[tokio::main]
@@ -81,7 +81,22 @@ async fn main() -> eyre::Result<()> {
     .context("Could not download weights.")?;
 
     let dataflow = Path::new("dataflow.yml");
-    dora_daemon::Daemon::run_dataflow(dataflow).await?;
+    run_dataflow(dataflow).await?;
 
     Ok(())
 }
+
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}

examples/python-operator-dataflow/run.rs

@@ -1,7 +1,7 @@
 use dora_core::{get_pip_path, get_python_path, run};
 use dora_download::download_file;
 use dora_tracing::set_up_tracing;
-use eyre::{ContextCompat, WrapErr};
+use eyre::{bail, ContextCompat, WrapErr};
 use std::path::Path;
 
 #[tokio::main]
@@ -81,7 +81,22 @@ async fn main() -> eyre::Result<()> {
     .await
     .context("Could not download weights.")?;
     let dataflow = Path::new("dataflow.yml");
-    dora_daemon::Daemon::run_dataflow(dataflow).await?;
+    run_dataflow(dataflow).await?;
 
     Ok(())
 }
+
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}

examples/python-ros2-dataflow/run.rs

@@ -74,7 +74,22 @@ async fn main() -> eyre::Result<()> {
     .context("maturin develop failed")?;
 
     let dataflow = Path::new("dataflow.yml");
-    dora_daemon::Daemon::run_dataflow(dataflow).await?;
+    run_dataflow(dataflow).await?;
 
     Ok(())
 }
+
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}

examples/rust-dataflow-url/run.rs

@@ -13,7 +13,7 @@ async fn main() -> eyre::Result<()> {
     let dataflow = Path::new("dataflow.yml");
     build_dataflow(dataflow).await?;
 
-    dora_daemon::Daemon::run_dataflow(dataflow).await?;
+    run_dataflow(dataflow).await?;
 
     Ok(())
 }
@@ -29,3 +29,18 @@ async fn build_dataflow(dataflow: &Path) -> eyre::Result<()> {
     };
     Ok(())
 }
+
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}

examples/rust-dataflow/run.rs

@@ -13,7 +13,7 @@ async fn main() -> eyre::Result<()> {
     let dataflow = Path::new("dataflow.yml");
     build_dataflow(dataflow).await?;
 
-    dora_daemon::Daemon::run_dataflow(dataflow).await?;
+    run_dataflow(dataflow).await?;
 
     Ok(())
 }
@@ -29,3 +29,18 @@ async fn build_dataflow(dataflow: &Path) -> eyre::Result<()> {
     };
     Ok(())
 }
+
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}

examples/rust-ros2-dataflow/run.rs

@@ -13,7 +13,7 @@ async fn main() -> eyre::Result<()> {
     let dataflow = Path::new("dataflow.yml");
     build_dataflow(dataflow).await?;
 
-    dora_daemon::Daemon::run_dataflow(dataflow).await?;
+    run_dataflow(dataflow).await?;
 
     Ok(())
 }
@@ -29,3 +29,18 @@ async fn build_dataflow(dataflow: &Path) -> eyre::Result<()> {
     };
     Ok(())
 }
+
+async fn run_dataflow(dataflow: &Path) -> eyre::Result<()> {
+    let cargo = std::env::var("CARGO").unwrap();
+    let mut cmd = tokio::process::Command::new(&cargo);
+    cmd.arg("run");
+    cmd.arg("--package").arg("dora-cli");
+    cmd.arg("--")
+        .arg("daemon")
+        .arg("--run-dataflow")
+        .arg(dataflow);
+    if !cmd.status().await?.success() {
+        bail!("failed to run dataflow");
+    };
+    Ok(())
+}