Merge branch 'master' into refactor/tasks

Author: jdantonio

lib/concurrent/atomic/atomic_fixnum.rb

@@ -1,38 +1,87 @@
 module Concurrent
 
-  # @!visibility private
-  module MutexAtomicFixnum # :nodoc:
+  # @!macro [attach] atomic_fixnum
+  #
+  #   A numeric value that can be updated atomically. Reads and writes to an atomic
+  #   fixnum and thread-safe and guaranteed to succeed. Reads and writes may block
+  #   briefly but no explicit locking is required.
+  #
+  #   @since 0.5.0
+  #   @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicLong.html java.util.concurrent.atomic.AtomicLong
+  class MutexAtomicFixnum
 
-    def allocate_storage(init)
+    # @!macro [attach] atomic_fixnum_method_initialize
+    #
+    # Creates a new `AtomicFixnum` with the given initial value.
+    #
+    # @param [Fixnum] init the initial value
+    # @raise [ArgumentError] if the initial value is not a `Fixnum`
+    def initialize(init = 0)
+      raise ArgumentError.new('initial value must be a Fixnum') unless init.is_a?(Fixnum)
       @value = init
       @mutex = Mutex.new
     end
 
+    # @!macro [attach] atomic_fixnum_method_value
+    #
+    #   Retrieves the current `Fixnum` value.
+    #
+    #   @return [Fixnum] the current value
     def value
       @mutex.synchronize do
         @value
       end
     end
 
+    # @!macro [attach] atomic_fixnum_method_value_eq
+    #
+    #   Explicitly sets the value.
+    #
+    #   @param [Fixnum] value the new value to be set
+    #
+    #   @return [Fixnum] the current value
+    #
+    #   @raise [ArgumentError] if the new value is not a `Fixnum`
     def value=(value)
       raise ArgumentError.new('value must be a Fixnum') unless value.is_a?(Fixnum)
       @mutex.synchronize do
         @value = value
       end
     end
 
+    # @!macro [attach] atomic_fixnum_method_increment
+    #
+    #   Increases the current value by 1.
+    #
+    #   @return [Fixnum] the current value after incrementation
     def increment
       @mutex.synchronize do
         @value += 1
       end
     end
+    alias_method :up, :increment
 
+    # @!macro [attach] atomic_fixnum_method_decrement
+    #
+    #   Decreases the current value by 1.
+    #
+    #   @return [Fixnum] the current value after decrementation
     def decrement
       @mutex.synchronize do
         @value -= 1
       end
     end
+    alias_method :down, :decrement
 
+    # @!macro [attach] atomic_fixnum_method_compare_and_set
+    # 
+    #   Atomically sets the value to the given updated value if the current
+    #   value == the expected value.
+    #
+    #   @param [Fixnum] expect the expected value
+    #   @param [Fixnum] update the new value
+    #
+    #   @return [Boolean] true if the value was updated else false 
     def compare_and_set(expect, update)
       @mutex.synchronize do
         if @value == expect
@@ -45,77 +94,60 @@ def compare_and_set(expect, update)
     end
   end
 
-  # @!visibility private
-  module JavaAtomicFixnum # :nodoc:
+  if RUBY_PLATFORM == 'java'
 
-    def allocate_storage(init)
-      @atomic = java.util.concurrent.atomic.AtomicLong.new(init)
-    end
+    # @!macro atomic_fixnum
+    class JavaAtomicFixnum
 
-    def value
-      @atomic.get
-    end
-
-    def value=(value)
-      raise ArgumentError.new('value must be a Fixnum') unless value.is_a?(Fixnum)
-      @atomic.set(value)
-    end
+      # @!macro atomic_fixnum_method_initialize
+      #
+      def initialize(init = 0)
+        raise ArgumentError.new('initial value must be a Fixnum') unless init.is_a?(Fixnum)
+        @atomic = java.util.concurrent.atomic.AtomicLong.new(init)
+      end
 
-    def increment
-      @atomic.increment_and_get
-    end
+      # @!macro atomic_fixnum_method_value
+      #
+      def value
+        @atomic.get
+      end
 
-    def decrement
-      @atomic.decrement_and_get
-    end
+      # @!macro atomic_fixnum_method_value_eq
+      #
+      def value=(value)
+        raise ArgumentError.new('value must be a Fixnum') unless value.is_a?(Fixnum)
+        @atomic.set(value)
+      end
 
