#812

Committed 23 May 2014 08:08PM UTC coverage: 39.344% (-57.9%) from 97.237%

Build # #812

Build Type

push

Committed by

Commit Message

Merge pull request #96 from jdantonio/refactor/errors

Moved all custom errors into a single file and into the Concurrent module

Coverage Stats

17 of 18 new or added lines in 10 files covered. (94.44%)

1435 existing lines in 57 files now uncovered.

1187 of 3017 relevant lines covered (39.34%)

0.5 hits per line

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

37.84

/lib/concurrent/agent.rb

require 'thread'

require 'concurrent/dereferenceable'
require 'concurrent/observable'
require 'concurrent/options_parser'
require 'concurrent/utility/timeout'

module Concurrent

  # An agent is a single atomic value that represents an identity. The current value
  # of the agent can be requested at any time (`#deref`). Each agent has a work queue and operates on
  # the global thread pool. Consumers can `#post` code blocks to the agent. The code block (function)
  # will receive the current value of the agent as its sole parameter. The return value of the block
  # will become the new value of the agent. Agents support two error handling modes: fail and continue.
  # A good example of an agent is a shared incrementing counter, such as the score in a video game.
  #
  # @example Basic usage
  #   score = Concurrent::Agent.new(10)
  #   score.value #=> 10
  #   
  #   score << proc{|current| current + 100 }
  #   sleep(0.1)
  #   score.value #=> 110
  #   
  #   score << proc{|current| current * 2 }
  #   sleep(0.1)
  #   score.value #=> 220
  #   
  #   score << proc{|current| current - 50 }
  #   sleep(0.1)
  #   score.value #=> 170
  #
  # @!attribute [r] timeout
  #   @return [Fixnum] the maximum number of seconds before an update is cancelled
  class Agent
    include Dereferenceable
    include Concurrent::Observable

    # The default timeout value (in seconds); used when no timeout option
    # is given at initialization
    TIMEOUT = 5

    attr_reader :timeout, :task_executor, :operation_executor

    # Initialize a new Agent with the given initial value and provided options.
    #
    # @param [Object] initial the initial value
    # @param [Hash] opts the options used to define the behavior at update and deref
    #
    # @option opts [Fixnum] :timeout (TIMEOUT) maximum number of seconds before an update is cancelled
    #
    # @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
    def initialize(initial, opts = {})
      @value              = initial
      @rescuers           = []
      @validator          = Proc.new { |result| true }
      @timeout            = opts.fetch(:timeout, TIMEOUT).freeze
      self.observers      = CopyOnWriteObserverSet.new
      @one_by_one         = OneByOne.new
      @task_executor      = OptionsParser.get_task_executor_from(opts)
      @operation_executor = OptionsParser.get_operation_executor_from(opts)
      init_mutex
      set_deref_options(opts)
    end

    # Specifies a block operation to be performed when an update operation raises
    # an exception. Rescue blocks will be checked in order they were added. The first
    # block for which the raised exception "is-a" subclass of the given `clazz` will
    # be called. If no `clazz` is given the block will match any caught exception.
    # This behavior is intended to be identical to Ruby's `begin/rescue/end` behavior.
    # Any number of rescue handlers can be added. If no rescue handlers are added then
    # caught exceptions will be suppressed.
    #
    # @param [Exception] clazz the class of exception to catch
    # @yield the block to be called when a matching exception is caught
    # @yieldparam [StandardError] ex the caught exception
    #
    # @example
    #   score = Concurrent::Agent.new(0).
    #             rescue(NoMethodError){|ex| puts "Bam!" }.
    #             rescue(ArgumentError){|ex| puts "Pow!" }.
    #             rescue{|ex| puts "Boom!" }
    #   
    #   score << proc{|current| raise ArgumentError }
    #   sleep(0.1)
    #   #=> puts "Pow!"
    def rescue(clazz = StandardError, &block)
      unless block.nil?
        mutex.synchronize do
          @rescuers << Rescuer.new(clazz, block)
        end
      end
      self
    end
    alias_method :catch, :rescue
    alias_method :on_error, :rescue

    # A block operation to be performed after every update to validate if the new
    # value is valid. If the new value is not valid then the current value is not
    # updated. If no validator is provided then all updates are considered valid.
    #
    # @yield the block to be called after every update operation to determine if
    #   the result is valid
    # @yieldparam [Object] value the result of the last update operation
    # @yieldreturn [Boolean] true if the value is valid else false
    def validate(&block)

      unless block.nil?
        begin
          mutex.lock
          @validator = block
        ensure
          mutex.unlock
        end
      end
      self
    end
    alias_method :validates, :validate
    alias_method :validate_with, :validate
    alias_method :validates_with, :validate

    # Update the current value with the result of the given block operation,
    # block should not do blocking calls, use #post_off for blocking calls
    #
    # @yield the operation to be performed with the current value in order to calculate
    #   the new value
    # @yieldparam [Object] value the current value
    # @yieldreturn [Object] the new value
    # @return [true, nil] nil when no block is given
    def post(&block)
      post_on(@task_executor, &block)
    end

    # Update the current value with the result of the given block operation,
    # block can do blocking calls
    #
    # @yield the operation to be performed with the current value in order to calculate
    #   the new value
    # @yieldparam [Object] value the current value
    # @yieldreturn [Object] the new value
    # @return [true, nil] nil when no block is given
    def post_off(&block)
      post_on(@operation_executor, &block)
    end

    # Update the current value with the result of the given block operation,
    # block should not do blocking calls, use #post_off for blocking calls
    #
    # @yield the operation to be performed with the current value in order to calculate
    #   the new value
    # @yieldparam [Object] value the current value
    # @yieldreturn [Object] the new value
    def <<(block)
      post(&block)
      self
    end

    # Waits/blocks until all the updates sent before this call are done.
    #
    # @param [Numeric] timeout the maximum time in second to wait.
    # @return [Boolean] false on timeout, true otherwise
    def await(timeout = nil)
      done = Event.new
      post { |val| done.set; val }
      done.wait timeout
    end

    private

    def post_on(executor, &block)
      return nil if block.nil?
      @one_by_one.post(executor) { work(&block) }
      true
    end

    # @!visibility private
    Rescuer = Struct.new(:clazz, :block) # :nodoc:

    # @!visibility private
    def try_rescue(ex) # :nodoc:
      rescuer = mutex.synchronize do
        @rescuers.find { |r| ex.is_a?(r.clazz) }
      end
      rescuer.block.call(ex) if rescuer
    rescue Exception => ex
      # supress
    end

    # @!visibility private
    def work(&handler) # :nodoc:
      validator, value = mutex.synchronize { [@validator, @value] }

      begin
        # FIXME creates second thread
        result, valid = Concurrent::timeout(@timeout) do
          result = handler.call(value)
          [result, validator.call(result)]
        end
      rescue Exception => ex
        exception = ex
      end

      begin
        mutex.lock
        should_notify = if !exception && valid
                          @value = result
                          true
                        end
      ensure
        mutex.unlock
      end

      if should_notify
        time = Time.now
        observers.notify_observers { [time, self.value] }
      end

      try_rescue(exception)
    end
  end
