diff --git a/src/Nerdbank.Streams/MultiplexingStream.Formatters.cs b/src/Nerdbank.Streams/MultiplexingStream.Formatters.cs index 732ed4db..9a6fa83c 100644 --- a/src/Nerdbank.Streams/MultiplexingStream.Formatters.cs +++ b/src/Nerdbank.Streams/MultiplexingStream.Formatters.cs @@ -20,6 +20,18 @@ public partial class MultiplexingStream { internal abstract class Formatter : System.IAsyncDisposable { + /// + /// The maximum number of bytes required to encode a payload: + /// a 1-element msgpack array header, plus a 9-byte (worst case) integer. + /// + protected const int MaxContentProcessedPayloadLength = 10; + + /// + /// 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). + /// + 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, }; } + + /// + /// An over a single, fixed-size array. + /// + /// + /// This is a lighter-weight alternative to for very small payloads + /// whose maximum size is known in advance. 's default constructor + /// creates a dedicated , which is far too expensive for hot paths. + /// + protected class FixedSizeBufferWriter : IBufferWriter + { + private readonly byte[] buffer; + private int written; + + /// + /// Initializes a new instance of the class. + /// + /// The maximum number of bytes that may be written. + internal FixedSizeBufferWriter(int capacity) + { + this.buffer = new byte[capacity]; + } + + /// + /// Gets a sequence over the bytes written so far. + /// + internal ReadOnlySequence WrittenSequence => new ReadOnlySequence(this.buffer, 0, this.written); + + /// + /// Thrown if is negative or exceeds the space previously returned by or . + public void Advance(int count) + { + Requires.Range(count >= 0 && count <= this.buffer.Length - this.written, nameof(count)); + this.written += count; + } + + /// + /// Thrown if is negative or exceeds the remaining capacity. + public Memory GetMemory(int sizeHint = 0) + { + this.CheckSizeHint(sizeHint); + return new Memory(this.buffer, this.written, this.buffer.Length - this.written); + } + + /// + /// Thrown if is negative or exceeds the remaining capacity. + public Span GetSpan(int sizeHint = 0) + { + this.CheckSizeHint(sizeHint); + return new Span(this.buffer, this.written, this.buffer.Length - this.written); + } + + /// + /// Verifies that this writer can satisfy a request for a given amount of space. + /// + /// The amount of space requested by the caller. + /// + /// 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 contract. + /// + 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 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 SerializeException(Exception? exception) internal override ReadOnlySequence SerializeContentProcessed(long bytesProcessed) { - var sequence = new Sequence(); - var writer = new MessagePackWriter(sequence); + // This method is on a very hot path (one call per frame of content received), + // so avoid Sequence, whose default constructor creates a dedicated ArrayPool 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 payload) @@ -649,6 +739,11 @@ internal override void WriteFrame(FrameHeader header, ReadOnlySequence 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; diff --git a/src/Nerdbank.Streams/MultiplexingStream.Options.cs b/src/Nerdbank.Streams/MultiplexingStream.Options.cs index 2a33e7bf..ad3a8aa0 100644 --- a/src/Nerdbank.Streams/MultiplexingStream.Options.cs +++ b/src/Nerdbank.Streams/MultiplexingStream.Options.cs @@ -25,10 +25,18 @@ public class Options /// which also serves as the minimum window size for any channel. /// /// + /// /// Using an integer multiple of ensures that the client can send full frames /// instead of ending with a partial frame when the remote window limit is reached. + /// + /// + /// 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 limit; memory is allocated on demand, so channels that carry little data cost little memory. + /// /// - private static readonly long RecommendedDefaultChannelReceivingWindowSize = 5 * FramePayloadMaxLength; + private static readonly long RecommendedDefaultChannelReceivingWindowSize = 50 * FramePayloadMaxLength; /// /// Backing field for the property. diff --git a/src/Nerdbank.Streams/MultiplexingStream.cs b/src/Nerdbank.Streams/MultiplexingStream.cs index 35c43891..ba2fecf0 100644 --- a/src/Nerdbank.Streams/MultiplexingStream.cs +++ b/src/Nerdbank.Streams/MultiplexingStream.cs @@ -32,6 +32,20 @@ public partial class MultiplexingStream : IDisposableObservable, System.IAsyncDi /// private const int FramePayloadMaxLength = 20 * 1024; + /// + /// Options for the that buffers frames on their way to the underlying transport. + /// + /// + /// The default 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. + /// + private static readonly PipeOptions TransportPipeOptions = new PipeOptions( + pauseWriterThreshold: 16 * FramePayloadMaxLength, + resumeWriterThreshold: 8 * FramePayloadMaxLength, + minimumSegmentSize: FramePayloadMaxLength, + useSynchronizationContext: false); + /// /// The encoding used for characters in control frames. /// @@ -324,7 +338,7 @@ public static async Task 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 { diff --git a/src/Nerdbank.Streams/Sequence`1.cs b/src/Nerdbank.Streams/Sequence`1.cs index 23306b76..c1dbcf2d 100644 --- a/src/Nerdbank.Streams/Sequence`1.cs +++ b/src/Nerdbank.Streams/Sequence`1.cs @@ -276,7 +276,10 @@ private SequenceSegment GetSegment(int sizeHint) Sequence.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 { diff --git a/test/Nerdbank.Streams.Benchmark/MultiplexingStreamBenchmark.cs b/test/Nerdbank.Streams.Benchmark/MultiplexingStreamBenchmark.cs new file mode 100644 index 00000000..acf5ea97 --- /dev/null +++ b/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; + + /// + /// Measures the throughput of a single carrying bulk data + /// over a loopback socket, which is representative of the "large file transfer" scenario. + /// + [Config(typeof(BenchmarkConfig))] + public class MultiplexingStreamBenchmark + { + private const int ChunkSize = 64 * 1024; + + private static readonly byte[] Chunk = new byte[ChunkSize]; + + /// + /// Gets or sets the major version of the multiplexing protocol to exercise. + /// + [Params(1, 2, 3)] + public int ProtocolMajorVersion { get; set; } + + /// + /// Gets or sets the number of bytes to transmit in each iteration. + /// + [Params(32 * 1024 * 1024)] + public int TransferSize { get; set; } + + /// + /// Transmits bytes across one channel and waits for it all to be received. + /// + /// A task that tracks the transfer. + [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 mx1Task = MultiplexingStream.CreateAsync(transport.Client, options1); + Task mx2Task = MultiplexingStream.CreateAsync(transport.Server, options2); + MultiplexingStream mx1 = await mx1Task; + MultiplexingStream mx2 = await mx2Task; + try + { + Task offer = mx1.OfferChannelAsync("bench"); + Task 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 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(); + } + } + } +} diff --git a/test/Nerdbank.Streams.Benchmark/Program.cs b/test/Nerdbank.Streams.Benchmark/Program.cs index 3eaea424..1c5ee825 100644 --- a/test/Nerdbank.Streams.Benchmark/Program.cs +++ b/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); }