fix/jit: Fixing deferred chains and better binary reuse

Makefile

@@ -18,13 +18,15 @@ JIT_BATCH_SIZE ?= 16
 # how many pending queue events to scan ahead when looking for fusion candidates
 FUSION_LOOKAHEAD ?= 32
 # horizontal fusion: max independent parallel chains per kernel pass
-MAX_HFUSE_CHAINS ?= 10
+MAX_HFUSE_CHAINS ?= 9
 # vertical fusion: max RPN opcodes per chain (caps how deep chains can be fused)
 MAX_VFUSE_OPS ?= 128
 # vertical fusion: max distinct input vectors referenced per kernel
 MAX_VFUSE_INPUTS ?= 11
 # vertical fusion: max live WRAM stack vectors used by the generic pipeline interpreter
-MAX_PIPELINE_STACK_DEPTH ?= 2
+# 3 lets the interpreter run depth-3 updates, so the hybrid fallback
+# engages instead of blocking on the compiler; costs one hfuse chain.
+MAX_PIPELINE_STACK_DEPTH ?= 3
 # DPU WRAM tile size is 2^BLOCK_SIZE_LOG2 elements per tasklet block
 BLOCK_SIZE_LOG2 ?= 4
 

benchmarks/_plot_common.py

@@ -34,7 +34,22 @@
 CONFIG = BENCHMARKS / "main-benchmarks" / _config_name
 SUITE_RESULTS = (RESULTS if _config_name == "benchmark.toml"
                  else RESULTS / Path(_config_name).stem)
-RUNS_CSV = SUITE_RESULTS / "runs.csv"
+# PLOT_RUNS_CSV plots another runs CSV; its stem tags the figures so two
+# inputs cannot overwrite each other.
+_runs_override = os.environ.get("PLOT_RUNS_CSV", "").strip()
+RUNS_CSV = Path(_runs_override) if _runs_override else SUITE_RESULTS / "runs.csv"
+if _runs_override and not RUNS_CSV.is_file():
+    raise SystemExit(f"PLOT_RUNS_CSV not found: {RUNS_CSV}")
+RUNS_TAG = "" if RUNS_CSV.stem == "runs" else f"-{RUNS_CSV.stem}"
+
+# Restrict the expected grid, or a partial sweep reads as unfinished and every
+# benchmark is dropped.  PLOT_ONLY_DPUS=256,2048  PLOT_ONLY_VARIANTS=polymerpim
+ONLY_DPUS = frozenset(int(part) for part
+                      in os.environ.get("PLOT_ONLY_DPUS", "").split(",")
+                      if part.strip())
+ONLY_VARIANTS = frozenset(part.strip() for part
+                          in os.environ.get("PLOT_ONLY_VARIANTS", "").split(",")
+                          if part.strip())
 
 BENCHMARK_ORDER = (
     "elementwise",
@@ -81,7 +96,7 @@ def path(self, stem, extension):
         directory.mkdir(parents=True, exist_ok=True)
         dropped = "-no-" + "-".join(sorted(self.without)) if self.without else ""
         scale = "-log" if self.log_y else ""
-        return directory / f"{stem}{dropped}{scale}{extension}"
+        return directory / f"{stem}{RUNS_TAG}{dropped}{scale}{extension}"
 
 
 def _env_names(key):
@@ -150,9 +165,12 @@ def load_selections():
         selections.append(BenchmarkSelection(
             name=name,
             elements_per_dpu=target_size,
-            dpus=tuple(int(value) for value in spec.get("dpus", defaults["dpus"])),
+            dpus=tuple(value for value
+                       in (int(v) for v in spec.get("dpus", defaults["dpus"]))
+                       if not ONLY_DPUS or value in ONLY_DPUS),
             variants=tuple(v for v in spec.get("variants", defaults["variants"])
-                           if VIEW.keeps(v)),
+                           if VIEW.keeps(v)
+                           and (not ONLY_VARIANTS or v in ONLY_VARIANTS)),
             warmup=int(spec.get("warmup", defaults["warmup"])),
             iterations=int(spec.get("iterations", defaults["iterations"])),
             ntrials=int(defaults["ntrials"]),

benchmarks/dynamic-benchmarks/benchmark.toml

@@ -20,7 +20,9 @@ parameters = { channels = 4, check_interval = 2, tolerance = 2 }
 [[dynamic_query]]
 dpus = [64]
 iterations = 10
-elements_per_dpu = [1024, 4096, 16384, 65536, 262144]
+# Below ~1M elements the fixed JIT compile cost dominates and every variant
+# just measures compiler overhead; start where execution dominates.
+elements_per_dpu = [16384, 65536, 262144, 524288]
 variants = [
   "polymerpim-jit",
   "polymerpim-hybrid",

benchmarks/dynamic-benchmarks/plot_adaptive_image.py

@@ -16,7 +16,9 @@
 RUNS_CSV = RESULTS / "adaptive-image.csv"
 SUMMARY_CSV = RESULTS / "adaptive-image-summary.csv"
 FIGURE = RESULTS / "adaptive-image.pdf"
-TRACE_TXT = RESULTS / "adaptive-image-trace.txt"
+# The regime where compile latency dominates the tail; the larger size inverts
+# the ordering because compile cost is fixed while execution scales.
+PLOT_SIZE = 262144
 
 MODEL_ORDER = (
     "polymerpim-jit",
@@ -87,28 +89,6 @@ def load_rows():
     return rows
 
 
-def load_traces():
-    # Per-iteration times from the focused 64-DPU runs; absent is fine.
-    collected = defaultdict(list)
-    if not TRACE_TXT.is_file():
-        return {}
-    for line in TRACE_TXT.read_text().splitlines():
-        fields = line.split()
-        if len(fields) < 4 or not fields[3].startswith("iteration_ms="):
-            continue
-        series = [float(value) for value
-                  in fields[3].split("=", 1)[1].split(";") if value]
-        if series:
-            collected[(f"polymerpim-{fields[0]}", int(fields[1]))].append(series)
-    # Median per iteration index, so one unlucky trial cannot set the shape.
-    traces = {}
-    for key, trials in collected.items():
-        width = min(len(series) for series in trials)
-        traces[key] = [median(series[index] for series in trials)
-                       for index in range(width)]
-    return traces
-
-
 def summarize(rows):
     grouped = defaultdict(list)
     for row in rows:
@@ -161,51 +141,34 @@ def draw(axis, rows, model, value, low=None, high=None):
 def plot(rows):
     plt = load_pyplot(FIGURE)
 
-    sizes = sorted({row.elements_per_dpu for row in rows})
-    if len(sizes) != 2:
-        raise SystemExit("adaptive_image plot expects exactly two problem sizes")
-
-    traces = load_traces()
-    rows_count = 3 if traces else 2
-    figure, axes = plt.subplots(rows_count, len(sizes),
-                                figsize=(10, 3.2 * rows_count))
-    for column, size in enumerate(sizes):
-        selected = [row for row in rows if row.elements_per_dpu == size]
-        column_axes = list(axes[:, column])
-        trace_axis = column_axes.pop(0) if traces else None
-        mean_axis, worst_axis = column_axes
-        for model in MODEL_ORDER:
-            draw(mean_axis, selected, model, "mean_ms")
-            draw(worst_axis, selected, model,
-                 "max_mean_ms", "max_min_ms", "max_max_ms")
-            if trace_axis is not None:
-                draw_series(trace_axis, model,
-                            list(enumerate(traces.get((model, size), []),
-                                           start=1)),
-                            linewidth=1.6)
-
-        top_axis = trace_axis if trace_axis is not None else mean_axis
-        top_axis.set_title(f"{size:,} elements/DPU", fontweight="bold")
-        mean_axis.set_ylabel("Mean iteration (ms)" if column == 0 else "")
-        worst_axis.set_ylabel(
-            "Worst iteration: mean, min-max (ms)" if column == 0 else "")
-        if trace_axis is not None:
-            trace_axis.set_ylabel(
-                "Iteration time, 64 DPUs (ms)" if column == 0 else "")
-            trace_axis.set_xlabel("Iteration")
-            trace_axis.grid(True, which="major", color="#d8d8d8",
-                            linewidth=0.8)
-            trace_axis.set_axisbelow(True)
-        dpus = sorted({row.dpus for row in selected})
-        for axis in (mean_axis, worst_axis):
-            configure_axis(axis, dpus, dpus, "DPUs")
-
-    figure.suptitle("Adaptive image: dynamic execution trade-offs",
-                    fontsize=15, fontweight="bold")
+    selected = [row for row in rows if row.elements_per_dpu == PLOT_SIZE]
+    if not selected:
+        raise SystemExit(f"no adaptive_image rows at {PLOT_SIZE} elements/DPU")
+
+    figure, axes = plt.subplots(1, 2, figsize=(10, 4.2))
+    mean_axis, worst_axis = axes
+    for model in MODEL_ORDER:
+        draw(mean_axis, selected, model, "mean_ms")
+        # Max over trials: a compile stall is a rare event, so averaging the
+        # per-trial maxima averages away exactly what this panel measures.
+        draw(worst_axis, selected, model, "max_max_ms")
+
+    mean_axis.set_title("Mean iteration", fontweight="bold")
+    worst_axis.set_title("Worst iteration (max over trials)",
+                         fontweight="bold")
+    mean_axis.set_ylabel("Time (ms)")
+    worst_axis.set_ylabel("Time (ms)")
+    dpus = sorted({row.dpus for row in selected})
+    for axis in (mean_axis, worst_axis):
+        configure_axis(axis, dpus, dpus, "DPUs")
+
+    figure.suptitle(
+        f"Adaptive image: dynamic execution trade-offs "
+        f"({PLOT_SIZE:,} elements/DPU)", fontsize=14, fontweight="bold")
     figure.legend(handles=legend_handles(MODEL_ORDER, linewidth=2.1),
                   loc="upper center", ncol=4, frameon=False,
-                  bbox_to_anchor=(0.5, 0.955))
-    figure.tight_layout(rect=(0, 0, 1, 0.925), h_pad=1.6, w_pad=1.25)
+                  bbox_to_anchor=(0.5, 0.90))
+    figure.tight_layout(rect=(0, 0, 1, 0.86), w_pad=1.6)
     figure.savefig(FIGURE)
 
 

benchmarks/dynamic-benchmarks/plot_query_sweep.py

@@ -25,6 +25,10 @@
 RUNS_CSV = RESULTS / "query-sweep.csv"
 SECTIONS_CSV = RESULTS / "query-sweep.sections.csv"
 SUMMARY_CSV = RESULTS / "query-sweep-summary.csv"
+# Below ~1M elements the fixed JIT compile cost dominates every variant, so the
+# curves measure compiler overhead rather than execution.  Smaller runs stay in
+# the CSV; they are simply not plotted.
+MIN_TOTAL_ELEMENTS = 1_000_000
 FIGURE = RESULTS / "query-sweep.pdf"
 
 RunKey = Tuple[str, str, str, str]
@@ -80,22 +84,31 @@ def load_measurements() -> Dict[MeasurementKey, Measurement]:
     sections = load_sections()
     grouped = defaultdict(lambda: defaultdict(list))
 
+    # A rerun appends rows rather than replacing them, so keep only the newest
+    # row per (variant, size, trial); otherwise old and new runs get averaged.
+    newest = {}
     with RUNS_CSV.open(newline="") as file:
         for row in csv.DictReader(file):
             if row["benchmark"] != BENCHMARK or row["status"] != "complete":
                 continue
-
-            query_ops = int(parse_parameters(row["parameters"])["query_ops"])
-            measured = sections.get(run_key(row), {})
-            if not {"query_first", "query_reuse"} <= measured.keys():
+            if int(row["total_elements"]) < MIN_TOTAL_ELEMENTS:
                 continue
+            newest[(row["variant"], row["elements_per_dpu"], row["dpus"],
+                    row["trial"])] = row
+
+    for row in newest.values():
+
+        query_ops = int(parse_parameters(row["parameters"])["query_ops"])
+        measured = sections.get(run_key(row), {})
+        if not {"query_first", "query_reuse"} <= measured.keys():
+            continue
 
-            key = row["variant"], int(row["total_elements"]), query_ops
-            grouped[key]["first_ms"].append(measured["query_first"])
-            grouped[key]["reuse_ms"].append(measured["query_reuse"])
-            grouped[key]["query_ms"].append(float(row["time"]))
-            if row["real_s"]:
-                grouped[key]["wall_s"].append(float(row["real_s"]))
+        key = row["variant"], int(row["total_elements"]), query_ops
+        grouped[key]["first_ms"].append(measured["query_first"])
+        grouped[key]["reuse_ms"].append(measured["query_reuse"])
+        grouped[key]["query_ms"].append(float(row["time"]))
+        if row["real_s"]:
+            grouped[key]["wall_s"].append(float(row["real_s"]))
 
     measurements = {}
     for key, samples in grouped.items():
@@ -198,7 +211,20 @@ def plot_results(boundaries, measurements):
     )
     boundary_axis.set_yscale("log")
     boundary_axis.set_ylabel("Batches to break even")
-    boundary_axis.axhline(1, color="#777777", linewidth=1)
+    # The benchmark reuses each compiled kernel this many times, so curves
+    # below the line break even within a single query.
+    reuse = {int(parse_parameters(row["parameters"])["batches_per_query"])
+             for row in csv.DictReader(RUNS_CSV.open(newline=""))
+             if row["benchmark"] == BENCHMARK and row["status"] == "complete"}
+    if len(reuse) == 1:
+        batches = reuse.pop()
+        boundary_axis.axhline(batches, color="#777777", linewidth=1.2,
+                              linestyle="--")
+        boundary_axis.annotate(f"batches per query = {batches}",
+                               xy=(0.02, batches), xycoords=("axes fraction",
+                                                             "data"),
+                               fontsize=8, color="#555555",
+                               va="bottom")
 
     figure.suptitle("Dynamic query performance", fontsize=15,
                     fontweight="bold")

benchmarks/plot_param_gain.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Tuned-vs-default fusion parameters, as gain over default per cell."""
+import collections
+import csv
+import random
+import statistics
+from pathlib import Path
+
+RESULTS = Path(__file__).resolve().parent / "results" / "main"
+FIGURE_STEM = "param-gain"
+VARIANT = "polymerpim"
+VARIANT_COLOR = "#3264a8"
+# One figure per DPU count.  Profiles are tuned at 256; at 2048 some tuned
+# builds go bimodal run to run, which the whiskers make visible.
+DPU_COUNTS = (256, 2048)
+
+
+def _gain_ci_width(default, tuned, keys, draws=4000):
+    """Width of the bootstrap 95% CI on the plotted bar, in points."""
+    rng = random.Random(1)
+    bars = []
+    for _ in range(draws):
+        per_size = []
+        for k in keys:
+            a = [rng.choice(default[k]) for _ in default[k]]
+            b = [rng.choice(tuned[k]) for _ in tuned[k]]
+            med_a = statistics.median(a)
+            per_size.append(100 * (med_a - statistics.median(b)) / med_a)
+        bars.append(statistics.median(per_size))
+    bars.sort()
+    return bars[int(0.975 * draws)] - bars[int(0.025 * draws)]
+
+
+def load(path):
+    cells = collections.defaultdict(list)
+    with open(path, newline="") as handle:
+        for row in csv.DictReader(handle):
+            if row["status"] != "complete" or not row["time"]:
+                continue
+            key = (row["benchmark"], row["variant"], int(row["dpus"]),
+                   int(row["elements_per_dpu"]))
+            cells[key].append(float(row["time"]))
+    return cells
+
+
+def main():
+    tuned, default = load(RESULTS / "tuned.csv"), load(RESULTS / "default.csv")
+    shared = sorted(set(tuned) & set(default))
+    if not shared:
+        raise SystemExit("no cells present in both runs")
+
+    import matplotlib
+    matplotlib.use("Agg")
+    import matplotlib.pyplot as plt
+
+    for dpus in DPU_COUNTS:
+        benchmarks = sorted({key[0] for key in shared if key[2] == dpus})
+        if not benchmarks:
+            print(f"no cells at {dpus} DPUs")
+            continue
+
+        figure, axis = plt.subplots(1, 1, figsize=(7.2, 4.4))
+        xs, gains, lows, highs, flags = [], [], [], [], []
+        for index, benchmark in enumerate(benchmarks):
+            keys = [k for k in shared
+                    if k[0] == benchmark and k[1] == VARIANT and k[2] == dpus]
+            if not keys:
+                continue
+            # Median of 5 trials on both sides: one stalled run (a random I/O
+            # or allocation hiccup) cannot move a median, but would dominate a
+            # min/max over raw trials.  Whiskers span the per-size gains.
+            per_size = [100 * (statistics.median(default[k])
+                               - statistics.median(tuned[k]))
+                        / statistics.median(default[k]) for k in keys]
+            middle = statistics.median(per_size)
+            # Flag uncertainty in the gain, not in the raw times: at 2048 DPUs a
+            # cell can be noisy on both sides and still pin the gain down, since
+            # common-mode noise cancels in the ratio.
+            unstable = _gain_ci_width(default, tuned, keys) > 10.0
+            xs.append(index)
+            gains.append(middle)
+            lows.append(middle - min(per_size))
+            highs.append(max(per_size) - middle)
+            flags.append(unstable)
+        bars = axis.bar(xs, gains, width=0.55, yerr=[lows, highs], capsize=4,
+                        color=VARIANT_COLOR)
+        for bar, unstable in zip(bars, flags):
+            if not unstable:
+                continue
+            bar.set_hatch("//")
+            bar.set_edgecolor("white")
+            axis.annotate("unstable", xy=(bar.get_x() + bar.get_width() / 2,
+                                          0), xytext=(0, -14),
+                          textcoords="offset points", ha="center",
+                          fontsize=7, color="#a0342c")
+        axis.axhline(0, color="#444444", linewidth=1)
+        axis.set_xticks(range(len(benchmarks)))
+        axis.set_xticklabels(benchmarks, rotation=30, ha="right")
+        axis.grid(True, axis="y", color="#dddddd", linewidth=0.8)
+        axis.set_axisbelow(True)
+        axis.set_ylabel("Gain from tuned parameters (%)")
+        figure.suptitle(f"Tuned fusion parameters vs defaults ({dpus} DPUs)",
+                        fontsize=14, fontweight="bold")
+        figure.tight_layout(rect=(0, 0, 1, 0.93))
+        path = RESULTS / f"{FIGURE_STEM}-{dpus}.pdf"
+        figure.savefig(path)
+        plt.close(figure)
+        cells = sum(1 for k in shared if k[1] == VARIANT and k[2] == dpus)
+        print(f"Wrote {path}  ({cells} cells, {len(benchmarks)} benchmarks)")
+    return
+
+
+if __name__ == "__main__":
+    main()