Break Synchronization::Object into Object and LockableObject

* JRuby broken for now

Author: pitr-ch

lib/concurrent/actor/context.rb

@@ -15,7 +15,7 @@ module Actor
     #     > {include:Actor::RestartingContext}
     #
     # Example of ac actor definition:
-    # 
+    #
     # {include:file:doc/actor/define.out.rb}
     #
     # See methods of {AbstractContext} what else can be tweaked, e.g {AbstractContext#default_reference_class}

lib/concurrent/actor/core.rb

@@ -10,7 +10,7 @@ module Actor
     # @note Whole class should be considered private. An user should use {Context}s and {Reference}s only.
     # @note devel: core should not block on anything, e.g. it cannot wait on children to terminate
     #   that would eat up all threads in task pool and deadlock
-    class Core < Synchronization::Object
+    class Core < Synchronization::LockableObject
       include TypeCheck
       include Concern::Logging
 

lib/concurrent/agent.rb

@@ -19,13 +19,13 @@ module Concurrent
   # and no blocking receive. The state of an Agent should be itself immutable
   # and the `#value` of an Agent is always immediately available for reading by
   # any thread without any messages, i.e. observation does not require
-  # cooperation or coordination. 
+  # cooperation or coordination.
   #
   # Agent action dispatches are made using the various `#send` methods. These
   # methods always return immediately. At some point later, in another thread,
   # the following will happen:
   #
-  # 1. The given `action` will be applied to the state of the Agent and the 
+  # 1. The given `action` will be applied to the state of the Agent and the
   #    `args`, if any were supplied.
   # 2. The return value of `action` will be passed to the validator lambda,
   #    if one has been set on the Agent.
@@ -55,7 +55,7 @@ module Concurrent
   # Unlike in Clojure, `Agent` cannot participate in `Concurrent::TVar` transactions.
   #
   # ## Example
-  # 
+  #
   # ```
   # def next_fibonacci(set = nil)
   #   return [0, 1] if set.nil?
@@ -130,7 +130,7 @@ module Concurrent
   # ```
   #
   # @!macro [new] agent_await_warning
-  # 
+  #
   #   **NOTE** Never, *under any circumstances*, call any of the "await" methods
   #   ({#await}, {#await_for}, {#await_for!}, and {#wait}) from within an action
   #   block/proc/lambda. The call will block the Agent and will always fail.
@@ -141,7 +141,7 @@ module Concurrent
   #
   # @see http://clojure.org/Agents Clojure Agents
   # @see http://clojure.org/state Values and Change - Clojure's approach to Identity and State
-  class Agent < Synchronization::Object
+  class Agent < Synchronization::LockableObject
     include Concern::Observable
 
     ERROR_MODES = [:continue, :fail].freeze
@@ -150,13 +150,13 @@ class Agent < Synchronization::Object
     AWAIT_FLAG = Object.new
     private_constant :AWAIT_FLAG
 
-    AWAIT_ACTION = ->(value, latch){ latch.count_down; AWAIT_FLAG }
+    AWAIT_ACTION = ->(value, latch) { latch.count_down; AWAIT_FLAG }
     private_constant :AWAIT_ACTION
 
-    DEFAULT_ERROR_HANDLER = ->(agent, error){ nil }
+    DEFAULT_ERROR_HANDLER = ->(agent, error) { nil }
     private_constant :DEFAULT_ERROR_HANDLER
 
-    DEFAULT_VALIDATOR = ->(value){ true }
+    DEFAULT_VALIDATOR = ->(value) { true }
     private_constant :DEFAULT_VALIDATOR
 
     Job = Struct.new(:action, :args, :executor, :caller)
@@ -226,6 +226,7 @@ def initialize(initial, opts = {})
     def value
       @current.value
     end
+
     alias_method :deref, :value
 
     # When {#failed?} and {#error_mode} is `:fail`, returns the error object
@@ -236,6 +237,7 @@ def value
     def error
       @error.value
     end
