cxxrtl: pull in CompilationUnit etc. from Chryse.

niar/build.py

@@ -1,4 +1,3 @@
-import importlib
 import inspect
 import logging
 import re
@@ -7,7 +6,7 @@ from typing import Optional
 
 from amaranth.build import Platform
 
-from .logger import logger, logtime
+from .logging import logger, logtime
 from .project import Project
 
 __all__ = ["add_arguments"]

niar/cmdrunner.py

@@ -0,0 +1,145 @@
+import hashlib
+import inspect
+import shlex
+import subprocess
+
+from .logging import logger
+
+__all__ = ["CompilationUnit", "CommandRunner"]
+
+
+class CompilationUnit:
+    def __init__(self, cmd, *, infs, outf, chdir):
+        if inspect.isfunction(cmd):
+            self.cmd = cmd
+        else:
+            self.cmd = [str(el) for el in cmd]
+        self.infs  = [str(p) for p in infs]
+        self.outf  = str(outf) if outf else None
+        self.chdir = chdir
+
+        self.forced = False
+        self.digest_path = f"{outf}.dig"
+
+    @property
+    def up_to_date(self):
+        if self.outf is None:
+            return False
+        try:
+            with open(self.digest_path, "r") as f:
+                return f.read() == self.digest_ins_with_cmd()
+        except FileNotFoundError:
+            return False
+
+    def mark_up_to_date(self):
+        if self.outf is None:
+            return
+        with open(self.digest_path, "w") as f:
+            f.write(self.digest_ins_with_cmd())
+
+    def digest_ins_with_cmd(self):
+        m = hashlib.sha256()
+
+        def digest_int(i):
+            m.update(f"{i:08x}".encode())
+
+        def digest_bytes(b):
+            digest_int(len(b))
+            m.update(b)
+
+        def digest_str(s):
+            digest_bytes(s.encode())
+
+        sorted_in_paths = list(self.infs)
+        sorted_in_paths.sort()
+
+        digest_int(len(sorted_in_paths))
+        for in_path in sorted_in_paths:
+            digest_str(in_path)
+            with open(in_path, "rb") as f:
+                digest_bytes(f.read())
+
+        if not inspect.isfunction(self.cmd):
+            digest_int(len(self.cmd))
+            for el in self.cmd:
+                digest_str(el)
+
+        return m.hexdigest()
+
+
+class CommandRunner:
+    cus: list[CompilationUnit]
+
+    def __init__(self, *, force=False):
+        self.cus = []
+        self.force = force
+
+    @property
+    def compile_commands(self):
+        return {cu.outf: cu.cmd for cu in self.cus}
+
+    def add_process(self, cmd, *, infs, outf, chdir=None):
+        self.cus.append(
+            CompilationUnit(cmd, infs=infs, outf=outf, chdir=chdir))
+
+    def run(self, step="compile"):
+        self.run_cus(self.cus, step)
+        self.cus = []
+
+    def run_cmd(self, cmd, *, step="compile", chdir=None):
+        cu = CompilationUnit(cmd, infs=[], outf=None, chdir=chdir)
+        self.run_cus([cu], step)
+
+    def run_cus(self, cus, step):
+        runnables = []
+        for cu in cus:
+            if cu.up_to_date:
+                if self.force:
+                    cu.forced = True
+                    runnables.append([cu, None])
+                else:
+                    self.report(cu, skip=True)
+            else:
+                runnables.append([cu, None])
+
+        for cu_proc in runnables:
+            cu = cu_proc[0]
+            self.report(cu)
+            if inspect.isfunction(cu.cmd):
+                cu.cmd()
+            else:
+                cu_proc[1] = subprocess.Popen(cu.cmd, cwd=cu.chdir)
+
+        failed = []
+        for cu, proc in runnables:
+            if proc is not None and proc.wait() != 0:
+                failed.append(cu)
+
+        if failed:
+            logger.error("the following process(es) failed:")
+            for cu in failed:
+                logger.error(f"  {formatted(cu)}")
+            raise RuntimeError(f"failed {step} step")
+
+        for cu, _ in runnables:
+            cu.mark_up_to_date()
+
+    def report(self, cu, *, skip=False):
+        if cu.forced:
+            assert(not skip)
+            action = "[force]"
+        elif skip:
+            action = "[skip] "
+        else:
+            action = "[run]  "
+        logger.info(f"{action} {formatted(cu)}")
+
+
+def formatted(cu):
+    if inspect.isfunction(cu.cmd):
+        return cu.cmd.__name__
+
+    cmd = shlex.join(cu.cmd)
+    if cu.chdir:
+        return f"(in {cu.chdir}/) {cmd}"
+    return cmd

