refactor(conformance): the Python runner is codeupipe pipelines

conformance/README.md

@@ -7,6 +7,11 @@ One set of language-neutral cases, a runner, and **one judge** (`run.py`).
 The judge is itself a codeupipe pipeline, in [judging/](judging/): one step
 reads the cases, one runs each target, one compares, and one writes the
 matrix. Each step is tested alone in `tests/conformance/test_judge_pipeline.py`.
+
+The Python runner is built the same way, in
+[runners/python/case_runner/](runners/python/case_runner/): one pipeline plays
+a case and one times a scenario. Its steps are tested alone in
+`tests/conformance/test_case_runner.py`.
 The dynamic run and each compiled form are held to exactly the same
 comparison.
 

conformance/runners/python/case_runner/__init__.py

@@ -0,0 +1,6 @@
+"""The Python conformance runner, as codeupipe pipelines.
+
+``run_case_pipeline.py`` plays one case on one target and reports what
+happened. ``bench_scenario_pipeline.py`` times one scenario. A runner never
+judges: ``conformance/judging`` does.
+"""

conformance/runners/python/case_runner/attach_noop_hook_filter.py

@@ -0,0 +1,19 @@
+"""AttachNoopHookFilter: a Hook that does nothing, for a scenario that measures the cost of Hooks."""
+
+import cup_ops as ops
+from codeupipe import Payload
+
+
+class AttachNoopHookFilter:
+    """Filter (sync).
+
+    Input keys: ``case`` (a scenario), ``pipeline``.
+    Output keys: ``hooks``: a list with one NoopHook for a scenario with
+    ``"hooks": "noop"``, or None. The pipeline gets its own NoopHook.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        if payload.get("case").get("hooks") != "noop":
+            return payload.insert("hooks", None)
+        payload.get("pipeline").use_hook(ops.NoopHook())
+        return payload.insert("hooks", [ops.NoopHook()])

conformance/runners/python/case_runner/attach_record_hook_filter.py

@@ -0,0 +1,22 @@
+"""AttachRecordHookFilter: the Hook that records the lifecycle, for a case that asks for it."""
+
+import cup_ops as ops
+from codeupipe import Payload
+
+
+class AttachRecordHookFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``pipeline``, ``compiled``, ``observe``.
+    Output keys: ``hooks``: the Hooks of the run, or None. A case with
+    ``"hooks"`` gets a RecordHook, on the pipeline too. Compiled code calls
+    Hooks only when it was compiled with ``observe``.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        wanted = payload.get("case").get("hooks") and (not payload.get("compiled") or payload.get("observe"))
+        if not wanted:
+            return payload.insert("hooks", None)
+        hook = ops.RecordHook()
+        payload.get("pipeline").use_hook(hook)
+        return payload.insert("hooks", [hook])

conformance/runners/python/case_runner/bench_each_scenario_filter.py

@@ -0,0 +1,38 @@
+"""BenchEachScenarioFilter: time each scenario, one after the other."""
+
+import pathlib
+import tempfile
+
+from codeupipe import Payload
+
+from .unsupported import ExportRefused
+
+
+class BenchEachScenarioFilter:
+    """Filter (sync).
+
+    Input keys: ``cases`` (the scenarios), ``target``.
+    Output keys: ``results``: for each scenario ``id``, ``us_per_run``,
+    ``relative`` and ``output``, or ``{"id", "unsupported": reason}`` when
+    the compiler refuses its pipeline.
+
+    ``each`` is the pipeline that times one scenario.
+    """
+
+    def __init__(self, each):
+        self.each = each
+
+    def call(self, payload: Payload) -> Payload:
+        target = payload.get("target")
+        results = [self.one(case, target, index) for index, case in enumerate(payload.get("cases"))]
+        return payload.insert("results", results)
+
+    def one(self, case, target, index) -> dict:
+        workdir = pathlib.Path(tempfile.mkdtemp(prefix="cup_bench_"))
+        try:
+            out = self.each.run(Payload({"case": case, "target": target, "workdir": workdir,
+                                         "module_name": f"bench_predicates_{index}"}))
+        except ExportRefused as reason:
+            return {"id": case["id"], "unsupported": str(reason)}
+        return {"id": case["id"], "us_per_run": out.get("us_per_run"), "relative": out.get("relative"),
+                "output": out.get("output")}

conformance/runners/python/case_runner/bench_scenario_pipeline.py

@@ -0,0 +1,30 @@
+"""BenchScenarioPipeline: time one scenario on one target."""
+
+from codeupipe import Pipeline
+
+from .attach_noop_hook_filter import AttachNoopHookFilter
+from .build_pipeline_filter import BuildPipelineFilter
+from .export_filter import ExportFilter
+from .is_compiled_predicate import IsCompiledPredicate
+from .make_run_function_filter import MakeRunFunctionFilter
+from .measure_filter import MeasureFilter
+from .parse_target_filter import ParseTargetFilter
+from .write_predicates_filter import WritePredicatesFilter
+
+TARGETS = ("run", "static-steps", "static-inline", "static-observe")
+
+
+class BenchScenarioPipeline(Pipeline):
+    """Input keys: ``case`` (a scenario), ``target``, ``workdir``, ``module_name``.
+    Output keys: ``us_per_run``, ``relative``, ``output``.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.add_filter(ParseTargetFilter(TARGETS, "bench target"))
+        self.add_filter(WritePredicatesFilter())
+        self.add_filter(BuildPipelineFilter())
+        self.add_filter(AttachNoopHookFilter())
+        self.add_valve(IsCompiledPredicate(), ExportFilter())
+        self.add_filter(MakeRunFunctionFilter())
+        self.add_filter(MeasureFilter())

