GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-29 07:07:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Server: fully asynchronous change-notification and event push path (#3941)

Browse Source 
# Description

Makes the server data-change **and** event push path fully asynchronous,
so a monitored node whose value is read through an asynchronous read
handler (`OnReadValueAsync` / `OnSimpleReadValueAsync`) is honored
without a thread ever blocking, and the per-node notification channel
applies back-pressure by awaiting rather than blocking a caller. This is
the deeper follow-up to the review discussion on #3922, where the
synchronous change-notification path was silently bypassing asynchronous
value read handlers.

The change is **additive and non-breaking**: every existing synchronous
API and sink keeps working unchanged.

**Asynchronous `NodeState` change-notification API**
(`Stack/Opc.Ua.Types`)
- New `ClearChangeMasksAsync`, `OnStateChangedAsync` /
`StateChangedAsync`, and `ReportEventAsync` / `OnReportEventAsync` (plus
the `BaseInstanceState` parent-recursion override).
- The synchronous `ClearChangeMasks` / `ReportEvent` now also drive the
asynchronous sinks: they complete inline for a synchronously-readable
node with channel capacity, and block the calling thread only when a
genuinely asynchronous read is in flight or the channel is full (so only
synchronous-API callers ever pay that cost).

**Asynchronous `MonitoredNode2` producers** (`Libraries/Opc.Ua.Server`)
- `OnMonitoredNodeChangedAsync` / `OnReportEventAsync` read each
monitored attribute at enqueue time via `ReadAttributeAsync` — honoring
an asynchronous read handler and materializing the value **before** the
change is enqueued (a strict enqueue-time snapshot) — and enqueue by
awaiting the bounded per-node channel. Synchronous wrappers are retained
for compatibility.
- FIFO/LIFO discard and the `Overflow` status bit remain a
per-monitored-item queue concern; the shared per-node channel stays a
lossless conduit.

**Asynchronous event source-report path**
- New `IServerInternal.ReportEventAsync` /
`ServerInternalData.ReportEventAsync`, and an asynchronous root-notifier
`OnReportEventAsync` in `AsyncCustomNodeManager` (event propagation to
the Server object is now awaited). The rare model-change
`Raise*ChangeEvent` helpers remain synchronous — they drive the async
producer via the sync wrapper, and events do not perform an asynchronous
read, so a full async cascade through the model-change/audit subsystems
would be near-zero value.

**Internal flush migration**
- The hot value-write flush (`AsyncCustomNodeManager.WriteAsync`) and
all other asynchronous-context `ClearChangeMasks` call sites now await
`ClearChangeMasksAsync`. Genuinely synchronous contexts (builders,
diagnostics timers, the synchronous `CustomNodeManager`, supervision)
remain on the synchronous wrapper (correct — not hot paths).

**Performance:** synchronous reads are unchanged (they complete inline);
asynchronous reads no longer block a thread (proven by a unit test
asserting the producer's `ValueTask` is not completed while a slow
asynchronous read is in flight); enqueue replaces a blocking wait /
per-event `AsTask` with an awaited write.

**Docs:** a new "Fully asynchronous change notifications" section in
`Docs/AsyncServerSupport.md`.

## Related Issues

Follow-up to the review feedback on #3922 (asynchronous value read
handlers were bypassed on the synchronous change-notification push
path).

- Fixes #

## Checklist

_Put an `x` in the boxes that apply. You can complete these step by step
after opening the PR._

