RUBY-3680 Close pipe fds

Author: comandeo-mongo

lib/mongo/server/connection_pool.rb

@@ -671,6 +671,7 @@ def close(options = nil)
 
           @max_connecting_cv.broadcast
           @size_cv.broadcast
+          @generation_manager.close_all_pipes
         end
 
         publish_cmap_event(

lib/mongo/server/connection_pool/generation_manager.rb

@@ -47,13 +47,15 @@ def generation_unlocked(service_id: nil)
         end
 
         def pipe_fds(service_id: nil)
-          @pipe_fds[service_id][@map[service_id]]
+          @pipe_fds.dig(service_id, @map[service_id])
         end
 
         def remove_pipe_fds(generation, service_id: nil)
           validate_service_id!(service_id)
 
           r, w = @pipe_fds[service_id].delete(generation)
+          return unless r && w
+
           w.close
           # Schedule the read end of the pipe to be closed. We cannot close it
           # immediately since we need to wait for any Kernel#select calls to
@@ -89,8 +91,31 @@ def bump(service_id: nil)
           end
         end
 
+        # Close all pipes in the generation manager.
+        #
+        # This method should be called only when the +ConnectionPool+ that
+        # owns this +GenerationManager+ is closed, to ensure that all
+        # pipes are closed properly.
+        def close_all_pipes
+          @lock.synchronize do
+            close_all_scheduled
+            @pipe_fds.keys.each do |service_id|
+              generations = @pipe_fds.delete(service_id)
+              generations.values.each do |(r, w)|
+                r.close
+                w.close
+              rescue IOError
+                # Ignore any IOError that occurs when closing the
+                # pipe, as there is nothing we can do about it.
+              end
+            end
+          end
+        end
+
+
         private
 
+
         def validate_service_id!(service_id)
           if service_id
             unless server.load_balancer?

lib/mongo/socket.rb

@@ -64,6 +64,9 @@ class Socket
     #   connection (for non-monitoring connections) that created this socket.
     # @option options [ true | false ] :monitor Whether this socket was
     #   created by a monitoring connection.
+    # @option options :pipe [ IO ] The file descriptor for the read end of the
+    #   pipe to listen on during the select system call when reading from the
+    #   socket.
     #
     # @api private
     def initialize(timeout, options)
@@ -106,6 +109,13 @@ def monitor?
       !!options[:monitor]
     end
 
+    # @return [ IO ] The file descriptor for the read end of the pipe to
+    # listen on during the select system call when reading from the
+    # socket.
+    def pipe
+      options[:pipe]
+    end
+
     # @return [ String ] Human-readable summary of the socket for debugging.
     #
     # @api private
@@ -161,7 +171,7 @@ def close
       begin
         # Sometimes it seems the close call can hang for a long time
         ::Timeout.timeout(5) do
-          @socket.close
+          @socket&.close
         end
       rescue
         # Silence all errors
@@ -390,7 +400,6 @@ def read_from_socket(length, socket_timeout: nil, csot: false)
             raise_timeout_error!("Took more than #{_timeout} seconds to receive data", csot)
           end
         end
-        pipe = options[:pipe]
         if exc.is_a?(IO::WaitReadable)
           if pipe
             select_args = [[@socket, pipe], nil, [@socket, pipe], select_timeout]

spec/mongo/server/connection_pool/generation_manager_spec.rb

@@ -0,0 +1,39 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Mongo::Server::ConnectionPool::GenerationManager do
+  describe '#close_all_pipes' do
+    let(:service_id) { 'test_service_id' }
+
+    let(:server) { instance_double('Mongo::Server') }
+
+    let(:manager) { described_class.new(server: server) }
+
+    before do
+      manager.pipe_fds(service_id: service_id)
+    end
+
+    it 'closes all pipes' do
+      expect(manager.pipe_fds(service_id: service_id).size).to eq(2)
+
+      manager.instance_variable_get(:@pipe_fds)[service_id].each do |gen, (r, w)|
+        expect(r).to receive(:close)
+        expect(w).to receive(:close)
+      end
+
+      manager.close_all_pipes
+    end
+
+    it 'removes all pipes from the map' do
+      expect(manager.pipe_fds(service_id: service_id).size).to eq(2)
+
+      manager.instance_variable_get(:@pipe_fds)[service_id].each do |gen, (r, w)|
+        expect(r).to receive(:close)
+        expect(w).to receive(:close)
+      end
+
+      manager.close_all_pipes
+    end
+  end
+end

spec/mongo/server/connection_pool_spec.rb

@@ -1215,6 +1215,11 @@ def create_pool(min_pool_size)
         expect(pool).to be_closed
       end
     end
+
+    it 'closes all pipes' do
+      expect(pool.generation_manager).to receive(:close_all_pipes).and_call_original
+      pool.close
+    end
   end
 
   describe '#inspect' do

spec/mongo/socket_spec.rb

@@ -167,4 +167,19 @@
       end
     end
   end
+
+  describe '#close' do
+    let(:pipe) { instance_double(IO) }
+
+    let(:socket) do
+      described_class.new(0, {pipe: pipe})
+    end
+
+    context 'when pipe is provided' do
+      it 'closes the pipe' do
+        expect(pipe).to receive(:close)
+        socket.close
+      end
+    end
+  end
 end