RUBY-416 do not checkout closed sockets

mongodb · Mar 7, 2012 · 06bc50f · 06bc50f
1 parent 8c3283f
commit 06bc50f
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 4 deletions.
diff --git a/lib/mongo/cursor.rb b/lib/mongo/cursor.rb
@@ -470,7 +470,7 @@ def send_initial_query
           Mongo::Constants::OP_QUERY, message, nil, sock, @command,
           nil, @options & OP_QUERY_EXHAUST != 0)
         rescue ConnectionFailure, OperationFailure, OperationTimeout => ex
-          force_checkin_socket(sock)
+          force_checkin_socket(sock) unless @socket
           raise ex
         end
         checkin_socket(sock) unless @socket

diff --git a/lib/mongo/repl_set_connection.rb b/lib/mongo/repl_set_connection.rb
@@ -355,15 +355,15 @@ def checkout_writer
 
     # Checkin a socket used for reading.
     def checkin_reader(socket)
-      if socket && socket.pool
+      if socket
         socket.pool.checkin(socket)
       end
       sync_refresh
     end
 
     # Checkin a socket used for writing.
     def checkin_writer(socket)
-      if socket && socket.pool
+      if socket
         socket.pool.checkin(socket)
       end
       sync_refresh

diff --git a/lib/mongo/util/pool.rb b/lib/mongo/util/pool.rb
@@ -19,7 +19,7 @@ module Mongo
   class Pool
     PING_ATTEMPTS  = 6
     MAX_PING_TIME  = 1_000_000
-    PRUNE_INTERVAL = 1000
+    PRUNE_INTERVAL = 10000
 
     attr_accessor :host, :port, :address,
       :size, :timeout, :safe, :checked_out, :connection
@@ -66,6 +66,7 @@ def initialize(connection, host, port, opts={})
     def close(opts={})
       @connection_mutex.synchronize do
         if opts[:soft] && !@checked_out.empty?
+          @closing = true
           close_sockets(@sockets - @checked_out)
         else
           close_sockets(@sockets)
@@ -282,6 +283,18 @@ def checkout
               op.call
             end
 
+            if socket.closed?
+              @checked_out.delete(socket)
+              @sockets.delete(socket)
+              @threads_to_sockets.each do |k,v|
+                if v == socket
+                  @threads_to_sockets.delete(k)
+                end
+              end
+
+              socket = checkout_new_socket
+            end
+
             return socket
           else
             # Otherwise, wait
@@ -295,6 +308,7 @@ def checkout
 
     def close_sockets(sockets)
       sockets.each do |socket|
+        @sockets.delete(socket)
         begin
           socket.close unless socket.closed?
         rescue IOError => ex