Client/transport: one recv per small chunk, ManagedSession lock fast path, write coalescing

perf/README.md

@@ -37,7 +37,7 @@ perf/ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe run --scenario
 | `--nodes` | 100 | Values per Read/Write request, or monitored items per subscription. |
 | `--inflight` | 1 | Outstanding requests per session (closed loop). |
 | `--security` | `none` | `none`, `sign` or `encrypt` (Basic256Sha256). |
-| `--client` | `classic` | `classic`: `Session` with the classic subscription engine. `managed`: `ManagedSession` with the V2 subscription engine (`DefaultSubscriptionEngineFactory`). |
+| `--client` | `classic` | `classic`: `Session` with the classic subscription engine. `managed`: `ManagedSession` with the V2 subscription engine (`DefaultSubscriptionEngineFactory`). `shared`: classic sessions on secure channels shared through one `ClientChannelManager`, created one at a time (`--connect-parallel`). |
 | `--callback` | `fast` | Classic data changes through `Subscription.FastDataChangeCallback` (`fast`) or the per-item `MonitoredItem.Notification` event (`item`). |
 | `--subs`, `--pub`, `--write-interval` | 5, 100, 100 | Subscriptions per session, publishing interval and writer interval in ms (`sub`). |
 | `--duration`, `--warmup` | 20, 10 | Measurement window and warm-up in seconds. |

perf/ServerLoadHarness/Program.cs

@@ -47,6 +47,7 @@
 using System.Globalization;
 using System.IO;
 using System.Linq;
+using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -301,25 +302,41 @@ await client
 
             // classic: Session from the default session factory (classic subscription engine).
             // managed: ManagedSession with the default V2 subscription engine.
+            // shared: classic Sessions whose secure channels come from one ClientChannelManager.
             string clientKind = Arg("client", "classic");
-            ConfiguredEndpoint? managedEndpoint = clientKind == "managed"
+            ConfiguredEndpoint? endpointForBuilder = clientKind is "managed" or "shared"
                 ? await client.GetEndpointAsync(new Uri(url), policy, selected).ConfigureAwait(false)
                 : null;
+            var channelManager = new ClientChannelManager(client.Config, telemetry);
+            await using ConfiguredAsyncDisposable disposeChannelManager = channelManager.ConfigureAwait(false);
             var sessions = new ISession[sessionCount];
             var connectTime = Stopwatch.StartNew();
+            // Shared sessions connect one at a time by default: concurrent creation
+            // on one ClientChannelManager fails with BadSecureChannelClosed.
+            int connectParallel = clientKind == "shared" ? ArgInt("connect-parallel", 1) : 32;
             await Parallel.ForAsync(
                 0,
                 sessionCount,
-                new ParallelOptions { MaxDegreeOfParallelism = 32 },
-                async (i, ct) => sessions[i] = managedEndpoint != null
-                    ? await new ManagedSessionBuilder(client.Config, telemetry)
-                        .UseEndpoint(managedEndpoint)
+                new ParallelOptions { MaxDegreeOfParallelism = connectParallel },
+                async (i, ct) => sessions[i] = clientKind switch
+                {
+                    "managed" => await new ManagedSessionBuilder(client.Config, telemetry)
+                        .UseEndpoint(endpointForBuilder!)
                         .WithSessionTimeout(TimeSpan.FromSeconds(60))
                         .ConnectAsync(ct)
-                        .ConfigureAwait(false)
-                    : await client
+                        .ConfigureAwait(false),
+                    "shared" => await Session.CreateAsync(
+                            channelManager,
+                            client.Config,
+                            endpointForBuilder!,
+                            updateBeforeConnect: false,
+                            sessionName: "PerfShared",
+                            ct: ct)
+                        .ConfigureAwait(false),
+                    _ => await client
                         .ConnectAsync(new Uri(url), policy, selected)
-                        .ConfigureAwait(false))
+                        .ConfigureAwait(false)
+                })
                 .ConfigureAwait(false);
             Console.WriteLine(FormattableString.Invariant(
                 $"connected {sessionCount} {clientKind} sessions ({security}) in {connectTime.ElapsedMilliseconds} ms"));

perf/ab.sh

@@ -1,10 +1,11 @@
 #!/bin/bash
 # Paired A/B runs: alternates a baseline and a candidate harness build per scenario and round.
 # BASELINE: harness built in a second worktree of the commit to compare against.
+# CANDIDATE: harness to compare (default: the build in this worktree).
 # SCENARIOS: file with one scenario per line (default: server-scenarios.txt).
 # ROUNDS: rounds per scenario (default 3). OUT: result file (default ab-results.txt).
 A=${BASELINE:-C:/uabase/perf/ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe}
-B=./ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe
+B=${CANDIDATE:-./ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe}
 O=${OUT:-ab-results.txt}; rm -f "$O"
 ROUNDS=${ROUNDS:-3}
 SCENARIOS=${SCENARIOS:-server-scenarios.txt}