Merge pull request #1124 from dotnet/perf/adaptive-receiving-window

Grow channel receiving windows on demand

Author: AArnott

src/Nerdbank.Streams/MultiplexingStream.Channel.cs

@@ -54,6 +54,30 @@ internal enum ControlCode : byte
             /// Recipients that do not recognize this code ignore it.
             /// </remarks>
             ContentReadingCompleted,
+
+            /// <summary>
+            /// Sent by a channel's receiver to enlarge the receiving window it previously advertised,
+            /// permitting the remote party to have more unacknowledged bytes in flight.
+            /// </summary>
+            /// <remarks>
+            /// The payload carries the new (absolute) window size, which is only ever larger than the prior value.
+            /// This code is additive to the protocol rather than a new version of it: recipients that predate it
+            /// ignore it, which is safe because the window they already agreed to remains valid.
+            /// It is only ever sent in reply to a <see cref="ChannelWindowGrowthRequest"/>, so it is never sent
+            /// to a party that would not understand it.
+            /// </remarks>
+            ChannelWindowAdjust,
+
+            /// <summary>
+            /// Sent by a channel's sender when it has run out of credit and still has data to send,
+            /// asking the receiver to enlarge the window it advertises.
+            /// </summary>
+            /// <remarks>
+            /// This code is additive to the protocol rather than a new version of it. A receiver that predates it
+            /// ignores it and never replies, which costs one small frame per channel and leaves throughput
+            /// exactly where a fixed window would have left it.
+            /// </remarks>
+            ChannelWindowGrowthRequest,
         }
     }
 }

src/Nerdbank.Streams/MultiplexingStream.ControlCode.cs

@@ -96,6 +96,23 @@ internal ValueTask<FlushResult> FlushAsync(CancellationToken cancellationToken)
 
             internal abstract ReadOnlySequence<byte> SerializeContentProcessed(long bytesProcessed);
 
+            /// <summary>
+            /// Serializes the payload of a <see cref="ControlCode.ChannelWindowAdjust"/> frame.
+            /// </summary>
+            /// <param name="windowSize">The new (absolute) size of the receiving window.</param>
+            /// <returns>The serialized payload.</returns>
+            /// <remarks>
+            /// The payload is a single integer, so it shares an encoding with <see cref="SerializeContentProcessed(long)"/>.
+            /// </remarks>
+            internal virtual ReadOnlySequence<byte> SerializeWindowSize(long windowSize) => this.SerializeContentProcessed(windowSize);
+
+            /// <summary>
+            /// Deserializes the payload of a <see cref="ControlCode.ChannelWindowAdjust"/> frame.
+            /// </summary>
+            /// <param name="payload">The payload to deserialize.</param>
+            /// <returns>The new (absolute) size of the receiving window.</returns>
+            internal virtual long DeserializeWindowSize(ReadOnlySequence<byte> payload) => this.DeserializeContentProcessed(payload);
+
             protected static bool IsOdd(ReadOnlySpan<byte> localRandomBuffer, ReadOnlySpan<byte> remoteRandomBuffer)
             {
                 bool? isOdd = null;

src/Nerdbank.Streams/MultiplexingStream.Formatters.cs

@@ -53,6 +53,16 @@ public class Options
             /// </summary>
             private long defaultChannelReceivingWindowSize = RecommendedDefaultChannelReceivingWindowSize;
 
+            /// <summary>
+            /// Backing field for the <see cref="MaxChannelReceivingWindowSize"/> property.
+            /// </summary>
+            private long maxChannelReceivingWindowSize = 16 * RecommendedDefaultChannelReceivingWindowSize;
+
+            /// <summary>
+            /// Backing field for the <see cref="MaxTotalChannelReceivingWindowSize"/> property.
+            /// </summary>
+            private long maxTotalChannelReceivingWindowSize = 64 * RecommendedDefaultChannelReceivingWindowSize;
+
             /// <summary>
             /// Backing field for the <see cref="DefaultChannelTraceSourceFactory"/> property.
             /// </summary>
@@ -91,6 +101,8 @@ public Options(Options copyFrom)
                 Requires.NotNull(copyFrom, nameof(copyFrom));
 
                 this.defaultChannelReceivingWindowSize = copyFrom.defaultChannelReceivingWindowSize;
+                this.maxChannelReceivingWindowSize = copyFrom.maxChannelReceivingWindowSize;
+                this.maxTotalChannelReceivingWindowSize = copyFrom.maxTotalChannelReceivingWindowSize;
                 this.traceSource = copyFrom.traceSource;
                 this.protocolMajorVersion = copyFrom.protocolMajorVersion;
                 this.defaultChannelTraceSourceFactory = copyFrom.defaultChannelTraceSourceFactory;
@@ -152,6 +164,61 @@ public long DefaultChannelReceivingWindowSize
                 }
             }
 
