CSHARP-6029: Make ICoreServerSessionPool disposable and send endSessions on dispose

specifications/connection-monitoring-and-pooling/tests/logging/connection-logging.json

@@ -201,6 +201,73 @@
                 }
               }
             },
+            {
+              "level": "debug",
+              "component": "connection",
+              "data": {
+                "message": "Connection checkout started",
+                "serverHost": {
+                  "$$type": "string"
+                },
+                "serverPort": {
+                  "$$type": [
+                    "int",
+                    "long"
+                  ]
+                }
+              }
+            },
+            {
+              "level": "debug",
+              "component": "connection",
+              "data": {
+                "message": "Connection checked out",
+                "driverConnectionId": {
+                  "$$type": [
+                    "int",
+                    "long"
+                  ]
+                },
+                "serverHost": {
+                  "$$type": "string"
+                },
+                "serverPort": {
+                  "$$type": [
+                    "int",
+                    "long"
+                  ]
+                },
+                "durationMS": {
+                  "$$type": [
+                    "double",
+                    "int",
+                    "long"
+                  ]
+                }
+              }
+            },
+            {
+              "level": "debug",
+              "component": "connection",
+              "data": {
+                "message": "Connection checked in",
+                "driverConnectionId": {
+                  "$$type": [
+                    "int",
+                    "long"
+                  ]
+                },
+                "serverHost": {
+                  "$$type": "string"
+                },
+                "serverPort": {
+                  "$$type": [
+                    "int",
+                    "long"
+                  ]
+                }
+              }
+            },
             {
               "level": "debug",
               "component": "connection",

specifications/connection-monitoring-and-pooling/tests/logging/connection-logging.yml

@@ -85,14 +85,40 @@ tests:
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
 
+          #  The next three expected logs are for ending a session.
+          - level: debug
+            component: connection
+            data:
+              message: "Connection checkout started"
+              serverHost: { $$type: string }
+              serverPort: { $$type: [int, long] }
+
+          - level: debug
+            component: connection
+            data:
+              message: "Connection checked out"
+              driverConnectionId: { $$type: [int, long] }
+              serverHost: { $$type: string }
+              serverPort: { $$type: [int, long] }
+              durationMS: { $$type: [double, int, long] }
+
+          - level: debug
+            component: connection
+            data:
+              message: "Connection checked in"
+              driverConnectionId: { $$type: [int, long] }
+              serverHost: { $$type: string }
+              serverPort: { $$type: [int, long] }
+
           - level: debug
             component: connection
             data:
               message: "Connection closed"
               driverConnectionId: { $$type: [int, long] }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
-              reason: "Connection pool was closed"
+              reason: "Unknown" # CSharp Driver does not support closeReason
+              #reason: "Connection pool was closed"
 
           - level: debug
             component: connection
@@ -185,8 +211,9 @@ tests:
               driverConnectionId: { $$type: [int, long] }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
-              reason: "An error occurred while using the connection"
-              error: { $$exists: true }
+              reason: "Unknown" # CSharp Driver does not support closeReason
+              # reason: "An error occurred while using the connection"
+              # error: { $$exists: true }
 
           - level: debug
             component: connection

src/MongoDB.Driver/Core/Bindings/CoreServerSessionPool.cs

@@ -14,78 +14,148 @@
 */
 
 using System;
-using System.Collections.Generic;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using Microsoft.Extensions.Logging;
+using MongoDB.Bson;
+using MongoDB.Bson.IO;
+using MongoDB.Bson.Serialization.Serializers;
+using MongoDB.Driver.Core.Bindings;
 using MongoDB.Driver.Core.Clusters;
+using MongoDB.Driver.Core.Logging;
 using MongoDB.Driver.Core.Misc;
+using MongoDB.Driver.Core.Servers;
+using MongoDB.Driver.Core.WireProtocol;
 
 namespace MongoDB.Driver
 {
     internal sealed class CoreServerSessionPool : ICoreServerSessionPool
     {
         // private fields
         private readonly ICluster _cluster;
-        private readonly object _lock = new object();
-        private readonly List<ICoreServerSession> _pool = new List<ICoreServerSession>();
+        private readonly ILogger<LogCategories.Client> _logger;
+        private readonly ConcurrentStack<ICoreServerSession> _pool = new();
+        private volatile bool _isDisposed = false;
+        private long _sessionsCreated;
+        private long _sessionsDisposed;
+        private long _sessionsAcquired;
+        private long _sessionsReturned;
 
         // constructors
-        public CoreServerSessionPool(ICluster cluster)
+        public CoreServerSessionPool(ICluster cluster, ILogger<LogCategories.Client> logger)
         {
+            _logger = logger;
             _cluster = Ensure.IsNotNull(cluster, nameof(cluster));
         }
 
         public ICoreServerSession AcquireSession()
         {
-            lock (_lock)
+            ThrowIfDisposed();
+            ICoreServerSession session = null;
+            while (session == null && _pool.TryPop(out session))
             {
-                for (var i = _pool.Count - 1; i >= 0; i--)
+                if (IsAboutToExpireOrDirty(session))
                 {
-                    var pooledSession = _pool[i];
-                    if (IsAboutToExpireOrDirty(pooledSession))
-                    {
-                        pooledSession.Dispose();
-                    }
-                    else
-                    {
-                        var removeCount = _pool.Count - i; // the one we're about to return and any about to expire ones we skipped over
-                        _pool.RemoveRange(i, removeCount);
-                        return new ReleaseOnDisposeCoreServerSession(pooledSession, this);
-                    }
+                    session.Dispose();
+                    session = null;
                 }
+            }
 
-                _pool.Clear(); // they're all about to expire
+            if (session == null)
+            {
+                Interlocked.Increment(ref _sessionsCreated);
+                session = new CoreServerSession();
             }
 
-            return new ReleaseOnDisposeCoreServerSession(new CoreServerSession(), this);
+            Interlocked.Increment(ref _sessionsAcquired);
+            return new ReleaseOnDisposeCoreServerSession(session, this);
         }
 
         public void ReleaseSession(ICoreServerSession session)
         {
-            lock (_lock)
+            if (_isDisposed)
+            {
+                session.Dispose();
+                _logger?.LogError("Cannot release session because the server session pool for cluster {clusterId} has been disposed.", _cluster.ClusterId);
+                return;
+            }
+
+            Interlocked.Increment(ref _sessionsReturned);
+
+            if (IsAboutToExpireOrDirty(session))
+            {
+                Interlocked.Increment(ref _sessionsDisposed);
+                session.Dispose();
+            }
+            else
+            {
+                _pool.Push(session);
+            }
+        }
+
+        public void CloseAndDispose(IServer server)
+        {
+            if (_isDisposed)
+            {
+                return;
+            }
+
+            _isDisposed = true;
+            var timestamp = Stopwatch.GetTimestamp();
+            _logger?.LogDebug(
+                "Closing server session pool for {clusterId}: total sessions created {sessionsCreated}, total sessions acquired {sessionsAcquired}, sessions returned {sessionsReturned}, sessions disposed {sessionsDisposed}, pooled sessions {pooledSessions}.",
+                _cluster.ClusterId, _sessionsCreated, _sessionsAcquired, _sessionsReturned, _sessionsDisposed, _pool.Count);
+
+            var sessionsEnded = 0;
+            try
             {
-                var removeCount = 0;
-                for (var i = 0; i < _pool.Count; i++)
+                while (true)
                 {
-                    var pooledSession = _pool[i];
-                    if (IsAboutToExpireOrDirty(pooledSession))
+                    var batchSize = Math.Min(10000, _pool.Count);
+                    var batch = new ICoreServerSession[batchSize];
+
+                    batchSize = _pool.TryPopRange(batch);
+                    if (batchSize == 0)
                     {
-                        pooledSession.Dispose();
-                        removeCount++;
+                        return;
                     }
-                    else
+
+                    var endSessionCommand = new BsonDocument("endSessions", new BsonArray(batch.Take(batchSize).Select(s => s.Id)));
+                    var operationContext = OperationContext.NoTimeout;
+                    using var channel = server.GetChannel(operationContext);
+                    channel.Command(
+                        operationContext,
+                        NoCoreSession.Instance,
+                        ReadPreference.PrimaryPreferred,
+                        DatabaseNamespace.Admin,
+                        endSessionCommand,
+                        null,
+                        NoOpElementNameValidator.Instance,
+                        null,
+                        null,
+                        CommandResponseHandling.Return,
+                        BsonDocumentSerializer.Instance,
+                        null);
+
+                    sessionsEnded += batchSize;
+
+                    for (var i = 0; i < batchSize; i++)
                     {
-                        break;
+                        batch[i].Dispose();
                     }
                 }
-                _pool.RemoveRange(0, removeCount);
-
-                if (IsAboutToExpireOrDirty(session))
-                {
-                    session.Dispose();
-                }
-                else
-                {
-                    _pool.Add(session);
-                }
+            }
+            catch(Exception ex)
+            {
+                _logger?.LogError(ex, "Error closing server session pool for {clusterId}.", _cluster.ClusterId);
+            }
+            finally
+            {
+                _logger?.LogDebug(
+                    "Closed server session pool for {clusterId} in {milliseconds}ms, total sessions ended {sessionsEnded}.",
+                    _cluster.ClusterId, (Stopwatch.GetTimestamp() - timestamp) / (double)Stopwatch.Frequency * 1000, sessionsEnded);
             }
         }
 
@@ -116,6 +186,14 @@ private bool IsAboutToExpireOrDirty(ICoreServerSession session)
             return IsAboutToExpire(session) || session.IsDirty;
         }
 
+        private void ThrowIfDisposed()
+        {
+            if (_isDisposed)
+            {
+                throw new ObjectDisposedException(nameof(CoreServerSessionPool));
+            }
+        }
+
         // nested types
         internal sealed class ReleaseOnDisposeCoreServerSession : WrappingCoreServerSession
         {

src/MongoDB.Driver/Core/Bindings/ICoreServerSessionPool.cs

@@ -13,11 +13,13 @@
 * limitations under the License.
 */
 
-namespace MongoDB.Driver
+using MongoDB.Driver.Core.Servers;
+
+namespace MongoDB.Driver;
+
+internal interface ICoreServerSessionPool
 {
-    internal interface ICoreServerSessionPool
-    {
-        ICoreServerSession AcquireSession();
-        void ReleaseSession(ICoreServerSession serverSession);
-    }
+    ICoreServerSession AcquireSession();
+    void ReleaseSession(ICoreServerSession serverSession);
+    void CloseAndDispose(IServer server);
 }

src/MongoDB.Driver/Core/Clusters/Cluster.cs

@@ -67,7 +67,7 @@ protected Cluster(ClusterSettings settings, IClusterableServerFactory serverFact
             _expirableClusterDescription = new (this, ClusterDescription.CreateInitial(_clusterId, _settings.DirectConnection));
             _latencyLimitingServerSelector = new LatencyLimitingServerSelector(settings.LocalThreshold);
             _serverSelectionWaitQueue = new ServerSelectionWaitQueue(this);
-            _serverSessionPool = new CoreServerSessionPool(this);
+            _serverSessionPool = new CoreServerSessionPool(this, loggerFactory?.CreateLogger<LogCategories.Client>());
             _clusterEventLogger = loggerFactory.CreateEventLogger<LogCategories.SDAM>(eventSubscriber);
             _serverSelectionEventLogger = loggerFactory.CreateEventLogger<LogCategories.ServerSelection>(eventSubscriber);
         }
@@ -131,6 +131,21 @@ public virtual void Initialize()
             }
         }
 
+        protected void ReleaseServerSessionPool()
+        {
+            // Do the sessionPool cleanup only if we have a server available immediately, do not have to wait here.
+            var server = Servers.FirstOrDefault(x => x.Description.State == ServerState.Connected && x.Description.Type == ServerType.ReplicaSetPrimary);
+            if (server == null)
+            {
+                server = Servers.FirstOrDefault(x => x.Description.State == ServerState.Connected);
+            }
+
+            if (server != null)
+            {
+                _serverSessionPool.CloseAndDispose(server);
+            }
+        }
+
         protected abstract void RequestHeartbeat();
 
         protected void OnDescriptionChanged(ClusterDescription oldDescription, ClusterDescription newDescription, bool shouldClusterDescriptionChangedEventBePublished)

src/MongoDB.Driver/Core/Clusters/LoadBalancedCluster.cs

@@ -86,7 +86,7 @@ public LoadBalancedCluster(
             _serverFactory = Ensure.IsNotNull(serverFactory, nameof(serverFactory));
             _serverReadyTaskCompletionSource = new TaskCompletionSource<bool>();
 
-            _serverSessionPool = new CoreServerSessionPool(this);
+            _serverSessionPool = new CoreServerSessionPool(this, loggerFactory?.CreateLogger<LogCategories.Client>());
 
             _state = new InterlockedInt32(State.Initial);
 
@@ -131,6 +131,8 @@ private void Dispose(bool disposing)
                     var stopwatch = Stopwatch.StartNew();
                     if (_server != null)
                     {
+                        _serverSessionPool.CloseAndDispose(_server);
+
                         _server.DescriptionChanged -= ServerDescriptionChangedHandler;
                         _server.Dispose();
                     }