Merge pull request #376 from alexdowad/experimental_tls_implementation

Experimental thread-local variable implementation

Author: jdantonio

examples/thread_local_memory_usage.rb

@@ -0,0 +1,74 @@
+#!/usr/bin/env ruby
+
+$: << File.expand_path('../../lib', __FILE__)
+
+$DEBUG_TLV = true
+require 'concurrent'
+require 'concurrent/atomic/thread_local_var'
+require 'benchmark'
+require 'thread'
+
+include Concurrent
+
+# if we hold on to vars, but threads die, space used for TLVs should be recovered
+
+def test_thread_gc(vars)
+  threads = 500.times.collect do
+    Thread.new do
+      vars.each do |var|
+        var.value = 1
+      end
+    end
+  end
+  threads.each(&:join)
+end
+
+puts "BEFORE THREAD GC TEST:"
+puts "Ruby heap pages: #{GC.stat[:heap_length]}, Other malloc'd bytes: #{GC.stat[:malloc_increase]}"
+
+vars = 500.times.collect { ThreadLocalVar.new(0) }
+200.times do
+  test_thread_gc(vars)
+  GC.start
+end
+
+puts "AFTER THREAD GC TEST:"
+puts "Ruby heap pages: #{GC.stat[:heap_length]}, Other malloc'd bytes: #{GC.stat[:malloc_increase]}"
+
+# if we hold on to threads, but drop TLVs, space used should be reused by allocated TLVs
+
+def tlv_gc_test_loop(queue)
+  while true
+    var = queue.pop
+    return if var.nil?
+    var.value = 1
+  end
+end
+
+def test_tlv_gc(queues)
+  500.times do
+    var = ThreadLocalVar.new(0)
+    queues.each { |q| q << var }
+  end
+end
+
+puts
+puts "BEFORE TLV GC TEST:"
+puts "Ruby heap pages: #{GC.stat[:heap_length]}, Other malloc'd bytes: #{GC.stat[:malloc_increase]}"
+
+queues  = 500.times.collect { Queue.new }
+threads = queues.map do |queue|
+  Thread.new do
+    tlv_gc_test_loop(queue)
+  end
+end
+
+200.times do
+  test_tlv_gc(queues)
+  GC.start
+end
+queues.each { |q| q << nil }
+threads.each(&:join)
+
+puts "AFTER TLV GC TEST:"
+puts "Ruby heap pages: #{GC.stat[:heap_length]}, Other malloc'd bytes: #{GC.stat[:malloc_increase]}"

examples/thread_local_var_bench.rb

@@ -0,0 +1,31 @@
+#!/usr/bin/env ruby
+
+$: << File.expand_path('../../lib', __FILE__)
+
+require 'concurrent'
+require 'concurrent/atomic/thread_local_var'
+require 'benchmark'
+
+include Concurrent
+
+N_THREADS = 100
+N_VARS    = 100
+
+vars = N_VARS.times.collect { ThreadLocalVar.new(0) }
+
+def test_threadlocal_perf(vars)
+  threads = N_THREADS.times.collect do
+    Thread.new do
+      10000.times do
+        index = rand(N_VARS)
+        var   = vars[index]
+        var.value = var.value + 1
+      end
+    end
+  end
+  threads.each(&:join)
+end
+
+Benchmark.bmbm do |bm|
+  bm.report('ThreadLocalVar') { test_threadlocal_perf(vars) }
+end

lib/concurrent/atomic/thread_local_var.rb

@@ -1,4 +1,4 @@
-require 'concurrent/atomic/thread_local_var/weak_key_map'
+require 'thread'
 
 module Concurrent
 
@@ -33,169 +33,132 @@ module Concurrent
   #
   #   @see https://docs.oracle.com/javase/7/docs/api/java/lang/ThreadLocal.html Java ThreadLocal
   #
-  # @!visibility private
-  class AbstractThreadLocalVar
+  class ThreadLocalVar
+
+    # Each thread has a (lazily initialized) array of thread-local variable values
+    # Each time a new thread-local var is created, we allocate an "index" for it
+    # For example, if the allocated index is 1, that means slot #1 in EVERY
+    #   thread's thread-local array will be used for the value of that TLV
+    #
+    # The good thing about using a per-THREAD structure to hold values, rather
+    #   than a per-TLV structure, is that no synchronization is needed when
+    #    reading and writing those values (since the structure is only ever
+    #    accessed by a single thread)
+    #
+    # Of course, when a TLV is GC'd, 1) we need to recover its index for use
+    #   by other new TLVs (otherwise the thread-local arrays could get bigger
+    #   and bigger with time), and 2) we need to null out all the references
+    #   held in the now-unused slots (both to avoid blocking GC of those objects,
+    #   and also to prevent "stale" values from being passed on to a new TLV
+    #   when the index is reused)
+    # Because we need to null out freed slots, we need to keep references to
+    #   ALL the thread-local arrays -- ARRAYS is for that
+    # But when a Thread is GC'd, we need to drop the reference to its thread-local
+    #   array, so we don't leak memory
 
     # @!visibility private
     NIL_SENTINEL = Object.new