conformance/runners/python/case_runner/build_pipeline_filter.py

@@ -0,0 +1,79 @@
+"""BuildPipelineFilter: the pipeline of a case, made from its steps."""
+
+import cup_ops as ops
+from codeupipe import Payload, Pipeline
+
+from .unsupported import Unsupported
+
+# The reference vocabulary (conformance/README.md): for each op, how its step is made from the case.
+_FILTERS = {
+    "set": lambda spec: ops.SetFilter(spec["key"], spec["value"]),
+    "copy": lambda spec: ops.CopyFilter(spec["from"], spec["to"]),
+    "add": lambda spec: ops.AddFilter(spec["key"], spec["by"]),
+    "append": lambda spec: ops.AppendFilter(spec["key"], spec["value"]),
+    "remove": lambda spec: ops.RemoveFilter(spec["key"]),
+    "merge": lambda spec: ops.MergeFilter(spec["value"]),
+    "has": lambda spec: ops.HasFilter(spec["key"], spec["to"]),
+    "mutate": lambda spec: ops.MutateFilter(spec["key"], spec["value"]),
+    "fail": lambda spec: ops.FailFilter(spec["message"]),
+    "run_again": lambda spec: ops.RunAgainFilter(spec["input"]),
+}
+_ASYNC_FILTERS = {**_FILTERS, "run_again": lambda spec: ops.AsyncRunAgainFilter(spec["input"])}
+_TAPS = {
+    "record": lambda spec: ops.RecordTap(spec["key"]),
+    "fail": lambda spec: ops.FailTap(spec["message"]),
+    "noop": lambda spec: ops.NoopTap(),
+}
+
+
+class BuildPipelineFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``predicates`` (from WritePredicatesFilter),
+    ``run_async``.
+    Output keys: ``pipeline``, and ``label``: for each step label that the
+    pipeline reports, the name that the case gives that step. Steps are
+    reported by label (conformance/API.md) and the cases name them, and this
+    Filter built each step, so it knows which is which. A label that is not
+    in the map stays as it is, and fails the comparison.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        names = {}
+        pipeline = self.build(payload.get("case")["steps"], Pipeline, payload.get("predicates"),
+                              names, payload.get("run_async"))
+        label = {step_label: names[id(step)] for step_label, step, _type in pipeline._steps}
+        return payload.insert("pipeline", pipeline).insert("label", label)
+
+    def build(self, steps, cls, predicates, names, run_async=False):
+        pipeline = cls()
+        for step in steps:
+            kind, name = step["kind"], step["name"]
+            if kind == "filter":
+                made = self.filter(step, run_async)
+                pipeline.add_filter(made)
+            elif kind == "tap":
+                made = self.tap(step)
+                pipeline.add_tap(made)
+            elif kind == "valve":
+                pipeline.add_valve(getattr(predicates, f"when_{name}"), self.filter(step["filter"]))
+                made = pipeline._steps[-1][1]  # the Valve that the pipeline made: the step that the label names
+            elif kind == "pipeline":
+                made = self.build(step["steps"], ops.NestedPipeline, predicates, names)
+                pipeline.add_pipeline(made)
+            else:
+                raise Unsupported(f"unknown step kind {kind!r}")
+            names[id(made)] = name
+        return pipeline
+
+    def filter(self, spec, run_async=False):
+        make = (_ASYNC_FILTERS if run_async else _FILTERS).get(spec["op"])
+        if make is None:
+            raise Unsupported(f"unknown filter op {spec['op']!r}")
+        return make(spec)
+
+    def tap(self, spec):
+        make = _TAPS.get(spec["op"])
+        if make is None:
+            raise Unsupported(f"unknown tap op {spec['op']!r}")
+        return make(spec)

