CSHARP-2188: Decrease likelihood of session leaks by releasing the

session as soon as possible in AsyncCursor.

Author: rstam

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

@@ -0,0 +1,105 @@
+/* Copyright 2018-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 MongoDB.Bson;
+using MongoDB.Driver.Core.Clusters;
+using MongoDB.Driver.Core.Misc;
+using MongoDB.Driver.Core.Operations;
+
+namespace MongoDB.Driver.Core.Bindings
+{
+    /// <summary>
+    /// Represents a session.
+    /// </summary>
+    /// <seealso cref="MongoDB.Driver.Core.Bindings.ICoreSession" />
+    internal sealed class CoreSession : ICoreSession
+    {
+        // private fields
+        private readonly IClusterClock _clusterClock = new ClusterClock();
+        private bool _disposed;
+        private readonly bool _isCausallyConsistent;
+        private readonly bool _isImplicit;
+        private readonly IOperationClock _operationClock = new OperationClock();
+        private readonly ICoreServerSession _serverSession;
+
+        // constructors
+        /// <summary>
+        /// Initializes a new instance of the <see cref="CoreSession" /> class.
+        /// </summary>
+        /// <param name="serverSession">The server session.</param>
+        /// <param name="isCausallyConsistent">if set to <c>true</c> [is causally consistent].</param>
+        /// <param name="isImplicit">if set to <c>true</c> [is implicit].</param>
+        public CoreSession(
+            ICoreServerSession serverSession,
+            bool isCausallyConsistent = false,
+            bool isImplicit = false)
+        {
+            _serverSession = Ensure.IsNotNull(serverSession, nameof(serverSession));
+            _isCausallyConsistent = isCausallyConsistent;
+            _isImplicit = isImplicit;
+        }
+
+        // public properties
+        /// <inheritdoc />
+        public BsonDocument ClusterTime => _clusterClock.ClusterTime;
+
+        /// <inheritdoc />
+        public BsonDocument Id => _serverSession.Id;
+
+        /// <inheritdoc />
+        public bool IsCausallyConsistent => _isCausallyConsistent;
+
+        /// <inheritdoc />
+        public bool IsImplicit => _isImplicit;
+
+        /// <inheritdoc />
+        public BsonTimestamp OperationTime => _operationClock.OperationTime;
+
+        // public methods
+        /// <inheritdoc />
+        public void AdvanceClusterTime(BsonDocument newClusterTime)
+        {
+            _clusterClock.AdvanceClusterTime(newClusterTime);
+        }
+
+        /// <inheritdoc />
+        public void AdvanceOperationTime(BsonTimestamp newOperationTime)
+        {
+            _operationClock.AdvanceOperationTime(newOperationTime);
+        }
+
+        /// <inheritdoc />
+        public long AdvanceTransactionNumber()
+        {
+            return _serverSession.AdvanceTransactionNumber();
+        }
+
+        /// <inheritdoc />
+        public void Dispose()
+        {
+            if (!_disposed)
+            {
+                _serverSession.Dispose();
+                _disposed = true;
+            }
+        }
+
+        /// <inheritdoc />
+        public void WasUsed()
+        {
+            _serverSession.WasUsed();
+        }
+    }
+}

src/MongoDB.Driver.Core/Core/Operations/AsyncCursor.cs

@@ -47,7 +47,7 @@ public class AsyncCursor<TDocument> : IAsyncCursor<TDocument>
         // fields
         private readonly int? _batchSize;
         private readonly CollectionNamespace _collectionNamespace;
-        private readonly IChannelSource _channelSource;
+        private IChannelSource _channelSource;
         private int _count;
         private IReadOnlyList<TDocument> _currentBatch;
         private long _cursorId;
@@ -104,12 +104,7 @@ public AsyncCursor(
             }
             _count = _firstBatch.Count;
 
-            // if we aren't going to need the channel source we can go ahead and Dispose it now
-            if (_cursorId == 0 && _channelSource != null)
-            {
-                _channelSource.Dispose();
-                _channelSource = null;
-            }
+            DisposeChannelSourceIfNoLongerNeeded();
         }
 
         // properties
@@ -276,6 +271,15 @@ protected virtual void Dispose(bool disposing)
             _disposed = true;
         }
 
+        private void DisposeChannelSourceIfNoLongerNeeded()
+        {
+            if (_channelSource != null && _cursorId == 0)
+            {
+                _channelSource.Dispose();
+                _channelSource = null;
+            }
+        }
+
         private CursorBatch<TDocument> GetNextBatch(CancellationToken cancellationToken)
         {
             using (EventContext.BeginOperation(_operationId))
@@ -373,6 +377,8 @@ private void SaveBatch(CursorBatch<TDocument> batch)
 
             _currentBatch = documents;
             _cursorId = batch.CursorId;
+
+            DisposeChannelSourceIfNoLongerNeeded();
         }
 
         private void ThrowIfDisposed()

src/MongoDB.Driver.Core/MongoDB.Driver.Core.csproj

@@ -96,6 +96,7 @@
     <Compile Include="Core\Bindings\ChannelSourceReadWriteBinding.cs" />
     <Compile Include="Core\Bindings\CoreServerSession.cs" />
     <Compile Include="Core\Bindings\CoreServerSessionPool.cs" />
+    <Compile Include="Core\Bindings\CoreSession.cs" />
     <Compile Include="Core\Bindings\CoreSessionHandle.cs" />
     <Compile Include="Core\Bindings\ICoreServerSessionPool.cs" />
     <Compile Include="Core\Bindings\ICoreServerSesssion.cs" />

tests/MongoDB.Driver.Core.TestHelpers/CoreTestConfiguration.cs

@@ -241,24 +241,9 @@ public static DatabaseNamespace GetDatabaseNamespaceForTestClass(Type testClassT
             return new DatabaseNamespace(databaseName);
         }
 
-        public static IReadBinding GetReadBinding(ICoreSessionHandle session)
-        {
-            return GetReadBinding(ReadPreference.Primary, session);
-        }
-
-        public static IReadBinding GetReadBinding(ReadPreference readPreference, ICoreSessionHandle session)
-        {
-            return new ReadPreferenceBinding(__cluster.Value, readPreference, session);
-        }
-
-        public static IReadWriteBinding GetReadWriteBinding(ICoreSessionHandle session)
-        {
-            return new WritableServerBinding(__cluster.Value, session);
-        }
-
         public static IEnumerable<string> GetModules()
         {
-            var session = CoreTestConfiguration.StartSession();
+            using (var session = StartSession())
             using (var binding = GetReadBinding(session))
             {
                 var command = new BsonDocument("buildinfo", 1);
@@ -278,7 +263,7 @@ public static IEnumerable<string> GetModules()
 
         public static string GetStorageEngine()
         {
-            var session = CoreTestConfiguration.StartSession();
+            using (var session = StartSession())
             using (var binding = GetReadWriteBinding(session))
             {
                 var command = new BsonDocument("serverStatus", 1);
@@ -305,7 +290,9 @@ public static ICoreSessionHandle StartSession(ICluster cluster)
         {
             if (AreSessionsSupported(cluster))
             {
-                return TestCoreSession.NewHandle();
+                var serverSession = cluster.AcquireServerSession();
+                var session = new CoreSession(serverSession);
+                return new CoreSessionHandle(session);
             }
             else
             {
@@ -371,7 +358,7 @@ private static string TruncateDatabaseNameIfTooLong(string databaseName)
         }
         #endregion
 
-        // methods
+        // private methods
         private static bool AreSessionsSupported(ICluster cluster)
         {
             SpinWait.SpinUntil(() => cluster.Description.Servers.Any(s => s.State == ServerState.Connected), TimeSpan.FromSeconds(30));
@@ -389,13 +376,28 @@ private static void DropDatabase()
         {
             var operation = new DropDatabaseOperation(__databaseNamespace, __messageEncoderSettings);
 
-            var session = CoreTestConfiguration.StartSession();
+            using (var session = StartSession())
             using (var binding = GetReadWriteBinding(session))
             {
                 operation.Execute(binding, CancellationToken.None);
             }
         }
 
+        private static IReadBinding GetReadBinding(ICoreSessionHandle session)
+        {
+            return GetReadBinding(ReadPreference.Primary, session);
+        }
+
+        private static IReadBinding GetReadBinding(ReadPreference readPreference, ICoreSessionHandle session)
+        {
+            return new ReadPreferenceBinding(__cluster.Value, readPreference, session.Fork());
+        }
+
+        private static IReadWriteBinding GetReadWriteBinding(ICoreSessionHandle session)
+        {
+            return new WritableServerBinding(__cluster.Value, session.Fork());
+        }
+
         public static void TearDown()
         {
             if (__cluster.IsValueCreated)

tests/MongoDB.Driver.Core.TestHelpers/MongoDB.Driver.Core.TestHelpers.csproj

@@ -63,7 +63,6 @@
     <Compile Include="FailPoint.cs" />
     <Compile Include="ICoreSessionHandleExtensions.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
-    <Compile Include="TestCoreSession.cs" />
     <Compile Include="XunitExtensions\RequireServer.cs" />
   </ItemGroup>
   <ItemGroup>

tests/MongoDB.Driver.Core.TestHelpers/TestCoreSession.cs

@@ -371,7 +371,7 @@ public void Execute_should_throw_when_maxTime_is_exceeded(
 
             var subject = new AggregateExplainOperation(_collectionNamespace, __pipeline, _messageEncoderSettings) { MaxTime = TimeSpan.FromSeconds(9001) };
 
-            using (var failPoint = FailPoint.ConfigureAlwaysOn(CoreTestConfiguration.Cluster, _session, FailPointName.MaxTimeAlwaysTimeout))
+            using (var failPoint = FailPoint.ConfigureAlwaysOn(_cluster, _session, FailPointName.MaxTimeAlwaysTimeout))
             {
                 var exception = Record.Exception(() => ExecuteOperation(subject, failPoint.Binding, async));
 

tests/MongoDB.Driver.Core.Tests/Core/Operations/AggregateExplainOperationTests.cs

@@ -569,7 +569,7 @@ public void Execute_should_throw_when_maxTime_is_exceeded(
 
             var subject = new AggregateOperation<BsonDocument>(_collectionNamespace, __pipeline, __resultSerializer, _messageEncoderSettings) { MaxTime = TimeSpan.FromSeconds(9001) };
 
-            using (var failPoint = FailPoint.ConfigureAlwaysOn(CoreTestConfiguration.Cluster, _session, FailPointName.MaxTimeAlwaysTimeout))
+            using (var failPoint = FailPoint.ConfigureAlwaysOn(_cluster, _session, FailPointName.MaxTimeAlwaysTimeout))
             {
                 var exception = Record.Exception(() => ExecuteOperation(subject, failPoint.Binding, async));
 

tests/MongoDB.Driver.Core.Tests/Core/Operations/AggregateOperationTests.cs

@@ -513,7 +513,7 @@ public void Execute_should_throw_when_maxTime_is_exceeded(
                 MaxTime = TimeSpan.FromSeconds(9001)
             };
 
-            using (var failPoint = FailPoint.ConfigureAlwaysOn(CoreTestConfiguration.Cluster, _session, FailPointName.MaxTimeAlwaysTimeout))
+            using (var failPoint = FailPoint.ConfigureAlwaysOn(_cluster, _session, FailPointName.MaxTimeAlwaysTimeout))
             {
                 var exception = Record.Exception(() => ExecuteOperation(subject, failPoint.Binding, async));