Initial implementation of ActorContext, ActorRef, and the #spawn factory.

Author: jdantonio

lib/concurrent.rb

@@ -32,6 +32,9 @@
 require 'concurrent/tvar'
 require 'concurrent/utilities'
 
+require 'concurrent/actor_context'
+require 'concurrent/simple_actor_ref'
+
 require 'concurrent/cached_thread_pool'
 require 'concurrent/fixed_thread_pool'
 require 'concurrent/immediate_executor'

lib/concurrent/actor_context.rb

@@ -0,0 +1,23 @@
+require 'concurrent/simple_actor_ref'
+
+module Concurrent
+
+  module ActorContext
+
+    def on_start
+    end
+
+    def on_restart
+    end
+
+    def on_shutdown
+    end
+
+    def self.included(base)
+
+      def base.spawn
+        Concurrent::SimpleActorRef.new(self.new)
+      end
+    end
+  end
+end

lib/concurrent/actor_ref.rb

@@ -0,0 +1,28 @@
+require 'concurrent/utilities'
+
+module Concurrent
+
+  module ActorRef
+
+    def running?
+      true
+    end
+
+    def shutdown?
+      false
+    end
+
+    def post(*msg, &block)
+      raise NotImplementedError
+    end
+
+    def post!(*msg)
+      raise NotImplementedError
+    end
+
+    def <<(message)
+      post(*message)
+      self
+    end
+  end
+end

lib/concurrent/simple_actor_ref.rb

@@ -0,0 +1,96 @@
+require 'thread'
+
+require 'concurrent/actor_ref'
+require 'concurrent/ivar'
+
+module Concurrent
+
+  class SimpleActorRef
+    include ActorRef
+
+    def initialize(actor)
+      @actor = actor
+      @mutex = Mutex.new
+      @queue = Queue.new
+      @thread = nil
+      @stopped = false
+    end
+
+    def running?
+      @mutex.synchronize{ @stopped == false }
+    end
+
+    def shutdown?
+      @mutex.synchronize{ @stopped == true }
+    end
+
+    def post(*msg, &block)
+      raise ArgumentError.new('message cannot be empty') if msg.empty?
+      @mutex.synchronize do
+        supervise unless @stopped == true
+      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
+        @stopped = true
+        if @thread && @thread.alive?
+          @thread.kill 
+          @actor.on_shutdown
+        end
+      end
+    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_restart
+        @thread = new_worker_thread
+      end
+    end
+
+    def new_worker_thread
+      Thread.new do
+        Thread.current.abort_on_exception = true
+        run_message_loop
+      end
+    end
+
+    def run_message_loop
+      loop do
+        message = @queue.pop
+        result = ex = nil
+
+        begin
+          result = @actor.receive(*message.payload)
+        rescue => ex
+          # suppress
+        ensure
+          message.ivar.complete(ex.nil?, result, ex)
+          message.callback.call(Time.now, result, ex) if message.callback
+        end
+      end
+    end
+  end
+end

spec/concurrent/actor_context_spec.rb

@@ -0,0 +1,31 @@
+require 'spec_helper'
+
+module Concurrent
+
+  describe ActorContext do
+
+    let(:described_class) do
+      Class.new do
+        include ActorContext
+      end
+    end
+
+    context 'callbacks' do
+
+      subject { described_class.new }
+
+      specify { subject.should respond_to :on_start }
+
+      specify { subject.should respond_to :on_restart }
+
+      specify { subject.should respond_to :on_shutdown }
+    end
+
+    context '#spawn' do
+
+      it 'returns an ActorRef' do
+        described_class.spawn.should be_a ActorRef
+      end
+    end
+  end
+end

spec/concurrent/actor_ref_shared.rb

