#2835

Committed 05 Oct 2014 10:16PM UTC coverage: 45.201% (-49.6%) from 94.81%

Build # #2835

Build Type

push

Committed by

Commit Message

Merge pull request #158 from obrok/promise-composition

Promise composition

Run Details

2 of 15 new or added lines in 1 file covered. (13.33%)

1514 existing lines in 84 files now uncovered.

1375 of 3042 relevant lines covered (45.2%)

0.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

52.17

/lib/concurrent/future.rb

require 'thread'

require 'concurrent/options_parser'
require 'concurrent/ivar'
require 'concurrent/executor/safe_task_executor'

module Concurrent

  # A `Future` represents a promise to complete an action at some time in the future.
  # The action is atomic and permanent. The idea behind a future is to send an operation
  # for asynchronous completion, do other stuff, then return and retrieve the result
  # of the async operation at a later time.
  #
  # A `Future` has four possible states: *:unscheduled*, *:pending*, *:rejected*, or *:fulfilled*.
  # When a `Future` is created its state is set to *:unscheduled*. Once the `#execute` method is
  # called the state becomes *:pending* and will remain in that state until processing is
  # complete. A completed `Future` is either *:rejected*, indicating that an exception was
  # thrown during processing, or *:fulfilled*, indicating success. If a `Future` is *:fulfilled*
  # its `value` will be updated to reflect the result of the operation. If *:rejected* the
  # `reason` will be updated with a reference to the thrown exception. The predicate methods
  # `#unscheduled?`, `#pending?`, `#rejected?`, and `fulfilled?` can be called at any time to
  # obtain the state of the `Future`, as can the `#state` method, which returns a symbol. 
  #
  # Retrieving the value of a `Future` is done through the `#value` (alias: `#deref`) method.
  # Obtaining the value of a `Future` is a potentially blocking operation. When a `Future` is
  # *:rejected* a call to `#value` will return `nil` immediately. When a `Future` is
  # *:fulfilled* a call to `#value` will immediately return the current value. When a
  # `Future` is *:pending* a call to `#value` will block until the `Future` is either
  # *:rejected* or *:fulfilled*. A *timeout* value can be passed to `#value` to limit how
  # long the call will block. If `nil` the call will block indefinitely. If `0` the call will
  # not block. Any other integer or float value will indicate the maximum number of seconds to block.
  #
  # The `Future` class also includes the behavior of the Ruby standard library `Observable` module,
  # but does so in a thread-safe way. On fulfillment or rejection all observers will be notified
  # according to the normal `Observable` behavior. The observer callback function will be called
  # with three parameters: the `Time` of fulfillment/rejection, the final `value`, and the final
  # `reason`. Observers added after fulfillment/rejection will still be notified as normal.
  #
  # @see http://ruby-doc.org/stdlib-2.1.1/libdoc/observer/rdoc/Observable.html Ruby Observable module
  # @see http://clojuredocs.org/clojure_core/clojure.core/future Clojure's future function
  # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Future.html java.util.concurrent.Future
  class Future < IVar

    # Create a new `Future` in the `:unscheduled` state.
    #
    # @yield the asynchronous operation to perform
    #
    # @param [Hash] opts the options controlling how the future will be processed
    # @option opts [Boolean] :operation (false) when `true` will execute the future on the global
    #   operation pool (for long-running operations), when `false` will execute the future on the
    #   global task pool (for short-running tasks)
    # @option opts [object] :executor when provided will run all operations on
    #   this executor rather than the global thread pool (overrides :operation)
    # @option opts [String] :dup_on_deref (false) call `#dup` before returning the data
    # @option opts [String] :freeze_on_deref (false) call `#freeze` before returning the data
    # @option opts [String] :copy_on_deref (nil) call the given `Proc` passing the internal value and
    #   returning the value returned from the proc
    #
    # @raise [ArgumentError] if no block is given
    def initialize(opts = {}, &block)
      raise ArgumentError.new('no block given') unless block_given?
      super(IVar::NO_VALUE, opts)
      @state = :unscheduled
      @task = block
      @executor = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_operation_pool
    end

    # Execute an `:unscheduled` `Future`. Immediately sets the state to `:pending` and
    # passes the block to a new thread/thread pool for eventual execution.
    # Does nothing if the `Future` is in any state other than `:unscheduled`.
    #
    # @return [Future] a reference to `self`
    #
    # @example Instance and execute in separate steps
    #   future = Concurrent::Future.new{ sleep(1); 42 }
    #   future.state #=> :unscheduled
    #   future.execute
    #   future.state #=> :pending
    #
    # @example Instance and execute in one line
    #   future = Concurrent::Future.new{ sleep(1); 42 }.execute
    #   future.state #=> :pending
    #
    # @since 0.5.0
    def execute
      if compare_and_set_state(:pending, :unscheduled)
        @executor.post{ work }
        self
      end
    end

    # Create a new `Future` object with the given block, execute it, and return the
    # `:pending` object.
    #
    # @yield the asynchronous operation to perform
    #
    # @option opts [String] :dup_on_deref (false) call `#dup` before returning the data
    # @option opts [String] :freeze_on_deref (false) call `#freeze` before returning the data
    # @option opts [String] :copy_on_deref (nil) call the given `Proc` passing the internal value and
    #   returning the value returned from the proc
    #
    # @return [Future] the newly created `Future` in the `:pending` state
    #
    # @raise [ArgumentError] if no block is given
    #
    # @example
    #   future = Concurrent::Future.execute{ sleep(1); 42 }
    #   future.state #=> :pending
    #
    # @since 0.5.0
    def self.execute(opts = {}, &block)
      Future.new(opts, &block).execute
    end

    protected :set, :fail, :complete

    private

    # @!visibility private
    def work # :nodoc:
      success, val, reason = SafeTaskExecutor.new(@task).execute
      complete(success, val, reason)
    end
  end
