mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-21 09:06:37 +08:00
romanett
8b360da7d2
Roadmap item A5 of the o6 benchmark gap analysis. Every Publish response deep-cloned its NotificationMessage in `SentMessageQueue` (PR #4707 kept the clone because a retained message returns its pooled payload when the client acknowledges it, which can happen before the response was sent). This PR replaces the clone with a send-completion lease, makes the activator pool that recycles the notifications O(1) and sizes it for the server's load, builds messages without intermediate copies (the notification array is rented), and lets monitored items tell their subscription when they have something to report so the publish timer no longer walks every item. ## Results Paired A/B against master `0dff94bd3` (includes #4742 and #4743; alternating builds, 3 rounds, medians; 20 sessions unless noted, workstation GC, 24-core machine shared with other sessions). Server = the reference server child process. | Scenario | Server alloc / op | Server GC pause | Server CPU / op | Write→notify p50 / p99 | Throughput | |---|---|---|---|---|---| | `sub`, 5 subs × 100 items, pub 100 ms | **−57 %** (373 → 159 B) | **−57 %** | −7 % ¹ | −1 % / −2 % | +1 % ² | | `sub`, 5 subs × 100 items, pub 10 ms | **−57 %** (375 → 163 B) | **−52 %** | +9 % ¹ | −22 % / +1 % | ±0 ² | | `sub`, pub 100 ms, SignAndEncrypt | **−57 %** (380 → 163 B) | **−53 %** | +4 % ¹ | −11 % / **−12 %** | ±0 ² | | `sub`, 2 sessions × 1 sub × 1000 items, 20 changing, pub 50 ms (9 rounds) | −2 % ³ | ±0 | −23 % ³ | ±0 / −2 % | ±0 | | `counters`, 4000 items, 10 ms sampling, pub 1 s (per notification) | −2 % | −6 % | −6 % | – | ±0 | | `o6counters`, 2048 items (per sample) | −3 % ⁴ (111 → 108 B) | −5 % | −7 % | – | ±0, all PASS | | `o6counters`, 4096 items (per sample) | −3 % ⁴ | +14 % ⁵ | −6 % | – | ±0, all PASS | ¹ CPU per notification varies by ±15 % between rounds on this machine; in an earlier 8-round batch against `071660354` the same scenarios measured −14 % / −1 %. Read it as "no regression". Client CPU per notification moved −18 %…0 %, within its spread. ² Bounded by the writer rate. ³ The server is nearly idle (0.05–0.12 cores, 360 notifications/s; its allocation is the writer and keep-alives). The first 3 rounds measured +31 % CPU, 6 more rounds −23 % (medians 969 → 742 ms per window): noise-bound, no regression. ⁴ With #4743's sampling permission cache a sample costs 111 B; one Publish per second carries 2048/4096 notifications. ⁵ The activator pools no longer allocate their 1023-slot arrays up front; the heap is a few MB smaller, which shifts the GC's generation budgets so that some low-allocation runs make every collection gen1 (26 of 26 instead of 2). Pause totals differ by milliseconds per 10 s window (37 → 42 ms here, 21 → 20 ms at 2048 items). Allocating the arrays eagerly restored master's gen1 counts in a separate test. In a single CPU trace of `sub` at pub 10 ms, recycling acknowledged payloads (pool `Return`) went from 16 % of server on-CPU time (master, the full pool scanned per return) to 5 %. ## Changes - **Send completion instead of a deep clone** (`NotificationMessageLease`, `LeasedPublishResponse`, `SentMessageQueue`): the retransmission queue and every Publish response that carries a message each own a reference. The transports already call `IPooledEncodeable.Reuse()` on a response once it is encoded or dropped; the Publish response is now a `LeasedPublishResponse` whose `Reuse()` releases its reference. The pooled payload (`NotificationMessage`, `DataChangeNotification`/`EventNotificationList`, `MonitoredItemNotification`/`EventFieldList`) is recycled when the last owner releases it, so an acknowledgement that overtakes the send no longer recycles a payload that is still being encoded, and the queue and the response share one message. A reference that is never released only leaves the payload to the GC; every owner releases at most once. Keep-alive messages are owned by their response alone. Republish, the durable snapshot and the retransmission mirror still copy. - **Transports**: the HTTPS (binary and JSON), WebSocket (UA-SC, JSON, OpenAPI) and Kestrel TCP listeners and the WebAPI endpoint dispatcher now release responses after encoding, as the TCP listener does, so leased payloads are recycled there too. A response retained for a reconnect is released when it is sent; a send that throws leaves the response to the GC. - **Activator pool** (`BoundedObjectPool`): rent and return scanned the 1023 slots linearly; with the pool full every returned notification scanned all of them before being dropped (16 % of server CPU at 94k notifications/s). It is now a `ConcurrentQueue` behind the lock-free first slot with an `Interlocked` bound (a locked stack spun under contention when transport threads return whole messages). The queue is created on first use, so activators of types that are never recycled no longer allocate a 1023-slot array each. The client decoder rents through the same pool. - **Pool bound** (`PooledEncodeableType<T>.EnsureMaximumRetained`, internal): the bound can be raised, never lowered. The subscriptions raise the `MonitoredItemNotification` and `EventFieldList` pools to twice the number of monitored items in the process (a notification is recycled about two publishing cycles after it was built), between the default 1024 and 65536. At the cap the retained notifications take about 5 MB; the bound stays raised after items are deleted. - **Message construction** (`Subscription.InnerPublish`/`ConstructMessage`, `MonitoredItem.Publish`): the per-Publish `Queue<>`/`List<>` objects are reused by the subscription; messages are built at their final size (no `List`→`ArrayOf` copy, no `AddItem` reallocation); the `MonitoredItemNotification[]` of a message is rented from `ArrayPool` and the message carries an exact-length slice of it; the lease returns the array (cleared) with the payload, and messages dropped on queue overflow return theirs at once; only arrays the subscription rented are tracked, so restored, mirrored and copied messages never reach the array pool; the `DiagnosticInfo` list is only allocated when a notification has one; timestamps are stripped on the struct before the pooled notification is filled. - **Ready list** (`IReadySignalingMonitoredItem`, `Subscription.PromoteReadyMonitoredItems`): `MonitoredItem` acquires a ready signal (compare-exchange, a full fence after its flags are set) when it queues a value or event or a resend is requested, and hands itself to the subscription's intake list. The publish timer checks only these candidates instead of every item under the subscription lock. An item stays a candidate while it has something pending that can become reportable without a new value (sampling deadline, aggregate interval, samples of a SAMPLING item, trigger flag); after releasing the signal the subscription checks again, so a value queued concurrently is never missed. Created, restored, modified, monitoring-mode and triggering changes mark candidates, as does an item that leaves the publish list with something pending. ### Behaviour notes (custom subscription or monitored-item extensions) - **Publish responses share the retained message.** `PublishResponse.NotificationMessage` returned by `SubscriptionManager.PublishAsync` is the retained message itself, not a copy, and its payload objects return to the activator pools after the response was sent and acknowledged. Code that keeps or modifies the message of a Publish response (for example in a `StandardServer.PublishAsync` override or a request interceptor) must copy it first (`CoreUtils.Clone`). `Republish` still returns a copy. - **Custom transports**: a transport that does not call `Reuse()` on responses keeps working; the payload is then collected by the GC instead of being pooled. - **Custom `IMonitoredItem` implementations** (not derived from `MonitoredItem`) cannot signal readiness: a subscription that contains one scans all its items on every publishing cycle, as before. `MonitoredItem` subclasses signal through the base class. - **`DataChangeNotification.MonitoredItems` of a published message is a slice of a pooled array**: read it through `ArrayOf<T>` (`Span`, indexer, `Count`); code that unwraps the underlying array would see a longer array. - **Order within a message**: notifications of different monitored items appear in the order the items became ready rather than in the internal list order. OPC UA does not define an order across items; the order of the values of one item is unchanged. ## Harness (`perf/`) - `--write-nodes K` for `sub`: the writer changes only the first K monitored nodes, for large subscriptions in which few items change. - `server-scenarios.txt`: publish-path scenarios (`--pub 10`, 1000 items per subscription with 20 changing). ## Testing - New `NotificationEncodingLifetimeRegressionTests` cases: an acknowledgement before send completion does not recycle the payload still being encoded (the send completion does); Republish after the send completed returns the original content and stays independent of the recycled original; a failed send that releases the response twice, plus a copy of the response, recycles the payload exactly once and only after the queue released it; a keep-alive is owned by its response; the rented notification array stays intact until both acknowledgement and send completion (either order), the message encodes only its slice, and queue overflow returns the arrays of dropped messages. The existing 30-case matrix (publish / queued / republish / mirrors × acknowledge / evict / clear × data / events) now asserts that a shared payload survives every queue mutation until its response is released. - New `ReadyListRegressionTests`: the same scenarios on a subscription that uses the ready list and on one that scans (it also has an item that cannot signal) must publish the same notifications, for triggering (sampling items, backlog of the triggering item), monitoring-mode changes (Sampling↔Reporting, Disabled, disabled before publish), disabled publishing with keep-alives and ResendData, a value queued while the item was in the publish list, and the backlog of an item that left Reporting while in the publish list. Mutation check: removing the item's signal fails 5 of 6, removing the subscription-side marks fails the backlog case. Plus: a published item without pending values is not tracked; creating 1200 items raises the notification pool bounds to at least 2400. - `PooledEncodeableTypeTests`: the raised bound is respected and never lowered. - net10.0 on the rebased branch: Server 8481 passed (5 skipped); Sessions 851 (379 skipped); Client 2788; Subscriptions.Classic 57; Subscriptions.Durable 36 (11 skipped); Redundancy.Server 884, Redundancy.Client 134, Redundancy.Samples 17; Stress Contract+Integration 45; Subscriptions 657 (6 skipped); Types pool tests 165. All 0 failed. - net48 build of `Opc.Ua.Server.Tests` and `Opc.Ua.Bindings.Https` and `Opc.Ua.Types.Tests` clean; `dotnet format whitespace` clean. ## Not changed (follow-ups) - Republish still copies the message (`ISubscriptionManager.Republish` returns a `NotificationMessage`, so the response cannot carry a lease without an API change). - `EventFieldList[]` per event message could be rented like the data change array. - Sampling remains the larger share of allocation and CPU in the `counters`/`o6counters` scenarios. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1035 lines
43 KiB
C#
1035 lines
43 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..];
|
|
string value = i + 1 < args.Length && !args[i + 1].StartsWith("--", StringComparison.Ordinal)
|
|
? args[++i]
|
|
: "1";
|
|
s_args[key] = value;
|
|
}
|
|
}
|
|
string mode = args.Length > 0 ? args[0] : "run";
|
|
return mode == "server"
|
|
? await RunServerAsync().ConfigureAwait(false)
|
|
: await RunDriverAsync(args).ConfigureAwait(false);
|
|
}
|
|
|
|
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;
|
|
string pkiRoot = Path.Combine(Path.GetTempPath(), "uaperf-pki-server");
|
|
string endpointUrl = 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("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 = 50;
|
|
// sized for the largest pipeline (sessions x inflight), as the o6 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)
|
|
: 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();
|
|
string? line;
|
|
while ((line = await Console.In.ReadLineAsync().ConfigureAwait(false)) != null)
|
|
{
|
|
switch (line.Trim())
|
|
{
|
|
case "mark":
|
|
GC.Collect();
|
|
GC.WaitForPendingFinalizers();
|
|
mark = ProcessStats.Capture();
|
|
(server as CounterServer)?.Counters?.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);
|
|
break;
|
|
case "quit":
|
|
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");
|
|
|
|
var startInfo = new ProcessStartInfo(
|
|
Environment.ProcessPath!,
|
|
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")}"))
|
|
{
|
|
RedirectStandardInput = true,
|
|
RedirectStandardOutput = true,
|
|
UseShellExecute = false
|
|
};
|
|
foreach (DictionaryEntry variable in Environment.GetEnvironmentVariables())
|
|
{
|
|
string name = (string)variable.Key;
|
|
if (name.StartsWith("SERVER_", StringComparison.Ordinal))
|
|
{
|
|
startInfo.Environment[name[7..]] = (string?)variable.Value;
|
|
}
|
|
}
|
|
|
|
using Process server = Process.Start(startInfo)!;
|
|
using var serverLines = new BlockingCollection<string>();
|
|
_ = Task.Run(() => PumpServerOutput(server, serverLines));
|
|
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);
|
|
var sessions = new ISession[sessionCount];
|
|
var connectTime = Stopwatch.StartNew();
|
|
// Shared sessions connect one at a time by default: concurrent creation
|
|
// on one ClientChannelManager fails with BadSecureChannelClosed.
|
|
int connectParallel = clientKind == "shared" ? ArgInt("connect-parallel", 1) : 32;
|
|
await Parallel.ForAsync(
|
|
0,
|
|
sessionCount,
|
|
new ParallelOptions { MaxDegreeOfParallelism = 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 {sessionCount} {clientKind} sessions ({security}) in {connectTime.ElapsedMilliseconds} ms"));
|
|
|
|
Workload workload = scenario == "counters"
|
|
? new CounterWorkload(sessions[0], ArgInt("nodes", 1), ArgInt("sampling", 10), ArgInt("pub", 1000))
|
|
: scenario == "sub"
|
|
? new SubscribeWorkload(
|
|
sessions,
|
|
ArgInt("nodes", 100),
|
|
ArgInt("subs", 5),
|
|
ArgInt("pub", 100),
|
|
ArgInt("write-interval", 100),
|
|
Arg("callback", "fast") == "item",
|
|
ArgInt("write-nodes", 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);
|
|
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..];
|
|
double clientCores = (Process.GetCurrentProcess().TotalProcessorTime - clientCpuStart).TotalSeconds /
|
|
window.Elapsed.TotalSeconds;
|
|
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 operations = workload.Operations;
|
|
if (operations > 0)
|
|
{
|
|
(double allocatedMb, double cpuMs) = ParseProcessReport(report);
|
|
result.Append(
|
|
CultureInfo.InvariantCulture,
|
|
$" per-op: allocB={allocatedMb * 1048576 / operations:F0} cpuUs={cpuMs * 1000 / operations:F1} ({workload.OperationName})\n");
|
|
(allocatedMb, 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;
|
|
}
|
|
|
|
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();
|
|
}
|
|
|
|
private 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
|
|
});
|
|
}
|
|
|
|
private 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; }
|
|
|
|
public abstract Task SetupAsync();
|
|
|
|
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,
|
|
int nodes,
|
|
int subscriptions,
|
|
int publishingInterval,
|
|
int writeInterval,
|
|
bool itemCallback,
|
|
int writeNodes)
|
|
{
|
|
m_sessions = sessions;
|
|
m_nodes = nodes;
|
|
m_subscriptions = subscriptions;
|
|
m_publishingInterval = publishingInterval;
|
|
m_writeInterval = writeInterval;
|
|
m_itemCallback = itemCallback;
|
|
m_writeNodes = writeNodes;
|
|
}
|
|
|
|
public override double Operations => Interlocked.Read(ref m_notifications);
|
|
|
|
public override string OperationName => "notification";
|
|
|
|
/// <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()
|
|
{
|
|
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, write {WrittenNodes} items every {m_writeInterval} ms; ") +
|
|
FormattableString.Invariant($"setup {setupTime.ElapsedMilliseconds} ms"));
|
|
}
|
|
|
|
public override async Task RunAsync(CancellationToken ct)
|
|
{
|
|
using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(m_writeInterval));
|
|
while (await timer.WaitForNextTickAsync(ct).ConfigureAwait(false))
|
|
{
|
|
int counter = Interlocked.Increment(ref m_counter);
|
|
var values = new WriteValue[WrittenNodes];
|
|
for (int i = 0; i < values.Length; i++)
|
|
{
|
|
values[i] = new WriteValue
|
|
{
|
|
NodeId = m_nodeIds[i],
|
|
AttributeId = Attributes.Value,
|
|
Value = new DataValue(new Variant(counter))
|
|
};
|
|
}
|
|
Volatile.Write(ref m_writeTicks[counter & 0xFFFF], Stopwatch.GetTimestamp());
|
|
await m_sessions[0].WriteAsync(null, values, ct).ConfigureAwait(false);
|
|
Interlocked.Increment(ref m_writes);
|
|
}
|
|
}
|
|
|
|
public override void ResetStats()
|
|
{
|
|
m_latency.Clear();
|
|
Interlocked.Exchange(ref m_notifications, 0);
|
|
Interlocked.Exchange(ref m_writes, 0);
|
|
}
|
|
|
|
public override string Report(double seconds)
|
|
{
|
|
long notifications = Interlocked.Read(ref m_notifications);
|
|
long expected = Interlocked.Read(ref m_writes) * m_sessions.Length * m_subscriptions * WrittenNodes;
|
|
double ratio = expected == 0 ? 0 : (double)notifications / expected;
|
|
return FormattableString.Invariant(
|
|
$"notif/s={notifications / seconds:F0} received/expected={ratio:P1} ") +
|
|
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 = 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 = 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 readonly int m_writeNodes;
|
|
}
|
|
}
|