RUBY-456 Read Semantics update for MongoDB 2.2

Updates read semantics for the Ruby driver in order to comply with the specification outlined here:

http://docs.mongodb.org/manual/applications/replication/#replica-set-read-preference

- Modified ReplSetConnection :read parameter to accept and respect the following read preference modes (Note :secondary behaves differently than before)

  :primary
  :primary_preferred
  :secondary
  :secondary_preferred
  :nearest

- Added :tag_sets parameter to ReplSetConnection in order for tag sets to be specified as part of read preference.

- Added :secondary_acceptable_latency_ms parameter to ReplSetConnection for setting maximum latency specification as part of read preference.

Author: TylerBrock

lib/mongo/collection.rb

@@ -23,6 +23,9 @@ class Collection
 
     attr_reader :db, :name, :pk_factory, :hint, :safe
 
+    # Read Preference
+    attr_accessor :read_preference, :tag_sets, :acceptable_latency
+
     # Initialize a collection object.
     #
     # @param [String, Symbol] name the name of the collection.
@@ -97,6 +100,8 @@ def initialize(name, db, opts={})
           value = @db.read_preference
         end
         @read_preference = value.is_a?(Hash) ? value.dup : value
+        @tag_sets = opts.fetch(:tag_sets, @db.tag_sets)
+        @acceptable_latency = opts.fetch(:acceptable_latency, @db.acceptable_latency)
       end
       @pk_factory = pk_factory || opts[:pk] || BSON::ObjectId
       @hint = nil
@@ -220,6 +225,8 @@ def find(selector={}, opts={})
       transformer = opts.delete(:transformer)
       show_disk_loc = opts.delete(:show_disk_loc)
       read          = opts.delete(:read) || @read_preference
+      tag_sets      = opts.delete(:tag_sets) || @tag_sets
+      acceptable_latency = opts.delete(:acceptable_latency) || @acceptable_latency
 
       if timeout == false && !block_given?
         raise ArgumentError, "Collection#find must be invoked with a block when timeout is disabled."
@@ -247,7 +254,9 @@ def find(selector={}, opts={})
         :max_scan    => max_scan,
         :show_disk_loc => show_disk_loc,
         :return_key    => return_key,
-        :read          => read
+        :read          => read,
+        :tag_sets      => tag_sets,
+        :acceptable_latency => acceptable_latency
       })
 
       if block_given?
@@ -763,13 +772,6 @@ def group(opts, condition={}, initial={}, reduce=nil, finalize=nil)
       end
     end
 
-    # The value of the read preference. This will be
-    # either +:primary+, +:secondary+, or an object
-    # representing the tags to be read from.
-    def read_preference
-      @read_preference
-    end
-
     private
 
     def new_group(opts={})

lib/mongo/connection.rb

@@ -42,7 +42,7 @@ class Connection
 
     attr_reader :logger, :size, :auths, :primary, :safe, :host_to_try,
       :pool_size, :connect_timeout, :pool_timeout,
-      :primary_pool, :socket_class, :op_timeout
+      :primary_pool, :socket_class, :op_timeout, :tag_sets, :acceptable_latency
 
     # Create a connection to single MongoDB instance.
     #
@@ -127,6 +127,10 @@ def initialize(host=nil, port=nil, opts={})
       @primary      = nil
       @primary_pool = nil
 
+      # Not set for direct connection
+      @tag_sets = {}
+      @acceptable_latency = 15
+
       check_opts(opts)
       setup(opts)
     end
@@ -334,6 +338,15 @@ def [](db_name)
       DB.new(db_name, self)
     end
 
+    def refresh
+    end
+
+    def pin_pool(pool)
+    end
+
+    def unpin_pool(pool)
+    end
+
     # Drop a database.
     #
     # @param [String] name name of an existing database.
@@ -471,7 +484,7 @@ def read_pool
     # The value of the read preference.
     def read_preference
       if slave_ok?
-        :secondary
+        :secondary_preferred
       else
         :primary
       end
@@ -492,15 +505,9 @@ def max_bson_size
       @max_bson_size
     end
 
-    # Prefer primary pool but fall back to secondary
-    def checkout_best
-      connect unless connected?
-      @primary_pool.checkout
-    end
-
     # Checkout a socket for reading (i.e., a secondary node).
     # Note: this is overridden in ReplSetConnection.
-    def checkout_reader
+    def checkout_reader(mode=:primary, tag_sets={}, acceptable_latency=15)
       connect unless connected?
       @primary_pool.checkout
     end
@@ -515,24 +522,11 @@ def checkout_writer
     # Check a socket back into its pool.
     # Note: this is overridden in ReplSetConnection.
     def checkin(socket)
-      if @primary_pool && socket
+      if @primary_pool && socket && socket.pool
         socket.pool.checkin(socket)
       end
     end
 
-    # Excecutes block with the best available socket
-    def best_available_socket
-      socket = nil
-      begin
-        socket = checkout_best
-        yield socket
-      ensure
-        if socket
-          socket.pool.checkin(socket)
-        end
-      end
-    end
-
     protected
 
     def valid_opts

lib/mongo/cursor.rb

@@ -77,6 +77,9 @@ def initialize(collection, opts={})
         value = collection.read_preference
       end
       @read_preference = value.is_a?(Hash) ? value.dup : value
+      @tag_sets = opts.fetch(:tag_sets, @collection.tag_sets)
+      @acceptable_latency = opts.fetch(:acceptable_latency, @collection.acceptable_latency)
+
       batch_size(opts[:batch_size] || 0)
 
       @full_collection_name = "#{@collection.db.name}.#{@collection.name}"
