/
agent.rb
587 lines (536 loc) · 21.5 KB
/
agent.rb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
require 'concurrent/configuration'
require 'concurrent/atomic/atomic_reference'
require 'concurrent/atomic/thread_local_var'
require 'concurrent/collection/copy_on_write_observer_set'
require 'concurrent/concern/observable'
require 'concurrent/synchronization'
module Concurrent
# `Agent` is inspired by Clojure's [agent](http://clojure.org/agents)
# 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:
#
# 1. The given `action` will be applied to the state of the Agent and the
# `args`, if any were supplied.
# 2. The return value of `action` will be passed to the validator lambda,
# if one has been set on the Agent.
# 3. 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.
# 4. If any observers were added to the Agent, they will be notified. See
# `#add_observer` for details.
# 5. 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 {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 = 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
# ```
#
# @!macro agent_await_warning
#
# **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.
#
# @!macro thread_safe_variable_comparison
#
# @see http://clojure.org/Agents Clojure Agents
# @see http://clojure.org/state Values and Change - Clojure's approach to Identity and State
class Agent < Synchronization::LockableObject
include Concern::Observable
ERROR_MODES = [:continue, :fail].freeze
private_constant :ERROR_MODES
AWAIT_FLAG = ::Object.new
private_constant :AWAIT_FLAG
AWAIT_ACTION = ->(value, latch) { latch.count_down; AWAIT_FLAG }
private_constant :AWAIT_ACTION
DEFAULT_ERROR_HANDLER = ->(agent, error) { nil }
private_constant :DEFAULT_ERROR_HANDLER
DEFAULT_VALIDATOR = ->(value) { true }
private_constant :DEFAULT_VALIDATOR
Job = Struct.new(:action, :args, :executor, :caller)
private_constant :Job
# Raised during action processing or any other time in an Agent's lifecycle.
class Error < StandardError
def initialize(message = nil)
message ||= 'agent must be restarted before jobs can post'
super(message)
end
end
# Raised when a new value obtained during action processing or at `#restart`
# fails validation.
class ValidationError < Error
def initialize(message = nil)
message ||= 'invalid value'
super(message)
end
end
# The error mode this Agent is operating in. See {#initialize} for details.
attr_reader :error_mode
# 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.
#
# @param [Object] initial the initial value
# @param [Hash] opts the configuration options
#
# @option opts [Symbol] :error_mode either `:continue` or `:fail`
# @option opts [nil, Proc] :error_handler the (optional) error handler
# @option opts [nil, Proc] :validator the (optional) validation procedure
def initialize(initial, opts = {})
super()
synchronize { ns_initialize(initial, opts) }
end
# The current value (state) of the Agent, irrespective of any pending or
# in-progress actions. The value is always available and is non-blocking.
#
# @return [Object] the current value
def value
@current.value # TODO (pitr 12-Sep-2015): broken unsafe read?
end
alias_method :deref, :value
# 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`.
#
# @return [nil, Error] the error which caused the failure when {#failed?}
def error
@error.value
end
alias_method :reason, :error
# @!macro agent_send
#
# 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
#
# @param [Array<Object>] args zero or more arguments to be passed to
# the action
# @param [Proc] action the action dispatch to be enqueued
#
# @yield [agent, value, *args] process the old value and return the new
# @yieldparam [Object] value the current {#value} of the Agent
# @yieldparam [Array<Object>] args zero or more arguments to pass to the
# action
# @yieldreturn [Object] the new value of the Agent
#
# @!macro send_return
# @return [Boolean] true if the action is successfully enqueued, false if
# the Agent is {#failed?}
def send(*args, &action)
enqueue_action_job(action, args, Concurrent.global_fast_executor)
end
# @!macro agent_send
#
# @!macro send_bang_return_and_raise
# @return [Boolean] true if the action is successfully enqueued
# @raise [Concurrent::Agent::Error] if the Agent is {#failed?}
def send!(*args, &action)
raise Error.new unless send(*args, &action)
true
end
# @!macro agent_send
# @!macro send_return
def send_off(*args, &action)
enqueue_action_job(action, args, Concurrent.global_io_executor)
end
alias_method :post, :send_off
# @!macro agent_send
# @!macro send_bang_return_and_raise
def send_off!(*args, &action)
raise Error.new unless send_off(*args, &action)
true
end
# @!macro agent_send
# @!macro send_return
# @param [Concurrent::ExecutorService] executor the executor on which the
# action is to be dispatched
def send_via(executor, *args, &action)
enqueue_action_job(action, args, executor)
end
# @!macro agent_send
# @!macro send_bang_return_and_raise
# @param [Concurrent::ExecutorService] executor the executor on which the
# action is to be dispatched
def send_via!(executor, *args, &action)
raise Error.new unless send_via(executor, *args, &action)
true
end
# 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.
#
# @param [Proc] action the action dispatch to be enqueued
# @return [Concurrent::Agent] self
# @see #send_off
def <<(action)
send_off(&action)
self
end
# 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
# ```
#
# @return [Boolean] self
#
# @!macro agent_await_warning
def await
wait(nil)
self
end
# 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.
#
# @param [Float] timeout the maximum number of seconds to wait
# @return [Boolean] true if all actions complete before timeout else false
#
# @!macro agent_await_warning
def await_for(timeout)
wait(timeout.to_f)
end
# 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.
#
# @param [Float] timeout the maximum number of seconds to wait
# @return [Boolean] true if all actions complete before timeout
#
# @raise [Concurrent::TimeoutError] when timout is reached
#
# @!macro agent_await_warning
def await_for!(timeout)
raise Concurrent::TimeoutError unless wait(timeout.to_f)
true
end
# 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.
#
# @param [Float] timeout the maximum number of seconds to wait
# @return [Boolean] true if all actions complete before timeout else false
#
# @!macro agent_await_warning
def wait(timeout = nil)
latch = Concurrent::CountDownLatch.new(1)
enqueue_await_job(latch)
latch.wait(timeout)
end
# Is the Agent in a failed state?
#
# @see #restart
def failed?
!@error.value.nil?
end
alias_method :stopped?, :failed?
# 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.
#
# @param [Object] new_value the new value for the Agent once restarted
# @param [Hash] opts the configuration options
# @option opts [Symbol] :clear_actions true if all enqueued but unprocessed
# actions should be discarded on restart, else false (default: false)
# @return [Boolean] true
#
# @raise [Concurrent:AgentError] when not failed
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
class << self
# 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.
#
# @param [Array<Concurrent::Agent>] agents the Agents on which to wait
# @return [Boolean] true
#
# @!macro agent_await_warning
def await(*agents)
agents.each { |agent| agent.await }
true
end
# 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.
#
# @param [Float] timeout the maximum number of seconds to wait
# @param [Array<Concurrent::Agent>] agents the Agents on which to wait
# @return [Boolean] true if all actions complete before timeout else false
#
# @!macro agent_await_warning
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
# 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.
#
# @param [Float] timeout the maximum number of seconds to wait
# @param [Array<Concurrent::Agent>] agents the Agents on which to wait
# @return [Boolean] true if all actions complete before timeout
#
# @raise [Concurrent::TimeoutError] when timout is reached
# @!macro agent_await_warning
def await_for!(timeout, *agents)
raise Concurrent::TimeoutError unless await_for(timeout, *agents)
true
end
end
private
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
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
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
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
def ns_post_next_job
@queue.first.executor.post { execute_next_job }
end
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
def ns_validate(value)
@validator.call(value)
rescue
false
end
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
def ns_find_last_job_for_thread
@queue.rindex { |job| job.caller == Thread.current.object_id }
end
end
end