add test for grain size

[concurrent.git] / lib / concurrent / futures.rb

#
# concurrent/futures - futures and lazy evaluation for Ruby
#
# Copyright (C) 2007  MenTaLguY <mental@rydia.net>
#
# This file is made available under the same terms as Ruby.
#

require 'concurrent/primitives'

module Concurrent
module Futures

# Encapsulates an exception raised asynchronously during evaluation of
# a future; this encapsulation is necessary to avoid ambiguity with similar
# exceptions synchronously raised.
#
class AsyncError < RuntimeError
  attr_reader :reason # the asynchronously raised exception

  def initialize( reason, desc=nil )
    super( desc )
    @reason = reason
  end
end

require 'concurrent/futures.so' # provides Thunk

# Raised in response to attempts to fulfill an already fulfilled promise
class AlreadyFulfilledError < StandardError
end

# A Promise is a cell which can be set to a value exactly once: initially,
# it is empty, and may be either "fulfilled" with a value or "failed" with
# an exception.
#
# <b>Scheduling</b>: Unfair
#
class Promise
  # A future which evaluates to the fulfilled value of the promise
  attr_reader :future

  class Value < Struct.new :value #:nodoc:
  end

  class Failure #:nodoc:
    def initialize( exception )
      @exception = exception
    end

    def value
      raise @exception
    end
  end

  # Creates a new unfulfilled promise
  def initialize
    @lock = Primitives::Semaphore.new 1
    @ready = Primitives::Latch.new
    @fulfilled = Primitives::Volatile.new
    @future = Thunk.new self
  end

  # Returns true if the promise has been fulfilled or failed, false otherwise.
  #
  # <b>Can Block</b>: No
  #
  def fulfilled?
    !@fulfilled.value.nil?
  end

  # Returns the value the promise was fulfilled with, raises the exception
  # it was failed with, and otherwise waits for the promise to be fulfilled.
  #
  # <b>Can Block</b>: Yes, if the promise is unfulfilled
  #
  def value
    value = @fulfilled.value
    @ready.wait unless value
    value.value
  end

  def fulfill_inner
    @lock.synchronize do
      if @fulfilled.value
        raise AlreadyFulfilledError, "promise already fulfilled"
      end
      @fulfilled.value = yield
      @ready.set
    end
    self
  end
  private :fulfill_inner

  # Fulfills the promise with +value+, or raises AlreadyFulfilledError if it
  # has already been fulfilled or failed.
  #
  # <b>Can Block</b>: Yes
  #
  def fulfill( value )
    fulfill_inner { Value.new value }
    self
  end

  # Fails the promise with +exception+, or raises AlreadyFulfilledError if it
  # has already been fulfilled or failed.
  #
  # <b>Can Block</b>: Yes
  #
  def fail( exception )
    fulfill_inner { Failure.new exception }
    self
  end
end

class Lazy #:nodoc:
  def initialize( &block )
    @block = block
    @promise = Promise.new
    @value = Primitives::Atomic.new nil
  end

  def value
    if @value.set_if_equal( nil, @promise )
      @promise.fulfill Future.async( &@block )
      @promise = nil
    end
    @value.value.future
  end

  def inspect
    "#<Lazy #{ @block.inspect }>"
  end
end

# A single-value cell similar to Primitives::Atomic except it has no
# compare-and-set operation, but the added ability to atomically assign
# a new value which is based on the old value.  See Ref#modify.
#
# <b>Scheduling</b>: Unfair
#
class Ref
  def initialize( value=nil )
    @value = Primitives::Atomic.new value
  end

  # Returns the cell's current value
  #
  # <b>Can Block</b>: No
  #
  def value ; @value.value ; end

  # Assigns +new_value+ to the cell, returning +new_value+.
  #
  # <b>Can Block</b>: No
  #
  def value=( new_value )
    @value.value = new_value
  end

  # Swaps the old value for a new one, returning the old value.
  #
  # See Primitives::Atomic#swap
  #
  # <b>Can Block</b>: No
  #
  def exchange( new_value )
    @value.swap( new_value )
  end
  alias swap exchange

  # Modifies the existing value, passing it to a block whose result will
  # become the new value.
  #
  # <b>Can Block</b>: Yes
  #
  def modify( &block )
    promise = Promise.new
    old_value = @value.swap( promise.future )
    promise.fulfill Future.async { block.call old_value }
    old_value
  end
end

# Futures are placeholders for the result of a pending computation.  When
# the computation is complete, they become nearly indistinguishable from the
# computation's result object.
#
# Calling a method on a future belonging to an unfinished computation will
# block until the computation completes.
#
# Once a computation is finished, the future will delegate the method call
# to the result object, or re-raise the exception that terminated the
# computation, wrapping it in AsyncError.  Subsequent method calls do not
# block, but delegate or raise immediately.
#
# Futures let you take a "buy now, pay later" approach to computation: you
# can fire off a background computation, get a handle (a future) for its
# result immediately, and nothing will have to wait for that computation to
# finish until or unless someone tries to call a method on its result.
#
# One caveat: a future will always appear to be true to Ruby, even if the
# result object is false or nil.  This can be worked around by using
# Future.await to unwrap the computation's real result object.
#
# <b>Scheduling</b>: Unfair
# <b>Write Visibility</b>: All writes by the computation become visible to a requesting thread
# <b>Blocking</b>: Yes, if the computation is not complete
#
class Future
  class << self
    undef allocate
    undef new

    # Spawns +block+ in a new thread, immediately returning a future for
    # its result.
    #
    # <b>Can Block</b>: Yes
    #
    def async( &block )
      Thunk.new( Thread.new( &block ) )
    end
    alias spawn async
    alias future async
 
    # Lazily spawns +block+ in a new thread, immediately returning a future
    # but waiting to actually spawn the block until the future's value is
    # required.
    #
    # <b>Can Block</b>: Yes
    #
    def lazy( &block )
      Thunk.new( Lazy.new( &block ) )
    end

    # Waits for the future's computation to finish, and then returns
    # its unwrapped result (or raises the exception which terminated it).
    # Calling it on an ordinary object will simply return that object.
    # 
    # <b>Write Visibility</b>: All writes by the computation become visible to the calling thread
    # <b>Can Block</b>: Yes, if the computation is unfinished
    #
    def await( future )
      Thunk.value future
    end
  end
end

end
end