-    def compare_and_set(expect, update)
-      @atomic.compare_and_set(expect, update)
-    end
-  end
+      # @!macro atomic_fixnum_method_increment
+      #
+      def increment
+        @atomic.increment_and_get
+      end
+      alias_method :up, :increment
 
-  # A numeric value that can be updated atomically. Reads and writes to an atomic
-  # fixnum and thread-safe and guaranteed to succeed. Reads and writes may block
-  # briefly but no explicit locking is required.
-  #
-  # @!method value()
-  #   Retrieves the current `Fixnum` value
-  #   @return [Fixnum] the current value
-  #
-  # @!method value=(value)
-  #   Explicitly sets the value
-  #   @param [Fixnum] value the new value to be set
-  #   @return [Fixnum] the current value
-  #   @raise [ArgumentError] if the new value is not a `Fixnum`
-  #
-  # @!method increment()
-  #   Increases the current value by 1
-  #   @return [Fixnum] the current value after incrementation
-  #
-  # @!method decrement()
-  #   Decreases the current value by 1
-  #   @return [Fixnum] the current value after decrementation
-  #
-  # @since 0.5.0
-  # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicLong.html java.util.concurrent.atomic.AtomicLong
-  class AtomicFixnum
+      # @!macro atomic_fixnum_method_decrement
+      #
+      def decrement
+        @atomic.decrement_and_get
+      end
+      alias_method :down, :decrement
 
-    # Creates a new `AtomicFixnum` with the given initial value.
-    #
-    # @param [Fixnum] init the initial value
-    # @raise [ArgumentError] if the initial value is not a `Fixnum`
-    def initialize(init = 0)
-      raise ArgumentError.new('initial value must be a Fixnum') unless init.is_a?(Fixnum)
-      allocate_storage(init)
+      # @!macro atomic_fixnum_method_compare_and_set
+      #
+      def compare_and_set(expect, update)
+        @atomic.compare_and_set(expect, update)
+      end
     end
 
-    if RUBY_PLATFORM == 'java'
-      include JavaAtomicFixnum
-    else
-      include MutexAtomicFixnum
+    # @!macro atomic_fixnum
+    class AtomicFixnum < JavaAtomicFixnum
     end
 
-    alias_method :up, :increment
-    alias_method :down, :decrement
+  else
+
+    # @!macro atomic_fixnum
+    class AtomicFixnum < MutexAtomicFixnum
+    end
   end
 end

lib/concurrent/atomic/count_down_latch.rb

@@ -2,20 +2,24 @@
 
 module Concurrent
 
-  # A synchronization object that allows one thread to wait on multiple other threads.
-  # The thread that will wait creates a `CountDownLatch` and sets the initial value
-  # (normally equal to the number of other threads). The initiating thread passes the
-  # latch to the other threads then waits for the other threads by calling the `#wait`
-  # method. Each of the other threads calls `#count_down` when done with its work.
-  # When the latch counter reaches zero the waiting thread is unblocked and continues
-  # with its work. A `CountDownLatch` can be used only once. Its value cannot be reset.
-  class CountDownLatch
+  # @!macro [attach] count_down_latch
+  #
+  #   A synchronization object that allows one thread to wait on multiple other threads.
+  #   The thread that will wait creates a `CountDownLatch` and sets the initial value
+  #   (normally equal to the number of other threads). The initiating thread passes the
+  #   latch to the other threads then waits for the other threads by calling the `#wait`
+  #   method. Each of the other threads calls `#count_down` when done with its work.
+  #   When the latch counter reaches zero the waiting thread is unblocked and continues
+  #   with its work. A `CountDownLatch` can be used only once. Its value cannot be reset.
+  class MutexCountDownLatch
 
-    # Create a new `CountDownLatch` with the initial `count`.
+    # @!macro [attach] count_down_latch_method_initialize
     #
-    # @param [Fixnum] count the initial count
+    #   Create a new `CountDownLatch` with the initial `count`.
     #
-    # @raise [ArgumentError] if `count` is not an integer or is less than zero
+    #   @param [Fixnum] count the initial count
+    #
+    #   @raise [ArgumentError] if `count` is not an integer or is less than zero
     def initialize(count)
       unless count.is_a?(Fixnum) && count >= 0
         raise ArgumentError.new('count must be in integer greater than or equal zero')
