1  #!/usr/bin/env python3
  2  
  3  import unittest
  4  import asyncio
  5  import argparse
  6  import json
  7  import re
  8  import sys
  9  import os
 10  import subprocess
 11  import tabulate
 12  from typing import Literal
 13  from pathlib import Path
 14  
 15  topdir = Path(__file__).parent.parent
 16  sys.path.insert(0, str(topdir))
 17  
 18  import test.regression.benchmark  # noqa: E402
 19  from test.regression.benchmark import BenchmarkResult  # noqa: E402
 20  from test.regression.pysim import PySimulation  # noqa: E402
 21  
 22  
 23  def cd_to_topdir():
 24      os.chdir(str(topdir))
 25  
 26  
 27  def load_benchmarks():
 28      all_tests = test.regression.benchmark.get_all_benchmark_names()
 29      if len(all_tests) == 0:
 30          res = subprocess.run(["make", "-C", "test/external/embench"])
 31          if res.returncode != 0:
 32              print("Couldn't build benchmarks")
 33              sys.exit(1)
 34  
 35          all_tests = test.regression.benchmark.get_all_benchmark_names()
 36  
 37      exclude = {
 38          "cubic",
 39          "huffbench",
 40          "nbody",
 41          "picojpeg",
 42          "primecount",
 43          "qrduino",
 44          "sglib-combined",
 45          "st",
 46          "wikisort",
 47          "matmult-int",
 48          "edn",
 49          "nettle-aes",
 50          "md5sum",
 51          "tarfind",
 52      }
 53  
 54      ret = list(set(all_tests) - exclude)
 55      ret.sort()
 56      return ret
 57  
 58  
 59  def run_benchmarks_with_cocotb(benchmarks: list[str], traces: bool) -> bool:
 60      arglist = ["make", "-C", "test/regression/cocotb", "-f", "benchmark.Makefile", "--no-print-directory"]
 61  
 62      test_cases = ",".join(benchmarks)
 63      arglist += [f"TESTCASE={test_cases}"]
 64  
 65      verilog_code = topdir.joinpath("core.v")
 66      gen_info_path = f"{verilog_code}.json"
 67  
 68      arglist += [f"VERILOG_SOURCES={verilog_code}"]
 69      arglist += [f"_COREBLOCKS_GEN_INFO={gen_info_path}"]
 70  
 71      if traces:
 72          arglist += ["TRACES=1"]
 73  
 74      res = subprocess.run(arglist)
 75  
 76      return res.returncode == 0
 77  
 78  
 79  def run_benchmarks_with_pysim(benchmarks: list[str], traces: bool) -> bool:
 80      suite = unittest.TestSuite()
 81  
 82      def _gen_test(test_name: str):
 83          def test_fn():
 84              traces_file = None
 85              if traces:
 86                  traces_file = "benchmark." + test_name
 87              asyncio.run(test.regression.benchmark.run_benchmark(PySimulation(traces_file=traces_file), test_name))
 88  
 89          test_fn.__name__ = test_name
 90          test_fn.__qualname__ = test_name
 91  
 92          return test_fn
 93  
 94      for test_name in benchmarks:
 95          suite.addTest(unittest.FunctionTestCase(_gen_test(test_name)))
 96  
 97      runner = unittest.TextTestRunner(verbosity=2)
 98      result = runner.run(suite)
 99  
100      return result.wasSuccessful()
101  
102  
103  def run_benchmarks(benchmarks: list[str], backend: Literal["pysim", "cocotb"], traces: bool) -> bool:
104      if backend == "cocotb":
105          return run_benchmarks_with_cocotb(benchmarks, traces)
106      elif backend == "pysim":
107          return run_benchmarks_with_pysim(benchmarks, traces)
108      return False
109  
110  
111  def build_result_table(results: dict[str, BenchmarkResult], tablefmt: str) -> str:
112      if len(results) == 0:
113          return ""
114  
115      header = ["Testbench name", "Cycles", "Instructions", "IPC"]
116  
117      # First fetch all metrics names to build the header
118      result = next(iter(results.values()))
119      for metric_name in sorted(result.metric_values.keys()):
120          regs = result.metric_values[metric_name]
121          for reg_name in regs:
122              header.append(f"{metric_name}/{reg_name}")
123  
124      columns = [header]
125      for benchmark_name, result in results.items():
126          ipc = result.instr / result.cycles
127  
128          column = [benchmark_name, result.cycles, result.instr, ipc]
129  
130          for metric_name in sorted(result.metric_values.keys()):
131              regs = result.metric_values[metric_name]
132              for reg_name in regs:
133                  column.append(regs[reg_name])
134  
135          columns.append(column)
136  
137      # Transpose the table, as the library expects to get a list of rows (and we have a list of columns).
138      rows = [list(i) for i in zip(*columns)]
139  
140      return tabulate.tabulate(rows, headers="firstrow", tablefmt=tablefmt)
141  
142  
143  def main():
144      parser = argparse.ArgumentParser()
145      parser.add_argument("-l", "--list", action="store_true", help="List all benchmarks")
146      parser.add_argument("-t", "--trace", action="store_true", help="Dump waveforms")
147      parser.add_argument("--log-level", default="WARNING", action="store", help="Level of messages to display.")
148      parser.add_argument("--log-filter", default=".*", action="store", help="Regexp used to filter out logs.")
149      parser.add_argument("-p", "--profile", action="store_true", help="Write execution profiles")
150      parser.add_argument("-b", "--backend", default="cocotb", choices=["cocotb", "pysim"], help="Simulation backend")
151      parser.add_argument(
152          "-o",
153          "--output",
154          default="benchmark.json",
155          help="Selects output file to write information to. Default: %(default)s",
156      )
157      parser.add_argument("--summary", default="", action="store", help="Write Markdown summary to this file")
158      parser.add_argument("benchmark_name", nargs="?")
159  
160      args = parser.parse_args()
161  
162      benchmarks = load_benchmarks()
163  
164      if args.list:
165          for name in benchmarks:
166              print(name)
167          return
168  
169      os.environ["__TRANSACTRON_LOG_LEVEL"] = args.log_level
170      os.environ["__TRANSACTRON_LOG_FILTER"] = args.log_filter
171  
172      if args.benchmark_name:
173          pattern = re.compile(args.benchmark_name)
174          benchmarks = [name for name in benchmarks if pattern.search(name)]
175  
176          if not benchmarks:
177              print(f"Could not find benchmark '{args.benchmark_name}'")
178              sys.exit(1)
179  
180      if args.profile:
181          os.environ["__TRANSACTRON_PROFILE"] = "1"
182  
183      success = run_benchmarks(benchmarks, args.backend, args.trace)
184      if not success:
185          print("Benchmark execution failed")
186          sys.exit(1)
187  
188      ipcs = []
189  
190      results: dict[str, BenchmarkResult] = {}
191  
192      for name in benchmarks:
193          with open(f"{str(test.regression.benchmark.results_dir)}/{name}.json", "r") as f:
194              result = BenchmarkResult.from_json(f.read())  # type: ignore
195  
196          results[name] = result
197  
198          ipc = result.instr / result.cycles
199          ipcs.append({"name": name, "unit": "Instructions Per Cycle", "value": ipc})
200  
201      print(build_result_table(results, "simple_outline"))
202  
203      if args.summary != "":
204          with open(args.summary, "w") as summary_file:
205              print(build_result_table(results, "github"), file=summary_file)
206  
207      with open(args.output, "w") as benchmark_file:
208          json.dump(ipcs, benchmark_file, indent=4)
209  
210  
211  if __name__ == "__main__":
212      cd_to_topdir()
213      main()