fix(hydro_deploy): handle -1 addresses from samply, fix _counter() rollover (#1814)

This fixes samply profiling on my "ancient" 2019 x86-64 macbook pro
15.3.2 (24D81)

This pull request aims to fix the handling of –1 address values from
samply by updating tracing filenames and refactoring related error and
type handling. Key changes include:
- Better error messages when `dtrace` or `samply` are not instaled.
- Fix integer rollover in `_counter()` by using `u64` instead of
inferred `i32`.
- Refactor samply profile conversion for asynchronous frame lookup.

<details>
<summary>Show a summary per file</summary>

| File | Description |
| ---- | ----------- |
| hydro_lang/src/rewrites/analyze_counter.rs | Adds custom panic with
measurement details if regex matching fails. (Used to diagnose
`_counter()` `i32` rollover) |
| hydro_deploy/core/src/localhost/samply.rs | Updates type for
addresses/resources, refactors frame lookup to use asynchronous
join_all, and adjusts string output for missing symbols. |
| hydro_deploy/core/src/localhost/mod.rs | Improves error handling
during command spawning with conditional context messages for when
`samply` or `dtrace` executables are not found. |
| hydro_deploy/core/src/localhost/launched_binary.rs | Uses
serde_path_to_error for improved deserialization error context. |
| dfir_lang/src/graph/ops/dest_sink.rs | Standardizes error messages by
removing extraneous punctuation. |
| dfir_lang/src/graph/ops/_counter.rs | Adds explicit type annotation
for a cell initialization to prevent `i32` rollover. |
</details>

Author: MingweiSamuel

.gitignore

@@ -15,3 +15,4 @@ flamegraph.svg
 /*.data.folded
 /*.perf.data
 /*.svg
+/*.profile

Cargo.lock

@@ -0,0 +1,2 @@
+#!/usr/bin/env bash
+rm *.data.folded *.perf.data *.svg *.profile || true

cleanup-perf.bash

@@ -68,7 +68,7 @@ pub const _COUNTER: OperatorConstraints = OperatorConstraints {
         let duration_ident = wc.make_ident("duration");
 
         let write_prologue = quote_spanned! {op_span=>
-            let #write_ident = ::std::rc::Rc::new(::std::cell::Cell::new(0));
+            let #write_ident = ::std::rc::Rc::new(::std::cell::Cell::new(0_u64));
 
             let #read_ident = ::std::rc::Rc::clone(&#write_ident);
             let #duration_ident = #duration_expr;

cleanup-tracing.bash

@@ -131,13 +131,13 @@ pub const DEST_SINK: OperatorConstraints = OperatorConstraints {
                     while let Some(item) = recv.recv().await {
                         sink.feed(item)
                             .await
-                            .expect("Error processing async sink item.");
+                            .expect("Error processing async sink item");
                         while let Ok(item) = recv.try_recv() {
                             sink.feed(item)
                                 .await
-                                .expect("Error processing async sink item.");
+                                .expect("Error processing async sink item");
                         }
-                        sink.flush().await.expect("Failed to flush sink.");
+                        sink.flush().await.expect("Failed to flush sink");
                     }
                 }
                 #df_ident
@@ -148,7 +148,7 @@ pub const DEST_SINK: OperatorConstraints = OperatorConstraints {
         let write_iterator = quote_spanned! {op_span=>
             let #ident = #root::pusherator::for_each::ForEach::new(|item| {
                 if let Err(err) = #send_ident.send(item) {
-                    panic!("Failed to send async write item for processing.: {}", err);
+                    panic!("Failed to send async write item for processing: {}", err);
                 }
             });
         };

dfir_lang/src/graph/ops/_counter.rs

@@ -33,6 +33,7 @@ nanoid = "0.4.0"
 nix = { version = "0.29.0", features = ["signal"] }
 serde = { version = "1.0.197", features = ["derive"] }
 serde_json = "1.0.115"
+serde_path_to_error = "0.1.0"
 shell-escape = "0.1.0"
 tempfile = "3.0.0"
 tokio = { version = "1.29.0", features = ["full"] }

dfir_lang/src/graph/ops/dest_sink.rs

@@ -188,9 +188,10 @@ impl LaunchedBinary for LaunchedLocalhostBinary {
                 }
 
                 let fold_data = if cfg!(target_os = "macos") {
-                    let loaded = serde_json::from_reader::<_, FxProfile>(std::fs::File::open(
-                        tracing_data.outfile.path(),
-                    )?)?;
+                    let deserializer = &mut serde_json::Deserializer::from_reader(
+                        std::fs::File::open(tracing_data.outfile.path())?,
+                    );
+                    let loaded = serde_path_to_error::deserialize::<_, FxProfile>(deserializer)?;
 
                     samply_to_folded(loaded).await.into()
                 } else if cfg!(target_family = "windows") {

hydro_deploy/core/Cargo.toml

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use anyhow::{Context, Result, bail};
+use anyhow::{Result, bail};
 use async_process::{Command, Stdio};
 use async_trait::async_trait;
 use hydro_deploy_integration::ServerBindConfig;
@@ -214,9 +214,14 @@ impl LaunchedHost for LaunchedLocalhost {
 
         ProgressTracker::println(format!("[{}] running command: `{:?}`", id, command));
 
-        let child = command
-            .spawn()
-            .with_context(|| format!("Failed to execute command: {:?}", command))?;
+        let child = command.spawn().map_err(|e| {
+            let msg = if maybe_perf_outfile.is_some() && std::io::ErrorKind::NotFound == e.kind() {
+                "Tracing executable not found, ensure it is installed"
+            } else {
+                "Failed to execute command"
+            };
+            anyhow::Error::new(e).context(format!("{}: {:?}", msg, command))
+        })?;
 
         Ok(Box::new(LaunchedLocalhostBinary::new(
             child,

hydro_deploy/core/src/localhost/launched_binary.rs

@@ -2,8 +2,10 @@ use std::collections::BTreeMap;
 use std::path::PathBuf;
 use std::str::FromStr;
 
+use futures::future::join_all;
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
+use serde_json::Number;
 use wholesym::debugid::DebugId;
 use wholesym::{LookupAddress, MultiArchDisambiguator, SymbolManager, SymbolManagerConfig};
 
@@ -47,13 +49,15 @@ pub struct StackTable {
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct FrameTable {
-    pub address: Vec<u64>,
+    /// `Vec` of `u64` or `-1`.
+    pub address: Vec<Number>,
     pub func: Vec<usize>,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct FuncTable {
-    pub resource: Vec<usize>,
+    // `Vec` of `usize` or `-1`.
+    pub resource: Vec<Number>,
 }
 
 pub async fn samply_to_folded(loaded: FxProfile) -> String {
@@ -74,40 +78,36 @@ pub async fn samply_to_folded(loaded: FxProfile) -> String {
         );
     }
 
-    let mut folded_frames: BTreeMap<Vec<String>, u64> = BTreeMap::new();
+    let mut folded_frames: BTreeMap<Vec<Option<String>>, u64> = BTreeMap::new();
     for thread in loaded.threads.into_iter().filter(|t| t.is_main_thread) {
-        let mut frame_lookuped = vec![];
-        for frame_id in 0..thread.frame_table.address.len() {
-            let address = thread.frame_table.address[frame_id];
-            let func_id = thread.frame_table.func[frame_id];
-            let resource_id = thread.func_table.resource[func_id];
-            let maybe_symbol_map = &symbol_maps[resource_id];
-
-            if let Some(symbols_map) = maybe_symbol_map {
-                if let Some(lookuped) = symbols_map
+        let frame_lookuped = join_all((0..thread.frame_table.address.len()).map(|frame_id| {
+            let fr_address = &thread.frame_table.address;
+            let fr_func = &thread.frame_table.func;
+            let fn_resource = &thread.func_table.resource;
+            let symbol_maps = &symbol_maps;
+            async move {
+                let address = fr_address[frame_id].as_u64()?;
+                let func_id = fr_func[frame_id];
+                let resource_id = fn_resource[func_id].as_u64()?;
+                let symbol_map = symbol_maps[resource_id as usize].as_ref()?;
+                let lookuped = symbol_map
                     .lookup(LookupAddress::Relative(address as u32))
-                    .await
-                {
-                    if let Some(inline_frames) = lookuped.frames {
-                        frame_lookuped.push(
-                            inline_frames
-                                .into_iter()
-                                .rev()
-                                .map(|inline| {
-                                    inline.function.unwrap_or_else(|| "unknown".to_string())
-                                })
-                                .join(";"),
-                        );
-                    } else {
-                        frame_lookuped.push(lookuped.symbol.name);
-                    }
+                    .await?;
+
+                if let Some(inline_frames) = lookuped.frames {
+                    Some(
+                        inline_frames
+                            .into_iter()
+                            .rev()
+                            .map(|inline| inline.function.unwrap_or_else(|| "unknown".to_string()))
+                            .join(";"),
+                    )
                 } else {
-                    frame_lookuped.push("unknown".to_string());
+                    Some(lookuped.symbol.name)
                 }
-            } else {
-                frame_lookuped.push("unknown".to_string());
             }
-        }
+        }))
+        .await;
 
         let all_leaves_grouped = thread
             .samples
@@ -116,10 +116,10 @@ pub async fn samply_to_folded(loaded: FxProfile) -> String {
             .enumerate()
             .filter_map(|(idx, s)| s.map(|s| (idx, s)))
             .map(|(idx, leaf)| (leaf, thread.samples.weight[idx]))
-            .chunk_by(|v| v.0)
+            .chunk_by(|&(leaf, _)| leaf)
             .into_iter()
             .map(|(leaf, group)| {
-                let weight = group.map(|t| t.1).sum();
+                let weight = group.map(|(_leaf, weight)| weight).sum();
                 (leaf, weight)
             })
             .collect::<Vec<(usize, u64)>>();
@@ -143,7 +143,7 @@ pub async fn samply_to_folded(loaded: FxProfile) -> String {
             if i != 0 {
                 output.push(';');
             }
-            output.push_str(s);
+            output.push_str(s.as_deref().unwrap_or("unknown"));
         }
 
         output.push_str(&format!(" {}\n", weight));

hydro_deploy/core/src/localhost/mod.rs

@@ -8,7 +8,12 @@ use crate::ir::*;
 /// Returns (op_id, count)
 pub fn parse_counter_usage(measurement: String) -> (usize, usize) {
     let regex = Regex::new(r"\((\d+)\): (\d+)").unwrap();
-    let matches = regex.captures_iter(&measurement).last().unwrap();
+    let matches = regex.captures_iter(&measurement).last().unwrap_or_else(|| {
+        ::core::panic!(
+            "Failed to parse counter usage from measurement: {:?}",
+            measurement
+        )
+    });
     let op_id = matches[1].parse::<usize>().unwrap();
     let count = matches[2].parse::<usize>().unwrap();
     (op_id, count)