Merge branch 'master' into promise-all

Author: jdantonio

doc/thread_pools.md

@@ -1,16 +1,16 @@
 # Thread Pools
 
-A Thread Pool is an abstraction that you can give a unit of work to, and the work will be executed by one of possibly several threads in the pool. One motivation for using thread pools is the overhead of creating and destroying threads. Creating a pool of reusable worker threads then repeatedly re-using threads from the pool can have huge performace benefits for a long-running application like a service.
+A Thread Pool is an abstraction that you can give a unit of work to, and the work will be executed by one of possibly several threads in the pool. One motivation for using thread pools is the overhead of creating and destroying threads. Creating a pool of reusable worker threads then repeatedly re-using threads from the pool can have huge performance benefits for a long-running application like a service.
 
 `concurrent-ruby` also offers some higher level abstractions than thread pools. For many problems, you will be better served by using one of these -- if you are thinking of using a thread pool, we especially recommend you look at and understand [Future](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Future.html)s before deciding to use thread pools directly instead.  Futures are implemented using thread pools, but offer a higher level abstraction.
 
 But there are some problems for which directly using a thread pool is an appropriate solution. Or, you may wish to make your own thread pool to run Futures on, to be separate or have different characteristics than the global thread pool that Futures run on by default.
 
-Thread pools are considered 'executors' -- an object you can give a unit of work to, to have it exeucted.  In fact, thread pools are the main kind of executor you will see, others are mainly for testing or odd edge cases. In some documentation or source code you'll see reference to an 'executor' -- this is commonly a thread pool, or else something similar that executes units of work (usually supplied as ruby blocks).
+Thread pools are considered 'executors' -- an object you can give a unit of work to, to have it executed.  In fact, thread pools are the main kind of executor you will see - others are mainly for testing or odd edge cases. In some documentation or source code you'll see reference to an 'executor' -- this is commonly a thread pool, or else something similar that executes units of work (usually supplied as Ruby blocks).
 
 ## FixedThreadPool
 
-A [FixedThreadPool](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/FixedThreadPool.html) contains a fixed number of threads. When you give a unit if work to it, an available thread will be used to execute.
+A [FixedThreadPool](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/FixedThreadPool.html) contains a fixed number of threads. When you give a unit of work to it, an available thread will be used to execute.
 
 ~~~ruby
 pool = Concurrent::FixedThreadPool.new(5) # 5 threads
@@ -21,15 +21,15 @@ end
 # while work is concurrently being done in the thread pool, at some possibly future point.
 ~~~
 
