mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-18 22:55:08 +08:00
romanett
849840be5d
Builds on #4815 (merged); rebased onto master. Two commits, touching only `perf/`. Extends `perf/ServerLoadHarness` and `perf/TraceAnalyzer` to profile publish-, write- and event-heavy server load that mirrors the explicit Sessions `LoadTest`. Built for the publish/write load investigation: [findings report](https://claude.ai/artifact/BWuEEPcvvbTAbjY44mJFsa). ## New options | Option | Scenario | Meaning | |---|---|---| | `--scenario events` | | Event subscriptions on the Server object; a server-side `EventGenerator` (in `EventWorkload.cs`) reports events, driven by the server child's new `events <rate>` / `events stop` stdin commands. | | `--event-rate` | `events` | Events/s the server reports; 0 = as fast as one thread can. | | `--writers N` | `sub` | Dedicated writer sessions in addition to `--sessions` (0: the first subscriber session writes, as before). | | `--write-interval 0` | `sub` | Closed-loop writers instead of a timer. | | `--write-batch` | `sub` | Values per Write; writers rotate through the nodes in slices. | | `--node-type all` | `sub` | The LoadTest's Int16…UInt64 mass node set (up to 600 nodes). | | `--queue-size` | `sub`, `events` | Monitored item queue size (default 1 / 10000). | | `--publish-requests` | all classic | Minimum outstanding Publish requests per session. | Also: - `Workload.Secondary` / `OnServerReport` hooks and a `per-op2:` result line: server cost per written value (`sub`) or per generated event (`events`). - TraceAnalyzer `threads` mode: busy share of each thread with its main roles (finds serialized stages). - TraceAnalyzer `cpu <trace> <top> <focus>` additionally lists the focus frame's direct callees. - README rows and paragraphs for all of the above. Conflicts with #4815's rewrite (native workers, `suspend`/`samples` modes) were resolved so #4815's code and text stay and these additions are layered on top. ## Verification Builds: `CustomTestTarget=net10.0 dotnet build perf/ServerLoadHarness -c Release` and `dotnet build perf/TraceAnalyzer -c Release`: 0 warnings, 0 errors. Smoke tests (`--duration 5 --warmup 3`, `DOTNET_gcServer=1 SERVER_DOTNET_gcServer=1`), each run only once foreign CPU was below 1.5 cores, with foreign CPU measured during the run (avg 0.5–0.6 cores): | Run | Result | |---|---| | `events --sessions 5 --subs 3 --event-rate 2000` | generated/s=2009, events/s=30413, received/expected=100.9 %; `per-op2` 588.9 µs CPU / 31.5 KB per generated event | | `sub --sessions 5 --subs 3 --nodes 600 --node-type all --writers 4 --write-interval 0 --write-batch 100` | 131k values/s written, 77k notif/s, writeErrors=0; received/expected=3.9 % (queue size 1 with closed-loop writers: values are overwritten before publishing, as intended); `per-op2` 50.5 µs / 1680 B per written value | | `sub --sessions 2 --subs 1 --nodes 100 --queue-size 2 --publish-requests 4` | received/expected=98.0 %, writeErrors=0 | | same as the second run with `--trace` | 23 MB trace; `TraceAnalyzer threads` lists 58 threads with busy shares and roles; `cpu … 10 MonitoredItem.Publish` prints callers, first Opc.Ua frame and the new "direct callees of" section | #4815's own scenarios still parse: `counters` passes; `native-read` (with `--server-affinity`, `--thread-states`) and `native-counters` start the contract server and fail only at launching the (absent) native worker executable, as expected without the workers. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Marc Schier <marcschier@users.noreply.github.com>
1279 lines
55 KiB
C#
1279 lines
55 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 classic|managed] [--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.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);
|
|
|
|
/// <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);
|
|
}
|
|
return NativeWorkerBenchmark.IsScenario(Arg("scenario", "read"))
|
|
? await NativeWorkerBenchmark.RunAsync(args).ConfigureAwait(false)
|
|
: await RunDriverAsync(args).ConfigureAwait(false);
|
|
}
|
|
|
|
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");
|
|
|
|
(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")}"));
|
|
using Process server = startedServer;
|
|
using BlockingCollection<string> serverLines = startedLines;
|
|
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);
|
|
}
|
|
}
|
|
|
|
// 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? 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))
|
|
.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();
|
|
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);
|
|
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);
|
|
}
|
|
|
|
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');
|
|
(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
|
|
};
|
|
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>();
|
|
_ = Task.Run(() => PumpServerOutput(server, lines));
|
|
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 samples (Stopwatch ticks) recorded by many threads. ConcurrentBag keeps a
|
|
/// list per thread, so recording does not contend between the load threads.
|
|
/// </summary>
|
|
internal sealed class LatencyRecorder
|
|
{
|
|
public void Record(long ticks)
|
|
{
|
|
m_samples.Add(ticks);
|
|
}
|
|
|
|
public void Clear()
|
|
{
|
|
m_samples.Clear();
|
|
}
|
|
|
|
public string Percentiles()
|
|
{
|
|
var all = new List<long>(m_samples);
|
|
if (all.Count == 0)
|
|
{
|
|
return "n=0";
|
|
}
|
|
all.Sort();
|
|
double Ms(double quantile)
|
|
{
|
|
return all[Math.Min(all.Count - 1, (int)(quantile * all.Count))] * 1000.0 / Stopwatch.Frequency;
|
|
}
|
|
return FormattableString.Invariant(
|
|
$"n={all.Count} 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={Ms(1):F3}ms");
|
|
}
|
|
|
|
private readonly ConcurrentBag<long> m_samples = [];
|
|
}
|
|
|
|
/// <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;
|
|
}
|
|
}
|