Merge pull request rails#46553 from Shopify/thread-lock

Make AbstractAdapter#lock thread local by default

Author: byroot

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

@@ -175,7 +175,7 @@ def lock_thread=(lock_thread)
         end
 
         if (active_connection = @thread_cached_conns[connection_cache_key(current_thread)])
-          active_connection.synchronized = lock_thread
+          active_connection.lock_thread = @lock_thread
         end
       end
 
@@ -357,7 +357,7 @@ def clear_reloadable_connections!
       # - ActiveRecord::ConnectionTimeoutError no connection can be obtained from the pool.
       def checkout(checkout_timeout = @checkout_timeout)
         connection = checkout_and_verify(acquire_connection(checkout_timeout))
-        connection.synchronized = @lock_thread
+        connection.lock_thread = @lock_thread
         connection
       end
 
@@ -375,6 +375,7 @@ def checkin(conn)
               conn.expire
             end
 
+            conn.lock_thread = nil
             @available.add conn
           end
         end

activerecord/lib/active_record/connection_adapters/abstract_adapter.rb

@@ -156,7 +156,7 @@ def initialize(config_or_deprecated_connection, deprecated_logger = nil, depreca
         @idle_since = Process.clock_gettime(Process::CLOCK_MONOTONIC)
         @visitor = arel_visitor
         @statements = build_statement_pool
-        self.synchronized = false
+        self.lock_thread = nil
 
         @prepared_statements = self.class.type_cast_config_to_boolean(
           @config.fetch(:prepared_statements) { default_prepared_statements }
@@ -172,8 +172,12 @@ def initialize(config_or_deprecated_connection, deprecated_logger = nil, depreca
         @verified = false
       end
 
-      def synchronized=(synchronized) # :nodoc:
-        @lock = if synchronized
+      def lock_thread=(lock_thread) # :nodoc:
+        @lock =
+        case lock_thread
+        when Thread
+          ActiveSupport::Concurrency::ThreadLoadInterlockAwareMonitor.new
+        when Fiber
           ActiveSupport::Concurrency::LoadInterlockAwareMonitor.new
         else
           ActiveSupport::Concurrency::NullLock

activerecord/test/cases/connection_pool_test.rb

@@ -801,6 +801,19 @@ def setup
         @connection_test_model_class = ThreadConnectionTestModel
       end
 
+      def test_lock_thread_allow_fiber_reentrency
+        @pool.lock_thread = true
+        connection = @pool.checkout
+        connection.transaction do
+          enumerator = Enumerator.new do |yielder|
+            connection.transaction do
+              yielder.yield 1
+            end
+          end
+          assert_equal 1, enumerator.next
+        end
+      end
+
       private
         def new_thread(...)
           Thread.new(...)

activesupport/lib/active_support/concurrency/load_interlock_aware_monitor.rb

@@ -4,9 +4,7 @@
 
 module ActiveSupport
   module Concurrency
-    # A monitor that will permit dependency loading while blocked waiting for
-    # the lock.
-    class LoadInterlockAwareMonitor < Monitor
+    module LoadInterlockAwareMonitorMixin # :nodoc:
       EXCEPTION_NEVER = { Exception => :never }.freeze
       EXCEPTION_IMMEDIATE = { Exception => :immediate }.freeze
       private_constant :EXCEPTION_NEVER, :EXCEPTION_IMMEDIATE
@@ -29,5 +27,46 @@ def synchronize(&block)
         end
       end
     end
+    # A monitor that will permit dependency loading while blocked waiting for
+    # the lock.
+    class LoadInterlockAwareMonitor < Monitor
+      include LoadInterlockAwareMonitorMixin
+    end
+
+    class ThreadLoadInterlockAwareMonitor # :nodoc:
+      prepend LoadInterlockAwareMonitorMixin
+
+      def initialize
+        @owner = nil
+        @count = 0
+        @mutex = Mutex.new
+      end
+
+      private
+        def mon_try_enter
+          if @owner != Thread.current
+            return false unless @mutex.try_lock
+            @owner = Thread.current
+          end
+          @count += 1
+        end
+
+        def mon_enter
+          @mutex.lock if @owner != Thread.current
+          @owner = Thread.current
+          @count += 1
+        end
+
+        def mon_exit
+          unless @owner == Thread.current
+            raise ThreadError, "current thread not owner"
+          end
+
+          @count -= 1
+          return unless @count == 0
+          @owner = nil
+          @mutex.unlock
+        end
+    end
   end
 end

activesupport/test/concurrency/load_interlock_aware_monitor_test.rb

@@ -6,11 +6,7 @@
 
 module ActiveSupport
   module Concurrency
-    class LoadInterlockAwareMonitorTest < ActiveSupport::TestCase
-      def setup
-        @monitor = ActiveSupport::Concurrency::LoadInterlockAwareMonitor.new
-      end
-
+    module LoadInterlockAwareMonitorTests
       def test_entering_with_no_blocking
         assert @monitor.mon_enter
       end
@@ -51,5 +47,32 @@ def test_entering_with_blocking
         assert able_to_load
       end
     end
+
+    class LoadInterlockAwareMonitorTest < ActiveSupport::TestCase
+      include LoadInterlockAwareMonitorTests
+
+      def setup
+        @monitor = ActiveSupport::Concurrency::LoadInterlockAwareMonitor.new
+      end
+    end
+
+    class ThreadLoadInterlockAwareMonitorTest < ActiveSupport::TestCase
+      include LoadInterlockAwareMonitorTests
+
+      def setup
+        @monitor = ActiveSupport::Concurrency::ThreadLoadInterlockAwareMonitor.new
+      end
+
+      def test_lock_owned_by_thread
+        @monitor.synchronize do
+          enumerator = Enumerator.new do |yielder|
+            @monitor.synchronize do
+              yielder.yield 42
+            end
+          end
+          assert_equal 42, enumerator.next
+        end
+      end
+    end
   end
 end