Intergrate roofline benchmark into rocprof-compute (#2015)

---------

Co-authored-by: Fei Zheng <44449748+feizheng10@users.noreply.github.com>
이 커밋은 다음에 포함됨:
Ben Richard
2025-12-03 10:51:46 -05:00
커밋한 사람 GitHub
부모 43f0a53fb0
커밋 2bfa9a4d4c
13개의 변경된 파일1810개의 추가작업 그리고 989개의 파일을 삭제
+453
파일 보기
@@ -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
+205
파일 보기
@@ -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)
+4 -2
파일 보기
@@ -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
파일 크기가 너무 크기때문에 변경 상태를 표시하지 않습니다. Diff 로드
바이너리 파일은 표시되지 않음.
바이너리 파일은 표시되지 않음.
바이너리 파일은 표시되지 않음.
바이너리 파일은 표시되지 않음.
바이너리 파일은 표시되지 않음.
바이너리 파일은 표시되지 않음.
바이너리 파일은 표시되지 않음.
-114
파일 보기
@@ -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
-873
파일 보기
@@ -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: