Fix configuration to be thread safe

by replacing ||= operator with Delay instances

Author: pitr-ch

lib/concurrent/configuration.rb

@@ -1,4 +1,5 @@
 require 'thread'
+require 'concurrent/delay'
 require 'concurrent/executor/thread_pool_executor'
 require 'concurrent/executor/timer_set'
 require 'concurrent/utility/processor_count'
@@ -32,33 +33,43 @@ class Configuration
 
     # Create a new configuration object.
     def initialize
-      @cores ||= Concurrent::processor_count
+      @cores = Concurrent::processor_count
+
+      @global_task_pool = Delay.new do
+        Concurrent::ThreadPoolExecutor.new(
+            min_threads:     [2, @cores].max,
+            max_threads:     [20, @cores * 15].max,
+            idletime:        2 * 60, # 2 minutes
+            max_queue:       0, # unlimited
+            overflow_policy: :abort # raise an exception
+        )
+      end
+
+      @global_operation_pool = Delay.new do
+        Concurrent::ThreadPoolExecutor.new(
+            min_threads:     [2, @cores].max,
+            max_threads:     [2, @cores].max,
+            idletime:        10 * 60, # 10 minutes
+            max_queue:       [20, @cores * 15].max,
+            overflow_policy: :abort # raise an exception
+        )
+      end
+
+      @global_timer_set = Delay.new { Concurrent::TimerSet.new }
     end
 
     # Global thread pool optimized for short *tasks*.
     #
     # @return [ThreadPoolExecutor] the thread pool
     def global_task_pool
-      @global_task_pool ||= Concurrent::ThreadPoolExecutor.new(
-        min_threads: [2, @cores].max,
-        max_threads: [20, @cores * 15].max,
-        idletime: 2 * 60,                   # 2 minutes
-        max_queue: 0,                       # unlimited
-        overflow_policy: :abort             # raise an exception
-      )
+      @global_task_pool.value
     end
 
     # Global thread pool optimized for long *operations*.
     #
     # @return [ThreadPoolExecutor] the thread pool
     def global_operation_pool
-      @global_operation_pool ||= Concurrent::ThreadPoolExecutor.new(
-        min_threads: [2, @cores].max,
-        max_threads: [2, @cores].max,
-        idletime: 10 * 60,                  # 10 minutes
-        max_queue: [20, @cores * 15].max,
-        overflow_policy: :abort             # raise an exception
-      )
+      @global_operation_pool.value
     end
 
     # Global thread pool optimized for *timers*
@@ -67,7 +78,7 @@ def global_operation_pool
     #
     # @see Concurrent::timer
     def global_timer_set
-      @global_timer_set ||= Concurrent::TimerSet.new
+      @global_timer_set.value
     end
 
     # Global thread pool optimized for short *tasks*.
@@ -85,8 +96,8 @@ def global_timer_set
     #
     # @raise [ConfigurationError] if this thread pool has already been set
     def global_task_pool=(executor)
-      raise ConfigurationError.new('global task pool was already set') unless @global_task_pool.nil?
-      @global_task_pool = executor
+      @global_task_pool.reconfigure { executor } or
+          raise ConfigurationError.new('global task pool was already set')
     end
 
     # Global thread pool optimized for long *operations*.
@@ -104,8 +115,8 @@ def global_task_pool=(executor)
     #
     # @raise [ConfigurationError] if this thread pool has already been set
     def global_operation_pool=(executor)
-      raise ConfigurationError.new('global operation pool was already set') unless @global_operation_pool.nil?
-      @global_operation_pool = executor
+      @global_operation_pool.reconfigure { executor } or
+          raise ConfigurationError.new('global operation pool was already set')
     end
   end