Skip to content

Commit 8c53008

Browse files
DmitryLukyanovDmitryLukyanov
authored
CSHARP-3906: Switch ConnectionPool background task to a dedicated thread. (#739)
CSHARP-3906: Switch ConnectionPool background task to a dedicated thread.
1 parent 3b7d210 commit 8c53008

File tree

4 files changed

+28
-53
lines changed

4 files changed

+28
-53
lines changed

‎src/MongoDB.Driver.Core/Core/ConnectionPools/ExclusiveConnectionPool.Helpers.cs‎

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -164,17 +164,17 @@ public void ThrowIfNotInitialized()
164164
internal sealed class MaintenanceHelper : IDisposable
165165
{
166166
private CancellationTokenSource _cancellationTokenSource = null;
167-
private Func<CancellationToken, Task> _maintenanceTaskCreator;
168-
private Task _maintenanceTask;
167+
private readonly Action<CancellationToken> _maintenanceAction;
168+
private Thread _maintenanceThread;
169169
private readonly TimeSpan _interval;
170170

171-
public MaintenanceHelper(Func<CancellationToken, Task> maintenanceTaskCreator, TimeSpan interval)
171+
public MaintenanceHelper(Action<CancellationToken> maintenanceAction, TimeSpan interval)
172172
{
173173
_interval = interval;
174-
_maintenanceTaskCreator = Ensure.IsNotNull(maintenanceTaskCreator, nameof(maintenanceTaskCreator));
174+
_maintenanceAction = Ensure.IsNotNull(maintenanceAction, nameof(maintenanceAction));
175175
}
176176

177-
public bool IsRunning => _maintenanceTask != null;
177+
public bool IsRunning => _maintenanceThread != null;
178178

179179
public void Cancel()
180180
{
@@ -185,7 +185,8 @@ public void Cancel()
185185

186186
CancelAndDispose();
187187
_cancellationTokenSource = null;
188-
_maintenanceTask = null;
188+
_maintenanceThread = null;
189+
// the previous _maintenanceThread might not be stopped yet, but it will be soon
189190
}
190191

191192
public void Start()
@@ -199,8 +200,13 @@ public void Start()
199200
_cancellationTokenSource = new CancellationTokenSource();
200201
var cancellationToken = _cancellationTokenSource.Token;
201202

202-
_maintenanceTask = Task.Run(() => _maintenanceTaskCreator(cancellationToken), cancellationToken);
203-
_maintenanceTask.ConfigureAwait(false);
203+
_maintenanceThread = new Thread(new ParameterizedThreadStart(ThreadStart)) { IsBackground = true };
204+
_maintenanceThread.Start(cancellationToken);
205+
206+
void ThreadStart(object cancellationToken)
207+
{
208+
_maintenanceAction((CancellationToken)cancellationToken);
209+
}
204210
}
205211

206212
public void Dispose()
@@ -867,12 +873,12 @@ public ConnectionCreator(ExclusiveConnectionPool pool, TimeSpan connectingTimeou
867873
_stopwatch = null;
868874
}
869875

870-
public async Task<PooledConnection> CreateOpenedAsync(CancellationToken cancellationToken)
876+
public PooledConnection CreateOpened(CancellationToken cancellationToken)
871877
{
872878
try
873879
{
874880
var stopwatch = Stopwatch.StartNew();
875-
_connectingWaitStatus = await _pool._maxConnectingQueue.WaitAsync(_connectingTimeout, cancellationToken).ConfigureAwait(false);
881+
_connectingWaitStatus = _pool._maxConnectingQueue.Wait(_connectingTimeout, cancellationToken);
876882
stopwatch.Stop();
877883

878884
_pool._poolState.ThrowIfNotReady();
@@ -882,7 +888,7 @@ public async Task<PooledConnection> CreateOpenedAsync(CancellationToken cancella
882888
_pool.CreateTimeoutException(stopwatch, $"Timed out waiting for in connecting queue after {stopwatch.ElapsedMilliseconds}ms.");
883889
}
884890

885-
var connection = await CreateOpenedInternalAsync(cancellationToken).ConfigureAwait(false);
891+
var connection = CreateOpenedInternal(cancellationToken);
886892
return connection;
887893
}
888894
catch (Exception ex)

‎src/MongoDB.Driver.Core/Core/ConnectionPools/ExclusiveConnectionPool.cs‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ public ExclusiveConnectionPool(
7676
_connectionExceptionHandler = Ensure.IsNotNull(connectionExceptionHandler, nameof(connectionExceptionHandler));
7777
Ensure.IsNotNull(eventSubscriber, nameof(eventSubscriber));
7878

79-
_maintenanceHelper = new MaintenanceHelper(token => MaintainSizeAsync(token), _settings.MaintenanceInterval);
79+
_maintenanceHelper = new MaintenanceHelper(MaintainSize, _settings.MaintenanceInterval);
8080
_poolState = new PoolState(EndPointHelper.ToString(_endPoint));
8181
_checkOutReasonCounter = new CheckOutReasonCounter();
8282

@@ -291,7 +291,7 @@ public int GetGeneration(ObjectId? serviceId)
291291
}
292292

293293
// private methods
294-
private async Task MaintainSizeAsync(CancellationToken cancellationToken)
294+
private void MaintainSize(CancellationToken cancellationToken)
295295
{
296296
try
297297
{
@@ -300,13 +300,13 @@ private async Task MaintainSizeAsync(CancellationToken cancellationToken)
300300
try
301301
{
302302
_connectionHolder.Prune(cancellationToken);
303-
await EnsureMinSizeAsync(cancellationToken).ConfigureAwait(false);
303+
EnsureMinSize(cancellationToken);
304304
}
305305
catch
306306
{
307307
// ignore exceptions
308308
}
309-
await Task.Delay(_settings.MaintenanceInterval, cancellationToken).ConfigureAwait(false);
309+
ThreadHelper.Sleep(_settings.MaintenanceInterval, cancellationToken);
310310
}
311311
}
312312
catch
@@ -315,23 +315,23 @@ private async Task MaintainSizeAsync(CancellationToken cancellationToken)
315315
}
316316
}
317317

318-
private async Task EnsureMinSizeAsync(CancellationToken cancellationToken)
318+
private void EnsureMinSize(CancellationToken cancellationToken)
319319
{
320320
var minTimeout = TimeSpan.FromMilliseconds(20);
321321

322322
while (CreatedCount < _settings.MinConnections && !cancellationToken.IsCancellationRequested)
323323
{
324324
using (var poolAwaiter = _maxConnectionsQueue.CreateAwaiter())
325325
{
326-
var entered = await poolAwaiter.WaitSignaledAsync(minTimeout, cancellationToken).ConfigureAwait(false);
326+
var entered = poolAwaiter.WaitSignaled(minTimeout, cancellationToken);
327327
if (!entered)
328328
{
329329
return;
330330
}
331331

332332
using (var connectionCreator = new ConnectionCreator(this, minTimeout))
333333
{
334-
var connection = await connectionCreator.CreateOpenedAsync(cancellationToken).ConfigureAwait(false);
334+
var connection = connectionCreator.CreateOpened(cancellationToken);
335335
_connectionHolder.Return(connection);
336336
}
337337
}

‎src/MongoDB.Driver.Core/Core/Misc/SemaphoreSlimSignalable.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -62,9 +62,9 @@ public SemaphoreSlimSignalableAwaiter(SemaphoreSlimSignalable semaphoreSlimSigna
6262
_enteredSemaphore = false;
6363
}
6464

65-
public async Task<bool> WaitSignaledAsync(TimeSpan timeout, CancellationToken cancellationToken)
65+
public bool WaitSignaled(TimeSpan timeout, CancellationToken cancellationToken)
6666
{
67-
var waitResult = await _semaphoreSlimSignalable.WaitSignaledAsync(timeout, cancellationToken).ConfigureAwait(false);
67+
var waitResult = _semaphoreSlimSignalable.WaitSignaled(timeout, cancellationToken);
6868
_enteredSemaphore = waitResult == SemaphoreWaitResult.Entered;
6969
return _enteredSemaphore;
7070
}

‎tests/MongoDB.Driver.Core.Tests/Core/ConnectionPools/ExclusiveConnectionPoolTests.cs‎

Lines changed: 2 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -1119,8 +1119,8 @@ public void Maintenance_should_call_connection_dispose_when_connection_authentic
11191119
var authenticationException = new MongoAuthenticationException(new ConnectionId(_serverId), "test message");
11201120
var authenticationFailedConnection = new Mock<IConnection>();
11211121
authenticationFailedConnection
1122-
.Setup(c => c.OpenAsync(It.IsAny<CancellationToken>())) // an authentication exception is thrown from _connectionInitializer.InitializeConnection
1123-
// that in turn is called from OpenAsync
1122+
.Setup(c => c.Open(It.IsAny<CancellationToken>())) // an authentication exception is thrown from _connectionInitializer.InitializeConnection
1123+
// that in turn is called from OpenAsync
11241124
.Throws(authenticationException);
11251125

11261126
_mockConnectionExceptionHandler
@@ -1144,32 +1144,6 @@ public void Maintenance_should_call_connection_dispose_when_connection_authentic
11441144
}
11451145
}
11461146

1147-
[Fact]
1148-
public void MaintainSizeAsync_should_not_try_new_attempt_after_failing_without_delay()
1149-
{
1150-
var settings = _settings.With(maintenanceInterval: TimeSpan.FromSeconds(10));
1151-
1152-
using (var subject = CreateSubject(settings))
1153-
{
1154-
var tokenSource = new CancellationTokenSource();
1155-
_mockConnectionFactory
1156-
.SetupSequence(f => f.CreateConnection(_serverId, _endPoint))
1157-
.Throws<Exception>() // failed attempt
1158-
.Returns(() => // successful attempt which should be delayed
1159-
{
1160-
// break the loop. With this line the MaintainSizeAsync will contain only 2 iterations
1161-
tokenSource.Cancel();
1162-
return new MockConnection(_serverId);
1163-
});
1164-
1165-
var testResult = Task.WaitAny(
1166-
subject.MaintainSizeAsync(tokenSource.Token), // if this task is completed first, it will mean that there was no delay (10 sec)
1167-
Task.Delay(TimeSpan.FromSeconds(1))); // time to be sure that delay is happening,
1168-
// if the method is running more than 1 second, then delay is happening
1169-
testResult.Should().Be(1);
1170-
}
1171-
}
1172-
11731147
[Theory]
11741148
[ParameterAttributeData]
11751149
public void Maintenance_should_run_with_finite_maintenanceInterval(
@@ -1655,11 +1629,6 @@ private void InitializeAndWait(ExclusiveConnectionPool pool = null, ConnectionPo
16551629

16561630
internal static class ExclusiveConnectionPoolReflector
16571631
{
1658-
public static Task MaintainSizeAsync(this ExclusiveConnectionPool obj, CancellationToken cancellationToken)
1659-
{
1660-
return (Task)Reflector.Invoke(obj, nameof(MaintainSizeAsync), cancellationToken);
1661-
}
1662-
16631632
public static int _waitQueueFreeSlots(this ExclusiveConnectionPool obj)
16641633
{
16651634
return (int)Reflector.GetFieldValue(obj, nameof(_waitQueueFreeSlots));

0 commit comments

Comments
 (0)