refactor(opcuaserver): async-ify ServeNode + Processed/AtTime HistoryRead internals, thread cancellation (03/S3)
This commit is contained in:
@@ -9,7 +9,7 @@
|
|||||||
{ "id": "R2-08-T6", "subject": "Outage-sim integration test (real library, unroutable host, via internal HandleImpersonation) + SDK m_eventLock contract doc + docs/security.md", "status": "completed", "blockedBy": ["R2-08-T4"] },
|
{ "id": "R2-08-T6", "subject": "Outage-sim integration test (real library, unroutable host, via internal HandleImpersonation) + SDK m_eventLock contract doc + docs/security.md", "status": "completed", "blockedBy": ["R2-08-T4"] },
|
||||||
{ "id": "R2-08-T7", "subject": "S3 sequential-serving repro test (RED): MaxObservedConcurrency >= 2 on an 8-node Raw batch — fails on current code", "status": "completed", "blockedBy": [] },
|
{ "id": "R2-08-T7", "subject": "S3 sequential-serving repro test (RED): MaxObservedConcurrency >= 2 on an 8-node Raw batch — fails on current code", "status": "completed", "blockedBy": [] },
|
||||||
{ "id": "R2-08-T8", "subject": "ServerHistorianOptions: HistoryReadBatchParallelism/MaxConcurrentHistoryReadBatches/HistoryReadDeadline + Validate warnings + node-manager properties + hosted-service wiring", "status": "completed", "blockedBy": [] },
|
{ "id": "R2-08-T8", "subject": "ServerHistorianOptions: HistoryReadBatchParallelism/MaxConcurrentHistoryReadBatches/HistoryReadDeadline + Validate warnings + node-manager properties + hosted-service wiring", "status": "completed", "blockedBy": [] },
|
||||||
{ "id": "R2-08-T9", "subject": "Async-ify ServeNode + Processed/AtTime arms (ct threading, OCE=>BadTimeout, behavior-neutral)", "status": "pending", "blockedBy": ["R2-08-T7", "R2-08-T8"] },
|
{ "id": "R2-08-T9", "subject": "Async-ify ServeNode + Processed/AtTime arms (ct threading, OCE=>BadTimeout, behavior-neutral)", "status": "completed", "blockedBy": ["R2-08-T7", "R2-08-T8"] },
|
||||||
{ "id": "R2-08-T10", "subject": "Async-ify ServeRawPaged (page read + tie-cluster over-fetch) + extract ServeEventsAsync", "status": "pending", "blockedBy": ["R2-08-T9"] },
|
{ "id": "R2-08-T10", "subject": "Async-ify ServeRawPaged (page read + tie-cluster over-fetch) + extract ServeEventsAsync", "status": "pending", "blockedBy": ["R2-08-T9"] },
|
||||||
{ "id": "R2-08-T11", "subject": "RunBounded fan-out in all four HistoryRead arms, single bridge per arm (turns T7 green) + upper-bound + isolation tests", "status": "pending", "blockedBy": ["R2-08-T9", "R2-08-T10"] },
|
{ "id": "R2-08-T11", "subject": "RunBounded fan-out in all four HistoryRead arms, single bridge per arm (turns T7 green) + upper-bound + isolation tests", "status": "pending", "blockedBy": ["R2-08-T9", "R2-08-T10"] },
|
||||||
{ "id": "R2-08-T12", "subject": "Per-request HistoryRead deadline CTS — hung backend surfaces BadTimeout per node", "status": "pending", "blockedBy": ["R2-08-T11"] },
|
{ "id": "R2-08-T12", "subject": "Per-request HistoryRead deadline CTS — hung backend surfaces BadTimeout per node", "status": "pending", "blockedBy": ["R2-08-T11"] },
|
||||||
|
|||||||
@@ -1861,13 +1861,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
|||||||
// and the single-shot backend returns every bucket in one read, so there is no "full page ⇒
|
// and the single-shot backend returns every bucket in one read, so there is no "full page ⇒
|
||||||
// maybe more" signal to page on. Returning the complete aggregate result with a null CP is
|
// maybe more" signal to page on. Returning the complete aggregate result with a null CP is
|
||||||
// spec-conformant (OPC UA Part 11 lets a server return all available data in one response).
|
// spec-conformant (OPC UA Part 11 lets a server return all available data in one response).
|
||||||
ServeNode(handle, results, errors, (source, tagname) => source.ReadProcessedAsync(
|
// Interim: still bridged sequentially per node (bounded fan-out lands in a later slice).
|
||||||
|
ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadProcessedAsync(
|
||||||
tagname,
|
tagname,
|
||||||
details.StartTime,
|
details.StartTime,
|
||||||
details.EndTime,
|
details.EndTime,
|
||||||
interval,
|
interval,
|
||||||
aggregate.Value,
|
aggregate.Value,
|
||||||
CancellationToken.None));
|
token), CancellationToken.None).GetAwaiter().GetResult();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1886,10 +1887,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
|||||||
var timestamps = details.ReqTimes?.ToList() ?? new List<DateTime>();
|
var timestamps = details.ReqTimes?.ToList() ?? new List<DateTime>();
|
||||||
foreach (var handle in nodesToProcess)
|
foreach (var handle in nodesToProcess)
|
||||||
{
|
{
|
||||||
ServeNode(handle, results, errors, (source, tagname) => source.ReadAtTimeAsync(
|
// Interim: still bridged sequentially per node (bounded fan-out lands in a later slice).
|
||||||
|
ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync(
|
||||||
tagname,
|
tagname,
|
||||||
timestamps,
|
timestamps,
|
||||||
CancellationToken.None));
|
token), CancellationToken.None).GetAwaiter().GetResult();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2061,11 +2063,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
|||||||
/// <param name="errors">The service-level errors list to fill at <c>handle.Index</c>.</param>
|
/// <param name="errors">The service-level errors list to fill at <c>handle.Index</c>.</param>
|
||||||
/// <param name="read">Invokes the resolved data-source read with the resolved tagname; only called
|
/// <param name="read">Invokes the resolved data-source read with the resolved tagname; only called
|
||||||
/// once the tagname is confirmed present.</param>
|
/// once the tagname is confirmed present.</param>
|
||||||
private void ServeNode(
|
private async Task ServeNodeAsync(
|
||||||
NodeHandle handle,
|
NodeHandle handle,
|
||||||
IList<SdkHistoryReadResult> results,
|
IList<SdkHistoryReadResult> results,
|
||||||
IList<ServiceResult> errors,
|
IList<ServiceResult> errors,
|
||||||
Func<IHistorianDataSource, string, Task<HistorianRead>> read)
|
Func<IHistorianDataSource, string, CancellationToken, Task<HistorianRead>> read,
|
||||||
|
CancellationToken ct)
|
||||||
{
|
{
|
||||||
var idString = handle.NodeId.Identifier?.ToString();
|
var idString = handle.NodeId.Identifier?.ToString();
|
||||||
if (idString is null || !TryGetHistorizedTagname(idString, out var tagname))
|
if (idString is null || !TryGetHistorizedTagname(idString, out var tagname))
|
||||||
@@ -2078,9 +2081,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
|||||||
|
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
// HistoryRead is NOT invoked under the node-manager Lock (unlike OnWriteValue), so blocking
|
// HistoryRead is NOT invoked under the node-manager Lock (unlike OnWriteValue), so awaiting
|
||||||
// on the async source here is safe and won't freeze the address space.
|
// the async source here is safe and won't freeze the address space.
|
||||||
var sourceResult = read(HistorianDataSource, tagname!).GetAwaiter().GetResult();
|
var sourceResult = await read(HistorianDataSource, tagname!, ct).ConfigureAwait(false);
|
||||||
var historyData = ToHistoryData(sourceResult);
|
var historyData = ToHistoryData(sourceResult);
|
||||||
|
|
||||||
results[handle.Index] = new SdkHistoryReadResult
|
results[handle.Index] = new SdkHistoryReadResult
|
||||||
@@ -2095,13 +2098,20 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
|||||||
};
|
};
|
||||||
errors[handle.Index] = ServiceResult.Good;
|
errors[handle.Index] = ServiceResult.Good;
|
||||||
}
|
}
|
||||||
|
catch (OperationCanceledException)
|
||||||
|
{
|
||||||
|
// The per-request deadline elapsed while this node was in flight — surface BadTimeout for
|
||||||
|
// THIS node only (a spec-conformant history-read status), ahead of the generic catch.
|
||||||
|
errors[handle.Index] = StatusCodes.BadTimeout;
|
||||||
|
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout };
|
||||||
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
// One node's backend failure (throw / timeout / cancellation) must not throw out of the
|
// One node's backend failure (throw / timeout) must not throw out of the batch — surface a
|
||||||
// batch — surface a Bad status for THIS node only. This CustomNodeManager2 carries no
|
// Bad status for THIS node only. This CustomNodeManager2 carries no ILogger (see
|
||||||
// ILogger (see ReportConditionEvent), so log through the SDK's static trace rather than
|
// ReportConditionEvent), so log through the SDK's static trace rather than swallowing silently.
|
||||||
// swallowing silently. Utils.LogError is [Obsolete] in 1.5.378 (favours an ITelemetryContext
|
// Utils.LogError is [Obsolete] in 1.5.378 (favours an ITelemetryContext this manager doesn't
|
||||||
// this manager doesn't wire) — suppress the deprecation, matching the existing pattern.
|
// wire) — suppress the deprecation, matching the existing pattern.
|
||||||
#pragma warning disable CS0618 // Type or member is obsolete
|
#pragma warning disable CS0618 // Type or member is obsolete
|
||||||
Utils.LogError(ex, "OtOpcUaNodeManager: HistoryRead failed for node {0}", handle.NodeId);
|
Utils.LogError(ex, "OtOpcUaNodeManager: HistoryRead failed for node {0}", handle.NodeId);
|
||||||
#pragma warning restore CS0618
|
#pragma warning restore CS0618
|
||||||
|
|||||||
Reference in New Issue
Block a user