Improve bulk-read throughput and reduce transport allocations

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:
 

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,13 +198,29 @@ public void WriteData(byte[] data, int dataLength)
 
             public void OnDataAvailable()
             {
+                DataAvailable?.Invoke();
             }
 
             public void OnSslError(int sslError, string sslState)
             {
             }
         }
 
+        private sealed class BorrowedLoopbackCallback : IS7TlsConnectorCallback, IS7TlsPlaintextCallback, IS7TlsRecordCallback
+        {
+            private readonly LoopbackConnectorCallback _owned;
+            private readonly Action<byte[], int> _receive;
+            internal BorrowedLoopbackCallback(LoopbackConnectorCallback owned, Action<byte[], int> 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<byte[]> _incoming = new BlockingCollection<byte[]>();

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();
+    }
+}

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; }

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<PValue>(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<InvalidDataException>(() => 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();
+        }
+    }
+}

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
+    }
+}