hwbench: Adding fio engine

This commit is adding a first cmdline engine_module to execute a
single fio command line.

This code has been tested on fio-3.19, defining the minimal release
version.

To enable this mode, engine_module must be set to "cmdline".
The expected command line to forward to fio must be provided in the engine_module_parameter_base.

The command line will be tweaked by hwbench to ensure:
- runtime consistency with other engines : --time_based and --runtime are added
- output consistency: --output-format=json+ is added
- job naming: --name is adjusted to match hwbench's job name

Please note that :
- Fio's runtime will inherit automatically from hwbench's runtime value.
- --numjobs value will be fed with 'stressor_range' making possible to
  study the scalability of a device with a minimal code.

If one of these values were already present in the
engine_module_parameter_base, hwbench will replace them by the values
that were computed based on the benchmark descrption.

A sample configuration file (configs/fio.conf) is provided as an example, it will:
- test /dev/sdp in a randread 4k profile
- two benchmarks are automatically created as per the stressor_range
  value ("4,6") :
-- one with numjobs=4
-- one with numjobs=6

The testing suite is added to ensure a proper parsing and benchmarking job creation.

Signed-off-by: Erwan Velu <[email protected]>

configs/fio.conf

@@ -0,0 +1,14 @@
+# This configuration will :
+# - load all cores with a matrixprod test during 15 sec.
+[global]
+runtime=15
+monitor=all
+
+[randread_cmdline]
+engine=fio
+engine_module=cmdline
+engine_module_parameter_base=--filename=/dev/sdp --direct=1 --rw=randread --bs=4k --ioengine=libaio --iodepth=256 --group_reporting --readonly
+hosting_cpu_cores=all
+hosting_cpu_cores_scaling=none
+stressor_range=4,6
+

hwbench/bench/benchmark.py

@@ -123,9 +123,9 @@ def pre_run(self):
         cpu_location = ""
         if p.get_pinned_cpu():
             if isinstance(p.get_pinned_cpu(), (int, str)):
-                cpu_location = " on CPU {:3d}".format(p.get_pinned_cpu())
+                cpu_location = " pinned on CPU {:3d}".format(p.get_pinned_cpu())
             elif isinstance(p.get_pinned_cpu(), list):
