Class: Concurrent::Agent

Relationships & Source Files
Namespace Children
Classes: `Job`
Exceptions: `Error`, `ValidationError`
Super Chains via Extension / Inclusion / Inheritance
Class Chain: self, `Synchronization::LockableObject`
Instance Chain: self, `Concern::Observable`, `Synchronization::LockableObject`
Inherits:	Concurrent::Synchronization::LockableObject `::Object` Concurrent::Synchronization::LockableObject Concurrent::Agent
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 the args, 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 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 = 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 Concurrent::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 = Agent.new(0) agent.send(agent) do |value, this|

this.send {|v| v + 42 }
3.14

end agent.value #=> 45.14 “‘

Over this:

“‘ agent = 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.

Constant Summary

AWAIT_ACTION = private
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 154
```
->(value, latch) { latch.count_down; AWAIT_FLAG }
```
AWAIT_FLAG = private
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 151
```
::Object.new
```
DEFAULT_ERROR_HANDLER = private
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 157
```
->(agent, error) { nil }
```
DEFAULT_VALIDATOR = private
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 160
```
->(value) { true }
```
ERROR_MODES = private
# File 'lib/concurrent-ruby/concurrent/agent.rb', line 148
```
[:continue, :fail].freeze
```

Class Method Summary

.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.
.new(initial, opts = {}) ⇒ Agent constructor

Create a new Agent with the given initial value and options.

Instance Attribute Summary

#error_mode readonly

The error mode this Agent is operating in.
#failed? ⇒ Boolean (also: #stopped?) readonly

Is the Agent in a failed state?
#stopped? readonly

Alias for #failed?.

Instance Method Summary

#<<(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.
#deref

Alias for #value.
#error ⇒ nil, Error (also: #reason)

When #failed? and #error_mode is :fail, returns the error object which caused the failure, else nil.
#post(*args, &action)

Alias for #send_off.
#reason

Alias for #error.
#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.
#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.
#enqueue_action_job(action, args, executor) private
#enqueue_await_job(latch) private
#execute_next_job private
#handle_error(error) private
#ns_enqueue_job(job, index = nil) private
#ns_find_last_job_for_thread private
#ns_initialize(initial, opts) private
#ns_post_next_job private
#ns_validate(value) private

`Concern::Observable` - Included

#add_observer	Adds an observer to this set.
#count_observers	Return the number of observers associated with this object.
#delete_observer	Remove `observer` as an observer on this object so that it will no longer receive notifications.
#delete_observers	Remove all observers associated with this object.
#with_observer	As `#add_observer` but can be used for chaining.

`Synchronization::LockableObject` - Inherited

#new_condition

Constructor Details

