SimpleActorRef now uses SingleThreadExecutor.

Author: jdantonio

lib/concurrent/actor/simple_actor_ref.rb

@@ -1,7 +1,6 @@
-require 'thread'
-
 require 'concurrent/actor/actor_ref'
 require 'concurrent/atomic/event'
+require 'concurrent/executor/single_thread_executor'
 require 'concurrent/ivar'
 
 module Concurrent
@@ -12,15 +11,14 @@ class SimpleActorRef
     def initialize(actor, opts = {})
       @actor = actor
       @mutex = Mutex.new
-      @queue = Queue.new
-      @thread = nil
+      @executor = SingleThreadExecutor.new
       @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
       self.observers = CopyOnNotifyObserverSet.new
 
       @actor.define_singleton_method(:shutdown, &method(:set_stop_event))
+      @actor.on_start
     end
 
     def running?
@@ -33,11 +31,8 @@ def shutdown?
 
     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))
+      @executor.post(Message.new(msg, ivar, block), &method(:process_message))
       ivar
     end
 
@@ -56,10 +51,8 @@ def post!(timeout, *msg)
     def shutdown
       @mutex.synchronize do
         return if shutdown?
-        if @thread && @thread.alive?
-          @thread.kill 
-          @actor.on_shutdown
-        end
+        @executor.shutdown 
+        @actor.on_shutdown
         @stop_event.set
       end
     end
@@ -76,49 +69,26 @@ def set_stop_event
       @stop_event.set
     end
 
-    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 process_message(message)
+      result = ex = nil
 
-    def run_message_loop
-      loop do
-        message = @queue.pop
-        result = ex = nil
+      begin
+        result = @actor.receive(*message.payload)
+      rescue @exception_class => ex
+        @actor.on_error(Time.now, message.payload, ex)
+        @actor.on_reset if @reset_on_error
+      ensure
+        now = Time.now
+        message.ivar.complete(ex.nil?, result, ex)
 
         begin
-          result = @actor.receive(*message.payload)
+          message.callback.call(now, result, ex) if message.callback
         rescue @exception_class => ex
-          @actor.on_error(Time.now, message.payload, ex)
-          @actor.on_reset if @reset_on_error
-        ensure
-          now = Time.now
-          message.ivar.complete(ex.nil?, result, ex)
-
-          begin
-            message.callback.call(now, result, ex) if message.callback
-          rescue @exception_class => ex
-            # suppress
-          end
-
-          observers.notify_observers(now, message.payload, result, ex)
+          # suppress
         end
 
-        break if @stop_event.set?
+        observers.notify_observers(now, message.payload, result, ex)
       end
-      @actor.on_shutdown
     end
   end
 end

spec/concurrent/actor/actor_ref_shared.rb

@@ -99,8 +99,9 @@ def receive(*msg)
     end
 
     it 'returns self' do
-      pending('intermittently failing on Travis CI -- SimpleActorRef')
-      (subject << [1,2,3,4]).should eq subject
+      expected = subject
+      result = subject << [1,2,3,4]
+      result.should eq expected
     end
   end
 

spec/concurrent/actor/simple_actor_ref_spec.rb

@@ -36,91 +36,11 @@ module Concurrent
         described_class.should_receive(:new).once.with(anything, opts)
         shared_actor_test_class.spawn(opts)
       end
-    end
-
-    context 'supervision' do
-
-      it 'does not start a new thread on construction' do
-        Thread.should_not_receive(:new).with(any_args)
-        subject = shared_actor_test_class.spawn
-      end
-
-      it 'starts a new thread on the first post' do
-        thread = Thread.new{ nil }
-        Thread.should_receive(:new).once.with(no_args).and_return(thread)
-        subject << :foo
-      end
-
-      it 'does not start a new thread after the first post' do
-        subject << :foo
-        sleep(0.1)
-        expected = Thread.list.length
-        5.times{ subject << :foo }
-        Thread.list.length.should eq expected
-      end
-
-      it 'starts a new thread when the prior thread has died' do
-        subject << :foo
-        sleep(0.1)
-
-        subject << :terminate
-        sleep(0.1)
-
-        thread = Thread.new{ nil }
-        Thread.should_receive(:new).once.with(no_args).and_return(thread)
-        subject << :foo
-      end
-
-      it 'does not reset the thread after shutdown' do
-        thread = Thread.new{ nil }
-        Thread.should_receive(:new).once.with(no_args).and_return(thread)
-        subject << :foo
-        sleep(0.1)
-
-        subject.shutdown
-        sleep(0.1)
-
-        subject << :foo
-      end
 
-      it 'calls #on_start when the thread is first started' do
-        actor = subject.instance_variable_get(:@actor)
+      it 'calls #on_start on the actor' do
+        actor = double(:shared_actor_test_class)
         actor.should_receive(:on_start).once.with(no_args)
-        subject << :foo
-      end
-
-      it 'calls #on_reset when the thread is started after the first time' do
-        actor = subject.instance_variable_get(:@actor)
-        actor.should_receive(:on_reset).once.with(no_args)
-        subject << :terminate
-        sleep(0.1)
-        subject << :foo
-      end
-    end
-
-    context 'abort_on_exception' do
-
-      after(:each) do
-        @ref.shutdown if @ref
-      end
-
-      it 'gets set on the actor thread' do
-        @ref = shared_actor_test_class.spawn(abort_on_exception: true)
-        @ref << :foo
-        sleep(0.1)
-        @ref.instance_variable_get(:@thread).abort_on_exception.should be_true
-
-        @ref = shared_actor_test_class.spawn(abort_on_exception: false)
-        @ref << :foo
-        sleep(0.1)
-        @ref.instance_variable_get(:@thread).abort_on_exception.should be_false
-      end
-
-      it 'defaults to true' do
-        @ref = shared_actor_test_class.spawn
-        @ref << :foo
-        sleep(0.1)
-        @ref.instance_variable_get(:@thread).abort_on_exception.should be_true
+        SimpleActorRef.new(actor)
       end
     end