Merge pull request #49 from jdantonio/channels

First channel implementation

Author: mighe

lib/concurrent.rb

@@ -32,6 +32,13 @@
 require 'concurrent/tvar'
 require 'concurrent/utilities'
 
+require 'concurrent/channel/probe'
+require 'concurrent/channel/channel'
+require 'concurrent/channel/unbuffered_channel'
+require 'concurrent/channel/buffered_channel'
+require 'concurrent/channel/ring_buffer'
+require 'concurrent/channel/blocking_ring_buffer'
+
 require 'concurrent/actor_context'
 require 'concurrent/simple_actor_ref'
 

lib/concurrent/channel/blocking_ring_buffer.rb

@@ -0,0 +1,60 @@
+module Concurrent
+  class BlockingRingBuffer
+
+    def initialize(capacity)
+      @buffer = RingBuffer.new(capacity)
+      @first = @last = 0
+      @count = 0
+      @mutex = Mutex.new
+      @condition = Condition.new
+    end
+
+    def capacity
+      @mutex.synchronize { @buffer.capacity }
+    end
+
+    def count
+      @mutex.synchronize { @buffer.count }
+    end
+
+    def full?
+      @mutex.synchronize { @buffer.full? }
+    end
+
+    def empty?
+      @mutex.synchronize { @buffer.empty? }
+    end
+
+    def put(value)
+      @mutex.synchronize do
+        wait_while_full
+        @buffer.offer(value)
+        @condition.signal
+      end
+    end
+
+    def take
+      @mutex.synchronize do
+        wait_while_empty
+        result = @buffer.poll
+        @condition.signal
+        result
+      end
+    end
+
+    def peek
+      @mutex.synchronize { @buffer.peek }
+    end
+
+    private
+
+    def wait_while_full
+      @condition.wait(@mutex) while @buffer.full?
+    end
+
+    def wait_while_empty
+      @condition.wait(@mutex) while @buffer.empty?
+    end
+
+  end
+end

lib/concurrent/channel/buffered_channel.rb

@@ -0,0 +1,83 @@
+require_relative 'waitable_list'
+
+module Concurrent
+  class BufferedChannel
+
+    def initialize(size)
+      @mutex = Mutex.new
+      @condition = Condition.new
+      @buffer_condition = Condition.new
+
+      @probe_set = WaitableList.new
+      @buffer = RingBuffer.new(size)
+    end
+
+    def probe_set_size
+      @probe_set.size
+    end
+
+    def buffer_queue_size
+      @mutex.synchronize { @buffer.count }
+    end
+
+    def push(value)
+      until set_probe_or_push_into_buffer(value)
+      end
+    end
+
+    def pop
+      probe = Probe.new
+      select(probe)
+      probe.value
+    end
+
+    def select(probe)
+      @mutex.synchronize do
+
+        if @buffer.empty?
+          @probe_set.put(probe)
+          true
+        else
+          shift_buffer if probe.set_unless_assigned peek_buffer
+        end
+
+      end
+    end
+
+    def remove_probe(probe)
+      @probe_set.delete(probe)
+    end
+
+    private
+
+    def push_into_buffer(value)
+      @buffer_condition.wait(@mutex) while @buffer.full?
+      @buffer.offer value
+      @buffer_condition.broadcast
+    end
+
+    def peek_buffer
+      @buffer_condition.wait(@mutex) while @buffer.empty?
+      @buffer.peek
+    end
+
+    def shift_buffer
+      @buffer_condition.wait(@mutex) while @buffer.empty?
+      result = @buffer.poll
+      @buffer_condition.broadcast
+      result
+    end
+
+    def set_probe_or_push_into_buffer(value)
+      @mutex.synchronize do
+        if @probe_set.empty?
+          push_into_buffer(value)
+          true
+        else
+          @probe_set.take.set_unless_assigned(value)
+        end
+      end
+    end
+
+  end
+end

lib/concurrent/channel/channel.rb

@@ -0,0 +1,11 @@
+module Concurrent
+  class Channel
+    def self.select(*channels)
+      probe = Probe.new
+      channels.each { |channel| channel.select(probe) }
+      result = probe.value
+      channels.each { |channel| channel.remove_probe(probe) }
+      result
+    end
+  end
+end

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/ring_buffer.rb

@@ -0,0 +1,54 @@
+module Concurrent
+
+  # not thread safe buffer
+  class RingBuffer
+
+    def initialize(capacity)
+      @buffer = Array.new(capacity)
+      @first = @last = 0
+      @count = 0
+    end
+
+    def capacity
+      @buffer.size
+    end
+
+    def count
+      @count
+    end
+
+    def empty?
+      @count == 0
+    end
+
+    def full?
+      @count == capacity
+    end
+
+    # @param [Object] value
+    # @return [Boolean] true if value has been inserted, false otherwise
+    def offer(value)
+      return false if full?
+
+      @buffer[@last] = value
+      @last = (@last + 1) % @buffer.size
+      @count += 1
+      true
+    end
+
+    # @return [Object] the first available value and removes it from the buffer. If buffer is empty returns nil
+    def poll
+      result = @buffer[@first]
+      @buffer[@first] = nil
+      @first = (@first + 1) % @buffer.size
+      @count -= 1
+      result
+    end
+
+    # @return [Object] the first available value and without removing it from the buffer. If buffer is empty returns nil
+    def peek
+      @buffer[@first]
+    end
+
+  end
+end

lib/concurrent/channel/unbuffered_channel.rb

@@ -0,0 +1,34 @@
+require_relative 'waitable_list'
+
+module Concurrent
+  class UnbufferedChannel
+
+    def initialize
+      @probe_set = WaitableList.new
+    end
+
+    def probe_set_size
+      @probe_set.size
+    end
+
+    def push(value)
+      until @probe_set.take.set_unless_assigned(value)
+      end
+    end
+
+    def pop
+      probe = Probe.new
+      select(probe)
+      probe.value
+    end
+
+    def select(probe)
+      @probe_set.put(probe)
+    end
+
+    def remove_probe(probe)
+      @probe_set.delete(probe)
+    end
+
+  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 put(value)
+      @mutex.synchronize do
+        @list << value
+        @condition.signal
+      end
+    end
+
+    def delete(value)
+      @mutex.synchronize { @list.delete(value) }
+    end
+
+    def take
+      @mutex.synchronize do
+        @condition.wait(@mutex) while @list.empty?
+        @list.shift
+      end
+    end
+
+  end
+end