diff --git a/.github/workflows/publish-python-package.yml b/.github/workflows/publish-python-package.yml index 2d90a46e..3381ee24 100644 --- a/.github/workflows/publish-python-package.yml +++ b/.github/workflows/publish-python-package.yml @@ -28,7 +28,7 @@ jobs: with: python-version: "3.12" - name: Setup Poetry - uses: Gr1N/setup-poetry@v9 + uses: Gr1N/setup-poetry@v10 - name: Build the source distribution and pure-Python wheel run: | poetry install diff --git a/.github/workflows/test-python-package.yml b/.github/workflows/test-python-package.yml index d1a15103..2538bce9 100644 --- a/.github/workflows/test-python-package.yml +++ b/.github/workflows/test-python-package.yml @@ -21,18 +21,18 @@ jobs: strategy: matrix: - os: [ubuntu-latest, macos-latest] + os: [ubuntu-26.04, macos-latest] # set to ubuntu-latest when it is updated to 26.04 steps: - uses: actions/checkout@v7 - name: Setup Nox uses: fjwillemsen/setup-nox2@v4 - name: Setup Poetry - uses: Gr1N/setup-poetry@v9 + uses: Gr1N/setup-poetry@v10 - run: poetry self add poetry-plugin-export - uses: julia-actions/setup-julia@v3 with: - version: '1.11' # when changed, also see `require_julia` in noxfile.py and the Julia version in Project.toml + version: '1.13' # when changed, also see `require_julia` in noxfile.py and the Julia version in Project.toml - name: Run tests with Nox run: | rm -rf .nox diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f8e66ef..49b6b146 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,16 @@ All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/). ## Unreleased -- Fix `np.shuffle` crash and wrong return type in `Searchspace` distributed random and LHS samplers +- Extended Julia functionality: + - Implemented Julia observers for Metal, Nvidia and AMD energy measurements. + - Improved timing stability. + - Include the new tests of #407, additional new and updated tests. +- Improved performance and data handling. +- Improved NVML and AMD observers. +- Improved signal handling. +- Added Python 3.15 support, dropped 3.10 support due to removed support from required dependencies. +- Removed tomli dependency in favor of built-in tomllib. +- Fix `np.shuffle` crash and wrong return type in `Searchspace` distributed random and LHS samplers. ## [1.5.0] - 2026-08-20 - Kernel Tuner can now be used from Julia, [see KernelTuner.jl](https://github.com/KernelTuner/KernelTuner.jl) diff --git a/README.md b/README.md index a58f51d1..6a2a4688 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ Create optimized GPU applications in any mainstream GPU -programming language (CUDA, HIP, OpenCL, OpenACC, OpenMP). +programming language (CUDA, HIP, OpenCL, OpenACC, OpenMP). For Julia, see [KernelTuner.jl](https://github.com/KernelTuner/KernelTuner.jl). What Kernel Tuner does: diff --git a/kernel_tuner/backends/julia.py b/kernel_tuner/backends/julia.py index c0ec8c24..a8c4b728 100644 --- a/kernel_tuner/backends/julia.py +++ b/kernel_tuner/backends/julia.py @@ -12,12 +12,46 @@ - Currently supports CuArray and scalar arguments; constant and texture memory are not implemented. """ +import os +import sys +import uuid from pathlib import Path from warnings import warn from dataclasses import dataclass +from time import perf_counter import numpy as np +# Julia stops its threads for garbage collection by letting them access a protected memory page, and handles the +# resulting signal itself, but only if JuliaCall lets Julia install its signal handlers. GPU packages run Julia code on +# threads of their own (e.g. Metal.jl waits for the GPU on GPUToolbox worker threads), even if Julia was started with a +# single thread, so without this a garbage collection can crash Python with a bus error or segmentation fault. +# JuliaCall reads this setting when it is first imported, so it is set here, before any import of JuliaCall. +# This also lets Julia take over Ctrl-C, which the backend hands back to Python (see `give_interrupts_to_python`). +JULIACALL_HANDLE_SIGNALS = "PYTHON_JULIACALL_HANDLE_SIGNALS" +if "juliacall" not in sys.modules: + os.environ.setdefault(JULIACALL_HANDLE_SIGNALS, "yes") + + + +def give_interrupts_to_python(jl): + """Let Ctrl-C raise a KeyboardInterrupt in Python again, after JuliaCall started Julia with signal handling. + + Julia only needs its own signal handlers for garbage collection (SIGSEGV / SIGBUS), but it also takes over the + interrupt signal (SIGINT), so that Ctrl-C would not stop tuning. Python's handler is installed again, so that Ctrl-C + raises a KeyboardInterrupt once the current call into Julia returns. Julia is told not to exit on Ctrl-C, in case + its handler is still in place (Python can only install signal handlers from the main thread). + """ + import signal + import threading + + jl.seval("Base.exit_on_sigint(false)") + if threading.current_thread() is threading.main_thread(): + handler = signal.getsignal(signal.SIGINT) # Python's handler (e.g. that of a Jupyter kernel), overridden by Julia + if handler is not None: + signal.signal(signal.SIGINT, handler) + + from kernel_tuner.backends.backend import GPUBackend from kernel_tuner.observers.julia import JuliaRuntimeObserver from kernel_tuner.util import SkippableFailure @@ -45,6 +79,14 @@ class JuliaFunctions(GPUBackend): def __init__(self, device=0, iterations=7, compiler_options=None, observers=None): """Initialize Julia backend using JuliaCall.""" # lazy import to avoid mixing e.g. CUDA drivers + # (when Julia runs Python instead, e.g. from KernelTuner.jl, JuliaCall did not start Julia and Julia handles signals) + juliacall_config = getattr(sys.modules.get("juliacall"), "CONFIG", {}) + if juliacall_config.get("inited") and juliacall_config.get("opt_handle_signals") != "yes": + warn( + "JuliaCall started Julia without signal handling, " + "which can crash Python with a bus error or segmentation fault during garbage collection. " + f"Set the environment variable {JULIACALL_HANDLE_SIGNALS}=yes before JuliaCall is first imported." + ) try: from juliacall import JuliaError from juliacall import Main as jl @@ -55,6 +97,8 @@ def __init__(self, device=0, iterations=7, compiler_options=None, observers=None jl = None if jl is None: raise ImportError("JuliaCall not installed. Please run `pip install juliacall`.") + if getattr(sys.modules.get("juliacall"), "CONFIG", {}).get("opt_handle_signals") == "yes": + give_interrupts_to_python(jl) # process passed options and backends self.process_compiler_options(compiler_options) @@ -69,13 +113,33 @@ def __init__(self, device=0, iterations=7, compiler_options=None, observers=None self.current_kernel = None self.smem_size = 0 + # Event tracking for async kernel launches + self._start_event = None + self._end_event = None + # Initialize Julia backend self.backend = None self.start_evt = None self.end_evt = None - self.host_time = None + self.launch_kernel_time = None + self.launch_kernel_time_start = None self.initialize_backend(device, backend_name=backend_name) + # set device environment info + self.env = { + "device_name": self.name, + "compute_capability": self.cc, + "iterations": iterations, + "compiler_options": self.compiler_options, + "uuid": self.uuid, + "pci_bus_id": self.pci_bus_id, + } + # PCI domain and device ID are only known for some backends (AMD), only add them when available + if self.pci_domain_id is not None: + self.env["pci_domain_id"] = self.pci_domain_id + if self.pci_device_id is not None: + self.env["pci_device_id"] = self.pci_device_id + # setup observers self.observers = observers or [] self.observers.append( @@ -116,20 +180,20 @@ def __init__(self, device=0, iterations=7, compiler_options=None, observers=None const kt_julia_backend = {self.backend_mod_instname}() const GPUArrayType = {self.GPUArrayType} include("{str(Path(__file__).parent / "julia_helper.jl")}") + include("{str(Path(__file__).parent / str("julia_helper_ext_" + self.backend_mod_name.lower() + ".jl"))}") end """ ) - self.to_gpuarray = jl.KernelTunerHelper.to_gpuarray self.launch_kernel = jl.KernelTunerHelper.launch_kernel + self.synchronize_gpu = jl.KernelTunerHelper.synchronize_gpu + self.to_gpuarray = jl.KernelTunerHelper.to_gpuarray - # env info - self.env = { - "device_name": self.name, - "compute_capability": self.cc, - "iterations": iterations, - "compiler_options": self.compiler_options, - } + # Metal specific functions + self.use_metal_buffer = self.backend_mod_name == "Metal" + if self.use_metal_buffer: + self.metal_kernel_finished = jl.KernelTunerHelper.metal_kernel_finished + self.metal_gpu_time_ms = jl.KernelTunerHelper.metal_gpu_time_ms def initialize_backend(self, device, backend_name): """Initialize for a choice of Julia backends by backend_name, one of 'cuda', 'amd', 'intel', 'metal'.""" @@ -194,6 +258,24 @@ def initialize_backend(self, device, backend_name): except JuliaError: self.max_threads = None + # Query UUID and PCI bus ID if available + self.uuid = None + self.pci_bus_id = None + if info["uuid"] is not None: + try: + self.uuid = str(jl.seval(info["uuid"](int(device)))) + except JuliaError: + self.uuid = None + if info["pci_bus_id"] is not None: + try: + self.pci_bus_id = str(jl.seval(info["pci_bus_id"](int(device)))) + except JuliaError: + self.pci_bus_id = None + self.pci_domain_id = None + self.pci_device_id = None + if backend_name == "AMD": + self.query_amd_device_identity() + # Get the device and context try: self.backend_device = self.backend_mod.device() @@ -208,10 +290,31 @@ def initialize_backend(self, device, backend_name): # f"ZeCommandQueue(ZeContext(first(drivers())), devices(first(drivers()))[{int(device) + 1}]))" # ) elif backend_name == "METAL": - self.contextqueue = self.backend_mod.MTLCommandQueue(self.backend_device) + self.contextqueue = self.backend_mod.global_queue(self.backend_device) self.setup_streams(backend_name) + def query_amd_device_identity(self): + """Query the UUID and PCI location of the selected AMDGPU device, used by e.g. the AMDSMIObserver to find the device. + + The UUID and PCI information follow the same format as the HIP backend. + """ + try: + uuid_bytes, pci_domain, pci_bus, pci_device = jl.seval( + """ + let props = AMDGPU.HIP.properties(AMDGPU.HIP.device()) + (collect(reinterpret(UInt8, collect(props.uuid.bytes))), + Int(props.pciDomainID), Int(props.pciBusID), Int(props.pciDeviceID)) + end + """ + ) + self.uuid = str(uuid.UUID(bytes=bytes(int(b) for b in uuid_bytes))) + self.pci_domain_id = int(pci_domain) + self.pci_bus_id = int(pci_bus) + self.pci_device_id = int(pci_device) + except (JuliaError, ValueError) as e: + warn(f"Could not query the UUID and PCI information of the AMDGPU device: {e}") + def __del__(self): # drop GPUArray references to let Julia GC handle them try: @@ -230,9 +333,14 @@ def ready_argument_list(self, arguments): gpu_args = [] for arg in arguments: try: - arr = self.to_gpuarray(arg) - gpu_args.append(arr) - self.allocations.append(arr) + if np.isscalar(arg): + # if not explicitly float64, it should be a float32 for GPU kernels + if isinstance(arg, float) and not isinstance(arg, np.float64): + arg = np.float32(arg) + else: + arg = self.to_gpuarray(arg) + gpu_args.append(arg) + self.allocations.append(arg) except Exception as e: raise RuntimeError(f"Failed to move array to GPU: {e}") return gpu_args @@ -246,7 +354,6 @@ def compile(self, kernel_instance): kernel_code = kernel_instance.kernel_string kernel_name = kernel_instance.name self.kernel_source = kernel_instance.kernel_source - self.host_time = None # reset host time for this kernel instance # Extract all 'using' statements and check for required packages uses = [] @@ -286,6 +393,8 @@ def compile(self, kernel_instance): def run_kernel(self, func, gpu_args, threads, grid, stream=None): """Launch a compiled Julia kernel.""" + self.launch_kernel_time = None # reset host time for this kernel instance + self.launch_kernel_time_start = None if func is None or not isinstance(func, JuliaKernel): raise RuntimeError("No Julia kernel compiled or provided.") @@ -310,17 +419,18 @@ def run_kernel(self, func, gpu_args, threads, grid, stream=None): # run the kernel try: - self.host_time = self.launch_kernel( + # the GPU timing events (CUDA, AMDGPU) are recorded in Julia directly around the launch + self.launch_kernel_time, self.launch_kernel_time_start, start_evt, end_evt = self.launch_kernel( julia_func, args_tuple, params, ndrange, workgroupsize, int(self.smem_size), - self.observers[-1].start, - self.observers[-1].end, - self.observers[-1].stream, + self.use_metal_buffer, ) + self._start_event = start_evt + self._end_event = end_evt except JuliaError as e: if self.raise_errors: raise e @@ -328,51 +438,46 @@ def run_kernel(self, func, gpu_args, threads, grid, stream=None): raise SkippableFailure(f"Julia kernel launch failed for {params=}: {e}") def start_event(self): - """Records the event that marks the start of a measurement.""" - if self.backend_mod_name == "CUDA": - evt = self.start_evt() - self.backend_mod.record(evt, self.stream) - self.backend_mod.synchronize(evt) - return evt - elif self.backend_mod_name == "AMDGPU": - evt = self.start_evt(self.stream, do_record=False, timing=True) - self.backend_mod.HIP.record(evt) - self.backend_mod.HIP.synchronize(evt) - return evt - elif self.backend_mod_name == "Metal": - # Because our kernel launch happens via Kernel Abstractions, we wrap our kernel between two command buffers. - # Normally you would just use one command buffer for the actual kernel and take GPUEndTime - GPUStartTime. - jl.start_buf = self.create_metal_buffer() - jl.seval("Metal.commit!(start_buf)") - self.backend_mod.wait_completed(jl.start_buf) - return float(jl.start_buf.GPUEndTime) + """Marks the start of a measurement (non-blocking). + + The GPU timing events (CUDA, AMDGPU) and Metal command buffers are recorded in Julia directly around the kernel + launch in `run_kernel`, so that host-side work before the launch (e.g. Python to Julia calls, GC pauses) is + not measured as GPU time. + """ + self._host_start_time = perf_counter() + self._start_event = None + self._end_event = None def stop_event(self): - """Records the event that marks the end of a measurement.""" - if self.backend_mod_name == "CUDA": - evt = self.end_evt() - self.backend_mod.record(evt, self.stream) - self.backend_mod.synchronize(evt) - return evt - elif self.backend_mod_name == "AMDGPU": - evt = self.end_evt(self.stream, do_record=False, timing=True) - self.backend_mod.HIP.record(evt) - self.backend_mod.HIP.synchronize(evt) - return evt - elif self.backend_mod_name == "Metal": - jl.end_buf = self.create_metal_buffer() - jl.seval("Metal.commit!(end_buf)") - self.backend_mod.wait_completed(jl.end_buf) - return float(jl.end_buf.GPUStartTime) + """Marks the end of a measurement (non-blocking); the events are recorded in `run_kernel`.""" + pass def kernel_finished(self): """Returns True if the kernel has finished, False otherwise.""" - return True # JuliaCall synchronizes on record + if self.backend_mod_name == "CUDA": + if self._end_event is not None: + return bool(self.backend_mod.isdone(self._end_event)) + return True + elif self.backend_mod_name == "AMDGPU": + if self._end_event is not None: + # AMDGPU.HIP.hipEventQuery returns (hipError_t, bool) + result = self.backend_mod.HIP.hipEventQuery(self._end_event) + if isinstance(result, tuple) and len(result) >= 2: + return bool(result[1]) + return bool(result) + return True + elif self.backend_mod_name == "Metal": + return bool(self.metal_kernel_finished()) + elif self.backend_mod_name in ("INTEL", "CPU"): + # No hardware events + return True + return True # @staticmethod def synchronize(self): try: - jl.Main.KernelAbstractions.synchronize(self.backend) + self._host_time_julia = self.synchronize_gpu(self.launch_kernel_time_start) + self._host_stop_time = perf_counter() except JuliaError as e: raise RuntimeError(f"Julia synchronize failed: {e}") @@ -394,6 +499,11 @@ def memset(allocation, value, size): def memcpy_dtoh(dest, src): """Perform a device to host memory copy.""" try: + if not isinstance(src, np.ndarray): + # NumPy cannot view device memory, so on a (wrapped) GPU array `np.copyto` falls back to a very slow + # generic conversion (seconds for a few million elements). Copy to a host `Array` in Julia instead, + # which NumPy can then read through the buffer protocol without further copies. + src = np.asarray(jl.Array(src)) np.copyto(dest, src) except JuliaError as e: raise RuntimeError(f"Julia memcpy_dtoh failed: {e}") @@ -446,14 +556,6 @@ def check_package_and_install(self, package, allow_install=True): f'{package} not found in your Julia environment. Run `using Pkg; Pkg.add(\"{package}\")` to install it.' ) from e - def create_metal_buffer(self): - """Create a Metal buffer in the command queue.""" - try: - buf = self.contextqueue.commandBuffer() - except Exception: - buf = self.backend_mod.MTLCommandBuffer(self.contextqueue) - return buf - def setup_streams(self, backend_name: str): """Set up stream and event attributes for observers.""" # Optional: common KernelAbstractions stream abstraction diff --git a/kernel_tuner/backends/julia_helper.jl b/kernel_tuner/backends/julia_helper.jl index 852421cf..bfff3bc4 100644 --- a/kernel_tuner/backends/julia_helper.jl +++ b/kernel_tuner/backends/julia_helper.jl @@ -1,63 +1,59 @@ -export to_gpuarray, launch_kernel +# Base file for Julia functions. For backend-specific functionality, see the extension files, e.g. `julia_helper_ext_cuda.jl`. + +export to_gpuarray, launch_kernel, synchronize_gpu, metal_kernel_finished, metal_gpu_time_ms + +# GPU timing events recorded directly around a kernel launch, specialized in the backend extension files. +# By default (backends without timing events), there are no events. +create_timing_events(backend) = (nothing, nothing) +record_timing_event(::Nothing) = nothing function to_gpuarray(a) - a = deepcopy(a) # ensure we have a separate copy of the array to avoid unintended side effects - if isa(a, AbstractArray) - a = GPUArrayType(a) - end - return a + # ensure we have a separate copy to avoid unintended side effects; for arrays, the conversion to a GPU array + # already copies, while a deepcopy of a view would needlessly copy its entire parent array + # (on the CPU backend, `Array(a)` returns `a` itself for an `Array`, in which case it is copied explicitly) + isa(a, AbstractArray) || return deepcopy(a) + b = GPUArrayType(a) + return b === a ? deepcopy(a) : b end -function launch_kernel(kernel, args::Tuple, params::Tuple, ndrange::Tuple, workgroupsize::Tuple, shmem::Int, start_evt::Any, end_evt::Any, stream::Any) - t = Inf +function launch_kernel(kernel, args::Tuple, params::Tuple, ndrange::Tuple, workgroupsize::Tuple, shmem::Int, use_metal_buffer::Bool) + launch_time = Inf + launch_time_start = nothing + start_evt = nothing + end_evt = nothing # Check if this is a KernelAbstractions kernel if isdefined(Main, :KernelAbstractions) && kt_julia_backend !== nothing && applicable(kernel, kt_julia_backend, workgroupsize) configured_kernel = kernel(kt_julia_backend, workgroupsize) - # Launch kernel + # Launch kernel asynchronously mktemp() do tmppath, _ open(tmppath, "w") do tmpio # kernel errors are printed to stdout, capture them redirect_stdout(tmpio) do try val_params = Val.(params) # convert parameters to Val types for kernel invocation - start_buff = nothing - end_buff = nothing - start = time_ns() # simple host-side timing as fallback in case of issues with GPU timing - if start_evt !== nothing - if isdefined(Main, :CUDA) && isa(start_evt, CuEvent) - Main.CUDA.record(start_evt, stream) - elseif isdefined(Main, :AMDGPU) && isa(start_evt, AMDGPU.HIP.HIPEvent) - Main.AMDGPU.HIP.record(start_evt) - elseif isdefined(Main, :Metal) - # prepare the next command buffers for timing as they can only be used once - start_buff = create_metal_buffer(Metal.device()) - end_buff = create_metal_buffer(Metal.device()) - Metal.commit!(start_buff) + start_evt, end_evt = create_timing_events(kt_julia_backend) + # Record the timing events directly around the launch, so that host-side work before the + # launch is not measured as GPU time. The GC is disabled meanwhile, as a GC pause between + # recording the start event and launching would otherwise be measured as GPU time. + gc_was_enabled = GC.enable(false) + try + launch_time_start = time_ns() + record_timing_event(start_evt) + if use_metal_buffer + # record the command buffers of this launch, used for GPU timing (see julia_helper_ext_metal.jl) + metal_launch_recorded(() -> configured_kernel(args..., val_params...; ndrange=ndrange)) else - error("Unsupported event type for timing: $(typeof(start_evt))") + configured_kernel(args..., val_params...; ndrange=ndrange) # launch the kernel (async) end + record_timing_event(end_evt) + launch_time = float((time_ns() - launch_time_start) / 1e6) # convert to milliseconds + finally + GC.enable(gc_was_enabled) end - configured_kernel(args..., val_params...; ndrange=ndrange) # launch the kernel - Main.KernelAbstractions.synchronize(kt_julia_backend) # synchronize to ensure kernel completion - if end_evt !== nothing - if isdefined(Main, :CUDA) && isa(end_evt, CuEvent) - Main.CUDA.record(end_evt, stream) - Main.CUDA.synchronize(end_evt) # ensure the event is recorded before we read it - elseif isdefined(Main, :AMDGPU) && isa(end_evt, AMDGPU.HIP.HIPEvent) - Main.AMDGPU.HIP.record(end_evt) - Main.AMDGPU.HIP.synchronize(end_evt) # ensure the event is recorded before we read it - elseif isdefined(Main, :Metal) - Metal.commit!(end_buff) - Metal.wait_completed(end_buff) # ensure the command buffer is completed before we read the time - t = (float(end_buff.GPUStartTime) - float(start_buff.GPUEndTime)) * 1000 - else - error("Unsupported event type for timing: $(typeof(end_evt))") - end - else - # host-side timing fallback if events are not available - t = float((time_ns() - start) / 1e6) # convert to milliseconds - end + # Note: kernel launch is asynchronous + # Synchronization and event recording is handled by the Python backend catch e + e isa InterruptException && rethrow() # Ctrl-C: Kernel Tuner stops tuning (see `util.is_interrupt`) redirect_stdout(stdout) # restore stdout close(tmpio) stdout_output = read(tmppath, String) @@ -67,22 +63,20 @@ function launch_kernel(kernel, args::Tuple, params::Tuple, ndrange::Tuple, workg end end end - # print any stdout output from the kernel - print(read(tmppath, String)) + # # print any stdout output from the kernel (for debugging only, disabled for performance) + # print(read(tmppath, String)) end else error("Only KernelAbstractions kernels are supported.") end - return t + # return the launch time and the GPU timing events (if any), which are used by the observer in the Python backend + return launch_time, launch_time_start, start_evt, end_evt end -function create_metal_buffer(device) - # Create a Metal buffer in the command queue for timing - if isdefined(Main, :Metal) - contextqueue = Main.Metal.MTLCommandQueue(device) - return Metal.MTLCommandBuffer(contextqueue) - # return contextqueue.commandBuffer() - else - error("Metal backend is not available.") +function synchronize_gpu(launch_time_start) + Main.KernelAbstractions.synchronize(kt_julia_backend) + if launch_time_start === nothing + return nothing end + return float((time_ns() - launch_time_start) / 1e6) end diff --git a/kernel_tuner/backends/julia_helper.py b/kernel_tuner/backends/julia_helper.py index 0b30bc7f..60e129d5 100644 --- a/kernel_tuner/backends/julia_helper.py +++ b/kernel_tuner/backends/julia_helper.py @@ -19,7 +19,10 @@ "name": "CUDA.name(CUDA.device())", "max_threads": "CUDA.attribute(CUDA.device(), CUDA.DEVICE_ATTRIBUTE_MAX_THREADS_PER_BLOCK)", "capability": "CUDA.capability(CUDA.device())", + "uuid": lambda d: f"NVML.uuid(CUDA.NVML.Device({d}))", + "pci_bus_id": None, "GPUArrayType": "CuArray", + "supported_float_types": ["Float16", "BFloat16", "Float32", "Float64"], }, "AMD": { "pkg": "AMDGPU", @@ -29,7 +32,10 @@ "name": "AMDGPU.HIP.name(AMDGPU.HIP.device())", "max_threads": "AMDGPU.HIP.attribute(AMDGPU.HIP.device(), AMDGPU.HIP.hipDeviceAttributeMaxThreadsPerBlock)", "capability": None, + "uuid": None, + "pci_bus_id": None, "GPUArrayType": "ROCArray", + "supported_float_types": ["Float16", "BFloat16", "Float32", "Float64"], }, "INTEL": { "pkg": "oneAPI", @@ -39,7 +45,10 @@ "name": "oneAPI.name(oneAPI.device())", "max_threads": "oneAPI.compute_properties(oneAPI.device()).maxTotalGroupSize", "capability": None, + "uuid": None, + "pci_bus_id": None, "GPUArrayType": "oneArray", + "supported_float_types": ["Float16", "BFloat16", "Float32", "Float64"], }, "METAL": { "pkg": "Metal", @@ -49,7 +58,10 @@ "name": "Metal.device().name", "max_threads": "Int(Metal.device().maxThreadsPerThreadgroup.width)", "capability": None, + "uuid": None, + "pci_bus_id": None, "GPUArrayType": "MtlArray", + "supported_float_types": ["Float16", "BFloat16", "Float32"], }, "CPU": { "pkg": None, @@ -57,8 +69,11 @@ "module_backend": "CPU", "device_select": lambda d: "nothing", "name": "CPU", - "max_threads": "1024", # arbitrary as CPU doesn't have a max threads per block concept + "max_threads": "1024", # arbitrary as CPU doesn't have a max threads per block concept "capability": None, + "uuid": None, + "pci_bus_id": None, + "supported_float_types": ["Float16", "Float32", "Float64"], "GPUArrayType": "Array", }, } @@ -113,10 +128,7 @@ def julia_backend_available_metal(): if "metal" in supported: version = regex_search(r".*metal([\d.]+)", supported).group(1) if float(version) < 3: - warn( - f"Metal backend detected, but {supported} < 3. " - "Metal.jl requires Metal version 3 or higher." - ) + warn(f"Metal backend detected, but {supported} < 3. Metal.jl requires Metal version 3 or higher.") else: return True return False @@ -126,9 +138,7 @@ def julia_backend_available_intel(): """Check if Intel backend is available. May give false positives if other backends are present.""" try: # not a perfect check but should work in most cases - subprocess.check_output( - "ls /dev/dri/by-path/".split() - ) + subprocess.check_output("ls /dev/dri/by-path/".split()) return True except (FileNotFoundError, subprocess.CalledProcessError): return False diff --git a/kernel_tuner/backends/julia_helper_ext_amdgpu.jl b/kernel_tuner/backends/julia_helper_ext_amdgpu.jl new file mode 100644 index 00000000..b0bb4a63 --- /dev/null +++ b/kernel_tuner/backends/julia_helper_ext_amdgpu.jl @@ -0,0 +1,7 @@ +# Extension of the julia_helper.jl file for the AMDGPU backend. You can assume `using AMDGPU` has ran. +# GPU timing events, recorded on the stream that the kernel is launched on +create_timing_events(::ROCBackend) = ( + AMDGPU.HIP.HIPEvent(AMDGPU.stream(); do_record=false, timing=true), + AMDGPU.HIP.HIPEvent(AMDGPU.stream(); do_record=false, timing=true), +) +record_timing_event(evt::AMDGPU.HIP.HIPEvent) = AMDGPU.HIP.record(evt) diff --git a/kernel_tuner/backends/julia_helper_ext_cuda.jl b/kernel_tuner/backends/julia_helper_ext_cuda.jl new file mode 100644 index 00000000..61798f83 --- /dev/null +++ b/kernel_tuner/backends/julia_helper_ext_cuda.jl @@ -0,0 +1,4 @@ +# Extension of the julia_helper.jl file for the CUDA backend. You can assume `using CUDA` has ran. +# GPU timing events, recorded on the stream that the kernel is launched on +create_timing_events(::CUDABackend) = (CuEvent(), CuEvent()) +record_timing_event(evt::CuEvent) = CUDA.record(evt, CUDA.stream()) diff --git a/kernel_tuner/backends/julia_helper_ext_metal.jl b/kernel_tuner/backends/julia_helper_ext_metal.jl new file mode 100644 index 00000000..5c0d6b8f --- /dev/null +++ b/kernel_tuner/backends/julia_helper_ext_metal.jl @@ -0,0 +1,37 @@ +# Extension of the julia_helper.jl file for the Metal backend. You can assume `using Metal` has ran. + +# GPU timing. Metal.jl batches kernel launches into shared command buffers, which are only committed later, so a kernel +# launch has no command buffer of its own to time. Instead, the command buffers that Metal.jl commits while launching are +# recorded through its profiling hook, and the batch is committed right after the launch so that the kernel starts running. +# Kernel Tuner synchronizes before every launch, so the recorded command buffers only contain this kernel. +const metal_launched_buffers = Any[] + +function metal_launch_recorded(launch) + empty!(metal_launched_buffers) + previous_hook = Metal.MTL.profile_hook[] + Metal.MTL.profile_hook[] = function (cmdbuf) + push!(metal_launched_buffers, cmdbuf) + previous_hook === nothing || previous_hook(cmdbuf) + end + try + launch() + Metal.flush!() # commit the batch that holds the kernel + finally + Metal.MTL.profile_hook[] = previous_hook + end +end + +metal_is_done(cmdbuf::Metal.MTL.MTLCommandBuffer) = cmdbuf.status >= Metal.MTL.MTLCommandBufferStatusCompleted # also when it failed +metal_is_done(submission) = Metal.MTL.is_completed(submission) # Metal 4 submissions + +# whether the command buffers of the last launch have finished +metal_kernel_finished() = all(metal_is_done, metal_launched_buffers) + +# the GPU time in milliseconds of the last launch, from the start of its first command buffer to the end of its last, +# or `nothing` if not available (e.g. not finished yet) +function metal_gpu_time_ms() + isempty(metal_launched_buffers) && return nothing + ranges = [Metal.MTL.gpu_time_range(cmdbuf) for cmdbuf in metal_launched_buffers] + any(isnothing, ranges) && return nothing + return (maximum(last, ranges) - minimum(first, ranges)) * 1e3 +end diff --git a/kernel_tuner/backends/julia_helper_ext_oneapi.jl b/kernel_tuner/backends/julia_helper_ext_oneapi.jl new file mode 100644 index 00000000..2b96839c --- /dev/null +++ b/kernel_tuner/backends/julia_helper_ext_oneapi.jl @@ -0,0 +1 @@ +# Extension of the julia_helper.jl file for the oneAPI backend. You can assume `using oneAPI` has ran. \ No newline at end of file diff --git a/kernel_tuner/core.py b/kernel_tuner/core.py index 4a715696..71cc2f6a 100644 --- a/kernel_tuner/core.py +++ b/kernel_tuner/core.py @@ -415,7 +415,7 @@ def __init__( # Take list of observers from self.dev because Backends tend to add their own observer self.benchmark_observers = [ - obs for obs in self.dev.observers if not isinstance(obs, (ContinuousObserver, PrologueObserver)) + obs for obs in self.dev.observers if (not isinstance(obs, (ContinuousObserver, PrologueObserver)) and not obs.passthrough_observer) ] self.iterations = iterations @@ -467,11 +467,27 @@ def benchmark_continuous(self, func, gpu_args, threads, grid, result, duration): for obs in self.continuous_observers: obs.before_start() self.dev.start_event() - for _ in range(iterations): + # for very short kernels, launching can take longer than running the kernel, so `iterations` launches would take + # much longer than `duration`; stop launching once `duration` has passed, the GPU has been as busy as it can be + launch_deadline = time.perf_counter() + duration + # the observers are polled while launching too, as the GPU may finish right after the last launch, which leaves + # no time to poll afterwards; at most once per `poll_interval`, so that polling hardly delays the launches + poll_interval = 1e-3 + last_poll = None + for i in range(iterations): self.dev.run_kernel(func, gpu_args, threads, grid) + now = time.perf_counter() + if i == 0: + for obs in self.continuous_observers: + obs.after_start() + last_poll = now + elif now - last_poll >= poll_interval: + for obs in self.continuous_observers: + obs.during() + last_poll = now + if now > launch_deadline: + break self.dev.stop_event() - for obs in self.continuous_observers: - obs.after_start() while not self.dev.kernel_finished(): for obs in self.continuous_observers: obs.during() @@ -531,6 +547,7 @@ def benchmark(self, func, gpu_args, instance, verbose, objective, skip_nvml_sett self.benchmark_continuous(func, gpu_args, instance.threads, instance.grid, result, duration) except Exception as e: + util.raise_if_interrupt(e) # Ctrl-C stops tuning instead of failing this configuration # some launches may fail because too many registers are required # to run the kernel given the current thread block size # the desired behavior is to simply skip over this configuration @@ -685,6 +702,7 @@ def compile_and_benchmark(self, kernel_source, gpu_args, params, kernel_options, last_benchmark_time = 1000 * (time.perf_counter() - start_benchmark) except Exception as e: + util.raise_if_interrupt(e) # Ctrl-C stops tuning instead of failing this configuration # dump kernel sources to temp file temp_filenames = instance.prepare_temp_files_for_error_msg() print("Error while compiling or benchmarking, see source files: " + " ".join(temp_filenames)) @@ -710,6 +728,7 @@ def compile_kernel(self, instance, verbose): try: func = self.dev.compile(instance) except Exception as e: + util.raise_if_interrupt(e) # Ctrl-C stops tuning instead of failing this configuration # compiles may fail because certain kernel configurations use too # much shared memory for example, the desired behavior is to simply # skip over this configuration and try the next one @@ -835,6 +854,7 @@ def run_kernel(self, func, gpu_args, instance): try: self.dev.run_kernel(func, gpu_args, instance.threads, instance.grid) except Exception as e: + util.raise_if_interrupt(e) # Ctrl-C stops tuning instead of failing this configuration if "too many resources requested for launch" in str(e) or "OUT_OF_RESOURCES" in str(e): logging.debug("ignoring runtime failure due to too many resources required") return False @@ -853,7 +873,8 @@ def retrieve_results_to_host(self, arguments: list, should_sync: list[bool], gpu cp = _get_cupy() cupy_ndarray = (cp.ndarray,) if cp is not None else () if isinstance(arg, (np.ndarray,) + cupy_ndarray) or util.is_julia_array(arg): - result_host.append(np.zeros_like(arg)) + # np.zeros_like on a wrapped Julia array is very slow (element-wise), while np.asarray is a zero-copy view + result_host.append(np.zeros_like(np.asarray(arg) if util.is_julia_array(arg) else arg)) self.dev.memcpy_dtoh(result_host[-1], gpu_args[i]) elif isinstance(arg, torch.Tensor) and isinstance(answer[i], torch.Tensor): if not answer[i].is_cuda: @@ -1009,6 +1030,7 @@ def _flatten(a): diff = lib.abs(expected - result) indices = lib.where(diff > atol) print(diff[indices]) + print(indices) correct = correct and output_test if not correct: diff --git a/kernel_tuner/interface.py b/kernel_tuner/interface.py index c1d72608..ebd43ed9 100644 --- a/kernel_tuner/interface.py +++ b/kernel_tuner/interface.py @@ -1037,7 +1037,8 @@ def run_kernel( elif isinstance(arg, torch.Tensor): results.append(arg.cpu()) else: - results.append(numpy.zeros_like(arg)) + # numpy.zeros_like on a wrapped Julia array is very slow (element-wise), while numpy.asarray is a zero-copy view + results.append(numpy.zeros_like(numpy.asarray(arg) if util.is_julia_array(arg) else arg)) dev.memcpy_dtoh(results[-1], gpu_args[i]) return results diff --git a/kernel_tuner/observers/amd.py b/kernel_tuner/observers/amd.py index bf73f218..05ea02d1 100644 --- a/kernel_tuner/observers/amd.py +++ b/kernel_tuner/observers/amd.py @@ -1,8 +1,9 @@ -import amdsmi import logging -import numpy as np import time +import amdsmi +import numpy as np + # Trapz was renamed to trapezoid in Numpy 2.0 try: from numpy import trapezoid @@ -10,6 +11,7 @@ from numpy import trapz as trapezoid from uuid import UUID + from kernel_tuner.observers import BenchmarkObserver, ContinuousObserver logger = logging.getLogger(__name__) @@ -77,7 +79,6 @@ def __init__(self, device): def total_energy_usage(self): """Returns total energy usage since startup.""" - result = amdsmi.amdsmi_get_energy_count(self.device) # This field changed name in rocm 6.4 @@ -92,18 +93,17 @@ def total_energy_usage(self): def current_power_usage(self): info = amdsmi.amdsmi_get_power_info(self.device) - if "current_socket_power" in info: - # For newer Mi300+ cards - return info["current_socket_power"] - elif "average_socket_power" in info: - # For older cards - return info["average_socket_power"] - else: - raise RuntimeError(f"invalid result from amdsmi_get_power_info: {info}") + # Newer MI300+ cards report "current_socket_power", older cards (e.g., MI210) only "average_socket_power". + # Unsupported fields are present but have the value "N/A", so check the value rather than the key. + for key in ("current_socket_power", "average_socket_power", "socket_power"): + value = info.get(key) + if value is not None and not isinstance(value, str): + return value + + raise RuntimeError(f"invalid result from amdsmi_get_power_info: {info}") def core_voltage(self): """Returns current voltage in Volt.""" - milli_volt = amdsmi.amdsmi_get_gpu_volt_metric( self.device, amdsmi.AmdSmiVoltageType.VDDGFX, @@ -227,8 +227,7 @@ def get_results(self): class AMDSMIObserver(BenchmarkObserver): - """ - BenchmarkObserver that uses amdsmi to monitor AMD GPUs and measure energy usage (`energy`), + """BenchmarkObserver that uses amdsmi to monitor AMD GPUs and measure energy usage (`energy`), core clock frequency (`core_freq`), memory clock frequency (`mem_freq`), temperature (`temperature`), and core voltage (`core_voltage`). """ @@ -242,8 +241,7 @@ def __init__( use_continuous_observer=True, continuous_duration=1.0, ): - """ - Initialize the AMDSMIObserver. + """Initialize the AMDSMIObserver. Supported observables are: `energy`, `core_freq`, `mem_freq`, `temperature`, and `core_voltage`. diff --git a/kernel_tuner/observers/julia.py b/kernel_tuner/observers/julia.py index 16fa1199..d90eedb5 100644 --- a/kernel_tuner/observers/julia.py +++ b/kernel_tuner/observers/julia.py @@ -1,10 +1,13 @@ from time import perf_counter from warnings import warn +import logging import numpy as np from kernel_tuner.observers.observer import BenchmarkObserver, PrologueObserver +logger = logging.getLogger(__name__) + class JuliaRuntimeObserver(BenchmarkObserver): """Cross-backend GPU timing for KernelAbstractions. @@ -12,7 +15,7 @@ class JuliaRuntimeObserver(BenchmarkObserver): - CUDA: CuEvent timing - ROCBackend: HIPEvent timing - OneAPI: host-side timing (less accurate, no events available) - - Metal: timing by wrapping the kernel launch between two command buffers and measuring the time between them + - Metal: GPU start and end times of the command buffers that ran the kernel """ def __init__( @@ -36,75 +39,108 @@ def __init__( self.start = start_event self.end = end_event self.times = [] - self.t0 = None - - if self.name == "cuda": - # initialize events for this instance of the observer - self.stream = self.backend_mod.stream() - self.start = self.start() - self.end = self.end() - elif self.name == "amdgpu": - self.stream = self.backend_mod.stream() - self.start = self.start(self.stream, do_record=False, timing=True) - self.end = self.end(self.stream, do_record=False, timing=True) def before_start(self): - if self.start is not None: - if self.name == "metal": - self.t0 = self.start() - elif self.name == "cuda": - # the events are recorded in the julia_helper kernel launch code - pass - elif self.name == "amdgpu": - # the events are recorded in the julia_helper kernel launch code - pass - else: - raise ValueError(f"Unsupported backend for timing: {self.name}") - else: - # fallback: host-side timestamp - self.t0 = perf_counter() + # Events are now recorded in the backend's start_event() method + pass def after_finish(self): ms = None - if self.end is not None: - if self.name == "metal": - ms_observer = float((self.end() - self.t0) * 1000.0) - ms_helper = self.kt_backend.host_time - # take the minimum of the two measurements to mitigate overhead of command buffer timing - ms = min( - ms_observer, ms_helper - ) - elif self.name == "cuda": - # the events are recorded in the julia_helper kernel launch code - ms = float(self.backend_mod.elapsed(self.start, self.end) * 1000.0) - elif self.name == "amdgpu": - # the events are recorded in the julia_helper kernel launch code - ms = float(self.backend_mod.HIP.elapsed(self.start, self.end) * 1000.0) - else: - raise ValueError(f"Unsupported backend for timing: {self.name}") - else: - self.kernelabstractions.synchronize(self.backend) - dt = perf_counter() - self.t0 - ms = dt * 1000.0 - warn(f"Using host-side timing for Julia {self.name} backend; results may be less accurate.") - - if ms > self.kt_backend.host_time: - if ms < 1 and (ms > 1.5 * self.kt_backend.host_time and self.end is not None): + ms_hostside = self.kt_backend._host_time_julia + + if self.name == "cuda": + # CUDA: use events stored in backend + if self.kt_backend._start_event is not None and self.kt_backend._end_event is not None: + ms = float(self.backend_mod.elapsed(self.kt_backend._start_event, self.kt_backend._end_event) * 1000.0) + elif self.name == "amdgpu": + # AMD: use events stored in backend + if self.kt_backend._start_event is not None and self.kt_backend._end_event is not None: + ms = float(self.backend_mod.HIP.elapsed(self.kt_backend._start_event, self.kt_backend._end_event) * 1000.0) + elif self.name == "metal": + # GPU time of the command buffers that ran the kernel + gpu_ms = self.kt_backend.metal_gpu_time_ms() + if gpu_ms is not None: + ms = float(gpu_ms) + elif self.name in ("intel", "cpu"): + # uses host-side timing + pass + + # If GPU timing failed or not available, fall back to host-side timing if available + if ms is None and ms_hostside is not None: + ms = ms_hostside + if self.name in ("cuda", "amdgpu", "metal"): + warn(f"Using host-side timing for Julia {self.name} backend; results may be less accurate.") + if ms is None: + if self.kt_backend._host_start_time is not None and self.kt_backend._host_stop_time is not None: + ms_hostside_python = (self.kt_backend._host_stop_time - self.kt_backend._host_start_time) * 1000.0 + ms = ms_hostside_python + warn(f"Using Python host-side timing for Julia {self.name} backend; results may be less accurate.") + if ms is None: + raise RuntimeError(f"Failed to measure GPU time for Julia {self.name} backend; no timing information available.") + + # If both GPU and host timing are available, check for discrepancies + if ms is not None and ms_hostside is not None and ms > ms_hostside and ms_hostside > 0.0: + if ms > 10 * ms_hostside: warn( - f"Measured GPU time {ms:.3f} ms is greater than host time {self.kt_backend.host_time:.3f} ms; " + f"Measured GPU time {ms:.3f} ms is more than 10x host time {ms_hostside:.3f} ms; " + "this may indicate an issue with the timing measurement. Replacing GPU time with host time for this measurement." + ) + ms = ms_hostside + elif ms < 1 and (ms > 1.2 * ms_hostside): + logger.info( + f"Measured GPU time {ms:.3f} ms is greater than host time {ms_hostside:.3f} ms; " "this may happen with very short execution times." ) - elif ms > 1.5 * self.kt_backend.host_time and self.end is not None: + elif ms > 1.2 * ms_hostside: warn( - f"Measured GPU time {ms:.3f} ms is substantially greater than host time {self.kt_backend.host_time:.3f} ms; " + f"Measured GPU time {ms:.3f} ms is substantially greater than host time {ms_hostside:.3f} ms; " "this may indicate an issue with the timing measurement." ) self.times.append(ms) + # a measured time is an outlier if it deviates from the median both statistically (robust z-score, based on the + # median absolute deviation) and substantially (relative to the median), so that tiny differences between otherwise + # consistent measurements (e.g. 0.031 vs. 0.033 ms) are not regarded as outliers + outlier_robust_z = 3.5 + outlier_rel_deviation = 0.25 + + @classmethod + def find_outliers(cls, arr): + """Returns a boolean mask of the times in `arr` that are outliers (only slower times are regarded as such).""" + median = np.median(arr) + deviation = arr - median + substantial = deviation > cls.outlier_rel_deviation * median + mad = np.median(np.abs(deviation)) + if mad == 0: + # at least half of the times are identical, any substantial deviation is an outlier + return substantial + robust_z = 0.6745 * deviation / mad + return substantial & (robust_z > cls.outlier_robust_z) + def get_results(self): + times_avg = self.times.copy() + if len(self.times) >= 5: + # due to interactivity, the Julia runtime observer may have rare outliers; remove them if there are any + arr = np.array(times_avg) + outliers = self.find_outliers(arr) + num_outliers = int(np.sum(outliers)) + if num_outliers > 0: + log_str = ( + f"Removed {num_outliers} of {len(arr)} measured times as outliers " + f"({', '.join(f'{t:.4f}' for t in arr[outliers])} ms vs. median {np.median(arr):.4f} ms)." + ) + # only warn when there is more than one outlier, as that indicates unreliable measurements rather than warmup effects + if num_outliers > 1: + logger.warning(log_str) + else: + logger.info(log_str) + times_avg = arr[~outliers] + else: + # remove the first to avoid JIT warmup effects + times_avg = self.times[1:] results = { - "time": np.average(self.times), + "time": np.average(times_avg), "times": self.times.copy(), } self.times = [] diff --git a/kernel_tuner/observers/julia_nvml.jl b/kernel_tuner/observers/julia_nvml.jl new file mode 100644 index 00000000..46f8594e --- /dev/null +++ b/kernel_tuner/observers/julia_nvml.jl @@ -0,0 +1,151 @@ +# NVML functionality for the JuliaNVMLObserver, using the NVML wrapper of CUDA.jl. +# Functions raise `NVML.NVMLError` on failure; `error_code` lets Python tell e.g. "not supported" from "no permission". +# Units follow NVML: power in milliwatt, energy in millijoule, clocks in MHz, temperature in degrees Celsius. +module KernelTunerNVML + +using CUDA: CUDA, NVML + +# NVML return code of an exception, or -1 if it is not an NVML error +error_code(err) = err isa NVML.NVMLError ? Int(err.code) : -1 + + +## device lookup + +# the device (not a MIG instance, as power and clocks are controlled per device) that runs the given CUDA device, +# found by UUID, as CUDA and NVML can number the devices differently (e.g. with CUDA_VISIBLE_DEVICES) +device_of(cuda_device::CUDA.CuDevice) = NVML.Device(CUDA.parent_uuid(cuda_device)) + +device_by_index(index::Integer) = NVML.Device(index) + +device_by_uuid(uuid::AbstractString) = NVML.Device(Base.UUID(replace(uuid, r"^(GPU|MIG)-" => ""))) + +function device_by_pci_bus(pci_bus::AbstractString) + handle = Ref{NVML.nvmlDevice_t}() + NVML.nvmlDeviceGetHandleByPciBusId_v2(pci_bus, handle) + index = Ref{Cuint}() + NVML.nvmlDeviceGetIndex(handle[], index) + return NVML.Device(Int(index[])) # NVML.Device has no constructor that takes a handle +end + +index(dev::NVML.Device) = NVML.index(dev) + +uuid(dev::NVML.Device) = "GPU-$(NVML.uuid(dev))" + +function pci_bus(dev::NVML.Device) + info = Ref{NVML.nvmlPciInfo_t}() + NVML.nvmlDeviceGetPciInfo_v3(dev, info) + chars = collect(info[].busId) + return String(UInt8.(chars[1:(something(findfirst(iszero, chars), length(chars)+1)-1)])) +end + + +## power and energy + +function power_limit(dev::NVML.Device) + limit = Ref{Cuint}() + NVML.nvmlDeviceGetPowerManagementLimit(dev, limit) + return Int(limit[]) +end + +function power_limit_constraints(dev::NVML.Device) + min_limit, max_limit = Ref{Cuint}(), Ref{Cuint}() + NVML.nvmlDeviceGetPowerManagementLimitConstraints(dev, min_limit, max_limit) + return [Int(min_limit[]), Int(max_limit[])] +end + +set_power_limit!(dev::NVML.Device, limit::Integer) = NVML.nvmlDeviceSetPowerManagementLimit(dev, limit) + +# current power, measured instantly where supported, as the regular power usage is averaged over about a second on +# recent GPUs, which is much longer than the run of a kernel +# reused for every call, so that reading the power during a measurement does not allocate and trigger garbage collection +const power_request = [NVML.nvmlFieldValue_t(ntuple(_ -> 0x00, sizeof(NVML.nvmlFieldValue_t)))] + +function power(dev::NVML.Device) + values = power_request + GC.@preserve values begin + ptr = pointer(values) + values[1] = NVML.nvmlFieldValue_t(ntuple(_ -> 0x00, sizeof(NVML.nvmlFieldValue_t))) + unsafe_store!(ptr.fieldId, NVML.NVML_FI_DEV_POWER_INSTANT) + NVML.nvmlDeviceGetFieldValues(dev, 1, ptr) + if unsafe_load(ptr.nvmlReturn) == NVML.NVML_SUCCESS + return Int(unsafe_load(ptr.value).uiVal) + end + end + # instant power is not supported, fall back to the regular power usage + usage = Ref{Cuint}() + NVML.nvmlDeviceGetPowerUsage(dev, usage) + return Int(usage[]) +end + +# total energy used since the driver was loaded +function energy(dev::NVML.Device) + total = Ref{Culonglong}() + NVML.nvmlDeviceGetTotalEnergyConsumption(dev, total) + return Int(total[]) +end + +temperature(dev::NVML.Device) = NVML.temperature(dev) + +# performance state, from 0 (maximum performance) to 15 (minimum performance), or 32 if unknown +function power_state(dev::NVML.Device) + state = Ref{NVML.nvmlPstates_t}() + NVML.nvmlDeviceGetPowerState(dev, state) + return Int(state[]) +end + + +## device settings + +function persistence_mode(dev::NVML.Device) + mode = Ref{NVML.nvmlEnableState_t}() + NVML.nvmlDeviceGetPersistenceMode(dev, mode) + return Int(mode[]) +end + +set_persistence_mode!(dev::NVML.Device, mode::Integer) = + NVML.nvmlDeviceSetPersistenceMode(dev, NVML.nvmlEnableState_t(mode)) + +function auto_boost(dev::NVML.Device) + enabled, default_enabled = Ref{NVML.nvmlEnableState_t}(), Ref{NVML.nvmlEnableState_t}() + NVML.nvmlDeviceGetAutoBoostedClocksEnabled(dev, enabled, default_enabled) + return Int(enabled[]) +end + +set_auto_boost!(dev::NVML.Device, setting::Integer) = + NVML.nvmlDeviceSetAutoBoostedClocksEnabled(dev, NVML.nvmlEnableState_t(setting)) + + +## clocks, `kind` is "graphics" or "mem" + +clock_type(kind::AbstractString) = kind == "graphics" ? NVML.NVML_CLOCK_GRAPHICS : + kind == "mem" ? NVML.NVML_CLOCK_MEM : + throw(ArgumentError("unknown clock kind: $kind")) + +function clock_query(f, dev::NVML.Device, kind::AbstractString) + mhz = Ref{Cuint}() + f(dev, clock_type(kind), mhz) + return Int(mhz[]) +end + +clock(dev::NVML.Device, kind) = clock_query(NVML.nvmlDeviceGetClockInfo, dev, kind) +applications_clock(dev::NVML.Device, kind) = clock_query(NVML.nvmlDeviceGetApplicationsClock, dev, kind) +default_applications_clock(dev::NVML.Device, kind) = clock_query(NVML.nvmlDeviceGetDefaultApplicationsClock, dev, kind) + +# supported clocks, from high to low +supported_memory_clocks(dev::NVML.Device) = NVML.supported_memory_clocks(dev) +supported_graphics_clocks(dev::NVML.Device, mem_clock::Integer) = NVML.supported_graphics_clocks(dev, mem_clock) + +set_applications_clocks!(dev::NVML.Device, mem_clock::Integer, gr_clock::Integer) = + NVML.nvmlDeviceSetApplicationsClocks(dev, mem_clock, gr_clock) + +function set_locked_clocks!(dev::NVML.Device, mem_clock::Integer, gr_clock::Integer) + NVML.nvmlDeviceSetGpuLockedClocks(dev, gr_clock, gr_clock) + NVML.nvmlDeviceSetMemoryLockedClocks(dev, mem_clock, mem_clock) +end + +function reset_locked_clocks!(dev::NVML.Device) + NVML.nvmlDeviceResetGpuLockedClocks(dev) + NVML.nvmlDeviceResetMemoryLockedClocks(dev) +end + +end diff --git a/kernel_tuner/observers/julia_nvml.py b/kernel_tuner/observers/julia_nvml.py new file mode 100644 index 00000000..24f161d6 --- /dev/null +++ b/kernel_tuner/observers/julia_nvml.py @@ -0,0 +1,452 @@ +"""NVML observer for the Julia backend, using the NVML wrapper of CUDA.jl instead of pynvml. + +The observables and tunable parameters are the same as those of the :class:`~kernel_tuner.observers.nvml.NVMLObserver`. +The NVML calls are made from the Julia process that runs the kernels, so pynvml is not needed. See ``julia_nvml.jl`` +for the Julia side. +""" + +import atexit +import re +import subprocess +import time +from pathlib import Path + +import numpy as np + +from kernel_tuner.observers.nvml import FIELD_TYPE_MAP, SUPPORTED_OBSERVABLES, NVMLObserver # noqa: F401 + +NVML_ERROR_NOT_SUPPORTED = 3 +NVML_ERROR_NO_PERMISSION = 4 + + +class NVMLError(Exception): + """NVML call failed, ``code`` is the NVML return code (-1 if the error did not come from NVML).""" + + def __init__(self, code, message): + super().__init__(message) + self.code = code + + +class NVMLError_NotSupported(NVMLError): + pass + + +class NVMLError_NoPermission(NVMLError): + pass + + +class _JuliaNVMLLibrary: + """Calls the functions of the ``KernelTunerNVML`` Julia module, and raises NVML errors as Python exceptions.""" + + def __init__(self): + # importing the Julia backend first sets up the signal handling for JuliaCall before it is first imported + import kernel_tuner.backends.julia # noqa: F401 + + try: + from juliacall import JuliaError + from juliacall import Main as jl + except ImportError as e: + raise ImportError("JuliaCall not installed. Please run `pip install juliacall`.") from e + self._error_type = JuliaError + if not jl.seval("isdefined(Main, :KernelTunerNVML)"): + try: + jl.include(str(Path(__file__).parent / "julia_nvml.jl")) + except JuliaError as e: + raise ImportError(f"Could not load NVML through CUDA.jl, is CUDA.jl installed? {e}") from e + self._module = jl.KernelTunerNVML + + def __getattr__(self, name): + function = getattr(self._module, name) + + def call(*args): + try: + return function(*args) + except self._error_type as e: + code = self._module.error_code(e.exception) + error_class = { + NVML_ERROR_NOT_SUPPORTED: NVMLError_NotSupported, + NVML_ERROR_NO_PERMISSION: NVMLError_NoPermission, + }.get(code, NVMLError) + raise error_class(code, f"NVML call {name} failed: {e}") from e + + return call + + +_library = None + + +def julia_nvml_library(): + """Load the Julia NVML functions on first use, so that Julia is only started when needed.""" + global _library + if _library is None: + _library = _JuliaNVMLLibrary() + return _library + + +class JuliaNVML: + """Class that gathers the NVML functionality for one device, through CUDA.jl. + + Has the same interface as :class:`kernel_tuner.observers.nvml.nvml`. The device is selected by exactly one of + ``cuda_device`` (a ``CUDA.CuDevice`` from Julia), ``device_id`` (the NVML index, as reported by nvidia-smi), + ``device_uuid``, or ``device_pci_bus``. + """ + + def __init__( + self, + device_id=None, + device_uuid=None, + device_pci_bus=None, + nvidia_smi_fallback=None, + use_locked_clocks=False, + cuda_device=None, + ): + """Create object to control device using NVML through CUDA.jl.""" + # set these first, as __del__ uses them and this __init__ may raise an exception midway + self.pwr_limit_default = None + self.modified_clocks = False + self.modified_pwr_limit = False + + self.lib = julia_nvml_library() + if cuda_device is not None: + self.dev = self.lib.device_of(cuda_device) + elif device_id is not None: + self.dev = self.lib.device_by_index(int(device_id)) + elif device_uuid is not None: + self.dev = self.lib.device_by_uuid(str(device_uuid)) + elif device_pci_bus is not None: + self.dev = self.lib.device_by_pci_bus(str(device_pci_bus)) + else: + raise ValueError("No device given, pass one of cuda_device, device_id, device_uuid, or device_pci_bus") + + self.id = self.lib.index(self.dev) + self.uuid = str(self.lib.uuid(self.dev)) + self.pci_bus = str(self.lib.pci_bus(self.dev)) + self.nvidia_smi = nvidia_smi_fallback or "nvidia-smi" + + if device_id is not None and self.id != device_id: + raise ValueError(f"NVML device ID does not match requested device: {device_id} != {self.id}") + + try: + self.pwr_limit_default = self.lib.power_limit(self.dev) + self.pwr_constraints = [int(x) for x in self.lib.power_limit_constraints(self.dev)] + except NVMLError_NotSupported: + # inverted range to make all range checks fail + self.pwr_limit_default = None + self.pwr_constraints = [1, 0] + + try: + self._persistence_mode = self.lib.persistence_mode(self.dev) + except NVMLError_NotSupported: + self._persistence_mode = None + + try: + self._auto_boost = self.lib.auto_boost(self.dev) + except NVMLError_NotSupported: + self._auto_boost = None + + # gather info on clock defaults + try: + self.gr_clock_default = self.lib.default_applications_clock(self.dev, "graphics") + self.mem_clock_default = self.lib.default_applications_clock(self.dev, "mem") + except NVMLError_NotSupported: + self.gr_clock_default = None + self.mem_clock_default = None + self.applications_gr_clock = self.gr_clock_default + self.applications_mem_clock = self.mem_clock_default + + # supported clocks, from high to low, with the supported gr clocks for each supported mem clock + self.supported_mem_clocks = [int(x) for x in self.lib.supported_memory_clocks(self.dev)] + self.supported_gr_clocks = { + mem_clock: [int(x) for x in self.lib.supported_graphics_clocks(self.dev, mem_clock)] + for mem_clock in self.supported_mem_clocks + } + + # test whether locked gr clocks and mem clocks are supported + self.use_locked_clocks = use_locked_clocks + if use_locked_clocks: + try: + # try to set highest supported clocks + mem_clock = self.supported_mem_clocks[0] + gr_clock = self.supported_gr_clocks[mem_clock][0] + self.locked_gr_clock = 0 + self.locked_mem_clock = 0 + self.set_clocks(mem_clock, gr_clock) + except NVMLError_NotSupported: + # switch to using application clocks + self.use_locked_clocks = False + + def __del__(self): + try: + self.restore_defaults() + except Exception: + # Julia or nvidia-smi may no longer be available while Python is shutting down, + # in which case the defaults were already restored at exit + pass + + def restore_defaults(self): + """Restore the power limit and clocks, if they were changed.""" + if self.modified_pwr_limit and self.pwr_limit_default is not None: + self.pwr_limit = self.pwr_limit_default + if self.modified_clocks: + self.reset_clocks() + + def _restore_at_exit(self): + if not (self.modified_clocks or self.modified_pwr_limit): + atexit.register(self.restore_defaults) + + def _sudo_nvidia_smi(self, *options): + subprocess.run(["sudo", self.nvidia_smi, "-i", str(self.id), *options], check=True) + + @property + def pwr_state(self): + """Get the Device current Power State.""" + return self.lib.power_state(self.dev) + + @property + def pwr_limit(self): + """Control the power limit in milliwatt (may require permission), check pwr_constraints for the allowed range.""" + return self.lib.power_limit(self.dev) + + @pwr_limit.setter + def pwr_limit(self, new_limit): + if not self.pwr_constraints[0] <= new_limit <= self.pwr_constraints[1]: + raise ValueError( + f"Power limit {new_limit} out of range [{self.pwr_constraints[0]}, {self.pwr_constraints[1]}]" + ) + if new_limit == self.pwr_limit: + return + self._restore_at_exit() + self.modified_pwr_limit = True + try: + self.lib.set_power_limit_b(self.dev, int(new_limit)) + except NVMLError_NoPermission: + # nvidia-smi expects Watts rather than milliwatts + self._sudo_nvidia_smi(f"--power-limit={int(new_limit / 1000.0)}") + + @property + def persistence_mode(self): + """Control persistence mode (may require permission), 0 for disabled, 1 for enabled.""" + return self._persistence_mode + + @persistence_mode.setter + def persistence_mode(self, new_mode): + if new_mode not in [0, 1]: + raise ValueError("Illegal value for persistence mode, should be either 0 or 1") + if self.persistence_mode == new_mode: + return + try: + self.lib.set_persistence_mode_b(self.dev, new_mode) + except NVMLError_NoPermission: + self._sudo_nvidia_smi("-pm", str(new_mode)) + self._persistence_mode = self.lib.persistence_mode(self.dev) + + def set_clocks(self, mem_clock, gr_clock): + """Set the memory and graphics clock for this device (may require permission).""" + mem_clock = min(self.supported_mem_clocks, key=lambda x: abs(x - mem_clock)) + if gr_clock not in self.supported_gr_clocks[mem_clock]: + raise ValueError( + f"Graphics clock incompatible with memory clock ({mem_clock}), compatible graphics clocks: {self.supported_gr_clocks[mem_clock]}" + ) + + # Without persistence mode, setting the clocks is not meaningful, so if setting it fails, setting the clocks fails + if self.persistence_mode != 1: + self.persistence_mode = 1 + + self._restore_at_exit() + self.modified_clocks = True + if self.use_locked_clocks: + try: + self.lib.set_locked_clocks_b(self.dev, int(mem_clock), int(gr_clock)) + except NVMLError_NoPermission: + self._sudo_nvidia_smi(f"--lock-gpu-clocks={gr_clock},{gr_clock}") + self._sudo_nvidia_smi(f"--lock-memory-clocks={mem_clock},{mem_clock}") + self.locked_gr_clock = gr_clock + self.locked_mem_clock = mem_clock + else: + try: + self.lib.set_applications_clocks_b(self.dev, int(mem_clock), int(gr_clock)) + except NVMLError_NoPermission: + self._sudo_nvidia_smi(f"--applications-clocks={mem_clock},{gr_clock}") + self.applications_gr_clock = gr_clock + self.applications_mem_clock = mem_clock + + def reset_clocks(self): + """Reset the clocks to the default clock if the device uses a non default clock.""" + if self.use_locked_clocks: + try: + self.lib.reset_locked_clocks_b(self.dev) + except NVMLError_NoPermission: + self._sudo_nvidia_smi("--reset-gpu-clocks") + self._sudo_nvidia_smi("--reset-memory-clocks") + elif self.gr_clock_default is not None: + gr_app_clock = self.lib.applications_clock(self.dev, "graphics") + mem_app_clock = self.lib.applications_clock(self.dev, "mem") + if gr_app_clock != self.gr_clock_default or mem_app_clock != self.mem_clock_default: + self.set_clocks(self.mem_clock_default, self.gr_clock_default) + + @property + def gr_clock(self): + """Control the graphics clock (may require permission), only values compatible with the memory clock can be set directly.""" + return self.lib.clock(self.dev, "graphics") + + @gr_clock.setter + def gr_clock(self, new_clock): + if self.use_locked_clocks: + if new_clock != self.locked_gr_clock: + self.set_clocks(self.mem_clock, new_clock) + elif new_clock != self.lib.applications_clock(self.dev, "graphics"): + self.set_clocks(self.applications_mem_clock, new_clock) + + @property + def mem_clock(self): + """Control the memory clock (may require permission), only values compatible with the graphics clock can be set directly.""" + return self.lib.clock(self.dev, "mem") + + @mem_clock.setter + def mem_clock(self, new_clock): + if self.use_locked_clocks: + if new_clock != self.locked_mem_clock: + self.set_clocks(new_clock, self.gr_clock) + elif new_clock != self.lib.applications_clock(self.dev, "mem"): + self.set_clocks(new_clock, self.applications_gr_clock) + + @property + def temperature(self): + """Get the GPU temperature.""" + return self.lib.temperature(self.dev) + + @property + def auto_boost(self): + """Control the auto boost setting (may require permission), 0 for disable, 1 for enabled.""" + return self._auto_boost + + @auto_boost.setter + def auto_boost(self, setting): + if setting not in [0, 1]: + raise ValueError("Illegal value for auto boost enabled, should be either 0 or 1") + self.lib.set_auto_boost_b(self.dev, setting) + self._auto_boost = self.lib.auto_boost(self.dev) + + def pwr_usage(self): + """Return current power usage in milliwatts.""" + return self.lib.power(self.dev) + + def energy_usage(self): + """Return total energy usage since the driver was loaded in millijoules.""" + return self.lib.energy(self.dev) + + def gr_voltage(self): + """Return current graphics voltage in millivolts, NVML does not report it, so it is read from nvidia-smi.""" + args = [self.nvidia_smi, "-i", str(self.id), "-q", "-d", "VOLTAGE"] + try: + result = subprocess.run(args, check=True, capture_output=True) + m = re.search(r"(\d+\.\d+) mV", result.stdout.decode()) + return float(m.group(1)) + except Exception: + return np.nan + + +class JuliaNVMLObserver(NVMLObserver): + """Observer for the Julia backend that uses NVML through CUDA.jl to monitor power, energy, clock frequencies, voltages and temperature. + + Takes the same arguments and supports the same observables and tunable parameters (``nvml_pwr_limit``, + ``nvml_gr_clock``, and ``nvml_mem_clock``) as the :class:`~kernel_tuner.observers.nvml.NVMLObserver`, see there. + By default, the observer uses the GPU that the Julia backend runs on, pass ``device`` (the index as reported by + nvidia-smi) to observe another one. + """ + + # Minimum time in seconds between two power readings. During continuous benchmarking, the power is polled every + # few microseconds, while NVML updates it much less often (e.g. every 500 ms on an RTX A6000). Every call into + # Julia allocates memory, which can trigger a garbage collection pause during the measurement. + power_read_interval = 1e-3 + _last_power = None + _last_power_time = 0.0 + + def read_power(self): + """Return power in Watt, read at most once every ``power_read_interval`` seconds.""" + now = time.perf_counter() + if self._last_power is None or now - self._last_power_time >= self.power_read_interval: + self._last_power, self._last_power_time = super().read_power(), now + return self._last_power + + def register_device(self, dev): + cuda_device = getattr(dev, "backend_device", None) if getattr(dev, "backend_mod_name", None) == "CUDA" else None + env = getattr(dev, "env", dict()) + uuid = env.get("uuid") + pci_bus = env.get("pci_bus_id") + + if self.device is not None: + self.nvml = JuliaNVML(device_id=self.device, **self.nvml_kwargs) + elif cuda_device is not None: + self.nvml = JuliaNVML(cuda_device=cuda_device, **self.nvml_kwargs) + elif uuid is not None: + self.nvml = JuliaNVML(device_uuid=uuid, **self.nvml_kwargs) + elif pci_bus is not None: + self.nvml = JuliaNVML(device_pci_bus=pci_bus, **self.nvml_kwargs) + else: + raise ValueError( + f"failed to detect NVIDIA device: not a Julia CUDA backend, and no UUID or PCI-bus-id in environment: {env}" + ) + + +# High-level Helper functions, the same as those in kernel_tuner.observers.nvml + + +def get_nvml_pwr_limits(device, n=None, quiet=False): + """Get tunable parameter for NVML power limits, n is desired number of values.""" + d = JuliaNVML(device) + # Min and Max converted to Watt + power_limit_min = d.pwr_constraints[0] * 1e-3 + power_limit_max = d.pwr_constraints[-1] * 1e-3 + power_limit_round = 5 + if n is None: + n = int((power_limit_max - power_limit_min) / power_limit_round) + 1 + + # Rounded power limit values + power_limits = power_limit_round * np.round((np.linspace(power_limit_min, power_limit_max, n) / power_limit_round)) + tune_params = {"nvml_pwr_limit": sorted(set(int(power_limit) for power_limit in power_limits))} + + if not quiet: + print("Using power limits:", tune_params["nvml_pwr_limit"]) + return tune_params + + +def get_nvml_gr_clocks(device, n=None, quiet=False): + """Get tunable parameter for NVML graphics clock, n is desired number of values.""" + d = JuliaNVML(device) + gr_clocks = d.supported_gr_clocks[max(d.supported_mem_clocks)] + + if n and (len(gr_clocks) > n): + indices = np.array(np.ceil(np.linspace(0, len(gr_clocks) - 1, n)), dtype=int) + gr_clocks = [gr_clocks[i] for i in indices] + + tune_params = {"nvml_gr_clock": list(gr_clocks)} + + if not quiet: + print("Using gr frequencies:", tune_params["nvml_gr_clock"]) + return tune_params + + +def get_nvml_mem_clocks(device, n=None, quiet=False): + """Get tunable parameter for NVML memory clock, n is desired number of values.""" + d = JuliaNVML(device) + mem_clocks = d.supported_mem_clocks + + if n and len(mem_clocks) > n: + mem_clocks = mem_clocks[:: int(len(mem_clocks) / n)] + + tune_params = {"nvml_mem_clock": mem_clocks} + + if not quiet: + print("Using mem frequencies:", tune_params["nvml_mem_clock"]) + return tune_params + + +def get_idle_power(device, n=5, sleep_s=0.1): + """Use NVML to measure device idle power consumption in Watt.""" + d = JuliaNVML(device) + readings = [] + for _ in range(n): + time.sleep(sleep_s) + readings.append(d.pwr_usage()) + return np.mean(readings) * 1e-3 diff --git a/kernel_tuner/observers/metal.py b/kernel_tuner/observers/metal.py new file mode 100644 index 00000000..29c44d04 --- /dev/null +++ b/kernel_tuner/observers/metal.py @@ -0,0 +1,322 @@ +import logging +import os +import plistlib +import select +import signal +import subprocess +import time +from warnings import warn + +import numpy as np + +from kernel_tuner.observers import BenchmarkObserver, ContinuousObserver + +logger = logging.getLogger(__name__) +PLIST_END = b"" + +SUPPORTED_OBSERVABLES = [ + "energy", + "power", + "freq_hz", + "load", +] + +FIELD_TYPE_MAP = { + "energy": float, + "power": float, + "freq_hz": int, + "load": float, +} + + +class MetalDevice: + """Wrapper for Metal GPU device metrics via powermetrics.""" + + def __init__(self, interval_ms=50): + self.interval_ms = interval_ms + self.process = None + self._samples = [] + self._buffer = b"" + self._window_start = None + self.cmd_powermetrics_options = [ + "-i", + str(self.interval_ms), + "--samplers", + "gpu_power", + "-a", + "0", + "-f", + "plist", + ] + self.has_noninteractive_permissions = self.check_permissions() + + def check_permissions(self) -> bool: + """Check if the user has non-interactive sudo permissions to run powermetrics.""" + try: + # check if we have non-interactive sudo rights on powermetrics + cmd_check = ["sudo", "-n", "powermetrics", "-n", "1", *self.cmd_powermetrics_options] + subprocess.run(cmd_check, capture_output=True, text=True, check=True) + except subprocess.CalledProcessError as e: + if "password is required" in e.stderr: + logger.warning( + f"Metal observers ideally have non-interactive sudo privileges for powermetrics; {e}, {e.stderr}" + ) + warn( + "Metal observers ideally have non-interactive sudo privileges for powermetrics. Please run `sudo visudo` and add to the bottom: `your_username ALL=(ALL) NOPASSWD: /usr/bin/powermetrics`" + ) + return False + return True + + def start_sampling(self): + """Start powermetrics and wait for its first sample, which marks the start of the measurement window. + + powermetrics emits a sample every ``interval_ms``. Each sample holds the energy used since the previous sample, + so the samples after the first one together cover the measurement window without gaps. + The output is read without blocking whenever the observer is called. No background thread is used, as that + can deadlock when Kernel Tuner runs inside Julia (juliacall). If the pipe fills up in between, powermetrics + waits, and its next sample covers the longer interval, so no energy is lost. + """ + if self.process: # left over from a measurement that was interrupted, e.g. by a failing kernel + self.stop_sampling() + cmd = ["sudo", "powermetrics", *self.cmd_powermetrics_options] + try: + self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) + except Exception as e: + logger.error(f"Failed to start powermetrics: {e}") + raise e + os.set_blocking(self.process.stdout.fileno(), False) + self._buffer = b"" + self._samples = [] + + # the first sample covers the time before the window, it only marks its start + timeout = max(5.0, 10 * self.interval_ms / 1e3) # starting sudo and powermetrics can take a while + if not self._read_until(lambda: len(self._samples) >= 1, timeout): + logger.error("powermetrics did not produce a sample in time") + self._window_start = time.perf_counter() + + def read_available(self): + """Read and parse the samples that powermetrics has written so far, without blocking.""" + if not self.process: + return + while True: + try: + data = os.read(self.process.stdout.fileno(), 65536) + except BlockingIOError: + break + if not data: # powermetrics has exited + break + self._buffer += data + while PLIST_END in self._buffer: + end = self._buffer.index(PLIST_END) + len(PLIST_END) + document, self._buffer = self._buffer[:end], self._buffer[end:] + start = document.find(b" 1 and sum(s["elapsed_s"] for s in self._samples[1:]) >= window_duration - 2e-3 + ) + if not self._read_until(covered, timeout=max(1.0, 3 * self.interval_ms / 1e3)): + logger.warning("powermetrics samples do not cover the full measurement window") + + self.process.terminate() + try: + self.process.wait(timeout=(self.interval_ms / 1e3) * 3) + except subprocess.TimeoutExpired: + logger.warning("powermetrics process did not terminate in time") + self.process.stdout.close() + self.process = None + return self._samples[1:] + + def _parse_sample(self, data): + """Parse a single powermetrics plist sample.""" + gpu_data = data.get("gpu", {}) + # units are documented in: https://docs.rs/pumas/latest/src/pumas/modules/powermetrics/plist_parsing.rs.html + return { + "energy_j": gpu_data.get("gpu_energy", 0) * 1e-3, # integer millijoules since the previous sample + "elapsed_s": data.get("elapsed_ns", self.interval_ms * 1e6) / 1e9, + "freq_hz": gpu_data.get("freq_hz", np.nan) * 1e6, # reported in MHz + "idle_ratio": gpu_data.get("idle_ratio", np.nan), + } + + @staticmethod + def summarize(samples, min_load=0.0): + """Combine the samples of a measurement window into the energy, average power, frequency, and load. + + Samples in which the GPU was mostly idle (load below ``min_load``) are left out, unless that would leave none. + Energy is reported in whole millijoules per sample, so it is summed over the window before computing the power. + """ + samples = [s for s in samples if s["elapsed_s"] > 0] + busy = [s for s in samples if not (1.0 - s["idle_ratio"] < min_load)] + samples = busy if busy else samples + if not samples: + return None + duration = sum(s["elapsed_s"] for s in samples) + energy = sum(s["energy_j"] for s in samples) + weighted = lambda key: sum(s[key] * s["elapsed_s"] for s in samples) / duration + return { + "energy": energy, + "power": energy / duration, + "freq_hz": weighted("freq_hz"), + "load": 1.0 - weighted("idle_ratio"), + "duration": duration, + } + + +class MetalContinuousObserver(ContinuousObserver): + """Continuous observer for Metal that wraps a parent observer. + + Kernel Tuner runs the kernel back-to-back for ``continuous_duration`` seconds. The parent measures the average + power over that run, and the energy per kernel execution is that power times the execution time of the kernel. + """ + + def __init__(self, parent, continuous_duration=1.0, interval_ms=50): + self.parent = parent + self.continuous_duration = continuous_duration + self.interval_ms = interval_ms + + # This assigned by Kernel Tuner's core + self.results = None + + def before_start(self): + self.parent.before_start() + + def after_start(self): + self.parent.after_start() + + def during(self): + self.parent.during() + + def after_finish(self): + self.parent.after_finish() + + def get_results(self): + results = self.parent.get_results() + energy_field = self.parent.field_name("energy") + if energy_field in results: + # the parent measured the energy of the entire run, scale it to a single kernel execution + time_sec = self.results["time"] * 1e-3 + results[energy_field] = FIELD_TYPE_MAP["energy"](self.parent.last_power * time_sec) + return results + + +class MetalObserver(BenchmarkObserver): + """BenchmarkObserver that uses powermetrics to monitor Apple Metal GPUs and measure + energy usage (`energy`), power (`power`), GPU frequency (`freq_hz`), and load (`load`). + """ + + def __init__( + self, + observables=SUPPORTED_OBSERVABLES, + *, + device_id=None, + prefix=None, + use_continuous_observer=True, + continuous_duration=1.0, + interval_ms=50, + min_load=0.1, + ): + """Initialize the MetalObserver. + + Supported observables are: `energy`, `power`, `freq_hz`, and `load`. + + :param observables: List of metrics to monitor. Defaults to all as there is no performance penalty for monitoring all metrics. + :param device_id: Not used for Metal (single GPU), kept for API compatibility. + :param prefix: Prefix used for name in the metrics. Defaults to "metal". + :param use_continuous_observer: Whether to use continuous observer. The continuous observer measures the average power while + the kernel runs back-to-back for `continuous_duration` seconds, and reports the energy per kernel execution as that power + times the execution time. Without it, each kernel execution is measured on its own, and the energy covers the whole + measurement window of at least one sampling interval, which is only meaningful for kernels that run much longer than `interval_ms`. + :param continuous_duration: Duration in seconds for continuous observation. + :param interval_ms: Sampling interval in milliseconds for powermetrics. Default 50ms as per https://arxiv.org/html/2511.07885v6. Going below 50ms may cause instability issues. + :param min_load: Minimum threshold for GPU load. Samples below this load are discarded. + """ + for obs in observables: + if obs not in SUPPORTED_OBSERVABLES: + raise ValueError(f"Observable {obs} not supported: {SUPPORTED_OBSERVABLES}") + + self.observables = set(observables) + self.prefix = prefix + self.device_id = device_id + self.device = None + self.use_continuous_observer = use_continuous_observer + self.passthrough_observer = ( + use_continuous_observer # This observer is only used as a passthrough for the continuous observer + ) + self.continuous_duration = continuous_duration + self.interval_ms = interval_ms + self.min_load = min_load + self.results_per_iteration = {self.field_name(k): [] for k in self.observables} + + if min_load < 0.0 or min_load > 1.0: + raise ValueError(f"min_load must be between 0.0 and 1.0, got {min_load}") + + def register_device(self, dev): + """Initialize the Metal device for monitoring.""" + self.device = MetalDevice(interval_ms=self.interval_ms) + + if self.use_continuous_observer: + self.continuous_observer = MetalContinuousObserver(self, continuous_duration=self.continuous_duration) + + def field_name(self, name): + if self.prefix: + return f"{self.prefix}_{name}" + else: + return name + + def before_start(self): + """Start the measurement window before the kernel is launched.""" + self.device.start_sampling() + + def during(self): + """Read the samples written so far, so that the output pipe of powermetrics does not fill up.""" + self.device.read_available() + + def after_finish(self): + """End the measurement window once the kernel has finished, and combine its samples.""" + summary = MetalDevice.summarize(self.device.stop_sampling(), self.min_load) + if summary is None: + logger.warning("No powermetrics samples were collected for this measurement") + return + self.last_power = summary["power"] + for key in self.observables: + self.results_per_iteration[self.field_name(key)].append(summary[key]) + + def get_results(self): + """Return averaged results across iterations.""" + results = dict() + + for key in self.results_per_iteration.keys(): + if self.results_per_iteration[key]: + field_name = key.replace(f"{self.prefix}_", "") if self.prefix else key + results[key] = FIELD_TYPE_MAP[field_name](np.average(self.results_per_iteration[key])) + self.results_per_iteration[key] = [] + + return results diff --git a/kernel_tuner/observers/nvml.py b/kernel_tuner/observers/nvml.py index eb5bbaf4..cf83a031 100644 --- a/kernel_tuner/observers/nvml.py +++ b/kernel_tuner/observers/nvml.py @@ -1,6 +1,7 @@ import re import subprocess import time +from warnings import warn import numpy as np @@ -12,16 +13,32 @@ pynvml = None +SUPPORTED_OBSERVABLES = [ + "power_readings", + "nvml_power", + "nvml_energy", + "core_freq", + "mem_freq", + "temperature", + "gr_voltage", +] + +FIELD_TYPE_MAP = { + "power_readings": list, + "nvml_power": float, + "nvml_energy": float, + "core_freq": int, + "mem_freq": int, + "temperature": float, + "gr_voltage": float, +} + + class nvml: """Class that gathers the NVML functionality for one device.""" def __init__( - self, - device_id=None, - device_uuid=None, - device_pci_bus=None, - nvidia_smi_fallback=None, - use_locked_clocks=False + self, device_id=None, device_uuid=None, device_pci_bus=None, nvidia_smi_fallback=None, use_locked_clocks=False ): """Create object to control device using NVML.""" # We set these first as __del__ checks these @@ -181,7 +198,6 @@ def persistence_mode(self, new_mode): def set_clocks(self, mem_clock, gr_clock): """Set the memory and graphics clock for this device (may require permission).""" - mem_clock = min(self.supported_mem_clocks, key=lambda x: abs(x - mem_clock)) if mem_clock not in self.supported_mem_clocks: raise ValueError("Illegal value for memory clock") @@ -222,6 +238,7 @@ def set_clocks(self, mem_clock, gr_clock): # Store the fact that we have modified the clocks if not self.modified_clocks: import atexit + atexit.register(self.reset_clocks) self.modified_clocks = True @@ -325,6 +342,65 @@ def gr_voltage(self): return np.nan +class NVMLContinuousObserver(ContinuousObserver): + """Continuous observer that measures the average power from the change in the energy counter of NVML. + + NVML updates its power readings infrequently (e.g. every 500 ms on an RTX A6000), so the median of the power + readings in a measurement window of about a second rests on only a few values. The cumulative energy counter is + updated more often (e.g. every 100 ms). The observer records when the counter changes during the window; the energy + between the first and the last change, divided by the time between them, is the average power. As both ends are + taken at a change of the counter, the delay with which the counter is updated cancels out. + + If the counter changes fewer than twice in the window (e.g. the window is too short, or the device has no energy + counter), the power is the median of the power readings instead. The same happens, with a warning, if the power + from the counter differs more than `max_disagreement` times from the mean of the power readings: on some systems + the counter is wrong (e.g. it increased about 15 times too slowly on an RTX A6000 with driver 595.91.07). + """ + + # minimum time in seconds between two readings of the energy counter, the counter changes much less often + energy_read_interval = 1e-3 + # maximum factor between the power from the counter and the mean of the power readings before the counter is distrusted + max_disagreement = 2.0 + _warned = False + + def before_start(self): + super().before_start() + self.energy_readings = [] # [time in s, energy counter in mJ], every time the counter changed + self._last_energy_read = None + + def during(self): + super().during() + now = time.perf_counter() + if self._last_energy_read is None or now - self._last_energy_read >= self.energy_read_interval: + self._last_energy_read = now + try: + energy = self.parent.nvml.energy_usage() + except Exception: + return # no energy counter, the power readings are used instead + if not self.energy_readings or energy != self.energy_readings[-1][1]: + self.energy_readings.append([now - self.t0, energy]) + + def after_finish(self): + super().after_finish() + # the first reading only marks the counter value before the first observed change + changes = self.energy_readings[1:] + if len(changes) < 2: + return + (t_first, e_first), (t_last, e_last) = changes[0], changes[-1] + counter_power = (e_last - e_first) / 1e3 / (t_last - t_first) + readings_power = np.mean([power for _, power in self.power_readings]) + if not readings_power / self.max_disagreement <= counter_power <= readings_power * self.max_disagreement: + if not self._warned: + warn( + f"The power from NVML's energy counter ({counter_power:.1f} W) differs more than {self.max_disagreement}x " + f"from its power readings ({readings_power:.1f} W), using the power readings instead" + ) + self._warned = True + return + self.power = counter_power + self.energy = self.power * self.results["time"] / 1e3 + + class NVMLObserver(BenchmarkObserver): """Observer that uses NVML to monitor power, energy, clock frequencies, voltages and temperature. @@ -333,7 +409,9 @@ class NVMLObserver(BenchmarkObserver): :param observables: List of quantities that should be observed during tuning, supported are: "power_readings", "nvml_power", "nvml_energy", "core_freq", "mem_freq", "temperature", "gr_voltage". If you want to measure the average power - consumption of a GPU kernel executing on the GPU use "nvml_power". The "power_readings" are the individual power readings + consumption of a GPU kernel executing on the GPU use "nvml_power", in Watt; "nvml_energy" is the energy of a single + kernel execution in Joule. Both are measured while the kernel runs back-to-back for `continuous_duration` seconds, + see also `use_energy_counter`. The "power_readings" are the individual power readings as reported by NVML and will return a lot of data if you are benchmarking many different kernel configurations. :type observables: list of strings @@ -357,6 +435,12 @@ class NVMLObserver(BenchmarkObserver): :param continuous_duration: Duration to use for energy/power measurements in seconds, default 1 second. :type continuous_duration: float + :param use_energy_counter: Measure "nvml_power" and "nvml_energy" from the change in NVML's energy counter, which is + updated more often than its power readings, instead of from the median of the power readings. If the counter + disagrees strongly with the power readings, as it does on some systems, the power readings are used with a warning. + See :class:`NVMLContinuousObserver`. Default is False. + :type use_energy_counter: boolean + """ def __init__( @@ -367,6 +451,7 @@ def __init__( nvidia_smi_fallback=None, use_locked_clocks=False, continuous_duration=1, + use_energy_counter=False, ): """Create an NVMLObserver.""" self.save_all = save_all @@ -375,18 +460,9 @@ def __init__( self.nvml_kwargs["use_locked_clocks"] = use_locked_clocks self.nvml_kwargs["nvidia_smi_fallback"] = nvidia_smi_fallback - supported = [ - "power_readings", - "nvml_power", - "nvml_energy", - "core_freq", - "mem_freq", - "temperature", - "gr_voltage", - ] for obs in observables: - if obs not in supported: - raise ValueError(f"Observable {obs} not in supported: {supported}") + if obs not in SUPPORTED_OBSERVABLES: + raise ValueError(f"Observable {obs} not in supported: {SUPPORTED_OBSERVABLES}") self.observables = observables self.measure_power = False @@ -394,7 +470,10 @@ def __init__( if any([obs in self.needs_power for obs in observables]): self.measure_power = True power_observables = [obs for obs in observables if obs in self.needs_power] - self.continuous_observer = ContinuousObserver("nvml", power_observables, self, continuous_duration=continuous_duration) + continuous_observer = NVMLContinuousObserver if use_energy_counter else ContinuousObserver + self.continuous_observer = continuous_observer( + "nvml", power_observables, self, continuous_duration=continuous_duration + ) # remove power observables self.observables = [obs for obs in observables if obs not in self.needs_power] @@ -428,14 +507,14 @@ def register_device(self, dev): elif pci_bus is not None: self.nvml = nvml(device_pci_bus=pci_bus, **self.nvml_kwargs) else: - raise ValueError("failed to detect NVIDIA device: no UUID or PCI-bus-id in environment") + raise ValueError(f"failed to detect NVIDIA device: no UUID or PCI-bus-id in environment: {env}") def read_power(self): - """ Return power in Watt """ + """Return power in Watt""" return self.nvml.pwr_usage() / 1e3 def read_energy(self): - """ Return cumulative energy usage in Joule """ + """Return cumulative energy usage in Joule""" now = self.nvml.energy_usage() if self.initial_energy_reading is None: @@ -473,12 +552,8 @@ def after_finish(self): self.results["mem_freqs"].append(np.average(self.iteration["mem_freq"])) if "gr_voltage" in self.observables: - execution_time = time.time() - self.t0 - gr_voltage_readings = self.gr_voltage_readings - gr_voltage_readings = [[0.0, gr_voltage_readings[0][1]]] + gr_voltage_readings - gr_voltage_readings = gr_voltage_readings + [[execution_time, gr_voltage_readings[-1][1]]] - # time in s, graphics voltage in millivolts - self.results["gr_voltages"].append(np.average(gr_voltage_readings[:][:][1])) + # readings are [time in s, graphics voltage in millivolts] + self.results["gr_voltages"].append(np.average([voltage for _, voltage in self.gr_voltage_readings])) def get_results(self): averaged_results = {} @@ -567,4 +642,3 @@ def get_idle_power(device, n=5, sleep_s=0.1): time.sleep(sleep_s) readings.append(d.pwr_usage()) return np.mean(readings) * 1e-3 # Watt - diff --git a/kernel_tuner/observers/observer.py b/kernel_tuner/observers/observer.py index bcf661c8..4cd14926 100644 --- a/kernel_tuner/observers/observer.py +++ b/kernel_tuner/observers/observer.py @@ -4,6 +4,7 @@ class BenchmarkObserver(ABC): """Base class for Benchmark Observers""" + passthrough_observer = False # do not execute as a benchmark observer, only used as a passthrough for other observers def register_device(self, dev): """Sets self.dev, for inspection by the observer at various points during benchmarking""" @@ -96,9 +97,9 @@ def during(self): def after_finish(self): self.parent.after_finish() - # safeguard in case we have no measurements, perhaps the kernel was too short to measure anything + # the run may have finished before the observer was polled, take at least one reading instead of reporting zero if not self.power_readings: - return + self.power_readings.append([time.perf_counter() - self.t0, self.parent.read_power()]) # convert to seconds from milliseconds execution_time = self.results["time"] / 1e3 diff --git a/kernel_tuner/util.py b/kernel_tuner/util.py index a917b68f..14fea809 100644 --- a/kernel_tuner/util.py +++ b/kernel_tuner/util.py @@ -86,7 +86,9 @@ def default(self, obj): return super(NpEncoder, self).default(obj) -def get_result_cost(result: dict, objectives: list[str], objective_higher_is_better: list[bool], invalid_value) -> list[float]: +def get_result_cost( + result: dict, objectives: list[str], objective_higher_is_better: list[bool], invalid_value +) -> list[float]: """Returns the cost of a result, taking the objective directions into account.""" # return the highest cost for invalid results if "__error__" in result: @@ -121,6 +123,40 @@ class StopCriterionReached(Exception): """Exception thrown when a stop criterion has been reached.""" +def is_interrupt(exception) -> bool: + """Whether `exception`, or an exception that caused it, is an interrupt (Ctrl-C). + + With the Julia backend, Julia handles the interrupt signal, so Ctrl-C arrives as a Julia `InterruptException`, + wrapped in a `JuliaError` by JuliaCall, and possibly re-raised as another exception by the backend. + """ + seen = set() + pending = [exception] + while pending: + exc = pending.pop() + if exc is None or id(exc) in seen: + continue + seen.add(id(exc)) + if isinstance(exc, KeyboardInterrupt): + return True + if type(exc).__name__ == "JuliaError": + julia_exception = repr(getattr(exc, "exception", "")) + # an interrupt in Julia, or a KeyboardInterrupt in Python code called from Julia + if "InterruptException" in julia_exception or "KeyboardInterrupt" in julia_exception: + return True + pending.extend([exc.__cause__, exc.__context__]) + return False + + +def raise_if_interrupt(exception): + """Raise a `KeyboardInterrupt` if `exception` is caused by an interrupt, so that it stops tuning. + + Use this in handlers that catch any exception, as an interrupt arriving through Julia (see `is_interrupt`) would + otherwise only make the current configuration fail. + """ + if is_interrupt(exception): + raise KeyboardInterrupt() from exception + + try: import torch except ImportError: @@ -1382,7 +1418,7 @@ def check_matching_problem_size(cached_problem_size, problem_size): def open_cachefile(cachefile, mode): - """ Open a cachefile for reading or writing (depending on mode) """ + """Open a cachefile for reading or writing (depending on mode)""" if cachefile[-3:] == ".gz": return gzip.open(cachefile, mode) return open(cachefile, mode) diff --git a/noxfile.py b/noxfile.py index f93da3e5..3b6cd5b5 100644 --- a/noxfile.py +++ b/noxfile.py @@ -21,7 +21,7 @@ # set the test parameters verbose = False -python_versions_to_test = ["3.14", "3.13", "3.12", "3.11"] +python_versions_to_test = ["3.15", "3.14", "3.13", "3.12", "3.11"] nox.options.stop_on_first_error = True nox.options.error_on_missing_interpreters = True nox.options.default_venv_backend = "virtualenv" @@ -72,9 +72,9 @@ def create_settings(session: Session) -> None: # obtain workspace level settings from the 'noxsettings.toml' file if settings_file_path.exists(): with settings_file_path.open(mode="rb") as fp: - import tomli + import tomllib - nox_settings = tomli.load(fp) + nox_settings = tomllib.load(fp) venvbackend = nox_settings["venvbackend"] envdir = nox_settings["envdir"] assert venvbackend in venvbackend_values, ( @@ -145,6 +145,7 @@ def tests(session: Session) -> None: small_disk = False skip_gpu = False github_action = False + session_posargs = [] if session.posargs: for arg in session.posargs: if arg.lower() == "skip-gpu": @@ -171,7 +172,8 @@ def tests(session: Session) -> None: elif arg.lower() == "github-action": github_action = True else: - raise ValueError(f"Unrecognized argument {arg}") + session_posargs.append(arg) + session.warn(f"Unrecognized argument {arg}, passing to pytest") if install_julia == False and julia_use_gpu == True: raise ValueError("Cannot use Julia GPU backend if Julia is disabled") if skip_gpu == True and julia_use_gpu == True: @@ -342,7 +344,7 @@ def tests(session: Session) -> None: # set the Julia version # when changed, also see `require_julia` in Project.toml and the Julia version in the GitHub Actions workflow if github_action: - preamble = "import juliapkg; juliapkg.require_julia('1.11')" # must match the setup-julia action + preamble = "import juliapkg; juliapkg.require_julia('1.13')" # must match the setup-julia action else: preamble = "import juliapkg; juliapkg.require_julia('1.11, 2')" # meaning >= 1.11, < 2.0 # call JuliaPKG to precompile packages in the session environment @@ -406,11 +408,11 @@ def tests(session: Session) -> None: # for the last Python version session if all optional dependencies are enabled: if session.python == python_versions_to_test[-1] and full_install: # run pytest on the package to generate the correct coverage report - session.run("pytest", external=False, env=env_vars) + session.run("pytest", *session_posargs, external=False, env=env_vars) else: # for the other Python version sessions: # run pytest without coverage reporting - session.run("pytest", "--no-cov", external=False, env=env_vars) + session.run("pytest", "--no-cov", *session_posargs, external=False, env=env_vars) # warn if no coverage report if not full_install: diff --git a/pyproject.toml b/pyproject.toml index 6977f175..ef9abdef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api" [project] name = "kernel_tuner" description = "An easy to use CUDA/OpenCL kernel tuner in Python" -version = "1.5.0" # adhere to PEP440 versioning: https://packaging.python.org/en/latest/guides/distributing-packages-using-setuptools/#id55 +version = "1.6.0" # adhere to PEP440 versioning: https://packaging.python.org/en/latest/guides/distributing-packages-using-setuptools/#id55 readme = "README.md" license = "Apache-2.0" authors = [ @@ -41,15 +41,15 @@ classifiers = [ "Topic :: Scientific/Engineering", "Topic :: Software Development", "Topic :: System :: Distributed Computing", - "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", - "Programming Language :: Python :: 3.14" + "Programming Language :: Python :: 3.14", # TODO when least supported Python version is 3.14, remove juliapkg[zstd] from the optional dependencies + "Programming Language :: Python :: 3.15" ] # ATTENTION: if anything is changed here, run `poetry update` -requires-python = ">=3.10,<4" # <4 is because of hip-python # NOTE when changing the Python versions, also change the test versions in the Noxfile and GitHub Actions +requires-python = ">=3.11,<4" # <4 is because of hip-python # NOTE when changing the Python versions, also change the test versions in the Noxfile and GitHub Actions dependencies = [ "numpy>=2.0.0", "scipy>=1.14.1", # Python >=3.13 needs scipy >=1.14 @@ -100,7 +100,11 @@ cuda = ["cuda-python>=12.6.0", "nvidia-ml-py>=12.535.108", "pynvml>=11.4.1"] # A opencl = ["pyopencl"] # Attention: if pyopencl is changed here, also change `session.install("pyopencl")` in the Noxfile cuda_opencl = ["cuda-python>=12.6.0", "pyopencl"] # Attention: if pycuda is changed here, also change `session.install("pycuda")` in the Noxfile hip = ["hip-python"] -julia = ["juliapkg>=0.1.23", "juliacall>=0.9.31"] +julia = [ + "juliapkg>=0.1.28; python_version >= '3.14'", + "juliapkg[zstd]>=0.1.28; python_version < '3.14'", # zstd is required for Julia>=1.13 and Python<3.14 due to zstd registry default + "juliacall>=0.9.37" +] tutorial = ["jupyter>=1.0.0", "matplotlib>=3.5.0", "nvidia-ml-py>=12.535.108"] # ATTENTION: if anything is changed here, run `poetry update` and `poetry export --with docs --without-hashes --format=requirements.txt --output doc/requirements.txt` @@ -130,7 +134,6 @@ nox = "^2024.4.15" nox-poetry = "^1.0.3" ruff = "^0.15.0" pep440 = "^0.1.2" -tomli = "^2.0.1" # held back by Python <= 3.10, can be replaced by built-in [tomllib](https://docs.python.org/3.11/library/tomllib.html) from Python 3.11 onwards scikit-optimize = "0.10.2" # development dependencies are unused for now, as this is already covered by test and docs diff --git a/test/test_julia_functions.py b/test/test_julia_functions.py index d8666cb5..fddf43f5 100644 --- a/test/test_julia_functions.py +++ b/test/test_julia_functions.py @@ -2,7 +2,8 @@ import numpy as np import pytest -from kernel_tuner import tune_kernel +from kernel_tuner.backends.julia_helper import detect_julia_gpu_backends +from kernel_tuner import run_kernel, tune_kernel from kernel_tuner.backends.julia import JuliaFunctions from kernel_tuner.core import KernelInstance, KernelSource @@ -24,6 +25,20 @@ end """ +axpy_kernel_name = "axpy!" +axpy_kernel_string = r""" + using KernelAbstractions + + @kernel function axpy!( + c, a, b, alpha::Float32, n, ::Val{block_size_x} = Val(128) + ) where {block_size_x} + i = @index(Global) + if i <= n + c[i] += alpha * a[i] + b[i] + end + end + """ + @skip_if_no_julia def test_ready_argument_list(): @@ -71,3 +86,134 @@ def test_tune_kernel(env): result, _ = tune_kernel(*env, lang="julia", verbose=True) assert len(result) > 0 + for r in result: + if '__error__' in r: + continue # Skip configurations that failed to compile or run + assert "time" in r + assert r["time"] > 0 + +@skip_if_no_julia +@pytest.mark.parametrize("env", [100, 100000000], indirect=True) # test with a larger problem size to ensure that we can measure meaningful power and energy usage +def test_tune_kernel_observers(env): + """Run a minimal Julia kernel tuner example with observers.""" + env[0] = kernel_name + env[1] = kernel_string + env[4] = list(env[4].items()) # convert from a dict to a list of tuples to preserve order + size = env[2] + + # create an observer based on the backend + backend = detect_julia_gpu_backends()[0] + if backend.lower() == "metal": + from kernel_tuner.observers.metal import MetalObserver, SUPPORTED_OBSERVABLES, FIELD_TYPE_MAP + observers = [MetalObserver(observables=SUPPORTED_OBSERVABLES)] + expected_observables = SUPPORTED_OBSERVABLES + elif backend.lower() == "cuda": + from kernel_tuner.observers.julia_nvml import JuliaNVMLObserver, SUPPORTED_OBSERVABLES, FIELD_TYPE_MAP + # gr_voltage is not supported on all GPUs + expected_observables = [o for o in SUPPORTED_OBSERVABLES if o != "gr_voltage"] + observers = [JuliaNVMLObserver(observables=expected_observables)] + else: + observers = [] + expected_observables = [] + + # tune the kernel with the observers + result, _ = tune_kernel(*env, observers=observers, lang="julia", verbose=True) + + # Define the numerical observables + expected_numerical_observables = [o for o in expected_observables if FIELD_TYPE_MAP[o] in [int, float]] or [] + + # check that the results contain the expected observables + assert len(result) > 0 + for r in result: + if '__error__' in r: + continue # Skip configurations that failed to compile or run + assert "time" in r + assert r["time"] > 0 + for observable in expected_observables: + assert observable in r, f"Expected {observable=} not found in config {r}" + for observable in expected_numerical_observables: + assert isinstance(r[observable], (int, float)) + if size >= 100000000: # only check for positive values if the problem size is large enough to produce meaningful measurements + assert r[observable] > 0, f"Expected {observable=} to be > 0 in config {r}" + + +@pytest.fixture +def axpy_env(): + """Arguments and expected output for a kernel that accumulates into c for the first n elements only.""" + size = 1000 + rng = np.random.default_rng(42) + a = rng.standard_normal(size).astype(np.float32) + b = rng.standard_normal(size).astype(np.float32) + c = rng.standard_normal(size).astype(np.float32) + alpha = np.float32(2.5) + n = np.int32(900) + + expected = c.copy() + expected[:n] += alpha * a[:n] + b[:n] + + return size, [c, a, b, alpha, n], expected + + +def to_julia_array(arr): + """Convert a numpy array into a Julia Vector{Float32}.""" + from juliacall import Main as jl + + return jl.seval("x -> Vector{Float32}(x)")(arr) + + +@skip_if_no_julia +@pytest.mark.parametrize("input_type", ["numpy", "julia"]) +def test_run_kernel_arguments_roundtrip(axpy_env, input_type): + """Check that arrays and scalars are passed to and from a Julia kernel correctly.""" + size, args, expected = axpy_env + host_args = [np.copy(arg) for arg in args] + if input_type == "julia": + args[:3] = [to_julia_array(arg) for arg in args[:3]] + + result = run_kernel(axpy_kernel_name, axpy_kernel_string, size, args, [("block_size_x", 128)], lang="julia") + + assert all(isinstance(r, np.ndarray) for r in result[:3]) + assert np.allclose(result[0], expected, atol=1e-6) + # elements beyond n are only correct if n was passed correctly + assert np.array_equal(result[0][args[4]:], host_args[0][args[4]:]) + assert np.array_equal(result[1], host_args[1]) + assert np.array_equal(result[2], host_args[2]) + assert result[3] == host_args[3] + assert result[4] == host_args[4] + # the user's arguments are not modified + for arg, host_arg in zip(args, host_args): + assert np.array_equal(np.asarray(arg), host_arg) + + +@skip_if_no_julia +@pytest.mark.parametrize("input_type", ["numpy", "julia"]) +@pytest.mark.parametrize("answer_type", ["numpy", "julia"]) +def test_tune_kernel_verifies_answer(axpy_env, input_type, answer_type): + """Check output verification with numpy and Julia arrays, output is reset between configurations.""" + size, args, expected = axpy_env + if input_type == "julia": + args[:3] = [to_julia_array(arg) for arg in args[:3]] + if answer_type == "julia": + expected = to_julia_array(expected) + answer = [expected, None, None, None, None] + tune_params = [("block_size_x", [64, 128, 256])] + + result, _ = tune_kernel( + axpy_kernel_name, axpy_kernel_string, size, args, tune_params, lang="julia", answer=answer, atol=1e-6 + ) + + assert len(result) == 3 + assert all("__error__" not in r for r in result) + + +@skip_if_no_julia +def test_tune_kernel_detects_wrong_answer(axpy_env, tmp_path, monkeypatch): + """Check that output verification fails when the kernel output does not match the answer.""" + monkeypatch.chdir(tmp_path) # a failed verification leaves the kernel source file behind + size, args, expected = axpy_env + answer = [expected + 1, None, None, None, None] + + with pytest.raises(RuntimeError, match="verification failed"): + tune_kernel( + axpy_kernel_name, axpy_kernel_string, size, args, [("block_size_x", [64])], lang="julia", answer=answer + ) diff --git a/test/test_julia_nvml_observer.py b/test/test_julia_nvml_observer.py new file mode 100644 index 00000000..4e85f2ca --- /dev/null +++ b/test/test_julia_nvml_observer.py @@ -0,0 +1,188 @@ +import numpy as np +import pytest + +from kernel_tuner import tune_kernel +from kernel_tuner.backends.julia_helper import julia_backend_available_cuda +from kernel_tuner.observers.julia_nvml import ( + JuliaNVML, + JuliaNVMLObserver, + NVMLError_NoPermission, + NVMLError_NotSupported, +) +from kernel_tuner.observers.nvml import NVMLObserver + +from .context import skip_if_no_julia + +skip_if_no_julia_cuda = pytest.mark.skipif(not julia_backend_available_cuda(), reason="No NVIDIA GPU available") + + +class FakeLibrary: + """Stands in for the Julia NVML functions, every setter raises NoPermission as for a user without root.""" + + def __init__(self): + self.app_clocks = {"graphics": 1800, "mem": 8001} + + def power_limit(self, dev): + return 300000 + + def applications_clock(self, dev, kind): + return self.app_clocks[kind] + + def persistence_mode(self, dev): + return 1 + + def __getattr__(self, name): + def no_permission(*args): + raise NVMLError_NoPermission(4, name) + + return no_permission + + +def fake_device(use_locked_clocks=False): + """A JuliaNVML for a device with two memory clocks, without starting Julia.""" + d = JuliaNVML.__new__(JuliaNVML) + d.lib, d.dev, d.id, d.nvidia_smi = FakeLibrary(), None, 2, "nvidia-smi" + d.pwr_limit_default, d.pwr_constraints = 300000, [100000, 300000] + d.modified_clocks = d.modified_pwr_limit = False + d._persistence_mode = 1 + d.gr_clock_default = d.applications_gr_clock = 1800 + d.mem_clock_default = d.applications_mem_clock = 8001 + d.supported_mem_clocks = [8001, 405] + d.supported_gr_clocks = {8001: [2100, 1800, 1200], 405: [405, 210]} + d.use_locked_clocks = use_locked_clocks + d.locked_gr_clock = d.locked_mem_clock = 0 + return d + + +@pytest.fixture +def sudo_calls(monkeypatch): + """Record the commands that would be run with sudo, and do not register exit handlers.""" + calls = [] + monkeypatch.setattr("kernel_tuner.observers.julia_nvml.subprocess.run", lambda args, check: calls.append(args)) + monkeypatch.setattr("kernel_tuner.observers.julia_nvml.atexit.register", lambda f: None) + return calls + + +def test_is_an_nvml_observer(): + # core and the checks on tunable parameters starting with nvml_ look for an NVMLObserver + assert issubclass(JuliaNVMLObserver, NVMLObserver) + + +def test_power_is_read_at_most_once_per_interval(monkeypatch): + observer = JuliaNVMLObserver(["nvml_power"]) + readings = iter([100_000, 200_000]) + observer.nvml = type("FakeNVML", (), {"pwr_usage": lambda self: next(readings)})() + clock = [10.0] + monkeypatch.setattr("kernel_tuner.observers.julia_nvml.time.perf_counter", lambda: clock[0]) + assert observer.read_power() == 100 + clock[0] += observer.power_read_interval / 2 + assert observer.read_power() == 100 + clock[0] += observer.power_read_interval + assert observer.read_power() == 200 + + +def test_power_limit_falls_back_to_nvidia_smi(sudo_calls): + d = fake_device() + d.pwr_limit = 150000 + assert sudo_calls == [["sudo", "nvidia-smi", "-i", "2", "--power-limit=150"]] + assert d.modified_pwr_limit + with pytest.raises(ValueError): + d.pwr_limit = 50000 + + +def test_application_clocks_fall_back_to_nvidia_smi(sudo_calls): + d = fake_device() + d.gr_clock = 1200 + assert sudo_calls == [["sudo", "nvidia-smi", "-i", "2", "--applications-clocks=8001,1200"]] + assert (d.applications_mem_clock, d.applications_gr_clock) == (8001, 1200) + with pytest.raises(ValueError): + d.gr_clock = 210 # only supported with the lower memory clock + + +def test_locked_clocks_fall_back_to_nvidia_smi(sudo_calls): + d = fake_device(use_locked_clocks=True) + d.set_clocks(8001, 1200) + d.reset_clocks() + assert [args[-1] for args in sudo_calls] == [ + "--lock-gpu-clocks=1200,1200", + "--lock-memory-clocks=8001,8001", + "--reset-gpu-clocks", + "--reset-memory-clocks", + ] + + +def test_restore_defaults_only_restores_what_was_changed(sudo_calls): + d = fake_device() + d.restore_defaults() + assert sudo_calls == [] + d.lib.app_clocks["graphics"] = 1200 # as if the clocks were set by an earlier call + d.modified_clocks = True + d.restore_defaults() + assert sudo_calls == [["sudo", "nvidia-smi", "-i", "2", "--applications-clocks=8001,1800"]] + + +@skip_if_no_julia +@skip_if_no_julia_cuda +def test_device_lookups_agree(): + d = JuliaNVML(device_id=0) + assert d.uuid.startswith("GPU-") + for other in [ + JuliaNVML(device_uuid=d.uuid), + JuliaNVML(device_uuid=d.uuid.removeprefix("GPU-")), + JuliaNVML(device_pci_bus=d.pci_bus), + ]: + assert other.id == d.id + + +@skip_if_no_julia +@skip_if_no_julia_cuda +def test_readings(): + d = JuliaNVML(device_id=0) + assert d.pwr_usage() > 0 + assert d.energy_usage() > 0 + assert d.gr_clock > 0 and d.mem_clock > 0 + assert 0 < d.temperature < 120 + assert d.pwr_constraints[0] <= d.pwr_limit <= d.pwr_constraints[1] + assert d.supported_mem_clocks == sorted(d.supported_mem_clocks, reverse=True) + + +@skip_if_no_julia +@skip_if_no_julia_cuda +def test_errors_are_translated(): + d = JuliaNVML(device_id=0) + with pytest.raises((NVMLError_NoPermission, NVMLError_NotSupported)): + # setting the power limit requires root, and is not supported on all GPUs + d.lib.set_power_limit_b(d.dev, d.pwr_limit) + + +@skip_if_no_julia +@skip_if_no_julia_cuda +def test_tune_kernel_with_julia_nvml_observer(): + kernel_string = """ + using KernelAbstractions + + @kernel function vector_add!(c, a, b, n, ::Val{block_size_x} = Val(128)) where {block_size_x} + i = @index(Global) + if i <= n + c[i] = a[i] + b[i] + end + end + """ + size = 50_000_000 + a, b = np.random.randn(size).astype(np.float32), np.random.randn(size).astype(np.float32) + args = [np.zeros_like(a), a, b, np.int32(size)] + tune_params = [("block_size_x", [128, 256])] + observables = ["nvml_power", "nvml_energy", "core_freq", "mem_freq", "temperature"] + observer = JuliaNVMLObserver(observables, continuous_duration=0.5) + + results, _ = tune_kernel("vector_add!", kernel_string, size, args, tune_params, lang="julia", observers=[observer]) + + from juliacall import Main as jl + + # the observer measures the GPU that the Julia backend runs on + assert observer.nvml.uuid == "GPU-" + str(jl.seval("CUDA.uuid(CUDA.device())")) + for r in results: + for observable in observables: + assert r[observable] > 0, f"{observable} not positive in {r}" + # the energy of a single run of the kernel + assert r["nvml_energy"] == pytest.approx(r["nvml_power"] * r["time"] / 1e3) diff --git a/test/test_metal_observer.py b/test/test_metal_observer.py new file mode 100644 index 00000000..65dd5640 --- /dev/null +++ b/test/test_metal_observer.py @@ -0,0 +1,48 @@ +import types + +import pytest + +from kernel_tuner.observers.metal import MetalContinuousObserver, MetalDevice, MetalObserver + + +def sample(energy_j, elapsed_s, idle_ratio=0.1, freq_hz=1e9): + return {"energy_j": energy_j, "elapsed_s": elapsed_s, "idle_ratio": idle_ratio, "freq_hz": freq_hz} + + +def test_summarize_sums_energy_over_the_window(): + # powermetrics reports whole millijoules per sample, so the power must come from the totals + samples = [sample(0.150, 0.050), sample(0.100, 0.050), sample(0.001, 0.0004)] + summary = MetalDevice.summarize(samples) + assert summary["energy"] == pytest.approx(0.251) + assert summary["duration"] == pytest.approx(0.1004) + assert summary["power"] == pytest.approx(0.251 / 0.1004) + + +def test_summarize_weights_by_duration(): + samples = [sample(0.1, 0.075, idle_ratio=0.0, freq_hz=1e9), sample(0.1, 0.025, idle_ratio=1.0, freq_hz=2e9)] + summary = MetalDevice.summarize(samples) + assert summary["load"] == pytest.approx(0.75) + assert summary["freq_hz"] == pytest.approx(1.25e9) + + +def test_summarize_skips_idle_samples_unless_all_are_idle(): + busy, idle = sample(0.3, 0.05, idle_ratio=0.1), sample(0.01, 0.05, idle_ratio=0.99) + assert MetalDevice.summarize([busy, idle], min_load=0.1)["power"] == pytest.approx(6.0) + assert MetalDevice.summarize([idle], min_load=0.1)["power"] == pytest.approx(0.2) + assert MetalDevice.summarize([]) is None + + +def test_continuous_observer_scales_energy_to_one_kernel(): + parent = MetalObserver.__new__(MetalObserver) # skip the powermetrics permission check + parent.prefix = None + parent.observables = {"energy", "power"} + parent.min_load = 0.0 + parent.results_per_iteration = {"energy": [], "power": []} + parent.device = types.SimpleNamespace(stop_sampling=lambda: [sample(1.0, 0.5), sample(1.0, 0.5)]) + parent.after_finish() + + observer = MetalContinuousObserver(parent) + observer.results = {"time": 0.5} # milliseconds per kernel execution + results = observer.get_results() + assert results["power"] == pytest.approx(2.0) + assert results["energy"] == pytest.approx(2.0 * 0.5e-3) diff --git a/test/test_nvml_continuous_observer.py b/test/test_nvml_continuous_observer.py new file mode 100644 index 00000000..5cb25a7f --- /dev/null +++ b/test/test_nvml_continuous_observer.py @@ -0,0 +1,94 @@ +import types + +import pytest + +from kernel_tuner.observers.nvml import NVMLContinuousObserver, NVMLObserver +from kernel_tuner.observers.observer import ContinuousObserver + + +class FakeParent: + """An NVMLObserver whose device reports a fixed power, and an energy counter that is updated every 100 ms.""" + + def __init__(self, clock, power_w=200.0, counter_power_w=250.0, has_energy_counter=True): + self.clock = clock + self.power_w = power_w + self.counter_power_w = counter_power_w + self.has_energy_counter = has_energy_counter + self.nvml = types.SimpleNamespace(energy_usage=self.energy_usage) + + def energy_usage(self): + if not self.has_energy_counter: + raise RuntimeError("not supported") + last_update = int(self.clock[0] * 10) / 10 # the counter is updated every 100 ms + return int(last_update * self.counter_power_w * 1e3) # millijoule + + def read_power(self): + return self.power_w + + def before_start(self): + pass + + def after_start(self): + pass + + def during(self): + pass + + def after_finish(self): + pass + + def get_results(self): + return {} + + +def measure(parent, clock, duration, kernel_time_ms=0.5): + """Run the hooks of the observer as core does, polling every millisecond for `duration` seconds.""" + observer = NVMLContinuousObserver("nvml", ["nvml_power", "nvml_energy"], parent) + observer.results = {"time": kernel_time_ms} + observer.before_start() + observer.after_start() + end = clock[0] + duration + while clock[0] < end: + observer.during() + clock[0] += 1e-3 + observer.after_finish() + return observer.get_results() + + +@pytest.fixture +def clock(monkeypatch): + clock = [10.05] + monkeypatch.setattr("kernel_tuner.observers.nvml.time.perf_counter", lambda: clock[0]) + monkeypatch.setattr("kernel_tuner.observers.observer.time.perf_counter", lambda: clock[0]) + return clock + + +def test_power_from_energy_counter(clock): + results = measure(FakeParent(clock), clock, duration=1.0) + # the power readings (200 W) are only a fallback, the energy counter (250 W) is used + assert results["nvml_power"] == pytest.approx(250.0, rel=0.02) + assert results["nvml_energy"] == pytest.approx(250.0 * 0.5e-3, rel=0.02) + + +def test_falls_back_to_power_readings_without_energy_counter(clock): + results = measure(FakeParent(clock, has_energy_counter=False), clock, duration=1.0) + assert results["nvml_power"] == pytest.approx(200.0) + assert results["nvml_energy"] == pytest.approx(200.0 * 0.5e-3) + + +def test_falls_back_to_power_readings_when_the_counter_is_wrong(clock): + # the counter increases about 15 times too slowly, as on an RTX A6000 with driver 595.91.07 + with pytest.warns(UserWarning, match="energy counter"): + results = measure(FakeParent(clock, counter_power_w=200.0 / 15), clock, duration=1.0) + assert results["nvml_power"] == pytest.approx(200.0) + + +def test_energy_counter_is_opt_in(): + assert type(NVMLObserver(["nvml_power"]).continuous_observer) is ContinuousObserver + assert type(NVMLObserver(["nvml_power"], use_energy_counter=True).continuous_observer) is NVMLContinuousObserver + + +def test_falls_back_to_power_readings_when_the_window_is_too_short(clock): + # one change of the counter is not enough to know how long it took to use that energy + results = measure(FakeParent(clock), clock, duration=0.15) + assert results["nvml_power"] == pytest.approx(200.0) diff --git a/test/test_runners.py b/test/test_runners.py index daadc743..e512379d 100644 --- a/test/test_runners.py +++ b/test/test_runners.py @@ -16,7 +16,7 @@ cache_filename_zipped = cache_filename + ".gz" @pytest.fixture -def env(): +def env(request): kernel_string = """ extern "C" __global__ void vector_add(float *c, float *a, float *b, int n) { int i = blockIdx.x * block_size_x + threadIdx.x; @@ -26,7 +26,7 @@ def env(): } """ - size = 100 + size = request.param if hasattr(request, 'param') else 100 a = np.random.randn(size).astype(np.float32) b = np.random.randn(size).astype(np.float32) c = np.zeros_like(b) diff --git a/test/test_time_budgets.py b/test/test_time_budgets.py index 16b10fd0..22e27194 100644 --- a/test/test_time_budgets.py +++ b/test/test_time_budgets.py @@ -1,6 +1,7 @@ # ruff: noqa from itertools import product from time import perf_counter +from warnings import warn import numpy as np import pytest @@ -66,7 +67,11 @@ def test_some_time_budget(env): assert 0 < len(res) < size_all # Ensure that the time limit was respected by some margin. - assert perf_counter() - start_time < (time_limit * 2) * (1.5 if running_on_ci else 1.0) # allow for more overhead on CI + if len(res) > 1: + print(f"Evaluated {len(res)} configurations in {perf_counter() - start_time:.2f} seconds, with a time limit of {time_limit:.2f} seconds.") + assert perf_counter() - start_time < (time_limit * 2) * (1.5 if running_on_ci else 1.0) # allow for more overhead on CI + else: + warn(f"Test skipped: Only one result was returned in {perf_counter() - start_time:.2f} seconds, taking longer than the time limit of {time_limit:.2f} seconds.") @skip_if_no_gcc diff --git a/test/test_toml_file.py b/test/test_toml_file.py index 29b353ae..f8c03503 100644 --- a/test/test_toml_file.py +++ b/test/test_toml_file.py @@ -2,13 +2,13 @@ from pathlib import Path -import tomli +import tomllib package_root = Path(".").parent.parent pyproject_toml_path = package_root / "pyproject.toml" assert pyproject_toml_path.exists() with pyproject_toml_path.open(mode="rb") as fp: - pyproject = tomli.load(fp) + pyproject = tomllib.load(fp) project = pyproject["project"] if "project" in pyproject else pyproject["tool"]["poetry"] diff --git a/test/test_util_functions.py b/test/test_util_functions.py index 2ab80253..28c14f3e 100644 --- a/test/test_util_functions.py +++ b/test/test_util_functions.py @@ -884,3 +884,34 @@ def test_convert_constraint_lambdas_illformatted(): print(result) except ValueError: pass + + +class JuliaError(Exception): + """Stand-in for JuliaCall's JuliaError, which holds the Julia exception in `exception`.""" + + def __init__(self, exception): + super().__init__(exception) + self.exception = exception + + +def test_is_interrupt(): + from kernel_tuner.util import is_interrupt, raise_if_interrupt + + assert is_interrupt(KeyboardInterrupt()) + assert is_interrupt(JuliaError("InterruptException()")) + assert is_interrupt(JuliaError("Python: KeyboardInterrupt")) # from Python code called from Julia + assert not is_interrupt(JuliaError("BoundsError()")) + assert not is_interrupt(RuntimeError("Kernel launch failed")) + + # an interrupt re-raised as another exception by a backend + try: + try: + raise JuliaError("InterruptException()") + except JuliaError as e: + raise RuntimeError("Julia synchronize failed") from e + except RuntimeError as e: + assert is_interrupt(e) + with pytest.raises(KeyboardInterrupt): + raise_if_interrupt(e) + + raise_if_interrupt(RuntimeError("not an interrupt")) # does nothing