Merge pull request #255 from ruby-concurrency/fix/thread-pool-shutdown

Global configuration and thread pool updates.

Author: jdantonio

.travis.yml

@@ -26,4 +26,4 @@ matrix:
     - rvm: jruby-head
     - rvm: 1.9.3
 
-script: "rake compile && bundle exec rspec --color --backtrace --tag ~unfinished --seed 1 --format documentation ./spec"
+script: "rake compile && bundle exec rspec --color --backtrace --tag ~unfinished --seed 1 ./spec"

CHANGELOG.md

@@ -15,6 +15,33 @@
 * Deprecated all clock-time based timer scheduling
   - Only support scheduling by delay
   - Effects `Concurrent.timer`, `TimerSet`, and `ScheduledTask`
+* Consistent `at_exit` behavior for Java and Ruby thread pools.
+* Added `at_exit` handler to Ruby thread pools (already in Java thread pools)
+  - Ruby handler stores the object id and retrieves from `ObjectSpace`
+  - JRuby disables `ObjectSpace` by default so that handler stores the object reference
+*	Added a `:stop_on_exit` option to thread pools to enable/disable `at_exit` handler
+* Updated thread pool docs to better explain shutting down thread pools
+* Simpler `:executor` option syntax for all abstractions which support this option
+* Added `Executor#auto_terminate?` predicate method (for thread pools)
+* Added `at_exit` handler to `TimerSet`
+* Simplified auto-termination of the global executors
+  - Can now disable auto-termination of global executors
+  - Added shutdown/kill/wait_for_termination variants for global executors
+* Can now disable auto-termination for *all* executors (the nuclear option)
+* Simplified auto-termination of the global executors
+* Deprecated terms "task pool" and "operation pool"
+  - New terms are "io executor" and "fast executor"
+  - New functions added with new names
+  - Deprecation warnings added to functions referencing old names
+* Moved all thread pool related functions from `Concurrent::Configuration` to `Concurrent`
+  - Old functions still exist with deprecation warnings
+  - New functions have updated names as appropriate
+* All high-level abstractions default to the "io executor"
+* Fixed bug in `Actor` causing it to prematurely warm global thread pools on gem load
+  - This also fixed a `RejectedExecutionError` bug when running with minitest/autorun via JRuby
+* Added `LazyReference`, a simpler and faster varition of `Delay`
+  - Updated most internal uses of `Delay` with `LazyReference`
+* Moved global logger up to the `Concurrent` namespace and refactored the code
 
 ## Current Release v0.8.0 (25 January 2015)
 

Gemfile

@@ -4,21 +4,20 @@ gemspec name: 'concurrent-ruby'
 
 group :development do
   gem 'rake', '~> 10.3.2'
-  gem 'rake-compiler', '~> 0.9.2'
+  gem 'rake-compiler', '~> 0.9.5'
   gem 'gem-compiler', '~> 0.3.0'
 end
 
 group :testing do
-  gem 'rspec', '~> 3.0.0'
-  gem 'simplecov', '~> 0.8.2', :require => false
-  gem 'coveralls', '~> 0.7.0', :require => false
-  gem 'timecop', '~> 0.7.1'
+  gem 'rspec', '~> 3.2.0'
+  gem 'simplecov', '~> 0.9.2', :require => false
+  gem 'coveralls', '~> 0.7.11', :require => false
+  gem 'timecop', '~> 0.7.3'
 end
 
 group :documentation do
   gem 'countloc', '~> 0.4.0', :platforms => :mri, :require => false
-  gem 'rubycritic', '~> 1.0.2', :platforms => :mri, require: false
-  gem 'yard', '~> 0.8.7.4', :require => false
-  gem 'inch', '~> 0.4.6', :platforms => :mri, :require => false
-  gem 'redcarpet', '~> 3.1.2', platforms: :mri # understands github markdown
+  gem 'yard', '~> 0.8.7.6', :require => false
+  gem 'inch', '~> 0.5.10', :platforms => :mri, :require => false
+  gem 'redcarpet', '~> 3.2.2', platforms: :mri # understands github markdown
 end

