SSSSSSSSSSSSSSSSSSSSSSSSSSSSMOKIN' FAST new implementation of thread locals

The previous implementation of thread-local variables would acquire a lock EVERY
time you read OR wrote a TLV. This completely eliminates the scalability advantage
of using thread-locals.

This implementation uses one array per thread to hold the values of thread-locals.
Obviously, the more thread-locals you create, the bigger the arrays will be. Since
each array is accessed by one thread only, no locking is required to read/write a
TLV.

Author: alexdowad

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
 
@@ -70,7 +70,7 @@ def value
     # @!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)
@@ -81,7 +81,7 @@ def value=(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
@@ -119,29 +119,129 @@ def set(value)
   # @!macro internal_implementation_note
   class RubyThreadLocalVar < AbstractThreadLocalVar
 
+    # 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
+
+    FREE = []
+    LOCK = Mutex.new
+    ARRAYS = {} # used as a hash set
+    @@next = 0
+
     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
+    def 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
 
-      @storage[key] = value
+    public
 
+    # @!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
+      if array = Thread.current[:__threadlocal_array__]
+        value = array[@index]
+        if value.nil?
+          @default
+        elsif value.equal?(NIL_SENTINEL)
+          nil
+        else
+          value
+        end
+      else
+        @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)
+      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
+      array[@index] = (value.nil? ? NIL_SENTINEL : value)
+      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 block_given?
+        old_value = self.value
         begin
+          self.value = value
           yield
         ensure
-          @storage.delete(key)
+          self.value = old_value
         end
       end
     end