Merge remote-tracking branch 'upstream/master' into actress

* upstream/master: (13 commits)
  Temporarily adding Rbx to Travis allowed failure list due to problem with the Travis service.
  ...

Conflicts:
	lib/concurrent/atomic/thread_local_var.rb

Author: pitr-ch

.gitignore

@@ -20,6 +20,7 @@ rdoc/*
 *.REMOTE.*
 git_pull.txt
 coverage
+critic
 .DS_Store
 TAGS
 tmtags

.travis.yml

@@ -14,6 +14,7 @@ branches:
     - master
 matrix:
   allow_failures:
+    - rvm: rbx-2
     - rvm: ruby-head
     - rvm: jruby-head
     - rvm: 1.9.3

Gemfile

@@ -16,6 +16,7 @@ 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

README.md

@@ -54,7 +54,7 @@ _NOTE: There is an old gem from 2007 called "concurrent" that does not appear to
 ## Features & Documentation
 
 Please see the [Concurrent Ruby Wiki](https://github.com/ruby-concurrency/concurrent-ruby/wiki)
-or the [API documentation](http://ruby-concurrency.github.io/concurrent-ruby/frames.html))
+or the [API documentation](http://ruby-concurrency.github.io/concurrent-ruby/frames.html)
 for more information or join our [mailing list](http://groups.google.com/group/concurrent-ruby).
 
 There are many concurrency abstractions in this library. These abstractions can be broadly categorized
@@ -146,6 +146,7 @@ task.value #=> 25.96
 * [Chip Miller](https://github.com/chip-miller)
 * [Giuseppe Capizzi](https://github.com/gcapizzi)
 * [Jamie Hodge](https://github.com/jamiehodge)
+* [Justin Lambert](https://github.com/mastfish)
 * [Larry Lv](https://github.com/larrylv)
 * [Maxim Chechel](https://github.com/maximchick)
 * [Ravil Bayramgalin](https://github.com/brainopia)

lib/concurrent/agent.rb

@@ -38,19 +38,13 @@ class Agent
     include Concurrent::Observable
     include Logging
 
-    # The default timeout value (in seconds); used when no timeout option
-    # is given at initialization
-    TIMEOUT = 5
-
     attr_reader :timeout, :task_executor, :operation_executor
 
     # Initialize a new Agent with the given initial value and provided options.
     #
     # @param [Object] initial the initial value
     # @param [Hash] opts the options used to define the behavior at update and deref
     #
-    # @option opts [Fixnum] :timeout (TIMEOUT) maximum number of seconds before an update is cancelled
-    #
     # @option opts [Boolean] :operation (false) when `true` will execute the future on the global
     #   operation pool (for long-running operations), when `false` will execute the future on the
     #   global task pool (for short-running tasks)
@@ -65,7 +59,6 @@ def initialize(initial, opts = {})
       @value                = initial
       @rescuers             = []
       @validator            = Proc.new { |result| true }
-      @timeout              = opts.fetch(:timeout, TIMEOUT).freeze
       self.observers        = CopyOnWriteObserverSet.new
       @serialized_execution = SerializedExecution.new
       @task_executor        = OptionsParser.get_task_executor_from(opts)
@@ -145,12 +138,19 @@ def post(&block)
     # Update the current value with the result of the given block operation,
     # block can do blocking calls
     #
+    # @param [Fixnum, nil] timeout maximum number of seconds before an update is cancelled
+    #
     # @yield the operation to be performed with the current value in order to calculate
     #   the new value
     # @yieldparam [Object] value the current value
     # @yieldreturn [Object] the new value
     # @return [true, nil] nil when no block is given
-    def post_off(&block)
+    def post_off(timeout = nil, &block)
+      block = if timeout
+                lambda { |value| Concurrent::timeout(timeout) { block.call(value) } }
+              else
+                block
+              end
       post_on(@operation_executor, &block)
     end
 
@@ -203,10 +203,8 @@ def work(&handler) # :nodoc:
       validator, value = mutex.synchronize { [@validator, @value] }
 
       begin
-        result, valid = Concurrent::timeout(@timeout) do
-          result = handler.call(value)
-          [result, validator.call(result)]
-        end
+        result = handler.call(value)
+        valid  = validator.call(result)
       rescue Exception => ex
         exception = ex
       end

lib/concurrent/atomic/thread_local_var.rb

@@ -1,56 +1,19 @@
-require 'concurrent/atomic/atomic_fixnum'
+require 'concurrent/atomic'
 
 module Concurrent
 
-  module ThreadLocalSymbolAllocator
-
-    COUNTER = Concurrent::AtomicFixnum.new
-
-    protected
-
-    def allocate_symbol
-      # Warning: this symbol may never be deallocated
-      @symbol = :"thread_local_symbol_#{COUNTER.increment}"
-    end
-
-  end
-
-  module ThreadLocalOldStorage
-
-    include ThreadLocalSymbolAllocator
-
-    protected
-
-    def allocate_storage
-      allocate_symbol
-    end
-
-    def get
-      Thread.current[@symbol]
-    end
-
-    def set(value)
-      Thread.current[@symbol] = value
-    end
-
-  end
-
-  module ThreadLocalNewStorage
-
-    include ThreadLocalSymbolAllocator
-
-    protected
+  module ThreadLocalRubyStorage
 
     def allocate_storage
-      allocate_symbol
+      @storage = Atomic.new Hash.new
     end
 
     def get
-      Thread.current.thread_variable_get(@symbol)
+      @storage.get[Thread.current]
     end
 
     def set(value)
-      Thread.current.thread_variable_set(@symbol, value)
+      @storage.update { |s| s.merge Thread.current => value }
     end
 
   end
@@ -111,10 +74,8 @@ def value=(value)
   class ThreadLocalVar < AbstractThreadLocalVar
     if RUBY_PLATFORM == 'java'
       include ThreadLocalJavaStorage
-    elsif Thread.current.respond_to?(:thread_variable_set)
-      include ThreadLocalNewStorage
     else
-      include ThreadLocalOldStorage
+      include ThreadLocalRubyStorage
     end
   end
 

lib/concurrent/dataflow.rb

@@ -61,17 +61,34 @@ def update(time, value, reason)
   # @raise [ArgumentError] if no block is given
   # @raise [ArgumentError] if any of the inputs are not `IVar`s
   def dataflow(*inputs, &block)
-    dataflow_with(Concurrent.configuration.global_task_pool, *inputs, &block)
+    dataflow_with(Concurrent.configuration.global_operation_pool, *inputs, &block)
   end
   module_function :dataflow
 
   def dataflow_with(executor, *inputs, &block)
+    call_dataflow(:value, executor, *inputs, &block)
+  end
+  module_function :dataflow_with
+  
+  def dataflow!(*inputs, &block)
+    dataflow_with!(Concurrent.configuration.global_task_pool, *inputs, &block)
+  end
+  module_function :dataflow!
+
+  def dataflow_with!(executor, *inputs, &block)
+    call_dataflow(:value!, executor, *inputs, &block)
+  end
+  module_function :dataflow_with!
+
+  private 
+
+  def call_dataflow(method, executor, *inputs, &block)
     raise ArgumentError.new('an executor must be provided') if executor.nil?
     raise ArgumentError.new('no block given') unless block_given?
     raise ArgumentError.new('not all dependencies are IVars') unless inputs.all? { |input| input.is_a? IVar }
 
     result = Future.new(executor: executor) do
-      values = inputs.map { |input| input.value }
+      values = inputs.map { |input| input.send(method) }
       block.call(*values)
     end
 
@@ -87,5 +104,5 @@ def dataflow_with(executor, *inputs, &block)
 
     result
   end
-  module_function :dataflow_with
+  module_function :call_dataflow
 end

lib/concurrent/executor/timer_set.rb

@@ -23,7 +23,7 @@ class TimerSet
     #   this executor rather than the global thread pool (overrides :operation)
     def initialize(opts = {})
       @queue = PriorityQueue.new(order: :min)
-      @task_executor = OptionsParser::get_executor_from(opts)
+      @task_executor = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_task_pool
       @timer_executor = SingleThreadExecutor.new
       @condition = Condition.new
       init_executor

lib/concurrent/future.rb

@@ -62,7 +62,7 @@ def initialize(opts = {}, &block)
       super(IVar::NO_VALUE, opts)
       @state = :unscheduled
       @task = block
-      @executor = OptionsParser::get_executor_from(opts)
+      @executor = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_operation_pool
     end
 
     # Execute an `:unscheduled` `Future`. Immediately sets the state to `:pending` and

lib/concurrent/options_parser.rb

@@ -10,14 +10,16 @@ module OptionsParser
     # @option opts [Boolean] :operation (`false`) when true use the global operation pool
     # @option opts [Boolean] :task (`true`) when true use the global task pool
     #
-    # @return [Executor] the requested thread pool (default: global task pool)
+    # @return [Executor, nil] the requested thread pool, or nil when no option specified
     def get_executor_from(opts = {})
       if opts[:executor]
         opts[:executor]
       elsif opts[:operation] == true || opts[:task] == false
         Concurrent.configuration.global_operation_pool
-      else
+      elsif opts[:operation] == false || opts[:task] == true
         Concurrent.configuration.global_task_pool
+      else
+        nil
       end
     end