-                cpu_location = " on CPU {}".format(str(p.get_pinned_cpu()))
+                cpu_location = " pinned on CPU {}".format(str(p.get_pinned_cpu()))
             else:
                 h.fatal(
                     "Unsupported get_pinned_cpu() format :{}".format(

hwbench/bench/test_fio.py

@@ -0,0 +1,38 @@
+from . import test_benchmarks_common as tbc
+
+
+class TestFio(tbc.TestCommon):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.load_mocked_hardware(
+            cpucores="./hwbench/tests/parsing/cpu_cores/v2321",
+            cpuinfo="./hwbench/tests/parsing/cpu_info/v2321",
+            numa="./hwbench/tests/parsing/numa/8domainsllc",
+        )
+        self.load_benches("./hwbench/config/fio.conf")
+        self.parse_jobs_config()
+        self.QUADRANT0 = list(range(0, 16)) + list(range(64, 80))
+        self.QUADRANT1 = list(range(16, 32)) + list(range(80, 96))
+        self.ALL = list(range(0, 128))
+
+    def test_fio(self):
+        """Check fio syntax."""
+        assert self.benches.count_benchmarks() == 2
+        assert self.benches.count_jobs() == 1
+        assert self.benches.runtime() == 30
+
+        for bench in self.benches.benchs:
+            self.assertIsNone(bench.validate_parameters())
+            bench.get_parameters().get_name() == "randread_cmdline"
+
+        bench_0 = self.get_bench_parameters(0)
+        assert (
+            bench_0.get_engine_module_parameter_base()
+            == "--runtime=15 --time_based --output-format=json+ --numjobs=4 --name=randread_cmdline_0 --time_based --numjobs=4 --filename=/dev/sdp --direct=1 --rw=randread --bs=4k --ioengine=libaio --iodepth=256 --group_reporting --readonly"
+        )
+
+        bench_1 = self.get_bench_parameters(1)
+        assert (
+            bench_1.get_engine_module_parameter_base()
+            == "--runtime=15 --time_based --output-format=json+ --numjobs=6 --name=randread_cmdline_1 --time_based --numjobs=6 --filename=/dev/sdp --direct=1 --rw=randread --bs=4k --ioengine=libaio --iodepth=256 --group_reporting --readonly"
+        )

hwbench/config/fio.conf

@@ -0,0 +1,16 @@
+# This configuration will :
+# - test /dev/sdp in 4k randread for 15 seconds
+# -- first with 4 stressors
+# -- then with 6 stressors
+[global]
+runtime=15
+monitor=all
+
+[randread_cmdline]
+engine=fio
+engine_module=cmdline
+engine_module_parameter_base=--filename=/dev/sdp --direct=1 --rw=randread --bs=4k --ioengine=libaio --iodepth=256 --group_reporting --readonly
+hosting_cpu_cores=all
+hosting_cpu_cores_scaling=none
+stressor_range=4,6
+

hwbench/config/test_parse_fio.py

@@ -0,0 +1,26 @@
+from unittest.mock import patch
+from ..environment.mock import MockHardware
+from ..bench import test_benchmarks_common as tbc
+
+
+class TestParseConfig(tbc.TestCommon):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.hw = MockHardware()
+        self.load_benches("./hwbench/config/fio.conf")
+
+    def test_sections_name(self):
+        """Check if sections names are properly detected."""
+        sections = self.get_jobs_config().get_sections()
+        assert sections == [
+            "randread_cmdline",
+        ]
+
+    def test_keywords(self):
+        """Check if all keywords are valid."""
+        try:
+            with patch("hwbench.utils.helpers.is_binary_available") as iba:
+                iba.return_value = True
+                self.get_jobs_config().validate_sections()
+        except Exception as exc:
+            assert False, f"'validate_sections' detected a syntax error {exc}"

hwbench/engines/fio.py

@@ -0,0 +1,184 @@
+import json
+from typing import Any
+
+from ..bench.parameters import BenchmarkParameters
+from ..bench.engine import EngineBase, EngineModuleBase
+from ..bench.benchmark import ExternalBench
+
+
+class EngineModuleCmdline(EngineModuleBase):
+    """This class implements the EngineModuleBase for fio"""
+
+    def __init__(self, engine: EngineBase, engine_module_name: str, fake_stdout=None):
+        super().__init__(engine, engine_module_name)
+        self.engine_module_name = engine_module_name
+        self.load_module_parameter(fake_stdout)
+
+    def load_module_parameter(self, fake_stdout=None):
+        # if needed add module parameters to your module
+        self.add_module_parameter("cmdline")
+
+    def validate_module_parameters(self, p: BenchmarkParameters):
+        msg = super().validate_module_parameters(p)
+        FioCmdLine(self, p).parse_parameters(True)
+        return msg
+
+    def run_cmd(self, p: BenchmarkParameters):
+        return FioCmdLine(self, p).run_cmd()
+
+    def run(self, p: BenchmarkParameters):
+        return FioCmdLine(self, p).run()
+
+    def fully_skipped_job(self, p) -> bool:
+        return FioCmdLine(self, p).fully_skipped_job()
+
+
+class Engine(EngineBase):
+    """The main fio class."""
+
+    def __init__(self, fake_stdout=None):
+        super().__init__("fio", "fio")
+        self.add_module(EngineModuleCmdline(self, "cmdline", fake_stdout))
+
+    def run_cmd_version(self) -> list[str]:
+        return [
+            self.get_binary(),
+            "--version",
+        ]
+
+    def run_cmd(self) -> list[str]:
+        return []
+
+    def parse_version(self, stdout: bytes, _stderr: bytes) -> bytes:
+        self.version = stdout.split(b"-")[1].strip()
+        return self.version
+
+    def version_major(self) -> int:
+        if self.version:
+            return int(self.version.split(b".")[0])
+        return 0
+
+    def version_minor(self) -> int:
+        if self.version:
+            return int(self.version.split(b".")[1])
+        return 0
+
+    def parse_cmd(self, stdout: bytes, stderr: bytes):
+        return {}
+
+
+class Fio(ExternalBench):
+    """The Fio stressor."""
+
+    def __init__(
+        self, engine_module: EngineModuleBase, parameters: BenchmarkParameters
+    ):
+        ExternalBench.__init__(self, engine_module, parameters)
+        self.parameters = parameters
+        self.engine_module = engine_module
+        self.parse_parameters()
+
+    def version_compatible(self) -> bool:
+        engine = self.engine_module.get_engine()
+        return engine.version_major() >= 3 and engine.version_minor() >= 19
+
+    def parse_parameters(self):
+        self.runtime = self.parameters.runtime
+
+    def need_skip_because_version(self):
+        if self.skip:
+            # we already skipped this benchmark, we can't know the reason anymore
+            # because we might not have run the version command.
+            return ["echo", "skipped benchmark"]
+        if not self.version_compatible():
+            print(f"WARNING: skipping benchmark {self.name}, needs fio >= 3.19")
+            self.skip = True
+            return ["echo", "skipped benchmark"]
+        return None
+
+    def run_cmd(self) -> list[str]:
+        skip = self.need_skip_because_version()
+        if skip:
+            return skip
+
+        # Let's build the command line to run the tool
+        args = [
+            self.engine_module.get_engine().get_binary(),
+        ]
+
+        return self.get_taskset(args)
+
+    def get_default_fio_command_line(self) -> str:
+        """Return the default fio arguments"""
+        cmdline = f"--runtime={self.parameters.get_runtime()}"
+        cmdline += " --time_based"
+        cmdline += " --output-format=json+"
+        cmdline += f" --numjobs={self.parameters.get_engine_instances_count()}"
+        cmdline += f" --name={self.parameters.get_name_with_position()}"
+        return f"{cmdline} "
+
+    def parse_cmd(self, stdout: bytes, stderr: bytes) -> dict[str, Any]:
+        if self.skip:
+            return self.parameters.get_result_format() | self.empty_result()
+        try:
+            ret = json.loads(stdout)
+        except json.decoder.JSONDecodeError:
+            print(
+                f"{self.parameters.get_name_with_position()}: Cannot load fio's JSON output"
+            )
+            return self.parameters.get_result_format() | self.empty_result()
+
+        return {"fio_results": ret} | self.parameters.get_result_format()
+
+    @property
+    def name(self) -> str:
+        return self.engine_module.get_engine().get_name()
+
+    def run_cmd_version(self) -> list[str]:
+        return self.engine_module.get_engine().run_cmd_version()
+
+    def parse_version(self, stdout: bytes, _stderr: bytes) -> bytes:
+        return self.engine_module.get_engine().parse_version(stdout, _stderr)
+
+    def empy_result(self):
+        """Default empty results for fio"""
+        return {
+            "effective_runtime": 0,
+            "skipped": self.skip,
+            "fio_results": {"jobs": []},
+        }
+
+
+class FioCmdLine(Fio):
+    def parse_parameters(self, fix_epmb=False):
+        """Removing fio arguments set by the engine"""
+        # If we only mean to check parameters, let's return
+        if not fix_epmb:
+            return
+
+        # We need to ensure we have a proper fio command line
+        # Let's remove duplicated and enforce some
+        args = self.parameters.get_engine_module_parameter_base().split()
+        for argument in args:
+            # These overrided arguments has a parameter, let's inform the user its dropped
+            for keyword in ["--runtime", "--name", "--numjobs", "--output-format"]:
+                if keyword in argument:
+                    args.remove(argument)
+                    print(
+                        f"{self.parameters.get_name_with_position()}: Overriding '{argument}' from engine module parameter base"
+                    )
+            # These overrided arguments has no parameter, just prevent duplicates
+            if argument.startswith("--time_based"):
+                args.remove(argument)
+
+        # Overriding empb to represent the real executed command
+        self.parameters.engine_module_parameter_base = (
+            self.get_default_fio_command_line() + " ".join(args)
+        )
+
+    def run_cmd(self) -> list[str]:
+        # Let's build the command line to run the tool
+        return (
+            super().run_cmd()
+            + self.parameters.get_engine_module_parameter_base().split()
+        )

hwbench/engines/test_parse_fio.py

@@ -0,0 +1,26 @@
+import pathlib
+import unittest
+from unittest.mock import patch
+
+from .fio import Engine as Fio
+
+
+def mock_engine() -> Fio:
+    with patch("hwbench.utils.helpers.is_binary_available") as iba:
+        iba.return_value = True
+        return Fio()
+
+
+class TestParse(unittest.TestCase):
+    def test_engine_parsing_version(self):
+        test_dir = pathlib.Path("./hwbench/tests/parsing/fio")
+        for d in test_dir.iterdir():
+            test_target = mock_engine()
+            if not d.is_dir():
+                continue
+            ver_stdout = (d / "version-stdout").read_bytes()
+            ver_stderr = (d / "version-stderr").read_bytes()
+            version = test_target.parse_version(ver_stdout, ver_stderr)
+            assert version == (d / "version").read_bytes().strip()
+            assert test_target.version_major() == 3
+            assert test_target.version_minor() == 19

hwbench/tests/parsing/fio/v319/version

@@ -0,0 +1 @@
+3.19

hwbench/tests/parsing/fio/v319/version-stdout

@@ -0,0 +1 @@
+fio-3.19