Ensure reaper threads respawn in forks

This broke in 3e2e8ee where it switched to one reaper thread per
process. However, the implementation will only spawn the reaping thread
if the reaping frequency has not been seen.  Since the reaping
frequencies are stored in a class instance variable, that variable is
copied when the process is forked. As such, a forked process will not
spawn a new reaping thread since the class instance variable would have
contain the reaping frequency that was seen in the parent process.

This commit tracks threads separately and checks that they both have
been spawned and are currently alive.

This adds a failing test to reaper_test.rb, based on the previous test
without forking. It also improves the test to return an failure instead
of hanging forever if the reaper is not started.

Co-authored-by: Guo Xiang Tan <tgx_world@hotmail.com>

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

@@ -316,14 +316,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
@@ -332,7 +333,7 @@ def register_pool(pool, frequency) # :nodoc:
             def spawn_thread(frequency)
               Thread.new(frequency) do |t|
                 running = true
-                while running do
+                while running
                   sleep t
                   @mutex.synchronize do
                     @pools[frequency].select!(&:weakref_alive?)
@@ -344,6 +345,7 @@ def spawn_thread(frequency)
 
                     if @pools[frequency].empty?
                       @pools.delete(frequency)
+                      @threads.delete(frequency)
                       running = false
                     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