end

1	require 'thread'	1✔
2
3	require 'concurrent/dereferenceable'	1✔
4	require 'concurrent/observable'	1✔
5	require 'concurrent/options_parser'	1✔
6	require 'concurrent/utility/timeout'	1✔
7
8	module Concurrent	1✔
9
10	# An agent is a single atomic value that represents an identity. The current value
11	# of the agent can be requested at any time (`#deref`). Each agent has a work queue and operates on
12	# the global thread pool. Consumers can `#post` code blocks to the agent. The code block (function)
13	# will receive the current value of the agent as its sole parameter. The return value of the block
14	# will become the new value of the agent. Agents support two error handling modes: fail and continue.
15	# A good example of an agent is a shared incrementing counter, such as the score in a video game.
16	#
17	# @example Basic usage
18	# score = Concurrent::Agent.new(10)
19	# score.value #=> 10
20	#
21	# score << proc{\|current\| current + 100 }
22	# sleep(0.1)
23	# score.value #=> 110
24	#
25	# score << proc{\|current\| current * 2 }
26	# sleep(0.1)
27	# score.value #=> 220
28	#
29	# score << proc{\|current\| current - 50 }
30	# sleep(0.1)
31	# score.value #=> 170
32	#
33	# @!attribute [r] timeout
34	# @return [Fixnum] the maximum number of seconds before an update is cancelled
35	class Agent	1✔
36	include Dereferenceable	1✔
37	include Concurrent::Observable	1✔
38
39	# The default timeout value (in seconds); used when no timeout option
40	# is given at initialization
41	TIMEOUT = 5	1✔
42
43	attr_reader :timeout, :task_executor, :operation_executor	1✔
44
45	# Initialize a new Agent with the given initial value and provided options.
46	#
47	# @param [Object] initial the initial value
48	# @param [Hash] opts the options used to define the behavior at update and deref
49	#
50	# @option opts [Fixnum] :timeout (TIMEOUT) maximum number of seconds before an update is cancelled
51	#
52	# @option opts [Boolean] :operation (false) when `true` will execute the future on the global
53	# operation pool (for long-running operations), when `false` will execute the future on the
54	# global task pool (for short-running tasks)
55	# @option opts [object] :executor when provided will run all operations on
56	# this executor rather than the global thread pool (overrides :operation)
57	#
58	# @option opts [String] :dup_on_deref (false) call `#dup` before returning the data
59	# @option opts [String] :freeze_on_deref (false) call `#freeze` before returning the data
60	# @option opts [String] :copy_on_deref (nil) call the given `Proc` passing the internal value and
61	# returning the value returned from the proc
62	def initialize(initial, opts = {})	1✔
UNCOV 63	@value = initial	×
UNCOV 64	@rescuers = []	×
UNCOV 65	@validator = Proc.new { \|result\| true }	×
UNCOV 66	@timeout = opts.fetch(:timeout, TIMEOUT).freeze	×
UNCOV 67	self.observers = CopyOnWriteObserverSet.new	×
UNCOV 68	@one_by_one = OneByOne.new	×
UNCOV 69	@task_executor = OptionsParser.get_task_executor_from(opts)	×
UNCOV 70	@operation_executor = OptionsParser.get_operation_executor_from(opts)	×
UNCOV 71	init_mutex	×
UNCOV 72	set_deref_options(opts)	×
73	end
74
75	# Specifies a block operation to be performed when an update operation raises
76	# an exception. Rescue blocks will be checked in order they were added. The first
77	# block for which the raised exception "is-a" subclass of the given `clazz` will
78	# be called. If no `clazz` is given the block will match any caught exception.
79	# This behavior is intended to be identical to Ruby's `begin/rescue/end` behavior.
80	# Any number of rescue handlers can be added. If no rescue handlers are added then
81	# caught exceptions will be suppressed.
82	#
83	# @param [Exception] clazz the class of exception to catch
84	# @yield the block to be called when a matching exception is caught
85	# @yieldparam [StandardError] ex the caught exception
86	#
87	# @example
88	# score = Concurrent::Agent.new(0).
89	# rescue(NoMethodError){\|ex\| puts "Bam!" }.
90	# rescue(ArgumentError){\|ex\| puts "Pow!" }.
91	# rescue{\|ex\| puts "Boom!" }
92	#
93	# score << proc{\|current\| raise ArgumentError }
94	# sleep(0.1)
95	# #=> puts "Pow!"
96	def rescue(clazz = StandardError, &block)	1✔
UNCOV 97	unless block.nil?	×
UNCOV 98	mutex.synchronize do	×
UNCOV 99	@rescuers << Rescuer.new(clazz, block)	×
100	end
101	end
UNCOV 102	self	×
103	end
104	alias_method :catch, :rescue	1✔
105	alias_method :on_error, :rescue	1✔
106
107	# A block operation to be performed after every update to validate if the new
108	# value is valid. If the new value is not valid then the current value is not
109	# updated. If no validator is provided then all updates are considered valid.
110	#
111	# @yield the block to be called after every update operation to determine if
112	# the result is valid
113	# @yieldparam [Object] value the result of the last update operation
114	# @yieldreturn [Boolean] true if the value is valid else false
115	def validate(&block)	1✔
116
UNCOV 117	unless block.nil?	×
118	begin
UNCOV 119	mutex.lock	×
UNCOV 120	@validator = block	×
121	ensure
UNCOV 122	mutex.unlock	×
123	end
124	end
UNCOV 125	self	×
126	end
127	alias_method :validates, :validate	1✔
128	alias_method :validate_with, :validate	1✔
129	alias_method :validates_with, :validate	1✔
130
131	# Update the current value with the result of the given block operation,
132	# block should not do blocking calls, use #post_off for blocking calls
133	#
134	# @yield the operation to be performed with the current value in order to calculate
135	# the new value
136	# @yieldparam [Object] value the current value
137	# @yieldreturn [Object] the new value
138	# @return [true, nil] nil when no block is given
139	def post(&block)	1✔
UNCOV 140	post_on(@task_executor, &block)	×
141	end
142
143	# Update the current value with the result of the given block operation,
144	# block can do blocking calls
145	#
146	# @yield the operation to be performed with the current value in order to calculate
147	# the new value
148	# @yieldparam [Object] value the current value
149	# @yieldreturn [Object] the new value
150	# @return [true, nil] nil when no block is given
151	def post_off(&block)	1✔
UNCOV 152	post_on(@operation_executor, &block)	×
153	end
154
155	# Update the current value with the result of the given block operation,
156	# block should not do blocking calls, use #post_off for blocking calls
157	#
158	# @yield the operation to be performed with the current value in order to calculate
159	# the new value
160	# @yieldparam [Object] value the current value
161	# @yieldreturn [Object] the new value
162	def <<(block)	1✔
163	post(&block)	×
164	self	×
165	end
166
167	# Waits/blocks until all the updates sent before this call are done.
168	#
169	# @param [Numeric] timeout the maximum time in second to wait.
170	# @return [Boolean] false on timeout, true otherwise
171	def await(timeout = nil)	1✔
UNCOV 172	done = Event.new	×
UNCOV 173	post { \|val\| done.set; val }	×
UNCOV 174	done.wait timeout	×
175	end
176
177	private	1✔
178
179	def post_on(executor, &block)	1✔
UNCOV 180	return nil if block.nil?	×
UNCOV 181	@one_by_one.post(executor) { work(&block) }	×
UNCOV 182	true	×
183	end
184
185	# @!visibility private
186	Rescuer = Struct.new(:clazz, :block) # :nodoc:	1✔
187
188	# @!visibility private
189	def try_rescue(ex) # :nodoc:	1✔
UNCOV 190	rescuer = mutex.synchronize do	×
UNCOV 191	@rescuers.find { \|r\| ex.is_a?(r.clazz) }	×
192	end
UNCOV 193	rescuer.block.call(ex) if rescuer	×
194	rescue Exception => ex
195	# supress
196	end
197
198	# @!visibility private
199	def work(&handler) # :nodoc:	1✔
UNCOV 200	validator, value = mutex.synchronize { [@validator, @value] }	×
201
202	begin
203	# FIXME creates second thread
UNCOV 204	result, valid = Concurrent::timeout(@timeout) do	×
UNCOV 205	result = handler.call(value)	×
UNCOV 206	[result, validator.call(result)]	×
207	end
208	rescue Exception => ex
UNCOV 209	exception = ex	×
210	end
211
212	begin
UNCOV 213	mutex.lock	×
UNCOV 214	should_notify = if !exception && valid	×
UNCOV 215	@value = result	×
UNCOV 216	true	×
217	end
218	ensure
UNCOV 219	mutex.unlock	×
220	end
221
UNCOV 222	if should_notify	×
UNCOV 223	time = Time.now	×
UNCOV 224	observers.notify_observers { [time, self.value] }	×
225	end
226
UNCOV 227	try_rescue(exception)	×
228	end
229	end
230	end

jdantonio / concurrent-ruby / #812

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