CSHARP-3556: Reconsider server pinning in GridFS

Author: sanych-sun

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

@@ -1,4 +1,4 @@
-/* Copyright 2013-present MongoDB Inc.
+/* 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.
@@ -15,27 +15,31 @@
 
 using System;
 using System.Collections.Generic;
-using System.Threading;
 using System.Threading.Tasks;
+using MongoDB.Driver.Core.Clusters;
+using MongoDB.Driver.Core.Clusters.ServerSelectors;
 using MongoDB.Driver.Core.Misc;
 using MongoDB.Driver.Core.Servers;
 
 namespace MongoDB.Driver.Core.Bindings
 {
     internal sealed class SingleServerReadBinding : IReadBinding
     {
+#pragma warning disable CA2213 // Disposable fields should be disposed
+        private readonly IClusterInternal _cluster;
+#pragma warning restore CA2213 // Disposable fields should be disposed
         private bool _disposed;
         private readonly ReadPreference _readPreference;
-        private readonly IServer _server;
-        private readonly TimeSpan _serverSelectionTimeout;
+        private readonly IServerSelector _serverSelector;
         private readonly ICoreSessionHandle _session;
 
-        public SingleServerReadBinding(IServer server, ReadPreference readPreference, ICoreSessionHandle session, TimeSpan serverSelectionTimeout)
+        public SingleServerReadBinding(IClusterInternal cluster, IServer server, ReadPreference readPreference, ICoreSessionHandle session)
         {
-            _server = Ensure.IsNotNull(server, nameof(server));
+            _cluster = Ensure.IsNotNull(cluster, nameof(cluster));
+            Ensure.IsNotNull(server, nameof(server));
+            _serverSelector = new EndPointServerSelector(server.EndPoint);
             _readPreference = Ensure.IsNotNull(readPreference, nameof(readPreference));
             _session = Ensure.IsNotNull(session, nameof(session));
-            _serverSelectionTimeout = Ensure.IsGreaterThanOrEqualToZero(serverSelectionTimeout, nameof(serverSelectionTimeout));
         }
 
         public ReadPreference ReadPreference
@@ -51,24 +55,22 @@ public ICoreSessionHandle Session
         public IChannelSourceHandle GetReadChannelSource(OperationContext operationContext)
         {
             ThrowIfDisposed();
-            return GetChannelSourceHelper(operationContext);
+            var server = _cluster.SelectServer(operationContext, _serverSelector);
+            return new ChannelSourceHandle(new ServerChannelSource(server, _session.Fork()));
         }
 
-        public Task<IChannelSourceHandle> GetReadChannelSourceAsync(OperationContext operationContext)
+        public async Task<IChannelSourceHandle> GetReadChannelSourceAsync(OperationContext operationContext)
         {
             ThrowIfDisposed();
-            return Task.FromResult(GetChannelSourceHelper(operationContext));
+            var server = await _cluster.SelectServerAsync(operationContext, _serverSelector).ConfigureAwait(false);
+            return new ChannelSourceHandle(new ServerChannelSource(server, _session.Fork()));
         }
 
-        public IChannelSourceHandle GetReadChannelSource(OperationContext operationContext, IReadOnlyCollection<ServerDescription> deprioritizedServers)
-        {
-            return GetReadChannelSource(operationContext);
-        }
+        public IChannelSourceHandle GetReadChannelSource(OperationContext operationContext, IReadOnlyCollection<ServerDescription> deprioritizedServers) =>
+            GetReadChannelSource(operationContext);
 
-        public Task<IChannelSourceHandle> GetReadChannelSourceAsync(OperationContext operationContext, IReadOnlyCollection<ServerDescription> deprioritizedServers)
-        {
-            return GetReadChannelSourceAsync(operationContext);
-        }
+        public Task<IChannelSourceHandle> GetReadChannelSourceAsync(OperationContext operationContext, IReadOnlyCollection<ServerDescription> deprioritizedServers) =>
+            GetReadChannelSourceAsync(operationContext);
 
         public void Dispose()
         {
@@ -79,31 +81,6 @@ public void Dispose()
             }
         }
 
-        private IChannelSourceHandle GetChannelSourceHelper(OperationContext operationContext)
-        {
-            // server might be in unknown state due to previous failed operation, allow description to be updated
-            // this is done instead of server selection
-            // TODO parameterize timeout and avoid busy wait, or offload waiting to server level
-            // Should be addressed by CSHARP-3556
-            if (_server.Description.State == ServerState.Disconnected)
-            {
-                using var serverSelectionContext = operationContext.WithTimeout(_serverSelectionTimeout);
-
-                if (SpinWait.SpinUntil(() =>
-                            _server.Description.State == ServerState.Connected || serverSelectionContext.CancellationToken.IsCancellationRequested,
-                        serverSelectionContext.RemainingTimeout))
-                {
-                    serverSelectionContext.CancellationToken.ThrowIfCancellationRequested();
-                }
-                else
-                {
-                    throw new TimeoutException($"A timeout occurred after {serverSelectionContext.Elapsed.TotalMilliseconds}ms waiting for the server to become available.");
-                }
-            }
-
-            return new ChannelSourceHandle(new ServerChannelSource(_server, _session.Fork()));
-        }
-
         private void ThrowIfDisposed()
         {
             if (_disposed)

src/MongoDB.Driver/Core/Clusters/ServerSelectors/EndPointServerSelector.cs

@@ -1,4 +1,4 @@
-/* Copyright 2013-present MongoDB Inc.
+/* 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.
@@ -39,6 +39,11 @@ public EndPointServerSelector(EndPoint endPoint)
             _endPoint = Ensure.IsNotNull(endPoint, nameof(endPoint));
         }
 
+        /// <summary>
+        /// Gets the endpoint associated with the server selector.
+        /// </summary>
+        public EndPoint EndPoint => _endPoint;
+
         // methods
         /// <inheritdoc/>
         public IEnumerable<ServerDescription> SelectServers(ClusterDescription cluster, IEnumerable<ServerDescription> servers)

src/MongoDB.Driver/GridFS/GridFSBucket.cs

@@ -1,4 +1,4 @@
-/* Copyright 2016-present MongoDB Inc.
+/* 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.
@@ -1041,7 +1041,7 @@ private IReadBindingHandle GetSingleServerReadBinding(OperationContext operation
             var readPreference = _options.ReadPreference ?? _database.Settings.ReadPreference;
             var selector = new ReadPreferenceServerSelector(readPreference);
             var server = _cluster.SelectServer(operationContext, selector);
-            var binding = new SingleServerReadBinding(server, readPreference, NoCoreSession.NewHandle(), _cluster.Settings.ServerSelectionTimeout);
+            var binding = new SingleServerReadBinding(_cluster, server, readPreference, NoCoreSession.NewHandle());
             return new ReadBindingHandle(binding);
         }
 
@@ -1050,7 +1050,7 @@ private async Task<IReadBindingHandle> GetSingleServerReadBindingAsync(Operation
             var readPreference = _options.ReadPreference ?? _database.Settings.ReadPreference;
             var selector = new ReadPreferenceServerSelector(readPreference);
             var server = await _cluster.SelectServerAsync(operationContext, selector).ConfigureAwait(false);
-            var binding = new SingleServerReadBinding(server, readPreference, NoCoreSession.NewHandle(), _cluster.Settings.ServerSelectionTimeout);
+            var binding = new SingleServerReadBinding(_cluster, server, readPreference, NoCoreSession.NewHandle());
             return new ReadBindingHandle(binding);
         }