end

1	require 'thread'	1✔
2
3	require 'concurrent/options_parser'	1✔
4	require 'concurrent/ivar'	1✔
5	require 'concurrent/executor/safe_task_executor'	1✔
6
7	module Concurrent	1✔
8
9	# A `Future` represents a promise to complete an action at some time in the future.
10	# The action is atomic and permanent. The idea behind a future is to send an operation
11	# for asynchronous completion, do other stuff, then return and retrieve the result
12	# of the async operation at a later time.
13	#
14	# A `Future` has four possible states: :unscheduled, :pending, :rejected, or :fulfilled.
15	# When a `Future` is created its state is set to :unscheduled. Once the `#execute` method is
16	# called the state becomes :pending and will remain in that state until processing is
17	# complete. A completed `Future` is either :rejected, indicating that an exception was
18	# thrown during processing, or :fulfilled, indicating success. If a `Future` is :fulfilled
19	# its `value` will be updated to reflect the result of the operation. If :rejected the
20	# `reason` will be updated with a reference to the thrown exception. The predicate methods
21	# `#unscheduled?`, `#pending?`, `#rejected?`, and `fulfilled?` can be called at any time to
22	# obtain the state of the `Future`, as can the `#state` method, which returns a symbol.
23	#
24	# Retrieving the value of a `Future` is done through the `#value` (alias: `#deref`) method.
25	# Obtaining the value of a `Future` is a potentially blocking operation. When a `Future` is
26	# :rejected a call to `#value` will return `nil` immediately. When a `Future` is
27	# :fulfilled a call to `#value` will immediately return the current value. When a
28	# `Future` is :pending a call to `#value` will block until the `Future` is either
29	# :rejected or :fulfilled. A timeout value can be passed to `#value` to limit how
30	# long the call will block. If `nil` the call will block indefinitely. If `0` the call will
31	# not block. Any other integer or float value will indicate the maximum number of seconds to block.
32	#
33	# The `Future` class also includes the behavior of the Ruby standard library `Observable` module,
34	# but does so in a thread-safe way. On fulfillment or rejection all observers will be notified
35	# according to the normal `Observable` behavior. The observer callback function will be called
36	# with three parameters: the `Time` of fulfillment/rejection, the final `value`, and the final
37	# `reason`. Observers added after fulfillment/rejection will still be notified as normal.
38	#
39	# @see http://ruby-doc.org/stdlib-2.1.1/libdoc/observer/rdoc/Observable.html Ruby Observable module
40	# @see http://clojuredocs.org/clojure_core/clojure.core/future Clojure's future function
41	# @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Future.html java.util.concurrent.Future
42	class Future < IVar	1✔
43
44	# Create a new `Future` in the `:unscheduled` state.
45	#
46	# @yield the asynchronous operation to perform
47	#
48	# @param [Hash] opts the options controlling how the future will be processed
49	# @option opts [Boolean] :operation (false) when `true` will execute the future on the global
50	# operation pool (for long-running operations), when `false` will execute the future on the
51	# global task pool (for short-running tasks)
52	# @option opts [object] :executor when provided will run all operations on
53	# this executor rather than the global thread pool (overrides :operation)
54	# @option opts [String] :dup_on_deref (false) call `#dup` before returning the data
55	# @option opts [String] :freeze_on_deref (false) call `#freeze` before returning the data
56	# @option opts [String] :copy_on_deref (nil) call the given `Proc` passing the internal value and
57	# returning the value returned from the proc
58	#
59	# @raise [ArgumentError] if no block is given
60	def initialize(opts = {}, &block)	1✔
UNCOV 61	raise ArgumentError.new('no block given') unless block_given?	×
UNCOV 62	super(IVar::NO_VALUE, opts)	×
UNCOV 63	@state = :unscheduled	×
UNCOV 64	@task = block	×
UNCOV 65	@executor = OptionsParser::get_executor_from(opts) \|\| Concurrent.configuration.global_operation_pool	×
66	end
67
68	# Execute an `:unscheduled` `Future`. Immediately sets the state to `:pending` and
69	# passes the block to a new thread/thread pool for eventual execution.
70	# Does nothing if the `Future` is in any state other than `:unscheduled`.
71	#
72	# @return [Future] a reference to `self`
73	#
74	# @example Instance and execute in separate steps
75	# future = Concurrent::Future.new{ sleep(1); 42 }
76	# future.state #=> :unscheduled
77	# future.execute
78	# future.state #=> :pending
79	#
80	# @example Instance and execute in one line
81	# future = Concurrent::Future.new{ sleep(1); 42 }.execute
82	# future.state #=> :pending
83	#
84	# @since 0.5.0
85	def execute	1✔
UNCOV 86	if compare_and_set_state(:pending, :unscheduled)	×
UNCOV 87	@executor.post{ work }	×
UNCOV 88	self	×
89	end
90	end
91
92	# Create a new `Future` object with the given block, execute it, and return the
93	# `:pending` object.
94	#
95	# @yield the asynchronous operation to perform
96	#
97	# @option opts [String] :dup_on_deref (false) call `#dup` before returning the data
98	# @option opts [String] :freeze_on_deref (false) call `#freeze` before returning the data
99	# @option opts [String] :copy_on_deref (nil) call the given `Proc` passing the internal value and
100	# returning the value returned from the proc
101	#
102	# @return [Future] the newly created `Future` in the `:pending` state
103	#
104	# @raise [ArgumentError] if no block is given
105	#
106	# @example
107	# future = Concurrent::Future.execute{ sleep(1); 42 }
108	# future.state #=> :pending
109	#
110	# @since 0.5.0
111	def self.execute(opts = {}, &block)	1✔
UNCOV 112	Future.new(opts, &block).execute	×
113	end
114
115	protected :set, :fail, :complete	1✔
116
117	private	1✔
118
119	# @!visibility private
120	def work # :nodoc:	1✔
UNCOV 121	success, val, reason = SafeTaskExecutor.new(@task).execute	×
UNCOV 122	complete(success, val, reason)	×
123	end
124	end
125	end

ruby-concurrency / concurrent-ruby / #2835

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous