first working implementation

Author: mighe

lib/concurrent/atomic/cyclic_barrier.rb

@@ -2,6 +2,11 @@ module Concurrent
 
   class CyclicBarrier
 
+    BrokenError = Class.new(StandardError)
+
+    Generation = Struct.new(:status)
+    private_constant :Generation
+
     # Create a new `CyclicBarrier` that waits for `parties` threads
     #
     # @param [Fixnum] parties the number of parties
@@ -15,6 +20,7 @@ def initialize(parties, &block)
       @condition = Condition.new
       @number_waiting = 0
       @action = block
+      @generation = Generation.new(:waiting)
     end
 
     # @return [Fixnum] the number of threads needed to pass the barrier
@@ -30,19 +36,37 @@ def number_waiting
     # Blocks on the barrier until the number of waiting threads is equal to `parties` or until `timeout` is reached or `reset` is called
     # If a block has been passed to the constructor, it will be executed once by the last arrived thread before releasing the others
     # @param [Fixnum] timeout the number of seconds to wait for the counter or `nil` to block indefinitely
-    # @return [Boolean] `true` if the `count` reaches zero else false on `timeout` or on `reset`
+    # @return [Boolean] `true` if the `count` reaches zero else false on `timeout` or on `reset` or on broken event
     def wait(timeout = nil)
       @mutex.synchronize do
+
+        return false unless @generation.status == :waiting
+
+        generation = @generation
+
         @number_waiting += 1
+
         if @number_waiting == @parties
           @action.call if @action
+          generation.status = :fulfilled
+          @generation = Generation.new(:waiting)
           @condition.broadcast
           @number_waiting = 0
+          true
         else
-          @condition.wait(@mutex, timeout)
-        end
+          remaining = Condition::Result.new(timeout)
+          while generation.status == :waiting && remaining.can_wait?
+            remaining = @condition.wait(@mutex, remaining.remaining_time)
+          end
 
-        true
+          if remaining.woken_up?
+            return true
+          else
+            generation.status = :broken
+            @condition.broadcast
+            return false
+          end
+        end
       end
     end
 
@@ -52,6 +76,12 @@ def wait(timeout = nil)
     #
     # @return [nil]
     def reset
+      @mutex.synchronize do
+        @generation.status = :reset
+        @condition.broadcast
+        @generation = Generation.new(:waiting)
+        @number_waiting = 0
+      end
     end
 
     # A barrier can be broken when:
@@ -61,7 +91,7 @@ def reset
     # A broken barrier can be restored using `reset` it's safer to create a new one
     # @return [Boolean] true if the barrier is broken otherwise false
     def broken?
-      false
+      @mutex.synchronize { @generation.status != :waiting }
     end
 
   end

spec/concurrent/atomic/cyclic_barrier_spec.rb

@@ -58,14 +58,20 @@ module Concurrent
         barrier.reset
         barrier.broken?.should eq false
       end
-
-      it 'should be broken when at least one thread timed out'
-      it 'should be restored when reset is called'
     end
 
     describe 'reset' do
-      it 'should release all waiting threads'
-      it 'should not execute the block'
+      it 'should release all waiting threads' do
+        latch = CountDownLatch.new(1)
+
+        Thread.new { barrier.wait; latch.count_down }
+        sleep(0.1)
+        barrier.reset
+        latch.wait(0.1).should be_true
+
+        barrier.should_not be_broken
+        barrier.number_waiting.should eq 0
+      end
     end
 
     describe '#wait' do
@@ -83,6 +89,7 @@ module Concurrent
           parties.times { Thread.new { barrier.wait; latch.count_down } }
           latch.wait(0.1).should be_true
           barrier.number_waiting.should eq 0
+          barrier.should_not be_broken
         end
 
         it 'returns true when released' do
@@ -103,6 +110,16 @@ module Concurrent
 
           counter.value.should eq 1
         end
+
+        it 'can be reused' do
+          first_latch = CountDownLatch.new(parties)
+          parties.times { Thread.new { barrier.wait; first_latch.count_down } }
+          first_latch.wait(0.1).should be_true
+
+          latch = CountDownLatch.new(parties)
+          parties.times { Thread.new { barrier.wait; latch.count_down } }
+          latch.wait(0.1).should be_true
+        end
       end
 
       context 'with timeout' do
@@ -132,18 +149,91 @@ module Concurrent
 
         context 'timeout expiring' do
 
-          it 'returns false'
-          it 'can return early and break the barrier'
-          it 'does not execute the block on timeout'
+          it 'returns false' do
+            latch = CountDownLatch.new(1)
+
+            Thread.new { latch.count_down if barrier.wait(0.1) == false }
+            latch.wait(0.2).should be_true
+          end
+
+          it 'breaks the barrier and release all other threads' do
+            latch = CountDownLatch.new(2)
+
+            Thread.new { barrier.wait(0.1); latch.count_down }
+            Thread.new { barrier.wait; latch.count_down }
+
+            latch.wait(0.2).should be_true
+            barrier.should be_broken
+          end
+
+          it 'does not execute the block on timeout' do
+            counter = AtomicFixnum.new
+            barrier = described_class.new(parties) { counter.increment }
+
+            barrier.wait(0.1)
+
+            counter.value.should eq 0
+          end
         end
       end
-    end
 
-    context 'spurious wakeups' do
-      it 'should resist'
+      context '#broken barrier' do
+        it 'should not accept new threads' do
+          Thread.new { barrier.wait(0.1) }
+          sleep(0.2)
+
+          barrier.should be_broken
+
+          barrier.wait.should be_false
+        end
+
+        it 'can be reset' do
+          Thread.new { barrier.wait(0.1) }
+          sleep(0.2)
+
+          barrier.should be_broken
+
+          barrier.reset
+
+          barrier.should_not be_broken
+        end
+      end
     end
 
-  end
+    context 'spurious wake ups' do
+
+      before(:each) do
+        def barrier.simulate_spurious_wake_up
+          @mutex.synchronize do
+            @condition.signal
+            @condition.broadcast
+          end
+        end
+      end
+
+      it 'should resist to spurious wake ups without timeout' do
+        @expected = false
+        Thread.new { barrier.wait; @expected = true }
+
+        sleep(0.1)
+        barrier.simulate_spurious_wake_up
+
+        sleep(0.1)
+        @expected.should be_false
+      end
 
+      it 'should resist to spurious wake ups with timeout' do
+        @expected = false
+        Thread.new { barrier.wait(0.5); @expected = true }
+
+        sleep(0.1)
+        barrier.simulate_spurious_wake_up
+
+        sleep(0.1)
+        @expected.should be_false
+
+      end
+    end
+  end
 
 end