first RingBuffer implementation

Author: mighe

lib/concurrent.rb

@@ -36,6 +36,7 @@
 require 'concurrent/channel/channel'
 require 'concurrent/channel/unbuffered_channel'
 require 'concurrent/channel/buffered_channel'
+require 'concurrent/channel/ring_buffer'
 
 require 'concurrent/cached_thread_pool'
 require 'concurrent/fixed_thread_pool'

lib/concurrent/channel/ring_buffer.rb

@@ -0,0 +1,53 @@
+module Concurrent
+  class RingBuffer
+
+    def initialize(capacity)
+      @buffer = Array.new(capacity)
+      @first = @last = 0
+      @count = 0
+      @mutex = Mutex.new
+      @condition = Condition.new
+    end
+
+    def capacity
+      @mutex.synchronize { @buffer.size }
+    end
+
+    def count
+      @mutex.synchronize { @count }
+    end
+
+    def put(value)
+      @mutex.synchronize do
+        wait_while_full
+        @buffer[@last] = value
+        @last += 1
+        @count += 1
+        @condition.signal
+      end
+    end
+
+    def take
+      @mutex.synchronize do
+        wait_while_empty
+        result = @buffer[@first]
+        @buffer[@first] = nil
+        @first += 1
+        @count -= 1
+        @condition.signal
+        result
+      end
+    end
+
+    private
+
+    def wait_while_full
+      @condition.wait(@mutex) while @count == @buffer.size
+    end
+
+    def wait_while_empty
+      @condition.wait(@mutex) while @count == 0
+    end
+
+  end
+end

spec/concurrent/channel/ring_buffer_spec.rb

@@ -0,0 +1,89 @@
+require 'spec_helper'
+
+module Concurrent
+
+  describe RingBuffer do
+
+    let(:capacity) { 3 }
+    let(:buffer) { RingBuffer.new( capacity ) }
+
+    describe '#capacity' do
+      it 'returns the value passed in constructor' do
+        buffer.capacity.should eq capacity
+      end
+    end
+
+    describe '#count' do
+      it 'is zero when created' do
+        buffer.count.should eq 0
+      end
+
+      it 'increases when an element is added' do
+        buffer.put 5
+        buffer.count.should eq 1
+
+        buffer.put 1
+        buffer.count.should eq 2
+      end
+
+      it 'decreases when an element is removed' do
+        buffer.put 10
+
+        buffer.take
+
+        buffer.count.should eq 0
+      end
+    end
+
+    describe '#put' do
+      it 'block when buffer is full' do
+        capacity.times { buffer.put 27 }
+
+        t = Thread.new { buffer.put 32 }
+
+        sleep(0.1)
+
+        t.status.should eq 'sleep'
+      end
+
+      it 'continues when an element is removed' do
+        latch = CountDownLatch.new(1)
+
+        Thread.new { (capacity + 1).times { buffer.put 'hi' }; latch.count_down }
+        Thread.new { sleep(0.1); buffer.take }
+
+        latch.wait(0.2).should be_true
+      end
+    end
+
+    describe '#take' do
+      it 'blocks when buffer is empty' do
+        t = Thread.new { buffer.take }
+
+        sleep(0.1)
+
+        t.status.should eq 'sleep'
+      end
+
+      it 'continues when an element is added' do
+        latch = CountDownLatch.new(1)
+
+        Thread.new { buffer.take; latch.count_down }
+        Thread.new { sleep(0.1); buffer.put 3 }
+
+        latch.wait(0.2).should be_true
+      end
+
+      it 'returns the first added value' do
+        buffer.put 'hi'
+        buffer.put 'foo'
+        buffer.put 'bar'
+
+        buffer.take.should eq 'hi'
+        buffer.take.should eq 'foo'
+        buffer.take.should eq 'bar'
+      end
+    end
+
+  end
+end