Various fixes for ThreadLocalVar and FiberLocalVar

Author: eregon

lib/concurrent-ruby/concurrent/atomic/fiber_local_var.rb

@@ -2,39 +2,50 @@
 require_relative 'locals'
 
 module Concurrent
-  class FiberLocalVar
-    LOCALS = FiberLocals.new(:concurrent_fiber_local_var)
-
-    # @!macro fiber_local_var_method_initialize
-    #
-    #   Creates a fiber local variable.
-    #
-    #   @param [Object] default the default value when otherwise unset
-    #   @param [Proc] default_block Optional block that gets called to obtain the
-    #     default value for each fiber
 
-    # @!macro fiber_local_var_method_get
-    #
-    #   Returns the value in the current fiber's copy of this fiber-local variable.
-    #
-    #   @return [Object] the current value
-
-    # @!macro fiber_local_var_method_set
-    #
-    #   Sets the current fiber's copy of this fiber-local variable to the specified value.
-    #
-    #   @param [Object] value the value to set
-    #   @return [Object] the new value
+  # A `FiberLocalVar` is a variable where the value is different for each fiber.
+  # Each variable may have a default value, but when you modify the variable only
+  # the current fiber will ever see that change.
+  #
+  # This is similar to Ruby's built-in fiber-local variables (`Thread.current[:name]`),
+  # but with these major advantages:
+  # * `FiberLocalVar` has its own identity, it doesn't need a Symbol.
+  # * Each Ruby's built-in fiber-local variable leaks some memory forever (it's a Symbol held forever on the fiber),
+  #   so it's only OK to create a small amount of them.
+  #   `FiberLocalVar` has no such issue and it is fine to create many of them.
+  # * Ruby's built-in fiber-local variables leak forever the value set on each fiber (unless set to nil explicitly).
+  #   `FiberLocalVar` automatically removes the mapping for each fiber once the `FiberLocalVar` instance is GC'd.
+  #
+  # @example
+  #   v = FiberLocalVar.new(14)
+  #   v.value #=> 14
+  #   v.value = 2
+  #   v.value #=> 2
+  #
+  # @example
+  #   v = FiberLocalVar.new(14)
+  #
+  #   Fiber.new do
+  #     v.value #=> 14
+  #     v.value = 1
+  #     v.value #=> 1
+  #   end.resume
+  #
+  #   Fiber.new do
+  #     v.value #=> 14
+  #     v.value = 2
+  #     v.value #=> 2
+  #   end.resume
+  #
+  #   v.value #=> 14
+  class FiberLocalVar
+    LOCALS = FiberLocals.new
 
-    # @!macro fiber_local_var_method_bind
+    # Creates a fiber local variable.
     #
-    #   Bind the given value to fiber local storage during
-    #   execution of the given block.
-    #
-    #   @param [Object] value the value to bind
-    #   @yield the operation to be performed with the bound variable
-    #   @return [Object] the value
-
+    # @param [Object] default the default value when otherwise unset
+    # @param [Proc] default_block Optional block that gets called to obtain the
+    #   default value for each fiber
     def initialize(default = nil, &default_block)
       if default && block_given?
         raise ArgumentError, "Cannot use both value and block as default value"
@@ -51,22 +62,32 @@ def initialize(default = nil, &default_block)
       @index = LOCALS.next_index(self)
     end
 
-    # @!macro fiber_local_var_method_get
+    # Returns the value in the current fiber's copy of this fiber-local variable.
+    #
+    # @return [Object] the current value
     def value
-      LOCALS.fetch(@index) {default}
+      LOCALS.fetch(@index) { default }
     end
 
-    # @!macro fiber_local_var_method_set
+    # Sets the current fiber's copy of this fiber-local variable to the specified value.
+    #
+    # @param [Object] value the value to set
+    # @return [Object] the new value
     def value=(value)
       LOCALS.set(@index, value)
     end
 
-    # @!macro fiber_local_var_method_bind
-    def bind(value, &block)
+    # Bind the given value to fiber local storage during
+    # execution of the given block.
+    #
+    # @param [Object] value the value to bind
+    # @yield the operation to be performed with the bound variable
+    # @return [Object] the value
+    def bind(value)
       if block_given?
         old_value = self.value
+        self.value = value
         begin
-          self.value = value
           yield
         ensure
           self.value = old_value

lib/concurrent-ruby/concurrent/atomic/locals.rb

@@ -32,13 +32,11 @@ module Concurrent
   # them. This is why we need to use a finalizer to clean up the locals array
   # when the EC goes out of scope.
   class AbstractLocals
-    def initialize(name_prefix = :concurrent_locals)
+    def initialize
       @free = []
       @lock = Mutex.new
-      @all_locals = {}
+      @all_arrays = {}
       @next = 0
-
-      @name = :"#{name_prefix}_#{object_id}"
     end
 
     def synchronize
@@ -53,7 +51,7 @@ def weak_synchronize
       alias_method :weak_synchronize, :synchronize
     end
 
-    def next_index(target)
+    def next_index(local)
       index = synchronize do
         if @free.empty?
           @next += 1
@@ -62,11 +60,11 @@ def next_index(target)
         end
       end
 
-      # When the target goes out of scope, we should free the associated index
+      # When the local goes out of scope, we should free the associated index
       # and all values stored into it.
-      ObjectSpace.define_finalizer(target, target_finalizer(index))
+      ObjectSpace.define_finalizer(local, local_finalizer(index))
 
-      return index
+      index
     end
 
     def free_index(index)
@@ -78,7 +76,7 @@ def free_index(index)
         #
         # DO NOT use each_value which might conflict with new pair assignment
         # into the hash in #set method.