+
     alias_method :reason, :error
 
     # @!macro [attach] agent_send
@@ -289,6 +291,7 @@ def send!(*args, &action)
     def send_off(*args, &action)
       enqueue_action_job(action, args, Concurrent.global_io_executor)
     end
+
     alias_method :post, :send_off
 
     # @!macro agent_send
@@ -396,6 +399,7 @@ def wait(timeout = nil)
     def failed?
       !@error.value.nil?
     end
+
     alias_method :stopped?, :failed?
 
     # When an Agent is {#failed?}, changes the Agent {#value} to `new_value`
@@ -420,7 +424,7 @@ def restart(new_value, opts = {})
         raise Error.new('agent is not failed') unless failed?
         raise ValidationError unless ns_validate(new_value)
         @current.value = new_value
-        @error.value = nil
+        @error.value   = nil
         @queue.clear if clear_actions
         ns_post_next_job unless @queue.empty?
       end
@@ -440,7 +444,7 @@ class << self
       #
       # @!macro agent_await_warning
       def await(*agents)
-        agents.each {|agent| agent.await }
+        agents.each { |agent| agent.await }
         true
       end
 
@@ -455,11 +459,11 @@ def await(*agents)
       # @!macro agent_await_warning
       def await_for(timeout, *agents)
         end_at = Concurrent.monotonic_time + timeout.to_f
-        ok = agents.length.times do |i|
+        ok     = agents.length.times do |i|
           break false if (delay = end_at - Concurrent.monotonic_time) < 0
           break false unless agents[i].await_for(delay)
         end
-        !! ok
+        !!ok
       end
 
       # Blocks the current thread until all actions dispatched thus far to all
@@ -481,7 +485,7 @@ def await_for!(timeout, *agents)
     private
 
     def ns_initialize(initial, opts)
-      @error_mode = opts[:error_mode]
+      @error_mode    = opts[:error_mode]
       @error_handler = opts[:error_handler]
 
       if @error_mode && !ERROR_MODES.include?(@error_mode)
@@ -491,11 +495,11 @@ def ns_initialize(initial, opts)
       end
 
       @error_handler ||= DEFAULT_ERROR_HANDLER
-      @validator       = opts.fetch(:validator, DEFAULT_VALIDATOR)
-      @current         = Concurrent::AtomicReference.new(initial)
-      @error           = Concurrent::AtomicReference.new(nil)
-      @caller          = Concurrent::ThreadLocalVar.new(nil)
-      @queue           = []
+      @validator     = opts.fetch(:validator, DEFAULT_VALIDATOR)
+      @current       = Concurrent::AtomicReference.new(initial)
+      @error         = Concurrent::AtomicReference.new(nil)
+      @caller        = Concurrent::ThreadLocalVar.new(nil)
+      @queue         = []
 
       self.observers = Collection::CopyOnNotifyObserverSet.new
     end
@@ -530,15 +534,15 @@ def ns_enqueue_job(job, index = nil)
     end
 
     def ns_post_next_job
-      @queue.first.executor.post{ execute_next_job }
+      @queue.first.executor.post { execute_next_job }
     end
 
     def execute_next_job
-      job = synchronize { @queue.first }
+      job       = synchronize { @queue.first }
       old_value = @current.value
 
       @caller.value = job.caller # for nested actions
-      new_value = job.action.call(old_value, *job.args)
+      new_value     = job.action.call(old_value, *job.args)
       @caller.value = nil
 
       if new_value != AWAIT_FLAG && ns_validate(new_value)
@@ -573,7 +577,7 @@ def handle_error(error)
     end
 
     def ns_find_last_job_for_thread
-      @queue.rindex {|job| job.caller == Thread.current.object_id }
+      @queue.rindex { |job| job.caller == Thread.current.object_id }
     end
   end
 end

lib/concurrent/atom.rb

@@ -20,7 +20,7 @@ module Concurrent
   # value validates.
   #
   # ## Example
