added waitable list

Author: mighe

lib/concurrent/channel/buffered_channel.rb

@@ -1,3 +1,5 @@
+require_relative 'waitable_list'
+
 module Concurrent
   class BufferedChannel
 
@@ -6,13 +8,13 @@ def initialize(size)
       @condition = Condition.new
       @buffer_condition = Condition.new
 
-      @probe_set = []
+      @probe_set = WaitableList.new
       @buffer = []
       @size = size
     end
 
     def probe_set_size
-      @mutex.synchronize { @probe_set.size }
+      @probe_set.size
     end
 
     def buffer_queue_size
@@ -34,7 +36,7 @@ def select(probe)
       @mutex.synchronize do
 
         if @buffer.empty?
-          @probe_set << probe
+          @probe_set.push(probe)
           true
         else
           shift_buffer if probe.set_unless_assigned peek_buffer
@@ -44,7 +46,7 @@ def select(probe)
     end
 
     def remove_probe(probe)
-      @mutex.synchronize { @probe_set.delete(probe) }
+      @probe_set.delete(probe)
     end
 
     private
@@ -81,7 +83,7 @@ def set_probe_or_push_into_buffer(value)
           push_into_buffer(value)
           true
         else
-          @probe_set.shift.set_unless_assigned(value)
+          @probe_set.first.set_unless_assigned(value)
         end
       end
     end

lib/concurrent/channel/unbuffered_channel.rb

@@ -1,19 +1,18 @@
+require_relative 'waitable_list'
+
 module Concurrent
   class UnbufferedChannel
 
     def initialize
-      @mutex = Mutex.new
-      @condition = Condition.new
-
-      @probe_set = []
+      @probe_set = WaitableList.new
     end
 
     def probe_set_size
-      @mutex.synchronize { @probe_set.size }
+      @probe_set.size
     end
 
     def push(value)
-      until first_waiting_probe.set_unless_assigned(value)
+      until @probe_set.first.set_unless_assigned(value)
       end
     end
 
@@ -24,22 +23,11 @@ def pop
     end
 
     def select(probe)
-      @mutex.synchronize do
-        @probe_set << probe
-        @condition.signal
-      end
+      @probe_set.push(probe)
     end
 
     def remove_probe(probe)
-      @mutex.synchronize { @probe_set.delete(probe) }
-    end
-
-    private
-    def first_waiting_probe
-      @mutex.synchronize do
-        @condition.wait(@mutex) while @probe_set.empty?
-        @probe_set.shift
-      end
+      @probe_set.delete(probe)
     end
 
   end

lib/concurrent/channel/waitable_list.rb

@@ -0,0 +1,38 @@
+module Concurrent
+  class WaitableList
+
+    def initialize
+      @mutex = Mutex.new
+      @condition = Condition.new
+
+      @list = []
+    end
+
+    def size
+      @mutex.synchronize { @list.size }
+    end
+
+    def empty?
+      @mutex.synchronize { @list.empty? }
+    end
+
+    def push(value)
+      @mutex.synchronize do
+        @list << value
+        @condition.signal
+      end
+    end
+
+    def delete(value)
+      @mutex.synchronize { @list.delete(value) }
+    end
+
+    def first
+      @mutex.synchronize do
+        @condition.wait(@mutex) while @list.empty?
+        @list.shift
+      end
+    end
+
+  end
+end