perf(s7): fail the rest of a pass fast once the handle is dead (R07:P1)
No read or write loop checked _plcDead between items: on a frozen-but- established peer the first op blew its deadline and marked the handle dead, and every remaining span / tag / write still went to the wire and could wait its own deadline under _gate (a 20-span pass ~100 s), with Degraded and the comms-loss watchdog's clock deferred until the pass ended. - The span loop, the multi-var chunk loop, the per-tag loop and the write loop stamp BadCommunicationError without I/O once _plcDead is set. - A multi-var chunk that dies on a transport loss (socket / IO / timeout / ConnectionError) marks the handle dead instead of falling back to one doomed per-tag read per member. The framing class is deliberately excluded there: snap7/S7.Net report a multi-var with one refused item as ReadData(WrongNumberReceivedBytes) over a healthy stream. - Live gate via an in-test freezable TCP proxy (counts request frames, not seconds). Mutation shows S7.Net 0.20 already self-closes on a cancelled op, so the gate pins the end-to-end property; the driver guard itself is proven by the unit tests (an op that ignores its token keeps the socket open). Refs #742 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,104 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// An in-test TCP forwarder in front of the S7 fixture that can be FROZEN: while frozen it keeps
|
||||
/// every socket open but forwards nothing in either direction — the frozen-but-established peer
|
||||
/// (cable pull, firewall DROP) that only a wire-op deadline can detect, produced without touching
|
||||
/// the shared fixture. It counts the ISO-on-TCP (TPKT) request frames the client sends while frozen,
|
||||
/// so a test can assert how many wire operations went out without timing anything.
|
||||
/// </summary>
|
||||
internal sealed class FreezableTcpProxy : IAsyncDisposable
|
||||
{
|
||||
private readonly string _targetHost;
|
||||
private readonly int _targetPort;
|
||||
private readonly TcpListener _listener = new(IPAddress.Loopback, 0);
|
||||
private readonly CancellationTokenSource _cts = new();
|
||||
private readonly List<TcpClient> _sockets = [];
|
||||
private volatile bool _frozen;
|
||||
private int _framesWhileFrozen;
|
||||
|
||||
/// <summary>Starts a proxy forwarding to <paramref name="targetHost"/>:<paramref name="targetPort"/>.</summary>
|
||||
/// <param name="targetHost">The fixture host.</param>
|
||||
/// <param name="targetPort">The fixture port.</param>
|
||||
public FreezableTcpProxy(string targetHost, int targetPort)
|
||||
{
|
||||
_targetHost = targetHost;
|
||||
_targetPort = targetPort;
|
||||
_listener.Start();
|
||||
_ = AcceptLoopAsync();
|
||||
}
|
||||
|
||||
/// <summary>The local port clients connect to.</summary>
|
||||
public int Port => ((IPEndPoint)_listener.LocalEndpoint).Port;
|
||||
|
||||
/// <summary>TPKT request frames the client sent while the proxy was frozen.</summary>
|
||||
public int FramesWhileFrozen => Volatile.Read(ref _framesWhileFrozen);
|
||||
|
||||
/// <summary>Stops forwarding (sockets stay open).</summary>
|
||||
public void Freeze() => _frozen = true;
|
||||
|
||||
/// <summary>Resumes forwarding for connections made from now on.</summary>
|
||||
public void Unfreeze() => _frozen = false;
|
||||
|
||||
private async Task AcceptLoopAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
while (!_cts.IsCancellationRequested)
|
||||
{
|
||||
var client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false);
|
||||
var upstream = new TcpClient();
|
||||
await upstream.ConnectAsync(_targetHost, _targetPort, _cts.Token).ConfigureAwait(false);
|
||||
lock (_sockets) { _sockets.Add(client); _sockets.Add(upstream); }
|
||||
// A connection accepted while frozen stays frozen for its whole life (its peer is "gone").
|
||||
var frozenAtBirth = _frozen;
|
||||
_ = PumpAsync(client.GetStream(), upstream.GetStream(), countFrames: true, frozenAtBirth);
|
||||
_ = PumpAsync(upstream.GetStream(), client.GetStream(), countFrames: false, frozenAtBirth);
|
||||
}
|
||||
}
|
||||
catch { /* disposed */ }
|
||||
}
|
||||
|
||||
private async Task PumpAsync(NetworkStream from, NetworkStream to, bool countFrames, bool frozenAtBirth)
|
||||
{
|
||||
var buffer = new byte[8192];
|
||||
var pending = new List<byte>();
|
||||
try
|
||||
{
|
||||
while (!_cts.IsCancellationRequested)
|
||||
{
|
||||
var n = await from.ReadAsync(buffer, _cts.Token).ConfigureAwait(false);
|
||||
if (n == 0) return;
|
||||
var frozen = _frozen || frozenAtBirth;
|
||||
if (countFrames && frozen)
|
||||
{
|
||||
// Count complete TPKT frames: 0x03 0x00 then a big-endian total length.
|
||||
pending.AddRange(buffer.AsSpan(0, n).ToArray());
|
||||
while (pending.Count >= 4)
|
||||
{
|
||||
var len = (pending[2] << 8) | pending[3];
|
||||
if (len < 4 || pending.Count < len) break;
|
||||
pending.RemoveRange(0, len);
|
||||
Interlocked.Increment(ref _framesWhileFrozen);
|
||||
}
|
||||
}
|
||||
if (frozen) continue; // swallow — the peer never answers
|
||||
await to.WriteAsync(buffer.AsMemory(0, n), _cts.Token).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch { /* socket closed / disposed */ }
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask DisposeAsync()
|
||||
{
|
||||
_cts.Cancel();
|
||||
_listener.Stop();
|
||||
lock (_sockets) foreach (var s in _sockets) s.Dispose();
|
||||
_cts.Dispose();
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
+71
@@ -0,0 +1,71 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests.S7_1500;
|
||||
|
||||
/// <summary>
|
||||
/// Live gate for R07:P1 over real S7comm: a frozen-but-established peer (produced by an in-test
|
||||
/// <see cref="FreezableTcpProxy"/>, so the shared fixture is never touched). The first wire op of
|
||||
/// the pass blows its deadline and marks the connection dead; the rest of the pass must NOT go to
|
||||
/// the wire. The assertion counts request frames, not seconds.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <b>What this gate does and does not prove.</b> Verified by mutation (2026-10-22): on the real
|
||||
/// stack this test also passes with the driver's dead-handle guard removed, because S7.Net 0.20
|
||||
/// honours the deadline's cancellation token and closes its own socket, so the next op fails at once
|
||||
/// with <c>ConnectionError</c> and sends nothing. The driver guard (proven by
|
||||
/// <c>S7DeadHandleShortCircuitTests</c> against a fake that does not self-close) is what covers an
|
||||
/// op that ignores its token — the <c>WaitAsync</c> leg of <c>S7OperationDeadline</c>, where the
|
||||
/// abandoned socket stays open and every further op would pay the full deadline. This test pins the
|
||||
/// end-to-end property (one request per frozen pass, then recovery) against S7.Net upgrades.
|
||||
/// </remarks>
|
||||
[Collection(Snap7ServerCollection.Name)]
|
||||
[Trait("Category", "Integration")]
|
||||
[Trait("Device", "S7_1500")]
|
||||
public sealed class S7_1500FrozenPeerTests(Snap7ServerFixture sim)
|
||||
{
|
||||
/// <summary>A pass against a frozen peer issues ONE request, then fails the rest fast; it recovers after.</summary>
|
||||
[Fact]
|
||||
public async Task A_frozen_peer_costs_one_wire_op_per_pass_not_one_per_tag()
|
||||
{
|
||||
if (sim.SkipReason is not null) Assert.Skip(sim.SkipReason);
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
|
||||
await using var proxy = new FreezableTcpProxy(sim.Host, sim.Port);
|
||||
var baseOptions = S7_1500Profile.BuildOptions("127.0.0.1", proxy.Port);
|
||||
var options = new S7DriverOptions
|
||||
{
|
||||
Host = baseOptions.Host, Port = baseOptions.Port, CpuType = baseOptions.CpuType,
|
||||
Rack = baseOptions.Rack, Slot = baseOptions.Slot,
|
||||
Timeout = TimeSpan.FromSeconds(1), // the per-op deadline that detects the frozen peer
|
||||
Probe = baseOptions.Probe,
|
||||
SpanReadMaxGapBytes = -1, // one wire request per tag, so "one per tag" is countable
|
||||
RawTags = baseOptions.RawTags,
|
||||
};
|
||||
string[] refs =
|
||||
[
|
||||
S7_1500Profile.ProbeTag, S7_1500Profile.SmokeI16Tag, S7_1500Profile.SmokeI32Tag,
|
||||
S7_1500Profile.SmokeF32Tag, S7_1500Profile.SmokeBoolTag, S7_1500Profile.WriteScratchTag,
|
||||
];
|
||||
|
||||
await using var drv = new S7Driver(options, driverInstanceId: "s7-frozen-peer-live");
|
||||
await drv.InitializeAsync("{}", ct);
|
||||
|
||||
// Positive control: through the proxy, every tag reads Good.
|
||||
(await drv.ReadAsync(refs, ct)).ShouldAllBe(s => s.StatusCode == 0u);
|
||||
|
||||
proxy.Freeze();
|
||||
var frozen = await drv.ReadAsync(refs, ct);
|
||||
frozen.ShouldAllBe(s => s.StatusCode != 0u);
|
||||
proxy.FramesWhileFrozen.ShouldBe(1, "only the request that discovered the dead connection may go out");
|
||||
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
|
||||
|
||||
// Recovery: the next pass reopens (a fresh proxy connection) and reads Good again.
|
||||
proxy.Unfreeze();
|
||||
var recovered = await drv.ReadAsync(refs, ct);
|
||||
recovered.ShouldAllBe(s => s.StatusCode == 0u);
|
||||
Convert.ToInt32(recovered[0].Value).ShouldBe(S7_1500Profile.ProbeSeedValue);
|
||||
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
|
||||
}
|
||||
}
|
||||
@@ -243,7 +243,9 @@ public sealed class S7ArrayWriteAndHealthTests
|
||||
var (drv, factory) = await TwoTagDriver();
|
||||
using var _ = drv;
|
||||
factory.Created[0].WriteFaults["DB1.DBW0"] = new PlcException(ErrorCode.WriteData, "Accessing object not allowed");
|
||||
factory.Created[0].WriteFaults["DB1.DBW2"] = new IOException("socket glitch");
|
||||
// A Degraded failure that is NOT connection-fatal: since R07:P1 a fatal one (an IOException)
|
||||
// would fail the rest of the batch without I/O, and C must reach the wire to be the success.
|
||||
factory.Created[0].WriteFaults["DB1.DBW2"] = new PlcException(ErrorCode.SendData, "device glitch");
|
||||
|
||||
var res = await drv.WriteAsync(
|
||||
[new WriteRequest("A", (ushort)1), new WriteRequest("B", (ushort)2), new WriteRequest("C", (ushort)3)],
|
||||
|
||||
@@ -0,0 +1,167 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.S7.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// R07:P1 — once a pass has marked the connection dead (a wire op blew its deadline on a frozen
|
||||
/// peer), the rest of that pass must not go to the wire: each further op would pay the full
|
||||
/// timeout under the gate, so a 20-span pass took ~100 s and the comms-loss watchdog's clock only
|
||||
/// started after it. The remaining items are stamped <c>BadCommunicationError</c> without I/O.
|
||||
/// Every assertion counts wire operations; nothing is timed.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class S7DeadHandleShortCircuitTests
|
||||
{
|
||||
private const uint BadCommunicationError = 0x80050000u;
|
||||
|
||||
private static S7DriverOptions Options(int maxGap, params S7TagDefinition[] tags) => new()
|
||||
{
|
||||
Host = "192.0.2.1",
|
||||
Timeout = TimeSpan.FromSeconds(5),
|
||||
Probe = new S7ProbeOptions { Enabled = false },
|
||||
SpanReadMaxGapBytes = maxGap,
|
||||
RawTags = S7RawTags.Entries(tags),
|
||||
};
|
||||
|
||||
/// <summary>Five isolated words — far enough apart that each is its own span.</summary>
|
||||
private static S7TagDefinition[] FarApartWords() =>
|
||||
[.. Enumerable.Range(0, 5).Select(i => new S7TagDefinition($"W{i}", $"DB{i + 1}.DBW0", S7DataType.UInt16, Writable: true))];
|
||||
|
||||
private static string[] Refs(int n) => [.. Enumerable.Range(0, n).Select(i => $"W{i}")];
|
||||
|
||||
private static async Task<(S7Driver, ScriptedS7PlcFactory)> StartAsync(S7DriverOptions options)
|
||||
{
|
||||
var factory = new ScriptedS7PlcFactory();
|
||||
var drv = new S7Driver(options, "s7-dead-short-circuit", factory);
|
||||
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
|
||||
return (drv, factory);
|
||||
}
|
||||
|
||||
/// <summary>Per-tag path: the first timeout ends the pass's wire traffic.</summary>
|
||||
[Fact]
|
||||
public async Task Per_tag_pass_stops_issuing_wire_ops_once_the_handle_is_dead()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var (drv, factory) = await StartAsync(Options(maxGap: -1, FarApartWords()));
|
||||
using var _ = drv;
|
||||
factory.Read = (_, _) => throw new TimeoutException("frozen peer");
|
||||
|
||||
var before = factory.WireOps;
|
||||
var r = await drv.ReadAsync(Refs(5), ct);
|
||||
|
||||
(factory.WireOps - before).ShouldBe(1, "only the op that discovered the dead handle may go to the wire");
|
||||
r.ShouldAllBe(s => s.StatusCode == BadCommunicationError);
|
||||
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
|
||||
|
||||
// Positive control: the next pass reopens and serves every tag.
|
||||
factory.Read = (_, _) => Task.FromResult<object?>((ushort)1);
|
||||
var recovered = await drv.ReadAsync(Refs(5), ct);
|
||||
recovered.ShouldAllBe(s => s.StatusCode == 0u);
|
||||
factory.Created.Count.ShouldBe(2);
|
||||
}
|
||||
|
||||
/// <summary>Span path: arrays are block reads even alone, so five arrays are five spans.</summary>
|
||||
[Fact]
|
||||
public async Task Span_pass_stops_issuing_wire_ops_once_the_handle_is_dead()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
S7TagDefinition[] arrays =
|
||||
[.. Enumerable.Range(0, 5).Select(i => new S7TagDefinition($"W{i}", $"DB{i + 1}.DBW0", S7DataType.UInt16, Writable: false, ArrayCount: 2))];
|
||||
var (drv, factory) = await StartAsync(Options(maxGap: 16, arrays));
|
||||
using var _ = drv;
|
||||
factory.ReadBytes = (_, _, _, _, _) => throw new TimeoutException("frozen peer");
|
||||
|
||||
var before = factory.WireOps;
|
||||
var r = await drv.ReadAsync(Refs(5), ct);
|
||||
|
||||
(factory.WireOps - before).ShouldBe(1);
|
||||
r.ShouldAllBe(s => s.StatusCode == BadCommunicationError);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Multi-var path: a chunk that dies on a transport loss used to fall back to one per-tag read
|
||||
/// per member — each paying the timeout again. Now it marks the handle dead and the fallback is
|
||||
/// short-circuited.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Multi_var_transport_loss_does_not_fall_back_to_per_tag_wire_reads()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var (drv, factory) = await StartAsync(Options(maxGap: 0, FarApartWords()));
|
||||
using var _ = drv;
|
||||
factory.ReadMulti = (_, _) => throw new TimeoutException("frozen peer");
|
||||
factory.Read = (_, _) => throw new TimeoutException("frozen peer");
|
||||
|
||||
var before = factory.WireOps;
|
||||
var r = await drv.ReadAsync(Refs(5), ct);
|
||||
|
||||
(factory.WireOps - before).ShouldBe(1);
|
||||
r.ShouldAllBe(s => s.StatusCode == BadCommunicationError);
|
||||
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Positive control for the classification: a multi-var REFUSAL (not a transport loss) still
|
||||
/// falls back per tag, so the members the CPU will serve are served.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Multi_var_refusal_still_falls_back_per_tag()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var (drv, factory) = await StartAsync(Options(maxGap: 0, FarApartWords()));
|
||||
using var _ = drv;
|
||||
// The live snap7 shape of a multi-var with one refused item.
|
||||
factory.ReadMulti = (_, _) => throw new global::S7.Net.PlcException(global::S7.Net.ErrorCode.ReadData,
|
||||
new global::S7.Net.PlcException(global::S7.Net.ErrorCode.WrongNumberReceivedBytes));
|
||||
factory.Read = (_, _) => Task.FromResult<object?>((ushort)9);
|
||||
|
||||
var before = factory.WireOps;
|
||||
var r = await drv.ReadAsync(Refs(5), ct);
|
||||
|
||||
(factory.WireOps - before).ShouldBe(6, "one multi-var request, then one per-tag read each");
|
||||
r.ShouldAllBe(s => s.StatusCode == 0u);
|
||||
factory.Created.Count.ShouldBe(1, "a refusal is not a reason to reopen");
|
||||
}
|
||||
|
||||
/// <summary>A refusal does not short-circuit the pass: the CPU answered, the handle is fine.</summary>
|
||||
[Fact]
|
||||
public async Task A_refusal_does_not_short_circuit_the_pass()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var (drv, factory) = await StartAsync(Options(maxGap: -1, FarApartWords()));
|
||||
using var _ = drv;
|
||||
factory.Read = (addr, _) => addr == "DB1.DBW0"
|
||||
? throw new Exception("Received error from PLC: Object does not exist.")
|
||||
: Task.FromResult<object?>((ushort)3);
|
||||
|
||||
var before = factory.WireOps;
|
||||
var r = await drv.ReadAsync(Refs(5), ct);
|
||||
|
||||
(factory.WireOps - before).ShouldBe(5);
|
||||
r.Skip(1).ShouldAllBe(s => s.StatusCode == 0u);
|
||||
}
|
||||
|
||||
/// <summary>Write batch: the first transport loss fails the rest of the batch without I/O.</summary>
|
||||
[Fact]
|
||||
public async Task Write_batch_stops_issuing_wire_ops_once_the_handle_is_dead()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var (drv, factory) = await StartAsync(Options(maxGap: -1, FarApartWords()));
|
||||
using var _ = drv;
|
||||
factory.Write = (_, _, _) => throw new System.IO.IOException("connection reset");
|
||||
|
||||
var before = factory.WireOps;
|
||||
var w = await drv.WriteAsync([.. Refs(5).Select(r => new WriteRequest(r, (ushort)1))], ct);
|
||||
|
||||
(factory.WireOps - before).ShouldBe(1);
|
||||
w.ShouldAllBe(x => x.StatusCode == BadCommunicationError);
|
||||
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
|
||||
|
||||
// Positive control: the next batch reopens and writes.
|
||||
factory.Write = (_, _, _) => Task.CompletedTask;
|
||||
(await drv.WriteAsync([.. Refs(5).Select(r => new WriteRequest(r, (ushort)1))], ct))
|
||||
.ShouldAllBe(x => x.StatusCode == 0u);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user