mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-25 03:43:05 +08:00
romanettandClaude Opus 5.5
bd38870cf7
## Summary The 1-core client stall in the server load harness (`--client-cores 1`, write->notify p99 0.7-1.7 s) was caused by the harness, not the OPC UA client. `StartServerChild` started `PumpServerOutput` with `Task.Run`. The pump blocks in a synchronous `StreamReader.ReadLine` on the server's stdout for the whole run, so it held one thread-pool worker permanently. With `DOTNET_PROCESSOR_COUNT=1` the pool's minimum is one worker, so all other work items waited for the pool's starvation injection. That is why `DOTNET_ThreadPool_ForceMinWorkerThreads=8` and two processors both hid the stall. **Evidence:** 14 `dotnet-stack` samples of the constrained client during the measurement window. In every sample, the only blocked pool worker was in `ReadFile <- StreamReader.ReadLine <- Program.PumpServerOutput <- StartServerChild`, and the other workers were idle. No `Opc.Ua` frame was blocked. **Fix:** run the pump on a dedicated background thread. ## Numbers `run --scenario sub --sessions 25 --subs 20 --nodes 100 --pub 500 --write-interval 100 --duration 10 --warmup 6 --client-cores 1 --client-memory 512`, 5 alternating rounds (ms): | round | base write->notify p99 | fix write->notify p99 | base write p99 | fix write p99 | |---|---|---|---|---| | 1 | 659 | 61 | 619 | 12 | | 2 | 885 | 310 | 859 | 15 | | 3 | 306 | 303 | 52 | 23 | | 4 | 1717 | 455 | 408 | 81 | | 5 | 234 | 193 | 148 | 67 | The baseline stalled (> 600 ms) in 3 of 5 runs and the fix in 0 of 5 (max 492 ms). The remaining ~300 ms p99 comes from the 500 ms publishing interval. ## Test plan - [x] Harness Release build (net10.0) - [x] Alternating A/B runs above 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1492 lines
65 KiB
C#
1492 lines
65 KiB
C#
/* ========================================================================
|
|
* Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
|
|
*
|
|
* OPC Foundation MIT License 1.00
|
|
*
|
|
* Permission is hereby granted, free of charge, to any person
|
|
* obtaining a copy of this software and associated documentation
|
|
* files (the "Software"), to deal in the Software without
|
|
* restriction, including without limitation the rights to use,
|
|
* copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
* copies of the Software, and to permit persons to whom the
|
|
* Software is furnished to do so, subject to the following
|
|
* conditions:
|
|
*
|
|
* The above copyright notice and this permission notice shall be
|
|
* included in all copies or substantial portions of the Software.
|
|
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
|
|
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
|
|
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
|
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
|
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
|
|
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
|
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
|
|
* OTHER DEALINGS IN THE SOFTWARE.
|
|
*
|
|
* The complete license agreement can be found here:
|
|
* http://opcfoundation.org/License/MIT/1.00/
|
|
* ======================================================================*/
|
|
|
|
|
|
// Server load / profiling harness. Not part of the product; see perf/README.md.
|
|
// Modes:
|
|
// server --port P [--diag 0|1] [--audit 0|1] [--threads N]
|
|
// Starts a ReferenceServer and reads commands from stdin: mark | report | quit
|
|
// run --scenario read|write|sub|browse [--sessions N] [--nodes K] [--inflight I]
|
|
// [--duration S] [--warmup S] [--security none|sign|encrypt] [--diag 0|1] [--audit 0|1]
|
|
// [--subs S] [--pub ms] [--write-interval ms] [--write-nodes K]
|
|
// [--trace file] [--trace-profile p] [--out file]
|
|
// [--client managed|classic|shared] [--callback fast|item] [--trace-side server|client]
|
|
// Spawns the server child, drives the load and prints a report of client latency
|
|
// and the CPU / GC / allocation deltas of both processes for the measurement window.
|
|
|
|
using System;
|
|
using System.Collections;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics;
|
|
using System.Globalization;
|
|
using System.IO;
|
|
using System.Linq;
|
|
using System.Numerics;
|
|
using System.Runtime.CompilerServices;
|
|
using System.Text;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Logging;
|
|
using Opc.Ua.Client;
|
|
using Opc.Ua.Client.TestFramework;
|
|
using Opc.Ua.Configuration;
|
|
using Opc.Ua.Server.TestFramework;
|
|
using Quickstarts.ReferenceServer;
|
|
using V2 = Opc.Ua.Client.Subscriptions;
|
|
|
|
namespace Opc.Ua.Perf.ServerLoadHarness
|
|
{
|
|
/// <summary>
|
|
/// Entry point of the harness.
|
|
/// </summary>
|
|
public static class Program
|
|
{
|
|
private static readonly Dictionary<string, string> s_args = new(StringComparer.OrdinalIgnoreCase);
|
|
|
|
// client GC events of the window (--gc-events 0 turns the listener off)
|
|
private static GcListener? s_gcListener;
|
|
|
|
// processor masks of a constrained client run (--client-cores), set by RunConstrainedAsync
|
|
private const string kClientAffinityVariable = "PERF_CLIENT_AFFINITY";
|
|
private const string kServerAffinityVariable = "PERF_SERVER_AFFINITY";
|
|
private const string kConstrainedVariable = "PERF_CONSTRAINED";
|
|
|
|
/// <summary>
|
|
/// Runs the server or the load driver.
|
|
/// </summary>
|
|
public static async Task<int> Main(string[] args)
|
|
{
|
|
for (int i = 1; i < args.Length; i++)
|
|
{
|
|
if (args[i].StartsWith("--", StringComparison.Ordinal))
|
|
{
|
|
string key = args[i][2..];
|
|
// --server-args takes the next argument verbatim, which may start with "--"
|
|
string value = i + 1 < args.Length &&
|
|
(key == "server-args" || !args[i + 1].StartsWith("--", StringComparison.Ordinal))
|
|
? args[++i]
|
|
: "1";
|
|
s_args[key] = value;
|
|
}
|
|
}
|
|
string mode = args.Length > 0 ? args[0] : "run";
|
|
if (mode == "server")
|
|
{
|
|
return await RunServerAsync().ConfigureAwait(false);
|
|
}
|
|
if ((s_args.ContainsKey("client-cores") || s_args.ContainsKey("client-memory")) &&
|
|
NativeWorkerBenchmark.IsScenario(Arg("scenario", "read")))
|
|
{
|
|
// the native workers are not .NET processes; they use --client-affinity
|
|
Console.WriteLine(
|
|
"--client-cores and --client-memory apply to the .NET client scenarios; " +
|
|
"use --client-affinity and --server-affinity with the native workers.");
|
|
return 2;
|
|
}
|
|
if ((s_args.ContainsKey("client-cores") || s_args.ContainsKey("client-memory")) &&
|
|
Environment.GetEnvironmentVariable(kConstrainedVariable) == null)
|
|
{
|
|
return await RunConstrainedAsync(args).ConfigureAwait(false);
|
|
}
|
|
return NativeWorkerBenchmark.IsScenario(Arg("scenario", "read"))
|
|
? await NativeWorkerBenchmark.RunAsync(args).ConfigureAwait(false)
|
|
: await RunDriverAsync(args).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Runs the harness again as a client confined like a container: <c>--client-cores N</c>
|
|
/// gives it the last N of the processors this process may use (the runtime sizes the
|
|
/// thread pool and GC for N through <c>DOTNET_PROCESSOR_COUNT</c>) and the server the
|
|
/// others; <c>--client-memory MB</c> sets the GC heap hard limit to 75 % of MB, the limit
|
|
/// .NET derives from a container memory limit.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Processor confinement uses the process affinity mask, which on Windows applies to all
|
|
/// threads of a process. On Linux it applies only to the main thread, so there the
|
|
/// client belongs in a cgroup (for example <c>docker --cpuset-cpus</c>) or under
|
|
/// <c>taskset</c> instead.
|
|
/// </remarks>
|
|
private static async Task<int> RunConstrainedAsync(string[] args)
|
|
{
|
|
var startInfo = new ProcessStartInfo(Environment.ProcessPath!) { UseShellExecute = false };
|
|
foreach (string arg in args)
|
|
{
|
|
startInfo.ArgumentList.Add(arg);
|
|
}
|
|
startInfo.Environment[kConstrainedVariable] = "1";
|
|
if (TryGetArg("client-cores", out string? coresArg))
|
|
{
|
|
if (!OperatingSystem.IsWindows())
|
|
{
|
|
Console.WriteLine(
|
|
"--client-cores is supported on Windows only, where the affinity mask applies to all threads; " +
|
|
"on Linux run the client in a cgroup (docker --cpuset-cpus) or under taskset.");
|
|
return 2;
|
|
}
|
|
// the processors this process may use, which can be a subset of the machine,
|
|
// within the processor group of the process (at most 64)
|
|
using Process self = Process.GetCurrentProcess();
|
|
long allowed = self.ProcessorAffinity;
|
|
var processors = new List<int>();
|
|
for (int bit = 0; bit < 64; bit++)
|
|
{
|
|
if ((allowed & (1L << bit)) != 0)
|
|
{
|
|
processors.Add(bit);
|
|
}
|
|
}
|
|
int cores = Math.Clamp(int.Parse(coresArg!, CultureInfo.InvariantCulture), 1, processors.Count);
|
|
long clientMask = 0;
|
|
foreach (int bit in processors.Skip(processors.Count - cores))
|
|
{
|
|
clientMask |= 1L << bit;
|
|
}
|
|
startInfo.Environment[kClientAffinityVariable] = clientMask.ToString("X", CultureInfo.InvariantCulture);
|
|
startInfo.Environment["DOTNET_PROCESSOR_COUNT"] = cores.ToString(CultureInfo.InvariantCulture);
|
|
if (cores < processors.Count)
|
|
{
|
|
startInfo.Environment[kServerAffinityVariable] =
|
|
(allowed & ~clientMask).ToString("X", CultureInfo.InvariantCulture);
|
|
}
|
|
}
|
|
if (TryGetArg("client-memory", out string? memory))
|
|
{
|
|
long limit = long.Parse(memory!, CultureInfo.InvariantCulture) * 1048576 * 3 / 4;
|
|
startInfo.Environment["DOTNET_GCHeapHardLimit"] = "0x" + limit.ToString("X", CultureInfo.InvariantCulture);
|
|
}
|
|
using Process child = Process.Start(startInfo)!;
|
|
await child.WaitForExitAsync().ConfigureAwait(false);
|
|
return child.ExitCode;
|
|
}
|
|
internal static bool TryGetArg(string name, out string? value)
|
|
{
|
|
return s_args.TryGetValue(name, out value);
|
|
}
|
|
|
|
internal static string Arg(string name, string defaultValue)
|
|
{
|
|
return s_args.TryGetValue(name, out string? value) ? value : defaultValue;
|
|
}
|
|
|
|
internal static int ArgInt(string name, int defaultValue)
|
|
{
|
|
return int.Parse(
|
|
Arg(name, defaultValue.ToString(CultureInfo.InvariantCulture)),
|
|
CultureInfo.InvariantCulture);
|
|
}
|
|
|
|
internal static List<NodeId> GetNodes(ISession session, int count, Type? onlyType = null)
|
|
{
|
|
var result = new List<NodeId>();
|
|
foreach ((Type type, ExpandedNodeId[] ids) in CommonTestWorkers.NodeIdTestSetStaticMassNumeric)
|
|
{
|
|
if (onlyType != null && type != onlyType)
|
|
{
|
|
continue;
|
|
}
|
|
foreach (ExpandedNodeId id in ids)
|
|
{
|
|
if (result.Count == count)
|
|
{
|
|
return result;
|
|
}
|
|
result.Add(ExpandedNodeId.ToNodeId(id, session.NamespaceUris));
|
|
}
|
|
}
|
|
return result;
|
|
}
|
|
|
|
private static ITelemetryContext CreateTelemetry(LogLevel level)
|
|
{
|
|
return DefaultTelemetry.Create(b => b.SetMinimumLevel(level).AddSimpleConsole(o => o.SingleLine = true));
|
|
}
|
|
|
|
private static async Task<int> RunServerAsync()
|
|
{
|
|
ITelemetryContext telemetry = CreateTelemetry(LogLevel.Error);
|
|
var application = new ApplicationInstance(telemetry)
|
|
{
|
|
ApplicationName = "PerfReferenceServer",
|
|
ApplicationType = ApplicationType.Server
|
|
};
|
|
try
|
|
{
|
|
return await RunServerAsync(application, telemetry).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
await application.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private static async Task<int> RunServerAsync(ApplicationInstance application, ITelemetryContext telemetry)
|
|
{
|
|
int port = ArgInt("port", 62555);
|
|
bool diagnostics = ArgInt("diag", 0) != 0;
|
|
bool auditing = ArgInt("audit", 0) != 0;
|
|
// the contract of the o6-automation opcua-benchmarks servers (see NativeWorkerBenchmark)
|
|
bool contract = ArgInt("contract", 0) != 0;
|
|
string pkiRoot = Path.Combine(Path.GetTempPath(), contract ? "uaperf-pki-contractserver" : "uaperf-pki-server");
|
|
string endpointUrl = contract
|
|
? FormattableString.Invariant($"opc.tcp://127.0.0.1:{port}")
|
|
: FormattableString.Invariant($"opc.tcp://localhost:{port}/PerfReferenceServer");
|
|
ArrayOf<CertificateIdentifier> certificates =
|
|
ApplicationConfigurationBuilder.CreateDefaultApplicationCertificates(
|
|
"CN=PerfReferenceServer, O=OPC Foundation, DC=localhost",
|
|
CertificateStoreType.Directory,
|
|
pkiRoot);
|
|
ApplicationConfiguration config = await application
|
|
.Build(contract ? NativeWorkerBenchmark.ContractServerUri : "urn:localhost:UA:PerfReferenceServer", "uri:opcfoundation.org:PerfReferenceServer")
|
|
.SetMaxByteStringLength(4 * 1024 * 1024)
|
|
.SetMaxArrayLength(1024 * 1024)
|
|
.SetMaxMessageSize(16 * 1024 * 1024)
|
|
.AsServer([endpointUrl])
|
|
.AddUnsecurePolicyNone()
|
|
.AddSignPolicies()
|
|
.AddSignAndEncryptPolicies()
|
|
.SetMaxChannelCount(2100)
|
|
.SetDiagnosticsEnabled(diagnostics)
|
|
.SetAuditingEnabled(auditing)
|
|
.SetShutdownDelay(0)
|
|
.AddSecurityConfiguration(certificates, pkiRoot)
|
|
.SetAutoAcceptUntrustedCertificates(true)
|
|
.CreateAsync()
|
|
.ConfigureAwait(false);
|
|
config.ServerConfiguration!.MaxSessionCount = 2000;
|
|
config.ServerConfiguration.MaxSubscriptionCount = 20000;
|
|
config.ServerConfiguration.MaxFailedAuthenticationAttempts = 0;
|
|
config.ServerConfiguration.MaxRequestThreadCount = ArgInt("threads", 200);
|
|
config.ServerConfiguration.MinRequestThreadCount = Math.Min(50, config.ServerConfiguration.MaxRequestThreadCount);
|
|
// sized for the largest pipeline (sessions x inflight), as the o6-automation benchmarks do
|
|
config.ServerConfiguration.MaxQueuedRequestCount = ArgInt("queue", 2000);
|
|
int counters = ArgInt("counters", 0);
|
|
int sampling = ArgInt("sampling", 10);
|
|
if (counters > 0)
|
|
{
|
|
// the subscription capacity workload of o6-automation/opcua-benchmarks
|
|
config.ServerConfiguration.MinPublishingInterval = 1;
|
|
config.ServerConfiguration.PublishingResolution = 1;
|
|
config.ServerConfiguration.MaxPublishingInterval = 86400_000;
|
|
config.ServerConfiguration.MaxNotificationQueueSize = 1;
|
|
config.ServerConfiguration.MaxNotificationsPerPublish = int.MaxValue;
|
|
config.ServerConfiguration.MaxPublishRequestCount = 4;
|
|
config.ServerConfiguration.AvailableSamplingRates = [new SamplingRateGroup(sampling, 1, 0)];
|
|
}
|
|
|
|
if (!await application
|
|
.CheckApplicationInstanceCertificatesAsync(true, CertificateFactory.DefaultLifeTime)
|
|
.ConfigureAwait(false))
|
|
{
|
|
Console.WriteLine("ERROR cert");
|
|
return 1;
|
|
}
|
|
|
|
using Opc.Ua.Server.StandardServer server = counters > 0
|
|
? new CounterServer(
|
|
telemetry,
|
|
counters,
|
|
contract ? NativeWorkerBenchmark.ContractServerUri : CounterNodeManager.Namespace,
|
|
ArgInt("counter-lock", 0) != 0,
|
|
contract && ArgInt("async-node-manager", 1) != 0)
|
|
: contract
|
|
? new BenchmarkContractServer(telemetry, ArgInt("async-node-manager", 1) != 0)
|
|
: new ReferenceServer(telemetry);
|
|
server.TransportBindings = TestTransportBindings.WithAllSchemes();
|
|
await application.StartAsync(server).ConfigureAwait(false);
|
|
Console.WriteLine(FormattableString.Invariant($"READY {endpointUrl} {Environment.ProcessId}"));
|
|
|
|
ProcessStats mark = ProcessStats.Capture();
|
|
using var events = new EventGenerator(server.CurrentInstance);
|
|
string? line;
|
|
while ((line = await Console.In.ReadLineAsync().ConfigureAwait(false)) != null)
|
|
{
|
|
line = line.Trim();
|
|
if (line.StartsWith("events ", StringComparison.Ordinal))
|
|
{
|
|
// events <rate>: start the event generator; events stop: stop it
|
|
string rate = line[7..];
|
|
if (rate == "stop")
|
|
{
|
|
events.Stop();
|
|
}
|
|
else
|
|
{
|
|
events.Start(int.Parse(rate, CultureInfo.InvariantCulture));
|
|
}
|
|
continue;
|
|
}
|
|
switch (line)
|
|
{
|
|
case "mark":
|
|
GC.Collect();
|
|
GC.WaitForPendingFinalizers();
|
|
mark = ProcessStats.Capture();
|
|
(server as CounterServer)?.Counters.Mark();
|
|
events.Mark();
|
|
Console.WriteLine("MARKED");
|
|
break;
|
|
case "report":
|
|
string counterReport = (server as CounterServer)?.Counters.Report(sampling) ?? string.Empty;
|
|
Console.WriteLine("REPORT " + ProcessStats.Capture().Delta(mark) + " " + counterReport + " " + events.Report());
|
|
break;
|
|
case "quit":
|
|
events.Stop();
|
|
await server.StopAsync().ConfigureAwait(false);
|
|
return 0;
|
|
}
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
private static async Task<int> RunDriverAsync(string[] args)
|
|
{
|
|
string scenario = Arg("scenario", "read");
|
|
int sessionCount = ArgInt("sessions", 20);
|
|
int durationSeconds = ArgInt("duration", 20);
|
|
int warmupSeconds = ArgInt("warmup", 10);
|
|
int port = ArgInt("port", 62555 + (Environment.ProcessId % 1000));
|
|
string security = Arg("security", "none");
|
|
if (ArgInt("gc-events", 1) != 0)
|
|
{
|
|
s_gcListener = new GcListener();
|
|
}
|
|
// a constrained client starts the server before it confines itself, so the server
|
|
// does not inherit the client's processors; the server runtime is sized for the
|
|
// processors it keeps (SERVER_DOTNET_PROCESSOR_COUNT still overrides)
|
|
string? clientAffinity = Environment.GetEnvironmentVariable(kClientAffinityVariable);
|
|
string? serverAffinity = Environment.GetEnvironmentVariable(kServerAffinityVariable);
|
|
Dictionary<string, string>? serverEnvironment = serverAffinity == null
|
|
? null
|
|
: new Dictionary<string, string>(StringComparer.Ordinal)
|
|
{
|
|
["DOTNET_PROCESSOR_COUNT"] = BitOperations.PopCount(Convert.ToUInt64(serverAffinity, 16))
|
|
.ToString(CultureInfo.InvariantCulture)
|
|
};
|
|
(Process startedServer, BlockingCollection<string> startedLines) = StartServerChild(
|
|
FormattableString.Invariant(
|
|
$"server --port {port} --diag {Arg("diag", "0")} --audit {Arg("audit", "0")} --threads {Arg("threads", "200")} --counters {(scenario == "counters" ? Arg("nodes", "1") : "0")} --sampling {Arg("sampling", "10")} --queue {Arg("queue", "2000")}"),
|
|
serverEnvironment);
|
|
using Process server = startedServer;
|
|
using BlockingCollection<string> serverLines = startedLines;
|
|
if (serverAffinity != null && OperatingSystem.IsWindows())
|
|
{
|
|
// the server keeps the processors the constrained client does not have
|
|
server.ProcessorAffinity = (nint)Convert.ToInt64(serverAffinity, 16);
|
|
}
|
|
if (clientAffinity != null && OperatingSystem.IsWindows())
|
|
{
|
|
// the process mask applies to the threads the runtime has already started, too
|
|
using Process self = Process.GetCurrentProcess();
|
|
self.ProcessorAffinity = (nint)Convert.ToInt64(clientAffinity, 16);
|
|
}
|
|
if (Environment.GetEnvironmentVariable(kConstrainedVariable) != null)
|
|
{
|
|
Console.WriteLine(FormattableString.Invariant(
|
|
$"constrained client: processors={Environment.ProcessorCount} mask=0x{clientAffinity ?? "all"} ") +
|
|
FormattableString.Invariant(
|
|
$"server mask=0x{serverAffinity ?? "all"} heapLimitMB={GC.GetGCMemoryInfo().TotalAvailableMemoryBytes / 1048576}"));
|
|
}
|
|
string ready = serverLines.Take();
|
|
if (!ready.StartsWith("READY", StringComparison.Ordinal))
|
|
{
|
|
Console.WriteLine(ready);
|
|
return 1;
|
|
}
|
|
string url = ready.Split(' ')[1];
|
|
Console.WriteLine(FormattableString.Invariant($"server pid={server.Id} url={url}"));
|
|
|
|
ITelemetryContext telemetry = CreateTelemetry(LogLevel.Error);
|
|
using var client = new ClientFixture(telemetry) { SessionTimeout = 60000, OperationTimeout = 60000 };
|
|
await client
|
|
.LoadClientConfigurationAsync(Path.Combine(Path.GetTempPath(), "uaperf-pki-client"), "PerfClient")
|
|
.ConfigureAwait(false);
|
|
// The test fixture asks the server for diagnostics in every response;
|
|
// an application does not by default.
|
|
client.SessionFactory.ReturnDiagnostics = DiagnosticsMasks.None;
|
|
string policy = security == "none" ? SecurityPolicies.None : SecurityPolicies.Basic256Sha256;
|
|
MessageSecurityMode mode = security switch
|
|
{
|
|
"sign" => MessageSecurityMode.Sign,
|
|
"encrypt" => MessageSecurityMode.SignAndEncrypt,
|
|
_ => MessageSecurityMode.None
|
|
};
|
|
ArrayOf<EndpointDescription> endpoints = await client.GetEndpointsAsync(new Uri(url)).ConfigureAwait(false);
|
|
var selected = new List<EndpointDescription>();
|
|
foreach (EndpointDescription endpoint in endpoints)
|
|
{
|
|
if (endpoint.SecurityPolicyUri == policy && endpoint.SecurityMode == mode)
|
|
{
|
|
selected.Add(endpoint);
|
|
}
|
|
}
|
|
|
|
// managed (default): ManagedSession with the default V2 subscription engine.
|
|
// classic: Session from the default session factory (classic subscription engine).
|
|
// shared: classic Sessions whose secure channels come from one ClientChannelManager.
|
|
string clientKind = Arg("client", "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);
|
|
// sub: --writers adds dedicated writer sessions after the subscriber sessions
|
|
int writerCount = scenario == "sub" ? ArgInt("writers", 0) : 0;
|
|
var sessions = new ISession[sessionCount + writerCount];
|
|
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,
|
|
sessions.Length,
|
|
new ParallelOptions { MaxDegreeOfParallelism = connectParallel },
|
|
async (i, ct) => sessions[i] = clientKind switch
|
|
{
|
|
"managed" => await new ManagedSessionBuilder(client.Config, telemetry)
|
|
.UseEndpoint(endpointForBuilder!)
|
|
.WithSessionTimeout(TimeSpan.FromSeconds(60))
|
|
.WithPoolNotifications(ArgInt("pool-notifications", 0) != 0)
|
|
.ConnectAsync(ct)
|
|
.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);
|
|
Console.WriteLine(FormattableString.Invariant(
|
|
$"connected {sessions.Length} {clientKind} sessions ({security}) in {connectTime.ElapsedMilliseconds} ms"));
|
|
if (TryGetArg("publish-requests", out string? publishRequests))
|
|
{
|
|
// outstanding Publish requests per session (the client default adapts to the subscriptions)
|
|
int count = int.Parse(publishRequests!, CultureInfo.InvariantCulture);
|
|
foreach (ISession session in sessions)
|
|
{
|
|
session.MaxPublishRequestCount = Math.Max(session.MaxPublishRequestCount, count);
|
|
session.MinPublishRequestCount = count;
|
|
}
|
|
}
|
|
|
|
Workload workload = scenario == "counters"
|
|
? new CounterWorkload(sessions[0], ArgInt("nodes", 1), ArgInt("sampling", 10), ArgInt("pub", 1000))
|
|
: scenario == "events"
|
|
? new EventWorkload(sessions, ArgInt("subs", 15), ArgInt("pub", 100), ArgInt("queue-size", 10000), ArgInt("event-rate", 0))
|
|
: scenario == "sub"
|
|
? new SubscribeWorkload(
|
|
sessions[..sessionCount],
|
|
writerCount == 0 ? [sessions[0]] : sessions[sessionCount..],
|
|
ArgInt("nodes", 100),
|
|
ArgInt("subs", 5),
|
|
ArgInt("pub", 100),
|
|
ArgInt("write-interval", 100),
|
|
Arg("callback", "fast") == "item",
|
|
ArgInt("write-nodes", 0),
|
|
Arg("node-type", "int") == "all",
|
|
ArgInt("write-batch", 0))
|
|
: new RequestWorkload(scenario, sessions, ArgInt("nodes", 100), ArgInt("inflight", 1));
|
|
await workload.SetupAsync().ConfigureAwait(false);
|
|
|
|
using var stop = new CancellationTokenSource();
|
|
Task run = workload.RunAsync(stop.Token);
|
|
if (workload is EventWorkload eventWorkload)
|
|
{
|
|
await server.StandardInput.WriteLineAsync(
|
|
FormattableString.Invariant($"events {eventWorkload.Rate}")).ConfigureAwait(false);
|
|
}
|
|
await Task.Delay(TimeSpan.FromSeconds(warmupSeconds)).ConfigureAwait(false);
|
|
|
|
bool traceClient = Arg("trace-side", "server") == "client";
|
|
using Process? tracer = StartTracer(
|
|
traceClient ? Environment.ProcessId : server.Id,
|
|
durationSeconds,
|
|
out string? trace);
|
|
if (tracer != null)
|
|
{
|
|
await Task.Delay(2000).ConfigureAwait(false);
|
|
}
|
|
|
|
await server.StandardInput.WriteLineAsync("mark").ConfigureAwait(false);
|
|
Take(serverLines, "MARKED");
|
|
GC.Collect();
|
|
GC.WaitForPendingFinalizers();
|
|
s_gcListener?.Mark();
|
|
ProcessStats clientMark = ProcessStats.Capture();
|
|
workload.ResetStats();
|
|
var window = Stopwatch.StartNew();
|
|
TimeSpan clientCpuStart = Process.GetCurrentProcess().TotalProcessorTime;
|
|
await Task.Delay(TimeSpan.FromSeconds(durationSeconds)).ConfigureAwait(false);
|
|
string clientProcess = ProcessStats.Capture().Delta(clientMark);
|
|
// the window ends here; GC events stamped before this arrive later and still count
|
|
s_gcListener?.End();
|
|
await server.StandardInput.WriteLineAsync("report").ConfigureAwait(false);
|
|
string report = Take(serverLines, "REPORT")[7..];
|
|
// freeze the client counts at the end of the server window
|
|
workload.OnServerReport(report);
|
|
double operations = workload.Operations;
|
|
(double Count, string Name)? secondary = workload.Secondary;
|
|
double clientCores = (Process.GetCurrentProcess().TotalProcessorTime - clientCpuStart).TotalSeconds /
|
|
window.Elapsed.TotalSeconds;
|
|
if (workload is EventWorkload)
|
|
{
|
|
await server.StandardInput.WriteLineAsync("events stop").ConfigureAwait(false);
|
|
}
|
|
string clientReport = workload.Report(window.Elapsed.TotalSeconds) +
|
|
FormattableString.Invariant($" clientCpuCores={clientCores:F2}");
|
|
await stop.CancelAsync().ConfigureAwait(false);
|
|
try
|
|
{
|
|
await run.ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// the workload stops by cancellation
|
|
}
|
|
|
|
if (tracer != null)
|
|
{
|
|
await tracer.WaitForExitAsync().ConfigureAwait(false);
|
|
Console.WriteLine("trace: " + trace);
|
|
}
|
|
|
|
string? clientGc = s_gcListener == null ? null : await s_gcListener.ReportAsync().ConfigureAwait(false);
|
|
var result = new StringBuilder();
|
|
result.Append(CultureInfo.InvariantCulture, $"RESULT scenario={scenario} security={security} sessions={sessionCount} ")
|
|
.Append(string.Join(' ', args.Skip(1)))
|
|
.Append('\n')
|
|
.Append(" client: ").Append(clientReport).Append('\n')
|
|
.Append(" server: ").Append(report).Append('\n')
|
|
.Append(" client-process: ").Append(clientProcess).Append('\n');
|
|
if (clientGc != null)
|
|
{
|
|
result.Append(" client-gc: ").Append(clientGc).Append('\n');
|
|
}
|
|
(double serverMb, double serverCpuMs) = ParseProcessReport(report);
|
|
if (operations > 0)
|
|
{
|
|
result.Append(
|
|
CultureInfo.InvariantCulture,
|
|
$" per-op: allocB={serverMb * 1048576 / operations:F0} cpuUs={serverCpuMs * 1000 / operations:F1} ({workload.OperationName})\n");
|
|
}
|
|
if (secondary is (double secondaryCount, string secondaryName) && secondaryCount > 0)
|
|
{
|
|
result.Append(
|
|
CultureInfo.InvariantCulture,
|
|
$" per-op2: allocB={serverMb * 1048576 / secondaryCount:F0} cpuUs={serverCpuMs * 1000 / secondaryCount:F1} ({secondaryName})\n");
|
|
}
|
|
if (operations > 0)
|
|
{
|
|
(double allocatedMb, double cpuMs) = ParseProcessReport(clientProcess);
|
|
result.Append(
|
|
CultureInfo.InvariantCulture,
|
|
$" client-per-op: allocB={allocatedMb * 1048576 / operations:F0} cpuUs={cpuMs * 1000 / operations:F1} ({workload.OperationName})\n");
|
|
}
|
|
Console.Write(result.ToString());
|
|
if (s_args.TryGetValue("out", out string? outFile))
|
|
{
|
|
await File.AppendAllTextAsync(outFile, result.ToString()).ConfigureAwait(false);
|
|
}
|
|
|
|
await Parallel.ForEachAsync(sessions, async (session, ct) =>
|
|
{
|
|
try
|
|
{
|
|
await session.CloseAsync(ct).ConfigureAwait(false);
|
|
}
|
|
catch (ServiceResultException)
|
|
{
|
|
// the server may already be shutting the session down
|
|
}
|
|
session.Dispose();
|
|
}).ConfigureAwait(false);
|
|
await server.StandardInput.WriteLineAsync("quit").ConfigureAwait(false);
|
|
if (!server.WaitForExit(10000))
|
|
{
|
|
server.Kill();
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Starts the harness itself as the server child; <c>SERVER_</c> variables are passed without the prefix.
|
|
/// </summary>
|
|
internal static (Process Server, BlockingCollection<string> Lines) StartServerChild(
|
|
string arguments,
|
|
IReadOnlyDictionary<string, string>? defaultEnvironment = null)
|
|
{
|
|
var startInfo = new ProcessStartInfo(Environment.ProcessPath!, arguments)
|
|
{
|
|
RedirectStandardInput = true,
|
|
RedirectStandardOutput = true,
|
|
UseShellExecute = false
|
|
};
|
|
// GC and processor settings of the harness configure the measured client; the server
|
|
// inherits the environment, so they are removed and only SERVER_ variables configure it
|
|
foreach (string name in startInfo.Environment.Keys.ToList())
|
|
{
|
|
if (name.StartsWith("DOTNET_GC", StringComparison.OrdinalIgnoreCase) ||
|
|
name.StartsWith("COMPlus_GC", StringComparison.OrdinalIgnoreCase) ||
|
|
name.StartsWith("DOTNET_ThreadPool_", StringComparison.OrdinalIgnoreCase) ||
|
|
name.Equals("DOTNET_PROCESSOR_COUNT", StringComparison.OrdinalIgnoreCase) ||
|
|
name.StartsWith("PERF_", StringComparison.Ordinal))
|
|
{
|
|
startInfo.Environment.Remove(name);
|
|
}
|
|
}
|
|
if (defaultEnvironment != null)
|
|
{
|
|
foreach (KeyValuePair<string, string> variable in defaultEnvironment)
|
|
{
|
|
startInfo.Environment[variable.Key] = variable.Value;
|
|
}
|
|
}
|
|
foreach (DictionaryEntry variable in Environment.GetEnvironmentVariables())
|
|
{
|
|
string name = (string)variable.Key;
|
|
if (name.StartsWith("SERVER_", StringComparison.Ordinal))
|
|
{
|
|
startInfo.Environment[name[7..]] = (string?)variable.Value;
|
|
}
|
|
}
|
|
Process server = Process.Start(startInfo)!;
|
|
var lines = new BlockingCollection<string>();
|
|
// LongRunning runs the pump on a dedicated thread: it blocks in ReadLine for the whole
|
|
// run, and on a pool thread it occupies the only minimum worker of a one-processor
|
|
// client, so every other work item waits for the pool's starvation injection
|
|
// (~0.5-1 s stalls). The task keeps a pump exception at shutdown (the collection or
|
|
// process disposed under it) from terminating the harness.
|
|
_ = Task.Factory.StartNew(
|
|
() => PumpServerOutput(server, lines),
|
|
CancellationToken.None,
|
|
TaskCreationOptions.LongRunning,
|
|
TaskScheduler.Default);
|
|
return (server, lines);
|
|
}
|
|
|
|
private static void PumpServerOutput(Process server, BlockingCollection<string> lines)
|
|
{
|
|
string? line;
|
|
while ((line = server.StandardOutput.ReadLine()) != null)
|
|
{
|
|
if (line.StartsWith("READY", StringComparison.Ordinal) ||
|
|
line.StartsWith("REPORT", StringComparison.Ordinal) ||
|
|
line.StartsWith("MARKED", StringComparison.Ordinal) ||
|
|
line.StartsWith("ERROR", StringComparison.Ordinal))
|
|
{
|
|
lines.Add(line);
|
|
}
|
|
else if (s_args.ContainsKey("verbose"))
|
|
{
|
|
Console.WriteLine("[server] " + line);
|
|
}
|
|
}
|
|
lines.CompleteAdding();
|
|
}
|
|
|
|
internal static Process? StartTracer(int processId, int durationSeconds, out string? trace)
|
|
{
|
|
if (!s_args.TryGetValue("trace", out trace))
|
|
{
|
|
return null;
|
|
}
|
|
string providers = Arg("trace-profile", "cpu") switch
|
|
{
|
|
// allocation sampling (GCAllocationTick, about every 100 KB) and GC events
|
|
"alloc" => "--providers Microsoft-Windows-DotNETRuntime:0x1:5",
|
|
"gc" => "--profile gc-verbose",
|
|
// the Contention keyword
|
|
"contention" => "--providers Microsoft-Windows-DotNETRuntime:0x4000:4",
|
|
_ => "--profile dotnet-sampled-thread-time"
|
|
};
|
|
return Process.Start(new ProcessStartInfo(
|
|
"dotnet-trace",
|
|
FormattableString.Invariant(
|
|
$"collect --process-id {processId} {providers} -o \"{trace}\" --duration 00:00:{durationSeconds:D2}"))
|
|
{
|
|
UseShellExecute = false,
|
|
RedirectStandardOutput = true,
|
|
RedirectStandardError = true
|
|
});
|
|
}
|
|
|
|
internal static string Take(BlockingCollection<string> lines, string prefix)
|
|
{
|
|
while (true)
|
|
{
|
|
string line = lines.Take();
|
|
if (line.StartsWith(prefix, StringComparison.Ordinal))
|
|
{
|
|
return line;
|
|
}
|
|
}
|
|
}
|
|
|
|
private static (double AllocatedMb, double CpuMs) ParseProcessReport(string report)
|
|
{
|
|
Dictionary<string, string> values = report
|
|
.Split(' ')
|
|
.Select(p => p.Split('='))
|
|
.Where(p => p.Length == 2)
|
|
.ToDictionary(p => p[0], p => p[1], StringComparer.Ordinal);
|
|
return (
|
|
double.Parse(values["allocMB"], CultureInfo.InvariantCulture),
|
|
double.Parse(values["cpuMs"], CultureInfo.InvariantCulture));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Process-wide counters, captured at the start and end of the measurement window.
|
|
/// </summary>
|
|
internal sealed record ProcessStats(
|
|
long Allocated,
|
|
int Gen0,
|
|
int Gen1,
|
|
int Gen2,
|
|
TimeSpan Pause,
|
|
TimeSpan Cpu,
|
|
long Contention,
|
|
long Timestamp,
|
|
int Threads,
|
|
long WorkingSet,
|
|
long Heap)
|
|
{
|
|
public static ProcessStats Capture()
|
|
{
|
|
using var process = Process.GetCurrentProcess();
|
|
return new ProcessStats(
|
|
GC.GetTotalAllocatedBytes(true),
|
|
GC.CollectionCount(0),
|
|
GC.CollectionCount(1),
|
|
GC.CollectionCount(2),
|
|
GC.GetTotalPauseDuration(),
|
|
process.TotalProcessorTime,
|
|
Monitor.LockContentionCount,
|
|
Stopwatch.GetTimestamp(),
|
|
ThreadPool.ThreadCount,
|
|
process.WorkingSet64,
|
|
GC.GetGCMemoryInfo().HeapSizeBytes);
|
|
}
|
|
|
|
public string Delta(ProcessStats mark)
|
|
{
|
|
double seconds = Stopwatch.GetElapsedTime(mark.Timestamp, Timestamp).TotalSeconds;
|
|
TimeSpan cpu = Cpu - mark.Cpu;
|
|
return FormattableString.Invariant(
|
|
$"sec={seconds:F2} allocMB={(Allocated - mark.Allocated) / 1048576.0:F1} gen0={Gen0 - mark.Gen0} ") +
|
|
FormattableString.Invariant(
|
|
$"gen1={Gen1 - mark.Gen1} gen2={Gen2 - mark.Gen2} pauseMs={(Pause - mark.Pause).TotalMilliseconds:F0} ") +
|
|
FormattableString.Invariant(
|
|
$"cpuMs={cpu.TotalMilliseconds:F0} cpuCores={cpu.TotalSeconds / seconds:F2} ") +
|
|
FormattableString.Invariant(
|
|
$"contention={Contention - mark.Contention} threads={Threads} wsMB={WorkingSet / 1048576} heapMB={Heap / 1048576}");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Latency histogram (Stopwatch ticks) recorded by many threads without allocating:
|
|
/// log-linear buckets with 64 steps per power of two (at most ~1.6 % error), so the
|
|
/// recorder does not add allocations, promotions or large arrays to the measured client.
|
|
/// </summary>
|
|
internal sealed class LatencyRecorder
|
|
{
|
|
public void Record(long ticks)
|
|
{
|
|
if (ticks < 0)
|
|
{
|
|
ticks = 0;
|
|
}
|
|
Interlocked.Increment(ref m_buckets[Index(ticks)]);
|
|
long max = Volatile.Read(ref m_max);
|
|
while (ticks > max)
|
|
{
|
|
long seen = Interlocked.CompareExchange(ref m_max, ticks, max);
|
|
if (seen == max)
|
|
{
|
|
break;
|
|
}
|
|
max = seen;
|
|
}
|
|
}
|
|
|
|
public void Clear()
|
|
{
|
|
for (int i = 0; i < m_buckets.Length; i++)
|
|
{
|
|
Interlocked.Exchange(ref m_buckets[i], 0);
|
|
}
|
|
Interlocked.Exchange(ref m_max, 0);
|
|
}
|
|
|
|
public string Percentiles()
|
|
{
|
|
long[] counts = new long[m_buckets.Length];
|
|
long total = 0;
|
|
for (int i = 0; i < counts.Length; i++)
|
|
{
|
|
counts[i] = Volatile.Read(ref m_buckets[i]);
|
|
total += counts[i];
|
|
}
|
|
if (total == 0)
|
|
{
|
|
return "n=0";
|
|
}
|
|
double Ms(double quantile)
|
|
{
|
|
long rank = Math.Min(total - 1, (long)(quantile * total));
|
|
long seen = 0;
|
|
for (int i = 0; i < counts.Length; i++)
|
|
{
|
|
seen += counts[i];
|
|
if (seen > rank)
|
|
{
|
|
return Value(i) * 1000.0 / Stopwatch.Frequency;
|
|
}
|
|
}
|
|
return Volatile.Read(ref m_max) * 1000.0 / Stopwatch.Frequency;
|
|
}
|
|
return FormattableString.Invariant(
|
|
$"n={total} p50={Ms(0.5):F3}ms p90={Ms(0.9):F3}ms p99={Ms(0.99):F3}ms ") +
|
|
FormattableString.Invariant(
|
|
$"p999={Ms(0.999):F3}ms max={Volatile.Read(ref m_max) * 1000.0 / Stopwatch.Frequency:F3}ms");
|
|
}
|
|
|
|
private static int Index(long ticks)
|
|
{
|
|
if (ticks < 2 * kSteps)
|
|
{
|
|
return (int)ticks;
|
|
}
|
|
int shift = 63 - BitOperations.LeadingZeroCount((ulong)ticks) - kBits;
|
|
return (2 * kSteps) + ((shift - 1) * kSteps) + (int)((ticks >> shift) - kSteps);
|
|
}
|
|
|
|
private static double Value(int index)
|
|
{
|
|
if (index < 2 * kSteps)
|
|
{
|
|
return index;
|
|
}
|
|
int shift = ((index - (2 * kSteps)) / kSteps) + 1;
|
|
long mantissa = ((index - (2 * kSteps)) % kSteps) + kSteps;
|
|
// bucket midpoint
|
|
return (mantissa << shift) + ((1L << shift) / 2.0);
|
|
}
|
|
|
|
private const int kBits = 6;
|
|
private const int kSteps = 1 << kBits;
|
|
private readonly long[] m_buckets = new long[(2 * kSteps) + (64 * kSteps)];
|
|
private long m_max;
|
|
}
|
|
|
|
/// <summary>
|
|
/// A load pattern driven against the server.
|
|
/// </summary>
|
|
internal abstract class Workload
|
|
{
|
|
public abstract double Operations { get; }
|
|
|
|
public abstract string OperationName { get; }
|
|
|
|
/// <summary>
|
|
/// A second operation count to report the server cost against, such as the writes
|
|
/// that caused the notifications.
|
|
/// </summary>
|
|
public virtual (double Count, string Name)? Secondary => null;
|
|
|
|
public abstract Task SetupAsync();
|
|
|
|
/// <summary>
|
|
/// Receives the server report of the window as soon as it arrives, before
|
|
/// <see cref="Operations"/>, <see cref="Secondary"/> and <see cref="Report"/> are read;
|
|
/// workloads that keep counting after the window freeze their counts here.
|
|
/// </summary>
|
|
public virtual void OnServerReport(string report)
|
|
{
|
|
}
|
|
|
|
public abstract Task RunAsync(CancellationToken ct);
|
|
|
|
public abstract void ResetStats();
|
|
|
|
public abstract string Report(double seconds);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Closed-loop Read, Write or Browse requests from every session.
|
|
/// </summary>
|
|
internal sealed class RequestWorkload : Workload
|
|
{
|
|
public RequestWorkload(string kind, ISession[] sessions, int nodes, int inflight)
|
|
{
|
|
m_kind = kind;
|
|
m_sessions = sessions;
|
|
m_nodes = nodes;
|
|
m_inflight = inflight;
|
|
}
|
|
|
|
public override double Operations => Interlocked.Read(ref m_requests);
|
|
|
|
public override string OperationName => "request";
|
|
|
|
public override Task SetupAsync()
|
|
{
|
|
List<NodeId> nodes = Program.GetNodes(m_sessions[0], m_nodes, m_kind == "write" ? typeof(int) : null);
|
|
m_read = nodes.Select(n => new ReadValueId { NodeId = n, AttributeId = Attributes.Value }).ToArray();
|
|
m_write = nodes.Select(n => new WriteValue
|
|
{
|
|
NodeId = n,
|
|
AttributeId = Attributes.Value,
|
|
Value = new DataValue(new Variant(42))
|
|
}).ToArray();
|
|
m_browse = new[]
|
|
{
|
|
new BrowseDescription
|
|
{
|
|
NodeId = ObjectIds.Server,
|
|
BrowseDirection = BrowseDirection.Forward,
|
|
ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences,
|
|
IncludeSubtypes = true,
|
|
ResultMask = (uint)BrowseResultMask.All
|
|
}
|
|
};
|
|
Console.WriteLine(FormattableString.Invariant(
|
|
$"{m_kind}: {nodes.Count} nodes/request, inflight {m_inflight}/session"));
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
public override async Task RunAsync(CancellationToken ct)
|
|
{
|
|
var tasks = new List<Task>();
|
|
foreach (ISession session in m_sessions)
|
|
{
|
|
for (int i = 0; i < m_inflight; i++)
|
|
{
|
|
tasks.Add(Task.Run(() => LoopAsync(session, ct), CancellationToken.None));
|
|
}
|
|
}
|
|
await Task.WhenAll(tasks).ConfigureAwait(false);
|
|
}
|
|
|
|
public override void ResetStats()
|
|
{
|
|
Interlocked.Increment(ref m_epoch);
|
|
m_latency.Clear();
|
|
Interlocked.Exchange(ref m_requests, 0);
|
|
Interlocked.Exchange(ref m_errors, 0);
|
|
}
|
|
|
|
public override string Report(double seconds)
|
|
{
|
|
long requests = Interlocked.Read(ref m_requests);
|
|
int valuesPerRequest = m_kind == "browse" ? 1 : m_nodes;
|
|
return FormattableString.Invariant(
|
|
$"req/s={requests / seconds:F0} values/s={requests * valuesPerRequest / seconds:F0} ") +
|
|
FormattableString.Invariant($"errors={Interlocked.Read(ref m_errors)} latency {m_latency.Percentiles()}");
|
|
}
|
|
|
|
private async Task LoopAsync(ISession session, CancellationToken ct)
|
|
{
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
long start = Stopwatch.GetTimestamp();
|
|
int epoch = Volatile.Read(ref m_epoch);
|
|
try
|
|
{
|
|
switch (m_kind)
|
|
{
|
|
case "write":
|
|
await session.WriteAsync(null, m_write, ct).ConfigureAwait(false);
|
|
break;
|
|
case "browse":
|
|
await session.BrowseAsync(null, null, 0, m_browse, ct).ConfigureAwait(false);
|
|
break;
|
|
default:
|
|
await session.ReadAsync(null, 0, TimestampsToReturn.Both, m_read, ct).ConfigureAwait(false);
|
|
break;
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
return;
|
|
}
|
|
catch (ServiceResultException ex)
|
|
{
|
|
if (Interlocked.Increment(ref m_errors) < 5)
|
|
{
|
|
Console.WriteLine("error: " + ex.Message);
|
|
}
|
|
continue;
|
|
}
|
|
if (epoch == Volatile.Read(ref m_epoch))
|
|
{
|
|
m_latency.Record(Stopwatch.GetTimestamp() - start);
|
|
Interlocked.Increment(ref m_requests);
|
|
}
|
|
}
|
|
}
|
|
|
|
private readonly string m_kind;
|
|
private readonly ISession[] m_sessions;
|
|
private readonly int m_nodes;
|
|
private readonly int m_inflight;
|
|
private readonly LatencyRecorder m_latency = new();
|
|
private int m_epoch;
|
|
private long m_requests;
|
|
private long m_errors;
|
|
private ArrayOf<ReadValueId> m_read;
|
|
private ArrayOf<WriteValue> m_write;
|
|
private ArrayOf<BrowseDescription> m_browse;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Data-change subscriptions on every session and one writer changing the monitored values.
|
|
/// The written value is a counter, so each notification maps back to the time of its write.
|
|
/// </summary>
|
|
internal sealed class SubscribeWorkload : Workload
|
|
{
|
|
public SubscribeWorkload(
|
|
ISession[] sessions,
|
|
ISession[] writers,
|
|
int nodes,
|
|
int subscriptions,
|
|
int publishingInterval,
|
|
int writeInterval,
|
|
bool itemCallback,
|
|
int writeNodes,
|
|
bool allTypes,
|
|
int writeBatch)
|
|
{
|
|
m_sessions = sessions;
|
|
m_writers = writers;
|
|
m_nodes = nodes;
|
|
m_subscriptions = subscriptions;
|
|
m_publishingInterval = publishingInterval;
|
|
m_writeInterval = writeInterval;
|
|
m_itemCallback = itemCallback;
|
|
m_writeNodes = writeNodes;
|
|
m_allTypes = allTypes;
|
|
m_writeBatch = writeBatch;
|
|
}
|
|
|
|
public override double Operations => m_final.Notifications;
|
|
|
|
public override string OperationName => "notification";
|
|
|
|
public override (double Count, string Name)? Secondary => (m_final.Values, "written value");
|
|
|
|
/// <summary>
|
|
/// Values per Write request: all written nodes, or slices of <c>--write-batch</c>.
|
|
/// </summary>
|
|
private int BatchSize => m_writeBatch > 0 ? Math.Min(m_writeBatch, WrittenNodes) : WrittenNodes;
|
|
|
|
/// <summary>
|
|
/// The number of monitored nodes the writer changes: all of them, or the first
|
|
/// <c>--write-nodes</c> (a large subscription in which few items change).
|
|
/// </summary>
|
|
private int WrittenNodes => m_writeNodes > 0 ? Math.Min(m_writeNodes, m_nodeIds.Count) : m_nodeIds.Count;
|
|
|
|
public override async Task SetupAsync()
|
|
{
|
|
if (m_allTypes)
|
|
{
|
|
// the mixed Int16..UInt64 node set of the Sessions LoadTest
|
|
m_nodeTypes = [];
|
|
foreach ((Type type, ExpandedNodeId[] ids) in CommonTestWorkers.NodeIdTestSetStaticMassNumeric)
|
|
{
|
|
foreach (ExpandedNodeId id in ids)
|
|
{
|
|
if (m_nodeTypes.Count < m_nodes)
|
|
{
|
|
m_nodeTypes.Add(type);
|
|
}
|
|
}
|
|
}
|
|
m_nodeIds = Program.GetNodes(m_sessions[0], m_nodes).ToArray();
|
|
}
|
|
else
|
|
{
|
|
m_nodeIds = Program.GetNodes(m_sessions[0], m_nodes, typeof(int)).ToArray();
|
|
}
|
|
var setupTime = Stopwatch.StartNew();
|
|
await Parallel.ForEachAsync(
|
|
m_sessions,
|
|
new ParallelOptions { MaxDegreeOfParallelism = 16 },
|
|
async (session, ct) =>
|
|
{
|
|
for (int i = 0; i < m_subscriptions; i++)
|
|
{
|
|
await CreateSubscriptionAsync(session, ct).ConfigureAwait(false);
|
|
}
|
|
}).ConfigureAwait(false);
|
|
Console.WriteLine(FormattableString.Invariant(
|
|
$"sub: {m_sessions.Length} sessions x {m_subscriptions} subs x {m_nodeIds.Count} items, ") +
|
|
FormattableString.Invariant(
|
|
$"pub {m_publishingInterval} ms, {m_writers.Length} writer(s) x {BatchSize} of {WrittenNodes} items ") +
|
|
FormattableString.Invariant(
|
|
$"{(m_writeInterval > 0 ? $"every {m_writeInterval} ms" : "closed loop")}; setup {setupTime.ElapsedMilliseconds} ms"));
|
|
}
|
|
|
|
public override async Task RunAsync(CancellationToken ct)
|
|
{
|
|
var tasks = new List<Task>();
|
|
for (int i = 0; i < m_writers.Length; i++)
|
|
{
|
|
ISession writer = m_writers[i];
|
|
// writers start on different slices when they write batches
|
|
int first = i * BatchSize % WrittenNodes;
|
|
tasks.Add(Task.Run(() => WriteLoopAsync(writer, first, ct), CancellationToken.None));
|
|
}
|
|
await Task.WhenAll(tasks).ConfigureAwait(false);
|
|
}
|
|
|
|
private async Task WriteLoopAsync(ISession writer, int first, CancellationToken ct)
|
|
{
|
|
using PeriodicTimer? timer = m_writeInterval > 0
|
|
? new PeriodicTimer(TimeSpan.FromMilliseconds(m_writeInterval))
|
|
: null;
|
|
int batch = BatchSize;
|
|
while (timer == null ? !ct.IsCancellationRequested : await timer.WaitForNextTickAsync(ct).ConfigureAwait(false))
|
|
{
|
|
int counter = Interlocked.Increment(ref m_counter);
|
|
var values = new WriteValue[batch];
|
|
for (int i = 0; i < values.Length; i++)
|
|
{
|
|
int index = (first + i) % WrittenNodes;
|
|
values[i] = new WriteValue
|
|
{
|
|
NodeId = m_nodeIds[index],
|
|
AttributeId = Attributes.Value,
|
|
Value = new DataValue(ToVariant(index, counter))
|
|
};
|
|
}
|
|
first = (first + batch) % WrittenNodes;
|
|
long start = Stopwatch.GetTimestamp();
|
|
Volatile.Write(ref m_writeTicks[counter & 0xFFFF], start);
|
|
WriteResponse response;
|
|
try
|
|
{
|
|
response = await writer.WriteAsync(null, values, ct).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
return;
|
|
}
|
|
catch (ServiceResultException ex)
|
|
{
|
|
if (Interlocked.Increment(ref m_writeErrors) < 5)
|
|
{
|
|
Console.WriteLine("write error: " + ex.Message);
|
|
}
|
|
continue;
|
|
}
|
|
m_writeLatency.Record(Stopwatch.GetTimestamp() - start);
|
|
// only values the server accepted change and notify
|
|
int written = 0;
|
|
StatusCode rejected = StatusCodes.BadUnexpectedError;
|
|
for (int i = 0; i < response.Results.Count; i++)
|
|
{
|
|
if (StatusCode.IsGood(response.Results[i]))
|
|
{
|
|
written++;
|
|
}
|
|
else
|
|
{
|
|
rejected = response.Results[i];
|
|
}
|
|
}
|
|
if (written < batch && Interlocked.Increment(ref m_writeErrors) < 5)
|
|
{
|
|
Console.WriteLine(FormattableString.Invariant(
|
|
$"write error: {batch - written} of {batch} values rejected ({rejected})"));
|
|
}
|
|
Interlocked.Increment(ref m_writes);
|
|
Interlocked.Add(ref m_writtenValues, written);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// The counter as the node's type; 16-bit types wrap, Int32 nodes carry the full
|
|
/// counter for the latency measurement.
|
|
/// </summary>
|
|
private Variant ToVariant(int index, int counter)
|
|
{
|
|
if (m_nodeTypes == null)
|
|
{
|
|
return new Variant(counter);
|
|
}
|
|
Type type = m_nodeTypes[index];
|
|
return type == typeof(short) ? new Variant((short)(counter & 0x7FFF))
|
|
: type == typeof(ushort) ? new Variant((ushort)(counter & 0x7FFF))
|
|
: type == typeof(int) ? new Variant(counter)
|
|
: type == typeof(uint) ? new Variant((uint)counter)
|
|
: type == typeof(long) ? new Variant((long)counter)
|
|
: new Variant((ulong)counter);
|
|
}
|
|
|
|
public override void ResetStats()
|
|
{
|
|
m_latency.Clear();
|
|
m_writeLatency.Clear();
|
|
Interlocked.Exchange(ref m_notifications, 0);
|
|
Interlocked.Exchange(ref m_writes, 0);
|
|
Interlocked.Exchange(ref m_writtenValues, 0);
|
|
Interlocked.Exchange(ref m_writeErrors, 0);
|
|
}
|
|
|
|
public override void OnServerReport(string report)
|
|
{
|
|
// closed-loop writers keep going: report the counts at the end of the server window
|
|
m_final = (
|
|
Interlocked.Read(ref m_notifications),
|
|
Interlocked.Read(ref m_writes),
|
|
Interlocked.Read(ref m_writtenValues),
|
|
Interlocked.Read(ref m_writeErrors));
|
|
}
|
|
|
|
public override string Report(double seconds)
|
|
{
|
|
(long notifications, long writes, long written, long writeErrors) = m_final;
|
|
long expected = written * m_sessions.Length * m_subscriptions;
|
|
double ratio = expected == 0 ? 0 : (double)notifications / expected;
|
|
return FormattableString.Invariant(
|
|
$"notif/s={notifications / seconds:F0} received/expected={ratio:P1} ") +
|
|
FormattableString.Invariant(
|
|
$"writes/s={writes / seconds:F0} values/s={written / seconds:F0} ") +
|
|
FormattableString.Invariant(
|
|
$"writeErrors={writeErrors} write latency {m_writeLatency.Percentiles()} ") +
|
|
FormattableString.Invariant($"write->notify latency {m_latency.Percentiles()}");
|
|
}
|
|
|
|
private async Task CreateSubscriptionAsync(ISession session, CancellationToken ct)
|
|
{
|
|
if (session is ManagedSession managed)
|
|
{
|
|
await CreateManagedSubscriptionAsync(managed, ct).ConfigureAwait(false);
|
|
return;
|
|
}
|
|
#pragma warning disable CA2000 // ownership transfers to the session in AddSubscription; disposed below on failure before that
|
|
var subscription = new Subscription(session.DefaultSubscription)
|
|
{
|
|
PublishingInterval = m_publishingInterval,
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100
|
|
};
|
|
#pragma warning restore CA2000
|
|
if (!m_itemCallback)
|
|
{
|
|
subscription.FastDataChangeCallback = OnDataChange;
|
|
}
|
|
try
|
|
{
|
|
foreach (NodeId nodeId in m_nodeIds)
|
|
{
|
|
var item = new MonitoredItem(subscription.DefaultItem)
|
|
{
|
|
StartNodeId = nodeId,
|
|
AttributeId = Attributes.Value,
|
|
MonitoringMode = MonitoringMode.Reporting,
|
|
SamplingInterval = 0,
|
|
QueueSize = (uint)Program.ArgInt("queue-size", 1)
|
|
};
|
|
if (m_itemCallback)
|
|
{
|
|
item.Notification += OnItemNotification;
|
|
}
|
|
subscription.AddItem(item);
|
|
}
|
|
// the session owns and disposes the subscription from here on
|
|
session.AddSubscription(subscription);
|
|
}
|
|
catch
|
|
{
|
|
subscription.Dispose();
|
|
throw;
|
|
}
|
|
await subscription.CreateAsync(ct).ConfigureAwait(false);
|
|
}
|
|
|
|
private void OnDataChange(Subscription subscription, DataChangeNotification notification, ArrayOf<string> stringTable)
|
|
{
|
|
long now = Stopwatch.GetTimestamp();
|
|
int count = 0;
|
|
foreach (MonitoredItemNotification item in notification.MonitoredItems)
|
|
{
|
|
RecordValue(now, item.Value);
|
|
count++;
|
|
}
|
|
Interlocked.Add(ref m_notifications, count);
|
|
}
|
|
|
|
private void OnItemNotification(MonitoredItem monitoredItem, MonitoredItemNotificationEventArgs e)
|
|
{
|
|
if (e.NotificationValue is MonitoredItemNotification item)
|
|
{
|
|
RecordValue(Stopwatch.GetTimestamp(), item.Value);
|
|
Interlocked.Increment(ref m_notifications);
|
|
}
|
|
}
|
|
|
|
private void RecordValue(long now, DataValue value)
|
|
{
|
|
if (value.WrappedValue.TryGetValue(out int counter) && counter > 0)
|
|
{
|
|
long written = Volatile.Read(ref m_writeTicks[counter & 0xFFFF]);
|
|
if (written != 0)
|
|
{
|
|
m_latency.Record(now - written);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task CreateManagedSubscriptionAsync(ManagedSession session, CancellationToken ct)
|
|
{
|
|
var setup = new SubscriptionSetup();
|
|
V2.ISubscription subscription = session.AddSubscription(
|
|
new NotificationHandler(this, setup),
|
|
new V2.SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(m_publishingInterval),
|
|
PublishingEnabled = true,
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100
|
|
});
|
|
for (int i = 0; i < m_nodeIds.Count; i++)
|
|
{
|
|
if (!subscription.TryAddMonitoredItem(
|
|
i.ToString(CultureInfo.InvariantCulture),
|
|
new V2.MonitoredItems.MonitoredItemOptions
|
|
{
|
|
StartNodeId = m_nodeIds[i],
|
|
AttributeId = Attributes.Value,
|
|
MonitoringMode = MonitoringMode.Reporting,
|
|
SamplingInterval = TimeSpan.Zero,
|
|
QueueSize = (uint)Program.ArgInt("queue-size", 1)
|
|
},
|
|
out _))
|
|
{
|
|
throw new InvalidOperationException(
|
|
$"Could not add monitored item '{m_nodeIds[i]}' during subscription setup.");
|
|
}
|
|
}
|
|
await setup.WaitForReadyAsync(subscription, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <summary>
|
|
/// V2 engine handler; data changes are counted like the classic fast callback.
|
|
/// </summary>
|
|
private sealed class NotificationHandler : V2.ISubscriptionNotificationHandler
|
|
{
|
|
public NotificationHandler(SubscribeWorkload owner, SubscriptionSetup setup)
|
|
{
|
|
m_owner = owner;
|
|
m_setup = setup;
|
|
}
|
|
|
|
public ValueTask OnDataChangeNotificationAsync(
|
|
V2.ISubscription subscription,
|
|
uint sequenceNumber,
|
|
DateTime publishTime,
|
|
ReadOnlyMemory<V2.DataValueChange> notification,
|
|
V2.PublishState publishStateMask,
|
|
IReadOnlyList<string> stringTable)
|
|
{
|
|
long now = Stopwatch.GetTimestamp();
|
|
foreach (V2.DataValueChange change in notification.Span)
|
|
{
|
|
m_owner.RecordValue(now, change.Value);
|
|
}
|
|
Interlocked.Add(ref m_owner.m_notifications, notification.Length);
|
|
return ValueTask.CompletedTask;
|
|
}
|
|
|
|
public ValueTask OnEventDataNotificationAsync(
|
|
V2.ISubscription subscription,
|
|
uint sequenceNumber,
|
|
DateTime publishTime,
|
|
ReadOnlyMemory<V2.EventNotification> notification,
|
|
V2.PublishState publishStateMask,
|
|
IReadOnlyList<string> stringTable)
|
|
{
|
|
return ValueTask.CompletedTask;
|
|
}
|
|
|
|
public ValueTask OnKeepAliveNotificationAsync(
|
|
V2.ISubscription subscription,
|
|
uint sequenceNumber,
|
|
DateTime publishTime,
|
|
V2.PublishState publishStateMask)
|
|
{
|
|
return ValueTask.CompletedTask;
|
|
}
|
|
|
|
public ValueTask OnSubscriptionStateChangedAsync(
|
|
V2.ISubscription subscription,
|
|
V2.SubscriptionState state,
|
|
V2.PublishState publishStateMask,
|
|
CancellationToken ct = default)
|
|
{
|
|
m_setup.OnStateChanged(state);
|
|
return ValueTask.CompletedTask;
|
|
}
|
|
|
|
private readonly SubscribeWorkload m_owner;
|
|
private readonly SubscriptionSetup m_setup;
|
|
}
|
|
|
|
private readonly ISession[] m_sessions;
|
|
private readonly int m_nodes;
|
|
private readonly int m_subscriptions;
|
|
private readonly int m_publishingInterval;
|
|
private readonly int m_writeInterval;
|
|
private readonly bool m_itemCallback;
|
|
private readonly long[] m_writeTicks = new long[1 << 16];
|
|
private readonly LatencyRecorder m_latency = new();
|
|
private long m_notifications;
|
|
private long m_writes;
|
|
private int m_counter;
|
|
private ArrayOf<NodeId> m_nodeIds;
|
|
private List<Type>? m_nodeTypes;
|
|
private readonly int m_writeNodes;
|
|
private readonly ISession[] m_writers;
|
|
private readonly bool m_allTypes;
|
|
private readonly int m_writeBatch;
|
|
private readonly LatencyRecorder m_writeLatency = new();
|
|
private long m_writtenValues;
|
|
private long m_writeErrors;
|
|
private (long Notifications, long Writes, long Values, long WriteErrors) m_final;
|
|
}
|
|
}
|