Reduce thread consumption for sync HTTP/2 reads (#32917)

The response Stream returned from SocketsHttpHandler.SendAsync supports all of the various forms of read operations, including synchronous reads.

For HTTP/1.x, we're able to implement these as sync all the way down to synchronous calls on the underlying Socket, as every request/response owns its own connection (at least for the duration of the operation; the connection may be returned to the pool for subsequent reuse after).

For HTTP/2, however, a single connection is shared across all requests multiplexed onto it, and as such, reads on those individual responses need to coordinate, which can mean waiting one's turn.  Such waiting for asynchronous operations is implemented via async/await and a ManualResetValueTaskSource-based implementation.  For the sync operations, though, they're implemented as simply blocking waiting on the async equivalent, aka sync-over-async.  This not only blocks the calling thread, but because the MRVTS-based implementation specifies RunContinuationsAsynchronously==true (which is important for reliability with how the async code paths are structured) it means that, in order to unblock this waiter, another thread pool thread needs to be available to run the small continuation that will in turn unblock the sync primitive being blocked on.

This PR tweaks the use of MRVTS so that when we perform a synchronous read, we set MRVTS.RunContinuationsAsynchronously to false.  Since the only action that will be taken as part of the continuation is to set a sync primitive, we don't need to worry about arbitrary code running as part of the completing call stack.  This in turn means that in the common case, sync reads will still end up blocking the calling thread _but_ won't require an additional thread pool thread to unblock it, other than for the task running the main connection loop.

Author: stephentoub

src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http2Stream.cs

@@ -836,7 +836,7 @@ public async Task ReadResponseHeadersAsync(CancellationToken cancellationToken)
                     (wait, emptyResponse) = TryEnsureHeaders();
                     if (wait)
                     {
-                        await GetWaiterTask(cancellationToken).ConfigureAwait(false);
+                        await WaitForDataAsync(cancellationToken).ConfigureAwait(false);
 
                         (wait, emptyResponse) = TryEnsureHeaders();
                         Debug.Assert(!wait);
@@ -893,7 +893,7 @@ private void ExtendWindow(int amount)
                 _connection.LogExceptions(_connection.SendWindowUpdateAsync(_streamId, windowUpdateSize));
             }
 
-            private (bool wait, int bytesRead) TryReadFromBuffer(Span<byte> buffer)
+            private (bool wait, int bytesRead) TryReadFromBuffer(Span<byte> buffer, bool partOfSyncRead = false)
             {
                 Debug.Assert(buffer.Length > 0);
 
@@ -920,6 +920,7 @@ private void ExtendWindow(int amount)
                     Debug.Assert(!_hasWaiter);
                     _hasWaiter = true;
                     _waitSource.Reset();
+                    _waitSource.RunContinuationsAsynchronously = !partOfSyncRead;
                     return (true, 0);
                 }
             }
@@ -931,13 +932,13 @@ public int ReadData(Span<byte> buffer, HttpResponseMessage responseMessage)
                     return 0;
                 }
 
-                (bool wait, int bytesRead) = TryReadFromBuffer(buffer);
+                (bool wait, int bytesRead) = TryReadFromBuffer(buffer, partOfSyncRead: true);
                 if (wait)
                 {
                     // Synchronously block waiting for data to be produced.
                     Debug.Assert(bytesRead == 0);
-                    GetWaiterTask(default).AsTask().GetAwaiter().GetResult();
-                    (wait, bytesRead) = TryReadFromBuffer(buffer);
+                    WaitForData();
+                    (wait, bytesRead) = TryReadFromBuffer(buffer, partOfSyncRead: true);
                     Debug.Assert(!wait);
                 }
 
