dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(opcuaclient): split the read gate from subscription management; no gate on writes or the discovery browse (R06:O-6)

Browse Source 
One SemaphoreSlim serialised reads, writes, subscribes, acks and the whole recursive
discovery browse, so a stalled write (up to the operation timeout) blocked every
read and subscription create, and a discovery browse would block the data plane for
its full duration. Reads/acks keep the read gate; writes and DiscoverAsync are
ungated (the SDK session is safe for concurrent service calls, the host bounds write
concurrency); subscription management takes its own gate. Dispose drains both
(DisposeAsync's drain no longer over-releases on a timed-out wait).

Refs #743

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 15:24:40 -04:00
parent afafd483dd
commit 66d100a9e9
3 changed files with 293 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
+125 -118
View File
@@ -158,9 +158,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <summary>Active OPC UA session. Null until <see cref="InitializeAsync"/> returns cleanly.</summary>
internal ISession? Session { get; private set; }
/// <summary>Per-connection gate. PRs 67+ serialize read/write/browse on this.</summary>
/// <summary>
/// R06:O-6 — subscription management (data + alarm subscribe/unsubscribe, ConditionRefresh sweeps) has its own
/// gate, so a slow subscription create no longer queues reads, and the reconcile reads never wait on it.
/// </summary>
private readonly SemaphoreSlim _subscriptionGate = new(1, 1);
/// <summary>Read gate: serialises reads (incl. the recovery / reconcile reads) and acknowledges. Writes and the
/// discovery browse take no gate (the SDK session is safe for concurrent service calls); subscription management
/// takes <see cref="SubscriptionGate"/>.</summary>
internal SemaphoreSlim Gate => _gate;
/// <summary>Test seam: the subscription-management gate.</summary>
internal SemaphoreSlim SubscriptionGate => _subscriptionGate;
/// <summary>
/// Test-only seam to swap the active <see cref="Session"/> the way a re-initialise does at
/// runtime, so the stale-session race the read/history paths guard against (re-read Session
@@ -1420,80 +1431,76 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
IReadOnlyList<Core.Abstractions.WriteRequest> writes, CancellationToken cancellationToken)
{
// See ReadAsync — the wire call must use the session current inside the gate, not a
// reference captured before WaitAsync.
// R06:O-6 — writes take NO gate: a write stalled for the operation timeout must not hold up every read
// and subscription create behind it (the host already bounds write concurrency and serialises per tag;
// the SDK session is safe for concurrent service calls). The session is read once, here.
_ = RequireSession();
var results = new WriteResult[writes.Count];
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
var session = Session;
if (session is null)
{
// Writes are non-idempotent — but here the request never reached the wire, so
// BadCommunicationError ("definitely did not happen") is the honest code.
for (var i = 0; i < writes.Count; i++)
results[i] = new WriteResult(StatusBadCommunicationError);
return results;
}
var toSend = new WriteValueCollection();
var indexMap = new List<int>(writes.Count);
for (var i = 0; i < writes.Count; i++)
{
if (!TryResolveTagNode(session, writes[i].FullReference, out var nodeId))
{
results[i] = new WriteResult(StatusBadNodeIdInvalid);
continue;
}
toSend.Add(new WriteValue
{
NodeId = nodeId,
AttributeId = Attributes.Value,
Value = new DataValue(new Variant(writes[i].Value)),
});
indexMap.Add(i);
}
if (toSend.Count == 0) return results;
try
{
var session = Session;
if (session is null)
{
// Writes are non-idempotent — but here the request never reached the wire, so
// BadCommunicationError ("definitely did not happen") is the honest code.
for (var i = 0; i < writes.Count; i++)
results[i] = new WriteResult(StatusBadCommunicationError);
return results;
}
var resp = await session.WriteAsync(
requestHeader: null,
nodesToWrite: toSend,
ct: cancellationToken).ConfigureAwait(false);
var toSend = new WriteValueCollection();
var indexMap = new List<int>(writes.Count);
for (var i = 0; i < writes.Count; i++)
var codes = resp.Results;
for (var w = 0; w < codes.Count; w++)
{
if (!TryResolveTagNode(session, writes[i].FullReference, out var nodeId))
{
results[i] = new WriteResult(StatusBadNodeIdInvalid);
continue;
}
toSend.Add(new WriteValue
{
NodeId = nodeId,
AttributeId = Attributes.Value,
Value = new DataValue(new Variant(writes[i].Value)),
});
indexMap.Add(i);
}
if (toSend.Count == 0) return results;
try
{
var resp = await session.WriteAsync(
requestHeader: null,
nodesToWrite: toSend,
ct: cancellationToken).ConfigureAwait(false);
var codes = resp.Results;
for (var w = 0; w < codes.Count; w++)
{
var r = indexMap[w];
// Pass upstream WriteResult StatusCode through verbatim. Success codes
// include Good (0) and any warning-level Good* variants; anything with
// the severity bits set is a Bad.
results[r] = new WriteResult(codes[w].Code);
}
}
catch (OperationCanceledException)
{
// Timeout / cancellation after the wire request may have been dispatched.
// Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly
// retry") is more honest than BadCommunicationError ("definitely did not
// happen"). Downstream callers that need retry semantics check for BadTimeout
// and can decide whether to re-issue.
const uint StatusBadTimeout = 0x800A0000u;
for (var w = 0; w < indexMap.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadTimeout);
}
catch (Exception)
{
// Pre-wire transport failure — the write definitely did not reach the server.
for (var w = 0; w < indexMap.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadCommunicationError);
var r = indexMap[w];
// Pass upstream WriteResult StatusCode through verbatim. Success codes
// include Good (0) and any warning-level Good* variants; anything with
// the severity bits set is a Bad.
results[r] = new WriteResult(codes[w].Code);
}
}
finally { _gate.Release(); }
catch (OperationCanceledException)
{
// Timeout / cancellation after the wire request may have been dispatched.
// Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly
// retry") is more honest than BadCommunicationError ("definitely did not
// happen"). Downstream callers that need retry semantics check for BadTimeout
// and can decide whether to re-issue.
const uint StatusBadTimeout = 0x800A0000u;
for (var w = 0; w < indexMap.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadTimeout);
}
catch (Exception)
{
// Pre-wire transport failure — the write definitely did not reach the server.
for (var w = 0; w < indexMap.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadCommunicationError);
}
return results;
}
@@ -1550,46 +1557,38 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(builder);
// Confirm a session exists before queuing; the session actually browsed is re-read
// inside the gate so a reconnect mid-wait can't leave us browsing a closed session.
_ = RequireSession();
// R06:O-6 — no gate across the recursive browse: it can run for as long as the remote tree takes to walk,
// and holding the read gate for that long would stall the data plane behind it. The browse reads the session
// once and uses it throughout (the SDK session is safe for concurrent service calls).
var session = Session
?? throw new InvalidOperationException("OpcUaClientDriver not initialized");
var rootFolder = builder.Folder("Remote", "Remote");
var visited = new HashSet<NodeId>();
var discovered = 0;
var pendingVariables = new List<PendingVariable>();
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
var session = Session
?? throw new InvalidOperationException(
"OpcUaClient session was lost before discovery could browse the remote server.");
var root = !string.IsNullOrEmpty(_options.BrowseRoot)
? NodeId.Parse(session.MessageContext, _options.BrowseRoot)
: ObjectIds.ObjectsFolder;
var root = !string.IsNullOrEmpty(_options.BrowseRoot)
? NodeId.Parse(session.MessageContext, _options.BrowseRoot)
: ObjectIds.ObjectsFolder;
// Pass 1: browse hierarchy + create folders inline, collect variables into a
// pending list. Defers variable registration until attributes are resolved — the
// address-space builder's Variable call is the one-way commit, so doing it only
// once per variable (with correct DataType/SecurityClass/IsArray) avoids the
// alternative (register with placeholders + mutate later) which the
// IAddressSpaceBuilder contract doesn't expose.
await BrowseRecursiveAsync(session, root, rootFolder, visited,
depth: 0,
discovered: () => discovered, increment: () => discovered++,
pendingVariables: pendingVariables,
ct: cancellationToken).ConfigureAwait(false);
// Pass 1: browse hierarchy + create folders inline, collect variables into a
// pending list. Defers variable registration until attributes are resolved — the
// address-space builder's Variable call is the one-way commit, so doing it only
// once per variable (with correct DataType/SecurityClass/IsArray) avoids the
// alternative (register with placeholders + mutate later) which the
// IAddressSpaceBuilder contract doesn't expose.
await BrowseRecursiveAsync(session, root, rootFolder, visited,
depth: 0,
discovered: () => discovered, increment: () => discovered++,
pendingVariables: pendingVariables,
ct: cancellationToken).ConfigureAwait(false);
// Pass 2: batch-read DataType + AccessLevel + ValueRank + Historizing per
// variable. One wire request for up to ~N variables; for 10k-node servers this is
// still a couple of hundred ms total since the SDK chunks ReadAsync automatically.
await EnrichAndRegisterVariablesAsync(session, pendingVariables, cancellationToken)
.ConfigureAwait(false);
}
finally { _gate.Release(); }
// Pass 2: batch-read DataType + AccessLevel + ValueRank + Historizing per
// variable. One wire request for up to ~N variables; for 10k-node servers this is
// still a couple of hundred ms total since the SDK chunks ReadAsync automatically.
await EnrichAndRegisterVariablesAsync(session, pendingVariables, cancellationToken)
.ConfigureAwait(false);
}
/// <summary>
@@ -1904,7 +1903,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
TimestampsToReturn = TimestampsToReturn.Both,
});
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
// Re-read Session inside the critical section: a reconnect completing while we were
@@ -1939,7 +1938,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
_subscriptions[id] = new RemoteSubscription(subscription, handle, itemHandlers);
}
finally { _gate.Release(); }
finally { _subscriptionGate.Release(); }
return handle;
@@ -1982,7 +1981,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (handle is not OpcUaSubscriptionHandle h) return;
if (!_subscriptions.TryRemove(h.Id, out var rs)) return;
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
// Detach Notification handlers BEFORE deleting the subscription so the SDK's
@@ -1994,7 +1993,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort — the subscription may already be gone on reconnect */ }
}
finally { _gate.Release(); }
finally { _subscriptionGate.Release(); }
}
private static void DetachNotificationHandlers(IReadOnlyList<MonitoredItemNotificationHandle> items)
@@ -2384,7 +2383,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
MonitoredItem? eventItem = null;
MonitoredItemNotificationEventHandler? notifHandler = null;
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
// Re-read Session inside the critical section: a reconnect completing while we were
@@ -2422,7 +2421,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// current state through the normal path (deduplicated against the last-told state).
await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false);
}
finally { _gate.Release(); }
finally { _subscriptionGate.Release(); }
return handle;
@@ -2463,7 +2462,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (handle is not OpcUaAlarmSubscriptionHandle h) return;
if (!_alarmSubscriptions.TryRemove(h.Id, out var rs)) return;
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
// Detach the Notification handler before deleting the subscription so the SDK's
@@ -2474,7 +2473,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); }
catch { /* best-effort — session may already be gone across a reconnect */ }
}
finally { _gate.Release(); }
finally { _subscriptionGate.Release(); }
}
/// <inheritdoc />
@@ -2761,7 +2760,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (_alarmSubscriptions.IsEmpty) return;
try
{
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
await _subscriptionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
}
catch (ObjectDisposedException) { return; }
try
@@ -2781,7 +2780,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
finally
{
try { _gate.Release(); } catch (ObjectDisposedException) { }
try { _subscriptionGate.Release(); } catch (ObjectDisposedException) { }
}
}
@@ -3560,13 +3559,16 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// DiscoverAsync) has definitely released before we dispose the gate. Without this
// drain, a background read that calls _gate.Release() after Dispose throws
// ObjectDisposedException.
try
foreach (var gate in new[] { _gate, _subscriptionGate })
{
if (_gate.Wait(TimeSpan.FromSeconds(2)))
_gate.Release();
try
{
if (gate.Wait(TimeSpan.FromSeconds(2)))
gate.Release();
}
catch { /* timeout or already disposed — proceed */ }
gate.Dispose();
}
catch { /* timeout or already disposed — proceed */ }
_gate.Dispose();
try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { }
SdkTelemetry.Release(_sdkTelemetry);
}
@@ -3590,13 +3592,18 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// Drain the gate before disposal so no in-flight _gate.Release() fires after
// Dispose.
try
foreach (var gate in new[] { _gate, _subscriptionGate })
{
await _gate.WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
_gate.Release();
try
{
// Short: the host's PostStop gives shutdown + dispose 5 s together, and ShutdownAsync above
// already spent its own budget; an op still holding a gate finishes on its own token.
if (await gate.WaitAsync(TimeSpan.FromMilliseconds(500)).ConfigureAwait(false))
gate.Release();
}
catch { /* timeout or already disposed */ }
gate.Dispose();
}
catch { /* timeout or already disposed */ }
_gate.Dispose();
try { _sessionWorkCts.Cancel(); } catch (ObjectDisposedException) { }
SdkTelemetry.Release(_sdkTelemetry);
}