.new(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.

Parameters:

initial (Object) —

the initial value
opts (Hash) (defaults to: {}) —

the configuration options

Options Hash (opts):

:error_mode (Symbol) —

either :continue or :fail
:error_handler (nil, Proc) —

the (optional) error handler
:validator (nil, Proc) —

the (optional) validation procedure

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 220


def initialize(initial, opts = {})
  super()
  synchronize { ns_initialize(initial, opts) }
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.

Parameters:

agents (Array<Agent>) —

the Agents on which to wait

Returns:

(Boolean) —

true

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 449


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.

Parameters:

timeout (Float) —

the maximum number of seconds to wait
agents (Array<Agent>) —

the Agents on which to wait

Returns:

(Boolean) —

true if all actions complete before timeout else false

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 463


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.

Parameters:

timeout (Float) —

the maximum number of seconds to wait
agents (Array<Agent>) —

the Agents on which to wait

Returns:

(Boolean) —

true if all actions complete before timeout

Raises:

(Concurrent::TimeoutError) —

when timeout is reached

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 482


def await_for!(timeout, *agents)
  raise Concurrent::TimeoutError unless await_for(timeout, *agents)
  true
end

Instance Attribute Details

#error_mode (readonly)

The error mode this Agent is operating in. See Concurrent#initialize for details.

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 184


attr_reader :error_mode

#failed? ⇒ `Boolean` (readonly) Also known as: #stopped?

Is the Agent in a failed state?

#stopped? (readonly)

Alias for #failed?.

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 406


alias_method :stopped?, :failed?

Instance Method Details

#<<(action) ⇒ `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.

Parameters:

action (Proc) —

the action dispatch to be enqueued

Returns:

(Agent) —

self

#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.

Returns:

(Boolean) —

self

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 350


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.

Parameters:

timeout (Float) —

the maximum number of seconds to wait

Returns:

(Boolean) —

true if all actions complete before timeout else false

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 363


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.

Parameters:

timeout (Float) —

the maximum number of seconds to wait

Returns:

(Boolean) —

true if all actions complete before timeout

Raises:

(Concurrent::TimeoutError) —

when timeout is reached

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 377


def await_for!(timeout)
  raise Concurrent::TimeoutError unless wait(timeout.to_f)
  true
end

#deref

Alias for #value.

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 233


alias_method :deref, :value

#enqueue_action_job(action, args, executor) (private)

Raises:

(ArgumentError)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 510


def enqueue_action_job(action, args, executor)
  raise ArgumentError.new('no action given') unless action
  job = Job.new(action, args, executor, @caller.value || Thread.current.object_id)
  synchronize { ns_enqueue_job(job) }
end

#enqueue_await_job(latch) (private)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 516


def enqueue_await_job(latch)
  synchronize do
    if (index = ns_find_last_job_for_thread)
      job = Job.new(AWAIT_ACTION, [latch], Concurrent.global_immediate_executor,
                    Thread.current.object_id)
      ns_enqueue_job(job, index+1)
    else
      latch.count_down
      true
    end
  end
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.

Returns:

(nil, Error) —

the error which caused the failure when #failed?

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 240


def error
  @error.value
end

#execute_next_job (private)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 543


def execute_next_job
  job       = synchronize { @queue.first }
  old_value = @current.value

  @caller.value = job.caller # for nested actions
  new_value     = job.action.call(old_value, *job.args)
  @caller.value = nil

  return if new_value == AWAIT_FLAG

  if ns_validate(new_value)
    @current.value = new_value
    observers.notify_observers(Time.now, old_value, new_value)
  else
    handle_error(ValidationError.new)
  end
rescue => error
  handle_error(error)
ensure
  synchronize do
    @queue.shift
    unless failed? || @queue.empty?
      ns_post_next_job
    end
  end
end

#handle_error(error) (private)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 576


def handle_error(error)
  # stop new jobs from posting
  @error.value = error if @error_mode == :fail
  @error_handler.call(self, error)
rescue
  # do nothing
end

#ns_enqueue_job(job, index = nil) (private)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 529


def ns_enqueue_job(job, index = nil)
  # a non-nil index means this is an await job
  return false if index.nil? && failed?
  index ||= @queue.length
  @queue.insert(index, job)
  # if this is the only job, post to executor
  ns_post_next_job if @queue.length == 1
  true
end

#ns_find_last_job_for_thread (private)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 584


def ns_find_last_job_for_thread
  @queue.rindex { |job| job.caller == Thread.current.object_id }
end

#ns_initialize(initial, opts) (private)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 490


def ns_initialize(initial, opts)
  @error_mode    = opts[:error_mode]
  @error_handler = opts[:error_handler]

  if @error_mode && !ERROR_MODES.include?(@error_mode)
    raise ArgumentError.new('unrecognized error mode')
  elsif @error_mode.nil?
    @error_mode = @error_handler ? :continue : :fail
  end

  @error_handler ||= DEFAULT_ERROR_HANDLER
  @validator     = opts.fetch(:validator, DEFAULT_VALIDATOR)
  @current       = Concurrent::AtomicReference.new(initial)
  @error         = Concurrent::AtomicReference.new(nil)
  @caller        = Concurrent::ThreadLocalVar.new(nil)
  @queue         = []

  self.observers = Collection::CopyOnNotifyObserverSet.new
end

#ns_post_next_job (private)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 539


def ns_post_next_job
  @queue.first.executor.post { execute_next_job }
end

#ns_validate(value) (private)

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 570


def ns_validate(value)
  @validator.call(value)
rescue
  false
end

#post(*args, &action)

Alias for #send_off.

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 298


alias_method :post, :send_off

#reason

Alias for #error.

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 244


alias_method :reason, :error

#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.

Parameters:

new_value (Object) —

the new value for the Agent once restarted
opts (Hash) (defaults to: {}) —

the configuration options

Options Hash (opts):

:clear_actions (Symbol) —

true if all enqueued but unprocessed actions should be discarded on restart, else false (default: false)

Returns:

(Boolean) —

true

Raises:

(Concurrent:AgentError) —

when not failed

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 424


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

Parameters:

args (Array<Object>) —

zero or more arguments to be passed to the action
action (Proc) —

the action dispatch to be enqueued

Yields:

(agent, value, *args) —

process the old value and return the new

Yield Parameters:

value (Object) —

the current #value of the Agent
args (Array<Object>) —

zero or more arguments to pass to the action

Yield Returns:

(Object) —

the new value of the Agent

Returns:

(Boolean) —

true if the action is successfully enqueued, false if the Agent is #failed?

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 278


def send(*args, &action)
  enqueue_action_job(action, args, Concurrent.global_fast_executor)
end

#send!(args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#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

Parameters:

args (Array<Object>) —

zero or more arguments to be passed to the action
action (Proc) —

the action dispatch to be enqueued

Yields:

(agent, value, *args) —

process the old value and return the new

Yield Parameters:

value (Object) —

the current #value of the Agent
args (Array<Object>) —

zero or more arguments to pass to the action

Yield Returns:

(Object) —

the new value of the Agent

Returns:

(Boolean) —

true if the action is successfully enqueued

Raises:

(Concurrent::Agent::Error) —

if the Agent is #failed?

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 287


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

#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

Parameters:

args (Array<Object>) —

zero or more arguments to be passed to the action
action (Proc) —

the action dispatch to be enqueued

Yields:

(agent, value, *args) —

process the old value and return the new

Yield Parameters:

value (Object) —

the current #value of the Agent
args (Array<Object>) —

zero or more arguments to pass to the action

Yield Returns:

(Object) —

the new value of the Agent

Returns:

(Boolean) —

true if the action is successfully enqueued, false if the Agent is #failed?

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 294


def send_off(*args, &action)
  enqueue_action_job(action, args, Concurrent.global_io_executor)
end

#send_off!(args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#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

Parameters:

args (Array<Object>) —

zero or more arguments to be passed to the action
action (Proc) —

the action dispatch to be enqueued

Yields:

(agent, value, *args) —

process the old value and return the new

Yield Parameters:

value (Object) —

the current #value of the Agent
args (Array<Object>) —

zero or more arguments to pass to the action

Yield Returns:

(Object) —

the new value of the Agent

Returns:

(Boolean) —

true if the action is successfully enqueued

Raises:

(Concurrent::Agent::Error) —

if the Agent is #failed?

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 302


def send_off!(*args, &action)
  raise Error.new unless send_off(*args, &action)
  true
end

#send_via(executor, args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#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

Parameters:

args (Array<Object>) —

zero or more arguments to be passed to the action
action (Proc) —

the action dispatch to be enqueued
executor (Concurrent::ExecutorService) —

the executor on which the action is to be dispatched

Yields:

(agent, value, *args) —

process the old value and return the new

Yield Parameters:

value (Object) —

the current #value of the Agent
args (Array<Object>) —

zero or more arguments to pass to the action

Yield Returns:

(Object) —

the new value of the Agent

Returns:

(Boolean) —

true if the action is successfully enqueued, false if the Agent is #failed?

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 311


def send_via(executor, *args, &action)
  enqueue_action_job(action, args, executor)
end

#send_via!(executor, args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#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

Parameters:

args (Array<Object>) —

zero or more arguments to be passed to the action
action (Proc) —

the action dispatch to be enqueued
executor (Concurrent::ExecutorService) —

the executor on which the action is to be dispatched

Yields:

(agent, value, *args) —

process the old value and return the new

Yield Parameters:

value (Object) —

the current #value of the Agent
args (Array<Object>) —

zero or more arguments to pass to the action

Yield Returns:

(Object) —

the new value of the Agent

Returns:

(Boolean) —

true if the action is successfully enqueued

Raises:

(Concurrent::Agent::Error) —

if the Agent is #failed?

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 319


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.

Returns:

(Object) —

the current value

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 229


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.

Parameters:

timeout (Float) (defaults to: nil) —

the maximum number of seconds to wait

Returns:

(Boolean) —

true if all actions complete before timeout else false

[ GitHub ]

# File 'lib/concurrent-ruby/concurrent/agent.rb', line 393


def wait(timeout = nil)
  latch = Concurrent::CountDownLatch.new(1)
  enqueue_await_job(latch)
  latch.wait(timeout)
end

Class: Concurrent::Agent

Overview

Example

create an agent with an initial value

send a few update requests

wait for them to complete

get the current value

Observation

Nested Actions

Constant Summary

Class Method Summary

Instance Attribute Summary

Instance Method Summary

Concern::Observable - Included

Synchronization::LockableObject - Inherited

Constructor Details

.new(initial, opts = {}) ⇒ Agent

Class Method Details

.await(*agents) ⇒ Boolean

.await_for(timeout, *agents) ⇒ Boolean

.await_for!(timeout, *agents) ⇒ Boolean

Instance Attribute Details

#error_mode (readonly)

#failed? ⇒ Boolean (readonly) Also known as: #stopped?

#stopped? (readonly)

Instance Method Details

#<<(action) ⇒ Agent

#await ⇒ Boolean

#await_for(timeout) ⇒ Boolean

#await_for!(timeout) ⇒ Boolean

#deref

#enqueue_action_job(action, args, executor) (private)

#enqueue_await_job(latch) (private)

#error ⇒ nil, Error Also known as: #reason

#execute_next_job (private)

#handle_error(error) (private)

#ns_enqueue_job(job, index = nil) (private)

#ns_find_last_job_for_thread (private)

#ns_initialize(initial, opts) (private)

#ns_post_next_job (private)

#ns_validate(value) (private)

#post(*args, &action)

#reason

#restart(new_value, opts = {}) ⇒ Boolean

#send(*args, &action) {|agent, value, *args| ... } ⇒ Boolean

#send!(*args, &action) {|agent, value, *args| ... } ⇒ Boolean

#send_off(*args, &action) {|agent, value, *args| ... } ⇒ Boolean Also known as: #post

#send_off!(*args, &action) {|agent, value, *args| ... } ⇒ Boolean

#send_via(executor, *args, &action) {|agent, value, *args| ... } ⇒ Boolean

#send_via!(executor, *args, &action) {|agent, value, *args| ... } ⇒ Boolean

#value ⇒ Object Also known as: #deref

#wait(timeout = nil) ⇒ Boolean

`Concern::Observable` - Included

`Synchronization::LockableObject` - Inherited

.new(initial, opts = {}) ⇒ `Agent`

.await(*agents) ⇒ `Boolean`

.await_for(timeout, *agents) ⇒ `Boolean`

.await_for!(timeout, *agents) ⇒ `Boolean`

#failed? ⇒ `Boolean` (readonly) Also known as: #stopped?

#<<(action) ⇒ `Agent`

#await ⇒ `Boolean`

#await_for(timeout) ⇒ `Boolean`

#await_for!(timeout) ⇒ `Boolean`

#error ⇒ `nil`, Error Also known as: #reason

#restart(new_value, opts = {}) ⇒ `Boolean`

#send(args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#send!(args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#send_off(args, &action) {|agent, value, args| ... } ⇒ `Boolean` Also known as: #post

#send_off!(args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#send_via(executor, args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#send_via!(executor, args, &action) {|agent, value, args| ... } ⇒ `Boolean`

#wait(timeout = nil) ⇒ `Boolean`