@@ -335,7 +338,7 @@ def close
         message.put_int(1)
         message.put_long(@cursor_id)
         log(:debug, "Cursor#close #{@cursor_id}")
-        @connection.send_message(Mongo::Constants::OP_KILL_CURSORS, message)
+        @connection.send_message(Mongo::Constants::OP_KILL_CURSORS, message, :socket => @socket)
       end
       @cursor_id = 0
       @closed    = true
@@ -458,19 +461,37 @@ def refresh
       end
     end
 
+    # Sends initial query -- which is always a read unless it is a command
+    #
+    # Upon ConnectionFailure, tries query 3 times if socket was not provided
+    # and the query is either not a command or is a secondary_ok command.
+    # 
+    # Pins pools upon successful read and unpins pool upon ConnectionFailure
+    #
     def send_initial_query
-      message = construct_query_message
-      sock    = @socket || checkout_socket_from_connection
+      tries = 0
       instrument(:find, instrument_payload) do
         begin
-        results, @n_received, @cursor_id = @connection.receive_message(
-          Mongo::Constants::OP_QUERY, message, nil, sock, @command,
-          nil, @options & OP_QUERY_EXHAUST != 0)
-        rescue ConnectionFailure, OperationFailure, OperationTimeout => ex
-          checkin_socket(sock) unless @socket
+          tries += 1
+          message = construct_query_message
+          sock    = @socket || checkout_socket_from_connection
+          results, @n_received, @cursor_id = @connection.receive_message(
+            Mongo::Constants::OP_QUERY, message, nil, sock, @command,
+            nil, @options & OP_QUERY_EXHAUST != 0)
+        rescue ConnectionFailure => ex
+          if tries < 3 && !@socket && (!@command || Mongo::Support.secondary_ok?(@selector))
+            @connection.unpin_pool(sock.pool) if sock
+            @connection.refresh
+            retry
+          else
+            raise ex
+          end
+        rescue OperationFailure, OperationTimeout => ex
           raise ex
+        ensure
+          checkin_socket(sock) unless @socket
         end
-        checkin_socket(sock) unless @socket
+        @connection.pin_pool(sock.pool) if !@command && !@socket
         @returned += @n_received
         @cache += results
         @query_run = true
@@ -498,38 +519,32 @@ def send_get_more
       # Cursor id.
       message.put_long(@cursor_id)
       log(:debug, "cursor.refresh() for cursor #{@cursor_id}") if @logger
+
       sock = @socket || checkout_socket_from_connection
 
       begin
-      results, @n_received, @cursor_id = @connection.receive_message(
-        Mongo::Constants::OP_GET_MORE, message, nil, sock, @command, nil)
-      rescue ConnectionFailure, OperationFailure, OperationTimeout => ex
+        results, @n_received, @cursor_id = @connection.receive_message(
+          Mongo::Constants::OP_GET_MORE, message, nil, sock, @command, nil)
+      ensure
         checkin_socket(sock) unless @socket
-        raise ex
       end
-      checkin_socket(sock) unless @socket
+      
       @returned += @n_received
       @cache += results
       close_cursor_if_query_complete
     end
 
     def checkout_socket_from_connection
-      socket = nil
       begin
-        @checkin_connection = true
-        if @command || @read_preference == :primary
-          socket = @connection.checkout_writer
-        elsif @read_preference == :secondary_only
-          socket = @connection.checkout_secondary
+        if @command && !Mongo::Support::secondary_ok?(@selector)
+          @connection.checkout_reader(:primary)
         else
-          socket = @connection.checkout_reader
+          @connection.checkout_reader(@read_preference, @tag_sets, @acceptable_latency)
         end
       rescue SystemStackError, NoMemoryError, SystemCallError => ex
         @connection.close
         raise ex
       end
-
-      socket
     end
 
     def checkin_socket(sock)

lib/mongo/db.rb

@@ -53,6 +53,9 @@ def strict?; @strict; end
     # The length of time that Collection.ensure_index should cache index calls
     attr_accessor :cache_time
 
+    # Read Preference
+    attr_accessor :read_preference, :tag_sets, :acceptable_latency
+
     # Instances of DB are normally obtained by calling Mongo#db.
     #
     # @param [String] name the database name.
@@ -72,6 +75,7 @@ def strict?; @strict; end
     #   value is provided, the default value set on this instance's Connection object will be used. This
     #   default can be overridden upon instantiation of any collection by explicity setting a :safe value
     #   on initialization
+    #
     # @option opts [Integer] :cache_time (300) Set the time that all ensure_index calls should cache the command.
     #
     # @core databases constructor_details
@@ -87,6 +91,8 @@ def initialize(name, connection, opts={})
         value = @connection.read_preference
       end
       @read_preference = value.is_a?(Hash) ? value.dup : value
+      @tag_sets = opts.fetch(:tag_sets, @connection.tag_sets)
+      @acceptable_latency = opts.fetch(:acceptable_latency, @connection.acceptable_latency)
       @cache_time = opts[:cache_time] || 300 #5 minutes.
     end
 
@@ -112,8 +118,11 @@ def authenticate(username, password, save_auth=true)
         end
       end
 
-      @connection.best_available_socket do |socket|
+      begin
+        socket = @connection.checkout_reader(:primary_preferred)
         issue_authentication(username, password, save_auth, :socket => socket)
+      ensure
+        socket.pool.checkin(socket) if socket
       end
 
       @connection.authenticate_pools
@@ -628,13 +637,6 @@ def validate_collection(name)
       doc
     end
 
-    # The value of the read preference. This will be
-    # either +:primary+, +:secondary+, or an object
-    # representing the tags to be read from.
-    def read_preference
-      @read_preference
-    end
-
     private
 
     def system_command_collection