+            /// <summary>
+            /// Gets or sets the largest value that <see cref="DefaultChannelReceivingWindowSize"/> may grow to
+            /// for an individual channel that demonstrates a need for it.
+            /// </summary>
+            /// <value>
+            /// Must be a positive value. The default is 16 times <see cref="DefaultChannelReceivingWindowSize"/>'s default value.
+            /// </value>
+            /// <exception cref="ArgumentOutOfRangeException">Thrown if set to a non-positive value.</exception>
+            /// <remarks>
+            /// <para>
+            /// This value is only used when <see cref="ProtocolMajorVersion"/> is at least 4, which adds the ability
+            /// for a receiver to enlarge a channel's receiving window after the channel has been established.
+            /// A channel's window only grows while its remote sender is actually blocked by it, so channels that never
+            /// saturate their window never consume more than <see cref="DefaultChannelReceivingWindowSize"/>.
+            /// </para>
+            /// <para>
+            /// Growth is additionally bounded across all channels by <see cref="MaxTotalChannelReceivingWindowSize"/>.
+            /// </para>
+            /// </remarks>
+            public long MaxChannelReceivingWindowSize
+            {
+                get => this.maxChannelReceivingWindowSize;
+                set
+                {
+                    Requires.Range(value > 0, nameof(value));
+                    this.ThrowIfFrozen();
+                    this.maxChannelReceivingWindowSize = value;
+                }
+            }
+
+            /// <summary>
+            /// Gets or sets the total number of bytes that all channels on this stream combined may commit to
+            /// receiving window growth beyond their initial <see cref="DefaultChannelReceivingWindowSize"/>.
+            /// </summary>
+            /// <value>
+            /// Must be a non-negative value. The default is 64 times <see cref="DefaultChannelReceivingWindowSize"/>'s default value.
+            /// </value>
+            /// <exception cref="ArgumentOutOfRangeException">Thrown if set to a negative value.</exception>
+            /// <remarks>
+            /// This value is only used when <see cref="ProtocolMajorVersion"/> is at least 4.
+            /// It bounds the worst case memory that automatic window growth may commit for the entire stream,
+            /// so that a stream with many busy channels cannot multiply <see cref="MaxChannelReceivingWindowSize"/>
+            /// by an unbounded number of channels.
+            /// </remarks>
+            public long MaxTotalChannelReceivingWindowSize
+            {
+                get => this.maxTotalChannelReceivingWindowSize;
+                set
+                {
+                    Requires.Range(value >= 0, nameof(value));
+                    this.ThrowIfFrozen();
+                    this.maxTotalChannelReceivingWindowSize = value;
+                }
+            }
+
             /// <summary>
             /// Gets or sets the protocol version to be used.
             /// </summary>
@@ -160,6 +227,7 @@ public long DefaultChannelReceivingWindowSize
             /// 1 is the original and default version.
             /// 2 is a protocol breaking change and adds backpressure support.
             /// 3 is a protocol breaking change that removes the initial handshake so no round-trip to establish the connection is necessary.
