Moved actor stress test from spec to script in examples.

Author: jdantonio

examples/actor_stress_test.rb

@@ -0,0 +1,139 @@
+#!/usr/bin/env ruby
+
+$: << File.expand_path('../../lib', __FILE__)
+
+require 'benchmark'
+require 'optparse'
+require 'thread'
+require 'rspec/expectations'
+
+require 'concurrent/actor'
+
+class ActorStressTester
+  include ::RSpec::Matchers
+
+  TESTS_PER_RUN = 5
+  THREADS_PER_TEST = 10
+  LOOPS_PER_THREAD = 25
+
+  class Ping < Concurrent::Actor::Context
+    def initialize(queue)
+      @queue = queue
+    end
+
+    def on_message(message)
+      case message
+      when :child
+        Concurrent::Actor::Utils::AdHoc.spawn(:pong, @queue) do |queue|
+          -> m { queue << m }
+        end
+      else
+        @queue << message
+        message
+      end
+    end
+  end
+
+  def initialize(opts = {})
+    @tests = opts.fetch(:tests, TESTS_PER_RUN)
+    @threads = opts.fetch(:threads, THREADS_PER_TEST)
+    @loops = opts.fetch(:loops, LOOPS_PER_THREAD)
+  end
+
+  def run
+    plural = ->(number){ number == 1 ? '' : 's' }
+
+    puts "Running #{@tests} test#{plural.call(@tests)} " +
+      "with #{@threads} thread#{plural.call(@threads)} each " +
+      "and #{@loops} loop#{plural.call(@loops)} per thread..."
+
+    Benchmark.bm do |bm|
+      @tests.times do
+        bm.report do
+          test(@threads, @loops)
+        end
+      end
+    end
+  end
+
+  def test(threads, loops)
+    (1..threads).collect do
+      Thread.new do
+        loops.times do
+
+          queue = Queue.new
+          actor = Ping.spawn(:ping, queue)
+
+          core = Concurrent::Actor.root.send(:core)
+          children = core.instance_variable_get(:@children)
+          expect(children).to include(actor)
+
+          actor << 'a' << 1
+          expect(queue.pop).to eq 'a'
+          expect(actor.ask(2).value).to eq 2
+
+          expect(actor.parent).to eq Concurrent::Actor.root
+          expect(Concurrent::Actor.root.path).to eq '/'
+          expect(actor.path).to eq '/ping'
+
+          child = actor.ask(:child).value
+          expect(child.path).to eq '/ping/pong'
+
+          queue.clear
+          child.ask(3)
+          expect(queue.pop).to eq 3
+
+          actor << :terminate!
+          expect(actor.ask(:blow_up).wait).to be_rejected
+          terminate_actors(actor, child)
+        end
+      end
+    end.each(&:join)
+  end
+
+  def terminate_actors(*actors)
+    actors.each do |actor|
+      unless actor.ask!(:terminated?)
+        actor.ask!(:terminate!)
+      end
+    end
+  end
+end
+
+# def trace!
+#   set_trace_func proc { |event, file, line, id, binding, classname|
+#     # thread = eval('Thread.current', binding).object_id.to_s(16)
+#     printf "%8s %20s %20s %s %s:%-2d\n", event, id, classname, nil, file, line
+#   }
+#   yield
+# ensure
+#   set_trace_func nil
+# end
+
+if $0 == __FILE__
+
+  options = {}
+
+  OptionParser.new do |opts|
+    opts.banner = "Usage: #{File.basename(__FILE__)} [options]"
+
+    opts.on("--tests=TESTS", "Number of tests per run") do |value|
+      options[:tests] = value.to_i
+    end
+
+    opts.on("--threads=THREADS", "Number of threads per test") do |value|
+      options[:threads] = value.to_i
+    end
+
+    opts.on("--loops=LOOPS", "Number of loops per thread") do |value|
+      options[:loops] = value.to_i
+    end
+
+    opts.on("-h", "--help", "Prints this help") do
+      puts opts
+      exit
+    end
+  end.parse!
+
+  ActorStressTester.new(options).run
+end

spec/concurrent/actor_spec.rb