-  # 
+  #
   # ```
   # def next_fibonacci(set = nil)
   #   return [0, 1] if set.nil?
@@ -57,7 +57,7 @@ module Concurrent
   #
   # @see http://clojure.org/atoms Clojure Atoms
   # @see http://clojure.org/state Values and Change - Clojure's approach to Identity and State
-  class Atom < Synchronization::Object
+  class Atom < Synchronization::LockableObject
     include Concern::Observable
 
     # Create a new atom with the given initial value.
@@ -70,7 +70,7 @@ class Atom < Synchronization::Object
     #   is acceptable else return false (preferrably) or raise an exception.
     #
     # @!macro deref_options
-    # 
+    #
     # @raise [ArgumentError] if the validator is not a `Proc` (when given)
     def initialize(value, opts = {})
       super()
@@ -102,7 +102,7 @@ def value
     # the application of the supplied block to a current value, atomically.
     # However, because the block might be called multiple times, it must be free
     # of side effects.
-    # 
+    #
     # @note The given block may be called multiple times, and thus should be free
     #   of side effects.
     #

lib/concurrent/atomic/cyclic_barrier.rb

@@ -4,7 +4,7 @@ module Concurrent
 
   # A synchronization aid that allows a set of threads to all wait for each
   # other to reach a common barrier point.
-  class CyclicBarrier < Synchronization::Object
+  class CyclicBarrier < Synchronization::LockableObject
 
     # @!visibility private
     Generation = Struct.new(:status)

lib/concurrent/atomic/event.rb

@@ -13,7 +13,7 @@ module Concurrent
   # `#reset` at any time once it has been set.
   #
   # @see http://msdn.microsoft.com/en-us/library/windows/desktop/ms682655.aspx
-  class Event < Synchronization::Object
+  class Event < Synchronization::LockableObject
 
     # Creates a new `Event` in the unset state. Threads calling `#wait` on the
     # `Event` will block.

lib/concurrent/atomic/mutex_atomic_boolean.rb

@@ -5,7 +5,7 @@ module Concurrent
   # @!macro atomic_boolean
   # @!visibility private
   # @!macro internal_implementation_note
-  class MutexAtomicBoolean < Synchronization::Object
+  class MutexAtomicBoolean < Synchronization::LockableObject
 
     # @!macro atomic_boolean_method_initialize
     def initialize(initial = false)

lib/concurrent/atomic/mutex_atomic_fixnum.rb

@@ -5,7 +5,7 @@ module Concurrent
   # @!macro atomic_fixnum
   # @!visibility private
   # @!macro internal_implementation_note
-  class MutexAtomicFixnum < Synchronization::Object
+  class MutexAtomicFixnum < Synchronization::LockableObject
 
     # http://stackoverflow.com/questions/535721/ruby-max-integer
     MIN_VALUE = -(2**(0.size * 8 - 2))

lib/concurrent/atomic/mutex_count_down_latch.rb

@@ -5,7 +5,7 @@ module Concurrent
   # @!macro count_down_latch
   # @!visibility private
   # @!macro internal_implementation_note
-  class MutexCountDownLatch < Synchronization::Object
+  class MutexCountDownLatch < Synchronization::LockableObject
 
     # @!macro count_down_latch_method_initialize
     def initialize(count = 1)

lib/concurrent/atomic/mutex_semaphore.rb

@@ -5,7 +5,7 @@ module Concurrent
   # @!macro semaphore
   # @!visibility private
   # @!macro internal_implementation_note
-  class MutexSemaphore < Synchronization::Object
+  class MutexSemaphore < Synchronization::LockableObject
 
     # @!macro semaphore_method_initialize
     def initialize(count)
@@ -78,7 +78,7 @@ def release(permits = 1)
     # @raise [ArgumentError] if `@free` - `@reduction` is less than zero
     #
     # @return [nil]
-    # 
+    #
     # @!visibility private
     def reduce_permits(reduction)
       unless reduction.is_a?(Fixnum) && reduction >= 0