[CK_TILE] Stream-K bridge: port #8123 driver/bench improvements

The Stream-K bridge keeps its own driver, worker and ctypes lib, so the
regular-GEMM bridge improvements that landed on #8123 after this branch
forked did not arrive via the merge. Port the Stream-K-specific analogues:

- streamk_gemm_ctypes_lib.cpp: benchmark knobs now default to old-TE's
  warmup=50/repeat=100 (was 3/10 -- a cold, un-ramped clock, the root of
  #8123's spurious "perf gap") and are env-overridable via
  CK_TILE_BENCH_WARMUP/REPEAT/FLUSH/ROTATING. Unlike the regular path,
  rotating_count defaults to 1: the Atomic preprocess re-zeros only the
  original C buffer, so rotating C would corrupt the accumulation.
- streamk_gemm_full_benchmark.py: fan the (kernel x problem) work across
  every visible GPU (device-pinned HIP_VISIBLE_DEVICES workers, --devices,
  device CSV column), add the --verify/--verify-tol fp32-reference gate, and
  constrain --dtype/--layout to the supported fp16/rcr surface. Also fixes a
  latent proc-unbound error in the batch handler.
- run_one_streamk_gemm_kernel.py: add the fp32 numpy reference check
  (global max|out-ref|/max|ref|, verified/max_rel) behind --verify.
- README: document the Stream-K bridge driver/worker, flags, _streamk name
  suffix, fp16 Atomic tolerance, and the rotating_count divergence.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: ozturkosu

projects/composablekernel/dispatcher/bindings/ctypes/streamk_gemm_ctypes_lib.cpp

@@ -32,6 +32,7 @@
 
 #include <hip/hip_runtime.h>
 #include <cstdint>
+#include <cstdlib>
 #include <cstring>
 #include <exception>
 #include <string>
@@ -47,6 +48,33 @@
 
 static bool g_initialized = false;
 
+// Read an integer benchmark knob from the environment, falling back to
+// `fallback` when unset or unparseable.
+static int env_int(const char* name, int fallback)
+{
+    const char* v = std::getenv(name);
+    if(v == nullptr || *v == '\0')
+        return fallback;
+    char* end      = nullptr;
+    const long out = std::strtol(v, &end, 10);
+    if(end == v)
+        return fallback;
+    return static_cast<int>(out);
+}
+
+// Read a boolean benchmark knob ("0"/"false"/"off", any case => false, else true).
+static bool env_bool(const char* name, bool fallback)
+{
+    const char* v = std::getenv(name);
+    if(v == nullptr || *v == '\0')
+        return fallback;
+    std::string s(v);
+    for(char& c : s)
+        if(c >= 'A' && c <= 'Z')
+            c = static_cast<char>(c - 'A' + 'a');
+    return !(s == "0" || s == "false" || s == "off");
+}
+
 extern "C" {
 
 /**
@@ -150,15 +178,29 @@ int dispatcher_run_gemm(
                                   /*stride_B=*/static_cast<ck_tile::index_t>(K),
                                   /*stride_C=*/static_cast<ck_tile::index_t>(N));
 
+    // Benchmark parameters. warmup/repeat default to old Tile Engine's values
+    // (warmup=50, repeat=100); a generous warmup keeps the GPU clock ramped, and
+    // 100 timed iterations give a stable median. These were the knobs behind the
+    // regular bridge's spurious "perf gap" (#8123): the old default of warmup=3/
+    // repeat=10 measured a cold, un-ramped clock. Each knob is env-overridable so
+    // a caller can match another harness without recompiling.
+    //
+    // Divergence from the regular path (generated_tile_backend.hpp): flush_cache_
+    // and rotating_count_ default OFF here. The Stream-K Atomic reduction
+    // accumulates into C, and the generated launch's launch_kernel_time_mask
+    // preprocess re-zeros only the original args.e_ptr -- rotating C across
+    // multiple buffers would leave the rotated copies un-zeroed and corrupt the
+    // accumulation. Leave rotating_count_=1 unless a caller knows the kernel
+    // re-zeros every rotated buffer.
     ck_tile::stream_config stream_cfg;
     stream_cfg.stream_id_      = nullptr;
     stream_cfg.time_kernel_    = true;
     stream_cfg.log_level_      = 0;
-    stream_cfg.cold_niters_    = 3;
-    stream_cfg.nrepeat_        = 10;
+    stream_cfg.cold_niters_    = env_int("CK_TILE_BENCH_WARMUP", 50);
+    stream_cfg.nrepeat_        = env_int("CK_TILE_BENCH_REPEAT", 100);
     stream_cfg.is_gpu_timer_   = true;
-    stream_cfg.flush_cache_    = false;
-    stream_cfg.rotating_count_ = 1;
+    stream_cfg.flush_cache_    = env_bool("CK_TILE_BENCH_FLUSH", false);
+    stream_cfg.rotating_count_ = env_int("CK_TILE_BENCH_ROTATING", 1);
 
     float exec_time = 0.0f;
     try

projects/composablekernel/tile_engine/ops/gemm/README.md

@@ -138,6 +138,40 @@ What that means for this PR:
 - Grouped GEMM and stream-K go through **separate bridge efforts** (stream-K in
   #8136, grouped GEMM on its own branch), not this PR.
 
+### Stream-K bridge
+
+Stream-K reuses the same bridge mechanics through its **own** driver/worker pair
+(it keeps a distinct ctypes lib that bypasses the registry; see
+`dispatcher/bindings/ctypes/streamk_gemm_ctypes_lib.cpp`):
+
+| Script | Role |
+|---|---|
+| `streamk_gemm_full_benchmark.py` | Stream-K driver — same 3-phase, multi-GPU, `--verify`/`--devices` flags as the regular driver; threads `variant="stream_k"`. Defaults to `gemm_streamk/configs/default_config.json`. |
+| `run_one_streamk_gemm_kernel.py` | Disposable per-kernel worker (same fp16 host path as the regular worker; the ABI matches). |
+
+```bash
+# Default Stream-K config, all visible GPUs, with correctness checking:
+python streamk_gemm_full_benchmark.py --verify
+
+# Explicit config, 4 GPUs, custom output:
+python streamk_gemm_full_benchmark.py gemm_streamk/configs/default_config.json \
+    --devices 4 --csv streamk_gemm_results.csv
+```
+
+Notes specific to Stream-K:
+- Kernel names carry a `_streamk` suffix (`GemmKernelConfig(variant="stream_k").name`).
+- Supported surface here is **fp16 / rcr**, same as the regular bridge.
+- The Atomic reduction does multiple fp16 atomic-adds per K-split, so it is noisier
+  than regular GEMM; `--verify` still uses the default `2e-2` gate (observed
+  `max_rel ≤ 2.5e-3`, well within it).
+- Benchmark knobs in `streamk_gemm_ctypes_lib.cpp` default to warmup=50/repeat=100
+  (env-overridable via `CK_TILE_BENCH_WARMUP`/`REPEAT`/`FLUSH`/`ROTATING`), matching
+  the regular path — **except** `rotating_count` defaults to 1, because the Atomic
+  preprocess only re-zeros the original C buffer and rotating C would corrupt the
+  accumulation.
+- Tiny problems (e.g. `257³`) have too few tiles to partition across CUs; the kernel
+  reports them unsupported (status `-2`) and the bridge surfaces that gracefully.
+
 ### Removal note
 
 The legacy regular-GEMM standalone build path has been **removed**, and the

projects/composablekernel/tile_engine/ops/gemm/run_one_streamk_gemm_kernel.py

@@ -15,8 +15,16 @@
     Single: {"so_path": "...", "problem": {"M":.., "N":.., "K":..}, "kernel_name": "..."}
     Batch:  {"items": [{"so_path": "...", "problem": {...}, "kernel_name": "..."}, ...]}
 
+Optional top-level keys ``verify`` (bool) and ``verify_tol`` (float) enable an
+fp32 numpy reference check; when set, each OK result also carries ``verified``
+and ``max_rel``. Stream-K's Atomic reduction does multiple fp16 atomic-adds (one
+per K-split partial), so it is inherently noisier than a single fp32->fp16 store;
+the default gate tolerance (2e-2) is loose enough to pass while still catching
+gross errors.
+
 Output JSON format (one line per kernel):
     {"idx": 0, "ok": true, "ms": 0.123, "tflops": 456.7, "non_zero": 1, "kernel": "..."}
+    {"idx": 0, "ok": true, ..., "verified": true, "max_rel": 1.1e-3}   # with --verify
     {"idx": 1, "ok": false, "error": "...", "kernel": "..."}
 """
 
@@ -35,8 +43,14 @@
 import numpy as np  # noqa: E402
 
 
-def _run_one(idx, so_path, prob_dict, kernel_name):
-    """Run a single kernel and emit its result as one JSON line."""
+def _run_one(idx, so_path, prob_dict, kernel_name, verify=False, verify_tol=2e-2):
+    """Run a single kernel and emit its result as one JSON line.
+
+    When ``verify`` is set, the kernel output is checked against an fp32 numpy
+    reference (``A @ B``) using the global relative metric
+    ``max|out - ref| / max|ref|``; the emitted ``verified`` field then reflects
+    correctness, not just liveness (``non_zero``).
+    """
     try:
         problem = GemmProblem.from_dict(prob_dict)
 
@@ -54,19 +68,22 @@ def _run_one(idx, so_path, prob_dict, kernel_name):
                 if result.output is not None
                 else 0
             )
-            print(
-                json.dumps(
-                    {
-                        "idx": idx,
-                        "ok": True,
-                        "ms": result.time_ms,
-                        "tflops": result.tflops,
-                        "non_zero": non_zero,
-                        "kernel": kernel_name,
-                    }
-                ),
-                flush=True,
-            )
+            out = {
+                "idx": idx,
+                "ok": True,
+                "ms": result.time_ms,
+                "tflops": result.tflops,
+                "non_zero": non_zero,
+                "kernel": kernel_name,
+            }
+            if verify:
+                ref = A.astype(np.float32) @ B.astype(np.float32)
+                got = result.output.astype(np.float32)
+                denom = float(np.max(np.abs(ref))) or 1.0
+                max_rel = float(np.max(np.abs(got - ref)) / denom)
+                out["max_rel"] = max_rel
+                out["verified"] = bool(max_rel <= verify_tol)
+            print(json.dumps(out), flush=True)
         else:
             print(
                 json.dumps(
@@ -100,13 +117,28 @@ def main():
         )
         sys.exit(1)
 
+    verify = bool(d.get("verify", False))
+    verify_tol = float(d.get("verify_tol", 2e-2))
+
     if "items" in d:
         for i, item in enumerate(d["items"]):
             _run_one(
-                i, item["so_path"], item["problem"], item.get("kernel_name", "unknown")
+                i,
+                item["so_path"],
+                item["problem"],
+                item.get("kernel_name", "unknown"),
+                verify=verify,
+                verify_tol=verify_tol,
             )
     else:
-        _run_one(0, d["so_path"], d["problem"], d.get("kernel_name", "unknown"))
+        _run_one(
+            0,
+            d["so_path"],
+            d["problem"],
+            d.get("kernel_name", "unknown"),
+            verify=verify,
+            verify_tol=verify_tol,
+        )
 
 
 if __name__ == "__main__":