diff --git a/e2e-tests/tests/backtrace_execution.rs b/e2e-tests/tests/backtrace_execution.rs index 6aa057fb..8c595bcd 100644 --- a/e2e-tests/tests/backtrace_execution.rs +++ b/e2e-tests/tests/backtrace_execution.rs @@ -1970,6 +1970,14 @@ trace dlopen_main_after_limit_heartbeat { refresh_count >= metadata_load_count, "each metadata load should have a corresponding identity resolution, got {metadata_load_count} loads and {refresh_count} resolutions\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}" ); + let stderr_after_limit = stderr + .split_once(&warning) + .map(|(_, remainder)| remainder) + .expect("the limit warning count was checked above"); + assert!( + !stderr_after_limit.contains("Resolving one backtrace runtime module in the background"), + "runtime module discovery should stop after the terminal limit warning\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}" + ); assert_eq!( metadata_load_count, MODULE_LIMIT as usize, @@ -2042,19 +2050,14 @@ trace dlopen_main_callback { } anyhow::ensure!(exit_code == 0, "stderr={stderr} stdout={stdout}"); - let blocks = backtrace_blocks_after(&stdout, "T_MODE_DLOPEN_CALLBACK_STACK", 6)?; - let refreshed_block = blocks - .iter() - .find(|block| block.contains("dlopen_lib_middle") && block.contains("dlopen_lib_driver")) - .ok_or_else(|| { - anyhow::anyhow!( - "expected a target-mode dlopen backtrace block to unwind inside the library\n\ - STDOUT:\n{stdout}\nSTDERR:\n{stderr}" - ) - })?; - assert_ordered_patterns( - refreshed_block, + let refreshed_block = matching_backtrace_block_with_ordered_patterns_after( + &stdout, + &stderr, + "T_MODE_DLOPEN_CALLBACK_STACK", + 6, + "a fully symbolized target-mode dlopen backtrace", &[ + "backtrace: truncated, 6 frames (max 6)", "#0 dlopen_main_callback", "#1 dlopen_lib_leaf", "#2 dlopen_lib_middle", @@ -2118,6 +2121,7 @@ trace dlopen_lib_leaf { anyhow::ensure!(exit_code == 0, "stderr={stderr} stdout={stdout}"); let expected_frames = [ + "backtrace: truncated, 6 frames (max 6)", "#0 dlopen_lib_leaf", "#1 dlopen_lib_middle", "#2 dlopen_lib_driver", @@ -2137,6 +2141,13 @@ trace dlopen_lib_leaf { "shared-library target dlopen backtrace should publish the first target mapping refresh\n\ BLOCK:\n{refreshed_block}\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}" ); + // The target-only periodic refresh must not undo a completed process range + // snapshot. Check steady-state output, not just the first successful event. + let blocks = backtrace_blocks_after(&stdout, "T_MODE_DLOPEN_SHARED_LIBRARY_STACK", 6)?; + assert!(blocks.len() >= 10, "expected steady-state backtrace events"); + for block in blocks.iter().rev().take(10) { + assert_ordered_patterns(block, &expected_frames)?; + } Ok(()) } diff --git a/e2e-tests/tests/backtrace_publication_execution.rs b/e2e-tests/tests/backtrace_publication_execution.rs new file mode 100644 index 00000000..ae11ffa7 --- /dev/null +++ b/e2e-tests/tests/backtrace_publication_execution.rs @@ -0,0 +1,113 @@ +//! Shared CFI allocation across initial loaders and background runtime appends. + +mod common; + +use aya::maps::{Array, HashMap, Map, MapData}; +use ghostscope_loader::GhostScopeLoader; +use ghostscope_process::pinned_bpf_maps::*; +use ghostscope_protocol::{BacktraceModuleRowRange, BacktraceUnwindRow}; +use std::sync::{Arc, Barrier}; + +#[tokio::test] +async fn test_shared_cfi_initialization_and_runtime_append_do_not_overlap() -> anyhow::Result<()> { + common::init(); + if std::env::var_os("E2E_RUN_CONTAINER_TOPOLOGY").is_some() { + return Ok(()); + } + cleanup_current_pinned_maps()?; + let _cleanup = scopeguard::guard((), |_| { + let _ = cleanup_current_pinned_maps(); + }); + ensure_pinned_proc_offsets_exists(4096)?; + ensure_pinned_pid_aliases_exists(4096)?; + ensure_pinned_proc_module_ranges_exist(4096)?; + ensure_pinned_backtrace_cfi_maps_exist(4096, 4096)?; + + let binary = common::FIXTURES.get_test_binary("backtrace_hot_program")?; + let analyzer = ghostscope_dwarf::DwarfAnalyzer::from_exec_path(&binary).await?; + let options = ghostscope_compiler::CompileOptions { + binary_path_hint: Some(binary.to_string_lossy().into_owned()), + backtrace_unwind_rows_max_entries: 4096, + ..Default::default() + }; + let compiled = ghostscope_compiler::compile_script( + "trace hot_bt_probe { bt; }", + &analyzer, + None, + Some(1), + &options, + )?; + anyhow::ensure!( + compiled.uprobe_configs.len() == 1, + "expected one backtrace config" + ); + let bytecode = &compiled.uprobe_configs[0].ebpf_bytecode; + let initial = GhostScopeLoader::new_with_shared_backtrace_maps(bytecode, true)?; + let runtime = GhostScopeLoader::new_with_shared_backtrace_maps(bytecode, true)?; + let start = Arc::new(Barrier::new(2)); + let row = |cookie, index| BacktraceUnwindRow { + pc_start: cookie * 0x1000 + index, + pc_end: cookie * 0x1000 + index + 1, + ..Default::default() + }; + std::thread::scope(|scope| -> anyhow::Result<()> { + let mut publishers = Vec::new(); + for (publisher, mut loader) in [(0, initial), (1, runtime)] { + let start = Arc::clone(&start); + publishers.push(scope.spawn(move || -> anyhow::Result<()> { + start.wait(); + for round in 0..16 { + let cookie = 1 + round * 2 + publisher; + let rows: Vec<_> = (0..32).map(|index| row(cookie, index)).collect(); + if publisher == 0 { + loader.populate_backtrace_unwind_rows_and_module_row_ranges( + &rows, + &[( + cookie, + BacktraceModuleRowRange { + row_start: 0, + row_end: 32, + }, + )], + )?; + } else { + loader.append_backtrace_unwind_rows_for_modules(&[(cookie, rows)])?; + } + } + Ok(()) + })); + } + for publisher in publishers { + publisher.join().unwrap()?; + } + Ok(()) + })?; + + let ranges = HashMap::<_, u64, BacktraceModuleRowRange>::try_from(Map::from_map_data( + MapData::from_pin(bt_module_row_ranges_pin_path()?)?, + )?)?; + let rows = Array::<_, BacktraceUnwindRow>::try_from(Map::from_map_data(MapData::from_pin( + bt_unwind_rows_pin_path()?, + )?)?)?; + let mut allocated = std::collections::BTreeSet::new(); + for cookie in 1..=32 { + let range = ranges.get(&cookie, 0)?; + assert_eq!(range.row_end - range.row_start, 32); + for index in range.row_start..range.row_end { + assert!( + allocated.insert(index), + "overlapping CFI row allocation at {index}" + ); + assert_eq!( + rows.get(&index, 0)?, + row(cookie, u64::from(index - range.row_start)) + ); + } + } + // A trace attached after all original loaders are gone sees completed modules. + let mut replacement = GhostScopeLoader::new_with_shared_backtrace_maps(bytecode, true)?; + let stats = replacement.append_backtrace_unwind_rows_for_modules(&[(1, vec![row(1, 0)])])?; + assert_eq!(stats.modules, 0); + assert_eq!(stats.rows, 0); + Ok(()) +} diff --git a/e2e-tests/tests/globals_target_execution.rs b/e2e-tests/tests/globals_target_execution.rs index 1173b048..5437d386 100644 --- a/e2e-tests/tests/globals_target_execution.rs +++ b/e2e-tests/tests/globals_target_execution.rs @@ -9,6 +9,11 @@ use serial_test::serial; use std::env; use std::os::unix::fs as unix_fs; use std::path::{Path, PathBuf}; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; +use std::thread::JoinHandle; use std::time::Duration; use tempfile::Builder; @@ -52,11 +57,64 @@ fn prepare_late_start_launcher( common::targets::ensure_target_binary_ready_for_default_sandbox(binary_path) } -fn ghostscope_log_path() -> anyhow::Result { - Ok(Path::new(env!("CARGO_MANIFEST_DIR")) - .parent() - .ok_or_else(|| anyhow::anyhow!("failed to resolve workspace root for ghostscope.log"))? - .join("ghostscope.log")) +fn ghostscope_log_path() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join("ghostscope.log") +} + +struct MapChangeChurn { + stop: Arc, + handle: Option>, +} + +impl MapChangeChurn { + fn start() -> Self { + const MAPS_PER_BATCH: usize = 64; + + let stop = Arc::new(AtomicBool::new(false)); + let worker_stop = Arc::clone(&stop); + let handle = std::thread::spawn(move || { + let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; + if page_size <= 0 { + return; + } + while !worker_stop.load(Ordering::Relaxed) { + for _ in 0..MAPS_PER_BATCH { + // SAFETY: The anonymous mapping is private to this worker and is unmapped + // immediately without exposing the pointer to Rust references. + let mapped = unsafe { + libc::mmap( + std::ptr::null_mut(), + page_size as usize, + libc::PROT_NONE, + libc::MAP_PRIVATE | libc::MAP_ANONYMOUS, + -1, + 0, + ) + }; + if mapped != libc::MAP_FAILED { + // SAFETY: `mapped` and the length exactly match the successful mmap above. + unsafe { + libc::munmap(mapped, page_size as usize); + } + } + } + std::thread::sleep(Duration::from_millis(1)); + } + }); + Self { + stop, + handle: Some(handle), + } + } +} + +impl Drop for MapChangeChurn { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + if let Some(handle) = self.handle.take() { + let _ = handle.join(); + } + } } fn workspace_root() -> anyhow::Result { @@ -559,6 +617,9 @@ async fn test_t_mode_library_late_start_globals_prints() -> anyhow::Result<()> { // only after GhostScope reports its late-start hooks are ready. let binary_path = FIXTURES.get_test_binary("globals_program")?; let _target_sandbox_guard = prepare_late_start_launcher(&binary_path)?; + // Keep unrelated mmap traffic flowing while GhostScope discovers the late-start target. This + // guards against sysmon maintenance becoming dependent on a globally quiet event stream. + let _map_change_churn = MapChangeChurn::start(); let bin_dir = binary_path.parent().unwrap().to_path_buf(); let lib_path = bin_dir.join("libgvars.so"); let script = r#" @@ -603,14 +664,14 @@ trace lib_tick { "Late-start: LIB_STATE.counter should +2 per tick. STDOUT: {stdout}" ); } else { - let log_dump = tokio::fs::read_to_string(ghostscope_log_path()?) + let log_dump = tokio::fs::read_to_string(ghostscope_log_path()) .await .unwrap_or_else(|_| "".to_string()); let msg = format!("Late-start: No events for our PID {pid}. STDOUT: {stdout}. LOG: {log_dump}"); assert!(!uniq.is_empty(), "{}", msg); } - let _ = tokio::fs::remove_file(ghostscope_log_path()?).await; + let _ = tokio::fs::remove_file(ghostscope_log_path()).await; Ok(()) } @@ -779,7 +840,7 @@ trace lib_tick { "Should not surface raw read_user errors for PID {pid}. STDOUT: {stdout}" ); - let _ = tokio::fs::remove_file(ghostscope_log_path()?).await; + let _ = tokio::fs::remove_file(ghostscope_log_path()).await; Ok(()) } diff --git a/ghostscope-dwarf/src/analyzer/mod.rs b/ghostscope-dwarf/src/analyzer/mod.rs index 812914d2..0c31d0d1 100644 --- a/ghostscope-dwarf/src/analyzer/mod.rs +++ b/ghostscope-dwarf/src/analyzer/mod.rs @@ -1049,9 +1049,22 @@ impl DwarfAnalyzer { ) -> Option { let symbols = self.runtime_text_symbols.get(&module_cookie)?; let upper = symbols.partition_point(|symbol| symbol.address <= address); - let symbol = symbols[..upper].iter().rev().find(|symbol| { - symbol.size == 0 || address < symbol.address.saturating_add(symbol.size) - })?; + let symbol = symbols[..upper] + .iter() + .enumerate() + .rev() + .find(|(index, symbol)| { + let end = if symbol.size == 0 { + symbols[index + 1..] + .iter() + .find(|next| next.address > symbol.address) + .map(|next| next.address) + } else { + Some(symbol.address.saturating_add(symbol.size)) + }; + end.is_none_or(|end| address < end) + }) + .map(|(_, symbol)| symbol)?; let rust_hashes = if full { RustSymbolHashDisplay::Shown } else { diff --git a/ghostscope-dwarf/src/analyzer/tests.rs b/ghostscope-dwarf/src/analyzer/tests.rs index c790533e..30bf9bb6 100644 --- a/ghostscope-dwarf/src/analyzer/tests.rs +++ b/ghostscope-dwarf/src/analyzer/tests.rs @@ -188,3 +188,39 @@ fn runtime_text_symbols_resolve_without_loading_module_dwarf() { .find_runtime_function_name_for_display(0x5678, 0x13f, false) .is_none()); } + +#[test] +fn zero_sized_runtime_text_symbol_stops_at_next_symbol() { + let mut analyzer = DwarfAnalyzer::from_modules(0, Vec::new()); + analyzer.add_runtime_text_symbols( + 0x1234, + vec![ + RuntimeTextSymbol { + name: "zero_sized".to_string(), + address: 0x100, + size: 0, + }, + RuntimeTextSymbol { + name: "next_function".to_string(), + address: 0x120, + size: 0x10, + }, + ], + ); + + assert_eq!( + analyzer + .find_runtime_function_name_for_display(0x1234, 0x110, false) + .as_deref(), + Some("zero_sized") + ); + assert_eq!( + analyzer + .find_runtime_function_name_for_display(0x1234, 0x125, false) + .as_deref(), + Some("next_function") + ); + assert!(analyzer + .find_runtime_function_name_for_display(0x1234, 0x130, false) + .is_none()); +} diff --git a/ghostscope-dwarf/src/runtime_unwind.rs b/ghostscope-dwarf/src/runtime_unwind.rs index dc66d977..424a1c15 100644 --- a/ghostscope-dwarf/src/runtime_unwind.rs +++ b/ghostscope-dwarf/src/runtime_unwind.rs @@ -1,7 +1,7 @@ //! Lightweight runtime unwind loading for modules discovered after tracing starts. use crate::{binary::MappedFile, objfile::ModuleUnwindInfo, CompactUnwindTable, ModuleId, Result}; -use anyhow::Context; +use ghostscope_process::module_probe::ModuleProbe; use object::{Object, ObjectSymbol, SymbolKind}; use std::{ path::PathBuf, @@ -68,14 +68,17 @@ impl RuntimeBacktraceLoadBudget { /// after the compact rows and bounded symbol list have been built. pub fn load_runtime_backtrace_metadata( module_path: PathBuf, + probe: ModuleProbe, max_rows: usize, budget: &RuntimeBacktraceLoadBudget, ) -> Result { budget.check()?; - let mapped_file = Arc::new( - MappedFile::open(&module_path) - .with_context(|| format!("Failed to map runtime module {}", module_path.display()))?, - ); + // Keep the exact descriptor-backed mapping validated during discovery. + // Reopening module_path here could parse a replacement under the old cookie. + let mapped_file = Arc::new(MappedFile { + data: probe.into_mmap(), + path: module_path.clone(), + }); budget.check()?; let text_symbols = collect_text_symbols(&mapped_file, budget)?; budget.check()?; @@ -169,11 +172,57 @@ fn append_text_symbol<'a>( mod tests { use super::*; + #[test] + fn pathname_replacement_does_not_replace_the_validated_elf_mapping() { + fn elf_with_symbol(name: &str) -> Vec { + let mut object = object::write::Object::new( + object::BinaryFormat::Elf, + object::Architecture::X86_64, + object::Endianness::Little, + ); + let text = object.section_id(object::write::StandardSection::Text); + object.append_section_data(text, &[0x90; 16], 1); + object.add_symbol(object::write::Symbol { + name: name.as_bytes().to_vec(), + value: 1, + size: 1, + kind: object::SymbolKind::Text, + scope: object::SymbolScope::Linkage, + weak: false, + section: object::write::SymbolSection::Section(text), + flags: object::SymbolFlags::None, + }); + object.write().unwrap() + } + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("module.so"); + let replacement = dir.path().join("replacement.so"); + std::fs::write(&path, elf_with_symbol("original_function")).unwrap(); + let probe = ModuleProbe::open(path.to_str().unwrap()).unwrap(); + std::fs::write(&replacement, elf_with_symbol("replacement_function")).unwrap(); + std::fs::rename(replacement, &path).unwrap(); + assert_ne!( + probe.cookie(), + ModuleProbe::open(path.to_str().unwrap()).unwrap().cookie() + ); + let budget = RuntimeBacktraceLoadBudget::new(Duration::from_secs(5)); + let metadata = load_runtime_backtrace_metadata(path, probe, 16, &budget).unwrap(); + assert!(metadata + .text_symbols + .iter() + .any(|s| s.name == "original_function")); + assert!(!metadata + .text_symbols + .iter() + .any(|s| s.name == "replacement_function")); + } + #[test] fn loads_bounded_backtrace_metadata_without_a_full_dwarf_analyzer() { let executable = std::env::current_exe().expect("test executable path"); let budget = RuntimeBacktraceLoadBudget::new(Duration::from_secs(5)); - let metadata = load_runtime_backtrace_metadata(executable, 1, &budget) + let probe = ModuleProbe::open(executable.to_str().unwrap()).unwrap(); + let metadata = load_runtime_backtrace_metadata(executable, probe, 1, &budget) .expect("runtime backtrace metadata load"); let table = metadata .unwind_table @@ -185,12 +234,13 @@ mod tests { } #[test] - fn cancelled_budget_stops_before_mapping_a_module() { + fn cancelled_budget_stops_before_parsing_a_module() { let executable = std::env::current_exe().expect("test executable path"); let budget = RuntimeBacktraceLoadBudget::new(Duration::from_secs(5)); budget.cancel(); - let error = load_runtime_backtrace_metadata(executable, 1, &budget) + let probe = ModuleProbe::open(executable.to_str().unwrap()).unwrap(); + let error = load_runtime_backtrace_metadata(executable, probe, 1, &budget) .expect_err("cancelled runtime load should fail"); assert!(error.to_string().contains("cancelled")); diff --git a/ghostscope-loader/src/lib.rs b/ghostscope-loader/src/lib.rs index 9eb0eac3..286f495c 100644 --- a/ghostscope-loader/src/lib.rs +++ b/ghostscope-loader/src/lib.rs @@ -40,6 +40,7 @@ use std::num::NonZeroU32; use std::os::unix::io::AsRawFd; use std::os::unix::io::RawFd; use std::path::Path; +use std::sync::{Mutex, MutexGuard}; use std::task::Poll; use std::time::Instant; use std::{io, ops::ControlFlow}; @@ -52,6 +53,20 @@ const MAX_RINGBUF_RECORDS_PER_WAIT: usize = 256; const PERF_READ_BATCH_SIZE: usize = 64; const EVENT_LOSS_OUTPUT_FAILURES_KEY: u32 = 0; +// The pinned CFI maps are scoped to this GhostScope process, not to one loader. +// Hold this across reading row_end, allocating rows, and publishing their range. +static BACKTRACE_CFI_PUBLICATION: Mutex<()> = Mutex::new(()); + +fn backtrace_cfi_publication_guard(shared: bool) -> Result>> { + shared + .then(|| { + BACKTRACE_CFI_PUBLICATION.lock().map_err(|_| { + LoaderError::Generic("shared backtrace CFI publication lock poisoned".into()) + }) + }) + .transpose() +} + #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct EventLossStats { pub output_failures: u64, @@ -1412,12 +1427,18 @@ impl GhostScopeLoader { return Ok(()); } - if ranges.is_empty() || !self.shared_backtrace_maps { + if !self.shared_backtrace_maps { self.populate_backtrace_unwind_rows(rows)?; self.populate_backtrace_module_row_ranges(ranges)?; return Ok(()); } + if ranges.is_empty() { + return Err(LoaderError::Generic( + "shared backtrace rows require module ranges for atomic publication".into(), + )); + } + let _publication = backtrace_cfi_publication_guard(true)?; self.sync_shared_backtrace_row_state()?; for (cookie, range) in ranges.iter().copied() { if self.backtrace_module_row_cookies.contains(&cookie) { @@ -1456,6 +1477,11 @@ impl GhostScopeLoader { } pub fn populate_backtrace_unwind_rows(&mut self, rows: &[BacktraceUnwindRow]) -> Result<()> { + if self.shared_backtrace_maps { + return Err(LoaderError::Generic( + "use combined row and range publication for shared backtrace maps".into(), + )); + } if rows.is_empty() { return Ok(()); } @@ -1490,6 +1516,11 @@ impl GhostScopeLoader { &mut self, ranges: &[(u64, BacktraceModuleRowRange)], ) -> Result<()> { + if self.shared_backtrace_maps { + return Err(LoaderError::Generic( + "use combined row and range publication for shared backtrace maps".into(), + )); + } if ranges.is_empty() { return Ok(()); } @@ -1524,6 +1555,7 @@ impl GhostScopeLoader { cookie: u64, rows: &[BacktraceUnwindRow], ) -> Result> { + let _publication = backtrace_cfi_publication_guard(self.shared_backtrace_maps)?; if self.shared_backtrace_maps { self.sync_shared_backtrace_row_state()?; } @@ -1624,6 +1656,7 @@ impl GhostScopeLoader { &mut self, modules: &[(u64, Vec)], ) -> Result { + let _publication = backtrace_cfi_publication_guard(self.shared_backtrace_maps)?; let mut stats = BacktraceUnwindRowsAppendStats::default(); if self.shared_backtrace_maps { self.sync_shared_backtrace_row_state()?; diff --git a/ghostscope-process/src/lib.rs b/ghostscope-process/src/lib.rs index db14439a..7c6dc528 100644 --- a/ghostscope-process/src/lib.rs +++ b/ghostscope-process/src/lib.rs @@ -4,7 +4,10 @@ pub mod pid; pub mod pinned_bpf_maps; pub mod proc_maps; pub mod target_arch; -pub use offsets::{PidOffsetsEntry, ProcessManager, SectionOffsets}; +pub use offsets::{ + PidOffsetsEntry, ProcessManager, ProcessManagerSnapshot, ProcessManagerSnapshotReader, + SectionOffsets, +}; pub use pid::{ build_runtime_pid_plan, detect_runtime_environment, host_pid_for_proc_pid, resolve_event_pid_for_proc, resolve_input_pid, resolve_pid_session, resolve_proc_pid, diff --git a/ghostscope-process/src/module_probe.rs b/ghostscope-process/src/module_probe.rs index f603a4be..c5307507 100644 --- a/ghostscope-process/src/module_probe.rs +++ b/ghostscope-process/src/module_probe.rs @@ -4,46 +4,41 @@ use memmap2::{Mmap, MmapOptions}; use object::Object; use std::fs::{self, OpenOptions}; use std::hash::{Hash, Hasher}; +use std::os::fd::AsRawFd; use std::os::unix::fs::MetadataExt; use std::os::unix::fs::OpenOptionsExt; -use std::path::PathBuf; +#[derive(Debug)] pub struct ModuleProbe { metadata_cookie: u64, - mmap: Mmap, -} - -struct ValidatedModulePath { - resolved_path: PathBuf, metadata: fs::Metadata, + mmap: Mmap, } impl ModuleProbe { pub fn open(module_path: &str) -> Result { let normalized_path = normalize_cookie_path(module_path); let validated = validate_module_path(&normalized_path)?; + Self::from_validated_file(validated) + } - // Open the resolved regular file so common launcher symlinks like - // `/bin/sh` and `/usr/bin/python3` continue to probe correctly. - // `O_NOFOLLOW` still protects the final open against a symlink swap - // after resolution. + fn from_validated_file(validated: fs::File) -> Result { + // Reopen the retained O_PATH descriptor, not a canonicalized pathname: + // resolving /proc//root in userspace can select a host-side file + // instead of the target's file in another mount namespace. The retained + // descriptor also prevents replacement or symlink swaps after validation. + let validated_metadata = validated.metadata()?; let file = OpenOptions::new() .read(true) - .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW) - .open(&validated.resolved_path)?; + .custom_flags(libc::O_CLOEXEC) + .open(format!("/proc/self/fd/{}", validated.as_raw_fd()))?; let meta = file.metadata()?; - if !meta.file_type().is_file() { - anyhow::bail!( - "refusing to read non-regular file {}", - validated.resolved_path.display() - ); - } - if meta.dev() != validated.metadata.dev() || meta.ino() != validated.metadata.ino() { - anyhow::bail!( - "module path changed while opening {}", - validated.resolved_path.display() - ); - } + anyhow::ensure!( + meta.file_type().is_file() + && meta.dev() == validated_metadata.dev() + && meta.ino() == validated_metadata.ino(), + "module identity changed while reopening validated descriptor" + ); let dev = meta.dev(); let ino = meta.ino(); @@ -56,10 +51,20 @@ impl ModuleProbe { Ok(Self { metadata_cookie, + metadata: meta, mmap, }) } + /// Identity of the descriptor backing this mapping, not a later path lookup. + pub fn metadata(&self) -> &fs::Metadata { + &self.metadata + } + + pub fn into_mmap(self) -> Mmap { + self.mmap + } + pub fn object(&self) -> Result> { Ok(object::File::parse(&self.mmap[..])?) } @@ -89,30 +94,42 @@ pub fn cookie_for_path(module_path: &str) -> u64 { .unwrap_or_else(|_| stable_hash(&normalize_cookie_path(module_path))) } -fn validate_module_path(path: &str) -> Result { +fn validate_module_path(path: &str) -> Result { // `/proc//maps` is not a trustworthy module list. Reject procfs/sysfs - // paths up front, then resolve symlinks to the final target and insist that - // the resolved object is a regular file before it reaches the ELF path. + // paths up front. O_PATH follows launcher symlinks without opening devices + // or blocking on FIFOs; inspect the retained object before reading any data. let input_is_proc_root = is_safe_proc_root_path(path); if is_filtered_module_prefix(path) && !input_is_proc_root { anyhow::bail!("refusing to read pseudo-filesystem path {path}"); } - let resolved_path = fs::canonicalize(path)?; - let resolved_str = resolved_path.to_string_lossy(); - let resolved_is_safe_proc_root = input_is_proc_root && is_safe_proc_root_path(&resolved_str); - if is_filtered_module_prefix(&resolved_str) && !resolved_is_safe_proc_root { - anyhow::bail!("refusing to read pseudo-filesystem path {resolved_str}"); - } - let meta = fs::metadata(&resolved_path)?; + let file = OpenOptions::new() + .read(true) + .custom_flags(libc::O_PATH | libc::O_CLOEXEC) + .open(path)?; + let meta = file.metadata()?; if !meta.file_type().is_file() { - anyhow::bail!("refusing to read non-regular file {resolved_str}"); + anyhow::bail!("refusing to read non-regular file {path}"); + } + + // Check the actual filesystem too: symlinks and proc-root paths can conceal + // a procfs/sysfs file behind an otherwise ordinary pathname. + let mut filesystem = std::mem::MaybeUninit::::uninit(); + // SAFETY: file owns a live descriptor and filesystem points to writable + // storage of the exact type and size expected by fstatfs. + if unsafe { libc::fstatfs(file.as_raw_fd(), filesystem.as_mut_ptr()) } != 0 { + return Err(std::io::Error::last_os_error().into()); + } + // SAFETY: successful fstatfs initialized the entire output structure. + let filesystem = unsafe { filesystem.assume_init() }; + if matches!( + filesystem.f_type, + libc::PROC_SUPER_MAGIC | libc::SYSFS_MAGIC + ) { + anyhow::bail!("refusing to read pseudo-filesystem path {path}"); } - Ok(ValidatedModulePath { - resolved_path, - metadata: meta, - }) + Ok(file) } fn is_safe_proc_root_path(path: &str) -> bool { @@ -221,4 +238,39 @@ mod tests { let _ = std::fs::remove_file(&link); let _ = std::fs::remove_file(&target); } + + #[test] + fn retains_validated_file_after_path_replacement() { + let base = std::env::temp_dir().join(format!( + "ghostscope-module-probe-replacement-{}", + std::process::id() + )); + let replaced = base.with_extension("replaced"); + std::fs::write(&base, b"original module").unwrap(); + let validated = validate_module_path(base.to_str().unwrap()).unwrap(); + std::fs::rename(&base, &replaced).unwrap(); + std::fs::write(&base, b"replacement module").unwrap(); + + let probe = ModuleProbe::from_validated_file(validated).unwrap(); + assert_eq!(&probe.mmap[..], b"original module"); + assert_ne!( + probe.metadata().ino(), + std::fs::metadata(&base).unwrap().ino() + ); + + std::fs::remove_file(&base).unwrap(); + std::fs::remove_file(&replaced).unwrap(); + } + + #[test] + fn rejects_symlinks_to_pseudo_filesystems() { + let link = std::env::temp_dir().join(format!( + "ghostscope-module-probe-proc-link-{}", + std::process::id() + )); + symlink("/proc/self/maps", &link).unwrap(); + let err = ModuleProbe::open(link.to_str().unwrap()).unwrap_err(); + assert!(err.to_string().contains("pseudo-filesystem path")); + std::fs::remove_file(&link).unwrap(); + } } diff --git a/ghostscope-process/src/offsets.rs b/ghostscope-process/src/offsets.rs index c8d5bb4a..dc12f0d6 100644 --- a/ghostscope-process/src/offsets.rs +++ b/ghostscope-process/src/offsets.rs @@ -10,7 +10,7 @@ use std::fs; use std::ops::ControlFlow; use std::os::unix::fs::MetadataExt; use std::path::Path; -// no extra imports +use std::sync::{Arc, RwLock}; /// Per-module section offsets (runtime bias) computed from /proc/PID/maps #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] @@ -38,6 +38,79 @@ pub struct ProcessManager { pid_cache: HashMap>, prefilled_pids: HashSet, runtime_pid_aliases: HashMap, + render_pid_cache: HashMap>, + render_snapshot: ProcessManagerSnapshotReader, + render_generation: u64, +} + +/// Immutable process-module state used by backtrace rendering. +/// +/// The mutable manager can spend noticeable time reading `/proc` and probing +/// ELF files. Renderers consume this separately published view so that work +/// never forces an event to fall back to an empty module map. +#[derive(Debug, Default)] +pub struct ProcessManagerSnapshot { + generation: u64, + pid_cache: HashMap>, + runtime_pid_aliases: HashMap, +} + +impl ProcessManagerSnapshot { + pub fn generation(&self) -> u64 { + self.generation + } + + pub fn cached_offsets_with_paths_for_pid(&self, pid: u32) -> Option<&[PidOffsetsEntry]> { + self.pid_cache.get(&pid).map(|entries| entries.as_ref()) + } + + pub fn resolve_runtime_proc_pid(&self, runtime_pid: u32) -> Option { + self.runtime_pid_aliases.get(&runtime_pid).copied() + } + + pub fn candidate_proc_pids_for_runtime_pid( + &self, + runtime_pid: u32, + proc_pid_hint: Option, + ) -> Vec { + let mut pids = Vec::with_capacity(3); + push_unique_pid(&mut pids, proc_pid_hint); + push_unique_pid(&mut pids, self.resolve_runtime_proc_pid(runtime_pid)); + push_unique_pid(&mut pids, Some(runtime_pid)); + pids + } +} + +/// Cheap reader for the most recently published immutable process-module view. +#[derive(Debug, Clone)] +pub struct ProcessManagerSnapshotReader { + current: Arc>>, +} + +impl Default for ProcessManagerSnapshotReader { + fn default() -> Self { + Self { + current: Arc::new(RwLock::new(Arc::new(ProcessManagerSnapshot::default()))), + } + } +} + +impl ProcessManagerSnapshotReader { + pub fn load(&self) -> Arc { + let snapshot = self + .current + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Arc::clone(&snapshot) + } + + fn publish(&self, snapshot: ProcessManagerSnapshot) { + let mut current = self + .current + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *current = Arc::new(snapshot); + } } impl Default for ProcessManager { @@ -187,9 +260,95 @@ impl ProcessManager { pid_cache: HashMap::new(), prefilled_pids: HashSet::new(), runtime_pid_aliases: HashMap::new(), + render_pid_cache: HashMap::new(), + render_snapshot: ProcessManagerSnapshotReader::default(), + render_generation: 0, } } + pub fn snapshot_reader(&self) -> ProcessManagerSnapshotReader { + self.render_snapshot.clone() + } + + /// Private working copy for discovery that may outlive its caller's deadline. + /// It shares neither mutable caches nor snapshot publication with this manager. + pub fn fork_for_runtime_discovery(&self) -> Self { + Self { + module_cache: self.module_cache.clone(), + prefilled_modules: self.prefilled_modules.clone(), + pid_cache: self.pid_cache.clone(), + prefilled_pids: self.prefilled_pids.clone(), + runtime_pid_aliases: self.runtime_pid_aliases.clone(), + ..Self::new() + } + } + + /// Apply discovery only if sysmon has not replaced this PID's mapping meanwhile. + pub fn apply_runtime_pid_snapshot( + &mut self, + pid: u32, + before: Option<&[PidOffsetsEntry]>, + after: &[PidOffsetsEntry], + ) -> bool { + if self.cached_offsets_with_paths_for_pid(pid) != before { + return false; + } + self.pid_cache.insert(pid, after.to_vec()); + self.prefilled_pids.insert(pid); + self.publish_render_pid_snapshot(pid); + true + } + + /// Retain the ELF mapping after validating the opened descriptor against /proc. + pub fn probe_runtime_module(pid: u32, entry: &PidOffsetsEntry) -> Result { + let maps = read_proc_maps(pid)?; + let probe = ModuleProbe::open(&entry.module_path)?; + let opened = ModuleMapIdentityKey::from_metadata(probe.metadata()); + let matches = maps + .iter() + .any(|mapped| runtime_module_matches_map(pid, entry, &opened, mapped)); + anyhow::ensure!( + matches && probe.cookie() == entry.cookie, + "runtime module identity changed while opening {}: opened {:?}, \ + mapping_match={}, expected_cookie={:#x}, opened_cookie={:#x}", + entry.module_path, + opened, + matches, + entry.cookie, + probe.cookie() + ); + Ok(probe) + } + + fn publish_render_snapshot(&mut self) { + self.render_generation = self.render_generation.wrapping_add(1); + self.render_snapshot.publish(ProcessManagerSnapshot { + generation: self.render_generation, + pid_cache: self.render_pid_cache.clone(), + runtime_pid_aliases: self.runtime_pid_aliases.clone(), + }); + } + + fn publish_render_pid_snapshot(&mut self, pid: u32) { + match self.pid_cache.get(&pid) { + Some(entries) + if self + .render_pid_cache + .get(&pid) + .is_some_and(|published| published.as_ref() == entries.as_slice()) => + { + return; + } + Some(entries) => { + self.render_pid_cache + .insert(pid, Arc::from(entries.clone())); + } + None if self.render_pid_cache.remove(&pid).is_none() => return, + None => {} + } + self.publish_render_snapshot(); + } + pub fn ensure_prefill_module(&mut self, module_path: &str) -> Result { if self.prefilled_modules.contains(module_path) { return Ok(0); @@ -359,6 +518,7 @@ impl ProcessManager { } self.pid_cache.insert(pid, list); self.prefilled_pids.insert(pid); + self.publish_render_pid_snapshot(pid); Ok(self.pid_cache.get(&pid).map(|v| v.len()).unwrap_or(0)) } @@ -366,7 +526,11 @@ impl ProcessManager { pub fn refresh_prefill_pid(&mut self, pid: u32) -> Result { self.prefilled_pids.remove(&pid); self.pid_cache.remove(&pid); - self.ensure_prefill_pid(pid) + let result = self.ensure_prefill_pid(pid); + if result.is_err() { + self.publish_render_pid_snapshot(pid); + } + result } /// Refresh only the module mapping that contains one runtime instruction pointer. @@ -393,7 +557,9 @@ impl ProcessManager { .path() .is_some_and(|path| !should_skip_mapped_module_path(path)) }) else { - self.remove_pid_offsets_containing(pid, raw_ip); + if self.remove_pid_offsets_containing(pid, raw_ip) { + self.publish_render_pid_snapshot(pid); + } return Ok(None); }; let Some(mapped_path) = containing.path().map(normalize_mapped_module_path) else { @@ -411,20 +577,32 @@ impl ProcessManager { // The current maps snapshot is authoritative. Remove any cached // identity for this address before opening the mapped file so a // failure cannot revive an unloaded module through stale cache data. - self.remove_pid_offsets_containing(pid, raw_ip); + let removed_stale_entry = self.remove_pid_offsets_containing(pid, raw_ip); let Some((module_path, summary)) = accessible_module_path_for_pid(pid, mapped_path, &summaries) else { + if removed_stale_entry { + self.publish_render_pid_snapshot(pid); + } return Ok(None); }; - let (cookie, offsets, base, size) = self.compute_section_offsets_from_candidates( + let computed = self.compute_section_offsets_from_candidates( pid, &module_path, &summary.candidates, summary.base(), summary.size(), - )?; + ); + let (cookie, offsets, base, size) = match computed { + Ok(computed) => computed, + Err(error) => { + if removed_stale_entry { + self.publish_render_pid_snapshot(pid); + } + return Err(error); + } + }; let entry = PidOffsetsEntry { module_path, cookie, @@ -436,12 +614,14 @@ impl ProcessManager { Ok(Some(entry)) } - fn remove_pid_offsets_containing(&mut self, pid: u32, raw_ip: u64) { + fn remove_pid_offsets_containing(&mut self, pid: u32, raw_ip: u64) -> bool { let Some(entries) = self.pid_cache.get_mut(&pid) else { - return; + return false; }; + let previous_len = entries.len(); entries .retain(|entry| raw_ip < entry.base || raw_ip >= entry.base.saturating_add(entry.size)); + entries.len() != previous_len } fn upsert_pid_offset(&mut self, pid: u32, entry: PidOffsetsEntry) { @@ -455,6 +635,7 @@ impl ProcessManager { entries.push(entry); entries.sort_by_key(|entry| (entry.base, entry.cookie)); self.prefilled_pids.insert(pid); + self.publish_render_pid_snapshot(pid); } fn compute_section_offsets_for_process( @@ -639,10 +820,13 @@ impl ProcessManager { } pub fn record_runtime_pid_alias(&mut self, runtime_pid: u32, proc_pid: u32) { - if runtime_pid == proc_pid { - self.runtime_pid_aliases.remove(&runtime_pid); + let changed = if runtime_pid == proc_pid { + self.runtime_pid_aliases.remove(&runtime_pid).is_some() } else { - self.runtime_pid_aliases.insert(runtime_pid, proc_pid); + self.runtime_pid_aliases.insert(runtime_pid, proc_pid) != Some(proc_pid) + }; + if changed { + self.publish_render_snapshot(); } } @@ -668,9 +852,11 @@ impl ProcessManager { self.pid_cache.remove(&pid); self.runtime_pid_aliases .retain(|runtime_pid, proc_pid| *runtime_pid != pid && *proc_pid != pid); + self.render_pid_cache.remove(&pid); for entries in self.module_cache.values_mut() { entries.retain(|entry| entry.pid != pid); } + self.publish_render_snapshot(); } } @@ -711,6 +897,29 @@ fn proc_root_module_path(pid: u32, mapped_path: &str) -> Option { .then(|| format!("/proc/{pid}/root{mapped_path}")) } +fn runtime_module_matches_map( + pid: u32, + entry: &PidOffsetsEntry, + opened: &ModuleMapIdentityKey, + mapped: &OwnedProcMapEntry, +) -> bool { + let identity = ModuleMapIdentityKey::from_entry(mapped); + // The access path can be rooted through /proc even though maps contains the + // target's absolute path. Strip only this exact PID's root for the existing + // overlayfs same-inode/path fallback; never relax inode or cookie checks. + let proc_root = format!("/proc/{pid}/root"); + let mapped_path = entry + .module_path + .strip_prefix(&proc_root) + .filter(|path| path.starts_with('/')) + .unwrap_or(&entry.module_path); + mapped.executable() + && mapped.start >= entry.base + && mapped.start < entry.base.saturating_add(entry.size) + && (identity == *opened + || (identity.inode == opened.inode && mapped.normalized_path() == Some(mapped_path))) +} + fn is_same_executable_as_current(pid: u32) -> bool { // Strongest signal: dev+ino equality on /proc/*/exe let self_meta = fs::metadata("/proc/self/exe"); @@ -748,6 +957,65 @@ fn is_same_executable_as_current(pid: u32) -> bool { #[cfg(test)] mod tests { use super::*; + + #[test] + fn detached_discovery_cannot_publish_or_overwrite_newer_pid_mappings() { + let entry = |cookie| PidOffsetsEntry { + module_path: "module.so".into(), + cookie, + offsets: Default::default(), + base: 0x1000, + size: 0x100, + }; + let mut manager = ProcessManager::new(); + manager.upsert_pid_offset(42, entry(1)); + let snapshots = manager.snapshot_reader(); + let baseline = manager.fork_for_runtime_discovery(); + let mut worker = baseline.fork_for_runtime_discovery(); + worker.upsert_pid_offset(42, entry(2)); + assert_eq!( + snapshots + .load() + .cached_offsets_with_paths_for_pid(42) + .unwrap()[0] + .cookie, + 1 + ); + manager.upsert_pid_offset(42, entry(3)); + assert!(!manager.apply_runtime_pid_snapshot( + 42, + baseline.cached_offsets_with_paths_for_pid(42), + worker.cached_offsets_with_paths_for_pid(42).unwrap(), + )); + assert_eq!( + manager.cached_offsets_with_paths_for_pid(42).unwrap()[0].cookie, + 3 + ); + } + + #[test] + fn runtime_probe_rejects_a_replacement_with_the_same_build_id() { + let pid = std::process::id(); + let executable = std::env::current_exe().unwrap(); + let mut manager = ProcessManager::new(); + manager.ensure_prefill_pid(pid).unwrap(); + let mut entry = manager + .cached_offsets_with_paths_for_pid(pid) + .unwrap() + .iter() + .find(|entry| Path::new(&entry.module_path) == executable) + .unwrap() + .clone(); + let temp = tempfile::tempdir().unwrap(); + let replacement = temp.path().join("replacement"); + fs::copy(&executable, &replacement).unwrap(); + entry.module_path = replacement.to_str().unwrap().into(); + assert_eq!( + ModuleProbe::open(&entry.module_path).unwrap().cookie(), + entry.cookie + ); + assert!(ProcessManager::probe_runtime_module(pid, &entry).is_err()); + } use crate::proc_maps::parse_maps_line; use std::time::{SystemTime, UNIX_EPOCH}; @@ -837,6 +1105,63 @@ mod tests { ); } + #[test] + fn render_snapshots_remain_available_and_immutable_during_manager_updates() { + let mut mgr = ProcessManager::new(); + mgr.pid_cache.insert( + 42, + vec![PidOffsetsEntry { + module_path: "/tmp/old.so".to_string(), + cookie: 1, + offsets: SectionOffsets::default(), + base: 0x1000, + size: 0x100, + }], + ); + mgr.record_runtime_pid_alias(4242, 42); + mgr.publish_render_pid_snapshot(42); + let snapshots = mgr.snapshot_reader(); + let mgr = Arc::new(std::sync::Mutex::new(mgr)); + + let old_snapshot = { + let _coordinator = mgr.lock().unwrap(); + snapshots.load() + }; + assert_eq!( + old_snapshot.candidate_proc_pids_for_runtime_pid(4242, None), + vec![42, 4242] + ); + assert_eq!( + old_snapshot.cached_offsets_with_paths_for_pid(42).unwrap()[0].module_path, + "/tmp/old.so" + ); + + let mut coordinator = mgr.lock().unwrap(); + coordinator.pid_cache.insert( + 42, + vec![PidOffsetsEntry { + module_path: "/tmp/new.so".to_string(), + cookie: 2, + offsets: SectionOffsets::default(), + base: 0x2000, + size: 0x100, + }], + ); + coordinator.publish_render_pid_snapshot(42); + drop(coordinator); + + let new_snapshot = snapshots.load(); + assert!(new_snapshot.generation() > old_snapshot.generation()); + assert_eq!( + old_snapshot.cached_offsets_with_paths_for_pid(42).unwrap()[0].module_path, + "/tmp/old.so" + ); + assert_eq!( + new_snapshot.cached_offsets_with_paths_for_pid(42).unwrap()[0].module_path, + "/tmp/new.so" + ); + } + #[test] fn targeted_runtime_refresh_replaces_an_overlapping_stale_mapping() { let mut mgr = ProcessManager::new(); @@ -1037,6 +1362,37 @@ mod tests { let _ = std::fs::remove_file(path); } + #[test] + fn runtime_probe_accepts_proc_root_overlay_identity_without_accepting_replacements() { + let mapped: OwnedProcMapEntry = + parse_maps_line("1000-2000 r-xp 00000000 00:01 42 /usr/lib/libc.so.6") + .unwrap() + .into(); + let mut opened = ModuleMapIdentityKey { + dev_major: 0, + dev_minor: 2, + inode: 42, + }; + let mut entry = PidOffsetsEntry { + module_path: "/proc/123/root/usr/lib/libc.so.6".into(), + cookie: 1, + offsets: Default::default(), + base: 0x1000, + size: 0x1000, + }; + assert!(runtime_module_matches_map(123, &entry, &opened, &mapped)); + assert!(!runtime_module_matches_map(124, &entry, &opened, &mapped)); + opened.inode = 43; + assert!(!runtime_module_matches_map(123, &entry, &opened, &mapped)); + opened.inode = 42; + entry.module_path = "/proc/123/root/usr/lib/replacement.so".into(); + assert!(!runtime_module_matches_map(123, &entry, &opened, &mapped)); + entry.module_path = "/proc/123/rooted/usr/lib/libc.so.6".into(); + assert!(!runtime_module_matches_map(123, &entry, &opened, &mapped)); + entry.module_path = "/usr/lib/libc.so.6".into(); + assert!(runtime_module_matches_map(123, &entry, &opened, &mapped)); + } + #[test] fn accessible_proc_root_path_rejects_replaced_file_identity_mismatch() { let suffix = SystemTime::now() diff --git a/ghostscope-process/src/pid/mod.rs b/ghostscope-process/src/pid/mod.rs index 99725755..87e0ed6a 100644 --- a/ghostscope-process/src/pid/mod.rs +++ b/ghostscope-process/src/pid/mod.rs @@ -11,6 +11,7 @@ pub use plan::{ pub use procfs::{ process_exists, read_nspid_chain, read_pid_ns_id, read_pid_ns_inode, INITIAL_PID_NAMESPACE_INO, }; +pub(crate) use resolve::EventProcPidResolver; pub use resolve::{ host_pid_for_proc_pid, resolve_event_pid_for_proc, resolve_input_pid, resolve_proc_pid, resolve_proc_pid_for_event, runtime_pid_candidates_for_proc, diff --git a/ghostscope-process/src/pid/resolve.rs b/ghostscope-process/src/pid/resolve.rs index 2654b22d..a05afa3e 100644 --- a/ghostscope-process/src/pid/resolve.rs +++ b/ghostscope-process/src/pid/resolve.rs @@ -2,6 +2,10 @@ use super::procfs::{ process_exists, read_nspid_chain, read_nspid_chain_from_status, read_pid_ns_id, read_status, }; use super::types::{PidResolveSource, PidViews}; +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +const EVENT_PID_INDEX_REFRESH_INTERVAL: Duration = Duration::from_millis(50); fn push_unique_pid(pids: &mut Vec, pid: u32) { if !pids.contains(&pid) { @@ -59,26 +63,93 @@ pub fn host_pid_for_proc_pid(proc_pid: u32) -> u32 { /// Resolve a kernel event PID (initial PID namespace) to the `/proc` PID in the /// current userspace namespace when possible. pub fn resolve_proc_pid_for_event(event_pid: u32) -> u32 { - if std::path::Path::new(&format!("/proc/{event_pid}")).exists() { + if process_exists(event_pid) { return event_pid; } - if let Ok(dir) = std::fs::read_dir("/proc") { - for ent in dir.flatten() { - let file_name = ent.file_name(); - let Ok(proc_pid) = file_name.to_string_lossy().parse::() else { - continue; - }; - let Some(chain) = read_nspid_chain(proc_pid) else { - continue; - }; - if chain.first().copied() == Some(event_pid) { - return proc_pid; + event_pid_index() + .get(&event_pid) + .copied() + .unwrap_or(event_pid) +} + +/// Cached resolver for bursts of sysmon events. +/// +/// Short-lived processes often disappear before userspace drains their map-change events. A +/// direct lookup then fails, but rebuilding the complete namespace PID index for every stale event +/// turns an event burst into repeated full `/proc` scans. Keep one short-lived snapshot so a burst +/// pays that cost at most once per refresh interval. +#[derive(Debug)] +pub(crate) struct EventProcPidResolver { + event_pid_to_proc_pid: HashMap, + refreshed_at: Option, + refresh_interval: Duration, + #[cfg(test)] + refresh_count: usize, +} + +impl EventProcPidResolver { + pub(crate) fn new() -> Self { + Self::with_refresh_interval(EVENT_PID_INDEX_REFRESH_INTERVAL) + } + + fn with_refresh_interval(refresh_interval: Duration) -> Self { + Self { + event_pid_to_proc_pid: HashMap::new(), + refreshed_at: None, + refresh_interval, + #[cfg(test)] + refresh_count: 0, + } + } + + pub(crate) fn resolve(&mut self, event_pid: u32) -> u32 { + if process_exists(event_pid) { + return event_pid; + } + + let now = Instant::now(); + let index_is_stale = self + .refreshed_at + .map(|refreshed_at| now.duration_since(refreshed_at) >= self.refresh_interval) + .unwrap_or(true); + if index_is_stale { + self.event_pid_to_proc_pid = event_pid_index(); + self.refreshed_at = Some(now); + #[cfg(test)] + { + self.refresh_count += 1; } } + + self.event_pid_to_proc_pid + .get(&event_pid) + .copied() + .unwrap_or(event_pid) } +} + +fn event_pid_index() -> HashMap { + let mut event_pid_to_proc_pid = HashMap::new(); + let Ok(dir) = std::fs::read_dir("/proc") else { + return event_pid_to_proc_pid; + }; - event_pid + for ent in dir.flatten() { + let file_name = ent.file_name(); + let Ok(proc_pid) = file_name.to_string_lossy().parse::() else { + continue; + }; + let Some(chain) = read_nspid_chain(proc_pid) else { + continue; + }; + let Some(event_pid) = chain.first().copied() else { + continue; + }; + event_pid_to_proc_pid.insert(event_pid, proc_pid); + } + + event_pid_to_proc_pid } /// Resolve a `/proc` PID back to the host-view event PID when possible. @@ -117,4 +188,13 @@ mod tests { vec![531, 1000, 17] ); } + + #[test] + fn event_pid_resolver_reuses_one_proc_snapshot_for_missing_pids() { + let mut resolver = EventProcPidResolver::with_refresh_interval(Duration::from_secs(1)); + + assert_eq!(resolver.resolve(u32::MAX), u32::MAX); + assert_eq!(resolver.resolve(u32::MAX - 1), u32::MAX - 1); + assert_eq!(resolver.refresh_count, 1); + } } diff --git a/ghostscope-process/src/sysmon/events.rs b/ghostscope-process/src/sysmon/events.rs index 94c050e8..7915f958 100644 --- a/ghostscope-process/src/sysmon/events.rs +++ b/ghostscope-process/src/sysmon/events.rs @@ -2,6 +2,180 @@ use super::offset_refresh::*; use super::pending::*; use super::pid_alias::*; use super::*; +use std::cell::RefCell; +use std::collections::{HashMap, VecDeque}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +struct MapChangeKey { + event_pid: u32, + host_pid: u32, +} + +impl MapChangeKey { + fn from_event(ev: SysEvent) -> Self { + Self { + event_pid: ev.tgid, + host_pid: sys_event_host_pid(&ev), + } + } +} + +#[derive(Debug, Clone, Copy)] +struct QueuedMapChange { + event: SysEvent, + ready_at: Instant, +} + +/// Coalesces noisy map-change events by process before any `/proc` work is performed. +/// +/// An entry remains queued during the debounce window, so a process repeatedly calling mmap does +/// not make sysmon rescan the same maps on every syscall. Queue overflow is safe because the +/// periodic target-module reconciliation remains the correctness fallback. +#[derive(Debug)] +pub(super) struct CoalescedMapChanges { + entries: HashMap, + order: VecDeque, + last_processed: HashMap, + capacity: usize, + debounce_interval: Duration, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum SysmonWorkEnqueueResult { + Queued, + LifecycleQueueFull, + MapChangeQueueFull, +} + +/// Bounded handoff between the eBPF event reader and the sysmon work thread. +/// +/// Lifecycle events retain FIFO order in their own queue. Map changes use the +/// existing per-process coalescer, so mmap-heavy workloads cannot crowd exec, +/// fork, or exit work out of the handoff queue. +#[derive(Debug)] +pub(super) struct SysmonWorkQueue { + lifecycle_events: VecDeque, + map_changes: CoalescedMapChanges, + lifecycle_capacity: usize, +} + +impl SysmonWorkQueue { + pub(super) fn new( + lifecycle_capacity: usize, + map_change_capacity: usize, + map_change_debounce_interval: Duration, + ) -> Self { + Self { + lifecycle_events: VecDeque::new(), + map_changes: CoalescedMapChanges::new( + map_change_capacity, + map_change_debounce_interval, + ), + lifecycle_capacity, + } + } + + pub(super) fn enqueue(&mut self, event: SysEvent, now: Instant) -> SysmonWorkEnqueueResult { + if event.event_kind() == Some(SysEventKind::MapChange) { + return if self.map_changes.enqueue(event, now) { + SysmonWorkEnqueueResult::Queued + } else { + SysmonWorkEnqueueResult::MapChangeQueueFull + }; + } + + if self.lifecycle_events.len() >= self.lifecycle_capacity { + return SysmonWorkEnqueueResult::LifecycleQueueFull; + } + self.lifecycle_events.push_back(event); + SysmonWorkEnqueueResult::Queued + } + + pub(super) fn pop_lifecycle(&mut self) -> Option { + self.lifecycle_events.pop_front() + } + + pub(super) fn pop_ready_map_change(&mut self, now: Instant) -> Option { + self.map_changes.pop_ready(now) + } + + #[cfg(test)] + fn lifecycle_len(&self) -> usize { + self.lifecycle_events.len() + } + + #[cfg(test)] + fn map_change_len(&self) -> usize { + self.map_changes.len() + } +} + +impl CoalescedMapChanges { + pub(super) fn new(capacity: usize, debounce_interval: Duration) -> Self { + Self { + entries: HashMap::new(), + order: VecDeque::new(), + last_processed: HashMap::new(), + capacity, + debounce_interval, + } + } + + pub(super) fn enqueue(&mut self, event: SysEvent, now: Instant) -> bool { + let key = MapChangeKey::from_event(event); + if let Some(queued) = self.entries.get_mut(&key) { + queued.event = event; + return true; + } + if self.entries.len() >= self.capacity { + return false; + } + + if self.last_processed.len() >= self.capacity { + self.last_processed.retain(|_, processed_at| { + now.saturating_duration_since(*processed_at) < self.debounce_interval + }); + } + let ready_at = self + .last_processed + .get(&key) + .and_then(|processed_at| processed_at.checked_add(self.debounce_interval)) + .unwrap_or(now); + self.entries + .insert(key, QueuedMapChange { event, ready_at }); + self.order.push_back(key); + true + } + + pub(super) fn pop_ready(&mut self, now: Instant) -> Option { + let queued_len = self.order.len(); + for _ in 0..queued_len { + let key = self.order.pop_front()?; + let Some(queued) = self.entries.get(&key).copied() else { + continue; + }; + if queued.ready_at > now { + self.order.push_back(key); + continue; + } + + self.entries.remove(&key); + self.last_processed.insert(key, now); + return Some(queued.event); + } + None + } + + #[cfg(test)] + fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + #[cfg(test)] + fn len(&self) -> usize { + self.entries.len() + } +} pub(super) fn try_publish_sys_event(tx: &mpsc::SyncSender, ev: SysEvent) -> bool { match tx.try_send(ev) { @@ -307,6 +481,22 @@ pub(super) fn sysmon_proc_pid_resolver( } } +pub(super) fn cached_sysmon_proc_pid_resolver( + watched_event_pid: Option, + watched_proc_pid: Option, +) -> impl Fn(u32) -> u32 { + let resolver = RefCell::new(EventProcPidResolver::new()); + move |event_pid| { + if watched_event_pid == Some(event_pid) { + if let Some(proc_pid) = watched_proc_pid { + return proc_pid; + } + } + + resolver.borrow_mut().resolve(event_pid) + } +} + pub(super) fn sys_event_host_pid(ev: &SysEvent) -> u32 { if ev.host_tgid != 0 { ev.host_tgid @@ -314,3 +504,108 @@ pub(super) fn sys_event_host_pid(ev: &SysEvent) -> u32 { ev.tgid } } + +#[cfg(test)] +mod tests { + use super::*; + + fn map_change(event_pid: u32, host_pid: u32) -> SysEvent { + SysEvent { + tgid: event_pid, + host_tgid: host_pid, + kind: SysEventKind::MapChange.as_u32(), + } + } + + #[test] + fn coalesced_map_changes_keep_one_entry_per_pid() { + let now = Instant::now(); + let mut changes = CoalescedMapChanges::new(4, Duration::from_millis(75)); + + assert!(changes.enqueue(map_change(10, 20), now)); + assert!(changes.enqueue(map_change(10, 20), now)); + assert_eq!(changes.len(), 1); + + let event = changes.pop_ready(now).expect("coalesced event"); + assert_eq!(event.tgid, 10); + assert_eq!(event.host_tgid, 20); + assert!(changes.is_empty()); + } + + #[test] + fn coalesced_map_changes_debounce_repeated_pid() { + let now = Instant::now(); + let debounce = Duration::from_millis(75); + let mut changes = CoalescedMapChanges::new(4, debounce); + + assert!(changes.enqueue(map_change(10, 10), now)); + assert!(changes.pop_ready(now).is_some()); + assert!(changes.enqueue(map_change(10, 10), now)); + assert!(changes.pop_ready(now + debounce / 2).is_none()); + assert!(changes.pop_ready(now + debounce).is_some()); + } + + #[test] + fn coalesced_map_changes_bound_unique_pid_queue() { + let now = Instant::now(); + let mut changes = CoalescedMapChanges::new(1, Duration::from_millis(75)); + + assert!(changes.enqueue(map_change(10, 10), now)); + assert!(changes.enqueue(map_change(10, 10), now)); + assert!(!changes.enqueue(map_change(11, 11), now)); + assert_eq!(changes.len(), 1); + } + + #[test] + fn work_queue_keeps_lifecycle_events_separate_from_map_churn() { + let now = Instant::now(); + let mut queue = SysmonWorkQueue::new(2, 1, Duration::from_millis(75)); + let exec = SysEvent { + tgid: 30, + host_tgid: 30, + kind: SysEventKind::Exec.as_u32(), + }; + + assert_eq!( + queue.enqueue(map_change(10, 10), now), + SysmonWorkEnqueueResult::Queued + ); + assert_eq!( + queue.enqueue(map_change(11, 11), now), + SysmonWorkEnqueueResult::MapChangeQueueFull + ); + assert_eq!(queue.enqueue(exec, now), SysmonWorkEnqueueResult::Queued); + + assert_eq!(queue.lifecycle_len(), 1); + assert_eq!(queue.map_change_len(), 1); + assert_eq!(queue.pop_lifecycle().map(|event| event.tgid), Some(30)); + assert_eq!( + queue.pop_ready_map_change(now).map(|event| event.tgid), + Some(10) + ); + } + + #[test] + fn work_queue_bounds_lifecycle_events() { + let now = Instant::now(); + let mut queue = SysmonWorkQueue::new(1, 1, Duration::from_millis(75)); + let lifecycle = |pid, kind| SysEvent { + tgid: pid, + host_tgid: pid, + kind: SysEventKind::from_u32(kind) + .expect("valid lifecycle kind") + .as_u32(), + }; + + assert_eq!( + queue.enqueue(lifecycle(10, 1), now), + SysmonWorkEnqueueResult::Queued + ); + assert_eq!( + queue.enqueue(lifecycle(11, 2), now), + SysmonWorkEnqueueResult::LifecycleQueueFull + ); + assert_eq!(queue.lifecycle_len(), 1); + assert_eq!(queue.pop_lifecycle().map(|event| event.tgid), Some(10)); + } +} diff --git a/ghostscope-process/src/sysmon/mod.rs b/ghostscope-process/src/sysmon/mod.rs index 4a2a5899..68fe9aaa 100644 --- a/ghostscope-process/src/sysmon/mod.rs +++ b/ghostscope-process/src/sysmon/mod.rs @@ -3,7 +3,7 @@ use crate::{ offsets::{PidOffsetsEntry, ProcessManager}, pid::{ resolve_event_pid_for_proc, resolve_proc_pid_for_event, runtime_pid_candidates_for_proc, - PidNamespaceId, + EventProcPidResolver, PidNamespaceId, }, pinned_bpf_maps, proc_maps::{ @@ -129,7 +129,7 @@ impl Default for SysmonEventMask { /// Keep repr(C), field order and sizes identical on both sides. Current /// layout (12 bytes): { tgid: u32, host_tgid: u32, kind: u32 }. #[repr(C)] -#[derive(Clone, Copy)] +#[derive(Debug, Clone, Copy)] pub struct SysEvent { /// Runtime TGID in the configured sysmon event namespace when available. pub tgid: u32, @@ -149,6 +149,22 @@ const PENDING_MAX_ATTEMPTS: u32 = 20; const MAP_CHANGE_DEBOUNCE_INTERVAL: Duration = Duration::from_millis(75); const MODULE_REFRESH_INTERVAL: Duration = Duration::from_millis(250); const SYSMON_EVENT_QUEUE_CAPACITY: usize = 1024; +#[cfg(feature = "sysmon-ebpf")] +const SYSMON_RING_DRAIN_EVENT_LIMIT: usize = 4096; +#[cfg(feature = "sysmon-ebpf")] +const SYSMON_RING_DRAIN_TIME_BUDGET: Duration = Duration::from_millis(2); +#[cfg(feature = "sysmon-ebpf")] +const SYSMON_WORK_LIFECYCLE_QUEUE_CAPACITY: usize = 16_384; +#[cfg(feature = "sysmon-ebpf")] +const SYSMON_WORK_LIFECYCLE_PROCESS_LIMIT: usize = 64; +#[cfg(feature = "sysmon-ebpf")] +const SYSMON_WORK_LIFECYCLE_TIME_BUDGET: Duration = Duration::from_millis(5); +#[cfg(feature = "sysmon-ebpf")] +const SYSMON_MAP_CHANGE_PROCESS_LIMIT: usize = 8; +#[cfg(feature = "sysmon-ebpf")] +const SYSMON_MAP_CHANGE_PROCESS_TIME_BUDGET: Duration = Duration::from_millis(5); +#[cfg(feature = "sysmon-ebpf")] +const SYSMON_MAP_CHANGE_QUEUE_CAPACITY: usize = 16_384; #[cfg(feature = "sysmon-ebpf")] const SYSMON_EVENT_MASK_EXEC: u32 = 1 << 0; diff --git a/ghostscope-process/src/sysmon/offset_refresh.rs b/ghostscope-process/src/sysmon/offset_refresh.rs index 39ba5cd0..4b7a55a5 100644 --- a/ghostscope-process/src/sysmon/offset_refresh.rs +++ b/ghostscope-process/src/sysmon/offset_refresh.rs @@ -557,7 +557,7 @@ pub(super) fn refresh_target_module_offsets( false } }; - match publish_offsets_for_runtime_pid_keys( + match publish_target_offsets_for_runtime_pid_keys( pid, event_pid, &runtime_pids, @@ -602,10 +602,9 @@ pub(super) fn refresh_target_module_offsets( if target_pid_map_signatures.get(pid) == Some(&maps_signature) { continue; } - target_pid_map_signatures.insert(*pid, maps_signature); - match refresh_full_offsets_for_pid(mgr, *pid, event_pid) { Ok(true) => { + target_pid_map_signatures.insert(*pid, maps_signature); newly_allowed_event_pids.insert(event_pid); } Ok(false) => {} diff --git a/ghostscope-process/src/sysmon/pending.rs b/ghostscope-process/src/sysmon/pending.rs index 93132eff..329b4a52 100644 --- a/ghostscope-process/src/sysmon/pending.rs +++ b/ghostscope-process/src/sysmon/pending.rs @@ -113,7 +113,7 @@ impl PendingOffsets { #[derive(Debug, Clone)] pub(super) struct PendingMapRefreshEntry { - pub(super) last_seen: Instant, + pub(super) ready_at: Instant, pub(super) event_pid: u32, pub(super) host_pid: u32, } @@ -145,29 +145,40 @@ impl PendingMapRefreshes { } pub(super) fn register(&mut self, event_pid: u32, host_pid: u32, proc_pid: u32) { - self.entries.insert( - proc_pid, - PendingMapRefreshEntry { - last_seen: Instant::now(), + self.register_at(event_pid, host_pid, proc_pid, Instant::now()); + } + + pub(super) fn take_due(&mut self) -> Vec { + self.take_due_at(Instant::now()) + } + + fn register_at(&mut self, event_pid: u32, host_pid: u32, proc_pid: u32, now: Instant) { + self.entries + .entry(proc_pid) + .and_modify(|entry| { + // Keep the original deadline so continuous mmap traffic cannot + // postpone this PID's refresh forever. The latest PID aliases + // are still used when the work becomes due. + entry.event_pid = event_pid; + entry.host_pid = host_pid; + }) + .or_insert_with(|| PendingMapRefreshEntry { + ready_at: now.checked_add(MAP_CHANGE_DEBOUNCE_INTERVAL).unwrap_or(now), event_pid, host_pid, - }, - ); + }); } - pub(super) fn take_due(&mut self) -> Vec { - let now = Instant::now(); + fn take_due_at(&mut self, now: Instant) -> Vec { let due: Vec = self .entries .iter() .filter_map(|(&proc_pid, entry)| { - (now.duration_since(entry.last_seen) >= MAP_CHANGE_DEBOUNCE_INTERVAL).then_some( - PendingMapRefreshDue { - event_pid: entry.event_pid, - host_pid: entry.host_pid, - proc_pid, - }, - ) + (now >= entry.ready_at).then_some(PendingMapRefreshDue { + event_pid: entry.event_pid, + host_pid: entry.host_pid, + proc_pid, + }) }) .collect(); for entry in &due { @@ -176,3 +187,35 @@ impl PendingMapRefreshes { due } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn repeated_map_changes_do_not_postpone_the_first_refresh_deadline() { + let started_at = Instant::now(); + let mut pending = PendingMapRefreshes::new(); + pending.register_at(10, 20, 30, started_at); + pending.register_at(11, 21, 30, started_at + MAP_CHANGE_DEBOUNCE_INTERVAL / 2); + + let due = pending.take_due_at(started_at + MAP_CHANGE_DEBOUNCE_INTERVAL); + assert_eq!(due.len(), 1); + assert_eq!(due[0].event_pid, 11); + assert_eq!(due[0].host_pid, 21); + assert_eq!(due[0].proc_pid, 30); + assert!(pending.entries.is_empty()); + } + + #[test] + fn map_refresh_waits_for_its_debounce_deadline() { + let started_at = Instant::now(); + let mut pending = PendingMapRefreshes::new(); + pending.register_at(10, 20, 30, started_at); + + assert!(pending + .take_due_at(started_at + MAP_CHANGE_DEBOUNCE_INTERVAL / 2) + .is_empty()); + assert_eq!(pending.entries.len(), 1); + } +} diff --git a/ghostscope-process/src/sysmon/pid_alias.rs b/ghostscope-process/src/sysmon/pid_alias.rs index 029d2725..91159452 100644 --- a/ghostscope-process/src/sysmon/pid_alias.rs +++ b/ghostscope-process/src/sysmon/pid_alias.rs @@ -102,6 +102,29 @@ pub(super) fn publish_offsets_for_runtime_pid_keys( runtime_pids: &[u32], items: &[(u64, crate::pinned_bpf_maps::ProcModuleOffsetsValue)], log_context: &str, +) -> anyhow::Result { + write_offsets_for_runtime_pid_keys(proc_pid, event_pid, runtime_pids, items, log_context, true) +} + +/// A target-only update is not a complete process map snapshot. Keep the +/// existing ranges until a full refresh can replace them, including unloads. +pub(super) fn publish_target_offsets_for_runtime_pid_keys( + proc_pid: u32, + event_pid: u32, + runtime_pids: &[u32], + items: &[(u64, crate::pinned_bpf_maps::ProcModuleOffsetsValue)], + log_context: &str, +) -> anyhow::Result { + write_offsets_for_runtime_pid_keys(proc_pid, event_pid, runtime_pids, items, log_context, false) +} + +fn write_offsets_for_runtime_pid_keys( + proc_pid: u32, + event_pid: u32, + runtime_pids: &[u32], + items: &[(u64, crate::pinned_bpf_maps::ProcModuleOffsetsValue)], + log_context: &str, + complete_snapshot: bool, ) -> anyhow::Result { use crate::pinned_bpf_maps::{insert_offsets_for_pid, replace_ranges_for_pid}; @@ -121,6 +144,9 @@ pub(super) fn publish_offsets_for_runtime_pid_keys( continue; } total_inserted += inserted; + if !complete_snapshot { + continue; + } if let Err(e) = replace_ranges_for_pid(*runtime_pid, items) { tracing::warn!( "Sysmon: failed to replace module ranges for {} runtime pid {} (event pid {}, proc pid {}): {}", diff --git a/ghostscope-process/src/sysmon/runtime_loop.rs b/ghostscope-process/src/sysmon/runtime_loop.rs index 9d40641f..b641144a 100644 --- a/ghostscope-process/src/sysmon/runtime_loop.rs +++ b/ghostscope-process/src/sysmon/runtime_loop.rs @@ -4,6 +4,380 @@ use super::offset_refresh::*; use super::pid_alias::*; use super::*; +#[cfg(feature = "sysmon-ebpf")] +struct SysmonLoopContext<'a, F, M> { + mgr: &'a Arc>, + target: &'a Option, + pending: &'a Arc>, + pending_map_refreshes: &'a Arc>, + proc_pid_for_event: &'a F, + proc_pid_for_map_change: &'a M, + tx: &'a mpsc::SyncSender, +} + +#[cfg(feature = "sysmon-ebpf")] +type SharedSysmonWorkQueue = Arc>; + +#[cfg(feature = "sysmon-ebpf")] +struct SysmonWorker { + queue: SharedSysmonWorkQueue, + wake_tx: Option>, + handle: Option>, + lifecycle_queue_overflow_reported: bool, + map_change_queue_overflow_reported: bool, +} + +#[cfg(feature = "sysmon-ebpf")] +impl SysmonWorker { + fn spawn( + mgr: Arc>, + cfg: SysmonConfig, + pending: Arc>, + pending_map_refreshes: Arc>, + tx: mpsc::SyncSender, + ) -> anyhow::Result { + let queue = Arc::new(Mutex::new(SysmonWorkQueue::new( + SYSMON_WORK_LIFECYCLE_QUEUE_CAPACITY, + SYSMON_MAP_CHANGE_QUEUE_CAPACITY, + MAP_CHANGE_DEBOUNCE_INTERVAL, + ))); + let worker_queue = Arc::clone(&queue); + let (wake_tx, wake_rx) = mpsc::sync_channel(1); + let handle = thread::Builder::new() + .name("gs-sysmon-work".to_string()) + .spawn(move || { + info!("Sysmon work thread started"); + if let Err(error) = run_sysmon_worker( + mgr, + cfg, + pending, + pending_map_refreshes, + tx, + worker_queue, + wake_rx, + ) { + error!("Sysmon work thread failed: {error:#}"); + } + info!("Sysmon work thread exiting"); + }) + .map_err(|error| anyhow::anyhow!("failed to spawn sysmon work thread: {error}"))?; + + Ok(Self { + queue, + wake_tx: Some(wake_tx), + handle: Some(handle), + lifecycle_queue_overflow_reported: false, + map_change_queue_overflow_reported: false, + }) + } + + fn enqueue(&mut self, event: SysEvent) -> anyhow::Result<()> { + let result = self + .queue + .lock() + .map_err(|_| anyhow::anyhow!("sysmon work queue lock poisoned"))? + .enqueue(event, Instant::now()); + + match result { + SysmonWorkEnqueueResult::Queued => {} + SysmonWorkEnqueueResult::LifecycleQueueFull => { + if !self.lifecycle_queue_overflow_reported { + warn!( + "Sysmon lifecycle work queue reached capacity {}; dropping events while periodic reconciliation remains active", + SYSMON_WORK_LIFECYCLE_QUEUE_CAPACITY + ); + self.lifecycle_queue_overflow_reported = true; + } + } + SysmonWorkEnqueueResult::MapChangeQueueFull => { + if !self.map_change_queue_overflow_reported { + warn!( + "Sysmon coalesced map-change queue reached capacity {}; relying on periodic reconciliation", + SYSMON_MAP_CHANGE_QUEUE_CAPACITY + ); + self.map_change_queue_overflow_reported = true; + } + } + } + + let wake_tx = self + .wake_tx + .as_ref() + .ok_or_else(|| anyhow::anyhow!("sysmon work thread is stopped"))?; + match wake_tx.try_send(()) { + Ok(()) | Err(mpsc::TrySendError::Full(())) => Ok(()), + Err(mpsc::TrySendError::Disconnected(())) => { + Err(anyhow::anyhow!("sysmon work thread stopped unexpectedly")) + } + } + } + + fn ensure_running(&self) -> anyhow::Result<()> { + if self.handle.as_ref().is_some_and(JoinHandle::is_finished) { + Err(anyhow::anyhow!("sysmon work thread stopped unexpectedly")) + } else { + Ok(()) + } + } +} + +#[cfg(feature = "sysmon-ebpf")] +impl Drop for SysmonWorker { + fn drop(&mut self) { + self.wake_tx.take(); + if let Some(handle) = self.handle.take() { + if handle.join().is_err() { + warn!("Sysmon work thread panicked"); + } + } + } +} + +#[cfg(feature = "sysmon-ebpf")] +fn pop_lifecycle_work(queue: &SharedSysmonWorkQueue) -> anyhow::Result> { + queue + .lock() + .map_err(|_| anyhow::anyhow!("sysmon work queue lock poisoned")) + .map(|mut queue| queue.pop_lifecycle()) +} + +#[cfg(feature = "sysmon-ebpf")] +fn pop_ready_map_change_work( + queue: &SharedSysmonWorkQueue, + now: Instant, +) -> anyhow::Result> { + queue + .lock() + .map_err(|_| anyhow::anyhow!("sysmon work queue lock poisoned")) + .map(|mut queue| queue.pop_ready_map_change(now)) +} + +#[cfg(feature = "sysmon-ebpf")] +fn process_lifecycle_events u32, M: Fn(u32) -> u32>( + context: &SysmonLoopContext<'_, F, M>, + queue: &SharedSysmonWorkQueue, +) -> anyhow::Result { + let started_at = Instant::now(); + let mut processed = 0; + while processed < SYSMON_WORK_LIFECYCLE_PROCESS_LIMIT { + let Some(event) = pop_lifecycle_work(queue)? else { + break; + }; + let matched = dispatch_sysmon_event( + context.mgr, + context.target, + context.pending, + context.pending_map_refreshes, + context.proc_pid_for_event, + &event, + ); + if matched { + try_publish_sys_event(context.tx, event); + } + processed += 1; + if started_at.elapsed() >= SYSMON_WORK_LIFECYCLE_TIME_BUDGET { + break; + } + } + Ok(processed) +} + +#[cfg(feature = "sysmon-ebpf")] +fn process_coalesced_map_changes u32, M: Fn(u32) -> u32>( + context: &SysmonLoopContext<'_, F, M>, + queue: &SharedSysmonWorkQueue, +) -> anyhow::Result { + let started_at = Instant::now(); + let mut processed = 0; + while processed < SYSMON_MAP_CHANGE_PROCESS_LIMIT { + let Some(ev) = pop_ready_map_change_work(queue, Instant::now())? else { + break; + }; + let matched = dispatch_sysmon_event( + context.mgr, + context.target, + context.pending, + context.pending_map_refreshes, + context.proc_pid_for_map_change, + &ev, + ); + if matched { + try_publish_sys_event(context.tx, ev); + } + processed += 1; + if started_at.elapsed() >= SYSMON_MAP_CHANGE_PROCESS_TIME_BUDGET { + break; + } + } + Ok(processed) +} + +#[cfg(feature = "sysmon-ebpf")] +fn service_sysmon_maintenance u32, M: Fn(u32) -> u32>( + context: &SysmonLoopContext<'_, F, M>, + queue: &SharedSysmonWorkQueue, + last_module_refresh: &mut Instant, + target_pid_map_signatures: &mut HashMap, +) -> anyhow::Result { + // Lifecycle work is the latency-sensitive path. It is always selected before + // scans, retries, and map-change refreshes, regardless of mmap event volume. + let processed_lifecycle = process_lifecycle_events(context, queue)?; + // Reconciliation is the correctness fallback when lifecycle or map-change events are delayed + // or dropped. Service it before lower-priority per-PID map work so event pressure cannot defer + // the refresh past its deadline. + refresh_target_module_offsets( + context.mgr, + context.target.as_deref(), + last_module_refresh, + target_pid_map_signatures, + context.tx, + ); + poll_pending_offsets(context.mgr, context.pending, context.proc_pid_for_event); + let processed_map_changes = process_coalesced_map_changes(context, queue)?; + poll_pending_map_refreshes( + context.mgr, + context.target.as_deref(), + context.pending_map_refreshes, + context.pending, + context.tx, + ); + Ok(processed_lifecycle + processed_map_changes) +} + +#[cfg(feature = "sysmon-ebpf")] +fn initialize_target_offsets( + mgr: &Arc>, + target: Option<&Path>, + proc_pid_for_event: &impl Fn(u32) -> u32, +) { + let Some(target_path) = target else { + return; + }; + + let mut initial_target_pids = BTreeSet::new(); + if let Ok(mut guard) = mgr.lock() { + if let Ok(prefilled) = guard.ensure_prefill_module(target_path.to_string_lossy().as_ref()) { + tracing::info!( + "Sysmon: initial prefill cached {} pid(s) for module {}", + prefilled, + target_path.display() + ); + let entries = guard.cached_offsets_for_module(target_path.to_string_lossy().as_ref()); + if !entries.is_empty() { + use crate::pinned_bpf_maps::ProcModuleOffsetsValue; + let mut by_pid: HashMap> = HashMap::new(); + for (pid, cookie, offsets, base, size) in entries { + if is_current_process_pid(pid) { + continue; + } + by_pid.entry(pid).or_default().push(( + cookie, + ProcModuleOffsetsValue::new( + offsets.text, + offsets.rodata, + offsets.data, + offsets.bss, + base, + size, + ), + )); + } + let mut total = 0; + for (pid, items) in by_pid { + initial_target_pids.insert(pid); + let event_pid = resolve_event_pid_for_proc(pid); + let runtime_pids = runtime_pid_keys_for_proc_event(pid, event_pid, []); + for runtime_pid in &runtime_pids { + write_pinned_runtime_pid_alias(*runtime_pid, pid); + guard.record_runtime_pid_alias(*runtime_pid, pid); + } + if let Ok(inserted) = publish_offsets_for_runtime_pid_keys( + pid, + event_pid, + &runtime_pids, + &items, + "initial prefill", + ) { + total += inserted; + } + insert_allowed_runtime_pid_keys(&runtime_pids); + } + tracing::info!( + "Sysmon: initial inserted {} offset entries for module {}", + total, + target_path.display() + ); + } + } + } + + for pid in initial_target_pids { + let event_pid = resolve_event_pid_for_proc(pid); + if let Err(error) = prefill_full_offsets_for_pid_if_new(mgr, event_pid, proc_pid_for_event) + { + tracing::debug!( + "Sysmon: initial full offset prefill failed for proc pid {} (event pid {}): {}", + pid, + event_pid, + error + ); + } + } +} + +#[cfg(feature = "sysmon-ebpf")] +fn run_sysmon_worker( + mgr: Arc>, + cfg: SysmonConfig, + pending: Arc>, + pending_map_refreshes: Arc>, + tx: mpsc::SyncSender, + queue: SharedSysmonWorkQueue, + wake_rx: mpsc::Receiver<()>, +) -> anyhow::Result<()> { + let target = cfg.target_module; + let proc_pid_for_event = sysmon_proc_pid_resolver(cfg.watched_pid, cfg.watched_proc_pid); + // Only noisy map-change handling uses a short-lived `/proc` index snapshot. Lifecycle events + // keep fresh resolution semantics so a newly visible short-lived process cannot hit a cached + // miss from an earlier unrelated event. + let proc_pid_for_map_change = + cached_sysmon_proc_pid_resolver(cfg.watched_pid, cfg.watched_proc_pid); + + initialize_target_offsets(&mgr, target.as_deref(), &proc_pid_for_event); + tracing::info!("Sysmon: setup complete"); + + // Initial prefill already ran above. Do not make the first periodic module + // refresh immediately due: for `-t executable`, the exec event is the fast + // path that inserts proc_module_offsets and allowed_pids. A fallback /proc + // scan here can delay a short-lived target past its only probe. + let mut last_module_refresh = Instant::now(); + let mut target_pid_map_signatures = HashMap::::new(); + let context = SysmonLoopContext { + mgr: &mgr, + target: &target, + pending: &pending, + pending_map_refreshes: &pending_map_refreshes, + proc_pid_for_event: &proc_pid_for_event, + proc_pid_for_map_change: &proc_pid_for_map_change, + tx: &tx, + }; + + loop { + let processed = service_sysmon_maintenance( + &context, + &queue, + &mut last_module_refresh, + &mut target_pid_map_signatures, + )?; + if processed == 0 { + match wake_rx.recv_timeout(Duration::from_millis(5)) { + Ok(()) | Err(mpsc::RecvTimeoutError::Timeout) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => return Ok(()), + } + } + } +} + #[cfg(feature = "sysmon-ebpf")] pub(super) fn run_sysmon_loop( mgr: Arc>, @@ -33,136 +407,44 @@ pub(super) fn run_sysmon_loop( warn!("sysmon-bpf object missing; running in stub mode (no realtime process events)"); return Ok(()); } - let target = cfg.target_module.clone(); let use_verbose = cfg!(debug_assertions) || log_enabled!(LogLevel::Trace) || log_enabled!(LogLevel::Debug); let mut bpf = load_and_attach_sysmon_bpf(obj, &cfg, use_verbose)?; - let proc_pid_for_event = sysmon_proc_pid_resolver(cfg.watched_pid, cfg.watched_proc_pid); - - // Using allowlist-based gating in kernel; userspace decides allow on exec. - - // Initial prefill for late-start cases: compute and insert offsets for already-running PIDs. - if let Some(tpath) = &target { - let mut initial_target_pids: BTreeSet = BTreeSet::new(); - if let Ok(mut guard) = mgr.lock() { - if let Ok(prefilled) = guard.ensure_prefill_module(tpath.to_string_lossy().as_ref()) { - tracing::info!( - "Sysmon: initial prefill cached {} pid(s) for module {}", - prefilled, - tpath.display() - ); - let entries = guard.cached_offsets_for_module(tpath.to_string_lossy().as_ref()); - if !entries.is_empty() { - use crate::pinned_bpf_maps::ProcModuleOffsetsValue; - let mut by_pid: HashMap> = - HashMap::new(); - for (pid, cookie, off, base, size) in entries { - if is_current_process_pid(pid) { - continue; - } - by_pid.entry(pid).or_default().push(( - cookie, - ProcModuleOffsetsValue::new( - off.text, off.rodata, off.data, off.bss, base, size, - ), - )); - } - let mut total = 0usize; - for (pid, items) in by_pid { - initial_target_pids.insert(pid); - // Add event PID (kernel namespace) to allowlist so subsequent - // fork/exit events are filtered in-kernel. - let event_pid = resolve_event_pid_for_proc(pid); - let runtime_pids = runtime_pid_keys_for_proc_event(pid, event_pid, []); - for runtime_pid in &runtime_pids { - write_pinned_runtime_pid_alias(*runtime_pid, pid); - guard.record_runtime_pid_alias(*runtime_pid, pid); - } - if let Ok(n) = publish_offsets_for_runtime_pid_keys( - pid, - event_pid, - &runtime_pids, - &items, - "initial prefill", - ) { - total += n; - } - insert_allowed_runtime_pid_keys(&runtime_pids); - } - tracing::info!( - "Sysmon: initial inserted {} offset entries for module {}", - total, - tpath.display() - ); - } - } - } - for pid in initial_target_pids { - let event_pid = resolve_event_pid_for_proc(pid); - if let Err(e) = - prefill_full_offsets_for_pid_if_new(&mgr, event_pid, &proc_pid_for_event) - { - tracing::debug!( - "Sysmon: initial full offset prefill failed for proc pid {} (event pid {}): {}", - pid, - event_pid, - e - ); - } - } - } - tracing::info!("Sysmon: setup complete"); - // Initial prefill already ran above. Do not make the first periodic module - // refresh immediately due: for `-t executable`, the exec event is the fast - // path that inserts proc_module_offsets and allowed_pids. A fallback /proc - // scan here can delay a short-lived target past its only probe. - let mut last_module_refresh = Instant::now(); - let mut target_pid_map_signatures = HashMap::::new(); // Event loop: prefer ringbuf; fallback to perf if let Some(map) = bpf.take_map("sysmon_events") { let mut rb: RingBuf = map.try_into()?; + let mut worker = SysmonWorker::spawn(mgr, cfg, pending, pending_map_refreshes, tx)?; loop { + worker.ensure_running()?; let mut had_event = false; - // Drain queued lifecycle events before periodic refresh. In the - // short-lived `-t executable` path, sched_process_exec must be - // handled promptly so offsets are ready before the first uprobe. - while let Some(item) = rb.next() { + let drain_started_at = Instant::now(); + let mut drained = 0; + // Keep the collector bounded so the work thread gets CPU even when + // the ring never becomes empty. All `/proc` and map work happens + // after this handoff. + while drained < SYSMON_RING_DRAIN_EVENT_LIMIT { + let Some(item) = rb.next() else { + break; + }; had_event = true; + drained += 1; if item.len() == core::mem::size_of::() { // SAFETY: The ring buffer sample length was checked to match SysEvent; // read_unaligned handles any alignment from the byte slice. let ev = unsafe { core::ptr::read_unaligned(item.as_ptr() as *const SysEvent) }; - let matched = dispatch_sysmon_event( - &mgr, - &target, - &pending, - &pending_map_refreshes, - &proc_pid_for_event, - &ev, - ); - if matched { - try_publish_sys_event(&tx, ev); - } + worker.enqueue(ev)?; + } + if drain_started_at.elapsed() >= SYSMON_RING_DRAIN_TIME_BUDGET { + break; } } - poll_pending_offsets(&mgr, &pending, &proc_pid_for_event); - poll_pending_map_refreshes( - &mgr, - target.as_deref(), - &pending_map_refreshes, - &pending, - &tx, - ); - refresh_target_module_offsets( - &mgr, - target.as_deref(), - &mut last_module_refresh, - &mut target_pid_map_signatures, - &tx, - ); if !had_event { std::thread::sleep(std::time::Duration::from_millis(5)); + } else if drained >= SYSMON_RING_DRAIN_EVENT_LIMIT + || drain_started_at.elapsed() >= SYSMON_RING_DRAIN_TIME_BUDGET + { + std::thread::yield_now(); } } } else if let Some(map) = bpf.take_map("sysmon_events_perf") { @@ -178,8 +460,11 @@ pub(super) fn run_sysmon_loop( if bufs.is_empty() { return Err(anyhow::anyhow!("No perf buffers opened")); } + let mut worker = SysmonWorker::spawn(mgr, cfg, pending, pending_map_refreshes, tx)?; loop { std::thread::sleep(std::time::Duration::from_millis(10)); + worker.ensure_running()?; + let mut enqueue_error = None; for buf in bufs.iter_mut() { if !buf.readable() { continue; @@ -203,16 +488,8 @@ pub(super) fn run_sysmon_loop( let ev = unsafe { core::ptr::read_unaligned(raw.as_ptr() as *const SysEvent) }; - let matched = dispatch_sysmon_event( - &mgr, - &target, - &pending, - &pending_map_refreshes, - &proc_pid_for_event, - &ev, - ); - if matched { - try_publish_sys_event(&tx, ev); + if enqueue_error.is_none() { + enqueue_error = worker.enqueue(ev).err(); } } } @@ -221,21 +498,9 @@ pub(super) fn run_sysmon_loop( } }); } - poll_pending_offsets(&mgr, &pending, &proc_pid_for_event); - poll_pending_map_refreshes( - &mgr, - target.as_deref(), - &pending_map_refreshes, - &pending, - &tx, - ); - refresh_target_module_offsets( - &mgr, - target.as_deref(), - &mut last_module_refresh, - &mut target_pid_map_signatures, - &tx, - ); + if let Some(error) = enqueue_error { + return Err(error); + } } } else { Err(anyhow::anyhow!("No sysmon events map found (ringbuf/perf)")) diff --git a/ghostscope/src/cli/script_runtime.rs b/ghostscope/src/cli/script_runtime.rs index 9d548a0f..8c8fc63e 100644 --- a/ghostscope/src/cli/script_runtime.rs +++ b/ghostscope/src/cli/script_runtime.rs @@ -391,7 +391,6 @@ async fn run_cli_with_session( let stdout = io::stdout(); let mut stdout = io::BufWriter::new(stdout.lock()); let mut backtrace_renderer = crate::trace::backtrace::BacktraceRenderer::default(); - let fallback_coordinator = ghostscope_process::ProcessManager::new(); let mut output_rate_limiter = ScriptOutputRateLimiter::new(config.script_output_events_per_sec); let mut ebpf_loss_report_ticker = tokio::time::interval(Duration::from_secs(1)); ebpf_loss_report_ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); @@ -405,6 +404,7 @@ async fn run_cli_with_session( Ok(events) => { let runtime_refresh_request = BacktraceRuntimeModuleRequest::from_events(&events); + let process_snapshot = session.process_manager_snapshot(); let mut wrote_output = false; let mut suppressed_output = false; @@ -416,25 +416,12 @@ async fn run_cli_with_session( ) { ScriptOutputRateDecision::Silent => {} ScriptOutputRateDecision::Render => { - let display_event = match session.coordinator.try_lock() { - Ok(coordinator) => backtrace_renderer.render_event_for_tui( - &event, - session.process_analyzer.as_ref(), - &coordinator, - session.proc_pid(), - ), - Err(std::sync::TryLockError::WouldBlock) => { - backtrace_renderer.render_event_for_tui( - &event, - session.process_analyzer.as_ref(), - &fallback_coordinator, - session.proc_pid(), - ) - } - Err(std::sync::TryLockError::Poisoned(_)) => { - panic!("coordinator mutex poisoned") - } - }; + let display_event = backtrace_renderer.render_event_for_tui( + &event, + session.process_analyzer.as_ref(), + &process_snapshot, + session.proc_pid(), + ); match output_renderer.write_display_event(&display_event, &mut stdout) { Ok(wrote) => wrote_output |= wrote, Err(e) => warn!("Failed to write event output: {e}"), diff --git a/ghostscope/src/core/session.rs b/ghostscope/src/core/session.rs index 1e045be6..ca20f7e9 100644 --- a/ghostscope/src/core/session.rs +++ b/ghostscope/src/core/session.rs @@ -6,10 +6,12 @@ use crate::trace::backtrace_runtime::{ }; use crate::trace::TraceManager; use anyhow::Result; +use futures::FutureExt; use ghostscope_debuginfod::{DebuginfodClient, DebuginfodConfig}; use ghostscope_dwarf::{DwarfAnalyzer, ExplicitDebugFile, ModuleStats, RuntimeBacktraceLoadBudget}; use ghostscope_process::{ - PidFilterSpec, PidNamespaceId, ProcessManager, ProcessSysmon, SysmonConfig, SysmonEventMask, + PidFilterSpec, PidNamespaceId, ProcessManager, ProcessManagerSnapshot, + ProcessManagerSnapshotReader, ProcessSysmon, SysmonConfig, SysmonEventMask, }; use ghostscope_protocol::{ParsedInstruction, ParsedTraceEvent}; use std::collections::{BTreeMap, BTreeSet}; @@ -21,6 +23,64 @@ use tracing::{info, warn}; const BACKTRACE_RUNTIME_MODULE_DISCOVERY_GRACE: Duration = Duration::from_millis(500); const BACKTRACE_RUNTIME_OBSERVATION_REQUEST_MAX: usize = 1_024; const BACKTRACE_RUNTIME_OBSERVATION_RETRY_DELAY: Duration = Duration::from_secs(1); +const BACKTRACE_RUNTIME_OBSERVATION_MAX_ATTEMPTS: u8 = 3; + +#[derive(Debug)] +struct BacktraceObservationState { + mapping: Option<(u32, ghostscope_process::PidOffsetsEntry)>, + attempts: u8, + retry_at: Option, +} + +impl BacktraceObservationState { + fn should_retry( + &self, + mapping: &Option<(u32, ghostscope_process::PidOffsetsEntry)>, + now: Instant, + ) -> bool { + self.mapping != *mapping || self.retry_at.is_some_and(|retry_at| now >= retry_at) + } + + fn finish(&mut self, terminal: bool, now: Instant) { + self.retry_at = if terminal || self.attempts >= BACKTRACE_RUNTIME_OBSERVATION_MAX_ATTEMPTS { + None + } else { + Some(now + BACKTRACE_RUNTIME_OBSERVATION_RETRY_DELAY * (1 << (self.attempts - 1))) + }; + } +} + +fn observation_mapping( + coordinator: &ProcessManager, + proc_pid: Option, + observation: BacktraceRuntimeModuleObservation, +) -> Option<(u32, ghostscope_process::PidOffsetsEntry)> { + for pid in coordinator.candidate_proc_pids_for_runtime_pid(observation.runtime_pid, proc_pid) { + if let Some(entry) = + coordinator + .cached_offsets_with_paths_for_pid(pid) + .and_then(|entries| { + entries.iter().find(|entry| { + if observation.raw_ip == 0 { + entry.cookie == observation.cookie_hint + } else { + observation.raw_ip >= entry.base + && observation.raw_ip < entry.base.saturating_add(entry.size) + } + }) + }) + { + return Some((pid, entry.clone())); + } + } + None +} + +fn new_process_manager_state() -> (Arc>, ProcessManagerSnapshotReader) { + let coordinator = ProcessManager::new(); + let snapshot_reader = coordinator.snapshot_reader(); + (Arc::new(Mutex::new(coordinator)), snapshot_reader) +} #[derive(Debug, Clone, Default)] pub struct BacktraceRuntimeModuleRequest { @@ -45,24 +105,34 @@ impl BacktraceRuntimeModuleRequest { if !requests_runtime_module { continue; } - let Some(frame) = frames.last() else { + let Some(stopping_frame_index) = frames.len().checked_sub(1) else { continue; }; - let observation = BacktraceRuntimeModuleObservation { - runtime_pid: event.pid, - raw_ip: frame.raw_ip, - cookie_hint: frame.module_cookie, - // When offsets are unavailable, the eBPF unwinder can only - // retain the previous frame's cookie. In all other runtime - // module failure modes the stopping cookie identifies the - // address-space mapping that produced the raw IP. - cookie_is_authoritative: !matches!( - status, - ghostscope_protocol::trace_event::BacktraceStatus::OffsetsUnavailable - ), - }; - if !observation.is_empty() { - request.observations.insert(observation); + let offsets_unavailable = matches!( + status, + ghostscope_protocol::trace_event::BacktraceStatus::OffsetsUnavailable + ); + for (frame_index, frame) in frames.iter().enumerate() { + let is_stopping_frame = frame_index == stopping_frame_index; + // Missing offsets can leave earlier, otherwise unwindable + // frames without a published runtime mapping. Preserve + // their exact PID/raw-IP/cookie tuples so each module can + // be recovered without mixing identities across frames. + if !offsets_unavailable && !is_stopping_frame { + continue; + } + let observation = BacktraceRuntimeModuleObservation { + runtime_pid: event.pid, + raw_ip: frame.raw_ip, + cookie_hint: frame.module_cookie, + // For an offsets failure only the stopping frame may + // retain the previous frame's cookie. Earlier frames, + // and all other failure modes, carry their own cookie. + cookie_is_authoritative: !offsets_unavailable || !is_stopping_frame, + }; + if !observation.is_empty() { + request.observations.insert(observation); + } } } } @@ -106,24 +176,12 @@ pub enum BacktraceRuntimeRefreshOutcome { #[derive(Debug)] enum BacktraceRuntimeRefreshTaskOutcome { - Loaded { + Prepared(BacktraceRuntimePreparedOutcome), + Published { resolved: ResolvedBacktraceRuntimeModule, runtime_symbols: Vec, append_result: crate::trace::manager::BacktraceUnwindRowsAppendResult, }, - AlreadyLoaded(ResolvedBacktraceRuntimeModule), - AlreadyAttempted, - ModuleNotFound, - Failed { - attempted_cookie: Option, - error: String, - }, - TimedOut { - attempted_cookie: Option, - }, - LimitReached { - limit: u32, - }, } #[derive(Debug)] @@ -134,7 +192,7 @@ enum BacktraceRuntimePreparedOutcome { unwind_modules: Vec<(u64, Vec)>, }, AlreadyLoaded(ResolvedBacktraceRuntimeModule), - AlreadyAttempted, + AlreadyAttempted(ResolvedBacktraceRuntimeModule), ModuleNotFound, Failed { attempted_cookie: Option, @@ -178,12 +236,14 @@ struct BacktraceRuntimeRefreshTask { observation: BacktraceRuntimeModuleObservation, timeout: Duration, budget: RuntimeBacktraceLoadBudget, + baseline: Arc, handle: tokio::task::JoinHandle, + result: Option, } #[allow(clippy::too_many_arguments)] fn prepare_backtrace_runtime_module( - coordinator: Arc>, + mut coordinator: ProcessManager, proc_pid: Option, target_binary: Option<&str>, observation: BacktraceRuntimeModuleObservation, @@ -194,7 +254,7 @@ fn prepare_backtrace_runtime_module( budget: &RuntimeBacktraceLoadBudget, ) -> BacktraceRuntimePreparedOutcome { let discovery_deadline = Instant::now() + BACKTRACE_RUNTIME_MODULE_DISCOVERY_GRACE; - let resolved = loop { + let mut resolved = loop { if budget.check().is_err() { return BacktraceRuntimePreparedOutcome::TimedOut { attempted_cookie: None, @@ -202,7 +262,6 @@ fn prepare_backtrace_runtime_module( } let resolution = { - let mut coordinator = coordinator.lock().expect("coordinator mutex poisoned"); if let Some(proc_pid) = proc_pid { BacktraceRuntimeRunner::resolve_pid_module_for_observation( &mut coordinator, @@ -260,7 +319,7 @@ fn prepare_backtrace_runtime_module( return BacktraceRuntimePreparedOutcome::AlreadyLoaded(resolved); } BacktraceRuntimeModuleLoadDecision::AlreadyAttempted => { - return BacktraceRuntimePreparedOutcome::AlreadyAttempted; + return BacktraceRuntimePreparedOutcome::AlreadyAttempted(resolved); } BacktraceRuntimeModuleLoadDecision::LimitReached => { return BacktraceRuntimePreparedOutcome::LimitReached { @@ -277,6 +336,10 @@ fn prepare_backtrace_runtime_module( ); match ghostscope_dwarf::load_runtime_backtrace_metadata( resolved.module.module_path.clone(), + *resolved + .probe + .take() + .expect("resolution retained a validated ELF mapping"), max_unwind_rows, budget, ) { @@ -310,6 +373,38 @@ fn prepare_backtrace_runtime_module( } } +/// A timed-out filesystem call must not keep Tokio's blocking pool alive at shutdown. +/// The detached worker owns only private discovery state and cannot publish anything. +/// A timeout disables further discovery for this session, bounding abandoned work to one. +fn spawn_backtrace_discovery( + timeout: Duration, + budget: RuntimeBacktraceLoadBudget, + work: impl FnOnce() -> BacktraceRuntimePreparedOutcome + Send + 'static, +) -> Result> { + let (sender, receiver) = tokio::sync::oneshot::channel(); + std::thread::Builder::new() + .name("backtrace-discovery".into()) + .spawn(move || { + let _ = sender.send(work()); + })?; + Ok(tokio::spawn(async move { + let outcome = match tokio::time::timeout(timeout, receiver).await { + Ok(Ok(prepared)) if !budget.expired_or_cancelled() => prepared, + Ok(Err(error)) => BacktraceRuntimePreparedOutcome::Failed { + attempted_cookie: None, + error: format!("backtrace runtime worker stopped: {error}"), + }, + _ => { + budget.cancel(); + BacktraceRuntimePreparedOutcome::TimedOut { + attempted_cookie: None, + } + } + }; + BacktraceRuntimeRefreshTaskOutcome::Prepared(outcome) + })) +} + fn sysmon_watch_from_config( config: &ResolvedConfig, fallback_host_pid: Option, @@ -376,12 +471,15 @@ pub struct GhostSession { pub debug_file: Option, // Optional debug file path pub config: Option, // Holds the resolved configuration pub coordinator: Arc>, // Manages PID/module offsets prefill and application + coordinator_snapshot: ProcessManagerSnapshotReader, pub sysmon: Option>>, // Realtime process monitor (exec/fork/exit) target_backtrace_runtime_modules_enabled: bool, backtrace_runtime_known_cookies: BTreeSet, backtrace_runtime_attempted_cookies: BTreeSet, backtrace_runtime_queued_observations: BTreeSet, - backtrace_runtime_recent_observations: BTreeMap, + backtrace_runtime_observations: + BTreeMap, + backtrace_runtime_unwind_modules: Vec<(u64, Vec)>, backtrace_runtime_refresh_task: Option, backtrace_runtime_refresh_timed_out: bool, backtrace_runtime_limit_reported: bool, @@ -391,6 +489,7 @@ impl GhostSession { /// Create a new ghost session with merged configuration pub fn new_with_config(config: &ResolvedConfig) -> Self { info!("Creating ghost session with merged configuration"); + let (coordinator, coordinator_snapshot) = new_process_manager_state(); let mut s = Self { process_analyzer: None, @@ -404,13 +503,15 @@ impl GhostSession { trace_manager: TraceManager::new(), source_path_resolver: SourcePathResolver::new(&config.source), config: Some(config.clone()), - coordinator: Arc::new(Mutex::new(ProcessManager::new())), + coordinator, + coordinator_snapshot, sysmon: None, target_backtrace_runtime_modules_enabled: false, backtrace_runtime_known_cookies: BTreeSet::new(), backtrace_runtime_attempted_cookies: BTreeSet::new(), backtrace_runtime_queued_observations: BTreeSet::new(), - backtrace_runtime_recent_observations: BTreeMap::new(), + backtrace_runtime_observations: BTreeMap::new(), + backtrace_runtime_unwind_modules: Vec::new(), backtrace_runtime_refresh_task: None, backtrace_runtime_refresh_timed_out: false, backtrace_runtime_limit_reported: false, @@ -523,6 +624,7 @@ impl GhostSession { #[allow(dead_code)] pub fn new(args: &ParsedArgs) -> Self { info!("Creating ghost session"); + let (coordinator, coordinator_snapshot) = new_process_manager_state(); let mut s = Self { process_analyzer: None, @@ -536,13 +638,15 @@ impl GhostSession { trace_manager: TraceManager::new(), source_path_resolver: SourcePathResolver::new(&Default::default()), config: None, - coordinator: Arc::new(Mutex::new(ProcessManager::new())), + coordinator, + coordinator_snapshot, sysmon: None, target_backtrace_runtime_modules_enabled: false, backtrace_runtime_known_cookies: BTreeSet::new(), backtrace_runtime_attempted_cookies: BTreeSet::new(), backtrace_runtime_queued_observations: BTreeSet::new(), - backtrace_runtime_recent_observations: BTreeMap::new(), + backtrace_runtime_observations: BTreeMap::new(), + backtrace_runtime_unwind_modules: Vec::new(), backtrace_runtime_refresh_task: None, backtrace_runtime_refresh_timed_out: false, backtrace_runtime_limit_reported: false, @@ -679,6 +783,50 @@ impl GhostSession { Ok(DwarfAnalyzer::runtime_modules_from_pid_offsets(entries)) } + /// Refresh the PID module snapshot and load newly mapped modules before + /// compiling another script in a long-lived session. + pub(crate) async fn refresh_pid_analyzer_before_compile(&mut self) -> Result { + let Some(proc_pid) = self.proc_pid() else { + return Ok(0); + }; + + let runtime_modules = { + let mut coordinator = self.coordinator.lock().expect("coordinator mutex poisoned"); + coordinator.refresh_prefill_pid(proc_pid)?; + let Some(entries) = coordinator.cached_offsets_with_paths_for_pid(proc_pid) else { + return Ok(0); + }; + DwarfAnalyzer::runtime_modules_from_pid_offsets(entries) + }; + if runtime_modules.is_empty() { + return Ok(0); + } + + let debug_search_paths = self.get_debug_search_paths(); + let allow_loose = self.get_allow_loose_debug_match(); + let debuginfod_client = self.build_debuginfod_client()?; + let Some(analyzer) = self.process_analyzer.as_mut() else { + return Ok(0); + }; + + let loaded = analyzer + .refresh_pid_runtime_modules_with_config_and_debuginfod( + runtime_modules, + &debug_search_paths, + allow_loose, + debuginfod_client, + |_| {}, + ) + .await?; + if loaded > 0 { + info!( + "Refreshed PID {} analyzer with {} newly mapped module(s) before compilation", + proc_pid, loaded + ); + } + Ok(loaded) + } + fn backtrace_runtime_modules_configured(&self) -> bool { self.config .as_ref() @@ -715,7 +863,9 @@ impl GhostSession { } fn backtrace_runtime_modules_allowed(&self) -> bool { - if !self.backtrace_runtime_modules_configured() || self.backtrace_runtime_refresh_timed_out + if !self.backtrace_runtime_modules_configured() + || self.backtrace_runtime_refresh_timed_out + || self.backtrace_runtime_limit_reported { return false; } @@ -733,6 +883,14 @@ impl GhostSession { self.backtrace_runtime_known_cookies.insert(cookie); } + pub(crate) fn seed_backtrace_runtime_rows( + &self, + loader: &mut ghostscope_loader::GhostScopeLoader, + ) -> Result<()> { + loader.append_backtrace_unwind_rows_for_modules(&self.backtrace_runtime_unwind_modules)?; + Ok(()) + } + pub fn schedule_backtrace_runtime_module_refresh( &mut self, request: BacktraceRuntimeModuleRequest, @@ -745,10 +903,10 @@ impl GhostSession { } let now = Instant::now(); - self.backtrace_runtime_recent_observations - .retain(|_, seen_at| { - now.saturating_duration_since(*seen_at) < BACKTRACE_RUNTIME_OBSERVATION_RETRY_DELAY - }); + let coordinator = Arc::clone(&self.coordinator); + let Ok(coordinator) = coordinator.try_lock() else { + return Ok(BacktraceRuntimeRefreshSchedule::NotNeeded); + }; let active_observation = self .backtrace_runtime_refresh_task .as_ref() @@ -760,25 +918,41 @@ impl GhostSession { || self .backtrace_runtime_queued_observations .contains(&observation) - || self - .backtrace_runtime_recent_observations - .contains_key(&observation) { continue; } - if self - .backtrace_runtime_queued_observations - .len() - .saturating_add(self.backtrace_runtime_recent_observations.len()) + let mapping = observation_mapping(&coordinator, self.proc_pid(), observation); + if let Some(state) = self.backtrace_runtime_observations.get(&observation) { + if !state.should_retry(&mapping, now) { + continue; + } + } + if self.backtrace_runtime_observations.len() >= BACKTRACE_RUNTIME_OBSERVATION_REQUEST_MAX + && !self + .backtrace_runtime_observations + .contains_key(&observation) { dropped += 1; continue; } + let state = self + .backtrace_runtime_observations + .entry(observation) + .or_insert(BacktraceObservationState { + mapping: mapping.clone(), + attempts: 0, + retry_at: Some(now), + }); + if state.mapping != mapping { + state.mapping = mapping; + state.attempts = 0; + } self.backtrace_runtime_queued_observations .insert(observation); queued += 1; } + drop(coordinator); if dropped > 0 { warn!( dropped, @@ -808,7 +982,9 @@ impl GhostSession { } fn start_next_backtrace_runtime_module_refresh(&mut self) -> Result> { - if self.backtrace_runtime_refresh_task.is_some() || self.backtrace_runtime_refresh_timed_out + if self.backtrace_runtime_refresh_task.is_some() + || self.backtrace_runtime_refresh_timed_out + || self.backtrace_runtime_limit_reported { return Ok(None); } @@ -816,6 +992,14 @@ impl GhostSession { return Ok(None); } + let baseline = match self.coordinator.try_lock() { + Ok(coordinator) => Arc::new(coordinator.fork_for_runtime_discovery()), + Err(std::sync::TryLockError::WouldBlock) => return Ok(None), + Err(std::sync::TryLockError::Poisoned(_)) => { + anyhow::bail!("coordinator mutex poisoned") + } + }; + let observation = self .backtrace_runtime_queued_observations .iter() @@ -824,98 +1008,42 @@ impl GhostSession { .expect("checked non-empty runtime module queue"); self.backtrace_runtime_queued_observations .remove(&observation); - self.backtrace_runtime_recent_observations - .insert(observation, Instant::now()); + if let Some(state) = self.backtrace_runtime_observations.get_mut(&observation) { + state.attempts += 1; + state.retry_at = None; + } - let coordinator = Arc::clone(&self.coordinator); + let worker_baseline = Arc::clone(&baseline); let proc_pid = self.proc_pid(); let target_binary = self.target_binary.clone(); let timeout = self.backtrace_runtime_module_timeout(); let max_unwind_rows = self.backtrace_runtime_unwind_rows_max(); - let unwind_rows_appender = self.trace_manager.backtrace_unwind_rows_appender(); let known_cookies = self.backtrace_runtime_known_cookies.clone(); let attempted_cookies = self.backtrace_runtime_attempted_cookies.clone(); let module_limit = self.backtrace_runtime_modules_max(); let budget = RuntimeBacktraceLoadBudget::new(timeout); let worker_budget = budget.clone(); - let timeout_budget = budget.clone(); - - let handle = tokio::spawn(async move { - let mut worker = tokio::task::spawn_blocking(move || { - prepare_backtrace_runtime_module( - coordinator, - proc_pid, - target_binary.as_deref(), - observation, - &known_cookies, - &attempted_cookies, - module_limit, - max_unwind_rows, - &worker_budget, - ) - }); - let prepared = tokio::select! { - result = &mut worker => match result { - Ok(prepared) => prepared, - Err(error) => BacktraceRuntimePreparedOutcome::Failed { - attempted_cookie: None, - error: format!("backtrace runtime blocking worker failed: {error}"), - }, - }, - () = tokio::time::sleep(timeout) => { - timeout_budget.cancel(); - // Drain the cooperative worker before reporting the timeout so - // no parsing continues invisibly after the warning is shown. - let _ = worker.await; - BacktraceRuntimePreparedOutcome::TimedOut { - attempted_cookie: None, - } - } - }; - - match prepared { - BacktraceRuntimePreparedOutcome::Loaded { - resolved, - runtime_symbols, - unwind_modules, - } => { - let append_result = unwind_rows_appender.append(unwind_modules).await; - BacktraceRuntimeRefreshTaskOutcome::Loaded { - resolved, - runtime_symbols, - append_result, - } - } - BacktraceRuntimePreparedOutcome::AlreadyLoaded(resolved) => { - BacktraceRuntimeRefreshTaskOutcome::AlreadyLoaded(resolved) - } - BacktraceRuntimePreparedOutcome::AlreadyAttempted => { - BacktraceRuntimeRefreshTaskOutcome::AlreadyAttempted - } - BacktraceRuntimePreparedOutcome::ModuleNotFound => { - BacktraceRuntimeRefreshTaskOutcome::ModuleNotFound - } - BacktraceRuntimePreparedOutcome::Failed { - attempted_cookie, - error, - } => BacktraceRuntimeRefreshTaskOutcome::Failed { - attempted_cookie, - error, - }, - BacktraceRuntimePreparedOutcome::TimedOut { attempted_cookie } => { - BacktraceRuntimeRefreshTaskOutcome::TimedOut { attempted_cookie } - } - BacktraceRuntimePreparedOutcome::LimitReached { limit } => { - BacktraceRuntimeRefreshTaskOutcome::LimitReached { limit } - } - } - }); + let handle = spawn_backtrace_discovery(timeout, budget.clone(), move || { + prepare_backtrace_runtime_module( + worker_baseline.fork_for_runtime_discovery(), + proc_pid, + target_binary.as_deref(), + observation, + &known_cookies, + &attempted_cookies, + module_limit, + max_unwind_rows, + &worker_budget, + ) + })?; self.backtrace_runtime_refresh_task = Some(BacktraceRuntimeRefreshTask { observation, timeout, budget, + baseline, handle, + result: None, }); info!( runtime_pid = observation.runtime_pid, @@ -930,40 +1058,160 @@ impl GhostSession { pub async fn poll_backtrace_runtime_module_refresh( &mut self, ) -> Result> { - let Some(task) = self.backtrace_runtime_refresh_task.as_ref() else { + let Some(task) = self.backtrace_runtime_refresh_task.as_mut() else { return Ok(None); }; - if !task.handle.is_finished() { - return Ok(None); + if task.result.is_none() { + if !task.handle.is_finished() { + return Ok(None); + } + task.result = Some( + match (&mut task.handle).now_or_never().expect("finished worker") { + Ok(result) => result, + Err(error) => BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::Failed { + attempted_cookie: None, + error: format!("backtrace runtime task failed: {error}"), + }, + ), + }, + ); } - + let coordinator = Arc::clone(&self.coordinator); + let needs_mapping = matches!( + task.result, + Some(BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::Loaded { .. } + | BacktraceRuntimePreparedOutcome::AlreadyLoaded(_) + | BacktraceRuntimePreparedOutcome::AlreadyAttempted(_) + )) + ); + let mut coordinator = if needs_mapping { + match coordinator.try_lock() { + Ok(guard) => Some(guard), + Err(std::sync::TryLockError::WouldBlock) => return Ok(None), + Err(std::sync::TryLockError::Poisoned(_)) => { + anyhow::bail!("coordinator mutex poisoned") + } + } + } else { + None + }; let task = self .backtrace_runtime_refresh_task .take() .expect("runtime refresh task was present"); - let _observation = task.observation; + let observation = task.observation; let timeout = task.timeout; - let task_result = match task.handle.await { - Ok(result) => result, - Err(error) => { - let next_started = self - .start_next_backtrace_runtime_module_refresh()? - .is_some(); - return Ok(Some(BacktraceRuntimeRefreshOutcome::Failed { - error: format!("backtrace runtime module task failed: {error}"), - next_started, - })); + let task_result = task.result.expect("completed worker result"); + let mut mapping_published = true; + if let BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::Loaded { resolved, .. } + | BacktraceRuntimePreparedOutcome::AlreadyLoaded(resolved) + | BacktraceRuntimePreparedOutcome::AlreadyAttempted(resolved), + ) = &task_result + { + let coordinator = coordinator.as_mut().expect("mapping publication guard"); + if coordinator.apply_runtime_pid_snapshot( + resolved.proc_pid, + task.baseline + .cached_offsets_with_paths_for_pid(resolved.proc_pid), + &resolved.entries, + ) { + BacktraceRuntimeRunner::publish_observation_pid_snapshot( + coordinator, + resolved.proc_pid, + observation, + &resolved.entries, + ); + if let Some(state) = self.backtrace_runtime_observations.get_mut(&observation) { + state.mapping = + observation_mapping(coordinator, Some(resolved.proc_pid), observation); + } + } else { + // A newer PID snapshot wins, but CFI from the validated mapping + // remains valid under its cookie. Retain it and charge the load + // budget; a bounded retry can refresh offsets without parsing again. + mapping_published = false; } - }; + } + drop(coordinator); + + if let Some(state) = self + .backtrace_runtime_observations + .get_mut(&observation) + .filter(|_| { + !matches!( + task_result, + BacktraceRuntimeRefreshTaskOutcome::Published { .. } + ) + }) + { + state.finish( + mapping_published + && !matches!( + task_result, + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::ModuleNotFound + | BacktraceRuntimePreparedOutcome::Failed { .. } + ) + ), + Instant::now(), + ); + } let outcome = match task_result { - BacktraceRuntimeRefreshTaskOutcome::Loaded { + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::Loaded { + resolved, + runtime_symbols, + mut unwind_modules, + }, + ) => { + self.backtrace_runtime_attempted_cookies + .insert(resolved.cookie); + // Retain a bounded set of completed rows to seed traces attached later, + // including replacements attached while this publication is in flight. + let retained: usize = self + .backtrace_runtime_unwind_modules + .iter() + .map(|(_, rows)| rows.len()) + .sum(); + let mut remaining = self + .backtrace_runtime_unwind_rows_max() + .saturating_sub(retained); + for (_, rows) in &mut unwind_modules { + rows.truncate(remaining); + remaining -= rows.len(); + } + self.backtrace_runtime_unwind_modules + .extend(unwind_modules.clone()); + let appender = self.trace_manager.backtrace_unwind_rows_appender(); + let handle = tokio::spawn(async move { + let append_result = appender.append(unwind_modules).await; + BacktraceRuntimeRefreshTaskOutcome::Published { + resolved, + runtime_symbols, + append_result, + } + }); + self.backtrace_runtime_refresh_task = Some(BacktraceRuntimeRefreshTask { + observation, + timeout, + budget: task.budget, + baseline: task.baseline, + handle, + result: None, + }); + return Ok(None); + } + BacktraceRuntimeRefreshTaskOutcome::Published { resolved, runtime_symbols, append_result, } => { - self.backtrace_runtime_attempted_cookies - .insert(resolved.cookie); + // Make symbols visible with the Loaded notification that clears + // renderer caches, never between CFI publication phases. if let Some(analyzer) = self.process_analyzer.as_mut() { analyzer.add_runtime_text_symbols(resolved.cookie, runtime_symbols); } @@ -980,7 +1228,9 @@ impl GhostSession { next_started, } } - BacktraceRuntimeRefreshTaskOutcome::AlreadyLoaded(resolved) => { + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::AlreadyLoaded(resolved), + ) => { self.record_backtrace_runtime_module(resolved.cookie); let next_started = self .start_next_backtrace_runtime_module_refresh()? @@ -991,7 +1241,9 @@ impl GhostSession { next_started, } } - BacktraceRuntimeRefreshTaskOutcome::AlreadyAttempted => { + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::AlreadyAttempted(_), + ) => { let next_started = self .start_next_backtrace_runtime_module_refresh()? .is_some(); @@ -1001,23 +1253,29 @@ impl GhostSession { next_started, } } - BacktraceRuntimeRefreshTaskOutcome::ModuleNotFound => { + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::ModuleNotFound, + ) => { let next_started = self .start_next_backtrace_runtime_module_refresh()? .is_some(); BacktraceRuntimeRefreshOutcome::ModuleNotFound { next_started } } - BacktraceRuntimeRefreshTaskOutcome::TimedOut { attempted_cookie } => { + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::TimedOut { attempted_cookie }, + ) => { self.backtrace_runtime_attempted_cookies .extend(attempted_cookie); self.backtrace_runtime_refresh_timed_out = true; self.backtrace_runtime_queued_observations.clear(); BacktraceRuntimeRefreshOutcome::TimedOut { timeout } } - BacktraceRuntimeRefreshTaskOutcome::Failed { - attempted_cookie, - error, - } => { + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::Failed { + attempted_cookie, + error, + }, + ) => { self.backtrace_runtime_attempted_cookies .extend(attempted_cookie); let next_started = self @@ -1028,15 +1286,15 @@ impl GhostSession { next_started, } } - BacktraceRuntimeRefreshTaskOutcome::LimitReached { limit } => { - let next_started = self - .start_next_backtrace_runtime_module_refresh()? - .is_some(); + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::LimitReached { limit }, + ) => { + self.backtrace_runtime_queued_observations.clear(); if self.backtrace_runtime_limit_reported { BacktraceRuntimeRefreshOutcome::Loaded { modules: 0, unwind_rows: 0, - next_started, + next_started: false, } } else { self.backtrace_runtime_limit_reported = true; @@ -1205,6 +1463,11 @@ impl GhostSession { .map(|pid_context| pid_context.proc_pid) } + /// Last complete process-module view published by the mutable coordinator. + pub(crate) fn process_manager_snapshot(&self) -> Arc { + self.coordinator_snapshot.load() + } + /// Host-view PID kept for logs, UI display, and host-TGID fallback paths. pub fn host_pid(&self) -> Option { self.pid_context @@ -1265,6 +1528,85 @@ mod tests { use crate::config::settings::{PathSubstitution, SourceConfig}; use crate::config::UserConfig; + #[test] + fn terminal_observations_only_retry_after_the_mapping_changes() { + let now = Instant::now(); + let mut state = BacktraceObservationState { + mapping: None, + attempts: 1, + retry_at: Some(now), + }; + state.finish(true, now); + assert!(!state.should_retry(&None, now + Duration::from_secs(3600))); + let changed = Some(( + 42, + ghostscope_process::PidOffsetsEntry { + module_path: "late.so".into(), + cookie: 1, + offsets: Default::default(), + base: 0x1000, + size: 0x100, + }, + )); + assert!(state.should_retry(&changed, now)); + } + + #[test] + fn unresolved_observations_have_bounded_exponential_retries() { + let now = Instant::now(); + let mut state = BacktraceObservationState { + mapping: None, + attempts: 1, + retry_at: None, + }; + state.finish(false, now); + assert!(!state.should_retry(&None, now)); + assert!(state.should_retry(&None, now + Duration::from_secs(1))); + state.attempts = 2; + state.finish(false, now); + assert!(!state.should_retry(&None, now + Duration::from_secs(1))); + assert!(state.should_retry(&None, now + Duration::from_secs(2))); + state.attempts = 3; + state.finish(false, now); + assert!(!state.should_retry(&None, now + Duration::from_secs(3600))); + } + + #[test] + fn runtime_timeout_and_shutdown_do_not_join_a_blocked_discovery_worker() { + let (release, blocked) = std::sync::mpsc::channel(); + let (finished, completion) = std::sync::mpsc::channel(); + let supervisor = std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let result = runtime.block_on(async { + let timeout = Duration::from_millis(20); + let budget = RuntimeBacktraceLoadBudget::new(timeout); + spawn_backtrace_discovery(timeout, budget, move || { + let _ = blocked.recv(); + BacktraceRuntimePreparedOutcome::ModuleNotFound + }) + .unwrap() + .await + .unwrap() + }); + drop(runtime); + finished + .send(matches!( + result, + BacktraceRuntimeRefreshTaskOutcome::Prepared( + BacktraceRuntimePreparedOutcome::TimedOut { .. } + ) + )) + .unwrap(); + }); + let completed_before_release = completion.recv_timeout(Duration::from_secs(2)); + let _ = release.send(()); + supervisor.join().unwrap(); + assert!(completed_before_release.unwrap()); + } + #[test] fn backtrace_runtime_request_preserves_stopping_frame_identity() { let event = ParsedTraceEvent { @@ -1307,6 +1649,56 @@ mod tests { ); } + #[test] + fn offsets_unavailable_requests_each_frame_with_only_stopping_cookie_relaxed() { + let event = ParsedTraceEvent { + generation: 0, + trace_id: 1, + timestamp: 2, + pid: 42, + tid: 43, + instructions: vec![ParsedInstruction::Backtrace { + requested_depth: 3, + flags: 0, + status: ghostscope_protocol::trace_event::BacktraceStatus::OffsetsUnavailable, + error_code: 0, + frames: vec![ + ghostscope_protocol::ParsedBacktraceFrame { + module_cookie: 0x11, + pc: 0x22, + raw_ip: 0x33, + flags: 0, + }, + ghostscope_protocol::ParsedBacktraceFrame { + module_cookie: 0x11, + pc: 0x44, + raw_ip: 0x55, + flags: 0, + }, + ], + }], + }; + + let request = BacktraceRuntimeModuleRequest::from_events(&[event]); + assert_eq!( + request.observations, + BTreeSet::from([ + BacktraceRuntimeModuleObservation { + runtime_pid: 42, + raw_ip: 0x33, + cookie_hint: 0x11, + cookie_is_authoritative: true, + }, + BacktraceRuntimeModuleObservation { + runtime_pid: 42, + raw_ip: 0x55, + cookie_hint: 0x11, + cookie_is_authoritative: false, + }, + ]) + ); + } + #[test] fn backtrace_runtime_request_keeps_same_raw_ip_separate_by_pid() { let event = |pid| ParsedTraceEvent { @@ -1364,8 +1756,186 @@ mod tests { ); } - #[test] - fn test_new_with_config_sets_source_resolver() { + #[tokio::test] + async fn runtime_symbols_wait_for_the_cache_invalidation_outcome() { + let mut session = test_session(); + session.process_analyzer = Some( + DwarfAnalyzer::from_pid_runtime_modules_with_config_and_debuginfod( + 0, + Vec::new(), + &[], + false, + None, + |_| {}, + ) + .await + .unwrap(), + ); + let observation = BacktraceRuntimeModuleObservation { + runtime_pid: 42, + raw_ip: 0x1010, + cookie_hint: 2, + cookie_is_authoritative: true, + }; + let prepared = BacktraceRuntimePreparedOutcome::Loaded { + resolved: ResolvedBacktraceRuntimeModule { + observation, + proc_pid: 42, + cookie: 2, + module: ghostscope_dwarf::LoadedModuleRuntimeInfo { + module_path: "runtime.so".into(), + loaded_address: Some(0x1000), + load_bias: Some(0x1000), + size: 0x100, + }, + entries: Vec::new(), + probe: None, + }, + runtime_symbols: vec![ghostscope_dwarf::RuntimeTextSymbol { + name: "late_symbol".into(), + address: 0x10, + size: 0x10, + }], + unwind_modules: Vec::new(), + }; + let timeout = Duration::from_secs(2); + session.backtrace_runtime_refresh_task = Some(BacktraceRuntimeRefreshTask { + observation, + timeout, + budget: RuntimeBacktraceLoadBudget::new(timeout), + baseline: Arc::new(ProcessManager::new()), + handle: tokio::spawn( + async move { BacktraceRuntimeRefreshTaskOutcome::Prepared(prepared) }, + ), + result: None, + }); + let published = tokio::time::timeout(timeout, async { + loop { + let outcome = session + .poll_backtrace_runtime_module_refresh() + .await + .unwrap(); + let symbol = session + .process_analyzer + .as_ref() + .unwrap() + .find_runtime_function_name_for_display(2, 0x10, false); + if let Some(outcome) = outcome { + assert_eq!(symbol.as_deref(), Some("late_symbol")); + break outcome; + } + // None does not invalidate renderer caches, even if parsing has + // finished and publication to the current actors has started. + assert!(symbol.is_none()); + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(matches!( + published, + BacktraceRuntimeRefreshOutcome::Loaded { modules: 1, .. } + )); + } + + #[tokio::test] + async fn completed_cfi_survives_a_newer_mapping_and_consumes_the_load_budget() { + let mut session = test_session(); + let entry = |cookie| ghostscope_process::PidOffsetsEntry { + module_path: "runtime.so".into(), + cookie, + offsets: Default::default(), + base: 0x1000, + size: 0x100, + }; + let observation = BacktraceRuntimeModuleObservation { + runtime_pid: 42, + raw_ip: 0x1010, + cookie_hint: 0, + cookie_is_authoritative: false, + }; + let baseline = Arc::new(ProcessManager::new()); + // Sysmon has published a newer mapping before the prepared result is consumed. + assert!(session + .coordinator + .lock() + .unwrap() + .apply_runtime_pid_snapshot(42, None, &[entry(3)])); + session.backtrace_runtime_observations.insert( + observation, + BacktraceObservationState { + mapping: None, + attempts: 1, + retry_at: None, + }, + ); + let resolved = ResolvedBacktraceRuntimeModule { + observation, + proc_pid: 42, + cookie: 2, + module: ghostscope_dwarf::LoadedModuleRuntimeInfo { + module_path: "runtime.so".into(), + loaded_address: Some(0x1000), + load_bias: Some(0x1000), + size: 0x100, + }, + entries: vec![entry(2)], + probe: None, + }; + let timeout = Duration::from_secs(2); + let handle = tokio::spawn(async move { + BacktraceRuntimeRefreshTaskOutcome::Prepared(BacktraceRuntimePreparedOutcome::Loaded { + resolved, + runtime_symbols: Vec::new(), + unwind_modules: vec![(2, vec![Default::default()])], + }) + }); + session.backtrace_runtime_refresh_task = Some(BacktraceRuntimeRefreshTask { + observation, + timeout, + budget: RuntimeBacktraceLoadBudget::new(timeout), + baseline, + handle, + result: None, + }); + tokio::time::timeout(timeout, async { + while session.backtrace_runtime_refresh_task.is_some() { + tokio::task::yield_now().await; + session + .poll_backtrace_runtime_module_refresh() + .await + .unwrap(); + } + }) + .await + .unwrap(); + assert!(session.backtrace_runtime_known_cookies.contains(&2)); + assert_eq!(session.backtrace_runtime_unwind_modules.len(), 1); + assert_eq!( + backtrace_runtime_module_load_decision( + 4, + &session.backtrace_runtime_known_cookies, + &session.backtrace_runtime_attempted_cookies, + 1, + ), + BacktraceRuntimeModuleLoadDecision::LimitReached + ); + assert!(session.backtrace_runtime_observations[&observation] + .retry_at + .is_some()); + assert_eq!( + session + .coordinator + .lock() + .unwrap() + .cached_offsets_with_paths_for_pid(42) + .unwrap()[0] + .cookie, + 3 + ); + } + + fn test_session() -> GhostSession { // Create a merged config with source settings let args = ParsedArgs { binary_path: None, @@ -1442,8 +2012,12 @@ mod tests { }, }; - // Create session with config - should automatically set resolver - let session = GhostSession::new_with_config(&resolved_config); + GhostSession::new_with_config(&resolved_config) + } + + #[test] + fn test_new_with_config_sets_source_resolver() { + let session = test_session(); // Verify resolver was set correctly from config let rules = session.source_path_resolver.get_all_rules(); diff --git a/ghostscope/src/script/attach.rs b/ghostscope/src/script/attach.rs index 18a504ba..9f28f6d3 100644 --- a/ghostscope/src/script/attach.rs +++ b/ghostscope/src/script/attach.rs @@ -170,6 +170,9 @@ pub(super) async fn create_and_attach_loader( &config.backtrace_module_row_ranges, ) .context("Failed to populate DWARF backtrace unwind rows")?; + session + .seed_backtrace_runtime_rows(&mut loader) + .context("Failed to seed completed runtime backtrace modules")?; loader .register_backtrace_tail_call_program( config diff --git a/ghostscope/src/script/cli.rs b/ghostscope/src/script/cli.rs index 68303ffc..8a4e40d1 100644 --- a/ghostscope/src/script/cli.rs +++ b/ghostscope/src/script/cli.rs @@ -57,7 +57,7 @@ pub async fn compile_and_load_script_for_cli( compile_options: &ghostscope_compiler::CompileOptions, ) -> Result<()> { let mut compile_options = compile_options.clone(); - prepare_runtime_modules_before_compile(script, session, &mut compile_options)?; + prepare_runtime_modules_before_compile(script, session, &mut compile_options).await?; let compilation_result = compile_script_for_cli(script, session, &compile_options)?; ensure_prefill_for_session_pid(session); diff --git a/ghostscope/src/script/runtime_prep.rs b/ghostscope/src/script/runtime_prep.rs index e15bd7ca..487b666d 100644 --- a/ghostscope/src/script/runtime_prep.rs +++ b/ghostscope/src/script/runtime_prep.rs @@ -33,11 +33,15 @@ fn script_contains_backtrace(script: &str) -> bool { .unwrap_or(false) } -pub(super) fn prepare_runtime_modules_before_compile( +pub(super) async fn prepare_runtime_modules_before_compile( script: &str, session: &mut GhostSession, compile_options: &mut ghostscope_compiler::CompileOptions, ) -> Result<()> { + if let Err(error) = session.refresh_pid_analyzer_before_compile().await { + warn!("Failed to refresh PID runtime modules before compilation: {error:#}"); + } + if session.is_target_mode() && script_contains_backtrace(script) { session.enable_target_backtrace_runtime_modules(); if let Err(error) = session.prepare_target_backtrace_module_mappings() { diff --git a/ghostscope/src/script/tui.rs b/ghostscope/src/script/tui.rs index 33f55b69..d30f06fd 100644 --- a/ghostscope/src/script/tui.rs +++ b/ghostscope/src/script/tui.rs @@ -17,7 +17,7 @@ pub async fn compile_and_load_script_for_tui( compile_options: &ghostscope_compiler::CompileOptions, ) -> Result { let mut compile_options = compile_options.clone(); - prepare_runtime_modules_before_compile(script, session, &mut compile_options)?; + prepare_runtime_modules_before_compile(script, session, &mut compile_options).await?; let binary_path = main_executable_path(session)?; // Compilation is synchronous and can be long-running. Mark the section as diff --git a/ghostscope/src/trace/actor.rs b/ghostscope/src/trace/actor.rs index ee5fe615..7294f0b8 100644 --- a/ghostscope/src/trace/actor.rs +++ b/ghostscope/src/trace/actor.rs @@ -71,6 +71,28 @@ pub(super) struct TraceActorHandle { } impl TraceActorHandle { + #[cfg(test)] + pub(super) fn recording_backtrace_actor() -> (Self, mpsc::UnboundedReceiver>) { + let (command_sender, mut commands) = mpsc::channel(TRACE_COMMAND_CHANNEL_CAPACITY); + let (record, records) = mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Some(command) = commands.recv().await { + match command { + TraceCommand::AppendBacktraceRows { modules, response } => { + let _ = record.send(modules.iter().map(|(cookie, _)| *cookie).collect()); + let _ = response.send(Ok(TraceActorBacktraceUpdate::default())); + } + TraceCommand::Delete(response) => { + let _ = response.send(Ok(())); + break; + } + _ => panic!("unexpected test actor command"), + } + } + }); + (Self { command_sender }, records) + } + pub async fn enable(&self) -> Result<()> { self.request(TraceCommand::Enable).await } diff --git a/ghostscope/src/trace/backtrace.rs b/ghostscope/src/trace/backtrace.rs index b17869e0..2123c85e 100644 --- a/ghostscope/src/trace/backtrace.rs +++ b/ghostscope/src/trace/backtrace.rs @@ -1,5 +1,9 @@ -use ghostscope_dwarf::{DwarfAnalyzer, FunctionParameter, ModuleAddress, PcContext}; +use ghostscope_dwarf::{ + DwarfAnalyzer, FunctionParameter, LoadedModuleRuntimeInfo, ModuleAddress, PcContext, +}; +#[cfg(test)] use ghostscope_process::ProcessManager; +use ghostscope_process::ProcessManagerSnapshot; #[cfg(test)] use ghostscope_protocol::trace_event::backtrace_error_label; use ghostscope_protocol::trace_event::{ @@ -21,6 +25,26 @@ struct ResolvedFrameModule { pc: u64, } +#[derive(Clone, Copy)] +struct FrameRenderInput<'a> { + frame: &'a ParsedBacktraceFrame, + pc_is_normalized: bool, +} + +impl<'a> FrameRenderInput<'a> { + fn new( + frame: &'a ParsedBacktraceFrame, + status: BacktraceStatus, + frame_index: usize, + frame_count: usize, + ) -> Self { + Self { + frame, + pc_is_normalized: backtrace_frame_pc_is_normalized(status, frame_index, frame_count), + } + } +} + #[derive(Debug)] pub struct BacktraceRenderer { #[cfg(test)] @@ -103,6 +127,7 @@ impl PidCacheKey { #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] struct FrameRenderCacheKey { + mapping_generation: u64, pids: PidCacheKey, analyzer_present: bool, index: u16, @@ -111,10 +136,12 @@ struct FrameRenderCacheKey { pc: u64, raw_ip: u64, frame_flags: u16, + pc_is_normalized: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] struct StatusCacheKey { + mapping_generation: u64, pids: PidCacheKey, analyzer_present: bool, module_cookie: u64, @@ -130,7 +157,7 @@ impl BacktraceRenderer { &mut self, event: &ParsedTraceEvent, analyzer: Option<&DwarfAnalyzer>, - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, proc_pid_hint: Option, ) -> UiTraceEvent { let mut items = Vec::new(); @@ -180,7 +207,7 @@ impl BacktraceRenderer { &mut self, event: &ParsedTraceEvent, analyzer: Option<&DwarfAnalyzer>, - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, proc_pid_hint: Option, ) -> ParsedTraceEvent { let mut changed = false; @@ -224,7 +251,7 @@ impl BacktraceRenderer { instruction: &ParsedInstruction, event_pid: u32, analyzer: Option<&DwarfAnalyzer>, - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, proc_pid_hint: Option, ) -> Vec { let ParsedInstruction::Backtrace { @@ -254,7 +281,14 @@ impl BacktraceRenderer { )]; for (index, frame) in frames.iter().enumerate() { - lines.extend(self.format_frame(index, frame, *flags, analyzer, coordinator, &pids)); + lines.extend(self.format_frame( + index, + FrameRenderInput::new(frame, *status, index, frames.len()), + *flags, + analyzer, + coordinator, + &pids, + )); } if display_status != BacktraceStatus::Complete { @@ -274,7 +308,7 @@ impl BacktraceRenderer { instruction: &ParsedInstruction, event_pid: u32, analyzer: Option<&DwarfAnalyzer>, - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, proc_pid_hint: Option, ) -> BacktraceDisplay { let ParsedInstruction::Backtrace { @@ -309,7 +343,7 @@ impl BacktraceRenderer { for (index, frame) in frames.iter().enumerate() { display_frames.extend(self.display_frame( index, - frame, + FrameRenderInput::new(frame, *status, index, frames.len()), *flags, analyzer, coordinator, @@ -333,7 +367,7 @@ impl BacktraceRenderer { error_code: u16, frames: &[ParsedBacktraceFrame], analyzer: Option<&DwarfAnalyzer>, - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, pids: &[u32], ) -> BacktraceStatus { if !matches!( @@ -351,6 +385,7 @@ impl BacktraceRenderer { }; let cache_key = StatusCacheKey { + mapping_generation: coordinator.generation(), pids: PidCacheKey::from_pids(pids), analyzer_present: true, module_cookie: last_frame.module_cookie, @@ -364,8 +399,15 @@ impl BacktraceRenderer { return cached; } - let Some(module) = resolve_frame_module(coordinator, Some(analyzer), pids, last_frame) - else { + let pc_is_normalized = + backtrace_frame_pc_is_normalized(status, frames.len() - 1, frames.len()); + let Some(module) = resolve_frame_module( + coordinator, + Some(analyzer), + pids, + last_frame, + pc_is_normalized, + ) else { return status; }; @@ -382,13 +424,18 @@ impl BacktraceRenderer { fn format_frame( &mut self, index: usize, - frame: &ParsedBacktraceFrame, + input: FrameRenderInput<'_>, flags: u8, analyzer: Option<&DwarfAnalyzer>, - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, pids: &[u32], ) -> Vec { + let FrameRenderInput { + frame, + pc_is_normalized, + } = input; let cache_key = FrameRenderCacheKey { + mapping_generation: coordinator.generation(), pids: PidCacheKey::from_pids(pids), analyzer_present: analyzer.is_some(), index: index.min(u16::MAX as usize) as u16, @@ -397,6 +444,7 @@ impl BacktraceRenderer { pc: frame.pc, raw_ip: frame.raw_ip, frame_flags: frame.flags, + pc_is_normalized, }; if let Some(cached) = self.frame_cache.get(&cache_key) { return cached; @@ -405,7 +453,7 @@ impl BacktraceRenderer { let raw = (flags & BACKTRACE_FLAG_RAW) != 0; let full = (flags & BACKTRACE_FLAG_FULL) != 0; let inline = (flags & BACKTRACE_FLAG_INLINE) != 0; - let module = resolve_frame_module(coordinator, analyzer, pids, frame); + let module = resolve_frame_module(coordinator, analyzer, pids, frame, pc_is_normalized); let frame_pc = module.as_ref().map(|module| module.pc).unwrap_or(frame.pc); let lookup_pc = if index == 0 { frame_pc @@ -487,13 +535,18 @@ impl BacktraceRenderer { fn display_frame( &mut self, index: usize, - frame: &ParsedBacktraceFrame, + input: FrameRenderInput<'_>, flags: u8, analyzer: Option<&DwarfAnalyzer>, - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, pids: &[u32], ) -> Vec { + let FrameRenderInput { + frame, + pc_is_normalized, + } = input; let cache_key = FrameRenderCacheKey { + mapping_generation: coordinator.generation(), pids: PidCacheKey::from_pids(pids), analyzer_present: analyzer.is_some(), index: index.min(u16::MAX as usize) as u16, @@ -502,6 +555,7 @@ impl BacktraceRenderer { pc: frame.pc, raw_ip: frame.raw_ip, frame_flags: frame.flags, + pc_is_normalized, }; if let Some(cached) = self.frame_display_cache.get(&cache_key) { return cached; @@ -510,7 +564,7 @@ impl BacktraceRenderer { let raw = (flags & BACKTRACE_FLAG_RAW) != 0; let full = (flags & BACKTRACE_FLAG_FULL) != 0; let inline = (flags & BACKTRACE_FLAG_INLINE) != 0; - let module = resolve_frame_module(coordinator, analyzer, pids, frame); + let module = resolve_frame_module(coordinator, analyzer, pids, frame, pc_is_normalized); let frame_pc = module.as_ref().map(|module| module.pc).unwrap_or(frame.pc); let lookup_pc = if index == 0 { frame_pc @@ -661,7 +715,7 @@ fn is_process_entry_frame( fn candidate_pids( event_pid: u32, proc_pid_hint: Option, - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, ) -> Vec { coordinator.candidate_proc_pids_for_runtime_pid(event_pid, proc_pid_hint) } @@ -679,10 +733,11 @@ fn format_backtrace_header(status: BacktraceStatus, frames: usize, requested_dep } fn resolve_frame_module( - coordinator: &ProcessManager, + coordinator: &ProcessManagerSnapshot, analyzer: Option<&DwarfAnalyzer>, pids: &[u32], frame: &ParsedBacktraceFrame, + pc_is_normalized: bool, ) -> Option { for pid in pids { if let Some(entries) = coordinator.cached_offsets_with_paths_for_pid(*pid) { @@ -707,12 +762,13 @@ fn resolve_frame_module( } } } - resolve_frame_module_from_analyzer(analyzer, frame) + resolve_frame_module_from_analyzer(analyzer, frame, pc_is_normalized) } fn resolve_frame_module_from_analyzer( analyzer: Option<&DwarfAnalyzer>, frame: &ParsedBacktraceFrame, + pc_is_normalized: bool, ) -> Option { let analyzer = analyzer?; analyzer @@ -724,20 +780,42 @@ fn resolve_frame_module_from_analyzer( if cookie != frame.module_cookie { return None; } - if module.loaded_address.is_none() && frame.raw_ip != 0 && frame.raw_ip == frame.pc { - return None; - } - if module.size != 0 && frame.pc >= module.size { - return None; - } + let pc = analyzer_module_pc(&module, frame, pc_is_normalized)?; Some(ResolvedFrameModule { module_path: module.module_path, cookie, - pc: frame.pc, + pc, }) }) } +fn backtrace_frame_pc_is_normalized( + status: BacktraceStatus, + frame_index: usize, + frame_count: usize, +) -> bool { + // OffsetsUnavailable describes the final stopping frame. Earlier frames + // were normalized before the unwinder reached the missing mapping. + status != BacktraceStatus::OffsetsUnavailable || frame_index < frame_count.saturating_sub(1) +} + +fn analyzer_module_pc( + module: &LoadedModuleRuntimeInfo, + frame: &ParsedBacktraceFrame, + pc_is_normalized: bool, +) -> Option { + // A path-only analyzer has no runtime bias with which to normalize an + // offsets-unavailable frame. Equality between raw_ip and pc cannot make + // that distinction because a non-PIE ET_EXEC legitimately has zero bias. + if !pc_is_normalized && module.loaded_address.is_none() { + return None; + } + if module.size != 0 && frame.pc >= module.size { + return None; + } + Some(frame.pc) +} + #[cfg(test)] fn format_raw_frame( index: usize, @@ -881,8 +959,9 @@ mod tests { ], }; let coordinator = ProcessManager::new(); + let snapshot = coordinator.snapshot_reader().load(); let rendered = - BacktraceRenderer::default().render_event_backtraces(&event, None, &coordinator, None); + BacktraceRenderer::default().render_event_backtraces(&event, None, &snapshot, None); let output = rendered.to_formatted_output(); assert_eq!(output[0], "before"); @@ -902,16 +981,52 @@ mod tests { ); } + #[test] + fn normalized_zero_bias_frame_matches_path_only_executable() { + let module = LoadedModuleRuntimeInfo { + module_path: PathBuf::from("/tmp/non-pie-executable"), + loaded_address: None, + load_bias: None, + size: 0, + }; + let frame = ParsedBacktraceFrame { + module_cookie: 0x1234, + pc: 0x401000, + raw_ip: 0x401000, + flags: 0, + }; + + assert_eq!(analyzer_module_pc(&module, &frame, true), Some(frame.pc)); + assert_eq!(analyzer_module_pc(&module, &frame, false), None); + } + + #[test] + fn offsets_unavailable_marks_only_the_stopping_frame_as_unnormalized() { + assert!(backtrace_frame_pc_is_normalized( + BacktraceStatus::OffsetsUnavailable, + 0, + 2 + )); + assert!(!backtrace_frame_pc_is_normalized( + BacktraceStatus::OffsetsUnavailable, + 1, + 2 + )); + assert!(backtrace_frame_pc_is_normalized( + BacktraceStatus::Complete, + 0, + 1 + )); + } + #[test] fn candidate_pids_include_runtime_alias_before_event_pid() { let mut coordinator = ProcessManager::new(); coordinator.record_runtime_pid_alias(4242, 42); + let snapshot = coordinator.snapshot_reader().load(); - assert_eq!(candidate_pids(4242, None, &coordinator), vec![42, 4242]); - assert_eq!( - candidate_pids(4242, Some(7), &coordinator), - vec![7, 42, 4242] - ); + assert_eq!(candidate_pids(4242, None, &snapshot), vec![42, 4242]); + assert_eq!(candidate_pids(4242, Some(7), &snapshot), vec![7, 42, 4242]); assert_eq!( PidCacheKey::from_pids(&[7, 42, 4242]), PidCacheKey { @@ -953,8 +1068,9 @@ mod tests { ], }; let coordinator = ProcessManager::new(); + let snapshot = coordinator.snapshot_reader().load(); let rendered = - BacktraceRenderer::default().render_event_for_tui(&event, None, &coordinator, None); + BacktraceRenderer::default().render_event_for_tui(&event, None, &snapshot, None); assert_eq!(rendered.items.len(), 3); assert!(matches!( diff --git a/ghostscope/src/trace/backtrace_runtime.rs b/ghostscope/src/trace/backtrace_runtime.rs index 569edef6..cd988943 100644 --- a/ghostscope/src/trace/backtrace_runtime.rs +++ b/ghostscope/src/trace/backtrace_runtime.rs @@ -21,16 +21,18 @@ impl BacktraceRuntimeModuleObservation { } /// A stopping-frame observation resolved against one exact process snapshot. -#[derive(Debug, Clone)] +#[derive(Debug)] pub struct ResolvedBacktraceRuntimeModule { pub observation: BacktraceRuntimeModuleObservation, pub proc_pid: u32, pub cookie: u64, pub module: LoadedModuleRuntimeInfo, + pub entries: Vec, + pub probe: Option>, } /// Result of matching an observation to a current process mapping. -#[derive(Debug, Clone)] +#[derive(Debug)] pub enum BacktraceRuntimeModuleResolution { Resolved(ResolvedBacktraceRuntimeModule), Unavailable, @@ -88,16 +90,6 @@ impl BacktraceRuntimeRunner { match coordinator.refresh_module_for_runtime_ip(proc_pid, observation.raw_ip) { Ok(Some(entry)) => Some(entry), Ok(None) => { - let entries = coordinator - .cached_offsets_with_paths_for_pid(proc_pid) - .unwrap_or_default() - .to_vec(); - Self::publish_observation_pid_snapshot( - coordinator, - proc_pid, - observation, - &entries, - ); return BacktraceRuntimeModuleResolution::Unavailable; } Err(error) => { @@ -107,16 +99,6 @@ impl BacktraceRuntimeRunner { observation.raw_ip, error ); - let entries = coordinator - .cached_offsets_with_paths_for_pid(proc_pid) - .unwrap_or_default() - .to_vec(); - Self::publish_observation_pid_snapshot( - coordinator, - proc_pid, - observation, - &entries, - ); return BacktraceRuntimeModuleResolution::Unavailable; } } @@ -142,7 +124,6 @@ impl BacktraceRuntimeRunner { } None => runtime_entry_for_observation(&entries, observation).cloned(), }; - Self::publish_observation_pid_snapshot(coordinator, proc_pid, observation, &entries); let Some(entry) = entry else { return BacktraceRuntimeModuleResolution::Unavailable; }; @@ -156,11 +137,20 @@ impl BacktraceRuntimeRunner { return BacktraceRuntimeModuleResolution::IdentityChanged; } + let probe = match ProcessManager::probe_runtime_module(proc_pid, &entry) { + Ok(probe) => probe, + Err(error) => { + tracing::debug!("Could not retain validated runtime module: {error:#}"); + return BacktraceRuntimeModuleResolution::IdentityChanged; + } + }; BacktraceRuntimeModuleResolution::Resolved(ResolvedBacktraceRuntimeModule { observation, proc_pid, cookie: entry.cookie, module: runtime_module_from_entry(&entry), + entries, + probe: Some(Box::new(probe)), }) } @@ -222,7 +212,7 @@ impl BacktraceRuntimeRunner { } } - fn publish_observation_pid_snapshot( + pub(crate) fn publish_observation_pid_snapshot( coordinator: &mut ProcessManager, proc_pid: u32, observation: BacktraceRuntimeModuleObservation, diff --git a/ghostscope/src/trace/manager.rs b/ghostscope/src/trace/manager.rs index ede1aada..bd1d93db 100644 --- a/ghostscope/src/trace/manager.rs +++ b/ghostscope/src/trace/manager.rs @@ -100,8 +100,9 @@ impl BacktraceUnwindRowsAppender { let modules = Arc::new(modules); let mut updates = Vec::with_capacity(self.targets.len()); - // Shared pinned maps require ordered publication so actors do not race - // while inserting the same module rows. + // Loaders serialize all shared-map publishers, including initialization + // of newly attached traces. Visit actors in order to advance each source + // generation after the maps have been updated. for (trace_id, actor) in self.targets { updates.push(( trace_id, @@ -711,6 +712,27 @@ mod tests { .is_enabled = true; } + #[tokio::test] + async fn completed_module_publication_uses_replacement_actors() { + use crate::trace::actor::TraceActorHandle; + let mut manager = TraceManager::new(); + add_test_trace(&mut manager, 1); + let (old_actor, mut old_records) = TraceActorHandle::recording_backtrace_actor(); + manager.traces.get_mut(&1).unwrap().actor = Some(old_actor); + // Loading starts here. All of its original traces disappear before it completes. + manager.delete_trace(1).await.unwrap(); + add_test_trace(&mut manager, 2); + let (replacement, mut records) = TraceActorHandle::recording_backtrace_actor(); + manager.traces.get_mut(&2).unwrap().actor = Some(replacement); + let result = manager + .backtrace_unwind_rows_appender() + .append(vec![(0x42, vec![Default::default()])]) + .await; + manager.apply_backtrace_unwind_rows_append(result); + assert_eq!(records.recv().await, Some(vec![0x42])); + assert!(old_records.recv().await.is_none()); + } + #[test] fn append_with_limit_preserves_capacity_under_limit() { let mut dst = vec![1, 2]; diff --git a/ghostscope/src/tui/coordinator.rs b/ghostscope/src/tui/coordinator.rs index 4aed85ce..582867ef 100644 --- a/ghostscope/src/tui/coordinator.rs +++ b/ghostscope/src/tui/coordinator.rs @@ -147,7 +147,6 @@ async fn run_runtime_coordinator( let trace_sender = runtime_channels.create_trace_sender(); let trace_channel_capacity = runtime_channels.trace_channel_capacity; let mut backtrace_renderer = crate::trace::backtrace::BacktraceRenderer::default(); - let fallback_coordinator = ghostscope_process::ProcessManager::new(); let mut backpressure_state = TraceBackpressureState::default(); let mut backpressure_report_ticker = tokio::time::interval(tokio::time::Duration::from_secs(1)); backpressure_report_ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); @@ -169,27 +168,15 @@ async fn run_runtime_coordinator( if let Some(ref mut session) = session { let runtime_refresh_request = BacktraceRuntimeModuleRequest::from_events(&events); + let process_snapshot = session.process_manager_snapshot(); tracing::debug!("Forwarding {} trace events to UI", events.len()); for event_data in events { - let event_data = match session.coordinator.try_lock() { - Ok(coordinator) => backtrace_renderer.render_event_for_tui( - &event_data, - session.process_analyzer.as_ref(), - &coordinator, - session.proc_pid(), - ), - Err(std::sync::TryLockError::WouldBlock) => { - backtrace_renderer.render_event_for_tui( - &event_data, - session.process_analyzer.as_ref(), - &fallback_coordinator, - session.proc_pid(), - ) - } - Err(std::sync::TryLockError::Poisoned(_)) => { - panic!("coordinator mutex poisoned") - } - }; + let event_data = backtrace_renderer.render_event_for_tui( + &event_data, + session.process_analyzer.as_ref(), + &process_snapshot, + session.proc_pid(), + ); match forward_trace_event( &trace_sender, event_data,