1use std::fs;
4use std::path::{Path, PathBuf};
5
6use serde::{Deserialize, Serialize};
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
9#[serde(rename_all = "snake_case")]
10pub enum GpuEvidenceStatus {
11 Available,
12 Unavailable,
13 Failed,
14}
15
16#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
17pub struct NsightSummaryRow {
18 pub name: String,
19 pub total_ns: u64,
20 pub count: u64,
21 pub average_ns: u64,
22 pub minimum_ns: u64,
23 pub maximum_ns: u64,
24 #[serde(default, skip_serializing_if = "Option::is_none")]
25 pub category: Option<String>,
26}
27
28#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
29pub struct NsightTimelineRow {
30 pub name: String,
31 pub kind: String,
32 #[serde(default, skip_serializing_if = "Option::is_none")]
33 pub device: Option<String>,
34 #[serde(default, skip_serializing_if = "Option::is_none")]
35 pub context: Option<String>,
36 #[serde(default, skip_serializing_if = "Option::is_none")]
37 pub stream: Option<String>,
38 #[serde(default, skip_serializing_if = "Option::is_none")]
39 pub correlation_id: Option<String>,
40 #[serde(default, skip_serializing_if = "Option::is_none")]
41 pub semantic_key: Option<String>,
42 pub start_ns: u64,
43 pub duration_ns: u64,
44 #[serde(default, skip_serializing_if = "Option::is_none")]
45 pub projected_start_ns: Option<u64>,
46 #[serde(default, skip_serializing_if = "Option::is_none")]
47 pub projected_duration_ns: Option<u64>,
48 #[serde(default, skip_serializing_if = "Option::is_none")]
49 pub gpu_operations: Option<u64>,
50}
51
52#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
53pub struct NsightCoverage {
54 pub kernel_summary: bool,
55 pub runtime_summary: bool,
56 pub memory_summary: bool,
57 pub nvtx_projection: bool,
58 pub gpu_timeline: bool,
59}
60
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62pub struct NsightCorrelation {
63 pub mode: String,
64 pub clock_aligned: bool,
65 pub complete: bool,
66 pub matched_ranges: usize,
67 pub total_ranges: usize,
68 #[serde(default, skip_serializing_if = "Option::is_none")]
69 pub reason: Option<String>,
70}
71
72#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
73pub struct ReportLimit {
74 pub total_rows: usize,
75 pub displayed_rows: usize,
76 pub truncated: bool,
77}
78
79#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
80pub struct NsightEvidence {
81 pub status: GpuEvidenceStatus,
82 #[serde(default, skip_serializing_if = "Option::is_none")]
83 pub reason: Option<String>,
84 #[serde(default, skip_serializing_if = "Option::is_none")]
85 pub raw_report: Option<String>,
86 #[serde(default)]
87 pub source_csv: Vec<String>,
88 pub coverage: NsightCoverage,
89 pub correlation: NsightCorrelation,
90 #[serde(default)]
91 pub diagnostics: Vec<String>,
92 #[serde(default)]
93 pub limits: std::collections::BTreeMap<String, ReportLimit>,
94 #[serde(default)]
95 pub kernels: Vec<NsightSummaryRow>,
96 #[serde(default)]
97 pub runtime_calls: Vec<NsightSummaryRow>,
98 #[serde(default)]
99 pub memory_operations: Vec<NsightSummaryRow>,
100 #[serde(default)]
101 pub nvtx_ranges: Vec<NsightTimelineRow>,
102 #[serde(default)]
103 pub gpu_timeline: Vec<NsightTimelineRow>,
104}
105
106impl NsightEvidence {
107 pub fn unavailable(reason: impl Into<String>) -> Self {
108 Self {
109 status: GpuEvidenceStatus::Unavailable,
110 reason: Some(reason.into()),
111 raw_report: None,
112 source_csv: Vec::new(),
113 coverage: NsightCoverage::default(),
114 correlation: NsightCorrelation {
115 mode: "none".into(),
116 clock_aligned: false,
117 complete: false,
118 matched_ranges: 0,
119 total_ranges: 0,
120 reason: Some("No projected NVTX ranges were normalized".into()),
121 },
122 diagnostics: Vec::new(),
123 limits: Default::default(),
124 kernels: Vec::new(),
125 runtime_calls: Vec::new(),
126 memory_operations: Vec::new(),
127 nvtx_ranges: Vec::new(),
128 gpu_timeline: Vec::new(),
129 }
130 }
131
132 pub fn load_optional(dir: Option<&Path>, expected_semantic_keys: &[String]) -> Self {
135 let Some(dir) = dir else {
136 return Self::unavailable("Nsight capture was not requested");
137 };
138 match Self::load(dir, expected_semantic_keys) {
139 Ok(evidence) => evidence,
140 Err(error) => Self {
141 status: GpuEvidenceStatus::Failed,
142 reason: Some(error.to_string()),
143 ..Self::unavailable("Nsight report normalization failed")
144 },
145 }
146 }
147
148 pub fn load(dir: &Path, expected_semantic_keys: &[String]) -> anyhow::Result<Self> {
149 anyhow::ensure!(
150 dir.is_dir(),
151 "Nsight report directory does not exist: {}",
152 dir.display()
153 );
154 let mut files: Vec<PathBuf> = fs::read_dir(dir)?
155 .filter_map(Result::ok)
156 .map(|entry| entry.path())
157 .filter(|path| path.is_file())
158 .collect();
159 files.sort();
160
161 let raw_report = files
162 .iter()
163 .find(|path| extension(path) == "nsys-rep")
164 .map(|path| path.display().to_string());
165 let mut result = Self {
166 status: GpuEvidenceStatus::Unavailable,
167 reason: None,
168 raw_report,
169 source_csv: Vec::new(),
170 coverage: NsightCoverage::default(),
171 correlation: NsightCorrelation {
172 mode: "none".into(),
173 clock_aligned: false,
174 complete: false,
175 matched_ranges: 0,
176 total_ranges: 0,
177 reason: Some("No projected NVTX ranges were normalized".into()),
178 },
179 diagnostics: Vec::new(),
180 limits: Default::default(),
181 kernels: Vec::new(),
182 runtime_calls: Vec::new(),
183 memory_operations: Vec::new(),
184 nvtx_ranges: Vec::new(),
185 gpu_timeline: Vec::new(),
186 };
187
188 for path in files.iter().filter(|path| extension(path) == "csv") {
189 let name = path
190 .file_name()
191 .and_then(|x| x.to_str())
192 .unwrap_or_default();
193 let parsed: anyhow::Result<Option<ReportLimit>> = if name.contains("cuda_gpu_kern_sum")
194 {
195 parse_summary(path).map(|(rows, limit)| {
196 result.kernels.extend(rows);
197 result.coverage.kernel_summary = true;
198 Some(limit)
199 })
200 } else if name.contains("cuda_api_sum") {
201 parse_summary(path).map(|(rows, limit)| {
202 result.runtime_calls.extend(rows);
203 result.coverage.runtime_summary = true;
204 Some(limit)
205 })
206 } else if name.contains("cuda_gpu_mem_time_sum") {
207 parse_summary(path).map(|(rows, limit)| {
208 result.memory_operations.extend(rows);
209 result.coverage.memory_summary = true;
210 Some(limit)
211 })
212 } else if name.contains("nvtx_gpu_proj_trace") {
213 parse_timeline(path, "nvtx_range", true).map(|(rows, limit)| {
214 result.nvtx_ranges.extend(rows);
215 result.coverage.nvtx_projection = true;
216 Some(limit)
217 })
218 } else if name.contains("cuda_gpu_trace") {
219 parse_timeline(path, "gpu_operation", false).map(|(rows, limit)| {
220 result.gpu_timeline.extend(rows);
221 result.coverage.gpu_timeline = true;
222 Some(limit)
223 })
224 } else {
225 Ok(None)
226 };
227 match parsed {
228 Ok(Some(limit)) => {
229 result.source_csv.push(path.display().to_string());
230 result.limits.insert(name.to_string(), limit);
231 }
232 Ok(None) => {}
233 Err(error) => result
234 .diagnostics
235 .push(format!("{}: {error}", path.display())),
236 }
237 }
238
239 let useful_rows = result.kernels.len()
240 + result.runtime_calls.len()
241 + result.memory_operations.len()
242 + result.nvtx_ranges.len()
243 + result.gpu_timeline.len();
244 if useful_rows == 0 {
245 result.reason = status_reason(dir)
246 .or_else(|| Some("No supported nsys stats CSV reports were found".into()));
247 } else {
248 result.status = GpuEvidenceStatus::Available;
249 }
250 if result.coverage.nvtx_projection {
251 let expected = expected_semantic_keys
252 .iter()
253 .collect::<std::collections::HashSet<_>>();
254 let matched_ranges = result
255 .nvtx_ranges
256 .iter()
257 .filter(|row| {
258 row.semantic_key
259 .as_ref()
260 .is_some_and(|key| expected.contains(key))
261 })
262 .count();
263 let total_ranges = result.nvtx_ranges.len();
264 let complete = total_ranges > 0
265 && matched_ranges == total_ranges
266 && result
267 .nvtx_ranges
268 .iter()
269 .all(|row| row.gpu_operations.is_some());
270 result.correlation = NsightCorrelation {
271 mode: "nvtx_projected_range".into(),
272 clock_aligned: false,
273 complete,
274 matched_ranges,
275 total_ranges,
276 reason: Some(if complete {
277 "Every projected NVTX range matched an exact application semantic label; Candle and Nsight clocks remain separate".into()
278 } else {
279 format!(
280 "{matched_ranges}/{total_ranges} projected NVTX ranges matched exact application semantic labels; clocks remain separate"
281 )
282 }),
283 };
284 }
285 Ok(result)
286 }
287}
288
289fn parse_summary(path: &Path) -> anyhow::Result<(Vec<NsightSummaryRow>, ReportLimit)> {
290 let mut reader = csv::ReaderBuilder::new().flexible(true).from_path(path)?;
291 let headers = normalized_headers(reader.headers()?);
292 anyhow::ensure!(
293 has_header(&headers, &["name", "operation", "range", "kernel_name"]),
294 "missing operation/name column"
295 );
296 anyhow::ensure!(
297 has_header(&headers, &["total_time_ns", "total_ns"]),
298 "missing total-time nanoseconds column"
299 );
300 let mut rows = Vec::new();
301 for record in reader.records() {
302 let record = record?;
303 let name = field(
304 &headers,
305 &record,
306 &["name", "operation", "range", "kernel_name"],
307 )
308 .unwrap_or_default()
309 .trim()
310 .to_string();
311 if name.is_empty() {
312 continue;
313 }
314 rows.push(NsightSummaryRow {
315 name,
316 total_ns: number(field(&headers, &record, &["total_time_ns", "total_ns"])),
317 count: number(field(
318 &headers,
319 &record,
320 &["instances", "num_calls", "operations", "count"],
321 )),
322 average_ns: number(field(&headers, &record, &["avg_ns", "average_ns"])),
323 minimum_ns: number(field(&headers, &record, &["min_ns", "minimum_ns"])),
324 maximum_ns: number(field(&headers, &record, &["max_ns", "maximum_ns"])),
325 category: field(&headers, &record, &["category"]).map(str::to_string),
326 });
327 }
328 rows.sort_by_key(|row| std::cmp::Reverse(row.total_ns));
329 let total_rows = rows.len();
330 rows.truncate(100);
331 Ok((
332 rows,
333 ReportLimit {
334 total_rows,
335 displayed_rows: total_rows.min(100),
336 truncated: total_rows > 100,
337 },
338 ))
339}
340
341fn parse_timeline(
342 path: &Path,
343 kind: &str,
344 projected: bool,
345) -> anyhow::Result<(Vec<NsightTimelineRow>, ReportLimit)> {
346 let mut reader = csv::ReaderBuilder::new().flexible(true).from_path(path)?;
347 let headers = normalized_headers(reader.headers()?);
348 anyhow::ensure!(
349 has_header(&headers, &["name", "operation", "range", "kernel_name"]),
350 "missing operation/name column"
351 );
352 anyhow::ensure!(
353 has_header(&headers, &["start_ns", "start"])
354 && has_header(&headers, &["duration_ns", "duration", "dur_ns"]),
355 "missing start/duration nanoseconds columns"
356 );
357 if projected {
358 anyhow::ensure!(
359 has_header(
360 &headers,
361 &["projected_start_ns", "projected_start", "proj_start_ns"]
362 ) && has_header(
363 &headers,
364 &["projected_duration_ns", "projected_duration", "proj_dur_ns"]
365 ),
366 "missing projected start/duration nanoseconds columns"
367 );
368 }
369 let mut rows = Vec::new();
370 for record in reader.records() {
371 let record = record?;
372 let name = field(
373 &headers,
374 &record,
375 &["name", "operation", "range", "kernel_name"],
376 )
377 .unwrap_or_default()
378 .trim()
379 .to_string();
380 if name.is_empty() {
381 continue;
382 }
383 let start_ns =
384 required_number(field(&headers, &record, &["start_ns", "start"]), "start_ns")?;
385 let duration_ns = required_number(
386 field(&headers, &record, &["duration_ns", "duration", "dur_ns"]),
387 "duration_ns",
388 )?;
389 anyhow::ensure!(duration_ns > 0, "timeline row `{name}` has zero duration");
390 let projected_start_ns = optional_number(field(
391 &headers,
392 &record,
393 &["projected_start_ns", "projected_start", "proj_start_ns"],
394 ));
395 let projected_duration_ns = optional_number(field(
396 &headers,
397 &record,
398 &["projected_duration_ns", "projected_duration", "proj_dur_ns"],
399 ));
400 if projected {
401 anyhow::ensure!(
402 projected_start_ns.is_some()
403 && projected_duration_ns.is_some_and(|duration| duration > 0),
404 "projected timeline row `{name}` has invalid projected timing"
405 );
406 }
407 rows.push(NsightTimelineRow {
408 semantic_key: (kind == "nvtx_range").then(|| name.clone()),
409 name,
410 kind: kind.into(),
411 device: field(&headers, &record, &["device", "device_id"]).map(str::to_string),
412 context: field(&headers, &record, &["context", "context_id"]).map(str::to_string),
413 stream: field(&headers, &record, &["stream", "stream_id"]).map(str::to_string),
414 correlation_id: field(&headers, &record, &["correlation_id", "corrid", "corr_id"])
415 .map(str::to_string),
416 start_ns,
417 duration_ns,
418 projected_start_ns,
419 projected_duration_ns,
420 gpu_operations: optional_number(field(
421 &headers,
422 &record,
423 &["num_gpu_ops", "numgpuops", "gpu_operations"],
424 )),
425 });
426 }
427 rows.sort_by_key(|row| row.start_ns);
428 let total_rows = rows.len();
429 rows.truncate(500);
430 Ok((
431 rows,
432 ReportLimit {
433 total_rows,
434 displayed_rows: total_rows.min(500),
435 truncated: total_rows > 500,
436 },
437 ))
438}
439
440fn has_header(headers: &[String], names: &[&str]) -> bool {
441 names
442 .iter()
443 .any(|name| headers.iter().any(|header| header == name))
444}
445
446fn normalized_headers(headers: &csv::StringRecord) -> Vec<String> {
447 headers
448 .iter()
449 .map(|header| {
450 header
451 .trim()
452 .to_ascii_lowercase()
453 .replace(['(', ')', '%'], "")
454 .replace([' ', '-', '/'], "_")
455 .trim_matches('_')
456 .to_string()
457 })
458 .collect()
459}
460
461fn field<'a>(headers: &[String], record: &'a csv::StringRecord, names: &[&str]) -> Option<&'a str> {
462 names.iter().find_map(|name| {
463 headers
464 .iter()
465 .position(|header| header == name)
466 .and_then(|index| record.get(index))
467 })
468}
469
470fn number(value: Option<&str>) -> u64 {
471 optional_number(value).unwrap_or(0)
472}
473
474fn required_number(value: Option<&str>, label: &str) -> anyhow::Result<u64> {
475 optional_number(value).ok_or_else(|| anyhow::anyhow!("invalid or missing {label} value"))
476}
477
478fn optional_number(value: Option<&str>) -> Option<u64> {
479 let cleaned = value?.trim().replace(',', "");
480 cleaned
481 .parse::<u64>()
482 .ok()
483 .or_else(|| cleaned.parse::<f64>().ok().map(|x| x.max(0.0) as u64))
484}
485
486fn extension(path: &Path) -> &str {
487 path.extension()
488 .and_then(|value| value.to_str())
489 .unwrap_or_default()
490}
491
492fn status_reason(dir: &Path) -> Option<String> {
493 let content = fs::read_to_string(dir.join("status.txt")).ok()?;
494 content
495 .lines()
496 .find_map(|line| line.strip_prefix("reason=").map(str::to_string))
497}
498
499#[cfg(test)]
500mod tests {
501 use super::*;
502
503 #[test]
504 fn normalizes_official_summary_headers() {
505 let dir = std::env::temp_dir().join(format!("candle-graph-nsys-{}", std::process::id()));
506 fs::create_dir_all(&dir).unwrap();
507 let path = dir.join("sample_cuda_gpu_kern_sum.csv");
508 fs::write(&path, "Time (%),Total Time (ns),Instances,Avg (ns),Min (ns),Max (ns),Name\n50.0,1200,3,400,200,600,gemm\n").unwrap();
509 let evidence = NsightEvidence::load(&dir, &[]).unwrap();
510 assert_eq!(evidence.status, GpuEvidenceStatus::Available);
511 assert_eq!(evidence.kernels[0].name, "gemm");
512 assert_eq!(evidence.kernels[0].total_ns, 1200);
513 let _ = fs::remove_dir_all(dir);
514 }
515
516 #[test]
517 fn malformed_report_is_diagnostic_not_available() {
518 let dir =
519 std::env::temp_dir().join(format!("candle-graph-nsys-bad-{}", std::process::id()));
520 fs::create_dir_all(&dir).unwrap();
521 fs::write(dir.join("bad_cuda_api_sum.csv"), "Unknown,Value\nx,1\n").unwrap();
522 let evidence = NsightEvidence::load(&dir, &[]).unwrap();
523 assert_eq!(evidence.status, GpuEvidenceStatus::Unavailable);
524 assert!(!evidence.diagnostics.is_empty());
525 let _ = fs::remove_dir_all(dir);
526 }
527
528 #[test]
529 fn reports_when_rows_are_truncated() {
530 let dir =
531 std::env::temp_dir().join(format!("candle-graph-nsys-limit-{}", std::process::id()));
532 fs::create_dir_all(&dir).unwrap();
533 let mut csv = String::from("Total Time (ns),Instances,Name\n");
534 for index in 0..101 {
535 csv.push_str(&format!("{},1,kernel-{index}\n", index + 1));
536 }
537 fs::write(dir.join("many_cuda_gpu_kern_sum.csv"), csv).unwrap();
538 let evidence = NsightEvidence::load(&dir, &[]).unwrap();
539 let limit = evidence.limits.values().next().unwrap();
540 assert_eq!(limit.total_rows, 101);
541 assert_eq!(limit.displayed_rows, 100);
542 assert!(limit.truncated);
543 let _ = fs::remove_dir_all(dir);
544 }
545
546 #[test]
547 fn projected_ranges_require_timing_and_exact_semantic_join() {
548 let dir =
549 std::env::temp_dir().join(format!("candle-graph-nsys-join-{}", std::process::id()));
550 fs::create_dir_all(&dir).unwrap();
551 fs::write(
552 dir.join("sample_nvtx_gpu_proj_trace.csv"),
553 "Name,Start (ns),Duration (ns),Projected Start (ns),Projected Duration (ns),Num GPU Ops\nrun/forward,1,10,2,8,3\n",
554 )
555 .unwrap();
556 let evidence = NsightEvidence::load(&dir, &["run/forward".into()]).unwrap();
557 assert!(evidence.correlation.complete);
558 assert_eq!(evidence.correlation.matched_ranges, 1);
559
560 fs::write(
561 dir.join("sample_nvtx_gpu_proj_trace.csv"),
562 "Name,Num GPU Ops\nrun/forward,3\n",
563 )
564 .unwrap();
565 let invalid = NsightEvidence::load(&dir, &["run/forward".into()]).unwrap();
566 assert_eq!(invalid.status, GpuEvidenceStatus::Unavailable);
567 assert!(!invalid.diagnostics.is_empty());
568 let _ = fs::remove_dir_all(dir);
569 }
570}