Class: Concurrent::TimerTask

Inherits:

RubyExecutorService

• Object
• Synchronization::LockableObject
• AbstractExecutorService
• RubyExecutorService
• Concurrent::TimerTask

Includes:

Concern::Dereferenceable, Concern::Observable

Defined in:

lib/concurrent-ruby/concurrent/timer_task.rb

Overview

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

## Copy Options

Object references in Ruby are mutable. This can lead to serious problems when the Concern::Dereferenceable#value of an object is a mutable reference. Which is always the case unless the value is a ‘Fixnum`, `Symbol`, or similar “primitive” data type. Each instance can be configured with a few options that can help protect the program from potentially dangerous operations. Each of these options can be optionally set when the object instance is created:

• ‘:dup_on_deref` When true the object will call the `#dup` method on the `value` object every time the `#value` method is called (default: false)

• ‘:freeze_on_deref` When true the object will call the `#freeze` method on the `value` object every time the `#value` method is called (default: false)

• ‘:copy_on_deref` When given a `Proc` object the `Proc` will be run every time the `#value` method is called. The `Proc` will be given the current `value` as its only argument and the result returned by the block will be the return value of the `#value` call. When `nil` this option will be ignored (default: nil)

When multiple deref options are set the order of operations is strictly defined. The order of deref operations is:

• ‘:copy_on_deref`

• ‘:dup_on_deref`

• ‘:freeze_on_deref`

Because of this ordering there is no need to ‘#freeze` an object created by a provided `:copy_on_deref` block. Simply set `:freeze_on_deref` to `true`. Setting both `:dup_on_deref` to `true` and `:freeze_on_deref` to `true` is as close to the behavior of a “pure” functional language (like Erlang, Clojure, or Haskell) as we are likely to get in Ruby.

Examples:

Basic usage

task = Concurrent::TimerTask.new{ puts 'Boom!' }
task.execute

task.execution_interval #=> 60 (default)

# wait 60 seconds...
#=> 'Boom!'

task.shutdown #=> true

Configuring ‘:execution_interval`

task = Concurrent::TimerTask.new(execution_interval: 5) do
       puts 'Boom!'
     end

task.execution_interval #=> 5

Immediate execution with ‘:run_now`

task = Concurrent::TimerTask.new(run_now: true){ puts 'Boom!' }
task.execute

#=> 'Boom!'

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

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

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

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 Also:

• http://ruby-doc.org/stdlib-2.0/libdoc/observer/rdoc/Observable.html
• http://docs.oracle.com/javase/7/docs/api/java/util/TimerTask.html

Constant Summary

EXECUTION_INTERVAL =

Default ‘:execution_interval` in seconds.

60

FIXED_DELAY =

Maintain the interval between the end of one execution and the start of the next execution.

:fixed_delay

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.

:fixed_rate

DEFAULT_INTERVAL_TYPE =

Default ‘:interval_type`

FIXED_DELAY

Constants inherited from AbstractExecutorService

AbstractExecutorService::FALLBACK_POLICIES

Instance Attribute Summary

• #execution_interval ⇒ Fixnum

  Number of seconds after the task completes before the task is performed again.

• #interval_type ⇒ Symbol readonly

  Method to calculate the interval between executions.

• #timeout_interval ⇒ Fixnum

  Number of seconds the task can run before it is considered to have failed.

Class Method Summary

• .execute(opts = {}) {|task| ... } ⇒ TimerTask

  Create and execute a new ‘TimerTask`.

Instance Method Summary

• #execute ⇒ TimerTask

  Execute a previously created ‘TimerTask`.

• #initialize(opts = {}) {|task| ... } ⇒ TimerTask constructor

  Create a new TimerTask with the given task and configuration.

• #running? ⇒ Boolean

  Is the executor running?.

Methods included from Concern::Observable

#add_observer, #count_observers, #delete_observer, #delete_observers, #with_observer

Methods included from Concern::Dereferenceable

#value

Constructor Details

#initialize(opts = {}) {|task| ... } ⇒ TimerTask

