RUBY-1547 Inline cluster rescans must be concurrent/done by monitor (#1142)

Author: p-mongo

lib/mongo/cluster.rb

@@ -426,17 +426,38 @@ def reconnect!
 
     # Force a scan of all known servers in the cluster.
     #
+    # If the sync parameter is true which is the default, the scan is
+    # performed synchronously in the thread which called this method.
+    # Each server in the cluster is checked sequentially. If there are
+    # many servers in the cluster or they are slow to respond, this
+    # can be a long running operation.
+    #
+    # If the sync parameter is false, this method instructs all server
+    # monitor threads to perform an immediate scan and returns without
+    # waiting for scan results.
+    #
+    # @note In both synchronous and asynchronous scans, each monitor
+    #   thread maintains a minimum interval between scans, meaning
+    #   calling this method may not initiate a scan on a particular server
+    #   the very next instant.
+    #
     # @example Force a full cluster scan.
     #   cluster.scan!
     #
-    # @note This operation is done synchronously. If servers in the cluster are
-    #   down or slow to respond this can potentially be a slow operation.
-    #
     # @return [ true ] Always true.
     #
     # @since 2.0.0
-    def scan!
-      servers_list.each{ |server| server.scan! } and true
+    def scan!(sync=true)
+      if sync
+        servers_list.each do |server|
+          server.scan!
+        end
+      else
+        servers_list.each do |server|
+          server.monitor.scan_semaphore.signal
+        end
+      end
+      true
     end
 
     # Determine if this cluster of servers is equal to another object. Checks the

lib/mongo/retryable.rb

@@ -44,7 +44,7 @@ def read_with_retry(session = nil)
       rescue Error::SocketError, Error::SocketTimeoutError => e
         raise(e) if attempt > cluster.max_read_retries || (session && session.in_transaction?)
         log_retry(e)
-        cluster.scan!
+        cluster.scan!(false)
         retry
       rescue Error::OperationFailure => e
         if cluster.sharded? && e.retryable? && !(session && session.in_transaction?)
@@ -133,17 +133,17 @@ def retry_write_allowed?(session, write_concern)
     end
 
     def retry_write(original_error, txn_num, &block)
-      cluster.scan!
+      cluster.scan!(false)
       server = cluster.next_primary
       raise original_error unless (server.retry_writes? && txn_num)
       log_retry(original_error)
       yield(server, txn_num)
     rescue Error::SocketError, Error::SocketTimeoutError => e
-      cluster.scan!
+      cluster.scan!(false)
       raise e
     rescue Error::OperationFailure => e
       raise original_error unless e.write_retryable?
-      cluster.scan!
+      cluster.scan!(false)
       raise e
     rescue
       raise original_error
@@ -162,7 +162,7 @@ def legacy_write_with_retry(server = nil, session = nil)
         raise(e) if attempt > Cluster::MAX_WRITE_RETRIES
         if e.write_retryable? && !(session && session.in_transaction?)
           log_retry(e)
-          cluster.scan!
+          cluster.scan!(false)
           retry
         else
           raise(e)

lib/mongo/semaphore.rb

@@ -36,5 +36,11 @@ def broadcast
         @cv.broadcast
       end
     end
+
+    def signal
+      @lock.synchronize do
+        @cv.signal
+      end
+    end
   end
 end

lib/mongo/server/monitor.rb

@@ -66,6 +66,7 @@ def initialize(address, event_listeners, monitoring, options = {})
         @monitoring = monitoring
         @options = options.freeze
         @round_trip_time_averager = RoundTripTimeAverager.new
+        @scan_semaphore = Semaphore.new
         # This is a Mongo::Server::Monitor::Connection
         @connection = Connection.new(address, options)
         @last_scan = nil
@@ -107,6 +108,9 @@ def heartbeat_frequency
         @heartbeat_frequency ||= options[:heartbeat_frequency] || HEARTBEAT_FREQUENCY
       end
 
+      # @api private
+      attr_reader :scan_semaphore
+
       # Runs the server monitor. Refreshing happens on a separate thread per
       # server.
       #
@@ -120,7 +124,7 @@ def run!
         @thread = Thread.new(heartbeat_frequency) do |i|
           loop do
             scan!
-            sleep(i)
+            @scan_semaphore.wait(i)
           end
         end
       end

lib/mongo/server_selector/selectable.rb

@@ -132,7 +132,7 @@ def select_server(cluster, ping = nil)
 
             return server
           end
-          cluster.scan!
+          cluster.scan!(false)
         end
         raise Error::NoServerAvailable.new(self, cluster)
       end

spec/integration/retryable_writes_spec.rb

@@ -164,7 +164,9 @@
             end
 
             before do
-              expect(client.cluster).to receive(:scan!).twice.and_call_original
+              # server selector can call scan! until it finds a server,
+              # hence more than two scan! calls may be issued
+              expect(client.cluster).to receive(:scan!).at_least(:twice).and_call_original
             end
 
             it 'does not retry writes and raises the second error' do
@@ -178,7 +180,9 @@
           context 'when the second error is a SocketTimeoutError' do
 
             before do
-              expect(client.cluster).to receive(:scan!).twice.and_call_original
+              # server selector can call scan! until it finds a server,
+              # hence more than two scan! calls may be issued
+              expect(client.cluster).to receive(:scan!).at_least(:twice).and_call_original
             end
 
             let(:second_error) do
@@ -196,7 +200,9 @@
           context 'when the second error is a retryable OperationFailure' do
 
             before do
-              expect(client.cluster).to receive(:scan!).twice.and_call_original
+              # server selector can call scan! until it finds a server,
+              # hence more than two scan! calls may be issued
+              expect(client.cluster).to receive(:scan!).at_least(:twice).and_call_original
             end
 
             let(:second_error) do

spec/mongo/operation/delete/op_msg_spec.rb

@@ -143,17 +143,21 @@
         context 'when an implicit session is created and the topology is then updated and the server does not support sessions' do
 
           let(:expected_global_args) do
-            global_args.delete(:lsid)
-            global_args
+            global_args.dup.tap do |args|
+              args.delete(:lsid)
+            end
           end
 
           before do
             session.instance_variable_set(:@options, { implicit: true })
+            # Topology is standalone, hence there is exactly one server
+            authorized_client.cluster.servers.first.monitor.stop!(true)
             allow(authorized_primary.features).to receive(:sessions_enabled?).and_return(false)
           end
 
           it 'creates the correct OP_MSG message' do
             authorized_client.command(ping:1)
+            expect(expected_global_args[:session]).to be nil
             expect(Mongo::Protocol::Msg).to receive(:new).with([:none], {}, expected_global_args, expected_payload_1)
             op.send(:message, authorized_primary)
           end
@@ -176,8 +180,10 @@
           context 'when the topology is replica set or sharded', if: test_sessions? do
 
             let(:expected_global_args) do
-              global_args.delete(:lsid)
-              global_args.merge!(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+              global_args.dup.tap do |args|
+                args.delete(:lsid)
+                args.merge!(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+              end
             end
 
             it 'does not send a session id in the command' do
@@ -190,8 +196,9 @@
           context 'when the topology is standalone', if: standalone? && sessions_enabled? do
 
             let(:expected_global_args) do
-              global_args.delete(:lsid)
-              global_args
+              global_args.dup.tap do |args|
+                args.delete(:lsid)
+              end
             end
 
             it 'creates the correct OP_MSG message' do
@@ -209,8 +216,10 @@
           end
 
           let(:expected_global_args) do
-            global_args.delete(:lsid)
-            global_args.merge!(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+            global_args.dup.tap do |args|
+              args.delete(:lsid)
+              args.merge!(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+            end
           end
 
           it 'does not send a session id in the command' do

spec/mongo/operation/insert/op_msg_spec.rb

@@ -152,17 +152,21 @@
         context 'when an implicit session is created and the topology is then updated and the server does not support sessions' do
 
           let(:expected_global_args) do
-            global_args.delete(:lsid)
-            global_args
+            global_args.dup.tap do |args|
+              args.delete(:lsid)
+            end
           end
 
           before do
             session.instance_variable_set(:@options, { implicit: true })
-            allow(authorized_primary.features).to receive(:sessions_enabled?).and_return(false)
+            # Topology is standalone, hence there is exactly one server
+            authorized_client.cluster.servers.first.monitor.stop!(true)
+            expect(authorized_primary.features).to receive(:sessions_enabled?).at_least(:once).and_return(false)
           end
 
           it 'creates the correct OP_MSG message' do
             authorized_client.command(ping:1)
+            expect(expected_global_args).not_to have_key(:lsid)
             expect(Mongo::Protocol::Msg).to receive(:new).with([:none],
                                                                { validating_keys: true },
                                                                expected_global_args,
@@ -188,8 +192,10 @@
           context 'when the topology is replica set or sharded', if: test_sessions? do
 
             let(:expected_global_args) do
-              global_args.delete(:lsid)
-              global_args.merge(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+              global_args.dup.tap do |args|
+                args.delete(:lsid)
+                args.merge!(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+              end
             end
 
             it 'does not send a session id in the command' do
@@ -205,8 +211,9 @@
           context 'when the topology is standalone', if: standalone? && sessions_enabled? do
 
             let(:expected_global_args) do
-              global_args.delete(:lsid)
-              global_args
+              global_args.dup.tap do |args|
+                args.delete(:lsid)
+              end
             end
 
             it 'creates the correct OP_MSG message' do
@@ -227,8 +234,10 @@
           end
 
           let(:expected_global_args) do
-            global_args.delete(:lsid)
-            global_args.merge(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+            global_args.dup.tap do |args|
+              args.delete(:lsid)
+              args.merge!(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+            end
           end
 
           it 'does not send a session id in the command' do

spec/mongo/operation/update/op_msg_spec.rb

@@ -148,12 +148,15 @@
         context 'when an implicit session is created and the topology is then updated and the server does not support sessions' do
 
           let(:expected_global_args) do
-            global_args.delete(:lsid)
-            global_args
+            global_args.dup.tap do |args|
+              args.delete(:lsid)
+            end
           end
 
           before do
             session.instance_variable_set(:@options, { implicit: true })
+            # Topology is standalone, hence there is exactly one server
+            authorized_client.cluster.servers.first.monitor.stop!(true)
             allow(authorized_primary.features).to receive(:sessions_enabled?).and_return(false)
           end
 
@@ -181,8 +184,10 @@
           context 'when the topology is replica set or sharded', if: test_sessions? do
 
             let(:expected_global_args) do
-              global_args.delete(:lsid)
-              global_args.merge(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+              global_args.dup.tap do |args|
+                args.delete(:lsid)
+                args.merge!(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+              end
             end
 
             it 'does not send a session id in the command' do
@@ -195,8 +200,9 @@
           context 'when the topology is standalone', if: standalone? && sessions_enabled? do
 
             let(:expected_global_args) do
-              global_args.delete(:lsid)
-              global_args
+              global_args.dup.tap do |args|
+                args.delete(:lsid)
+              end
             end
 
             it 'creates the correct OP_MSG message' do
@@ -214,8 +220,10 @@
           end
 
           let(:expected_global_args) do
-            global_args.delete(:lsid)
-            global_args.merge(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+            global_args.dup.tap do |args|
+              args.delete(:lsid)
+              args.merge!(Mongo::Operation::CLUSTER_TIME => authorized_client.cluster.cluster_time)
+            end
           end
 
           it 'does not send a session id in the command' do