diff --git a/docs/adapters.md b/docs/adapters.md index 24c240f..522bf63 100644 --- a/docs/adapters.md +++ b/docs/adapters.md @@ -377,57 +377,123 @@ not exact timestamps. #### `flameox.kernel-validation.v1` -Producer-neutral kernel-correctness evidence is imported as strict JSON conforming to the -`flameox.kernel-validation.v1` schema. The published JSON Schema lives at -`src/flameox/schemas/kernel-validation-v1.schema.json` and is generated from the same Pydantic -model used for validation. The artifact kind is `validation_output`; import does not require a -GPU, CUDA, or a kernel runner. - -The contract binds producer and reference identity, bounded case inputs, seed and device, -declared metrics and tolerances, output and case outcomes, up to eight representative failures, -and coverage limitations. Supported metrics are `max_abs_error`, `max_rel_error`, `mse`, `rmse`, -`psnr`, and `cosine_similarity`. Lower-is-better metrics require `<=`; higher-is-better metrics -require `>=`. Metric, output, case, and aggregate statuses are checked for consistency. Passing -the document requires complete coverage; non-finite values and incomplete coverage without a -limitation are rejected. - -Extraction publishes additive schema-minor-9 tables `kernel_validation_cases` and -`kernel_validation_metrics`. Native JSON remains authoritative, extracted tables are rebuildable, -and repeated extraction reuses the existing generation. Fixtures are project-owned synthetic JSON -documents generated in `tests/adapters/test_kernel_validation.py` under the project MIT license. - -Observed claims are the declared case outcomes, metric values and thresholds, coverage flag, and -representative failure coordinates. Aggregate statuses are derived. The extractor makes no inferred -correctness claim beyond the declared metrics, tolerances, and coverage. +Producer-neutral kernel-correctness evidence is imported as strict JSON +conforming to the `flameox.kernel-validation.v1` schema. The published JSON +Schema lives at `src/flameox/schemas/kernel-validation-v1.schema.json` and is +generated from the same Pydantic model used for validation, so import and +schema publication cannot drift. The artifact kind is `validation_output`; +import does not require a GPU, CUDA, or any kernel runner. + +The document binds producer and reference-implementation identity, per-case +inputs (dtype, shape, role, seed, device), declared metric definitions and +tolerances, per-output and per-case observed metrics, pass/fail/inconclusive/ +unsupported outcomes, bounded representative failures, and coverage +limitations. Supported metrics are `max_abs_error`, `max_rel_error`, `mse`, +`rmse`, `psnr`, and `cosine_similarity`; each carries a comparator (`<=` for +lower-is-better, `>=` for higher-is-better), threshold, unit, and status. The +model enforces three aggregate-verdict invariants: a metric status cannot +contradict its value and threshold; an output status cannot contradict its +metrics; and the document status cannot contradict its case outcomes or +declared coverage completeness. Non-finite values, unknown schema versions, +ambiguous aggregates, and incomplete coverage without a stated limitation are +rejected explicitly. Failed outputs require at least one bounded representative +failure; non-failed outputs reject representative failures. + +Extraction publishes two evidence tables: +`kernel_validation_cases` (case/output identity, status, dimensions, inputs, +device, seed, representative failures, limitations) and +`kernel_validation_metrics` (per-output metric name, value, comparator, +threshold, unit, status, limitation). Both are schema minor 9 additions; the +raw artifact remains authoritative and rows are rebuildable. Repeating +extraction for the same artifact and extractor identity reuses the existing +normalized generation. The trial-level `flameox.oracle-receipt.v1` remains the +decisive verdict and may reference this artifact by diagnostic role; the +detailed artifact does not duplicate benchmark samples, run provenance, or +experiment structure. + +Ownership: `tests/adapters/test_kernel_validation.py`, lane `adapters`, +markers `unit`. Fixtures are synthetic JSON documents generated in-test; no +vendor-produced artifact is required. The test suite covers exact agreement, +tolerance-bound agreement, numerical failure with bounded representative +failures, non-finite rejection, contradictory aggregate rejection, unknown +schema version rejection, and idempotent re-extraction. + +Observed claims: per-case status, per-metric value and threshold, declared +tolerances, coverage completeness flag, representative failure coordinates. +Derived claims: aggregate output status from metric statuses, aggregate case +status from output statuses, document status from case outcomes and coverage. +Inferred claims: none — the extractor does not infer correctness beyond the +declared metrics and tolerances. #### `compute-sanitizer` -The maintained adapter runs NVIDIA Compute Sanitizer around a declared workload and preserves its -XML report as `sanitizer_report`. It supports the official `memcheck`, `racecheck`, `initcheck`, and -`synccheck` tools and fixes output to `--xml --save `. Strict options cover launch skip/count, -target-process scope and bounded filter, kernel name, demangling, a distinct finding exit code, and -an optional project-relative suppression file whose SHA-256 digest enters capture-plan identity. -Suppression files must be regular, non-linked, project-contained files; arbitrary flags are refused. - -Compatibility family `compute-sanitizer.xml.2026.v1` was observed with Compute Sanitizer 2026.2.1. -NVIDIA does not publish a stable XML XSD, so extraction is version-bounded and unknown record shapes -or tags become limitations. XML parsing runs in an isolated bounded worker using `defusedxml`, caps -host stacks at 64 frames, and normalizes source paths. A configured finding exit plus parsed records -is a completed failed validation, while other nonzero exits, missing or malformed reports, and -timeouts remain failed attempts with preserved partial evidence. - -Linux and Windows are supported; GPU access and a CUDA toolkit installation are required and are -not provisioned by Flameox. Overhead depends on the selected sanitizer tool and input. A clean report -covers only the selected tool, launches, processes, and filters and does not prove numerical -correctness. - -The live fixture `tests/fixtures/compute_sanitizer/kernel_probe.cu` is project-owned MIT-licensed -CUDA C++. The optional live test compiles it with `nvcc -lineinfo` and checks both in-bounds and -out-of-bounds captures. Deterministic tests use synthetic XML and cover clean, memory, API, -sanitizer, malformed, truncated, unknown, and oversized reports. Observed claims come from XML; -classification and path normalization are deterministic derivations; no root cause is inferred. - -+#### `nvbench` +The maintained capture adapter runs NVIDIA Compute Sanitizer around an already +declared workload and preserves the XML report as a `sanitizer_report` +artifact. The adapter supports the four official tools — `memcheck`, +`racecheck`, `initcheck`, and `synccheck` — and emits XML through `--xml` with +`--save`. It does not require a separate import converter; existing XML reports +may also be imported directly with producer `compute-sanitizer`. + +Official output: `compute-sanitizer --tool --xml --save -- `. +The adapter binds tool choice, `launch_skip`, `launch_count`, +`target_processes` (`application-only` or `all`), `target_processes_filter`, +`kernel_name`, `demangle`, `finding_exit_code`, and an optional +project-relative suppression file whose SHA-256 digest is recorded in the +capture plan. Suppression files must be regular, non-linked, project-contained +files; absolute paths, `..` traversal, and symlinks are rejected. Arbitrary +flag injection is refused — only the declared options are accepted. + +Platform: Linux and Windows. Permissions: GPU access and the CUDA toolkit; +the executable is detected at runtime and not provisioned by Flameox. Overhead: +GPU instrumentation overhead whose exact cost depends on the selected sanitizer +tool. Containment: follows the workload's selected execution policy; the +sanitizer wraps the workload argv and Flameox owns process execution, +containment, quotas, and cancellation. The adapter preserves artifacts on +nonzero exit because sanitizer findings produce a nonzero exit code by design. + +Local evidence: Compute Sanitizer `2026.2.1` was observed locally. The +compatibility family is `compute-sanitizer.xml.2026.v1`; extraction is +version-bounded because NVIDIA does not publish a stable XSD for the XML +format. The parser runs in a bounded subprocess worker +(`flameox.workers.compute_sanitizer`) behind the canonical broker and uses +`defusedxml` rather than implementing XML entity defenses locally. It rejects +DTD and entity declarations, requires a `ComputeSanitizerOutput` root element, +and reports unknown record elements or XML tags as explicit limitations rather +than silently accepting them. Host stacks are truncated to 64 frames. Source +paths are normalized to project-relative form; external paths are reported as +`/`. The extractor distinguishes `clean` (zero findings, +no limitations), `findings` (one or more records), and `inconclusive` +(limitations without records). A clean report covers only the selected tool, +launches, processes, and filters; it does not prove numerical correctness. An +oracle receipt may reference the sanitizer report by diagnostic role, but a +clean sanitizer run never proves numerical equivalence. + +Fixture provenance and licensing: `tests/fixtures/compute_sanitizer/kernel_probe.cu` +is a Flameox-authored CUDA C++ source file under the project MIT license. It +compiles with `nvcc -lineinfo` into a small probe that performs an in-bounds or +out-of-bounds global memory write depending on a runtime argument. No +vendor-produced XML fixture is committed; XML fixtures are synthetic and +generated in-test. The live test (`tests/adapters/test_compute_sanitizer_live.py`) +compiles the probe with `nvcc`, runs `compute-sanitizer --tool memcheck` around +it, imports the XML, and extracts findings; it skips when `compute-sanitizer` +or `nvcc` is absent. + +Ownership: `tests/adapters/test_compute_sanitizer.py` is owned by `adapters`, +lane `adapters`, markers `unit`. `tests/adapters/test_compute_sanitizer_live.py` +is owned by `compute-sanitizer-live`, lane `adapters`, markers `integration`, +`optional`, `process`, `serial`, `requires_compute_sanitizer`. + +Observed claims: error kind, level, message, memory space, access size, +direction, error class, function, source path, line, PC, thread/block indices, +and host stack frames — all read directly from the XML record. Derived claims: +classification (`memory_access`, `race`, `uninitialized_memory`, +`synchronization`, `api_error`, `sanitizer_error`, `unknown`) is derived from +the record's kind, message, and error text; project-relative path normalization +is a deterministic derivation from the reported path. Inferred claims: none — +the extractor does not infer the root cause of a memory error or claim +correctness beyond the retained findings. + +#### `nvbench` The NVBench integration targets the JSON schema and JSON-binary behavior verified at `NVIDIA/nvbench@c18488992e313240166f588b9ee4da3e0de76004`. NVBench is linked into each benchmark @@ -501,6 +567,102 @@ staging paths, manifest outcome, and pipeline identity. Cache status remains `unknown` unless the producer supplies evidence. Inferred claims: none — an available stage does not prove semantic correctness, optimal code generation, or a cache hit. +#### `nsight.compute` + +The maintained adapter preserves official `.ncu-rep` and `.ncu-repz` reports as immutable +`kernel_profile` artifacts. Extraction runs in the isolated artifact worker and exclusively uses +the `ncu_report` Python interface shipped with the detected Nsight Compute installation. NVIDIA +owns the native report format and reader; Flameox neither decodes the binary format nor adds a +runtime PyPI dependency. The schema fingerprint binds the report-interface version to the +observed metric and section identities. Roofline evidence is published only when a metric, rule, +or section in the report explicitly identifies roofline data; Flameox does not synthesize a +roofline or bottleneck conclusion. + +Capture requires exactly one named section set or 1–32 exact section identifiers, with an +optional bounded kernel name, launch skip from 0 to 1,000,000, launch count from 1 to 1,000,000, +and replay mode `kernel`, `application`, `range`, or `app-range`. Regex sections, arbitrary flags, +external section directories, and source import are not accepted. Linux and Windows are declared +capture platforms; a supported NVIDIA GPU, driver, Nsight Compute installation, and +performance-counter permission are required. Kernel replay can impose substantial, +workload-dependent overhead. The standard execution policy owns process containment, staging, +quotas, timeout, and +cancellation, while `ncu` owns the native report bytes. Flameox never changes privileges. NVIDIA's +`ERR_NVGPUCTRPERM` maps to `permission_required` with remediation rather than a privilege attempt. + +Normalization is bounded by the workspace generation row quota, a 120-second worker timeout, +1,000 ranges, 10,000 actions, and separate metric and observation budgets derived from the row +quota. Numeric metrics enter `measurements`; string attributes, rules and result tables, section +identities, and source/SASS/PTX references enter observations. Source bodies are not copied out of +the report. Unknown metric value kinds, truncated collections, and exceptions from optional +official-interface access become explicit limitations. Missing optional methods degrade +gracefully; corrupt reports, a missing official interface, and invalid required interface results +fail extraction with bounded recovery guidance. + +Local compatibility evidence used Nsight Compute `2026.2.1` and its installed, vendor-produced +`extras/samples/instructionMix/sobelFloat.ncu-rep`. The sample was read successfully through that +installation's `ncu_report` interface and is not redistributed or committed by Flameox; it remains +covered by the locally installed NVIDIA product's license. Deterministic tests use a +Flameox-authored fake interface under the project MIT license to exercise type handling, bounds, +corruption, and provenance. The local driver initially produced the official +`ERR_NVGPUCTRPERM` diagnostic, proving the `permission_required` mapping. After an administrator +enabled non-admin counter access, the managed adapter profiled the pinned official NVBench stream +benchmark with the `basic` set on the `sm_86` RTX 3060, preserved the `.ncu-rep`, and extracted +actions and numeric metrics through the installed official interface. Both the denied capability +path and successful live counter capture are therefore observed on this host. + +Observed claims are report/range/action names exposed by the official interface, kernel and device +identity attributes, numeric and string metric values and units, section and rule identities, +result tables, and source/SASS/PTX references. Derived claims are bounded row normalization, +counts, schema fingerprint, and the explicit-identity test for `roofline_present`. Inferred claims: +none — metric presence or magnitude does not by itself prove causality, correctness, a bottleneck, +or an optimization opportunity. + +#### `rocprofv3` + +The maintained Linux capture adapter invokes the official ROCprofiler-SDK CLI +with `--output-format pftrace`, a Flameox-owned `-d` staging directory, and the +fixed `-o rocprofv3` basename. It accepts only the documented `--hip-trace`, +`--kernel-trace`, `--memory-copy-trace`, `--memory-allocation-trace`, +`--scratch-memory-trace`, and `--marker-trace` domains; at least one must be +enabled. The resulting `rocprofv3_results.pftrace` remains an immutable +`execution_trace` artifact with producer `rocprofv3`. Flameox does not load the +raw ROCprofiler SDK or decode PFTrace: extraction uses the existing bounded +Perfetto worker and accelerator recipe. + +Compatibility floor: ROCm 6.2 / ROCprofiler-SDK 0.4 is the documented minimum +for rocprofv3 HIP tracing and PFTrace output. The complete option set above and +the `_results.pftrace` naming convention were verified against current official +ROCm 7.x documentation. An older CLI that rejects a selected domain is reported +as a failed attempt with its stdout, stderr, process identity, and any non-empty +PFTrace retained; Flameox does not silently substitute another domain. + +Platform and requirements: Linux, a supported AMD GPU and ROCm installation, +and workload access to the host GPU device nodes (normally `/dev/kfd` and +`/dev/dri`). Flameox never changes device permissions or privileges. Tracing +overhead depends on the enabled domains and workload event rate. The normal +execution policy owns containment, quotas, timeout, cancellation, and the +bounded output directory; rocprofv3 owns the native PFTrace bytes. + +Fixture provenance and proof gap: `tests/adapters/test_rocprofv3.py` creates a +Flameox-authored fake CLI under the project MIT license. It verifies exact argv, +output naming, strict option rejection, and partial-artifact preservation, but +its sentinel output is not presented as a valid PFTrace. The project-owned +`project-owned-rocm-shaped-perfetto.json` fixture is a valid Perfetto-compatible +Chrome trace containing HIP-runtime-shaped and kernel events. It covers import, +bounded Perfetto extraction, and accelerator summarization, but it was not +produced by rocprofv3 and therefore does not establish native PFTrace +compatibility. No AMD host or vendor-produced trace is in scope, so rocprofv3 +capture compatibility remains fixture/process-simulation backed. + +Perfetto extraction inherits the workspace row quota, worker timeout, curated +standard-table queries, truncation reporting, and malformed-trace errors. +Unsupported or absent ROCm events degrade to incomplete standard summaries; +Flameox does not add a second parser or infer missing activity. Observed claims +are the selected domains, process result, native artifact identity, and events +returned by Trace Processor. Derived claims are bounded normalized slices, +counts, durations, and accelerator summaries. Inferred claims: none — neither a +trace nor an absent event establishes causality, correctness, or a bottleneck. + ### Candidate adapters - GDB/LLDB and elfutils for core metadata, with user init files, autoload, and @@ -511,7 +673,6 @@ available stage does not prove semantic correctness, optimal code generation, or - `rr` recording references; - Nsight Systems Arrow, JSONL, or Parquet exports beyond the maintained SQLite subset; -- `rocprofv3` Perfetto-compatible exports; - heaptrack or platform-native heap profiles; - VizTracer or Python monitoring integrations when ordered call evidence is necessary and Perfetto annotations are insufficient; diff --git a/docs/interfaces.md b/docs/interfaces.md index 232cd8b..d657a04 100644 --- a/docs/interfaces.md +++ b/docs/interfaces.md @@ -410,7 +410,7 @@ The supported tools are grouped as follows: | Family | Tools | | --- | --- | | Workspace | `initialize_workspace`, `workspace_status`, `workload_configuration_status`, `configure_workload`, `list_capabilities`, `start_capability_setup`, `get_capability_setup`, `cancel_capability_setup`, `prepare_adapter`, `prepare_workload_dependencies`, `validate_workspace` | -| Capture and import | `plan_capture`, `execute_capture_plan`, `import_artifact`, `import_nvbench`, `import_kernel_build`, `extract_benchmark_samples`, `extract_pyperf`, `extract_python_startup`, `extract_pytest`, `extract_coverage`, `extract_memray`, `extract_perfetto`, `extract_nsight_systems`, `extract_kernel_validation`, `extract_compute_sanitizer`, `extract_nvbench`, `extract_observations` | +| Capture and import | `plan_capture`, `execute_capture_plan`, `import_artifact`, `import_nvbench`, `import_kernel_build`, `extract_benchmark_samples`, `extract_pyperf`, `extract_python_startup`, `extract_pytest`, `extract_coverage`, `extract_memray`, `extract_perfetto`, `extract_nsight_systems`, `extract_kernel_validation`, `extract_compute_sanitizer`, `extract_nvbench`, `extract_nsight_compute`, `extract_observations` | | Detached capture | `start_detached_capture`, `get_detached_capture`, `cancel_detached_capture` | | Discovery | `list_declared_workflows`, `get_declared_workflow`, `list_runs`, `list_findings` | | Investigations | `create_investigation`, `list_investigations`, `get_investigation`, `record_hypothesis`, `get_hypothesis` | diff --git a/pyproject.toml b/pyproject.toml index c6d3657..6738e8d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,9 +10,9 @@ readme = "README.md" license = "MIT" requires-python = ">=3.12" dependencies = [ - "defusedxml>=0.7.1,<0.8", "anyio>=4.9,<5", "duckdb>=1.5.4,<1.6", + "defusedxml>=0.7.1,<0.8", "ijson>=3.4,<4", "mcp==2.0.0", "mcp-types==2.0.0", @@ -87,7 +87,6 @@ packages = ["src/flameox"] addopts = "-ra --strict-config --strict-markers --durations=10 -p no:randomly -m 'not performance and not optional and not process'" testpaths = ["tests"] markers = [ - "requires_compute_sanitizer: requires NVIDIA Compute Sanitizer and a compatible GPU", "unit: deterministic tests for one semantic owner", "integration: tests crossing two or more Flameox services", "process: tests that spawn, transport to, or cancel another process", @@ -104,6 +103,9 @@ markers = [ "requires_systemd: requires a systemd user manager", "requires_toxiproxy: requires the pinned Toxiproxy server binary", "requires_torch: requires the optional PyTorch provider", + "requires_compute_sanitizer: requires NVIDIA Compute Sanitizer and a compatible GPU", + "requires_ncu: requires NVIDIA Nsight Compute and its official Python report interface", + "requires_rocprofv3: requires rocprofv3 and a compatible AMD GPU", "requires_triton: requires the optional Triton provider and a compatible GPU", "requires_nvbench: requires a configured NVBench benchmark executable and compatible GPU", "requires_cute: requires the optional CuTe DSL provider and a compatible GPU", diff --git a/src/flameox/adapters/__init__.py b/src/flameox/adapters/__init__.py index 4519d85..ea52b4b 100644 --- a/src/flameox/adapters/__init__.py +++ b/src/flameox/adapters/__init__.py @@ -14,6 +14,7 @@ from flameox.adapters.kernel_build import * # noqa: F403 from flameox.adapters.kernel_validation import * # noqa: F403 from flameox.adapters.memray import * # noqa: F403 + from flameox.adapters.nsight_compute import * # noqa: F403 from flameox.adapters.nsight_systems import * # noqa: F403 from flameox.adapters.nvbench import * # noqa: F403 from flameox.adapters.observations import * # noqa: F403 @@ -31,13 +32,14 @@ _MODULES = ( "benchmark_samples", "client_setup", - "compute_sanitizer", "coverage", + "compute_sanitizer", "inference", "kernel_build", "kernel_validation", "memray", "nsight_systems", + "nsight_compute", "nvbench", "observations", "options", @@ -89,6 +91,9 @@ "MooncakeRequestRow", "MooncakeTraceParser", "MooncakeTraceSummary", + "NsightComputeExtractionResult", + "NsightComputeExtractor", + "NsightComputeOptions", "NsightSystemsExtractionResult", "NsightSystemsExtractor", "NvbenchBenchmark", diff --git a/src/flameox/adapters/builtins.py b/src/flameox/adapters/builtins.py index 2633285..7ad14c0 100644 --- a/src/flameox/adapters/builtins.py +++ b/src/flameox/adapters/builtins.py @@ -11,7 +11,9 @@ compute_sanitizer_options, compute_sanitizer_suppression_path, cute_compiler_options, + nsight_compute_options, nvbench_options, + rocprofv3_options, triton_compiler_options, ) from flameox.adapters.torch_profiler import SdkTorchProfilerOptions, torch_profiler_options @@ -282,6 +284,63 @@ class CaptureInvocation: ), preserve_artifact_on_nonzero=True, ), + BuiltinAdapter( + name="rocprofv3", + dependency_kind="executable", + dependency="rocprofv3", + supported_modes=("pftrace",), + supported_formats=("pftrace",), + features=( + "hip_api", + "kernel_dispatch", + "memory_copy", + "memory_allocation", + "scratch_memory", + "marker_ranges", + ), + remediation=( + "Install rocprofiler-sdk with rocprofv3 from ROCm and verify AMD GPU access.", + ), + version_args=("--version",), + supported_platforms=("linux",), + output_filename="rocprofv3_results.pftrace", + artifact_kinds=(ArtifactKind.EXECUTION_TRACE,), + expected_overhead=( + "ROCm tracing overhead depends on the selected API, dispatch, memory, scratch, " + "and marker domains." + ), + capture_limitations=( + "PFTrace contains only the explicitly selected rocprofv3 trace domains.", + "Counter collection and the raw rocprofiler SDK are not enabled.", + ), + preserve_artifact_on_nonzero=True, + ), + BuiltinAdapter( + name="nsight.compute", + dependency_kind="executable", + dependency="ncu", + supported_modes=("profile",), + supported_formats=("ncu-rep", "ncu-repz"), + features=("gpu_metrics", "report_sections", "rules", "source_correlation"), + remediation=( + "Install NVIDIA Nsight Compute and grant access to NVIDIA GPU performance " + "counters.", + ), + version_args=("--version",), + permissions=("nvidia_gpu_performance_counters",), + supported_platforms=("linux", "windows"), + output_filename="nsight-compute.ncu-rep", + artifact_kinds=(ArtifactKind.KERNEL_PROFILE,), + expected_overhead=( + "Kernel replay and metric collection can substantially change execution time." + ), + capture_limitations=( + "Only the selected set or explicit sections and bounded launches are profiled.", + "Counter availability depends on GPU, driver, and system permissions.", + "Roofline evidence is exposed only when present in the official report.", + ), + preserve_artifact_on_nonzero=True, + ), BuiltinAdapter( name="triton.compiler", dependency_kind="internal", @@ -353,7 +412,7 @@ def builtin_adapter(name: str) -> BuiltinAdapter | None: return BUILTIN_ADAPTERS.get(name) -def build_capture_invocation( +def build_capture_invocation( # noqa: C901 - provider routing is intentionally explicit adapter_name: str, workload_argv: tuple[str, ...], output_root: Path, @@ -451,6 +510,22 @@ def build_capture_invocation( executable=executable, options=options, ) + elif adapter_name == "rocprofv3": + return _rocprofv3_capture_invocation( + adapter, + workload_argv, + output_root, + executable=executable, + options=options, + ) + elif adapter_name == "nsight.compute": + return _nsight_compute_capture_invocation( + adapter, + workload_argv, + output, + executable=executable, + options=options, + ) elif adapter_name == "triton.compiler": return _triton_compiler_capture_invocation( adapter, @@ -767,6 +842,84 @@ def _nvbench_capture_invocation( ) +def _rocprofv3_capture_invocation( + adapter: BuiltinAdapter, + workload_argv: tuple[str, ...], + output_root: Path, + *, + executable: str | None, + options: dict[str, object] | None, +) -> CaptureInvocation: + selected = rocprofv3_options(options) + argv_parts: list[str] = [ + _require_executable(adapter.name, executable), + "--output-format", + "pftrace", + "-o", + "rocprofv3", + "-d", + str(output_root), + ] + for enabled, flag in ( + (selected.hip_trace, "--hip-trace"), + (selected.kernel_trace, "--kernel-trace"), + (selected.memory_copy_trace, "--memory-copy-trace"), + (selected.memory_allocation_trace, "--memory-allocation-trace"), + (selected.scratch_memory_trace, "--scratch-memory-trace"), + (selected.marker_trace, "--marker-trace"), + ): + if enabled: + argv_parts.append(flag) + argv = (*argv_parts, "--", *workload_argv) + return CaptureInvocation( + argv=argv, + artifact_kinds=adapter.artifact_kinds, + expected_overhead=adapter.expected_overhead or "", + limitations=adapter.capture_limitations, + environment={}, + ) + + +def _nsight_compute_capture_invocation( + adapter: BuiltinAdapter, + workload_argv: tuple[str, ...], + output: str, + *, + executable: str | None, + options: dict[str, object] | None, +) -> CaptureInvocation: + selected = nsight_compute_options(options) + argv_parts: list[str] = [ + _require_executable(adapter.name, executable), + "--export", + output, + "--force-overwrite", + "--replay-mode", + selected.replay_mode, + "--launch-skip", + str(selected.launch_skip), + "--launch-count", + str(selected.launch_count), + ] + if selected.set is not None: + argv_parts.extend(("--set", selected.set)) + else: + for section in selected.sections or (): + argv_parts.extend(("--section", section)) + if selected.kernel_name is not None: + argv_parts.extend( + ("--kernel-name-base", "demangled", "--kernel-name", selected.kernel_name) + ) + argv = (*argv_parts, *workload_argv) + return CaptureInvocation( + argv=argv, + artifact_kinds=adapter.artifact_kinds, + expected_overhead=adapter.expected_overhead or "", + limitations=adapter.capture_limitations, + environment={}, + ) + + def _triton_compiler_capture_invocation( adapter: BuiltinAdapter, workload_argv: tuple[str, ...], diff --git a/src/flameox/adapters/nsight_compute.py b/src/flameox/adapters/nsight_compute.py new file mode 100644 index 0000000..a8aa123 --- /dev/null +++ b/src/flameox/adapters/nsight_compute.py @@ -0,0 +1,297 @@ +from __future__ import annotations + +import json +import re +import shutil +from pathlib import Path +from typing import Any + +from flameox.application.artifact_workers import ArtifactWorker +from flameox.domain import ArtifactKind, DomainError, ErrorCode, digest_model +from flameox.evidence import GenerationPublisher +from flameox.models import ContractModel +from flameox.storage import ArtifactStore, RunStore, Workspace + + +class NsightComputeExtractionResult(ContractModel): + schema_version: int = 1 + run_id: str + artifact_id: str + producer_version: str | None + report_version: str + range_count: int + action_count: int + metric_count: int + observation_count: int + roofline_present: bool + schema_fingerprint: str + corpus_commit_id: str + limitations: tuple[str, ...] + + +def find_ncu_report_interface( + *, + executable: str | Path | None = None, + producer_version: str | None = None, +) -> Path | None: + """Find the Python interface shipped with Nsight Compute, never a PyPI substitute.""" + + candidates: set[Path] = set() + resolved_executable = Path(executable).resolve() if executable else None + if resolved_executable is not None: + for parent in resolved_executable.parents: + candidate = parent / "extras" / "python" / "ncu_report.py" + if candidate.is_file(): + candidates.add(candidate) + for base in (Path("/opt/nvidia/nsight-compute"), Path("/usr/local/NVIDIA-Nsight-Compute")): + if base.is_dir(): + candidates.update(base.glob("*/extras/python/ncu_report.py")) + direct = base / "extras" / "python" / "ncu_report.py" + if direct.is_file(): + candidates.add(direct) + if not candidates: + return None + + def key(path: Path) -> tuple[int, tuple[int, ...], str]: + version_text = path.parents[2].name + matches = tuple(int(part) for part in re.findall(r"\d+", version_text)) + exact = int(bool(producer_version and version_text == producer_version)) + return exact, matches, path.as_posix() + + return max(candidates, key=key) + + +class NsightComputeExtractor: + name = "nsight.compute.report" + version = "1" + compatibility_family = "nsight-compute.ncu-report-api.v1" + + def __init__(self, workspace: Workspace) -> None: + self.workspace = workspace + self.publisher = GenerationPublisher(workspace) + + def extract(self, run_id: str) -> NsightComputeExtractionResult: + run = RunStore(self.workspace).read(run_id) + matches = tuple(item for item in run.artifacts if item.kind is ArtifactKind.KERNEL_PROFILE) + if len(matches) != 1: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The run must contain exactly one Nsight Compute report.", + run_id=run_id, + ) + registration = matches[0] + if registration.producer not in {"nsight.compute", "ncu", "flameox.import"}: + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The kernel profile is not registered as Nsight Compute output.", + run_id=run_id, + details={"registered_producer": registration.producer}, + ) + if not registration.display_name.casefold().endswith((".ncu-rep", ".ncu-repz")): + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "Only unchanged .ncu-rep and .ncu-repz reports are supported.", + ) + interface = find_ncu_report_interface( + executable=shutil.which("ncu"), + producer_version=registration.producer_version, + ) + if interface is None: + raise DomainError( + ErrorCode.CAPABILITY_UNAVAILABLE, + "The official ncu_report Python interface was not found.", + remediation=( + "Install Nsight Compute with its extras/python interface, then retry; " + "FlameOx does not decode NVIDIA report binaries.", + ), + ) + artifact = ArtifactStore(self.workspace).get(registration.artifact_id) + maximum = self.workspace.config.storage.max_rows_per_generation + if maximum < 3: + raise DomainError( + ErrorCode.QUERY_BUDGET_EXCEEDED, + "Nsight Compute extraction requires room for metrics, observations, " + "and provenance.", + ) + max_metrics = (maximum - 1) // 2 + max_observations = maximum - 1 - max_metrics + response = ArtifactWorker(self.workspace).run_sync( + "flameox.workers.nsight_compute", + { + "artifact_path": str(artifact.payload_path), + "interface_path": str(interface), + "max_ranges": min(1_000, max_observations), + "max_actions": min(10_000, max_observations), + "max_metrics": max_metrics, + "max_observations": max_observations, + }, + name="Nsight Compute", + timeout_seconds=120, + ) + measurements = _dict_list(response.get("measurements"), "measurements") + observations = _dict_list(response.get("observations"), "observations") + metric_ids = _string_list(response.get("metric_ids"), "metric_ids") + section_ids = _string_list(response.get("section_ids"), "section_ids") + limitations = _string_list(response.get("limitations"), "limitations") + report_version = response.get("report_version") + if not isinstance(report_version, str): + raise DomainError( + ErrorCode.ARTIFACT_PARSE_FAILED, + "The ncu_report worker omitted its report version.", + ) + schema_fingerprint = digest_model( + { + "compatibility_family": self.compatibility_family, + "report_version": report_version, + "metric_ids": metric_ids, + "section_ids": section_ids, + } + ) + measurement_rows = [ + self._measurement_row(run_id, registration.artifact_id, index, item) + for index, item in enumerate(measurements) + ] + observation_rows = [ + self._observation_row(run_id, registration.artifact_id, index, item) + for index, item in enumerate(observations) + ] + observation_rows.append( + self._observation_row( + run_id, + registration.artifact_id, + len(observations), + { + "kind": "profile.extraction", + "name": self.compatibility_family, + "value": { + "metric_ids": metric_ids, + "producer_version": registration.producer_version, + "report_version": report_version, + "roofline_present": bool(response.get("roofline_present", False)), + "schema_fingerprint": schema_fingerprint, + "section_ids": section_ids, + }, + }, + ) + ) + published = self.publisher.publish_rows_idempotent( + {"measurements": measurement_rows, "observations": observation_rows}, + publisher=self.name, + publisher_version=self.version, + input_run_ids=(run_id,), + input_artifact_ids=(registration.artifact_id,), + operation_identity={ + "compatibility_family": self.compatibility_family, + "report_version": report_version, + "schema_fingerprint": schema_fingerprint, + "max_metrics": max_metrics, + "max_observations": max_observations, + }, + ) + return NsightComputeExtractionResult( + run_id=run_id, + artifact_id=registration.artifact_id, + producer_version=registration.producer_version, + report_version=report_version, + range_count=_nonnegative_int(response.get("range_count"), "range_count"), + action_count=_nonnegative_int(response.get("action_count"), "action_count"), + metric_count=len(measurements), + observation_count=len(observations), + roofline_present=bool(response.get("roofline_present", False)), + schema_fingerprint=schema_fingerprint, + corpus_commit_id=published.commit.commit_id, + limitations=tuple(limitations), + ) + + @staticmethod + def _measurement_row( + run_id: str, + artifact_id: str, + index: int, + metric: dict[str, Any], + ) -> dict[str, object]: + value = metric.get("value") + if isinstance(value, bool) or not isinstance(value, int | float): + raise DomainError(ErrorCode.ARTIFACT_PARSE_FAILED, "Invalid numeric metric value.") + name = metric.get("name") + unit = metric.get("unit") + if not isinstance(name, str) or not isinstance(unit, str): + raise DomainError(ErrorCode.ARTIFACT_PARSE_FAILED, "Invalid metric identity.") + dimensions = { + "action_index": str(metric.get("action_index")), + "action_name": str(metric.get("action_name")), + "range_index": str(metric.get("range_index")), + "report_provider": "nsight.compute", + } + return { + "measurement_id": digest_model( + {"artifact_id": artifact_id, "metric_index": index, "metric": metric} + ), + "run_id": run_id, + "artifact_id": artifact_id, + "name": name, + "value_int": value if isinstance(value, int) else None, + "value_float": value if isinstance(value, float) else None, + "unit": unit or "unknown", + "aggregation": "reported", + "scope": "device", + "trial_id": None, + "worker_id": None, + "worker_run_index": None, + "value_index": index, + "loop_count": None, + "is_warmup": False, + "block_id": None, + "variant_id": None, + "order_in_block": None, + "phase": None, + "dimensions": dimensions, + "evidence_level": "observed", + } + + @staticmethod + def _observation_row( + run_id: str, + artifact_id: str, + index: int, + observation: dict[str, Any], + ) -> dict[str, object]: + kind = observation.get("kind") + name = observation.get("name") + if not isinstance(kind, str) or not isinstance(name, str): + raise DomainError(ErrorCode.ARTIFACT_PARSE_FAILED, "Invalid profile observation.") + return { + "observation_id": digest_model( + {"artifact_id": artifact_id, "observation_index": index, "value": observation} + ), + "run_id": run_id, + "artifact_id": artifact_id, + "kind": kind, + "name": name, + "value_json": json.dumps( + observation.get("value"), allow_nan=False, separators=(",", ":"), sort_keys=True + ), + "file": None, + "line_from": None, + "line_to": None, + "context": "extractor_provenance" if kind == "profile.extraction" else None, + "evidence_level": "observed", + } + + +def _dict_list(value: object, name: str) -> list[dict[str, Any]]: + if not isinstance(value, list) or any(not isinstance(item, dict) for item in value): + raise DomainError(ErrorCode.ARTIFACT_PARSE_FAILED, f"Invalid ncu_report {name} payload.") + return value + + +def _string_list(value: object, name: str) -> list[str]: + if not isinstance(value, list) or any(not isinstance(item, str) for item in value): + raise DomainError(ErrorCode.ARTIFACT_PARSE_FAILED, f"Invalid ncu_report {name} payload.") + return value + + +def _nonnegative_int(value: object, name: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise DomainError(ErrorCode.ARTIFACT_PARSE_FAILED, f"Invalid ncu_report {name} value.") + return value diff --git a/src/flameox/adapters/options.py b/src/flameox/adapters/options.py index d20ebd2..2ced780 100644 --- a/src/flameox/adapters/options.py +++ b/src/flameox/adapters/options.py @@ -4,7 +4,7 @@ import os import stat from pathlib import Path -from typing import Annotated, Literal, cast +from typing import Annotated, Any, Literal, cast from pydantic import Field, JsonValue, StringConstraints, field_validator, model_validator @@ -61,6 +61,80 @@ class NvbenchOptions(ContractModel): devices: Annotated[str, StringConstraints(min_length=1, max_length=200)] | None = None +class NsightComputeOptions(ContractModel): + """Bounded selections accepted by the managed Nsight Compute capture.""" + + set: ( + Annotated[ + str, + StringConstraints(min_length=1, max_length=100, pattern=r"^[A-Za-z0-9_.-]+$"), + ] + | None + ) = "basic" + sections: ( + Annotated[ + tuple[ + Annotated[ + str, + StringConstraints( + min_length=1, + max_length=100, + pattern=r"^[A-Za-z0-9_.-]+$", + ), + ], + ..., + ], + Field(min_length=1, max_length=32), + ] + | None + ) = None + kernel_name: BoundedFilter | None = None + launch_skip: Annotated[int, Field(ge=0, le=1_000_000)] = 0 + launch_count: Annotated[int, Field(ge=1, le=1_000_000)] = 1 + replay_mode: Literal["kernel", "application", "range", "app-range"] = "kernel" + + @model_validator(mode="before") + @classmethod + def explicit_sections_replace_default_set(cls, value: Any) -> Any: + if isinstance(value, dict) and value.get("sections") is not None and "set" not in value: + return {**value, "set": None} + return value + + @model_validator(mode="after") + def one_profile_selection(self) -> NsightComputeOptions: + if (self.set is None) == (self.sections is None): + raise ValueError("exactly one of set or sections must be selected") + if self.sections is not None and len(set(self.sections)) != len(self.sections): + raise ValueError("sections must not contain duplicates") + return self + + +class Rocprofv3Options(ContractModel): + """Trace domains exposed by rocprofv3's documented PFTrace CLI.""" + + hip_trace: bool = False + kernel_trace: bool = True + memory_copy_trace: bool = False + memory_allocation_trace: bool = False + scratch_memory_trace: bool = False + marker_trace: bool = False + + @model_validator(mode="after") + def at_least_one_domain(self) -> Rocprofv3Options: + if not any( + ( + self.hip_trace, + self.kernel_trace, + self.memory_copy_trace, + self.memory_allocation_trace, + self.scratch_memory_trace, + self.marker_trace, + ) + ): + raise ValueError("at least one rocprofv3 trace domain must be enabled") + return self + + _BoundedSubdir = Annotated[ str, StringConstraints(min_length=1, max_length=200, pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]*$"), @@ -117,7 +191,9 @@ def unique_and_consistent_allowlist(self) -> CuteCompilerOptions: _ADAPTER_OPTION_MODELS: dict[str, type[ContractModel]] = { "compute-sanitizer": ComputeSanitizerOptions, "cute.compiler": CuteCompilerOptions, + "nsight.compute": NsightComputeOptions, "nvbench": NvbenchOptions, + "rocprofv3": Rocprofv3Options, "triton.compiler": TritonCompilerOptions, } @@ -252,6 +328,14 @@ def nvbench_options(options: dict[str, object] | None) -> NvbenchOptions: return cast(NvbenchOptions, _validate_adapter_options("nvbench", options)) +def nsight_compute_options(options: dict[str, object] | None) -> NsightComputeOptions: + return cast(NsightComputeOptions, _validate_adapter_options("nsight.compute", options)) + + +def rocprofv3_options(options: dict[str, object] | None) -> Rocprofv3Options: + return cast(Rocprofv3Options, _validate_adapter_options("rocprofv3", options)) + + def triton_compiler_options(options: dict[str, object] | None) -> TritonCompilerOptions: return cast(TritonCompilerOptions, _validate_adapter_options("triton.compiler", options)) diff --git a/src/flameox/application/capabilities.py b/src/flameox/application/capabilities.py index ee272dc..21128a1 100644 --- a/src/flameox/application/capabilities.py +++ b/src/flameox/application/capabilities.py @@ -2,6 +2,7 @@ import asyncio import json +import os import platform import re import shutil @@ -20,6 +21,7 @@ from platformdirs import user_data_path from flameox.adapters.builtins import BUILTIN_ADAPTERS, BuiltinAdapter, builtin_adapter +from flameox.adapters.nsight_compute import find_ncu_report_interface from flameox.adapters.registry import AdapterRegistry from flameox.adapters.setup_runtime import install_trace_processor from flameox.adapters.toxiproxy import ToxiproxyClient, ToxiproxyToolManager, ToxiproxyToolReceipt @@ -220,7 +222,7 @@ def _list(self, *, recommendation_adapter: str | None = None) -> CapabilityList: permissions=adapter.permissions, permission_status=( "unknown_until_active_probe" - if adapter.name in {"py-spy", "perf"} and resolved + if adapter.name in {"py-spy", "perf", "nsight.compute"} and resolved else None ), restrictions=self._platform_restrictions(adapter), @@ -403,6 +405,10 @@ async def probe(self, adapter: str, *, refresh: bool = False) -> CapabilityRepor result = result.model_copy(update={"version": version_report.version}) self._active_cache[adapter] = result return result + if adapter == "nsight.compute": + result = await self._probe_nsight_compute(version_report) + self._active_cache[adapter] = result + return result result = version_report if adapter == "py-spy": result = result.model_copy(update={"permission_status": "not_exercised"}) @@ -432,14 +438,23 @@ async def _probe_version( "utf-8", errors="replace", ) - first_line = next((line.strip() for line in output.splitlines() if line.strip()), None) + lines = tuple(line.strip() for line in output.splitlines() if line.strip()) + version_line = next( + ( + line + for line in lines + if passive.adapter in {"compute-sanitizer", "nsight.compute"} + and line.casefold().startswith("version ") + ), + lines[0] if lines else None, + ) succeeded = outcome.process.exit_code == 0 return passive.model_copy( update={ "status": CapabilityStatus.AVAILABLE if succeeded else CapabilityStatus.DEGRADED, - "version": first_line or passive.version, + "version": version_line or passive.version, "limitations": ( () if succeeded @@ -542,6 +557,169 @@ async def _probe_perf(self, passive: CapabilityReport) -> CapabilityReport: return self._perf_permission_failure(passive, diagnostic) return self._perf_failure(passive, diagnostic or "perf sampling probe failed.") + async def _probe_nsight_compute(self, passive: CapabilityReport) -> CapabilityReport: + """Check the shipped report interface and map NVIDIA's counter diagnostic.""" + if self.workspace is None or passive.executable is None: + return passive.model_copy(update={"permission_status": "not_exercised"}) + if find_ncu_report_interface(executable=passive.executable) is None: + return passive.model_copy( + update={ + "status": CapabilityStatus.DEGRADED, + "permission_status": "unknown", + "limitations": ("The official ncu_report Python interface is unavailable.",), + "remediation": ( + "Install the Nsight Compute extras/python interface; FlameOx does not " + "decode NVIDIA report binaries.", + ), + "probe_kind": "active", + "probed_at": utc_now(), + } + ) + restriction = self._nvidia_counter_access_restriction() + if restriction is not None: + return self._nsight_compute_permission_failure(passive, restriction) + staging_root = self.workspace.paths.staging + staging_root.mkdir(parents=True, exist_ok=True) + try: + with tempfile.TemporaryDirectory( + dir=staging_root, prefix="capability-ncu-" + ) as temporary: + output = Path(temporary) / "probe.ncu-rep" + request = ExecutionRequest( + argv=( + passive.executable, + "--set", + "basic", + "--launch-count", + "1", + "--export", + str(output), + "--force-overwrite", + sys.executable, + "-I", + "-S", + "-c", + "pass", + ), + cwd=self.workspace.project_root, + environment_allowlist=(), + allowed_working_roots=(self.workspace.project_root,), + timeout_seconds=10, + max_output_bytes=self.workspace.config.execution.max_output_bytes, + resource_policy=ResourcePolicy( + filesystem_path=self.workspace.paths.root, + staging_root=Path(temporary), + minimum_free_bytes=self.workspace.config.storage.min_free_bytes, + sampling_interval_ms=( + self.workspace.config.execution.resource_sampling_interval_ms + ), + max_observed_files=( + self.workspace.config.execution.max_resource_observed_files + ), + ), + ) + try: + outcome = await self.broker.run(request) + except DomainError as error: + process = error.details.get("process") + diagnostic = error.message + if isinstance(process, dict): + diagnostic = ( + " ".join( + str(value) + for value in (process.get("stdout"), process.get("stderr")) + if value + ) + or diagnostic + ) + diagnostic = self._bounded_diagnostic(diagnostic) + if "ERR_NVGPUCTRPERM" in diagnostic: + return self._nsight_compute_permission_failure(passive, diagnostic) + return self._nsight_compute_probe_failure(passive, diagnostic) + except (OSError, ValueError) as error: + return self._nsight_compute_probe_failure(passive, str(error)) + diagnostic = self._bounded_diagnostic( + (outcome.stdout + b"\n" + outcome.stderr).decode("utf-8", errors="replace") + ) + if "ERR_NVGPUCTRPERM" in diagnostic: + return self._nsight_compute_permission_failure(passive, diagnostic) + if outcome.process.exit_code != 0: + return self._nsight_compute_probe_failure(passive, diagnostic) + return passive.model_copy( + update={ + "permission_status": "not_exercised", + "limitations": ( + "The official report interface is available, but the bounded probe did not " + "execute a CUDA kernel; counter permission remains unexercised.", + ), + "probe_kind": "active", + "probed_at": utc_now(), + } + ) + + @staticmethod + def _nsight_compute_permission_failure( + passive: CapabilityReport, + diagnostic: str, + ) -> CapabilityReport: + return passive.model_copy( + update={ + "status": CapabilityStatus.PERMISSION_REQUIRED, + "permission_status": "denied", + "limitations": (diagnostic,), + "remediation": ( + "Enable NVIDIA GPU performance-counter access following NVIDIA's " + "ERR_NVGPUCTRPERM guidance, then refresh capabilities; FlameOx will not " + "change system privileges.", + ), + "probe_kind": "active", + "probed_at": utc_now(), + } + ) + + @staticmethod + def _nvidia_counter_access_restriction() -> str | None: + """Read NVIDIA's Linux driver policy without attempting a privilege change.""" + parameters = Path("/proc/driver/nvidia/params") + if not parameters.is_file() or os.geteuid() == 0: + return None + try: + admin_only = any( + line.strip() == "RmProfilingAdminOnly: 1" + for line in parameters.read_text(encoding="utf-8").splitlines() + ) + status_lines = Path("/proc/self/status").read_text(encoding="utf-8").splitlines() + effective = next( + int(line.split(":", 1)[1].strip(), 16) + for line in status_lines + if line.startswith("CapEff:") + ) + except (OSError, StopIteration, ValueError): + return None + cap_sys_admin = bool(effective & (1 << 21)) + if admin_only and not cap_sys_admin: + return ( + "ERR_NVGPUCTRPERM: NVIDIA driver reports RmProfilingAdminOnly=1 and this " + "process lacks CAP_SYS_ADMIN." + ) + return None + + @staticmethod + def _nsight_compute_probe_failure( + passive: CapabilityReport, + diagnostic: str, + ) -> CapabilityReport: + return passive.model_copy( + update={ + "status": CapabilityStatus.DEGRADED, + "permission_status": "unknown", + "limitations": (diagnostic or "Nsight Compute active probe failed.",), + "remediation": ("Inspect the bounded Nsight Compute probe diagnostic.",), + "probe_kind": "active", + "probed_at": utc_now(), + } + ) + def _perf_permission_failure( self, passive: CapabilityReport, diff --git a/src/flameox/cli.py b/src/flameox/cli.py index c34c250..05f5c51 100644 --- a/src/flameox/cli.py +++ b/src/flameox/cli.py @@ -24,6 +24,7 @@ InferenceArtifactExtractor, KernelValidationExtractor, MemrayExtractor, + NsightComputeExtractor, NsightSystemsExtractor, NvbenchExtractor, ObservationExtractor, @@ -2397,6 +2398,23 @@ def extract_nvbench( _emit(result, as_json=json_output) +@extract_app.command("nsight-compute") +def extract_nsight_compute( + run_id: Annotated[ + str, + typer.Argument(help="Run containing an unchanged .ncu-rep or .ncu-repz report."), + ], + workspace: WorkspaceOption = None, + json_output: JsonOption = False, +) -> None: + """Extract bounded metrics through NVIDIA's installed ncu_report interface.""" + try: + result = NsightComputeExtractor(_workspace(workspace)).extract(run_id) + except DomainError as error: + _fail(error) + _emit(result, as_json=json_output) + + @extract_app.command("inference-trace") def extract_inference_trace( run_id: Annotated[str, typer.Argument(help="Import run containing Mooncake JSONL.")], diff --git a/src/flameox/domain/models.py b/src/flameox/domain/models.py index b7c3fc8..d90e75d 100644 --- a/src/flameox/domain/models.py +++ b/src/flameox/domain/models.py @@ -104,6 +104,7 @@ class ArtifactKind(StrEnum): INFERENCE_REQUEST_TRACE = "inference_request_trace" INFERENCE_RESULT = "inference_result" KERNEL_BUILD = "kernel_build" + KERNEL_PROFILE = "kernel_profile" class RunType(StrEnum): diff --git a/src/flameox/mcp/server.py b/src/flameox/mcp/server.py index a883d98..8ef54ef 100644 --- a/src/flameox/mcp/server.py +++ b/src/flameox/mcp/server.py @@ -31,6 +31,8 @@ KernelValidationExtractor, MemrayExtractionResult, MemrayExtractor, + NsightComputeExtractionResult, + NsightComputeExtractor, NsightSystemsExtractionResult, NsightSystemsExtractor, NvbenchExtractionResult, @@ -2088,6 +2090,8 @@ async def import_artifact_tool( "memray", "coverage", "compute-sanitizer", + "nsight.compute", + "rocprofv3", "pyperf", "pytest", "aiperf", @@ -3401,6 +3405,42 @@ async def extract_nvbench_tool( except DomainError as error: return _failure(error) + @server.tool(name="extract_nsight_compute", annotations=ADDITIVE) + async def extract_nsight_compute_tool( + run_id: Annotated[str, Field(min_length=1, max_length=200)], + ctx: Context[AppContext], + ) -> Annotated[CallToolResult, ToolPayload[NsightComputeExtractionResult]]: + """Extract bounded metrics through NVIDIA's installed ncu_report interface.""" + try: + await ctx.report_progress(0, 2, "Nsight Compute extraction started") + result = await run_atomic_thread( + lambda: NsightComputeExtractor( + ctx.request_context.lifespan_context.require_workspace() + ).extract(run_id) + ) + await ctx.report_progress(1, 2, "Nsight Compute evidence published") + await ctx.report_progress(2, 2, "Nsight Compute result ready") + return _success( + result, + f"Extracted {result.metric_count} Nsight Compute metrics.", + resource_links=( + ResourceLink( + name=f"Run {result.run_id}", + uri=f"flameox://runs/{result.run_id}", + description="Authoritative Nsight Compute run manifest.", + mime_type="application/json", + ), + ResourceLink( + name=f"Artifact {result.artifact_id}", + uri=f"flameox://artifacts/{result.artifact_id}", + description="Authoritative native Nsight Compute report metadata.", + mime_type="application/json", + ), + ), + ) + except DomainError as error: + return _failure(error) + @server.tool(name="extract_inference_trace", annotations=ADDITIVE) async def extract_inference_trace_tool( run_id: Annotated[str, Field(min_length=1, max_length=200)], diff --git a/src/flameox/workers/nsight_compute.py b/src/flameox/workers/nsight_compute.py new file mode 100644 index 0000000..4a88e0c --- /dev/null +++ b/src/flameox/workers/nsight_compute.py @@ -0,0 +1,386 @@ +from __future__ import annotations + +import argparse +import importlib +import json +import math +import sys +from collections.abc import Iterable +from itertools import islice +from pathlib import Path +from typing import Any, cast + + +def _bounded_text(value: object, limit: int = 2_000) -> str: + return str(value)[:limit] + + +def _safe_json(value: object, *, depth: int = 0) -> object: + if depth >= 5: + return _bounded_text(value) + if value is None or isinstance(value, bool | int | str): + return value if not isinstance(value, str) else value[:2_000] + if isinstance(value, float): + return value if math.isfinite(value) else _bounded_text(value) + if isinstance(value, dict): + return { + _bounded_text(key, 200): _safe_json(item, depth=depth + 1) + for key, item in islice(value.items(), 100) + } + if isinstance(value, list | tuple): + return [_safe_json(item, depth=depth + 1) for item in value[:100]] + try: + return [ + _safe_json(item, depth=depth + 1) for item in islice(cast(Iterable[object], value), 100) + ] + except (TypeError, ValueError): + return _bounded_text(value) + + +def _call( + value: object, + name: str, + default: object = None, + *, + limitations: list[str] | None = None, +) -> Any: + method = getattr(value, name, None) + if not callable(method): + return default + try: + return method() + except Exception as exc: # Official bindings can throw SWIG-specific exception classes. + if limitations is not None: + limitations.append( + f"Official ncu_report access {name} failed with {type(exc).__name__}." + ) + return default + + +def _metric_value( + metric: object, + *, + limitations: list[str], +) -> tuple[str, int | float | str | None]: + has_value = _call(metric, "has_value", True, limitations=limitations) + if has_value is False: + return "missing", None + kind = _call(metric, "kind", limitations=limitations) + if kind in { + getattr(metric, "ValueKind_UINT32", object()), + getattr(metric, "ValueKind_UINT64", object()), + }: + value = _call(metric, "as_uint64", limitations=limitations) + return ( + ("integer", value) + if isinstance(value, int) and not isinstance(value, bool) + else ( + "unknown", + None, + ) + ) + if kind in { + getattr(metric, "ValueKind_FLOAT", object()), + getattr(metric, "ValueKind_DOUBLE", object()), + }: + value = _call(metric, "as_double", limitations=limitations) + return ( + ("float", value) + if isinstance(value, int | float) and math.isfinite(value) + else ( + "unknown", + None, + ) + ) + if kind == getattr(metric, "ValueKind_STRING", object()): + value = _call(metric, "as_string", limitations=limitations) + return ("string", _bounded_text(value)) if value is not None else ("missing", None) + value = _call(metric, "value", limitations=limitations) + if isinstance(value, int) and not isinstance(value, bool): + return "integer", value + if isinstance(value, float) and math.isfinite(value): + return "float", value + if isinstance(value, str): + return "string", value[:2_000] + return "unknown", None + + +def _extract( # noqa: C901 - bounded traversal mirrors the official report hierarchy + report_path: Path, + *, + interface_path: Path, + max_ranges: int, + max_actions: int, + max_metrics: int, + max_observations: int, +) -> dict[str, object]: + if interface_path.name == "ncu_report.py": + sys.path.insert(0, str(interface_path.parent)) + else: + sys.path.insert(0, str(interface_path)) + module = importlib.import_module("ncu_report") + report = module.load_report(str(report_path)) + if report is None: + raise ValueError("ncu_report.load_report returned no report") + + measurements: list[dict[str, object]] = [] + observations: list[dict[str, object]] = [] + limitations: list[str] = [] + section_ids: set[str] = set() + metric_ids: set[str] = set() + roofline_present = False + range_total = int(report.num_ranges()) + action_seen = 0 + total_actions = 0 + metric_seen = 0 + + for range_index in range(min(range_total, max_ranges)): + selected_range = report.range_by_idx(range_index) + action_total = int(selected_range.num_actions()) + total_actions += action_total + range_name = _call( + selected_range, + "name", + f"range-{range_index}", + limitations=limitations, + ) + if len(observations) < max_observations: + observations.append( + { + "kind": "profile.range", + "name": _bounded_text(range_name, 500), + "value": {"range_index": range_index, "action_count": action_total}, + } + ) + for action_index in range(min(action_total, max_actions - action_seen)): + action = selected_range.action_by_idx(action_index) + action_seen += 1 + action_name = _bounded_text(action.name(), 500) + workload_type = _call(action, "workload_type", limitations=limitations) + action_identity: dict[str, object] = {} + action_value = { + "range_index": range_index, + "range_name": _bounded_text(range_name, 500), + "action_index": action_index, + "workload_type": _safe_json(workload_type), + "identity": action_identity, + } + if len(observations) < max_observations: + observations.append( + { + "kind": "profile.action", + "name": action_name, + "value": action_value, + } + ) + + remaining_metrics = max_metrics - metric_seen + raw_metric_names = action.metric_names() or () + metric_names = tuple( + str(item) for item in islice(raw_metric_names, remaining_metrics + 1) + ) + for metric_name in metric_names[:remaining_metrics]: + metric = action.metric_by_name(metric_name) + value_kind, value = _metric_value(metric, limitations=limitations) + metric_seen += 1 + metric_ids.add(metric_name) + unit = _bounded_text( + _call(metric, "unit", "", limitations=limitations) or "unknown", 100 + ) + lower_identity = metric_name.casefold() + roofline_present = roofline_present or "roofline" in lower_identity + if ( + any( + marker in lower_identity + for marker in ( + "device__attribute", + "context_id", + "stream_id", + "process_id", + "launch__grid", + "launch__block", + ) + ) + and value is not None + ): + action_identity[metric_name[:500]] = value + common = { + "range_index": range_index, + "action_index": action_index, + "action_name": action_name, + "name": metric_name[:500], + "unit": unit, + "description": _bounded_text( + _call(metric, "description", "", limitations=limitations), 2_000 + ), + "metric_type": _safe_json( + _call(metric, "metric_type", limitations=limitations) + ), + "metric_subtype": _safe_json( + _call(metric, "metric_subtype", limitations=limitations) + ), + "rollup_operation": _safe_json( + _call(metric, "rollup_operation", limitations=limitations) + ), + "roofline": "roofline" in lower_identity, + } + if value_kind in {"integer", "float"} and isinstance(value, int | float): + measurements.append({**common, "value_kind": value_kind, "value": value}) + elif value_kind == "string" and len(observations) < max_observations: + observations.append( + { + "kind": "profile.attribute", + "name": metric_name[:500], + "value": {**common, "value": value}, + } + ) + elif value_kind == "unknown": + limitations.append(f"Unsupported metric value type: {metric_name[:200]}.") + if len(metric_names) > remaining_metrics: + limitations.append(f"Metrics were truncated to {max_metrics} entries.") + + rules = _call(action, "rule_results_as_dicts", (), limitations=limitations) or () + rule_budget = max(0, max_observations - len(observations)) + bounded_rules = tuple(islice(cast(Iterable[object], rules), rule_budget + 1)) + for rule in bounded_rules[:rule_budget]: + normalized = _safe_json(rule) + if isinstance(normalized, dict): + section = normalized.get("section_identifier") + if isinstance(section, str): + section_ids.add(section) + roofline_present = roofline_present or "roofline" in section.casefold() + identifier = normalized.get("rule_identifier") + name = identifier if isinstance(identifier, str) else "rule" + else: + name = "rule" + observations.append({"kind": "profile.rule", "name": name, "value": normalized}) + if len(bounded_rules) > rule_budget: + limitations.append("Rule results were truncated by the observation budget.") + + source_files = _call(action, "source_files", {}, limitations=limitations) or {} + try: + source_names = [ + _bounded_text(item, 500) + for item in islice(cast(Iterable[object], source_files), 100) + ] + except TypeError: + source_names = [] + if source_names and len(observations) < max_observations: + observations.append( + { + "kind": "profile.source_files", + "name": action_name, + # Preserve identities only. Source bodies remain in the native report and + # are not imported into normalized evidence. + "value": {"files": source_names}, + } + ) + markers = _call(action, "source_markers", (), limitations=limitations) or () + marker_budget = max(0, max_observations - len(observations)) + for marker in islice(cast(Iterable[object], markers), marker_budget): + normalized_marker = _safe_json(marker) + observations.append( + { + "kind": "profile.source_reference", + "name": action_name, + "value": normalized_marker, + } + ) + if not isinstance(marker, dict): + continue + address = marker.get("source_address") + if not isinstance(address, int) or len(observations) >= max_observations: + continue + reference = { + "address": address, + "source": _safe_json( + _call_with_arg(action, "source_info", address, limitations=limitations) + ), + "sass": _safe_json( + _call_with_arg(action, "sass_by_pc", address, limitations=limitations) + ), + "ptx": _safe_json( + _call_with_arg(action, "ptx_by_pc", address, limitations=limitations) + ), + } + observations.append( + { + "kind": "profile.source_sass_reference", + "name": action_name, + "value": reference, + } + ) + if action_seen >= max_actions: + break + + if range_total > max_ranges: + limitations.append(f"Ranges were truncated to {max_ranges} entries.") + if total_actions > max_actions: + limitations.append(f"Actions were bounded to {max_actions} entries.") + if len(observations) >= max_observations: + limitations.append(f"Observations were bounded to {max_observations} entries.") + return { + "ok": True, + "report_version": _bounded_text(report.get_version(), 200), + "measurements": measurements, + "observations": observations, + "metric_ids": sorted(metric_ids), + "section_ids": sorted(section_ids), + "range_count": min(range_total, max_ranges), + "action_count": action_seen, + "roofline_present": roofline_present, + "limitations": list(dict.fromkeys(limitations)), + } + + +def _call_with_arg( + value: object, + name: str, + argument: object, + *, + limitations: list[str] | None = None, +) -> Any: + method = getattr(value, name, None) + if not callable(method): + return None + try: + return method(argument) + except Exception as exc: # Official bindings can throw SWIG-specific exception classes. + if limitations is not None: + limitations.append( + f"Official ncu_report access {name} failed with {type(exc).__name__}." + ) + return None + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--request", required=True, type=Path) + parser.add_argument("--response", required=True, type=Path) + args = parser.parse_args() + try: + request = json.loads(args.request.read_text(encoding="utf-8")) + response = _extract( + Path(request["artifact_path"]), + interface_path=Path(request["interface_path"]), + max_ranges=int(request["max_ranges"]), + max_actions=int(request["max_actions"]), + max_metrics=int(request["max_metrics"]), + max_observations=int(request["max_observations"]), + ) + except Exception as exc: + response = { + "ok": False, + "code": "ARTIFACT_PARSE_FAILED", + "message": f"Nsight Compute report extraction failed: {type(exc).__name__}: {exc}", + } + temporary = args.response.with_suffix(".tmp") + temporary.write_text(json.dumps(response, allow_nan=False, sort_keys=True), encoding="utf-8") + temporary.replace(args.response) + # The response envelope, not the process status, carries parser failures so the + # parent can preserve the bounded provider diagnostic. + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/adapters/test_nsight_compute.py b/tests/adapters/test_nsight_compute.py new file mode 100644 index 0000000..24e552b --- /dev/null +++ b/tests/adapters/test_nsight_compute.py @@ -0,0 +1,220 @@ +# ruff: noqa: E501 - the fake official interface is embedded source, not product code +from __future__ import annotations + +from pathlib import Path +from typing import cast + +import pytest + +from flameox.adapters.builtins import build_capture_invocation +from flameox.adapters.nsight_compute import NsightComputeExtractor +from flameox.adapters.options import bind_adapter_options +from flameox.application import ImportArtifactRequest, ImportService +from flameox.domain import ArtifactKind, DomainError, ErrorCode +from flameox.storage import ArtifactStore, RunStore, Workspace + + +def _fake_interface(path: Path, *, corrupt: bool = False) -> Path: + path.mkdir() + body = """ +class Metric: + ValueKind_UINT32 = 1 + ValueKind_UINT64 = 2 + ValueKind_FLOAT = 3 + ValueKind_DOUBLE = 4 + ValueKind_STRING = 5 + def __init__(self, name, kind, value, unit): self._name, self._kind, self._value, self._unit = name, kind, value, unit + def has_value(self): return True + def kind(self): return self._kind + def as_uint64(self): return int(self._value) + def as_double(self): return float(self._value) + def as_string(self): return str(self._value) + def value(self): return self._value + def unit(self): return self._unit + def description(self): return 'official metric description' + def metric_type(self): return 2 + def metric_subtype(self): + if self._name == 'unknown': raise RuntimeError('optional subtype unavailable') + return None + def rollup_operation(self): return 1 + +class Action: + def __init__(self): + self.metrics = { + 'sm__cycles_elapsed.avg': Metric('cycles', 2, 42, 'cycle'), + 'device__attribute_display_name': Metric('device', 5, 'Fake GPU', ''), + 'unsupported.metric': Metric('unknown', 99, object(), ''), + } + def name(self): return 'vector_add' + def workload_type(self): + def values(): + for index in range(100): yield index + raise AssertionError('provider iterable was consumed past its bound') + return values() + def metric_names(self): return tuple(self.metrics) + def metric_by_name(self, name): return self.metrics[name] + def rule_results_as_dicts(self): + return [{'rule_identifier': 'ExplicitRoofline', 'section_identifier': 'SpeedOfLight_RooflineChart', 'result_table': {'rows': [1, 2]}}] + def source_files(self): return {'0': 'kernel.cu'} + def source_markers(self): return [{'source_address': 1234, 'message': 'source marker'}] + def source_info(self, address): return {'file': 'kernel.cu', 'line': 7, 'address': address} + def sass_by_pc(self, address): return {'address': address, 'instruction': 'LDG.E'} + def ptx_by_pc(self, address): return {'address': address, 'instruction': 'ld.global'} + +class Range: + def num_actions(self): return 1 + def action_by_idx(self, index): return Action() + +class Report: + def get_version(self): return '2099.1' + def num_ranges(self): return 1 + def range_by_idx(self, index): return Range() + +def load_report(path): + return Report() +""" + if corrupt: + body = "def load_report(path):\n raise RuntimeError('corrupt official report')\n" + interface = path / "ncu_report.py" + interface.write_text(body, encoding="utf-8") + return interface + + +def _import_report(workspace: Workspace, path: Path) -> str: + return ( + ImportService(workspace) + .import_artifact( + ImportArtifactRequest( + path=path, + kind=ArtifactKind.KERNEL_PROFILE, + producer="nsight.compute", + producer_version="2099.1", + ) + ) + .run.run_id + ) + + +def test_strict_options_build_only_documented_bounded_arguments(tmp_path: Path) -> None: + selected = bind_adapter_options( + "nsight.compute", + { + "sections": ["LaunchStats", "SpeedOfLight"], + "kernel_name": "vector_add", + "launch_skip": 2, + "launch_count": 3, + "replay_mode": "application", + }, + project_root=tmp_path, + ) + invocation = build_capture_invocation( + "nsight.compute", + ("./workload",), + tmp_path, + executable="/opt/nvidia/ncu", + options=cast(dict[str, object], selected), + project_root=tmp_path, + ) + + assert invocation.artifact_kinds == (ArtifactKind.KERNEL_PROFILE,) + assert invocation.argv == ( + "/opt/nvidia/ncu", + "--export", + str(tmp_path / "nsight-compute.ncu-rep"), + "--force-overwrite", + "--replay-mode", + "application", + "--launch-skip", + "2", + "--launch-count", + "3", + "--section", + "LaunchStats", + "--section", + "SpeedOfLight", + "--kernel-name-base", + "demangled", + "--kernel-name", + "vector_add", + "./workload", + ) + + with pytest.raises(DomainError) as unknown: + bind_adapter_options( + "nsight.compute", + {"set": "basic", "arbitrary_flags": ["--import"]}, + project_root=tmp_path, + ) + assert unknown.value.code is ErrorCode.INVALID_CAPTURE_PLAN + with pytest.raises(DomainError): + bind_adapter_options( + "nsight.compute", + {"set": None, "sections": ["regex:.*"]}, + project_root=tmp_path, + ) + + +def test_extractor_uses_fake_official_interface_in_isolated_worker( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = Workspace.initialize(tmp_path) + source = tmp_path / "fake.ncu-rep" + source.write_bytes(b"opaque NVIDIA report bytes") + before = source.read_bytes() + run_id = _import_report(workspace, source) + interface = _fake_interface(tmp_path / "official-interface") + monkeypatch.setattr( + "flameox.adapters.nsight_compute.find_ncu_report_interface", + lambda **_: interface, + ) + + result = NsightComputeExtractor(workspace).extract(run_id) + repeated = NsightComputeExtractor(workspace).extract(run_id) + + assert result.report_version == "2099.1" + assert result.range_count == 1 + assert result.action_count == 1 + assert result.metric_count == 1 + assert result.observation_count >= 5 + assert result.roofline_present is True + assert any("Unsupported metric value type" in item for item in result.limitations) + assert any("metric_subtype failed with RuntimeError" in item for item in result.limitations) + assert repeated.corpus_commit_id == result.corpus_commit_id + registration = next(item for item in RunStore(workspace).read(run_id).artifacts) + assert ( + ArtifactStore(workspace).get(registration.artifact_id).payload_path.read_bytes() == before + ) + + +def test_corrupt_report_from_official_interface_is_bounded_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = Workspace.initialize(tmp_path) + source = tmp_path / "corrupt.ncu-rep" + source.write_bytes(b"opaque") + run_id = _import_report(workspace, source) + interface = _fake_interface(tmp_path / "corrupt-interface", corrupt=True) + monkeypatch.setattr( + "flameox.adapters.nsight_compute.find_ncu_report_interface", + lambda **_: interface, + ) + + with pytest.raises(DomainError) as failure: + NsightComputeExtractor(workspace).extract(run_id) + + assert failure.value.code is ErrorCode.ARTIFACT_PARSE_FAILED + assert "RuntimeError" in failure.value.message + + +def test_rejects_non_native_report_extension(tmp_path: Path) -> None: + workspace = Workspace.initialize(tmp_path) + source = tmp_path / "report.bin" + source.write_bytes(b"opaque") + run_id = _import_report(workspace, source) + + with pytest.raises(DomainError) as failure: + NsightComputeExtractor(workspace).extract(run_id) + + assert failure.value.code is ErrorCode.ARTIFACT_PARSE_FAILED diff --git a/tests/adapters/test_nsight_compute_live.py b/tests/adapters/test_nsight_compute_live.py new file mode 100644 index 0000000..5bb892b --- /dev/null +++ b/tests/adapters/test_nsight_compute_live.py @@ -0,0 +1,103 @@ +import json +import os +from pathlib import Path + +import pytest + +from flameox.adapters.nsight_compute import NsightComputeExtractor +from flameox.application import ( + CapabilityService, + CaptureService, + ExecutionPolicy, + ImportArtifactRequest, + ImportService, +) +from flameox.domain import ( + ArtifactKind, + CapabilityStatus, + CaptureStatus, + ExecutionStatus, +) +from flameox.storage import Workspace +from tests.support.capture import disable_containment + + +@pytest.mark.anyio +@pytest.mark.optional +@pytest.mark.requires_ncu +@pytest.mark.requires_nvbench +async def test_managed_capture_collects_and_extracts_live_counters(tmp_path: Path) -> None: + workload = os.environ.get("FLAMEOX_NVBENCH_EXECUTABLE") + assert workload is not None + workspace = Workspace.initialize(tmp_path) + disable_containment(workspace) + (tmp_path / "flameox.toml").write_text( + f""" +schema_version = 1 +[workloads.profile] +argv = [{json.dumps(workload)}, "--timeout", "0.1", "--min-time", "1e-5"] +timeout_seconds = 120 +""" + ) + + service = CaptureService(workspace) + plan = await service.plan( + workload_name="profile", + adapter="nsight.compute", + execution_policy=ExecutionPolicy.TRUSTED_LOCAL, + ) + result = await service.execute(plan.plan_id) + + assert result.run.execution_status is ExecutionStatus.SUCCEEDED + assert result.run.capture_status is CaptureStatus.REGISTERED + assert any( + registration.kind is ArtifactKind.KERNEL_PROFILE and registration.role == "primary" + for registration in result.run.artifacts + ) + extracted = NsightComputeExtractor(workspace).extract(result.run.run_id) + assert extracted.action_count >= 1 + assert extracted.metric_count >= 1 + + +@pytest.mark.optional +@pytest.mark.requires_ncu +def test_installed_official_interface_extracts_bundled_sample(tmp_path: Path) -> None: + sample = Path( + "/opt/nvidia/nsight-compute/2026.2.1/extras/samples/instructionMix/sobelFloat.ncu-rep" + ) + if not sample.is_file(): + pytest.skip("installed Nsight Compute sample report is unavailable") + workspace = Workspace.initialize(tmp_path) + run_id = ( + ImportService(workspace) + .import_artifact( + ImportArtifactRequest( + path=sample, + kind=ArtifactKind.KERNEL_PROFILE, + producer="nsight.compute", + producer_version="2026.2.1", + allow_external_path=True, + ) + ) + .run.run_id + ) + + result = NsightComputeExtractor(workspace).extract(run_id) + + assert result.report_version == "2026.2.1" + assert result.action_count >= 1 + assert result.metric_count >= 1 + + +@pytest.mark.anyio +@pytest.mark.optional +@pytest.mark.requires_ncu +async def test_local_driver_policy_reports_counter_permission_requirement(tmp_path: Path) -> None: + service = CapabilityService(Workspace.initialize(tmp_path)) + report = await service.probe("nsight.compute", refresh=True) + if report.status is not CapabilityStatus.PERMISSION_REQUIRED: + pytest.skip("local NVIDIA driver does not expose a restricted counter policy") + + assert report.status is CapabilityStatus.PERMISSION_REQUIRED + assert report.permission_status == "denied" + assert "ERR_NVGPUCTRPERM" in report.limitations[0] diff --git a/tests/adapters/test_rocprofv3.py b/tests/adapters/test_rocprofv3.py new file mode 100644 index 0000000..59e8078 --- /dev/null +++ b/tests/adapters/test_rocprofv3.py @@ -0,0 +1,243 @@ +from __future__ import annotations + +import os +import shutil +import subprocess +from pathlib import Path +from typing import cast + +import pytest + +from flameox.adapters import PerfettoExtractor +from flameox.adapters.builtins import build_capture_invocation +from flameox.adapters.options import bind_adapter_options +from flameox.analysis import RecipeService +from flameox.application import ( + CaptureService, + ExecutionPolicy, + ImportArtifactRequest, + ImportService, +) +from flameox.domain import ( + ArtifactKind, + CaptureStatus, + DomainError, + ErrorCode, + ExecutionStatus, +) +from flameox.storage import ArtifactStore, Workspace +from tests.support.capture import disable_containment +from tests.support.providers import require_trace_processor + +_FIXTURE_ROOT = Path(__file__).parents[1] / "fixtures" / "rocprofv3" + + +def test_rocprofv3_invocation_uses_only_selected_pftrace_domains(tmp_path: Path) -> None: + bound = bind_adapter_options( + "rocprofv3", + { + "hip_trace": True, + "kernel_trace": False, + "memory_copy_trace": True, + "memory_allocation_trace": True, + "scratch_memory_trace": True, + "marker_trace": True, + }, + project_root=tmp_path, + ) + + invocation = build_capture_invocation( + "rocprofv3", + ("python", "workload.py"), + tmp_path / "capture", + executable="/opt/rocm/bin/rocprofv3", + options=cast(dict[str, object], bound), + ) + + assert invocation.argv == ( + "/opt/rocm/bin/rocprofv3", + "--output-format", + "pftrace", + "-o", + "rocprofv3", + "-d", + str(tmp_path / "capture"), + "--hip-trace", + "--memory-copy-trace", + "--memory-allocation-trace", + "--scratch-memory-trace", + "--marker-trace", + "--", + "python", + "workload.py", + ) + assert invocation.artifact_kinds == (ArtifactKind.EXECUTION_TRACE,) + + +def test_rocprofv3_options_reject_unknown_fields_and_empty_domain_set(tmp_path: Path) -> None: + with pytest.raises(DomainError) as unknown: + bind_adapter_options( + "rocprofv3", + {"arbitrary_flags": "--sys-trace"}, + project_root=tmp_path, + ) + assert unknown.value.code is ErrorCode.INVALID_CAPTURE_PLAN + + with pytest.raises(DomainError) as empty: + bind_adapter_options( + "rocprofv3", + { + "hip_trace": False, + "kernel_trace": False, + "memory_copy_trace": False, + "memory_allocation_trace": False, + "scratch_memory_trace": False, + "marker_trace": False, + }, + project_root=tmp_path, + ) + assert empty.value.code is ErrorCode.INVALID_CAPTURE_PLAN + + +def test_rocprofv3_process_fixture_writes_expected_pftrace_path(tmp_path: Path) -> None: + executable = _write_fake_rocprofv3(tmp_path / "rocprofv3", exit_code=0) + output = tmp_path / "output" + output.mkdir() + invocation = build_capture_invocation( + "rocprofv3", + ("/bin/true",), + output, + executable=str(executable), + options={"kernel_trace": True}, + ) + + completed = subprocess.run(invocation.argv, check=False, capture_output=True) + + assert completed.returncode == 0 + assert (output / "rocprofv3_results.pftrace").read_bytes() == b"fixture-pftrace" + assert completed.stdout == b"rocprof fixture stdout" + assert completed.stderr == b"rocprof fixture stderr" + + +@pytest.mark.anyio +async def test_rocprofv3_capture_preserves_partial_pftrace_on_nonzero_exit( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + executable_directory = tmp_path / "bin" + executable_directory.mkdir() + _write_fake_rocprofv3(executable_directory / "rocprofv3", exit_code=7) + monkeypatch.setenv("PATH", f"{executable_directory}{os.pathsep}{os.environ['PATH']}") + workspace = Workspace.initialize(tmp_path) + (tmp_path / "flameox.toml").write_text( + "schema_version = 1\n" + "[workloads.probe]\n" + "argv = ['/bin/true']\n" + "cwd = '.'\n" + "timeout_seconds = 5\n" + ) + disable_containment(workspace) + service = CaptureService(workspace) + plan = await service.plan( + workload_name="probe", + adapter="rocprofv3", + adapter_options={"kernel_trace": True, "hip_trace": True}, + execution_policy=ExecutionPolicy.TRUSTED_LOCAL, + ) + + result = await service.execute(plan.plan_id) + + assert result.run.execution_status is ExecutionStatus.FAILED + assert result.run.capture_status is CaptureStatus.REGISTERED + trace = next( + registration + for registration in result.run.artifacts + if registration.kind is ArtifactKind.EXECUTION_TRACE + ) + assert trace.producer == "rocprofv3" + assert ArtifactStore(workspace).get(trace.artifact_id).payload_path.read_bytes() == ( + b"fixture-pftrace" + ) + roles = {registration.role for registration in result.run.artifacts} + assert {"primary", "stdout", "stderr"} <= roles + assert any(detail.code == "nonzero_exit" for detail in result.run.limitation_details) + + +@pytest.mark.anyio +@pytest.mark.optional +@pytest.mark.requires_perfetto +async def test_project_owned_rocm_shaped_trace_reaches_accelerator_summary( + tmp_path: Path, +) -> None: + """Exercise the shared path without presenting this fixture as rocprofv3 output.""" + binary = require_trace_processor() + trace = tmp_path / "project-owned-rocm-shaped-perfetto.json" + shutil.copyfile(_FIXTURE_ROOT / trace.name, trace) + workspace = Workspace.initialize(tmp_path) + config = workspace.config.model_copy( + update={ + "analysis": workspace.config.analysis.model_copy( + update={"trace_processor_path": str(binary)} + ) + } + ) + workspace.paths.config.write_text(config.to_toml()) + imported = ImportService(workspace).import_artifact( + ImportArtifactRequest( + path=trace, + kind=ArtifactKind.EXECUTION_TRACE, + producer="project-owned-rocm-shaped-perfetto", + ) + ) + + extracted = await PerfettoExtractor(workspace).extract(imported.run.run_id) + summary = RecipeService(workspace).accelerator_launches( + imported.run.run_id, + phase="decode", + ) + + assert extracted.slice_count == 3 + assert extracted.trace_event_count == 3 + assert summary.coverage == { + "runtime_launches": True, + "accelerator_kernels": True, + "phase_annotations": True, + "correlation_ids": True, + "host_to_device_correlation": True, + "stream_identity": True, + } + assert summary.total == 1 + region = summary.regions[0] + assert region.direct_launch_count == 1 + assert region.kernel_count == 2 + assert region.kernel_duration_ns == 8_000 + assert region.correlated_kernel_count == 1 + assert region.idle_gap_total_ns == 3_000 + assert region.streams[0].device == "amd:0" + assert region.streams[0].stream == "7" + + +def _write_fake_rocprofv3(path: Path, *, exit_code: int) -> Path: + path.write_text( + "#!/bin/sh\n" + 'if [ "$1" = "--version" ]; then\n' + " printf 'rocprofv3 fixture 7.0\\n'\n" + " exit 0\n" + "fi\n" + "output_name=''\n" + "output_directory=''\n" + 'while [ "$#" -gt 0 ]; do\n' + ' case "$1" in\n' + " -o) output_name=$2; shift 2 ;;\n" + " -d) output_directory=$2; shift 2 ;;\n" + " --) break ;;\n" + " *) shift ;;\n" + " esac\n" + "done\n" + "printf 'fixture-pftrace' > \"$output_directory/${output_name}_results.pftrace\"\n" + "printf 'rocprof fixture stdout'\n" + "printf 'rocprof fixture stderr' >&2\n" + f"exit {exit_code}\n" + ) + path.chmod(0o755) + return path diff --git a/tests/application/test_nsight_compute_capability.py b/tests/application/test_nsight_compute_capability.py new file mode 100644 index 0000000..9edabcb --- /dev/null +++ b/tests/application/test_nsight_compute_capability.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from flameox.application import CapabilityService +from flameox.domain import CapabilityStatus, ProcessResult +from flameox.execution import ExecutionOutcome, ExecutionRequest, SubprocessBroker +from flameox.storage import Workspace + + +class _NcuProbeBroker(SubprocessBroker): + def __init__( + self, + *, + probe_stderr: bytes = ( + b"==ERROR== ERR_NVGPUCTRPERM - Permission to access GPU counters denied\n" + ), + ) -> None: + self.requests: list[ExecutionRequest] = [] + self.probe_stderr = probe_stderr + + async def run(self, request: ExecutionRequest, **_: Any) -> ExecutionOutcome: + self.requests.append(request) + if len(self.requests) == 1: + stdout = b"NVIDIA (R) Nsight Compute\nVersion 2026.2.1.0\n" + stderr = b"" + exit_code = 0 + else: + stdout = b"" + stderr = self.probe_stderr + exit_code = 1 + return ExecutionOutcome( + process=ProcessResult(exit_code=exit_code, cleanup_complete=True), + stdout=stdout, + stderr=stderr, + resolved_executable=Path(request.argv[0]), + containment="process_group", + ) + + +@pytest.mark.anyio +async def test_ncu_probe_maps_counter_denial_to_permission_required( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = Workspace.initialize(tmp_path) + broker = _NcuProbeBroker() + service = CapabilityService(workspace, broker=broker) + monkeypatch.setattr( + service, + "_resolved_executable", + lambda adapter, executable: "/usr/bin/ncu" if adapter == "nsight.compute" else None, + ) + monkeypatch.setattr( + "flameox.application.capabilities.find_ncu_report_interface", + lambda **_: Path("/opt/nvidia/nsight-compute/extras/python/ncu_report.py"), + ) + monkeypatch.setattr(service, "_nvidia_counter_access_restriction", lambda: None) + + report = await service.probe("nsight.compute") + + assert report.status is CapabilityStatus.PERMISSION_REQUIRED + assert report.permission_status == "denied" + assert report.version == "Version 2026.2.1.0" + assert "ERR_NVGPUCTRPERM" in report.limitations[0] + assert "will not change system privileges" in report.remediation[0] + assert broker.requests[1].argv[1:6] == ( + "--set", + "basic", + "--launch-count", + "1", + "--export", + ) + assert broker.requests[1].resource_policy is not None + assert broker.requests[1].resource_policy.staging_root is not None + assert not broker.requests[1].resource_policy.staging_root.exists() + + +@pytest.mark.anyio +async def test_ncu_probe_reports_non_permission_failure_as_degraded( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = Workspace.initialize(tmp_path) + service = CapabilityService( + workspace, + broker=_NcuProbeBroker(probe_stderr=b"==ERROR== CUDA driver initialization failed\n"), + ) + monkeypatch.setattr( + service, + "_resolved_executable", + lambda adapter, executable: "/usr/bin/ncu" if adapter == "nsight.compute" else None, + ) + monkeypatch.setattr( + "flameox.application.capabilities.find_ncu_report_interface", + lambda **_: Path("/opt/nvidia/nsight-compute/extras/python/ncu_report.py"), + ) + monkeypatch.setattr(service, "_nvidia_counter_access_restriction", lambda: None) + + report = await service.probe("nsight.compute") + + assert report.status is CapabilityStatus.DEGRADED + assert report.permission_status == "unknown" + assert "CUDA driver initialization failed" in report.limitations[0] diff --git a/tests/collection-baseline.toml b/tests/collection-baseline.toml index 0eb283f..edd150e 100644 --- a/tests/collection-baseline.toml +++ b/tests/collection-baseline.toml @@ -1,4 +1,4 @@ -expected_test_count = 1015 +expected_test_count = 1029 # New files are normalized to their pre-split path so this receipt proves # test identity and parametrized case preservation across the decomposition. @@ -44,8 +44,11 @@ expected_test_count = 1015 'tests/adapters/test_kernel_build.py' = { count = 13, digest = 'b6c8efe8c729909e6f3554c1b83554e62f5cf289728588e2c6b4a3877d59637a' } 'tests/adapters/test_kernel_build_capture.py' = { count = 23, digest = 'ba3671771c7bffa170869d7fca5e86a2e098603d461788bccc7e931eb6e7431a' } 'tests/adapters/test_kernel_validation.py' = { count = 12, digest = '86a12a5fb7f6a3cf94a68be67fd68cb1ea34736f5be7a68a7bee2b9145bf0f37' } +'tests/adapters/test_nsight_compute.py' = { count = 4, digest = '3135414e6ce11321c6fc2aec2c7fbbf06343eeca695c72b4a0bf48811df229ca' } +'tests/adapters/test_nsight_compute_live.py' = { count = 3, digest = 'bf6c37d9ff8691a3624795dfd1de809e407e2b5b7cc4bfc8522c5b147036d18f' } 'tests/adapters/test_nvbench.py' = { count = 42, digest = '74e6d6e9e6cb31e5a018ac0ed4c0d11b0d0077b6f6dbc56124accce2b0c230d0' } 'tests/adapters/test_nvbench_live.py' = { count = 1, digest = '95b6165af0c766dac662211d5065d9f3854aa99a038fba53c28cba1a8f5b5bf9' } +'tests/adapters/test_rocprofv3.py' = { count = 5, digest = '70dcedb9448c59b72bfd9b538c1413620dff1ba50c57967e9187a416d032dd5a' } 'tests/adapters/test_triton_compiler_live.py' = { count = 1, digest = '0dfa64a2bfaaf94a441d80fac025d703bc267101e9ab748c31d25c8ee9ebfd5e' } 'tests/adapters/test_inference_artifacts.py' = { count = 86, digest = '9a9ccc9c7cfa6204fa5dc02a04b1b23c4899b5b719ed98ce846608d80a43d739' } 'tests/adapters/test_coverage.py' = { count = 3, digest = 'a9aa85687b6f34101ebcf2d6b69c22c866f9279436bdb9cbaff2793dba714891' } @@ -85,6 +88,7 @@ expected_test_count = 1015 'tests/application/test_inference_profiling.py' = { count = 14, digest = 'a6e5535fe44590c12e8ef2b07ae8d4dad077a5c9f30c562826c34c03e097fb15' } 'tests/application/test_inference_providers.py' = { count = 12, digest = '12253262c2a2a5aa282a6c57d13eef4cfbbf578a60434778433b4bdd22df1b52' } 'tests/application/test_native_reducer_properties.py' = { count = 6, digest = '474b6937339c51536863778fea2fa73b4fc7508bdd81d6f29eca5f618a8c3065' } +'tests/application/test_nsight_compute_capability.py' = { count = 2, digest = '6546a4ba71940cbc8ef193896798f70d85c56dd847f2945431d6f94e5bfbfdb5' } 'tests/application/test_gc.py' = { count = 8, digest = '2ef5cddb403f867124821c2cfde9df70896e2cd8d284c2e563b6455f0ad998d4' } 'tests/application/test_integrity.py' = { count = 1, digest = 'c44a2cb51b7f5a90c106b4a59e9661551a47ae8d0a21df308387618b99e277aa' } 'tests/application/test_path_containment.py' = { count = 9, digest = '3336e2349910166ada6bd0519a015e74fe405dc3a0389d63a43e41d706dcfefc' } diff --git a/tests/fixtures/rocprofv3/project-owned-rocm-shaped-perfetto.json b/tests/fixtures/rocprofv3/project-owned-rocm-shaped-perfetto.json new file mode 100644 index 0000000..0e1fc21 --- /dev/null +++ b/tests/fixtures/rocprofv3/project-owned-rocm-shaped-perfetto.json @@ -0,0 +1,47 @@ +{ + "traceEvents": [ + { + "name": "hipLaunchKernelGGL", + "cat": "hip_runtime", + "ph": "X", + "ts": 0, + "dur": 2, + "pid": 17, + "tid": 3, + "args": { + "phase": "decode", + "correlation": 41 + } + }, + { + "name": "vector_add", + "cat": "kernel", + "ph": "X", + "ts": 4, + "dur": 5, + "pid": 17, + "tid": 9, + "args": { + "phase": "decode", + "correlation": 41, + "device": "amd:0", + "stream": 7 + } + }, + { + "name": "vector_add", + "cat": "kernel", + "ph": "X", + "ts": 12, + "dur": 3, + "pid": 17, + "tid": 9, + "args": { + "phase": "decode", + "correlation": 42, + "device": "amd:0", + "stream": 7 + } + } + ] +} diff --git a/tests/mcp/test_workflows.py b/tests/mcp/test_workflows.py index 8cfe18b..d4f9058 100644 --- a/tests/mcp/test_workflows.py +++ b/tests/mcp/test_workflows.py @@ -319,7 +319,11 @@ async def test_mcp_kernel_validation_extraction_reports_progress_and_resources( Workspace.initialize(tmp_path) recorded_progress: list[tuple[float, float | None, str | None]] = [] - async def record(progress: float, total: float | None, message: str | None) -> None: + async def record( + progress: float, + total: float | None, + message: str | None, + ) -> None: recorded_progress.append((progress, total, message)) async with Client(create_server(tmp_path), raise_exceptions=True) as client: diff --git a/tests/ownership.toml b/tests/ownership.toml index bb7d089..791823f 100644 --- a/tests/ownership.toml +++ b/tests/ownership.toml @@ -23,16 +23,24 @@ markers = ["integration", "optional", "requires_memray"] owner = "adapters" lane = "adapters" paths = [ - "tests/adapters/test_benchmark_samples.py", - "tests/adapters/test_compute_sanitizer.py", - "tests/adapters/test_kernel_build.py", - "tests/adapters/test_kernel_validation.py", - "tests/adapters/test_nvbench.py", - "tests/adapters/test_pyperf.py", + "tests/adapters/test_benchmark_samples.py", + "tests/adapters/test_compute_sanitizer.py", + "tests/adapters/test_kernel_build.py", + "tests/adapters/test_kernel_validation.py", + "tests/adapters/test_nvbench.py", + "tests/adapters/test_pyperf.py", "tests/adapters/test_python_startup.py", ] markers = ["unit"] + +[[ownership]] +owner = "rocprofv3-adapter" +lane = "adapters" +paths = ["tests/adapters/test_rocprofv3.py"] +markers = ["integration", "process"] + + [[ownership]] owner = "compute-sanitizer-live" lane = "adapters" @@ -640,6 +648,24 @@ lane = "process" paths = ["tests/adapters/test_nsight_systems.py"] markers = ["integration", "process", "serial"] +[[ownership]] +owner = "nsight-compute-adapter" +lane = "adapters" +paths = ["tests/adapters/test_nsight_compute.py"] +markers = ["integration", "process"] + +[[ownership]] +owner = "nsight-compute-capability" +lane = "application" +paths = ["tests/application/test_nsight_compute_capability.py"] +markers = ["integration"] + +[[ownership]] +owner = "nsight-compute-live" +lane = "adapters" +paths = ["tests/adapters/test_nsight_compute_live.py"] +markers = ["integration", "optional", "process", "serial", "requires_ncu"] + [[ownership]] owner = "kernel-build-capture" lane = "adapters" diff --git a/tests/support/providers.py b/tests/support/providers.py index 0218f43..37e2207 100644 --- a/tests/support/providers.py +++ b/tests/support/providers.py @@ -19,10 +19,12 @@ EXECUTABLE_PROVIDERS = { "requires_bwrap": "bwrap", "requires_cargo": "cargo", - "requires_compute_sanitizer": "compute-sanitizer", "requires_perf": "perf", "requires_pyspy": "py-spy", "requires_systemd": "systemd-run", + "requires_compute_sanitizer": "compute-sanitizer", + "requires_ncu": "ncu", + "requires_rocprofv3": "rocprofv3", } PROVIDER_MARKERS = frozenset( { diff --git a/tests/test_test_runner.py b/tests/test_test_runner.py index fab43c0..a3cffcc 100644 --- a/tests/test_test_runner.py +++ b/tests/test_test_runner.py @@ -40,7 +40,8 @@ def test_list_reports_lanes_and_metadata_commands() -> None: assert result.returncode == 0, result.stderr assert " golden" in result.stdout - assert " optional-nvbench" in result.stdout + assert " optional-ncu" in result.stdout + assert " optional-rocprofv3" in result.stdout assert "Metadata commands:" in result.stdout assert " capabilities validate managed setup metadata against extras" in result.stdout diff --git a/tools/test.py b/tools/test.py index d54509e..7309086 100644 --- a/tools/test.py +++ b/tools/test.py @@ -68,7 +68,9 @@ "optional-coverage": "optional and requires_coverage", "optional-compute-sanitizer": "optional and requires_compute_sanitizer", "optional-cute": "optional and requires_cute", + "optional-ncu": "optional and requires_ncu", "optional-nvbench": "optional and requires_nvbench", + "optional-rocprofv3": "optional and requires_rocprofv3", "optional-triton": "optional and requires_triton", "optional-memray": "optional and requires_memray", "optional-perfetto": "optional and requires_perfetto", @@ -76,7 +78,8 @@ "optional-torch": "optional and requires_torch", "optional-host": ( "optional and not requires_compute_sanitizer and not requires_cute " - "and not requires_nvbench and not requires_triton " + "and not requires_ncu and not requires_nvbench and not requires_rocprofv3 " + "and not requires_triton " "and not requires_coverage " "and not requires_memray " "and not requires_perfetto and not requires_pyspy and not requires_torch" @@ -86,7 +89,9 @@ { "optional-compute-sanitizer", "optional-cute", + "optional-ncu", "optional-nvbench", + "optional-rocprofv3", "optional-triton", } ) @@ -728,7 +733,9 @@ def affected_plan( # noqa: C901 ("requires_coverage", "optional-coverage"), ("requires_compute_sanitizer", "optional-compute-sanitizer"), ("requires_cute", "optional-cute"), + ("requires_ncu", "optional-ncu"), ("requires_nvbench", "optional-nvbench"), + ("requires_rocprofv3", "optional-rocprofv3"), ("requires_triton", "optional-triton"), ("requires_memray", "optional-memray"), ("requires_perfetto", "optional-perfetto"), diff --git a/uv.lock b/uv.lock index c210d72..4319bd8 100644 --- a/uv.lock +++ b/uv.lock @@ -1088,8 +1088,8 @@ name = "flameox" version = "0.1.13" source = { editable = "." } dependencies = [ - { name = "defusedxml" }, { name = "anyio" }, + { name = "defusedxml" }, { name = "duckdb" }, { name = "ijson" }, { name = "mcp" }, @@ -1176,12 +1176,12 @@ dev = [ [package.metadata] requires-dist = [ - { name = "defusedxml", specifier = ">=0.7.1,<0.8" }, { name = "aiperf", marker = "extra == 'all'", specifier = ">=0.12,<0.13" }, { name = "aiperf", marker = "extra == 'inference'", specifier = ">=0.12,<0.13" }, { name = "anyio", specifier = ">=4.9,<5" }, { name = "coverage", marker = "extra == 'all'", specifier = ">=7.14,<8" }, { name = "coverage", marker = "extra == 'execution'", specifier = ">=7.14,<8" }, + { name = "defusedxml", specifier = ">=0.7.1,<0.8" }, { name = "deptry", marker = "extra == 'dev'", specifier = ">=0.23" }, { name = "duckdb", specifier = ">=1.5.4,<1.6" }, { name = "hypothesis", marker = "extra == 'dev'", specifier = ">=6.130" },