AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Intergrate roofline benchmark into rocprof-compute (#2015)

Browse Source 
---------

Co-authored-by: Fei Zheng <44449748+feizheng10@users.noreply.github.com>
This commit is contained in:
Ben Richard
2025-12-03 10:51:46 -05:00
committed by GitHub
parent 43f0a53fb0
commit 2bfa9a4d4c
13 changed files with 1810 additions and 989 deletions
Download Patch File Download Diff File Expand all files Collapse all files
projects/rocprofiler-compute/src/hip/hip.py
+453
View File
@@ -0,0 +1,453 @@
##############################################################################
# MIT License
#
# Copyright (c) 2025 Advanced Micro Devices, Inc. All Rights Reserved.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
##############################################################################
import ctypes
from ctypes import (
POINTER,
Structure,
byref,
c_char,
c_char_p,
c_float,
c_int,
c_size_t,
c_uint,
c_uint8,
c_void_p,
)
_lib = ctypes.CDLL("libamdhip64.so")
# Mirrors struct hipUUID_t
class HIPUUID(Structure):
_fields_ = [
("uuid", c_uint8 * 16),
]
# Mirrors hipDeviceArch_t
class HIPDeviceArch(Structure):
_fields_ = [
# 32-bit Atomics
("hasGlobalInt32Atomics", c_uint, 1),
("hasGlobalFloatAtomicExch", c_uint, 1),
("hasSharedInt32Atomics", c_uint, 1),
("hasSharedFloatAtomicExch", c_uint, 1),
("hasFloatAtomicAdd", c_uint, 1),
# 64-bit Atomics
("hasGlobalInt64Atomics", c_uint, 1),
("hasSharedInt64Atomics", c_uint, 1),
# Doubles
("hasDoubles", c_uint, 1),
# Warp cross-lane operations
("hasWarpVote", c_uint, 1),
("hasWarpBallot", c_uint, 1),
("hasWarpShuffle", c_uint, 1),
("hasFunnelShift", c_uint, 1),
# Sync
("hasThreadFenceSystem", c_uint, 1),
("hasSyncThreadsExt", c_uint, 1),
# Misc
("hasSurfaceFuncs", c_uint, 1),
("has3dGrid", c_uint, 1),
("hasDynamicParallelism", c_uint, 1),
]
# Mirrors hipDeviceProp_tR0600
class HIPDeviceProperties(Structure):
_fields_ = [
("name_str", c_char * 256),
("uuid", HIPUUID),
("luid", c_char * 8),
("luidDeviceNodeMask", c_uint),
("totalGlobalMem", c_size_t),
("sharedMemPerBlock", c_size_t),
("regsPerBlock", c_int),
("warpSize", c_int),
("memPitch", c_size_t),
("maxThreadsPerBlock", c_int),
("maxThreadsDim", c_int * 3),
("maxGridSize", c_int * 3),
("clockRate", c_int),
("totalConstMem", c_size_t),
("major", c_int),
("minor", c_int),
("textureAlignment", c_size_t),
("texturePitchAlignment", c_size_t),
("deviceOverlap", c_int),
("multiProcessorCount", c_int),
("kernelExecTimeoutEnabled", c_int),
("integrated", c_int),
("canMapHostMemory", c_int),
("computeMode", c_int),
("maxTexture1D", c_int),
("maxTexture1DMipmap", c_int),
("maxTexture1DLinear", c_int),
("maxTexture2D", c_int * 2),
("maxTexture2DMipmap", c_int * 2),
("maxTexture2DLinear", c_int * 3),
("maxTexture2DGather", c_int * 2),
("maxTexture3D", c_int * 3),
("maxTexture3DAlt", c_int * 3),
("maxTextureCubemap", c_int),
("maxTexture1DLayered", c_int * 2),
("maxTexture2DLayered", c_int * 3),
("maxTextureCubemapLayered", c_int * 2),
("maxSurface1D", c_int),
("maxSurface2D", c_int * 2),
("maxSurface3D", c_int * 3),
("maxSurface1DLayered", c_int * 2),
("maxSurface2DLayered", c_int * 3),
("maxSurfaceCubemap", c_int),
("maxSurfaceCubemapLayered", c_int * 2),
("surfaceAlignment", c_size_t),
("concurrentKernels", c_int),
("ECCEnabled", c_int),
("pciBusID", c_int),
("pciDeviceID", c_int),
("pciDomainID", c_int),
("tccDriver", c_int),
("asyncEngineCount", c_int),
("unifiedAddressing", c_int),
("memoryClockRate", c_int),
("memoryBusWidth", c_int),
("l2CacheSize", c_int),
("persistingL2CacheMaxSize", c_int),
("maxThreadsPerMultiProcessor", c_int),
("streamPrioritiesSupported", c_int),
("globalL1CacheSupported", c_int),
("localL1CacheSupported", c_int),
("sharedMemPerMultiprocessor", c_size_t),
("regsPerMultiprocessor", c_int),
("managedMemory", c_int),
("isMultiGpuBoard", c_int),
("multiGpuBoardGroupID", c_int),
("hostNativeAtomicSupported", c_int),
("singleToDoublePrecisionPerfRatio", c_int),
("pageableMemoryAccess", c_int),
("concurrentManagedAccess", c_int),
("computePreemptionSupported", c_int),
("canUseHostPointerForRegisteredMem", c_int),
("cooperativeLaunch", c_int),
("cooperativeMultiDeviceLaunch", c_int),
("sharedMemPerBlockOptin", c_size_t),
("pageableMemoryAccessUsesHostPageTables", c_int),
("directManagedMemAccessFromHost", c_int),
("maxBlocksPerMultiProcessor", c_int),
("accessPolicyMaxWindowSize", c_int),
("reservedSharedMemPerBlock", c_size_t),
("hostRegisterSupported", c_int),
("sparseHipArraySupported", c_int),
("hostRegisterReadOnlySupported", c_int),
("timelineSemaphoreInteropSupported", c_int),
("memoryPoolsSupported", c_int),
("gpuDirectRDMASupported", c_int),
("gpuDirectRDMAFlushWritesOptions", c_uint),
("gpuDirectRDMAWritesOrdering", c_int),
("memoryPoolSupportedHandleTypes", c_uint),
("deferredMappingHipArraySupported", c_int),
("ipcEventSupported", c_int),
("clusterLaunch", c_int),
("unifiedFunctionPointers", c_int),
("reserved", c_int * 63),
("hipReserved", c_int * 32),
# HIP-only
("gcnArchName_str", c_char * 256),
("maxSharedMemoryPerMultiProcessor", c_size_t),
("clockInstructionRate", c_int),
("arch", HIPDeviceArch),
("hdpMemFlushCntl", POINTER(c_uint)),
("hdpRegFlushCntl", POINTER(c_uint)),
("cooperativeMultiDeviceUnmatchedFunc", c_int),
("cooperativeMultiDeviceUnmatchedGridDim", c_int),
("cooperativeMultiDeviceUnmatchedBlockDim", c_int),
("cooperativeMultiDeviceUnmatchedSharedMem", c_int),
("isLargeBar", c_int),
("asicRevision", c_int),
]
# Add properties as needed
@property
def name(self) -> str:
return self.name_str.decode("utf-8")
@property
def gcnArchName(self) -> str:
return self.gcnArchName_str.decode("utf-8")
# Declare HIP functions here
_lib.hipGetDeviceCount.restype = c_int
_lib.hipGetDeviceCount.argtypes = [POINTER(c_int)]
_lib.hipGetDevicePropertiesR0600.restype = c_int
_lib.hipGetDevicePropertiesR0600.argtypes = [POINTER(HIPDeviceProperties), c_int]
_lib.hipMalloc.restype = c_int
_lib.hipMalloc.argtypes = [POINTER(c_void_p), c_size_t]
_lib.hipFree.restype = c_int
_lib.hipFree.argtypes = [c_void_p]
_lib.hipMemcpyHtoD.restype = c_int
_lib.hipMemcpyHtoD.argtypes = [c_void_p, c_void_p, c_size_t]
_lib.hipMemcpyDtoH.restype = c_int
_lib.hipMemcpyDtoH.argtypes = [c_void_p, c_void_p, c_size_t]
_lib.hipSetDevice.restype = c_int
_lib.hipSetDevice.argtypes = [c_int]
_lib.hipModuleLoadData.restype = c_int
_lib.hipModuleLoadData.argtypes = [POINTER(c_void_p), c_char_p]
_lib.hipModuleUnload.restype = c_int
_lib.hipModuleUnload.argtypes = [c_void_p]
_lib.hipModuleGetFunction.restype = c_int
_lib.hipModuleGetFunction.argtypes = [POINTER(c_void_p), c_void_p, c_char_p]
_lib.hipDeviceSynchronize.restype = c_int
_lib.hipDeviceSynchronize.argtypes = []
_lib.hipModuleLaunchKernel.restype = c_int
_lib.hipModuleLaunchKernel.argtypes = [
c_void_p,
c_uint,
c_uint,
c_uint,
c_uint,
c_uint,
c_uint,
c_uint,
c_void_p,
POINTER(c_void_p),
POINTER(c_void_p),
]
_lib.hipEventCreate.restype = c_int
_lib.hipEventCreate.argtypes = [POINTER(c_void_p)]
_lib.hipEventDestroy.restype = c_int
_lib.hipEventDestroy.argtypes = [c_void_p]
_lib.hipEventRecord.restype = c_int
_lib.hipEventRecord.argtypes = [c_void_p, c_void_p]
_lib.hipEventElapsedTime.restype = c_int
_lib.hipEventElapsedTime.argtypes = [POINTER(c_float), c_void_p, c_void_p]
class HIPError(Exception):
def __init__(self, code: int) -> None:
self.code = code
self.message = f"HIP Error {self.code}"
def __str__(self) -> str:
return self.message
class HIPDeviceMemory:
def __init__(self, ptr: POINTER) -> None:
self.ptr = ptr
def __del__(self) -> None:
_lib.hipFree(self.ptr)
class HIPEvent:
def __init__(self, handle: POINTER) -> None:
self.handle = handle
def __del__(self) -> None:
_lib.hipEventDestroy(self.handle)
class HIPModule:
def __init__(self, handle: POINTER) -> None:
self.handle = handle
def __del__(self) -> None:
_lib.hipModuleUnload(self.handle)
# Implement HIP functions here
def hipGetDeviceCount() -> int:
device_count = c_int()
status = _lib.hipGetDeviceCount(byref(device_count))
if status != 0:
raise HIPError(status)
return device_count.value
def hipGetDeviceProperties(device_id: int) -> HIPDeviceProperties:
props = HIPDeviceProperties()
res = _lib.hipGetDevicePropertiesR0600(byref(props), device_id)
if res != 0:
raise HIPError(res)
return props
def hipMalloc(size: int) -> HIPDeviceMemory:
buf_size = c_size_t(size)
ptr = c_void_p()
status = _lib.hipMalloc(byref(ptr), buf_size)
if status != 0:
raise HIPError(status)
return HIPDeviceMemory(ptr)
def hipMemcpyHtoD(dst: HIPDeviceMemory, src: POINTER, size: int) -> None:
res = _lib.hipMemcpyHtoD(dst.ptr, src, size)
if res != 0:
raise HIPError(res)
def hipMemcpyDtoH(dst: POINTER, src: HIPDeviceMemory, size: int) -> None:
res = _lib.hipMemcpyDtoH(dst, src.ptr, size)
if res != 0:
raise HIPError(res)
def hipSetDevice(id: int) -> None:
status = _lib.hipSetDevice(id)
if status != 0:
raise HIPError(status)
def hipDeviceSynchronize() -> None:
res = _lib.hipDeviceSynchronize()
if res != 0:
raise HIPError(res)
def hipModuleLoadData(code: POINTER) -> HIPModule:
module = c_void_p()
res = _lib.hipModuleLoadData(byref(module), code)
if res != 0:
raise HIPError(res)
return HIPModule(module)
def hipModuleGetFunction(module: POINTER, name: str) -> POINTER:
name_bytes = name.encode("utf-8")
func = c_void_p()
res = _lib.hipModuleGetFunction(byref(func), module.handle, name_bytes)
if res != 0:
raise HIPError(res)
return func
def hipModuleLaunchKernel(
func: POINTER,
grid_dim_x: int,
grid_dim_y: int,
grid_dim_z: int,
block_dim_x: int,
block_dim_y: int,
block_dim_z: int,
shared_mem_size: int,
stream: POINTER,
kernel_params: POINTER,
extra: POINTER = None,
) -> None:
res = _lib.hipModuleLaunchKernel(
func,
grid_dim_x,
grid_dim_y,
grid_dim_z,
block_dim_x,
block_dim_y,
block_dim_z,
shared_mem_size,
stream,
kernel_params,
extra,
)
if res != 0:
raise HIPError(res)
def hipEventCreate() -> HIPEvent:
handle = c_void_p()
res = _lib.hipEventCreate(byref(handle))
if res != 0:
raise HIPError(res)
return HIPEvent(handle)
def hipEventRecord(event: HIPEvent, stream: POINTER = None) -> None:
res = _lib.hipEventRecord(event.handle, stream)
if res != 0:
raise HIPError(res)
def hipEventElapsedTime(start: HIPEvent, stop: HIPEvent) -> float:
ms = c_float()
res = _lib.hipEventElapsedTime(byref(ms), start.handle, stop.handle)
if res != 0:
raise HIPError(res)
return ms.value
projects/rocprofiler-compute/src/hip/hiprtc.py
+205
View File
@@ -0,0 +1,205 @@
##############################################################################
# MIT License
#
# Copyright (c) 2025 Advanced Micro Devices, Inc. All Rights Reserved.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
##############################################################################
import ctypes
from ctypes import (
POINTER,
byref,
c_char,
c_char_p,
c_int,
c_size_t,
c_void_p,
)
_lib = ctypes.CDLL("libhiprtc.so")
_lib.hiprtcCreateProgram.restype = c_int
_lib.hiprtcCreateProgram.argtypes = [
POINTER(c_void_p),
c_char_p,
c_char_p,
c_int,
POINTER(c_char_p),
POINTER(c_char_p),
]
_lib.hiprtcDestroyProgram.restype = c_int
_lib.hiprtcDestroyProgram.argtypes = [
POINTER(c_void_p),
]
_lib.hiprtcCompileProgram.restype = c_int
_lib.hiprtcCompileProgram.argtypes = [
c_void_p,
c_int,
POINTER(c_char_p),
]
_lib.hiprtcGetProgramLogSize.restype = c_int
_lib.hiprtcGetProgramLogSize.argtypes = [
c_void_p,
POINTER(c_size_t),
]
_lib.hiprtcGetProgramLog.restype = c_int
_lib.hiprtcGetProgramLog.argtypes = [
c_void_p,
c_char_p,
]
_lib.hiprtcGetCodeSize.restype = c_int
_lib.hiprtcGetCodeSize.argtypes = [
c_void_p,
POINTER(c_size_t),
]
_lib.hiprtcGetCode.restype = c_int
_lib.hiprtcGetCode.argtypes = [
c_void_p,
c_char_p,
]
_lib.hiprtcAddNameExpression.restype = c_int
_lib.hiprtcAddNameExpression.argtypes = [
c_void_p,
c_char_p,
]
_lib.hiprtcGetLoweredName.restype = c_int
_lib.hiprtcGetLoweredName.argtypes = [c_void_p, c_char_p, POINTER(c_char_p)]
class HIPRTCError(Exception):
def __init__(self, code: int) -> None:
self.code = code
self.message = f"HIP Error {self.code}"
def __str__(self) -> str:
return self.message
class HIPRTCProgram:
def __init__(self, handle: POINTER) -> None:
self.handle = handle
def __del__(self) -> None:
_lib.hiprtcDestroyProgram(self.handle)
# TODO: Handle headers
def hiprtcCreateProgram(src: str, name: str) -> HIPRTCProgram:
src_bytes = src.encode("utf-8")
name_bytes = name.encode("utf-8")
prog = c_void_p()
res = _lib.hiprtcCreateProgram(byref(prog), src_bytes, name_bytes, 0, None, None)
if res != 0:
raise HIPRTCError(res)
return HIPRTCProgram(prog)
# TODO: Handle compile options
def hiprtcCompileProgram(prog: HIPRTCProgram) -> None:
res = _lib.hiprtcCompileProgram(prog.handle, 0, None)
if res != 0:
raise HIPRTCError(res)
def hiprtcGetProgramLogSize(prog: HIPRTCProgram) -> int:
size = c_size_t(0)
res = _lib.hiprtcGetProgramLogSize(prog.handle, byref(size))
if res != 0:
raise HIPRTCError(res)
return size.value
def hiprtcGetProgramLog(prog: HIPRTCProgram) -> str:
size = hiprtcGetProgramLogSize(prog)
buf = (ctypes.c_char * size)()
res = _lib.hiprtcGetProgramLog(prog.handle, buf)
if res != 0:
raise HIPRTCError(res)
return ctypes.string_at(buf, size).decode("utf-8", errors="ignore")
def hiprtcGetCodeSize(prog: HIPRTCProgram) -> int:
size = c_size_t(0)
res = _lib.hiprtcGetCodeSize(prog.handle, byref(size))
if res != 0:
raise HIPRTCError(res)
return size.value
def hiprtcGetCode(prog: HIPRTCProgram) -> POINTER:
size = hiprtcGetCodeSize(prog)
buf = (c_char * size)()
res = _lib.hiprtcGetCode(prog.handle, buf)
if res != 0:
raise HIPRTCError(res)
return buf
def hiprtcGetLoweredName(prog: HIPRTCProgram, name_expression: str) -> str:
expr_bytes = name_expression.encode("utf-8")
name_bytes = c_char_p()
res = _lib.hiprtcGetLoweredName(prog.handle, expr_bytes, name_bytes)
if res != 0:
raise HIPRTCError(res)
return name_bytes.value.decode("utf-8")
def hiprtcAddNameExpression(prog: HIPRTCProgram, name_expression: str) -> None:
expr_bytes = name_expression.encode("utf-8")
res = _lib.hiprtcAddNameExpression(prog.handle, expr_bytes)
if res != 0:
raise HIPRTCError(res)
projects/rocprofiler-compute/src/rocprof_compute_soc/soc_base.py
+4 -2
View File
@@ -36,6 +36,7 @@ import yaml
import config
from roofline import Roofline
from utils import benchmark
from utils.amdsmi_interface import amdsmi_ctx, get_gpu_model, get_mem_max_clock
from utils.logger import (
console_debug,
@@ -53,7 +54,6 @@ from utils.utils import (
convert_metric_id_to_panel_info,
get_panel_alias,
is_tcc_channel_counter,
mibench,
parse_sets_yaml,
)
@@ -680,7 +680,9 @@ class OmniSoC_Base:
"roofline", f"Checking for roofline.csv in {self.get_args().path}"
)
if not (Path(self.get_args().path) / "roofline.csv").is_file():
mibench(self.get_args(), self._mspec)
result = benchmark.run_on_devices([self.get_args().device])
benchmark.dump_csv(result, f"{self.get_args().path}/roofline.csv")
self.roofline_obj.post_processing()
@abstractmethod
projects/rocprofiler-compute/src/utils/benchmark.py
+1148
View File
File diff suppressed because it is too large Load Diff
projects/rocprofiler-compute/src/utils/rooflines/roofline-azurelinux3-rocm7
BIN
View File
Binary file not shown.
projects/rocprofiler-compute/src/utils/rooflines/roofline-rhel8-rocm6
BIN
View File
Binary file not shown.
projects/rocprofiler-compute/src/utils/rooflines/roofline-rhel8-rocm7
BIN
View File
Binary file not shown.
projects/rocprofiler-compute/src/utils/rooflines/roofline-sles15sp6-rocm6
BIN
View File
Binary file not shown.
projects/rocprofiler-compute/src/utils/rooflines/roofline-sles15sp6-rocm7
BIN
View File
Binary file not shown.
projects/rocprofiler-compute/src/utils/rooflines/roofline-ubuntu22_04-rocm6
BIN
View File
Binary file not shown.
projects/rocprofiler-compute/src/utils/rooflines/roofline-ubuntu22_04-rocm7
BIN
View File
Binary file not shown.
projects/rocprofiler-compute/src/utils/utils.py
-114
View File
@@ -1267,120 +1267,6 @@ def gen_sysinfo(
df.to_csv(workload_dir + "/" + "sysinfo.csv", index=False)
def detect_roofline(mspec: Any) -> dict[str, str]: # noqa: ANN401
from utils import specs
rocm_ver = int(mspec.rocm_version[:1])
target_binary: dict[str, Any] = {
"rocm_ver": rocm_ver,
"distro": "override",
"path": None,
}
# Create distro ID list based off of ID (a string, containing a single distro)
# and ID_LIKE (a string, listing at least one distro, separated by a single space)
# from the system /etc/os-release file
os_release = Path("/etc/os-release").read_text()
id_list = specs.search(r'^ID_LIKE="?(.*?)"?$', os_release) or ""
id = specs.search(r'^ID="?(.*?)"?$', os_release) or ""
id_list = id_list.split() + [id]
if "ROOFLINE_BIN" in os.environ.keys():
rooflineBinary = os.environ["ROOFLINE_BIN"]
if Path(rooflineBinary).exists():
console_warning(
"roofline",
f"Detected user-supplied binary --> ROOFLINE_BIN = {rooflineBinary}\n",
)
# distro stays marked as override and path value is substituted in
target_binary["path"] = rooflineBinary
return target_binary
else:
console_error(
"roofline",
"user-supplied path to binary not accessible --> "
f"ROOFLINE_BIN = {rooflineBinary}\n",
)
# check that the system OS is based off of one of the following distributions
elif "azurelinux" in id_list:
distro = "azurelinux"
elif "debian" in id_list:
distro = "22.04"
elif ("fedora" in id_list) or ("tencentos" in id_list):
distro = "platform:el8"
elif "suse" in id_list:
distro = "15.6"
else:
console_error(
"roofline", "Cannot find a valid binary for your operating system"
)
# distro gets assigned, to follow default roofline bin location and nomenclature
target_binary["distro"] = distro
return target_binary
def mibench(args: argparse.Namespace, mspec: Any) -> None: # noqa: ANN401
"""Run roofline microbenchmark to generate peek BW and FLOP measurements."""
console_log("roofline", "No roofline data found. Generating...")
distro_map = {
"platform:el8": "rhel8",
"15.6": "sles15sp6",
"22.04": "ubuntu22_04",
"azurelinux": "azurelinux3",
}
binary_paths: list[str] = []
target_binary = detect_roofline(mspec)
if target_binary["distro"] == "override":
binary_paths.append(target_binary["path"])
else:
# check two potential locations for roofline binaries due to differences in
# development usage vs formal install
potential_paths = [
config.rocprof_compute_home / "utils" / "rooflines" / "roofline",
config.rocprof_compute_home.parent.parent / "bin" / "roofline",
]
for directory in potential_paths:
path_to_binary = (
f"{directory}-{distro_map[target_binary['distro']]}"
f"-rocm{target_binary['rocm_ver']}"
)
binary_paths.append(path_to_binary)
# Distro is valid but cant find rocm ver
found = False
for binary_path in binary_paths:
if Path(binary_path).exists():
found = True
path_to_binary = binary_path
break
if not found:
console_error("roofline", f"Unable to locate expected binary ({binary_paths}).")
my_args = [
path_to_binary,
"-o",
f"{args.path}/roofline.csv",
"-d",
str(args.device),
]
if args.quiet:
my_args += "--quiet"
subprocess.run(my_args, check=True)
def get_submodules(package_name: str) -> list[str]:
"""List all submodules for a target package"""
import importlib
projects/rocprofiler-compute/tests/test_utils.py
-873
View File
@@ -32,10 +32,8 @@ import logging
import os
import re
import shutil
import subprocess
import tempfile
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
import pandas as pd
@@ -4429,877 +4427,6 @@ def test_process_hip_trace_output_invalid_fbase_characters(tmp_path, monkeypatch
utils_mod.process_hip_trace_output(workload_dir, fbase)
# ==============================================================================
# ROOFLINE DETECTION TESTS
# ==============================================================================
def test_ubuntu_detection(monkeypatch):
"""
Test Ubuntu detection.
Args:
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching
Returns:
Verifies that the function correctly identifies Ubuntu and
returns the appropriate distro
"""
mock_os_release = "ID=ubuntu\nID_LIKE=debian"
def mock_path_read_text(self):
return mock_os_release
monkeypatch.setattr("os.environ", {"keys": lambda: []})
monkeypatch.setattr("pathlib.Path.read_text", mock_path_read_text)
def mock_search(pattern, text):
if "ID_LIKE" in pattern:
return "debian"
return None
monkeypatch.setattr("utils.specs.search", mock_search)
import utils.utils as utils_mod
# Create an object with attribute value = 1
result = utils_mod.detect_roofline(SimpleNamespace(rocm_version="0.x.x"))
assert result["rocm_ver"] == 0
def test_debian_detection(monkeypatch):
"""
Test Debian detection.
Args:
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching
Returns:
Verifies that the function correctly identifies Debian
and returns the appropriate distro
"""
mock_os_release = "ID=debian"
def mock_path_read_text(self):
return mock_os_release
monkeypatch.setattr("os.environ", {"keys": lambda: []})
monkeypatch.setattr("pathlib.Path.read_text", mock_path_read_text)
def mock_search(pattern, text):
if "ID" in pattern:
return "debian"
return None
monkeypatch.setattr("utils.specs.search", mock_search)
import utils.utils as utils_mod
# Create an object with attribute value = 1
result = utils_mod.detect_roofline(SimpleNamespace(rocm_version="0.x.x"))
assert result["rocm_ver"] == 0
def test_rhel_detection(monkeypatch):
"""
Test RHEL distro detection.
Args:
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching
Returns:
Verifies that the function correctly identifies RHEL
and returns the appropriate distro
"""
mock_os_release = 'ID_LIKE="rhel fedora"\nID="rhel"'
def mock_path_read_text(self):
return mock_os_release
monkeypatch.setattr("os.environ", {"keys": lambda: []})
monkeypatch.setattr("pathlib.Path.read_text", mock_path_read_text)
monkeypatch.setattr("pathlib.Path.exists", lambda *a, **k: True)
def mock_search(pattern, text):
if "ID_LIKE" in pattern:
return "rhel fedora"
return None
monkeypatch.setattr("utils.specs.search", mock_search)
import utils.utils as utils_mod
result = utils_mod.detect_roofline(SimpleNamespace(rocm_version="7.x.x"))
assert result["rocm_ver"] == 7
def test_azl_detection(monkeypatch):
"""
Test Azure Linux distro detection.
Args:
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching
Returns:
Verifies that the function correctly identifies AZL
and returns the appropriate distro
"""
mock_os_release = "ID=azurelinux"
def mock_path_read_text(self):
return mock_os_release
monkeypatch.setattr("os.environ", {"keys": lambda: []})
monkeypatch.setattr("pathlib.Path.read_text", mock_path_read_text)
monkeypatch.setattr("pathlib.Path.exists", lambda *a, **k: True)
def mock_search(pattern, text):
if "ID" in pattern:
return "azurelinux"
return None
monkeypatch.setattr("utils.specs.search", mock_search)
import utils.utils as utils_mod
result = utils_mod.detect_roofline(SimpleNamespace(rocm_version="7.x.x"))
assert result["rocm_ver"] == 7
def test_sles_detection(monkeypatch):
"""
Test SLES detection.
Args:
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching
Returns:
Verifies that the function correctly identifies SLES
and returns the appropriate distro
"""
mock_os_release = 'ID="opensuse-leap"\nID_LIKE="suse opensuse"'
def mock_path_read_text(self):
return mock_os_release
monkeypatch.setattr("os.environ", {"keys": lambda: []})
monkeypatch.setattr("pathlib.Path.read_text", mock_path_read_text)
def mock_search(pattern, text):
if "ID_LIKE" in pattern:
return "suse openuse"
return None
monkeypatch.setattr("utils.specs.search", mock_search)
import utils.utils as utils_mod
result = utils_mod.detect_roofline(SimpleNamespace(rocm_version="0.x.x"))
assert result["rocm_ver"] == 0
# =============================================================================
# TESTS FOR MIBENCH OUTPUT
# =============================================================================
def test_mibench_override_distro_success(tmp_path, monkeypatch):
"""
Test mibench with override distro that successfully finds and executes binary.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that override path is used and subprocess is called correctly.
"""
class MockArgs:
path = str(tmp_path)
device = 0
quiet = False
class MockMspec:
pass
override_binary_path = tmp_path / "custom_roofline"
override_binary_path.write_text("#!/bin/bash\necho 'roofline executed'")
override_binary_path.chmod(0o755)
def mock_detect_roofline(mspec):
return {
"distro": "override",
"path": str(override_binary_path),
"rocm_ver": "0.x.x",
}
subprocess_calls = []
def mock_subprocess_run(args, check=True):
subprocess_calls.append((args, check))
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("subprocess.run", mock_subprocess_run)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
import utils.utils as utils_mod
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
assert len(subprocess_calls) == 1
expected_args = [ # noqa
str(override_binary_path),
"-o",
str(tmp_path) + "/roofline.csv",
"-d",
"0",
]
assert subprocess_calls[0][1] is True
def test_mibench_standard_distro_first_path_exists(tmp_path, monkeypatch):
"""
Test mibench with standard distro where first potential path exists.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that first path is used when it exists.
"""
class MockArgs:
path = str(tmp_path)
device = 1
quiet = True
class MockMspec:
pass
rocprof_home = tmp_path / "rocprof_home"
install_root = tmp_path / "install_root"
rocprof_home.mkdir(parents=True)
install_root.mkdir(parents=True)
first_path = rocprof_home / "utils" / "rooflines"
first_path.mkdir(parents=True)
binary_path = first_path / "roofline-ubuntu22_04"
binary_path.write_text("#!/bin/bash\necho 'roofline executed'")
binary_path.chmod(0o755)
class MockConfig:
def __init__(self):
self.rocprof_compute_home = self.MockPath(rocprof_home, install_root)
class MockPath:
def __init__(self, home_path, install_path):
self._home_path = home_path
self._install_path = install_path
self.parent = self.MockParent(install_path)
def __str__(self):
return str(self._home_path)
def __truediv__(self, other):
return self._home_path / other
class MockParent:
def __init__(self, install_path):
self.parent = install_path
def __truediv__(self, other):
return self.parent / other
mock_config = MockConfig()
def mock_detect_roofline(mspec):
return {"distro": "22.04", "rocm_ver": "0.x.x"}
subprocess_calls = []
def mock_subprocess_run(args, check=True):
subprocess_calls.append((args, check))
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("utils.utils.config", mock_config)
monkeypatch.setattr("subprocess.run", mock_subprocess_run)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
monkeypatch.setattr("pathlib.Path.exists", lambda *a, **k: True)
import utils.utils as utils_mod
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
assert len(subprocess_calls) == 1
def test_mibench_standard_distro_second_path_exists(tmp_path, monkeypatch):
"""
Test mibench with standard distro where second potential path exists.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that second path is used when first doesn't exist.
"""
class MockArgs:
path = str(tmp_path)
device = 2
quiet = False
class MockMspec:
pass
rocprof_home = tmp_path / "rocprof_home"
install_root = tmp_path / "install_root"
rocprof_home.mkdir(parents=True)
install_root.mkdir(parents=True)
second_path = install_root / "bin"
second_path.mkdir(parents=True)
binary_path = second_path / "roofline-rhel8"
binary_path.write_text("#!/bin/bash\necho 'roofline executed'")
binary_path.chmod(0o755)
class MockConfig:
def __init__(self):
self.rocprof_compute_home = self.MockPath(rocprof_home, install_root)
class MockPath:
def __init__(self, home_path, install_path):
self._home_path = home_path
self._install_path = install_path
self.parent = self.MockParent(install_path)
def __str__(self):
return str(self._home_path)
def __truediv__(self, other):
return self._home_path / other
class MockParent:
def __init__(self, install_path):
self.parent = install_path
def __truediv__(self, other):
return self.parent / other
mock_config = MockConfig()
def mock_detect_roofline(mspec):
return {"distro": "platform:el8", "rocm_ver": "0.x.x"}
subprocess_calls = []
def mock_subprocess_run(args, check=True):
subprocess_calls.append((args, check))
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("utils.utils.config", mock_config)
monkeypatch.setattr("subprocess.run", mock_subprocess_run)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
monkeypatch.setattr("pathlib.Path.exists", lambda *a, **k: True)
import utils.utils as utils_mod
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
assert len(subprocess_calls) == 1
expected_args = [ # noqa: F841
str(binary_path),
"-o",
str(tmp_path) + "/roofline.csv",
"-d",
"2",
]
def test_mibench_no_binary_found_error(tmp_path, monkeypatch):
"""
Test mibench when no binary paths exist, should call console_error.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that console_error is called when no binaries are found.
"""
class MockArgs:
path = str(tmp_path)
device = 0
quiet = False
class MockMspec:
pass
rocprof_home = tmp_path / "rocprof_home"
install_root = tmp_path / "install_root"
rocprof_home.mkdir(parents=True)
install_root.mkdir(parents=True)
class MockConfig:
def __init__(self):
self.rocprof_compute_home = self.MockPath(rocprof_home, install_root)
class MockPath:
def __init__(self, home_path, install_path):
self._home_path = home_path
self._install_path = install_path
self.parent = self.MockParent(install_path)
def __str__(self):
return str(self._home_path)
def __truediv__(self, other):
return self._home_path / other
class MockParent:
def __init__(self, install_path):
self.parent = install_path
def __truediv__(self, other):
return self.parent / other
mock_config = MockConfig()
def mock_detect_roofline(mspec):
return {"distro": "15.6", "rocm_ver": "0.x.x"}
console_error_calls = []
def mock_console_error(category, msg):
console_error_calls.append((category, msg))
raise RuntimeError("console_error called")
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("utils.utils.config", mock_config)
monkeypatch.setattr("utils.utils.console_error", mock_console_error)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
import utils.utils as utils_mod
with pytest.raises(RuntimeError, match="console_error called"):
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
assert len(console_error_calls) == 1
assert console_error_calls[0][0] == "roofline"
assert "Unable to locate expected binary" in console_error_calls[0][1]
def test_mibench_quiet_flag_handling_bug(tmp_path, monkeypatch):
"""
Test mibench quiet flag handling demonstrates the bug where += splits the string.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that the bug exists and characters are split.
"""
rocprof_home = tmp_path / "rocprof_home"
install_root = tmp_path / "install_root"
rocprof_home.mkdir(parents=True)
install_root.mkdir(parents=True)
first_path = rocprof_home / "utils" / "rooflines"
first_path.mkdir(parents=True)
binary_path = first_path / "roofline-ubuntu22_04"
binary_path.write_text("#!/bin/bash\necho 'roofline executed'")
binary_path.chmod(0o755)
class MockConfig:
def __init__(self):
self.rocprof_compute_home = self.MockPath(rocprof_home, install_root)
class MockPath:
def __init__(self, home_path, install_path):
self._home_path = home_path
self._install_path = install_path
self.parent = self.MockParent(install_path)
def __str__(self):
return str(self._home_path)
def __truediv__(self, other):
return self._home_path / other
class MockParent:
def __init__(self, install_path):
self.parent = install_path
def __truediv__(self, other):
return self.parent / other
mock_config = MockConfig()
def mock_detect_roofline(mspec):
return {"distro": "22.04", "rocm_ver": "0.x.x"}
subprocess_calls = []
def mock_subprocess_run(args, check=True):
subprocess_calls.append((args, check))
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("utils.utils.config", mock_config)
monkeypatch.setattr("subprocess.run", mock_subprocess_run)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
monkeypatch.setattr("pathlib.Path.exists", lambda *a, **k: True)
import utils.utils as utils_mod
class MockArgsQuiet:
path = str(tmp_path)
device = 0
quiet = True
class MockMspecQuiet:
pass
utils_mod.mibench(MockArgsQuiet(), SimpleNamespace(rocm_version="0.x.x"))
expected_base_args = [
str(binary_path),
"-o",
str(tmp_path) + "/roofline.csv",
"-d",
"0",
]
expected_full_args = expected_base_args + [ # noqa: F841
"-",
"-",
"q",
"u",
"i",
"e",
"t",
]
subprocess_calls.clear()
class MockArgsNotQuiet:
path = str(tmp_path)
device = 0
quiet = False
class MockMspecNotQuiet:
pass
utils_mod.mibench(MockArgsQuiet(), SimpleNamespace(rocm_version="0.x.x"))
expected_args = [ # noqa: F841
str(binary_path),
"-o",
str(tmp_path) + "/roofline.csv",
"-d",
"0",
]
def test_mibench_sles_distro_mapping(tmp_path, monkeypatch):
"""
Test mibench with SLES distro mapping.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that SLES distro is correctly mapped.
"""
class MockArgs:
path = str(tmp_path)
device = 3
quiet = False
class MockMspec:
pass
rocprof_home = tmp_path / "rocprof_home"
install_root = tmp_path / "install_root"
rocprof_home.mkdir(parents=True)
install_root.mkdir(parents=True)
first_path = rocprof_home / "utils" / "rooflines"
first_path.mkdir(parents=True)
binary_path = first_path / "roofline-sles15sp6"
binary_path.write_text("#!/bin/bash\necho 'roofline executed'")
binary_path.chmod(0o755)
class MockConfig:
def __init__(self):
self.rocprof_compute_home = self.MockPath(rocprof_home, install_root)
class MockPath:
def __init__(self, home_path, install_path):
self._home_path = home_path
self._install_path = install_path
self.parent = self.MockParent(install_path)
def __str__(self):
return str(self._home_path)
def __truediv__(self, other):
return self._home_path / other
class MockParent:
def __init__(self, install_path):
self.parent = install_path
def __truediv__(self, other):
return self.parent / other
mock_config = MockConfig()
def mock_detect_roofline(mspec):
return {"distro": "15.6", "rocm_ver": "0.x.x"}
subprocess_calls = []
def mock_subprocess_run(args, check=True):
subprocess_calls.append((args, check))
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("utils.utils.config", mock_config)
monkeypatch.setattr("subprocess.run", mock_subprocess_run)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
monkeypatch.setattr("pathlib.Path.exists", lambda *a, **k: True)
import utils.utils as utils_mod
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
assert len(subprocess_calls) == 1
def test_mibench_subprocess_run_failure(tmp_path, monkeypatch):
"""
Test mibench when subprocess.run raises an exception.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that subprocess exceptions are properly propagated.
"""
class MockArgs:
path = str(tmp_path)
device = 0
quiet = False
class MockMspec:
pass
override_binary_path = tmp_path / "failing_roofline"
override_binary_path.write_text("#!/bin/bash\nexit 1")
override_binary_path.chmod(0o755)
def mock_detect_roofline(mspec):
return {
"distro": "override",
"path": str(override_binary_path),
"rocm_ver": "0.x.x",
}
def mock_subprocess_run(args, check=True):
raise subprocess.CalledProcessError(1, args)
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("subprocess.run", mock_subprocess_run)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
import utils.utils as utils_mod
with pytest.raises(subprocess.CalledProcessError):
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
def test_mibench_device_string_conversion(tmp_path, monkeypatch):
"""
Test mibench correctly converts device ID to string.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that device ID is converted to string in subprocess args.
"""
class MockArgs:
path = str(tmp_path)
device = 42
quiet = False
class MockMspec:
pass
override_binary_path = tmp_path / "test_roofline"
override_binary_path.write_text("#!/bin/bash\necho 'success'")
override_binary_path.chmod(0o755)
def mock_detect_roofline(mspec):
return {
"distro": "override",
"path": str(override_binary_path),
"rocm_ver": "0.x.x",
}
subprocess_calls = []
def mock_subprocess_run(args, check=True):
subprocess_calls.append(args)
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("subprocess.run", mock_subprocess_run)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
import utils.utils as utils_mod
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
assert len(subprocess_calls) == 1
device_arg_index = subprocess_calls[0].index("-d") + 1
assert subprocess_calls[0][device_arg_index] == "42"
assert isinstance(subprocess_calls[0][device_arg_index], str)
def test_mibench_unknown_distro_mapping(tmp_path, monkeypatch):
"""
Test mibench behavior with unknown distro (should cause KeyError).
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that KeyError is raised for unknown distro.
"""
class MockArgs:
path = str(tmp_path)
device = 0
quiet = False
class MockMspec:
pass
rocprof_home = tmp_path / "rocprof_home"
install_root = tmp_path / "install_root"
rocprof_home.mkdir(parents=True)
install_root.mkdir(parents=True)
class MockConfig:
def __init__(self):
self.rocprof_compute_home = self.MockPath(rocprof_home, install_root)
class MockPath:
def __init__(self, home_path, install_path):
self._home_path = home_path
self._install_path = install_path
self.parent = self.MockParent(install_path)
def __str__(self):
return str(self._home_path)
def __truediv__(self, other):
return self._home_path / other
class MockParent:
def __init__(self, install_path):
self.parent = install_path
def __truediv__(self, other):
return self.parent / other
mock_config = MockConfig()
def mock_detect_roofline(mspec):
return {"distro": "unknown_distro", "rocm_ver": "0.x.x"} # Not in distro_map
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("utils.utils.config", mock_config)
monkeypatch.setattr("utils.utils.console_log", lambda *a, **k: None)
import utils.utils as utils_mod
with pytest.raises(KeyError):
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
def test_mibench_console_log_called(tmp_path, monkeypatch):
"""
Test mibench calls console_log with correct message.
Args:
tmp_path (Path): Temporary directory for test files.
monkeypatch (pytest.MonkeyPatch): Pytest fixture for patching.
Returns:
None: Asserts that console_log is called with expected message.
"""
class MockArgs:
path = str(tmp_path)
device = 0
quiet = False
class MockMspec:
pass
override_binary_path = tmp_path / "test_roofline"
override_binary_path.write_text("#!/bin/bash\necho 'success'")
override_binary_path.chmod(0o755)
def mock_detect_roofline(mspec):
return {
"distro": "override",
"path": str(override_binary_path),
"rocm_ver": "0.x.x",
}
console_log_calls = []
def mock_console_log(category, message):
console_log_calls.append((category, message))
def mock_subprocess_run(args, check=True):
pass
monkeypatch.setattr("utils.utils.detect_roofline", mock_detect_roofline)
monkeypatch.setattr("subprocess.run", mock_subprocess_run)
monkeypatch.setattr("utils.utils.console_log", mock_console_log)
import utils.utils as utils_mod
utils_mod.mibench(MockArgs(), SimpleNamespace(rocm_version="0.x.x"))
assert len(console_log_calls) == 1
assert console_log_calls[0][0] == "roofline"
assert console_log_calls[0][1] == "No roofline data found. Generating..."
"""
Normal Functionality: