Scale receive-window ACK replenishment with the window size

src/Nerdbank.Streams/MultiplexingStream.Channel.cs

@@ -1160,6 +1160,7 @@ private class WindowPipeReader : PipeReader
             {
                 private readonly Channel owner;
                 private readonly PipeReader inner;
+                private readonly long ackThreshold;
                 private ReadResult lastReadResult;
                 private long bytesProcessed;
                 private SequencePosition lastExaminedPosition;
@@ -1168,6 +1169,15 @@ internal WindowPipeReader(Channel owner, PipeReader inner)
                 {
                     this.owner = owner;
                     this.inner = inner;
+
+                    // Acknowledge in fractions of the window rather than in fixed-size frames.
+                    // Every acknowledgement costs a frame on the wire and a wake-up on both sides, so tying the
+                    // rate to the window keeps that cost proportional to the window instead of growing with it.
+                    // The frame length floor preserves the original behavior for small windows, and dividing
+                    // (rather than subtracting) guarantees the threshold never exceeds the window itself, so a
+                    // sender that has filled the window always earns credit once the reader drains it.
+                    Assumes.True(owner.localWindowSize.HasValue);
+                    this.ackThreshold = Math.Max(FramePayloadMaxLength, owner.localWindowSize.Value / 8);
                 }
 
                 public override void AdvanceTo(SequencePosition consumed)
@@ -1229,11 +1239,11 @@ private long Consumed(SequencePosition consumed, SequencePosition examined)
 
                     this.bytesProcessed += bytesJustProcessed;
 
-                    // Only send the 'more bytes please' message if we've consumed at least a max frame's worth of data
-                    // or if our reader indicates that more data is required before it will examine any more.
+                    // Only send the 'more bytes please' message if we've examined a large enough fraction of the window
+                    // to be worth a frame, or if our reader indicates that more data is required before it will examine any more.
                     // Or in some cases of very small receiving windows, when the entire window is empty.
                     long result = 0;
-                    if (this.bytesProcessed >= FramePayloadMaxLength || this.bytesProcessed == this.owner.localWindowSize)
+                    if (this.bytesProcessed >= this.ackThreshold || this.bytesProcessed == this.owner.localWindowSize)
                     {
                         result = this.bytesProcessed;
                         this.bytesProcessed = 0;

test/Nerdbank.Streams.Benchmark/BulkTransferBenchmark.cs

@@ -0,0 +1,107 @@
+// 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.Pipelines;
+    using System.Threading.Tasks;
+    using BenchmarkDotNet.Attributes;
+
+    /// <summary>
+    /// Measures throughput when one channel carries a large, one-way payload,
+    /// which is the "large file transfer" scenario reported in
+    /// <see href="https://github.com/dotnet/Nerdbank.Streams/issues/505">issue #505</see>.
+    /// </summary>
+    /// <remarks>
+    /// This scenario is dominated by flow control: the sender must stop and wait whenever it has
+    /// filled the receiver's window, so it is the most sensitive benchmark to window sizing and to
+    /// how often the receiver acknowledges the data it has consumed.
+    /// </remarks>
+    public class BulkTransferBenchmark : MultiplexingStreamBenchmarkBase
+    {
+        private const int ChunkSize = 64 * 1024;
+
+        private static readonly byte[] Chunk = new byte[ChunkSize];
+
+        private int channelCounter;
+
+        /// <summary>
+        /// Gets or sets the number of bytes to transmit in each operation.
+        /// </summary>
+        [Params(32 * 1024 * 1024)]
+        public int TransferSize { get; set; }
+
+        /// <summary>
+        /// Gets or sets the receiving window size to configure for each channel,
+        /// or 0 to use the default.
+        /// </summary>
+        /// <remarks>
+        /// A deliberately small window is included because it stresses the acknowledgement path far
+        /// harder than the default does, and is therefore the case most likely to regress when the
+        /// acknowledgement policy changes.
+        /// </remarks>
+        [Params(0, 100 * 1024)]
+        public int WindowSize { get; set; }
+
+        /// <summary>
+        /// Transmits <see cref="TransferSize"/> bytes across a fresh channel and waits for it all to arrive.
+        /// </summary>
+        /// <returns>A task that tracks the transfer.</returns>
+        [Benchmark]
+        public async Task TransmitBulkData()
+        {
+            (MultiplexingStream.Channel Sender, MultiplexingStream.Channel Receiver) channel =
+                await this.CreateChannelAsync($"bulk{this.channelCounter++}");
+            try
+            {
+                long transferSize = this.TransferSize;
+                Task receiveTask = Task.Run(async delegate
+                {
+                    long bytesReceived = 0;
+                    while (bytesReceived < transferSize)
+                    {
+                        ReadResult readResult = await channel.Receiver.Input.ReadAsync();
+                        if (readResult.Buffer.IsEmpty && readResult.IsCompleted)
+                        {
+                            break;
+                        }
+
+                        bytesReceived += readResult.Buffer.Length;
+                        channel.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(channel.Sender.Output.GetSpan(bytesThisRound));
+                    channel.Sender.Output.Advance(bytesThisRound);
+                    await channel.Sender.Output.FlushAsync();
+                    bytesSent += bytesThisRound;
+                }
+
+                await receiveTask;
+            }
+            finally
+            {
+                channel.Sender.Dispose();
+                channel.Receiver.Dispose();
+            }
+        }
+
+        /// <inheritdoc/>
+        protected override MultiplexingStream.Options CreateOptions()
+        {
+            MultiplexingStream.Options options = base.CreateOptions();
+
+            // Version 1 has no backpressure, so it has no window to configure.
+            if (this.WindowSize > 0 && this.ProtocolMajorVersion > 1)
+            {
+                options.DefaultChannelReceivingWindowSize = this.WindowSize;
+            }
+
+            return options;
+        }
+    }
+}

test/Nerdbank.Streams.Benchmark/ChannelLifetimeBenchmark.cs

@@ -0,0 +1,48 @@
+// 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.Threading.Tasks;
+    using BenchmarkDotNet.Attributes;
+
+    /// <summary>
+    /// Measures the cost of opening and closing many channels that carry little or no data.
+    /// </summary>
+    /// <remarks>
+    /// This is the shape of a large IDE session, where most channels exist to carry occasional small
+    /// messages rather than bulk data. Paired with the memory diagnoser, it is the guard against
+    /// changes that buy throughput by making every channel more expensive to own.
+    /// </remarks>
+    public class ChannelLifetimeBenchmark : MultiplexingStreamBenchmarkBase
+    {
+        private int roundCounter;
+
+        /// <summary>
+        /// Gets or sets the number of channels to open and close in each operation.
+        /// </summary>
+        [Params(500)]
+        public int ChannelCount { get; set; }
+
+        /// <summary>
+        /// Opens <see cref="ChannelCount"/> channels, then closes them all.
+        /// </summary>
+        /// <returns>A task that tracks the work.</returns>
+        [Benchmark]
+        public async Task OpenAndCloseChannels()
+        {
+            int round = this.roundCounter++;
+            var channels = new (MultiplexingStream.Channel Sender, MultiplexingStream.Channel Receiver)[this.ChannelCount];
+            for (int i = 0; i < this.ChannelCount; i++)
+            {
+                channels[i] = await this.CreateChannelAsync($"lifetime{round}-{i}");
+            }
+
+            foreach ((MultiplexingStream.Channel Sender, MultiplexingStream.Channel Receiver) channel in channels)
+            {
+                channel.Sender.Dispose();
+                channel.Receiver.Dispose();
+            }
+        }
+    }
+}

test/Nerdbank.Streams.Benchmark/ContendedChannelsBenchmark.cs

@@ -0,0 +1,100 @@
+// 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.Pipelines;
+    using System.Threading.Tasks;
+    using BenchmarkDotNet.Attributes;
+
+    /// <summary>
+    /// Measures aggregate throughput when several channels transmit at once.
+    /// </summary>
+    /// <remarks>
+    /// All frames on a connection are serialized through a single send path, so this benchmark is the
+    /// one that exposes contention there. A change that helps a single channel but serializes poorly
+    /// across many will show up here and nowhere else.
+    /// </remarks>
+    public class ContendedChannelsBenchmark : MultiplexingStreamBenchmarkBase
+    {
+        private const int ChunkSize = 64 * 1024;
+
+        private static readonly byte[] Chunk = new byte[ChunkSize];
+
+        private int roundCounter;
+
+        /// <summary>
+        /// Gets or sets the number of channels transmitting concurrently.
+        /// </summary>
+        [Params(8)]
+        public int ChannelCount { get; set; }
+
+        /// <summary>
+        /// Gets or sets the total number of bytes to transmit across all channels in each operation.
+        /// </summary>
+        [Params(32 * 1024 * 1024)]
+        public int TotalTransferSize { get; set; }
+
+        /// <summary>
+        /// Transmits <see cref="TotalTransferSize"/> bytes divided evenly across <see cref="ChannelCount"/> channels.
+        /// </summary>
+        /// <returns>A task that tracks the transfers.</returns>
+        [Benchmark]
+        public async Task TransmitOverManyChannels()
+        {
+            int round = this.roundCounter++;
+            var channels = new (MultiplexingStream.Channel Sender, MultiplexingStream.Channel Receiver)[this.ChannelCount];
+            for (int i = 0; i < this.ChannelCount; i++)
+            {
+                channels[i] = await this.CreateChannelAsync($"contended{round}-{i}");
+            }
+
+            try
+            {
+                long perChannel = this.TotalTransferSize / this.ChannelCount;
+                var tasks = new Task[this.ChannelCount * 2];
+                for (int i = 0; i < this.ChannelCount; i++)
+                {
+                    (MultiplexingStream.Channel Sender, MultiplexingStream.Channel Receiver) channel = channels[i];
+                    tasks[i] = Task.Run(async delegate
+                    {
+                        long bytesReceived = 0;
+                        while (bytesReceived < perChannel)
+                        {
+                            ReadResult readResult = await channel.Receiver.Input.ReadAsync();
+                            if (readResult.Buffer.IsEmpty && readResult.IsCompleted)
+                            {
+                                break;
+                            }
+
+                            bytesReceived += readResult.Buffer.Length;
+                            channel.Receiver.Input.AdvanceTo(readResult.Buffer.End);
+                        }
+                    });
+                    tasks[this.ChannelCount + i] = Task.Run(async delegate
+                    {
+                        for (long bytesSent = 0; bytesSent < perChannel;)
+                        {
+                            int bytesThisRound = (int)Math.Min(ChunkSize, perChannel - bytesSent);
+                            Chunk.AsSpan(0, bytesThisRound).CopyTo(channel.Sender.Output.GetSpan(bytesThisRound));
+                            channel.Sender.Output.Advance(bytesThisRound);
+                            await channel.Sender.Output.FlushAsync();
+                            bytesSent += bytesThisRound;
+                        }
+                    });
+                }
+
+                await Task.WhenAll(tasks);
+            }
+            finally
+            {
+                foreach ((MultiplexingStream.Channel Sender, MultiplexingStream.Channel Receiver) channel in channels)
+                {
+                    channel.Sender.Dispose();
+                    channel.Receiver.Dispose();
+                }
+            }
+        }
+    }
+}

test/Nerdbank.Streams.Benchmark/DuplexMessagingBenchmark.cs

@@ -0,0 +1,94 @@
+// 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.IO.Pipelines;
+    using System.Threading.Tasks;
+    using BenchmarkDotNet.Attributes;
+
+    /// <summary>
+    /// Measures round-trip latency for small messages, which is representative of an RPC workload
+    /// such as the one StreamJsonRpc layers on top of this library.
+    /// </summary>
+    /// <remarks>
+    /// Messages here are far too small to ever fill a channel's window, so this benchmark is
+    /// deliberately insensitive to window sizing. Its purpose is to guard the latency of a single
+    /// frame's journey through the send path, which bulk throughput benchmarks hide.
+    /// </remarks>
+    public class DuplexMessagingBenchmark : MultiplexingStreamBenchmarkBase
+    {
+        private const int MessageSize = 128;
+
+        private MultiplexingStream.Channel? requester;
+        private MultiplexingStream.Channel? responder;
+        private Task? echoTask;
+
+        /// <summary>
+        /// Gets or sets the number of round trips to perform in each operation.
+        /// </summary>
+        [Params(1000)]
+        public int RoundTrips { get; set; }
+
+        /// <summary>
+        /// Sends a small message and waits for it to be echoed back, <see cref="RoundTrips"/> times.
+        /// </summary>
+        /// <returns>A task that tracks the round trips.</returns>
+        [Benchmark]
+        public async Task RoundTripSmallMessages()
+        {
+            for (int i = 0; i < this.RoundTrips; i++)
+            {
+                this.requester!.Output.GetSpan(MessageSize).Slice(0, MessageSize).Clear();
+                this.requester.Output.Advance(MessageSize);
+                await this.requester.Output.FlushAsync();
+
+                int bytesRead = 0;
+                while (bytesRead < MessageSize)
+                {
+                    ReadResult readResult = await this.requester.Input.ReadAsync();
+                    if (readResult.Buffer.IsEmpty && readResult.IsCompleted)
+                    {
+                        return;
+                    }
+
+                    bytesRead += (int)readResult.Buffer.Length;
+                    this.requester.Input.AdvanceTo(readResult.Buffer.End);
+                }
+            }
+        }
+
+        /// <summary>
+        /// Opens the channel used for all iterations and starts the echo loop on the far end.
+        /// </summary>
+        /// <returns>A task that tracks the setup.</returns>
+        protected override async Task OnConnectedAsync()
+        {
+            (MultiplexingStream.Channel Sender, MultiplexingStream.Channel Receiver) channel = await this.CreateChannelAsync("duplex");
+            this.requester = channel.Sender;
+            this.responder = channel.Receiver;
+
+            // Echo everything back for as long as the channel is open.
+            this.echoTask = Task.Run(async delegate
+            {
+                while (true)
+                {
+                    ReadResult readResult = await this.responder.Input.ReadAsync();
+                    if (readResult.Buffer.IsEmpty && readResult.IsCompleted)
+                    {
+                        return;
+                    }
+
+                    foreach (System.ReadOnlyMemory<byte> segment in readResult.Buffer)
+                    {
+                        segment.Span.CopyTo(this.responder.Output.GetSpan(segment.Length));
+                        this.responder.Output.Advance(segment.Length);
+                    }
+
+                    this.responder.Input.AdvanceTo(readResult.Buffer.End);
+                    await this.responder.Output.FlushAsync();
+                }
+            });
+        }
+    }
+}