Merge pull request rails#52731 from tgwizard/thread-pool-for-live-action-controller

Use thread pool for ActionController::Live

Author: rafaelfranca

actionpack/CHANGELOG.md

@@ -1,3 +1,7 @@
+*   Update `ActionController::Live` to use a thread-pool to reuse threads across requests.
+
+    *Adam Renberg Tamm*
+
 *   Introduce safer, more explicit params handling method with `params#expect` such that
     `params.expect(table: [ :attr ])` replaces `params.require(:table).permit(:attr)`
 

actionpack/lib/action_controller/metal/live.rb

@@ -304,6 +304,10 @@ def process(name)
               error = e
             end
           ensure
+            # Ensure we clean up any thread locals we copied so that the thread can reused.
+            ActiveSupport::IsolatedExecutionState.clear
+            locals.each { |k, _| t2[k] = nil }
+
             @_response.commit!
           end
         end
@@ -368,11 +372,15 @@ def send_stream(filename:, disposition: "attachment", type: nil)
       # data from the response bodies. Nobody should call this method except in Rails
       # internals. Seriously!
       def new_controller_thread # :nodoc:
-        Thread.new {
+        ActionController::Live.live_thread_pool_executor.post do
           t2 = Thread.current
           t2.abort_on_exception = true
           yield
-        }
+        end
+      end
+
+      def self.live_thread_pool_executor
+        @live_thread_pool_executor ||= Concurrent::CachedThreadPool.new(name: "action_controller.live")
       end
 
       def log_error(exception)

actionpack/lib/action_controller/test_case.rb

@@ -22,6 +22,8 @@ module Live
     # database on the main thread, so they could open a txn, then the controller
     # thread will open a new connection and try to access data that's only visible
     # to the main thread's txn. This is the problem in #23483.
+    alias_method :original_new_controller_thread, :new_controller_thread
+
     silence_redefinition_of_method :new_controller_thread
     def new_controller_thread # :nodoc:
       yield

actionpack/test/controller/live_stream_test.rb

@@ -228,10 +228,24 @@ def thread_locals
         response.stream.close
       end
 
+      def no_thread_locals
+        tc.assert_nil Thread.current[:setting]
+
+        response.headers["Content-Type"] = "text/event-stream"
+        %w{ hello world }.each do |word|
+          response.stream.write word
+        end
+        response.stream.close
+      end
+
       def isolated_state
         render plain: CurrentState.id.inspect
       end
 
+      def raw_isolated_state
+        render plain: ActiveSupport::IsolatedExecutionState[:raw_isolated_state].inspect
+      end
+
       def with_stale
         render plain: "stale" if stale?(etag: "123", template: false)
       end
@@ -344,7 +358,7 @@ def setup
       super
 
       def @controller.new_controller_thread(&block)
-        Thread.new(&block)
+        original_new_controller_thread(&block)
       end
     end
 
@@ -525,6 +539,34 @@ def test_isolated_state_get_copied
       assert_stream_closed
     end
 
+    def test_thread_locals_get_reset
+      @controller.tc = self
+
+      Thread.current[:originating_thread] = Thread.current.object_id
+      Thread.current[:setting]            = "aaron"
+
+      get :thread_locals
+
+      Thread.current[:setting] = nil
+
+      get :no_thread_locals
+    end
+
+    def test_isolated_state_get_reset
+      @controller.tc = self
+      ActiveSupport::IsolatedExecutionState[:raw_isolated_state] = "buffy"
+
+      get :raw_isolated_state
+      assert_equal "buffy".inspect, response.body
+      assert_stream_closed
+
+      ActiveSupport::IsolatedExecutionState.clear
+
+      get :isolated_state
+      assert_equal nil.inspect, response.body
+      assert_stream_closed
+    end
+
     def test_live_stream_default_header
       get :default_header
       assert response.headers["Content-Type"]
@@ -565,14 +607,9 @@ def test_exception_handling_plain_text
     end
 
     def test_exception_callback_when_committed
-      current_threads = Thread.list
-
       capture_log_output do |output|
         get :exception_with_callback, format: "text/event-stream"
 
-        # Wait on the execution of all threads
-        (Thread.list - current_threads).each(&:join)
-
         assert_equal %(data: "500 Internal Server Error"\n\n), response.body
         assert_match "An exception occurred...", output.rewind && output.read
         assert_stream_closed