CSHARP-2964: Reduce Client Time To Recovery On Topology Changes.

Author: DmitryLukyanov

src/MongoDB.Driver.Core/Core/Configuration/ClusterBuilder.cs

@@ -272,14 +272,17 @@ private IServerMonitorFactory CreateServerMonitorFactory()
                 );
 
             var serverMonitorStreamFactory = CreateTcpStreamFactory(serverMonitorTcpStreamSettings);
+            var serverMonitorSettings = new ServerMonitorSettings(
+                connectTimeout: serverMonitorTcpStreamSettings.ConnectTimeout,
+                heartbeatInterval: _serverSettings.HeartbeatInterval);
 
             var serverMonitorConnectionFactory = new BinaryConnectionFactory(
                 serverMonitorConnectionSettings,
                 serverMonitorStreamFactory,
                 new EventAggregator());
 
             return new ServerMonitorFactory(
-                _serverSettings,
+                serverMonitorSettings,
                 serverMonitorConnectionFactory,
                 _eventAggregator);
         }

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

@@ -612,6 +612,11 @@ public Task SendMessagesAsync(IEnumerable<RequestMessage> messages, MessageEncod
             {
                 return _connection.SendMessagesAsync(messages, messageEncoderSettings, cancellationToken);
             }
+
+            public void SetReadTimeout(TimeSpan timeout)
+            {
+                _connection.SetReadTimeout(timeout);
+            }
         }
 
         private sealed class AcquiredConnection : IConnectionHandle
@@ -711,6 +716,12 @@ public Task SendMessagesAsync(IEnumerable<RequestMessage> messages, MessageEncod
                 return _reference.Instance.SendMessagesAsync(messages, messageEncoderSettings, cancellationToken);
             }
 
