class Concurrent::TimerTask
A very common concurrency pattern is to run a thread that performs a task at regular intervals. The thread that performs the task sleeps for the given interval then wakes up and performs the task. Lather, rinse, repeat… This pattern causes two problems. First, it is difficult to test the business logic of the task because the task itself is tightly coupled with the concurrency logic. Second, an exception raised while performing the task can cause the entire thread to abend. In a long-running application where the task thread is intended to run for days/weeks/years a crashed task thread can pose a significant problem. ‘TimerTask` alleviates both problems.
When a ‘TimerTask` is launched it starts a thread for monitoring the execution interval. The `TimerTask` thread does not perform the task, however. Instead, the TimerTask
launches the task on a separate thread. Should the task experience an unrecoverable crash only the task thread will crash. This makes the `TimerTask` very fault tolerant. Additionally, the `TimerTask` thread can respond to the success or failure of the task, performing logging or ancillary operations.
One other advantage of ‘TimerTask` is that it forces the business logic to be completely decoupled from the concurrency logic. The business logic can be tested separately then passed to the `TimerTask` for scheduling and running.
A ‘TimerTask` supports two different types of interval calculations. A fixed delay will always wait the same amount of time between the completion of one task and the start of the next. A fixed rate will attempt to maintain a constant rate of execution regardless of the duration of the task. For example, if a fixed rate task is scheduled to run every 60 seconds but the task itself takes 10 seconds to complete, the next task will be scheduled to run 50 seconds after the start of the previous task. If the task takes 70 seconds to complete, the next task will be start immediately after the previous task completes. Tasks will not be executed concurrently.
In some cases it may be necessary for a ‘TimerTask` to affect its own execution cycle. To facilitate this, a reference to the TimerTask
instance is passed as an argument to the provided block every time the task is executed.
The ‘TimerTask` class includes the `Dereferenceable` mixin module so the result of the last execution is always available via the `#value` method. Dereferencing options can be passed to the `TimerTask` during construction or at any later time using the `#set_deref_options` method.
‘TimerTask` supports notification through the Ruby standard library {ruby-doc.org/stdlib-2.0/libdoc/observer/rdoc/Observable.html Observable} module. On execution the `TimerTask` will notify the observers with three arguments: time of execution, the result of the block (or nil on failure), and any raised exceptions (or nil on success).
@!macro copy_options
@example Basic usage
task = Concurrent::TimerTask.new{ puts 'Boom!' } task.execute task.execution_interval #=> 60 (default) # wait 60 seconds... #=> 'Boom!' task.shutdown #=> true
@example Configuring ‘:execution_interval`
task = Concurrent::TimerTask.new(execution_interval: 5) do puts 'Boom!' end task.execution_interval #=> 5
@example Immediate execution with ‘:run_now`
task = Concurrent::TimerTask.new(run_now: true){ puts 'Boom!' } task.execute #=> 'Boom!'
@example Configuring ‘:interval_type` with either :fixed_delay or :fixed_rate, default is :fixed_delay
task = Concurrent::TimerTask.new(execution_interval: 5, interval_type: :fixed_rate) do puts 'Boom!' end task.interval_type #=> :fixed_rate
@example Last ‘#value` and `Dereferenceable` mixin
task = Concurrent::TimerTask.new( dup_on_deref: true, execution_interval: 5 ){ Time.now } task.execute Time.now #=> 2013-11-07 18:06:50 -0500 sleep(10) task.value #=> 2013-11-07 18:06:55 -0500
@example Controlling execution from within the block
timer_task = Concurrent::TimerTask.new(execution_interval: 1) do |task| task.execution_interval.to_i.times{ print 'Boom! ' } print "\n" task.execution_interval += 1 if task.execution_interval > 5 puts 'Stopping...' task.shutdown end end timer_task.execute #=> Boom! #=> Boom! Boom! #=> Boom! Boom! Boom! #=> Boom! Boom! Boom! Boom! #=> Boom! Boom! Boom! Boom! Boom! #=> Stopping...
@example Observation
class TaskObserver def update(time, result, ex) if result print "(#{time}) Execution successfully returned #{result}\n" else print "(#{time}) Execution failed with error #{ex}\n" end end end task = Concurrent::TimerTask.new(execution_interval: 1){ 42 } task.add_observer(TaskObserver.new) task.execute sleep 4 #=> (2013-10-13 19:08:58 -0400) Execution successfully returned 42 #=> (2013-10-13 19:08:59 -0400) Execution successfully returned 42 #=> (2013-10-13 19:09:00 -0400) Execution successfully returned 42 task.shutdown task = Concurrent::TimerTask.new(execution_interval: 1){ sleep } task.add_observer(TaskObserver.new) task.execute #=> (2013-10-13 19:07:25 -0400) Execution timed out #=> (2013-10-13 19:07:27 -0400) Execution timed out #=> (2013-10-13 19:07:29 -0400) Execution timed out task.shutdown task = Concurrent::TimerTask.new(execution_interval: 1){ raise StandardError } task.add_observer(TaskObserver.new) task.execute #=> (2013-10-13 19:09:37 -0400) Execution failed with error StandardError #=> (2013-10-13 19:09:38 -0400) Execution failed with error StandardError #=> (2013-10-13 19:09:39 -0400) Execution failed with error StandardError task.shutdown
@see ruby-doc.org/stdlib-2.0/libdoc/observer/rdoc/Observable.html @see docs.oracle.com/javase/7/docs/api/java/util/TimerTask.html
Constants
- DEFAULT_INTERVAL_TYPE
-
Default ‘:interval_type`
- EXECUTION_INTERVAL
-
Default ‘:execution_interval` in seconds.
- FIXED_DELAY
-
Maintain the interval between the end of one execution and the start of the next execution.
- FIXED_RATE
-
Maintain the interval between the start of one execution and the start of the next. If execution time exceeds the interval, the next execution will start immediately after the previous execution finishes. Executions will not run concurrently.
Attributes
@!attribute [r] interval_type
@return [Symbol] method to calculate the interval between executions
Public Class Methods
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 252 def self.execute(opts = {}, &task) TimerTask.new(opts, &task).execute end
Create and execute a new ‘TimerTask`.
@!macro timer_task_initialize
@example
task = Concurrent::TimerTask.execute(execution_interval: 10){ print "Hello World\n" } task.running? #=> true
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 209 def initialize(opts = {}, &task) raise ArgumentError.new('no block given') unless block_given? super set_deref_options opts end
Create a new TimerTask
with the given task and configuration.
@!macro timer_task_initialize
@param [Hash] opts the options defining task execution. @option opts [Float] :execution_interval number of seconds between task executions (default: EXECUTION_INTERVAL) @option opts [Boolean] :run_now Whether to run the task immediately upon instantiation or to wait until the first # execution_interval has passed (default: false) @options opts [Symbol] :interval_type method to calculate the interval between executions, can be either :fixed_rate or :fixed_delay. (default: :fixed_delay) @option opts [Executor] executor, default is `global_io_executor` @!macro deref_options @raise ArgumentError when no block is given. @yield to the block after :execution_interval seconds have passed since the last yield @yieldparam task a reference to the `TimerTask` instance so that the block can control its own lifecycle. Necessary since `self` will refer to the execution context of the block rather than the running `TimerTask`. @return [TimerTask] the new `TimerTask`
Concurrent::RubyExecutorService::new
Public Instance Methods
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 235 def execute synchronize do if @running.false? @running.make_true schedule_next_task(@run_now ? 0 : @execution_interval) end end self end
Execute a previously created ‘TimerTask`.
@return [TimerTask] a reference to ‘self`
@example Instance and execute in separate steps
task = Concurrent::TimerTask.new(execution_interval: 10){ print "Hello World\n" } task.running? #=> false task.execute task.running? #=> true
@example Instance and execute in one line
task = Concurrent::TimerTask.new(execution_interval: 10){ print "Hello World\n" }.execute task.running? #=> true
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 259 def execution_interval synchronize { @execution_interval } end
@!attribute [rw] execution_interval
@return [Fixnum] Number of seconds after the task completes before the
task is performed again.
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 266 def execution_interval=(value) if (value = value.to_f) <= 0.0 raise ArgumentError.new('must be greater than zero') else synchronize { @execution_interval = value } end end
@!attribute [rw] execution_interval
@return [Fixnum] Number of seconds after the task completes before the
task is performed again.
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 218 def running? @running.true? end
Is the executor running?
@return [Boolean] ‘true` when running, `false` when shutting down or shutdown
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 281 def timeout_interval warn 'TimerTask timeouts are now ignored as these were not able to be implemented correctly' end
@!attribute [rw] timeout_interval
@return [Fixnum] Number of seconds the task can run before it is
considered to have failed.
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 288 def timeout_interval=(value) warn 'TimerTask timeouts are now ignored as these were not able to be implemented correctly' end
@!attribute [rw] timeout_interval
@return [Fixnum] Number of seconds the task can run before it is
considered to have failed.
Private Instance Methods
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 352 def calculate_next_interval(start_time) if @interval_type == FIXED_RATE run_time = Concurrent.monotonic_time - start_time [execution_interval - run_time, 0].max else # FIXED_DELAY execution_interval end end
@!visibility private
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 336 def execute_task(completion) return nil unless @running.true? start_time = Concurrent.monotonic_time _success, value, reason = @task.execute(self) if completion.try? self.value = value schedule_next_task(calculate_next_interval(start_time)) time = Time.now observers.notify_observers do [time, self.value, reason] end end nil end
@!visibility private
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 296 def ns_initialize(opts, &task) set_deref_options(opts) self.execution_interval = opts[:execution] || opts[:execution_interval] || EXECUTION_INTERVAL if opts[:interval_type] && ![FIXED_DELAY, FIXED_RATE].include?(opts[:interval_type]) raise ArgumentError.new('interval_type must be either :fixed_delay or :fixed_rate') end if opts[:timeout] || opts[:timeout_interval] warn 'TimeTask timeouts are now ignored as these were not able to be implemented correctly' end @run_now = opts[:now] || opts[:run_now] @interval_type = opts[:interval_type] || DEFAULT_INTERVAL_TYPE @task = Concurrent::SafeTaskExecutor.new(task) @executor = opts[:executor] || Concurrent.global_io_executor @running = Concurrent::AtomicBoolean.new(false) @value = nil self.observers = Collection::CopyOnNotifyObserverSet.new end
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 324 def ns_kill_execution @running.make_false super end
@!visibility private
Concurrent::AbstractExecutorService#ns_kill_execution
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 318 def ns_shutdown_execution @running.make_false super end
@!visibility private
Concurrent::RubyExecutorService#ns_shutdown_execution
Source
# File lib/concurrent-ruby/concurrent/timer_task.rb, line 330 def schedule_next_task(interval = execution_interval) ScheduledTask.execute(interval, executor: @executor, args: [Concurrent::Event.new], &method(:execute_task)) nil end
@!visibility private