RUBY-3368 Measure minimum RTT (#2837)

Author: comandeo-mongo

lib/mongo/server.rb

@@ -80,7 +80,7 @@ def initialize(address, cluster, monitoring, event_listeners, options = {})
         include Id
       end
       @scan_semaphore = DistinguishingSemaphore.new
-      @round_trip_time_averager = RoundTripTimeAverager.new
+      @round_trip_time_calculator = RoundTripTimeCalculator.new
       @description = Description.new(address, {},
         load_balancer: !!@options[:load_balancer],
         force_load_balancer: force_load_balancer?,
@@ -228,9 +228,9 @@ def compressor
     # @api private
     attr_reader :scan_semaphore
 
-    # @return [ RoundTripTimeAverager ] Round trip time averager object.
+    # @return [ RoundTripTimeCalculator ] Round trip time calculator object.
     # @api private
-    attr_reader :round_trip_time_averager
+    attr_reader :round_trip_time_calculator
 
     # Is this server equal to another?
     #
@@ -701,5 +701,5 @@ def update_last_scan
 require 'mongo/server/connection_pool'
 require 'mongo/server/description'
 require 'mongo/server/monitor'
-require 'mongo/server/round_trip_time_averager'
+require 'mongo/server/round_trip_time_calculator'
 require 'mongo/server/push_monitor'

lib/mongo/server/monitor.rb

@@ -238,7 +238,7 @@ def run_sdam_flow(result, awaited: false, scan_error: nil)
           old_description = server.description
 
           new_description = Description.new(server.address, result,
-            average_round_trip_time: server.round_trip_time_averager.average_round_trip_time
+            average_round_trip_time: server.round_trip_time_calculator.average_round_trip_time
           )
 
           server.cluster.run_sdam_flow(server.description, new_description, awaited: awaited, scan_error: scan_error)
@@ -306,7 +306,7 @@ def check
         end
 
         if @connection
-          result = server.round_trip_time_averager.measure do
+          result = server.round_trip_time_calculator.measure do
             begin
               doc = @connection.check_document
               cmd = Protocol::Query.new(
@@ -323,7 +323,7 @@ def check
         else
           connection = Connection.new(server.address, options)
           connection.connect!
-          result = server.round_trip_time_averager.measure do
+          result = server.round_trip_time_calculator.measure do
             connection.handshake!
           end
           @connection = connection

lib/mongo/server/pending_connection.rb

@@ -131,7 +131,7 @@ def handshake!(speculative_auth_doc: nil)
         doc = nil
         @server.handle_handshake_failure! do
           begin
-            response = @server.round_trip_time_averager.measure do
+            response = @server.round_trip_time_calculator.measure do
               add_server_diagnostics do
                 socket.write(hello_command.serialize.to_s)
                 Protocol::Message.deserialize(socket, Protocol::Message::MAX_MESSAGE_SIZE)
@@ -155,7 +155,7 @@ def handshake!(speculative_auth_doc: nil)
           doc['serviceId'] ||= "fake:#{rand(2**32-1)+1}"
         end
 
-        post_handshake(doc, @server.round_trip_time_averager.average_round_trip_time)
+        post_handshake(doc, @server.round_trip_time_calculator.average_round_trip_time)
 
         doc
       end

lib/mongo/server/round_trip_time_averager.rb renamed to lib/mongo/server/round_trip_time_calculator.rb

@@ -18,20 +18,29 @@
 module Mongo
   class Server
     # @api private
-    class RoundTripTimeAverager
+    class RoundTripTimeCalculator
 
       # The weighting factor (alpha) for calculating the average moving
       # round trip time.
       RTT_WEIGHT_FACTOR = 0.2.freeze
       private_constant :RTT_WEIGHT_FACTOR
 
+      RTT_SAMPLES_FOR_MINIMUM = 10
+      private_constant :RTT_SAMPLES_FOR_MINIMUM
+
+      MIN_SAMPLES = 3
+      private_constant :MIN_SAMPLES
+
       def initialize
         @last_round_trip_time = nil
         @average_round_trip_time = nil
+        @minimum_round_trip_time = 0
+        @rtts = []
       end
 
       attr_reader :last_round_trip_time
       attr_reader :average_round_trip_time
+      attr_reader :minimum_round_trip_time
 
       def measure
         start = Utils.monotonic_time
@@ -44,14 +53,15 @@ def measure
         rescue Error, Error::AuthError => exc
           # For other errors, RTT is valid.
         end
-        last_round_trip_time = Utils.monotonic_time - start
+        last_rtt = Utils.monotonic_time - start
 
         # If hello fails, we need to return the last round trip time
         # because it is used in the heartbeat failed SDAM event,
         # but we must not update the round trip time recorded in the server.
         unless exc
-          @last_round_trip_time = last_round_trip_time
+          @last_round_trip_time = last_rtt
           update_average_round_trip_time
+          update_minimum_round_trip_time
         end
 
         if exc
@@ -61,16 +71,21 @@ def measure
         end
       end
 
-      private
-
-      # This method is separate for testing purposes.
       def update_average_round_trip_time
         @average_round_trip_time = if average_round_trip_time
           RTT_WEIGHT_FACTOR * last_round_trip_time + (1 - RTT_WEIGHT_FACTOR) * average_round_trip_time
         else
           last_round_trip_time
         end
       end
+
+      def update_minimum_round_trip_time
+        @rtts.push(@last_round_trip_time) unless @last_round_trip_time.nil?
+        @minimum_round_trip_time = 0 and return if @rtts.size < MIN_SAMPLES
+
+        @rtts.shift if @rtts.size > RTT_SAMPLES_FOR_MINIMUM
+        @minimum_round_trip_time = @rtts.compact.min
+      end
     end
   end
 end

spec/mongo/server/round_trip_time_averager_spec.rb

@@ -0,0 +1,120 @@
+# frozen_string_literal: true
+# rubocop:todo all
+
+require 'spec_helper'
+
+describe Mongo::Server::RoundTripTimeCalculator do
+  let(:calculator) { Mongo::Server::RoundTripTimeCalculator.new }
+
+  describe '#update_average_round_trip_time' do
+    context 'no existing average rtt' do
+      it 'updates average rtt' do
+        calculator.instance_variable_set('@last_round_trip_time', 5)
+        calculator.update_average_round_trip_time
+        expect(calculator.average_round_trip_time).to eq(5)
+      end
+    end
+
+    context 'with existing average rtt' do
+      it 'averages with existing average rtt' do
+        calculator.instance_variable_set('@last_round_trip_time', 5)
+        calculator.instance_variable_set('@average_round_trip_time', 10)
+        calculator.update_average_round_trip_time
+        expect(calculator.average_round_trip_time).to eq(9)
+      end
+    end
+  end
+
+  describe '#update_minimum_round_trip_time' do
+    context 'with no samples' do
+      it 'sets minimum_round_trip_time to zero' do
+        calculator.update_minimum_round_trip_time
+        expect(calculator.minimum_round_trip_time).to eq(0)
+      end
+    end
+
+    context 'with one sample' do
+      before do
+        calculator.instance_variable_set('@last_round_trip_time', 5)
+      end
+
+      it 'sets minimum_round_trip_time to zero' do
+        calculator.update_minimum_round_trip_time
+        expect(calculator.minimum_round_trip_time).to eq(0)
+      end
+    end
+
+    context 'with two samples' do
+      before do
+        calculator.instance_variable_set('@last_round_trip_time', 10)
+        calculator.instance_variable_set('@rtts', [5])
+      end
+
+      it 'sets minimum_round_trip_time to zero' do
+        calculator.update_minimum_round_trip_time
+        expect(calculator.minimum_round_trip_time).to eq(0)
+      end
+    end
+
+    context 'with samples less than maximum' do
+      before do
+        calculator.instance_variable_set('@last_round_trip_time', 10)
+        calculator.instance_variable_set('@rtts', [5, 4, 120])
+      end
+
+      it 'properly sets minimum_round_trip_time' do
+        calculator.update_minimum_round_trip_time
+        expect(calculator.minimum_round_trip_time).to eq(4)
+      end
+    end
+
+    context 'with more than maximum samples' do
+      before do
+        calculator.instance_variable_set('@last_round_trip_time', 2)
+        calculator.instance_variable_set('@rtts', [1, 20, 15, 4, 5, 6, 7, 39, 8, 4])
+      end
+
+      it 'properly sets minimum_round_trip_time' do
+        calculator.update_minimum_round_trip_time
+        expect(calculator.minimum_round_trip_time).to eq(2)
+      end
+    end
+
+  end
+
+  describe '#measure' do
+    context 'block does not raise' do
+      it 'updates average rtt' do
+        expect(calculator).to receive(:update_average_round_trip_time)
+        calculator.measure do
+        end
+      end
+
+      it 'updates minimum rtt' do
+        expect(calculator).to receive(:update_minimum_round_trip_time)
+        calculator.measure do
+        end
+      end
+    end
+
+    context 'block raises' do
+      it 'does not update average rtt' do
+        expect(calculator).not_to receive(:update_average_round_trip_time)
+        expect do
+          calculator.measure do
+            raise "Problem"
+          end
+        end.to raise_error(/Problem/)
+      end
+
+      it 'does not update minimum rtt' do
+        expect(calculator).not_to receive(:update_minimum_round_trip_time)
+        expect do
+          calculator.measure do
+            raise "Problem"
+          end
+        end.to raise_error(/Problem/)
+      end
+    end
+  end
+end

spec/mongo/server/round_trip_time_calculator_spec.rb

@@ -15,18 +15,18 @@
 
     context(spec.description) do
 
-      let(:averager) do
-        Mongo::Server::RoundTripTimeAverager.new
+      let(:calculator) do
+        Mongo::Server::RoundTripTimeCalculator.new
       end
 
       before do
-        averager.instance_variable_set(:@average_round_trip_time, spec.average_rtt)
-        averager.instance_variable_set(:@last_round_trip_time, spec.new_rtt)
-        averager.send(:update_average_round_trip_time)
+        calculator.instance_variable_set(:@average_round_trip_time, spec.average_rtt)
+        calculator.instance_variable_set(:@last_round_trip_time, spec.new_rtt)
+        calculator.update_average_round_trip_time
       end
 
       it 'correctly calculates the moving average round trip time' do
-        expect(averager.average_round_trip_time).to eq(spec.new_average_rtt)
+        expect(calculator.average_round_trip_time).to eq(spec.new_average_rtt)
       end
     end
   end