Merge pull request #17 from chrisseaton/mvars

MVar: initial implementation.

Author: jdantonio

README.md

@@ -34,6 +34,7 @@ The design goals of this gem are:
   loosely based on the [MailboxProcessor](http://blogs.msdn.com/b/dsyme/archive/2010/02/15/async-and-parallel-design-patterns-in-f-part-3-agents.aspx)
   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
 
 ### Semantic Versioning
 

lib/concurrent.rb

@@ -13,6 +13,7 @@
 require 'concurrent/dereferenceable'
 require 'concurrent/event'
 require 'concurrent/future'
+require 'concurrent/mvar'
 require 'concurrent/obligation'
 require 'concurrent/postable'
 require 'concurrent/promise'

lib/concurrent/dereferenceable.rb

@@ -32,19 +32,23 @@ def set_deref_options(opts = {})
     # by the #set_deref_options method.
     def value
       mutex.synchronize do
-        return nil if @value.nil?
-        return @value if @do_nothing_on_deref
-        value = @value
-        value = @copy_on_deref.call(value) if @copy_on_deref
-        value = value.dup if @dup_on_deref
-        value = value.freeze if @freeze_on_deref
-        value
+        apply_deref_options(@value)
       end
     end
     alias_method :deref, :value
 
     protected
 
+    def apply_deref_options(value)
+      return nil if value.nil?
+      return value if @do_nothing_on_deref
+      value = value
+      value = @copy_on_deref.call(value) if @copy_on_deref
+      value = value.dup if @dup_on_deref
+      value = value.freeze if @freeze_on_deref
+      value
+    end
+
     def mutex # :nodoc:
       @mutex
     end

lib/concurrent/mvar.rb

@@ -0,0 +1,131 @@
+require 'concurrent/event'
+
+module Concurrent
+
+  class MVar
+
+    include Dereferenceable
+
+    EMPTY = Object.new
+    TIMEOUT = Object.new
+
+    def initialize(value = EMPTY, opts = {})
+      @value = value
+      @mutex = Mutex.new
+      @empty_condition = ConditionVariable.new
+      @full_condition = ConditionVariable.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?
+
+        # If we timed out we'll still be empty
+        if full?
+          value = @value
+          @value = EMPTY
+          @empty_condition.signal
+          apply_deref_options(value)
+        else
+          TIMEOUT
+        end
+      end
+    end
+
+    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?
+
+        # If we timed out we won't be empty
+        if empty?
+          @value = value
+          @full_condition.signal
+          apply_deref_options(value)
+        else
+          TIMEOUT
+        end
+      end
+    end
+
+    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?
+
+        # If we timed out we'll still be empty
+        if full?
+          value = @value
+          @value = yield value
+          @full_condition.signal
+          apply_deref_options(value)
+        else
+          TIMEOUT
+        end
+      end
+    end
+
+    def try_take!
+      @mutex.synchronize do
+        if full?
+          value = @value
+          @value = EMPTY
+          @empty_condition.signal
+          apply_deref_options(value)
+        else
+          EMPTY
+        end
+      end
+    end
+
+    def try_put!(value)
+      @mutex.synchronize do
+        if empty?
+          @value = value
+          @full_condition.signal
+          true
+        else
+          false
+        end
+      end
+    end
+
+    def set!(value)
+      @mutex.synchronize do
+        old_value = @value
+        @value = value
+        @full_condition.signal
+        apply_deref_options(old_value)
+      end
+    end
+
+    def modify!(timeout = nil)
+      raise ArgumentError.new('no block given') unless block_given?
+
+      @mutex.synchronize do
+        value = @value
+        @value = yield value
+        if @value == EMPTY
+          @empty_condition.signal
+        else
+          @full_condition.signal
+        end
+        apply_deref_options(value)
+      end
+    end
+
+    def empty?
+      @value == EMPTY
+    end
+
+    def full?
+      not empty?
+    end
+
+  end
+
+end

md/mvar.md

@@ -0,0 +1,32 @@
+# MVar
+
+An `MVar` is a synchronized single element container. They are empty or contain
+one item. Taking a value from an empty `MVar` blocks, as does putting a value
+into a full one. You can either think of them as blocking queue of length one,
+or a special kind of mutable variable.
+
+On top of the fundamental `#put` and `#take` operations, we also provide a
+`#mutate` that is atomic with respect to operations on the same instance. These
+operations all support timeouts.
+
+We also support non-blocking operations `#try_put!` and `#try_take!`, a `#set!`
+that ignores existing values, a `#value` that returns the value without removing
+it or returns `MVar::EMPTY`, and a `#modify!` that yields `MVar::EMPTY` if the
+`MVar` is empty and can be used to set `MVar::EMPTY`. You shouldn't use these
+operations in the first instance.
+
+`MVar` is a Dereferenceable.
+
+`MVar` is related to M-structures in Id, `MVar` in Haskell and `SyncVar` in
+Scala.
+
+See:
+
+1.  P. Barth, R. Nikhil, and Arvind. M-Structures: Extending a parallel, non-
+strict, functional language with state. In Proceedings of the 5th ACM Conference
+on Functional Programming Languages and Computer Architecture (FPCA), 1991.
+
+2.  S. Peyton Jones, A. Gordon, and S. Finne. Concurrent Haskell. In Proceedings of the 23rd Symposium on Principles of Programming Languages (PoPL), 1996.
+
+Note that unlike the original Haskell paper, our `#take` is blocking. This is
+how Haskell and Scala do it today.