refactor(hydro_deploy)!: end-to-end flamegraph generation, fix #1365

hydro_deploy/core/Cargo.toml

@@ -22,6 +22,7 @@ futures = "0.3.26"
 futures-core = "0.3.26"
 hydroflow_cli_integration = { path = "../hydroflow_cli_integration", version = "^0.5.2" }
 indicatif = "0.17.8"
+inferno = "0.11.20"
 memo-map = "0.3.2"
 nanoid = "0.4.0"
 nix = "0.26.2"
@@ -31,4 +32,4 @@ shell-escape = "0.1.5"
 tempfile = "3.3.0"
 tokio = { version = "1.16", features = [ "full" ] }
 tokio-stream = { version = "0.1.15", default-features = false }
-tokio-util = { version = "0.7.7" }
+tokio-util = { version = "0.7.7", features = [ "io-util" ] }

hydro_deploy/core/src/hydroflow_crate/perf_options.rs

@@ -1,7 +1,18 @@
 use std::path::PathBuf;
 
-#[derive(Clone)]
+#[derive(Clone, buildstructor::Builder)]
+#[non_exhaustive] // Prevent direct construction.
 pub struct PerfOptions {
-    pub output_file: PathBuf,
     pub frequency: u32,
+
+    /// TODO(mingwei): Only used by localhost, for now.
+    pub perf_outfile: Option<PathBuf>,
+
+    /// If set, what the write the folded output to.
+    pub fold_outfile: Option<PathBuf>,
+    pub fold_options: Option<inferno::collapse::perf::Options>,
+    /// If set, what to write the output flamegraph SVG file to.
+    pub flamegraph_outfile: Option<PathBuf>,
+    // This type is super annoying and isn't `clone` and has a lifetime... so wrap in fn pointer for now.
+    pub flamegraph_options: Option<fn() -> inferno::flamegraph::Options<'static>>,
 }

hydro_deploy/core/src/hydroflow_crate/service.rs

@@ -303,24 +303,31 @@ impl Service for HydroflowCrateService {
     }
 
     async fn stop(&mut self) -> Result<()> {
-        let launched_binary = self.launched_binary.as_mut().unwrap();
-        launched_binary.stdin().send("stop\n".to_string())?;
-
-        let timeout_result = ProgressTracker::leaf(
-            self.display_id
+        ProgressTracker::with_group(
+            &self
+                .display_id
                 .clone()
-                .unwrap_or_else(|| format!("service/{}", self.id))
-                + " / waiting for exit",
-            tokio::time::timeout(Duration::from_secs(60), launched_binary.wait()),
-        )
-        .await;
-        match timeout_result {
-            Err(_timeout) => {} // `wait()` timed out, but stop will force quit.
-            Ok(Err(unexpected_error)) => return Err(unexpected_error), // `wait()` errored.
-            Ok(Ok(_exit_status)) => {}
-        }
-        launched_binary.stop().await?;
+                .unwrap_or_else(|| format!("service/{}", self.id)),
+            None,
+            || async {
+                let launched_binary = self.launched_binary.as_mut().unwrap();
+                launched_binary.stdin().send("stop\n".to_string())?;
 
-        Ok(())
+                let timeout_result = ProgressTracker::leaf(
+                    "waiting for exit".to_owned(),
+                    tokio::time::timeout(Duration::from_secs(60), launched_binary.wait()),
+                )
+                .await;
+                match timeout_result {
+                    Err(_timeout) => {} // `wait()` timed out, but stop will force quit.
+                    Ok(Err(unexpected_error)) => return Err(unexpected_error), // `wait()` errored.
+                    Ok(Ok(_exit_status)) => {}
+                }
+                launched_binary.stop().await?;
+
+                Ok(())
+            },
+        )
+        .await
     }
 }

hydro_deploy/core/src/localhost/mod.rs

@@ -163,7 +163,10 @@ impl LaunchedHost for LaunchedLocalhost {
                 "dwarf,64000",
                 "-o",
             ])
-            .arg(&perf.output_file)
+            .arg(
+                perf.perf_outfile
+                    .expect("perf_outfile should be set for localhost perf."),
+            )
             .arg(&binary.bin_path)
             .args(args);
             tmp