-        @all_locals.values.each do |locals|
+        @all_arrays.values.each do |locals|
           locals[index] = nil
         end
 
@@ -87,18 +85,13 @@ def free_index(index)
       end
     end
 
-    def fetch(index, default = nil)
-      if locals = self.locals
-        value = locals[index]
-      end
+    def fetch(index)
+      locals = self.locals
+      value = locals ? locals[index] : nil
 
-      if value.nil?
-        if block_given?
-          yield
-        else
-          default
-        end
-      elsif value.equal?(NULL)
+      if nil == value
+        yield
+      elsif NULL.equal?(value)
         nil
       else
         value
@@ -107,25 +100,25 @@ def fetch(index, default = nil)
 
     def set(index, value)
       locals = self.locals!
-      locals[index] = (value.nil? ? NULL : value)
+      locals[index] = (nil == value ? NULL : value)
 
       value
     end
 
     private
 
-    # When the target index goes out of scope, clean up that slot across all locals currently assigned.
-    def target_finalizer(index)
+    # When the local goes out of scope, clean up that slot across all locals currently assigned.
+    def local_finalizer(index)
       proc do
         free_index(index)
       end
     end
 
-    # When a target (locals) goes out of scope, delete the locals from all known locals.
-    def locals_finalizer(locals_object_id)
-      proc do |locals_id|
+    # When a thread/fiber goes out of scope, remove the array from @all_arrays.
+    def thread_fiber_finalizer(array_object_id)
+      proc do
         weak_synchronize do
-          @all_locals.delete(locals_object_id)
+          @all_arrays.delete(array_object_id)
         end
       end
     end
@@ -146,23 +139,23 @@ def locals!
   # An array-backed storage of indexed variables per thread.
   class ThreadLocals < AbstractLocals
     def locals
-      Thread.current.thread_variable_get(@name)
+      Thread.current.thread_variable_get(:concurrent_thread_locals)
     end
 
     def locals!
       thread = Thread.current
-      locals = thread.thread_variable_get(@name)
+      locals = thread.thread_variable_get(:concurrent_thread_locals)
 
       unless locals
-        locals = thread.thread_variable_set(@name, [])
+        locals = thread.thread_variable_set(:concurrent_thread_locals, [])
         weak_synchronize do
-          @all_locals[locals.object_id] = locals
-          # When the thread goes out of scope, we should delete the associated locals:
-          ObjectSpace.define_finalizer(thread, locals_finalizer(locals.object_id))
+          @all_arrays[locals.object_id] = locals
         end
+        # When the thread goes out of scope, we should delete the associated locals:
+        ObjectSpace.define_finalizer(thread, thread_fiber_finalizer(locals.object_id))
       end
 
-      return locals
+      locals
     end
   end
 
@@ -171,23 +164,25 @@ def locals!
   # An array-backed storage of indexed variables per fiber.
   class FiberLocals < AbstractLocals
     def locals
-      Thread.current[@name]
+      Thread.current[:concurrent_fiber_locals]
     end
 
     def locals!
       thread = Thread.current
-      locals = thread[@name]
+      locals = thread[:concurrent_fiber_locals]
 
       unless locals
-        locals = thread[@name] = []
+        locals = thread[:concurrent_fiber_locals] = []
         weak_synchronize do
-          @all_locals[locals.object_id] = locals
-          # When the thread goes out of scope, we should delete the associated locals:
-          ObjectSpace.define_finalizer(Fiber.current, locals_finalizer(locals.object_id))
+          @all_arrays[locals.object_id] = locals
         end
+        # When the fiber goes out of scope, we should delete the associated locals:
+        ObjectSpace.define_finalizer(Fiber.current, thread_fiber_finalizer(locals.object_id))
       end
 
-      return locals
+      locals
     end
   end
+
+  private_constant :AbstractLocals, :ThreadLocals, :FiberLocals
 end

lib/concurrent-ruby/concurrent/atomic/lock_local_var.rb

@@ -1,15 +1,17 @@
+require 'concurrent/utility/engine'
 require_relative 'fiber_local_var'
 require_relative 'thread_local_var'
 
 module Concurrent
   # @!visibility private
   def self.mutex_owned_per_thread?
-    mutex = Mutex.new
+    return false if Concurrent.on_jruby? || Concurrent.on_truffleruby?
 
+    mutex = Mutex.new
     # Lock the mutex:
     mutex.synchronize do
       # Check if the mutex is still owned in a child fiber:
-      Fiber.new{mutex.owned?}.resume
+      Fiber.new { mutex.owned? }.resume
     end
   end
 
@@ -18,4 +20,9 @@ def self.mutex_owned_per_thread?
   else
     LockLocalVar = FiberLocalVar
   end
+
+  # Either {FiberLocalVar} or {ThreadLocalVar} depending on whether Mutex (and Monitor)
+  # are held, respectively, per Fiber or per Thread.
+  class LockLocalVar
+  end
 end

lib/concurrent-ruby/concurrent/atomic/reentrant_read_write_lock.rb

@@ -111,7 +111,7 @@ def initialize
       @Counter    = AtomicFixnum.new(0)       # single integer which represents lock state
       @ReadQueue  = Synchronization::Lock.new # used to queue waiting readers
       @WriteQueue = Synchronization::Lock.new # used to queue waiting writers
-      @HeldCount  = LockLocalVar.new(0)       # indicates # of R & W locks held by this thread
+      @HeldCount  = LockLocalVar.new(0) # indicates # of R & W locks held by this thread
     end
 
     # Execute a block operation within a read lock.