improved MVar to deal with spurious wake ups and fixed a potential issue in empty? and full? methods

Author: mighe

lib/concurrent/mvar.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

spec/concurrent/event_spec.rb

@@ -148,7 +148,7 @@ module Concurrent
     context 'spurious wake ups' do
 
       before(:each) do
-        def subject.wake_up
+        def subject.simulate_spurious_wake_up
           @mutex.synchronize do
             @condition.signal
             @condition.broadcast
@@ -161,7 +161,7 @@ def subject.wake_up
         Thread.new { subject.wait; @expected = true }
 
         sleep(0.1)
-        subject.wake_up
+        subject.simulate_spurious_wake_up
 
         sleep(0.1)
         @expected.should be_false
@@ -172,7 +172,7 @@ def subject.wake_up
         Thread.new { subject.wait(0.5); @expected = true }
 
         sleep(0.1)
-        subject.wake_up
+        subject.simulate_spurious_wake_up
 
         sleep(0.1)
         @expected.should be_false

spec/concurrent/mvar_spec.rb

@@ -1,4 +1,5 @@
 require 'spec_helper'
+require_relative 'dereferenceable_shared'
 
 module Concurrent
 
@@ -16,7 +17,7 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
-    context '#initialize' do
+    describe '#initialize' do
 
       it 'accepts no initial value' do
         m = MVar.new
@@ -40,7 +41,7 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
-    context '#take' do
+    describe '#take' do
 
       it 'sets the MVar to empty' do
         m = MVar.new(14)
@@ -57,21 +58,21 @@ def dereferenceable_subject(value, opts = {})
         m = MVar.new
 
         putter = Thread.new {
-          sleep(0.5)
-          m.put 14 
+          sleep(0.1)
+          m.put 14
         }
 
         m.take.should eq 14
       end
 
       it 'returns TIMEOUT on timeout on an empty MVar' do
         m = MVar.new
-        m.take(0.5).should eq MVar::TIMEOUT
+        m.take(0.1).should eq MVar::TIMEOUT
       end
 
     end
 
-    context '#put' do
+    describe '#put' do
 
       it 'sets the MVar to be empty' do
         m = MVar.new(14)
@@ -89,7 +90,7 @@ def dereferenceable_subject(value, opts = {})
         m = MVar.new(14)
 
         putter = Thread.new {
-          sleep(0.5)
+          sleep(0.1)
           m.take
         }
 
@@ -98,7 +99,7 @@ def dereferenceable_subject(value, opts = {})
 
       it 'returns TIMEOUT on timeout on a full MVar' do
         m = MVar.new(14)
-        m.put(14, 0.5).should eq MVar::TIMEOUT
+        m.put(14, 0.1).should eq MVar::TIMEOUT
       end
 
       it 'returns the value' do
@@ -108,7 +109,7 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
-    context '#empty?' do
+    describe '#empty?' do
 
       it 'returns true on an empty MVar' do
         m = MVar.new
@@ -122,7 +123,7 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
-    context '#full?' do
+    describe '#full?' do
 
       it 'returns false on an empty MVar' do
         m = MVar.new
@@ -136,7 +137,7 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
-    context '#modify' do
+    describe '#modify' do
 
       it 'raises an exception when no block given' do
         m = MVar.new(14)
@@ -158,7 +159,7 @@ def dereferenceable_subject(value, opts = {})
         m = MVar.new
 
         putter = Thread.new {
-          sleep(0.5)
+          sleep(0.1)
           m.put 14 
         }
 
@@ -173,24 +174,24 @@ def dereferenceable_subject(value, opts = {})
 
         modifier = Thread.new {
           m.modify do |v|
-            sleep(1)
+            sleep(0.5)
             1
           end
         }
 
-        sleep(0.5)
-        m.put(2, 1).should eq MVar::TIMEOUT
+        sleep(0.1)
+        m.put(2, 0.5).should eq MVar::TIMEOUT
         m.take.should eq 1
       end
 
       it 'returns TIMEOUT on timeout on an empty MVar' do
         m = MVar.new
-        m.modify(0.5){ |v| v + 2 }.should eq MVar::TIMEOUT
+        m.modify(0.1){ |v| v + 2 }.should eq MVar::TIMEOUT
       end
 
     end
 
-    context '#try_put!' do
+    describe '#try_put!' do
 
       it 'returns true an empty MVar' do
         m = MVar.new
@@ -210,7 +211,7 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
-    context '#try_take!' do
+    describe '#try_take!' do
 
       it 'returns EMPTY an empty MVar' do
         m = MVar.new
@@ -230,7 +231,7 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
-    context '#set!' do
+    describe '#set!' do
 
       it 'sets an empty MVar to be full' do
         m = MVar.new
@@ -257,7 +258,7 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
-    context '#modify!' do
+    describe '#modify!' do
 
       it 'raises an exception when no block given' do
         m = MVar.new(14)
@@ -295,6 +296,86 @@ def dereferenceable_subject(value, opts = {})
 
     end
 
+    context 'spurious wake ups' do
+
+      let(:m) { MVar.new }
+
+      before(:each) do
+        def m.simulate_spurious_wake_up
+          @mutex.synchronize do
+            @full_condition.broadcast
+            @empty_condition.broadcast
+          end
+        end
+      end
+
+      describe '#take' do
+        it 'waits for another thread to #put' do
+          Thread.new { sleep(0.5); m.put 14 }
+          Thread.new { sleep(0.1); m.simulate_spurious_wake_up }
+
+          m.take.should eq 14
+        end
+
+        it 'returns TIMEOUT on timeout on an empty MVar' do
+          result = nil
+          Thread.new { result = m.take(0.3) }
+          sleep(0.1)
+          Thread.new { m.simulate_spurious_wake_up }
+          sleep(0.1)
+          result.should be_nil
+          sleep(0.2)
+          result.should eq MVar::TIMEOUT
+        end
+      end
+
+      describe '#modify' do
+
+        it 'waits for another thread to #put' do
+          Thread.new { sleep(0.5); m.put 14 }
+          Thread.new { sleep(0.1); m.simulate_spurious_wake_up }
+
+          m.modify{ |v| v + 2 }.should eq 14
+        end
+
+        it 'returns TIMEOUT on timeout on an empty MVar' do
+          result = nil
+          Thread.new { result = m.modify(0.3){ |v| v + 2 } }
+          sleep(0.1)
+          Thread.new { m.simulate_spurious_wake_up }
+          sleep(0.1)
+          result.should be_nil
+          sleep(0.2)
+          result.should eq MVar::TIMEOUT
+        end
+      end
+
+      describe '#put' do
+
+        before(:each) { m.put(42) }
+
+        it 'waits for another thread to #take' do
+          Thread.new { sleep(0.5); m.take }
+          Thread.new { sleep(0.1); m.simulate_spurious_wake_up }
+
+          m.put(14).should eq 14
+        end
+
+        it 'returns TIMEOUT on timeout on a full MVar' do
+          result = nil
+          Thread.new { result = m.put(14, 0.3) }
+          sleep(0.1)
+          Thread.new { m.simulate_spurious_wake_up }
+          sleep(0.1)
+          result.should be_nil
+          sleep(0.2)
+          result.should eq MVar::TIMEOUT
+        end
+      end
+
+
+    end
+
   end
 
 end