Skip to content

Commit 5ff1804

Browse files
JamesNKBrennanConroy
JamesNK
and
BrennanConroy
authored
Use SocketsHttpHandler.ConnectTimeout on connect (#1829)
Co-authored-by: Brennan <brecon@microsoft.com>
1 parent 9d0919f commit 5ff1804

File tree

9 files changed

+183
-55
lines changed

9 files changed

+183
-55
lines changed

‎Grpc.DotNet.sln

Lines changed: 12 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -39,9 +39,9 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "benchmarkapps", "benchmarka
3939
EndProject
4040
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Shared", "Shared", "{310E5783-455A-4D09-A7AE-39DC2AB09504}"
4141
ProjectSection(SolutionItems) = preProject
42-
perf\benchmarkapps\Shared\benchmark_service.proto = perf\benchmarkapps\Shared\benchmark_service.proto
4342
perf\benchmarkapps\Shared\BenchmarkConfigurationHelpers.cs = perf\benchmarkapps\Shared\BenchmarkConfigurationHelpers.cs
4443
perf\benchmarkapps\Shared\BenchmarkServiceImpl.cs = perf\benchmarkapps\Shared\BenchmarkServiceImpl.cs
44+
perf\benchmarkapps\Shared\benchmark_service.proto = perf\benchmarkapps\Shared\benchmark_service.proto
4545
perf\benchmarkapps\Shared\messages.proto = perf\benchmarkapps\Shared\messages.proto
4646
EndProjectSection
4747
EndProject
@@ -124,23 +124,21 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "LinkerTestsWebsite", "testa
124124
EndProject
125125
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Grpc.Net.Client.Web.Tests", "test\Grpc.Net.Client.Web.Tests\Grpc.Net.Client.Web.Tests.csproj", "{14B1CA94-1222-4D2E-B37A-1FF8676E233E}"
126126
EndProject
127-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "QpsWorker", "perf\benchmarkapps\QpsWorker\QpsWorker.csproj", "{430F312C-550B-4D1F-907F-01A72F3E5CF2}"
128-
EndProject
129-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Grpc.Core.Api", "src\Grpc.Core.Api\Grpc.Core.Api.csproj", "{BAE7C213-5950-4916-B456-A482828D89A0}"
127+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "QpsWorker", "perf\benchmarkapps\QpsWorker\QpsWorker.csproj", "{430F312C-550B-4D1F-907F-01A72F3E5CF2}"
130128
EndProject
131-
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "test", "test", "{8545CBA9-0780-46C5-8A98-18D9E0B958C1}"
129+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Grpc.Core.Api", "src\Grpc.Core.Api\Grpc.Core.Api.csproj", "{BAE7C213-5950-4916-B456-A482828D89A0}"
132130
EndProject
133-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Grpc.Core.Api.Tests", "test\Grpc.Core.Api.Tests\Grpc.Core.Api.Tests.csproj", "{CAEA2276-BCB8-4B1E-9E6F-E546E4BDBCA8}"
131+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Grpc.Core.Api.Tests", "test\Grpc.Core.Api.Tests\Grpc.Core.Api.Tests.csproj", "{CAEA2276-BCB8-4B1E-9E6F-E546E4BDBCA8}"
134132
EndProject
135-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Grpc.Auth", "src\Grpc.Auth\Grpc.Auth.csproj", "{0BF42289-98E1-4490-8934-AF56B05ADB8B}"
133+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Grpc.Auth", "src\Grpc.Auth\Grpc.Auth.csproj", "{0BF42289-98E1-4490-8934-AF56B05ADB8B}"
136134
EndProject
137-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Grpc.HealthCheck", "src\Grpc.HealthCheck\Grpc.HealthCheck.csproj", "{6A70E853-DFAF-4484-90FC-ACA755CD02A4}"
135+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Grpc.HealthCheck", "src\Grpc.HealthCheck\Grpc.HealthCheck.csproj", "{6A70E853-DFAF-4484-90FC-ACA755CD02A4}"
138136
EndProject
139-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Grpc.Reflection", "src\Grpc.Reflection\Grpc.Reflection.csproj", "{B4153E7F-5CF3-4DFB-A9D1-5E77A2FB2C48}"
137+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Grpc.Reflection", "src\Grpc.Reflection\Grpc.Reflection.csproj", "{B4153E7F-5CF3-4DFB-A9D1-5E77A2FB2C48}"
140138
EndProject
141-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Grpc.HealthCheck.Tests", "test\Grpc.HealthCheck.Tests\Grpc.HealthCheck.Tests.csproj", "{25544326-C145-4D05-A4C3-AC7D59E17196}"
139+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Grpc.HealthCheck.Tests", "test\Grpc.HealthCheck.Tests\Grpc.HealthCheck.Tests.csproj", "{25544326-C145-4D05-A4C3-AC7D59E17196}"
142140
EndProject
143-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Grpc.Reflection.Tests", "test\Grpc.Reflection.Tests\Grpc.Reflection.Tests.csproj", "{857C5B4B-E2A8-4ACA-98FB-5E592E2224CC}"
141+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Grpc.Reflection.Tests", "test\Grpc.Reflection.Tests\Grpc.Reflection.Tests.csproj", "{857C5B4B-E2A8-4ACA-98FB-5E592E2224CC}"
144142
EndProject
145143
Global
146144
GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -351,12 +349,12 @@ Global
351349
{14B1CA94-1222-4D2E-B37A-1FF8676E233E} = {CECC4AE8-9C4E-4727-939B-517CC2E58D65}
352350
{430F312C-550B-4D1F-907F-01A72F3E5CF2} = {1B8B6117-CE39-4580-BAFA-D8026102767A}
353351
{BAE7C213-5950-4916-B456-A482828D89A0} = {8C62055F-8CD7-4859-9001-634D544DF2AE}
354-
{CAEA2276-BCB8-4B1E-9E6F-E546E4BDBCA8} = {8545CBA9-0780-46C5-8A98-18D9E0B958C1}
352+
{CAEA2276-BCB8-4B1E-9E6F-E546E4BDBCA8} = {CECC4AE8-9C4E-4727-939B-517CC2E58D65}
355353
{0BF42289-98E1-4490-8934-AF56B05ADB8B} = {8C62055F-8CD7-4859-9001-634D544DF2AE}
356354
{6A70E853-DFAF-4484-90FC-ACA755CD02A4} = {8C62055F-8CD7-4859-9001-634D544DF2AE}
357355
{B4153E7F-5CF3-4DFB-A9D1-5E77A2FB2C48} = {8C62055F-8CD7-4859-9001-634D544DF2AE}
358-
{25544326-C145-4D05-A4C3-AC7D59E17196} = {8545CBA9-0780-46C5-8A98-18D9E0B958C1}
359-
{857C5B4B-E2A8-4ACA-98FB-5E592E2224CC} = {8545CBA9-0780-46C5-8A98-18D9E0B958C1}
356+
{25544326-C145-4D05-A4C3-AC7D59E17196} = {CECC4AE8-9C4E-4727-939B-517CC2E58D65}
357+
{857C5B4B-E2A8-4ACA-98FB-5E592E2224CC} = {CECC4AE8-9C4E-4727-939B-517CC2E58D65}
360358
EndGlobalSection
361359
GlobalSection(ExtensibilityGlobals) = postSolution
362360
SolutionGuid = {CD5C2B19-49B4-480A-990C-36D98A719B07}

‎src/Grpc.Net.Client/Balancer/Internal/ISubchannelTransport.cs

Lines changed: 37 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ namespace Grpc.Net.Client.Balancer.Internal
2727
internal interface ISubchannelTransport : IDisposable
2828
{
2929
BalancerAddress? CurrentAddress { get; }
30+
TimeSpan? ConnectTimeout { get; }
3031

3132
#if NET5_0_OR_GREATER
3233
ValueTask<Stream> GetStreamAsync(BalancerAddress address, CancellationToken cancellationToken);
@@ -37,11 +38,46 @@ internal interface ISubchannelTransport : IDisposable
3738
#else
3839
Task<bool>
3940
#endif
40-
TryConnectAsync(CancellationToken cancellationToken);
41+
TryConnectAsync(ConnectContext context);
4142

4243
void Disconnect();
4344
}
4445

46+
internal sealed class ConnectContext
47+
{
48+
private readonly CancellationTokenSource _cts;
49+
private bool _disposed;
50+
51+
// This flag allows the transport to determine why the cancellation token was canceled.
52+
// - Explicit cancellation, e.g. the channel was disposed.
53+
// - Connection timeout, e.g. SocketsHttpHandler.ConnectTimeout was exceeded.
54+
public bool IsConnectCanceled { get; private set; }
55+
56+
public CancellationToken CancellationToken => _cts.Token;
57+
58+
public ConnectContext(TimeSpan connectTimeout)
59+
{
60+
_cts = new CancellationTokenSource(connectTimeout);
61+
}
62+
63+
public void CancelConnect()
64+
{
65+
// Check disposed because CTS.Cancel throws if the CTS is disposed.
66+
if (!_disposed)
67+
{
68+
IsConnectCanceled = true;
69+
_cts.Cancel();
70+
}
71+
}
72+
73+
public void Dispose()
74+
{
75+
// Dispose the CTS because it could be created with an internal timer.
76+
_cts.Dispose();
77+
_disposed = true;
78+
}
79+
}
80+
4581
internal interface ISubchannelTransportFactory
4682
{
4783
ISubchannelTransport Create(Subchannel subchannel);

‎src/Grpc.Net.Client/Balancer/Internal/PassiveSubchannelTransport.cs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ public PassiveSubchannelTransport(Subchannel subchannel)
4343
}
4444

4545
public BalancerAddress? CurrentAddress => _currentAddress;
46+
public TimeSpan? ConnectTimeout { get; }
4647

4748
public void Disconnect()
4849
{
@@ -56,7 +57,7 @@ public void Disconnect()
5657
#else
5758
Task<bool>
5859
#endif
59-
TryConnectAsync(CancellationToken cancellationToken)
60+
TryConnectAsync(ConnectContext context)
6061
{
6162
Debug.Assert(_subchannel._addresses.Count == 1);
6263
Debug.Assert(CurrentAddress == null);

‎src/Grpc.Net.Client/Balancer/Internal/SocketConnectivitySubchannelTransport.cs

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -63,17 +63,19 @@ internal class SocketConnectivitySubchannelTransport : ISubchannelTransport, IDi
6363
private bool _disposed;
6464
private BalancerAddress? _currentAddress;
6565

66-
public SocketConnectivitySubchannelTransport(Subchannel subchannel, TimeSpan socketPingInterval, ILoggerFactory loggerFactory)
66+
public SocketConnectivitySubchannelTransport(Subchannel subchannel, TimeSpan socketPingInterval, TimeSpan? connectTimeout, ILoggerFactory loggerFactory)
6767
{
6868
_logger = loggerFactory.CreateLogger<SocketConnectivitySubchannelTransport>();
6969
_subchannel = subchannel;
7070
_socketPingInterval = socketPingInterval;
71+
ConnectTimeout = connectTimeout;
7172
_activeStreams = new List<ActiveStream>();
7273
_socketConnectedTimer = new Timer(OnCheckSocketConnection, state: null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
7374
}
7475

7576
public object Lock => _subchannel.Lock;
7677
public BalancerAddress? CurrentAddress => _currentAddress;
78+
public TimeSpan? ConnectTimeout { get; }
7779
public bool HasStream { get; }
7880

7981
// For testing. Take a copy under lock for thread-safety.
@@ -112,7 +114,7 @@ private void DisconnectUnsynchronized()
112114
_currentAddress = null;
113115
}
114116

115-
public async ValueTask<bool> TryConnectAsync(CancellationToken cancellationToken)
117+
public async ValueTask<bool> TryConnectAsync(ConnectContext context)
116118
{
117119
Debug.Assert(CurrentAddress == null);
118120

@@ -135,7 +137,7 @@ public async ValueTask<bool> TryConnectAsync(CancellationToken cancellationToken
135137
try
136138
{
137139
SocketConnectivitySubchannelTransportLog.ConnectingSocket(_logger, _subchannel.Id, currentAddress);
138-
await socket.ConnectAsync(currentAddress.EndPoint, cancellationToken).ConfigureAwait(false);
140+
await socket.ConnectAsync(currentAddress.EndPoint, context.CancellationToken).ConfigureAwait(false);
139141
SocketConnectivitySubchannelTransportLog.ConnectedSocket(_logger, _subchannel.Id, currentAddress);
140142

141143
lock (Lock)
@@ -158,9 +160,23 @@ public async ValueTask<bool> TryConnectAsync(CancellationToken cancellationToken
158160
{
159161
firstConnectionError = ex;
160162
}
163+
164+
// Stop trying to connect to addresses on cancellation.
165+
if (context.CancellationToken.IsCancellationRequested)
166+
{
167+
break;
168+
}
161169
}
162170
}
163171

172+
// Check if cancellation happened because of timeout.
173+
if (firstConnectionError is OperationCanceledException oce &&
174+
oce.CancellationToken == context.CancellationToken &&
175+
!context.IsConnectCanceled)
176+
{
177+
firstConnectionError = new TimeoutException("A connection could not be established within the configured ConnectTimeout.", firstConnectionError);
178+
}
179+
164180
// All connections failed
165181
_subchannel.UpdateConnectivityState(
166182
ConnectivityState.TransientFailure,

‎src/Grpc.Net.Client/Balancer/Subchannel.cs

Lines changed: 31 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ public sealed class Subchannel : IDisposable
5959
private readonly ConnectionManager _manager;
6060
private readonly ILogger _logger;
6161

62-
private CancellationTokenSource? _connectCts;
62+
private ConnectContext? _connectContext;
6363
private ConnectivityState _state;
6464
private TaskCompletionSource<object?>? _delayInterruptTcs;
6565

@@ -172,7 +172,7 @@ public void UpdateAddresses(IReadOnlyList<BalancerAddress> addresses)
172172

173173
if (requireReconnect)
174174
{
175-
_connectCts?.Cancel();
175+
CancelInProgressConnect();
176176
Transport.Disconnect();
177177
RequestConnection();
178178
}
@@ -213,12 +213,26 @@ public void RequestConnection()
213213
_ = ConnectTransportAsync();
214214
}
215215

216+
private void CancelInProgressConnect()
217+
{
218+
var connectContext = _connectContext;
219+
if (connectContext != null)
220+
{
221+
lock (Lock)
222+
{
223+
// Cancel connect cancellation token.
224+
connectContext.CancelConnect();
225+
connectContext.Dispose();
226+
}
227+
}
228+
}
229+
216230
private async Task ConnectTransportAsync()
217231
{
218232
// There shouldn't be a previous connect in progress, but cancel the CTS to ensure they're no longer running.
219-
_connectCts?.Cancel();
233+
CancelInProgressConnect();
220234

221-
_connectCts = new CancellationTokenSource();
235+
var connectContext = _connectContext = new ConnectContext(Transport.ConnectTimeout ?? Timeout.InfiniteTimeSpan);
222236

223237
var backoffPolicy = _manager.BackoffPolicyFactory.Create();
224238

@@ -236,12 +250,12 @@ private async Task ConnectTransportAsync()
236250
}
237251
}
238252

239-
if (await Transport.TryConnectAsync(_connectCts.Token).ConfigureAwait(false))
253+
if (await Transport.TryConnectAsync(connectContext).ConfigureAwait(false))
240254
{
241255
return;
242256
}
243257

244-
_connectCts.Token.ThrowIfCancellationRequested();
258+
connectContext.CancellationToken.ThrowIfCancellationRequested();
245259

246260
_delayInterruptTcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
247261
var delayCts = new CancellationTokenSource();
@@ -284,6 +298,15 @@ private async Task ConnectTransportAsync()
284298

285299
UpdateConnectivityState(ConnectivityState.TransientFailure, "Error connecting to subchannel.");
286300
}
301+
finally
302+
{
303+
lock (Lock)
304+
{
305+
// Dispose context because it might have been created with a connect timeout.
306+
// Want to clean up the connect timeout timer.
307+
connectContext.Dispose();
308+
}
309+
}
287310
}
288311

289312
internal bool UpdateConnectivityState(ConnectivityState state, string successDetail)
@@ -357,8 +380,9 @@ public void Dispose()
357380
{
358381
UpdateConnectivityState(ConnectivityState.Shutdown, "Subchannel disposed.");
359382
_stateChangedRegistrations.Clear();
383+
384+
CancelInProgressConnect();
360385
Transport.Dispose();
361-
_connectCts?.Cancel();
362386
}
363387
}
364388

‎src/Grpc.Net.Client/GrpcChannel.cs

Lines changed: 30 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ public sealed class GrpcChannel : ChannelBase, IDisposable
5959

6060
internal Uri Address { get; }
6161
internal HttpMessageInvoker HttpInvoker { get; }
62+
internal TimeSpan? ConnectTimeout { get; }
6263
internal HttpHandlerType HttpHandlerType { get; }
6364
internal TimeSpan InitialReconnectBackoff { get; }
6465
internal TimeSpan? MaxReconnectBackoff { get; }
@@ -112,7 +113,7 @@ internal GrpcChannel(Uri address, GrpcChannelOptions channelOptions) : base(addr
112113
Address = address;
113114
LoggerFactory = channelOptions.LoggerFactory ?? channelOptions.ResolveService<ILoggerFactory>(NullLoggerFactory.Instance);
114115
RandomGenerator = channelOptions.ResolveService<IRandomGenerator>(new RandomGenerator());
115-
HttpHandlerType = CalculateHandlerType(channelOptions);
116+
(HttpHandlerType, ConnectTimeout) = CalculateHandlerContext(channelOptions);
116117

117118
#if SUPPORT_LOAD_BALANCING
118119
InitialReconnectBackoff = channelOptions.InitialReconnectBackoff;
@@ -217,48 +218,58 @@ private bool IsHttpOrHttpsAddress()
217218
return Address.Scheme == Uri.UriSchemeHttps || Address.Scheme == Uri.UriSchemeHttp;
218219
}
219220

220-
private static HttpHandlerType CalculateHandlerType(GrpcChannelOptions channelOptions)
221+
private static HttpHandlerContext CalculateHandlerContext(GrpcChannelOptions channelOptions)
221222
{
222223
if (channelOptions.HttpHandler == null)
223224
{
224225
// No way to know what handler a HttpClient is using so assume custom.
225-
return channelOptions.HttpClient == null
226+
var type = channelOptions.HttpClient == null
226227
? HttpHandlerType.SocketsHttpHandler
227228
: HttpHandlerType.Custom;
229+
230+
return new HttpHandlerContext(type);
228231
}
229232

230233
if (HttpRequestHelpers.HasHttpHandlerType(channelOptions.HttpHandler, "System.Net.Http.WinHttpHandler"))
231234
{
232-
return HttpHandlerType.WinHttpHandler;
235+
return new HttpHandlerContext(HttpHandlerType.WinHttpHandler);
233236
}
234237
if (HttpRequestHelpers.HasHttpHandlerType(channelOptions.HttpHandler, "System.Net.Http.SocketsHttpHandler"))
235238
{
239+
HttpHandlerType type;
240+
TimeSpan? connectTimeout;
241+
236242
#if NET5_0_OR_GREATER
237243
var socketsHttpHandler = HttpRequestHelpers.GetHttpHandlerType<SocketsHttpHandler>(channelOptions.HttpHandler)!;
238244

245+
type = HttpHandlerType.SocketsHttpHandler;
246+
connectTimeout = socketsHttpHandler.ConnectTimeout;
247+
239248
// Check if the SocketsHttpHandler is being shared by channels.
240249
// It has already been setup by another channel (i.e. ConnectCallback is set) then
241250
// additional channels can use advanced connectivity features.
242-
if (BalancerHttpHandler.IsSocketsHttpHandlerSetup(socketsHttpHandler))
243-
{
244-
return HttpHandlerType.SocketsHttpHandler;
245-
}
246-
247-
// Someone has already configured the handler callback.
248-
// This channel can't support advanced connectivity features.
249-
if (socketsHttpHandler.ConnectCallback != null)
251+
if (!BalancerHttpHandler.IsSocketsHttpHandlerSetup(socketsHttpHandler))
250252
{
251-
return HttpHandlerType.Custom;
253+
// Someone has already configured the handler callback.
254+
// This channel can't support advanced connectivity features.
255+
if (socketsHttpHandler.ConnectCallback != null)
256+
{
257+
type = HttpHandlerType.Custom;
258+
connectTimeout = null;
259+
}
252260
}
261+
#else
262+
type = HttpHandlerType.SocketsHttpHandler;
263+
connectTimeout = null;
253264
#endif
254-
return HttpHandlerType.SocketsHttpHandler;
265+
return new HttpHandlerContext(type, connectTimeout);
255266
}
256267
if (HttpRequestHelpers.GetHttpHandlerType<HttpClientHandler>(channelOptions.HttpHandler) != null)
257268
{
258-
return HttpHandlerType.HttpClientHandler;
269+
return new HttpHandlerContext(HttpHandlerType.HttpClientHandler);
259270
}
260271

261-
return HttpHandlerType.Custom;
272+
return new HttpHandlerContext(HttpHandlerType.Custom);
262273
}
263274

264275
#if SUPPORT_LOAD_BALANCING
@@ -715,7 +726,7 @@ public ISubchannelTransport Create(Subchannel subchannel)
715726
{
716727
if (_channel.HttpHandlerType == HttpHandlerType.SocketsHttpHandler)
717728
{
718-
return new SocketConnectivitySubchannelTransport(subchannel, TimeSpan.FromSeconds(5), _channel.LoggerFactory);
729+
return new SocketConnectivitySubchannelTransport(subchannel, TimeSpan.FromSeconds(5), _channel.ConnectTimeout, _channel.LoggerFactory);
719730
}
720731

721732
return new PassiveSubchannelTransport(subchannel);
@@ -754,6 +765,8 @@ public static void AddressPathUnused(ILogger logger, string address)
754765
_addressPathUnused(logger, address, null);
755766
}
756767
}
768+
769+
private readonly record struct HttpHandlerContext(HttpHandlerType HttpHandlerType, TimeSpan? ConnectTimeout = null);
757770
}
758771

759772
internal enum HttpHandlerType

0 commit comments

Comments
 (0)