From 4841158f94c0a894074c5272acadb4c844cd79ed Mon Sep 17 00:00:00 2001 From: jogibear9988 Date: Wed, 14 Oct 2026 12:44:49 +0200 Subject: [PATCH 1/3] fix: avoid redundant PLC round trips in bulk reads Large symbolic read batches were split twice: first at the PLC's negotiated item limit, then again to fit each logical request into a single transport frame. Transport fragmentation already supports larger requests, so the second split added avoidable request/response exchanges and cycle latency. A captured 10,000-signal cycle previously required 200 responses; it now requires 100 while preserving the PLC's 100-item request limit. Build read batches against the PLC item limit and the protocol's 16-bit PDU data-length bound. Reserve space for the legacy digest and growth of the integrity identifier assigned during send. Size ordinary batches with one serialization and retain the bounded payload search for unusually long addresses. Send GetMultiVariables requests as one logical PDU and leave fragmentation to the transport. Remove the TLS callback's additional fixed 960-byte splitting. Fragment records once at the negotiated COTP boundary, preserve the supplied data length, and set EOT only on the final fragment. Write batch sizing and the explicit write-enable gate remain unchanged. Add regression coverage for full and final partial read batches across transport frames, protocol length limits with long addresses, and payload and EOT preservation for both plain sends and TLS callback data. Validation: - 301 tests passed on .NET 8; 296 passed on .NET Framework 4.8. - Release builds passed for net48, net6.0, net8.0 and net9.0 with no warnings. - Read-only live validation covered 21,464 signals across 59 data blocks. Median cycle time decreased from 499 ms to 331 ms across 20 measured cycles. No value mismatches were found among 21,421 unchanged signals; 43 signals changed between comparison reads. No PLC writes were made. --- .../S7ClientTransportTests.cs | 11 +++- .../S7CommPlusProtocolSessionReceiveTests.cs | 33 +++++++++++ src/S7CommPlusDriver/Net/S7Client.cs | 18 +++--- .../S7CommPlusProtocolSession.cs | 58 ++++++++++++++----- 4 files changed, 93 insertions(+), 27 deletions(-) diff --git a/src/S7CommPlusDriver.Tests/S7ClientTransportTests.cs b/src/S7CommPlusDriver.Tests/S7ClientTransportTests.cs index 1e71a42..e54912c 100644 --- a/src/S7CommPlusDriver.Tests/S7ClientTransportTests.cs +++ b/src/S7CommPlusDriver.Tests/S7ClientTransportTests.cs @@ -104,8 +104,10 @@ public void DisconnectClosesInjectedTransportOnlyOnce() Assert.False(transport.Connected); } - [Fact] - public void SendFragmentsPayloadAtNegotiatedCotpBoundary() + [Theory] + [InlineData(false)] + [InlineData(true)] + public void SendFragmentsPayloadAtNegotiatedCotpBoundary(bool tlsRecord) { var transport = new FakeS7Transport { EmptyReceiveDelayMilliseconds = 500 }; transport.EnqueueReceive(new byte[] { 0x03, 0x00, 0x00, 0x23 }); @@ -120,7 +122,10 @@ public void SendFragmentsPayloadAtNegotiatedCotpBoundary() transport.Sent.Clear(); var payload = Enumerable.Range(0, 2000).Select(value => (byte)value).ToArray(); - client.Send(payload); + if (tlsRecord) + client.WriteData(payload.Concat(new byte[13]).ToArray(), payload.Length); + else + client.Send(payload); client.Disconnect(50); Assert.Equal(2, transport.Sent.Count); diff --git a/src/S7CommPlusDriver.Tests/S7CommPlusProtocolSessionReceiveTests.cs b/src/S7CommPlusDriver.Tests/S7CommPlusProtocolSessionReceiveTests.cs index ef2aa7b..b4d3e40 100644 --- a/src/S7CommPlusDriver.Tests/S7CommPlusProtocolSessionReceiveTests.cs +++ b/src/S7CommPlusDriver.Tests/S7CommPlusProtocolSessionReceiveTests.cs @@ -2,6 +2,7 @@ using System.Linq; using System.IO; using System.Security.Cryptography; +using S7CommPlusDriver.Internal; using Xunit; namespace S7CommPlusDriver.Tests @@ -207,6 +208,38 @@ public void SmallGetMultiVariablesRequestFitsLegacySingleFrameLimit() Assert.False(exceedsLimit); } + [Fact] + public void ReadBatchUsesPlcItemLimitEvenWhenItNeedsMultipleTransportFrames() + { + var connection = new S7CommPlusProtocolSession(); + var addresses = Enumerable.Range(1, 107) + .Select(i => new ItemAddress($"8A0E{i:X4}.1.2.3.4.5.6.7.8")) + .ToArray(); + + var first = connection.DebugCreateReadRequestBatchForTests(addresses, 0, 100); + var last = connection.DebugCreateReadRequestBatchForTests(addresses, first.ItemCount, 100); + + Assert.Equal(100, first.ItemCount); + Assert.True(first.SerializedLength > S7CommPlusProtocolConstants.DefaultIsoTpduSize); + Assert.Equal(7, last.ItemCount); + } + + [Fact] + public void ReadBatchKeepsTheProtocolLengthBoundForUnusuallyLongAddresses() + { + var connection = new S7CommPlusProtocolSession(); + var addresses = Enumerable.Range(1, 20) + .Select(i => new ItemAddress($"8A0E{i:X4}.1") + { + LID = Enumerable.Repeat(uint.MaxValue, 2000).ToList() + }).ToArray(); + + var batch = connection.DebugCreateReadRequestBatchForTests(addresses, 0, 100); + + Assert.InRange(batch.ItemCount, 1, addresses.Length - 1); + Assert.InRange(batch.SerializedLength, 1025, ushort.MaxValue - 33 - sizeof(uint)); + } + [Fact] public void WriteBatchStopsBeforeSerializedPayloadExceedsLimit() { diff --git a/src/S7CommPlusDriver/Net/S7Client.cs b/src/S7CommPlusDriver/Net/S7Client.cs index 8868aba..b545d25 100644 --- a/src/S7CommPlusDriver/Net/S7Client.cs +++ b/src/S7CommPlusDriver/Net/S7Client.cs @@ -91,20 +91,20 @@ public void WriteData(byte[] pData, int dataLength) { // SSL fordert Daten zum Absenden an // System.Diagnostics.Trace.WriteLine("S7Client - OpenSSL WriteData: dataLength=" + dataLength); - int offset = 0; - int maxPayloadSize = Math.Min(MaxPduSizeToRequest, PDU.Length - IsoHSize); - if (maxPayloadSize <= 0) + if (pData == null || dataLength < 0 || dataLength > pData.Length) { _LastError = S7Consts.errIsoInvalidPDU; return; } - while (offset < dataLength && _LastError == 0) + // Fragment the TLS record once at the negotiated COTP boundary. Splitting + // it into fixed 960-byte messages first adds packets and premature EOT flags. + if (dataLength == pData.Length) + SendIsoPacket(pData); + else { - int chunkSize = Math.Min(maxPayloadSize, dataLength - offset); - byte[] sendData = new byte[chunkSize]; - Array.Copy(pData, offset, sendData, 0, chunkSize); - SendIsoPacket(sendData); - offset += chunkSize; + var data = new byte[dataLength]; + Array.Copy(pData, data, dataLength); + SendIsoPacket(data); } } diff --git a/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs b/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs index 62fd277..699f40f 100644 --- a/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs +++ b/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs @@ -212,7 +212,8 @@ private int SendS7plusFunctionObject(IS7pRequest funcObj) MemoryStream stream = new MemoryStream(); funcObj.Serialize(stream); - return SendS7plusPDUdata(stream.ToArray(), (int)stream.Length, funcObj.ProtocolVersion); + return SendS7plusPDUdata(stream.ToArray(), (int)stream.Length, funcObj.ProtocolVersion, + allowTransportFragmentation: funcObj is GetMultiVariablesRequest); } private int SendS7plusFunctionObjectSerialized(IS7pRequest funcObj) @@ -530,7 +531,7 @@ private void ClearNotificationQueues() } } - private int SendS7plusPDUdata(byte[] sendPduData, int bytesToSend, byte protoVersion) + private int SendS7plusPDUdata(byte[] sendPduData, int bytesToSend, byte protoVersion, bool allowTransportFragmentation = false) { m_LastError = 0; @@ -539,7 +540,9 @@ private int SendS7plusPDUdata(byte[] sendPduData, int bytesToSend, byte protoVer int sendLen; bool useLegacyDigest = ShouldUseLegacyDigest(protoVersion); int legacyDigestLength = useLegacyDigest ? LegacyDigestFieldLength : 0; - int MaxSize = GetMaxS7CommPlusPayloadSize(legacyDigestLength); + int MaxSize = allowTransportFragmentation + ? Math.Min(bytesToSend, UInt16.MaxValue - legacyDigestLength) + : GetMaxS7CommPlusPayloadSize(legacyDigestLength); if (MaxSize <= 0) { return S7Consts.errIsoInvalidPDU; @@ -1032,6 +1035,13 @@ internal bool DebugGetMultiVariablesRequestExceedsLegacySingleFrameForTests(IEnu return (itemCount, GetSerializedRequestLengthForBatching(request)); } + internal (int ItemCount, long SerializedLength) DebugCreateReadRequestBatchForTests( + IReadOnlyList addresses, int startIndex, int maxItems) + { + var request = CreateReadRequestBatch(addresses, startIndex, maxItems, out var itemCount); + return (itemCount, GetSerializedRequestLengthForBatching(request)); + } + internal int DebugSendLegacyPayloadForTests(byte[] payload) { return SendS7plusPDUdata(payload, payload.Length, ProtocolVersion.V3); @@ -1546,24 +1556,17 @@ public int ReadValues(List addresslist, out List values, ou return 0; } - // Split requests by both the PLC's item limit and the negotiated TPDU payload. + // A logical read request may span transport frames. Restricting it to one + // TPDU creates extra PLC round trips even though the send path fragments it. + // Keep the PLC item limit and the protocol's 16-bit data-length limit. int chunk_startIndex = 0; int count_perChunk = 0; int maxTagsPerRequest = Math.Max(1, m_CommunicationResources.TagsPerReadRequestMax); - int maxPayloadSize = GetSingleFramePayloadLimit(ProtocolVersion.V2); do { int res; - var getMultiVarReq = new GetMultiVariablesRequest(ProtocolVersion.V2); - - getMultiVarReq.AddressList.Clear(); - var maxCandidates = Math.Min(maxTagsPerRequest, addresslist.Count - chunk_startIndex); - count_perChunk = AppendItemsWithinPayloadLimit( - maxCandidates, - maxPayloadSize, - getMultiVarReq, - offset => getMultiVarReq.AddressList.Add(addresslist[chunk_startIndex + offset]), - () => getMultiVarReq.AddressList.RemoveAt(getMultiVarReq.AddressList.Count - 1)); + var getMultiVarReq = CreateReadRequestBatch(addresslist, chunk_startIndex, + maxTagsPerRequest, out count_perChunk); res = SendS7plusFunctionObjectAndWait(getMultiVarReq, m_ReadTimeout); if (res != 0) @@ -1603,6 +1606,31 @@ public int ReadValues(List addresslist, out List values, ou return m_LastError; } + private static GetMultiVariablesRequest CreateReadRequestBatch( + IReadOnlyList addresses, int startIndex, int maxItems, out int itemCount) + { + var request = new GetMultiVariablesRequest(ProtocolVersion.V2); + var maxCandidates = Math.Min(maxItems, addresses.Count - startIndex); + // Reserve the legacy digest and growth of the VLQ integrity id assigned on send. + var maxPayloadSize = UInt16.MaxValue - LegacyDigestFieldLength - sizeof(uint); + for (var offset = 0; offset < maxCandidates; offset++) + { + request.AddressList.Add(addresses[startIndex + offset]); + } + // Ordinary reads fit in one logical PDU: measure the complete batch once. + // Only unusually long addresses need the bounded payload search. + if (maxCandidates <= 1 || !ExceedsSingleFramePayload(request, maxPayloadSize)) + { + itemCount = maxCandidates; + return request; + } + request.AddressList.Clear(); + itemCount = AppendItemsWithinPayloadLimit(maxCandidates, maxPayloadSize, request, + offset => request.AddressList.Add(addresses[startIndex + offset]), + () => request.AddressList.RemoveAt(request.AddressList.Count - 1)); + return request; + } + public int WriteValues(List addresslist, List values, out List errors) { int res; From f7c44fd142209c2dddfa90e9007bc27a01a0de89 Mon Sep 17 00:00:00 2001 From: jogibear9988 Date: Wed, 14 Oct 2026 14:08:22 +0200 Subject: [PATCH 2/3] feat: pipeline bulk reads on a single PLC connection Add opt-in ReadPipelineDepth (1-16, default 1) to keep a bounded number of read requests outstanding on the existing TCP/TLS connection. Pass complete bulk reads to the protocol session so high-level batching does not serialize the pipeline prematurely. Each request still respects the PLC item limit and protocol PDU length limit. Match responses by sequence number, validate integrity identifiers, and restore values and per-item errors to caller order even when responses arrive out of order. Abort the logical read and close the transport on transport or validation failure, preventing pending responses from being reused by a subsequent operation. Do not publish partial tag updates when the bulk operation fails. Reserve response dispatch before sending acknowledged exchanges and throughout pipelined reads. This prevents subscription polling from consuming foreground responses; encountered notifications remain queued for their subscribers. Other operations retain existing serialization and write safety gates. Remove per-field UInt32 VLQ scratch-array allocations with a one-byte fast path and stack-based encoding. Preserve wire bytes and support all target frameworks, including the .NET Framework stream API. Document configuration, per-CPU measurement, and refreshing communication resources after connecting, especially for legacy fallback item limits. Validation: - 329 tests passed on net8.0 and 323 on net48, with no skips. - Release driver builds passed for net48/net6.0/net8.0/net9.0 with no warnings or errors; staged whitespace checks passed. - Cover bounded windows, out-of-order responses, sequence wrap, partial batches, item failures, malformed responses, integrity/transport failures, option validation, and response-dispatch reservation. - Verify UInt32 VLQ boundary encodings and allocation-free modern path. - On a TLS CPU, 21,464 identical signals improved from a 338 ms median sequential cycle to 181 ms at depth two across 20 samples per mode. - On a legacy CPU, 7,787 signals measured 1,094 ms sequentially versus 1,136 ms at depth two across 10 samples; retain depth one there. - Stable live values matched, and foreground reads with tag/alarm subscriptions succeeded on the same connection on both CPUs. - Live validation performed reads only, without writes or alarm acknowledgements. Active alarm text was not exercised. --- README.md | 28 +++ .../FakeS7CommPlusSession.cs | 2 +- .../ReadPipelineTests.cs | 187 ++++++++++++++++++ .../ResponseDispatchReservationTests.cs | 51 +++++ .../UInt32VlqEncodingTests.cs | 53 +++++ src/S7CommPlusDriver/Core/S7p.cs | 31 +-- .../IS7CommPlusBulkReadSession.cs | 11 ++ src/S7CommPlusDriver/README.md | 10 + src/S7CommPlusDriver/S7CommPlusClient.cs | 6 + .../S7CommPlusClientOptions.cs | 9 + .../S7CommPlusProtocolSession.cs | 54 ++++- .../S7CommPlusReadPipeline.cs | 79 ++++++++ 12 files changed, 504 insertions(+), 17 deletions(-) create mode 100644 src/S7CommPlusDriver.Tests/ReadPipelineTests.cs create mode 100644 src/S7CommPlusDriver.Tests/ResponseDispatchReservationTests.cs create mode 100644 src/S7CommPlusDriver.Tests/UInt32VlqEncodingTests.cs create mode 100644 src/S7CommPlusDriver/IS7CommPlusBulkReadSession.cs create mode 100644 src/S7CommPlusDriver/S7CommPlusReadPipeline.cs diff --git a/README.md b/README.md index 1130b67..2745cb7 100644 --- a/README.md +++ b/README.md @@ -386,6 +386,34 @@ matching TIA's standard element-type syntax. ## Communication Limits +### Bulk read pipelining + +For PLCs that process multiple outstanding read requests efficiently, set +`ReadPipelineDepth = 2` before creating the client: + +```csharp +await using var client = new S7CommPlusClient(new S7CommPlusClientOptions +{ + Address = "10.0.100.10", + ReadPipelineDepth = 2, + WriteEnabled = false +}); +await client.ConnectAsync(); +await client.GetCommunicationResourcesAsync(); +``` + +The default depth is `1`; valid values are `1` through `16`. Measure on the target +CPU: two in-flight requests can substantially reduce large read cycles, while an +older CPU may be faster with sequential reads. Querying communication resources +also refreshes the actual item limits on legacy connections that still have +conservative fallback limits after connecting. + +The pipeline preserves caller item order, checks response sequence/integrity +identifiers, respects each request's PLC item and protocol length limits, and routes +subscription notifications while reading. A transport or validation failure closes +the connection to discard outstanding responses before reuse/reconnect. Other +operations remain serialized; writes are not pipelined or enabled by this option. + PLC communication limits are available through the production client. This is useful before creating subscriptions or planning large batch reads: diff --git a/src/S7CommPlusDriver.Tests/FakeS7CommPlusSession.cs b/src/S7CommPlusDriver.Tests/FakeS7CommPlusSession.cs index f54d8aa..4235f4c 100644 --- a/src/S7CommPlusDriver.Tests/FakeS7CommPlusSession.cs +++ b/src/S7CommPlusDriver.Tests/FakeS7CommPlusSession.cs @@ -8,7 +8,7 @@ namespace S7CommPlusDriver.Tests { - internal sealed class FakeS7CommPlusSession : IS7CommPlusSession + internal class FakeS7CommPlusSession : IS7CommPlusSession { public bool IsConnected { get; private set; } public int ConnectCount { get; private set; } diff --git a/src/S7CommPlusDriver.Tests/ReadPipelineTests.cs b/src/S7CommPlusDriver.Tests/ReadPipelineTests.cs new file mode 100644 index 0000000..5d93a32 --- /dev/null +++ b/src/S7CommPlusDriver.Tests/ReadPipelineTests.cs @@ -0,0 +1,187 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading.Tasks; +using S7CommPlusDriver.ClientApi; +using S7CommPlusDriver.Internal; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class ReadPipelineTests + { + [Theory] + [InlineData(1, 3)] + [InlineData(2, 1)] + public async Task ProductionClientUsesBulkContractOnlyWhenPipeliningIsEnabled(int depth, int expectedCalls) + { + var session = new BulkReadSession + { + CachedCommunicationResources = new S7CommPlusCommunicationResourceSnapshot(100, 100), + ReadHandler = addresses => (0, addresses.Select(_ => (object?)new ValueBool(true)).ToList(), + addresses.Select(_ => 0UL).ToList()) + }; + await using var client = new S7CommPlusClient(new S7CommPlusClientOptions + { Address = "127.0.0.1", ReadPipelineDepth = depth }, () => session); + var tags = Enumerable.Range(0, 250).Select(index => (PlcTag)new PlcTagBool( + $"Tag{index}", new ItemAddress($"8A0E0001.{index:X}"), Softdatatype.S7COMMP_SOFTDATATYPE_BOOL)).ToArray(); + + var result = await client.ReadAsync(tags); + + Assert.Equal(expectedCalls, session.ReadCount); + Assert.Equal(250, result.Items.Count); + Assert.All(result.Items, item => Assert.True(item.IsSuccess)); + Assert.All(tags, tag => Assert.True(Assert.IsType(tag).Value)); + } + + [Fact] + public async Task ProductionClientDoesNotPublishPartialTagsAfterBulkTransportFailure() + { + var session = new BulkReadSession + { + ReadHandler = addresses => (S7Consts.errTCPConnectionReset, + addresses.Select(_ => (object?)new ValueBool(true)).ToList(), addresses.Select(_ => 0UL).ToList()) + }; + await using var client = new S7CommPlusClient(new S7CommPlusClientOptions + { Address = "127.0.0.1", ReadPipelineDepth = 2, AutoReconnect = false }, () => session); + var tags = Enumerable.Range(0, 250).Select(index => new PlcTagBool( + $"Tag{index}", new ItemAddress($"8A0E0001.{index:X}"), Softdatatype.S7COMMP_SOFTDATATYPE_BOOL)).ToArray(); + + await Assert.ThrowsAsync(() => client.ReadAsync(tags)); + Assert.All(tags, tag => Assert.False(tag.Value)); + } + + private sealed class BulkReadSession : FakeS7CommPlusSession, IS7CommPlusBulkReadSession { } + + [Theory] + [InlineData(1, false)] + [InlineData(2, false)] + [InlineData(2, true)] + [InlineData(4, true)] + public void WindowIsBoundedAndOutOfOrderValuesReturnInCallerOrder(int depth, bool reverse) + { + var harness = new PipelineHarness { Reverse = reverse, NextSequence = ushort.MaxValue - 1 }; + var result = harness.Execute(107, 10, depth, out var values, out var errors); + + Assert.Equal(0, result); + Assert.Equal(107, values.Count); + Assert.Equal(Enumerable.Range(0, 107).Select(i => (uint)i), values.Select(v => Assert.IsType(v).GetValue())); + Assert.All(errors, error => Assert.Equal(0UL, error)); + Assert.InRange(harness.MaximumPending, 1, depth); + Assert.Equal(11, harness.RequestSizes.Count); + Assert.Equal(7, harness.RequestSizes.Last()); + Assert.Empty(harness.Pending); + } + + [Fact] + public void PerItemFailuresRemainAtTheOriginalIndex() + { + var harness = new PipelineHarness { FailedItem = 12, Reverse = true }; + Assert.Equal(0, harness.Execute(23, 10, 2, out var values, out var errors)); + Assert.Null(values[12]); + Assert.Equal(0x8000UL, errors[12]); + Assert.Equal(22, values.Count(v => v != null)); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void MissingOrInvalidItemNumbersCannotMasqueradeAsSuccessfulValues(bool invalidIndex) + { + var harness = new PipelineHarness { InvalidIndex = invalidIndex, MissingItem = !invalidIndex }; + var result = harness.Execute(23, 10, 2, out var values, out var errors); + if (invalidIndex) Assert.Equal(S7Consts.errIsoInvalidPDU, result); + else + { + Assert.Equal(0, result); + Assert.Null(values[0]); + Assert.Equal(ulong.MaxValue, errors[0]); + } + } + + [Theory] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + public void TransportValidationAndIntegrityFailuresAbortTheLogicalRead(int failure) + { + var harness = new PipelineHarness { Failure = failure }; + Assert.NotEqual(0, harness.Execute(107, 10, 2, out _, out _)); + Assert.InRange(harness.RequestSizes.Count, 1, 2); + } + + [Theory] + [InlineData(0)] + [InlineData(17)] + public void PipelineDepthIsValidatedBeforeConnecting(int depth) + { + Assert.Throws(() => new S7CommPlusClient( + new S7CommPlusClientOptions { Address = "127.0.0.1", ReadPipelineDepth = depth })); + } + + private sealed class PipelineHarness + { + internal readonly List Pending = new List(); + internal readonly List RequestSizes = new List(); + internal int MaximumPending; + internal bool Reverse, InvalidIndex, MissingItem; + internal int FailedItem = -1, Failure; + internal ushort NextSequence; + + internal int Execute(int count, int limit, int depth, out List values, out List errors) + { + var addresses = Enumerable.Range(0, count).Select(i => new ItemAddress($"8A0E0001.{i:X}")).ToArray(); + return S7CommPlusReadPipeline.Execute(addresses, limit, depth, Send, Receive, + (_, _) => Failure == 2 ? S7Consts.errIsoInvalidPDU5 : 0, out values, out errors); + } + + private int Send(GetMultiVariablesRequest request) + { + request.SequenceNumber = unchecked(++NextSequence); + request.IntegrityId = request.SequenceNumber; + Pending.Add(request); + RequestSizes.Add(request.AddressList.Count); + MaximumPending = Math.Max(MaximumPending, Pending.Count); + return 0; + } + + private (int Error, MemoryStream Pdu) Receive(GetMultiVariablesRequest expected, ISet accepted) + { + if (Failure == 1) return (S7Consts.errCliJobTimeout, null!); + var index = Reverse ? Pending.Count - 1 : 0; + var request = Pending[index]; + Assert.Contains(request.SequenceNumber, accepted); + Pending.RemoveAt(index); + var stream = new MemoryStream(); + S7p.EncodeByte(stream, ProtocolVersion.V2); + S7p.EncodeByte(stream, Opcode.Response); + S7p.EncodeUInt16(stream, 0); + S7p.EncodeUInt16(stream, Functioncode.GetMultiVariables); + S7p.EncodeUInt16(stream, 0); + S7p.EncodeUInt16(stream, request.SequenceNumber); + S7p.EncodeByte(stream, S7CommPlusProtocolConstants.RequestWithResponseTransportFlags); + S7p.EncodeUInt64Vlq(stream, FailedItem >= 0 ? 1UL : 0UL); + for (var item = 0; item < request.AddressList.Count; item++) + { + var value = request.AddressList[item].GetLocalId(0); + if (value == FailedItem || (MissingItem && value == 0)) continue; + S7p.EncodeUInt32Vlq(stream, InvalidIndex ? (uint)request.AddressList.Count + 1 : (uint)item + 1); + new ValueUDInt(value).Serialize(stream); + if (InvalidIndex) break; + } + S7p.EncodeUInt32Vlq(stream, 0); + for (var item = 0; item < request.AddressList.Count; item++) + if (request.AddressList[item].GetLocalId(0) == FailedItem) + { + S7p.EncodeUInt32Vlq(stream, (uint)item + 1); + S7p.EncodeUInt64Vlq(stream, 0x8000); + } + S7p.EncodeUInt32Vlq(stream, 0); + S7p.EncodeUInt32Vlq(stream, Failure == 3 ? 0u : unchecked(request.IntegrityId + request.SequenceNumber)); + stream.Position = 0; + return (0, stream); + } + } + } +} diff --git a/src/S7CommPlusDriver.Tests/ResponseDispatchReservationTests.cs b/src/S7CommPlusDriver.Tests/ResponseDispatchReservationTests.cs new file mode 100644 index 0000000..de2d268 --- /dev/null +++ b/src/S7CommPlusDriver.Tests/ResponseDispatchReservationTests.cs @@ -0,0 +1,51 @@ +using System; +using System.Reflection; +using System.Threading; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class ResponseDispatchReservationTests + { + [Theory] + [InlineData(false)] + [InlineData(true)] + public void AcknowledgedSendWaitsForNotificationDispatcherBeforeSending(bool raw) + { + var session = new S7CommPlusProtocolSession(); + var flags = BindingFlags.Instance | BindingFlags.NonPublic; + var dispatchLock = typeof(S7CommPlusProtocolSession).GetField("m_ReceiveDispatchLock", flags)!.GetValue(session)!; + var method = typeof(S7CommPlusProtocolSession).GetMethod(raw + ? "SendRawS7plusPduAndWait" : "SendS7plusFunctionObjectAndWait", flags)!; + using var started = new ManualResetEventSlim(); + using var completed = new ManualResetEventSlim(); + Exception? failure = null; + Thread worker; + // Simulate a notification poller that currently owns response dispatch. + Monitor.Enter(dispatchLock); + try + { + worker = new Thread(() => + { + started.Set(); + failure = Record.Exception(() => method.Invoke(session, raw + ? new object[] { new byte[] { 0x31 }, 1, ProtocolVersion.V2, 100 } + : new object[] { new GetMultiVariablesRequest(ProtocolVersion.V2), 100 })); + completed.Set(); + }) { IsBackground = true }; + worker.Start(); + Assert.True(started.Wait(2000)); + Assert.False(completed.Wait(50)); + } + finally + { + Monitor.Exit(dispatchLock); + } + Assert.True(completed.Wait(2000)); + Assert.True(worker.Join(2000)); + // There is intentionally no transport: reaching the send after releasing + // dispatch fails immediately, proving it did not start while the poller owned it. + Assert.IsType(failure); + } + } +} diff --git a/src/S7CommPlusDriver.Tests/UInt32VlqEncodingTests.cs b/src/S7CommPlusDriver.Tests/UInt32VlqEncodingTests.cs new file mode 100644 index 0000000..d8789a9 --- /dev/null +++ b/src/S7CommPlusDriver.Tests/UInt32VlqEncodingTests.cs @@ -0,0 +1,53 @@ +using System; +using System.IO; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class UInt32VlqEncodingTests + { + [Theory] + [InlineData(0u, new byte[] { 0x00 })] + [InlineData(127u, new byte[] { 0x7f })] + [InlineData(128u, new byte[] { 0x81, 0x00 })] + [InlineData(16383u, new byte[] { 0xff, 0x7f })] + [InlineData(16384u, new byte[] { 0x81, 0x80, 0x00 })] + [InlineData(2097151u, new byte[] { 0xff, 0xff, 0x7f })] + [InlineData(2097152u, new byte[] { 0x81, 0x80, 0x80, 0x00 })] + [InlineData(268435455u, new byte[] { 0xff, 0xff, 0xff, 0x7f })] + [InlineData(268435456u, new byte[] { 0x81, 0x80, 0x80, 0x80, 0x00 })] + [InlineData(uint.MaxValue, new byte[] { 0x8f, 0xff, 0xff, 0xff, 0x7f })] + public void EncoderPreservesWireBytesAndFollowingValue(uint value, byte[] expected) + { + using var stream = new MemoryStream(); + var length = S7p.EncodeUInt32Vlq(stream, value); + Assert.Equal(expected.Length, length); + Assert.Equal(expected, stream.ToArray()); + stream.WriteByte(42); + stream.Position = 0; + Assert.Equal(length, S7p.DecodeUInt32Vlq(stream, out var decoded)); + Assert.Equal(value, decoded); + Assert.Equal(42, stream.ReadByte()); + } + +#if NET6_0_OR_GREATER + [Fact] + public void RepeatedAddressFieldEncodingDoesNotAllocateScratchArrays() + { + using var stream = new MemoryStream(5); + for (var i = 0; i < 100; i++) + { + stream.Position = 0; + S7p.EncodeUInt32Vlq(stream, uint.MaxValue); + } + var before = GC.GetAllocatedBytesForCurrentThread(); + for (var i = 0; i < 10000; i++) + { + stream.Position = 0; + S7p.EncodeUInt32Vlq(stream, (uint)i); + } + Assert.Equal(before, GC.GetAllocatedBytesForCurrentThread()); + } +#endif + } +} diff --git a/src/S7CommPlusDriver/Core/S7p.cs b/src/S7CommPlusDriver/Core/S7p.cs index b5de3b0..b668b13 100644 --- a/src/S7CommPlusDriver/Core/S7p.cs +++ b/src/S7CommPlusDriver/Core/S7p.cs @@ -230,22 +230,29 @@ public static int DecodeInt64(System.IO.Stream buffer, out Int64 value) public static int EncodeUInt32Vlq(System.IO.Stream buffer, UInt32 value) { - byte[] bytes = new byte[5]; - int i, j; - for (i = 4; i > 0; i--) + if (value < 0x80) { - if ((value & (0x7f << (i * 7))) > 0) - { - break; - } + buffer.WriteByte((byte)value); + return 1; } - for (j = 0; j <= i; j++) + + // Address fields are encoded for every read request. Keep the scratch + // bytes on the stack rather than allocating an array for each field. + Span bytes = stackalloc byte[5]; + var start = bytes.Length - 1; + bytes[start] = (byte)(value & 0x7f); + while ((value >>= 7) != 0) { - bytes[j] = (byte)(((value >> ((i - j) * 7)) & 0x7f) | 0x80); + bytes[--start] = (byte)((value & 0x7f) | 0x80); } - bytes[i] ^= 0x80; - buffer.Write(bytes, 0, i + 1); - return i + 1; + var length = bytes.Length - start; +#if NET6_0_OR_GREATER + buffer.Write(bytes.Slice(start, length)); +#else + for (var index = start; index < bytes.Length; index++) + buffer.WriteByte(bytes[index]); +#endif + return length; } public static int DecodeUInt32Vlq(System.IO.Stream buffer, out UInt32 value) diff --git a/src/S7CommPlusDriver/IS7CommPlusBulkReadSession.cs b/src/S7CommPlusDriver/IS7CommPlusBulkReadSession.cs new file mode 100644 index 0000000..2c962f2 --- /dev/null +++ b/src/S7CommPlusDriver/IS7CommPlusBulkReadSession.cs @@ -0,0 +1,11 @@ +using System.Collections.Generic; + +namespace S7CommPlusDriver +{ + // Sessions implementing this contract own item-count batching for a complete + // logical read, allowing the production protocol session to overlap requests. + internal interface IS7CommPlusBulkReadSession + { + int ReadValues(List addresses, out List values, out List errors); + } +} diff --git a/src/S7CommPlusDriver/README.md b/src/S7CommPlusDriver/README.md index 28199da..b239657 100644 --- a/src/S7CommPlusDriver/README.md +++ b/src/S7CommPlusDriver/README.md @@ -153,6 +153,16 @@ Each `S7CommPlusAlarm` exposes `SourceRelationId` and `SourceAlarmId`, decoded f Communication limits are exposed through `GetCommunicationResourcesAsync()`, including max read/write batch sizes, available PLC subscription slots, and subscription memory. +For large reads, `S7CommPlusClientOptions.ReadPipelineDepth = 2` can overlap two +requests on one connection. The default is `1` (sequential), with supported depths +from `1` through `16`. Benchmark the target CPU before increasing it; older CPUs +can be faster at depth one. After connecting, `GetCommunicationResourcesAsync()` +refreshes negotiated item limits, including conservative legacy fallback limits. +The pipeline preserves caller result order, validates response identifiers, routes +notifications, and closes the connection on transport/validation failure so pending +responses cannot contaminate the next operation. This setting applies to reads; +other operations retain their existing serialization and write safety gates. + ## Block Metadata and Online View Block metadata is available through `BrowseBlocksAsync()`, `GetPlcStructureXmlAsync()`, `BrowseBlockStructureAsync()`, and `GetBlockContentAsync(relid)`. diff --git a/src/S7CommPlusDriver/S7CommPlusClient.cs b/src/S7CommPlusDriver/S7CommPlusClient.cs index 66eed87..5e525e4 100644 --- a/src/S7CommPlusDriver/S7CommPlusClient.cs +++ b/src/S7CommPlusDriver/S7CommPlusClient.cs @@ -1493,6 +1493,12 @@ public Task> WriteAsync(IEnumerable IS7CommPlusSession session, IReadOnlyList requestTags) { + if (_options.ReadPipelineDepth > 1 && session is IS7CommPlusBulkReadSession bulkSession) + { + var error = bulkSession.ReadValues(requestTags.Select(tag => tag.Address).ToList(), out var bulkValues, out var bulkErrors); + ThrowIfError("ReadTags", error); + return (bulkValues, bulkErrors); + } var values = new List(requestTags.Count); var itemErrors = new List(requestTags.Count); for (var batchStart = 0; batchStart < requestTags.Count; batchStart += _tagsPerReadRequestMax) diff --git a/src/S7CommPlusDriver/S7CommPlusClientOptions.cs b/src/S7CommPlusDriver/S7CommPlusClientOptions.cs index f5fa475..edafd77 100644 --- a/src/S7CommPlusDriver/S7CommPlusClientOptions.cs +++ b/src/S7CommPlusDriver/S7CommPlusClientOptions.cs @@ -39,6 +39,13 @@ public string RemoteTsap public bool AutoReconnect { get; set; } = true; public bool WriteEnabled { get; set; } = false; /// + /// Maximum number of read requests in flight on one connection. The default + /// of one keeps sequential request behavior. Two can improve bulk reads on + /// CPUs that process multiple outstanding requests; measure on the target PLC. + /// Does not enable writes or parallelize other operations. + /// + public int ReadPipelineDepth { get; set; } = 1; + /// /// Enables periodic renewal of the integrity key for legacy challenge-authenticated sessions. /// public bool LegacySessionKeyRefreshEnabled { get; set; } = true; @@ -114,6 +121,8 @@ internal string GetLegacyPublicKeyFingerprint(string plcFingerprint) internal void Validate() { + if (ReadPipelineDepth < 1 || ReadPipelineDepth > 16) + throw new ArgumentOutOfRangeException(nameof(ReadPipelineDepth), "Read pipeline depth must be between 1 and 16."); if (string.IsNullOrWhiteSpace(Address)) { throw new ArgumentException("PLC address is required.", nameof(Address)); diff --git a/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs b/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs index 699f40f..e3bff99 100644 --- a/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs +++ b/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs @@ -29,7 +29,7 @@ namespace S7CommPlusDriver { - internal partial class S7CommPlusProtocolSession + internal partial class S7CommPlusProtocolSession : IS7CommPlusBulkReadSession { #region Private Members private S7Client m_client; @@ -53,6 +53,7 @@ public UInt32 SessionId2 } private int m_ReadTimeout = 5000; + private int m_ReadPipelineDepth = 1; private UInt16 m_SequenceNumber = 0; private UInt32 m_IntegrityId = 0; private UInt32 m_IntegrityId_Set = 0; @@ -237,6 +238,7 @@ private int SendS7plusFunctionObjectSerialized(IS7pRequest funcObj) private int SendS7plusFunctionObjectAndWait(IS7pRequest funcObj, int timeout) { lock (m_RequestLock) + lock (m_ReceiveDispatchLock) { try { @@ -262,6 +264,7 @@ private int SendS7plusFunctionObjectAndWait(IS7pRequest funcObj, int timeout) private int SendRawS7plusPduAndWait(byte[] sendPduData, int bytesToSend, byte protoVersion, int timeout) { lock (m_RequestLock) + lock (m_ReceiveDispatchLock) { var result = SendS7plusPDUdata(sendPduData, bytesToSend, protoVersion); if (result != 0) @@ -335,7 +338,7 @@ private int WaitForNotification(uint subscriptionObjectId, int timeout, out Noti return S7Consts.errCliJobTimeout; } - private int DispatchOneReceivedPdu(int timeout, IS7pRequest expectedResponse, uint notificationSubscriptionObjectId, out MemoryStream responsePdu, out Notification matchingNotification) + private int DispatchOneReceivedPdu(int timeout, IS7pRequest expectedResponse, uint notificationSubscriptionObjectId, out MemoryStream responsePdu, out Notification matchingNotification, ISet acceptedResponseSequences = null) { responsePdu = null; matchingNotification = null; @@ -359,7 +362,8 @@ private int DispatchOneReceivedPdu(int timeout, IS7pRequest expectedResponse, ui { if (expectedResponse != null && function == expectedResponse.FunctionCode - && sequenceNumber == expectedResponse.SequenceNumber) + && (sequenceNumber == expectedResponse.SequenceNumber + || acceptedResponseSequences?.Contains(sequenceNumber) == true)) { pdu.Position = 0; responsePdu = pdu; @@ -1116,6 +1120,7 @@ internal int Connect(S7CommPlusClientOptions options) { return S7Consts.errCliInvalidParams; } + m_ReadPipelineDepth = options.ReadPipelineDepth; switch (options.SecurityMode) { @@ -1541,6 +1546,33 @@ private int DeleteObject(uint deleteObjectId) public int ReadValues(List addresslist, out List values, out List errors) { + if (m_ReadPipelineDepth > 1 && addresslist.Count > Math.Max(1, m_CommunicationResources.TagsPerReadRequestMax)) + { + lock (m_RequestLock) + // Reserve response dispatch before sending. Subscription pollers + // otherwise can consume a read response while expecting notifications. + lock (m_ReceiveDispatchLock) + { + try + { + var result = S7CommPlusReadPipeline.Execute(addresslist, + Math.Max(1, m_CommunicationResources.TagsPerReadRequestMax), m_ReadPipelineDepth, + request => SendS7plusFunctionObject(request), ReceivePipelinedReadResponse, + checkResponseWithIntegrity, out values, out errors); + if (result != 0) CloseTransport(2000); + return result; + } + catch + { + CloseTransport(2000); + throw; + } + finally + { + m_LastSentRequestForWait = null; + } + } + } // The requester must pass the internal type with the request, otherwise not all return values can be converted automatically. // For example, strings are transmitted as UInt-Array. values = new List(); @@ -1606,7 +1638,7 @@ public int ReadValues(List addresslist, out List values, ou return m_LastError; } - private static GetMultiVariablesRequest CreateReadRequestBatch( + internal static GetMultiVariablesRequest CreateReadRequestBatch( IReadOnlyList addresses, int startIndex, int maxItems, out int itemCount) { var request = new GetMultiVariablesRequest(ProtocolVersion.V2); @@ -1631,6 +1663,20 @@ private static GetMultiVariablesRequest CreateReadRequestBatch( return request; } + private (int Error, MemoryStream Pdu) ReceivePipelinedReadResponse( + GetMultiVariablesRequest expected, ISet acceptedSequences) + { + var deadline = RuntimeCompatibility.TickCount64 + Math.Max(1, m_ReadTimeout); + while (RuntimeCompatibility.TickCount64 < deadline) + { + var remaining = (int)Math.Min(50, Math.Max(1, deadline - RuntimeCompatibility.TickCount64)); + var result = DispatchOneReceivedPdu(remaining, expected, 0, out var response, out _, acceptedSequences); + if (result == S7Consts.errCliJobTimeout || result == S7Consts.errTCPReceiveTimeout) continue; + if (result != 0 || response != null) return (result, response); + } + return (S7Consts.errCliJobTimeout, null); + } + public int WriteValues(List addresslist, List values, out List errors) { int res; diff --git a/src/S7CommPlusDriver/S7CommPlusReadPipeline.cs b/src/S7CommPlusDriver/S7CommPlusReadPipeline.cs new file mode 100644 index 0000000..5a3a581 --- /dev/null +++ b/src/S7CommPlusDriver/S7CommPlusReadPipeline.cs @@ -0,0 +1,79 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; + +namespace S7CommPlusDriver +{ + internal static class S7CommPlusReadPipeline + { + private sealed class PendingRead + { + internal GetMultiVariablesRequest Request; + internal int Start; + internal int Count; + } + + internal static int Execute( + IReadOnlyList addresses, int maxItems, int depth, + Func send, + Func, (int Error, MemoryStream Pdu)> receive, + Func validate, + out List values, out List errors) + { + values = Enumerable.Repeat(null, addresses.Count).ToList(); + errors = Enumerable.Repeat(ulong.MaxValue, addresses.Count).ToList(); + var pending = new Dictionary(); + var sequences = new HashSet(); + var next = 0; + + int FillWindow() + { + while (pending.Count < depth && next < addresses.Count) + { + var request = S7CommPlusProtocolSession.CreateReadRequestBatch(addresses, next, maxItems, out var count); + var error = send(request); + if (error != 0) return error; + if (!sequences.Add(request.SequenceNumber)) return S7Consts.errIsoInvalidPDU5; + pending.Add(request.SequenceNumber, new PendingRead { Request = request, Start = next, Count = count }); + next += count; + } + return 0; + } + + var result = FillWindow(); + if (result != 0) return result; + while (pending.Count > 0) + { + var received = receive(pending.Values.First().Request, sequences); + if (received.Error != 0) return received.Error; + var response = GetMultiVariablesResponse.DeserializeFromPdu(received.Pdu); + if (response == null || !pending.TryGetValue(response.SequenceNumber, out var completed)) + return S7Consts.errIsoInvalidPDU5; + result = validate(completed.Request, response); + if (result != 0) return result; + if (response.IntegrityId != unchecked(completed.Request.IntegrityId + completed.Request.SequenceNumber)) + return S7Consts.errIsoInvalidPDU; + // Preserve item failures and missing-value status; never expose partial + // data as a successful logical read after a transport/validation failure. + foreach (var value in response.Values) + { + if (value.Key == 0 || value.Key > completed.Count) return S7Consts.errIsoInvalidPDU; + var index = completed.Start + (int)value.Key - 1; + values[index] = value.Value; + errors[index] = 0; + } + foreach (var error in response.ErrorValues) + { + if (error.Key == 0 || error.Key > completed.Count) return S7Consts.errIsoInvalidPDU; + errors[completed.Start + (int)error.Key - 1] = error.Value; + } + pending.Remove(response.SequenceNumber); + sequences.Remove(response.SequenceNumber); + result = FillWindow(); + if (result != 0) return result; + } + return 0; + } + } +} From ae8f9af23979969b832e6604d07bdf35b3b248de Mon Sep 17 00:00:00 2001 From: jogibear9988 Date: Wed, 14 Oct 2026 16:41:41 +0200 Subject: [PATCH 3/3] perf: reduce bulk-read allocations and transport copies Decode bulk-read responses directly into ordered staging buffers instead of building per-response dictionaries. Validate item indices and duplicates, preserve per-item errors and missing values, and retain sequence/integrity checks before publishing the complete read result. Replace the asynchronous channel bridge with a monitor-based receive FIFO for the synchronous protocol worker. Preserve timeout, completion and receive-error behavior while removing per-response task and cancellation-source allocations. Remove the now-unused net48 channel dependency. Calculate serialized read-request lengths without encoding temporary payloads, preallocate request/result buffers, send the active MemoryStream buffer directly, and reuse immutable flag-zero scalar boolean values. Remove numeric and VLQ scratch arrays with targeted spans, stack storage and allocation-free IEEE bit conversion. Decode UTF-8 directly from exposed stream storage; use stack or pooled buffers for UTF-8 encoding. Preserve byte order, VLQ boundary behavior, stream origins and floating-point bit patterns. Deliver TLS plaintext and encrypted record slices through internal synchronous borrowed-buffer callbacks. Preserve owned exact-length buffers for public callbacks. Pool COTP fragments, send only their actual wire lengths, and validate active receive lengths so spare buffer capacity cannot become parsed payload. Fix CreateObject response parsing to honor the optional response-object flag. After long read runs, an integrity counter can begin with an object-like marker such as 0xAB; it must not be decoded as a variable-type list when no object is present. Keep diagnostic formatting safe for responses without objects. Validation: - 822 tests passed across net8.0 and net48, with no skips. - Release builds passed for net48/net6.0/net8.0/net9.0 without warnings or errors. - All 110 captured CreateObject response fixtures replayed successfully. - Read-only live checks passed on TLS and legacy PLCs, with zero stable value mismatches and 20/5 additional subscription create/read/dispose cycles. - The 21,464-signal depth-two workload allocated approximately 4.29 MB per cycle versus 6.93 MB before the span/buffer pass, a reduction of about 38%. - Latest read medians were 167.9 ms at depth two and 304.6 ms sequentially. Separate live runs vary; these timings do not establish a universal speedup. No PLC values were written or alarms acknowledged. --- .../BouncyCastleTlsConnectorTests.cs | 44 +++- .../CreateObjectResponseFlagsTests.cs | 61 ++++++ .../OrderedReadResponseTests.cs | 100 +++++++++ .../PlaintextBufferOwnershipTests.cs | 73 +++++++ .../ReadRequestLengthTests.cs | 60 ++++++ .../ReceiveQueueTests.cs | 69 ++++++ .../S7ClientTransportTests.cs | 14 ++ .../S7CommPlusProtocolSessionReceiveTests.cs | 18 ++ .../ScalarBooleanReadTests.cs | 35 ++++ src/S7CommPlusDriver.Tests/SpanCodecTests.cs | 197 ++++++++++++++++++ src/S7CommPlusDriver/ClientApi/ItemAddress.cs | 11 + .../Core/CreateObjectResponse.cs | 8 +- .../Core/GetMultiVariablesReadResponse.cs | 69 ++++++ .../Core/GetMultiVariablesRequest.cs | 17 ++ .../Core/GetMultiVariablesResponse.cs | 10 +- src/S7CommPlusDriver/Core/PValue.cs | 10 +- src/S7CommPlusDriver/Core/S7p.cs | 188 +++++++++++------ .../Internal/S7CommPlusReceiveQueue.cs | 58 ++++++ src/S7CommPlusDriver/Net/IS7TlsConnector.cs | 12 ++ src/S7CommPlusDriver/Net/S7Client.cs | 94 ++++++--- src/S7CommPlusDriver/S7CommPlusDriver.csproj | 1 - .../S7CommPlusProtocolSession.cs | 113 +++++----- .../S7CommPlusReadPipeline.cs | 18 +- .../Tls/BouncyCastleTlsConnector.cs | 20 +- 24 files changed, 1120 insertions(+), 180 deletions(-) create mode 100644 src/S7CommPlusDriver.Tests/CreateObjectResponseFlagsTests.cs create mode 100644 src/S7CommPlusDriver.Tests/OrderedReadResponseTests.cs create mode 100644 src/S7CommPlusDriver.Tests/PlaintextBufferOwnershipTests.cs create mode 100644 src/S7CommPlusDriver.Tests/ReadRequestLengthTests.cs create mode 100644 src/S7CommPlusDriver.Tests/ReceiveQueueTests.cs create mode 100644 src/S7CommPlusDriver.Tests/ScalarBooleanReadTests.cs create mode 100644 src/S7CommPlusDriver.Tests/SpanCodecTests.cs create mode 100644 src/S7CommPlusDriver/Core/GetMultiVariablesReadResponse.cs create mode 100644 src/S7CommPlusDriver/Internal/S7CommPlusReceiveQueue.cs diff --git a/src/S7CommPlusDriver.Tests/BouncyCastleTlsConnectorTests.cs b/src/S7CommPlusDriver.Tests/BouncyCastleTlsConnectorTests.cs index c94966c..28d6704 100644 --- a/src/S7CommPlusDriver.Tests/BouncyCastleTlsConnectorTests.cs +++ b/src/S7CommPlusDriver.Tests/BouncyCastleTlsConnectorTests.cs @@ -24,15 +24,32 @@ public sealed class BouncyCastleTlsConnectorTests private const string OmsExporterLabel = "EXPERIMENTAL_OMS"; private const int OmsExporterSecretLength = 32; - [Fact] - public void OmsExporterSecretRemainsAvailableAfterHandshakeCompletion() + [Theory] + [InlineData(false)] + [InlineData(true)] + public void OmsExporterSecretAndApplicationDataWorkWithOwnedAndBorrowedCallbacks(bool borrowed) { var serverInput = new BlockingInputStream(); BouncyCastleTlsConnector? connector = null; var serverOutput = new CallbackOutputStream( (data, offset, count) => connector!.ReadCompleted(Copy(data, offset, count), count)); var callback = new LoopbackConnectorCallback(serverInput); - connector = new BouncyCastleTlsConnector(callback); + using var plaintextReady = new ManualResetEventSlim(); + byte[]? plaintext = null; + var marker = System.Text.Encoding.UTF8.GetBytes("TLS application data"); + callback.DataAvailable = () => + { + var buffer = new byte[4096]; + var count = connector!.Receive(ref buffer, buffer.Length); + plaintext = Copy(buffer, 0, count); + plaintextReady.Set(); + }; + IS7TlsConnectorCallback sink = borrowed ? new BorrowedLoopbackCallback(callback, (data, count) => + { + plaintext = Copy(data, 0, count); + plaintextReady.Set(); + }) : callback; + connector = new BouncyCastleTlsConnector(sink); var server = new ExporterTlsServer(); Exception? serverException = null; @@ -42,6 +59,8 @@ public void OmsExporterSecretRemainsAvailableAfterHandshakeCompletion() { var protocol = new TlsServerProtocol(serverInput, serverOutput); protocol.Accept(server); + protocol.Stream.Write(marker, 0, marker.Length); + protocol.Stream.Flush(); } catch (Exception ex) { @@ -60,6 +79,8 @@ public void OmsExporterSecretRemainsAvailableAfterHandshakeCompletion() Assert.True(serverThread.Join(TimeSpan.FromSeconds(10)), "TLS server handshake did not complete."); Assert.Null(serverException); + Assert.True(plaintextReady.Wait(TimeSpan.FromSeconds(10)), "TLS application data was not delivered."); + Assert.Equal(marker, plaintext); var firstExport = connector.GetOmsExporterSecret(); var secondExport = connector.GetOmsExporterSecret(); @@ -163,6 +184,7 @@ private static byte[] CreateCertificate(AsymmetricCipherKeyPair keyPair) private sealed class LoopbackConnectorCallback : IS7TlsConnectorCallback { private readonly BlockingInputStream _serverInput; + internal Action? DataAvailable; public LoopbackConnectorCallback(BlockingInputStream serverInput) { @@ -176,6 +198,7 @@ public void WriteData(byte[] data, int dataLength) public void OnDataAvailable() { + DataAvailable?.Invoke(); } public void OnSslError(int sslError, string sslState) @@ -183,6 +206,21 @@ public void OnSslError(int sslError, string sslState) } } + private sealed class BorrowedLoopbackCallback : IS7TlsConnectorCallback, IS7TlsPlaintextCallback, IS7TlsRecordCallback + { + private readonly LoopbackConnectorCallback _owned; + private readonly Action _receive; + internal BorrowedLoopbackCallback(LoopbackConnectorCallback owned, Action receive) + { + _owned = owned; _receive = receive; + } + public void WriteData(byte[] data, int dataLength) => _owned.WriteData(data, dataLength); + public void WriteEncryptedData(byte[] data, int offset, int count) => _owned.WriteData(Copy(data, offset, count), count); + public void OnPlaintextData(byte[] data, int dataLength) => _receive(data, dataLength); + public void OnDataAvailable() => throw new InvalidOperationException("Borrowed TLS delivery must not queue an owned payload."); + public void OnSslError(int sslError, string sslState) => _owned.OnSslError(sslError, sslState); + } + private sealed class BlockingInputStream : Stream { private readonly BlockingCollection _incoming = new BlockingCollection(); diff --git a/src/S7CommPlusDriver.Tests/CreateObjectResponseFlagsTests.cs b/src/S7CommPlusDriver.Tests/CreateObjectResponseFlagsTests.cs new file mode 100644 index 0000000..fe66efc --- /dev/null +++ b/src/S7CommPlusDriver.Tests/CreateObjectResponseFlagsTests.cs @@ -0,0 +1,61 @@ +using System; +using System.IO; +using System.Linq; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class CreateObjectResponseFlagsTests + { + [Fact] + public void CapturedSubscriptionResponseDoesNotParseIntegrityAsVartypeList() + { + // Real response after a long bulk-read run: trailing integrity ID starts 0xAB. + using var stream = new MemoryStream(Hex("0232000004CA000015E93400018780809F72AB6A00000000")); + var response = CreateObjectResponse.DeserializeFromPdu(stream)!; + Assert.NotNull(response); Assert.Equal(0UL, response.ReturnValue); + Assert.Single(response.ObjectIds); Assert.Null(response.ResponseObject); + Assert.Equal(18, stream.Position); + } + + [Theory] + [InlineData(0x1080u)] + [InlineData(0x1100u)] + [InlineData(0x1180u)] + [InlineData(0x15eau)] + [InlineData(0x1600u)] + public void AllObjectLikeCounterPrefixesAreIgnoredWhenObjectFlagIsAbsent(uint integrity) + { + using var stream = Header(0x34); + S7p.EncodeUInt32Vlq(stream, integrity); S7p.EncodeUInt32(stream, 0); + stream.Position = 0; + var response = CreateObjectResponse.DeserializeFromPdu(stream)!; + Assert.Null(response.ResponseObject); Assert.Equal(new uint[] { 42 }, response.ObjectIds); + } + + [Fact] + public void DeclaredResponseObjectIsStillDecoded() + { + using var stream = Header(0x36); + S7p.EncodeByte(stream, ElementID.StartOfObject); S7p.EncodeUInt32(stream, 123); + S7p.EncodeUInt32Vlq(stream, 456); S7p.EncodeUInt32Vlq(stream, 0); S7p.EncodeUInt32Vlq(stream, 0); + S7p.EncodeByte(stream, ElementID.TerminatingObject); + S7p.EncodeByte(stream, 0); // Terminates the enclosing object decoder. + stream.Position = 0; + var response = CreateObjectResponse.DeserializeFromPdu(stream)!; + Assert.NotNull(response.ResponseObject); Assert.Equal(stream.Length, stream.Position); + } + + private static MemoryStream Header(byte flags) + { + var stream = new MemoryStream(); + S7p.EncodeByte(stream, ProtocolVersion.V2); S7p.EncodeByte(stream, Opcode.Response); + S7p.EncodeUInt16(stream, 0); S7p.EncodeUInt16(stream, Functioncode.CreateObject); S7p.EncodeUInt16(stream, 0); + S7p.EncodeUInt16(stream, 1); S7p.EncodeByte(stream, flags); S7p.EncodeUInt64Vlq(stream, 0); + S7p.EncodeByte(stream, 1); S7p.EncodeUInt32Vlq(stream, 42); + return stream; + } + private static byte[] Hex(string value) => Enumerable.Range(0, value.Length / 2) + .Select(i => Convert.ToByte(value.Substring(i * 2, 2), 16)).ToArray(); + } +} diff --git a/src/S7CommPlusDriver.Tests/OrderedReadResponseTests.cs b/src/S7CommPlusDriver.Tests/OrderedReadResponseTests.cs new file mode 100644 index 0000000..73b4be6 --- /dev/null +++ b/src/S7CommPlusDriver.Tests/OrderedReadResponseTests.cs @@ -0,0 +1,100 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class OrderedReadResponseTests + { + [Fact] + public void MixedValuesErrorsAndMissingItemsMatchDictionaryDecoder() + { + var data = Response(new[] + { + (4u, (PValue)new ValueWString("hello")), + (1u, (PValue)new ValueBool(true)), + (3u, (PValue)new ValueDInt(-123)), + (2u, (PValue)new ValueUDInt(uint.MaxValue)) + }, new[] { (3u, 0x8000UL), (5u, 0x9000UL) }); + using var referenceStream = new MemoryStream(data); + var reference = GetMultiVariablesResponse.DeserializeFromPdu(referenceStream)!; + using var stream = new MemoryStream(data); + var response = GetMultiVariablesReadResponse.ReadHeader(stream)!; + var values = new object[8]; + var errors = Enumerable.Repeat(ulong.MaxValue, 8).ToArray(); + + Assert.Equal(0, response.ReadItems(stream, values, errors, 1, 6)); + Assert.Equal(reference.SequenceNumber, response.SequenceNumber); + Assert.Equal(reference.IntegrityId, response.IntegrityId); + Assert.Equal(reference.ReturnValue, response.ReturnValue); + foreach (var item in reference.Values) + { + using var expected = new MemoryStream(); item.Value.Serialize(expected); + using var actual = new MemoryStream(); Assert.IsAssignableFrom(values[item.Key]).Serialize(actual); + Assert.Equal(expected.ToArray(), actual.ToArray()); + Assert.Equal(reference.ErrorValues.TryGetValue(item.Key, out var error) ? error : 0UL, errors[item.Key]); + } + Assert.Equal(0x9000UL, errors[5]); + Assert.Null(values[6]); Assert.Equal(ulong.MaxValue, errors[6]); + Assert.Null(values[0]); Assert.Null(values[7]); + Assert.Equal(ulong.MaxValue, errors[0]); Assert.Equal(ulong.MaxValue, errors[7]); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public void DuplicateAndOutOfRangeItemsAreRejected(bool errors, bool duplicate) + { + var valueItems = errors ? Array.Empty<(uint, PValue)>() + : duplicate ? new[] { (1u, (PValue)new ValueBool(true)), (1u, (PValue)new ValueBool(false)) } + : new[] { (2u, (PValue)new ValueBool(true)) }; + var errorItems = !errors ? Array.Empty<(uint, ulong)>() + : duplicate ? new[] { (1u, ulong.MaxValue), (1u, 1UL) } : new[] { (2u, 1UL) }; + using var stream = new MemoryStream(Response(valueItems, errorItems)); + var response = GetMultiVariablesReadResponse.ReadHeader(stream)!; + Assert.Equal(S7Consts.errIsoInvalidPDU, response.ReadItems(stream, + new object[1], new[] { ulong.MaxValue }, 0, 1)); + } + + [Fact] + public void EveryTruncatedPrefixFailsInsteadOfProducingACompleteRead() + { + var data = Response(new[] { (1u, (PValue)new ValueBool(true)) }, Array.Empty<(uint, ulong)>()); + for (var length = 0; length < data.Length; length++) + { + using var stream = new MemoryStream(data, 0, length); + var response = GetMultiVariablesReadResponse.ReadHeader(stream); + if (response == null) continue; + Assert.Throws(() => response.ReadItems(stream, + new object[1], new[] { ulong.MaxValue }, 0, 1)); + } + } + + [Fact] + public void UnexpectedFunctionIsRejectedBeforeDecodingItems() + { + var data = Response(Array.Empty<(uint, PValue)>(), Array.Empty<(uint, ulong)>()); + data[5] ^= 1; + using var stream = new MemoryStream(data); + Assert.Null(GetMultiVariablesReadResponse.ReadHeader(stream)); + } + + private static byte[] Response(IEnumerable<(uint Index, PValue Value)> values, IEnumerable<(uint Index, ulong Error)> errors) + { + using var stream = new MemoryStream(); + S7p.EncodeByte(stream, ProtocolVersion.V2); S7p.EncodeByte(stream, Opcode.Response); + S7p.EncodeUInt16(stream, 0); S7p.EncodeUInt16(stream, Functioncode.GetMultiVariables); + S7p.EncodeUInt16(stream, 0); S7p.EncodeUInt16(stream, 0x1234); S7p.EncodeByte(stream, 0x34); + S7p.EncodeUInt64Vlq(stream, 1); + foreach (var item in values) { S7p.EncodeUInt32Vlq(stream, item.Index); item.Value.Serialize(stream); } + S7p.EncodeUInt32Vlq(stream, 0); + foreach (var item in errors) { S7p.EncodeUInt32Vlq(stream, item.Index); S7p.EncodeUInt64Vlq(stream, item.Error); } + S7p.EncodeUInt32Vlq(stream, 0); S7p.EncodeUInt32Vlq(stream, 0x89abcdef); + return stream.ToArray(); + } + } +} diff --git a/src/S7CommPlusDriver.Tests/PlaintextBufferOwnershipTests.cs b/src/S7CommPlusDriver.Tests/PlaintextBufferOwnershipTests.cs new file mode 100644 index 0000000..6f9fc83 --- /dev/null +++ b/src/S7CommPlusDriver.Tests/PlaintextBufferOwnershipTests.cs @@ -0,0 +1,73 @@ +using System; +using System.IO; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class PlaintextBufferOwnershipTests + { + [Fact] + public void PublicCallbackReceivesAnOwnedExactLengthCopy() + { + using var client = new S7Client(); + byte[]? received = null; + client.OnDataReceived = (data, count) => { Assert.Equal(3, count); received = data; }; + var source = new byte[] { 1, 2, 3, 99, 99 }; + ((IS7TlsPlaintextCallback)client).OnPlaintextData(source, 3); + source[0] = 42; + Assert.Equal(new byte[] { 1, 2, 3 }, received); + Assert.NotSame(source, received); + } + + [Fact] + public void BorrowedProtocolCallbackConsumesOnlyTheActivePrefix() + { + using var client = new S7Client(); + byte[]? received = null; var countReceived = 0; + client.OnBorrowedDataReceived = (data, count) => { received = data; countReceived = count; }; + var source = new byte[] { 1, 2, 3, 99, 99 }; + ((IS7TlsPlaintextCallback)client).OnPlaintextData(source, 3); + Assert.Same(source, received); Assert.Equal(3, countReceived); + } + + [Fact] + public void ProtocolCopiesBorrowedPayloadBeforeTheBufferIsReused() + { + var session = new S7CommPlusProtocolSession(); + session.DebugResetReceiveDispatcherForTests(); + using var client = new S7Client(); + client.OnBorrowedDataReceived = (data, count) => session.DebugOnBorrowedDataReceivedForTests(data, count); + var source = new byte[] { 0x72, ProtocolVersion.V1, 0, 1, 0xAA, 0x72, ProtocolVersion.V1, 0, 0, 99 }; + ((IS7TlsPlaintextCallback)client).OnPlaintextData(source, 9); + Array.Clear(source, 0, source.Length); + Assert.Equal(0, session.DebugReceiveNextS7plusPduForTests(100, out MemoryStream pdu)); + Assert.Equal(new byte[] { ProtocolVersion.V1, 0xAA }, pdu.ToArray()); + } + + [Fact] + public void DeclaredLengthCannotReadInactiveBorrowedBufferTail() + { + var session = new S7CommPlusProtocolSession(); + session.DebugResetReceiveDispatcherForTests(); + var source = new byte[] { 0x72, ProtocolVersion.V1, 0, 2, 0xAA, 0xBB, 0x72, ProtocolVersion.V1, 0, 0 }; + session.DebugOnBorrowedDataReceivedForTests(source, 5); + Assert.Equal(S7Consts.errIsoInvalidPDU, session.DebugReceiveNextS7plusPduForTests(100, out var pdu)); + Assert.Null(pdu); + } + +#if NET6_0_OR_GREATER + [Fact] + public void BorrowedDeliveryDoesNotAllocateAnEightKilobyteScratchArray() + { + using var client = new S7Client(); + client.OnBorrowedDataReceived = (_, _) => { }; + var source = new byte[128]; + var sink = (IS7TlsPlaintextCallback)client; + sink.OnPlaintextData(source, source.Length); + var before = GC.GetAllocatedBytesForCurrentThread(); + for (var i = 0; i < 1000; i++) sink.OnPlaintextData(source, source.Length); + Assert.Equal(0, GC.GetAllocatedBytesForCurrentThread() - before); + } +#endif + } +} diff --git a/src/S7CommPlusDriver.Tests/ReadRequestLengthTests.cs b/src/S7CommPlusDriver.Tests/ReadRequestLengthTests.cs new file mode 100644 index 0000000..c7e3251 --- /dev/null +++ b/src/S7CommPlusDriver.Tests/ReadRequestLengthTests.cs @@ -0,0 +1,60 @@ +using System; +using System.IO; +using S7CommPlusDriver.ClientApi; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class ReadRequestLengthTests + { + [Theory] + [InlineData(0u)] + [InlineData(127u)] + [InlineData(128u)] + [InlineData(16383u)] + [InlineData(16384u)] + [InlineData(2097152u)] + [InlineData(268435456u)] + [InlineData(uint.MaxValue)] + public void ComputedLengthMatchesWireBytesAcrossVlqBoundariesAndMutations(uint boundary) + { + var request = new GetMultiVariablesRequest(ProtocolVersion.V3) + { IntegrityId = boundary, SessionId = uint.MaxValue, SequenceNumber = ushort.MaxValue, LinkId = 123 }; + var address = new ItemAddress { AccessArea = boundary, AccessSubArea = boundary, SymbolCrc = boundary }; + address.LID.Add(boundary); + request.AddressList.Add(address); + AssertLength(request); + // Exercise the mutable compatibility API rather than relying on cached sizes. + address.SymbolCrc = boundary ^ uint.MaxValue; + address.LID.Add(uint.MaxValue); + AssertLength(request); + for (var i = 0; i < 128; i++) request.AddressList.Add(address); + AssertLength(request); + request.WithIntegrityId = false; + AssertLength(request); + } + + [Fact] + public void LongAddressAndWideFieldCountMatchActualSerialization() + { + var request = new GetMultiVariablesRequest(ProtocolVersion.V2) { IntegrityId = uint.MaxValue }; + var address = new ItemAddress { AccessArea = uint.MaxValue }; + for (var i = 0; i < 16384; i++) address.LID.Add((uint)i); + request.AddressList.Add(address); + AssertLength(request); + } + + private static void AssertLength(GetMultiVariablesRequest request) + { + using var stream = new MemoryStream(); + request.Serialize(stream); + Assert.Equal(stream.Length, request.GetSerializedLength()); + var integrity = request.IntegrityId; + request.IntegrityId = uint.MaxValue; + using var widest = new MemoryStream(); + request.Serialize(widest); + request.IntegrityId = integrity; + Assert.Equal(widest.Length, request.GetSerializedLength(uint.MaxValue)); + } + } +} diff --git a/src/S7CommPlusDriver.Tests/ReceiveQueueTests.cs b/src/S7CommPlusDriver.Tests/ReceiveQueueTests.cs new file mode 100644 index 0000000..85b1c55 --- /dev/null +++ b/src/S7CommPlusDriver.Tests/ReceiveQueueTests.cs @@ -0,0 +1,69 @@ +using System; +using System.Threading; +using S7CommPlusDriver.Internal; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class ReceiveQueueTests + { + [Fact] + public void CompletionDrainsQueuedItemsAndRejectsNewWrites() + { + var queue = new S7CommPlusReceiveQueue(); + Assert.True(queue.TryWrite(1)); Assert.True(queue.TryWrite(2)); + queue.Complete(); queue.Complete(); + Assert.False(queue.TryWrite(3)); + Assert.Equal(0, queue.Read(100, out var first)); Assert.Equal(1, first); + Assert.Equal(0, queue.Read(100, out var second)); Assert.Equal(2, second); + Assert.Equal(S7Consts.errTCPNotConnected, queue.Read(100, out _)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void WaitingReaderIsWokenByDataOrCompletion(bool complete) + { + var queue = new S7CommPlusReceiveQueue(); + using var started = new ManualResetEventSlim(); + using var done = new ManualResetEventSlim(); + var result = -1; var value = -1; + var reader = new Thread(() => + { + started.Set(); result = queue.Read(5000, out value); done.Set(); + }) { IsBackground = true }; + reader.Start(); Assert.True(started.Wait(2000)); + if (complete) queue.Complete(); else Assert.True(queue.TryWrite(42)); + Assert.True(done.Wait(2000)); Assert.True(reader.Join(2000)); + Assert.Equal(complete ? S7Consts.errTCPNotConnected : 0, result); + Assert.Equal(complete ? 0 : 42, value); + } + + [Fact] + public void EmptyQueueTimesOutWithoutCompletingTheQueue() + { + var queue = new S7CommPlusReceiveQueue(); + Assert.Equal(S7Consts.errCliJobTimeout, queue.Read(1, out _)); + Assert.True(queue.TryWrite(42)); + Assert.Equal(0, queue.Read(100, out var value)); Assert.Equal(42, value); + } + + [Fact] + public void ConcurrentProducerPreservesEveryItemInOrder() + { + var queue = new S7CommPlusReceiveQueue(); + var producer = new Thread(() => + { + for (var i = 0; i < 10000; i++) queue.TryWrite(i); + queue.Complete(); + }) { IsBackground = true }; + producer.Start(); + for (var i = 0; i < 10000; i++) + { + Assert.Equal(0, queue.Read(2000, out var value)); Assert.Equal(i, value); + } + Assert.True(producer.Join(2000)); + Assert.Equal(S7Consts.errTCPNotConnected, queue.Read(100, out _)); + } + } +} diff --git a/src/S7CommPlusDriver.Tests/S7ClientTransportTests.cs b/src/S7CommPlusDriver.Tests/S7ClientTransportTests.cs index e54912c..3cfd5da 100644 --- a/src/S7CommPlusDriver.Tests/S7ClientTransportTests.cs +++ b/src/S7CommPlusDriver.Tests/S7ClientTransportTests.cs @@ -135,5 +135,19 @@ public void SendFragmentsPayloadAtNegotiatedCotpBoundary(bool tlsRecord) Assert.Equal(0x80, transport.Sent[1][6]); Assert.Equal(payload, transport.Sent.SelectMany(packet => packet.Skip(7)).ToArray()); } + + [Fact] + public void BorrowedEncryptedSliceUsesItsOffsetCountAndExactWireLength() + { + var transport = new FakeS7Transport { Connected = true }; + using var client = new S7Client(() => transport); + var payload = Enumerable.Range(0, 3000).Select(value => (byte)value).ToArray(); + ((IS7TlsRecordCallback)client).WriteEncryptedData(payload, 17, 2000); + Assert.Equal(2000, transport.Sent.Sum(packet => packet.Length - 7)); + Assert.All(transport.Sent, packet => Assert.Equal(packet.Length, (packet[2] << 8) | packet[3])); + Assert.All(transport.Sent.Take(transport.Sent.Count - 1), packet => Assert.Equal(0, packet[6])); + Assert.Equal(0x80, transport.Sent.Last()[6]); + Assert.Equal(payload.Skip(17).Take(2000), transport.Sent.SelectMany(packet => packet.Skip(7))); + } } } diff --git a/src/S7CommPlusDriver.Tests/S7CommPlusProtocolSessionReceiveTests.cs b/src/S7CommPlusDriver.Tests/S7CommPlusProtocolSessionReceiveTests.cs index b4d3e40..14b623b 100644 --- a/src/S7CommPlusDriver.Tests/S7CommPlusProtocolSessionReceiveTests.cs +++ b/src/S7CommPlusDriver.Tests/S7CommPlusProtocolSessionReceiveTests.cs @@ -73,6 +73,24 @@ public void ReceiveTimeoutReturnsJobTimeout() Assert.Null(pdu); } +#if NET6_0_OR_GREATER + [Fact] + public void AlreadyQueuedResponsesDoNotAllocateTimeoutOrTaskObjects() + { + var connection = new S7CommPlusProtocolSession(); + connection.DebugResetReceiveDispatcherForTests(); + var packet = new byte[] { 0x72, ProtocolVersion.V1, 0x00, 0x01, 0xAA, 0x72, ProtocolVersion.V1, 0x00, 0x00 }; + connection.DebugOnDataReceivedForTests(packet); + Assert.Equal(0, connection.DebugReceiveNextS7plusPduForTests(100, out _)); + for (var i = 0; i < 100; i++) connection.DebugOnDataReceivedForTests(packet); + var before = GC.GetAllocatedBytesForCurrentThread(); + for (var i = 0; i < 100; i++) + if (connection.DebugReceiveNextS7plusPduForTests(100, out _) != 0) + throw new InvalidOperationException("Queued response was lost."); + Assert.Equal(0, GC.GetAllocatedBytesForCurrentThread() - before); + } +#endif + [Fact] public void LegacyDigestMismatchPublishesReceiveError() { diff --git a/src/S7CommPlusDriver.Tests/ScalarBooleanReadTests.cs b/src/S7CommPlusDriver.Tests/ScalarBooleanReadTests.cs new file mode 100644 index 0000000..e65709d --- /dev/null +++ b/src/S7CommPlusDriver.Tests/ScalarBooleanReadTests.cs @@ -0,0 +1,35 @@ +using System; +using System.IO; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class ScalarBooleanReadTests + { + [Theory] + [InlineData(0, 0)] + [InlineData(0, 1)] + [InlineData(1, 1)] + [InlineData(2, 0)] + public void BooleanReadPreservesValueAndFlags(int flags, int value) + { + using var stream = new MemoryStream(new[] { (byte)flags, (byte)Datatype.Bool, (byte)value }); + var result = Assert.IsType(PValue.Deserialize(stream)); + Assert.Equal(value != 0, result.GetValue()); + using var encoded = new MemoryStream(); result.Serialize(encoded); + Assert.Equal(new[] { (byte)flags, (byte)Datatype.Bool, (byte)value }, encoded.ToArray()); + } + +#if NET6_0_OR_GREATER + [Fact] + public void OrdinaryBooleanDecodingDoesNotAllocatePerSignal() + { + using var stream = new MemoryStream(new[] { (byte)0, (byte)Datatype.Bool, (byte)1 }); + _ = PValue.Deserialize(stream); + var before = GC.GetAllocatedBytesForCurrentThread(); + for (var i = 0; i < 1000; i++) { stream.Position = 0; _ = PValue.Deserialize(stream); } + Assert.Equal(0, GC.GetAllocatedBytesForCurrentThread() - before); + } +#endif + } +} diff --git a/src/S7CommPlusDriver.Tests/SpanCodecTests.cs b/src/S7CommPlusDriver.Tests/SpanCodecTests.cs new file mode 100644 index 0000000..89acf7e --- /dev/null +++ b/src/S7CommPlusDriver.Tests/SpanCodecTests.cs @@ -0,0 +1,197 @@ +using System; +using System.IO; +using System.Linq; +using System.Runtime.InteropServices; +using System.Text; +using Xunit; + +namespace S7CommPlusDriver.Tests +{ + public sealed class SpanCodecTests + { + [StructLayout(LayoutKind.Explicit)] + private struct SingleBits + { + [FieldOffset(0)] internal uint Bits; + [FieldOffset(0)] internal float Value; + } + [StructLayout(LayoutKind.Explicit)] + private struct DoubleBits + { + [FieldOffset(0)] internal ulong Bits; + [FieldOffset(0)] internal double Value; + } + + [Theory] + [InlineData("0000000000000000")] + [InlineData("0102030405060708")] + [InlineData("8000000000000000")] + [InlineData("FFFFFFFFFFFFFFFF")] + public void UInt64AndInt64PreserveBigEndianWireBits(string hex) + { + var bytes = Hex(hex); + ulong expected = 0; + foreach (var b in bytes) expected = (expected << 8) | b; + using var stream = new MemoryStream(bytes); + Assert.Equal(8, S7p.DecodeUInt64(stream, out var unsigned)); Assert.Equal(expected, unsigned); + stream.Position = 0; + Assert.Equal(8, S7p.DecodeInt64(stream, out var signed)); Assert.Equal(unchecked((long)expected), signed); + } + + [Theory] + [InlineData("00000000")] + [InlineData("80000000")] + [InlineData("3F800000")] + [InlineData("7F800000")] + [InlineData("FF800000")] + [InlineData("7FC01234")] + [InlineData("00000001")] + public void FloatPreservesZeroInfinityNanAndSubnormalBits(string hex) + { + var bytes = Hex(hex); + using var input = new MemoryStream(bytes); + Assert.Equal(4, S7p.DecodeFloat(input, out var value)); + var bits = new SingleBits { Value = value }.Bits; + Assert.Equal(Convert.ToUInt32(hex, 16), bits); + using var output = new MemoryStream(); + Assert.Equal(4, S7p.EncodeFloat(output, value)); Assert.Equal(bytes, output.ToArray()); + } + + [Theory] + [InlineData("0000000000000000")] + [InlineData("8000000000000000")] + [InlineData("3FF0000000000000")] + [InlineData("7FF0000000000000")] + [InlineData("FFF0000000000000")] + [InlineData("7FF8123456789ABC")] + [InlineData("0000000000000001")] + public void DoublePreservesZeroInfinityNanAndSubnormalBits(string hex) + { + var bytes = Hex(hex); + using var input = new MemoryStream(bytes); + Assert.Equal(8, S7p.DecodeDouble(input, out var value)); + Assert.Equal(Convert.ToUInt64(hex, 16), new DoubleBits { Value = value }.Bits); + using var output = new MemoryStream(); + Assert.Equal(8, S7p.EncodeDouble(output, value)); Assert.Equal(bytes, output.ToArray()); + } + + [Theory] + [InlineData(0, "00")] + [InlineData(63, "3F")] + [InlineData(64, "8040")] + [InlineData(-64, "FF40")] + [InlineData(-255, "FE01")] + [InlineData(int.MaxValue, "87FFFFFF7F")] + [InlineData(int.MinValue, "F880808000")] + public void SignedIntVlqPreservesGoldenWireEncoding(int value, string hex) + { + using var stream = new MemoryStream(); + S7p.EncodeInt32Vlq(stream, value); Assert.Equal(Hex(hex), stream.ToArray()); + stream.Position = 0; S7p.DecodeInt32Vlq(stream, out var decoded); Assert.Equal(value, decoded); + } + + [Theory] + [InlineData(0UL, "00")] + [InlineData(127UL, "7F")] + [InlineData(128UL, "8100")] + [InlineData(0x00FFFFFFFFFFFFFFUL, "FFFFFFFFFFFFFF7F")] + [InlineData(0x0100000000000000UL, "80C080808080808000")] + [InlineData(ulong.MaxValue, "FFFFFFFFFFFFFFFFFF")] + public void UInt64VlqPreservesTheNinthByteBoundary(ulong value, string hex) + { + using var stream = new MemoryStream(); + S7p.EncodeUInt64Vlq(stream, value); Assert.Equal(Hex(hex), stream.ToArray()); + stream.Position = 0; S7p.DecodeUInt64Vlq(stream, out var decoded); Assert.Equal(value, decoded); + } + + [Theory] + [InlineData(0L)] + [InlineData(63L)] + [InlineData(-64L)] + [InlineData(0x007FFFFFFFFFFFFFL)] + [InlineData(0x0080000000000000L)] + [InlineData(-0x0080000000000000L)] + [InlineData(long.MinValue)] + [InlineData(long.MaxValue)] + public void SignedLongVlqPreservesExtremeValues(long value) + { + using var stream = new MemoryStream(); + S7p.EncodeInt64Vlq(stream, value); + stream.Position = 0; S7p.DecodeInt64Vlq(stream, out var decoded); Assert.Equal(value, decoded); + } + + [Fact] + public void Utf8DecodeHonorsNonzeroMemoryStreamOriginAndPosition() + { + const string text = "Grüße 日本語 😀"; + var utf8 = Encoding.UTF8.GetBytes(text); + var bytes = new byte[utf8.Length + 10]; Array.Copy(utf8, 0, bytes, 5, utf8.Length); + using var stream = new MemoryStream(bytes, 3, utf8.Length + 4, false, true); + stream.Position = 2; + Assert.Equal(utf8.Length, S7p.DecodeWString(stream, utf8.Length, out var decoded)); + Assert.Equal(text, decoded); Assert.Equal(2 + utf8.Length, stream.Position); + } + + [Fact] + public void FixedWidthSpanViewsHonorTheMemoryStreamOriginAndActiveRange() + { + var data = Hex("FFEE0102030405060708AABB"); + using var stream = new MemoryStream(data, 1, 10, false, true); + stream.Position = 1; + Assert.Equal(8, S7p.DecodeUInt64(stream, out var value)); + Assert.Equal(0x0102030405060708UL, value); Assert.Equal(9, stream.Position); + Assert.Equal(0, S7p.DecodeUInt64(stream, out _)); Assert.Equal(9, stream.Position); + stream.Position = 1; + Assert.Equal(4, S7p.DecodeFloat(stream, out var single)); + Assert.Equal(0x01020304u, new SingleBits { Value = single }.Bits); Assert.Equal(5, stream.Position); + } + + [Theory] + [InlineData(0)] + [InlineData(10)] + [InlineData(5000)] + public void Utf8EncodingMatchesStandardEncodingForStackAndPoolPaths(int length) + { + var text = new string('ü', length) + "😀"; + using var stream = new MemoryStream(); + Assert.Equal(Encoding.UTF8.GetByteCount(text), S7p.EncodeWString(stream, text)); + Assert.Equal(Encoding.UTF8.GetBytes(text), stream.ToArray()); + } + + [Fact] + public void FixedWidthTruncationCannotProduceNumericValues() + { + using var stream = new MemoryStream(new byte[3]); + Assert.Equal(0, S7p.DecodeFloat(stream, out var single)); Assert.Equal(0, single); + Assert.Equal(0, S7p.DecodeDouble(stream, out var wide)); Assert.Equal(0, wide); + Assert.Equal(0, S7p.DecodeUInt64(stream, out _)); Assert.Equal(0, S7p.DecodeInt64(stream, out _)); + Assert.Equal(0, stream.Position); + } + +#if NET6_0_OR_GREATER + [Fact] + public void NumericCodecsDoNotAllocateTemporaryArrays() + { + using var stream = new MemoryStream(128); + void Run() + { + stream.SetLength(0); + S7p.EncodeFloat(stream, 1.25f); S7p.EncodeDouble(stream, -2.5); + S7p.EncodeUInt64(stream, ulong.MaxValue); S7p.EncodeInt64(stream, long.MinValue); + S7p.EncodeInt32Vlq(stream, int.MinValue); S7p.EncodeUInt64Vlq(stream, ulong.MaxValue); + S7p.EncodeInt64Vlq(stream, long.MinValue); + stream.Position = 0; + S7p.DecodeFloat(stream, out _); S7p.DecodeDouble(stream, out _); + S7p.DecodeUInt64(stream, out _); S7p.DecodeInt64(stream, out _); + } + Run(); + var before = GC.GetAllocatedBytesForCurrentThread(); + for (var i = 0; i < 1000; i++) Run(); + Assert.Equal(0, GC.GetAllocatedBytesForCurrentThread() - before); + } +#endif + + private static byte[] Hex(string value) => Enumerable.Range(0, value.Length / 2) + .Select(i => Convert.ToByte(value.Substring(i * 2, 2), 16)).ToArray(); + } +} diff --git a/src/S7CommPlusDriver/ClientApi/ItemAddress.cs b/src/S7CommPlusDriver/ClientApi/ItemAddress.cs index 9b70d3c..ac6a5b6 100644 --- a/src/S7CommPlusDriver/ClientApi/ItemAddress.cs +++ b/src/S7CommPlusDriver/ClientApi/ItemAddress.cs @@ -146,6 +146,17 @@ public void SetAccessAreaToDatablock(UInt32 number) AccessArea = (UInt16)number + 0x8a0e0000; } + internal long GetSerializedLength() + { + long length = S7p.GetUInt32VlqLength(SymbolCrc) + + S7p.GetUInt32VlqLength(AccessArea) + + S7p.GetUInt32VlqLength((uint)LocalIdCount + 1) + + S7p.GetUInt32VlqLength(AccessSubArea); + for (var index = 0; index < LocalIdCount; index++) + length += S7p.GetUInt32VlqLength(GetLocalId(index)); + return length; + } + public int Serialize(Stream buffer) { int ret = 0; diff --git a/src/S7CommPlusDriver/Core/CreateObjectResponse.cs b/src/S7CommPlusDriver/Core/CreateObjectResponse.cs index cac7c89..2b2398e 100644 --- a/src/S7CommPlusDriver/Core/CreateObjectResponse.cs +++ b/src/S7CommPlusDriver/Core/CreateObjectResponse.cs @@ -57,7 +57,11 @@ public int Deserialize(Stream buffer) ret += S7p.DecodeUInt32Vlq(buffer, out object_id); ObjectIds.Add(object_id); } - ret += S7p.DecodeObject(buffer, ref ResponseObject); + // Bit 1 declares the optional response object. Without it, the next + // bytes are an integrity counter, whose VLQ prefix can resemble an + // object element (for example 0xAB) after many read requests. + if ((TransportFlags & 0x02) != 0) + ret += S7p.DecodeObject(buffer, ref ResponseObject); return ret; } @@ -75,7 +79,7 @@ public override string ToString() { s += "" + id.ToString() + "" + Environment.NewLine; } - s += "" + ResponseObject.ToString() + "" + Environment.NewLine; + s += "" + ResponseObject?.ToString() + "" + Environment.NewLine; s += "" + Environment.NewLine; s += "" + Environment.NewLine; return s; diff --git a/src/S7CommPlusDriver/Core/GetMultiVariablesReadResponse.cs b/src/S7CommPlusDriver/Core/GetMultiVariablesReadResponse.cs new file mode 100644 index 0000000..2e11cea --- /dev/null +++ b/src/S7CommPlusDriver/Core/GetMultiVariablesReadResponse.cs @@ -0,0 +1,69 @@ +using System; +using System.Collections.Generic; +using System.IO; + +namespace S7CommPlusDriver +{ + /// Decodes a bulk read into ordered staging buffers without per-response dictionaries. + internal sealed class GetMultiVariablesReadResponse : IS7pResponse + { + public byte TransportFlags; + public ulong ReturnValue; + public byte ProtocolVersion { get; set; } + public ushort FunctionCode => Functioncode.GetMultiVariables; + public ushort SequenceNumber { get; set; } + public uint IntegrityId { get; set; } + public bool WithIntegrityId { get; set; } = true; + + internal static GetMultiVariablesReadResponse ReadHeader(Stream pdu) + { + if (pdu.Length - pdu.Position < 11) return null; + S7p.DecodeByte(pdu, out var version); + S7p.DecodeByte(pdu, out var opcode); + S7p.DecodeUInt16(pdu, out _); + S7p.DecodeUInt16(pdu, out var function); + S7p.DecodeUInt16(pdu, out _); + S7p.DecodeUInt16(pdu, out var sequence); + S7p.DecodeByte(pdu, out var flags); + if (opcode != Opcode.Response || function != Functioncode.GetMultiVariables) return null; + return new GetMultiVariablesReadResponse + { + ProtocolVersion = version, SequenceNumber = sequence, TransportFlags = flags + }; + } + + // Each request owns a disjoint, initially empty range. The owner publishes + // tag values only after the complete logical read succeeds. + internal int ReadItems(Stream pdu, IList values, IList errors, int start, int count) + { + if (start < 0 || count < 0 || start > values.Count - count || start > errors.Count - count) + throw new ArgumentOutOfRangeException(nameof(count)); + S7p.DecodeUInt64Vlq(pdu, out ReturnValue); + S7p.DecodeUInt32Vlq(pdu, out var item); + while (item != 0) + { + if (item > count) return S7Consts.errIsoInvalidPDU; + var index = start + (int)item - 1; + if (values[index] != null) return S7Consts.errIsoInvalidPDU; + values[index] = PValue.Deserialize(pdu); + errors[index] = 0; + S7p.DecodeUInt32Vlq(pdu, out item); + } + + S7p.DecodeUInt32Vlq(pdu, out item); + HashSet seenErrors = null; + while (item != 0) + { + if (item > count) return S7Consts.errIsoInvalidPDU; + if (seenErrors == null) seenErrors = new HashSet(); + if (!seenErrors.Add(item)) return S7Consts.errIsoInvalidPDU; + S7p.DecodeUInt64Vlq(pdu, out var error); + errors[start + (int)item - 1] = error; + S7p.DecodeUInt32Vlq(pdu, out item); + } + S7p.DecodeUInt32Vlq(pdu, out var integrity); + IntegrityId = integrity; + return 0; + } + } +} diff --git a/src/S7CommPlusDriver/Core/GetMultiVariablesRequest.cs b/src/S7CommPlusDriver/Core/GetMultiVariablesRequest.cs index 59f4a56..d749d27 100644 --- a/src/S7CommPlusDriver/Core/GetMultiVariablesRequest.cs +++ b/src/S7CommPlusDriver/Core/GetMultiVariablesRequest.cs @@ -25,6 +25,7 @@ class GetMultiVariablesRequest : IS7pRequest byte TransportFlags = S7CommPlusProtocolConstants.RequestWithResponseTransportFlags; public UInt32 LinkId = 0; // for reading variables, this should be 0 public List AddressList = new List(); + private static readonly int ObjectQualifierLength = S7p.EncodeObjectQualifier(Stream.Null); public uint SessionId { get; set; } public byte ProtocolVersion { get; set; } @@ -44,6 +45,22 @@ public byte GetProtocolVersion() return ProtocolVersion; } + // Calculate from current fields; ItemAddress remains mutable, so do not cache + // address bytes or lengths across reads. Batching reserves the widest integrity ID. + internal long GetSerializedLength(uint? integrityId = null) + { + uint fieldCount = 0; + long addressLength = 0; + foreach (var address in AddressList) + { + fieldCount += address.GetNumberOfFields(); + addressLength += address.GetSerializedLength(); + } + return 18L + S7p.GetUInt32VlqLength((uint)AddressList.Count) + + S7p.GetUInt32VlqLength(fieldCount) + addressLength + ObjectQualifierLength + + (WithIntegrityId ? S7p.GetUInt32VlqLength(integrityId ?? IntegrityId) : 0) + 4; + } + public int Serialize(Stream buffer) { int ret = 0; diff --git a/src/S7CommPlusDriver/Core/GetMultiVariablesResponse.cs b/src/S7CommPlusDriver/Core/GetMultiVariablesResponse.cs index 6e9bab6..d558a59 100644 --- a/src/S7CommPlusDriver/Core/GetMultiVariablesResponse.cs +++ b/src/S7CommPlusDriver/Core/GetMultiVariablesResponse.cs @@ -32,11 +32,13 @@ internal class GetMultiVariablesResponse : IS7pResponse public uint IntegrityId { get; set; } public bool WithIntegrityId { get; set; } - public GetMultiVariablesResponse(byte protocolVersion) + public GetMultiVariablesResponse(byte protocolVersion, int expectedValueCount = 0) { ProtocolVersion = protocolVersion; ErrorValues = new Dictionary(); - Values = new Dictionary(); + // The originating request supplies the capacity; avoid repeated dictionary + // growth for every batch without trusting a count from the wire. + Values = new Dictionary(expectedValueCount); WithIntegrityId = true; } @@ -111,7 +113,7 @@ public override string ToString() return s; } - public static GetMultiVariablesResponse DeserializeFromPdu(Stream pdu) + public static GetMultiVariablesResponse DeserializeFromPdu(Stream pdu, int expectedValueCount = 0) { byte protocolVersion; byte opcode; @@ -131,7 +133,7 @@ public static GetMultiVariablesResponse DeserializeFromPdu(Stream pdu) { return null; } - GetMultiVariablesResponse resp = new GetMultiVariablesResponse(protocolVersion); + GetMultiVariablesResponse resp = new GetMultiVariablesResponse(protocolVersion, expectedValueCount); resp.Deserialize(pdu); return resp; diff --git a/src/S7CommPlusDriver/Core/PValue.cs b/src/S7CommPlusDriver/Core/PValue.cs index 84aacac..e299921 100644 --- a/src/S7CommPlusDriver/Core/PValue.cs +++ b/src/S7CommPlusDriver/Core/PValue.cs @@ -307,9 +307,11 @@ internal static ValueNullArray Deserialize(Stream buffer, byte flags, bool disab } } - internal class ValueBool : PValue + internal sealed class ValueBool : PValue { - bool Value; + private static readonly ValueBool False = new ValueBool(false); + private static readonly ValueBool True = new ValueBool(true); + readonly bool Value; public ValueBool(bool value) : this(value, 0) { @@ -344,7 +346,9 @@ public static ValueBool Deserialize(Stream buffer, byte flags) { byte value; S7p.DecodeByte(buffer, out value); - return new ValueBool(Convert.ToBoolean(value), flags); + // Ordinary scalar booleans are immutable. Reuse their two representations + // without sharing array values or values carrying other wire flags. + return flags == 0 ? (value == 0 ? False : True) : new ValueBool(Convert.ToBoolean(value), flags); } } diff --git a/src/S7CommPlusDriver/Core/S7p.cs b/src/S7CommPlusDriver/Core/S7p.cs index b668b13..5876774 100644 --- a/src/S7CommPlusDriver/Core/S7p.cs +++ b/src/S7CommPlusDriver/Core/S7p.cs @@ -14,6 +14,9 @@ #endregion using System; +using System.Buffers; +using System.Buffers.Binary; +using System.Runtime.InteropServices; using System.Text; using System.Collections.Generic; using S7CommPlusDriver.Internal; @@ -22,6 +25,55 @@ namespace S7CommPlusDriver { internal class S7p { + [StructLayout(LayoutKind.Explicit)] + private struct SingleBits + { + [FieldOffset(0)] internal uint Bits; + [FieldOffset(0)] internal float Value; + } + + [StructLayout(LayoutKind.Explicit)] + private struct DoubleBits + { + [FieldOffset(0)] internal ulong Bits; + [FieldOffset(0)] internal double Value; + } + + private static bool TryReadMemoryBytes(System.IO.Stream buffer, int length, out ReadOnlySpan bytes) + { + if (buffer is System.IO.MemoryStream memory && memory.TryGetBuffer(out var segment) + && memory.Length - memory.Position >= length) + { + var position = checked((int)memory.Position); + bytes = segment.Array.AsSpan(segment.Offset + position, length); + memory.Position = position + length; + return true; + } + bytes = default; + return false; + } + + private static bool TryReadExactly(System.IO.Stream buffer, Span value) + { +#if NET6_0_OR_GREATER + var totalRead = 0; + while (totalRead < value.Length) + { + var read = buffer.Read(value.Slice(totalRead)); + if (read == 0) return false; + totalRead += read; + } +#else + for (var i = 0; i < value.Length; i++) + { + var read = buffer.ReadByte(); + if (read < 0) return false; + value[i] = (byte)read; + } +#endif + return true; + } + private static bool TryReadExactly(System.IO.Stream buffer, byte[] value, int length) { var totalRead = 0; @@ -194,38 +246,40 @@ public static int DecodeInt32(System.IO.Stream buffer, out Int32 value) public static int DecodeUInt64(System.IO.Stream buffer, out UInt64 value) { - if (buffer.Length - buffer.Position < 8) + if (TryReadMemoryBytes(buffer, 8, out var view)) { - value = 0; - return 0; + value = BinaryPrimitives.ReadUInt64BigEndian(view); + return 8; } - byte[] b = new byte[8]; - if (!TryReadExactly(buffer, b, 8)) + if (buffer.Length - buffer.Position < 8) { value = 0; return 0; } - Array.Reverse(b, 0, 8); - value = BitConverter.ToUInt64(b, 0); + var b0 = buffer.ReadByte(); var b1 = buffer.ReadByte(); + var b2 = buffer.ReadByte(); var b3 = buffer.ReadByte(); + var b4 = buffer.ReadByte(); var b5 = buffer.ReadByte(); + var b6 = buffer.ReadByte(); var b7 = buffer.ReadByte(); + if ((b0 | b1 | b2 | b3 | b4 | b5 | b6 | b7) < 0) { value = 0; return 0; } + value = ((ulong)b0 << 56) | ((ulong)b1 << 48) | ((ulong)b2 << 40) | ((ulong)b3 << 32) + | ((ulong)b4 << 24) | ((ulong)b5 << 16) | ((ulong)b6 << 8) | (uint)b7; return 8; } public static int DecodeInt64(System.IO.Stream buffer, out Int64 value) { - if (buffer.Length - buffer.Position < 8) - { - value = 0; - return 0; - } - byte[] b = new byte[8]; - if (!TryReadExactly(buffer, b, 8)) - { - value = 0; - return 0; - } - Array.Reverse(b, 0, 8); - value = BitConverter.ToInt64(b, 0); - return 8; + var count = DecodeUInt64(buffer, out var bits); + value = unchecked((long)bits); + return count; + } + + internal static int GetUInt32VlqLength(uint value) + { + if (value < 0x80) return 1; + if (value < 0x4000) return 2; + if (value < 0x200000) return 3; + if (value < 0x10000000) return 4; + return 5; } public static int EncodeUInt32Vlq(System.IO.Stream buffer, UInt32 value) @@ -339,7 +393,7 @@ public static int EncodeInt32Vlq(System.IO.Stream buffer, Int32 value) // The actual method writes the values in the 2nd compact variant. // The decode algorithms can handle both variants. - byte[] b = new byte[5]; + Span b = stackalloc byte[5]; UInt32 abs_v; if (value == Int32.MinValue) // Handle this, otherwise Math.Abs() will fail { @@ -370,7 +424,7 @@ public static int EncodeInt32Vlq(System.IO.Stream buffer, Int32 value) // Reverse order of bytes for (int i = length - 1; i >= 0; i--) { - buffer.Write(b, i, 1); + buffer.WriteByte(b[i]); } return length; } @@ -383,7 +437,7 @@ public static int EncodeUInt64Vlq(System.IO.Stream buffer, UInt64 value) // Every byte looses 1 bit for the "continue" flag. For 8 bytes loosing 1 bit, we need 1 more byte. // Which in the normal variant would have only 7 bit of space, because 1 bit is for the "continue" flag. // The special handling allows to use all 8 bits in the additional 9th byte. - byte[] b = new byte[9]; + Span b = stackalloc byte[9]; bool special = value > 0x00ffffffffffffff; if (special) @@ -435,7 +489,7 @@ public static int EncodeUInt64Vlq(System.IO.Stream buffer, UInt64 value) // Reverse order of bytes for (int i = length - 1; i >= 0; i--) { - buffer.Write(b, i, 1); + buffer.WriteByte(b[i]); } return length; } @@ -533,7 +587,7 @@ public static int DecodeInt64Vlq(System.IO.Stream buffer, out Int64 value) public static int EncodeInt64Vlq(System.IO.Stream buffer, Int64 value) { - byte[] b = new byte[9]; + Span b = stackalloc byte[9]; UInt64 abs_v; if (value == Int64.MinValue) // Handle this, otherwise Math.Abs() will fail { @@ -597,66 +651,69 @@ public static int EncodeInt64Vlq(System.IO.Stream buffer, Int64 value) // Reverse order of bytes for (int i = length - 1; i >= 0; i--) { - buffer.Write(b, i, 1); + buffer.WriteByte(b[i]); } return length; } public static int EncodeFloat(System.IO.Stream buffer, float value) { - byte[] v = BitConverter.GetBytes(value); - buffer.WriteByte(v[3]); - buffer.WriteByte(v[2]); - buffer.WriteByte(v[1]); - buffer.WriteByte(v[0]); - return 4; + return EncodeUInt32(buffer, new SingleBits { Value = value }.Bits); } public static int DecodeFloat(System.IO.Stream buffer, out float value) { - byte[] v = new byte[4]; - v[3] = (byte)buffer.ReadByte(); - v[2] = (byte)buffer.ReadByte(); - v[1] = (byte)buffer.ReadByte(); - v[0] = (byte)buffer.ReadByte(); - value = BitConverter.ToSingle(v, 0); + value = 0; + if (buffer.CanSeek) + { + var read = DecodeUInt32(buffer, out var bits); + value = new SingleBits { Bits = bits }.Value; + return read; + } + Span bytes = stackalloc byte[4]; + if (!TryReadExactly(buffer, bytes)) return 0; + value = new SingleBits { Bits = BinaryPrimitives.ReadUInt32BigEndian(bytes) }.Value; return 4; } public static int EncodeDouble(System.IO.Stream buffer, double value) { - byte[] v = BitConverter.GetBytes(value); - buffer.WriteByte(v[7]); - buffer.WriteByte(v[6]); - buffer.WriteByte(v[5]); - buffer.WriteByte(v[4]); - buffer.WriteByte(v[3]); - buffer.WriteByte(v[2]); - buffer.WriteByte(v[1]); - buffer.WriteByte(v[0]); - return 8; + return EncodeUInt64(buffer, new DoubleBits { Value = value }.Bits); } public static int DecodeDouble(System.IO.Stream buffer, out double value) { - byte[] v = new byte[8]; - v[7] = (byte)buffer.ReadByte(); - v[6] = (byte)buffer.ReadByte(); - v[5] = (byte)buffer.ReadByte(); - v[4] = (byte)buffer.ReadByte(); - v[3] = (byte)buffer.ReadByte(); - v[2] = (byte)buffer.ReadByte(); - v[1] = (byte)buffer.ReadByte(); - v[0] = (byte)buffer.ReadByte(); - value = BitConverter.ToDouble(v, 0); + value = 0; + if (buffer.CanSeek && buffer.Length - buffer.Position < 8) return 0; + Span bytes = stackalloc byte[8]; + if (!TryReadExactly(buffer, bytes)) return 0; + value = new DoubleBits { Bits = BinaryPrimitives.ReadUInt64BigEndian(bytes) }.Value; return 8; } public static int EncodeWString(System.IO.Stream buffer, string value) { - byte[] bytes = Encoding.UTF8.GetBytes(value); - buffer.Write(bytes, 0, bytes.Length); - return bytes.Length; + var length = Encoding.UTF8.GetByteCount(value); +#if NET6_0_OR_GREATER + if (length <= 256) + { + Span bytes = stackalloc byte[256]; + var count = Encoding.UTF8.GetBytes(value.AsSpan(), bytes); + buffer.Write(bytes.Slice(0, count)); + return count; + } +#endif + var rented = ArrayPool.Shared.Rent(length); + try + { + var count = Encoding.UTF8.GetBytes(value, 0, value.Length, rented, 0); + buffer.Write(rented, 0, count); + return count; + } + finally + { + ArrayPool.Shared.Return(rented, clearArray: true); + } } public static int DecodeWString(System.IO.Stream buffer, int length, out string value) @@ -666,6 +723,13 @@ public static int DecodeWString(System.IO.Stream buffer, int length, out string value = string.Empty; return 0; } + if (buffer is System.IO.MemoryStream memory && memory.TryGetBuffer(out var segment)) + { + var offset = segment.Offset + checked((int)memory.Position); + value = Encoding.UTF8.GetString(segment.Array, offset, length); + memory.Position += length; + return length; + } byte[] tmp = new byte[length]; if (!TryReadExactly(buffer, tmp, length)) { diff --git a/src/S7CommPlusDriver/Internal/S7CommPlusReceiveQueue.cs b/src/S7CommPlusDriver/Internal/S7CommPlusReceiveQueue.cs new file mode 100644 index 0000000..6a29af6 --- /dev/null +++ b/src/S7CommPlusDriver/Internal/S7CommPlusReceiveQueue.cs @@ -0,0 +1,58 @@ +using System; +using System.Collections.Generic; +using System.Threading; + +namespace S7CommPlusDriver.Internal +{ + /// A synchronous receive handoff for the synchronous protocol worker. + internal sealed class S7CommPlusReceiveQueue + { + private readonly Queue _items = new Queue(); + private bool _completed; + + internal bool TryWrite(T item) + { + lock (_items) + { + if (_completed) return false; + _items.Enqueue(item); + Monitor.Pulse(_items); + return true; + } + } + + internal int Read(int timeout, out T item) + { + var deadline = RuntimeCompatibility.TickCount64 + Math.Max(1, timeout); + lock (_items) + { + while (_items.Count == 0 && !_completed) + { + var remaining = deadline - RuntimeCompatibility.TickCount64; + if (remaining <= 0) + { + item = default; + return S7Consts.errCliJobTimeout; + } + Monitor.Wait(_items, (int)Math.Min(int.MaxValue, remaining)); + } + if (_items.Count != 0) + { + item = _items.Dequeue(); + return 0; + } + item = default; + return S7Consts.errTCPNotConnected; + } + } + + internal void Complete() + { + lock (_items) + { + _completed = true; + Monitor.PulseAll(_items); + } + } + } +} diff --git a/src/S7CommPlusDriver/Net/IS7TlsConnector.cs b/src/S7CommPlusDriver/Net/IS7TlsConnector.cs index 671b76c..10d87e6 100644 --- a/src/S7CommPlusDriver/Net/IS7TlsConnector.cs +++ b/src/S7CommPlusDriver/Net/IS7TlsConnector.cs @@ -16,4 +16,16 @@ internal interface IS7TlsConnectorCallback void OnDataAvailable(); void OnSslError(int sslError, string sslState); } + + /// The plaintext buffer is borrowed and valid only for the duration of the callback. + internal interface IS7TlsPlaintextCallback + { + void OnPlaintextData(byte[] data, int dataLength); + } + + /// The encrypted buffer slice is borrowed for a synchronous transport send. + internal interface IS7TlsRecordCallback + { + void WriteEncryptedData(byte[] data, int offset, int count); + } } diff --git a/src/S7CommPlusDriver/Net/S7Client.cs b/src/S7CommPlusDriver/Net/S7Client.cs index b545d25..92929a4 100644 --- a/src/S7CommPlusDriver/Net/S7Client.cs +++ b/src/S7CommPlusDriver/Net/S7Client.cs @@ -11,6 +11,7 @@ using S7CommPlusDriver.Internal; using S7CommPlusDriver.Tls; using System; +using System.Buffers; using System.IO; using System.Threading; @@ -21,7 +22,7 @@ namespace S7CommPlusDriver // | it under the terms of the Lesser GNU General Public License as published by | // | the Free Software Foundation, either version 3 of the License, or | // | (at your option) any later version. | - public class S7Client : IS7TlsConnectorCallback, IDisposable + public class S7Client : IS7TlsConnectorCallback, IS7TlsPlaintextCallback, IS7TlsRecordCallback, IDisposable { public static bool WriteSslKeyToFile; public static string WriteSslKeyPath; @@ -98,26 +99,50 @@ public void WriteData(byte[] pData, int dataLength) } // Fragment the TLS record once at the negotiated COTP boundary. Splitting // it into fixed 960-byte messages first adds packets and premature EOT flags. - if (dataLength == pData.Length) - SendIsoPacket(pData); - else - { - var data = new byte[dataLength]; - Array.Copy(pData, data, dataLength); - SendIsoPacket(data); - } + SendIsoPacket(pData, 0, dataLength); + } + + void IS7TlsRecordCallback.WriteEncryptedData(byte[] data, int offset, int count) + { + SendIsoPacket(data, offset, count); } // OpenSSL meldet fertige Daten (decrypted) zum einlesen public void OnDataAvailable() { - // Netzwerk meldet eintreffende Daten - byte[] buf = new byte[8192]; - int bytesRead = m_sslconn.Receive(ref buf, buf.Length); - // System.Diagnostics.Trace.WriteLine("S7Client - OpenSSL OnDataAvailable: bytesRead=" + bytesRead); - byte[] readData = new byte[bytesRead]; - Array.Copy(buf, readData, bytesRead); - OnDataReceived?.Invoke(readData, bytesRead); + var rented = ArrayPool.Shared.Rent(8192); + try + { + var buffer = rented; + int bytesRead = m_sslconn.Receive(ref buffer, 8192); + DeliverPlaintext(buffer, bytesRead); + } + finally + { + ArrayPool.Shared.Return(rented, clearArray: true); + } + } + + void IS7TlsPlaintextCallback.OnPlaintextData(byte[] data, int dataLength) + { + DeliverPlaintext(data, dataLength); + } + + private void DeliverPlaintext(byte[] data, int dataLength) + { + if (dataLength <= 0) return; + if (data == null || dataLength > data.Length) + throw new ArgumentOutOfRangeException(nameof(dataLength)); + // The protocol parser consumes and copies the payload synchronously. Public + // callbacks retain their existing ownership contract and receive an exact copy. + OnBorrowedDataReceived?.Invoke(data, dataLength); + var callback = OnDataReceived; + if (callback != null) + { + var owned = new byte[dataLength]; + Buffer.BlockCopy(data, 0, owned, 0, dataLength); + callback(owned, dataLength); + } } public void OnSslError(int sslError, string sslState) @@ -345,6 +370,7 @@ private void ReceiveLoop() m_sslconn.ReadCompleted(Buffer, Size); } else { // Wenn etwas gelesen werden konnte, Client benachrichtigen + OnBorrowedDataReceived?.Invoke(Buffer, Size); OnDataReceived?.Invoke(Buffer, Size); } } @@ -357,6 +383,7 @@ private void ReceiveLoop() } public _OnDataReceived OnDataReceived; + internal _OnDataReceived OnBorrowedDataReceived; public delegate void _OnDataReceived(byte[] PDU, int len); public Action OnReceiveError; @@ -472,9 +499,14 @@ internal int SendEmptyDtData() } private int SendIsoPacket(byte[] Buffer) + { + return SendIsoPacket(Buffer, 0, Buffer?.Length ?? 0); + } + + private int SendIsoPacket(byte[] Buffer, int sourceOffset, int count) { _LastError = 0; - if (Buffer == null) + if (Buffer == null || sourceOffset < 0 || count < 0 || sourceOffset > Buffer.Length - count) return _LastError = S7Consts.errIsoInvalidPDU; int negotiatedTpduSize = _PDULength > 0 @@ -487,18 +519,26 @@ private int SendIsoPacket(byte[] Buffer) int offset = 0; do { - int chunkSize = Math.Min(maxPayloadSize, Buffer.Length - offset); - byte[] packet = new byte[TPKT_ISO.Length + chunkSize]; - Array.Copy(TPKT_ISO, 0, packet, 0, TPKT_ISO.Length); - SetWordAt(packet, 2, (ushort)packet.Length); - // The COTP EOT flag is set only on the final transport fragment. - packet[6] = offset + chunkSize >= Buffer.Length ? (byte)0x80 : (byte)0x00; - if (chunkSize > 0) - Array.Copy(Buffer, offset, packet, TPKT_ISO.Length, chunkSize); - SendPacket(packet); + int chunkSize = Math.Min(maxPayloadSize, count - offset); + int packetLength = TPKT_ISO.Length + chunkSize; + var packet = ArrayPool.Shared.Rent(packetLength); + try + { + Array.Copy(TPKT_ISO, 0, packet, 0, TPKT_ISO.Length); + SetWordAt(packet, 2, (ushort)packetLength); + // Pool capacity is not wire length. EOT belongs only to the last fragment. + packet[6] = offset + chunkSize >= count ? (byte)0x80 : (byte)0x00; + if (chunkSize > 0) + Array.Copy(Buffer, sourceOffset + offset, packet, TPKT_ISO.Length, chunkSize); + SendPacket(packet, packetLength); + } + finally + { + ArrayPool.Shared.Return(packet, clearArray: true); + } offset += chunkSize; } - while (_LastError == 0 && offset < Buffer.Length); + while (_LastError == 0 && offset < count); return _LastError; } diff --git a/src/S7CommPlusDriver/S7CommPlusDriver.csproj b/src/S7CommPlusDriver/S7CommPlusDriver.csproj index 9326e25..6de55b2 100644 --- a/src/S7CommPlusDriver/S7CommPlusDriver.csproj +++ b/src/S7CommPlusDriver/S7CommPlusDriver.csproj @@ -24,7 +24,6 @@ - diff --git a/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs b/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs index e3bff99..e977b4d 100644 --- a/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs +++ b/src/S7CommPlusDriver/S7CommPlusProtocolSession.cs @@ -17,7 +17,6 @@ using System.Collections.Generic; using System.Text; using System.Threading; -using System.Threading.Channels; using System.IO; using System.Linq; using System.Diagnostics; @@ -35,7 +34,7 @@ internal partial class S7CommPlusProtocolSession : IS7CommPlusBulkReadSession private S7Client m_client; private MemoryStream m_ReceivedPDU; private MemoryStream m_ReceivedTempPDU; - private Channel m_ReceivedPDUs = CreateReceiveChannel(); + private S7CommPlusReceiveQueue m_ReceivedPDUs = CreateReceiveQueue(); private readonly object m_RequestLock = new object(); private readonly object m_ReceiveDispatchLock = new object(); private readonly object m_NotificationQueueLock = new object(); @@ -89,14 +88,9 @@ public ReceivedS7PlusPdu(MemoryStream pdu, int errorCode = 0) public int ErrorCode { get; } } - private static Channel CreateReceiveChannel() + private static S7CommPlusReceiveQueue CreateReceiveQueue() { - return Channel.CreateUnbounded(new UnboundedChannelOptions - { - SingleReader = true, - SingleWriter = false, - AllowSynchronousContinuations = false - }); + return new S7CommPlusReceiveQueue(); } private UInt16 GetNextSequenceNumber() @@ -165,25 +159,11 @@ private void WaitForNewS7plusReceived(int Timeout) private int ReceiveNextS7plusPdu(int timeout, out MemoryStream pdu) { pdu = null; - try - { - using var cts = new CancellationTokenSource(Math.Max(1, timeout)); - var received = m_ReceivedPDUs.Reader.ReadAsync(cts.Token).AsTask().GetAwaiter().GetResult(); - if (received.ErrorCode != 0) - { - return received.ErrorCode; - } - pdu = received.Pdu; - return 0; - } - catch (OperationCanceledException) - { - return S7Consts.errCliJobTimeout; - } - catch (ChannelClosedException) - { - return S7Consts.errTCPNotConnected; - } + var result = m_ReceivedPDUs.Read(timeout, out var received); + if (result != 0) return result; + if (received.ErrorCode != 0) return received.ErrorCode; + pdu = received.Pdu; + return 0; } private int SendS7plusFunctionObject(IS7pRequest funcObj) @@ -211,9 +191,11 @@ private int SendS7plusFunctionObject(IS7pRequest funcObj) } m_LastSentRequestForWait = funcObj; - MemoryStream stream = new MemoryStream(); + using var stream = funcObj is GetMultiVariablesRequest read + ? new MemoryStream(checked((int)read.GetSerializedLength())) + : new MemoryStream(); funcObj.Serialize(stream); - return SendS7plusPDUdata(stream.ToArray(), (int)stream.Length, funcObj.ProtocolVersion, + return SendS7plusPDUdata(stream.GetBuffer(), (int)stream.Length, funcObj.ProtocolVersion, allowTransportFragmentation: funcObj is GetMultiVariablesRequest); } @@ -789,6 +771,8 @@ private static int AppendItemsWithinPayloadLimit( private static long GetSerializedRequestLengthForBatching(IS7pRequest request) { + if (request is GetMultiVariablesRequest read) + return read.GetSerializedLength(UInt32.MaxValue); uint sessionId = request.SessionId; ushort sequenceNumber = request.SequenceNumber; uint integrityId = request.IntegrityId; @@ -841,9 +825,17 @@ private void OnDataReceived(byte[] PDU, int len) // First step we push the complete PDU to a queue. // Receive errors are published through the receive queue so request waiters fail immediately. + // Borrowed TLS buffers may have spare capacity. Validate the active length + // before parsing so truncated frames cannot expose stale buffer contents. + if (PDU == null || len < 4 || len > PDU.Length) + { + m_ReceivedNeedMoreDataForCompletePDU = false; + PublishReceiveError(S7Consts.errIsoInvalidPDU); + return; + } if (!m_ReceivedNeedMoreDataForCompletePDU) { - m_ReceivedTempPDU = new MemoryStream(); + m_ReceivedTempPDU = new MemoryStream(len); } // S7comm-plus byte protoVersion; @@ -874,6 +866,12 @@ private void OnDataReceived(byte[] PDU, int len) // Read the length of the data-part from header s7HeaderDataLen = GetWordAt(PDU, pos); pos += 2; + if (s7HeaderDataLen > len - pos) + { + m_ReceivedNeedMoreDataForCompletePDU = false; + PublishReceiveError(S7Consts.errIsoInvalidPDU); + return; + } if (s7HeaderDataLen > 0) { // Special handling for SystemEvent 0xfe PDUs: @@ -973,7 +971,7 @@ private void OnDataReceived(byte[] PDU, int len) // If a complete (usable) PDU is received, add to the queue (threadsafe) for readout if (m_NewS7CommPlusReceived) { - m_ReceivedPDUs.Writer.TryWrite(new ReceivedS7PlusPdu(m_ReceivedTempPDU)); + m_ReceivedPDUs.TryWrite(new ReceivedS7PlusPdu(m_ReceivedTempPDU)); m_NewS7CommPlusReceived = false; } } @@ -981,7 +979,7 @@ private void OnDataReceived(byte[] PDU, int len) private void PublishReceiveError(int errorCode) { m_LastError = errorCode; - m_ReceivedPDUs.Writer.TryWrite(new ReceivedS7PlusPdu(null, errorCode)); + m_ReceivedPDUs.TryWrite(new ReceivedS7PlusPdu(null, errorCode)); } private byte[] BuildLegacyAccumulatedDigestData(byte[] pdu, int bodyPos, int bodyLen, int previousBodyLen) @@ -998,6 +996,11 @@ internal void DebugOnDataReceivedForTests(byte[] pdu) OnDataReceived(pdu, pdu.Length); } + internal void DebugOnBorrowedDataReceivedForTests(byte[] pdu, int activeLength) + { + OnDataReceived(pdu, activeLength); + } + internal int DebugReceiveNextS7plusPduForTests(int timeout, out MemoryStream pdu) { return ReceiveNextS7plusPdu(timeout, out pdu); @@ -1005,8 +1008,8 @@ internal int DebugReceiveNextS7plusPduForTests(int timeout, out MemoryStream pdu internal void DebugResetReceiveDispatcherForTests() { - m_ReceivedPDUs.Writer.TryComplete(); - m_ReceivedPDUs = CreateReceiveChannel(); + m_ReceivedPDUs.Complete(); + m_ReceivedPDUs = CreateReceiveQueue(); m_ReceivedTempPDU = null; m_ReceivedPDU = null; m_ReceivedNeedMoreDataForCompletePDU = false; @@ -1203,10 +1206,10 @@ private void PrepareClient(string address, int timeoutMs, int port, ushort local m_IntegrityId_Set = 0; dbInfoList = null; typeInfoList.Clear(); - m_ReceivedPDUs.Writer.TryComplete(); - m_ReceivedPDUs = CreateReceiveChannel(); + m_ReceivedPDUs.Complete(); + m_ReceivedPDUs = CreateReceiveQueue(); m_client = new S7Client(); - m_client.OnDataReceived = this.OnDataReceived; + m_client.OnBorrowedDataReceived = this.OnDataReceived; m_client.OnReceiveError = this.PublishReceiveError; m_client.PLCPort = port; m_client.ConnTimeout = timeoutMs; @@ -1389,6 +1392,7 @@ private int DisposeClient(int timeoutMs) // A receive that is still unwinding must not publish into a new session. client.OnDataReceived = null; + client.OnBorrowedDataReceived = null; client.OnReceiveError = null; return client.Dispose(timeoutMs); } @@ -1435,8 +1439,8 @@ public int TryDisconnect(int timeoutMs = 2000) m_ReadTimeout = oldTimeout; m_ReceivedPDU = null; m_ReceivedTempPDU = null; - m_ReceivedPDUs.Writer.TryComplete(); - m_ReceivedPDUs = CreateReceiveChannel(); + m_ReceivedPDUs.Complete(); + m_ReceivedPDUs = CreateReceiveQueue(); m_ReceivedNeedMoreDataForCompletePDU = false; m_NewS7CommPlusReceived = false; m_LastSentRequestForWait = null; @@ -1477,8 +1481,8 @@ internal int CloseTransport(int timeoutMs = 2000) m_ReadTimeout = oldTimeout; m_ReceivedPDU = null; m_ReceivedTempPDU = null; - m_ReceivedPDUs.Writer.TryComplete(); - m_ReceivedPDUs = CreateReceiveChannel(); + m_ReceivedPDUs.Complete(); + m_ReceivedPDUs = CreateReceiveQueue(); m_ReceivedNeedMoreDataForCompletePDU = false; m_NewS7CommPlusReceived = false; m_LastSentRequestForWait = null; @@ -1575,8 +1579,8 @@ public int ReadValues(List addresslist, out List values, ou } // The requester must pass the internal type with the request, otherwise not all return values can be converted automatically. // For example, strings are transmitted as UInt-Array. - values = new List(); - errors = new List(); + values = new List(addresslist.Count); + errors = new List(addresslist.Count); // Initialize error fields to error value for (int i = 0; i < addresslist.Count; i++) { @@ -1606,7 +1610,10 @@ public int ReadValues(List addresslist, out List values, ou return res; } - var getMultiVarRes = GetMultiVariablesResponse.DeserializeFromPdu(m_ReceivedPDU); + var getMultiVarRes = GetMultiVariablesReadResponse.ReadHeader(m_ReceivedPDU); + if (getMultiVarRes == null) return S7Consts.errIsoInvalidPDU4; + res = getMultiVarRes.ReadItems(m_ReceivedPDU, values, errors, chunk_startIndex, count_perChunk); + if (res != 0) return res; res = checkResponseWithIntegrity(getMultiVarReq, getMultiVarRes); if (res != 0) { @@ -1618,19 +1625,6 @@ public int ReadValues(List addresslist, out List values, ou Trace.WriteLine("S7CommPlusProtocolSession - ReadValues: Executed with Error! ReturnValue=" + getMultiVarRes.ReturnValue); } - // If a variable could not be read, there is no value, but there is an ErrorValue. - // The production client maps this to per-item batch status. - foreach (var v in getMultiVarRes.Values) - { - values[chunk_startIndex + (int)v.Key - 1] = v.Value; - // Initialize error to 0, will be overwritten below if there was an error on an item. - errors[chunk_startIndex + (int)v.Key - 1] = 0; - } - - foreach (var ev in getMultiVarRes.ErrorValues) - { - errors[chunk_startIndex + (int)ev.Key - 1] = ev.Value; - } chunk_startIndex += count_perChunk; } while (chunk_startIndex < addresslist.Count); @@ -1649,7 +1643,8 @@ internal static GetMultiVariablesRequest CreateReadRequestBatch( { request.AddressList.Add(addresses[startIndex + offset]); } - // Ordinary reads fit in one logical PDU: measure the complete batch once. + // Ordinary reads fit in one logical PDU: calculate the complete batch length + // without encoding a throwaway payload. // Only unusually long addresses need the bounded payload search. if (maxCandidates <= 1 || !ExceedsSingleFramePayload(request, maxPayloadSize)) { diff --git a/src/S7CommPlusDriver/S7CommPlusReadPipeline.cs b/src/S7CommPlusDriver/S7CommPlusReadPipeline.cs index 5a3a581..633d934 100644 --- a/src/S7CommPlusDriver/S7CommPlusReadPipeline.cs +++ b/src/S7CommPlusDriver/S7CommPlusReadPipeline.cs @@ -47,27 +47,15 @@ int FillWindow() { var received = receive(pending.Values.First().Request, sequences); if (received.Error != 0) return received.Error; - var response = GetMultiVariablesResponse.DeserializeFromPdu(received.Pdu); + var response = GetMultiVariablesReadResponse.ReadHeader(received.Pdu); if (response == null || !pending.TryGetValue(response.SequenceNumber, out var completed)) return S7Consts.errIsoInvalidPDU5; + result = response.ReadItems(received.Pdu, values, errors, completed.Start, completed.Count); + if (result != 0) return result; result = validate(completed.Request, response); if (result != 0) return result; if (response.IntegrityId != unchecked(completed.Request.IntegrityId + completed.Request.SequenceNumber)) return S7Consts.errIsoInvalidPDU; - // Preserve item failures and missing-value status; never expose partial - // data as a successful logical read after a transport/validation failure. - foreach (var value in response.Values) - { - if (value.Key == 0 || value.Key > completed.Count) return S7Consts.errIsoInvalidPDU; - var index = completed.Start + (int)value.Key - 1; - values[index] = value.Value; - errors[index] = 0; - } - foreach (var error in response.ErrorValues) - { - if (error.Key == 0 || error.Key > completed.Count) return S7Consts.errIsoInvalidPDU; - errors[completed.Start + (int)error.Key - 1] = error.Value; - } pending.Remove(response.SequenceNumber); sequences.Remove(response.SequenceNumber); result = FillWindow(); diff --git a/src/S7CommPlusDriver/Tls/BouncyCastleTlsConnector.cs b/src/S7CommPlusDriver/Tls/BouncyCastleTlsConnector.cs index 8e3d0c1..1f61042 100644 --- a/src/S7CommPlusDriver/Tls/BouncyCastleTlsConnector.cs +++ b/src/S7CommPlusDriver/Tls/BouncyCastleTlsConnector.cs @@ -111,10 +111,17 @@ private void ReadDecryptedData() break; } - var data = new byte[bytesRead]; - Buffer.BlockCopy(buffer, 0, data, 0, bytesRead); - _decryptedData.Add(data); - _dataSink.OnDataAvailable(); + if (_dataSink is IS7TlsPlaintextCallback plaintextSink) + { + plaintextSink.OnPlaintextData(buffer, bytesRead); + } + else + { + var data = new byte[bytesRead]; + Buffer.BlockCopy(buffer, 0, data, 0, bytesRead); + _decryptedData.Add(data); + _dataSink.OnDataAvailable(); + } } } catch (Exception ex) @@ -285,6 +292,11 @@ public override int Read(byte[] buffer, int offset, int count) public override void Write(byte[] buffer, int offset, int count) { + if (_dataSink is IS7TlsRecordCallback recordSink) + { + recordSink.WriteEncryptedData(buffer, offset, count); + return; + } var data = new byte[count]; Buffer.BlockCopy(buffer, offset, data, 0, count); _dataSink.WriteData(data, data.Length);