Merge pull request #36998 from jhawthorn/reaper_fork

Ensure connection reaper threads respawn in forks

activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb

@@ -317,14 +317,15 @@ def initialize(pool, frequency)
 
         @mutex = Mutex.new
         @pools = {}
+        @threads = {}
 
         class << self
           def register_pool(pool, frequency) # :nodoc:
             @mutex.synchronize do
-              unless @pools.key?(frequency)
-                @pools[frequency] = []
-                spawn_thread(frequency)
+              unless @threads[frequency]&.alive?
+                @threads[frequency] = spawn_thread(frequency)
               end
+              @pools[frequency] ||= []
               @pools[frequency] << WeakRef.new(pool)
             end
           end
@@ -333,7 +334,8 @@ def register_pool(pool, frequency) # :nodoc:
 
             def spawn_thread(frequency)
               Thread.new(frequency) do |t|
-                loop do
+                running = true
+                while running
                   sleep t
                   @mutex.synchronize do
                     @pools[frequency].select!(&:weakref_alive?)
@@ -342,6 +344,12 @@ def spawn_thread(frequency)
                       p.flush
                     rescue WeakRef::RefError
                     end
+
+                    if @pools[frequency].empty?
+                      @pools.delete(frequency)
+                      @threads.delete(frequency)
+                      running = false
+                    end
                   end
                 end
               end

activerecord/test/cases/reaper_test.rb

@@ -72,21 +72,60 @@ def test_connection_pool_starts_reaper
 
         pool = ConnectionPool.new spec
 
+        conn, child = new_conn_in_thread(pool)
+
+        assert_predicate conn, :in_use?
+
+        child.terminate
+
+        wait_for_conn_idle(conn)
+        assert_not_predicate conn, :in_use?
+      end
+
+      def test_connection_pool_starts_reaper_in_fork
+        spec = ActiveRecord::Base.connection_pool.spec.dup
+        spec.config[:reaping_frequency] = "0.0001"
+
+        pool = ConnectionPool.new spec
+        pool.checkout
+
+        pid = fork do
+          pool = ConnectionPool.new spec
+
+          conn, child = new_conn_in_thread(pool)
+          child.terminate
+
+          wait_for_conn_idle(conn)
+          if conn.in_use?
+            exit!(1)
+          else
+            exit!(0)
+          end
+        end
+
+        Process.waitpid(pid)
+        assert $?.success?
+      end
+
+      def new_conn_in_thread(pool)
+        event = Concurrent::Event.new
         conn = nil
+
         child = Thread.new do
           conn = pool.checkout
+          event.set
           Thread.stop
         end
-        Thread.pass while conn.nil?
-
-        assert_predicate conn, :in_use?
 
-        child.terminate
+        event.wait
+        [conn, child]
+      end
 
-        while conn.in_use?
+      def wait_for_conn_idle(conn, timeout = 5)
+        start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        while conn.in_use? && Process.clock_gettime(Process::CLOCK_MONOTONIC) - start < timeout
           Thread.pass
         end
-        assert_not_predicate conn, :in_use?
       end
     end
   end