Merge pull request #397 from ruby-concurrency/thread_safe

jdantonio · jdantonio · commit 4ffdb7e9e75e · 2015-08-19T11:13:26.000-04:00
More thread_safe cleanups
diff --git a/ext/com/concurrent_ruby/ext/JRubyMapBackendLibrary.java b/ext/com/concurrent_ruby/ext/JRubyMapBackendLibrary.java
@@ -26,7 +26,7 @@ public class JRubyMapBackendLibrary implements Library {
   public void load(Ruby runtime, boolean wrap) throws IOException {
 
     RubyModule concurrentMod = runtime.defineModule("Concurrent");
-    RubyModule thread_safeMod = concurrentMod.defineModuleUnder("ThreadSafe");
+    RubyModule thread_safeMod = concurrentMod.defineModuleUnder("Collection");
     RubyClass jrubyRefClass = thread_safeMod.defineClassUnder("JRubyMapBackend", runtime.getObject(), BACKEND_ALLOCATOR);
     jrubyRefClass.setAllocator(BACKEND_ALLOCATOR);
     jrubyRefClass.defineAnnotatedMethods(JRubyMapBackend.class);
diff --git a/lib/concurrent.rb b/lib/concurrent.rb
@@ -11,25 +11,29 @@
 require 'concurrent/atomic/atomic_reference'
 require 'concurrent/atom'
 require 'concurrent/array'
+require 'concurrent/hash'
+require 'concurrent/map'
+require 'concurrent/tuple'
 require 'concurrent/async'
 require 'concurrent/dataflow'
 require 'concurrent/delay'
 require 'concurrent/exchanger'
 require 'concurrent/future'
-require 'concurrent/hash'
 require 'concurrent/immutable_struct'
 require 'concurrent/ivar'
-require 'concurrent/map'
 require 'concurrent/maybe'
 require 'concurrent/mutable_struct'
 require 'concurrent/mvar'
 require 'concurrent/promise'
 require 'concurrent/scheduled_task'
 require 'concurrent/settable_struct'
 require 'concurrent/timer_task'
-require 'concurrent/tuple'
 require 'concurrent/tvar'
 
+require 'concurrent/thread_safe/synchronized_delegator'
+require 'concurrent/thread_safe/util'
+
+
 # @!macro [new] internal_implementation_note
 #
 #   @note **Private Implementation:** This abstraction is a private, internal
@@ -122,4 +126,9 @@
 # * Be small, lean, and loosely coupled
 module Concurrent
 
+  # Various classes within allows for +nil+ values to be stored,
+  # so a special +NULL+ token is required to indicate the "nil-ness".
+  # @!visibility private
+  NULL = Object.new
+
 end
diff --git a/lib/concurrent/array.rb b/lib/concurrent/array.rb
@@ -1 +1,38 @@
-require 'concurrent/thread_safe'
+require 'concurrent/utility/engine'
+require 'concurrent/thread_safe/util'
+
+module Concurrent
+  if Concurrent.on_cruby?
+
+    # Because MRI never runs code in parallel, the existing
+    # non-thread-safe structures should usually work fine.
+
+    # @!macro [attach] concurrent_array
+    #
+    #   A thread-safe subclass of Array. This version locks against the object
+    #   itself for every method call, ensuring only one thread can be reading
+    #   or writing at a time. This includes iteration methods like `#each`.
+    #
+    #   @see http://ruby-doc.org/core-2.2.0/Array.html Ruby standard library `Array`
+    class Array < ::Array;
+    end
+
+  elsif Concurrent.on_jruby?
+    require 'jruby/synchronized'
+
+    # @!macro concurrent_array
+    class Array < ::Array
+      include JRuby::Synchronized
+    end
+
+  elsif Concurrent.on_rbx?
+    require 'monitor'
+
+    # @!macro concurrent_array
+    class Array < ::Array
+    end
+
+    ThreadSafe::Util.make_synchronized_on_rbx Array
+  end
+end
+
diff --git a/lib/concurrent/collection/map/atomic_reference_map_backend.rb b/lib/concurrent/collection/map/atomic_reference_map_backend.rb
@@ -3,7 +3,7 @@
 module Concurrent
 
   # @!visibility private
-  module ThreadSafe
+  module Collection
 
     # A Ruby port of the Doug Lea's jsr166e.ConcurrentHashMapV8 class version 1.59
     # available in public domain.
@@ -186,7 +186,7 @@ module ThreadSafe
     #
     # @!visibility private
     class AtomicReferenceMapBackend
-      
+
       # @!visibility private
       class Table < Concurrent::ThreadSafe::Util::PowerOfTwoTuple
         def cas_new_node(i, hash, key, value)
diff --git a/lib/concurrent/collection/map/mri_map_backend.rb b/lib/concurrent/collection/map/mri_map_backend.rb
@@ -1,10 +1,10 @@
 require 'thread'
-require 'concurrent/thread_safe/non_concurrent_map_backend'
+require 'concurrent/collection/map/non_concurrent_map_backend'
 
 module Concurrent
 
   # @!visibility private
-  module ThreadSafe
+  module Collection
 
     # @!visibility private
     class MriMapBackend < NonConcurrentMapBackend
diff --git a/lib/concurrent/collection/map/non_concurrent_map_backend.rb b/lib/concurrent/collection/map/non_concurrent_map_backend.rb
@@ -1,7 +1,7 @@
 module Concurrent
 
   # @!visibility private
-  module ThreadSafe
+  module Collection
 
     # @!visibility private
     class NonConcurrentMapBackend
diff --git a/lib/concurrent/collection/map/synchronized_map_backend.rb b/lib/concurrent/collection/map/synchronized_map_backend.rb
@@ -1,9 +1,9 @@
-require 'concurrent/thread_safe/non_concurrent_map_backend'
+require 'concurrent/collection/map/non_concurrent_map_backend'
 
 module Concurrent
 
   # @!visibility private
-  module ThreadSafe
+  module Collection
 
     # @!visibility private
     class SynchronizedMapBackend < NonConcurrentMapBackend
diff --git a/lib/concurrent/future.rb b/lib/concurrent/future.rb
@@ -27,7 +27,7 @@ class Future < IVar
     # @raise [ArgumentError] if no block is given
     def initialize(opts = {}, &block)
       raise ArgumentError.new('no block given') unless block_given?
-      super(IVar::NO_VALUE, opts.merge(__task_from_block__: block), &nil)
+      super(NULL, opts.merge(__task_from_block__: block), &nil)
     end
 
     # Execute an `:unscheduled` `Future`. Immediately sets the state to `:pending` and
@@ -74,7 +74,7 @@ def self.execute(opts = {}, &block)
     end
 
     # @!macro ivar_set_method
-    def set(value = IVar::NO_VALUE, &block)
+    def set(value = NULL, &block)
       check_for_block_or_value!(block_given?, value)
       synchronize do
         if @state != :unscheduled
diff --git a/lib/concurrent/hash.rb b/lib/concurrent/hash.rb
@@ -1 +1,35 @@
-require 'concurrent/thread_safe'
+require 'concurrent/utility/engine'
+require 'concurrent/thread_safe/util'
+
+module Concurrent
+  if Concurrent.on_cruby?
+
+    # @!macro [attach] concurrent_hash
+    #
+    #   A thread-safe subclass of Hash. This version locks against the object
+    #   itself for every method call, ensuring only one thread can be reading
+    #   or writing at a time. This includes iteration methods like `#each`.
+    #
+    #   @see http://ruby-doc.org/core-2.2.0/Hash.html Ruby standard library `Hash`
+    class Hash < ::Hash;
+    end
+
+  elsif Concurrent.on_jruby?
+    require 'jruby/synchronized'
+
+    # @!macro concurrent_hash
+    class Hash < ::Hash
+      include JRuby::Synchronized
+    end
+
+  elsif Concurrent.on_rbx?
+    require 'monitor'
+
+    # @!macro concurrent_hash
+    class Hash < ::Hash
+    end
+
+    ThreadSafe::Util.make_synchronized_on_rbx Hash
+
+  end
+end
diff --git a/lib/concurrent/ivar.rb b/lib/concurrent/ivar.rb
@@ -47,9 +47,6 @@ class IVar < Synchronization::Object
     include Concern::Obligation
     include Concern::Observable
 
-    # @!visibility private
-    NO_VALUE = Object.new # :nodoc:
-
     # Create a new `IVar` in the `:pending` state with the (optional) initial value.
     #
     # @param [Object] value the initial value
@@ -60,8 +57,8 @@ class IVar < Synchronization::Object
     #   returning the data
     # @option opts [String] :copy_on_deref (nil) call the given `Proc` passing
     #   the internal value and returning the value returned from the proc
-    def initialize(value = NO_VALUE, opts = {}, &block)
-      if value != NO_VALUE && block_given?
+    def initialize(value = NULL, opts = {}, &block)
+      if value != NULL && block_given?
         raise ArgumentError.new('provide only a value or a block')
       end
       super(&nil)
@@ -102,16 +99,16 @@ def add_observer(observer = nil, func = :update, &block)
 
     # @!macro [attach] ivar_set_method
     #   Set the `IVar` to a value and wake or notify all threads waiting on it.
-    #  
+    #
     #   @!macro [attach] ivar_set_parameters_and_exceptions
     #     @param [Object] value the value to store in the `IVar`
     #     @yield A block operation to use for setting the value
     #     @raise [ArgumentError] if both a value and a block are given
     #     @raise [Concurrent::MultipleAssignmentError] if the `IVar` has already
     #       been set or otherwise completed
-    #  
+    #
     #   @return [IVar] self
-    def set(value = NO_VALUE)
+    def set(value = NULL)
       check_for_block_or_value!(block_given?, value)
       raise MultipleAssignmentError unless compare_and_set_state(:processing, :pending)
 
@@ -128,7 +125,7 @@ def set(value = NO_VALUE)
 
     # @!macro [attach] ivar_fail_method
     #   Set the `IVar` to failed due to some error and wake or notify all threads waiting on it.
-    #  
+    #
     #   @param [Object] reason for the failure
     #   @raise [Concurrent::MultipleAssignmentError] if the `IVar` has already
     #     been set or otherwise completed
@@ -143,7 +140,7 @@ def fail(reason = StandardError.new)
     # @!macro ivar_set_parameters_and_exceptions
     #
     # @return [Boolean] true if the value was set else false
-    def try_set(value = NO_VALUE, &block)
+    def try_set(value = NULL, &block)
       set(value, &block)
       true
     rescue MultipleAssignmentError
@@ -159,7 +156,7 @@ def ns_initialize(value, opts)
       self.observers = Collection::CopyOnWriteObserverSet.new
       set_deref_options(opts)
 
-      if value == NO_VALUE
+      if value == NULL
         @state = :pending
       else
         ns_complete_without_notification(true, value, nil)
@@ -202,7 +199,7 @@ def ns_complete_without_notification(success, value, reason)
 
     # @!visibility private
     def check_for_block_or_value!(block_given, value) # :nodoc:
-      if (block_given && value != NO_VALUE) || (! block_given && value == NO_VALUE)
+      if (block_given && value != NULL) || (! block_given && value == NULL)
         raise ArgumentError.new('must set with either a value or a block')
       end
     end
diff --git a/lib/concurrent/map.rb b/lib/concurrent/map.rb
@@ -2,22 +2,23 @@
 
 module Concurrent
   # @!visibility private
-  module ThreadSafe
+  module Collection
 
     # @!visibility private
-    MapBackend = if defined?(RUBY_ENGINE)
+    MapImplementation = if defined?(RUBY_ENGINE)
                    case RUBY_ENGINE
                    when 'jruby'
+                     # noinspection RubyResolve
                      JRubyMapBackend
                    when 'ruby'
-                     require 'concurrent/thread_safe/mri_map_backend'
+                     require 'concurrent/collection/map/mri_map_backend'
                      MriMapBackend
                    when 'rbx'
-                     require 'concurrent/thread_safe/atomic_reference_map_backend'
+                     require 'concurrent/collection/map/atomic_reference_map_backend'
                      AtomicReferenceMapBackend
                    else
                      warn 'Concurrent::Map: unsupported Ruby engine, using a fully synchronized Concurrent::Map implementation' if $VERBOSE
-                     require 'concurrent/thread_safe/synchronized_map_backend'
+                     require 'concurrent/collection/map/synchronized_map_backend'
                      SynchronizedMapBackend
                    end
                  else
@@ -36,7 +37,7 @@ module ThreadSafe
   # >
   # > map = Concurrent::Map.new
 
-  class Map < ThreadSafe::MapBackend
+  class Map < Collection::MapImplementation
     def initialize(options = nil, &block)
       if options.kind_of?(::Hash)
         validate_options_hash!(options)
diff --git a/lib/concurrent/promise.rb b/lib/concurrent/promise.rb
@@ -205,7 +205,7 @@ class Promise < IVar
     # @see http://promises-aplus.github.io/promises-spec/
     def initialize(opts = {}, &block)
       opts.delete_if { |k, v| v.nil? }
-      super(IVar::NO_VALUE, opts.merge(__promise_body_from_block__: block), &nil)
+      super(NULL, opts.merge(__promise_body_from_block__: block), &nil)
     end
 
     # Create a new `Promise` and fulfill it immediately.
@@ -254,7 +254,7 @@ def execute
     # @!macro ivar_set_method
     #
     # @raise [Concurrent::PromiseExecutionError] if not the root promise
-    def set(value = IVar::NO_VALUE, &block)
+    def set(value = NULL, &block)
       raise PromiseExecutionError.new('supported only on root promise') unless root?
       check_for_block_or_value!(block_given?, value)
       synchronize do
diff --git a/lib/concurrent/scheduled_task.rb b/lib/concurrent/scheduled_task.rb
@@ -163,7 +163,7 @@ def initialize(delay, opts = {}, &task)
       raise ArgumentError.new('no block given') unless block_given?
       raise ArgumentError.new('seconds must be greater than zero') if delay.to_f < 0.0
 
-      super(IVar::NO_VALUE, opts, &nil)
+      super(NULL, opts, &nil)
 
       synchronize do
         ns_set_state(:unscheduled)
diff --git a/lib/concurrent/thread_safe.rb b/lib/concurrent/thread_safe.rb
diff --git a/lib/concurrent/thread_safe/readme.txt b/lib/concurrent/thread_safe/readme.txt
diff --git a/lib/concurrent/thread_safe/util.rb b/lib/concurrent/thread_safe/util.rb
diff --git a/lib/concurrent/thread_safe/util/array_hash_rbx.rb b/lib/concurrent/thread_safe/util/array_hash_rbx.rb
diff --git a/lib/concurrent/thread_safe/util/power_of_two_tuple.rb b/lib/concurrent/thread_safe/util/power_of_two_tuple.rb
diff --git a/spec/concurrent/ivar_spec.rb b/spec/concurrent/ivar_spec.rb
