From 7efff9c96354da47bf58f95c9ce93983c27a30c9 Mon Sep 17 00:00:00 2001 From: itdevwu Date: Mon, 9 Nov 2026 18:00:34 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20refactor:=20move=20CUP?= =?UTF-8?q?TI=20filter=20and=20GPU=20aggregate=20logic=20out=20of=20CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Selector-to-Agent filter lowering now lives in collector::cupti and aggregate merging, selector hints, and result construction live in core::gpu_aggregate, mirroring syscall_aggregate::analyze. The CLI keeps only collector lifecycle checks and type mapping. A kernel aggregate without a name or a zero-count group now returns an explicit error instead of panicking or emitting a NaN mean. Also correct the stale discover help text. --- docs/architecture.md | 4 +- xprobe/cli/src/main.rs | 465 +++------------------------ xprobe/collector/src/cupti.rs | 4 + xprobe/collector/src/cupti/filter.rs | 249 ++++++++++++++ xprobe/core/src/gpu_aggregate.rs | 446 +++++++++++++++++++++++++ xprobe/core/src/lib.rs | 1 + 6 files changed, 752 insertions(+), 417 deletions(-) create mode 100644 xprobe/collector/src/cupti/filter.rs create mode 100644 xprobe/core/src/gpu_aggregate.rs diff --git a/docs/architecture.md b/docs/architecture.md index 9ceb5df..93b5dfc 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -28,9 +28,9 @@ attachment, collection, correlation, logical CUPTI shutdown, and cleanup. | Path | Responsibility | | --- | --- | | `xprobe/cli` | Four public commands, rendering, exit codes, orchestration | -| `xprobe/core` | doctor, process identity, discovery, ELF resolution, validation, ptrace injection | +| `xprobe/core` | doctor, process identity, discovery, ELF resolution, validation, ptrace injection, inventory normalization | | `xprobe/protocol` | Strict serde contracts and generated JSON Schema | -| `xprobe/collector` | eBPF collection, CUPTI control protocol and ABI decoding | +| `xprobe/collector` | eBPF collection, CUPTI selector filters, control protocol and ABI decoding | | `xprobe/correlator` | Selector matching, pair evidence, statistics and quality | | `xprobe/exporter` | Event JSONL and Chrome Trace Event Format | | `bpf/` | PID-scoped uprobe, USDT, syscall, tracepoint, and aggregate programs | diff --git a/xprobe/cli/src/main.rs b/xprobe/cli/src/main.rs index d1df995..1adc0e3 100644 --- a/xprobe/cli/src/main.rs +++ b/xprobe/cli/src/main.rs @@ -28,20 +28,19 @@ use xprobe_collector::{ use xprobe_core::{ cpu_sampling, cpu_sampling_analysis::{self, CpuCaptureData, RawCpuSample}, - cupti_compat, discover, doctor, inject, inspect, resolve, syscall_aggregate, validate, + cupti_compat, discover, doctor, gpu_aggregate, inject, inspect, resolve, syscall_aggregate, + validate, }; use xprobe_correlator::{MeasureError, MeasureOptions, measure}; use xprobe_exporter::{events_to_chrome_trace, events_to_jsonl}; use xprobe_protocol::{ - AggregateActivity, AggregateCollectionSummary, AggregateDuration, AggregateGroup, - AggregateInventory, AggregateInventoryResult, CapabilityReport, CaptureCompleteness, - CheckResult, ClockDomain, CpuSampleEvent, CpuSampleInventoryResult, CpuSamplingSpec, - CpuSamplingValidationResult, CuptiCollectionSummary, DiscoveryResult, ErrorCode, ErrorResponse, - Event, EventSource, EventType, ExportFormat, HostCaptureResult, MatchPolicy, MeasurementMode, - MeasurementResult, MeasurementSpec, MemcpyKind, ProcessReport, ResolvedCudaSelector, - ResolvedLinuxSelector, ResolvedProbe, SchemaVersion, SessionStatus, SyscallAggregateResult, - SyscallAggregateSpec, SyscallAggregateValidationResult, TargetIdentity, TraceExportResult, - ValidationResult, Warning, XprobeError, + AggregateActivity, AggregateInventoryResult, CapabilityReport, CheckResult, ClockDomain, + CpuSampleEvent, CpuSampleInventoryResult, CpuSamplingSpec, CpuSamplingValidationResult, + CuptiCollectionSummary, DiscoveryResult, ErrorCode, ErrorResponse, Event, EventSource, + EventType, ExportFormat, HostCaptureResult, MatchPolicy, MeasurementMode, MeasurementResult, + MeasurementSpec, ProcessReport, ResolvedLinuxSelector, ResolvedProbe, SchemaVersion, + SessionStatus, SyscallAggregateResult, SyscallAggregateSpec, SyscallAggregateValidationResult, + TargetIdentity, TraceExportResult, ValidationResult, Warning, XprobeError, }; #[derive(Debug, Parser)] @@ -55,7 +54,7 @@ struct Cli { enum Command { /// Inspect local tracing and GPU capabilities. Doctor(DoctorArgs), - /// Discover event selectors available in a target process. + /// List NVML-confirmed CUDA context holders under a process-tree root. Discover(DiscoverArgs), /// Inspect a target process without attaching probes. #[command(hide = true)] @@ -2291,7 +2290,7 @@ fn validate_aggregate_selection(validation: &ValidationResult) -> Result<(), Com if let Some(pattern) = start.kernel_name_regex.as_deref() && pattern != ".*" && matches!( - bounded_kernel_name_filter(pattern), + cupti::bounded_name_filter(pattern), cupti::CuptiNameFilter::Any ) { @@ -2385,7 +2384,11 @@ fn cupti_arm_config( let filters = [&validation.start, &validation.end] .into_iter() .filter_map(|endpoint| endpoint.cuda.as_ref()) - .map(cupti_event_filter) + .map(|selector| { + cupti::event_filter(selector).map_err(|error| { + CommandFailure::new(ErrorCode::InvalidEventSelector, error.to_string(), false) + }) + }) .collect::, _>>()?; let record_capacity = u64::try_from(max_events).map_err(|error| { CommandFailure::new( @@ -2404,122 +2407,6 @@ fn cupti_arm_config( })) } -fn cupti_event_filter( - selector: &ResolvedCudaSelector, -) -> Result { - let record_kind = match selector.event_type { - EventType::CudaApiEntry => cupti::CuptiRecordKind::CudaApiEntry, - EventType::CudaApiExit => cupti::CuptiRecordKind::CudaApiExit, - EventType::GpuKernelStart => cupti::CuptiRecordKind::GpuKernelStart, - EventType::GpuKernelEnd => cupti::CuptiRecordKind::GpuKernelEnd, - EventType::GpuMemcpyStart => cupti::CuptiRecordKind::GpuMemcpyStart, - EventType::GpuMemcpyEnd => cupti::CuptiRecordKind::GpuMemcpyEnd, - EventType::GpuMemsetStart => cupti::CuptiRecordKind::GpuMemsetStart, - EventType::GpuMemsetEnd => cupti::CuptiRecordKind::GpuMemsetEnd, - EventType::NvtxRangeStart => cupti::CuptiRecordKind::NvtxRangeStart, - EventType::NvtxRangeEnd => cupti::CuptiRecordKind::NvtxRangeEnd, - _ => { - return Err(CommandFailure::new( - ErrorCode::InvalidEventSelector, - "validated CUDA endpoint has a non-CUDA event type", - false, - )); - } - }; - let api_domain = match selector.api_domain.as_deref() { - None => cupti::CuptiApiDomain::Any, - Some("driver_api") => cupti::CuptiApiDomain::Driver, - Some("runtime_api") => cupti::CuptiApiDomain::Runtime, - Some(domain) => { - return Err(CommandFailure::new( - ErrorCode::InvalidEventSelector, - format!("validated CUDA endpoint has unsupported API domain {domain:?}"), - false, - )); - } - }; - let memcpy_kind = match selector.memcpy_kind { - None | Some(MemcpyKind::Unknown) => cupti::CuptiMemcpyKind::Any, - Some(MemcpyKind::HostToDevice) => cupti::CuptiMemcpyKind::HostToDevice, - Some(MemcpyKind::DeviceToHost) => cupti::CuptiMemcpyKind::DeviceToHost, - Some(MemcpyKind::DeviceToDevice) => cupti::CuptiMemcpyKind::DeviceToDevice, - Some(MemcpyKind::HostToHost) => cupti::CuptiMemcpyKind::HostToHost, - Some(MemcpyKind::PeerToPeer) => cupti::CuptiMemcpyKind::PeerToPeer, - }; - let name = if let Some(name) = selector.api_name.as_ref() { - cupti::CuptiNameFilter::Exact(name.clone()) - } else if let Some(pattern) = selector.kernel_name_regex.as_deref() { - bounded_kernel_name_filter(pattern) - } else if let Some(pattern) = selector.nvtx_name_regex.as_deref() { - bounded_kernel_name_filter(pattern) - } else { - cupti::CuptiNameFilter::Any - }; - Ok(cupti::CuptiEventFilter { - record_kind, - api_domain, - memcpy_kind, - name, - }) -} - -fn bounded_kernel_name_filter(pattern: &str) -> cupti::CuptiNameFilter { - if pattern == ".*" { - return cupti::CuptiNameFilter::Any; - } - let candidates = [ - ("^", "$", 1_u8), - ("^", ".*", 2), - (".*", "$", 3), - (".*", ".*", 4), - ("^", "", 2), - ("", "$", 3), - (".*", "", 4), - ("", "", 4), - ]; - for (prefix, suffix, kind) in candidates { - let Some(inner) = pattern - .strip_prefix(prefix) - .and_then(|value| value.strip_suffix(suffix)) - else { - continue; - }; - let Some(literal) = regex_literal(inner) else { - continue; - }; - if literal.is_empty() || literal.len() >= cupti::CUPTI_NAME_CAPACITY { - continue; - } - return match kind { - 1 => cupti::CuptiNameFilter::Exact(literal), - 2 => cupti::CuptiNameFilter::Prefix(literal), - 3 => cupti::CuptiNameFilter::Suffix(literal), - 4 => cupti::CuptiNameFilter::Contains(literal), - _ => unreachable!("fixed kernel filter kind"), - }; - } - cupti::CuptiNameFilter::Any -} - -fn regex_literal(pattern: &str) -> Option { - let mut literal = String::with_capacity(pattern.len()); - let mut characters = pattern.chars(); - while let Some(character) = characters.next() { - if character == '\\' { - let escaped = characters.next()?; - if escaped.is_ascii_alphanumeric() { - return None; - } - literal.push(escaped); - } else if ".+*?()|[]{}^$".contains(character) { - return None; - } else { - literal.push(character); - } - } - Some(literal) -} - struct CuptiActivation { socket: Option, managed: bool, @@ -2991,67 +2878,41 @@ fn aggregate_inventory_result( request: &LiveMeasureRequest, ) -> Result { validate_aggregate_capture(capture)?; - let raw_groups = u64::try_from(capture.aggregate_groups.len()).map_err(|error| { - CommandFailure::new( - ErrorCode::TraceExportFailed, - format!("aggregate group count exceeds u64: {error}"), - false, - ) - })?; - let groups = merge_aggregate_groups(capture.aggregate_groups.clone())?; - let grouped_activities = groups.iter().try_fold(0_u64, |total, group| { - total.checked_add(group.count).ok_or_else(|| { - CommandFailure::new( - ErrorCode::TraceExportFailed, - "aggregate activity count overflowed", - false, - ) - }) - })?; - if grouped_activities != capture.observed_records { - return Err(CommandFailure::new( - ErrorCode::TraceExportFailed, - format!( - "aggregate activity counters are inconsistent: grouped={grouped_activities}, observed={}", - capture.observed_records - ), - false, - )); - } - let distinct_groups = u64::try_from(groups.len()).map_err(|error| { - CommandFailure::new( - ErrorCode::TraceExportFailed, - format!("distinct aggregate group count exceeds u64: {error}"), - false, - ) - })?; - let groups = groups - .into_iter() - .map(protocol_aggregate_group) - .collect::>(); - Ok(AggregateInventoryResult { - schema_version: SchemaVersion::current(), - ok: true, - session_id: request.options.session_id.clone(), - status: SessionStatus::Completed, - inventory: AggregateInventory { - name: request.options.name.clone(), - start_selector: request.options.start_selector.clone(), - end_selector: request.options.end_selector.clone(), - groups, - }, - collection: AggregateCollectionSummary { - completeness: CaptureCompleteness::Complete, - observed_activities: capture.observed_records, - grouped_activities, - dropped_activities: 0, - group_capacity: capture.record_capacity, - groups: distinct_groups, - occupied_slots: raw_groups, - table_utilization: utilization(raw_groups, capture.record_capacity), + let capture = gpu_aggregate::GpuAggregateCapture { + groups: capture + .aggregate_groups + .iter() + .map(|group| gpu_aggregate::RawGpuAggregateGroup { + activity: match group.activity { + cupti::CuptiAggregateActivity::Kernel => AggregateActivity::Kernel, + cupti::CuptiAggregateActivity::Memcpy => AggregateActivity::Memcpy, + cupti::CuptiAggregateActivity::Memset => AggregateActivity::Memset, + }, + name: group.name.clone(), + name_complete: group.name_complete, + device_id: group.device_id, + memcpy_kind: group.memcpy_kind.clone(), + count: group.count, + total_duration_ns: group.total_duration_ns, + min_duration_ns: group.min_duration_ns, + max_duration_ns: group.max_duration_ns, + total_bytes: group.total_bytes, + }) + .collect(), + observed_activities: capture.observed_records, + group_capacity: capture.record_capacity, + }; + let options = &request.options; + gpu_aggregate::analyze( + gpu_aggregate::GpuAggregateRequest { + session_id: options.session_id.clone(), + name: options.name.clone(), + start_selector: options.start_selector.clone(), + end_selector: options.end_selector.clone(), }, - warnings: Vec::new(), - }) + capture, + ) + .map_err(|error| CommandFailure::new(error.code(), error.to_string(), error.recoverable())) } fn validate_aggregate_capture(capture: &cupti::CuptiCapture) -> Result<(), CommandFailure> { @@ -3105,170 +2966,6 @@ fn validate_aggregate_capture(capture: &cupti::CuptiCapture) -> Result<(), Comma Ok(()) } -fn merge_aggregate_groups( - mut groups: Vec, -) -> Result, CommandFailure> { - groups.sort_by(|left, right| { - aggregate_activity_rank(left.activity) - .cmp(&aggregate_activity_rank(right.activity)) - .then_with(|| left.name.cmp(&right.name)) - .then_with(|| left.device_id.cmp(&right.device_id)) - .then_with(|| { - memcpy_kind_rank(left.memcpy_kind.as_ref()) - .cmp(&memcpy_kind_rank(right.memcpy_kind.as_ref())) - }) - }); - let mut merged: Vec = Vec::with_capacity(groups.len()); - for group in groups { - let Some(previous) = merged.last_mut() else { - merged.push(group); - continue; - }; - if previous.activity != group.activity - || previous.name != group.name - || previous.device_id != group.device_id - || previous.memcpy_kind != group.memcpy_kind - { - merged.push(group); - continue; - } - previous.count = previous.count.checked_add(group.count).ok_or_else(|| { - CommandFailure::new( - ErrorCode::TraceExportFailed, - "aggregate group count overflowed while merging", - false, - ) - })?; - previous.total_duration_ns = previous - .total_duration_ns - .checked_add(group.total_duration_ns) - .ok_or_else(|| { - CommandFailure::new( - ErrorCode::TraceExportFailed, - "aggregate duration overflowed while merging", - false, - ) - })?; - previous.min_duration_ns = previous.min_duration_ns.min(group.min_duration_ns); - previous.max_duration_ns = previous.max_duration_ns.max(group.max_duration_ns); - previous.total_bytes = match (previous.total_bytes, group.total_bytes) { - (Some(left), Some(right)) => Some(left.checked_add(right).ok_or_else(|| { - CommandFailure::new( - ErrorCode::TraceExportFailed, - "aggregate byte count overflowed while merging", - false, - ) - })?), - (None, None) => None, - _ => { - return Err(CommandFailure::new( - ErrorCode::TraceExportFailed, - "aggregate byte counters are inconsistent", - false, - )); - } - }; - } - Ok(merged) -} - -const fn aggregate_activity_rank(activity: cupti::CuptiAggregateActivity) -> u8 { - match activity { - cupti::CuptiAggregateActivity::Kernel => 0, - cupti::CuptiAggregateActivity::Memcpy => 1, - cupti::CuptiAggregateActivity::Memset => 2, - } -} - -const fn memcpy_kind_rank(kind: Option<&MemcpyKind>) -> u8 { - match kind { - None => 0, - Some(&MemcpyKind::Unknown) => 1, - Some(&MemcpyKind::HostToDevice) => 2, - Some(&MemcpyKind::DeviceToHost) => 3, - Some(&MemcpyKind::DeviceToDevice) => 4, - Some(&MemcpyKind::HostToHost) => 5, - Some(&MemcpyKind::PeerToPeer) => 6, - } -} - -#[allow(clippy::cast_precision_loss)] -fn protocol_aggregate_group(group: cupti::CuptiAggregateGroup) -> AggregateGroup { - let (start_selector_hint, end_selector_hint) = aggregate_selector_hints(&group); - AggregateGroup { - activity: match group.activity { - cupti::CuptiAggregateActivity::Kernel => AggregateActivity::Kernel, - cupti::CuptiAggregateActivity::Memcpy => AggregateActivity::Memcpy, - cupti::CuptiAggregateActivity::Memset => AggregateActivity::Memset, - }, - name: group.name, - name_complete: group.name_complete, - device_id: group.device_id, - memcpy_kind: group.memcpy_kind, - start_selector_hint, - end_selector_hint, - count: group.count, - duration_ns: AggregateDuration { - min: group.min_duration_ns, - mean: group.total_duration_ns as f64 / group.count as f64, - max: group.max_duration_ns, - total: group.total_duration_ns, - }, - total_bytes: group.total_bytes, - } -} - -fn aggregate_selector_hints(group: &cupti::CuptiAggregateGroup) -> (String, String) { - match group.activity { - cupti::CuptiAggregateActivity::Kernel => { - let name = group.name.as_deref().expect("kernel aggregate has a name"); - let escaped = escape_selector_regex(name); - let suffix = if group.name_complete == Some(true) { - "$" - } else { - "" - }; - ( - format!("cuda:kernel_start:name~^{escaped}{suffix}"), - format!("cuda:kernel_end:name~^{escaped}{suffix}"), - ) - } - cupti::CuptiAggregateActivity::Memcpy => { - let kind = match group.memcpy_kind.as_ref() { - Some(MemcpyKind::HostToDevice) => Some("HtoD"), - Some(MemcpyKind::DeviceToHost) => Some("DtoH"), - Some(MemcpyKind::DeviceToDevice) => Some("DtoD"), - Some(MemcpyKind::HostToHost) => Some("HtoH"), - Some(MemcpyKind::PeerToPeer) => Some("PtoP"), - Some(MemcpyKind::Unknown) | None => None, - }; - kind.map_or_else( - || ("cuda:memcpy_start".to_owned(), "cuda:memcpy_end".to_owned()), - |kind| { - ( - format!("cuda:memcpy_start:kind={kind}"), - format!("cuda:memcpy_end:kind={kind}"), - ) - }, - ) - } - cupti::CuptiAggregateActivity::Memset => { - ("cuda:memset_start".to_owned(), "cuda:memset_end".to_owned()) - } - } -} - -fn escape_selector_regex(name: &str) -> String { - let mut escaped = String::with_capacity(name.len()); - for character in name.chars() { - if "\\.^$|?*+()[]{}".contains(character) { - escaped.push('\\'); - } - escaped.push(character); - } - escaped -} - fn injection_warning(collection: &LiveCollection) -> Warning { let mut details = BTreeMap::new(); if let Some(major) = collection.agent_major { @@ -4263,65 +3960,3 @@ const fn schema_version(version: SchemaVersion) -> &'static str { SchemaVersion::V2 => "2.0", } } - -#[cfg(test)] -mod tests { - use xprobe_collector::cupti::{CuptiAggregateActivity, CuptiAggregateGroup, CuptiNameFilter}; - - use super::{aggregate_selector_hints, bounded_kernel_name_filter, regex_literal}; - - #[test] - fn lowers_only_equivalent_bounded_kernel_patterns() { - assert_eq!( - bounded_kernel_name_filter("^flash_.*"), - CuptiNameFilter::Prefix("flash_".to_owned()) - ); - assert_eq!( - bounded_kernel_name_filter(".*_kernel$"), - CuptiNameFilter::Suffix("_kernel".to_owned()) - ); - assert_eq!( - bounded_kernel_name_filter("cudaLaunch"), - CuptiNameFilter::Contains("cudaLaunch".to_owned()) - ); - assert_eq!( - bounded_kernel_name_filter(".*xprobe_multisource_kernel.*"), - CuptiNameFilter::Contains("xprobe_multisource_kernel".to_owned()) - ); - assert_eq!( - bounded_kernel_name_filter("^(flash|attention)$"), - CuptiNameFilter::Any - ); - } - - #[test] - fn decodes_escaped_regex_literals_conservatively() { - assert_eq!(regex_literal("kernel\\.v1"), Some("kernel.v1".to_owned())); - assert_eq!(regex_literal("kernel\\d"), None); - assert_eq!(regex_literal("kernel+"), None); - } - - #[test] - fn aggregate_hints_use_prefixes_for_incomplete_kernel_names() { - let group = CuptiAggregateGroup { - activity: CuptiAggregateActivity::Kernel, - name: Some("long_kernel_prefix".to_owned()), - name_complete: Some(false), - device_id: Some(0), - memcpy_kind: None, - count: 1, - total_duration_ns: 10, - min_duration_ns: 10, - max_duration_ns: 10, - total_bytes: None, - }; - - assert_eq!( - aggregate_selector_hints(&group), - ( - "cuda:kernel_start:name~^long_kernel_prefix".to_owned(), - "cuda:kernel_end:name~^long_kernel_prefix".to_owned(), - ) - ); - } -} diff --git a/xprobe/collector/src/cupti.rs b/xprobe/collector/src/cupti.rs index 2446a28..3deaa75 100644 --- a/xprobe/collector/src/cupti.rs +++ b/xprobe/collector/src/cupti.rs @@ -16,6 +16,10 @@ use xprobe_protocol::{ NvtxRangeKind, SchemaVersion, }; +mod filter; + +pub use filter::{CuptiFilterError, bounded_name_filter, event_filter}; + const OUTPUT_MAGIC: &[u8; 8] = b"XPCUPTI\0"; const CONTROL_MAGIC: &[u8; 8] = b"XPCTRL\0\0"; const CONTROL_VERSION: u32 = 4; diff --git a/xprobe/collector/src/cupti/filter.rs b/xprobe/collector/src/cupti/filter.rs new file mode 100644 index 0000000..16ff007 --- /dev/null +++ b/xprobe/collector/src/cupti/filter.rs @@ -0,0 +1,249 @@ +//! Lowering of validated CUDA selectors into fixed CUPTI Agent filters. + +use std::{error::Error, fmt}; + +use xprobe_protocol::{EventType, MemcpyKind, ResolvedCudaSelector}; + +use super::{ + CUPTI_NAME_CAPACITY, CuptiApiDomain, CuptiEventFilter, CuptiMemcpyKind, CuptiNameFilter, + CuptiRecordKind, +}; + +type NameLowering = fn(String) -> CuptiNameFilter; + +/// A validated CUDA selector that cannot be represented as an Agent filter. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CuptiFilterError { + NonCudaEventType(EventType), + UnsupportedApiDomain(String), +} + +impl fmt::Display for CuptiFilterError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::NonCudaEventType(event_type) => write!( + formatter, + "validated CUDA endpoint has a non-CUDA event type {event_type:?}" + ), + Self::UnsupportedApiDomain(domain) => write!( + formatter, + "validated CUDA endpoint has unsupported API domain {domain:?}" + ), + } + } +} + +impl Error for CuptiFilterError {} + +/// Lower one validated CUDA endpoint into the Agent's fixed filter layout. +/// +/// Name regexes that cannot be expressed as a bounded literal match lower to +/// [`CuptiNameFilter::Any`]; the correlator still applies the full regex. +/// +/// # Errors +/// +/// Returns [`CuptiFilterError`] when the selector is not a CUDA event type or +/// names an API domain the Agent does not support. +pub fn event_filter(selector: &ResolvedCudaSelector) -> Result { + let record_kind = match selector.event_type { + EventType::CudaApiEntry => CuptiRecordKind::CudaApiEntry, + EventType::CudaApiExit => CuptiRecordKind::CudaApiExit, + EventType::GpuKernelStart => CuptiRecordKind::GpuKernelStart, + EventType::GpuKernelEnd => CuptiRecordKind::GpuKernelEnd, + EventType::GpuMemcpyStart => CuptiRecordKind::GpuMemcpyStart, + EventType::GpuMemcpyEnd => CuptiRecordKind::GpuMemcpyEnd, + EventType::GpuMemsetStart => CuptiRecordKind::GpuMemsetStart, + EventType::GpuMemsetEnd => CuptiRecordKind::GpuMemsetEnd, + EventType::NvtxRangeStart => CuptiRecordKind::NvtxRangeStart, + EventType::NvtxRangeEnd => CuptiRecordKind::NvtxRangeEnd, + ref other => return Err(CuptiFilterError::NonCudaEventType(other.clone())), + }; + let api_domain = match selector.api_domain.as_deref() { + None => CuptiApiDomain::Any, + Some("driver_api") => CuptiApiDomain::Driver, + Some("runtime_api") => CuptiApiDomain::Runtime, + Some(domain) => return Err(CuptiFilterError::UnsupportedApiDomain(domain.to_owned())), + }; + let memcpy_kind = match selector.memcpy_kind { + None | Some(MemcpyKind::Unknown) => CuptiMemcpyKind::Any, + Some(MemcpyKind::HostToDevice) => CuptiMemcpyKind::HostToDevice, + Some(MemcpyKind::DeviceToHost) => CuptiMemcpyKind::DeviceToHost, + Some(MemcpyKind::DeviceToDevice) => CuptiMemcpyKind::DeviceToDevice, + Some(MemcpyKind::HostToHost) => CuptiMemcpyKind::HostToHost, + Some(MemcpyKind::PeerToPeer) => CuptiMemcpyKind::PeerToPeer, + }; + let name = if let Some(name) = selector.api_name.as_ref() { + CuptiNameFilter::Exact(name.clone()) + } else if let Some(pattern) = selector + .kernel_name_regex + .as_deref() + .or(selector.nvtx_name_regex.as_deref()) + { + bounded_name_filter(pattern) + } else { + CuptiNameFilter::Any + }; + Ok(CuptiEventFilter { + record_kind, + api_domain, + memcpy_kind, + name, + }) +} + +/// Lower a name regex to an equivalent bounded literal filter when possible. +/// +/// Returns [`CuptiNameFilter::Any`] for `.*` and for any pattern that is not +/// exactly an anchored or unanchored escaped literal fitting the Agent buffer. +#[must_use] +pub fn bounded_name_filter(pattern: &str) -> CuptiNameFilter { + if pattern == ".*" { + return CuptiNameFilter::Any; + } + let candidates: [(&str, &str, NameLowering); 8] = [ + ("^", "$", CuptiNameFilter::Exact), + ("^", ".*", CuptiNameFilter::Prefix), + (".*", "$", CuptiNameFilter::Suffix), + (".*", ".*", CuptiNameFilter::Contains), + ("^", "", CuptiNameFilter::Prefix), + ("", "$", CuptiNameFilter::Suffix), + (".*", "", CuptiNameFilter::Contains), + ("", "", CuptiNameFilter::Contains), + ]; + for (prefix, suffix, filter) in candidates { + let Some(literal) = pattern + .strip_prefix(prefix) + .and_then(|value| value.strip_suffix(suffix)) + .and_then(regex_literal) + else { + continue; + }; + if !literal.is_empty() && literal.len() < CUPTI_NAME_CAPACITY { + return filter(literal); + } + } + CuptiNameFilter::Any +} + +fn regex_literal(pattern: &str) -> Option { + let mut literal = String::with_capacity(pattern.len()); + let mut characters = pattern.chars(); + while let Some(character) = characters.next() { + if character == '\\' { + let escaped = characters.next()?; + if escaped.is_ascii_alphanumeric() { + return None; + } + literal.push(escaped); + } else if ".+*?()|[]{}^$".contains(character) { + return None; + } else { + literal.push(character); + } + } + Some(literal) +} + +#[cfg(test)] +mod tests { + use xprobe_protocol::{EventType, MemcpyKind, ResolvedCudaSelector}; + + use super::{ + CuptiApiDomain, CuptiFilterError, CuptiMemcpyKind, CuptiNameFilter, CuptiRecordKind, + bounded_name_filter, event_filter, regex_literal, + }; + + fn selector(event_type: EventType) -> ResolvedCudaSelector { + ResolvedCudaSelector { + event_type, + api_domain: None, + api_name: None, + kernel_name_regex: None, + nvtx_name_regex: None, + memcpy_kind: None, + } + } + + #[test] + fn lowers_only_equivalent_bounded_name_patterns() { + assert_eq!( + bounded_name_filter("^flash_.*"), + CuptiNameFilter::Prefix("flash_".to_owned()) + ); + assert_eq!( + bounded_name_filter(".*_kernel$"), + CuptiNameFilter::Suffix("_kernel".to_owned()) + ); + assert_eq!( + bounded_name_filter("^gemm$"), + CuptiNameFilter::Exact("gemm".to_owned()) + ); + assert_eq!( + bounded_name_filter("cudaLaunch"), + CuptiNameFilter::Contains("cudaLaunch".to_owned()) + ); + assert_eq!( + bounded_name_filter(".*xprobe_multisource_kernel.*"), + CuptiNameFilter::Contains("xprobe_multisource_kernel".to_owned()) + ); + assert_eq!( + bounded_name_filter("^(flash|attention)$"), + CuptiNameFilter::Any + ); + assert_eq!(bounded_name_filter(".*"), CuptiNameFilter::Any); + assert_eq!( + bounded_name_filter(&"k".repeat(super::CUPTI_NAME_CAPACITY)), + CuptiNameFilter::Any + ); + } + + #[test] + fn decodes_escaped_regex_literals_conservatively() { + assert_eq!(regex_literal("kernel\\.v1"), Some("kernel.v1".to_owned())); + assert_eq!(regex_literal("kernel\\d"), None); + assert_eq!(regex_literal("kernel+"), None); + } + + #[test] + fn lowers_cuda_selectors_to_agent_filters() { + let mut api = selector(EventType::CudaApiExit); + api.api_domain = Some("runtime_api".to_owned()); + api.api_name = Some("cudaLaunchKernel".to_owned()); + let filter = event_filter(&api).expect("runtime API selector lowers"); + assert_eq!(filter.record_kind, CuptiRecordKind::CudaApiExit); + assert_eq!(filter.api_domain, CuptiApiDomain::Runtime); + assert_eq!( + filter.name, + CuptiNameFilter::Exact("cudaLaunchKernel".to_owned()) + ); + + let mut memcpy = selector(EventType::GpuMemcpyEnd); + memcpy.memcpy_kind = Some(MemcpyKind::DeviceToHost); + let filter = event_filter(&memcpy).expect("memcpy selector lowers"); + assert_eq!(filter.memcpy_kind, CuptiMemcpyKind::DeviceToHost); + assert_eq!(filter.name, CuptiNameFilter::Any); + + let mut nvtx = selector(EventType::NvtxRangeStart); + nvtx.nvtx_name_regex = Some("^step$".to_owned()); + let filter = event_filter(&nvtx).expect("NVTX selector lowers"); + assert_eq!(filter.name, CuptiNameFilter::Exact("step".to_owned())); + } + + #[test] + fn rejects_selectors_the_agent_cannot_filter() { + assert_eq!( + event_filter(&selector(EventType::HostFunctionEntry)), + Err(CuptiFilterError::NonCudaEventType( + EventType::HostFunctionEntry + )) + ); + let mut unknown = selector(EventType::CudaApiEntry); + unknown.api_domain = Some("graph_api".to_owned()); + assert_eq!( + event_filter(&unknown), + Err(CuptiFilterError::UnsupportedApiDomain( + "graph_api".to_owned() + )) + ); + } +} diff --git a/xprobe/core/src/gpu_aggregate.rs b/xprobe/core/src/gpu_aggregate.rs new file mode 100644 index 0000000..caa40e5 --- /dev/null +++ b/xprobe/core/src/gpu_aggregate.rs @@ -0,0 +1,446 @@ +//! Deterministic normalization of bounded CUPTI aggregate captures. + +use std::{error::Error, fmt}; + +use xprobe_protocol::{ + AggregateActivity, AggregateCollectionSummary, AggregateDuration, AggregateGroup, + AggregateInventory, AggregateInventoryResult, CaptureCompleteness, ErrorCode, MemcpyKind, + SchemaVersion, SessionStatus, +}; + +/// One Agent aggregate table slot. Several slots may describe the same group. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RawGpuAggregateGroup { + pub activity: AggregateActivity, + pub name: Option, + pub name_complete: Option, + pub device_id: Option, + pub memcpy_kind: Option, + pub count: u64, + pub total_duration_ns: u64, + pub min_duration_ns: u64, + pub max_duration_ns: u64, + pub total_bytes: Option, +} + +/// A completed aggregate capture whose collector lifecycle was already checked. +#[derive(Debug, Clone)] +pub struct GpuAggregateCapture { + pub groups: Vec, + pub observed_activities: u64, + pub group_capacity: u64, +} + +/// Identity of the inventory request echoed into the result. +#[derive(Debug, Clone)] +pub struct GpuAggregateRequest { + pub session_id: String, + pub name: Option, + pub start_selector: String, + pub end_selector: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum GpuAggregateError { + CountOverflow, + DurationOverflow, + ByteOverflow, + InconsistentBytes, + InconsistentActivityCount { grouped: u64, observed: u64 }, + UnnamedKernel, + EmptyGroup, +} + +impl GpuAggregateError { + #[must_use] + pub const fn code(&self) -> ErrorCode { + ErrorCode::TraceExportFailed + } + + #[must_use] + pub const fn recoverable(&self) -> bool { + false + } +} + +impl fmt::Display for GpuAggregateError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::CountOverflow => formatter.write_str("aggregate activity count overflowed"), + Self::DurationOverflow => formatter.write_str("aggregate duration overflowed"), + Self::ByteOverflow => formatter.write_str("aggregate byte count overflowed"), + Self::InconsistentBytes => { + formatter.write_str("aggregate byte counters are inconsistent") + } + Self::InconsistentActivityCount { grouped, observed } => write!( + formatter, + "aggregate activity counters are inconsistent: grouped={grouped}, observed={observed}" + ), + Self::UnnamedKernel => formatter.write_str("aggregate kernel group has no name"), + Self::EmptyGroup => formatter.write_str("aggregate group has zero activities"), + } + } +} + +impl Error for GpuAggregateError {} + +/// Merge Agent aggregate slots into deterministic groups with selector hints. +/// +/// # Errors +/// +/// Returns [`GpuAggregateError`] when counters overflow, disagree with the +/// observed activity count, or describe a malformed group. +pub fn analyze( + request: GpuAggregateRequest, + capture: GpuAggregateCapture, +) -> Result { + let occupied_slots = + u64::try_from(capture.groups.len()).map_err(|_| GpuAggregateError::CountOverflow)?; + let groups = merge_groups(capture.groups)?; + let grouped_activities = groups.iter().try_fold(0_u64, |total, group| { + total + .checked_add(group.count) + .ok_or(GpuAggregateError::CountOverflow) + })?; + if grouped_activities != capture.observed_activities { + return Err(GpuAggregateError::InconsistentActivityCount { + grouped: grouped_activities, + observed: capture.observed_activities, + }); + } + let distinct_groups = + u64::try_from(groups.len()).map_err(|_| GpuAggregateError::CountOverflow)?; + let groups = groups + .into_iter() + .map(protocol_group) + .collect::, _>>()?; + Ok(AggregateInventoryResult { + schema_version: SchemaVersion::current(), + ok: true, + session_id: request.session_id, + status: SessionStatus::Completed, + inventory: AggregateInventory { + name: request.name, + start_selector: request.start_selector, + end_selector: request.end_selector, + groups, + }, + collection: AggregateCollectionSummary { + completeness: CaptureCompleteness::Complete, + observed_activities: capture.observed_activities, + grouped_activities, + dropped_activities: 0, + group_capacity: capture.group_capacity, + groups: distinct_groups, + occupied_slots, + table_utilization: utilization(occupied_slots, capture.group_capacity), + }, + warnings: Vec::new(), + }) +} + +fn merge_groups( + mut groups: Vec, +) -> Result, GpuAggregateError> { + groups.sort_by(|left, right| { + activity_rank(left.activity) + .cmp(&activity_rank(right.activity)) + .then_with(|| left.name.cmp(&right.name)) + .then_with(|| left.device_id.cmp(&right.device_id)) + .then_with(|| { + memcpy_kind_rank(left.memcpy_kind.as_ref()) + .cmp(&memcpy_kind_rank(right.memcpy_kind.as_ref())) + }) + }); + let mut merged: Vec = Vec::with_capacity(groups.len()); + for group in groups { + let Some(previous) = merged.last_mut() else { + merged.push(group); + continue; + }; + if previous.activity != group.activity + || previous.name != group.name + || previous.device_id != group.device_id + || previous.memcpy_kind != group.memcpy_kind + { + merged.push(group); + continue; + } + previous.count = previous + .count + .checked_add(group.count) + .ok_or(GpuAggregateError::CountOverflow)?; + previous.total_duration_ns = previous + .total_duration_ns + .checked_add(group.total_duration_ns) + .ok_or(GpuAggregateError::DurationOverflow)?; + previous.min_duration_ns = previous.min_duration_ns.min(group.min_duration_ns); + previous.max_duration_ns = previous.max_duration_ns.max(group.max_duration_ns); + previous.total_bytes = match (previous.total_bytes, group.total_bytes) { + (Some(left), Some(right)) => Some( + left.checked_add(right) + .ok_or(GpuAggregateError::ByteOverflow)?, + ), + (None, None) => None, + _ => return Err(GpuAggregateError::InconsistentBytes), + }; + } + Ok(merged) +} + +const fn activity_rank(activity: AggregateActivity) -> u8 { + match activity { + AggregateActivity::Kernel => 0, + AggregateActivity::Memcpy => 1, + AggregateActivity::Memset => 2, + } +} + +const fn memcpy_kind_rank(kind: Option<&MemcpyKind>) -> u8 { + match kind { + None => 0, + Some(&MemcpyKind::Unknown) => 1, + Some(&MemcpyKind::HostToDevice) => 2, + Some(&MemcpyKind::DeviceToHost) => 3, + Some(&MemcpyKind::DeviceToDevice) => 4, + Some(&MemcpyKind::HostToHost) => 5, + Some(&MemcpyKind::PeerToPeer) => 6, + } +} + +#[allow(clippy::cast_precision_loss)] +fn protocol_group(group: RawGpuAggregateGroup) -> Result { + if group.count == 0 { + return Err(GpuAggregateError::EmptyGroup); + } + let (start_selector_hint, end_selector_hint) = selector_hints(&group)?; + Ok(AggregateGroup { + activity: group.activity, + name: group.name, + name_complete: group.name_complete, + device_id: group.device_id, + memcpy_kind: group.memcpy_kind, + start_selector_hint, + end_selector_hint, + count: group.count, + duration_ns: AggregateDuration { + min: group.min_duration_ns, + mean: group.total_duration_ns as f64 / group.count as f64, + max: group.max_duration_ns, + total: group.total_duration_ns, + }, + total_bytes: group.total_bytes, + }) +} + +fn selector_hints(group: &RawGpuAggregateGroup) -> Result<(String, String), GpuAggregateError> { + Ok(match group.activity { + AggregateActivity::Kernel => { + let name = group + .name + .as_deref() + .ok_or(GpuAggregateError::UnnamedKernel)?; + let escaped = escape_selector_regex(name); + let suffix = if group.name_complete == Some(true) { + "$" + } else { + "" + }; + ( + format!("cuda:kernel_start:name~^{escaped}{suffix}"), + format!("cuda:kernel_end:name~^{escaped}{suffix}"), + ) + } + AggregateActivity::Memcpy => { + let kind = match group.memcpy_kind.as_ref() { + Some(MemcpyKind::HostToDevice) => Some("HtoD"), + Some(MemcpyKind::DeviceToHost) => Some("DtoH"), + Some(MemcpyKind::DeviceToDevice) => Some("DtoD"), + Some(MemcpyKind::HostToHost) => Some("HtoH"), + Some(MemcpyKind::PeerToPeer) => Some("PtoP"), + Some(MemcpyKind::Unknown) | None => None, + }; + kind.map_or_else( + || ("cuda:memcpy_start".to_owned(), "cuda:memcpy_end".to_owned()), + |kind| { + ( + format!("cuda:memcpy_start:kind={kind}"), + format!("cuda:memcpy_end:kind={kind}"), + ) + }, + ) + } + AggregateActivity::Memset => ("cuda:memset_start".to_owned(), "cuda:memset_end".to_owned()), + }) +} + +fn escape_selector_regex(name: &str) -> String { + let mut escaped = String::with_capacity(name.len()); + for character in name.chars() { + if "\\.^$|?*+()[]{}".contains(character) { + escaped.push('\\'); + } + escaped.push(character); + } + escaped +} + +#[allow(clippy::cast_precision_loss)] +fn utilization(occupied: u64, capacity: u64) -> f64 { + if capacity == 0 { + 0.0 + } else { + occupied as f64 / capacity as f64 + } +} + +#[cfg(test)] +mod tests { + use xprobe_protocol::{AggregateActivity, MemcpyKind}; + + use super::{ + GpuAggregateCapture, GpuAggregateError, GpuAggregateRequest, RawGpuAggregateGroup, analyze, + selector_hints, + }; + + fn kernel(name: &str, complete: bool, count: u64, total: u64) -> RawGpuAggregateGroup { + RawGpuAggregateGroup { + activity: AggregateActivity::Kernel, + name: Some(name.to_owned()), + name_complete: Some(complete), + device_id: Some(0), + memcpy_kind: None, + count, + total_duration_ns: total, + min_duration_ns: total / count, + max_duration_ns: total / count, + total_bytes: None, + } + } + + fn memcpy(kind: MemcpyKind, count: u64, bytes: Option) -> RawGpuAggregateGroup { + RawGpuAggregateGroup { + activity: AggregateActivity::Memcpy, + name: None, + name_complete: None, + device_id: Some(0), + memcpy_kind: Some(kind), + count, + total_duration_ns: 10 * count, + min_duration_ns: 10, + max_duration_ns: 10, + total_bytes: bytes, + } + } + + fn request() -> GpuAggregateRequest { + GpuAggregateRequest { + session_id: "s".to_owned(), + name: None, + start_selector: "cuda:kernel_start".to_owned(), + end_selector: "cuda:kernel_end".to_owned(), + } + } + + #[test] + fn merges_duplicate_slots_deterministically() { + let capture = GpuAggregateCapture { + groups: vec![ + memcpy(MemcpyKind::HostToDevice, 1, Some(64)), + kernel("b", true, 2, 40), + kernel("a", true, 1, 10), + kernel("b", true, 1, 50), + ], + observed_activities: 5, + group_capacity: 8, + }; + let result = analyze(request(), capture).expect("consistent capture"); + let groups = &result.inventory.groups; + assert_eq!(groups.len(), 3); + assert_eq!(groups[0].name.as_deref(), Some("a")); + assert_eq!(groups[1].name.as_deref(), Some("b")); + assert_eq!(groups[1].count, 3); + assert_eq!(groups[1].duration_ns.total, 90); + assert_eq!(groups[1].duration_ns.min, 20); + assert_eq!(groups[1].duration_ns.max, 50); + assert_eq!(groups[2].activity, AggregateActivity::Memcpy); + assert_eq!(groups[2].start_selector_hint, "cuda:memcpy_start:kind=HtoD"); + assert_eq!(result.collection.groups, 3); + assert_eq!(result.collection.occupied_slots, 4); + assert!((result.collection.table_utilization - 0.5).abs() < f64::EPSILON); + } + + #[test] + fn rejects_inconsistent_counters() { + let capture = GpuAggregateCapture { + groups: vec![kernel("a", true, 2, 20)], + observed_activities: 3, + group_capacity: 8, + }; + assert_eq!( + analyze(request(), capture), + Err(GpuAggregateError::InconsistentActivityCount { + grouped: 2, + observed: 3 + }) + ); + + let capture = GpuAggregateCapture { + groups: vec![ + memcpy(MemcpyKind::DeviceToHost, 1, Some(8)), + memcpy(MemcpyKind::DeviceToHost, 1, None), + ], + observed_activities: 2, + group_capacity: 8, + }; + assert_eq!( + analyze(request(), capture), + Err(GpuAggregateError::InconsistentBytes) + ); + } + + #[test] + fn rejects_malformed_groups_instead_of_panicking() { + let mut unnamed = kernel("a", true, 1, 10); + unnamed.name = None; + let capture = GpuAggregateCapture { + groups: vec![unnamed], + observed_activities: 1, + group_capacity: 8, + }; + assert_eq!( + analyze(request(), capture), + Err(GpuAggregateError::UnnamedKernel) + ); + + let mut empty = kernel("a", true, 1, 10); + empty.count = 0; + let capture = GpuAggregateCapture { + groups: vec![empty], + observed_activities: 0, + group_capacity: 8, + }; + assert_eq!( + analyze(request(), capture), + Err(GpuAggregateError::EmptyGroup) + ); + } + + #[test] + fn hints_escape_names_and_use_prefixes_for_incomplete_names() { + assert_eq!( + selector_hints(&kernel("long_kernel_prefix", false, 1, 10)), + Ok(( + "cuda:kernel_start:name~^long_kernel_prefix".to_owned(), + "cuda:kernel_end:name~^long_kernel_prefix".to_owned(), + )) + ); + assert_eq!( + selector_hints(&kernel("void k<1>(float*)", true, 1, 10)), + Ok(( + "cuda:kernel_start:name~^void k<1>\\(float\\*\\)$".to_owned(), + "cuda:kernel_end:name~^void k<1>\\(float\\*\\)$".to_owned(), + )) + ); + } +} diff --git a/xprobe/core/src/lib.rs b/xprobe/core/src/lib.rs index e8fb8a7..1f6520c 100644 --- a/xprobe/core/src/lib.rs +++ b/xprobe/core/src/lib.rs @@ -5,6 +5,7 @@ pub mod cpu_sampling_analysis; pub mod cupti_compat; pub mod discover; pub mod doctor; +pub mod gpu_aggregate; pub mod inject; pub mod inspect; pub mod resolve; From 85c649e758f84bc6d63a68ca9b24b82951fc5526 Mon Sep 17 00:00:00 2001 From: itdevwu Date: Mon, 9 Nov 2026 18:03:12 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=F0=9F=94=96=20chore:=20prepare=20version?= =?UTF-8?q?=200.5.2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CMakeLists.txt | 2 +- Cargo.lock | 12 ++++++------ Cargo.toml | 2 +- README.md | 2 +- docs/agent-integration.md | 2 +- docs/installation.md | 12 ++++++------ install.sh | 4 ++-- skills/xprobe-measure-latency/references/setup.md | 8 ++++---- tests/agent-contract/test_contract.py | 2 +- 9 files changed, 23 insertions(+), 23 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 4873901..397723b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,6 +1,6 @@ cmake_minimum_required(VERSION 3.20) -project(xprobe VERSION 0.5.1 LANGUAGES C) +project(xprobe VERSION 0.5.2 LANGUAGES C) include(CTest) diff --git a/Cargo.lock b/Cargo.lock index 128e728..5dd424d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -732,7 +732,7 @@ dependencies = [ [[package]] name = "xprobe-cli" -version = "0.5.1" +version = "0.5.2" dependencies = [ "clap", "serde_json", @@ -745,7 +745,7 @@ dependencies = [ [[package]] name = "xprobe-collector" -version = "0.5.1" +version = "0.5.2" dependencies = [ "libbpf-rs", "perf-event-open", @@ -755,7 +755,7 @@ dependencies = [ [[package]] name = "xprobe-core" -version = "0.5.1" +version = "0.5.2" dependencies = [ "cpp_demangle", "nix", @@ -766,7 +766,7 @@ dependencies = [ [[package]] name = "xprobe-correlator" -version = "0.5.1" +version = "0.5.2" dependencies = [ "regex", "serde_json", @@ -775,7 +775,7 @@ dependencies = [ [[package]] name = "xprobe-exporter" -version = "0.5.1" +version = "0.5.2" dependencies = [ "serde_json", "xprobe-protocol", @@ -783,7 +783,7 @@ dependencies = [ [[package]] name = "xprobe-protocol" -version = "0.5.1" +version = "0.5.2" dependencies = [ "schemars", "serde", diff --git a/Cargo.toml b/Cargo.toml index f5a2558..e6fd5d7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ members = [ resolver = "2" [workspace.package] -version = "0.5.1" +version = "0.5.2" edition = "2024" license = "Apache-2.0" repository = "https://github.com/itdevwu/xprobe" diff --git a/README.md b/README.md index 81cb6d1..fe7f8eb 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ only installation action required from the user: ```bash npx skills@1 add \ - https://github.com/itdevwu/xprobe/tree/v0.5.1/skills/xprobe-measure-latency \ + https://github.com/itdevwu/xprobe/tree/v0.5.2/skills/xprobe-measure-latency \ --global ``` diff --git a/docs/agent-integration.md b/docs/agent-integration.md index 6184b8c..598a413 100644 --- a/docs/agent-integration.md +++ b/docs/agent-integration.md @@ -17,7 +17,7 @@ repairs the matching xprobe CLI itself: ```bash npx skills@1 add \ - https://github.com/itdevwu/xprobe/tree/v0.5.1/skills/xprobe-measure-latency \ + https://github.com/itdevwu/xprobe/tree/v0.5.2/skills/xprobe-measure-latency \ --global ``` diff --git a/docs/installation.md b/docs/installation.md index 66e3f31..8d118b5 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -13,7 +13,7 @@ user needs to run: ```bash npx skills@1 add \ - https://github.com/itdevwu/xprobe/tree/v0.5.1/skills/xprobe-measure-latency \ + https://github.com/itdevwu/xprobe/tree/v0.5.2/skills/xprobe-measure-latency \ --global ``` @@ -29,7 +29,7 @@ The versioned bootstrap installs to `~/.local` without root access: ```bash curl --proto '=https' --tlsv1.2 -fsSL \ - https://raw.githubusercontent.com/itdevwu/xprobe/v0.5.1/install.sh | sh + https://raw.githubusercontent.com/itdevwu/xprobe/v0.5.2/install.sh | sh ``` The bootstrap downloads the release archive and its SHA256 file, verifies the @@ -48,7 +48,7 @@ prefix, download the script and pass `--prefix`: ```bash curl --proto '=https' --tlsv1.2 -fsSLO \ - https://raw.githubusercontent.com/itdevwu/xprobe/v0.5.1/install.sh + https://raw.githubusercontent.com/itdevwu/xprobe/v0.5.2/install.sh sh install.sh --prefix /opt/xprobe ``` @@ -60,7 +60,7 @@ script with `sudo`. The installer never elevates privileges itself. For a fully explicit archive workflow: ```bash -version=0.5.1 +version=0.5.2 base=https://github.com/itdevwu/xprobe/releases/download/v$version archive=xprobe-$version-linux-x86_64.tar.gz @@ -106,7 +106,7 @@ missing or damaged, manually install the complete version-matched directory: ```bash npx skills@1 add \ - https://github.com/itdevwu/xprobe/tree/v0.5.1/skills/xprobe-measure-latency \ + https://github.com/itdevwu/xprobe/tree/v0.5.2/skills/xprobe-measure-latency \ --global ``` @@ -115,7 +115,7 @@ installation names the target explicitly: ```bash npx --yes skills@1 add \ - https://github.com/itdevwu/xprobe/tree/v0.5.1/skills/xprobe-measure-latency \ + https://github.com/itdevwu/xprobe/tree/v0.5.2/skills/xprobe-measure-latency \ --agent codex --global --copy --yes ``` diff --git a/install.sh b/install.sh index 3fe0d1c..0d6314e 100755 --- a/install.sh +++ b/install.sh @@ -2,7 +2,7 @@ set -eu repository=${XPROBE_REPOSITORY:-itdevwu/xprobe} -version=${XPROBE_VERSION:-0.5.1} +version=${XPROBE_VERSION:-0.5.2} if [ -n "${XPROBE_PREFIX:-}" ]; then prefix=$XPROBE_PREFIX elif [ -n "${HOME:-}" ]; then @@ -19,7 +19,7 @@ Install a released xprobe binary and its CUDA Agents. Usage: install.sh [--version VERSION] [--prefix DIR] [--uninstall] Options: - --version VERSION Release to install (default: 0.5.1) + --version VERSION Release to install (default: 0.5.2) --prefix DIR Installation prefix (default: $HOME/.local) --uninstall Remove xprobe from the selected prefix -h, --help Show this help diff --git a/skills/xprobe-measure-latency/references/setup.md b/skills/xprobe-measure-latency/references/setup.md index 31348a4..b0ee4ae 100644 --- a/skills/xprobe-measure-latency/references/setup.md +++ b/skills/xprobe-measure-latency/references/setup.md @@ -23,7 +23,7 @@ installing under `~/.local`: ```bash curl --proto '=https' --tlsv1.2 -fsSL \ - https://raw.githubusercontent.com/itdevwu/xprobe/v0.5.1/install.sh \ + https://raw.githubusercontent.com/itdevwu/xprobe/v0.5.2/install.sh \ -o /tmp/xprobe-install.sh sh /tmp/xprobe-install.sh export PATH="$HOME/.local/bin:$PATH" @@ -69,7 +69,7 @@ host glibc instead. This is a local-use fallback, not permission to weaken the release package's `GLIBC_2.34` ceiling. ```bash -git clone --depth 1 --branch v0.5.1 https://github.com/itdevwu/xprobe.git +git clone --depth 1 --branch v0.5.2 https://github.com/itdevwu/xprobe.git cd xprobe mamba env create --file environment.yml mamba run -n xprobe-dev just build @@ -112,7 +112,7 @@ release directory through the Agent Skills CLI: ```bash npx skills@1 add \ - https://github.com/itdevwu/xprobe/tree/v0.5.1/skills/xprobe-measure-latency \ + https://github.com/itdevwu/xprobe/tree/v0.5.2/skills/xprobe-measure-latency \ --global ``` @@ -120,7 +120,7 @@ For non-interactive automation, select the host explicitly: ```bash npx --yes skills@1 add \ - https://github.com/itdevwu/xprobe/tree/v0.5.1/skills/xprobe-measure-latency \ + https://github.com/itdevwu/xprobe/tree/v0.5.2/skills/xprobe-measure-latency \ --agent codex --global --copy --yes ``` diff --git a/tests/agent-contract/test_contract.py b/tests/agent-contract/test_contract.py index c5933bb..b7f1232 100755 --- a/tests/agent-contract/test_contract.py +++ b/tests/agent-contract/test_contract.py @@ -224,7 +224,7 @@ def check_skill(workspace: pathlib.Path) -> None: ): assert required in normalized_cli_contract for required in ( - "v0.5.1/install.sh", + "v0.5.2/install.sh", "xprobe `0.5.x`", "npx skills@1 add", "xprobe --version", From 90d05cb39f3c0c1bbaba01fc0b11105bc3474d76 Mon Sep 17 00:00:00 2001 From: itdevwu Date: Mon, 9 Nov 2026 20:16:02 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=F0=9F=A7=AA=20test:=20show=20values=20in?= =?UTF-8?q?=20emptiness=20assertions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rust 1.99 clippy adds `assert_is_empty`, which rejects `assert!(x.is_empty())` under `-D warnings`. Rewrite the affected test assertions as `assert_eq!`/`assert_ne!` so failures print the offending value; CLI stderr checks now print the decoded text. --- xprobe/cli/tests/cpu_sampling.rs | 5 ++++- xprobe/cli/tests/cupti.rs | 4 ++-- xprobe/cli/tests/discover.rs | 2 +- xprobe/cli/tests/doctor.rs | 2 +- xprobe/cli/tests/inspect.rs | 8 ++++---- xprobe/cli/tests/measure.rs | 12 ++++++------ xprobe/cli/tests/resolve.rs | 2 +- xprobe/cli/tests/uprobe.rs | 2 +- xprobe/cli/tests/validate.rs | 9 ++++++--- xprobe/collector/src/cupti.rs | 2 +- xprobe/collector/src/linux.rs | 2 +- xprobe/core/src/inspect.rs | 2 +- xprobe/core/src/validate.rs | 15 ++++++++++++--- xprobe/correlator/src/lib.rs | 2 +- 14 files changed, 42 insertions(+), 27 deletions(-) diff --git a/xprobe/cli/tests/cpu_sampling.rs b/xprobe/cli/tests/cpu_sampling.rs index 0140b03..0c217d0 100644 --- a/xprobe/cli/tests/cpu_sampling.rs +++ b/xprobe/cli/tests/cpu_sampling.rs @@ -44,7 +44,10 @@ fn samples_and_symbolizes_a_busy_native_process() { let result: CpuSampleInventoryResult = serde_json::from_slice(&output.stdout).expect("stdout must contain CPU inventory JSON"); assert!(result.collection.observed_samples > 0); - assert!(!result.inventory.stack_groups.is_empty()); + assert_ne!( + result.inventory.stack_groups, + Vec::::new() + ); assert!( result.inventory.hotspots.iter().any(|hotspot| { hotspot.frame.language == CpuFrameLanguage::Native && hotspot.frame.symbol.is_some() diff --git a/xprobe/cli/tests/cupti.rs b/xprobe/cli/tests/cupti.rs index f3f3d01..14a62b0 100644 --- a/xprobe/cli/tests/cupti.rs +++ b/xprobe/cli/tests/cupti.rs @@ -63,7 +63,7 @@ fn cupti_capture_is_emitted_as_event_jsonl() { "{}", String::from_utf8_lossy(&output.stderr) ); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let stdout = String::from_utf8(output.stdout).expect("JSONL must be UTF-8"); let lines: Vec<_> = stdout.lines().collect(); assert_eq!(lines.len(), 1); @@ -96,7 +96,7 @@ fn malformed_cupti_capture_returns_structured_error() { fs::remove_file(path).expect("capture fixture must be removed"); assert_eq!(output.status.code(), Some(1)); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let response: ErrorResponse = serde_json::from_slice(&output.stdout).expect("error must be JSON"); assert_eq!(response.error.code, ErrorCode::TraceExportFailed); diff --git a/xprobe/cli/tests/discover.rs b/xprobe/cli/tests/discover.rs index 64c425d..3fe55d1 100644 --- a/xprobe/cli/tests/discover.rs +++ b/xprobe/cli/tests/discover.rs @@ -61,7 +61,7 @@ fn discovers_only_nvml_confirmed_cuda_workers() { "{}", String::from_utf8_lossy(&output.stderr) ); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let result: DiscoveryResult = serde_json::from_slice(&output.stdout).unwrap(); assert_eq!(result.root.pid, pid); assert_eq!(result.total_candidates, 1); diff --git a/xprobe/cli/tests/doctor.rs b/xprobe/cli/tests/doctor.rs index 4b9424e..3a77499 100644 --- a/xprobe/cli/tests/doctor.rs +++ b/xprobe/cli/tests/doctor.rs @@ -14,7 +14,7 @@ fn doctor_json_is_machine_readable() { "{}", String::from_utf8_lossy(&output.stderr) ); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let report: CapabilityReport = serde_json::from_slice(&output.stdout).expect("stdout must contain only capability JSON"); diff --git a/xprobe/cli/tests/inspect.rs b/xprobe/cli/tests/inspect.rs index cf6c4d4..2f056d6 100644 --- a/xprobe/cli/tests/inspect.rs +++ b/xprobe/cli/tests/inspect.rs @@ -22,14 +22,14 @@ fn inspect_json_describes_the_target_process() { "{}", String::from_utf8_lossy(&output.stderr) ); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let report: ProcessReport = serde_json::from_slice(&output.stdout).expect("stdout must contain only process JSON"); assert_eq!(report.schema_version, SchemaVersion::current()); assert_eq!(report.target.pid, pid); - assert!(!report.executable.is_empty()); - assert!(!report.namespace_pids.is_empty()); + assert_ne!(report.executable, ""); + assert_ne!(report.namespace_pids, Vec::::new()); } #[test] @@ -40,7 +40,7 @@ fn inspect_missing_target_returns_a_structured_error() { .expect("xprobe inspect must run"); assert_eq!(output.status.code(), Some(3)); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let response: ErrorResponse = serde_json::from_slice(&output.stdout).expect("stdout must contain only error JSON"); diff --git a/xprobe/cli/tests/measure.rs b/xprobe/cli/tests/measure.rs index 511f807..ba23b10 100644 --- a/xprobe/cli/tests/measure.rs +++ b/xprobe/cli/tests/measure.rs @@ -166,7 +166,7 @@ fn measures_exact_kernel_durations_from_a_completed_capture() { "{}", String::from_utf8_lossy(&output.stderr) ); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let result: MeasurementResult = serde_json::from_slice(&output.stdout).expect("stdout must contain measurement JSON"); assert_eq!(result.status, SessionStatus::Completed); @@ -245,7 +245,7 @@ fn rejects_a_capture_that_reached_the_agent_record_limit() { ); assert_eq!(response.error.details["record_capacity"], 2); assert_eq!(response.error.details["observed_records"], 3); - assert!(!response.error.hints.is_empty()); + assert_ne!(response.error.hints, Vec::::new()); } #[test] @@ -286,7 +286,7 @@ fn rejects_an_active_capture_as_incomplete() { let response: ErrorResponse = serde_json::from_slice(&output.stdout).unwrap(); assert_eq!(response.error.code, ErrorCode::CuptiNotAvailable); assert_eq!(response.error.details["capture_state"], "active"); - assert!(!response.error.hints.is_empty()); + assert_ne!(response.error.hints, Vec::::new()); } #[test] @@ -334,7 +334,7 @@ fn rejects_unaligned_api_to_kernel_latency() { evidence_path.to_string_lossy().as_ref() ); assert_eq!(error.error.details["artifact_event_count"], 2); - assert!(!error.error.hints.is_empty()); + assert_ne!(error.error.hints, Vec::::new()); let evidence = fs::read_to_string(&evidence_path).expect("failure evidence must be written"); fs::remove_file(evidence_path).expect("failure evidence must be removed"); assert_eq!(evidence.lines().count(), 2); @@ -426,7 +426,7 @@ fn reports_artifact_failure_without_hiding_measurement_failure() { error.error.details["original_error_code"], "NO_MATCHED_SAMPLES" ); - assert!(!error.error.hints.is_empty()); + assert_ne!(error.error.hints, Vec::::new()); } #[test] @@ -506,7 +506,7 @@ fn measures_host_to_kernel_latency_from_merged_captures() { assert_eq!(result.error.code, ErrorCode::EventsDropped); assert!(result.error.message.contains("dropped 2 events")); assert_eq!(result.error.details["dropped_events"], 2); - assert!(!result.error.hints.is_empty()); + assert_ne!(result.error.hints, Vec::::new()); } #[test] diff --git a/xprobe/cli/tests/resolve.rs b/xprobe/cli/tests/resolve.rs index 7717840..56cf884 100644 --- a/xprobe/cli/tests/resolve.rs +++ b/xprobe/cli/tests/resolve.rs @@ -120,7 +120,7 @@ fn resolve(pid: u32, selector: &str) -> ResolvedProbe { "{}", String::from_utf8_lossy(&output.stdout) ); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); serde_json::from_slice(&output.stdout).expect("stdout must contain resolved probe JSON") } diff --git a/xprobe/cli/tests/uprobe.rs b/xprobe/cli/tests/uprobe.rs index 165566d..7c566f2 100644 --- a/xprobe/cli/tests/uprobe.rs +++ b/xprobe/cli/tests/uprobe.rs @@ -99,7 +99,7 @@ fn uprobe_jsonl_failure_is_machine_readable() { .expect("xprobe dev uprobe must run"); assert_eq!(output.status.code(), Some(1)); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let response: ErrorResponse = serde_json::from_slice(&output.stdout).expect("stdout must contain only error JSON"); assert_eq!(response.error.code, ErrorCode::InvalidEventSelector); diff --git a/xprobe/cli/tests/validate.rs b/xprobe/cli/tests/validate.rs index b2ec2ab..5be41ab 100644 --- a/xprobe/cli/tests/validate.rs +++ b/xprobe/cli/tests/validate.rs @@ -30,7 +30,7 @@ fn validate_reports_environment_requirements_without_attaching() { "{}", String::from_utf8_lossy(&output.stderr) ); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let result: ValidationResult = serde_json::from_slice(&output.stdout).expect("stdout must contain validation JSON"); assert_eq!(result.match_policy, MatchPolicy::Exact); @@ -50,7 +50,10 @@ fn validate_reports_environment_requirements_without_attaching() { AgentActivation::InjectionRequired ); assert!(result.requirements.target_mutation); - assert!(result.issues.is_empty()); + assert_eq!( + result.issues, + Vec::::new() + ); assert!( result .warnings @@ -103,7 +106,7 @@ fn validate_accepts_pid_scoped_cpu_sampling_without_ebpf() { "{}", String::from_utf8_lossy(&output.stderr) ); - assert!(output.stderr.is_empty()); + assert_eq!(String::from_utf8_lossy(&output.stderr), ""); let result: CpuSamplingValidationResult = serde_json::from_slice(&output.stdout).expect("stdout must contain validation JSON"); assert_eq!(result.valid, result.issues.is_empty()); diff --git a/xprobe/collector/src/cupti.rs b/xprobe/collector/src/cupti.rs index 3deaa75..6ba708f 100644 --- a/xprobe/collector/src/cupti.rs +++ b/xprobe/collector/src/cupti.rs @@ -1349,7 +1349,7 @@ mod tests { let decoded = decode_capture(&bytes, "unused").expect("aggregate capture must decode"); - assert!(decoded.events.is_empty()); + assert_eq!(decoded.events, Vec::::new()); assert_eq!(decoded.aggregate_groups.len(), 2); assert_eq!( decoded.aggregate_groups[0].activity, diff --git a/xprobe/collector/src/linux.rs b/xprobe/collector/src/linux.rs index b6197d9..af66a38 100644 --- a/xprobe/collector/src/linux.rs +++ b/xprobe/collector/src/linux.rs @@ -740,7 +740,7 @@ mod tests { assert_eq!(host.probe_kind, HostProbeKind::Usdt); assert_eq!(host.binary_path.as_deref(), Some("/usr/bin/python3")); assert_eq!(host.symbol.as_deref(), Some("gc__start")); - assert!(host.arguments.is_empty()); + assert_eq!(host.arguments, Vec::::new()); assert_eq!(event.attributes["usdt_provider"], "python"); assert!(validate_request(&request).is_ok()); } diff --git a/xprobe/core/src/inspect.rs b/xprobe/core/src/inspect.rs index 0ee7e89..6b599b1 100644 --- a/xprobe/core/src/inspect.rs +++ b/xprobe/core/src/inspect.rs @@ -406,7 +406,7 @@ mod tests { fn parses_cgroup_v2_entry() { let entries = parse_cgroups("0::/user.slice\n", 1).unwrap(); assert_eq!(entries[0].hierarchy_id, 0); - assert!(entries[0].controllers.is_empty()); + assert_eq!(entries[0].controllers, Vec::::new()); assert_eq!(entries[0].path, "/user.slice"); } } diff --git a/xprobe/core/src/validate.rs b/xprobe/core/src/validate.rs index 69c4299..93b4d7f 100644 --- a/xprobe/core/src/validate.rs +++ b/xprobe/core/src/validate.rs @@ -1219,7 +1219,10 @@ mod tests { AgentActivation::InjectionRequired ); assert!(result.requirements.target_mutation); - assert!(result.issues.is_empty()); + assert_eq!( + result.issues, + Vec::::new() + ); assert!( result .warnings @@ -1248,7 +1251,10 @@ mod tests { ); assert!(!result.requirements.target_mutation); assert!(!result.requirements.needs_clock_alignment); - assert!(result.issues.is_empty()); + assert_eq!( + result.issues, + Vec::::new() + ); } #[test] @@ -1355,6 +1361,9 @@ mod tests { .unwrap(); assert!(result.valid); assert!(result.requirements.needs_clock_alignment); - assert!(result.issues.is_empty()); + assert_eq!( + result.issues, + Vec::::new() + ); } } diff --git a/xprobe/correlator/src/lib.rs b/xprobe/correlator/src/lib.rs index f5a5ab1..418038a 100644 --- a/xprobe/correlator/src/lib.rs +++ b/xprobe/correlator/src/lib.rs @@ -1908,7 +1908,7 @@ mod tests { let result = measure(&events, &options(MatchPolicy::Exact)).unwrap(); assert_eq!(result.clock.alignment, "cupti_normalized_to_host_monotonic"); assert_eq!(result.clock.estimated_error_ns, Some(7)); - assert!(result.warnings.is_empty()); + assert_eq!(result.warnings, Vec::::new()); } #[test]