@@ -25,11 +29,13 @@ def initialize(count)
       @count = count
     end
 
-    # Block on the latch until the counter reaches zero or until `timeout` is reached.
+    # @!macro [attach] count_down_latch_method_wait
+    #
+    #   Block on the latch until the counter reaches zero or until `timeout` is reached.
     #
-    # @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`
+    #   @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`
     def wait(timeout = nil)
       @mutex.synchronize do
 
@@ -42,21 +48,69 @@ def wait(timeout = nil)
       end
     end
 
-    # Signal the latch to decrement the counter. Will signal all blocked threads when
-    # the `count` reaches zero.
+    # @!macro [attach] count_down_latch_method_count_down
+    #
+    #   Signal the latch to decrement the counter. Will signal all blocked threads when
+    #   the `count` reaches zero.
     def count_down
       @mutex.synchronize do
         @count -= 1 if @count > 0
         @condition.broadcast if @count == 0
       end
     end
 
-    # The current value of the counter.
+    # @!macro [attach] count_down_latch_method_count
     #
-    # @return [Fixnum] the current value of the counter
+    #   The current value of the counter.
+    #
+    #   @return [Fixnum] the current value of the counter
     def count
       @mutex.synchronize { @count }
     end
+  end
+
+  if RUBY_PLATFORM == 'java'
+
+    # @!macro count_down_latch
+    class JavaCountDownLatch
+
+      # @!macro count_down_latch_method_initialize
+      def initialize(count)
+        unless count.is_a?(Fixnum) && count >= 0
+          raise ArgumentError.new('count must be in integer greater than or equal zero')
+        end
+        @latch = java.util.concurrent.CountDownLatch.new(count)
+      end
+
+      # @!macro count_down_latch_method_wait
+      def wait(timeout = nil)
+        if timeout.nil?
+          @latch.await
+          true
+        else
+          @latch.await(1000 * timeout, java.util.concurrent.TimeUnit::MILLISECONDS)
+        end
+      end
+
+      # @!macro count_down_latch_method_count_down
+      def count_down
+        @latch.countDown
+      end
+
+      # @!macro count_down_latch_method_count
+      def count
+        @latch.getCount
+      end
+    end
+
+    # @!macro count_down_latch
+    class CountDownLatch < JavaCountDownLatch
+    end
 
+  else
+
+    # @!macro count_down_latch
+    class CountDownLatch < MutexCountDownLatch
+    end
   end
 end

lib/concurrent/channels.rb

@@ -1,5 +1,5 @@
+require 'concurrent/collections'
+
 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'

lib/concurrent/channel/blocking_ring_buffer.rb renamed to lib/concurrent/collection/blocking_ring_buffer.rb

@@ -11,30 +11,38 @@ def initialize(capacity)
       @condition = Condition.new
     end
 
+    # @return [Integer] the capacity of the buffer
     def capacity
       @mutex.synchronize { @buffer.capacity }
     end
 
+    # @return [Integer] the number of elements currently in the buffer
     def count
       @mutex.synchronize { @buffer.count }
     end
 
-    def full?
-      @mutex.synchronize { @buffer.full? }
-    end
-
+    # @return [Boolean] true if buffer is empty, false otherwise
     def empty?
       @mutex.synchronize { @buffer.empty? }
     end
 
+    # @return [Boolean] true if buffer is full, false otherwise
+    def full?
+      @mutex.synchronize { @buffer.full? }
+    end
+
+    # @param [Object] value. This methods blocks until an empty slot is available
+    # @return [Boolean] true if value has been inserted, false otherwise
     def put(value)
       @mutex.synchronize do
         wait_while_full
         @buffer.offer(value)
         @condition.signal
+        true
       end
     end
 
+    # @return [Object] the first available value and removes it from the buffer. If buffer is empty it blocks until an element is available
     def take
       @mutex.synchronize do
         wait_while_empty
@@ -44,6 +52,7 @@ def take
       end
     end
 
+    # @return [Object] the first available value and without removing it from the buffer. If buffer is empty returns nil
     def peek
       @mutex.synchronize { @buffer.peek }
     end