CSHARP-2427: Implement Connection Monitoring and Pooling.

Author: DmitryLukyanov

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

@@ -136,6 +136,7 @@ public static ClusterBuilder ConfigureWithConnectionString(this ClusterBuilder b
             {
                 builder = builder.ConfigureConnectionPool(s => s.With(minConnections: connectionString.MinPoolSize.Value));
             }
+#pragma warning disable 618
             if (connectionString.WaitQueueSize != null)
             {
                 builder = builder.ConfigureConnectionPool(s => s.With(waitQueueSize: connectionString.WaitQueueSize.Value));
@@ -146,6 +147,7 @@ public static ClusterBuilder ConfigureWithConnectionString(this ClusterBuilder b
                 var waitQueueSize = (int)Math.Round(maxConnections * connectionString.WaitQueueMultiple.Value);
                 builder = builder.ConfigureConnectionPool(s => s.With(waitQueueSize: waitQueueSize));
             }
+#pragma warning restore 618
             if (connectionString.WaitQueueTimeout != null)
             {
                 builder = builder.ConfigureConnectionPool(s => s.With(waitQueueTimeout: connectionString.WaitQueueTimeout.Value));

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

@@ -93,6 +93,7 @@ public int MinConnections
         /// <value>
         /// The size of the wait queue.
         /// </value>
+        [Obsolete("This property will be removed in a later release.")]
         public int WaitQueueSize
         {
             get { return _waitQueueSize; }

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

@@ -451,6 +451,7 @@ public GuidRepresentation? UuidRepresentation
         /// <summary>
         /// Gets the wait queue multiple.
         /// </summary>
+        [Obsolete("This property will be removed in a later release.")]
         public double? WaitQueueMultiple
         {
             get { return _waitQueueMultiple; }
@@ -459,6 +460,7 @@ public double? WaitQueueMultiple
         /// <summary>
         /// Gets the wait queue size.
         /// </summary>
+        [Obsolete("This property will be removed in a later release.")]
         public int? WaitQueueSize
         {
             get { return _waitQueueSize; }

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

@@ -54,6 +54,9 @@ internal sealed class ExclusiveConnectionPool : IConnectionPool
         private readonly Action<ConnectionPoolOpenedEvent> _openedEventHandler;
         private readonly Action<ConnectionPoolClosingEvent> _closingEventHandler;
         private readonly Action<ConnectionPoolClosedEvent> _closedEventHandler;
+        private readonly Action<ConnectionPoolClearingEvent> _clearingEventHandler;
+        private readonly Action<ConnectionPoolClearedEvent> _clearedEventHandler;
+        private readonly Action<ConnectionCreatedEvent> _connectionCreatedEventHandler;
 
         // constructors
         public ExclusiveConnectionPool(
@@ -71,7 +74,9 @@ public ExclusiveConnectionPool(
 
             _connectionHolder = new ListConnectionHolder(eventSubscriber);
             _poolQueue = new WaitQueue(settings.MaxConnections);
+#pragma warning disable 618
             _waitQueue = new SemaphoreSlim(settings.WaitQueueSize);
+#pragma warning restore 618
             _maintenanceCancellationTokenSource = new CancellationTokenSource();
             _state = new InterlockedInt32(State.Initial);
 
@@ -88,6 +93,9 @@ public ExclusiveConnectionPool(
             eventSubscriber.TryGetEventHandler(out _closedEventHandler);
             eventSubscriber.TryGetEventHandler(out _addingConnectionEventHandler);
             eventSubscriber.TryGetEventHandler(out _addedConnectionEventHandler);
+            eventSubscriber.TryGetEventHandler(out _clearingEventHandler);
+            eventSubscriber.TryGetEventHandler(out _clearedEventHandler);
+            eventSubscriber.TryGetEventHandler(out _connectionCreatedEventHandler);
         }
 
         // properties
@@ -140,12 +148,12 @@ public int UsedCount
         // public methods
         public IConnectionHandle AcquireConnection(CancellationToken cancellationToken)
         {
-            ThrowIfNotOpen();
-
             var helper = new AcquireConnectionHelper(this);
             try
             {
                 helper.CheckingOutConnection();
+                ThrowIfNotOpen();
+                helper.EnterWaitQueue();
                 var enteredPool = _poolQueue.Wait(_settings.WaitQueueTimeout, cancellationToken);
                 return helper.EnteredPool(enteredPool);
             }
@@ -162,12 +170,12 @@ public IConnectionHandle AcquireConnection(CancellationToken cancellationToken)
 
         public async Task<IConnectionHandle> AcquireConnectionAsync(CancellationToken cancellationToken)
         {
-            ThrowIfNotOpen();
-
             var helper = new AcquireConnectionHelper(this);
             try
             {
                 helper.CheckingOutConnection();
+                ThrowIfNotOpen();
+                helper.EnterWaitQueue();
                 var enteredPool = await _poolQueue.WaitAsync(_settings.WaitQueueTimeout, cancellationToken).ConfigureAwait(false);
                 return helper.EnteredPool(enteredPool);
             }
@@ -185,13 +193,20 @@ public async Task<IConnectionHandle> AcquireConnectionAsync(CancellationToken ca
         public void Clear()
         {
             ThrowIfNotOpen();
+
+            _clearingEventHandler?.Invoke(new ConnectionPoolClearingEvent(_serverId, _settings));
+
             Interlocked.Increment(ref _generation);
+
+            _clearedEventHandler?.Invoke(new ConnectionPoolClearedEvent(_serverId, _settings));
         }
 
         private PooledConnection CreateNewConnection()
         {
             var connection = _connectionFactory.CreateConnection(_serverId, _endPoint);
-            return new PooledConnection(this, connection);
+            var pooledConnection = new PooledConnection(this, connection);
+            _connectionCreatedEventHandler?.Invoke(new ConnectionCreatedEvent(connection.ConnectionId, connection.Settings, EventContext.OperationId));
+            return pooledConnection;
         }
 
         public void Initialize()
@@ -204,12 +219,12 @@ public void Initialize()
                     _openingEventHandler(new ConnectionPoolOpeningEvent(_serverId, _settings));
                 }
 
-                MaintainSizeAsync().ConfigureAwait(false);
-
                 if (_openedEventHandler != null)
                 {
                     _openedEventHandler(new ConnectionPoolOpenedEvent(_serverId, _settings));
                 }
+
+                MaintainSizeAsync().ConfigureAwait(false);
             }
         }
 
@@ -335,25 +350,28 @@ private async Task EnsureMinSizeAsync(CancellationToken cancellationToken)
 
         private void ReleaseConnection(PooledConnection connection)
         {
-            if (_state.Value == State.Disposed)
+            if (_checkingInConnectionEventHandler != null)
             {
-                connection.Dispose();
-                return;
+                _checkingInConnectionEventHandler(new ConnectionPoolCheckingInConnectionEvent(connection.ConnectionId, EventContext.OperationId));
             }
 
-            if (_checkingInConnectionEventHandler != null)
+            if (_checkedInConnectionEventHandler != null)
             {
-                _checkingInConnectionEventHandler(new ConnectionPoolCheckingInConnectionEvent(connection.ConnectionId, EventContext.OperationId));
+                _checkedInConnectionEventHandler(new ConnectionPoolCheckedInConnectionEvent(connection.ConnectionId, TimeSpan.Zero, EventContext.OperationId));
             }
 
-            var stopwatch = Stopwatch.StartNew();
-            _connectionHolder.Return(connection);
-            _poolQueue.Release();
-            stopwatch.Stop();
+            if (!connection.IsExpired && _state.Value != State.Disposed)
+            {
+                _connectionHolder.Return(connection);
+            }
+            else
+            {
+                _connectionHolder.RemoveConnection(connection);
+            }
 
-            if (_checkedInConnectionEventHandler != null)
+            if (_state.Value != State.Disposed)
             {
-                _checkedInConnectionEventHandler(new ConnectionPoolCheckedInConnectionEvent(connection.ConnectionId, stopwatch.Elapsed, EventContext.OperationId));
+                _poolQueue.Release();
             }
         }
 
@@ -404,7 +422,10 @@ public void CheckingOutConnection()
                 {
                     handler(new ConnectionPoolCheckingOutConnectionEvent(_pool._serverId, EventContext.OperationId));
                 }
+            }
 
+            public void EnterWaitQueue()
+            {
                 _enteredWaitQueue = _pool._waitQueue.Wait(0); // don't wait...
                 if (!_enteredWaitQueue)
                 {
@@ -467,7 +488,14 @@ public void HandleException(Exception ex)
                 var handler = _pool._checkingOutConnectionFailedEventHandler;
                 if (handler != null)
                 {
-                    handler(new ConnectionPoolCheckingOutConnectionFailedEvent(_pool._serverId, ex, EventContext.OperationId));
+                    ConnectionCheckOutFailedReason reason;
+                    switch (ex)
+                    {
+                        case ObjectDisposedException _: reason = ConnectionCheckOutFailedReason.PoolClosed; break;
+                        case TimeoutException _: reason = ConnectionCheckOutFailedReason.Timeout; break;
+                        default: reason = ConnectionCheckOutFailedReason.ConnectionError; break;
+                    }
+                    handler(new ConnectionPoolCheckingOutConnectionFailedEvent(_pool._serverId, ex, EventContext.OperationId, reason));
                 }
             }
 
@@ -788,19 +816,13 @@ public PooledConnection Acquire()
 
             public void Return(PooledConnection connection)
             {
-                if (connection.IsExpired)
-                {
-                    RemoveConnection(connection);
-                    return;
-                }
-
                 lock (_lock)
                 {
                     _connections.Add(connection);
                 }
             }
 
-            private void RemoveConnection(PooledConnection connection)
+            public void RemoveConnection(PooledConnection connection)
             {
                 if (_removingConnectionEventHandler != null)
                 {

src/MongoDB.Driver.Core/Core/Events/ClusterAddedServerEvent.cs

@@ -19,7 +19,6 @@
 
 namespace MongoDB.Driver.Core.Events
 {
-    /// <preliminary/>
     /// <summary>
     /// Occurs after a server is added to the cluster.
     /// </summary>

src/MongoDB.Driver.Core/Core/Events/ClusterAddingServerEvent.cs

@@ -18,7 +18,6 @@
 
 namespace MongoDB.Driver.Core.Events
 {
-    /// <preliminary/>
     /// <summary>
     /// Occurs before a server is added to the cluster.
     /// </summary>

src/MongoDB.Driver.Core/Core/Events/ClusterClosedEvent.cs

@@ -18,7 +18,6 @@
 
 namespace MongoDB.Driver.Core.Events
 {
-    /// <preliminary/>
     /// <summary>
     /// Occurs after a cluster is closed.
     /// </summary>

src/MongoDB.Driver.Core/Core/Events/ClusterClosingEvent.cs

@@ -17,7 +17,6 @@
 
 namespace MongoDB.Driver.Core.Events
 {
-    /// <preliminary/>
     /// <summary>
     /// Occurs before a cluster is closed.
     /// </summary>

src/MongoDB.Driver.Core/Core/Events/ClusterDescriptionChangedEvent.cs

@@ -18,7 +18,6 @@
 
 namespace MongoDB.Driver.Core.Events
 {
-    /// <preliminary/>
     /// <summary>
     /// Occurs when a cluster has changed.
     /// </summary>

src/MongoDB.Driver.Core/Core/Events/ClusterOpenedEvent.cs

@@ -19,7 +19,6 @@
 
 namespace MongoDB.Driver.Core.Events
 {
-    /// <preliminary/>
     /// <summary>
     /// Occurs after a cluster is opened.
     /// </summary>