dial9_perf_self_profile/
symbolize_processor.rs

1use dial9_core::pipeline::{Payload, ProcessError, SegmentData, SegmentProcessor};
2use dial9_core::rate_limited;
3use std::future::Future;
4use std::pin::Pin;
5use std::time::Duration;
6
7// ---------------------------------------------------------------------------
8// SymbolizeProcessor — resolves stack frame addresses to symbol names
9// ---------------------------------------------------------------------------
10
11/// Resolves stack-frame addresses in the segment to symbol names using
12/// the current process's `/proc/self/maps`.
13///
14/// Owns a long-lived
15/// [`OfflineSymbolizer`](crate::offline_symbolize::OfflineSymbolizer)
16/// running on a dedicated thread, so blazesym's per-ELF DWARF cache
17/// stays warm across segments. Without this, every segment paid the
18/// full ELF parse cost (hundreds of ms — see #462).
19pub struct SymbolizeProcessor {
20    symbolizer: std::sync::Arc<crate::offline_symbolize::OfflineSymbolizer>,
21}
22
23impl SymbolizeProcessor {
24    pub fn new() -> Self {
25        Self {
26            symbolizer: std::sync::Arc::new(crate::offline_symbolize::OfflineSymbolizer::new()),
27        }
28    }
29}
30
31impl Default for SymbolizeProcessor {
32    fn default() -> Self {
33        Self::new()
34    }
35}
36
37impl std::fmt::Debug for SymbolizeProcessor {
38    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
39        f.debug_struct("SymbolizeProcessor").finish_non_exhaustive()
40    }
41}
42
43impl SegmentProcessor for SymbolizeProcessor {
44    fn name(&self) -> &'static str {
45        "Symbolize"
46    }
47
48    fn process(
49        &mut self,
50        mut data: SegmentData,
51    ) -> Pin<Box<dyn Future<Output = Result<SegmentData, ProcessError>> + Send + '_>> {
52        let symbolizer = self.symbolizer.clone();
53        Box::pin(async move {
54            // Skip already-compressed segments (e.g. leftover from a previous run).
55            if data.payload().starts_with(&[0x1f, 0x8b]) {
56                tracing::debug!(target: "dial9_worker", "segment is gzip-compressed, skipping symbolization");
57                return Ok(data);
58            }
59            // The symbolize FFI reads `&[u8]`, so we materialize a single
60            // contiguous `Bytes`. When there's only one chunk this is a
61            // zero-copy `Bytes::clone`-equivalent; the `BytesMut` concat
62            // path runs only on already-segmented input (rare).
63            let input = data.take_payload().into_bytes();
64            // Hand off to a blocking thread because `OfflineSymbolizer::symbolize`
65            // is itself a blocking call (it sends to its dedicated symbolizer
66            // thread and waits for the response).
67            let result = tokio::task::spawn_blocking(move || {
68                let maps = crate::read_proc_maps();
69                let output = symbolizer.symbolize_bytes(input.clone(), &maps)?;
70                // Hand back the original bytes plus the symbol output as two
71                // chunks — no copy of `input`.
72                let mut combined = Payload::new();
73                combined.push(input);
74                combined.push(bytes::Bytes::from(output));
75                Ok::<_, std::io::Error>(combined)
76            })
77            .await;
78            match result {
79                Ok(Ok(payload)) => {
80                    data.set_payload(payload);
81                    Ok(data)
82                }
83                Ok(Err(e)) => {
84                    rate_limited!(Duration::from_secs(60), {
85                        tracing::warn!(target: "dial9_worker", error = %e, "symbolization failed, preserving original bytes");
86                    });
87                    Err(ProcessError::io(data, e))
88                }
89                Err(e) => Err(ProcessError::io(data, std::io::Error::other(e))),
90            }
91        })
92    }
93}
dial9_perf_self_profile/symbolize_processor.rs

dial9_perf_self_profile/
symbolize_processor.rs