conformance/runners/python/case_runner/export_filter.py

@@ -0,0 +1,48 @@
+"""ExportFilter: compile the pipeline of the case and import the package from disk, as a customer would."""
+
+import importlib
+import sys
+
+from codeupipe import Payload
+
+from .unsupported import ExportRefused
+
+
+class ExportFilter:
+    """Filter (sync).
+
+    Input keys: ``pipeline``, ``lowering``, ``observe``, ``workdir``.
+    Output keys: ``package``: the compiled code, with its ``run`` function.
+
+    The compiler refuses a pipeline that it cannot keep equal. For the fast
+    forms that is an ExportRefused: the target cannot express the case. With
+    ``observe`` the error is given as it is.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        try:
+            package = self.load(payload.get("pipeline"), payload.get("lowering"), payload.get("observe"),
+                                payload.get("workdir"))
+        except Exception as error:  # noqa: BLE001 — the compiler refused the shape (for example Hooks)
+            if payload.get("observe"):
+                raise
+            raise ExportRefused(f"{type(error).__name__}: {error}") from error
+        return payload.insert("package", package)
+
+    def load(self, pipeline, lowering, observe, workdir):
+        from codeupipe.compile import build_export_pipeline_v2
+
+        out = build_export_pipeline_v2(write=False).run(
+            Payload({"pipeline": pipeline, "pattern": "flat", "lowering": lowering, "observe": observe}))
+        for file in out.get("files") or []:
+            target = workdir / file["path"]
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text(file["content"])
+        package_parent = str(workdir / "src")
+        for name in [m for m in sys.modules if m == "steps" or m.startswith("steps.")]:
+            del sys.modules[name]
+        sys.path.insert(0, package_parent)
+        try:
+            return importlib.import_module("steps")
+        finally:
+            sys.path.remove(package_parent)

conformance/runners/python/case_runner/finish_result_filter.py

@@ -0,0 +1,16 @@
+"""FinishResultFilter: the id of the case and what its Taps recorded."""
+
+import conformance_sink as sink
+from codeupipe import Payload
+
+
+class FinishResultFilter:
+    """Filter (sync).
+
+    Input keys: ``case``, ``result``.
+    Output keys: ``result``, with ``id`` and ``taps``.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        result = {**payload.get("result"), "id": payload.get("case")["id"], "taps": list(sink.TAPS)}
+        return payload.insert("result", result)

