Merge master into tvars.

Author: chrisseaton

.travis.yml

@@ -8,6 +8,10 @@ rvm:
   - jruby-19mode
   - jruby-head
   - rbx-2
+branches:
+  only:
+    - master
+    - tvars
 matrix:
   allow_failures:
     - rvm: ruby-head

Gemfile

@@ -4,7 +4,6 @@ gemspec
 
 group :development do
   gem 'rake'
-  gem 'eventmachine'
   gem 'countloc', :platforms => :mri
 end
 

README.md

@@ -35,6 +35,7 @@ The design goals of this gem are:
   agent in [F#](http://msdn.microsoft.com/en-us/library/ee370357.aspx)
 * [Dataflow](https://github.com/jdantonio/concurrent-ruby/blob/master/md/dataflow.md) loosely based on the syntax of Akka and Habanero Java
 * [MVar](https://github.com/jdantonio/concurrent-ruby/blob/master/md/mvar.md) inspired by Haskell
+* [Thread local variables](https://github.com/jdantonio/concurrent-ruby/blob/master/md/threadlocalvar.md) with default value
 
 ### Semantic Versioning
 

lib/concurrent.rb

@@ -1,6 +1,8 @@
 require 'concurrent/version'
 
+require 'concurrent/atomic_counter'
 require 'concurrent/count_down_latch'
+require 'concurrent/condition'
 require 'concurrent/copy_on_notify_observer_set'
 require 'concurrent/copy_on_write_observer_set'
 require 'concurrent/safe_task_executor'
@@ -21,6 +23,7 @@
 require 'concurrent/scheduled_task'
 require 'concurrent/stoppable'
 require 'concurrent/supervisor'
+require 'concurrent/threadlocalvar'
 require 'concurrent/timer_task'
 require 'concurrent/tvar'
 require 'concurrent/utilities'
@@ -31,8 +34,6 @@
 require 'concurrent/fixed_thread_pool'
 require 'concurrent/immediate_executor'
 
-require 'concurrent/event_machine_defer_proxy' if defined?(EventMachine)
-
 # Modern concurrency tools for Ruby. Inspired by Erlang, Clojure, Scala, Haskell,
 # F#, C#, Java, and classic concurrency patterns.
 # 

lib/concurrent/atomic_counter.rb

@@ -0,0 +1,31 @@
+module Concurrent
+
+  class AtomicCounter
+
+    def initialize(init = 0)
+      raise ArgumentError.new('initial value must be an integer') unless init.is_a?(Integer)
+      @counter = init
+      @mutex = Mutex.new
+    end
+
+    def value
+      @mutex.synchronize do
+        @counter
+      end
+    end
+
+    def increment
+      @mutex.synchronize do
+        @counter += 1
+      end
+    end
+    alias_method :up, :increment
+
+    def decrement
+      @mutex.synchronize do
+        @counter -= 1
+      end
+    end
+    alias_method :down, :decrement
+  end
+end

lib/concurrent/condition.rb

@@ -0,0 +1,49 @@
+module Concurrent
+  class Condition
+
+    class Result
+      def initialize(remaining_time)
+        @remaining_time = remaining_time
+      end
+
+      attr_reader :remaining_time
+
+      def woken_up?
+        @remaining_time.nil? || @remaining_time > 0
+      end
+
+      def timed_out?
+        @remaining_time != nil && @remaining_time <= 0
+      end
+
+      alias_method :can_wait?, :woken_up?
+
+    end
+
+    def initialize
+      @condition = ConditionVariable.new
+    end
+
+    def wait(mutex, timeout = nil)
+      start_time = Time.now.to_f
+      @condition.wait(mutex, timeout)
+
+      if timeout.nil?
+        Result.new(nil)
+      else
+        Result.new(start_time + timeout - Time.now.to_f)
+      end
+    end
+
+    def signal
+      @condition.signal
+      true
+    end
+
+    def broadcast
+      @condition.broadcast
+      true
+    end
+
+  end
+end

lib/concurrent/dataflow.rb

@@ -1,21 +1,8 @@
+require 'concurrent/atomic_counter'
 require 'concurrent/future'
 
 module Concurrent
 
-  class AtomicCounter
-
-    def initialize(init)
-      @counter = init
-      @mutex = Mutex.new
-    end
-
-    def decrement
-      @mutex.synchronize do
-        @counter -= 1
-      end
-    end
-  end
-
   class DependencyCounter
 
     def initialize(count, &block)

lib/concurrent/event.rb

@@ -1,5 +1,6 @@
 require 'thread'
 require 'concurrent/utilities'
+require 'concurrent/condition'
 
 module Concurrent
 
@@ -20,7 +21,7 @@ class Event
     def initialize
       @set = false
       @mutex = Mutex.new
-      @condition = ConditionVariable.new
+      @condition = Condition.new
     end
 
     # Is the object in the set state?
@@ -66,7 +67,12 @@ def reset
     def wait(timeout = nil)
       @mutex.synchronize do
         return true if @set
-        @condition.wait(@mutex, timeout)
+
+        remaining = Condition::Result.new(timeout)
+        while !@set && remaining.can_wait?
+          remaining = @condition.wait(@mutex, remaining.remaining_time)
+        end
+
         @set
       end
     end

lib/concurrent/event_machine_defer_proxy.rb

@@ -12,18 +12,17 @@ class MVar
     def initialize(value = EMPTY, opts = {})
       @value = value
       @mutex = Mutex.new
-      @empty_condition = ConditionVariable.new
-      @full_condition = ConditionVariable.new
+      @empty_condition = Condition.new
+      @full_condition = Condition.new
       set_deref_options(opts)
     end
 
     def take(timeout = nil)
       @mutex.synchronize do
-        # If the value isn't empty, wait for full to be signalled
-        @full_condition.wait(@mutex, timeout) if empty?
+        wait_for_full(timeout)
 
         # If we timed out we'll still be empty
-        if full?
+        if unlocked_full?
           value = @value
           @value = EMPTY
           @empty_condition.signal
@@ -36,11 +35,10 @@ def take(timeout = nil)
 
     def put(value, timeout = nil)
       @mutex.synchronize do
-        # Unless the value is empty, wait for empty to be signalled
-        @empty_condition.wait(@mutex, timeout) if full?
+        wait_for_empty(timeout)
 
         # If we timed out we won't be empty
-        if empty?
+        if unlocked_empty?
           @value = value
           @full_condition.signal
           apply_deref_options(value)
@@ -54,11 +52,10 @@ def modify(timeout = nil)
       raise ArgumentError.new('no block given') unless block_given?
 
       @mutex.synchronize do
-        # If the value isn't empty, wait for full to be signalled
-        @full_condition.wait(@mutex, timeout) if empty?
+        wait_for_full(timeout)
 
         # If we timed out we'll still be empty
-        if full?
+        if unlocked_full?
           value = @value
           @value = yield value
           @full_condition.signal
@@ -71,7 +68,7 @@ def modify(timeout = nil)
 
     def try_take!
       @mutex.synchronize do
-        if full?
+        if unlocked_full?
           value = @value
           @value = EMPTY
           @empty_condition.signal
@@ -84,7 +81,7 @@ def try_take!
 
     def try_put!(value)
       @mutex.synchronize do
-        if empty?
+        if unlocked_empty?
           @value = value
           @full_condition.signal
           true
@@ -103,13 +100,13 @@ def set!(value)
       end
     end
 
-    def modify!(timeout = nil)
+    def modify!
       raise ArgumentError.new('no block given') unless block_given?
 
       @mutex.synchronize do
         value = @value
         @value = yield value
-        if @value == EMPTY
+        if unlocked_empty?
           @empty_condition.signal
         else
           @full_condition.signal
@@ -119,13 +116,37 @@ def modify!(timeout = nil)
     end
 
     def empty?
-      @value == EMPTY
+      @mutex.synchronize { @value == EMPTY }
     end
 
     def full?
       not empty?
     end
 
+    private
+
+    def unlocked_empty?
+      @value == EMPTY
+    end
+
+    def unlocked_full?
+      ! unlocked_empty?
+    end
+
+    def wait_for_full(timeout)
+      remaining = Condition::Result.new(timeout)
+      while unlocked_empty? && remaining.can_wait?
+        remaining = @full_condition.wait(@mutex, remaining.remaining_time)
+      end
+    end
+
+    def wait_for_empty(timeout)
+      remaining = Condition::Result.new(timeout)
+      while unlocked_full? && remaining.can_wait?
+        remaining = @empty_condition.wait(@mutex, remaining.remaining_time)
+      end
+    end
+
   end
 
 end