<commit_before>﻿// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.

using System.Buffers.Text;
using System.Collections.Generic;
using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.Net.Http.Headers;
using System.Net.Security;
using System.Net.Sockets;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace System.Net.Http
{
    internal partial class HttpConnection : IDisposable
    {
        /// <summary>Default size of the read buffer used for the connection.</summary>
        private const int InitialReadBufferSize =
#if DEBUG
            10;
#else
            4096;
#endif
        /// <summary>Default size of the write buffer used for the connection.</summary>
        private const int InitialWriteBufferSize = InitialReadBufferSize;
        /// <summary>
        /// Size after which we'll close the connection rather than send the payload in response
        /// to final error status code sent by the server when using Expect: 100-continue.
        /// </summary>
        private const int Expect100ErrorSendThreshold = 1024;

        private static readonly byte[] s_contentLength0NewlineAsciiBytes = Encoding.ASCII.GetBytes("Content-Length: 0\r\n");
        private static readonly byte[] s_spaceHttp10NewlineAsciiBytes = Encoding.ASCII.GetBytes(" HTTP/1.0\r\n");
        private static readonly byte[] s_spaceHttp11NewlineAsciiBytes = Encoding.ASCII.GetBytes(" HTTP/1.1\r\n");
        private static readonly byte[] s_httpSchemeAndDelimiter = Encoding.ASCII.GetBytes(Uri.UriSchemeHttp + Uri.SchemeDelimiter);
        private static readonly byte[] s_http1DotBytes = Encoding.ASCII.GetBytes("HTTP/1.");
        private static readonly ulong s_http10Bytes = BitConverter.ToUInt64(Encoding.ASCII.GetBytes("HTTP/1.0"));
        private static readonly ulong s_http11Bytes = BitConverter.ToUInt64(Encoding.ASCII.GetBytes("HTTP/1.1"));

        private readonly HttpConnectionPool _pool;
        private readonly Socket _socket; // used for polling; _stream should be used for all reading/writing. _stream owns disposal.
        private readonly Stream _stream;
        private readonly TransportContext _transportContext;
        private readonly bool _usingProxy;
        private readonly WeakReference<HttpConnection> _weakThisRef;

        private HttpRequestMessage _currentRequest;
        private readonly byte[] _writeBuffer;
        private int _writeOffset;
        private int _allowedReadLineBytes;

        private ValueTask<int>? _readAheadTask;
        private byte[] _readBuffer;
        private int _readOffset;
        private int _readLength;

        private bool _inUse;
        private bool _canRetry;
        private bool _connectionClose; // Connection: close was seen on last response
        private int _disposed; // 1 yes, 0 no

        public HttpConnection(
            HttpConnectionPool pool,
            Socket socket,
            Stream stream,
            TransportContext transportContext)
        {
            Debug.Assert(pool != null);
            Debug.Assert(stream != null);

            _pool = pool;
            _socket = socket; // may be null in cases where we couldn't easily get the underlying socket
            _stream = stream;
            _transportContext = transportContext;
            _usingProxy = pool.UsingProxy;

            _writeBuffer = new byte[InitialWriteBufferSize];
            _readBuffer = new byte[InitialReadBufferSize];

            _weakThisRef = new WeakReference<HttpConnection>(this);

            if (NetEventSource.IsEnabled)
            {
                if (pool.IsSecure)
                {
                    var sslStream = (SslStream)_stream;
                    Trace(
                        $"Secure connection created to {pool}. " +
                        $"SslProtocol:{sslStream.SslProtocol}, " +
                        $"CipherAlgorithm:{sslStream.CipherAlgorithm}, CipherStrength:{sslStream.CipherStrength}, " +
                        $"HashAlgorithm:{sslStream.HashAlgorithm}, HashStrength:{sslStream.HashStrength}, " +
                        $"KeyExchangeAlgorithm:{sslStream.KeyExchangeAlgorithm}, KeyExchangeStrength:{sslStream.KeyExchangeStrength}, " +
                        $"LocalCert:{sslStream.LocalCertificate}, RemoteCert:{sslStream.RemoteCertificate}");
                }
                else
                {
                    Trace($"Connection created to {pool}.");
                }
            }
        }

        public void Dispose() => Dispose(disposing: true);

        protected void Dispose(bool disposing)
        {
            // Ensure we're only disposed once.  Dispose could be called concurrently, for example,
            // if the request and the response were running concurrently and both incurred an exception.
            if (Interlocked.Exchange(ref _disposed, 1) == 0)
            {
                if (NetEventSource.IsEnabled) Trace("Connection closing.");
                _pool.DecrementConnectionCount();
                if (disposing)
                {
                    GC.SuppressFinalize(this);
                    _stream.Dispose();

                    // Eat any exceptions from the read-ahead task.  We don't need to log, as we expect
                    // failures from this task due to closing the connection while a read is in progress.
                    if (_readAheadTask != null)
                    {
                        ValueTask<int> t = _readAheadTask.GetValueOrDefault();
                        if (t.IsCompleted)
                        {
                            if (!t.IsCompletedSuccessfully)
                            {
                                Exception ignored = t.AsTask().Exception; // accessing Exception prop is sufficient to suppress unobserved exception events
                            }
                        }
                        else
                        {
                            t.AsTask().ContinueWith(p =>
                            {
                                Exception ignored = p.Exception;
                            }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);
                        }
                    }
                }
            }
        }

        /// <summary>Do a non-blocking poll to see whether the connection has data available or has been closed.</summary>
        /// <remarks>If we don't have direct access to the underlying socket, we instead use a read-ahead task.</remarks>
        public bool PollRead() => _socket != null ?
            _socket.Poll(0, SelectMode.SelectRead) :
            EnsureReadAheadAndPollRead();

        /// <summary>
        /// Issues a read-ahead on the connection, which will serve both as the first read on the
        /// response as well as a polling indication of whether the connection is usable.
        /// </summary>
        /// <returns>true if there's data available on the connection or it's been closed; otherwise, false.</returns>
        public bool EnsureReadAheadAndPollRead()
        {
            try
            {
                Debug.Assert(_readAheadTask == null || _socket == null, "Should only already have a read-ahead task if we don't have a socket to poll");
                if (_readAheadTask == null)
                {
                    _readAheadTask = _stream.ReadAsync(new Memory<byte>(_readBuffer));
                }
            }
            catch (Exception error)
            {
                // If reading throws, eat the error and don't pool the connection.
                if (NetEventSource.IsEnabled) Trace($"Error performing read ahead: {error}");
                Dispose();
                _readAheadTask = new ValueTask<int>(0);
            }

            return _readAheadTask.Value.IsCompleted; // equivalent to polling
        }

        public bool IsNewConnection
        {
            get
            {
                // This is only valid when we are not actually processing a request.
                Debug.Assert(_currentRequest == null);
                return (_readAheadTask == null);
            }
        }

        public bool CanRetry
        {
            get
            {
                // Should only be called when we have been disposed.
                Debug.Assert(_disposed != 0);
                return _canRetry;
            }
        }

        public DateTimeOffset CreationTime { get; } = DateTimeOffset.UtcNow;

        public TransportContext TransportContext => _transportContext;

        public bool UsingProxy => _usingProxy;

        private int ReadBufferSize => _readBuffer.Length;

        private ReadOnlyMemory<byte> RemainingBuffer => new ReadOnlyMemory<byte>(_readBuffer, _readOffset, _readLength - _readOffset);

        private void ConsumeFromRemainingBuffer(int bytesToConsume)
        {
            Debug.Assert(bytesToConsume <= _readLength - _readOffset, $"{bytesToConsume} > {_readLength} - {_readOffset}");
            _readOffset += bytesToConsume;
        }

        private async Task WriteHeadersAsync(HttpHeaders headers, string cookiesFromContainer)
        {
            foreach (KeyValuePair<HeaderDescriptor, string[]> header in headers.GetHeaderDescriptorsAndValues())
            {
                if (header.Key.KnownHeader != null)
                {
                    await WriteBytesAsync(header.Key.KnownHeader.AsciiBytesWithColonSpace).ConfigureAwait(false);
                }
                else
                {
                    await WriteAsciiStringAsync(header.Key.Name).ConfigureAwait(false);
                    await WriteTwoBytesAsync((byte)':', (byte)' ').ConfigureAwait(false);
                }

                Debug.Assert(header.Value.Length > 0, "No values for header??");
                if (header.Value.Length > 0)
                {
                    await WriteStringAsync(header.Value[0]).ConfigureAwait(false);

                    if (cookiesFromContainer != null && header.Key.KnownHeader == KnownHeaders.Cookie)
                    {
                        await WriteTwoBytesAsync((byte)';', (byte)' ').ConfigureAwait(false);
                        await WriteStringAsync(cookiesFromContainer).ConfigureAwait(false);

                        cookiesFromContainer = null;
                    }

                    for (int i = 1; i < header.Value.Length; i++)
                    {
                        await WriteTwoBytesAsync((byte)',', (byte)' ').ConfigureAwait(false);
                        await WriteStringAsync(header.Value[i]).ConfigureAwait(false);
                    }
                }

                await WriteTwoBytesAsync((byte)'\r', (byte)'\n').ConfigureAwait(false);
            }

            if (cookiesFromContainer != null)
            {
                await WriteAsciiStringAsync(HttpKnownHeaderNames.Cookie).ConfigureAwait(false);
                await WriteTwoBytesAsync((byte)':', (byte)' ').ConfigureAwait(false);
                await WriteAsciiStringAsync(cookiesFromContainer).ConfigureAwait(false);
                await WriteTwoBytesAsync((byte)'\r', (byte)'\n').ConfigureAwait(false);
            }
        }

        private async Task WriteHostHeaderAsync(Uri uri)
        {
            await WriteBytesAsync(KnownHeaders.Host.AsciiBytesWithColonSpace).ConfigureAwait(false);

            if (_pool.HostHeaderValueBytes != null)
            {
                Debug.Assert(!_pool.UsingProxy);
                await WriteBytesAsync(_pool.HostHeaderValueBytes).ConfigureAwait(false);
            }
            else
            {
                Debug.Assert(_pool.UsingProxy);
                await WriteAsciiStringAsync(uri.IdnHost).ConfigureAwait(false);

                if (!uri.IsDefaultPort)
                {
                    await WriteByteAsync((byte)':').ConfigureAwait(false);
                    await WriteDecimalInt32Async(uri.Port).ConfigureAwait(false);
                }
            }

            await WriteTwoBytesAsync((byte)'\r', (byte)'\n').ConfigureAwait(false);
        }

        private Task WriteDecimalInt32Async(int value)
        {
            // Try to format into our output buffer directly.
            if (Utf8Formatter.TryFormat(value, new Span<byte>(_writeBuffer, _writeOffset, _writeBuffer.Length - _writeOffset), out int bytesWritten))
            {
                _writeOffset += bytesWritten;
                return Task.CompletedTask;
            }

            // If we don't have enough room, do it the slow way.
            return WriteAsciiStringAsync(value.ToString());
        }

        private Task WriteHexInt32Async(int value)
        {
            // Try to format into our output buffer directly.
            if (Utf8Formatter.TryFormat(value, new Span<byte>(_writeBuffer, _writeOffset, _writeBuffer.Length - _writeOffset), out int bytesWritten, 'X'))
            {
                _writeOffset += bytesWritten;
                return Task.CompletedTask;
            }

            // If we don't have enough room, do it the slow way.
            return WriteAsciiStringAsync(value.ToString("X", CultureInfo.InvariantCulture));
        }

        public async Task<HttpResponseMessage> SendAsyncCore(HttpRequestMessage request, CancellationToken cancellationToken)
        {
            TaskCompletionSource<bool> allowExpect100ToContinue = null;
            Debug.Assert(_currentRequest == null, $"Expected null {nameof(_currentRequest)}.");
            Debug.Assert(RemainingBuffer.Length == 0, "Unexpected data in read buffer");

            _currentRequest = request;
            bool isConnectMethod = (request.Method == HttpMethod.Connect);

            Debug.Assert(!_canRetry);
            _canRetry = true;

            // Send the request.
            if (NetEventSource.IsEnabled) Trace($"Sending request: {request}");
            CancellationTokenRegistration cancellationRegistration = RegisterCancellation(cancellationToken);
            try
            {
                // Write request line
                await WriteStringAsync(request.Method.Method).ConfigureAwait(false);
                await WriteByteAsync((byte)' ').ConfigureAwait(false);

                if (isConnectMethod)
                {
                    // RFC 7231 #section-4.3.6.
                    // Write only CONNECT foo.com:345 HTTP/1.1
                    if (!request.HasHeaders || request.Headers.Host == null)
                    {
                        throw new HttpRequestException(SR.net_http_request_no_host);
                    }
                    await WriteAsciiStringAsync(request.Headers.Host).ConfigureAwait(false);
                }
                else
                {
                    if (_usingProxy)
                    {
                        // Proxied requests contain full URL
                        Debug.Assert(request.RequestUri.Scheme == Uri.UriSchemeHttp);
                        await WriteBytesAsync(s_httpSchemeAndDelimiter).ConfigureAwait(false);
                        await WriteAsciiStringAsync(request.RequestUri.IdnHost).ConfigureAwait(false);
                    }
                    await WriteStringAsync(request.RequestUri.GetComponents(UriComponents.PathAndQuery | UriComponents.Fragment, UriFormat.UriEscaped)).ConfigureAwait(false);
                }

                // Fall back to 1.1 for all versions other than 1.0
                Debug.Assert(request.Version.Major >= 0 && request.Version.Minor >= 0); // guaranteed by Version class
                bool isHttp10 = request.Version.Minor == 0 && request.Version.Major == 1;
                await WriteBytesAsync(isHttp10 ? s_spaceHttp10NewlineAsciiBytes : s_spaceHttp11NewlineAsciiBytes).ConfigureAwait(false);

                // Determine cookies to send
                string cookiesFromContainer = null;
                if (_pool.Settings._useCookies)
                {
                    cookiesFromContainer = _pool.Settings._cookieContainer.GetCookieHeader(request.RequestUri);
                    if (cookiesFromContainer == "")
                    {
                        cookiesFromContainer = null;
                    }
                }

                // Write request headers
                if (request.HasHeaders || cookiesFromContainer != null)
                {
                    await WriteHeadersAsync(request.Headers, cookiesFromContainer).ConfigureAwait(false);
                }

                if (request.Content == null)
                {
                    // Write out Content-Length: 0 header to indicate no body,
                    // unless this is a method that never has a body.
                    if (request.Method != HttpMethod.Get && request.Method != HttpMethod.Head && !isConnectMethod)
                    {
                        await WriteBytesAsync(s_contentLength0NewlineAsciiBytes).ConfigureAwait(false);
                    }
                }
                else
                {
                    // Write content headers
                    await WriteHeadersAsync(request.Content.Headers, cookiesFromContainer: null).ConfigureAwait(false);
                }

                // Write special additional headers.  If a host isn't in the headers list, then a Host header
                // wasn't sent, so as it's required by HTTP 1.1 spec, send one based on the Request Uri.
                if (!request.HasHeaders || request.Headers.Host == null)
                {
                    await WriteHostHeaderAsync(request.RequestUri).ConfigureAwait(false);
                }

                // CRLF for end of headers.
                await WriteTwoBytesAsync((byte)'\r', (byte)'\n').ConfigureAwait(false);

                Task sendRequestContentTask = null;
                if (request.Content == null)
                {
                    // We have nothing more to send, so flush out any headers we haven't yet sent.
                    await FlushAsync().ConfigureAwait(false);
                }
                else
                {
                    // Send the body if there is one.  We prefer to serialize the sending of the content before
                    // we try to receive any response, but if ExpectContinue has been set, we allow the sending
                    // to run concurrently until we receive the final status line, at which point we wait for it.
                    if (!request.HasHeaders || request.Headers.ExpectContinue != true)
                    {
                        await SendRequestContentAsync(request, CreateRequestContentStream(request), cancellationToken).ConfigureAwait(false);
                    }
                    else
                    {
                        // We're sending an Expect: 100-continue header. We need to flush headers so that the server receives
                        // all of them, and we need to do so before initiating the send, as once we do that, it effectively
                        // owns the right to write, and we don't want to concurrently be accessing the write buffer.
                        await FlushAsync().ConfigureAwait(false);

                        // Create a TCS we'll use to block the request content from being sent, and create a timer that's used
                        // as a fail-safe to unblock the request content if we don't hear back from the server in a timely manner.
                        // Then kick off the request.  The TCS' result indicates whether content should be sent or not.
                        allowExpect100ToContinue = new TaskCompletionSource<bool>();
                        var expect100Timer = new Timer(
                            s => ((TaskCompletionSource<bool>)s).TrySetResult(true),
                            allowExpect100ToContinue, _pool.Settings._expect100ContinueTimeout, Timeout.InfiniteTimeSpan);
                        sendRequestContentTask = SendRequestContentWithExpect100ContinueAsync(
                            request, allowExpect100ToContinue.Task, CreateRequestContentStream(request), expect100Timer, cancellationToken);
                    }
                }

                // Start to read response.
                _allowedReadLineBytes = _pool.Settings._maxResponseHeadersLength * 1024;

                // We should not have any buffered data here; if there was, it should have been treated as an error
                // by the previous request handling.  (Note we do not support HTTP pipelining.)
                Debug.Assert(_readOffset == _readLength);

                // When the connection was put back into the pool, a pre-emptive read was performed
                // into the read buffer.  That read should not complete prior to us using the
                // connection again, as that would mean the connection was either closed or had
                // erroneous data sent on it by the server in response to no request from us.
                // We need to consume that read prior to issuing another read request.
                ValueTask<int>? t = _readAheadTask;
                if (t != null)
                {
                    _readAheadTask = null;

                    int bytesRead = await t.GetValueOrDefault().ConfigureAwait(false);
                    if (NetEventSource.IsEnabled) Trace($"Received {bytesRead} bytes.");

                    if (bytesRead == 0)
                    {
                        throw new IOException(SR.net_http_invalid_response);
                    }

                    _readOffset = 0;
                    _readLength = bytesRead;
                }

                // The request is no longer retryable; either we received data from the _readAheadTask,
                // or there was no _readAheadTask because this is the first request on the connection.
                // (We may have already set this as well if we sent request content.)<commit_msg>Fix race condition on read ahead task in SocketsHttpHandler (#28637)<commit_after>﻿// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.

using System.Buffers.Text;
using System.Collections.Generic;
using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.Net.Http.Headers;
using System.Net.Security;
using System.Net.Sockets;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace System.Net.Http
{
    internal partial class HttpConnection : IDisposable
    {
        /// <summary>Default size of the read buffer used for the connection.</summary>
        private const int InitialReadBufferSize =
#if DEBUG
            10;
#else
            4096;
#endif
        /// <summary>Default size of the write buffer used for the connection.</summary>
        private const int InitialWriteBufferSize = InitialReadBufferSize;
        /// <summary>
        /// Size after which we'll close the connection rather than send the payload in response
        /// to final error status code sent by the server when using Expect: 100-continue.
        /// </summary>
        private const int Expect100ErrorSendThreshold = 1024;

        private static readonly byte[] s_contentLength0NewlineAsciiBytes = Encoding.ASCII.GetBytes("Content-Length: 0\r\n");
        private static readonly byte[] s_spaceHttp10NewlineAsciiBytes = Encoding.ASCII.GetBytes(" HTTP/1.0\r\n");
        private static readonly byte[] s_spaceHttp11NewlineAsciiBytes = Encoding.ASCII.GetBytes(" HTTP/1.1\r\n");
        private static readonly byte[] s_httpSchemeAndDelimiter = Encoding.ASCII.GetBytes(Uri.UriSchemeHttp + Uri.SchemeDelimiter);
        private static readonly byte[] s_http1DotBytes = Encoding.ASCII.GetBytes("HTTP/1.");
        private static readonly ulong s_http10Bytes = BitConverter.ToUInt64(Encoding.ASCII.GetBytes("HTTP/1.0"));
        private static readonly ulong s_http11Bytes = BitConverter.ToUInt64(Encoding.ASCII.GetBytes("HTTP/1.1"));

        private readonly HttpConnectionPool _pool;
        private readonly Socket _socket; // used for polling; _stream should be used for all reading/writing. _stream owns disposal.
        private readonly Stream _stream;
        private readonly TransportContext _transportContext;
        private readonly bool _usingProxy;
        private readonly WeakReference<HttpConnection> _weakThisRef;

        private HttpRequestMessage _currentRequest;
        private readonly byte[] _writeBuffer;
        private int _writeOffset;
        private int _allowedReadLineBytes;

        private ValueTask<int>? _readAheadTask;
        private int _readAheadTaskLock = 0; // 0 == free, 1 == held
        private byte[] _readBuffer;
        private int _readOffset;
        private int _readLength;

        private bool _inUse;
        private bool _canRetry;
        private bool _connectionClose; // Connection: close was seen on last response
        private int _disposed; // 1 yes, 0 no

        public HttpConnection(
            HttpConnectionPool pool,
            Socket socket,
            Stream stream,
            TransportContext transportContext)
        {
            Debug.Assert(pool != null);
            Debug.Assert(stream != null);

            _pool = pool;
            _socket = socket; // may be null in cases where we couldn't easily get the underlying socket
            _stream = stream;
            _transportContext = transportContext;
            _usingProxy = pool.UsingProxy;

            _writeBuffer = new byte[InitialWriteBufferSize];
            _readBuffer = new byte[InitialReadBufferSize];

            _weakThisRef = new WeakReference<HttpConnection>(this);

            if (NetEventSource.IsEnabled)
            {
                if (pool.IsSecure)
                {
                    var sslStream = (SslStream)_stream;
                    Trace(
                        $"Secure connection created to {pool}. " +
                        $"SslProtocol:{sslStream.SslProtocol}, " +
                        $"CipherAlgorithm:{sslStream.CipherAlgorithm}, CipherStrength:{sslStream.CipherStrength}, " +
                        $"HashAlgorithm:{sslStream.HashAlgorithm}, HashStrength:{sslStream.HashStrength}, " +
                        $"KeyExchangeAlgorithm:{sslStream.KeyExchangeAlgorithm}, KeyExchangeStrength:{sslStream.KeyExchangeStrength}, " +
                        $"LocalCert:{sslStream.LocalCertificate}, RemoteCert:{sslStream.RemoteCertificate}");
                }
                else
                {
                    Trace($"Connection created to {pool}.");
                }
            }
        }

        public void Dispose() => Dispose(disposing: true);

        protected void Dispose(bool disposing)
        {
            // Ensure we're only disposed once.  Dispose could be called concurrently, for example,
            // if the request and the response were running concurrently and both incurred an exception.
            if (Interlocked.Exchange(ref _disposed, 1) == 0)
            {
                if (NetEventSource.IsEnabled) Trace("Connection closing.");
                _pool.DecrementConnectionCount();
                if (disposing)
                {
                    GC.SuppressFinalize(this);
                    _stream.Dispose();

                    // Eat any exceptions from the read-ahead task.  We don't need to log, as we expect
                    // failures from this task due to closing the connection while a read is in progress.
                    ValueTask<int>? readAheadTask = ConsumeReadAheadTask();
                    if (readAheadTask != null)
                    {
                        IgnoreExceptionsAsync(readAheadTask.GetValueOrDefault());
                    }
                }
            }
        }

        /// <summary>Awaits a task, ignoring any resulting exceptions.</summary>
        private static async void IgnoreExceptionsAsync(ValueTask<int> task)
        {
            try { await task.ConfigureAwait(false); } catch { }
        }

        /// <summary>Do a non-blocking poll to see whether the connection has data available or has been closed.</summary>
        /// <remarks>If we don't have direct access to the underlying socket, we instead use a read-ahead task.</remarks>
        public bool PollRead()
        {
            if (_socket != null) // may be null if we don't have direct access to the socket
            {
                try
                {
                    return _socket.Poll(0, SelectMode.SelectRead);
                }
                catch (Exception e) when (e is SocketException || e is ObjectDisposedException)
                {
                    // Poll can throw when used on a closed socket.
                    return true;
                }
            }
            else
            {
                return EnsureReadAheadAndPollRead();
            }
        }

        /// <summary>
        /// Issues a read-ahead on the connection, which will serve both as the first read on the
        /// response as well as a polling indication of whether the connection is usable.
        /// </summary>
        /// <returns>true if there's data available on the connection or it's been closed; otherwise, false.</returns>
        public bool EnsureReadAheadAndPollRead()
        {
            try
            {
                Debug.Assert(_readAheadTask == null || _socket == null, "Should only already have a read-ahead task if we don't have a socket to poll");
                if (_readAheadTask == null)
                {
                    _readAheadTask = _stream.ReadAsync(new Memory<byte>(_readBuffer));
                }
            }
            catch (Exception error)
            {
                // If reading throws, eat the error and don't pool the connection.
                if (NetEventSource.IsEnabled) Trace($"Error performing read ahead: {error}");
                Dispose();
                _readAheadTask = new ValueTask<int>(0);
            }

            return _readAheadTask.Value.IsCompleted; // equivalent to polling
        }

        private ValueTask<int>? ConsumeReadAheadTask()
        {
            if (Interlocked.CompareExchange(ref _readAheadTaskLock, 1, 0) == 0)
            {
                ValueTask<int>? t = _readAheadTask;
                _readAheadTask = null;
                Volatile.Write(ref _readAheadTaskLock, 0);
                return t;
            }

            // We couldn't get the lock, which means it must already be held
            // by someone else who will consume the task.
            return null;
        }

        public bool IsNewConnection
        {
            get
            {
                // This is only valid when we are not actually processing a request.
                Debug.Assert(_currentRequest == null);
                return (_readAheadTask == null);
            }
        }

        public bool CanRetry
        {
            get
            {
                // Should only be called when we have been disposed.
                Debug.Assert(_disposed != 0);
                return _canRetry;
            }
        }

        public DateTimeOffset CreationTime { get; } = DateTimeOffset.UtcNow;

        public TransportContext TransportContext => _transportContext;

        public bool UsingProxy => _usingProxy;

        private int ReadBufferSize => _readBuffer.Length;

        private ReadOnlyMemory<byte> RemainingBuffer => new ReadOnlyMemory<byte>(_readBuffer, _readOffset, _readLength - _readOffset);

        private void ConsumeFromRemainingBuffer(int bytesToConsume)
        {
            Debug.Assert(bytesToConsume <= _readLength - _readOffset, $"{bytesToConsume} > {_readLength} - {_readOffset}");
            _readOffset += bytesToConsume;
        }

        private async Task WriteHeadersAsync(HttpHeaders headers, string cookiesFromContainer)
        {
            foreach (KeyValuePair<HeaderDescriptor, string[]> header in headers.GetHeaderDescriptorsAndValues())
            {
                if (header.Key.KnownHeader != null)
                {
                    await WriteBytesAsync(header.Key.KnownHeader.AsciiBytesWithColonSpace).ConfigureAwait(false);
                }
                else
                {
                    await WriteAsciiStringAsync(header.Key.Name).ConfigureAwait(false);
                    await WriteTwoBytesAsync((byte)':', (byte)' ').ConfigureAwait(false);
                }

                Debug.Assert(header.Value.Length > 0, "No values for header??");
                if (header.Value.Length > 0)
                {
                    await WriteStringAsync(header.Value[0]).ConfigureAwait(false);

                    if (cookiesFromContainer != null && header.Key.KnownHeader == KnownHeaders.Cookie)
                    {
                        await WriteTwoBytesAsync((byte)';', (byte)' ').ConfigureAwait(false);
                        await WriteStringAsync(cookiesFromContainer).ConfigureAwait(false);

                        cookiesFromContainer = null;
                    }

                    for (int i = 1; i < header.Value.Length; i++)
                    {
                        await WriteTwoBytesAsync((byte)',', (byte)' ').ConfigureAwait(false);
                        await WriteStringAsync(header.Value[i]).ConfigureAwait(false);
                    }
                }

                await WriteTwoBytesAsync((byte)'\r', (byte)'\n').ConfigureAwait(false);
            }

            if (cookiesFromContainer != null)
            {
                await WriteAsciiStringAsync(HttpKnownHeaderNames.Cookie).ConfigureAwait(false);
                await WriteTwoBytesAsync((byte)':', (byte)' ').ConfigureAwait(false);
                await WriteAsciiStringAsync(cookiesFromContainer).ConfigureAwait(false);
                await WriteTwoBytesAsync((byte)'\r', (byte)'\n').ConfigureAwait(false);
            }
        }

        private async Task WriteHostHeaderAsync(Uri uri)
        {
            await WriteBytesAsync(KnownHeaders.Host.AsciiBytesWithColonSpace).ConfigureAwait(false);

            if (_pool.HostHeaderValueBytes != null)
            {
                Debug.Assert(!_pool.UsingProxy);
                await WriteBytesAsync(_pool.HostHeaderValueBytes).ConfigureAwait(false);
            }
            else
            {
                Debug.Assert(_pool.UsingProxy);
                await WriteAsciiStringAsync(uri.IdnHost).ConfigureAwait(false);

                if (!uri.IsDefaultPort)
                {
                    await WriteByteAsync((byte)':').ConfigureAwait(false);
                    await WriteDecimalInt32Async(uri.Port).ConfigureAwait(false);
                }
            }

            await WriteTwoBytesAsync((byte)'\r', (byte)'\n').ConfigureAwait(false);
        }

        private Task WriteDecimalInt32Async(int value)
        {
            // Try to format into our output buffer directly.
            if (Utf8Formatter.TryFormat(value, new Span<byte>(_writeBuffer, _writeOffset, _writeBuffer.Length - _writeOffset), out int bytesWritten))
            {
                _writeOffset += bytesWritten;
                return Task.CompletedTask;
            }

            // If we don't have enough room, do it the slow way.
            return WriteAsciiStringAsync(value.ToString());
        }

        private Task WriteHexInt32Async(int value)
        {
            // Try to format into our output buffer directly.
            if (Utf8Formatter.TryFormat(value, new Span<byte>(_writeBuffer, _writeOffset, _writeBuffer.Length - _writeOffset), out int bytesWritten, 'X'))
            {
                _writeOffset += bytesWritten;
                return Task.CompletedTask;
            }

            // If we don't have enough room, do it the slow way.
            return WriteAsciiStringAsync(value.ToString("X", CultureInfo.InvariantCulture));
        }

        public async Task<HttpResponseMessage> SendAsyncCore(HttpRequestMessage request, CancellationToken cancellationToken)
        {
            TaskCompletionSource<bool> allowExpect100ToContinue = null;
            Debug.Assert(_currentRequest == null, $"Expected null {nameof(_currentRequest)}.");
            Debug.Assert(RemainingBuffer.Length == 0, "Unexpected data in read buffer");

            _currentRequest = request;
            bool isConnectMethod = (request.Method == HttpMethod.Connect);

            Debug.Assert(!_canRetry);
            _canRetry = true;

            // Send the request.
            if (NetEventSource.IsEnabled) Trace($"Sending request: {request}");
            CancellationTokenRegistration cancellationRegistration = RegisterCancellation(cancellationToken);
            try
            {
                // Write request line
                await WriteStringAsync(request.Method.Method).ConfigureAwait(false);
                await WriteByteAsync((byte)' ').ConfigureAwait(false);

                if (isConnectMethod)
                {
                    // RFC 7231 #section-4.3.6.
                    // Write only CONNECT foo.com:345 HTTP/1.1
                    if (!request.HasHeaders || request.Headers.Host == null)
                    {
                        throw new HttpRequestException(SR.net_http_request_no_host);
                    }
                    await WriteAsciiStringAsync(request.Headers.Host).ConfigureAwait(false);
                }
                else
                {
                    if (_usingProxy)
                    {
                        // Proxied requests contain full URL
                        Debug.Assert(request.RequestUri.Scheme == Uri.UriSchemeHttp);
                        await WriteBytesAsync(s_httpSchemeAndDelimiter).ConfigureAwait(false);
                        await WriteAsciiStringAsync(request.RequestUri.IdnHost).ConfigureAwait(false);
                    }
                    await WriteStringAsync(request.RequestUri.GetComponents(UriComponents.PathAndQuery | UriComponents.Fragment, UriFormat.UriEscaped)).ConfigureAwait(false);
                }

                // Fall back to 1.1 for all versions other than 1.0
                Debug.Assert(request.Version.Major >= 0 && request.Version.Minor >= 0); // guaranteed by Version class
                bool isHttp10 = request.Version.Minor == 0 && request.Version.Major == 1;
                await WriteBytesAsync(isHttp10 ? s_spaceHttp10NewlineAsciiBytes : s_spaceHttp11NewlineAsciiBytes).ConfigureAwait(false);

                // Determine cookies to send
                string cookiesFromContainer = null;
                if (_pool.Settings._useCookies)
                {
                    cookiesFromContainer = _pool.Settings._cookieContainer.GetCookieHeader(request.RequestUri);
                    if (cookiesFromContainer == "")
                    {
                        cookiesFromContainer = null;
                    }
                }

                // Write request headers
                if (request.HasHeaders || cookiesFromContainer != null)
                {
                    await WriteHeadersAsync(request.Headers, cookiesFromContainer).ConfigureAwait(false);
                }

                if (request.Content == null)
                {
                    // Write out Content-Length: 0 header to indicate no body,
                    // unless this is a method that never has a body.
                    if (request.Method != HttpMethod.Get && request.Method != HttpMethod.Head && !isConnectMethod)
                    {
                        await WriteBytesAsync(s_contentLength0NewlineAsciiBytes).ConfigureAwait(false);
                    }
                }
                else
                {
                    // Write content headers
                    await WriteHeadersAsync(request.Content.Headers, cookiesFromContainer: null).ConfigureAwait(false);
                }

                // Write special additional headers.  If a host isn't in the headers list, then a Host header
                // wasn't sent, so as it's required by HTTP 1.1 spec, send one based on the Request Uri.
                if (!request.HasHeaders || request.Headers.Host == null)
                {
                    await WriteHostHeaderAsync(request.RequestUri).ConfigureAwait(false);
                }

                // CRLF for end of headers.
                await WriteTwoBytesAsync((byte)'\r', (byte)'\n').ConfigureAwait(false);

                Task sendRequestContentTask = null;
                if (request.Content == null)
                {
                    // We have nothing more to send, so flush out any headers we haven't yet sent.
                    await FlushAsync().ConfigureAwait(false);
                }
                else
                {
                    // Send the body if there is one.  We prefer to serialize the sending of the content before
                    // we try to receive any response, but if ExpectContinue has been set, we allow the sending
                    // to run concurrently until we receive the final status line, at which point we wait for it.
                    if (!request.HasHeaders || request.Headers.ExpectContinue != true)
                    {
                        await SendRequestContentAsync(request, CreateRequestContentStream(request), cancellationToken).ConfigureAwait(false);
                    }
                    else
                    {
                        // We're sending an Expect: 100-continue header. We need to flush headers so that the server receives
                        // all of them, and we need to do so before initiating the send, as once we do that, it effectively
                        // owns the right to write, and we don't want to concurrently be accessing the write buffer.
                        await FlushAsync().ConfigureAwait(false);

                        // Create a TCS we'll use to block the request content from being sent, and create a timer that's used
                        // as a fail-safe to unblock the request content if we don't hear back from the server in a timely manner.
                        // Then kick off the request.  The TCS' result indicates whether content should be sent or not.
                        allowExpect100ToContinue = new TaskCompletionSource<bool>();
                        var expect100Timer = new Timer(
                            s => ((TaskCompletionSource<bool>)s).TrySetResult(true),
                            allowExpect100ToContinue, _pool.Settings._expect100ContinueTimeout, Timeout.InfiniteTimeSpan);
                        sendRequestContentTask = SendRequestContentWithExpect100ContinueAsync(
                            request, allowExpect100ToContinue.Task, CreateRequestContentStream(request), expect100Timer, cancellationToken);
                    }
                }

                // Start to read response.
                _allowedReadLineBytes = _pool.Settings._maxResponseHeadersLength * 1024;

                // We should not have any buffered data here; if there was, it should have been treated as an error
                // by the previous request handling.  (Note we do not support HTTP pipelining.)
                Debug.Assert(_readOffset == _readLength);

                // When the connection was taken out of the pool, a pre-emptive read was performed
                // into the read buffer. We need to consume that read prior to issuing another read.
                ValueTask<int>? t = ConsumeReadAheadTask();
                if (t != null)
                {
                    int bytesRead = await t.GetValueOrDefault().ConfigureAwait(false);
                    if (NetEventSource.IsEnabled) Trace($"Received {bytesRead} bytes.");

                    if (bytesRead == 0)
                    {
                        throw new IOException(SR.net_http_invalid_response);
                    }

                    _readOffset = 0;
                    _readLength = bytesRead;
                }

                // The request is no longer retryable; either we received data from the _readAheadTask,
                // or there was no _readAheadTask because this is the first request on the connection.
                // (We may have already set this as well if we sent request content.)
{'fraction_non_alphanumeric': 0.07858484658735128, 'fraction_numerical': 0.005859200286251007, 'mean_word_length': 1.7140689487739742, 'pattern_counts': {'":': 0, '<': 72, '<?xml version=': 0, '>': 92, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7153226', 'n_tokens_mistral': 11028, 'n_tokens_neox': 10221, 'n_words': 2976}