@@ -43,57 +43,10 @@ def on_message(message)
         end
       end
 
-      # def trace!
-      #   set_trace_func proc { |event, file, line, id, binding, classname|
-      #     # thread = eval('Thread.current', binding).object_id.to_s(16)
-      #     printf "%8s %20s %20s %s %s:%-2d\n", event, id, classname, nil, file, line
-      #   }
-      #   yield
-      # ensure
-      #   set_trace_func nil
-      # end
-
       it 'forbids Immediate executor' do
         expect { Utils::AdHoc.spawn name: 'test', executor: ImmediateExecutor.new }.to raise_error
       end
 
-      #describe 'stress test' do
-        #1.times do |i|
-          #it format('run %3d', i) do
-            ## puts format('run %3d', i)
-            #Array.new(10).map do
-              #Thread.new do
-                #10.times do
-                  ## trace! do
-                  #queue = Queue.new
-                  #actor = Ping.spawn :ping, queue
-
-                  ## when spawn returns children are set
-                  #expect(Concurrent::Actor.root.send(:core).instance_variable_get(:@children)).to include(actor)
-
-                  #actor << 'a' << 1
-                  #expect(queue.pop).to eq 'a'
-                  #expect(actor.ask(2).value).to eq 2
-
-                  #expect(actor.parent).to eq Concurrent::Actor.root
-                  #expect(Concurrent::Actor.root.path).to eq '/'
-                  #expect(actor.path).to eq '/ping'
-                  #child = actor.ask(:child).value
-                  #expect(child.path).to eq '/ping/pong'
-                  #queue.clear
-                  #child.ask(3)
-                  #expect(queue.pop).to eq 3
-
-                  #actor << :terminate!
-                  #expect(actor.ask(:blow_up).wait).to be_rejected
-                  #terminate_actors actor, child
-                #end
-              #end
-            #end.each(&:join)
-          #end
-        #end
-      #end
-
       describe 'spawning' do
         describe 'Actor#spawn' do
           behaviour = -> v { -> _ { v } }
@@ -292,7 +245,7 @@ def on_message(message)
 
           test = AdHoc.spawn name: :tester, behaviour_definition: resuming_behaviour do
             actor = AdHoc.spawn name:                 :pausing,
-                                behaviour_definition: Behaviour.restarting_behaviour_definition do
+              behaviour_definition: Behaviour.restarting_behaviour_definition do
               queue << :init
               -> m { m == :add ? 1 : pass }
             end
@@ -314,22 +267,22 @@ def on_message(message)
           terminate_actors test
 
           test = AdHoc.spawn name:                 :tester,
-                             behaviour_definition: Behaviour.restarting_behaviour_definition do
+            behaviour_definition: Behaviour.restarting_behaviour_definition do
             actor = AdHoc.spawn name:                 :pausing,
-                                supervise:            true,
-                                behaviour_definition: Behaviour.restarting_behaviour_definition do
-              queue << :init
-              -> m { m == :object_id ? self.object_id : pass }
-            end
+              supervise:            true,
+              behaviour_definition: Behaviour.restarting_behaviour_definition do
+                queue << :init
+                -> m { m == :object_id ? self.object_id : pass }
+              end
 
-            queue << actor.ask!(:supervisor)
-            queue << actor.ask!(:object_id)
-            actor << nil
-            queue << actor.ask(:object_id)
+              queue << actor.ask!(:supervisor)
+              queue << actor.ask!(:object_id)
+              actor << nil
+              queue << actor.ask(:object_id)
 
-            -> m do
-              queue << m
-            end
+              -> m do
+                queue << m
+              end
           end
 
           expect(queue.pop).to eq :init
@@ -352,7 +305,7 @@ def on_message(message)
 
           test = AdHoc.spawn name: :tester, behaviour_definition: resuming_behaviour do
             actor = AdHoc.spawn name:                 :pausing,
-                                behaviour_definition: Behaviour.restarting_behaviour_definition do
+              behaviour_definition: Behaviour.restarting_behaviour_definition do
               queue << :init
               -> m { m == :add ? 1 : pass }
             end