Merge branch 'master' into channels

mighe · mighe · commit e78e7d7b14ab · 2014-04-07T20:06:15.000+02:00
Conflicts:
	lib/concurrent.rb
diff --git a/README.md b/README.md
@@ -56,7 +56,7 @@ into several general groups:
 * Thread-safe variables including [M-Structures](https://github.com/jdantonio/concurrent-ruby/wiki/MVar-(M-Structure)),
   [I-Structures](https://github.com/jdantonio/concurrent-ruby/wiki/IVar-(I-Structure)),
   [thread-local variables](https://github.com/jdantonio/concurrent-ruby/wiki/ThreadLocalVar),
-  and atomic counters
+  atomic counters, and [software transactional memory](https://github.com/jdantonio/concurrent-ruby/wiki/TVar-(STM))
 * Thread synchronization classes and algorithms including [dataflow](https://github.com/jdantonio/concurrent-ruby/wiki/Dataflow), 
   timeout, condition, countdown latch, dependency counter, and event
 * Java-inspired [thread pools](https://github.com/jdantonio/concurrent-ruby/wiki/Thread%20Pools)
diff --git a/lib/concurrent.rb b/lib/concurrent.rb
@@ -39,6 +39,9 @@
 require 'concurrent/channel/ring_buffer'
 require 'concurrent/channel/blocking_ring_buffer'
 
+require 'concurrent/actor_context'
+require 'concurrent/simple_actor_ref'
+
 require 'concurrent/cached_thread_pool'
 require 'concurrent/fixed_thread_pool'
 require 'concurrent/immediate_executor'
diff --git a/lib/concurrent/actor.rb b/lib/concurrent/actor.rb
@@ -179,6 +179,7 @@ def initialize(queue)
     #
     # @deprecated +Actor+ is being replaced with a completely new framework prior to v1.0.0
     def self.pool(count, *args, &block)
+      warn '[DEPRECATED] `Actor` is deprecated and will be replaced with `ActorContext`.'
       raise ArgumentError.new('count must be greater than zero') unless count > 0
       mailbox = Queue.new
       actors = count.times.collect do
@@ -211,13 +212,15 @@ def self.pool(count, *args, &block)
     # 
     # @!visibility public
     def act(*message)
+      warn '[DEPRECATED] `Actor` is deprecated and will be replaced with `ActorContext`.'
       raise NotImplementedError.new("#{self.class} does not implement #act")
     end
 
     # @!visibility private
     #
     # @deprecated +Actor+ is being replaced with a completely new framework prior to v1.0.0
     def on_run # :nodoc:
+      warn '[DEPRECATED] `Actor` is deprecated and will be replaced with `ActorContext`.'
       queue.clear
     end
 
diff --git a/lib/concurrent/actor_context.rb b/lib/concurrent/actor_context.rb
@@ -0,0 +1,28 @@
+require 'concurrent/simple_actor_ref'
+
+module Concurrent
+
+  module ActorContext
+
+    def on_start
+    end
+
+    def on_reset
+    end
+
+    def on_shutdown
+    end
+
+    def self.included(base)
+
+      class << base
+        protected :new
+
+        def spawn(opts = {})
+          args = opts.fetch(:args, [])
+          Concurrent::SimpleActorRef.new(self.new(*args), opts)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/concurrent/actor_ref.rb b/lib/concurrent/actor_ref.rb
@@ -0,0 +1,39 @@
+require 'concurrent/copy_on_notify_observer_set'
+
+module Concurrent
+
+  module ActorRef
+
+    #NOTE: Required API methods
+    #      Must be implemented in all subclasses
+    #def post(*msg, &block)
+    #def post!(*msg)
+    #def running?
+    #def shutdown?
+    #def shutdown
+    #def join(timeout = nil)
+
+    def <<(message)
+      post(*message)
+      self
+    end
+
+    def add_observer(*args)
+      @observers.add_observer(*args)
+    end
+
+    def delete_observer(*args)
+      @observers.delete_observer(*args)
+    end
+
+    def delete_observers
+      @observers.delete_observers
+    end
+
+    protected
+
+    def observers
+      @observers ||= CopyOnNotifyObserverSet.new
+    end
+  end
+end
diff --git a/lib/concurrent/atomic.rb b/lib/concurrent/atomic.rb
@@ -109,7 +109,7 @@ def initialize(init = 0)
       allocate_storage(init)
     end
 
-    if defined? java.util
+    if RUBY_PLATFORM == 'java'
       include JavaAtomicFixnum
     else
       include MutexAtomicFixnum
diff --git a/lib/concurrent/cached_thread_pool.rb b/lib/concurrent/cached_thread_pool.rb
@@ -2,35 +2,35 @@
 
 module Concurrent
 
-  if defined? java.util
+  if RUBY_PLATFORM == 'java'
     require 'concurrent/java_cached_thread_pool'
     # @!macro [attach] cached_thread_pool
     #   A thread pool that dynamically grows and shrinks to fit the current workload.
     #   New threads are created as needed, existing threads are reused, and threads
     #   that remain idle for too long are killed and removed from the pool. These
     #   pools are particularly suited to applications that perform a high volume of
     #   short-lived tasks.
-    #  
+    #
     #   On creation a +CachedThreadPool+ has zero running threads. New threads are
     #   created on the pool as new operations are +#post+. The size of the pool
     #   will grow until +#max_length+ threads are in the pool or until the number
     #   of threads exceeds the number of running and pending operations. When a new
     #   operation is post to the pool the first available idle thread will be tasked
     #   with the new operation.
-    #  
+    #
     #   Should a thread crash for any reason the thread will immediately be removed
     #   from the pool. Similarly, threads which remain idle for an extended period
     #   of time will be killed and reclaimed. Thus these thread pools are very
     #   efficient at reclaiming unused resources.
-    #  
+    #
     #   The API and behavior of this class are based on Java's +CachedThreadPool+
     #
     #   @note When running on the JVM (JRuby) this class will inherit from +JavaCachedThreadPool+.
     #     On all other platforms it will inherit from +RubyCachedThreadPool+.
     #
     #   @see Concurrent::RubyCachedThreadPool
     #   @see Concurrent::JavaCachedThreadPool
-    #  
+    #
     #   @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
     #   @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
     #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
diff --git a/lib/concurrent/fixed_thread_pool.rb b/lib/concurrent/fixed_thread_pool.rb
@@ -2,7 +2,7 @@
 
 module Concurrent
 
-  if defined? java.util
+  if RUBY_PLATFORM == 'java'
     require 'concurrent/java_fixed_thread_pool'
     # @!macro [attach] fixed_thread_pool
     #
@@ -11,15 +11,15 @@ module Concurrent
     #   tasks +#post+ to the thread pool are enqueued until a thread becomes available.
     #   Should a thread crash for any reason the thread will immediately be removed
     #   from the pool and replaced.
-    #  
+    #
     #   The API and behavior of this class are based on Java's +FixedThreadPool+
     #
     #   @note When running on the JVM (JRuby) this class will inherit from +JavaFixedThreadPool+.
     #     On all other platforms it will inherit from +RubyFixedThreadPool+.
     #
     #   @see Concurrent::RubyFixedThreadPool
     #   @see Concurrent::JavaFixedThreadPool
-    #  
+    #
     #   @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
     #   @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
     #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
diff --git a/lib/concurrent/java_cached_thread_pool.rb b/lib/concurrent/java_cached_thread_pool.rb
@@ -1,4 +1,4 @@
-if defined? java.util
+if RUBY_PLATFORM == 'java'
 
   require 'concurrent/java_thread_pool_executor'
 
diff --git a/lib/concurrent/java_fixed_thread_pool.rb b/lib/concurrent/java_fixed_thread_pool.rb
@@ -1,4 +1,4 @@
-if defined? java.util
+if RUBY_PLATFORM == 'java'
 
   require 'concurrent/java_thread_pool_executor'
 
diff --git a/lib/concurrent/java_thread_pool_executor.rb b/lib/concurrent/java_thread_pool_executor.rb
@@ -1,4 +1,4 @@
-if defined? java.util
+if RUBY_PLATFORM == 'java'
 
   module Concurrent
 
@@ -102,6 +102,15 @@ def remaining_capacity
         @max_queue == 0 ? -1 : @executor.getQueue.remainingCapacity
       end
 
+      # This method is deprecated and will be removed soon.
+      # This method is supost to return the threads status, but Java API doesn't
+      # provide a way to get the thread status. So we return an empty Array instead.
+      def status
+        warn '[DEPRECATED] `status` is deprecated and will be removed soon.'
+        warn "Calls to `status` return an empty Array. Java ThreadPoolExecutor does not provide thread's status."
+        []
+      end
+
       # Is the thread pool running?
       #
       # @return [Boolean] +true+ when running, +false+ when shutting down or shutdown
diff --git a/lib/concurrent/processor_count.rb b/lib/concurrent/processor_count.rb
@@ -34,7 +34,7 @@ module Concurrent
   # @see http://docs.oracle.com/javase/6/docs/api/java/lang/Runtime.html#availableProcessors()
   # @see http://msdn.microsoft.com/en-us/library/aa394373(v=vs.85).aspx
   def processor_count
-    if defined? java.lang
+    if RUBY_PLATFORM == 'java'
       java.lang.Runtime.getRuntime.availableProcessors
     else
       @@processor_count ||= begin
diff --git a/lib/concurrent/ruby_thread_pool_executor.rb b/lib/concurrent/ruby_thread_pool_executor.rb
@@ -90,6 +90,14 @@ def running?
       @mutex.synchronize { @state == :running }
     end
 
+    # Returns an array with the status of each thread in the pool
+    #
+    # This method is deprecated and will be removed soon.
+    def status
+      warn '[DEPRECATED] `status` is deprecated and will be removed soon.'
+      @mutex.synchronize { @pool.collect { |worker| worker.status } }
+    end
+    
     # Is the thread pool shutdown?
     #
     # @return [Boolean] +true+ when shutdown, +false+ when shutting down or running
diff --git a/lib/concurrent/ruby_thread_pool_worker.rb b/lib/concurrent/ruby_thread_pool_worker.rb
@@ -25,6 +25,13 @@ def last_activity # :nodoc:
       @mutex.synchronize { @last_activity }
     end
 
+    def status
+      @mutex.synchronize do
+        return 'not running' if @thread.nil?
+        @thread.status
+      end
+    end
+
     # @!visibility private
     def kill # :nodoc:
       @mutex.synchronize do
diff --git a/lib/concurrent/simple_actor_ref.rb b/lib/concurrent/simple_actor_ref.rb
@@ -0,0 +1,108 @@
+require 'thread'
+
+require 'concurrent/actor_ref'
+require 'concurrent/event'
+require 'concurrent/ivar'
+
+module Concurrent
+
+  class SimpleActorRef
+    include ActorRef
+
+    def initialize(actor, opts = {})
+      @actor = actor
+      @mutex = Mutex.new
+      @queue = Queue.new
+      @thread = nil
+      @stop_event = Event.new
+      @abort_on_exception = opts.fetch(:abort_on_exception, true)
+      @reset_on_error = opts.fetch(:reset_on_error, true)
+      @exception_class = opts.fetch(:rescue_exception, false) ? Exception : StandardError
+      @observers = CopyOnNotifyObserverSet.new
+    end
+
+    def running?
+      ! @stop_event.set?
+    end
+
+    def shutdown?
+      @stop_event.set?
+    end
+
+    def post(*msg, &block)
+      raise ArgumentError.new('message cannot be empty') if msg.empty?
+      @mutex.synchronize do
+        supervise unless shutdown?
+      end
+      ivar = IVar.new
+      @queue.push(Message.new(msg, ivar, block))
+      ivar
+    end
+
+    def post!(seconds, *msg)
+      raise Concurrent::TimeoutError if seconds == 0
+      ivar = self.post(*msg)
+      ivar.value(seconds)
+      if ivar.incomplete?
+        raise Concurrent::TimeoutError
+      elsif ivar.reason
+        raise ivar.reason
+      end
+      ivar.value
+    end
+
+    def shutdown
+      @mutex.synchronize do
+        return if shutdown?
+        if @thread && @thread.alive?
+          @thread.kill 
+          @actor.on_shutdown
+        end
+        @stop_event.set
+      end
+    end
+
+    def join(timeout = nil)
+      @stop_event.wait(timeout)
+    end
+
+    private
+
+    Message = Struct.new(:payload, :ivar, :callback)
+
+    def supervise
+      if @thread.nil?
+        @actor.on_start
+        @thread = new_worker_thread
+      elsif ! @thread.alive?
+        @actor.on_reset
+        @thread = new_worker_thread
+      end
+    end
+
+    def new_worker_thread
+      Thread.new do
+        Thread.current.abort_on_exception = @abort_on_exception
+        run_message_loop
+      end
+    end
+
+    def run_message_loop
+      loop do
+        message = @queue.pop
+        result = ex = nil
+
+        begin
+          result = @actor.receive(*message.payload)
+        rescue @exception_class => ex
+          @actor.on_reset if @reset_on_error
+        ensure
+          now = Time.now
+          message.ivar.complete(ex.nil?, result, ex)
+          message.callback.call(now, result, ex) if message.callback
+          observers.notify_observers(now, message.payload, result, ex)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/concurrent/thread_local_var.rb b/lib/concurrent/thread_local_var.rb
@@ -75,7 +75,7 @@ class ThreadLocalVar
 
     NIL_SENTINEL = Object.new
 
-    if defined? java.lang
+    if RUBY_PLATFORM == 'java'
       include ThreadLocalJavaStorage
     elsif Thread.current.respond_to?(:thread_variable_set)
       include ThreadLocalNewStorage
diff --git a/lib/concurrent/thread_pool_executor.rb b/lib/concurrent/thread_pool_executor.rb
diff --git a/spec/concurrent/actor_context_spec.rb b/spec/concurrent/actor_context_spec.rb
diff --git a/spec/concurrent/actor_ref_shared.rb b/spec/concurrent/actor_ref_shared.rb
diff --git a/spec/concurrent/cached_thread_pool_shared.rb b/spec/concurrent/cached_thread_pool_shared.rb
diff --git a/spec/concurrent/fixed_thread_pool_shared.rb b/spec/concurrent/fixed_thread_pool_shared.rb
diff --git a/spec/concurrent/simple_actor_ref_spec.rb b/spec/concurrent/simple_actor_ref_spec.rb
diff --git a/spec/concurrent/thread_pool_shared.rb b/spec/concurrent/thread_pool_shared.rb

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-if defined? java.util`
	`1`	`+if RUBY_PLATFORM == 'java'`
`2`	`2`
`3`	`3`	`require 'concurrent/java_thread_pool_executor'`
`4`	`4`