conformance/runners/python/case_runner/is_compiled_predicate.py

@@ -0,0 +1,13 @@
+"""IsCompiledPredicate: does the case run as compiled code?"""
+
+from codeupipe import Payload
+
+
+class IsCompiledPredicate:
+    """True when ``compiled`` is the value that was given (True unless said)."""
+
+    def __init__(self, compiled=True):
+        self.compiled = compiled
+
+    def test(self, payload: Payload) -> bool:
+        return payload.get("compiled") is self.compiled

conformance/runners/python/case_runner/make_run_function_filter.py

@@ -0,0 +1,33 @@
+"""MakeRunFunctionFilter: the one function that a measurement calls again and again."""
+
+from codeupipe import Payload
+
+
+class MakeRunFunctionFilter:
+    """Filter (sync).
+
+    Input keys: ``case`` (a scenario), ``compiled``, ``observe``, ``hooks``,
+    and ``pipeline`` or ``package``.
+    Output keys: ``fn``: a function with no arguments that does one run of
+    the scenario on a new copy of its input. ``output``: what one run gives.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        fn = self.make(payload)
+        return payload.insert("fn", fn).insert("output", self.output(fn(), payload.get("compiled")))
+
+    def make(self, payload: Payload):
+        data = payload.get("case")["input"]
+        if not payload.get("compiled"):
+            pipeline = payload.get("pipeline")
+            return lambda: pipeline.run(Payload(dict(data)))
+        run = payload.get("package").run
+        if not payload.get("observe"):
+            return lambda: run(dict(data))
+        hooks = payload.get("hooks")
+        state = {}
+        return lambda: run(dict(data), state=state, hooks=hooks)
+
+    def output(self, out, compiled):
+        """The dynamic run gives a Payload, and compiled code gives a dict."""
+        return out if compiled else out.to_dict()

conformance/runners/python/case_runner/measure_filter.py

@@ -0,0 +1,61 @@
+"""MeasureFilter: how long one run takes."""
+
+import statistics
+import time
+
+from codeupipe import Payload
+
+WARMUP = 20000
+RUNS = 20000
+REPEATS = 5
+
+_REFERENCE_KEYS = tuple("k%d" % i for i in range(10))
+
+
+def _reference_write(data, key):
+    data[key] = data.get(key, 0) + 1
+    return data
+
+
+class MeasureFilter:
+    """Filter (sync).
+
+    Input keys: ``fn`` (from MakeRunFunctionFilter).
+    Output keys: ``us_per_run`` (µs for one run: the best of REPEATS x RUNS)
+    and ``relative`` (the time of ``fn`` as a ratio to a reference workload).
+
+    The relative time is for comparing runs on different machines and under
+    different load: each repeat times the reference workload and then
+    ``fn``, back to back, so whatever slows one slows the other, and the
+    median of the ratios of the repeats is reported.
+    """
+
+    def call(self, payload: Payload) -> Payload:
+        us_per_run, relative = self.measure(payload.get("fn"))
+        return payload.insert("us_per_run", us_per_run).insert("relative", relative)
+
+    def measure(self, fn):
+        for _ in range(WARMUP):
+            fn()
+            self.reference()
+        best, ratios = None, []
+        for _ in range(REPEATS):
+            reference = self.timed(self.reference)
+            elapsed = self.timed(fn)
+            ratios.append(elapsed / reference)
+            best = elapsed if best is None else min(best, elapsed)
+        return best / RUNS * 1e6, statistics.median(ratios)
+
+    def timed(self, fn) -> float:
+        started = time.perf_counter()
+        for _ in range(RUNS):
+            fn()
+        return time.perf_counter() - started
+
+    @staticmethod
+    def reference():
+        """The reference workload: ten dict writes behind ten function calls, with no framework."""
+        data = {"a": 1}
+        for key in _REFERENCE_KEYS:
+            data = _reference_write(data, key)
+        return data