README.md

@@ -66,36 +66,29 @@ This library contains a variety of concurrency abstractions at high and low leve
 
 * See [ThreadPool](http://ruby-concurrency.github.io/concurrent-ruby/file.thread_pools.html) overview, which also contains a list of other Executors available.
 
-### Thread-safe Observers
-
-* [Concurrent::Observable](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Observable.html) mixin module
-* [CopyOnNotifyObserverSet](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CopyOnNotifyObserverSet.html)
-* [CopyOnWriteObserverSet](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CopyOnWriteObserverSet.html)
-
 ### Thread synchronization classes and algorithms
 
-Lower-level abstractions mainly used as building blocks. 
-
-* [condition](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Condition.html)
-* [countdown latch](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CountDownLatch.html)
-* [cyclic barrier](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CyclicBarrier.html)
-* [event](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Event.html)
-* [exchanger](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Exchanger.html)
-* [semaphore](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Semaphore.html)
-* [timeout](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent.html#timeout-class_method)
-* [timer](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent.html#timer-class_method)
+* [Condition](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Condition.html)
+* [CountdownLatch](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CountDownLatch.html)
+* [CyclicBarrier](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CyclicBarrier.html)
+* [Event](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Event.html)
+* [Exchanger](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Exchanger.html)
+* [Semaphore](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Semaphore.html)
+* [Timeout](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent.html#timeout-class_method)
+* [Timer](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent.html#timer-class_method)
 
 ### Thread-safe variables
 
-Lower-level abstractions mainly used as building blocks. 
-
 * [AtomicBoolean](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/AtomicBoolean.html)
 * [AtomicFixnum](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/AtomicFixnum.html)
-* AtomicReference (no docs currently available, check source)
+* [AtomicReference](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/MutexAtomic.html)
+* [Delay](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Delay.html)
+* [LazyReference](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/LazyReference.html)
+* [LazyRegister](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/LazyRegister.html)
 * [I-Structures](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/IVar.html) (IVar)
 * [M-Structures](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/MVar.html) (MVar)
-* [thread-local variables](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/ThreadLocalVar.html)
-* [software transactional memory](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/TVar.html) (TVar)
+* [Thread-local variables](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/ThreadLocalVar.html)
+* [Software transactional memory](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/TVar.html) (TVar)
 
 ## Usage
 
@@ -128,6 +121,8 @@ require 'concurrent/delay'          # Concurrent::Delay
 require 'concurrent/exchanger'      # Concurrent::Exchanger
 require 'concurrent/future'         # Concurrent::Future
 require 'concurrent/ivar'           # Concurrent::IVar
+require 'concurrent/lazy_register'  # Concurrent::LazyRegister
+require 'concurrent/lazy_reference' # Concurrent::LazyReference
 require 'concurrent/mvar'           # Concurrent::MVar
 require 'concurrent/promise'        # Concurrent::Promise
 require 'concurrent/scheduled_task' # Concurrent::ScheduledTask

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

examples/lazy_and_delay.rb

@@ -0,0 +1,23 @@
+#!/usr/bin/env ruby
+
+$: << File.expand_path('../../lib', __FILE__)
+
+require 'benchmark'
+
+require 'concurrent/delay'
+require 'concurrent/lazy_reference'
+
+n = 500_000
+
+delay = Concurrent::Delay.new{ nil }
+lazy = Concurrent::LazyReference.new{ nil }
+
+delay.value
+lazy.value
+
+Benchmark.bm do |x|
+  puts 'Benchmarking Delay...'
+  x.report { n.times{ delay.value } }
+  puts 'Benchmarking Lazy...'
+  x.report { n.times{ lazy.value } }
+end

lib/concurrent.rb

@@ -2,28 +2,26 @@
 
 require 'concurrent/configuration'
 
+require 'concurrent/actor'
 require 'concurrent/atomics'
 require 'concurrent/channels'
 require 'concurrent/collections'
+require 'concurrent/errors'
 require 'concurrent/executors'
 require 'concurrent/utilities'
 
-require 'concurrent/actor'
 require 'concurrent/atomic'
-require 'concurrent/lazy_register'
 require 'concurrent/agent'
 require 'concurrent/async'
+require 'concurrent/atomic'
 require 'concurrent/dataflow'
 require 'concurrent/delay'
-require 'concurrent/dereferenceable'
-require 'concurrent/errors'
 require 'concurrent/exchanger'
 require 'concurrent/future'
 require 'concurrent/ivar'
+require 'concurrent/lazy_reference'
+require 'concurrent/lazy_register'
 require 'concurrent/mvar'
-require 'concurrent/obligation'
-require 'concurrent/observable'
-require 'concurrent/options_parser'
 require 'concurrent/promise'
 require 'concurrent/scheduled_task'
 require 'concurrent/timer_task'

lib/concurrent/actor.rb

@@ -1,4 +1,5 @@
 require 'concurrent/configuration'
+require 'concurrent/delay'
 require 'concurrent/executor/serialized_execution'
 require 'concurrent/ivar'
 require 'concurrent/logging'
@@ -39,7 +40,7 @@ def self.current
       Thread.current[:__current_actor__]
     end
 
-    @root = Delay.new do
+    @root = Delay.new(executor: :immediate) do
       Core.new(parent: nil, name: '/', class: Root, initialized: ivar = IVar.new).reference.tap do
         ivar.no_error!
       end
@@ -59,7 +60,7 @@ def self.root
     #   Actor.spawn name:     :ping3,
     #                 class:    AdHoc,
     #                 args:     [1]
-    #                 executor: Concurrent.configuration.global_task_pool do |add|
+    #                 executor: Concurrent.global_io_executor do |add|
     #     lambda { |number| number + add }
     #   end
     #

lib/concurrent/actor/core.rb

@@ -35,7 +35,7 @@ class Core
       # @option opts [Class] reference a custom descendant of {Reference} to use
       # @option opts [Context] actor_class a class to be instantiated defining Actor's behaviour
       # @option opts [Array<Object>] args arguments for actor_class instantiation
-      # @option opts [Executor] executor, default is `Concurrent.configuration.global_task_pool`
+      # @option opts [Executor] executor, default is `Concurrent.global_io_executor`
       # @option opts [true, false] link, atomically link the actor to its parent
       # @option opts [true, false] supervise, atomically supervise the actor by its parent
       # @option opts [Array<Array(Behavior::Abstract, Array<Object>)>]
@@ -56,7 +56,7 @@ def initialize(opts = {}, &block)
           @context_class = Child! opts.fetch(:class), AbstractContext
           allocate_context
 
-          @executor = Type! opts.fetch(:executor, Concurrent.configuration.global_task_pool), Executor
+          @executor = Type! opts.fetch(:executor, Concurrent.global_io_executor), Executor
           raise ArgumentError, 'ImmediateExecutor is not supported' if @executor.is_a? ImmediateExecutor
 
           @reference = (Child! opts[:reference_class] || @context.default_reference_class, Reference).new self

lib/concurrent/actor/reference.rb

@@ -27,8 +27,8 @@ def tell(message)
 
       # @note it's a good practice to use tell whenever possible. Ask should be used only for
       # testing and when it returns very shortly. It can lead to deadlock if all threads in
-      # global_task_pool will block on while asking. It's fine to use it form outside of actors and
-      # global_task_pool.
+      # global_io_executor will block on while asking. It's fine to use it form outside of actors and
+      # global_io_executor.
       #
       # sends message to the actor and asks for the result of its processing, returns immediately
       # @param [Object] message
@@ -40,8 +40,8 @@ def ask(message, ivar = IVar.new)
 
       # @note it's a good practice to use tell whenever possible. Ask should be used only for
       # testing and when it returns very shortly. It can lead to deadlock if all threads in
-      # global_task_pool will block on while asking. It's fine to use it form outside of actors and
-      # global_task_pool.
+      # global_io_executor will block on while asking. It's fine to use it form outside of actors and
+      # global_io_executor.
       #
       # sends message to the actor and asks for the result of its processing, blocks
       # @param [Object] message