-What happens if you post new work when all (eg) 5 threads are currently busy? It will be added to a queue, and executed when a thread becomes available.  In a `FixedThreadPool`, if you post work to the pool much faster than the work can be completed, the queue may grow without bounds, as the work piles up in the holding queue, using up memory without bounds.  To limit the queue and apply some form of 'back pressure' instead, you can use the more configurable `ThreadPoolExecutor` (See below).
+What happens if you post new work when all (e.g.) 5 threads are currently busy? It will be added to a queue, and executed when a thread becomes available.  In a `FixedThreadPool`, if you post work to the pool much faster than the work can be completed, the queue may grow without bounds, as the work piles up in the holding queue, using up memory without bounds.  To limit the queue and apply some form of 'back pressure' instead, you can use the more configurable `ThreadPoolExecutor` (See below).
 
 If you'd like to base the number of threads in the pool on the number of processors available, your code can consult [Concurrent.processor_count](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/ProcessorCounter.html#processor_count-instance_method).
 
 The `FixedThreadPool` is based on the semantics used in Java for [java.util.concurrent.Executors.newFixedThreadPool(int nThreads)](https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html#newFixedThreadPool(int))
 
 ## CachedThreadPool
 
-A [CachedThreadPool](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CachedThreadPool.html) will create as many threads as neccesary for work posted to it. If you post work to a `CachedThreadPool` when all it's existing threads are busy, it will create a new thread to execute that work, and then keep that thread cached for future work. Cached threads are reclaimed (destroyed) after they are idle for a while.
+A [CachedThreadPool](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/CachedThreadPool.html) will create as many threads as necessary for work posted to it. If you post work to a `CachedThreadPool` when all its existing threads are busy, it will create a new thread to execute that work, and then keep that thread cached for future work. Cached threads are reclaimed (destroyed) after they are idle for a while.
 
 CachedThreadPools typically improve the performance of programs that execute many short-lived asynchronous tasks.
 
@@ -58,18 +58,18 @@ pool = Concurrent::ThreadPoolExecutor.new(
 )
 ~~~
 
-If you want to provide a maximum queue size, you may also consider the `overflow_policy` -- what will happen if work is posted to a pool when the queue of waiting work has reached the maximum size? Available policies:
+If you want to provide a maximum queue size, you may also consider the `fallback_policy` -- what will happen if work is posted to a pool when the queue of waiting work has reached the maximum size? Available policies:
 
 * abort: Raise a `Concurrent::RejectedExecutionError` exception and discard the task. (default policy)
 * discard: Silently discard the task and return nil as the task result.
-* caller_runs: The work will be executed in the thread of the caller, instead of being given to another thread in the pool
+* caller_runs: The work will be executed in the thread of the caller, instead of being given to another thread in the pool.
 
 ~~~ruby
 pool = Concurrent::ThreadPoolExecutor.new(
    min_threads: 5,
    max_threads: 5,
    max_queue: 100,
-   overflow_policy: :caller_runs
+   fallback_policy: :caller_runs
 )
 ~~~
 
@@ -83,14 +83,14 @@ pool = Concurrent::ThreadPoolExecutor.new(
 )
 ~~~
 
-Or, with a variable number of threads like a CachedThreadPool, but with a bounded queue and an overflow_policy:
+Or, with a variable number of threads like a CachedThreadPool, but with a bounded queue and a fallback_policy:
 
 ~~~ruby
 pool = Concurrent::ThreadPoolExecutor.new(
    min_threads: 3, # create 3 threads at startup
    max_threads: 10, # create at most 10 threads
    max_queue: 100, # at most 100 jobs waiting in the queue,
-   overflow_policy: :abort
+   fallback_policy: :abort
 )
 ~~~
 
@@ -106,11 +106,11 @@ ThreadPoolExecutors with `min_threads` and `max_threads` set to different values
 
 A running thread pool can be shutdown in an orderly or disruptive manner. Once a thread pool has been shutdown it cannot be started again.
 
-The `shutdown` method can be used to initiate an orderly shutdown of the thread pool. All new post calls will reject the given block and immediately return false. Threads in the pool will continue to process all in-progress work and will process all tasks still in the queue.
+The `shutdown` method can be used to initiate an orderly shutdown of the thread pool. All new post calls will be handled according to the `fallback_policy` (i.e. failing with a RejectedExecutionError by default). Threads in the pool will continue to process all in-progress work and will process all tasks still in the queue.
 
-The `kill` method can be used to immediately shutdown the pool. All new post calls will reject the given block and immediately return false. Ruby's Thread.kill will be called on all threads in the pool, aborting all in-progress work. Tasks in the queue will be discarded.
+The `kill` method can be used to immediately shutdown the pool. All new post calls will be handled according to the `fallback_policy`. Ruby's `Thread.kill` will be called on all threads in the pool, aborting all in-progress work. Tasks in the queue will be discarded.
 
-The method `wait_for_termination` can be used to block and wait for pool shutdown to complete. This is useful when shutting down an application and ensuring the app doesn't exit before pool processing is complete. The method wait_for_termination will block for a maximum of the given number of seconds then return true if shutdown completed successfully or false. When the timeout value is nil the call will block indefinitely. Calling wait_for_termination on a stopped thread pool will immediately return true.
+The method `wait_for_termination` can be used to block and wait for pool shutdown to complete. This is useful when shutting down an application and ensuring the app doesn't exit before pool processing is complete. The method wait_for_termination will block for a maximum of the given number of seconds then return true (if shutdown completed successfully) or false (if it was still ongoing). When the timeout value is `nil` the call will block indefinitely. Calling `wait_for_termination` on a stopped thread pool will immediately return true.
 
 ~~~ruby
 # tell the pool to shutdown in an orderly fashion, allowing in progress work to complete
@@ -155,10 +155,10 @@ pool = Concurrent::ThreadPoolExecutor.new(
   :min_threads => [2, Concurrent.processor_count].max,
   :max_threads => [2, Concurrent.processor_count].max,
   :max_queue   => [2, Concurrent.processor_count].max * 5,
-  :overflow_policy => :caller_runs
+  :fallback_policy => :caller_runs
 )
 
 future = Future.new(:executor => pool).execute do
    #work
 end
-~~~
+~~~

lib/concurrent/async.rb

@@ -81,13 +81,13 @@ def method_missing(method, *args, &block)
         super unless @delegate.respond_to?(method)
         Async::validate_argc(@delegate, method, *args)
 
-        self.define_singleton_method(method) do |*args|
-          Async::validate_argc(@delegate, method, *args)
+        self.define_singleton_method(method) do |*args2|
+          Async::validate_argc(@delegate, method, *args2)
           ivar = Concurrent::IVar.new
           value, reason = nil, nil
           @serializer.post(@executor.value) do
             begin
-              value = @delegate.send(method, *args, &block)
+              value = @delegate.send(method, *args2, &block)
             rescue => reason
               # caught
             ensure

lib/concurrent/atomic/atomic_fixnum.rb

@@ -25,8 +25,8 @@ module Concurrent
   class MutexAtomicFixnum
 
     # http://stackoverflow.com/questions/535721/ruby-max-integer
-    MIN_VALUE = -(2**(0.size * 8 -2))
-    MAX_VALUE = (2**(0.size * 8 -2) -1)
+    MIN_VALUE = -(2**(0.size * 8 - 2))
+    MAX_VALUE = (2**(0.size * 8 - 2) - 1)
 
     # @!macro [attach] atomic_fixnum_method_initialize
     #

lib/concurrent/executor/indirect_immediate_executor.rb

@@ -28,7 +28,7 @@ def post(*args, &task)
       return false unless running?
 
       event = Concurrent::Event.new
-      internal_executor.post do
+      @internal_executor.post do
         begin
           task.call(*args)
         ensure
@@ -39,8 +39,5 @@ def post(*args, &task)
 
       true
     end
-
-    private
-    attr_reader :internal_executor
   end
 end

lib/concurrent/executor/ruby_thread_pool_worker.rb

@@ -13,6 +13,7 @@ def initialize(queue, parent)
       @parent = parent
       @mutex = Mutex.new
       @last_activity = Time.now.to_f
+      @thread = nil
     end
 
     # @!visibility private

lib/concurrent/executor/serialized_execution.rb

@@ -12,7 +12,7 @@ class SerializedExecution
 
     Job = Struct.new(:executor, :args, :block) do
       def call
-        block.call *args
+        block.call(*args)
       end
     end
 

lib/concurrent/lazy_register.rb

@@ -49,7 +49,7 @@ def register(key, &block)
     # @return self
     # @param [Object] key
     def unregister(key)
-      @data.update { |h| h.dup.tap { |h| h.delete(key) } }
+      @data.update { |h| h.dup.tap { |j| j.delete(key) } }
       self
     end
 

lib/concurrent/observable.rb

@@ -46,7 +46,7 @@ def add_observer(*args, &block)
     # as #add_observer but it can be used for chaining
     # @return [Observable] self
     def with_observer(*args, &block)
-      add_observer *args, &block
+      add_observer(*args, &block)
       self
     end
 

lib/concurrent/promise.rb

@@ -260,7 +260,7 @@ def then(rescuer = nil, &block)
     # @return [Promise]
     def on_success(&block)
       raise ArgumentError.new('no block given') unless block_given?
-      self.then &block
+      self.then(&block)
     end
 
     # @return [Promise]

lib/concurrent/scheduled_task.rb

@@ -26,7 +26,7 @@ def initialize(intended_time, opts = {}, &block)
     def execute
       if compare_and_set_state(:pending, :unscheduled)
         @schedule_time = TimerSet.calculate_schedule_time(@intended_time)
-        Concurrent::timer(@schedule_time.to_f - Time.now.to_f) { @executor.post &method(:process_task) }
+        Concurrent::timer(@schedule_time.to_f - Time.now.to_f) { @executor.post(&method(:process_task)) }
         self
       end
     end