niar/cxxrtl.py

@@ -2,19 +2,17 @@ import json
 import logging
 import os
 import shutil
-import subprocess
 from enum import Enum
 from functools import partial
 from pathlib import Path
 from typing import Any
 
-from amaranth import Elaboratable
 from amaranth._toolchain.yosys import YosysBinary, find_yosys
 from amaranth.back import rtlil
 
 from .build import construct_top
-from .cxxrtl_platform import CxxrtlPlatform
+from .cmdrunner import CommandRunner
-from .logger import logger, logtime
+from .logging import logtime
 from .project import Project
 
 __all__ = ["add_arguments"]
@@ -89,6 +87,12 @@ def add_arguments(np: Project, parser):
         type=str,
         help="output a VCD file",
     )
+    parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        help="don't use cached compilations",
+    )
 
 
 def main(np: Project, args):
@@ -98,138 +102,123 @@ def main(np: Project, args):
 
     platform = np.cxxrtl_target_by_name(args.target)
     design = construct_top(np, platform)
+    cr = CommandRunner(force=args.force)
 
-    cxxrtl_cc_path = np.path.build(f"{np.name}.cc")
     with logtime(logging.DEBUG, "elaboration"):
-        _cxxrtl_convert_with_header(
+        il_path = np.path.build(f"{np.name}.il")
-            yosys,
+        rtlil_text = rtlil.convert(design, name=np.name, platform=platform)
-            cxxrtl_cc_path,
+        with open(il_path, "w") as f:
-            design,
+            f.write(rtlil_text)
-            np.name,
-            platform,
-            black_boxes={},
-        )
 
-    # TODO: bring in Chryse's CompilationUnit stuff to reduce rework.
+        cxxrtl_cc_path = np.path.build(f"{np.name}.cc")
+        yosys_script_path = _make_absolute(np.path.build(f"{np.name}-cxxrtl.ys"))
+        black_boxes = {}
 
-    cc_o_paths = {
+        with open(yosys_script_path, "w") as f:
-        cxxrtl_cc_path: np.path.build(f"{np.name}.o"),
+            for box_source in black_boxes.values():
-    }
+                f.write(f"read_rtlil <<rtlil\n{box_source}\nrtlil\n")
-    for path in np.path("cxxrtl").glob("*.cc"):
+            f.write(f"read_rtlil {_make_absolute(il_path)}\n")
-        cc_o_paths[path] = np.path.build(f"{path.stem}.o")
+            # TODO: do we want to call any opt passes here?
+            f.write(f"write_cxxrtl -header {_make_absolute(cxxrtl_cc_path)}\n")
 
-    cxxflags = CXXFLAGS + [
+        def rtlil_to_cc():
-        f"-DCLOCK_HZ={int(platform.default_clk_frequency)}",
+            yosys.run(["-q", yosys_script_path])
-        *(["-O3"] if args.optimize.opt_rtl else ["-O0"]),
-        *(["-g"] if args.debug else []),
-    ]
-    if platform.uses_zig:
-        cxxflags += [
-            "-DCXXRTL_INCLUDE_CAPI_IMPL",
-            "-DCXXRTL_INCLUDE_VCD_CAPI_IMPL",
-        ]
 