@@ -0,0 +1,162 @@
+require 'spec_helper'
+
+def shared_actor_test_class
+  Class.new do
+    include Concurrent::ActorContext
+    def receive(*msg)
+      case msg.first
+      when :poison
+        raise StandardError
+      when :terminate
+        Thread.current.kill
+      when :sleep
+        sleep(msg.last)
+      when :check
+        msg[1].set(msg.last)
+      else
+        msg.first
+      end
+    end
+  end
+end
+
+share_examples_for :actor_ref do
+
+  it 'includes ActorRef' do
+    subject.should be_a Concurrent::ActorRef
+  end
+
+  context 'running and shutdown' do
+
+    specify { subject.should respond_to :shutdown }
+
+    specify { subject.should be_running }
+
+    specify { subject.should_not be_shutdown }
+
+    specify do
+      subject.shutdown
+      sleep(0.1)
+      subject.should be_shutdown
+    end
+  end
+
+  context '#post' do
+
+    it 'raises an exception when the message is empty' do
+      expect {
+        subject.post
+      }.to raise_error(ArgumentError)
+    end
+
+    it 'returns an IVar' do
+      subject.post(:foo).should be_a Concurrent::IVar
+    end
+
+    it 'fulfills the IVar when message is processed' do
+      ivar = subject.post(:foo)
+      sleep(0.1)
+      ivar.should be_fulfilled
+      ivar.value.should eq :foo
+    end
+
+    it 'rejects the IVar when message processing fails' do
+      ivar = subject.post(:poison)
+      sleep(0.1)
+      ivar.should be_rejected
+      ivar.reason.should be_a StandardError
+    end
+  end
+
+  context '#<<' do
+
+    it 'posts the message' do
+      ivar = Concurrent::IVar.new
+      subject << [:check, ivar, :foo]
+      ivar.value(0.1).should eq :foo
+    end
+
+    it 'returns self' do
+      (subject << [1,2,3,4]).should eq subject
+    end
+  end
+
+  context '#post with callback' do
+
+    specify 'on success calls the callback with time and value' do
+      expected_value = expected_reason = nil
+      subject.post(:foo) do |time, value, reason|
+        expected_value = value
+        expected_reason = reason
+      end
+      sleep(0.1)
+
+      expected_value.should eq :foo
+      expected_reason.should be_nil
+    end
+
+    specify 'on failure calls the callback with time and reason' do
+      expected_value = expected_reason = nil
+      subject.post(:poison) do |time, value, reason|
+        expected_value = value
+        expected_reason = reason
+      end
+      sleep(0.1)
+
+      expected_value.should be_nil
+      expected_reason.should be_a StandardError
+    end
+  end
+
+  context '#post!' do
+
+    it 'raises an exception when the message is empty' do
+      expect {
+        subject.post!(1)
+      }.to raise_error(ArgumentError)
+    end
+
+    it 'blocks for up to the given number of seconds' do
+      start = Time.now.to_f
+      begin
+        subject.post!(1, :sleep, 2)
+      rescue
+      end
+      delta = Time.now.to_f - start
+      delta.should >= 1
+      delta.should <= 2
+    end
+
+    it 'blocks forever when the timeout is nil' do
+      start = Time.now.to_f
+      begin
+        subject.post!(nil, :sleep, 1)
+      rescue
+      end
+      delta = Time.now.to_f - start
+      delta.should > 1
+    end
+
+    it 'raises a TimeoutError when timeout is zero' do
+      expect {
+        subject.post!(0, :foo)
+      }.to raise_error(Concurrent::TimeoutError)
+    end
+
+    it 'raises a TimeoutError when the timeout is reached' do
+      expect {
+        subject.post!(1, :sleep, 10)
+      }.to raise_error(Concurrent::TimeoutError)
+    end
+
+    it 'returns the result of success processing' do
+      subject.post!(1, :foo).should eq :foo
+    end
+
+    it 'bubbles exceptions thrown during processing' do
+      expect {
+        subject.post!(1, :poison)
+      }.to raise_error(StandardError)
+    end
+  end
+end