+            public void SetReadTimeout(TimeSpan timeout)
+            {
+                ThrowIfDisposed();
+                _reference.Instance.SetReadTimeout(timeout);
+            }
+
             private void ThrowIfDisposed()
             {
                 if (_disposed)

src/MongoDB.Driver.Core/Core/Connections/BinaryConnection.cs

@@ -40,8 +40,6 @@ namespace MongoDB.Driver.Core.Connections
     internal class BinaryConnection : IConnection
     {
         // fields
-        private readonly CancellationToken _backgroundTaskCancellationToken;
-        private readonly CancellationTokenSource _backgroundTaskCancellationTokenSource;
         private readonly CommandEventHelper _commandEventHelper;
         private readonly ICompressorSource _compressorSource;
         private ConnectionId _connectionId;
@@ -85,9 +83,6 @@ public BinaryConnection(ServerId serverId, EndPoint endPoint, ConnectionSettings
             _connectionInitializer = Ensure.IsNotNull(connectionInitializer, nameof(connectionInitializer));
             Ensure.IsNotNull(eventSubscriber, nameof(eventSubscriber));
 
-            _backgroundTaskCancellationTokenSource = new CancellationTokenSource();
-            _backgroundTaskCancellationToken = _backgroundTaskCancellationTokenSource.Token;
-
             _connectionId = new ConnectionId(serverId);
             _receiveLock = new SemaphoreSlim(1);
             _sendLock = new SemaphoreSlim(1);
@@ -199,8 +194,6 @@ private void Dispose(bool disposing)
                     }
 
                     var stopwatch = Stopwatch.StartNew();
-                    _backgroundTaskCancellationTokenSource.Cancel();
-                    _backgroundTaskCancellationTokenSource.Dispose();
                     _receiveLock.Dispose();
                     _sendLock.Dispose();
 
@@ -237,7 +230,7 @@ private void EnsureMessageSizeIsValid(int messageSize)
 
         public void Open(CancellationToken cancellationToken)
         {
-            ThrowIfDisposed();
+            ThrowIfCancelledOrDisposed(cancellationToken);
 
             TaskCompletionSource<bool> taskCompletionSource = null;
             var connecting = false;
@@ -274,7 +267,7 @@ public void Open(CancellationToken cancellationToken)
 
         public Task OpenAsync(CancellationToken cancellationToken)
         {
-            ThrowIfDisposed();
+            ThrowIfCancelledOrDisposed(cancellationToken);
 
             lock (_openLock)
             {
@@ -329,19 +322,19 @@ private async Task OpenHelperAsync(CancellationToken cancellationToken)
             }
         }
 
-        private IByteBuffer ReceiveBuffer()
+        private IByteBuffer ReceiveBuffer(CancellationToken cancellationToken)
         {
             try
             {
                 var messageSizeBytes = new byte[4];
-                _stream.ReadBytes(messageSizeBytes, 0, 4, _backgroundTaskCancellationToken);
+                _stream.ReadBytes(messageSizeBytes, 0, 4, cancellationToken);
                 var messageSize = BitConverter.ToInt32(messageSizeBytes, 0);
                 EnsureMessageSizeIsValid(messageSize);
                 var inputBufferChunkSource = new InputBufferChunkSource(BsonChunkPool.Default);
                 var buffer = ByteBufferFactory.Create(inputBufferChunkSource, messageSize);
                 buffer.Length = messageSize;
                 buffer.SetBytes(0, messageSizeBytes, 0, 4);
-                _stream.ReadBytes(buffer, 4, messageSize - 4, _backgroundTaskCancellationToken);
+                _stream.ReadBytes(buffer, 4, messageSize - 4, cancellationToken);
                 _lastUsedAtUtc = DateTime.UtcNow;
                 buffer.MakeReadOnly();
                 return buffer;
@@ -370,7 +363,7 @@ private IByteBuffer ReceiveBuffer(int responseTo, CancellationToken cancellation
                     receiveLockRequest.Task.GetAwaiter().GetResult(); // propagate exceptions
                     while (true)
                     {
-                        var buffer = ReceiveBuffer();
+                        var buffer = ReceiveBuffer(cancellationToken);
                         _dropbox.AddMessage(buffer);
 
                         if (messageTask.IsCompleted)
@@ -391,20 +384,20 @@ private IByteBuffer ReceiveBuffer(int responseTo, CancellationToken cancellation
             }
         }
 
-        private async Task<IByteBuffer> ReceiveBufferAsync()
+        private async Task<IByteBuffer> ReceiveBufferAsync(CancellationToken cancellationToken)
         {
             try
             {
                 var messageSizeBytes = new byte[4];
                 var readTimeout = _stream.CanTimeout ? TimeSpan.FromMilliseconds(_stream.ReadTimeout) : Timeout.InfiniteTimeSpan;
-                await _stream.ReadBytesAsync(messageSizeBytes, 0, 4, readTimeout, _backgroundTaskCancellationToken).ConfigureAwait(false);
+                await _stream.ReadBytesAsync(messageSizeBytes, 0, 4, readTimeout, cancellationToken).ConfigureAwait(false);
                 var messageSize = BitConverter.ToInt32(messageSizeBytes, 0);
                 EnsureMessageSizeIsValid(messageSize);
                 var inputBufferChunkSource = new InputBufferChunkSource(BsonChunkPool.Default);
                 var buffer = ByteBufferFactory.Create(inputBufferChunkSource, messageSize);
                 buffer.Length = messageSize;
                 buffer.SetBytes(0, messageSizeBytes, 0, 4);
-                await _stream.ReadBytesAsync(buffer, 4, messageSize - 4, readTimeout, _backgroundTaskCancellationToken).ConfigureAwait(false);
+                await _stream.ReadBytesAsync(buffer, 4, messageSize - 4, readTimeout, cancellationToken).ConfigureAwait(false);
                 _lastUsedAtUtc = DateTime.UtcNow;
                 buffer.MakeReadOnly();
                 return buffer;
@@ -433,7 +426,7 @@ private async Task<IByteBuffer> ReceiveBufferAsync(int responseTo, CancellationT
                     receiveLockRequest.Task.GetAwaiter().GetResult(); // propagate exceptions
                     while (true)
                     {
-                        var buffer = await ReceiveBufferAsync().ConfigureAwait(false);
+                        var buffer = await ReceiveBufferAsync(cancellationToken).ConfigureAwait(false);
                         _dropbox.AddMessage(buffer);
 
                         if (messageTask.IsCompleted)
@@ -461,7 +454,7 @@ public ResponseMessage ReceiveMessage(
             CancellationToken cancellationToken)
         {
             Ensure.IsNotNull(encoderSelector, nameof(encoderSelector));
-            ThrowIfDisposedOrNotOpen();
+            ThrowIfCancelledOrDisposedOrNotOpen(cancellationToken);
 
             var helper = new ReceiveMessageHelper(this, responseTo, messageEncoderSettings, _compressorSource);
             try
@@ -477,7 +470,7 @@ public ResponseMessage ReceiveMessage(
             catch (Exception ex)
             {
                 helper.FailedReceivingMessage(ex);
-                throw;
+                throw WrapInOperationCanceledExceptionIfRequired(ex, cancellationToken);
             }
         }
 
@@ -488,7 +481,7 @@ public async Task<ResponseMessage> ReceiveMessageAsync(
             CancellationToken cancellationToken)
         {
             Ensure.IsNotNull(encoderSelector, nameof(encoderSelector));
-            ThrowIfDisposedOrNotOpen();
+            ThrowIfCancelledOrDisposedOrNotOpen(cancellationToken);
 
             var helper = new ReceiveMessageHelper(this, responseTo, messageEncoderSettings, _compressorSource);
             try
@@ -504,7 +497,7 @@ public async Task<ResponseMessage> ReceiveMessageAsync(
             catch (Exception ex)
             {
                 helper.FailedReceivingMessage(ex);
-                throw;
+                throw WrapInOperationCanceledExceptionIfRequired(ex, cancellationToken);
             }
         }
 
@@ -520,8 +513,7 @@ private void SendBuffer(IByteBuffer buffer, CancellationToken cancellationToken)
 
                 try
                 {
-                    // don't use the caller's cancellationToken because once we start writing a message we have to write the whole thing
-                    _stream.WriteBytes(buffer, 0, buffer.Length, _backgroundTaskCancellationToken);
+                    _stream.WriteBytes(buffer, 0, buffer.Length, cancellationToken);
                     _lastUsedAtUtc = DateTime.UtcNow;
                 }
                 catch (Exception ex)
@@ -549,9 +541,8 @@ private async Task SendBufferAsync(IByteBuffer buffer, CancellationToken cancell
 
                 try
                 {
-                    // don't use the caller's cancellationToken because once we start writing a message we have to write the whole thing
                     var writeTimeout = _stream.CanTimeout ? TimeSpan.FromMilliseconds(_stream.WriteTimeout) : Timeout.InfiniteTimeSpan;
-                    await _stream.WriteBytesAsync(buffer, 0, buffer.Length, writeTimeout, _backgroundTaskCancellationToken).ConfigureAwait(false);
+                    await _stream.WriteBytesAsync(buffer, 0, buffer.Length, writeTimeout, cancellationToken).ConfigureAwait(false);
                     _lastUsedAtUtc = DateTime.UtcNow;
                 }
                 catch (Exception ex)
@@ -570,7 +561,7 @@ private async Task SendBufferAsync(IByteBuffer buffer, CancellationToken cancell
         public void SendMessages(IEnumerable<RequestMessage> messages, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
         {
             Ensure.IsNotNull(messages, nameof(messages));
-            ThrowIfDisposedOrNotOpen();
+            ThrowIfCancelledOrDisposedOrNotOpen(cancellationToken);
 
             var helper = new SendMessagesHelper(this, messages, messageEncoderSettings);
             try
@@ -599,14 +590,14 @@ public void SendMessages(IEnumerable<RequestMessage> messages, MessageEncoderSet
             catch (Exception ex)
             {
                 helper.FailedSendingMessages(ex);
-                throw;
+                throw WrapInOperationCanceledExceptionIfRequired(ex, cancellationToken);
             }
         }
 
         public async Task SendMessagesAsync(IEnumerable<RequestMessage> messages, MessageEncoderSettings messageEncoderSettings, CancellationToken cancellationToken)
         {
             Ensure.IsNotNull(messages, nameof(messages));
-            ThrowIfDisposedOrNotOpen();
+            ThrowIfCancelledOrDisposedOrNotOpen(cancellationToken);
 
             var helper = new SendMessagesHelper(this, messages, messageEncoderSettings);
             try
@@ -635,10 +626,16 @@ public async Task SendMessagesAsync(IEnumerable<RequestMessage> messages, Messag
             catch (Exception ex)
             {
                 helper.FailedSendingMessages(ex);
-                throw;
+                throw WrapInOperationCanceledExceptionIfRequired(ex, cancellationToken);
             }
         }
 
+        public void SetReadTimeout(TimeSpan timeout)
+        {
+            ThrowIfDisposed();
+            _stream.ReadTimeout = (int)timeout.TotalMilliseconds;
+        }
+
         // private methods
         private bool AnyMessageNeedsToBeCompressed(IEnumerable<RequestMessage> messages)
         {
@@ -698,17 +695,15 @@ private void CompressMessage(
             compressedMessageEncoder.WriteMessage(compressedMessage);
         }
 
-        private void ThrowIfDisposed()
+        private void ThrowIfCancelledOrDisposed(CancellationToken cancellationToken = default)
         {
-            if (_state.Value == State.Disposed)
-            {
-                throw new ObjectDisposedException(GetType().Name);
-            }
+            cancellationToken.ThrowIfCancellationRequested();
+            ThrowIfDisposed();
         }
 
-        private void ThrowIfDisposedOrNotOpen()
+        private void ThrowIfCancelledOrDisposedOrNotOpen(CancellationToken cancellationToken)
         {
-            ThrowIfDisposed();
+            ThrowIfCancelledOrDisposed(cancellationToken);
             if (_state.Value == State.Failed)
             {
                 throw new MongoConnectionClosedException(_connectionId);
@@ -719,6 +714,14 @@ private void ThrowIfDisposedOrNotOpen()
             }
         }
 
+        private void ThrowIfDisposed()
+        {
+            if (_state.Value == State.Disposed)
+            {
+                throw new ObjectDisposedException(GetType().Name);
+            }
+        }
+
         private Exception WrapException(Exception ex, string action)
         {
             if (
@@ -727,7 +730,9 @@ ex is ThreadAbortException ||
                 ex is StackOverflowException ||
 #endif
                 ex is MongoAuthenticationException ||
-                ex is OutOfMemoryException)
+                ex is OutOfMemoryException ||
+                ex is OperationCanceledException ||
+                ex is ObjectDisposedException)
             {
                 return ex;
             }
@@ -738,6 +743,23 @@ ex is MongoAuthenticationException ||
             }
         }
 
+        private Exception WrapInOperationCanceledExceptionIfRequired(Exception exception, CancellationToken cancellationToken)
+        {
+            if (exception is ObjectDisposedException objectDisposedException)
+            {
+                // We expect two cases here:
+                //      objectDisposedException.ObjectName == GetType().Name
+                //      objectDisposedException.Message == "The semaphore has been disposed."
+                // but since the last one is language-specific, the only option we have is avoiding any additional conditions for ObjectDisposedException
+                // TODO: this logic should be reviewed in the scope of https://jira.mongodb.org/browse/CSHARP-3165
+                return new OperationCanceledException($"The {nameof(BinaryConnection)} operation has been cancelled.", exception);
+            }
+            else
+            {
+                return exception;
+            }
+        }
+
         // nested classes
         private class Dropbox
         {

src/MongoDB.Driver.Core/Core/Connections/IConnection.cs

@@ -15,12 +15,9 @@
 
 using System;
 using System.Collections.Generic;
-using System.Linq;
 using System.Net;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
-using MongoDB.Bson.Serialization;
 using MongoDB.Driver.Core.Configuration;
 using MongoDB.Driver.Core.WireProtocol.Messages;
 using MongoDB.Driver.Core.WireProtocol.Messages.Encoders;
@@ -81,6 +78,12 @@ public interface IConnection : IDisposable
         /// </value>
         ConnectionSettings Settings { get; }
 
+        /// <summary>
+        /// Set read timeout value.
+        /// </summary>
+        /// <param name="timeout">The timeout.</param>
+        void SetReadTimeout(TimeSpan timeout);
+
         // methods
         /// <summary>
         /// Opens the connection.