added support for expired probes

Author: mighe

lib/concurrent.rb

@@ -32,6 +32,7 @@
 require 'concurrent/tvar'
 require 'concurrent/utilities'
 
+require 'concurrent/channel/probe'
 require 'concurrent/channel/unbuffered_channel'
 
 require 'concurrent/cached_thread_pool'

lib/concurrent/channel/probe.rb

@@ -0,0 +1,19 @@
+module Concurrent
+  class Probe < IVar
+
+    def initialize(value = NO_VALUE, opts = {})
+      super(value, opts)
+    end
+
+    def set_unless_assigned(value)
+      mutex.synchronize do
+        return false if [:fulfilled, :rejected].include? @state
+
+        set_state(true, value, nil)
+        event.set
+        true
+      end
+
+    end
+  end
+end

lib/concurrent/channel/unbuffered_channel.rb

@@ -9,30 +9,33 @@ def initialize
     end
 
     def push(value)
-      probe = @mutex.synchronize do
-        @condition.wait(@mutex) while @wait_set.empty?
-        @wait_set.shift
+      until first_waiting_probe.set_unless_assigned(value)
       end
-
-      probe.set(value)
     end
 
     def pop
-      probe = IVar.new
+      probe = Probe.new
+      select(probe)
+      probe.value
+    end
 
+    def select(probe)
       @mutex.synchronize do
         @wait_set << probe
         @condition.signal
       end
-
-      probe.value
-    end
-
-    def select(probe)
     end
 
     def remove_probe(probe)
     end
 
+    private
+      def first_waiting_probe
+        @mutex.synchronize do
+          @condition.wait(@mutex) while @wait_set.empty?
+          @wait_set.shift
+        end
+      end
+
   end
 end

spec/concurrent/channel/probe_spec.rb

@@ -0,0 +1,8 @@
+require 'spec_helper'
+
+module Concurrent
+
+  describe Probe do
+    it 'should be written'
+  end
+end

spec/concurrent/channel/unbuffered_channel_spec.rb

@@ -31,6 +31,7 @@ module Concurrent
     end
 
     context 'cooperating threads' do
+
       it 'passes the pushed value to thread waiting on pop' do
         result = nil
 
@@ -41,7 +42,60 @@ module Concurrent
 
         result.should eq 42
       end
+
+      it 'passes the pushed value to only one thread' do
+        result = []
+
+        Thread.new { channel.push 37 }
+        Thread.new { result << channel.pop }
+        Thread.new { result << channel.pop }
+
+        sleep(0.05)
+
+        result.should have(1).items
+      end
     end
 
+    describe 'select' do
+
+      let(:probe) { Probe.new }
+
+      it 'does not block' do
+        t = Thread.new { channel.select(probe) }
+
+        sleep(0.05)
+
+        t.status.should eq false
+      end
+
+      it 'gets notified by writer thread' do
+        channel.select(probe)
+
+        Thread.new { channel.push 82 }
+
+        probe.value.should eq 82
+      end
+
+      it 'ignores already set probes and waits for a new one' do
+        probe.set(27)
+
+        channel.select(probe)
+
+        t = Thread.new { channel.push 72 }
+
+        sleep(0.05)
+
+        t.status.should eq 'sleep'
+
+        new_probe = Probe.new
+
+        channel.select(new_probe)
+
+        sleep(0.05)
+
+        new_probe.value.should eq 72
+      end
+
+    end
   end
 end