Simplify the run.rs scripts of our dataflow examples using xshell

Cargo.toml

@@ -86,7 +86,6 @@ ros2-examples = []
 
 [dev-dependencies]
 eyre = "0.6.8"
-tokio = "1.24.2"
 dora-coordinator = { workspace = true }
 dora-core = { workspace = true }
 dora-tracing = { workspace = true }
@@ -96,7 +95,7 @@ serde_yaml = "0.8.23"
 uuid = { version = "1.7", features = ["v7", "serde"] }
 tracing = "0.1.36"
 futures = "0.3.25"
-tokio-stream = "0.1.11"
+xshell = "0.2.6"
 
 [[example]]
 name = "c-dataflow"

binaries/cli/src/main.rs

@@ -14,7 +14,7 @@ use dora_daemon::Daemon;
 use dora_tracing::set_up_tracing;
 use duration_str::parse;
 use eyre::{bail, Context};
-use std::net::SocketAddr;
+use std::{collections::BTreeSet, net::SocketAddr};
 use std::{
  net::{IpAddr, Ipv4Addr},
  path::PathBuf,
@@ -102,6 +102,8 @@ enum Command {
  dataflow: Option<String>,
  node: String,
  },
+ #[clap(hide = true)]
+ ConnectedMachines,
  // Metrics,
  // Stats,
  // Get,
@@ -273,6 +275,12 @@ fn run() -> eyre::Result<()> {
  bail!("No dora coordinator seems to be running.");
  }
  },
+ Command::ConnectedMachines => match connect_to_coordinator() {
+ Ok(mut session) => connected_machines(&mut *session)?,
+ Err(_) => {
+ bail!("No dora coordinator seems to be running.");
+ }
+ },
  Command::Stop {
  uuid,
  name,
@@ -466,6 +474,37 @@ fn query_running_dataflows(
  Ok(ids)
 }
 
+fn connected_machines(session: &mut TcpRequestReplyConnection) -> Result<(), eyre::ErrReport> {
+ let machines = query_connected_machines(session)?;
+
+ if machines.is_empty() {
+ eprintln!("No machines are connected");
+ } else {
+ for id in machines {
+ println!("{id}");
+ }
+ }
+
+ Ok(())
+}
+
+fn query_connected_machines(
+ session: &mut TcpRequestReplyConnection,
+) -> eyre::Result<BTreeSet<String>> {
+ let reply_raw = session
+ .request(&serde_json::to_vec(&ControlRequest::ConnectedMachines).unwrap())
+ .wrap_err("failed to send connected machines message")?;
+ let reply: ControlRequestReply =
+ serde_json::from_slice(&reply_raw).wrap_err("failed to parse reply")?;
+ let ids = match reply {
+ ControlRequestReply::ConnectedMachines(ids) => ids,
+ ControlRequestReply::Error(err) => bail!("{err}"),
+ other => bail!("unexpected connected machines reply: {other:?}"),
+ };
+
+ Ok(ids)
+}
+
 fn connect_to_coordinator() -> std::io::Result<Box<TcpRequestReplyConnection>> {
  TcpLayer::new().connect(control_socket_addr())
 }

examples/benchmark/run.rs

@@ -1,46 +1,43 @@
-use dora_tracing::set_up_tracing;
-use eyre::{bail, Context};
-use std::path::Path;
+use std::path::{Path, PathBuf};
+use xshell::{cmd, Shell};
 
-#[tokio::main]
-async fn main() -> eyre::Result<()> {
- set_up_tracing("benchmark-runner").wrap_err("failed to set up tracing subscriber")?;
+fn main() -> eyre::Result<()> {
+ // create a new shell in this folder
+ let sh = prepare_shell()?;
+ // build the `dora` binary (you can skip this if you use `cargo install dora-cli`)
+ let dora = prepare_dora_optimized(&sh)?;
 
- let root = Path::new(env!("CARGO_MANIFEST_DIR"));
- std::env::set_current_dir(root.join(file!()).parent().unwrap())
- .wrap_err("failed to set working dir")?;
+ // build the dataflow using `dora build`
+ cmd!(sh, "{dora} build dataflow.yml").run()?;
+
+ // start up the dora daemon and coordinator
+ cmd!(sh, "{dora} up").run()?;
 
- let dataflow = Path::new("dataflow.yml");
- build_dataflow(dataflow).await?;
+ // start running the dataflow.yml
+ cmd!(sh, "{dora} start dataflow.yml --attach").run()?;
 
- run_dataflow(dataflow).await?;
+ // stop the dora daemon and coordinator again
+ cmd!(sh, "{dora} destroy").run()?;
 
  Ok(())
 }
 
-async fn build_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("build").arg(dataflow);
- if !cmd.status().await?.success() {
- bail!("failed to build dataflow");
- };
- Ok(())
+/// Prepares a shell and set the working directory to the parent folder of this file.
+///
+/// You can use your system shell instead (e.g. `bash`);
+fn prepare_shell() -> Result<Shell, eyre::Error> {
+ let sh = Shell::new()?;
+ let root = Path::new(env!("CARGO_MANIFEST_DIR"));
+ sh.change_dir(root.join(file!()).parent().unwrap());
+ Ok(sh)
 }
 
-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(())
+/// Build the `dora` command-line executable from this repo.
+///
+/// You can skip this step and run `cargo install dora-cli --locked` instead.
+fn prepare_dora_optimized(sh: &Shell) -> eyre::Result<PathBuf> {
+ cmd!(sh, "cargo build --package dora-cli --release").run()?;
+ let root = Path::new(env!("CARGO_MANIFEST_DIR"));
+ let dora = root.join("target").join("release").join("dora");
+ Ok(dora)
 }