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 行删除
-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: