diff --git a/Makefile b/Makefile index 6bb3e70..78c58b6 100644 --- a/Makefile +++ b/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 diff --git a/benchmarks/_plot_common.py b/benchmarks/_plot_common.py index 36fa478..a582ac5 100644 --- a/benchmarks/_plot_common.py +++ b/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"]), diff --git a/benchmarks/dynamic-benchmarks/benchmark.toml b/benchmarks/dynamic-benchmarks/benchmark.toml index eec0a06..75a5b2d 100644 --- a/benchmarks/dynamic-benchmarks/benchmark.toml +++ b/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", diff --git a/benchmarks/dynamic-benchmarks/plot_adaptive_image.py b/benchmarks/dynamic-benchmarks/plot_adaptive_image.py index 258d515..86a776a 100755 --- a/benchmarks/dynamic-benchmarks/plot_adaptive_image.py +++ b/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) diff --git a/benchmarks/dynamic-benchmarks/plot_query_sweep.py b/benchmarks/dynamic-benchmarks/plot_query_sweep.py index e88b440..87a0bb1 100755 --- a/benchmarks/dynamic-benchmarks/plot_query_sweep.py +++ b/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") diff --git a/benchmarks/plot_param_gain.py b/benchmarks/plot_param_gain.py new file mode 100644 index 0000000..3cbf959 --- /dev/null +++ b/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() diff --git a/benchmarks/src/tuning.jl b/benchmarks/src/tuning.jl index 26be7e5..8bdf80b 100644 --- a/benchmarks/src/tuning.jl +++ b/benchmarks/src/tuning.jl @@ -1,6 +1,6 @@ const DEFAULT_FUSION_BUILD = Dict( "FUSION_LOOKAHEAD" => 128, - "MAX_HFUSE_CHAINS" => 10, + "MAX_HFUSE_CHAINS" => 9, "JIT_BATCH_SIZE" => 16, "MAX_VFUSE_OPS" => 128, "MAX_VFUSE_INPUTS" => 11, @@ -9,7 +9,8 @@ const DEFAULT_FUSION_BUILD = Dict( const DEFAULT_FUSION_SEARCH = Dict( "FUSION_LOOKAHEAD" => [0, 1, 2, 4, 8, 16, 32, 64, 128], - "MAX_HFUSE_CHAINS" => [1, 2, 4, 6, 8, 10], + # 9 is the ceiling: a tenth chain overflows WRAM at stack depth 3. + "MAX_HFUSE_CHAINS" => [1, 2, 4, 6, 8, 9], "JIT_BATCH_SIZE" => [0, 1, 2, 4, 8, 16, 32], "MAX_VFUSE_OPS" => [1, 8, 16, 32, 64, 96, 128, 192], ) @@ -494,6 +495,15 @@ function tune_target(config::RunnerConfig, spec::BenchmarkSpec, options::TuneOpt end seed = copy(DEFAULT_FUSION_BUILD) + # The seed must be buildable: vector_search at dim=16 needs 16 inputs. + if !isempty(options.workspace_profiles) + seed["MAX_VFUSE_INPUTS"], seed["BLOCK_SIZE_LOG2"] = + options.workspace_profiles[1] + end + for (knob, candidates) in options.search + haskey(seed, knob) && !(seed[knob] in candidates) && + (seed[knob] = maximum(candidates)) + end initial = evaluate(seed) record!(checkpoint, seed, initial, "initial", "", "") isfinite(initial.objective) || error("initial configuration failed for $(spec.name)") diff --git a/benchmarks/test/tests/timings.jl b/benchmarks/test/tests/timings.jl index 2c9ab24..531ddaf 100644 --- a/benchmarks/test/tests/timings.jl +++ b/benchmarks/test/tests/timings.jl @@ -43,7 +43,7 @@ const SAMPLE_APP_TIME = @test run["min"] == "3.1" @test run["max"] == "3.4" @test [run[knob] for knob in BenchmarkRunner.FUSION_BUILD_KNOBS] == - ["128", "10", "16", "128", "11", "4"] + ["128", "9", "16", "128", "11", "4"] sections = csv_records(BenchmarkRunner.sections_csv(path)) @test length(sections) == 4 diff --git a/benchmarks/test/tests/tuning.jl b/benchmarks/test/tests/tuning.jl index 3411894..ff3637e 100644 --- a/benchmarks/test/tests/tuning.jl +++ b/benchmarks/test/tests/tuning.jl @@ -1,6 +1,7 @@ @testset "fusion search" begin + # 9 is the ceiling: a tenth chain overflows WRAM at stack depth 3. @test BenchmarkRunner.DEFAULT_FUSION_SEARCH["MAX_HFUSE_CHAINS"] == - [1, 2, 4, 6, 8, 10] + [1, 2, 4, 6, 8, 9] seed = copy(BenchmarkRunner.DEFAULT_FUSION_BUILD) search = Dict(knob => [value] for (knob, value) in seed) search["FUSION_LOOKAHEAD"] = [1, 2, 3] diff --git a/host/detail/queue.cc b/host/detail/queue.cc index b60b7e5..c8c35a9 100644 --- a/host/detail/queue.cc +++ b/host/detail/queue.cc @@ -423,6 +423,15 @@ void EventQueue::finalize_finished_events() { std::lock_guard lock(mtx_); while (!running_events_.empty() && running_events_.front()->finished) { auto e = running_events_.front(); + // A reduction result is terminal -- the host reads it, nothing re-derives + // it -- so once the reduce has run its absorbed input list is dead weight. + // Holding it keeps every full-length intermediate alive for as long as the + // caller keeps the future, which a batch of reductions cannot afford. + if (e->output && e->output->is_reduction_result) { + e->output->absorbed_inputs.clear(); + e->output->absorbed_rpn.clear(); + e->output->absorbed_scalars.clear(); + } last_finished_id_.store(e->max_id); trace::execution_end(); running_events_.pop_front(); @@ -508,6 +517,7 @@ void EventQueue::await_jit_binary(const std::shared_ptr& e) { std::future_status::ready) { if (detail::pipeline_can_interpret(e->rpn_ops)) { e->jit_pipeline_fallback = true; + ++unused_compiles_; VECTORDPU_NOTE(jit_pipeline_fallbacks); return; } diff --git a/host/jit.cc b/host/jit.cc index 9706adc..2ffc18d 100644 --- a/host/jit.cc +++ b/host/jit.cc @@ -323,6 +323,15 @@ void EventQueue::flush_jit_batch() { std::vector, std::string>> batch = pending_unique_kernels_; + // Re-link the previous kernels too, so a mix of old and new shapes does not + // ping-pong between binaries. New kernels stay in front to keep slots. + for (const auto& prior : last_linked_kernels_) { + if (batch.size() >= jit_link_batch_limit()) break; + bool present = false; + for (const auto& have : batch) present = present || (have == prior); + if (!present) batch.push_back(prior); + } + #if ENABLE_DPU_LOGGING >= 1 auto log = DpuRuntime::get().get_logger().lock(logcat::QUEUE_JIT); log.first() << "flush JIT batch"; @@ -338,10 +347,10 @@ void EventQueue::flush_jit_batch() { #endif [batch]() { return jit_compile(batch); }); for (auto& ev : pending_jit_events_) ev->jit_future = future; -#if JIT_PIPELINE_FALLBACK + // Every kernel in the superset now resolves to this binary. for (size_t i = 0; i < batch.size(); ++i) inflight_jit_kernels_[batch[i]] = {future, (int)i}; -#endif + last_linked_kernels_ = batch; pending_jit_events_.clear(); pending_unique_kernels_.clear(); @@ -351,6 +360,9 @@ void EventQueue::lock_for_jit(std::shared_ptr e) { if (e->op != Event::OperationType::COMPUTE || e->is_locked_for_jit || !e->jit_binary_path.empty()) return; +#if JIT_PIPELINE_FALLBACK + if (unused_compiles_ >= kStopCompilingAfter) return; +#endif if (e->inputs.empty()) return; // no input: no RPN form (e.g. a fill) e->is_locked_for_jit = true; @@ -392,8 +404,7 @@ void EventQueue::lock_for_jit(std::shared_ptr e) { } } -#if JIT_PIPELINE_FALLBACK - // Relink cached objects into an active batch to avoid binary ping-pong. + // Reuse the binary a kernel already sits in, to avoid binary ping-pong. auto inflight = inflight_jit_kernels_.find(sig); if (inflight != inflight_jit_kernels_.end() && pending_unique_kernels_.empty()) { @@ -401,7 +412,6 @@ void EventQueue::lock_for_jit(std::shared_ptr e) { e->jit_sub_kernel_idx = inflight->second.slot; return; } -#endif if (pending_unique_kernels_.size() >= jit_link_batch_limit()) flush_jit_batch(); diff --git a/host/polymerpim.cc b/host/polymerpim.cc index 94a715a..46461d8 100644 --- a/host/polymerpim.cc +++ b/host/polymerpim.cc @@ -43,6 +43,10 @@ struct Node { bool structural = false; // Cached subtree op count; the deferral cap reads it once per op. size_t ops = 0; + // Cached RPN stack depth this subtree needs to evaluate. + size_t stack_depth = 0; + // Upper bound on distinct inputs; shared vectors are counted once per use. + size_t input_count = 0; std::vector> children; }; @@ -55,6 +59,17 @@ NodeRef node(ExprOp op, std::vector children = {}) { for (const auto& child : children) { if (child) result->ops += child->ops; } + result->input_count = (op == ExprOp::input) ? 1 : 0; + for (const auto& child : children) { + if (child) result->input_count += child->input_count; + } + // Child i sits above i already-pushed slots. + result->stack_depth = 1; + for (size_t i = 0; i < children.size(); ++i) { + if (!children[i]) continue; + const size_t needed = i + children[i]->stack_depth; + if (needed > result->stack_depth) result->stack_depth = needed; + } result->children = std::move(children); return result; } @@ -112,6 +127,16 @@ uint8_t binary_opcode(ExprOp op) { // Ops a pending expression contributes to a fused program. size_t expression_ops(const NodeRef& value) { return value ? value->ops : 0; } +// RPN stack slots needed; past the interpreter's stack it cannot run at all. +size_t expression_depth(const NodeRef& value) { + return value ? value->stack_depth : 0; +} + +// Deferring past the input budget makes lowering throw; materialise instead. +size_t expression_inputs(const NodeRef& value) { + return value ? value->input_count : 0; +} + uint8_t scalar_opcode(ExprOp op) { switch (op) { case ExprOp::add: @@ -439,8 +464,19 @@ struct DPUVector::Impl { BackendVector& storage() const { if (!pending) return value; - value = is_fill() ? fill_vector(pending->scalar, length, name) - : materialize(pending); + const bool fill = is_fill(); + value = fill ? fill_vector(pending->scalar, length, name) + : materialize(pending); + // A consumer that already fused this chain shares the node: point it at + // the buffer so it is not evaluated twice. A fill stays scalar. + if (consumed && !fill) { + pending->op = ExprOp::input; + pending->input = value; + pending->children.clear(); + pending->ops = 0; + pending->stack_depth = 1; + pending->input_count = 1; + } pending = nullptr; return value; } @@ -590,7 +626,9 @@ DPUVector::operator DpuLazy() const { // Past the cap the chain cannot fuse into one kernel, so extending it only // mints another program shape for the JIT to compile. if (impl_->pending && !impl_->consumed && - expression_ops(impl_->pending) < MAX_VFUSE_OPS) { + expression_ops(impl_->pending) < MAX_VFUSE_OPS && + expression_depth(impl_->pending) <= MAX_PIPELINE_STACK_DEPTH && + expression_inputs(impl_->pending) < MAX_COMBINED_INPUTS) { impl_->consumed = true; return DpuLazy(std::make_shared::Impl>(impl_->pending)); } diff --git a/host/queue.cc b/host/queue.cc index 3a7bf07..39977d7 100644 --- a/host/queue.cc +++ b/host/queue.cc @@ -85,7 +85,14 @@ bool EventQueue::process_next() { log_launch(e); begin_running(e); +#if JIT_PIPELINE_FALLBACK + // Gate both compile paths together: skipping only the batch path diverts + // here, which compiles synchronously and blocks. + if (unused_compiles_ < kStopCompilingAfter) compile_kernel_if_unbatched(e); + if (e->is_locked_for_jit && !e->jit_pipeline_fallback) unused_compiles_ = 0; +#else compile_kernel_if_unbatched(e); +#endif switch_dpu_binary(e); try { diff --git a/host/queue.h b/host/queue.h index e7c3548..93f8779 100644 --- a/host/queue.h +++ b/host/queue.h @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -11,16 +12,10 @@ #include #include -#include "config.h" - -#if JIT_PIPELINE_FALLBACK -#include - -#include "jit.h" -#endif - #include "common.h" +#include "config.h" #include "detail/vector_desc.h" +#include "jit.h" class Event : public std::enable_shared_from_this { public: @@ -184,12 +179,19 @@ class EventQueue { std::vector, std::string>> pending_unique_kernels_; std::vector> pending_jit_events_; -#if JIT_PIPELINE_FALLBACK struct InflightJitKernel { std::shared_future binary; int slot; }; std::map inflight_jit_kernels_; + // Last binary's kernels; the next link re-includes them so it is a superset. + std::vector, std::string>> + last_linked_kernels_; +#if JIT_PIPELINE_FALLBACK + // Interpreted launches in a row whose compile never landed; past the + // threshold compiling is pure overhead. + static constexpr uint32_t kStopCompilingAfter = 8; + uint32_t unused_compiles_ = 0; #endif std::string current_binary_path_;