-    procs = []
+        cr.add_process(rtlil_to_cc,
-    compile_commands = {}
+            infs=[il_path, yosys_script_path],
-    for cc_path, o_path in cc_o_paths.items():
+            outf=cxxrtl_cc_path)
-        cmd = [
+        cr.run()
-            "c++",
+
-            *cxxflags,
+    with logtime(logging.DEBUG, "compilation"):
-            "-I" + str(np.path("build")),
+        cc_o_paths = {cxxrtl_cc_path: np.path.build(f"{np.name}.o")}
-            "-I"
+        for path in np.path("cxxrtl").glob("**/*.cc"):
-            + str(yosys.data_dir() / "include" / "backends" / "cxxrtl" / "runtime"),
+            # XXX: we make no effort to distinguish cxxrtl/a.cc and cxxrtl/dir/a.cc.
-            "-c",
+            cc_o_paths[path] = np.path.build(f"{path.stem}.o")
-            str(cc_path),
+
-            "-o",
+        cxxflags = CXXFLAGS + [
-            str(o_path),
+            f"-DCLOCK_HZ={int(platform.default_clk_frequency)}",
+            *(["-O3"] if args.optimize.opt_rtl else ["-O0"]),
+            *(["-g"] if args.debug else []),
         ]
         if platform.uses_zig:
-            cmd = ["zig"] + cmd
+            cxxflags += [
-        compile_commands[o_path] = cmd
+                "-DCXXRTL_INCLUDE_CAPI_IMPL",
-        logger.debug(" ".join(str(e) for e in cmd))
+                "-DCXXRTL_INCLUDE_VCD_CAPI_IMPL",
-        procs.append((cc_path, subprocess.Popen(cmd)))
+            ]
 
-    with open(np.path.build("compile_commands.json"), "w") as f:
+        depfs = list(np.path("cxxrtl").glob("**/*.h"))
-        json.dump(
+
-            [
+        for cc_path, o_path in cc_o_paths.items():
-                {
+            cmd = [
+                "c++",
+                *cxxflags,
+                f"-I{np.path("build")}",
+                f"-I{yosys.data_dir() / "include" / "backends" / "cxxrtl" / "runtime"}",
+                "-c",
+                cc_path,
+                "-o",
+                o_path,
+            ]
+            if platform.uses_zig:
+                cmd = ["zig"] + cmd
+            cr.add_process(cmd, infs=[cc_path] + depfs, outf=o_path)
+
+        with open(np.path.build("compile_commands.json"), "w") as f:
+            json.dump(
+                [{
                     "directory": str(np.path()),
-                    "file": str(file),
+                    "file": file,
                     "arguments": arguments,
-                }
+                } for file, arguments in cr.compile_commands.items()],
-                for file, arguments in compile_commands.items()
+                f,
-            ],
+            )
-            f,
-        )
 
-    failed = []
+        cr.run()
-    for cc_path, p in procs:
-        if p.wait() != 0:
-            failed.append(cc_path)
 
-    if failed:
+        exe_o_path = np.path.build("cxxrtl")
-        logger.error("Failed to build paths:")
+        if platform.uses_zig:
-        for p in failed:
+            cmd = [
-            logger.error(f"- {p}")
+                "zig",
-        raise RuntimeError("failed compile step")
+                "build",
-
+                f"-Dclock_hz={int(platform.default_clk_frequency)}",
-    exe_o_path = np.path.build("cxxrtl")
+                f"-Dyosys_data_dir={yosys.data_dir()}",
-    if platform.uses_zig:
+            ] + [
-        cmd = [
+                # Zig really wants relative paths.
-            "zig",
+                f"-Dcxxrtl_o_path=../{p.relative_to(np.path())}" for p in cc_o_paths.values()
-            "build",
+            ]
-            f"-Dclock_hz={int(platform.default_clk_frequency)}",
+            if args.optimize.opt_app:
-            f"-Dyosys_data_dir={yosys.data_dir()}",
+                cmd += ["-Doptimize=ReleaseFast"]
-        ] + [
+            outf = "cxxrtl/zig-out/bin/cxxrtl"
-            # Zig really wants relative paths.
+            cr.add_process(cmd,
-            f"-Dcxxrtl_o_path=../{p.relative_to(np.path())}" for p in cc_o_paths.values()
+                infs=cc_o_paths.values() + np.path("cxxrtl").glob("**/*.zig"),
-        ]
+                outf=outf,
-        if args.optimize.opt_app:
+                chdir="cxxrtl")
-            cmd += ["-Doptimize=ReleaseFast"]
+            cr.run()
-        logger.debug(" ".join(str(e) for e in cmd))
+            shutil.copy(outf, exe_o_path)
-        subprocess.run(cmd, cwd="cxxrtl", check=True)
+        else:
-        shutil.copy("cxxrtl/zig-out/bin/cxxrtl", exe_o_path)
+            cmd = [
-    else:
+                "c++",
-        cmd = [
+                *cxxflags,
-            "c++",
+                *cc_o_paths.values(),
-            *cxxflags,
+                "-o",
-            *cc_o_paths.values(),
+                exe_o_path,
-            "-o",
+            ]
-            exe_o_path,
+            cr.add_process(cmd,
-        ]
+                infs=cc_o_paths.values(),
-        logger.debug(" ".join(str(e) for e in cmd))
+                outf=exe_o_path)
-        subprocess.run(cmd, check=True)
+            cr.run()
 
     if not args.compile:
         cmd = [exe_o_path]
         if args.vcd:
             cmd += ["--vcd", args.vcd]
-        logger.debug(" ".join(str(e) for e in cmd))
+        cr.run_cmd(cmd, step="run")
-        subprocess.run(cmd, check=True)
 
 
-def _cxxrtl_convert_with_header(
+def _make_absolute(path):
-    yosys: YosysBinary,
+    if path.is_absolute():
-    cc_out: Path,
-    design: Elaboratable,
-    name: str,
-    platform: CxxrtlPlatform,
-    *,
-    black_boxes: dict[Any, str],
-):
-    if cc_out.is_absolute():
         try:
-            cc_out = cc_out.relative_to(Path.cwd())
+            path = path.relative_to(Path.cwd())
         except ValueError:
-            raise AssertionError(
+            raise AssertionError("path must be relative to cwd for builtin-yosys to access it")
-                "cc_out must be relative to cwd for builtin-yosys to write to it"
+    return path
-            )
-    rtlil_out = f"{cc_out}.il"
-    rtlil_text = rtlil.convert(design, name=name, platform=platform)
-    script = []
-    for box_source in black_boxes.values():
-        script.append(f"read_rtlil <<rtlil\n{box_source}\nrtlil")
-    script.append(f"read_rtlil <<rtlil\n{rtlil_text}\nrtlil")
-    script.append(f"write_rtlil {rtlil_out}")
-    script.append(f"write_cxxrtl -header {cc_out}")
-    yosys.run(["-q", "-"], "\n".join(script))

niar/test_cli.py

@@ -4,7 +4,7 @@ from argparse import ArgumentParser
 from amaranth import Elaboratable, Module
 from amaranth_boards.icebreaker import ICEBreakerPlatform
 
-from . import Project, build, logger
+from . import Project, build, logging
 
 calling_test = False
 
@@ -22,8 +22,8 @@ class FixtureProject(Project):
 
 class TestCLI(unittest.TestCase):
     def setUp(self):
-        logger.disable()
+        logging.disable()
-        self.addCleanup(logger.enable)
+        self.addCleanup(logging.enable)
 
     def test_build_works(self):
         parser = ArgumentParser()