+            /// 4 is a protocol breaking change that allows a receiver to enlarge a channel's receiving window after the channel is established.
             /// </remarks>
             public int ProtocolMajorVersion
             {

src/Nerdbank.Streams/MultiplexingStream.Options.cs

@@ -119,12 +119,26 @@ public partial class MultiplexingStream : IDisposableObservable, System.IAsyncDi
         /// </summary>
         private readonly int protocolMajorVersion;
 
+        /// <summary>
+        /// The largest receiving window any single channel may grow to via <see cref="ControlCode.ChannelWindowAdjust"/>.
+        /// </summary>
+        private readonly long maxChannelReceivingWindowSize;
+
         /// <summary>
         /// A value indicating whether any open channels should be faulted (i.e. their <see cref="Channel.Completion"/> task will be faulted)
         /// when the <see cref="MultiplexingStream"/> is disposed.
         /// </summary>
         private readonly bool faultOpenChannelsOnStreamDisposal;
 
+        /// <summary>
+        /// The number of bytes of receiving window growth (beyond each channel's initial window)
+        /// that remain available to be committed across all channels of this stream.
+        /// </summary>
+        /// <remarks>
+        /// Accessed only via <see cref="TryReserveWindowGrowth(long)"/> and <see cref="ReleaseWindowGrowth(long)"/>.
+        /// </remarks>
+        private long remainingWindowGrowthBudget;
+
         /// <summary>
         /// The last number assigned to a channel.
         /// Each use of this should increment by two, if <see cref="isOdd"/> has a value.
@@ -162,6 +176,8 @@ private MultiplexingStream(Formatter formatter, bool? isOdd, Options options)
 
             this.DefaultChannelReceivingWindowSize = options.DefaultChannelReceivingWindowSize;
             this.protocolMajorVersion = options.ProtocolMajorVersion;
+            this.maxChannelReceivingWindowSize = Math.Max(options.MaxChannelReceivingWindowSize, options.DefaultChannelReceivingWindowSize);
+            this.remainingWindowGrowthBudget = options.MaxTotalChannelReceivingWindowSize;
 
             // Set up seed channels
             for (int i = 0; i < options.SeededChannels.Count; i++)
@@ -866,6 +882,12 @@ private async Task ReadStreamAsync()
                         case ControlCode.ContentProcessed:
                             this.OnContentProcessed(header, frame.Value.Payload);
                             break;
+                        case ControlCode.ChannelWindowAdjust:
+                            this.OnChannelWindowAdjust(header, frame.Value.Payload);
+                            break;
+                        case ControlCode.ChannelWindowGrowthRequest:
+                            this.OnChannelWindowGrowthRequest(header);
+                            break;
                         case ControlCode.ContentWritingCompleted:
                             this.OnContentWritingCompleted(header.RequiredChannelId);
                             break;
@@ -1086,6 +1108,72 @@ private void OnContentProcessed(FrameHeader header, ReadOnlySequence<byte> paylo
             channel.OnContentProcessed(bytesProcessed);
         }
 
+        private void OnChannelWindowAdjust(FrameHeader header, ReadOnlySequence<byte> payloadBuffer)
+        {
+            Channel? channel;
+            lock (this.syncObject)
+            {
+                this.openChannels.TryGetValue(header.RequiredChannelId, out channel);
+            }
+
+            if (channel is null)
+            {
+                // The channel closed concurrently with the remote party's decision to enlarge its window.
+                // Dropping the adjustment is safe: there is nothing left to send.
+                return;
+            }
+
+            long newWindowSize = this.formatter.DeserializeWindowSize(payloadBuffer);
+            channel.OnWindowAdjust(newWindowSize);
+        }
+
+        private void OnChannelWindowGrowthRequest(FrameHeader header)
+        {
+            Channel? channel;
+            lock (this.syncObject)
+            {
+                this.openChannels.TryGetValue(header.RequiredChannelId, out channel);
+            }
+
+            // A request for a channel that has since closed needs no answer.
+            channel?.OnWindowGrowthRequested();
+        }
+
+        /// <summary>
+        /// Attempts to claim a portion of this stream's total budget for receiving window growth.
+        /// </summary>
+        /// <param name="bytes">The number of additional bytes of window the caller wants to commit to.</param>
+        /// <returns><see langword="true"/> if the budget had room and has been debited; otherwise <see langword="false"/>.</returns>
+        private bool TryReserveWindowGrowth(long bytes)
+        {
+            long remaining = Volatile.Read(ref this.remainingWindowGrowthBudget);
+            while (remaining >= bytes)
+            {
+                long candidate = Interlocked.CompareExchange(ref this.remainingWindowGrowthBudget, remaining - bytes, remaining);
+                if (candidate == remaining)
+                {
+                    return true;
+                }
+
+                remaining = candidate;
+            }
+
+            return false;
+        }
+
+        /// <summary>
+        /// Returns window growth budget previously claimed by <see cref="TryReserveWindowGrowth(long)"/>,
+        /// so that other channels may use it after a channel closes.
+        /// </summary>
+        /// <param name="bytes">The number of bytes to return to the budget.</param>
+        private void ReleaseWindowGrowth(long bytes)
+        {
+            if (bytes > 0)
+            {
+                Interlocked.Add(ref this.remainingWindowGrowthBudget, bytes);
+            }
+        }
+
         private void OnOffer(QualifiedChannelId channelId, ReadOnlySequence<byte> payloadBuffer)
         {
             Channel.OfferParameters? offerParameters = this.formatter.DeserializeOfferParameters(payloadBuffer);

src/Nerdbank.Streams/MultiplexingStream.cs

@@ -0,0 +1,116 @@
+// 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.Threading.Tasks;
+    using BenchmarkDotNet.Attributes;
+
+    /// <summary>
+    /// Measures bulk transfer over a transport with artificial latency.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The other benchmarks all run over loopback, where a round trip is essentially free. That makes
+    /// them blind to the central trade-off in the receive window's flow control: returning credit less
+    /// often costs fewer frames but makes the sender wait a full round trip when it does run out.
+    /// On loopback the second half of that trade-off is invisible, so tuning against loopback alone
+    /// will happily pick a setting that behaves badly on a real network.
+    /// </para>
+    /// <para>
+    /// Throughput here is expected to approach the window divided by the round trip time whenever the
+    /// window is the binding constraint, so the interesting signal is how much larger a window must be
+    /// as latency grows.
+    /// </para>
+    /// </remarks>
+    public class LatencyBulkTransferBenchmark : MultiplexingStreamBenchmarkBase
+    {
+        private byte[] payload = null!;
+
+        /// <summary>
+        /// Gets or sets the one-way delay applied to the transport, in milliseconds.
+        /// </summary>
+        /// <remarks>
+        /// The round trip time is twice this. Zero disables the wrapper entirely, giving a
+        /// direct comparison against the plain loopback benchmarks. The platform timer cannot
+        /// faithfully reproduce delays below about a millisecond, so no smaller value is offered.
+        /// </remarks>
+        [Params(0, 1, 8)]
+        public int OneWayLatencyMs { get; set; }
+
+        /// <summary>
+        /// Gets or sets the receiving window size to configure, or 0 to use the default.
+        /// </summary>
+        [Params(0, 4 * 1024 * 1024)]
+        public int WindowSize { get; set; }
+
+        /// <summary>
+        /// Gets or sets the number of bytes to transfer.
+        /// </summary>
+        /// <remarks>
+        /// This is much smaller than the loopback benchmarks use, because a latent connection
+        /// takes far longer to move the same volume.
+        /// </remarks>
+        [Params(4 * 1024 * 1024)]
+        public int TransferSize { get; set; }
+
+        /// <summary>
+        /// Transfers <see cref="TransferSize"/> bytes over a single channel.
+        /// </summary>
+        /// <returns>A task that tracks the transfer.</returns>
+        [Benchmark]
+        public async Task TransmitBulkData()
+        {
+            (MultiplexingStream.Channel sender, MultiplexingStream.Channel receiver) = await this.CreateChannelAsync(Guid.NewGuid().ToString("n"));
+
+            Task writeTask = Task.Run(async delegate
+            {
+                await sender.Output.WriteAsync(this.payload);
+                await sender.Output.CompleteAsync();
+            });
+
+            long bytesRead = 0;
+            while (bytesRead < this.TransferSize)
+            {
+                ReadResult readResult = await receiver.Input.ReadAsync();
+                if (readResult.Buffer.IsEmpty && readResult.IsCompleted)
+                {
+                    break;
+                }
+
+                bytesRead += readResult.Buffer.Length;
+                receiver.Input.AdvanceTo(readResult.Buffer.End);
+            }
+
+            await writeTask;
+            sender.Dispose();
+            receiver.Dispose();
+        }
+
+        /// <inheritdoc/>
+        protected override MultiplexingStream.Options CreateOptions()
+        {
+            MultiplexingStream.Options options = base.CreateOptions();
+            if (this.WindowSize > 0)
+            {
+                options.DefaultChannelReceivingWindowSize = this.WindowSize;
+            }
+
+            return options;
+        }
+
+        /// <inheritdoc/>
+        protected override Stream WrapTransport(Stream transport)
+            => this.OneWayLatencyMs == 0 ? transport : new LatencyStream(transport, TimeSpan.FromMilliseconds(this.OneWayLatencyMs));
+
+        /// <inheritdoc/>
+        protected override Task OnConnectedAsync()
+        {
+            this.payload = new byte[this.TransferSize];
+            return Task.CompletedTask;
+        }
+    }
+}