-    private_constant :NIL_SENTINEL
+    FREE = []
+    LOCK = Mutex.new
+    ARRAYS = {} # used as a hash set
+    @@next = 0
+    private_constant :NIL_SENTINEL, :FREE, :LOCK, :ARRAYS
 
-    # @!macro [attach] thread_local_var_method_initialize
-    #
     #   Creates a thread local variable.
     #
     #   @param [Object] default the default value when otherwise unset
     def initialize(default = nil)
       @default = default
-      allocate_storage
+      @index = LOCK.synchronize do
+        FREE.pop || begin
+          result = @@next
+          @@next += 1
+          result
+        end
+      end
+      ObjectSpace.define_finalizer(self, self.class.threadlocal_finalizer(@index))
     end
 
-    # @!macro [attach] thread_local_var_method_get
-    #
     #   Returns the value in the current thread's copy of this thread-local variable.
     #
     #   @return [Object] the current value
     def value
-      value = get
-
-      if value.nil?
-        @default
-      elsif value == NIL_SENTINEL
-        nil
+      if array = Thread.current[:__threadlocal_array__]
+        value = array[@index]
+        if value.nil?
+          @default
+        elsif value.equal?(NIL_SENTINEL)
+          nil
+        else
+          value
+        end
       else
-        value
+        @default
       end
     end
 
-    # @!macro [attach] thread_local_var_method_set
-    #
     #   Sets the current thread's copy of this thread-local variable to the specified value.
-    #   
+    #
     #   @param [Object] value the value to set
     #   @return [Object] the new value
     def value=(value)
-      bind value
-    end
-
-    # @!macro [attach] thread_local_var_method_bind
-    #
-    #   Bind the given value to thread 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, &block)
-      if value.nil?
-        stored_value = NIL_SENTINEL
-      else
-        stored_value = value
+      me = Thread.current
+      # We could keep the thread-local arrays in a hash, keyed by Thread
+      # But why? That would require locking
+      # Using Ruby's built-in thread-local storage is faster
+      unless array = me[:__threadlocal_array__]
+        array = me[:__threadlocal_array__] = []
+        LOCK.synchronize { ARRAYS[array.object_id] = array }
+        ObjectSpace.define_finalizer(me, self.class.thread_finalizer(array))
       end
-
-      set(stored_value, &block)
-
+      array[@index] = (value.nil? ? NIL_SENTINEL : value)
       value
     end
 
     protected
 
     # @!visibility private
-    def allocate_storage
-      raise NotImplementedError
-    end
-
-    # @!visibility private
-    def get
-      raise NotImplementedError
-    end
-
-    # @!visibility private
-    def set(value)
-      raise NotImplementedError
-    end
-  end
-
-  # @!visibility private
-  # @!macro internal_implementation_note
-  class RubyThreadLocalVar < AbstractThreadLocalVar
-
-    protected
-
-    # @!visibility private
-    def allocate_storage
-      @storage = WeakKeyMap.new
+    def self.threadlocal_finalizer(index)
+      proc do
+        LOCK.synchronize do
+          FREE.push(index)
+          # The cost of GC'ing a TLV is linear in the number of threads using TLVs
+          # But that is natural! More threads means more storage is used per TLV
+          # So naturally more CPU time is required to free more storage
+          ARRAYS.each_value do |array|
+            array[index] = nil
+          end
+        end
+      end
     end
 
     # @!visibility private
-    def get
-      @storage[Thread.current]
+    def self.thread_finalizer(array)
+      proc do
+        LOCK.synchronize do
+          # The thread which used this thread-local array is now gone
+          # So don't hold onto a reference to the array (thus blocking GC)
+          ARRAYS.delete(array.object_id)
+        end
+      end
     end
 
-    # @!visibility private
-    def set(value)
-      key = Thread.current
-
-      @storage[key] = value
-
+    #   Bind the given value to thread 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, &block)
       if block_given?
+        old_value = self.value
         begin
+          self.value = value
           yield
         ensure
-          @storage.delete(key)
+          self.value = old_value
         end
       end
     end
   end
-
-  if Concurrent.on_jruby?
-
-    # @!visibility private
-    # @!macro internal_implementation_note
-    class JavaThreadLocalVar < AbstractThreadLocalVar
-
-      protected
-
-      # @!visibility private
-      def allocate_storage
-        @var = java.lang.ThreadLocal.new
-      end
-
-      # @!visibility private
-      def get
-        @var.get
-      end
-
-      # @!visibility private
-      def set(value)
-        @var.set(value)
-      end
-    end
-  end
-
-  # @!visibility private
-  # @!macro internal_implementation_note
-  ThreadLocalVarImplementation = case
-                                 when Concurrent.on_jruby?
-                                   JavaThreadLocalVar
-                                 else
-                                   RubyThreadLocalVar
-                                 end
-  private_constant :ThreadLocalVarImplementation
-
-  # @!macro thread_local_var
-  class ThreadLocalVar < ThreadLocalVarImplementation
-
-    # @!method initialize(default = nil)
-    #   @!macro thread_local_var_method_initialize
-
-    # @!method value
-    #   @!macro thread_local_var_method_get
-
-    # @!method value=(value)
-    #   @!macro thread_local_var_method_set
-
-    # @!method bind(value, &block)
-    #   @!macro thread_local_var_method_bind
-
-  end
 end