- [ ] I have signed the
[CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf)
and read the
[CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md)
doc.
- [x] I have added tests that prove my fix is effective or that my
feature works and increased code coverage.
- [x] I have added all necessary documentation.
- [x] I have verified that my changes do not introduce (new) build or
analyzer warnings.
- [ ] I ran **all** tests locally using the **UA.slnx** solution against
at least .net **framework** and .net **10**, and all passed.
- [ ] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings.
- [ ] I have addressed **all** PR feedback received.
This commit is contained in:
Marc Schier
2026-07-03 14:42:24 +02:00
committed by GitHub
parent d4714bd8ff
commit 26ef554283
15 changed files with 954 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Docs/AsyncServerSupport.md
+21 -1
View File
@@ -169,13 +169,33 @@ public MyNodeManager(IServerInternal server, ApplicationConfiguration config)
```
- **`MonitoredNodeMonitoredItemManager`** (default): node value changes are propagated to
subscribers immediately by calling `NodeState.ClearChangeMasks` after every successful write.
subscribers immediately by awaiting `NodeState.ClearChangeMasksAsync` after every successful
write (see [Fully asynchronous change notifications](#fully-asynchronous-change-notifications)).
No background threads are created per subscription.
- **`SamplingGroupMonitoredItemManager`**: a background timer thread samples the current node
value at the negotiated `SamplingInterval`. Write changes are *not* pushed immediately; instead
the next scheduled sample detects and delivers them. Choose this mode when the data source
produces values independently of OPC UA write requests (e.g. hardware polling).
### Fully asynchronous change notifications
`AsyncCustomNodeManager` propagates value changes and events to monitored items over a fully asynchronous push path, so no thread is blocked when a node exposes an asynchronous value read handler or when the per-node notification channel applies back-pressure.
`NodeState` exposes asynchronous counterparts of its change-notification API alongside the existing synchronous members:
| Synchronous | Asynchronous |
|-------------|--------------|
| `ClearChangeMasks(context, includeChildren)` | `ClearChangeMasksAsync(context, includeChildren, ct)` |
| `OnStateChanged` / `StateChanged` | `OnStateChangedAsync` / `StateChangedAsync` |
| `ReportEvent(context, e)` | `ReportEventAsync(context, e, ct)` |
| `OnReportEvent` | `OnReportEventAsync` |
The additions are non-breaking: existing synchronous callers and sinks continue to work unchanged. `MonitoredNodeMonitoredItemManager` wires the asynchronous sinks, so after a successful write `AsyncCustomNodeManager.WriteAsync` flushes changes with `await handle.Node.ClearChangeMasksAsync(...)`. The value for each monitored attribute is read at that moment through `NodeState.ReadAttributeAsync` — honoring an asynchronous `OnReadValueAsync` / `OnSimpleReadValueAsync` handler when one is registered — and the resulting snapshot is enqueued by awaiting the bounded per-node channel. Because the read is materialized before the change is enqueued, every notification carries the node value at the instant of the change (an enqueue-time snapshot); because both the read and the enqueue are awaited, the producing thread is never blocked.
The synchronous `ClearChangeMasks` / `ReportEvent` still drive the asynchronous sinks: they complete inline when the node is synchronously readable and the channel has capacity, and block the calling thread only when a genuinely asynchronous read is in flight or the channel is full. A synchronous node manager therefore keeps working without any change, paying a blocking cost only in those two cases.
Overflow and discard — the FIFO/LIFO `DiscardOldest` behavior and the `Overflow` status bit — remain a property of each monitored item's own queue. The shared per-node channel is a lossless conduit and does not drop notifications, so a monitored item's configured queue policy is always applied to the full stream of changes.
### Creating a custom node manager
Derive from `AsyncCustomNodeManager` and override only the virtual methods you need:
Docs/Benchmarks.md
+2
View File
@@ -424,6 +424,8 @@ activator pool system.
The `[Explicit]` macro test `ServerManySessionsLoadTestAsync(int sessionCount)` (`Tests/Opc.Ua.Sessions.Tests/LoadTest.cs`) exercises the reference server under many concurrent sessions. Each session opens its own secure channel, creates one slow-publishing subscription (1000 ms) with a single monitored item on a shared value node, and a separate writer session changes that value periodically; every session is expected to receive value-change notifications over a steady-state window. It runs over `Basic256Sha256` (sign & encrypt) and asserts that all sessions connect and all receive notifications. It is parameterized from a `500` baseline up to a `10000` stress case (500, 1000, 1500, 2000, 2500, 4000, 10000), selected by name (e.g. `ServerManySessionsLoadTestAsync(2000)`).
> For a deep, code-referenced analysis of *why* a single node tops out here — the establishment vs steady-state boundaries and a prioritized admission-control / rate-limiting roadmap for moving beyond ~2000 — see [Server Session Scalability](ServerScalability.md).
| Tested configuration | Value |
| --- | --- |
| Concurrent sessions | 500 baseline, up to 10000 (selectable case) |
Docs/README.md
+1
View File
@@ -16,6 +16,7 @@ Here is a list of available documentation for different topics:
* Support for the [TransferSubscriptions](TransferSubscription.md) service set.
* [Diagnostics](Diagnostics.md) — logging, telemetry, server audit events, server diagnostics nodes, and packet capture.
* [Performance Benchmarks](Benchmarks.md) — BenchmarkDotNet methodology, the 2.0 (`master`) vs 1.5.378 (`master378`) comparison, root-cause analysis of the encoder/decoder/session regressions and their real-world impact, the subscription-notification (pooled encodeable) micro-benchmarks, server session scalability (the 500-session capability, sizing and bottlenecks), and planned future work.
* [Server Session Scalability](ServerScalability.md) — why a single node tops out at ~2000 concurrent sessions, the establishment vs steady-state boundaries (socket backlog, the `BadTcpInternalError` retry-storm amplifier, the O(N²) diagnostics rescan, CreateSession crypto-under-lock, the RSA CPU wall, and the held-Publish worker-accounting coupling) with code references, and a prioritized admission-control / rate-limiting roadmap for moving beyond it.
* Support for [WellKnownRoles & RoleBasedUserManagement](RoleBasedUserManagement.md).
* Pluggable [Identity Providers](IdentityProviders.md) — interfaces (`IClientIdentityProvider`, `IUserTokenAuthenticator`, `IAccessTokenProvider`, `ITokenIssuer`, `IIdentityClaims`) plus the OPC 10000-6 §6.5.2.2 `IssuerEndpointUrl` JSON parser for OAuth2 / OIDC / Entra / JWT flows.
* Support for [ECC Certificates](EccProfiles.md).
Docs/ServerScalability.md
+125
View File
@@ -0,0 +1,125 @@
# Server session scalability — boundaries and how to move beyond them
This document explains what limits the reference server's **concurrent-session** scale on a single node, where the boundaries are in the code, and a prioritized roadmap for moving beyond them. It is the architectural companion to the measured numbers in [Performance Benchmarks — Server session scalability](Benchmarks.md#server-session-scalability).
## TL;DR
- On a 6-core developer machine the server establishes **~1500 concurrent sessions cleanly** and delivers 100 % of steady-state notifications; beyond that it hits a wall at **~2000**.
- **That wall is an *establishment* (connect-storm) ceiling, not a steady-state one.** Once sessions are up, the async publish path keeps delivering; the bottleneck is the CPU-bound secure-channel handshake burst plus a positive-feedback retry loop.
- The single most effective lever to raise and *stabilize* the ceiling is **explicit admission control / rate limiting** at connection accept and session establishment, so the server sheds load with a fast deterministic rejection instead of aborting mid-handshake and inviting client retries.
- Beyond a single node's core count the RSA handshake cost is fundamental: scale **out** using the distributed mirroring already in the stack, rather than only up.
## The request lifecycle (where each stage can bottleneck)
A client session is brought up in four stages, each with its own scaling boundary:
1. **TCP accept** — the OS accepts the socket and the listener creates a channel object.
2. **Secure channel** — `OpenSecureChannel` performs the asymmetric (RSA) handshake: decrypt + signature-verify with the server certificate's private key, generate nonces.
3. **CreateSession** — admission checks (session/channel limits), server-nonce generation and signing, per-session diagnostics node creation.
4. **ActivateSession** — verify the client signature and the user identity token, activate the session.
After activation the session enters **steady state**: it keeps one long-polled `Publish` request outstanding and receives notifications as the subscription's monitored items change.
## Why 2000 collapses: the connect-storm feedback loop
The failure at 2000 is not a hard limit that returns a clean error — it is a *positive-feedback collapse*:
> overload → a handshake times out or is aborted → the client **retries** → the retry creates a **new** server session while the previous one is orphaned → more load → more aborts.
In a 2000-session run on a 6-core box the server created **~4432 sessions** for the 2000 target (2.2× — the duplicates are retries) and abandoned **~31,779 subscriptions**. The extra sessions are pure waste that competes for the same CPU as the useful work, so the server never converges on the full set within the connect budget. Fixing the collapse is therefore less about a single hard limit and more about **breaking the feedback loop** with backpressure.
## Establishment boundaries (the current ceiling)
### B1 — The listener socket backlog is hard-coded to 10
`TcpTransportListener` calls `Listen(kSocketBacklog)` with `kSocketBacklog = 10` (`Stack/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs`, `kSocketBacklog` constant near line 261, applied in the listener start paths). A burst of 2000 simultaneous connects instantly overflows a backlog of 10, so the OS drops or resets connections before the accept loop ever sees them — which the client experiences as a transport failure and retries. This is the first wall and is not configurable today.
### B2 — Every transient handshake error becomes `BadTcpInternalError`, which triggers a client retry
The secure-channel receive path maps *any* exception during the Hello or `OpenSecureChannel` processing to `BadTcpInternalError`:
- `TcpServerChannel.ProcessHelloMessage` and `ProcessOpenSecureChannelRequest` (`Stack/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs`, faults at lines 589, 895).
- The base receive/write loop in `UaSCBinaryChannel` (`.../UaSCBinaryChannel.cs`, faults at lines 581, 760, 840, 881).
Under load a transient decrypt/parse/IO hiccup aborts the channel, the client retries, and the retry creates a duplicate server session. This catch-all is the amplifier that turns a 2000 target into 4432 creates. Post-#3930 (caller-owned certificate handles) the previously dominant private-key handle race is largely gone — a 2000 run now shows a *single* `BadTcpInternalError` rather than tens of thousands — so the remaining aborts are increasingly timeout/backpressure driven rather than crypto-handle driven.
### B3 — Session-diagnostics creation is globally serialized and forces an O(N) rescan per create
Creating a session adds a per-session diagnostics node under the **global address-space semaphore** and then forces a full diagnostics rescan:
- `DiagnosticsNodeManager.CreateSessionDiagnosticsAsync` runs under `m_modifyAddressSpaceSemaphoreSlim` and sets `m_forceDiagnosticsScan = true` (`Libraries/Opc.Ua.Server/Diagnostics/DiagnosticsNodeManager.cs`, around lines 903–1014).
- `DoScan` (same file, from line 1820) rebuilds the full `SessionDiagnosticsDataType[N]` and `SessionSecurityDiagnosticsDataType[N]` arrays over **all** sessions (lines 1843–1875).
- The `m_forceDiagnosticsScan` flag *defeats* the 1-second scan throttle that would otherwise coalesce updates (checks at lines 1710–1715 and 1736–1742).
So N concurrent creates each trigger an O(N) rebuild → **~O(N²)** diagnostics work during a bulk connect, performed under a global lock that also serializes address-space mutation. This is gated by `DiagnosticsEnabled`: servers that do not need the live session-diagnostics arrays can disable them to remove this cost entirely.
### B4 — CreateSession holds a global semaphore across admission *and* RSA signing
`SessionManager.CreateSessionAsync` takes `m_semaphoreSlim` and holds it across the `MaxSessionCount` check (→ `BadTooManySessions`), nonce-uniqueness check, token/nonce generation, `Session` construction, and the table reservation (`Libraries/Opc.Ua.Server/Session/SessionManager.cs`, roughly lines 218–320). `StandardServer.CreateSessionAsync` additionally performs the CPU-bound **server-signature RSA** inside its own `m_semaphoreSlim` region (`Libraries/Opc.Ua.Server/Server/StandardServer.cs`, around lines 504–535).
`ActivateSession` already demonstrates the better pattern — it keeps the CPU-bound *client-signature verify* **outside** the semaphore (`SessionManager.cs`, note around lines 443–447) — but `CreateSession` does its crypto inside the lock, so concurrent creates largely serialize behind one signing operation at a time.
### B5 — Establishment crypto is CPU-bound and competes with steady state on the same cores
Each connect costs, cumulatively: an RSA decrypt of the `OpenSecureChannel` request using the server private key (`Stack/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.Rsa.cs`, `GetRSAPrivateKey()` at line 118), a client certificate-chain validation (`StandardServer.cs`, around lines 349–405), the server-nonce signature, and on activate the client-signature verify. Handshakes run per-channel on the thread pool, so they parallelize across cores — but on 6 cores a 2000-connect burst saturates every core. That starves the steady-state publish sweep and the servicing of held `Publish` requests, so keep-alives are missed and subscriptions drift to *expired/abandoned* (the ~31,779 abandonments). **CPU is the fundamental single-node ceiling**, and it is why establishment (not delivery) is the wall on modest hardware.
## Steady-state boundaries (the next wall, once establishment is fixed)
### S1 — Held Publishes are async-parked, but the request-queue accounting couples the worker/thread budget to session count
A common assumption is "one blocked thread per held `Publish`". That is **not** what happens: when no subscription is ready, `SessionPublishQueue.PublishAsync` returns a `TaskCompletionSource`-backed task (`Libraries/Opc.Ua.Server/Subscription/SessionPublishQueue.cs`, lines 114–156) and the worker `await`s it — the OS thread is released, not spinning.
However, the request-queue **accounting** still couples the worker budget to the number of concurrently-held Publishes. In `ServerBase.RequestQueue.WorkerLoopAsync` the active-worker counter is incremented for the *entire* duration of the awaited request, including the whole long-poll wait (`Stack/Opc.Ua.Core/Stack/Server/ServerBase.RequestQueue.cs`, lines 203–214). `ScheduleIncomingRequest` spawns additional worker loops up to `MaxRequestThreadCount` whenever the active count reaches the current worker count (lines 174–186), and the queue constructor raises the process-wide `ThreadPool` maximum to `MaxRequestThreadCount` (lines 84–89).
The practical consequence: to serve N sessions you must size `MaxRequestThreadCount ≥ N` and the process `ThreadPool` ceiling balloons accordingly (the load test sets 10500 for a 10000-session case). A parked long-poll should not count against the active-worker budget — decoupling it would let a small worker pool hold many thousands of outstanding Publishes.
### S2 — Per-session outstanding-Publish cap
Each session's `SessionPublishQueue` rejects additional outstanding requests once the queue reaches `MaxPublishRequestCount`, returning `BadTooManyPublishRequests` (`SessionPublishQueue.cs`, lines 136–155). This is a correct per-session guard, but it must be sized for the client's publish-pipelining depth.
### S3 — The publish sweep is O(subscriptions) per cycle
`SubscriptionManager.PublishSubscriptionsAsync` snapshots the session queues and sweeps them each cycle, using `Parallel.For` above a threshold and a serial loop below it (`Libraries/Opc.Ua.Server/Subscription/SubscriptionManager.cs`, around lines 2066–2101). The work is O(subscriptions) per cycle and parallelizes across cores without a manager-wide lock, so it scales with cores — but it is still per-cycle O(N) and shares cores with establishment crypto (B5).
### S4 — The ABANDONED path is a symptom, not an independent limit
A subscription expires when its lifetime counter reaches the maximum while it is still waiting to publish (`Subscription.cs`, around lines 503–543 and 841–860; cleanup in `SubscriptionManager.cs`, around lines 2103–2133). Mass abandonment during a connect storm is a *downstream effect* of B5/S1 CPU starvation — the publishes are not serviced in time — rather than a boundary to fix on its own.
## Where to linearize: admission control and rate limiting
Because the collapse is a positive-feedback loop, the highest-leverage fix is **explicit backpressure** so the server degrades gracefully — a fast, deterministic rejection the client can honor — instead of aborting mid-handshake and inviting a retry that doubles the load:
1. **Connection-admission rate limit at the listener** (token bucket) plus a larger, configurable socket backlog. Bound the rate of new secure-channel handshakes to what the RSA-handshake CPU can absorb; hold or fast-reject the excess with a `BadServerTooBusy`/Retry-After rather than letting it fail deep in the handshake. Addresses B1, B2, B5.
2. **Bounded session-establishment admission**, sized to core count, that returns a *fast deterministic* `BadTooManySessions` / `BadServerTooBusy` (ideally with a Retry-After hint) so a client backs off instead of retrying into duplicate sessions. Addresses B2, B4.
3. **Move crypto out of the establishment locks** so the global semaphore only guards the table reservation, not the RSA signing (mirror what ActivateSession already does). Addresses B4.
4. **Coalesce or throttle the diagnostics rescan** — honor the existing 1-second throttle under bulk connect, make the array update incremental (append one entry) instead of a full O(N) rebuild, and/or allow disabling session diagnostics under load. Addresses B3.
5. **Decouple held Publishes from the worker/thread accounting** so a parked long-poll does not consume a worker slot and the `ThreadPool` ceiling does not scale with session count. Addresses S1.
6. **Client-side backoff / honor Retry-After** so a well-behaved client ramps its connects instead of hammering, which linearizes the whole system end-to-end.
The unifying principle: **linearize the *admission* of expensive work (handshakes, session creation) to the available CPU, and give overflow an immediate, honest "busy" answer** — never a slow, ambiguous abort that a client will retry.
## Roadmap
The mitigations group into three phases of increasing effort:
- **Option A — Quick wins (raise the ceiling on existing hardware).** Make the socket backlog configurable and larger (B1); move the server-nonce RSA signing out of the CreateSession semaphores (B4); coalesce/short-circuit the forced diagnostics rescan and allow disabling session diagnostics under load (B3). Low risk; directly attacks the storm.
- **Option B — Admission control / rate limiting (graceful degradation).** A connection rate-limiter at accept plus bounded session-establishment admission with a fast `BadServerTooBusy` and Retry-After, and an audit of the `BadTcpInternalError` catch-alls to avoid spurious aborts (B2). Turns the collapse into backpressure and lets clients ramp without creating duplicate sessions.
- **Option C — Steady-state scaling (well beyond a few thousand).** Decouple held Publishes from the worker-count accounting (S1) and make the diagnostics update incremental, so a small worker pool holds many thousands of long-polls. Needed to go far past the establishment ceiling once A and B land.
Beyond these, **horizontal scale-out** is the architectural answer: a single node's RSA-handshake throughput is bounded by its cores, so scale out across nodes behind a connection-distributing front. The stack already provides distributed mirroring of the address space, sessions, and subscriptions for exactly this high-availability / distributed topology.
## Configuration knobs available today
Until the above land, the following existing settings let you push a single node as far as its hardware allows (see also the sizing notes in [Benchmarks.md](Benchmarks.md#server-session-scalability)):
- **`MaxSessionCount`** (default 100) caps concurrently open sessions. Size **`MaxChannelCount`** (one channel per session) and **`MaxSubscriptionCount`** above the target session count.
- **`MaxRequestThreadCount`** caps concurrent request processing; because a held `Publish` occupies a worker slot for its wait (S1), a server serving N sessions needs `MaxRequestThreadCount` well above N or held Publishes starve other services with `BadRequestTimeout`. **`MinRequestThreadCount`** pre-warms the pool so a connect burst is not throttled by thread-pool cold-start.
- **`MaxFailedAuthenticationAttempts`** (default 5; `0` disables) is the per-certificate brute-force lockout; a single-certificate client opening many sessions can trip it on transient handshake failures and then be rejected with `BadUserAccessDenied`. Disable or raise it for bulk-connect clients.
- **Stagger the client connects.** Establishment is CPU-bound but parallelizes across cores; throttling and staggering concurrent connects (rather than bursting all at once) avoids self-inflicting the storm and is the single most effective client-side mitigation today.
## See also
- [Performance Benchmarks](Benchmarks.md#server-session-scalability) — the measured 500 / 1000 / 1500 / 2000 scaling table and the `ServerManySessionsLoadTestAsync` macro test.
- [Sessions, Reconnection, and Subscription Engines](Sessions.md) — session and subscription-engine architecture.
- [Subscriptions and Monitored Items Service Set](Subscriptions.md) — the V2 subscription engine.
- [Diagnostics](Diagnostics.md) — server diagnostics nodes (the source of the B3 cost) and how to control them.
Libraries/Opc.Ua.Server/Configuration/ConfigurationNodeManager.cs
+2 -1
View File
@@ -478,7 +478,8 @@ namespace Opc.Ua.Server
// add node as child of ServerNamespaces and in predefined nodes
serverNamespacesNode.AddChild(namespaceMetadataState);
serverNamespacesNode.ClearChangeMasks(SystemContext, true);
await serverNamespacesNode.ClearChangeMasksAsync(SystemContext, true, cancellationToken)
.ConfigureAwait(false);
await AddPredefinedNodeAsync(SystemContext, namespaceMetadataState, cancellationToken)
.ConfigureAwait(false);
}
Libraries/Opc.Ua.Server/KeyCredential/KeyCredentialPushSubject.cs
+3 -2
View File
@@ -138,7 +138,7 @@ namespace Opc.Ua.Server
}
}
folder.ClearChangeMasks(context, includeChildren: true);
await folder.ClearChangeMasksAsync(context, includeChildren: true, ct).ConfigureAwait(false);
}
private async ValueTask<CreateCredentialMethodStateResult> OnCreateCredentialAsync(
@@ -240,7 +240,8 @@ namespace Opc.Ua.Server
state.CredentialId.Value = credentialId;
state.ServiceStatus ??= state.CreateOrReplaceServiceStatus(context, state.ServiceStatus!);
state.ServiceStatus.Value = StatusCodes.Good;
state.ClearChangeMasks(context, includeChildren: true);
await state.ClearChangeMasksAsync(context, includeChildren: true, ct)
.ConfigureAwait(false);
}
return new KeyCredentialUpdateMethodStateResult { ServiceResult = ServiceResult.Good };
Libraries/Opc.Ua.Server/NodeManager/AsyncCustomNodeManager.cs
+26 -6
View File
@@ -1170,7 +1170,7 @@ namespace Opc.Ua.Server
{
lock (activeNode)
{
activeNode.OnReportEvent = OnReportEvent;
activeNode.OnReportEventAsync = OnReportEventAsync;
if (!activeNode.ReferenceExists(
ReferenceTypeIds.HasNotifier,
@@ -1215,7 +1215,7 @@ namespace Opc.Ua.Server
return;
}
node.UpdateChangeMasks(NodeStateChangeMasks.Deleted);
node.ClearChangeMasks(context, false);
await node.ClearChangeMasksAsync(context, false, cancellationToken).ConfigureAwait(false);
await OnNodeRemovedAsync(node, cancellationToken).ConfigureAwait(false);
// remove from the parent.
@@ -2691,7 +2691,8 @@ namespace Opc.Ua.Server
if (m_monitoredItemManager is MonitoredNodeMonitoredItemManager)
{
// updates to source finished - report changes to monitored items.
handle.Node.ClearChangeMasks(systemContext, true);
await handle.Node.ClearChangeMasksAsync(systemContext, true, cancellationToken)
.ConfigureAwait(false);
}
}
@@ -3124,7 +3125,8 @@ namespace Opc.Ua.Server
cancellationToken).ConfigureAwait(false);
// updates to source finished - report changes to monitored items.
source.ClearChangeMasks(context, false);
await source.ClearChangeMasksAsync(context, false, cancellationToken)
.ConfigureAwait(false);
}
finally
{
@@ -4523,7 +4525,7 @@ namespace Opc.Ua.Server
{
lock (notifier)
{
notifier.OnReportEvent = OnReportEvent;
notifier.OnReportEventAsync = OnReportEventAsync;
if (!notifier.ReferenceExists(
ReferenceTypeIds.HasNotifier,
@@ -4564,7 +4566,7 @@ namespace Opc.Ua.Server
{
lock (notifier)
{
notifier.OnReportEvent = null;
notifier.OnReportEventAsync = null;
if (notifier.ReferenceExists(
ReferenceTypeIds.HasNotifier,
@@ -4592,6 +4594,24 @@ namespace Opc.Ua.Server
Server.ReportEvent(context, e);
}
/// <summary>
/// Asynchronously reports an event for a root notifier, awaiting the server report so the
/// producing thread is never blocked. This is the sink wired to the root notifier; the
/// synchronous <see cref="OnReportEvent"/> is retained for compatibility.
/// </summary>
/// <param name="context">The system context.</param>
/// <param name="node">The notifier node.</param>
/// <param name="e">The event.</param>
/// <param name="cancellationToken">The cancellation token.</param>
protected virtual ValueTask OnReportEventAsync(
ISystemContext context,
NodeState node,
IFilterTarget e,
CancellationToken cancellationToken = default)
{
return Server.ReportEventAsync(context, e, cancellationToken);
}
/// <summary>
/// Subscribes to events.
/// </summary>
Libraries/Opc.Ua.Server/NodeManager/MonitoredItem/MonitoredNode.cs
+72 -38
View File
@@ -170,7 +170,7 @@ namespace Opc.Ua.Server
bool wasEmpty = DataChangeMonitoredItems.IsEmpty;
DataChangeMonitoredItems.TryAdd(datachangeItem.Id, datachangeItem);
Node.OnStateChanged = OnMonitoredNodeChanged;
Node.OnStateChangedAsync = OnMonitoredNodeChangedAsync;
// Subscribe to namespace default permission changes when the first item is added.
if (wasEmpty && m_server.ConfigurationNodeManager != null)
@@ -194,7 +194,7 @@ namespace Opc.Ua.Server
if (DataChangeMonitoredItems.IsEmpty)
{
Node.OnStateChanged = null;
Node.OnStateChangedAsync = null;
// Unsubscribe from namespace default permission changes when the last item is removed.
m_server.ConfigurationNodeManager?.DefaultPermissionsChanged -= OnDefaultPermissionsChanged;
@@ -209,7 +209,7 @@ namespace Opc.Ua.Server
{
EventMonitoredItems.TryAdd(eventItem.Id, eventItem);
Node.OnReportEvent = OnReportEvent;
Node.OnReportEventAsync = OnReportEventAsync;
// Scale up: add a consumer task for each new event MI beyond the first.
if (m_useMultipleConsumers && m_additionalConsumers != null)
@@ -238,7 +238,7 @@ namespace Opc.Ua.Server
if (EventMonitoredItems.IsEmpty)
{
Node.OnReportEvent = null;
Node.OnReportEventAsync = null;
}
// Scale down: remove a consumer task when MIs decrease (keep at least 1 total = primary).
@@ -257,12 +257,19 @@ namespace Opc.Ua.Server
}
/// <summary>
/// Called when a Node produces an event.
/// Called asynchronously when a Node produces an event. Clones the event state and enqueues
/// it for the consumer, awaiting the (bounded) channel so back-pressure is preserved without
/// blocking a thread.
/// </summary>
/// <param name="context">The system context.</param>
/// <param name="node">The affected node.</param>
/// <param name="e">The event.</param>
public void OnReportEvent(ISystemContext context, NodeState node, IFilterTarget e)
/// <param name="cancellationToken">The cancellation token.</param>
public async ValueTask OnReportEventAsync(
ISystemContext context,
NodeState node,
IFilterTarget e,
CancellationToken cancellationToken = default)
{
if (m_disposed)
{
@@ -284,7 +291,7 @@ namespace Opc.Ua.Server
try
{
m_channel.Writer.WriteAsync(notification).AsTask().GetAwaiter().GetResult();
await m_channel.Writer.WriteAsync(notification, cancellationToken).ConfigureAwait(false);
}
catch (ChannelClosedException)
{
@@ -293,15 +300,37 @@ namespace Opc.Ua.Server
}
/// <summary>
/// Called when the state of a Node changes.
/// Synchronous wrapper over <see cref="OnReportEventAsync"/>. Completes inline when the
/// channel has capacity; blocks only when the bounded channel is full (only synchronous
/// callers pay that cost).
/// </summary>
/// <param name="context">The system context.</param>
/// <param name="node">The affected node.</param>
/// <param name="e">The event.</param>
public void OnReportEvent(ISystemContext context, NodeState node, IFilterTarget e)
{
ValueTask report = OnReportEventAsync(context, node, e, default);
if (!report.IsCompletedSuccessfully)
{
report.AsTask().GetAwaiter().GetResult();
}
}
/// <summary>
/// Called asynchronously when the state of a Node changes. Reads each monitored attribute at
/// enqueue time - through the async entry point, so a genuinely asynchronous read handler is
/// honored - and enqueues the snapshot, awaiting the bounded channel for back-pressure. No
/// thread is blocked.
/// </summary>
/// <param name="context">The system context.</param>
/// <param name="node">The affected node.</param>
/// <param name="changes">The mask indicating what changes have occurred.</param>
public void OnMonitoredNodeChanged(
/// <param name="cancellationToken">The cancellation token.</param>
public async ValueTask OnMonitoredNodeChangedAsync(
ISystemContext context,
NodeState node,
NodeStateChangeMasks changes)
NodeStateChangeMasks changes,
CancellationToken cancellationToken = default)
{
if (m_disposed)
{
@@ -342,24 +371,17 @@ namespace Opc.Ua.Server
m_timeProvider.GetUtcNow().UtcDateTime,
DateTime.MinValue);
// Read through the async entry point so a node that registers a
// genuinely asynchronous value read handler (OnReadValueAsync /
// OnSimpleReadValueAsync) is honored. ReadAttributeAsync is the
// unified path: when no async handler is registered it just locks
// the node and calls the synchronous ReadAttribute, completing
// synchronously - so the common path unwraps an already-completed
// ValueTask without blocking, and only a node that opts into a
// real async read incurs a (its own) synchronous wait here.
ValueTask<(ServiceResult Result, DataValue Value)> readTask = node.ReadAttributeAsync(
// Read at enqueue time via the async entry point: ReadAttributeAsync honors an
// asynchronous value read handler (OnReadValueAsync) when one is registered and
// otherwise completes synchronously. The value is therefore materialised for this
// change before it is enqueued (strict enqueue-time snapshot) without blocking.
(_, attributeSnapshots[attributeId]) = await node.ReadAttributeAsync(
context,
attributeId,
default,
QualifiedName.Null,
dataValue);
attributeSnapshots[attributeId] = readTask.IsCompletedSuccessfully
? readTask.Result.Value
: readTask.AsTask().GetAwaiter().GetResult().Value;
dataValue,
cancellationToken).ConfigureAwait(false);
}
var notification = new DataChangeSnapshot
@@ -370,21 +392,33 @@ namespace Opc.Ua.Server
AttributeSnapshots = attributeSnapshots
};
// Enqueue without blocking on the common path. The bounded channel only
// makes the caller wait when the consumer is severely backed up
// (capacity is per node); fall back to the blocking write in that rare
// case so backpressure is preserved without a sync-over-async on the
// hot notification path.
if (!m_channel.Writer.TryWrite(notification))
try
{
try
{
m_channel.Writer.WriteAsync(notification).AsTask().GetAwaiter().GetResult();
}
catch (ChannelClosedException)
{
// The channel was completed during shutdown/disposal.
}
await m_channel.Writer.WriteAsync(notification, cancellationToken).ConfigureAwait(false);
}
catch (ChannelClosedException)
{
// The channel was completed during shutdown/disposal.
}
}
/// <summary>
/// Synchronous wrapper over <see cref="OnMonitoredNodeChangedAsync"/>. Completes inline for a
/// synchronously-readable node with channel capacity; blocks only when a read is genuinely
/// asynchronous or the bounded channel is full (only synchronous callers pay that cost).
/// </summary>
/// <param name="context">The system context.</param>
/// <param name="node">The affected node.</param>
/// <param name="changes">The mask indicating what changes have occurred.</param>
public void OnMonitoredNodeChanged(
ISystemContext context,
NodeState node,
NodeStateChangeMasks changes)
{
ValueTask change = OnMonitoredNodeChangedAsync(context, node, changes, default);
if (!change.IsCompletedSuccessfully)
{
change.AsTask().GetAwaiter().GetResult();
}
}
Libraries/Opc.Ua.Server/Server/IServerInternal.cs
+20
View File
@@ -295,6 +295,26 @@ namespace Opc.Ua.Server
/// <param name="e">The event.</param>
void ReportEvent(ISystemContext context, IFilterTarget e);
/// <summary>
/// Asynchronously reports a global event, awaiting an asynchronous report sink so the caller
/// is never blocked.
/// </summary>
/// <param name="e">The event.</param>
/// <param name="cancellationToken">The cancellation token.</param>
ValueTask ReportEventAsync(IFilterTarget e, CancellationToken cancellationToken = default);
/// <summary>
/// Asynchronously reports a global event, awaiting an asynchronous report sink so the caller
/// is never blocked.
/// </summary>
/// <param name="context">The context.</param>
/// <param name="e">The event.</param>
/// <param name="cancellationToken">The cancellation token.</param>
ValueTask ReportEventAsync(
ISystemContext context,
IFilterTarget e,
CancellationToken cancellationToken = default);
/// <summary>
/// Refreshes the conditions for the specified subscription.
/// </summary>
Libraries/Opc.Ua.Server/Server/ServerInternalData.cs
+37
View File
@@ -684,6 +684,43 @@ namespace Opc.Ua.Server
ServerObject?.ReportEvent(context, e);
}
/// <summary>
/// Called by any component to asynchronously report a global event.
/// </summary>
/// <param name="e">The event.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public ValueTask ReportEventAsync(IFilterTarget e, CancellationToken cancellationToken = default)
{
return ReportEventAsync(DefaultSystemContext, e, cancellationToken);
}
/// <summary>
/// Called by any component to asynchronously report a global event, awaiting an asynchronous
/// report sink so the caller is never blocked.
/// </summary>
/// <param name="context">The context.</param>
/// <param name="e">The event.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public ValueTask ReportEventAsync(
ISystemContext context,
IFilterTarget e,
CancellationToken cancellationToken = default)
{
if ((!Auditing) && (e is AuditEventState))
{
// do not report auditing events if server Auditing flag is false
return default;
}
NodeState? serverObject = ServerObject;
if (serverObject == null)
{
return default;
}
return serverObject.ReportEventAsync(context, e, cancellationToken);
}
/// <summary>
/// Refreshes the conditions for the specified subscription.
/// </summary>
Stack/Opc.Ua.Types/State/BaseInstanceState.cs
+23
View File
@@ -30,6 +30,8 @@
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Opc.Ua.Types;
namespace Opc.Ua
@@ -295,6 +297,27 @@ namespace Opc.Ua
Parent?.ReportEvent(context, e);
}
/// <summary>
/// Asynchronously reports an event and recursively notifies the parent, awaiting any
/// asynchronous report sink so the caller is never blocked.
/// </summary>
/// <param name="context">The system context.</param>
/// <param name="e">The event to report.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public override async ValueTask ReportEventAsync(
ISystemContext context,
IFilterTarget e,
CancellationToken cancellationToken = default)
{
await base.ReportEventAsync(context, e, cancellationToken).ConfigureAwait(false);
// recursively notify the parent.
if (Parent != null)
{
await Parent.ReportEventAsync(context, e, cancellationToken).ConfigureAwait(false);
}
}
/// <summary>
/// Sets the minimum sampling interval for the node an all of its child variables..
/// </summary>
Stack/Opc.Ua.Types/State/NodeState.cs
+207
View File
@@ -2256,6 +2256,12 @@ namespace Opc.Ua
/// </summary>
public event NodeStateChangedHandler? StateChanged;
/// <summary>
/// Asynchronous counterpart of <see cref="StateChanged"/>, awaited by
/// <see cref="ClearChangeMasksAsync"/>.
/// </summary>
public event NodeStateChangedAsyncHandler? StateChangedAsync;
/// <summary>
/// Called when the Validate method is called
/// </summary>
@@ -2266,6 +2272,13 @@ namespace Opc.Ua
/// </summary>
public NodeStateChangedHandler? OnStateChanged;
/// <summary>
/// Asynchronous counterpart of <see cref="OnStateChanged"/>. When set it is awaited by
/// <see cref="ClearChangeMasksAsync"/>; the synchronous <see cref="ClearChangeMasks"/>
/// still invokes it but blocks until it completes (only synchronous callers pay that cost).
/// </summary>
public NodeStateChangedAsyncHandler? OnStateChangedAsync;
/// <summary>
/// Called when a reference gets added to the node
/// </summary>
@@ -2281,6 +2294,13 @@ namespace Opc.Ua
/// </summary>
public NodeStateReportEventHandler? OnReportEvent;
/// <summary>
/// Asynchronous counterpart of <see cref="OnReportEvent"/>. When set it is awaited by
/// <see cref="ReportEventAsync"/>; the synchronous <see cref="ReportEvent"/> still invokes
/// it but blocks until it completes (only synchronous callers pay that cost).
/// </summary>
public NodeStateReportEventAsyncHandler? OnReportEventAsync;
/// <summary>
/// Called when ClearChangeMasks is called and the ChangeMask is not None.
/// </summary>
@@ -2510,6 +2530,33 @@ namespace Opc.Ua
{
OnReportEvent?.Invoke(context, this, e);
// Drive an asynchronous report sink too, so a node whose only report sink is
// asynchronous is still notified when the synchronous API is used. Completes inline
// in the common case; a genuinely asynchronous sink blocks here - only synchronous
// callers pay that cost.
NodeStateReportEventAsyncHandler? onReportEventAsync = OnReportEventAsync;
if (onReportEventAsync != null)
{
if (SynchronizationContext.Current == null)
{
// No ambient synchronization context (the normal server case): drive inline
// and block only on the rare genuinely-asynchronous sink.
ValueTask report = onReportEventAsync(context, this, e, default);
if (!report.IsCompletedSuccessfully)
{
report.AsTask().GetAwaiter().GetResult();
}
}
else
{
// A synchronization context is present (e.g. a UI / legacy ASP.NET thread):
// run the sink on the thread pool so a context-capturing continuation cannot
// deadlock the blocking wait below.
Task.Run(() => onReportEventAsync(context, this, e, CancellationToken.None).AsTask())
.GetAwaiter().GetResult();
}
}
List<Notifier>? notifiers;
lock (m_notifiersLock)
@@ -2531,6 +2578,49 @@ namespace Opc.Ua
}
}
/// <summary>
/// Asynchronously reports an event produced by the node, awaiting an asynchronous report
/// sink (see <see cref="OnReportEventAsync"/>) so the caller is never blocked. A synchronous
/// sink is invoked as well.
/// </summary>
/// <param name="context">The system context.</param>
/// <param name="e">The event to report.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public virtual async ValueTask ReportEventAsync(
ISystemContext context,
IFilterTarget e,
CancellationToken cancellationToken = default)
{
OnReportEvent?.Invoke(context, this, e);
NodeStateReportEventAsyncHandler? onReportEventAsync = OnReportEventAsync;
if (onReportEventAsync != null)
{
await onReportEventAsync(context, this, e, cancellationToken).ConfigureAwait(false);
}
List<Notifier>? notifiers;
lock (m_notifiersLock)
{
notifiers = m_notifiers != null ? [.. m_notifiers] : null;
}
// report event to notifier sources.
if (notifiers != null)
{
for (int ii = 0; ii < notifiers.Count; ii++)
{
NodeState? node = notifiers[ii].Node;
if (notifiers[ii].IsInverse && node != null)
{
await node.ReportEventAsync(context, e, cancellationToken)
.ConfigureAwait(false);
}
}
}
}
/// <summary>
/// Adds a notifier relationship to the node.
/// </summary>
@@ -2773,10 +2863,105 @@ namespace Opc.Ua
{
OnStateChanged?.Invoke(context, this, changeMasks);
StateChanged?.Invoke(context, this, changeMasks);
// Drive any asynchronous sinks too, so a node whose only state-changed sink is
// asynchronous (for example a monitored-item manager) is still notified when the
// synchronous API is used. Completes inline in the common case; a genuinely
// asynchronous sink blocks here - only synchronous callers pay that cost.
if (OnStateChangedAsync != null || StateChangedAsync != null)
{
if (SynchronizationContext.Current == null)
{
// No ambient synchronization context (the normal server case): drive inline
// and block only on the rare genuinely-asynchronous sink.
ValueTask raise = RaiseStateChangedAsync(context, changeMasks, default);
if (!raise.IsCompletedSuccessfully)
{
raise.AsTask().GetAwaiter().GetResult();
}
}
else
{
// A synchronization context is present (e.g. a UI / legacy ASP.NET thread):
// run the sinks on the thread pool so a context-capturing continuation
// cannot deadlock the blocking wait below.
Task.Run(() =>
RaiseStateChangedAsync(context, changeMasks, CancellationToken.None)
.AsTask())
.GetAwaiter().GetResult();
}
}
m_changeMasks = NodeStateChangeMasks.None;
}
}
/// <summary>
/// Asynchronously clears the change masks, awaiting any asynchronous state-changed sinks
/// (see <see cref="OnStateChangedAsync"/> / <see cref="StateChangedAsync"/>) so the caller
/// is never blocked. Synchronous sinks are invoked as well.
/// </summary>
/// <param name="context">The context that describes how to access the system containing the data.</param>
/// <param name="includeChildren">if set to <c>true</c> clear masks recursively for all children.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public async ValueTask ClearChangeMasksAsync(
ISystemContext context,
bool includeChildren,
CancellationToken cancellationToken = default)
{
if (includeChildren)
{
var children = new List<BaseInstanceState>();
GetChildren(context, children);
for (int ii = 0; ii < children.Count; ii++)
{
await children[ii]
.ClearChangeMasksAsync(context, true, cancellationToken)
.ConfigureAwait(false);
}
}
NodeStateChangeMasks changeMasks = m_changeMasks;
if (changeMasks != NodeStateChangeMasks.None)
{
OnStateChanged?.Invoke(context, this, changeMasks);
StateChanged?.Invoke(context, this, changeMasks);
await RaiseStateChangedAsync(context, changeMasks, cancellationToken)
.ConfigureAwait(false);
m_changeMasks = NodeStateChangeMasks.None;
}
}
/// <summary>
/// Awaits the asynchronous state-changed sinks (<see cref="OnStateChangedAsync"/> and every
/// sink registered on the <see cref="StateChangedAsync"/> event) in registration order.
/// </summary>
private async ValueTask RaiseStateChangedAsync(
ISystemContext context,
NodeStateChangeMasks changeMasks,
CancellationToken cancellationToken)
{
NodeStateChangedAsyncHandler? onStateChangedAsync = OnStateChangedAsync;
if (onStateChangedAsync != null)
{
await onStateChangedAsync(context, this, changeMasks, cancellationToken)
.ConfigureAwait(false);
}
NodeStateChangedAsyncHandler? stateChangedAsync = StateChangedAsync;
if (stateChangedAsync != null)
{
foreach (Delegate handler in stateChangedAsync.GetInvocationList())
{
await ((NodeStateChangedAsyncHandler)handler)(
context, this, changeMasks, cancellationToken)
.ConfigureAwait(false);
}
}
}
/// <summary>
/// Recursively sets the status code and timestamp for the node and all child variables.
/// </summary>
@@ -5325,6 +5510,17 @@ namespace Opc.Ua
NodeState node,
NodeStateChangeMasks changes);
/// <summary>
/// Asynchronous counterpart of <see cref="NodeStateChangedHandler"/>, invoked by
/// <see cref="NodeState.ClearChangeMasksAsync"/> so a sink can perform asynchronous work
/// (for example an asynchronous attribute read) without blocking the caller.
/// </summary>
public delegate ValueTask NodeStateChangedAsyncHandler(
ISystemContext context,
NodeState node,
NodeStateChangeMasks changes,
CancellationToken cancellationToken);
/// <summary>
/// Used to receive notifications when a reference get added to the node
/// </summary>
@@ -5351,6 +5547,17 @@ namespace Opc.Ua
NodeState node,
IFilterTarget e);
/// <summary>
/// Asynchronous counterpart of <see cref="NodeStateReportEventHandler"/>, invoked by
/// <see cref="NodeState.ReportEventAsync"/> so a sink can enqueue the event without
/// blocking the caller.
/// </summary>
public delegate ValueTask NodeStateReportEventAsyncHandler(
ISystemContext context,
NodeState node,
IFilterTarget e,
CancellationToken cancellationToken);
/// <summary>
/// Used to receive notifications when a node needs to refresh its conditions.
/// </summary>
Tests/Opc.Ua.Server.Tests/AsyncCustomNodeManagerTests.cs
+22 -6
View File
@@ -2186,7 +2186,9 @@ namespace Opc.Ua.Server.Tests
await manager.AddRootNotifierPublicAsync(notifier).ConfigureAwait(false);
Assert.That(notifier.OnReportEvent, Is.Not.Null);
Assert.That(
notifier.OnReportEvent != null || notifier.OnReportEventAsync != null,
Is.True);
}
[Test]
@@ -2221,7 +2223,9 @@ namespace Opc.Ua.Server.Tests
await manager.AddRootNotifierPublicAsync(serverNode).ConfigureAwait(false);
Assert.That(manager.RootNotifiers.ContainsKey(ObjectIds.Server), Is.True);
Assert.That(serverNode.OnReportEvent, Is.Null);
Assert.That(
serverNode.OnReportEvent == null && serverNode.OnReportEventAsync == null,
Is.True);
Assert.That(
serverNode.ReferenceExists(ReferenceTypeIds.HasNotifier, true, ObjectIds.Server),
Is.False);
@@ -2281,11 +2285,15 @@ namespace Opc.Ua.Server.Tests
notifier.BrowseName = new QualifiedName("Notifier", nsIdx);
await manager.AddRootNotifierPublicAsync(notifier).ConfigureAwait(false);
Assert.That(notifier.OnReportEvent, Is.Not.Null);
Assert.That(
notifier.OnReportEvent != null || notifier.OnReportEventAsync != null,
Is.True);
await manager.RemoveRootNotifierPublicAsync(notifier).ConfigureAwait(false);
Assert.That(notifier.OnReportEvent, Is.Null);
Assert.That(
notifier.OnReportEvent == null && notifier.OnReportEventAsync == null,
Is.True);
}
[Test]
@@ -2354,12 +2362,20 @@ namespace Opc.Ua.Server.Tests
[Test]
public async Task OnReportEventIsInvokedWhenNodeReportsEventAsync()
{
// Verifies that the callback wired by AddRootNotifierAsync routes events
// through to IServerInternal.ReportEvent.
// Verifies that the callback wired by AddRootNotifierAsync routes events through to
// IServerInternal - via ReportEvent for the synchronous manager and ReportEventAsync
// for the asynchronous manager.
IFilterTarget capturedEvent = null;
m_mockServer
.Setup(s => s.ReportEvent(It.IsAny<ISystemContext>(), It.IsAny<IFilterTarget>()))
.Callback<ISystemContext, IFilterTarget>((_, e) => capturedEvent = e);
m_mockServer
.Setup(s => s.ReportEventAsync(
It.IsAny<ISystemContext>(),
It.IsAny<IFilterTarget>(),
It.IsAny<CancellationToken>()))
.Returns(default(ValueTask))
.Callback<ISystemContext, IFilterTarget, CancellationToken>((_, e, _) => capturedEvent = e);
using ITestNodeManager manager = CreateManager();
ushort nsIdx = manager.NamespaceIndexes[0];
Tests/Opc.Ua.Server.Tests/MonitoredNode2Tests.cs
+195 -4
View File
@@ -200,7 +200,7 @@ namespace Opc.Ua.Server.Tests
// Assert – QueueValue should never have been called (permission denied)
monitoredItemMock.Verify(
m => m.QueueValue(It.IsAny<DataValue>(), It.IsAny<ServiceResult>()),
m => m.QueueValue(It.Ref<DataValue>.IsAny, It.IsAny<ServiceResult>()),
Times.Never);
// And validate was only called once (cached bad result)
@@ -325,7 +325,7 @@ namespace Opc.Ua.Server.Tests
using var firstItemProcessed = new ManualResetEventSlim(false);
Mock<IDataChangeMonitoredItem2> monitoredItemMock = CreateDataChangeMonitoredItemMock(1u, Attributes.Value);
monitoredItemMock
.Setup(m => m.QueueValue(It.IsAny<DataValue>(), It.IsAny<ServiceResult>()))
.Setup(m => m.QueueValue(It.Ref<DataValue>.IsAny, It.IsAny<ServiceResult>()))
.Callback(firstItemProcessed.Set);
var monitoredNode = new MonitoredNode2(nodeManagerMock.Object, serverMock.Object, node);
@@ -1129,6 +1129,197 @@ namespace Opc.Ua.Server.Tests
Assert.That(queueCount, Is.Zero);
}
/// <summary>
/// Verifies that a node which exposes a genuinely asynchronous value read handler
/// (<c>OnSimpleReadValueAsync</c>) has that handler honored on the change-notification
/// push path: the value delivered to the monitored item is the asynchronously-read value,
/// not the cached synchronous value.
/// </summary>
[Test]
public async Task OnMonitoredNodeChangedAsyncHonorsAsyncReadHandler()
{
var node = new BaseDataVariableState(null)
{
NodeId = new NodeId("asyncNode", 1),
BrowseName = new QualifiedName("asyncNode", 1),
DataType = DataTypeIds.Int32,
AccessLevel = AccessLevels.CurrentRead,
UserAccessLevel = AccessLevels.CurrentRead,
Value = 0
};
// The asynchronous read handler returns a value that differs from the cached one.
node.OnSimpleReadValueAsync = (c, n, ct) => new ValueTask<AttributeSimpleReadResult>(
new AttributeSimpleReadResult(ServiceResult.Good, new Variant(777)));
var nodeManagerMock = new Mock<IAsyncNodeManager>();
nodeManagerMock
.Setup(m => m.ValidateRolePermissionsAsync(
It.IsAny<OperationContext>(),
It.IsAny<NodeId>(),
It.IsAny<PermissionType>(),
It.IsAny<CancellationToken>()))
.Returns(new ValueTask<ServiceResult>(ServiceResult.Good));
var serverMock = new Mock<IServerInternal>();
serverMock.Setup(s => s.Auditing).Returns(false);
using var delivered = new ManualResetEventSlim(false);
Mock<IDataChangeMonitoredItem2> itemMock = CreateDataChangeMonitoredItemMock(1u, Attributes.Value);
itemMock
.Setup(m => m.QueueValue(It.Ref<DataValue>.IsAny, It.IsAny<ServiceResult>()))
.Callback(delivered.Set);
var monitoredNode = new MonitoredNode2(nodeManagerMock.Object, serverMock.Object, node);
monitoredNode.Add(itemMock.Object);
ISystemContext context = new Mock<ISystemContext>().Object;
await monitoredNode
.OnMonitoredNodeChangedAsync(context, node, NodeStateChangeMasks.Value)
.ConfigureAwait(false);
Assert.That(
delivered.Wait(TimeSpan.FromSeconds(30)),
Is.True,
"the value change was not delivered within the timeout");
monitoredNode.Dispose();
var deliveredValues = itemMock.Invocations
.Where(i => i.Method.Name == nameof(IDataChangeMonitoredItem2.QueueValue))
.Select(i => ((DataValue)i.Arguments[0]).WrappedValue.AsBoxedObject())
.ToList();
Assert.That(deliveredValues, Has.Count.EqualTo(1));
Assert.That(deliveredValues[0], Is.EqualTo(777));
}
/// <summary>
/// Verifies that the asynchronous change-notification producer does not block a thread while
/// a genuinely asynchronous read handler is in flight: the returned <see cref="ValueTask"/>
/// is not yet completed while the read is pending, and the value is delivered once the read
/// completes.
/// </summary>
[Test]
public async Task OnMonitoredNodeChangedAsyncDoesNotBlockOnSlowAsyncRead()
{
using var readStarted = new ManualResetEventSlim(false);
using var releaseRead = new ManualResetEventSlim(false);
var node = new BaseDataVariableState(null)
{
NodeId = new NodeId("slowAsyncNode", 1),
BrowseName = new QualifiedName("slowAsyncNode", 1),
DataType = DataTypeIds.Int32,
AccessLevel = AccessLevels.CurrentRead,
UserAccessLevel = AccessLevels.CurrentRead,
Value = 0
};
node.OnReadValueAsync = async (c, n, range, encoding, ct) =>
{
readStarted.Set();
// Wait off-thread so the producer cannot be completing synchronously.
await Task.Run(() => releaseRead.Wait(TimeSpan.FromSeconds(30)), ct).ConfigureAwait(false);
return new AttributeReadResult(
ServiceResult.Good, new Variant(555), StatusCodes.Good, DateTimeUtc.Now);
};
var nodeManagerMock = new Mock<IAsyncNodeManager>();
nodeManagerMock
.Setup(m => m.ValidateRolePermissionsAsync(
It.IsAny<OperationContext>(),
It.IsAny<NodeId>(),
It.IsAny<PermissionType>(),
It.IsAny<CancellationToken>()))
.Returns(new ValueTask<ServiceResult>(ServiceResult.Good));
var serverMock = new Mock<IServerInternal>();
serverMock.Setup(s => s.Auditing).Returns(false);
using var delivered = new ManualResetEventSlim(false);
Mock<IDataChangeMonitoredItem2> itemMock = CreateDataChangeMonitoredItemMock(1u, Attributes.Value);
itemMock
.Setup(m => m.QueueValue(It.Ref<DataValue>.IsAny, It.IsAny<ServiceResult>()))
.Callback(delivered.Set);
var monitoredNode = new MonitoredNode2(nodeManagerMock.Object, serverMock.Object, node);
monitoredNode.Add(itemMock.Object);
ISystemContext context = new Mock<ISystemContext>().Object;
ValueTask produce = monitoredNode.OnMonitoredNodeChangedAsync(
context, node, NodeStateChangeMasks.Value);
Assert.That(readStarted.Wait(TimeSpan.FromSeconds(30)), Is.True);
Assert.That(
produce.IsCompleted,
Is.False,
"the producer must not block a thread while an asynchronous read is in flight");
releaseRead.Set();
await produce.ConfigureAwait(false);
Assert.That(
delivered.Wait(TimeSpan.FromSeconds(30)),
Is.True,
"the value change was not delivered within the timeout");
monitoredNode.Dispose();
var deliveredValues = itemMock.Invocations
.Where(i => i.Method.Name == nameof(IDataChangeMonitoredItem2.QueueValue))
.Select(i => ((DataValue)i.Arguments[0]).WrappedValue.AsBoxedObject())
.ToList();
Assert.That(deliveredValues, Has.Count.EqualTo(1));
Assert.That(deliveredValues[0], Is.EqualTo(555));
}
/// <summary>
/// Verifies that an event reported through the asynchronous <c>OnReportEventAsync</c> path is
/// delivered to the registered event monitored item via the channel consumer.
/// </summary>
[Test]
public async Task OnReportEventAsyncDeliversEventToMonitoredItem()
{
var node = new BaseDataVariableState(null)
{
NodeId = new NodeId("eventNode", 1),
BrowseName = new QualifiedName("eventNode", 1),
DataType = DataTypeIds.Int32
};
var nodeManagerMock = new Mock<IAsyncNodeManager>();
nodeManagerMock
.Setup(m => m.ValidateEventRolePermissionsAsync(
It.IsAny<IEventMonitoredItem>(),
It.IsAny<IFilterTarget>(),
It.IsAny<CancellationToken>()))
.Returns(new ValueTask<ServiceResult>(ServiceResult.Good));
var serverMock = new Mock<IServerInternal>();
serverMock.Setup(s => s.Auditing).Returns(false);
using var delivered = new ManualResetEventSlim(false);
Mock<IEventMonitoredItem> eventItemMock = CreateEventMonitoredItemMock(1u);
eventItemMock
.Setup(m => m.QueueEvent(It.IsAny<IFilterTarget>()))
.Callback(delivered.Set);
var monitoredNode = new MonitoredNode2(nodeManagerMock.Object, serverMock.Object, node);
monitoredNode.Add(eventItemMock.Object);
var eventState = new BaseEventState(null);
ISystemContext context = new Mock<ISystemContext>().Object;
await monitoredNode.OnReportEventAsync(context, node, eventState).ConfigureAwait(false);
Assert.That(
delivered.Wait(TimeSpan.FromSeconds(30)),
Is.True,
"the event was not delivered within the timeout");
monitoredNode.Dispose();
int queueCount = eventItemMock.Invocations
.Count(i => i.Method.Name == nameof(IEventMonitoredItem.QueueEvent));
Assert.That(queueCount, Is.EqualTo(1));
}
private static Mock<IEventMonitoredItem> CreateEventMonitoredItemMock(uint id)
{
var sessionMock = new Mock<ISession>();
@@ -1212,7 +1403,7 @@ namespace Opc.Ua.Server.Tests
Mock<IDataChangeMonitoredItem2> monitoredItemMock =
CreateDataChangeMonitoredItemMockWithSession(1u, Attributes.Value, sessionId);
monitoredItemMock
.Setup(m => m.QueueValue(It.IsAny<DataValue>(), It.IsAny<ServiceResult>()))
.Setup(m => m.QueueValue(It.Ref<DataValue>.IsAny, It.IsAny<ServiceResult>()))
.Callback(firstItemProcessed.Set);
var monitoredNode = new MonitoredNode2(nodeManagerMock.Object, serverMock.Object, node);
@@ -1278,7 +1469,7 @@ namespace Opc.Ua.Server.Tests
Mock<IDataChangeMonitoredItem2> monitoredItemMock =
CreateDataChangeMonitoredItemMockWithSession(1u, Attributes.Value, sessionId);
monitoredItemMock
.Setup(m => m.QueueValue(It.IsAny<DataValue>(), It.IsAny<ServiceResult>()))
.Setup(m => m.QueueValue(It.Ref<DataValue>.IsAny, It.IsAny<ServiceResult>()))
.Callback(firstItemProcessed.Set);
var monitoredNode = new MonitoredNode2(nodeManagerMock.Object, serverMock.Object, node);
Tests/Opc.Ua.Types.Tests/State/NodeStateChangeNotificationTests.cs
+198
View File
@@ -0,0 +1,198 @@
/* ========================================================================
* 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/
* ======================================================================*/
using System.Threading.Tasks;
using NUnit.Framework;
using Opc.Ua.Tests;
namespace Opc.Ua.Types.Tests.State
{
/// <summary>
/// Tests for the asynchronous change-notification API on <see cref="NodeState"/>:
/// <see cref="NodeState.ClearChangeMasksAsync"/> / <see cref="NodeState.OnStateChangedAsync"/>
/// and <see cref="NodeState.ReportEventAsync"/> / <see cref="NodeState.OnReportEventAsync"/>,
/// including that the synchronous entry points still drive the asynchronous sinks.
/// </summary>
[TestFixture]
[Category("NodeState")]
[SetCulture("en-us")]
[SetUICulture("en-us")]
[Parallelizable]
public class NodeStateChangeNotificationTests
{
private ITelemetryContext m_telemetry;
private ServiceMessageContext m_messageContext;
[OneTimeSetUp]
protected void OneTimeSetUp()
{
m_telemetry = NUnitTelemetryContext.Create();
m_messageContext = ServiceMessageContext.CreateEmpty(m_telemetry);
}
[OneTimeTearDown]
protected void OneTimeTearDown()
{
(m_messageContext as System.IDisposable)?.Dispose();
}
private SystemContext CreateSystemContext()
{
return new SystemContext(m_telemetry)
{
NamespaceUris = m_messageContext.NamespaceUris,
TypeTable = new TypeTable(m_messageContext.NamespaceUris)
};
}
private static BaseDataVariableState CreateVariable()
{
return new BaseDataVariableState(null)
{
NodeId = new NodeId("Var", 0),
BrowseName = new QualifiedName("Var", 0),
DisplayName = new LocalizedText("Var"),
DataType = DataTypeIds.Double,
ValueRank = ValueRanks.Scalar,
AccessLevel = AccessLevels.CurrentReadOrWrite,
UserAccessLevel = AccessLevels.CurrentReadOrWrite
};
}
[Test]
public async Task ClearChangeMasksAsyncInvokesAsyncAndSyncSinks()
{
SystemContext ctx = CreateSystemContext();
BaseDataVariableState v = CreateVariable();
// Flush any change masks set while constructing the node so the assertion below
// observes only the mask raised by this test.
v.ClearChangeMasks(ctx, includeChildren: false);
NodeStateChangeMasks syncMask = NodeStateChangeMasks.None;
NodeStateChangeMasks asyncMask = NodeStateChangeMasks.None;
v.OnStateChanged = (c, n, mask) => syncMask = mask;
v.OnStateChangedAsync = (c, n, mask, ct) =>
{
asyncMask = mask;
return default;
};
v.UpdateChangeMasks(NodeStateChangeMasks.Value);
await v.ClearChangeMasksAsync(ctx, includeChildren: false).ConfigureAwait(false);
Assert.That(syncMask, Is.EqualTo(NodeStateChangeMasks.Value));
Assert.That(asyncMask, Is.EqualTo(NodeStateChangeMasks.Value));
}
[Test]
public void ClearChangeMasksSyncDrivesAsyncSink()
{
SystemContext ctx = CreateSystemContext();
BaseDataVariableState v = CreateVariable();
// Flush any change masks set while constructing the node.
v.ClearChangeMasks(ctx, includeChildren: false);
NodeStateChangeMasks asyncMask = NodeStateChangeMasks.None;
v.OnStateChangedAsync = (c, n, mask, ct) =>
{
asyncMask = mask;
return default;
};
v.UpdateChangeMasks(NodeStateChangeMasks.Value);
v.ClearChangeMasks(ctx, includeChildren: false);
Assert.That(asyncMask, Is.EqualTo(NodeStateChangeMasks.Value));
}
[Test]
public async Task ClearChangeMasksAsyncAwaitsStateChangedAsyncEvent()
{
SystemContext ctx = CreateSystemContext();
BaseDataVariableState v = CreateVariable();
bool invoked = false;
v.StateChangedAsync += (c, n, mask, ct) =>
{
invoked = true;
return default;
};
v.UpdateChangeMasks(NodeStateChangeMasks.Value);
await v.ClearChangeMasksAsync(ctx, includeChildren: false).ConfigureAwait(false);
Assert.That(invoked, Is.True);
}
[Test]
public async Task ReportEventAsyncInvokesAsyncAndSyncSinks()
{
SystemContext ctx = CreateSystemContext();
BaseDataVariableState v = CreateVariable();
var target = new BaseObjectState(null);
IFilterTarget syncTarget = null;
IFilterTarget asyncTarget = null;
v.OnReportEvent = (c, n, e) => syncTarget = e;
v.OnReportEventAsync = (c, n, e, ct) =>
{
asyncTarget = e;
return default;
};
await v.ReportEventAsync(ctx, target).ConfigureAwait(false);
Assert.That(syncTarget, Is.SameAs(target));
Assert.That(asyncTarget, Is.SameAs(target));
}
[Test]
public void ReportEventSyncDrivesAsyncSink()
{
SystemContext ctx = CreateSystemContext();
BaseDataVariableState v = CreateVariable();
var target = new BaseObjectState(null);
IFilterTarget asyncTarget = null;
v.OnReportEventAsync = (c, n, e, ct) =>
{
asyncTarget = e;
return default;
};
v.ReportEvent(ctx, target);
Assert.That(asyncTarget, Is.SameAs(target));
}
}
}