first working wait implementation

Author: mighe

lib/concurrent/atomic/cyclic_barrier.rb

@@ -8,12 +8,13 @@ class CyclicBarrier
     # @yield an optional block that will be executed that will be executed after the last thread arrives and before the others are released
     #
     # @raise [ArgumentError] if `parties` is not an integer or is less than zero
-    def initialize(parties)
+    def initialize(parties, &block)
       raise ArgumentError.new('count must be in integer greater than or equal zero') if !parties.is_a?(Fixnum) || parties < 1
       @parties = parties
       @mutex = Mutex.new
       @condition = Condition.new
       @number_waiting = 0
+      @action = block
     end
 
     # @return [Fixnum] the number of threads needed to pass the barrier
@@ -27,17 +28,21 @@ def number_waiting
     end
 
     # 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`
     def wait(timeout = nil)
       @mutex.synchronize do
         @number_waiting += 1
         if @number_waiting == @parties
+          @action.call if @action
           @condition.broadcast
           @number_waiting = 0
         else
-          @condition.wait(@mutex)
+          @condition.wait(@mutex, timeout)
         end
+
+        true
       end
     end
 
@@ -56,6 +61,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
     end
 
   end

spec/concurrent/atomic/cyclic_barrier_spec.rb

@@ -50,8 +50,15 @@ module Concurrent
     end
 
     describe '#broken?' do
-      it 'should not be broken when created'
-      it 'should not be broken when reset is called without waiting thread'
+      it 'should not be broken when created' do
+        barrier.broken?.should eq false
+      end
+
+      it 'should not be broken when reset is called without waiting thread' do
+        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
@@ -74,18 +81,61 @@ module Concurrent
           latch = CountDownLatch.new(parties)
 
           parties.times { Thread.new { barrier.wait; latch.count_down } }
-          latch.wait(0.2).should be_true
+          latch.wait(0.1).should be_true
           barrier.number_waiting.should eq 0
         end
 
-        it 'executes the block'
+        it 'returns true when released' do
+          latch = CountDownLatch.new(parties)
+
+          parties.times { Thread.new { latch.count_down if barrier.wait == true } }
+          latch.wait(0.1).should be_true
+        end
+
+        it 'executes the block once' do
+          counter = AtomicFixnum.new
+          barrier = described_class.new(parties) { counter.increment }
+
+          latch = CountDownLatch.new(parties)
+
+          parties.times { Thread.new { latch.count_down if barrier.wait == true } }
+          latch.wait(0.1).should be_true
+
+          counter.value.should eq 1
+        end
       end
 
       context 'with timeout' do
-        it 'should block the thread'
-        it 'should release all threads when their number matches the desired one'
-        it 'can return early and break the barrier'
-        it 'does not execute the block on timeout'
+        context 'timeout not expiring' do
+          it 'should block the thread' do
+            t = Thread.new { barrier.wait(1) }
+            sleep(0.1)
+
+            t.status.should eq 'sleep'
+          end
+
+          it 'should release all threads when their number matches the desired one' do
+            latch = CountDownLatch.new(parties)
+
+            parties.times { Thread.new { barrier.wait(1); latch.count_down } }
+            latch.wait(0.2).should be_true
+            barrier.number_waiting.should eq 0
+          end
+
+          it 'returns true when released' do
+            latch = CountDownLatch.new(parties)
+
+            parties.times { Thread.new { latch.count_down if barrier.wait(1) == true } }
+            latch.wait(0.1).should be_true
+          end
+        end
+
+        context 'timeout expiring' do
+
+          it 'returns false'
+          it 'can return early and break the barrier'
+          it 'does not execute the block on timeout'
+        end
       end
     end