@@ -965,7 +966,7 @@ public async ValueTask<int> ReadDataAsync(Memory<byte> buffer, HttpResponseMessa
                 if (wait)
                 {
                     Debug.Assert(bytesRead == 0);
-                    await GetWaiterTask(cancellationToken).ConfigureAwait(false);
+                    await WaitForDataAsync(cancellationToken).ConfigureAwait(false);
                     (wait, bytesRead) = TryReadFromBuffer(buffer.Span);
                     Debug.Assert(!wait);
                 }
@@ -983,6 +984,42 @@ public async ValueTask<int> ReadDataAsync(Memory<byte> buffer, HttpResponseMessa
                 return bytesRead;
             }
 
+            public void CopyTo(HttpResponseMessage responseMessage, Stream destination, int bufferSize)
+            {
+                byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
+                try
+                {
+                    // Generallly the same logic as in ReadData, but wrapped in a loop where every read segment is written to the destination.
+                    while (true)
+                    {
+                        (bool wait, int bytesRead) = TryReadFromBuffer(buffer, partOfSyncRead: true);
+                        if (wait)
+                        {
+                            Debug.Assert(bytesRead == 0);
+                            WaitForData();
+                            (wait, bytesRead) = TryReadFromBuffer(buffer, partOfSyncRead: true);
+                            Debug.Assert(!wait);
+                        }
+
+                        if (bytesRead != 0)
+                        {
+                            ExtendWindow(bytesRead);
+                            destination.Write(new ReadOnlySpan<byte>(buffer, 0, bytesRead));
+                        }
+                        else
+                        {
+                            // We've hit EOF.  Pull in from the Http2Stream any trailers that were temporarily stored there.
+                            CopyTrailersToResponseMessage(responseMessage);
+                            return;
+                        }
+                    }
+                }
+                finally
+                {
+                    ArrayPool<byte>.Shared.Return(buffer);
+                }
+            }
+
             public async Task CopyToAsync(HttpResponseMessage responseMessage, Stream destination, int bufferSize, CancellationToken cancellationToken)
             {
                 byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
@@ -995,7 +1032,7 @@ public async Task CopyToAsync(HttpResponseMessage responseMessage, Stream destin
                         if (wait)
                         {
                             Debug.Assert(bytesRead == 0);
-                            await GetWaiterTask(cancellationToken).ConfigureAwait(false);
+                            await WaitForDataAsync(cancellationToken).ConfigureAwait(false);
                             (wait, bytesRead) = TryReadFromBuffer(buffer);
                             Debug.Assert(!wait);
                         }
@@ -1122,13 +1159,27 @@ private void CloseResponseBody()
             ValueTaskSourceStatus IValueTaskSource.GetStatus(short token) => _waitSource.GetStatus(token);
             void IValueTaskSource.OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags) => _waitSource.OnCompleted(continuation, state, token, flags);
             void IValueTaskSource.GetResult(short token) => _waitSource.GetResult(token);
-            private ValueTask GetWaiterTask(CancellationToken cancellationToken)
+
+            private void WaitForData()
             {
+                // See comments in WaitAsync.
+                Debug.Assert(!_waitSource.RunContinuationsAsynchronously);
+                new ValueTask(this, _waitSource.Version).AsTask().GetAwaiter().GetResult();
+            }
+
+            private ValueTask WaitForDataAsync(CancellationToken cancellationToken)
+            {
+                Debug.Assert(_waitSource.RunContinuationsAsynchronously);
+
                 // No locking is required here to access _waitSource.  To be here, we've already updated _hasWaiter (while holding the lock)
                 // to indicate that we would be creating this waiter, and at that point the only code that could be await'ing _waitSource or
                 // Reset'ing it is this code here.  It's possible for this to race with the _waitSource being completed, but that's ok and is
                 // handled by _waitSource as one of its primary purposes.  We can't assert _hasWaiter here, though, as once we released the
                 // lock, a producer could have seen _hasWaiter as true and both set it to false and signaled _waitSource.
+                if (!cancellationToken.CanBeCanceled)
+                {
+                    return new ValueTask(this, _waitSource.Version);
+                }
 
                 // With HttpClient, the supplied cancellation token will always be cancelable, as HttpClient supplies a token that
                 // will have cancellation requested if CancelPendingRequests is called (or when a non-infinite Timeout expires).
@@ -1137,36 +1188,34 @@ private ValueTask GetWaiterTask(CancellationToken cancellationToken)
                 // this pay-for-play: if the token isn't cancelable, return a ValueTask wrapping this object directly, and only
                 // if it is cancelable, then register for the cancellation callback, allocate a task for the asynchronously
                 // completing case, etc.
-                return cancellationToken.CanBeCanceled ?
-                    GetCancelableWaiterTask(cancellationToken) :
-                    new ValueTask(this, _waitSource.Version);
-            }
+                return GetCancelableWaiterTask(cancellationToken);
 
-            private async ValueTask GetCancelableWaiterTask(CancellationToken cancellationToken)
-            {
-                using (cancellationToken.UnsafeRegister(s =>
+                async ValueTask GetCancelableWaiterTask(CancellationToken cancellationToken)
                 {
-                    var thisRef = (Http2Stream)s;
-
-                    bool signalWaiter;
-                    Debug.Assert(!Monitor.IsEntered(thisRef.SyncObject));
-                    lock (thisRef.SyncObject)
+                    using (cancellationToken.UnsafeRegister(s =>
                     {
-                        signalWaiter = thisRef._hasWaiter;
-                        thisRef._hasWaiter = false;
-                    }
+                        var thisRef = (Http2Stream)s;
 
-                    if (signalWaiter)
+                        bool signalWaiter;
+                        Debug.Assert(!Monitor.IsEntered(thisRef.SyncObject));
+                        lock (thisRef.SyncObject)
+                        {
+                            signalWaiter = thisRef._hasWaiter;
+                            thisRef._hasWaiter = false;
+                        }
+
+                        if (signalWaiter)
+                        {
+                            // Wake up the wait.  It will then immediately check whether cancellation was requested and throw if it was.
+                            thisRef._waitSource.SetResult(true);
+                        }
+                    }, this))
                     {
-                        // Wake up the wait.  It will then immediately check whether cancellation was requested and throw if it was.
-                        thisRef._waitSource.SetResult(true);
+                        await new ValueTask(this, _waitSource.Version).ConfigureAwait(false);
                     }
-                }, this))
-                {
-                    await new ValueTask(this, _waitSource.Version).ConfigureAwait(false);
-                }
 
-                CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
+                    CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
+                }
             }
 
             public void Trace(string message, [CallerMemberName] string memberName = null) =>
@@ -1262,6 +1311,13 @@ public override ValueTask<int> ReadAsync(Memory<byte> destination, CancellationT
                     return http2Stream.ReadDataAsync(destination, _responseMessage, cancellationToken);
                 }
 
+                public override void CopyTo(Stream destination, int bufferSize)
+                {
+                    StreamHelpers.ValidateCopyToArgs(this, destination, bufferSize);
+                    Http2Stream http2Stream = _http2Stream ?? throw ExceptionDispatchInfo.SetCurrentStackTrace(new ObjectDisposedException(nameof(Http2ReadStream)));
+                    http2Stream.CopyTo(_responseMessage, destination, bufferSize);
+                }
+
                 public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
                 {
                     StreamHelpers.ValidateCopyToArgs(this, destination, bufferSize);