Create a new TimerTask with the given task and configuration.

Parameters:

• opts (Hash) (defaults to: {}) —

  the options defining task execution.

Options Hash (opts):

• :execution_interval (Float) —

  number of seconds between task executions (default: EXECUTION_INTERVAL)

• :run_now (Boolean) —

  Whether to run the task immediately upon instantiation or to wait until the first # execution_interval has passed (default: false)

• executor, (Executor) —

  default is ‘global_io_executor`

• :dup_on_deref (Boolean) — default: false —

  Call ‘#dup` before returning the data from Concern::Dereferenceable#value

• :freeze_on_deref (Boolean) — default: false —

  Call ‘#freeze` before returning the data from Concern::Dereferenceable#value

• :copy_on_deref (Proc) — default: nil —

  When calling the Concern::Dereferenceable#value method, call the given proc passing the internal value as the sole argument then return the new value returned from the proc.

Yields:

• to the block after :execution_interval seconds have passed since the last yield

Yield Parameters:

• 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`.

Raises:

• ArgumentError when no block is given.

# 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

Instance Attribute Details

#execution_interval ⇒ Fixnum

Returns Number of seconds after the task completes before the task is performed again.

Returns:

• (Fixnum) —

  Number of seconds after the task completes before the task is performed again.

# File 'lib/concurrent-ruby/concurrent/timer_task.rb', line 259

def execution_interval
  synchronize { @execution_interval }
end

#interval_type ⇒ Symbol (readonly)

Returns method to calculate the interval between executions.

Returns:

• (Symbol) —

  method to calculate the interval between executions

# File 'lib/concurrent-ruby/concurrent/timer_task.rb', line 276

def interval_type
  @interval_type
end

#timeout_interval ⇒ Fixnum

Returns Number of seconds the task can run before it is considered to have failed.

Returns:

• (Fixnum) —

  Number of seconds the task can run before it is considered to have failed.

# 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

Class Method Details

.execute(opts = {}) {|task| ... } ⇒ TimerTask

Create and execute a new ‘TimerTask`.

Examples:

task = Concurrent::TimerTask.execute(execution_interval: 10){ print "Hello World\n" }
task.running? #=> true

Parameters:

• opts (Hash) (defaults to: {}) —

  the options defining task execution.

Options Hash (opts):

• :execution_interval (Float) —

  number of seconds between task executions (default: EXECUTION_INTERVAL)

• :run_now (Boolean) —

  Whether to run the task immediately upon instantiation or to wait until the first # execution_interval has passed (default: false)

• executor, (Executor) —

  default is ‘global_io_executor`

• :dup_on_deref (Boolean) — default: false —

  Call ‘#dup` before returning the data from Concern::Dereferenceable#value

• :freeze_on_deref (Boolean) — default: false —

  Call ‘#freeze` before returning the data from Concern::Dereferenceable#value

• :copy_on_deref (Proc) — default: nil —

  When calling the Concern::Dereferenceable#value method, call the given proc passing the internal value as the sole argument then return the new value returned from the proc.

Yields:

• to the block after :execution_interval seconds have passed since the last yield

Yield Parameters:

• 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`.

Returns:

• (TimerTask) —

  the new ‘TimerTask`

Raises:

• ArgumentError when no block is given.

# File 'lib/concurrent-ruby/concurrent/timer_task.rb', line 252

def self.execute(opts = {}, &task)
  TimerTask.new(opts, &task).execute
end

Instance Method Details

#execute ⇒ TimerTask

Execute a previously created ‘TimerTask`.

Examples:

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

Instance and execute in one line

task = Concurrent::TimerTask.new(execution_interval: 10){ print "Hello World\n" }.execute
task.running? #=> true

Returns:

• (TimerTask) —

  a reference to ‘self`

# 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

#running? ⇒ Boolean

Is the executor running?

Returns:

• (Boolean) —

  ‘true` when running, `false` when shutting down or shutdown

# File 'lib/concurrent-ruby/concurrent/timer_task.rb', line 218

def running?
  @running.true?
end