Improve MultiplexingStream v2/v3 throughput ~2.3x

src/Nerdbank.Streams/MultiplexingStream.Formatters.cs

@@ -20,6 +20,18 @@ public partial class MultiplexingStream
     {
         internal abstract class Formatter : System.IAsyncDisposable
         {
+            /// <summary>
+            /// The maximum number of bytes required to encode a <see cref="ControlCode.ContentProcessed"/> payload:
+            /// a 1-element msgpack array header, plus a 9-byte (worst case) integer.
+            /// </summary>
+            protected const int MaxContentProcessedPayloadLength = 10;
+
+            /// <summary>
+            /// The maximum number of bytes required to encode everything in a frame that precedes its payload
+            /// (the msgpack array header, control code, channel ID, channel source, and binary payload header).
+            /// </summary>
+            protected const int MaxFramePrologueLength = 25;
+
             protected Formatter(PipeWriter writer)
             {
                 this.PipeWriter = writer;
@@ -134,6 +146,77 @@ protected FrameHeader CreateFrameHeader(ControlCode code, ulong? channelId, Chan
                     ChannelId = qualifiedId,
                 };
             }
+
+            /// <summary>
+            /// An <see cref="IBufferWriter{T}"/> over a single, fixed-size array.
+            /// </summary>
+            /// <remarks>
+            /// This is a lighter-weight alternative to <see cref="Sequence{T}"/> for very small payloads
+            /// whose maximum size is known in advance. <see cref="Sequence{T}"/>'s default constructor
+            /// creates a dedicated <see cref="ArrayPool{T}"/>, which is far too expensive for hot paths.
+            /// </remarks>
+            protected class FixedSizeBufferWriter : IBufferWriter<byte>
+            {
+                private readonly byte[] buffer;
+                private int written;
+
+                /// <summary>
+                /// Initializes a new instance of the <see cref="FixedSizeBufferWriter"/> class.
+                /// </summary>
+                /// <param name="capacity">The maximum number of bytes that may be written.</param>
+                internal FixedSizeBufferWriter(int capacity)
+                {
+                    this.buffer = new byte[capacity];
+                }
+
+                /// <summary>
+                /// Gets a sequence over the bytes written so far.
+                /// </summary>
+                internal ReadOnlySequence<byte> WrittenSequence => new ReadOnlySequence<byte>(this.buffer, 0, this.written);
+
+                /// <inheritdoc/>
+                /// <exception cref="ArgumentOutOfRangeException">Thrown if <paramref name="count"/> is negative or exceeds the space previously returned by <see cref="GetSpan(int)"/> or <see cref="GetMemory(int)"/>.</exception>
+                public void Advance(int count)
+                {
+                    Requires.Range(count >= 0 && count <= this.buffer.Length - this.written, nameof(count));
+                    this.written += count;
+                }
+
+                /// <inheritdoc/>
+                /// <exception cref="ArgumentOutOfRangeException">Thrown if <paramref name="sizeHint"/> is negative or exceeds the remaining capacity.</exception>
+                public Memory<byte> GetMemory(int sizeHint = 0)
+                {
+                    this.CheckSizeHint(sizeHint);
+                    return new Memory<byte>(this.buffer, this.written, this.buffer.Length - this.written);
+                }
+
+                /// <inheritdoc/>
+                /// <exception cref="ArgumentOutOfRangeException">Thrown if <paramref name="sizeHint"/> is negative or exceeds the remaining capacity.</exception>
+                public Span<byte> GetSpan(int sizeHint = 0)
+                {
+                    this.CheckSizeHint(sizeHint);
+                    return new Span<byte>(this.buffer, this.written, this.buffer.Length - this.written);
+                }
+
+                /// <summary>
+                /// Verifies that this writer can satisfy a request for a given amount of space.
+                /// </summary>
+                /// <param name="sizeHint">The amount of space requested by the caller.</param>
+                /// <remarks>
+                /// This writer cannot grow, so a request it cannot satisfy indicates that the capacity
+                /// this instance was created with is too small for what is being serialized.
+                /// Failing loudly here is far preferable to silently returning a short buffer,
+                /// which would violate the <see cref="IBufferWriter{T}"/> contract.
+                /// </remarks>
+                private void CheckSizeHint(int sizeHint)
+                {
+                    Requires.Range(sizeHint >= 0, nameof(sizeHint));
+                    if (sizeHint > this.buffer.Length - this.written)
+                    {
+                        throw new ArgumentOutOfRangeException(nameof(sizeHint), $"Requested {sizeHint} bytes but only {this.buffer.Length - this.written} remain of this writer's {this.buffer.Length} byte capacity.");
+                    }
+                }
+            }
         }
 
         internal class V1Formatter : Formatter
@@ -369,6 +452,11 @@ internal override void WriteFrame(FrameHeader header, ReadOnlySequence<byte> pay
             {
                 Verify.NotDisposed(!this.IsDisposed, this);
 
+                // Reserve enough contiguous space for the entire frame up-front.
+                // Without this, the payload lands in a buffer of its own, separate from the frame's msgpack header,
+                // which fragments the frame across multiple buffers and thus multiple writes to the transport.
+                this.PipeWriter.GetSpan(checked(MaxFramePrologueLength + (int)payload.Length));
+
                 var writer = new MessagePackWriter(this.PipeWriter);
 
                 int elementCount = !payload.IsEmpty ? 3 : header.ChannelId.HasValue ? 2 : 1;
@@ -463,12 +551,14 @@ internal ReadOnlySequence<byte> SerializeException(Exception? exception)
 
             internal override ReadOnlySequence<byte> SerializeContentProcessed(long bytesProcessed)
             {
-                var sequence = new Sequence<byte>();
-                var writer = new MessagePackWriter(sequence);
+                // This method is on a very hot path (one call per frame of content received),
+                // so avoid Sequence<byte>, whose default constructor creates a dedicated ArrayPool<byte> each time.
+                var bufferWriter = new FixedSizeBufferWriter(MaxContentProcessedPayloadLength);
+                var writer = new MessagePackWriter(bufferWriter);
                 writer.WriteArrayHeader(1);
                 writer.Write(bytesProcessed);
                 writer.Flush();
-                return sequence;
+                return bufferWriter.WrittenSequence;
             }
 
             internal override long DeserializeContentProcessed(ReadOnlySequence<byte> payload)
@@ -649,6 +739,11 @@ internal override void WriteFrame(FrameHeader header, ReadOnlySequence<byte> pay
             {
                 Verify.NotDisposed(!this.IsDisposed, this);
 
+                // Reserve enough contiguous space for the entire frame up-front.
+                // Without this, the payload lands in a buffer of its own, separate from the frame's msgpack header,
+                // which fragments the frame across multiple buffers and thus multiple writes to the transport.
+                this.PipeWriter.GetSpan(checked(MaxFramePrologueLength + (int)payload.Length));
+
                 var writer = new MessagePackWriter(this.PipeWriter);
 
                 int elementCount = !payload.IsEmpty ? 4 : header.ChannelId.HasValue ? 3 : 1;

src/Nerdbank.Streams/MultiplexingStream.Options.cs

@@ -25,10 +25,18 @@ public class Options
             /// which also serves as the minimum window size for any channel.
             /// </summary>
             /// <remarks>
+            /// <para>
             /// Using an integer multiple of <see cref="FramePayloadMaxLength"/> ensures that the client can send full frames
             /// instead of ending with a partial frame when the remote window limit is reached.
+            /// </para>
+            /// <para>
+            /// This window is the number of bytes a channel's sender may transmit before it must stop and wait for the
+            /// receiver to acknowledge that it has processed some of them. Each such wait costs a round-trip through both
+            /// processes' thread pools, so a small window severely limits throughput on fast transports. This value is only
+            /// a <em>limit</em>; memory is allocated on demand, so channels that carry little data cost little memory.
+            /// </para>
             /// </remarks>
-            private static readonly long RecommendedDefaultChannelReceivingWindowSize = 5 * FramePayloadMaxLength;
+            private static readonly long RecommendedDefaultChannelReceivingWindowSize = 50 * FramePayloadMaxLength;
 
             /// <summary>
             /// Backing field for the <see cref="TraceSource"/> property.

src/Nerdbank.Streams/MultiplexingStream.cs

@@ -32,6 +32,20 @@ public partial class MultiplexingStream : IDisposableObservable, System.IAsyncDi
         /// </summary>
         private const int FramePayloadMaxLength = 20 * 1024;
 
+        /// <summary>
+        /// Options for the <see cref="Pipe"/> that buffers frames on their way to the underlying transport.
+        /// </summary>
+        /// <remarks>
+        /// The default <see cref="PipeOptions"/> allow only 64KB of buffering, which is barely three frames.
+        /// A multiplexing transport benefits from buffering many more frames so that several may be written
+        /// to the transport at once and so that frame producers rarely have to wait on the transport.
+        /// </remarks>
+        private static readonly PipeOptions TransportPipeOptions = new PipeOptions(
+            pauseWriterThreshold: 16 * FramePayloadMaxLength,
+            resumeWriterThreshold: 8 * FramePayloadMaxLength,
+            minimumSegmentSize: FramePayloadMaxLength,
+            useSynchronizationContext: false);
+
         /// <summary>
         /// The encoding used for characters in control frames.
         /// </summary>
@@ -324,7 +338,7 @@ public static async Task<MultiplexingStream> CreateAsync(Stream stream, Options?
 
             // Do NOT specify our own cancellationToken parameter in UsePipeWriter, since this PipeWriter
             // must outlive this method and therefore should not be canceled later if that token is eventually canceled.
-            PipeWriter? streamWriter = stream.UsePipeWriter(cancellationToken: CancellationToken.None);
+            PipeWriter? streamWriter = stream.UsePipeWriter(TransportPipeOptions, cancellationToken: CancellationToken.None);
 
             Formatter? formatter = options.ProtocolMajorVersion switch
             {

src/Nerdbank.Streams/Sequence`1.cs

@@ -276,7 +276,10 @@ private SequenceSegment GetSegment(int sizeHint)
                 Sequence<T>.SequenceSegment? segment = this.segmentPool.Count > 0 ? this.segmentPool.Pop() : new SequenceSegment();
                 if (this.arrayPool != null)
                 {
-                    segment.Assign(this.arrayPool.Rent(minBufferSize.Value == -1 ? DefaultLengthFromArrayPool : minBufferSize.Value));
+                    // When the caller gave no size hint, we still honor MinimumSpanLength when it exceeds the pool's
+                    // own default, since callers that fill buffers from a stream benefit from larger buffers
+                    // just as much as those that ask for a specific size. Never shrink below the pool's default.
+                    segment.Assign(this.arrayPool.Rent(minBufferSize.Value == -1 ? Math.Max(DefaultLengthFromArrayPool, this.MinimumSpanLength) : minBufferSize.Value));
                 }
                 else
                 {

test/Nerdbank.Streams.Benchmark/MultiplexingStreamBenchmark.cs

@@ -0,0 +1,113 @@
+// Copyright (c) Andrew Arnott. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+namespace Nerdbank.Streams.Benchmark
+{
+    using System;
+    using System.IO;
+    using System.IO.Pipelines;
+    using System.Net;
+    using System.Net.Sockets;
+    using System.Threading.Tasks;
+    using BenchmarkDotNet.Attributes;
+
+    /// <summary>
+    /// Measures the throughput of a single <see cref="MultiplexingStream.Channel"/> carrying bulk data
+    /// over a loopback socket, which is representative of the "large file transfer" scenario.
+    /// </summary>
+    [Config(typeof(BenchmarkConfig))]
+    public class MultiplexingStreamBenchmark
+    {
+        private const int ChunkSize = 64 * 1024;
+
+        private static readonly byte[] Chunk = new byte[ChunkSize];
+
+        /// <summary>
+        /// Gets or sets the major version of the multiplexing protocol to exercise.
+        /// </summary>
+        [Params(1, 2, 3)]
+        public int ProtocolMajorVersion { get; set; }
+
+        /// <summary>
+        /// Gets or sets the number of bytes to transmit in each iteration.
+        /// </summary>
+        [Params(32 * 1024 * 1024)]
+        public int TransferSize { get; set; }
+
+        /// <summary>
+        /// Transmits <see cref="TransferSize"/> bytes across one channel and waits for it all to be received.
+        /// </summary>
+        /// <returns>A task that tracks the transfer.</returns>
+        [Benchmark]
+        public async Task TransmitBulkDataOverOneChannel()
+        {
+            (Stream Client, Stream Server) transport = await CreateLoopbackStreamPairAsync();
+            MultiplexingStream.Options options1 = new() { ProtocolMajorVersion = this.ProtocolMajorVersion };
+            MultiplexingStream.Options options2 = new() { ProtocolMajorVersion = this.ProtocolMajorVersion };
+            Task<MultiplexingStream> mx1Task = MultiplexingStream.CreateAsync(transport.Client, options1);
+            Task<MultiplexingStream> mx2Task = MultiplexingStream.CreateAsync(transport.Server, options2);
+            MultiplexingStream mx1 = await mx1Task;
+            MultiplexingStream mx2 = await mx2Task;
+            try
+            {
+                Task<MultiplexingStream.Channel> offer = mx1.OfferChannelAsync("bench");
+                Task<MultiplexingStream.Channel> accept = mx2.AcceptChannelAsync("bench");
+                MultiplexingStream.Channel sender = await offer;
+                MultiplexingStream.Channel receiver = await accept;
+
+                long transferSize = this.TransferSize;
+                Task receiveTask = Task.Run(async delegate
+                {
+                    long bytesReceived = 0;
+                    while (bytesReceived < transferSize)
+                    {
+                        ReadResult readResult = await receiver.Input.ReadAsync();
+                        if (readResult.Buffer.IsEmpty && readResult.IsCompleted)
+                        {
+                            break;
+                        }
+
+                        bytesReceived += readResult.Buffer.Length;
+                        receiver.Input.AdvanceTo(readResult.Buffer.End);
+                    }
+                });
+
+                for (long bytesSent = 0; bytesSent < transferSize;)
+                {
+                    int bytesThisRound = (int)Math.Min(ChunkSize, transferSize - bytesSent);
+                    Chunk.AsSpan(0, bytesThisRound).CopyTo(sender.Output.GetSpan(bytesThisRound));
+                    sender.Output.Advance(bytesThisRound);
+                    await sender.Output.FlushAsync();
+                    bytesSent += bytesThisRound;
+                }
+
+                await receiveTask;
+            }
+            finally
+            {
+                await mx1.DisposeAsync();
+                await mx2.DisposeAsync();
+            }
+        }
+
+        private static async Task<(Stream Client, Stream Server)> CreateLoopbackStreamPairAsync()
+        {
+            TcpListener listener = new(IPAddress.Loopback, 0);
+            listener.Start();
+            try
+            {
+                Task<TcpClient> acceptTask = listener.AcceptTcpClientAsync();
+                TcpClient client = new();
+                await client.ConnectAsync(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
+                TcpClient server = await acceptTask;
+                client.NoDelay = true;
+                server.NoDelay = true;
+                return (client.GetStream(), server.GetStream());
+            }
+            finally
+            {
+                listener.Stop();
+            }
+        }
+    }
+}

test/Nerdbank.Streams.Benchmark/Program.cs

@@ -13,6 +13,7 @@ private static void Main(string[] args)
             var switcher = new BenchmarkSwitcher(new[]
             {
                 typeof(SequenceBenchmark),
+                typeof(MultiplexingStreamBenchmark),
             });
             switcher.Run(args);
         }