Class: Concurrent::Agent
- Inherits:
-
Synchronization::LockableObject
- Object
- Synchronization::LockableObject
- Concurrent::Agent
- Includes:
- Concern::Observable
- Defined in:
- lib/concurrent-ruby/concurrent/agent.rb
Overview
Agent
is inspired by Clojure's agent
function. An agent is a shared, mutable variable providing independent,
uncoordinated, asynchronous change of individual values. Best used when
the value will undergo frequent, complex updates. Suitable when the result
of an update does not need to be known immediately. Agent
is (mostly)
functionally equivalent to Clojure's agent, except where the runtime
prevents parity.
Agents are reactive, not autonomous - there is no imperative message loop
and no blocking receive. The state of an Agent should be itself immutable
and the #value
of an Agent is always immediately available for reading by
any thread without any messages, i.e. observation does not require
cooperation or coordination.
Agent action dispatches are made using the various #send
methods. These
methods always return immediately. At some point later, in another thread,
the following will happen:
- The given
action
will be applied to the state of the Agent and theargs
, if any were supplied. - The return value of
action
will be passed to the validator lambda, if one has been set on the Agent. - If the validator succeeds or if no validator was given, the return value
of the given
action
will become the new#value
of the Agent. See#initialize
for details. - If any observers were added to the Agent, they will be notified. See
#add_observer
for details. - If during the
action
execution any other dispatches are made (directly or indirectly), they will be held until after the#value
of the Agent has been changed.
If any exceptions are thrown by an action function, no nested dispatches
will occur, and the exception will be cached in the Agent itself. When an
Agent has errors cached, any subsequent interactions will immediately throw
an exception, until the agent's errors are cleared. Agent errors can be
examined with #error
and the agent restarted with #restart
.
The actions of all Agents get interleaved amongst threads in a thread pool.
At any point in time, at most one action for each Agent is being executed.
Actions dispatched to an agent from another single agent or thread will
occur in the order they were sent, potentially interleaved with actions
dispatched to the same agent from other sources. The #send
method should
be used for actions that are CPU limited, while the #send_off
method is
appropriate for actions that may block on IO.
Unlike in Clojure, Agent
cannot participate in Concurrent::TVar
transactions.
Example
def next_fibonacci(set = nil)
return [0, 1] if set.nil?
set + [set[-2..-1].reduce{|sum,x| sum + x }]
end
# create an agent with an initial value
agent = Concurrent::Agent.new(next_fibonacci)
# send a few update requests
5.times do
agent.send{|set| next_fibonacci(set) }
end
# wait for them to complete
agent.await
# get the current value
agent.value #=> [0, 1, 1, 2, 3, 5, 8]
Observation
Agents support observers through the Observable mixin module. Notification of observers occurs every time an action dispatch returns and the new value is successfully validated. Observation will not occur if the action raises an exception, if validation fails, or when a #restart occurs.
When notified the observer will receive three arguments: time
, old_value
,
and new_value
. The time
argument is the time at which the value change
occurred. The old_value
is the value of the Agent when the action began
processing. The new_value
is the value to which the Agent was set when the
action completed. Note that old_value
and new_value
may be the same.
This is not an error. It simply means that the action returned the same
value.
Nested Actions
It is possible for an Agent action to post further actions back to itself. The nested actions will be enqueued normally then processed after the outer action completes, in the order they were sent, possibly interleaved with action dispatches from other threads. Nested actions never deadlock with one another and a failure in a nested action will never affect the outer action.
Nested actions can be called using the Agent reference from the enclosing
scope or by passing the reference in as a "send" argument. Nested actions
cannot be post using self
from within the action block/proc/lambda; self
in this context will not reference the Agent. The preferred method for
dispatching nested actions is to pass the Agent as an argument. This allows
Ruby to more effectively manage the closing scope.
Prefer this:
agent = Concurrent::Agent.new(0)
agent.send(agent) do |value, this|
this.send {|v| v + 42 }
3.14
end
agent.value #=> 45.14
Over this:
agent = Concurrent::Agent.new(0)
agent.send do |value|
agent.send {|v| v + 42 }
3.14
end
NOTE Never, under any circumstances, call any of the "await" methods
(#await, #await_for, #await_for!, and #wait) from within an action
block/proc/lambda. The call will block the Agent and will always fail.
Calling either #await or #wait (with a timeout of nil
) will
hopelessly deadlock the Agent with no possibility of recovery.
Thread-safe Variable Classes
Each of the thread-safe variable classes is designed to solve a different problem. In general:
- Agent: Shared, mutable variable providing independent, uncoordinated, asynchronous change of individual values. Best used when the value will undergo frequent, complex updates. Suitable when the result of an update does not need to be known immediately.
- Atom: Shared, mutable variable providing independent, uncoordinated, synchronous change of individual values. Best used when the value will undergo frequent reads but only occasional, though complex, updates. Suitable when the result of an update must be known immediately.
- AtomicReference: A simple object reference that can be updated atomically. Updates are synchronous but fast. Best used when updates a simple set operations. Not suitable when updates are complex. AtomicBoolean and AtomicFixnum are similar but optimized for the given data type.
- Exchanger: Shared, stateless synchronization point. Used when two or more threads need to exchange data. The threads will pair then block on each other until the exchange is complete.
- MVar: Shared synchronization point. Used when one thread must give a value to another, which must take the value. The threads will block on each other until the exchange is complete.
- ThreadLocalVar: Shared, mutable, isolated variable which holds a different value for each thread which has access. Often used as an instance variable in objects which must maintain different state for different threads.
- TVar: Shared, mutable variables which provide coordinated, synchronous, change of many stated. Used when multiple value must change together, in an all-or-nothing transaction.
Defined Under Namespace
Classes: Error, ValidationError
Instance Attribute Summary collapse
-
#error_mode ⇒ undocumented
readonly
The error mode this Agent is operating in.
Class Method Summary collapse
-
.await(*agents) ⇒ Boolean
Blocks the current thread (indefinitely!) until all actions dispatched thus far to all the given Agents, from this thread or nested by the given Agents, have occurred.
-
.await_for(timeout, *agents) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far to all the given Agents, from this thread or nested by the given Agents, have occurred, or the timeout (in seconds) has elapsed.
-
.await_for!(timeout, *agents) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far to all the given Agents, from this thread or nested by the given Agents, have occurred, or the timeout (in seconds) has elapsed.
Instance Method Summary collapse
-
#<<(action) ⇒ Concurrent::Agent
Dispatches an action to the Agent and returns immediately.
-
#await ⇒ Boolean
Blocks the current thread (indefinitely!) until all actions dispatched thus far, from this thread or nested by the Agent, have occurred.
-
#await_for(timeout) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far, from this thread or nested by the Agent, have occurred, or the timeout (in seconds) has elapsed.
-
#await_for!(timeout) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far, from this thread or nested by the Agent, have occurred, or the timeout (in seconds) has elapsed.
-
#error ⇒ nil, Error
(also: #reason)
When #failed? and #error_mode is
:fail
, returns the error object which caused the failure, elsenil
. -
#failed? ⇒ Boolean
(also: #stopped?)
Is the Agent in a failed state?.
-
#initialize(initial, opts = {}) ⇒ Agent
constructor
Create a new
Agent
with the given initial value and options. - #restart(new_value, opts = {}) ⇒ Boolean
-
#send(*args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately.
-
#send!(*args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately.
-
#send_off(*args, &action) {|agent, value, *args| ... } ⇒ Boolean
(also: #post)
Dispatches an action to the Agent and returns immediately.
-
#send_off!(*args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately.
-
#send_via(executor, *args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately.
-
#send_via!(executor, *args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately.
-
#value ⇒ Object
(also: #deref)
The current value (state) of the Agent, irrespective of any pending or in-progress actions.
-
#wait(timeout = nil) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far, from this thread or nested by the Agent, have occurred, or the timeout (in seconds) has elapsed.
-
#add_observer(observer = nil, func = :update, &block) ⇒ Object
included
from Concern::Observable
Adds an observer to this set.
-
#count_observers ⇒ Integer
included
from Concern::Observable
Return the number of observers associated with this object.
-
#delete_observer(observer) ⇒ Object
included
from Concern::Observable
Remove
observer
as an observer on this object so that it will no longer receive notifications. -
#delete_observers ⇒ Observable
included
from Concern::Observable
Remove all observers associated with this object.
-
#with_observer(observer = nil, func = :update, &block) ⇒ Observable
included
from Concern::Observable
As
#add_observer
but can be used for chaining.
Constructor Details
#initialize(initial, opts = {}) ⇒ Agent
Create a new Agent
with the given initial value and options.
The :validator
option must be nil
or a side-effect free proc/lambda
which takes one argument. On any intended value change the validator, if
provided, will be called. If the new value is invalid the validator should
return false
or raise an error.
The :error_handler
option must be nil
or a proc/lambda which takes two
arguments. When an action raises an error or validation fails, either by
returning false or raising an error, the error handler will be called. The
arguments to the error handler will be a reference to the agent itself and
the error object which was raised.
The :error_mode
may be either :continue
(the default if an error
handler is given) or :fail
(the default if error handler nil or not
given).
If an action being run by the agent throws an error or doesn't pass
validation the error handler, if present, will be called. After the
handler executes if the error mode is :continue
the Agent will continue
as if neither the action that caused the error nor the error itself ever
happened.
If the mode is :fail
the Agent will become #failed? and will stop
accepting new action dispatches. Any previously queued actions will be
held until #restart is called. The #value method will still work,
returning the value of the Agent before the error.
219 220 221 222 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 219 def initialize(initial, opts = {}) super() synchronize { ns_initialize(initial, opts) } end |
Instance Attribute Details
#error_mode ⇒ undocumented (readonly)
The error mode this Agent is operating in. See #initialize for details.
183 184 185 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 183 def error_mode @error_mode end |
Class Method Details
.await(*agents) ⇒ Boolean
Blocks the current thread (indefinitely!) until all actions dispatched
thus far to all the given Agents, from this thread or nested by the
given Agents, have occurred. Will block when any of the agents are
failed. Will never return if a failed Agent is restart with
:clear_actions
true.
NOTE Never, under any circumstances, call any of the "await" methods
(#await, #await_for, #await_for!, and #wait) from within an action
block/proc/lambda. The call will block the Agent and will always fail.
Calling either #await or #wait (with a timeout of nil
) will
hopelessly deadlock the Agent with no possibility of recovery.
448 449 450 451 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 448 def await(*agents) agents.each { |agent| agent.await } true end |
.await_for(timeout, *agents) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far to all the given Agents, from this thread or nested by the given Agents, have occurred, or the timeout (in seconds) has elapsed.
NOTE Never, under any circumstances, call any of the "await" methods
(#await, #await_for, #await_for!, and #wait) from within an action
block/proc/lambda. The call will block the Agent and will always fail.
Calling either #await or #wait (with a timeout of nil
) will
hopelessly deadlock the Agent with no possibility of recovery.
462 463 464 465 466 467 468 469 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 462 def await_for(timeout, *agents) end_at = Concurrent.monotonic_time + timeout.to_f ok = agents.length.times do |i| break false if (delay = end_at - Concurrent.monotonic_time) < 0 break false unless agents[i].await_for(delay) end !!ok end |
.await_for!(timeout, *agents) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far to all the given Agents, from this thread or nested by the given Agents, have occurred, or the timeout (in seconds) has elapsed.
NOTE Never, under any circumstances, call any of the "await" methods
(#await, #await_for, #await_for!, and #wait) from within an action
block/proc/lambda. The call will block the Agent and will always fail.
Calling either #await or #wait (with a timeout of nil
) will
hopelessly deadlock the Agent with no possibility of recovery.
481 482 483 484 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 481 def await_for!(timeout, *agents) raise Concurrent::TimeoutError unless await_for(timeout, *agents) true end |
Instance Method Details
#<<(action) ⇒ Concurrent::Agent
Dispatches an action to the Agent and returns immediately. Subsequently, in a thread from a thread pool, the #value will be set to the return value of the action. Appropriate for actions that may block on IO.
330 331 332 333 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 330 def <<(action) send_off(&action) self end |
#await ⇒ Boolean
Blocks the current thread (indefinitely!) until all actions dispatched
thus far, from this thread or nested by the Agent, have occurred. Will
block when #failed?. Will never return if a failed Agent is #restart
with :clear_actions
true.
Returns a reference to self
to support method chaining:
current_value = agent.await.value
NOTE Never, under any circumstances, call any of the "await" methods
(#await, #await_for, #await_for!, and #wait) from within an action
block/proc/lambda. The call will block the Agent and will always fail.
Calling either #await or #wait (with a timeout of nil
) will
hopelessly deadlock the Agent with no possibility of recovery.
349 350 351 352 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 349 def await wait(nil) self end |
#await_for(timeout) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far, from this thread or nested by the Agent, have occurred, or the timeout (in seconds) has elapsed.
NOTE Never, under any circumstances, call any of the "await" methods
(#await, #await_for, #await_for!, and #wait) from within an action
block/proc/lambda. The call will block the Agent and will always fail.
Calling either #await or #wait (with a timeout of nil
) will
hopelessly deadlock the Agent with no possibility of recovery.
362 363 364 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 362 def await_for(timeout) wait(timeout.to_f) end |
#await_for!(timeout) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far, from this thread or nested by the Agent, have occurred, or the timeout (in seconds) has elapsed.
NOTE Never, under any circumstances, call any of the "await" methods
(#await, #await_for, #await_for!, and #wait) from within an action
block/proc/lambda. The call will block the Agent and will always fail.
Calling either #await or #wait (with a timeout of nil
) will
hopelessly deadlock the Agent with no possibility of recovery.
376 377 378 379 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 376 def await_for!(timeout) raise Concurrent::TimeoutError unless wait(timeout.to_f) true end |
#error ⇒ nil, Error Also known as: reason
When #failed? and #error_mode is :fail
, returns the error object
which caused the failure, else nil
. When #error_mode is :continue
will always return nil
.
239 240 241 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 239 def error @error.value end |
#failed? ⇒ Boolean Also known as: stopped?
Is the Agent in a failed state?
401 402 403 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 401 def failed? !@error.value.nil? end |
#restart(new_value, opts = {}) ⇒ Boolean
When an Agent is #failed?, changes the Agent #value to new_value
then un-fails the Agent so that action dispatches are allowed again. If
the :clear_actions
option is give and true, any actions queued on the
Agent that were being held while it was failed will be discarded,
otherwise those held actions will proceed. The new_value
must pass the
validator if any, or restart
will raise an exception and the Agent will
remain failed with its old #value and #error. Observers, if any, will
not be notified of the new state.
423 424 425 426 427 428 429 430 431 432 433 434 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 423 def restart(new_value, opts = {}) clear_actions = opts.fetch(:clear_actions, false) synchronize do raise Error.new('agent is not failed') unless failed? raise ValidationError unless ns_validate(new_value) @current.value = new_value @error.value = nil @queue.clear if clear_actions ns_post_next_job unless @queue.empty? end true end |
#send(*args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately. Subsequently, in a thread from a thread pool, the #value will be set to the return value of the action. Action dispatches are only allowed when the Agent is not #failed?.
The action must be a block/proc/lambda which takes 1 or more arguments.
The first argument is the current #value of the Agent. Any arguments
passed to the send method via the args
parameter will be passed to the
action as the remaining arguments. The action must return the new value
of the Agent.
- #send and #send! should be used for actions that are CPU limited
- #send_off, #send_off!, and #<< are appropriate for actions that may block on IO
- #send_via and #send_via! are used when a specific executor is to be used for the action
277 278 279 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 277 def send(*args, &action) enqueue_action_job(action, args, Concurrent.global_fast_executor) end |
#send!(*args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately. Subsequently, in a thread from a thread pool, the #value will be set to the return value of the action. Action dispatches are only allowed when the Agent is not #failed?.
The action must be a block/proc/lambda which takes 1 or more arguments.
The first argument is the current #value of the Agent. Any arguments
passed to the send method via the args
parameter will be passed to the
action as the remaining arguments. The action must return the new value
of the Agent.
- #send and #send! should be used for actions that are CPU limited
- #send_off, #send_off!, and #<< are appropriate for actions that may block on IO
- #send_via and #send_via! are used when a specific executor is to be used for the action
286 287 288 289 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 286 def send!(*args, &action) raise Error.new unless send(*args, &action) true end |
#send_off(*args, &action) {|agent, value, *args| ... } ⇒ Boolean Also known as: post
Dispatches an action to the Agent and returns immediately. Subsequently, in a thread from a thread pool, the #value will be set to the return value of the action. Action dispatches are only allowed when the Agent is not #failed?.
The action must be a block/proc/lambda which takes 1 or more arguments.
The first argument is the current #value of the Agent. Any arguments
passed to the send method via the args
parameter will be passed to the
action as the remaining arguments. The action must return the new value
of the Agent.
- #send and #send! should be used for actions that are CPU limited
- #send_off, #send_off!, and #<< are appropriate for actions that may block on IO
- #send_via and #send_via! are used when a specific executor is to be used for the action
293 294 295 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 293 def send_off(*args, &action) enqueue_action_job(action, args, Concurrent.global_io_executor) end |
#send_off!(*args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately. Subsequently, in a thread from a thread pool, the #value will be set to the return value of the action. Action dispatches are only allowed when the Agent is not #failed?.
The action must be a block/proc/lambda which takes 1 or more arguments.
The first argument is the current #value of the Agent. Any arguments
passed to the send method via the args
parameter will be passed to the
action as the remaining arguments. The action must return the new value
of the Agent.
- #send and #send! should be used for actions that are CPU limited
- #send_off, #send_off!, and #<< are appropriate for actions that may block on IO
- #send_via and #send_via! are used when a specific executor is to be used for the action
301 302 303 304 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 301 def send_off!(*args, &action) raise Error.new unless send_off(*args, &action) true end |
#send_via(executor, *args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately. Subsequently, in a thread from a thread pool, the #value will be set to the return value of the action. Action dispatches are only allowed when the Agent is not #failed?.
The action must be a block/proc/lambda which takes 1 or more arguments.
The first argument is the current #value of the Agent. Any arguments
passed to the send method via the args
parameter will be passed to the
action as the remaining arguments. The action must return the new value
of the Agent.
- #send and #send! should be used for actions that are CPU limited
- #send_off, #send_off!, and #<< are appropriate for actions that may block on IO
- #send_via and #send_via! are used when a specific executor is to be used for the action
310 311 312 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 310 def send_via(executor, *args, &action) enqueue_action_job(action, args, executor) end |
#send_via!(executor, *args, &action) {|agent, value, *args| ... } ⇒ Boolean
Dispatches an action to the Agent and returns immediately. Subsequently, in a thread from a thread pool, the #value will be set to the return value of the action. Action dispatches are only allowed when the Agent is not #failed?.
The action must be a block/proc/lambda which takes 1 or more arguments.
The first argument is the current #value of the Agent. Any arguments
passed to the send method via the args
parameter will be passed to the
action as the remaining arguments. The action must return the new value
of the Agent.
- #send and #send! should be used for actions that are CPU limited
- #send_off, #send_off!, and #<< are appropriate for actions that may block on IO
- #send_via and #send_via! are used when a specific executor is to be used for the action
318 319 320 321 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 318 def send_via!(executor, *args, &action) raise Error.new unless send_via(executor, *args, &action) true end |
#value ⇒ Object Also known as: deref
The current value (state) of the Agent, irrespective of any pending or in-progress actions. The value is always available and is non-blocking.
228 229 230 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 228 def value @current.value # TODO (pitr 12-Sep-2015): broken unsafe read? end |
#wait(timeout = nil) ⇒ Boolean
Blocks the current thread until all actions dispatched thus far, from this thread or nested by the Agent, have occurred, or the timeout (in seconds) has elapsed. Will block indefinitely when timeout is nil or not given.
Provided mainly for consistency with other classes in this library. Prefer
the various await
methods instead.
NOTE Never, under any circumstances, call any of the "await" methods
(#await, #await_for, #await_for!, and #wait) from within an action
block/proc/lambda. The call will block the Agent and will always fail.
Calling either #await or #wait (with a timeout of nil
) will
hopelessly deadlock the Agent with no possibility of recovery.
392 393 394 395 396 |
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 392 def wait(timeout = nil) latch = Concurrent::CountDownLatch.new(1) enqueue_await_job(latch) latch.wait(timeout) end |
#add_observer(observer = nil, func = :update, &block) ⇒ Object Originally defined in module Concern::Observable
Adds an observer to this set. If a block is passed, the observer will be created by this method and no other params should be passed.
#count_observers ⇒ Integer Originally defined in module Concern::Observable
Return the number of observers associated with this object.
#delete_observer(observer) ⇒ Object Originally defined in module Concern::Observable
Remove observer
as an observer on this object so that it will no
longer receive notifications.
#delete_observers ⇒ Observable Originally defined in module Concern::Observable
Remove all observers associated with this object.
#with_observer(observer = nil, func = :update, &block) ⇒ Observable Originally defined in module Concern::Observable
As #add_observer
but can be used for chaining.