diff --git a/BUILDGUIDE.md b/BUILDGUIDE.md index 1751348b1e..f1d1372a84 100644 --- a/BUILDGUIDE.md +++ b/BUILDGUIDE.md @@ -648,3 +648,64 @@ $ dotnet run -c Release -f net10.0 $ RUNNER_CONFIG=~/.configs/runnerconfig.jsonc dotnet run -c Release -f net10.0 ``` + +### Sustained async connectivity scorecard + +Enable `Benchmarks.ConnectivityLoadRunnerConfig` in your external runner config +and select only this unit: + +```bash +PERF_BENCHMARK=ConnectivityLoad RUNNER_CONFIG=~/.configs/runnerconfig.jsonc \ + dotnet run -c Release -f net10.0 +``` + +This benchmark runs 1, 16, and 64 concurrent async worker loops by default against the +configured SQL Server. Each loop awaits `OpenAsync` and disposes the connection; +it does not manufacture network delays or synchronously wait on async tasks. +`Pooling=false` measures repeated physical connections, while `Pooling=true` +measures mostly warm-pool checkout/return after the initial ramp. Each iteration +starts with its own pool cleared. No tables or special server are required. +`Concurrency` and `Pooling` arrays in the load config select the parameter cases. +For example, `"Concurrency": [16], "Pooling": [false]` runs only physical opens +at concurrency 16, allowing each case to be measured in its own fresh process. + +`DurationSeconds` (default 10) controls how long loops admit new operations. +Already-started opens finish before the measurement ends; their time and +successful completions are included. An open failure fails the invocation, +stops new admissions and drains in-flight operations instead of reporting a +misleading improvement. Keep a finite connection timeout in the connection string. +Failed benchmark cases make the `ConnectivityLoad` process exit nonzero. +`SampleIntervalMilliseconds` (default 10) controls a dedicated observer thread, +so sampling can continue even when the thread pool is starved. Sampling integrates +observed occupancy using actual timestamps, including a final sample. + +The custom diagnoser adds these values to the BenchmarkDotNet summary and full +JSON reports, excluding warmup invocations: + +| Metric | Meaning | +| --- | --- | +| Avg workers | Time-weighted occupied thread-pool workers. | +| Worker ms/open | Integrated occupied worker time divided by successful opens. | +| Opens/sec | Successful opens divided by actual measured elapsed time. | +| Peak workers | Largest sampled occupied-worker count. | +| Peak TP threads | Largest sampled existing thread-pool count, including idle threads. | +| Peak process threads | Largest sampled process thread count, including dedicated threads and the observer. | +| CPU ms/open | Process CPU time divided by successful opens. | +| Opens / Samples / Measured sec | Totals across measured invocations, for checking normalization and sampling. | + +Occupancy is `GetMaxThreads` minus `GetAvailableThreads`, **not a blocked-thread +counter**: it includes CPU work, other process activity, and connection disposal. +Interpret lower occupancy alongside throughput and CPU cost. This is a +closed-loop workload, so improved async can increase throughput without lowering +average occupancy. Longer runs reduce sampling noise but do not remove systematic +server/network variation or exercise cold/expired authentication caches by themselves. +The process-thread peak guards against moving blocking off the thread pool onto +dedicated threads. All peaks and integrated occupancy are sampling estimates; +short waits between samples can be missed. + +Use fresh benchmark processes for baseline and candidate, keep concurrency, +duration, sampling interval, SDK/CPU settings, authentication, and server/network +conditions matched, and repeat runs. The suite's in-process toolchain preserves +driver switches but does not reset thread-pool history between parameter cases. +The existing perf comparison script emits custom metric deltas in JSON and a +separate Markdown table; its execution-time regression gate is unchanged. diff --git a/TESTGUIDE.md b/TESTGUIDE.md index a87d6aa028..4e08a95ceb 100644 --- a/TESTGUIDE.md +++ b/TESTGUIDE.md @@ -5,6 +5,9 @@ tests. For build prerequisites and general `build.proj` usage, see [BUILDGUIDE.md](BUILDGUIDE.md). +For isolated worker-starvation, cancellation, and transaction-lock reproductions, +see [Async connectivity reproductions](doc/async-connectivity-reproductions.md). + ## Test Projects The primary test projects for Microsoft.Data.SqlClient are under diff --git a/doc/async-connectivity-reproductions.md b/doc/async-connectivity-reproductions.md new file mode 100644 index 0000000000..d42b461633 --- /dev/null +++ b/doc/async-connectivity-reproductions.md @@ -0,0 +1,86 @@ +# Async connectivity reproductions + +This focused suite retains five demonstrated async connectivity mechanisms +(worker starvation, transport cancellation, and transaction lock inversion), +plus the controls and test infrastructure needed to distinguish defects from +inadmissible setups. The related issue for each mechanism is linked below. +It changes no production behavior or public APIs. + +## Retained reproductions + +Test classes are under +`src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/`. + +| Mechanism | Desired-behavior test | Local baseline and essential controls | +|---|---|---| +| Async authentication starvation | `ConnectivityProcessTests.OpenAsync_TokenBurst_WithCappedWorkers_Completes` | Every V2 worker reaches held Login7. After Login7 release, zero workers are available and the token callback cannot enter. Generous-worker V2 and capped V1/non-pooled controls complete. Relevant to [#3459](https://github.com/dotnet/SqlClient/issues/3459). | +| Additional user-thread OpenAsync behind occupied workers | `ConnectivityProcessTests.OpenAsync_AdditionalOpen_WithCappedWorkers_MakesProgress` | A non-pool caller obtains a Task, but physical login to a separate, unheld endpoint cannot progress until the original Login7 replies release workers. The recovery control completes every open under the unchanged cap. This covers the starvation mechanism in [#3118](https://github.com/dotnet/SqlClient/issues/3118), not its exact TLS/error signature. | +| Synchronous provider-authentication worker starvation | `ConnectivityAuthenticationPressureTests.Open_ProviderAuthentication_WithCappedWorkerCallers_Completes` | Cold and genuinely expired fake managed-identity tokens stall both pools before provider entry with zero workers available. Increasing workers during cleanup completes the same callers. Warm-cache, generous-worker, dedicated-caller, and direct-async controls distinguish worker pressure from provider/cache failures. Relevant to [#2152](https://github.com/dotnet/SqlClient/issues/2152)/[#2470](https://github.com/dotnet/SqlClient/issues/2470), which primarily report synchronous Open. | +| Physical TCP attempts continue after public cancellation | `ConnectivityTransportCancellationTests.OpenAsync_RefusedConnect_AfterPublicCancellation_StartsNoNewAttempt` (.NET only; it observes managed-SNI traces) | Gate the first actual refused TCP connect, confirm public cancellation, then release physical work. Attempts grow from 1 to 13; the desired count remains 1. An uncancelled control proves the internal retry path is reachable with configurable retries disabled. Relevant to [#1619](https://github.com/dotnet/SqlClient/issues/1619). | +| Cancelled-scope rollback/physical-open lock inversion | `ConnectivityTransactionTests.OpenAsync_AmbientTransaction_CancelledAfterBegin_ScopeAndPhysicalWorkComplete` | Real BEGIN completes before a post-BEGIN gate. After public cancellation, rollback owns the connection monitor while physical opening owns the parser monitor. Releasing the gate produces `LOCK_CYCLE_STALLED`. Post-BEGIN, pre-enlistment cancellation, held-BEGIN, and actual sync/async commit/rollback controls establish phase ordering and wire validity. Relevant to [#4696](https://github.com/dotnet/SqlClient/issues/4696). | + +**Keep both `ConnectivityProcessTests` anchors.** Their worker caps are accepted +and verified, and an async-only yield workload first proves that the capped +runtime can schedule genuinely asynchronous work. The peer runs in the parent, +outside the capped pool. Caps cover thread-pool workers, not every managed thread. +Worker availability and pending-task state are diagnostics, not requirements that +a future fix must preserve exhaustion. + +The fake provider uses the real connection-string authentication-provider path, +but does not contact Azure. Expiry ages physical and shared cached tokens without +substituting pool clearing. The transaction peer supports only the local +BEGIN/COMMIT/ROLLBACK subset needed here, not DTC or a general SQL engine. +Request round-trip/truncation tests protect that new wire support. + +## Running + +Use the repository's SDK and build instructions in [BUILDGUIDE](../BUILDGUIDE.md). +The unit project builds and copies `tests/tools/ConnectivityTestHost` automatically. +Run healthy controls first: + +```bash +dotnet test src/Microsoft.Data.SqlClient/tests/UnitTests/Microsoft.Data.SqlClient.UnitTests.csproj \ + -f net10.0 --filter 'FullyQualifiedName~Connectivity&category!=failing' +``` + +The desired-behavior reproductions are deliberately `category=failing`, excluded +by the repository's normal filter. Select them explicitly to capture the current +defects; all eight cases are expected to fail on the current macOS baseline: + +```bash +dotnet test src/Microsoft.Data.SqlClient/tests/UnitTests/Microsoft.Data.SqlClient.UnitTests.csproj \ + -f net10.0 --filter 'FullyQualifiedName~Connectivity&category=failing' \ + --logger 'console;verbosity=normal' +``` + +To examine either anchor alone, replace the filter with +`FullyQualifiedName~OpenAsync_TokenBurst_WithCappedWorkers_Completes` or +`FullyQualifiedName~OpenAsync_AdditionalOpen_WithCappedWorkers_MakesProgress`. +Use `--list-tests` with the same filter to check selection. Once a driver fix +passes a desired-behavior test and its controls, remove that test's `failing` +trait and retain it as ordinary regression coverage. + +Every protocol read/phase/completion is bounded by an independent parent +watchdog. Missed setup phases, EOF, child faults, or rejected runtime limits are +test failures, not passing reproductions. Teardown releases gates and kills/reaps +only the owned child when it cannot recover. A lifecycle control explicitly +verifies missing-phase timeout and reaping. After a timed-out protocol read, +cleanup must not issue another read on that stream. + +## Evidence limits + +Local verification uses macOS/.NET 10, ephemeral loopback peers, and fake tokens: +no SQL Server, Azure credentials, certificates, or machine network changes. +Trace gates and reflection probe the current source's physical-open tasks, +authentication caches, transaction promoter, and monitors. Missing members or +trace boundaries fail explicitly; these probes are not a general released-package +comparison harness. Monitor ownership and captured thread states establish the +local lock cycle, but external opposing blocked stacks were not collected. +These mechanisms do not establish every customer's exact environment or cause. + +The historical [#2192](https://github.com/dotnet/SqlClient/issues/2192) defect was +addressed by [#2915](https://github.com/dotnet/SqlClient/pull/2915), starting with +6.0.0-preview2; it is not an outstanding reproduction target here. Windows UI, +native-SNI, real-service/network confirmations, broad correctness coverage, +benchmarks, and separate timeout-budget/pool-retirement investigations are +intentionally outside this focused change. diff --git a/eng/pipelines/perf/scripts/compare_perf.py b/eng/pipelines/perf/scripts/compare_perf.py index 609af7e8bd..70ae8d5b64 100644 --- a/eng/pipelines/perf/scripts/compare_perf.py +++ b/eng/pipelines/perf/scripts/compare_perf.py @@ -4,6 +4,7 @@ Reads every ``*-report-full.json`` under a baseline directory and a current directory, matches benchmarks by (Type, Method, Parameters), and computes the per-benchmark delta in mean execution time and allocated memory. +Custom diagnoser metrics are also compared by their stable descriptor IDs. Outputs: * a GitHub-flavoured markdown table (``--out-md``), and @@ -49,6 +50,18 @@ def _load_benchmarks(directory): continue memory = bench.get("Memory") or {} alloc = memory.get("BytesAllocatedPerOperation") + metrics = {} + for metric in bench.get("Metrics", []) or []: + descriptor = metric.get("Descriptor") or {} + metric_id = descriptor.get("Id") + value = metric.get("Value") + if metric_id is not None and value is not None: + metrics[metric_id] = { + "value": float(value), + "displayName": descriptor.get("DisplayName") or metric_id, + "unit": descriptor.get("Unit") or "", + "greaterIsBetter": descriptor.get("TheGreaterTheBetter"), + } key = f"{btype}.{method}({params})" records[key] = { @@ -57,6 +70,7 @@ def _load_benchmarks(directory): "parameterSignature": params, "meanNs": float(mean_ns), "allocatedBytes": float(alloc) if alloc is not None else None, + "metrics": metrics, } return records @@ -90,7 +104,25 @@ def build_comparison(baseline_dir, current_dir, threshold_pct): "currentMeanMs": (c["meanNs"] / NS_PER_MS) if c else None, "baselineAllocBytes": b["allocatedBytes"] if b else None, "currentAllocBytes": c["allocatedBytes"] if c else None, + "metrics": [], } + baseline_metrics = b["metrics"] if b else {} + current_metrics = c["metrics"] if c else {} + for metric_id in sorted(set(baseline_metrics) | set(current_metrics)): + bm = baseline_metrics.get(metric_id) + cm = current_metrics.get(metric_id) + metadata = cm or bm + base_value = bm["value"] if bm else None + current_value = cm["value"] if cm else None + entry["metrics"].append({ + "id": metric_id, + "displayName": metadata["displayName"], + "unit": metadata["unit"], + "greaterIsBetter": metadata["greaterIsBetter"], + "baselineValue": base_value, + "currentValue": current_value, + "deltaPct": _pct(base_value, current_value), + }) if b and c: entry["meanDeltaPct"] = _pct(b["meanNs"], c["meanNs"]) @@ -203,6 +235,28 @@ def render_markdown(entries, baseline_version, threshold_pct): ) ) lines.append("") + metric_entries = [(entry, metric) for entry in entries for metric in entry.get("metrics", [])] + if metric_entries: + lines.extend([ + "## Diagnoser metrics", + "", + "Values use the diagnoser's units and normalization; these deltas do not change " + "the execution-time regression gate.", + "", + "| Benchmark | Method | Params | Metric | Unit | Baseline | Current | Delta |", + "| --------- | ------ | ------ | ------ | ---- | -------- | ------- | ----- |", + ]) + for entry, metric in metric_entries: + baseline = metric["baselineValue"] + current = metric["currentValue"] + lines.append( + f"| {entry['benchmarkName']} | {entry['methodName']} | " + f"{entry['parameterSignature'] or '-'} | {metric['displayName']} | " + f"{metric['unit'] or '-'} | " + f"{f'{baseline:.4f}' if baseline is not None else '-'} | " + f"{f'{current:.4f}' if current is not None else '-'} | " + f"{_fmt_pct(metric['deltaPct'])} |") + lines.append("") return "\n".join(lines) diff --git a/eng/pipelines/perf/scripts/tests/test_compare_metrics.py b/eng/pipelines/perf/scripts/tests/test_compare_metrics.py new file mode 100644 index 0000000000..698a1b129f --- /dev/null +++ b/eng/pipelines/perf/scripts/tests/test_compare_metrics.py @@ -0,0 +1,90 @@ +# Licensed to the .NET Foundation under one or more agreements. +# The .NET Foundation licenses this file to you under the MIT license. +# See the LICENSE file in the project root for more information. + +"""Verify custom scorecard metrics survive baseline comparison and Kusto conversion.""" + +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +import compare_perf +import perf_to_kusto + + +class CompareMetricsTests(unittest.TestCase): + """Compare by stable IDs, preserving zero and missing values explicitly.""" + + def report(self, directory, metrics): + benchmark = { + "Type": "ConnectivityLoadRunner", + "Method": "OpenAsyncLoop", + "Parameters": "Concurrency=16, Pooling=False", + "Statistics": {"Mean": 1_000_000}, + "Metrics": [ + { + "Descriptor": { + "Id": metric_id, + "DisplayName": display_name, + "Legend": display_name, + "Unit": "workers", + "TheGreaterTheBetter": False, + }, + "Value": value, + } + for metric_id, display_name, value in metrics + ], + } + with open(os.path.join(directory, "load-report-full.json"), "w") as handle: + json.dump({"Benchmarks": [benchmark]}, handle) + return benchmark + + def test_metrics_match_by_id_not_display_name(self): + with tempfile.TemporaryDirectory() as baseline, tempfile.TemporaryDirectory() as current: + self.report(baseline, [("AverageOccupiedWorkers", "Old label", 8)]) + self.report(current, [("AverageOccupiedWorkers", "Avg workers", 2)]) + entries = compare_perf.build_comparison(baseline, current, 10) + self.assertEqual(entries[0]["status"], "unchanged") + metric = entries[0]["metrics"][0] + self.assertEqual(metric["baselineValue"], 8) + self.assertEqual(metric["currentValue"], 2) + self.assertEqual(metric["deltaPct"], -75) + markdown = compare_perf.render_markdown(entries, "baseline", 10) + self.assertIn("## Diagnoser metrics", markdown) + self.assertIn("Avg workers", markdown) + self.assertIn("-75.00%", markdown) + + def test_zero_and_one_sided_metrics_remain_visible(self): + with tempfile.TemporaryDirectory() as baseline, tempfile.TemporaryDirectory() as current: + self.report(baseline, [("zero", "Zero", 0), ("removed", "Removed", 4)]) + self.report(current, [("zero", "Zero", 2), ("new", "New", 3)]) + entries = compare_perf.build_comparison(baseline, current, 10) + metrics = {metric["id"]: metric for metric in entries[0]["metrics"]} + self.assertEqual(metrics["zero"]["baselineValue"], 0) + self.assertEqual(metrics["zero"]["currentValue"], 2) + self.assertIsNone(metrics["zero"]["deltaPct"]) + self.assertIsNone(metrics["new"]["baselineValue"]) + self.assertIsNone(metrics["removed"]["currentValue"]) + markdown = compare_perf.render_markdown(entries, "baseline", 10) + self.assertIn("| 0.0000 | 2.0000 | - |", markdown) + + def test_kusto_converter_preserves_scorecard_values(self): + with tempfile.TemporaryDirectory() as directory: + benchmark = self.report(directory, [("AverageOccupiedWorkers", "Avg workers", 0)]) + self.assertEqual(perf_to_kusto._driver_specific_metrics(benchmark)["Avg workers"], 0) + + def test_existing_reports_without_metrics_keep_timing_comparison(self): + with tempfile.TemporaryDirectory() as baseline, tempfile.TemporaryDirectory() as current: + self.report(baseline, []) + self.report(current, []) + entries = compare_perf.build_comparison(baseline, current, 10) + self.assertEqual(entries[0]["metrics"], []) + self.assertNotIn("## Diagnoser metrics", compare_perf.render_markdown(entries, "baseline", 10)) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/Microsoft.Data.SqlClient/tests/Common/ConnectivityProcess.cs b/src/Microsoft.Data.SqlClient/tests/Common/ConnectivityProcess.cs new file mode 100644 index 0000000000..3bdb4b0149 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/Common/ConnectivityProcess.cs @@ -0,0 +1,65 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Diagnostics; +using System.Globalization; + +namespace Microsoft.Data.SqlClient.Tests.Common; + +/// Owns launching, bounded protocol reads, and reaping hazardous connectivity clients. +public static class ConnectivityProcess +{ + /// Launches the copied helper against a fake loopback peer, never a shell or remote server. + public static Process Start(string scenario, int argument) + { + string directory = Path.Combine(AppContext.BaseDirectory, "ConnectivityTestHost"); +#if NETFRAMEWORK + string executable = Path.Combine(directory, "ConnectivityTestHost.exe"); + string arguments = $"{scenario} {argument.ToString(CultureInfo.InvariantCulture)}"; +#else + string executable = Environment.GetEnvironmentVariable("DOTNET_HOST_PATH") ?? "dotnet"; + string arguments = $"\"{Path.Combine(directory, "ConnectivityTestHost.dll")}\" {scenario} {argument.ToString(CultureInfo.InvariantCulture)}"; +#endif + var start = new ProcessStartInfo(executable, arguments) + { + WorkingDirectory = directory, + UseShellExecute = false, + RedirectStandardOutput = true, + RedirectStandardError = true, + RedirectStandardInput = true, + CreateNoWindow = true + }; + return Process.Start(start) ?? throw new InvalidOperationException("Failed to launch the connectivity test helper."); + } + + /// Fails on EOF or a missed watchdog; a stalled or missing phase is not a passing reproduction. + public static async Task ReadAsync(Process process, TimeSpan watchdog) + { + Task read = process.StandardOutput.ReadLineAsync(); + if (!ReferenceEquals(read, await Task.WhenAny(read, Task.Delay(watchdog)))) + { + _ = read.ContinueWith(failed => { _ = failed.Exception; }, CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + throw new TimeoutException("Connectivity child missed its expected phase/completion within the parent watchdog."); + } + return await read ?? throw new InvalidOperationException("Connectivity client exited before the expected phase."); + } + + /// Kills only the owned client if needed, then waits for it to exit before test teardown continues. + public static void Stop(Process process) + { + if (!process.HasExited) + { +#if NETFRAMEWORK + process.Kill(); +#else + process.Kill(entireProcessTree: true); +#endif + } + if (!process.WaitForExit(30_000)) + { + throw new TimeoutException("Connectivity child could not be reaped."); + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/BenchmarkRunners/ConnectivityLoadRunner.cs b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/BenchmarkRunners/ConnectivityLoadRunner.cs new file mode 100644 index 0000000000..15f0685adf --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/BenchmarkRunners/ConnectivityLoadRunner.cs @@ -0,0 +1,89 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using BenchmarkDotNet.Attributes; + +namespace Microsoft.Data.SqlClient.PerformanceTests +{ + /// Measures sustained OpenAsync/open-close throughput and worker occupancy against the configured SQL Server. + [Config(typeof(ConnectivityLoadConfig))] + public class ConnectivityLoadRunner : BaseRunner + { + [ParamsSource(nameof(ConcurrencyValues))] + public int Concurrency { get; set; } + + [ParamsSource(nameof(PoolingValues))] + public bool Pooling { get; set; } + + [ParamsSource(nameof(DurationValues))] + public int DurationSeconds { get; set; } + + [ParamsSource(nameof(SampleIntervalValues))] + public int SampleIntervalMilliseconds { get; set; } + + public IEnumerable ConcurrencyValues => LoadJob.Concurrency; + public IEnumerable PoolingValues => LoadJob.Pooling; + public IEnumerable DurationValues => + new[] { LoadJob.DurationSeconds }; + + public IEnumerable SampleIntervalValues => + new[] { LoadJob.SampleIntervalMilliseconds }; + + private static ConnectivityLoadJob LoadJob => s_config.Benchmarks.ConnectivityLoadRunnerConfig; + private string _connectionString; + + [GlobalSetup] + public void Setup() + { + if (Concurrency <= 0 || DurationSeconds <= 0 || SampleIntervalMilliseconds <= 0 + || SampleIntervalMilliseconds > DurationSeconds * 1000L) + { + throw new InvalidOperationException("Connectivity load requires positive concurrency, duration and a sampling interval within it."); + } + + var builder = new SqlConnectionStringBuilder(s_config.ConnectionString) + { + Pooling = Pooling, + MinPoolSize = 0, + MaxPoolSize = Math.Max(Concurrency, 200) + }; + if (builder.ConnectTimeout == 0) + { + throw new InvalidOperationException("Connectivity load requires a finite connection timeout."); + } + _connectionString = builder.ConnectionString; + } + + [IterationSetup] + public void ClearPool() + { + using var connection = new SqlConnection(_connectionString); + SqlConnection.ClearPool(connection); + } + + /// Runs concurrent awaited loops, including operation drain in the occupancy and throughput denominator. + [Benchmark] + public async Task OpenAsyncLoop() + { + ConnectivityLoadResult result = await ConnectivityLoad.RunAsync( + OpenAndCloseAsync, + Concurrency, + TimeSpan.FromSeconds(DurationSeconds), + TimeSpan.FromMilliseconds(SampleIntervalMilliseconds)); + ConnectivityLoadDiagnoser.Instance.Record(result); + } + + [GlobalCleanup] + public void Cleanup() => ClearPool(); + + private async Task OpenAndCloseAsync() + { + using var connection = new SqlConnection(_connectionString); + await connection.OpenAsync().ConfigureAwait(false); + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Config/Config.cs b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Config/Config.cs index 2fd7d4f9fd..6287f42ea2 100644 --- a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Config/Config.cs +++ b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Config/Config.cs @@ -74,6 +74,15 @@ public class Benchmarks public RunnerJob ConnectionPoolChurnRunnerConfig; public RunnerJob ConnectionPoolRampRunnerConfig; public RunnerJob ConnectionPoolThreadPoolPressureRunnerConfig; + public ConnectivityLoadJob ConnectivityLoadRunnerConfig; + } + + public class ConnectivityLoadJob : RunnerJob + { + public int[] Concurrency = new[] { 1, 16, 64 }; + public bool[] Pooling = new[] { false, true }; + public int DurationSeconds = 10; + public int SampleIntervalMilliseconds = 10; } public class RunnerJob diff --git a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Config/ConnectivityLoadDiagnoser.cs b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Config/ConnectivityLoadDiagnoser.cs new file mode 100644 index 0000000000..2e02f4ab7d --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Config/ConnectivityLoadDiagnoser.cs @@ -0,0 +1,121 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections.Generic; +using BenchmarkDotNet.Analysers; +using BenchmarkDotNet.Columns; +using BenchmarkDotNet.Configs; +using BenchmarkDotNet.Diagnosers; +using BenchmarkDotNet.Engines; +using BenchmarkDotNet.Exporters; +using BenchmarkDotNet.Loggers; +using BenchmarkDotNet.Reports; +using BenchmarkDotNet.Running; +using BenchmarkDotNet.Toolchains.InProcess.Emit; +using BenchmarkDotNet.Validators; + +namespace Microsoft.Data.SqlClient.PerformanceTests +{ + public sealed class ConnectivityLoadConfig : ManualConfig + { + public ConnectivityLoadConfig() => AddDiagnoser(ConnectivityLoadDiagnoser.Instance); + } + + internal sealed class ConnectivityLoadDiagnoser : IDiagnoser + { + internal static readonly ConnectivityLoadDiagnoser Instance = new(); + private readonly Dictionary _results = new(); + private ConnectivityLoadResult _current; + private bool _capturing; + + public IEnumerable Ids => new[] { nameof(ConnectivityLoadDiagnoser) }; + public IEnumerable Exporters => Array.Empty(); + public IEnumerable Analysers => Array.Empty(); + public RunMode GetRunMode(BenchmarkCase benchmarkCase) => RunMode.NoOverhead; + public void DisplayResults(ILogger logger) { } + + public void Handle(HostSignal signal, DiagnoserActionParameters parameters) + { + if (signal == HostSignal.BeforeActualRun) + { + _current = default; + _capturing = true; + } + else if (signal == HostSignal.AfterActualRun) + { + _capturing = false; + if (_current.SuccessfulOpens > 0) + { + BenchmarkCase benchmark = parameters.BenchmarkCase; + _results[benchmark] = _results.TryGetValue(benchmark, out ConnectivityLoadResult previous) + ? previous.Combine(_current) + : _current; + } + } + else if (signal == HostSignal.AfterAll) + { + _capturing = false; + } + } + + internal void Record(ConnectivityLoadResult result) + { + if (_capturing) + { + _current = _current.Combine(result); + } + } + + public IEnumerable ProcessResults(DiagnoserResults results) + { + if (!_results.TryGetValue(results.BenchmarkCase, out ConnectivityLoadResult result)) + { + throw new InvalidOperationException("No measured connectivity scorecard was captured."); + } + + yield return Create("AverageOccupiedWorkers", "Avg workers", "Average occupied thread-pool workers", "workers", result.AverageOccupiedWorkers); + yield return Create("WorkerMillisecondsPerOpen", "Worker ms/open", "Occupied worker milliseconds per successful open", "ms/open", result.WorkerMillisecondsPerOpen); + yield return Create("OpensPerSecond", "Opens/sec", "Successful opens per second", "opens/s", result.OpensPerSecond, greaterIsBetter: true); + yield return Create("PeakOccupiedWorkers", "Peak workers", "Peak occupied thread-pool workers", "workers", result.PeakOccupiedWorkers); + yield return Create("PeakPoolThreads", "Peak TP threads", "Peak existing thread-pool threads including idle threads", "threads", result.PeakPoolThreads); + yield return Create("PeakProcessThreads", "Peak process threads", "Peak process threads including dedicated threads and the observer", "threads", result.PeakProcessThreads); + yield return Create("CpuMillisecondsPerOpen", "CPU ms/open", "Process CPU milliseconds per successful open", "ms/open", result.CpuMillisecondsPerOpen); + yield return Create("SuccessfulOpens", "Opens", "Total successful opens in measured invocations", "opens", result.SuccessfulOpens, greaterIsBetter: true); + yield return Create("OccupancySamples", "Samples", "Total occupancy samples in measured invocations", "samples", result.Samples, greaterIsBetter: true); + yield return Create("MeasuredSeconds", "Measured sec", "Total measured seconds including in-flight operation drain", "s", result.ElapsedSeconds, greaterIsBetter: true); + } + + public IEnumerable Validate(ValidationParameters validationParameters) + { + foreach (BenchmarkCase benchmark in validationParameters.Benchmarks) + { + if (benchmark.Descriptor.Type == typeof(ConnectivityLoadRunner) + && benchmark.Job.Infrastructure.Toolchain is not InProcessEmitToolchain) + { + yield return new ValidationError( + true, "ConnectivityLoadDiagnoser requires the suite's in-process toolchain.", benchmark); + } + } + } + + private static Metric Create( + string id, string displayName, string legend, string unit, double value, bool greaterIsBetter = false) => + new(new Descriptor(id, displayName, legend, unit, greaterIsBetter), value); + + private sealed class Descriptor( + string id, string displayName, string legend, string unit, bool greaterIsBetter) : IMetricDescriptor + { + public string Id => id; + public string DisplayName => displayName; + public string Legend => legend; + public string NumberFormat => "0.000"; + public UnitType UnitType => UnitType.Dimensionless; + public string Unit => unit; + public bool TheGreaterTheBetter => greaterIsBetter; + public int PriorityInCategory => 0; + public bool GetIsAvailable(Metric metric) => true; + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/ConnectivityLoad.cs b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/ConnectivityLoad.cs new file mode 100644 index 0000000000..5b1f72ae4a --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/ConnectivityLoad.cs @@ -0,0 +1,217 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +#nullable enable + +using System; +using System.Diagnostics; +using System.Runtime.ExceptionServices; +using System.Threading; +using System.Threading.Tasks; + +namespace Microsoft.Data.SqlClient.PerformanceTests +{ + internal readonly record struct ConnectivityLoadResult( + double ElapsedSeconds, + double WorkerSeconds, + long SuccessfulOpens, + int PeakOccupiedWorkers, + int PeakPoolThreads, + int PeakProcessThreads, + long Samples, + double CpuSeconds) + { + public double AverageOccupiedWorkers => WorkerSeconds / ElapsedSeconds; + public double WorkerMillisecondsPerOpen => WorkerSeconds * 1000 / SuccessfulOpens; + public double OpensPerSecond => SuccessfulOpens / ElapsedSeconds; + public double CpuMillisecondsPerOpen => CpuSeconds * 1000 / SuccessfulOpens; + + /// Combines totals so different invocation lengths retain their correct weight. + public ConnectivityLoadResult Combine(ConnectivityLoadResult other) => new( + ElapsedSeconds + other.ElapsedSeconds, + WorkerSeconds + other.WorkerSeconds, + checked(SuccessfulOpens + other.SuccessfulOpens), + Math.Max(PeakOccupiedWorkers, other.PeakOccupiedWorkers), + Math.Max(PeakPoolThreads, other.PeakPoolThreads), + Math.Max(PeakProcessThreads, other.PeakProcessThreads), + checked(Samples + other.Samples), + CpuSeconds + other.CpuSeconds); + } + + /// Integrates occupied workers using actual sample timestamps, not the requested timer interval. + internal sealed class WorkerOccupancy + { + private long _firstTimestamp; + private long _lastTimestamp; + private int _previousOccupiedWorkers; + private double _workerTicks; + private int _peakOccupiedWorkers; + private int _peakPoolThreads; + private int _peakProcessThreads; + private long _samples; + + /// Adds a monotonic sample and trapezoidally integrates the interval since the previous sample. + public void Sample(long timestamp, int occupiedWorkers, int poolThreads, int processThreads = 0) + { + ArgumentOutOfRangeException.ThrowIfNegative(occupiedWorkers); + ArgumentOutOfRangeException.ThrowIfNegative(poolThreads); + ArgumentOutOfRangeException.ThrowIfNegative(processThreads); + if (_samples > 0 && timestamp < _lastTimestamp) + { + throw new ArgumentOutOfRangeException(nameof(timestamp), "Sample timestamps must be monotonic."); + } + + if (_samples == 0) + { + _firstTimestamp = timestamp; + } + else + { + _workerTicks += (_previousOccupiedWorkers + (double)occupiedWorkers) / 2 + * (timestamp - _lastTimestamp); + } + + _lastTimestamp = timestamp; + _previousOccupiedWorkers = occupiedWorkers; + _peakOccupiedWorkers = Math.Max(_peakOccupiedWorkers, occupiedWorkers); + _peakPoolThreads = Math.Max(_peakPoolThreads, poolThreads); + _peakProcessThreads = Math.Max(_peakProcessThreads, processThreads); + _samples++; + } + + /// Produces raw totals, rejecting an empty or unsuccessful measurement instead of dividing by zero. + public ConnectivityLoadResult Complete(long successfulOpens, double cpuSeconds) + { + if (_samples < 2 || _lastTimestamp <= _firstTimestamp || successfulOpens <= 0 || cpuSeconds < 0) + { + throw new InvalidOperationException("A scorecard requires elapsed time, samples and successful opens."); + } + + return new ConnectivityLoadResult( + (_lastTimestamp - _firstTimestamp) / (double)Stopwatch.Frequency, + _workerTicks / Stopwatch.Frequency, + successfulOpens, + _peakOccupiedWorkers, + _peakPoolThreads, + _peakProcessThreads, + _samples, + cpuSeconds); + } + } + + /// Runs awaited open/close loops and observes their process-wide worker occupancy from a dedicated thread. + internal static class ConnectivityLoad + { + /// Samples concurrent awaited operations until admissions stop and all in-flight operations finish. + public static async Task RunAsync( + Func openAndClose, + int concurrency, + TimeSpan duration, + TimeSpan sampleInterval) + { + ArgumentNullException.ThrowIfNull(openAndClose); + ArgumentOutOfRangeException.ThrowIfNegativeOrZero(concurrency); + if (duration <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException(nameof(duration)); + } + if (sampleInterval < TimeSpan.FromMilliseconds(1) || sampleInterval > duration) + { + throw new ArgumentOutOfRangeException(nameof(sampleInterval)); + } + + using var ready = new ManualResetEventSlim(); + using var stopObserver = new ManualResetEventSlim(); + using Process process = Process.GetCurrentProcess(); + var start = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var occupancy = new WorkerOccupancy(); + ExceptionDispatchInfo? observerError = null; + double cpuSeconds = 0; + int failed = 0; + + var observer = new Thread(() => + { + try + { + TimeSpan cpuStart = process.TotalProcessorTime; + Sample(); + ready.Set(); + while (!stopObserver.Wait(sampleInterval)) + { + Sample(); + } + Sample(); + cpuSeconds = (process.TotalProcessorTime - cpuStart).TotalSeconds; + } + catch (Exception error) + { + observerError = ExceptionDispatchInfo.Capture(error); + Interlocked.Exchange(ref failed, 1); + ready.Set(); + } + }) + { + IsBackground = true, + Name = "SqlClient connectivity occupancy observer" + }; + + observer.Start(); + long[] completed; + try + { + ready.Wait(); + observerError?.Throw(); + var workers = new Task[concurrency]; + for (int i = 0; i < workers.Length; i++) + { + workers[i] = Task.Run(async () => + { + long started = await start.Task.ConfigureAwait(false); + long successfulOpens = 0; + try + { + while (Volatile.Read(ref failed) == 0 && Stopwatch.GetElapsedTime(started) < duration) + { + await openAndClose().ConfigureAwait(false); + successfulOpens++; + } + return successfulOpens; + } + catch + { + // Stop admitting new opens, but drain every already-started operation. + Interlocked.Exchange(ref failed, 1); + throw; + } + }); + } + start.SetResult(Stopwatch.GetTimestamp()); + completed = await Task.WhenAll(workers).ConfigureAwait(false); + } + finally + { + stopObserver.Set(); + observer.Join(); + } + + observerError?.Throw(); + long totalOpens = 0; + foreach (long count in completed) + { + totalOpens = checked(totalOpens + count); + } + return occupancy.Complete(totalOpens, cpuSeconds); + + void Sample() + { + ThreadPool.GetMaxThreads(out int maximumWorkers, out _); + ThreadPool.GetAvailableThreads(out int availableWorkers, out _); + long timestamp = Stopwatch.GetTimestamp(); + process.Refresh(); + occupancy.Sample( + timestamp, maximumWorkers - availableWorkers, ThreadPool.ThreadCount, process.Threads.Count); + } + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Program.cs b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Program.cs index 5b69dfc8b0..e233ba7ba3 100644 --- a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Program.cs +++ b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/Program.cs @@ -58,6 +58,7 @@ public BenchmarkUnit(string name, Func selector, Type run new BenchmarkUnit("ConnectionPoolChurn", b => b.ConnectionPoolChurnRunnerConfig, typeof(ConnectionPoolChurnRunner)), new BenchmarkUnit("ConnectionPoolRamp", b => b.ConnectionPoolRampRunnerConfig, typeof(ConnectionPoolRampRunner)), new BenchmarkUnit("ConnectionPoolThreadPoolPressure", b => b.ConnectionPoolThreadPoolPressureRunnerConfig, typeof(ConnectionPoolThreadPoolPressureRunner)), + new BenchmarkUnit("ConnectivityLoad", b => b.ConnectivityLoadRunnerConfig, typeof(ConnectivityLoadRunner)), }; /// @@ -106,7 +107,13 @@ public Program() foreach (BenchmarkUnit unit in toRun) { - BenchmarkRunner.Run(unit.RunnerType, BenchmarkConfig.s_instance(unit.Selector(_config.Benchmarks))); + var summary = BenchmarkRunner.Run(unit.RunnerType, BenchmarkConfig.s_instance(unit.Selector(_config.Benchmarks))); + if (unit.RunnerType == typeof(ConnectivityLoadRunner) + && (summary.HasCriticalValidationErrors || !summary.Reports.Any() || summary.Reports.Any(report => !report.Success))) + { + Console.Error.WriteLine("Connectivity load failed; its scorecard cannot be used for comparison."); + Environment.ExitCode = 1; + } } // TODOs: diff --git a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/runnerconfig.jsonc b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/runnerconfig.jsonc index 083d001b11..71576f42a1 100644 --- a/src/Microsoft.Data.SqlClient/tests/PerformanceTests/runnerconfig.jsonc +++ b/src/Microsoft.Data.SqlClient/tests/PerformanceTests/runnerconfig.jsonc @@ -199,6 +199,21 @@ "InvocationCount": 1, "WarmupCount": 1, "RowCount": 0 + }, + // Opt-in sustained real-server load; no artificial server delays. + "ConnectivityLoadRunnerConfig": { + "Enabled": false, + "LaunchCount": 1, + "IterationCount": 5, + "InvocationCount": 1, + "WarmupCount": 1, + "RowCount": 0, + "RunStrategy": "Monitoring", + "Concurrency": [1, 16, 64], + "Pooling": [false, true], + "DurationSeconds": 10, + "SampleIntervalMilliseconds": 10, + "TimeoutMinutes": 10 } } } diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/Microsoft.Data.SqlClient.UnitTests.csproj b/src/Microsoft.Data.SqlClient/tests/UnitTests/Microsoft.Data.SqlClient.UnitTests.csproj index 135bb28eff..cd2386695f 100644 --- a/src/Microsoft.Data.SqlClient/tests/UnitTests/Microsoft.Data.SqlClient.UnitTests.csproj +++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/Microsoft.Data.SqlClient.UnitTests.csproj @@ -38,6 +38,7 @@ + @@ -112,6 +113,7 @@ + diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityAuthenticationPressureTests.cs b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityAuthenticationPressureTests.cs new file mode 100644 index 0000000000..6ae5cfb2ed --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityAuthenticationPressureTests.cs @@ -0,0 +1,140 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Diagnostics; +using System.Globalization; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient.Tests.Common; +using Microsoft.SqlServer.TDS.PreLogin; +using Microsoft.SqlServer.TDS.Servers; +using Xunit; +using Xunit.Abstractions; + +namespace Microsoft.Data.SqlClient.UnitTests.SimulatedServerTests; + +/// Reproduces #2152/#2470's synchronous worker callers, rather than only isolated dedicated-thread authentication. +[Collection(SimulatedServerTestCollection.Name)] +public class ConnectivityAuthenticationPressureTests +{ + private readonly ITestOutputHelper _output; + private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30); + + /// Records phase evidence without token values or connection strings. + public ConnectivityAuthenticationPressureTests(ITestOutputHelper output) => _output = output; + + /// Known defect: synchronous worker callers must not prevent cold/expired provider work from being scheduled. + [Theory] + [Trait("category", "failing")] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public Task Open_ProviderAuthentication_WithCappedWorkerCallers_Completes(bool poolV2, bool expired) => + Run($"auth-sync-{(expired ? "expired" : "cold")}-capped-{(poolV2 ? "v2" : "v1")}"); + + /// Generous workers and dedicated sync callers preserve provider progress; warm checkout needs no token work. + [Theory] + [InlineData("auth-sync-cold-generous-v1")] + [InlineData("auth-sync-expired-generous-v1")] + [InlineData("auth-sync-cold-generous-v2")] + [InlineData("auth-sync-expired-generous-v2")] + [InlineData("auth-dedicated-expired-capped-v1")] + [InlineData("auth-dedicated-expired-capped-v2")] + [InlineData("auth-sync-warm-capped-v1")] + [InlineData("auth-sync-warm-capped-v2")] + [InlineData("auth-async-expired-generous-v1")] + [InlineData("auth-async-expired-generous-v2")] + public Task Open_ProviderAuthentication_HealthyControls_Complete(string scenario) => Run(scenario); + + /// Observes real fed-auth traffic before testing scheduling and always frees child workers during recovery. + private async Task Run(string scenario) + { + bool expired = scenario.Contains("-expired-"); + bool warm = scenario.Contains("-warm-"); + using var server = new GatedConnectionTdsServer(TdsConnectionPhase.FederatedAuthenticationInfo, + new TdsServerArguments { FedAuthRequiredPreLoginOption = TdsPreLoginFedAuthRequiredOption.FedAuthRequired }); + server.Release(); + server.Start(); + using Process child = ConnectivityProcess.Start(scenario, server.EndPoint.Port); + Task errors = child.StandardError.ReadToEndAsync(); + bool releaseSent = false; + bool observing = false; + int width = 0; + try + { + string ready = await Read(child); + if (expired) + { + Assert.StartsWith("EXPIRED_PHYSICAL_AND_CACHE ", ready); + ready = await Read(child); + } + string[] parts = ready.Split(' '); + Assert.Equal("READY", parts[0]); + width = int.Parse(parts[1], CultureInfo.InvariantCulture); + Assert.Equal(scenario.Contains("-generous-") ? width * 4 : width, + int.Parse(parts[2], CultureInfo.InvariantCulture)); + int primed = warm || expired ? width : 0; + Assert.Equal(primed, server.Login7Count); + child.StandardInput.WriteLine("START"); + Assert.Equal($"CALLERS_STARTED {width}", await Read(child)); + if (!warm) + { + await Bounded(server.WaitForArrivalsAsync(primed + 1)); + } + child.StandardInput.WriteLine("OBSERVE"); + observing = true; + string tokenEntry = await Read(child); + Assert.StartsWith("AVAILABLE ", await Read(child)); + Assert.Equal(warm ? "WARM_CALLBACKS 0" : "TOKEN_ENTERED", tokenEntry); + child.StandardInput.WriteLine("RELEASE"); + releaseSent = true; + string complete = await Read(child); + Assert.StartsWith($"COMPLETE {width} CALLBACKS ", complete); + int callbacks = int.Parse(complete.Substring(complete.LastIndexOf(' ') + 1), CultureInfo.InvariantCulture); + Assert.InRange(callbacks, warm ? 0 : 1, warm ? 0 : width); + Assert.Equal(warm ? width : primed + width, server.Login7Count); + await Bounded(Task.Factory.StartNew(child.WaitForExit, CancellationToken.None, + TaskCreationOptions.LongRunning, TaskScheduler.Default)); + Assert.Equal(0, child.ExitCode); + Assert.Equal(string.Empty, await errors); + } + finally + { + try + { + if (observing && !releaseSent && !child.HasExited) + { + child.StandardInput.WriteLine("RECOVER"); + Assert.Equal("WORKERS_RELEASED", await Read(child)); + Assert.StartsWith($"COMPLETE {width} CALLBACKS ", await Read(child)); + await Bounded(Task.Factory.StartNew(child.WaitForExit, CancellationToken.None, + TaskCreationOptions.LongRunning, TaskScheduler.Default)); + Assert.Equal(0, child.ExitCode); + } + } + finally + { + ConnectivityProcess.Stop(child); + _output.WriteLine(await errors); + } + } + } + + /// Protocol reads are bounded by the independent parent, not the exhausted child's workers. + private async Task Read(Process child) + { + string line = await ConnectivityProcess.ReadAsync(child, Watchdog); + _output.WriteLine(line); + return line; + } + + /// Bounds parent-side phase observation and surfaces missing phases. + private static async Task Bounded(Task task) + { + Assert.Same(task, await Task.WhenAny(task, Task.Delay(Watchdog))); + await task; + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityLoadTests.cs b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityLoadTests.cs new file mode 100644 index 0000000000..9b8a045977 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityLoadTests.cs @@ -0,0 +1,67 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +#if NET +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient.PerformanceTests; +using Microsoft.SqlServer.TDS.Servers; +using Xunit; + +namespace Microsoft.Data.SqlClient.UnitTests.SimulatedServerTests; + +/// Checks the sustained workload's successful-open accounting against ordinary, ungated TDS logins. +[Collection(SimulatedServerTestCollection.Name)] +public class ConnectivityLoadTests +{ + /// Non-pooled operations each perform a physical login; pooled operations may reuse connections. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task OpenAsyncLoop_CountsCompletedOpens(bool pooling) + { + using var server = new TdsServer(); + server.Start(); + var builder = new SqlConnectionStringBuilder + { + DataSource = $"127.0.0.1,{server.EndPoint.Port}", + Encrypt = SqlConnectionEncryptOption.Optional, + Pooling = pooling, + Enlist = false, + ConnectTimeout = 10 + }; + string connectionString = builder.ConnectionString; + long completed = 0; + try + { + ConnectivityLoadResult result = await ConnectivityLoad.RunAsync( + async () => + { + using var connection = new SqlConnection(connectionString); + await connection.OpenAsync(); + Interlocked.Increment(ref completed); + }, + 2, TimeSpan.FromMilliseconds(250), TimeSpan.FromMilliseconds(10)); + + Assert.Equal(completed, result.SuccessfulOpens); + Assert.True(completed > 0); + Assert.True(server.Login7Count > 0); + if (pooling) + { + Assert.True(server.Login7Count <= completed); + } + else + { + Assert.Equal(completed, server.Login7Count); + } + } + finally + { + using var connection = new SqlConnection(connectionString); + SqlConnection.ClearPool(connection); + } + } +} +#endif diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityProcessTests.cs b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityProcessTests.cs new file mode 100644 index 0000000000..10b9f78555 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityProcessTests.cs @@ -0,0 +1,232 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient.Tests.Common; +using Microsoft.SqlServer.TDS.PreLogin; +using Microsoft.SqlServer.TDS.Servers; +using Xunit; +using Xunit.Abstractions; + +namespace Microsoft.Data.SqlClient.UnitTests.SimulatedServerTests; + +/// Parent-owned isolation for the two retained worker-starvation reproductions. +[Collection(SimulatedServerTestCollection.Name)] +public class ConnectivityProcessTests +{ + private readonly ITestOutputHelper _output; + private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30); + + /// Routes child output into the existing xUnit result stream. + public ConnectivityProcessTests(ITestOutputHelper output) => _output = output; + + /// A ready child deliberately receives no command: a missed protocol phase fails and cleanup reaps it. + [Fact] + public async Task Process_MissingProtocolPhase_TimesOutAndIsReaped() + { + using Process child = Start("protocol-idle", 0); + Task errors = child.StandardError.ReadToEndAsync(); + try + { + Assert.Equal("READY", await Read(child)); + await Assert.ThrowsAsync(() => + ConnectivityProcess.ReadAsync(child, TimeSpan.FromSeconds(1))); + } + finally + { + Stop(child); + Assert.True(child.HasExited); + Assert.Equal(string.Empty, await errors); + } + } + + /// The same cold token burst completes when enough workers remain for token tasks. + [Theory] + [InlineData(true, true)] + [InlineData(false, true)] + [InlineData(false, false)] + public Task OpenAsync_TokenBurst_WithSufficientWorkers_Completes(bool poolV2, bool pooling) => + TokenBurst(false, poolV2, pooling); + + /// Dedicated V1/non-pooled creation is compared with V2 under the same accepted cap. + [Theory] + [InlineData(false, true)] + [InlineData(false, false)] + public Task OpenAsync_TokenBurst_WithCappedDedicatedCreation_Completes(bool poolV2, bool pooling) => + TokenBurst(true, poolV2, pooling); + + /// + /// Known #3459 repro: all V2 creation workers reach Login7 before authentication is released. + /// Keep explicitly selected until physical open/token acquisition no longer blocks workers. + /// + [Fact] + [Trait("category", "failing")] + public Task OpenAsync_TokenBurst_WithCappedWorkers_Completes() => TokenBurst(true, true, true); + + /// Covers the #3459/#3118 worker-starvation mechanism: an independent open must progress while earlier logins hold every V2 worker. + [Fact] + [Trait("category", "failing")] + public Task OpenAsync_AdditionalOpen_WithCappedWorkers_MakesProgress() => AdditionalOpen(false); + + /// Releasing held plain logins frees workers and permits the additional call to finish without changing the cap. + [Fact] + public Task OpenAsync_AdditionalOpen_AfterHeldLoginsReleased_Completes() => AdditionalOpen(true); + + /// Separates invocation return from wire progress; a second unheld endpoint excludes pool capacity and server gating. + private async Task AdditionalOpen(bool releaseHeldLogins) + { + using var held = new GatedConnectionTdsServer(TdsConnectionPhase.Login7); + using var additional = new GatedConnectionTdsServer(TdsConnectionPhase.Login7); + additional.Release(); + held.Start(); + additional.Start(); + using Process child = Start("workers-additional-v2", held.EndPoint.Port); + Task errors = child.StandardError.ReadToEndAsync(); + Task additionalArrival = additional.WaitForArrivalsAsync(1); + bool callReturned = false; + bool additionalCompleted = false; + int width = 0; + try + { + string[] ready = (await Read(child)).Split(' '); + Assert.Equal("READY", ready[0]); + width = int.Parse(ready[1], System.Globalization.CultureInfo.InvariantCulture); + Assert.Equal(width.ToString(System.Globalization.CultureInfo.InvariantCulture), ready[2]); + Assert.Equal(width.ToString(System.Globalization.CultureInfo.InvariantCulture), ready[3]); + Assert.Equal("CALLS_RETURNED", await Read(child)); + await Bounded(held.WaitForArrivalsAsync(width)); + child.StandardInput.WriteLine("PROBE"); + string occupancy = await Read(child); + Assert.StartsWith("AVAILABLE ", occupancy); + Assert.InRange(int.Parse(occupancy.Split(' ')[1], System.Globalization.CultureInfo.InvariantCulture), 0, width); + + child.StandardInput.WriteLine($"OPEN {additional.EndPoint.Port}"); + Assert.Equal("ADDITIONAL_CALL_ENTERED", await Read(child)); + Assert.Equal("ADDITIONAL_CALL_RETURNED", await Read(child)); + callReturned = true; + string pending = await Read(child); + Assert.StartsWith("ADDITIONAL_TASK_PENDING ", pending); + Assert.True(bool.TryParse(pending.Split(' ')[1], out _)); + _output.WriteLine($"ADDITIONAL_LOGIN_ARRIVALS_BEFORE_RELEASE {additional.ArrivalCount}"); + if (releaseHeldLogins) + { + _output.WriteLine("RELEASING_HELD_LOGINS"); + held.Release(); + } + await Bounded(additionalArrival); + Assert.Equal("ADDITIONAL_COMPLETE", await Read(child)); + additionalCompleted = true; + } + finally + { + held.Release(); + _output.WriteLine("HELD_LOGINS_RELEASED_DURING_CLEANUP"); + try + { + if (callReturned) + { + await Bounded(additionalArrival); + if (!additionalCompleted) + { + Assert.Equal("ADDITIONAL_COMPLETE", await Read(child)); + } + Assert.Equal($"COMPLETE {width + 1}", await Read(child)); + Assert.Equal(width, held.ArrivalCount); + Assert.Equal(1, additional.ArrivalCount); + await Bounded(Task.Factory.StartNew(child.WaitForExit, CancellationToken.None, + TaskCreationOptions.LongRunning, TaskScheduler.Default)); + Assert.Equal(0, child.ExitCode); + Assert.Equal(string.Empty, await errors); + } + } + finally + { + Stop(child); + _output.WriteLine(await errors); + } + } + } + + /// Runs a gated cold burst, checking accepted limits and actual occupied workers. + private async Task TokenBurst(bool capped, bool poolV2, bool pooling) + { + using var server = new GatedConnectionTdsServer(TdsConnectionPhase.Login7, new TdsServerArguments + { + FedAuthRequiredPreLoginOption = TdsPreLoginFedAuthRequiredOption.FedAuthRequired + }); + server.Start(); + string mode = pooling ? poolV2 ? "v2" : "v1" : "nonpooled"; + using Process child = Start($"workers-{(capped ? "capped" : "generous")}-{mode}", server.EndPoint.Port); + Task errors = child.StandardError.ReadToEndAsync(); + try + { + string[] ready = (await Read(child)).Split(' '); + Assert.Equal("READY", ready[0]); + int width = int.Parse(ready[1], System.Globalization.CultureInfo.InvariantCulture); + Assert.Equal(width.ToString(System.Globalization.CultureInfo.InvariantCulture), ready[2]); + Assert.Equal(capped ? width : width * 4, int.Parse(ready[3], System.Globalization.CultureInfo.InvariantCulture)); + Assert.Equal("CALLS_RETURNED", await Read(child)); + await Bounded(server.WaitForArrivalsAsync(poolV2 && pooling ? width : 1)); + child.StandardInput.WriteLine("PROBE"); + string occupancy = await Read(child); + Assert.StartsWith("AVAILABLE ", occupancy); + Assert.InRange(int.Parse(occupancy.Split(' ')[1], System.Globalization.CultureInfo.InvariantCulture), + 0, capped ? width : width * 4); + server.Release(); + Assert.Equal("TOKEN_ENTERED", await Read(child)); + child.StandardInput.WriteLine("RELEASE"); + int tokenCalls = 1; + string progress; + while ((progress = await Read(child)) == "TOKEN_ENTERED") + { + tokenCalls++; + } + Assert.InRange(tokenCalls, 1, width); + Assert.Equal($"COMPLETE {width}", progress); + Assert.Equal(width, server.ArrivalCount); + await Bounded(Task.Factory.StartNew(child.WaitForExit, CancellationToken.None, + TaskCreationOptions.LongRunning, TaskScheduler.Default)); + Assert.Equal(0, child.ExitCode); + Assert.Equal(string.Empty, await errors); + } + finally + { + Stop(child); + server.Release(); + _output.WriteLine(await errors); + } + } + + /// Launches only the copied test helper, never a shell or a server on a remote interface. + private static Process Start(string scenario, int port) + => ConnectivityProcess.Start(scenario, port); + + /// Reads bounded protocol lines and fails on EOF; missing phases are never a passing repro. + private async Task Read(Process process) + { + string line = await ConnectivityProcess.ReadAsync(process, Watchdog); + _output.WriteLine(line); + return line; + } + + /// Uses a parent watchdog which is independent of the capped/deadlocked client. + private static async Task Bounded(Task task) + { + Task winner = await Task.WhenAny(task, Task.Delay(Watchdog)); + if (!ReferenceEquals(task, winner)) + { + _ = task.ContinueWith(failed => { _ = failed.Exception; }, CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + } + Assert.True(ReferenceEquals(task, winner), + "Connectivity client missed the expected phase/completion within the parent watchdog."); + await task; + } + + /// Reaps the owned process on success or failure before stopping the separately hosted server. + private static void Stop(Process process) => ConnectivityProcess.Stop(process); +} diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityTransactionTests.cs b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityTransactionTests.cs new file mode 100644 index 0000000000..34b925fa1d --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityTransactionTests.cs @@ -0,0 +1,216 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.IO; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using System.Transactions; +using Microsoft.SqlServer.TDS.Servers; +using Microsoft.Data.SqlClient.Tests.Common; +using Microsoft.SqlServer.TDS.TransactionManager; +using Xunit; +using Xunit.Abstractions; + +namespace Microsoft.Data.SqlClient.UnitTests.SimulatedServerTests; + +/// Validates the real transaction wire boundary needed for #4696 instead of substituting a Login7 delay. +[Collection(SimulatedServerTestCollection.Name)] +public class ConnectivityTransactionTests +{ + private readonly ITestOutputHelper _output; + public ConnectivityTransactionTests(ITestOutputHelper output) => _output = output; + + /// Releasing the actual post-BEGIN gate allows ordinary scope and physical-open completion. + [Fact] + public Task OpenAsync_AmbientTransaction_PostBeginGate_HealthyControl_Completes() => GatedEnlistment(false); + + /// Known defect #4696: actual rollback holds the connection monitor against login's proven parser lock. + [Fact] + [Trait("category", "failing")] + public Task OpenAsync_AmbientTransaction_CancelledAfterBegin_ScopeAndPhysicalWorkComplete() => GatedEnlistment(true); + + /// Before BEGIN, public cancellation unwinds the scope even while physical login remains held. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task OpenAsync_AmbientTransaction_CancelledBeforeEnlistment_DisposesScope(bool enlist) + { + using var server = new GatedConnectionTdsServer(TdsConnectionPhase.Login7); + server.Start(); + using Process child = ConnectivityProcess.Start(enlist ? "enlist-before-cancel" : "enlist-disabled-cancel", server.EndPoint.Port); + Task errors = child.StandardError.ReadToEndAsync(); + try + { + Assert.Equal("CALL_RETURNED", await Read(child)); + Task arrival = server.WaitForArrivalsAsync(1); + Assert.Same(arrival, await Task.WhenAny(arrival, Task.Delay(TimeSpan.FromSeconds(15)))); + await arrival; + child.StandardInput.WriteLine("CANCEL"); + Assert.Equal("PUBLIC_CANCELLED", await Read(child)); + Assert.Equal("SCOPE_COMPLETE", await Read(child)); + server.Release(); + child.StandardInput.WriteLine("FINISH"); + Assert.Equal("SCOPE_AND_PHYSICAL_COMPLETE", await Read(child)); + Assert.True(child.WaitForExit(30_000), "Pre-enlistment child did not exit."); + Assert.Equal(0, child.ExitCode); + Assert.Equal(string.Empty, await errors); + Assert.Equal(1, server.ArrivalCount); + } + finally + { + server.Release(); + ConnectivityProcess.Stop(child); + _output.WriteLine(await errors); + } + } + + private async Task GatedEnlistment(bool cancel) + { + using var server = new GatedTransactionTdsServer(); + server.Start(); + using Process child = ConnectivityProcess.Start(cancel ? "enlist-cancel" : "enlist-control", server.EndPoint.Port); + Task errors = child.StandardError.ReadToEndAsync(); + try + { + Assert.Equal("CALL_RETURNED", await Read(child)); + Assert.Same(server.BeginArrived, await Task.WhenAny(server.BeginArrived, Task.Delay(TimeSpan.FromSeconds(15)))); + await server.BeginArrived; + Assert.Equal(1, server.BeginCount); + server.ReleaseBegin(); + Assert.StartsWith("ENLISTED_PARSER_HELD ", await Read(child)); + child.StandardInput.WriteLine(cancel ? "CANCEL" : "FINISH"); + if (cancel) + { + Assert.Equal("PUBLIC_CANCELLED", await Read(child)); + Assert.Contains(await Read(child), new[] { "ROLLBACK_OWNS_CONNECTION", "ROLLBACK_RELEASED_CONNECTION" }); + child.StandardInput.WriteLine("FINISH"); + } + string completion = await Read(child); + Assert.Equal("SCOPE_AND_PHYSICAL_COMPLETE", completion); + Assert.True(child.WaitForExit(30_000), "Enlistment child did not exit."); + Assert.Equal(0, child.ExitCode); + Assert.Equal(string.Empty, await errors); + Assert.Equal(cancel ? 0 : 1, server.CommitCount); + } + finally + { + server.ReleaseBegin(); + ConnectivityProcess.Stop(child); + _output.WriteLine(await errors); + } + } + + private async Task Read(Process child) + { + string line = await ConnectivityProcess.ReadAsync(child, TimeSpan.FromSeconds(30)); + _output.WriteLine(line); + return line; + } + + /// Local transaction requests preserve descriptor and operation, with strict truncation handling. + [Theory] + [InlineData(TDSTransactionManagerRequest.Begin)] + [InlineData(TDSTransactionManagerRequest.Commit)] + [InlineData(TDSTransactionManagerRequest.Rollback)] + public void TransactionManager_Request_RoundTripsAndRejectsTruncation(TDSTransactionManagerRequest request) + { + var original = new TDSTransactionManagerToken { Request = request, Descriptor = 42, IsolationLevel = 2 }; + using var stream = new MemoryStream(); + original.Deflate(stream); + stream.Position = 0; + var parsed = new TDSTransactionManagerToken(); + Assert.True(parsed.Inflate(stream)); + Assert.Equal(original.Request, parsed.Request); + Assert.Equal(original.Descriptor, parsed.Descriptor); + Assert.Equal(original.OutstandingRequests, parsed.OutstandingRequests); + stream.SetLength(stream.Length - 1); + stream.Position = 0; + Assert.Throws(() => parsed.Inflate(stream)); + } + + /// Both APIs genuinely enlist via BEGIN and finish with COMMIT or ROLLBACK, with and without pooling. + [Theory] + [InlineData(false, false, false)] + [InlineData(false, false, true)] + [InlineData(false, true, false)] + [InlineData(false, true, true)] + [InlineData(true, false, false)] + [InlineData(true, false, true)] + [InlineData(true, true, false)] + [InlineData(true, true, true)] + public async Task Open_AmbientTransaction_UsesActualBeginAndCompletion(bool async, bool pooling, bool complete) + { + using var server = new GatedTransactionTdsServer(); + server.ReleaseBegin(); + server.Start(); + using var connection = new SqlConnection(ConnectionString(server, pooling)); + try + { + using (var scope = new TransactionScope(TransactionScopeOption.RequiresNew, + new TransactionOptions { IsolationLevel = IsolationLevel.ReadCommitted }, TransactionScopeAsyncFlowOption.Enabled)) + { + if (async) + { + await connection.OpenAsync(); + } + else + { + connection.Open(); + } + Assert.Equal(1, server.BeginCount); + if (complete) + { + scope.Complete(); + } + } + Assert.Equal(complete ? 1 : 0, server.CommitCount); + Assert.Equal(complete ? 0 : 1, server.RollbackCount); + } + finally + { + connection.Close(); + SqlConnection.ClearPool(connection); + } + } + + /// The gate observes transaction BEGIN only after login; releasing it lets the actual ambient open finish. + [Fact] + public async Task OpenAsync_AmbientTransaction_HeldBegin_CompletesAfterRelease() + { + using var server = new GatedTransactionTdsServer(); + server.Start(); + using var connection = new SqlConnection(ConnectionString(server, false)); + using var scope = new TransactionScope(TransactionScopeOption.RequiresNew, + new TransactionOptions { IsolationLevel = IsolationLevel.ReadCommitted }, TransactionScopeAsyncFlowOption.Enabled); + Task open = connection.OpenAsync(); + try + { + Assert.Same(server.BeginArrived, await Task.WhenAny(server.BeginArrived, Task.Delay(TimeSpan.FromSeconds(30)))); + await server.BeginArrived; + Assert.False(open.IsCompleted); + server.ReleaseBegin(); + Assert.Same(open, await Task.WhenAny(open, Task.Delay(TimeSpan.FromSeconds(30)))); + await open; + scope.Complete(); + } + finally + { + server.ReleaseBegin(); + await open; + } + } + + /// Uses only a fake local peer; ambient enlistment remains explicitly enabled. + internal static string ConnectionString(GatedTransactionTdsServer server, bool pooling) => new SqlConnectionStringBuilder + { + DataSource = $"127.0.0.1,{server.EndPoint.Port}", + Pooling = pooling, + Enlist = true, + Encrypt = SqlConnectionEncryptOption.Optional, + ConnectRetryCount = 0, + ConnectTimeout = 30 + }.ConnectionString; +} diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityTransportCancellationTests.cs b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityTransportCancellationTests.cs new file mode 100644 index 0000000000..ec8d77596e --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/SimulatedServerTests/ConnectivityTransportCancellationTests.cs @@ -0,0 +1,101 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +// The attempt counter observes managed-SNI SniTcpHandle traces; .NET Framework always uses native SNI. +#if NET +using System; +using System.Diagnostics; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient.Tests.Common; +using Xunit; +using Xunit.Abstractions; + +namespace Microsoft.Data.SqlClient.UnitTests.SimulatedServerTests; + +/// Counts real TCP attempts after public cancellation, rather than equating a cancelled task with stopped physical work (#1619). +[Collection(SimulatedServerTestCollection.Name)] +public class ConnectivityTransportCancellationTests +{ + private readonly ITestOutputHelper _output; + private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30); + + /// Records physical counts at the cancellation boundary and at actual creation completion. + public ConnectivityTransportCancellationTests(ITestOutputHelper output) => _output = output; + + /// Proves that refused TCP connects reach internal retries with configured retries disabled. + [Fact] + public Task OpenAsync_RefusedConnect_InternalRetryControl_ReachesMultipleAttempts() => Run(false); + + /// Known defect: no new attempt for a cancelled caller should start after its held first connect failure. + [Fact] + [Trait("category", "failing")] + public Task OpenAsync_RefusedConnect_AfterPublicCancellation_StartsNoNewAttempt() => Run(true); + + /// Only releases the first failed transport after public cancellation has been confirmed. + private async Task Run(bool cancel) + { + using var endpoint = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + endpoint.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + int port = ((IPEndPoint)endpoint.LocalEndPoint!).Port; + // A bound, non-listening socket can silently drop SYNs on macOS; close it to produce refusal. + endpoint.Dispose(); + using Process child = ConnectivityProcess.Start(cancel ? "attempt-cancel" : "attempt-control", port); + Task errors = child.StandardError.ReadToEndAsync(); + try + { + Assert.Equal("CALL_RETURNED", await Read(child)); + Assert.Equal("FAILURE_HELD ATTEMPTS 1", await Read(child)); + child.StandardInput.WriteLine(cancel ? "CANCEL" : "CONTINUE"); + Assert.Equal(cancel ? "PUBLIC_CANCELLED" : "RETRY_ALLOWED", await Read(child)); + const int before = 1; + _output.WriteLine($"ATTEMPTS_AT_PUBLIC_BOUNDARY {before}"); + child.StandardInput.WriteLine("FINISH"); + if (!cancel) + { + Assert.StartsWith("NETWORK_FAILURE ", await Read(child)); + } + string completion = await Read(child); + Assert.StartsWith("PHYSICAL_COMPLETE ATTEMPTS ", completion); + int attempts = int.Parse(completion.Substring(completion.LastIndexOf(' ') + 1), + System.Globalization.CultureInfo.InvariantCulture); + await Bounded(Task.Factory.StartNew(child.WaitForExit, CancellationToken.None, + TaskCreationOptions.LongRunning, TaskScheduler.Default)); + Assert.Equal(0, child.ExitCode); + Assert.Equal(string.Empty, await errors); + _output.WriteLine($"ATTEMPTS_AT_PHYSICAL_COMPLETION {attempts}"); + if (cancel) + { + Assert.Equal(before, attempts); + } + else + { + Assert.True(attempts >= 2, "Transport faults did not reach the intended internal retry path."); + } + } + finally + { + ConnectivityProcess.Stop(child); + _output.WriteLine(await errors); + } + } + + /// Bounds child protocol reads independently of the physical connection operation. + private async Task Read(Process child) + { + string line = await ConnectivityProcess.ReadAsync(child, Watchdog); + _output.WriteLine(line); + return line; + } + + /// Fails on a missed phase; transport-loop failures are observed rather than swallowed. + private static async Task Bounded(Task task) + { + Assert.Same(task, await Task.WhenAny(task, Task.Delay(Watchdog))); + await task; + } +} +#endif diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/WorkerOccupancyTests.cs b/src/Microsoft.Data.SqlClient/tests/UnitTests/WorkerOccupancyTests.cs new file mode 100644 index 0000000000..9b3c6be234 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/WorkerOccupancyTests.cs @@ -0,0 +1,178 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +#if NET +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient.PerformanceTests; +using Xunit; + +namespace Microsoft.Data.SqlClient.UnitTests; + +/// Serializes process-wide thread measurements and temporary worker-floor changes. +[CollectionDefinition("Connectivity scorecard", DisableParallelization = true)] +public class ConnectivityScorecardCollection { } + +/// Verifies the connectivity scorecard's time weighting, normalization and failure propagation without SQL Server. +[Collection("Connectivity scorecard")] +public class WorkerOccupancyTests +{ + /// Uneven sample intervals are weighted by actual elapsed time rather than sample count. + [Fact] + public void Samples_UseElapsedTimeAndNormalizePerOpen() + { + var occupancy = new WorkerOccupancy(); + occupancy.Sample(0, 2, 4, 10); + occupancy.Sample(Stopwatch.Frequency, 2, 5, 12); + occupancy.Sample(Stopwatch.Frequency * 4, 6, 8, 11); + ConnectivityLoadResult result = occupancy.Complete(10, 0.5); + + Assert.Equal(4, result.ElapsedSeconds); + Assert.Equal(14, result.WorkerSeconds); + Assert.Equal(3.5, result.AverageOccupiedWorkers); + Assert.Equal(1400, result.WorkerMillisecondsPerOpen); + Assert.Equal(2.5, result.OpensPerSecond); + Assert.Equal(50, result.CpuMillisecondsPerOpen); + Assert.Equal(6, result.PeakOccupiedWorkers); + Assert.Equal(8, result.PeakPoolThreads); + Assert.Equal(12, result.PeakProcessThreads); + Assert.Equal(3, result.Samples); + } + + /// Combining invocations normalizes their totals instead of averaging differently-sized rates. + [Fact] + public void Invocations_CombineWeightedTotals() + { + var first = new ConnectivityLoadResult(1, 2, 10, 2, 4, 10, 3, 0.1); + var second = new ConnectivityLoadResult(3, 18, 30, 6, 8, 12, 7, 0.3); + ConnectivityLoadResult combined = first.Combine(second); + + Assert.Equal(5, combined.AverageOccupiedWorkers); + Assert.Equal(500, combined.WorkerMillisecondsPerOpen); + Assert.Equal(10, combined.OpensPerSecond); + Assert.Equal(10, combined.CpuMillisecondsPerOpen); + Assert.Equal(6, combined.PeakOccupiedWorkers); + Assert.Equal(8, combined.PeakPoolThreads); + Assert.Equal(12, combined.PeakProcessThreads); + Assert.Equal(10, combined.Samples); + } + + /// Missing observations, nonmonotonic timestamps and zero successful opens cannot produce success-shaped metrics. + [Fact] + public void InvalidMeasurements_FailExplicitly() + { + var occupancy = new WorkerOccupancy(); + Assert.Throws(() => occupancy.Complete(1, 0)); + Assert.Throws(() => occupancy.Sample(0, -1, 0)); + occupancy.Sample(10, 1, 1); + Assert.Throws(() => occupancy.Sample(9, 1, 1)); + occupancy.Sample(20, 1, 1); + Assert.Throws(() => occupancy.Complete(0, 0)); + } + + /// Both truly asynchronous waits and synchronous worker waits complete and are included in the sampling window. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Load_AwaitsEveryOperationAndStopsObserver(bool blocking) + { + long completed = 0; + ConnectivityLoadResult result = await ConnectivityLoad.RunAsync( + async () => + { + if (blocking) + { + Thread.Sleep(10); + } + else + { + await Task.Delay(10); + } + Interlocked.Increment(ref completed); + }, + 4, TimeSpan.FromMilliseconds(100), TimeSpan.FromMilliseconds(5)); + + Assert.Equal(Interlocked.Read(ref completed), result.SuccessfulOpens); + Assert.True(result.SuccessfulOpens > 0); + Assert.True(result.ElapsedSeconds >= 0.1); + Assert.True(result.Samples >= 2); + Assert.True(double.IsFinite(result.AverageOccupiedWorkers)); + Assert.True(double.IsFinite(result.WorkerMillisecondsPerOpen)); + } + + /// An open failure invalidates the invocation and does not become a reduced-throughput success. + [Fact] + public async Task Load_OpenFailurePropagates() + { + var error = new InvalidOperationException("open failed"); + Exception actual = await Assert.ThrowsAsync(() => + ConnectivityLoad.RunAsync( + () => Task.FromException(error), + 4, TimeSpan.FromMilliseconds(100), TimeSpan.FromMilliseconds(5))); + Assert.Same(error, actual); + } + + /// Stopping admissions does not stop sampling while an already-started operation is still pending. + [Fact] + public async Task Load_InFlightOperationIsIncludedBeyondDuration() + { + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Task load = ConnectivityLoad.RunAsync( + () => + { + entered.TrySetResult(true); + return release.Task; + }, + 1, TimeSpan.FromMilliseconds(100), TimeSpan.FromMilliseconds(5)); + try + { + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await Task.Delay(150); + Assert.False(load.IsCompleted); + } + finally + { + release.TrySetResult(true); + } + + ConnectivityLoadResult result = await load; + Assert.Equal(1, result.SuccessfulOpens); + Assert.True(result.ElapsedSeconds >= 0.15); + Assert.True(result.Samples > 2); + } + + /// The same wait holds workers when synchronous and releases them when awaited. + [Fact] + public async Task Load_BlockingControlOccupiesMoreWorkersThanAsyncWait() + { + ThreadPool.GetMinThreads(out int originalWorkers, out int originalIo); + Assert.True(ThreadPool.SetMinThreads(Math.Max(originalWorkers, 8), originalIo)); + try + { + ConnectivityLoadResult asynchronous = await ConnectivityLoad.RunAsync( + () => Task.Delay(25), + 4, TimeSpan.FromMilliseconds(300), TimeSpan.FromMilliseconds(5)); + ConnectivityLoadResult blocking = await ConnectivityLoad.RunAsync( + () => + { + Thread.Sleep(25); + return Task.CompletedTask; + }, + 4, TimeSpan.FromMilliseconds(300), TimeSpan.FromMilliseconds(5)); + Assert.True(blocking.AverageOccupiedWorkers > asynchronous.AverageOccupiedWorkers + 2, + $"Blocking average {blocking.AverageOccupiedWorkers}, async average {asynchronous.AverageOccupiedWorkers}."); + } + finally + { + Assert.True(ThreadPool.SetMinThreads(originalWorkers, originalIo)); + ThreadPool.GetMinThreads(out int restoredWorkers, out int restoredIo); + Assert.Equal(originalWorkers, restoredWorkers); + Assert.Equal(originalIo, restoredIo); + } + } +} +#endif diff --git a/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/AuthenticationPressureScenario.cs b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/AuthenticationPressureScenario.cs new file mode 100644 index 0000000000..7cc66cf511 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/AuthenticationPressureScenario.cs @@ -0,0 +1,231 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections; +using System.Linq; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient; + +namespace ConnectivityTestHost; + +/// Combines synchronous pool-worker callers with cold/expired provider authentication, isolated from xUnit. +internal static class AuthenticationPressureScenario +{ + /// Runs the selected caller/cache/worker/pool dimensions; stdin controls start, token release and recovery. + internal static void Run(int port, string scenario) + { + string[] mode = scenario.Split('-'); + if (mode.Length != 5 || + (mode[1] != "sync" && mode[1] != "dedicated" && mode[1] != "async") || + (mode[2] != "cold" && mode[2] != "expired" && mode[2] != "warm") || + (mode[3] != "capped" && mode[3] != "generous") || + (mode[4] != "v1" && mode[4] != "v2")) + { + throw new ArgumentException("Unsupported authentication-pressure scenario."); + } + int width = Math.Max(Environment.ProcessorCount, 4); + var provider = new PressureProvider(); + if (!SqlAuthenticationProvider.SetProvider(SqlAuthenticationMethod.ActiveDirectoryManagedIdentity, provider)) + { + throw new InvalidOperationException("Could not register the isolated managed-identity test provider."); + } + string connectionString = new SqlConnectionStringBuilder + { + DataSource = $"127.0.0.1,{port}", + Encrypt = SqlConnectionEncryptOption.Optional, + Authentication = SqlAuthenticationMethod.ActiveDirectoryManagedIdentity, + MaxPoolSize = width + 1, + ConnectTimeout = 120, + ConnectRetryCount = 0, + Enlist = false + }.ConnectionString; + SqlConnection[] connections = Enumerable.Range(0, width) + .Select(_ => new SqlConnection(connectionString)).ToArray(); + try + { + if (mode[2] != "cold") + { + foreach (SqlConnection connection in connections) + { + connection.Open(); + } + if (mode[2] == "expired") + { + AgeTokens(connections); + } + foreach (SqlConnection connection in connections) + { + connection.Close(); + } + } + provider.Measuring = true; + int maximum = mode[3] == "capped" ? width : width * 4; + ConfigureWorkers(width, maximum); + Task.WhenAll(Enumerable.Range(0, width).Select(async _ => await Task.Yield())).GetAwaiter().GetResult(); + Console.WriteLine($"READY {width} {maximum} {mode[2]}"); + Require("START"); + using var start = new ManualResetEventSlim(); + using var admitted = new CountdownEvent(width); + Task[] opens; + if (mode[1] == "async") + { + opens = connections.Select(connection => connection.OpenAsync()).ToArray(); + } + else + { + opens = connections.Select(connection => Task.Factory.StartNew(() => + { + admitted.Signal(); + start.Wait(); + connection.Open(); + }, CancellationToken.None, mode[1] == "sync" ? TaskCreationOptions.None : TaskCreationOptions.LongRunning, + TaskScheduler.Default)).ToArray(); + if (!admitted.Wait(TimeSpan.FromSeconds(30))) + { + throw new TimeoutException("Not every caller reached the explicit worker start barrier."); + } + start.Set(); + } + Console.WriteLine($"CALLERS_STARTED {width}"); + Require("OBSERVE"); + if (mode[2] == "warm") + { + Task.WhenAll(opens).GetAwaiter().GetResult(); + Console.WriteLine($"WARM_CALLBACKS {provider.Calls}"); + } + else + { + Console.WriteLine(provider.Entered.Wait(TimeSpan.FromSeconds(10)) + ? "TOKEN_ENTERED" : "TOKEN_ENTRY_TIMEOUT"); + } + ThreadPool.GetAvailableThreads(out int available, out _); + Console.WriteLine($"AVAILABLE {available}"); + string release = Console.ReadLine(); + if (release == "RECOVER") + { + ConfigureWorkers(width, width * 4); + Console.WriteLine("WORKERS_RELEASED"); + } + else if (release != "RELEASE") + { + throw new InvalidOperationException("Parent did not release authentication."); + } + provider.Token.SetResult(NewToken()); + Task.WhenAll(opens).GetAwaiter().GetResult(); + Console.WriteLine($"COMPLETE {width} CALLBACKS {provider.Calls}"); + } + finally + { + SqlConnection.ClearPool(connections[0]); + foreach (SqlConnection connection in connections) + { + connection.Dispose(); + } + } + } + + /// Uses checked runtime limits, not assumptions about processor count or thread injection. + private static void ConfigureWorkers(int width, int maximum) + { + ThreadPool.GetMinThreads(out int minimum, out int minIo); + ThreadPool.GetMaxThreads(out _, out int maxIo); + if (!ThreadPool.SetMinThreads(Math.Min(minimum, width), minIo) || + !ThreadPool.SetMaxThreads(maximum, maxIo) || !ThreadPool.SetMinThreads(width, minIo)) + { + throw new InvalidOperationException("Runtime rejected the authentication worker limits."); + } + ThreadPool.GetMinThreads(out int actualMin, out _); + ThreadPool.GetMaxThreads(out int actualMax, out _); + if (actualMin != width || actualMax != maximum) + { + throw new InvalidOperationException("Runtime did not retain the requested authentication worker limits."); + } + } + + /// Ages physical and cached token metadata independently, without clearing the warmed pool or waiting for wall-clock expiry. + private static void AgeTokens(SqlConnection[] connections) + { + const BindingFlags flags = BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public; + PropertyInfo innerProperty = typeof(SqlConnection).GetProperty("InnerConnection", flags) + ?? throw new MissingMemberException("SqlConnection.InnerConnection"); + DateTimeOffset expired = DateTimeOffset.UtcNow.AddMinutes(-1); + foreach (SqlConnection connection in connections) + { + object inner = innerProperty.GetValue(connection) + ?? throw new InvalidOperationException("Priming did not produce a physical connection."); + FieldInfo field = inner.GetType().GetField("_fedAuthToken", flags) + ?? throw new MissingFieldException("Physical connection token field is unavailable."); + ConstructorInfo constructor = field.FieldType.GetConstructor(flags, null, new[] { typeof(SqlAuthenticationToken) }, null) + ?? throw new MissingMethodException("Physical token constructor is unavailable."); + field.SetValue(inner, constructor.Invoke(new object[] { new SqlAuthenticationToken("test-expired", expired) })); + PropertyInfo expiredProperty = inner.GetType().GetProperty("IsAccessTokenExpired", flags) + ?? throw new MissingMemberException("Physical token expiry property is unavailable."); + if (!Equals(true, expiredProperty.GetValue(inner))) + { + throw new InvalidOperationException("Physical token aging did not establish expiry."); + } + object pool = inner.GetType().GetProperty("Pool", flags)?.GetValue(inner) + ?? throw new MissingMemberException("Physical connection pool is unavailable."); + if (pool.GetType().GetProperty("AuthenticationContexts", flags)?.GetValue(pool) is not IDictionary cache || + cache.Count != 1) + { + throw new InvalidOperationException("Priming did not establish exactly one shared authentication context."); + } + IDictionaryEnumerator enumerator = cache.GetEnumerator(); + if (!enumerator.MoveNext()) + { + throw new InvalidOperationException("Authentication context disappeared during preconditioning."); + } + DictionaryEntry entry = enumerator.Entry; + object context = entry.Value ?? throw new InvalidOperationException("Authentication context is missing."); + byte[] accessToken = context.GetType().GetProperty("AccessToken", flags)?.GetValue(context) as byte[] + ?? throw new MissingMemberException("Authentication context token is unavailable."); + ConstructorInfo contextConstructor = context.GetType().GetConstructor(flags, null, + new[] { typeof(byte[]), typeof(DateTime) }, null) + ?? throw new MissingMethodException("Authentication context constructor is unavailable."); + cache[entry.Key] = contextConstructor.Invoke(new object[] { accessToken, expired.UtcDateTime }); + } + Console.WriteLine($"EXPIRED_PHYSICAL_AND_CACHE {connections.Length}"); + } + + /// Rejects absent/out-of-order controller commands instead of continuing with a default scenario. + private static void Require(string expected) + { + if (Console.ReadLine() != expected) + { + throw new InvalidOperationException($"Parent did not issue {expected}."); + } + } + + /// Returns only a fake token accepted by the loopback peer. + private static SqlAuthenticationToken NewToken() => new("test-token", DateTimeOffset.UtcNow.AddHours(2)); + + /// Priming is immediate; measured acquisition is genuinely incomplete and requires a worker continuation. + private sealed class PressureProvider : SqlAuthenticationProvider + { + internal bool Measuring; + internal int Calls; + internal readonly ManualResetEventSlim Entered = new(); + internal readonly TaskCompletionSource Token = new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// Only the explicitly selected managed-identity provider path is supported. + public override bool IsSupported(SqlAuthenticationMethod authenticationMethod) => + authenticationMethod == SqlAuthenticationMethod.ActiveDirectoryManagedIdentity; + + /// Signals measured entry separately from warm-pool preconditioning. + public override Task AcquireTokenAsync(SqlAuthenticationParameters parameters) + { + if (!Measuring) + { + return Task.FromResult(NewToken()); + } + Interlocked.Increment(ref Calls); + Entered.Set(); + return Token.Task; + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/ConnectivityTestHost.csproj b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/ConnectivityTestHost.csproj new file mode 100644 index 0000000000..38e321a2ae --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/ConnectivityTestHost.csproj @@ -0,0 +1,17 @@ + + + Exe + net462;net10.0 + false + + + + + + + + + + + + diff --git a/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/ConnectivityTestHost.targets b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/ConnectivityTestHost.targets new file mode 100644 index 0000000000..3e19f28c0a --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/ConnectivityTestHost.targets @@ -0,0 +1,19 @@ + + + + + + + + + + <_ConnectivityHostDirectory>@(_ConnectivityHostAssembly->'%(RootDir)%(Directory)') + + + <_ConnectivityHostFiles Include="$(_ConnectivityHostDirectory)**/*" /> + + + + diff --git a/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/EnlistmentCancellationScenario.cs b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/EnlistmentCancellationScenario.cs new file mode 100644 index 0000000000..8facdfe288 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/EnlistmentCancellationScenario.cs @@ -0,0 +1,264 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Diagnostics; +using System.Diagnostics.Tracing; +using System.Linq; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using System.Transactions; +using Microsoft.Data.SqlClient; + +namespace ConnectivityTestHost; + +/// Orders actual promotable enlistment against cancelled-scope disposal without adding driver synchronization hooks. +internal static class EnlistmentCancellationScenario +{ + /// Cancels at Login7, before enlistment can acquire either transaction lock. + internal static void RunBeforeEnlistment(int port, bool enlist) + { + using var cancellation = new CancellationTokenSource(); + using var started = new ManualResetEventSlim(); + Task caller = Task.Run(async () => + { + using var connection = new SqlConnection(ConnectionString(port, enlist)); + using (var scope = new TransactionScope(TransactionScopeOption.RequiresNew, + new TransactionOptions { IsolationLevel = IsolationLevel.ReadCommitted }, + TransactionScopeAsyncFlowOption.Enabled)) + { + Task open = connection.OpenAsync(cancellation.Token); + started.Set(); + try + { + await open.ConfigureAwait(false); + throw new InvalidOperationException("The held open completed without cancellation."); + } + catch (OperationCanceledException) when (cancellation.IsCancellationRequested) + { + Console.WriteLine("PUBLIC_CANCELLED"); + } + } + Console.WriteLine("SCOPE_COMPLETE"); + }); + if (!started.Wait(TimeSpan.FromSeconds(15))) + { + throw new TimeoutException("Ambient caller did not obtain an OpenAsync task."); + } + Console.WriteLine("CALL_RETURNED"); + if (Console.ReadLine() != "CANCEL") + { + throw new InvalidOperationException("Parent did not cancel the pre-enlistment open."); + } + cancellation.Cancel(); + if (!caller.Wait(TimeSpan.FromSeconds(15))) + { + throw new TimeoutException("Pre-enlistment cancellation did not finish scope disposal."); + } + if (Console.ReadLine() != "FINISH") + { + throw new InvalidOperationException("Parent did not release the held login."); + } + TransportAttemptScenario.WaitForPhysicalCompletion(); + Console.WriteLine("SCOPE_AND_PHYSICAL_COMPLETE"); + } + + internal static void Run(int port, bool cancel) + { + using var cancellation = new CancellationTokenSource(); + using var started = new ManualResetEventSlim(); + Transaction transaction = null; + using var trace = new EnlistmentTrace(() => transaction); + Task caller = Task.Run(async () => + { + using var connection = new SqlConnection(ConnectionString(port, true)); + using var scope = new TransactionScope(TransactionScopeOption.RequiresNew, + new TransactionOptions { IsolationLevel = IsolationLevel.ReadCommitted, Timeout = TimeSpan.FromMinutes(2) }, + TransactionScopeAsyncFlowOption.Enabled); + transaction = Transaction.Current; + Task open = connection.OpenAsync(cancellation.Token); + started.Set(); + try + { + await open.ConfigureAwait(false); + scope.Complete(); + } + catch (OperationCanceledException) when (cancel && cancellation.IsCancellationRequested) + { + Console.WriteLine("PUBLIC_CANCELLED"); + } + }); + if (!started.Wait(TimeSpan.FromSeconds(15))) + { + throw new TimeoutException("Ambient caller did not obtain an OpenAsync task."); + } + Console.WriteLine("CALL_RETURNED"); + if (!trace.Enlisted.Wait(TimeSpan.FromSeconds(15))) + { + throw new TimeoutException("Actual BEGIN did not complete into the post-enlistment trace gate."); + } + if (trace.Failure != null) + { + throw new InvalidOperationException("The source-specific enlistment lock probe is unavailable.", trace.Failure); + } + if (!trace.OpenStack) + { + throw new InvalidOperationException("The gate did not observe the actual physical enlistment stack."); + } + Console.WriteLine($"ENLISTED_PARSER_HELD {trace.ParserOwned}"); + if (Console.ReadLine() != (cancel ? "CANCEL" : "FINISH")) + { + throw new InvalidOperationException("Parent did not issue the selected enlistment command."); + } + if (cancel) + { + cancellation.Cancel(); + if (!SpinWait.SpinUntil(() => caller.IsCompleted || (trace.RollbackEntered.IsSet && + (trace.RollbackThread.ThreadState & System.Threading.ThreadState.WaitSleepJoin) != 0), + TimeSpan.FromSeconds(15))) + { + throw new TimeoutException("The cancelled scope did not complete or reach its actual rollback wait."); + } + if (!caller.IsCompleted && !trace.RollbackStack) + { + throw new InvalidOperationException("Rollback did not run from the cancelled caller's actual TransactionScope disposal."); + } + bool rollbackOwnsConnection = trace.ConnectionField.GetValue(trace.Promoter) == null && + IsMonitorOwned(trace.PhysicalConnection); + Console.WriteLine(rollbackOwnsConnection ? "ROLLBACK_OWNS_CONNECTION" : "ROLLBACK_RELEASED_CONNECTION"); + if (Console.ReadLine() != "FINISH") + { + throw new InvalidOperationException("Parent did not release the login-owned parser lock path."); + } + } + trace.Release.Set(); + if (!caller.Wait(TimeSpan.FromSeconds(20))) + { + if (trace.RollbackThread != null && + IsMonitorOwned(trace.PhysicalConnection) && IsMonitorOwned(trace.ParserSemaphore) && + (trace.OpenThread.ThreadState & System.Threading.ThreadState.WaitSleepJoin) != 0 && + (trace.RollbackThread.ThreadState & System.Threading.ThreadState.WaitSleepJoin) != 0) + { + Console.WriteLine("LOCK_CYCLE_STALLED"); + } + throw new TimeoutException("The cancelled caller's scope disposal did not finish after releasing physical enlistment."); + } + TransportAttemptScenario.WaitForPhysicalCompletion(); + Console.WriteLine("SCOPE_AND_PHYSICAL_COMPLETE"); + if (trace.GateTimedOut) + { + throw new TimeoutException("The controller missed the enlistment trace release."); + } + } + + private static string ConnectionString(int port, bool enlist) => new SqlConnectionStringBuilder + { + DataSource = $"127.0.0.1,{port}", + Pooling = false, + Enlist = enlist, + Encrypt = SqlConnectionEncryptOption.Optional, + ConnectRetryCount = 0, + ConnectTimeout = 60 + }.ConnectionString; + + private static bool IsMonitorOwned(object connection) + { + if (!Monitor.TryEnter(connection)) + { + return true; + } + Monitor.Exit(connection); + return false; + } + + /// Uses existing post-BEGIN and rollback traces. No trace entry alone is accepted as proof of lock ownership. + private sealed class EnlistmentTrace : EventListener + { + private readonly Func _transaction; + internal readonly ManualResetEventSlim Enlisted = new(); + internal readonly ManualResetEventSlim RollbackEntered = new(); + internal readonly ManualResetEventSlim Release = new(); + internal object Promoter; + internal object PhysicalConnection; + internal object ParserSemaphore; + internal Thread OpenThread; + internal Thread RollbackThread; + internal FieldInfo ConnectionField; + internal bool ParserOwned; + internal bool OpenStack; + internal bool RollbackStack; + internal bool GateTimedOut; + internal Exception Failure; + + internal EnlistmentTrace(Func transaction) => _transaction = transaction; + + protected override void OnEventSourceCreated(EventSource source) + { + if (source.Name == "Microsoft.Data.SqlClient.EventSource") + { + EnableEvents(source, EventLevel.Verbose, (EventKeywords)(128 | 2)); + } + } + + protected override void OnEventWritten(EventWrittenEventArgs data) + { + if (data.Payload?.FirstOrDefault() is not string message) + { + return; + } + if (message.StartsWith("SqlDelegatedTransaction.Rollback ", StringComparison.Ordinal)) + { + RollbackStack = new StackTrace().GetFrames().Any(frame => + frame.GetMethod().DeclaringType == typeof(TransactionScope) && frame.GetMethod().Name == "Dispose"); + RollbackThread = Thread.CurrentThread; + RollbackEntered.Set(); + } + if (message.StartsWith("SqlInternalConnection.EnlistNonNull ", StringComparison.Ordinal) && + message.Contains("delegated to transaction")) + { + try + { + object internalTransaction = Field(typeof(Transaction), "_internalTransaction", "internalTransaction").GetValue(_transaction()); + Promoter = Field(internalTransaction.GetType(), "_promoter", "promoter").GetValue(internalTransaction) + ?? throw new InvalidOperationException("The ambient transaction has no promotable enlistment."); + if (Promoter.GetType().FullName != "Microsoft.Data.SqlClient.SqlDelegatedTransaction") + { + throw new InvalidOperationException("The ambient transaction promoter is not SqlClient."); + } + ConnectionField = Field(Promoter.GetType(), "_connection"); + PhysicalConnection = ConnectionField.GetValue(Promoter) + ?? throw new InvalidOperationException("Promotable enlistment has no physical connection."); + object parserLock = Field(PhysicalConnection.GetType(), "_parserLock").GetValue(PhysicalConnection); + ParserSemaphore = Field(parserLock.GetType(), "_semaphore").GetValue(parserLock); + ParserOwned = Monitor.IsEntered(ParserSemaphore); + OpenThread = Thread.CurrentThread; + OpenStack = new StackTrace().GetFrames().Any(frame => frame.GetMethod().Name == "CompleteLogin"); + } + catch (Exception error) when (error is MissingFieldException || error is InvalidOperationException) + { + Failure = error; + } + Enlisted.Set(); + GateTimedOut = !Release.Wait(TimeSpan.FromSeconds(45)); + } + } + + private static FieldInfo Field(Type type, params string[] names) + { + for (Type current = type; current != null; current = current.BaseType) + { + foreach (string name in names) + { + FieldInfo field = current.GetField(name, BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public | BindingFlags.DeclaredOnly); + if (field != null) + { + return field; + } + } + } + throw new MissingFieldException(type.FullName, string.Join("/", names)); + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/Program.cs b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/Program.cs new file mode 100644 index 0000000000..5bd612353c --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/Program.cs @@ -0,0 +1,169 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Globalization; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient; + +namespace ConnectivityTestHost; + +/// Test-owned client process: global worker caps and deadlocks must not poison xUnit. +internal static class Program +{ + /// Executes a local reproduction. The parent owns gates, assertions and the watchdog. + [STAThread] + private static int Main(string[] args) + { + if (args.Length != 2) + { + throw new ArgumentException("Expected a scenario and one numeric argument."); + } + if (args[0] == "protocol-idle") + { + Console.WriteLine("READY"); + if (Console.ReadLine() != "STOP") + { + throw new InvalidOperationException("Unexpected protocol command."); + } + return 0; + } + int port = int.Parse(args[1], CultureInfo.InvariantCulture); + if (port <= 0 || port > 65535) + { + throw new ArgumentOutOfRangeException(nameof(port)); + } + bool poolV2 = args[0].EndsWith("-v2", StringComparison.Ordinal); + bool pooling = !args[0].EndsWith("-nonpooled", StringComparison.Ordinal); + AppContext.SetSwitch("Switch.Microsoft.Data.SqlClient.UseConnectionPoolV2", poolV2); + AppContext.SetSwitch("Switch.Microsoft.Data.SqlClient.UseManagedNetworkingOnWindows", true); + if (args[0] == "enlist-control" || args[0] == "enlist-cancel") + { + EnlistmentCancellationScenario.Run(port, args[0] == "enlist-cancel"); + return 0; + } + if (args[0] == "enlist-before-cancel" || args[0] == "enlist-disabled-cancel") + { + EnlistmentCancellationScenario.RunBeforeEnlistment(port, args[0] == "enlist-before-cancel"); + return 0; + } + if (args[0] == "attempt-cancel" || args[0] == "attempt-control") + { + TransportAttemptScenario.Run(port, args[0] == "attempt-cancel"); + return 0; + } + if (args[0].StartsWith("auth-", StringComparison.Ordinal)) + { + AuthenticationPressureScenario.Run(port, args[0]); + return 0; + } + string[] mode = args[0].Split('-'); + if (mode.Length != 3 || mode[0] != "workers" || + (mode[1] != "capped" && mode[1] != "generous" && mode[1] != "additional") || + (mode[1] == "additional" && mode[2] != "v2") || + (mode[2] != "v1" && mode[2] != "v2" && mode[2] != "nonpooled")) + { + throw new ArgumentException("Unsupported connectivity scenario."); + } + int width = Math.Max(Environment.ProcessorCount, 4); + ThreadPool.GetMinThreads(out int oldMin, out int minIo); + ThreadPool.GetMaxThreads(out _, out int maxIo); + int maximum = mode[1] == "generous" ? width * 4 : width; + if (!ThreadPool.SetMinThreads(Math.Min(oldMin, width), minIo) || + !ThreadPool.SetMaxThreads(maximum, maxIo) || + !ThreadPool.SetMinThreads(width, minIo)) + { + throw new InvalidOperationException("Runtime rejected the requested worker limits."); + } + ThreadPool.GetMinThreads(out int actualMin, out _); + ThreadPool.GetMaxThreads(out int actualMax, out _); + if (actualMin != width || actualMax != maximum) + { + throw new InvalidOperationException("Runtime worker limits differ from the requested limits."); + } + + // An async-only workload must progress under the same accepted cap. + Task.WhenAll(Enumerable.Range(0, width).Select(async _ => await Task.Yield())).GetAwaiter().GetResult(); + Console.WriteLine($"READY {width} {actualMin} {actualMax}"); + var token = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + SqlConnection[] connections = Enumerable.Range(0, width).Select(_ => + { + var connection = new SqlConnection(new SqlConnectionStringBuilder(ConnectionString(port)) { Pooling = pooling }.ConnectionString); + if (mode[1] != "additional") + { + connection.AccessTokenCallback = (_, _) => + { + Console.WriteLine("TOKEN_ENTERED"); + return token.Task; + }; + } + return connection; + }).ToArray(); + Task[] opens = connections.Select(connection => connection.OpenAsync()).ToArray(); + Console.WriteLine("CALLS_RETURNED"); + if (Console.ReadLine() != "PROBE") + { + throw new InvalidOperationException("Parent did not request the occupancy probe."); + } + ThreadPool.GetAvailableThreads(out int available, out _); + Console.WriteLine($"AVAILABLE {available}"); + if (mode[1] == "additional") + { + RunAdditionalOpen(connections, opens); + return 0; + } + if (Console.ReadLine() != "RELEASE") + { + throw new InvalidOperationException("Parent did not release authentication."); + } + token.SetResult(new SqlAuthenticationToken("test-token", DateTimeOffset.UtcNow.AddHours(2))); + Task.WhenAll(opens).GetAwaiter().GetResult(); + foreach (SqlConnection connection in connections) + { + connection.Dispose(); + } + Console.WriteLine($"COMPLETE {width}"); + return 0; + } + + /// Calls OpenAsync on the non-pool main thread after earlier opens occupy every worker. + private static void RunAdditionalOpen(SqlConnection[] connections, Task[] opens) + { + string command = Console.ReadLine(); + string[] parts = command?.Split(' '); + if (parts is null || parts.Length != 2 || parts[0] != "OPEN" || + !int.TryParse(parts[1], NumberStyles.None, CultureInfo.InvariantCulture, out int port) || + port <= 0 || port > 65535) + { + throw new InvalidOperationException("Parent did not provide the additional loopback endpoint."); + } + using var additional = new SqlConnection(ConnectionString(port)); + Console.WriteLine("ADDITIONAL_CALL_ENTERED"); + Task open = additional.OpenAsync(); + Console.WriteLine("ADDITIONAL_CALL_RETURNED"); + Console.WriteLine($"ADDITIONAL_TASK_PENDING {!open.IsCompleted}"); + open.GetAwaiter().GetResult(); + Console.WriteLine("ADDITIONAL_COMPLETE"); + Task.WhenAll(opens).GetAwaiter().GetResult(); + foreach (SqlConnection connection in connections) + { + connection.Dispose(); + } + Console.WriteLine($"COMPLETE {connections.Length + 1}"); + } + + /// Uses only the parent's fake loopback peer; no credentials enter process arguments. + private static string ConnectionString(int port) => new SqlConnectionStringBuilder + { + DataSource = $"127.0.0.1,{port}", + Encrypt = SqlConnectionEncryptOption.Optional, + Pooling = true, + MaxPoolSize = Math.Max(200, Environment.ProcessorCount * 4), + ConnectTimeout = 120, + ConnectRetryCount = 0, + Enlist = false + }.ConnectionString; +} diff --git a/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/TransportAttemptScenario.cs b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/TransportAttemptScenario.cs new file mode 100644 index 0000000000..d4a35cbe1d --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/ConnectivityTestHost/TransportAttemptScenario.cs @@ -0,0 +1,154 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Diagnostics.Tracing; +using System.Linq; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient; + +namespace ConnectivityTestHost; + +/// Separates public cancellation from completion of the actual non-pooled physical-creation task. +internal static class TransportAttemptScenario +{ + /// Counts actual socket connect starts against an unused loopback port; no configurable retry or query is involved. + internal static void Run(int port, bool cancel) + { + using var cancellation = new CancellationTokenSource(); + using var trace = new ConnectTrace(); + using var connection = new SqlConnection(new SqlConnectionStringBuilder + { + DataSource = $"127.0.0.1,{port}", + Encrypt = SqlConnectionEncryptOption.Optional, + Pooling = false, + Enlist = false, + ConnectRetryCount = 0, + ConnectTimeout = 10 + }.ConnectionString); + Task open = connection.OpenAsync(cancellation.Token); + Console.WriteLine("CALL_RETURNED"); + if (!trace.FirstFailure.Wait(TimeSpan.FromSeconds(15))) + { + throw new TimeoutException($"The endpoint did not produce a traced connect failure. Attempts: {trace.Attempts}; last phase: {trace.LastPhase}."); + } + Console.WriteLine($"FAILURE_HELD ATTEMPTS {trace.Attempts}"); + string command = Console.ReadLine(); + if (command != (cancel ? "CANCEL" : "CONTINUE")) + { + throw new InvalidOperationException("Parent did not issue the selected attempt command."); + } + if (cancel) + { + cancellation.Cancel(); + try + { + open.GetAwaiter().GetResult(); + throw new InvalidOperationException("The public open unexpectedly succeeded."); + } + catch (OperationCanceledException) when (cancellation.IsCancellationRequested) + { + Console.WriteLine("PUBLIC_CANCELLED"); + } + } + else + { + Console.WriteLine("RETRY_ALLOWED"); + } + if (Console.ReadLine() != "FINISH") + { + throw new InvalidOperationException("Parent did not release the transport failure."); + } + trace.Release.Set(); + WaitForPhysicalCompletion(); + if (!cancel) + { + try + { + open.GetAwaiter().GetResult(); + throw new InvalidOperationException("The unused endpoint unexpectedly accepted login."); + } + catch (SqlException exception) + { + Console.WriteLine($"NETWORK_FAILURE {exception.Number}"); + } + } + if (trace.GateTimedOut) + { + throw new TimeoutException("The controller did not release the traced connect failure."); + } + Console.WriteLine($"PHYSICAL_COMPLETE ATTEMPTS {trace.Attempts}"); + } + + /// Observes the source's actual non-pooled creation tasks, not just a cancelled public task. + internal static void WaitForPhysicalCompletion() + { + Type factory = typeof(SqlConnection).Assembly.GetType("Microsoft.Data.SqlClient.SqlConnectionFactory", throwOnError: true); + FieldInfo field = factory.GetField("s_pendingOpenNonPooled", BindingFlags.Static | BindingFlags.NonPublic) + ?? throw new MissingFieldException("Non-pooled physical-creation tasks are unavailable."); + if (field.GetValue(null) is not Array pending) + { + throw new InvalidOperationException("Non-pooled creation task array is unavailable."); + } + Task[] physical = pending.Cast().Where(task => task != null).ToArray(); + if (physical.Length == 0) + { + throw new InvalidOperationException("No physical-creation task was observed."); + } + if (!SpinWait.SpinUntil(() => physical.All(task => task.IsCompleted), TimeSpan.FromSeconds(15))) + { + throw new TimeoutException("Physical creation did not finish within the cleanup watchdog."); + } + foreach (Task task in physical.Where(task => task.IsFaulted)) + { + _ = task.Exception; + } + } + + /// Gates an existing error trace after a real refused connect; it does not replace transport or configure SQL-error retries. + private sealed class ConnectTrace : EventListener + { + internal readonly ManualResetEventSlim FirstFailure = new(); + internal readonly ManualResetEventSlim Release = new(); + private int _attempts; + private int _failed; + internal int Attempts => Volatile.Read(ref _attempts); + internal bool GateTimedOut; + internal string LastPhase = "none"; + + /// Enables only managed-SNI tracing; payloads are filtered in memory and never printed. + protected override void OnEventSourceCreated(EventSource source) + { + if (source.Name == "Microsoft.Data.SqlClient.EventSource") + { + EnableEvents(source, EventLevel.Verbose, (EventKeywords)(2048 | 128)); + } + } + + /// Counts actual handle connect starts and holds only the first observed socket failure. + protected override void OnEventWritten(EventWrittenEventArgs data) + { + if ((data.EventName != "SNITrace" && data.EventName != "AdvancedTrace") || data.Payload?.FirstOrDefault() is not string message || + !message.StartsWith("SniTcpHandle.", StringComparison.Ordinal)) + { + return; + } + if (message.Contains("Connecting to IP address")) + { + Interlocked.Increment(ref _attempts); + } + int separator = message.IndexOf('|'); + LastPhase = separator >= 0 ? message.Substring(0, separator) : data.EventName; + if ((message.IndexOf("ConnectionRefused", StringComparison.OrdinalIgnoreCase) >= 0 || + message.IndexOf("Connection refused", StringComparison.OrdinalIgnoreCase) >= 0) && + Interlocked.CompareExchange(ref _failed, 1, 0) == 0) + { + FirstFailure.Set(); + GateTimedOut = !Release.Wait(TimeSpan.FromSeconds(30)); + } + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.EndPoint/ITDSTransactionServer.cs b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.EndPoint/ITDSTransactionServer.cs new file mode 100644 index 0000000000..1b43b5a7b8 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.EndPoint/ITDSTransactionServer.cs @@ -0,0 +1,13 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +namespace Microsoft.SqlServer.TDS.EndPoint +{ + /// Optional transaction-manager support; existing fake peers are not required to implement it. + public interface ITDSTransactionServer + { + /// Processes an actual transaction-manager request on its originating session. + TDSMessageCollection OnTransactionManagerRequest(ITDSServerSession session, TDSMessage message); + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.EndPoint/TDSServerParser.cs b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.EndPoint/TDSServerParser.cs index 05078d7e96..96d831fbc0 100644 --- a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.EndPoint/TDSServerParser.cs +++ b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.EndPoint/TDSServerParser.cs @@ -116,6 +116,15 @@ public void Run() responseMessages = Server.OnAttention(Session, MessageBeingReceived); break; } + case TDSMessageType.TransactionManager: + { + if (!(Server is ITDSTransactionServer transactionServer)) + { + throw new NotSupportedException("This test peer does not support transaction-manager requests."); + } + responseMessages = transactionServer.OnTransactionManagerRequest(Session, MessageBeingReceived); + break; + } case TDSMessageType.FederatedAuthenticationToken: { // Call into the subscriber to process the packet diff --git a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GatedConnectionTdsServer.cs b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GatedConnectionTdsServer.cs new file mode 100644 index 0000000000..fa539bd89b --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GatedConnectionTdsServer.cs @@ -0,0 +1,137 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections.Generic; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.SqlServer.TDS.EndPoint; + +namespace Microsoft.SqlServer.TDS.Servers +{ + /// Connection phases which can be held without delaying unrelated requests. + public enum TdsConnectionPhase + { + Login7, + FederatedAuthenticationInfo + } + + /// + /// Holds a selected connection phase until the test releases it. Arrival signals establish + /// ordering without sleeps; release precedes server disposal so teardown cannot strand a handler. + /// + public sealed class GatedConnectionTdsServer : TdsServer + { + private readonly object _gateLock = new object(); + private readonly ManualResetEventSlim _release = new ManualResetEventSlim(); + private readonly List>> _arrivals = + new List>>(); + private readonly TdsConnectionPhase _phase; + private int _arrivalCount; + private bool _disposed; + + /// Creates a server whose selected phase is initially held. + public GatedConnectionTdsServer(TdsConnectionPhase phase, TdsServerArguments arguments = null) + : base(arguments ?? new TdsServerArguments()) + { + _phase = phase; + ListenAddress = IPAddress.Loopback; + } + + /// Number of requests that have reached the selected phase. + public int ArrivalCount => Volatile.Read(ref _arrivalCount); + + /// + /// Signals once the requested number of actual wire requests has arrived. The caller + /// must bound its wait; a missed phase is not a successful reproduction. + /// + public Task WaitForArrivalsAsync(int count) + { + if (count <= 0) + { + throw new ArgumentOutOfRangeException(nameof(count)); + } + + lock (_gateLock) + { + if (_disposed) + { + throw new ObjectDisposedException(nameof(GatedConnectionTdsServer)); + } + + if (_arrivalCount >= count) + { + return Task.CompletedTask; + } + + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _arrivals.Add(new KeyValuePair>(count, completion)); + return completion.Task; + } + } + + /// Releases current and subsequent requests, including on an assertion failure. + public void Release() => _release.Set(); + + /// Releases handlers and observes server shutdown before disposing synchronization state. + public override void Dispose() + { + lock (_gateLock) + { + if (_disposed) + { + return; + } + _disposed = true; + foreach (var arrival in _arrivals) + { + arrival.Value.TrySetException(new ObjectDisposedException(nameof(GatedConnectionTdsServer))); + } + _arrivals.Clear(); + } + + Release(); + base.Dispose(); + _release.Dispose(); + } + + /// Holds Login7 before the server prepares its login response. + public override TDSMessageCollection OnLogin7Request(ITDSServerSession session, TDSMessage request) + { + Enter(TdsConnectionPhase.Login7); + return base.OnLogin7Request(session, request); + } + + /// Holds the fed-auth information needed before invoking the client's provider. + protected override TDSMessageCollection OnFederatedAuthenticationInfoRequest(ITDSServerSession session) + { + Enter(TdsConnectionPhase.FederatedAuthenticationInfo); + return base.OnFederatedAuthenticationInfoRequest(session); + } + + /// Publishes arrival before blocking the server-side handler only. + private void Enter(TdsConnectionPhase phase) + { + if (phase != _phase) + { + return; + } + + lock (_gateLock) + { + Interlocked.Increment(ref _arrivalCount); + for (int i = _arrivals.Count - 1; i >= 0; i--) + { + if (_arrivals[i].Key <= _arrivalCount) + { + _arrivals[i].Value.TrySetResult(true); + _arrivals.RemoveAt(i); + } + } + } + _release.Wait(); + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GatedTransactionTdsServer.cs b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GatedTransactionTdsServer.cs new file mode 100644 index 0000000000..e3f7519ef6 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GatedTransactionTdsServer.cs @@ -0,0 +1,108 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections.Concurrent; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.SqlServer.TDS.Done; +using Microsoft.SqlServer.TDS.EndPoint; +using Microsoft.SqlServer.TDS.EnvChange; +using Microsoft.SqlServer.TDS.TransactionManager; + +namespace Microsoft.SqlServer.TDS.Servers +{ + /// Supports real local enlistment and can hold its BEGIN response; it does not implement DTC or SQL transaction semantics. + public sealed class GatedTransactionTdsServer : TdsServer, ITDSTransactionServer + { + private readonly ConcurrentDictionary _transactions = new ConcurrentDictionary(); + private readonly ManualResetEventSlim _release = new ManualResetEventSlim(); + private readonly TaskCompletionSource _begin = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private int _beginCount; + private int _commitCount; + private int _rollbackCount; + private int _disposed; + + public Task BeginArrived => _begin.Task; + public int BeginCount => Volatile.Read(ref _beginCount); + public int CommitCount => Volatile.Read(ref _commitCount); + public int RollbackCount => Volatile.Read(ref _rollbackCount); + + /// Creates a loopback peer with an initially held BEGIN response. + public GatedTransactionTdsServer() : base(new TdsServerArguments()) + { + ListenAddress = IPAddress.Loopback; + } + + /// Releases present and future BEGIN responses, including during cleanup. + public void ReleaseBegin() => _release.Set(); + + /// Tracks descriptors per session and returns genuine transaction ENVCHANGE and DONE tokens. + public TDSMessageCollection OnTransactionManagerRequest(ITDSServerSession session, TDSMessage message) + { + var request = (TDSTransactionManagerToken)message[0]; + ulong descriptor; + TDSEnvChangeToken change; + bool begin = request.Request == TDSTransactionManagerRequest.Begin; + if (begin) + { + descriptor = session.SessionID + 1UL; + if (request.Descriptor != 0 || !_transactions.TryAdd(session.SessionID, descriptor)) + { + throw new InvalidOperationException("Nested or duplicate local transaction BEGIN."); + } + Interlocked.Increment(ref _beginCount); + _begin.TrySetResult(true); + _release.Wait(); + change = new TDSEnvChangeToken(TDSEnvChangeTokenType.BeginTransaction, BitConverter.GetBytes(descriptor), null); + } + else + { + if (!_transactions.TryRemove(session.SessionID, out descriptor) || descriptor != request.Descriptor) + { + throw new InvalidOperationException("Transaction completion did not match the session descriptor."); + } + if (request.Request == TDSTransactionManagerRequest.Commit) + { + Interlocked.Increment(ref _commitCount); + change = new TDSEnvChangeToken(TDSEnvChangeTokenType.CommitTransaction, null, BitConverter.GetBytes(descriptor)); + } + else if (request.Request == TDSTransactionManagerRequest.Rollback) + { + Interlocked.Increment(ref _rollbackCount); + change = new TDSEnvChangeToken(TDSEnvChangeTokenType.RollbackTransaction, null, BitConverter.GetBytes(descriptor)); + } + else + { + throw new NotSupportedException("Unsupported local transaction-manager operation."); + } + } + var response = new TDSMessage(TDSMessageType.Response); + response.Add(change); + response.Add(new TDSDoneToken(begin ? TDSDoneTokenStatusType.TransactionInProgress : TDSDoneTokenStatusType.Final)); + return new TDSMessageCollection(response); + } + + /// Removes only this session's transaction bookkeeping. + public override void CloseSession(ITDSServerSession session) + { + _transactions.TryRemove(session.SessionID, out _); + base.CloseSession(session); + } + + /// Unblocks transport handlers before joining peer threads and disposing the gate. + public override void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + ReleaseBegin(); + _begin.TrySetException(new ObjectDisposedException(nameof(GatedTransactionTdsServer))); + base.Dispose(); + _release.Dispose(); + } + } +} diff --git a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GenericTdsServer.cs b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GenericTdsServer.cs index c3fadc3e0a..366113766e 100644 --- a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GenericTdsServer.cs +++ b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS.Servers/GenericTdsServer.cs @@ -127,6 +127,9 @@ public GenericTdsServer(T arguments, QueryEngine queryEngine) public IPEndPoint EndPoint => _endpoint.ServerEndPoint; + /// Listener address; local-only reproductions select loopback before starting. + public IPAddress ListenAddress { get; set; } = IPAddress.Any; + /// /// Server configuration /// @@ -174,7 +177,7 @@ public void Start([CallerMemberName] string methodName = "") } _endpoint = new TDSServerEndPoint(this) { - ServerEndPoint = new IPEndPoint(IPAddress.Any, 0), + ServerEndPoint = new IPEndPoint(ListenAddress, 0), EndpointName = methodName, EventLog = Arguments.Log }; diff --git a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/EnvChange/TDSEnvChangeToken.cs b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/EnvChange/TDSEnvChangeToken.cs index 1c2837de31..dd4e97bc01 100644 --- a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/EnvChange/TDSEnvChangeToken.cs +++ b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/EnvChange/TDSEnvChangeToken.cs @@ -191,6 +191,9 @@ public override bool Inflate(Stream source) break; } case TDSEnvChangeTokenType.SQLCollation: + case TDSEnvChangeTokenType.BeginTransaction: + case TDSEnvChangeTokenType.CommitTransaction: + case TDSEnvChangeTokenType.RollbackTransaction: { // Read new value length byte valueLength = (byte)source.ReadByte(); @@ -348,6 +351,9 @@ public override void Deflate(Stream destination) break; } case TDSEnvChangeTokenType.SQLCollation: + case TDSEnvChangeTokenType.BeginTransaction: + case TDSEnvChangeTokenType.CommitTransaction: + case TDSEnvChangeTokenType.RollbackTransaction: { // Write new value length cache.WriteByte((byte)(NewValue != null ? ((byte[])NewValue).Length : 0)); diff --git a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/TDSMessage.cs b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/TDSMessage.cs index 65ff06e8ad..fbabc83292 100644 --- a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/TDSMessage.cs +++ b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/TDSMessage.cs @@ -314,6 +314,13 @@ private void _InflateClientTokens() // Do nothing break; } + case TDSMessageType.TransactionManager: + { + var transaction = new TransactionManager.TDSTransactionManagerToken(); + transaction.Inflate(_dataStream); + Add(transaction); + break; + } case TDSMessageType.FederatedAuthenticationToken: { Add(new TDSFedAuthToken(_dataStream)); diff --git a/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/TransactionManager/TDSTransactionManagerToken.cs b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/TransactionManager/TDSTransactionManagerToken.cs new file mode 100644 index 0000000000..1149aea9a4 --- /dev/null +++ b/src/Microsoft.Data.SqlClient/tests/tools/TDS/TDS/TransactionManager/TDSTransactionManagerToken.cs @@ -0,0 +1,105 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.IO; +using System.Text; + +namespace Microsoft.SqlServer.TDS.TransactionManager +{ + /// Local transaction-manager operations supported by the connectivity test peer. + public enum TDSTransactionManagerRequest : ushort + { + Begin = 5, + Commit = 7, + Rollback = 8 + } + + /// Parses the TDS 7.4 transaction descriptor header and local BEGIN/COMMIT/ROLLBACK payload. + public sealed class TDSTransactionManagerToken : TDSPacketToken + { + public ulong Descriptor { get; set; } + public uint OutstandingRequests { get; set; } = 1; + public TDSTransactionManagerRequest Request { get; set; } + public byte IsolationLevel { get; set; } + public string Name { get; set; } = string.Empty; + + /// Inflates a complete buffered request; truncated/unsupported headers and transaction operations fail explicitly. + public override bool Inflate(Stream source) + { + using (var reader = new BinaryReader(source, Encoding.Unicode, leaveOpen: true)) + { + if (reader.ReadUInt32() != 22 || reader.ReadUInt32() != 18 || reader.ReadUInt16() != 2) + { + throw new NotSupportedException("The local transaction peer requires one transaction-descriptor header."); + } + Descriptor = reader.ReadUInt64(); + OutstandingRequests = reader.ReadUInt32(); + Request = (TDSTransactionManagerRequest)reader.ReadUInt16(); + if (Request == TDSTransactionManagerRequest.Begin) + { + IsolationLevel = reader.ReadByte(); + if (IsolationLevel < 1 || IsolationLevel > 5) + { + throw new NotSupportedException("Unsupported local transaction isolation level."); + } + } + else if (Request != TDSTransactionManagerRequest.Commit && Request != TDSTransactionManagerRequest.Rollback) + { + throw new NotSupportedException("The local peer supports only BEGIN, COMMIT, and ROLLBACK."); + } + int length = reader.ReadByte(); + if ((length & 1) != 0) + { + throw new InvalidDataException("Transaction names must contain complete UTF-16 code units."); + } + byte[] name = reader.ReadBytes(length); + if (name.Length != length) + { + throw new EndOfStreamException("Transaction name was truncated."); + } + Name = Encoding.Unicode.GetString(name); + if (Request != TDSTransactionManagerRequest.Begin && reader.ReadByte() != 0) + { + throw new NotSupportedException("Chained transaction flags are not supported by the local peer."); + } + if (source.Position != source.Length) + { + throw new InvalidDataException("Unexpected trailing transaction-manager payload."); + } + return true; + } + } + + /// Serializes the supported request shape for parser round-trip tests. + public override void Deflate(Stream destination) + { + if (Request != TDSTransactionManagerRequest.Begin && + Request != TDSTransactionManagerRequest.Commit && Request != TDSTransactionManagerRequest.Rollback) + { + throw new NotSupportedException("Unsupported local transaction request."); + } + byte[] name = Encoding.Unicode.GetBytes(Name); + using (var writer = new BinaryWriter(destination, Encoding.Unicode, leaveOpen: true)) + { + writer.Write(22u); + writer.Write(18u); + writer.Write((ushort)2); + writer.Write(Descriptor); + writer.Write(OutstandingRequests); + writer.Write((ushort)Request); + if (Request == TDSTransactionManagerRequest.Begin) + { + writer.Write(IsolationLevel); + } + writer.Write(checked((byte)name.Length)); + writer.Write(name); + if (Request != TDSTransactionManagerRequest.Begin) + { + writer.Write((byte)0); + } + } + } + } +}