CSHARP-3761: Reimplement SDAM to use a dedicated thread. (mongodb#722)

CSHARP-3761: Reimplement SDAM to use a dedicated thread.

Author: DmitryLukyanov

src/MongoDB.Driver.Core/Core/Misc/ThreadHelper.cs

@@ -0,0 +1,29 @@
+/* Copyright 2010-present MongoDB Inc.
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace MongoDB.Driver.Core.Misc
+{
+    internal static class ThreadHelper
+    {
+        public static void Sleep(TimeSpan timeout, CancellationToken cancellationToken)
+        {
+            Task.Delay(timeout, cancellationToken).Wait(cancellationToken);
+        }
+    }
+}

src/MongoDB.Driver.Core/Core/Servers/HeartbeatDelay.cs

@@ -74,6 +74,11 @@ public void RequestHeartbeat()
             }
         }
 
+        public void Wait(CancellationToken cancellationToken)
+        {
+            _taskCompletionSource.Task.Wait();
+        }
+
         private void TimerCallback(object state)
         {
             _taskCompletionSource.TrySetResult(true);

src/MongoDB.Driver.Core/Core/Servers/RoundTripTimeMonitor.cs

@@ -17,7 +17,6 @@
 using System.Diagnostics;
 using System.Net;
 using System.Threading;
-using System.Threading.Tasks;
 using MongoDB.Driver.Core.Connections;
 using MongoDB.Driver.Core.Misc;
 
@@ -28,7 +27,7 @@ internal interface IRoundTripTimeMonitor : IDisposable
         TimeSpan Average { get; }
         void AddSample(TimeSpan roundTripTime);
         void Reset();
-        Task RunAsync();
+        void Start();
     }
 
     internal sealed class RoundTripTimeMonitor : IRoundTripTimeMonitor
@@ -42,6 +41,7 @@ internal sealed class RoundTripTimeMonitor : IRoundTripTimeMonitor
         private readonly TimeSpan _heartbeatInterval;
         private readonly object _lock = new object();
         private IConnection _roundTripTimeConnection;
+        private Thread _roundTripTimeMonitorThread;
         private readonly ServerApi _serverApi;
         private readonly ServerId _serverId;
 
@@ -85,7 +85,26 @@ public void Dispose()
             }
         }
 
-        public async Task RunAsync()
+        public void Start()
+        {
+            _roundTripTimeMonitorThread = new Thread(ThreadStart) { IsBackground = true };
+            _roundTripTimeMonitorThread.Start();
+
+            void ThreadStart()
+            {
+                try
+                {
+                    MonitorServer();
+                }
+                catch (OperationCanceledException)
+                {
+                    // ignore OperationCanceledException
+                }
+            }
+        }
+
+        // private methods
+        private void MonitorServer()
         {
             var helloOk = false;
             while (!_cancellationToken.IsCancellationRequested)
@@ -94,15 +113,15 @@ public async Task RunAsync()
                 {
                     if (_roundTripTimeConnection == null)
                     {
-                        await InitializeConnectionAsync().ConfigureAwait(false); // sets _roundTripTimeConnection
+                        InitializeConnection(); // sets _roundTripTimeConnection
                     }
                     else
                     {
                         var helloCommand = HelloHelper.CreateCommand(_serverApi, helloOk);
                         var helloProtocol = HelloHelper.CreateProtocol(helloCommand, _serverApi);
 
                         var stopwatch = Stopwatch.StartNew();
-                        var helloResult = await HelloHelper.GetResultAsync(_roundTripTimeConnection, helloProtocol, _cancellationToken).ConfigureAwait(false);
+                        var helloResult = HelloHelper.GetResult(_roundTripTimeConnection, helloProtocol, _cancellationToken);
                         stopwatch.Stop();
                         AddSample(stopwatch.Elapsed);
                         helloOk = helloResult.HelloOk;
@@ -118,13 +137,11 @@ public async Task RunAsync()
                     }
                     toDispose?.Dispose();
                 }
-
-                await Task.Delay(_heartbeatInterval, _cancellationToken).ConfigureAwait(false);
+                ThreadHelper.Sleep(_heartbeatInterval, _cancellationToken);
             }
         }
 
-        // private methods
-        private async Task InitializeConnectionAsync()
+        private void InitializeConnection()
         {
             _cancellationToken.ThrowIfCancellationRequested();
 
@@ -135,7 +152,7 @@ private async Task InitializeConnectionAsync()
             {
                 // if we are cancelling, it's because the server has
                 // been shut down and we really don't need to wait.
-                await roundTripTimeConnection.OpenAsync(_cancellationToken).ConfigureAwait(false);
+                roundTripTimeConnection.Open(_cancellationToken);
                 _cancellationToken.ThrowIfCancellationRequested();
             }
             catch

src/MongoDB.Driver.Core/Core/Servers/ServerMonitor.cs

@@ -14,10 +14,10 @@
 */
 
 using System;
+using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Net;
 using System.Threading;
-using System.Threading.Tasks;
 using MongoDB.Bson;
 using MongoDB.Driver.Core.Connections;
 using MongoDB.Driver.Core.Events;
@@ -43,6 +43,8 @@ internal sealed class ServerMonitor : IServerMonitor
         private readonly InterlockedInt32 _state;
         private readonly ServerMonitorSettings _serverMonitorSettings;
 
+        private Thread _serverMonitorThread;
+
         private readonly Action<ServerHeartbeatStartedEvent> _heartbeatStartedEventHandler;
         private readonly Action<ServerHeartbeatSucceededEvent> _heartbeatSucceededEventHandler;
         private readonly Action<ServerHeartbeatFailedEvent> _heartbeatFailedEventHandler;
@@ -144,15 +146,21 @@ public void Initialize()
         {
             if (_state.TryChange(State.Initial, State.Open))
             {
-                // the call to Task.Factory.StartNew is not normally recommended or necessary
-                // we are using it temporarily to work around a race condition in some of our tests
-                // the issue is that we set up some mocked async methods to return results immediately synchronously
-                // which results in the MonitorServerAsync method making more progress synchronously than the test expected
-                // by using Task.Factory.StartNew we introduce a short delay before the MonitorServerAsync Task starts executing
-                // the delay is whatever time it takes for the new Task to be activated and scheduled
-                // and the delay is usually long enough for the test to get past the race condition (though not guaranteed)
-                _ = Task.Factory.StartNew(() => _ = MonitorServerAsync().ConfigureAwait(false)).ConfigureAwait(false);
-                _ = _roundTripTimeMonitor.RunAsync().ConfigureAwait(false);
+                _roundTripTimeMonitor.Start();
+                _serverMonitorThread = new Thread(ThreadStart) { IsBackground = true };
+                _serverMonitorThread.Start();
+            }
+
+            void ThreadStart()
+            {
+                try
+                {
+                    MonitorServer();
+                }
+                catch (OperationCanceledException)
+                {
+                    // ignore OperationCanceledException
+                }
             }
         }
 
@@ -187,7 +195,7 @@ private CommandWireProtocol<BsonDocument> InitializeHelloProtocol(IConnection co
             return HelloHelper.CreateProtocol(helloCommand, _serverApi, commandResponseHandling);
         }
 
-        private async Task<IConnection> InitializeConnectionAsync(CancellationToken cancellationToken) // called setUpConnection in spec
+        private IConnection InitializeConnection(CancellationToken cancellationToken) // called setUpConnection in spec
         {
             var connection = _connectionFactory.CreateConnection(_serverId, _endPoint);
 
@@ -196,7 +204,7 @@ private async Task<IConnection> InitializeConnectionAsync(CancellationToken canc
             {
                 // if we are cancelling, it's because the server has
                 // been shut down and we really don't need to wait.
-                await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
+                connection.Open(cancellationToken);
             }
             catch
             {
@@ -210,7 +218,7 @@ private async Task<IConnection> InitializeConnectionAsync(CancellationToken canc
             return connection;
         }
 
-        private async Task MonitorServerAsync()
+        private void MonitorServer()
         {
             var metronome = new Metronome(_serverMonitorSettings.HeartbeatInterval);
             var monitorCancellationToken = _monitorCancellationTokenSource.Token;
@@ -227,7 +235,7 @@ private async Task MonitorServerAsync()
 
                     try
                     {
-                        await HeartbeatAsync(cachedHeartbeatCancellationToken).ConfigureAwait(false);
+                        Heartbeat(cachedHeartbeatCancellationToken);
                     }
                     catch (OperationCanceledException) when (cachedHeartbeatCancellationToken.IsCancellationRequested)
                     {
@@ -244,7 +252,7 @@ private async Task MonitorServerAsync()
                             {
                                 handler.Invoke(new SdamInformationEvent(() =>
                                     string.Format(
-                                        "Unexpected exception in ServerMonitor.MonitorServerAsync: {0}",
+                                        "Unexpected exception in ServerMonitor.MonitorServer: {0}",
                                         unexpectedException.ToString())));
                             }
                             catch
@@ -276,7 +284,7 @@ private async Task MonitorServerAsync()
                         }
                         _heartbeatDelay = newHeartbeatDelay;
                     }
-                    await newHeartbeatDelay.Task.ConfigureAwait(false); // corresponds to wait method in spec
+                    newHeartbeatDelay.Wait(monitorCancellationToken); // corresponds to wait method in spec
                 }
                 catch
                 {
@@ -285,7 +293,7 @@ private async Task MonitorServerAsync()
             }
         }
 
-        private async Task HeartbeatAsync(CancellationToken cancellationToken)
+        private void Heartbeat(CancellationToken cancellationToken)
         {
             CommandWireProtocol<BsonDocument> helloProtocol = null;
             bool processAnother = true;
@@ -304,7 +312,7 @@ private async Task HeartbeatAsync(CancellationToken cancellationToken)
                     }
                     if (connection == null)
                     {
-                        var initializedConnection = await InitializeConnectionAsync(cancellationToken).ConfigureAwait(false);
+                        var initializedConnection = InitializeConnection(cancellationToken);
                         lock (_lock)
                         {
                             if (_state.Value == State.Disposed)
@@ -326,7 +334,7 @@ private async Task HeartbeatAsync(CancellationToken cancellationToken)
                         {
                             helloProtocol = InitializeHelloProtocol(connection, previousDescription?.HelloOk ?? false);
                         }
-                        heartbeatHelloResult = await GetHelloResultAsync(connection, helloProtocol, cancellationToken).ConfigureAwait(false);
+                        heartbeatHelloResult = GetHelloResult(connection, helloProtocol, cancellationToken);
                     }
                 }
                 catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
@@ -420,7 +428,7 @@ bool IsNetworkError(Exception ex)
             }
         }
 
-        private async Task<HelloResult> GetHelloResultAsync(
+        private HelloResult GetHelloResult(
             IConnection connection,
             CommandWireProtocol<BsonDocument> helloProtocol,
             CancellationToken cancellationToken)
@@ -434,7 +442,7 @@ private async Task<HelloResult> GetHelloResultAsync(
             try
             {
                 var stopwatch = Stopwatch.StartNew();
-                var helloResult = await HelloHelper.GetResultAsync(connection, helloProtocol, cancellationToken).ConfigureAwait(false);
+                var helloResult = HelloHelper.GetResult(connection, helloProtocol, cancellationToken);
                 stopwatch.Stop();
 
                 if (_heartbeatSucceededEventHandler != null)