Add perf experiment pipeline and fix v2 connection pool regressions

.github/instructions/connection-pooling.instructions.md

@@ -124,13 +124,20 @@ private readonly ChannelWriter<DbConnectionInternal?> _idleConnectionWriter;
 await _idleConnectionReader.WaitToReadAsync(token);
 ```
 
-### Sync Over Async Protection
+### Sync Over Async Blocking
 ```csharp
-// Prevent thread pool starvation
-private static SemaphoreSlim _syncOverAsyncSemaphore = 
-    new(Math.Max(1, Environment.ProcessorCount / 2));
+// Block via Task, not an opaque primitive, so the thread pool learns this
+// thread is stalled and can inject a worker to run the read's continuation.
+return _idleChannel.ReadAsync(cancellationToken).AsTask().GetAwaiter().GetResult();
 ```
 
+The idle channel is created without `AllowSynchronousContinuations`, so the continuation
+that completes a pending read is queued to the thread pool. Blocking on an opaque primitive
+(`ManualResetEventSlim`, `SemaphoreSlim`) hides the stall from the thread pool, so under
+saturation every parked thread waits on a continuation the pool does not know it needs to
+run. Do not reintroduce a semaphore around this wait: it cannot stop sync work from
+consuming thread pool threads, since waiting on it blocks just as opaquely.
+
 ## Best Practices
 
 ### Application Design

eng/pipelines/perf/scripts/interleave_perf.py

@@ -101,17 +101,22 @@ def apply_affinity(proc, cpus):
 # --------------------------------------------------------------------------------------------------
 # Running one unit and collecting its artifacts.
 # --------------------------------------------------------------------------------------------------
-def run_unit_process(exe_dir, assembly, unit, cwd, cpus, log_path):
+def run_unit_process(exe_dir, assembly, unit, cwd, cpus, log_path, env_overrides=None):
     """Run one benchmark *unit* from the build at *exe_dir* in *cwd*.
 
     Returns the subprocess return code.  Kept as a small seam so tests can substitute
-    a fake runner.
+    a fake runner.  *env_overrides*, when given, is applied on top of the inherited
+    environment (e.g. a per-variant RUNNER_CONFIG so baseline and current can run with
+    different SqlClient behaviour flags, such as comparing the legacy vs new connection
+    pool from the SAME build).
     """
     os.makedirs(cwd, exist_ok=True)
     cmd = ["dotnet", os.path.join(exe_dir, assembly)]
     env = dict(os.environ)
     env["PERF_BENCHMARK"] = unit
     env.pop("PERF_LIST_BENCHMARKS", None)
+    if env_overrides:
+        env.update(env_overrides)
 
     with open(log_path, "w", encoding="utf-8") as log:
         proc = subprocess.Popen(cmd, cwd=cwd, env=env, stdout=log,
@@ -154,12 +159,18 @@ def collect_results(cwd, dest):
 class Runner:
     """Holds the invariant run parameters and performs interleaved unit passes."""
 
-    def __init__(self, baseline_dir, current_dir, assembly, work_dir, cpus):
+    def __init__(self, baseline_dir, current_dir, assembly, work_dir, cpus,
+                 baseline_runner_config=None, current_runner_config=None):
         self.baseline_dir = baseline_dir
         self.current_dir = current_dir
         self.assembly = assembly
         self.work_dir = work_dir
         self.cpus = cpus
+        # Optional per-variant RUNNER_CONFIG override (e.g. --switch-under-test needs baseline
+        # and current to run with different SqlClient behaviour flags even though they share the
+        # same build). None means "no override" -> both variants use the ambient RUNNER_CONFIG.
+        self.baseline_runner_config = baseline_runner_config
+        self.current_runner_config = current_runner_config
 
     def list_units(self):
         cmd = ["dotnet", os.path.join(self.current_dir, self.assembly)]
@@ -175,7 +186,11 @@ def _run_one(self, variant, exe_dir, unit, rep, agg_dir):
             shutil.rmtree(cwd)
         os.makedirs(cwd, exist_ok=True)
         log_path = os.path.join(cwd, "run.log")
-        rc = run_unit_process(exe_dir, self.assembly, unit, cwd, self.cpus, log_path)
+        runner_config = (self.baseline_runner_config if variant == "baseline"
+                         else self.current_runner_config)
+        env_overrides = {"RUNNER_CONFIG": runner_config} if runner_config else None
+        rc = run_unit_process(exe_dir, self.assembly, unit, cwd, self.cpus, log_path,
+                              env_overrides=env_overrides)
         if rc != 0:
             _tail(log_path)
             raise RuntimeError(f"benchmark unit '{unit}' ({variant}, rep {rep}) failed (exit {rc}).")
@@ -279,7 +294,6 @@ def orchestrate(runner, units, results_dir, threshold, reps):
     unconfirmed = [e for e in entries if e["status"] == "regression-unconfirmed"]
     return entries, confirmed, unconfirmed
 
-
 # --------------------------------------------------------------------------------------------------
 # Rendering.
 # --------------------------------------------------------------------------------------------------
@@ -332,6 +346,7 @@ def render_markdown(entries, confirmed, unconfirmed, baseline_version, threshold
             )
         )
     lines.append("")
+
     return "\n".join(lines)
 
 
@@ -353,6 +368,14 @@ def main(argv=None):
     parser.add_argument("--reps", type=int, default=3,
                         help="Total interleaved passes for a flagged unit (best-of-N). 1 disables confirmation.")
     parser.add_argument("--baseline-version", default="baseline")
+    parser.add_argument("--baseline-runner-config", default=None,
+                        help="Override RUNNER_CONFIG for baseline-variant subprocesses only "
+                             "(e.g. to force a different SqlClient behaviour flag for the "
+                             "baseline, such as comparing the legacy vs new connection pool "
+                             "from the same build). Omit to use the ambient RUNNER_CONFIG for "
+                             "both variants (default behaviour).")
+    parser.add_argument("--current-runner-config", default=None,
+                        help="Override RUNNER_CONFIG for current-variant subprocesses only.")
     parser.add_argument("--client-cpus", default=os.environ.get("PERF_CLIENT_CPUS", ""),
                         help="CPU set to pin the benchmark client to, e.g. '16-31'.")
     parser.add_argument("--fail-on-regression", action="store_true",
@@ -376,7 +399,9 @@ def main(argv=None):
 
     runner = Runner(os.path.abspath(args.baseline_exe_dir),
                     os.path.abspath(args.current_exe_dir),
-                    args.assembly, work_dir, cpus)
+                    args.assembly, work_dir, cpus,
+                    baseline_runner_config=args.baseline_runner_config,
+                    current_runner_config=args.current_runner_config)
 
     units = runner.list_units()
     if not units:
@@ -389,7 +414,8 @@ def main(argv=None):
 
     # Outputs.
     comparison_dir = os.path.join(results_dir, "comparison")
-    md = render_markdown(entries, confirmed, unconfirmed, args.baseline_version, args.threshold, args.reps)
+    md = render_markdown(entries, confirmed, unconfirmed,
+                         args.baseline_version, args.threshold, args.reps)
     with open(os.path.join(comparison_dir, "comparison.md"), "w", encoding="utf-8") as fh:
         fh.write(md + "\n")
     with open(os.path.join(comparison_dir, "comparison.json"), "w", encoding="utf-8") as fh: