Fix polling on https connections in HttpConnectionPool (dotnet#49474)

stephentoub · scalablecory · web-flow · commit d1da07868439 · 2021-03-11T20:41:07.000-05:00
* Fix polling on https connections in HttpConnectionPool

Avoid performing read-aheads on such connections, which can result in long-pinned buffers and unnecessary faulted tasks when the connections are torn down.

* Update src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnectionPool.cs

Co-authored-by: Cory Nelson &lt;phrosty@gmail.com&gt;

Co-authored-by: Cory Nelson &lt;phrosty@gmail.com&gt;
diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnection.cs b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnection.cs
@@ -71,6 +71,7 @@ internal partial class HttpConnection : HttpConnectionBase, IDisposable
 
         public HttpConnection(
             HttpConnectionPool pool,
+            Socket? socket,
             Stream stream,
             TransportContext? transportContext)
         {
@@ -79,10 +80,7 @@ public HttpConnection(
 
             _pool = pool;
             _stream = stream;
-            if (stream is NetworkStream networkStream)
-            {
-                _socket = networkStream.Socket;
-            }
+            _socket = socket;
 
             _transportContext = transportContext;
 
diff --git a/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnectionPool.cs b/src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnectionPool.cs
@@ -567,6 +567,7 @@ private static bool IsUsableHttp11Connection(HttpConnection connection, long now
             }
 
             // Try to establish an HTTP2 connection
+            Socket? socket = null;
             Stream? stream = null;
             SslStream? sslStream = null;
             TransportContext? transportContext = null;
@@ -591,7 +592,7 @@ private static bool IsUsableHttp11Connection(HttpConnection connection, long now
 
                     HttpResponseMessage? failureResponse;
 
-                    (stream, transportContext, failureResponse) =
+                    (socket, stream, transportContext, failureResponse) =
                         await ConnectAsync(request, async, cancellationToken).ConfigureAwait(false);
 
                     if (failureResponse != null)
@@ -681,7 +682,7 @@ private static bool IsUsableHttp11Connection(HttpConnection connection, long now
 
                 if (canUse)
                 {
-                    return (await ConstructHttp11ConnectionAsync(async, stream!, transportContext, request, cancellationToken).ConfigureAwait(false), true, null);
+                    return (await ConstructHttp11ConnectionAsync(async, socket, stream!, transportContext, request, cancellationToken).ConfigureAwait(false), true, null);
                 }
                 else
                 {
@@ -1213,7 +1214,7 @@ public ValueTask<HttpResponseMessage> SendAsync(HttpRequestMessage request, bool
             return SendWithProxyAuthAsync(request, async, doRequestAuth, cancellationToken);
         }
 
-        private async ValueTask<(Stream?, TransportContext?, HttpResponseMessage?)> ConnectAsync(HttpRequestMessage request, bool async, CancellationToken cancellationToken)
+        private async ValueTask<(Socket?, Stream?, TransportContext?, HttpResponseMessage?)> ConnectAsync(HttpRequestMessage request, bool async, CancellationToken cancellationToken)
         {
             // If a non-infinite connect timeout has been set, create and use a new CancellationToken that will be canceled
             // when either the original token is canceled or a connect timeout occurs.
@@ -1228,17 +1229,18 @@ public ValueTask<HttpResponseMessage> SendAsync(HttpRequestMessage request, bool
             try
             {
                 Stream? stream = null;
+                Socket? socket = null;
                 switch (_kind)
                 {
                     case HttpConnectionKind.Http:
                     case HttpConnectionKind.Https:
                     case HttpConnectionKind.ProxyConnect:
                         Debug.Assert(_originAuthority != null);
-                        stream = await ConnectToTcpHostAsync(_originAuthority.IdnHost, _originAuthority.Port, request, async, cancellationToken).ConfigureAwait(false);
+                        (socket, stream) = await ConnectToTcpHostAsync(_originAuthority.IdnHost, _originAuthority.Port, request, async, cancellationToken).ConfigureAwait(false);
                         break;
 
                     case HttpConnectionKind.Proxy:
-                        stream = await ConnectToTcpHostAsync(_proxyUri!.IdnHost, _proxyUri.Port, request, async, cancellationToken).ConfigureAwait(false);
+                        (socket, stream) = await ConnectToTcpHostAsync(_proxyUri!.IdnHost, _proxyUri.Port, request, async, cancellationToken).ConfigureAwait(false);
                         break;
 
                     case HttpConnectionKind.ProxyTunnel:
@@ -1249,12 +1251,18 @@ public ValueTask<HttpResponseMessage> SendAsync(HttpRequestMessage request, bool
                         {
                             // Return non-success response from proxy.
                             response.RequestMessage = request;
-                            return (null, null, response);
+                            return (null, null, null, response);
                         }
                         break;
                 }
 
                 Debug.Assert(stream != null);
+                if (socket is null && stream is NetworkStream ns)
+                {
+                    // We weren't handed a socket directly.  But if we're able to extract one, do so.
+                    // Most likely case here is a ConnectCallback was used and returned a NetworkStream.
+                    socket = ns.Socket;
+                }
 
                 TransportContext? transportContext = null;
                 if (IsSecure)
@@ -1264,20 +1272,21 @@ public ValueTask<HttpResponseMessage> SendAsync(HttpRequestMessage request, bool
                     stream = sslStream;
                 }
 
-                return (stream, transportContext, null);
+                return (socket, stream, transportContext, null);
             }
             finally
             {
                 cancellationWithConnectTimeout?.Dispose();
             }
         }
 
-        private async ValueTask<Stream> ConnectToTcpHostAsync(string host, int port, HttpRequestMessage initialRequest, bool async, CancellationToken cancellationToken)
+        private async ValueTask<(Socket?, Stream)> ConnectToTcpHostAsync(string host, int port, HttpRequestMessage initialRequest, bool async, CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested();
 
             var endPoint = new DnsEndPoint(host, port);
             Socket? socket = null;
+            Stream? stream = null;
             try
             {
                 // If a ConnectCallback was supplied, use that to establish the connection.
@@ -1294,7 +1303,7 @@ private async ValueTask<Stream> ConnectToTcpHostAsync(string host, int port, Htt
                         Trace($"{nameof(SocketsHttpHandler.ConnectCallback)} completing asynchronously for a synchronous request.");
                     }
 
-                    return await streamTask.ConfigureAwait(false) ?? throw new HttpRequestException(SR.net_http_null_from_connect_callback);
+                    stream = await streamTask.ConfigureAwait(false) ?? throw new HttpRequestException(SR.net_http_null_from_connect_callback);
                 }
                 else
                 {
@@ -1313,8 +1322,10 @@ private async ValueTask<Stream> ConnectToTcpHostAsync(string host, int port, Htt
                         }
                     }
 
-                    return new NetworkStream(socket, ownsSocket: true);
+                    stream = new NetworkStream(socket, ownsSocket: true);
                 }
+
+                return (socket, stream);
             }
             catch (Exception ex)
             {
@@ -1327,15 +1338,15 @@ private async ValueTask<Stream> ConnectToTcpHostAsync(string host, int port, Htt
 
         internal async ValueTask<(HttpConnection?, HttpResponseMessage?)> CreateHttp11ConnectionAsync(HttpRequestMessage request, bool async, CancellationToken cancellationToken)
         {
-            (Stream? stream, TransportContext? transportContext, HttpResponseMessage? failureResponse) =
+            (Socket? socket, Stream? stream, TransportContext? transportContext, HttpResponseMessage? failureResponse) =
                 await ConnectAsync(request, async, cancellationToken).ConfigureAwait(false);
 
             if (failureResponse != null)
             {
                 return (null, failureResponse);
             }
 
-            return (await ConstructHttp11ConnectionAsync(async, stream!, transportContext, request, cancellationToken).ConfigureAwait(false), null);
+            return (await ConstructHttp11ConnectionAsync(async, socket, stream!, transportContext, request, cancellationToken).ConfigureAwait(false), null);
         }
 
         private SslClientAuthenticationOptions GetSslOptionsForRequest(HttpRequestMessage request)
@@ -1393,10 +1404,15 @@ private async ValueTask<Stream> ApplyPlaintextFilterAsync(bool async, Stream str
             return newStream;
         }
 
-        private async ValueTask<HttpConnection> ConstructHttp11ConnectionAsync(bool async, Stream stream, TransportContext? transportContext, HttpRequestMessage request, CancellationToken cancellationToken)
+        private async ValueTask<HttpConnection> ConstructHttp11ConnectionAsync(bool async, Socket? socket, Stream stream, TransportContext? transportContext, HttpRequestMessage request, CancellationToken cancellationToken)
         {
-            stream = await ApplyPlaintextFilterAsync(async, stream, HttpVersion.Version11, request, cancellationToken).ConfigureAwait(false);
-            return new HttpConnection(this, stream, transportContext);
+            Stream newStream = await ApplyPlaintextFilterAsync(async, stream, HttpVersion.Version11, request, cancellationToken).ConfigureAwait(false);
+            if (newStream != stream)
+            {
+                // If a plaintext filter created a new stream, we can't trust that the socket is still applicable.
+                socket = null;
+            }
+            return new HttpConnection(this, socket, newStream, transportContext);
         }
 
         private async ValueTask<Http2Connection> ConstructHttp2ConnectionAsync(Stream stream, HttpRequestMessage request, CancellationToken cancellationToken)

Original file line number	Diff line number	Diff line change
`@@ -567,6 +567,7 @@ private static bool IsUsableHttp11Connection(HttpConnection connection, long now`
`567`	`567`	`}`
`568`	`568`
`569`	`569`	`// Try to establish an HTTP2 connection`
	`570`	`+ Socket? socket = null;`
`570`	`571`	`Stream? stream = null;`
`571`	`572`	`SslStream? sslStream = null;`
`572`	`573`	`TransportContext? transportContext = null;`
`@@ -591,7 +592,7 @@ private static bool IsUsableHttp11Connection(HttpConnection connection, long now`
`591`	`592`
`592`	`593`	`HttpResponseMessage? failureResponse;`
`593`	`594`
`594`		`- (stream, transportContext, failureResponse) =`
	`595`	`+ (socket, stream, transportContext, failureResponse) =`
`595`	`596`	`await ConnectAsync(request, async, cancellationToken).ConfigureAwait(false);`
`596`	`597`
`597`	`598`	`if (failureResponse != null)`
`@@ -681,7 +682,7 @@ private static bool IsUsableHttp11Connection(HttpConnection connection, long now`
`681`	`682`
`682`	`683`	`if (canUse)`
`683`	`684`	`{`
`684`		`- return (await ConstructHttp11ConnectionAsync(async, stream!, transportContext, request, cancellationToken).ConfigureAwait(false), true, null);`
	`685`	`+ return (await ConstructHttp11ConnectionAsync(async, socket, stream!, transportContext, request, cancellationToken).ConfigureAwait(false), true, null);`
`685`	`686`	`}`
`686`	`687`	`else`
`687`	`688`	`{`
`@@ -1213,7 +1214,7 @@ public ValueTask<HttpResponseMessage> SendAsync(HttpRequestMessage request, bool`
`1213`	`1214`	`return SendWithProxyAuthAsync(request, async, doRequestAuth, cancellationToken);`
`1214`	`1215`	`}`
`1215`	`1216`
`1216`		`- private async ValueTask<(Stream?, TransportContext?, HttpResponseMessage?)> ConnectAsync(HttpRequestMessage request, bool async, CancellationToken cancellationToken)`
	`1217`	`+ private async ValueTask<(Socket?, Stream?, TransportContext?, HttpResponseMessage?)> ConnectAsync(HttpRequestMessage request, bool async, CancellationToken cancellationToken)`
`1217`	`1218`	`{`
`1218`	`1219`	`// If a non-infinite connect timeout has been set, create and use a new CancellationToken that will be canceled`
`1219`	`1220`	`// when either the original token is canceled or a connect timeout occurs.`
`@@ -1228,17 +1229,18 @@ public ValueTask<HttpResponseMessage> SendAsync(HttpRequestMessage request, bool`
`1228`	`1229`	`try`
`1229`	`1230`	`{`
`1230`	`1231`	`Stream? stream = null;`
	`1232`	`+ Socket? socket = null;`
`1231`	`1233`	`switch (_kind)`
`1232`	`1234`	`{`
`1233`	`1235`	`case HttpConnectionKind.Http:`
`1234`	`1236`	`case HttpConnectionKind.Https:`
`1235`	`1237`	`case HttpConnectionKind.ProxyConnect:`
`1236`	`1238`	`Debug.Assert(_originAuthority != null);`
`1237`		`- stream = await ConnectToTcpHostAsync(_originAuthority.IdnHost, _originAuthority.Port, request, async, cancellationToken).ConfigureAwait(false);`
	`1239`	`+ (socket, stream) = await ConnectToTcpHostAsync(_originAuthority.IdnHost, _originAuthority.Port, request, async, cancellationToken).ConfigureAwait(false);`
`1238`	`1240`	`break;`
`1239`	`1241`
`1240`	`1242`	`case HttpConnectionKind.Proxy:`
`1241`		`- stream = await ConnectToTcpHostAsync(_proxyUri!.IdnHost, _proxyUri.Port, request, async, cancellationToken).ConfigureAwait(false);`
	`1243`	`+ (socket, stream) = await ConnectToTcpHostAsync(_proxyUri!.IdnHost, _proxyUri.Port, request, async, cancellationToken).ConfigureAwait(false);`
`1242`	`1244`	`break;`
`1243`	`1245`
`1244`	`1246`	`case HttpConnectionKind.ProxyTunnel:`
`@@ -1249,12 +1251,18 @@ public ValueTask<HttpResponseMessage> SendAsync(HttpRequestMessage request, bool`
`1249`	`1251`	`{`
`1250`	`1252`	`// Return non-success response from proxy.`
`1251`	`1253`	`response.RequestMessage = request;`
`1252`		`- return (null, null, response);`
	`1254`	`+ return (null, null, null, response);`
`1253`	`1255`	`}`
`1254`	`1256`	`break;`
`1255`	`1257`	`}`
`1256`	`1258`
`1257`	`1259`	`Debug.Assert(stream != null);`
	`1260`	`+ if (socket is null && stream is NetworkStream ns)`
	`1261`	`+ {`
	`1262`	`+ // We weren't handed a socket directly. But if we're able to extract one, do so.`
	`1263`	`+ // Most likely case here is a ConnectCallback was used and returned a NetworkStream.`
	`1264`	`+ socket = ns.Socket;`
	`1265`	`+ }`
`1258`	`1266`
`1259`	`1267`	`TransportContext? transportContext = null;`
`1260`	`1268`	`if (IsSecure)`
`@@ -1264,20 +1272,21 @@ public ValueTask<HttpResponseMessage> SendAsync(HttpRequestMessage request, bool`
`1264`	`1272`	`stream = sslStream;`
`1265`	`1273`	`}`
`1266`	`1274`
`1267`		`- return (stream, transportContext, null);`
	`1275`	`+ return (socket, stream, transportContext, null);`
`1268`	`1276`	`}`
`1269`	`1277`	`finally`
`1270`	`1278`	`{`
`1271`	`1279`	`cancellationWithConnectTimeout?.Dispose();`
`1272`	`1280`	`}`
`1273`	`1281`	`}`
`1274`	`1282`
`1275`		`- private async ValueTask<Stream> ConnectToTcpHostAsync(string host, int port, HttpRequestMessage initialRequest, bool async, CancellationToken cancellationToken)`
	`1283`	`+ private async ValueTask<(Socket?, Stream)> ConnectToTcpHostAsync(string host, int port, HttpRequestMessage initialRequest, bool async, CancellationToken cancellationToken)`
`1276`	`1284`	`{`
`1277`	`1285`	`cancellationToken.ThrowIfCancellationRequested();`
`1278`	`1286`
`1279`	`1287`	`var endPoint = new DnsEndPoint(host, port);`
`1280`	`1288`	`Socket? socket = null;`
	`1289`	`+ Stream? stream = null;`
`1281`	`1290`	`try`
`1282`	`1291`	`{`
`1283`	`1292`	`// If a ConnectCallback was supplied, use that to establish the connection.`
`@@ -1294,7 +1303,7 @@ private async ValueTask<Stream> ConnectToTcpHostAsync(string host, int port, Htt`
`1294`	`1303`	`Trace($"{nameof(SocketsHttpHandler.ConnectCallback)} completing asynchronously for a synchronous request.");`
`1295`	`1304`	`}`
`1296`	`1305`
`1297`		`- return await streamTask.ConfigureAwait(false) ?? throw new HttpRequestException(SR.net_http_null_from_connect_callback);`
	`1306`	`+ stream = await streamTask.ConfigureAwait(false) ?? throw new HttpRequestException(SR.net_http_null_from_connect_callback);`
`1298`	`1307`	`}`
`1299`	`1308`	`else`
`1300`	`1309`	`{`
`@@ -1313,8 +1322,10 @@ private async ValueTask<Stream> ConnectToTcpHostAsync(string host, int port, Htt`
`1313`	`1322`	`}`
`1314`	`1323`	`}`
`1315`	`1324`
`1316`		`- return new NetworkStream(socket, ownsSocket: true);`
	`1325`	`+ stream = new NetworkStream(socket, ownsSocket: true);`
`1317`	`1326`	`}`
	`1327`	`+`
	`1328`	`+ return (socket, stream);`
`1318`	`1329`	`}`
`1319`	`1330`	`catch (Exception ex)`
`1320`	`1331`	`{`
`@@ -1327,15 +1338,15 @@ private async ValueTask<Stream> ConnectToTcpHostAsync(string host, int port, Htt`
`1327`	`1338`
`1328`	`1339`	`internal async ValueTask<(HttpConnection?, HttpResponseMessage?)> CreateHttp11ConnectionAsync(HttpRequestMessage request, bool async, CancellationToken cancellationToken)`
`1329`	`1340`	`{`
`1330`		`- (Stream? stream, TransportContext? transportContext, HttpResponseMessage? failureResponse) =`
	`1341`	`+ (Socket? socket, Stream? stream, TransportContext? transportContext, HttpResponseMessage? failureResponse) =`
`1331`	`1342`	`await ConnectAsync(request, async, cancellationToken).ConfigureAwait(false);`
`1332`	`1343`
`1333`	`1344`	`if (failureResponse != null)`
`1334`	`1345`	`{`
`1335`	`1346`	`return (null, failureResponse);`
`1336`	`1347`	`}`
`1337`	`1348`
`1338`		`- return (await ConstructHttp11ConnectionAsync(async, stream!, transportContext, request, cancellationToken).ConfigureAwait(false), null);`
	`1349`	`+ return (await ConstructHttp11ConnectionAsync(async, socket, stream!, transportContext, request, cancellationToken).ConfigureAwait(false), null);`
`1339`	`1350`	`}`
`1340`	`1351`
`1341`	`1352`	`private SslClientAuthenticationOptions GetSslOptionsForRequest(HttpRequestMessage request)`
`@@ -1393,10 +1404,15 @@ private async ValueTask<Stream> ApplyPlaintextFilterAsync(bool async, Stream str`
`1393`	`1404`	`return newStream;`
`1394`	`1405`	`}`
`1395`	`1406`
`1396`		`- private async ValueTask<HttpConnection> ConstructHttp11ConnectionAsync(bool async, Stream stream, TransportContext? transportContext, HttpRequestMessage request, CancellationToken cancellationToken)`
	`1407`	`+ private async ValueTask<HttpConnection> ConstructHttp11ConnectionAsync(bool async, Socket? socket, Stream stream, TransportContext? transportContext, HttpRequestMessage request, CancellationToken cancellationToken)`
`1397`	`1408`	`{`
`1398`		`- stream = await ApplyPlaintextFilterAsync(async, stream, HttpVersion.Version11, request, cancellationToken).ConfigureAwait(false);`
`1399`		`- return new HttpConnection(this, stream, transportContext);`
	`1409`	`+ Stream newStream = await ApplyPlaintextFilterAsync(async, stream, HttpVersion.Version11, request, cancellationToken).ConfigureAwait(false);`
	`1410`	`+ if (newStream != stream)`
	`1411`	`+ {`
	`1412`	`+ // If a plaintext filter created a new stream, we can't trust that the socket is still applicable.`
	`1413`	`+ socket = null;`
	`1414`	`+ }`
	`1415`	`+ return new HttpConnection(this, socket, newStream, transportContext);`
`1400`	`1416`	`}`
`1401`	`1417`
`1402`	`1418`	`private async ValueTask<Http2Connection> ConstructHttp2ConnectionAsync(Stream stream, HttpRequestMessage request, CancellationToken cancellationToken)`