Class: Concurrent::TimerTask

Inherits:
Object
  • Object
show all
Includes:
Dereferenceable, Observable, RubyExecutor
Defined in:
lib/concurrent/timer_task.rb

Overview

A very common currency 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. TimerTask can also be configured with a timeout value allowing it to kill a task that runs too long.

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.

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. Derefencing 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). If the timeout interval is exceeded the observer will receive a Concurrent::TimeoutError object as the third argument.

Examples:

Basic usage

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

task.execution_interval #=> 60 (default)
task.timeout_interval   #=> 30 (default)

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

task.shutdown #=> true

Configuring :execution_interval and :timeout_interval

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

task.execution_interval #=> 5
task.timeout_interval   #=> 5

Immediate execution with :run_now

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

#=> 'Boom!'

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.times{ print 'Boom! ' }
  print "\n"
  task.execution_interval += 1
  if task.execution_interval > 5
    puts 'Stopping...'
    task.shutdown
  end
end

timer_task.execute # blocking call - this task will stop itself
#=> 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"
    elsif ex.is_a?(Concurrent::TimeoutError)
      print "(#{time}) Execution timed out\n"
    else
      print "(#{time}) Execution failed with error #{ex}\n"
    end
  end
end

task = Concurrent::TimerTask.new(execution_interval: 1, timeout_interval: 1){ 42 }
task.add_observer(TaskObserver.new)
task.execute

#=> (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, timeout_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:

Constant Summary collapse

EXECUTION_INTERVAL =

Default :execution_interval in seconds.

60
TIMEOUT_INTERVAL =

Default :timeout_interval in seconds.

30

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

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

Note:

Calls Concurrent::Dereferenceable# set_deref_options passing opts. All options supported by Concurrent::Dereferenceable can be set during object initialization.

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 (Integer)

    number of seconds between task executions (default: EXECUTION_INTERVAL)

  • :timeout_interval (Integer)

    number of seconds a task can run before it is considered to have failed (default: TIMEOUT_INTERVAL)

  • :run_now (Boolean)

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

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.

See Also:



183
184
185
186
187
188
189
190
191
192
193
194
195
196
# File 'lib/concurrent/timer_task.rb', line 183

def initialize(opts = {}, &task)
  raise ArgumentError.new('no block given') unless block_given?

  init_executor
  set_deref_options(opts)

  self.execution_interval = opts[:execution] || opts[:execution_interval] || EXECUTION_INTERVAL
  self.timeout_interval = opts[:timeout] || opts[:timeout_interval] || TIMEOUT_INTERVAL
  @run_now = opts[:now] || opts[:run_now]
  @executor = Concurrent::SafeTaskExecutor.new(task)
  @running = Concurrent::AtomicBoolean.new(false)

  self.observers = CopyOnNotifyObserverSet.new
end

Instance Attribute Details

#execution_intervalFixnum

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.



246
247
248
249
250
251
# File 'lib/concurrent/timer_task.rb', line 246

def execution_interval
  mutex.lock
  @execution_interval
ensure
  mutex.unlock
end

#fallback_policyObject (readonly) Originally defined in module Executor

The policy defining how rejected tasks (tasks received once the queue size reaches the configured max_queue, or after the executor has shut down) are handled. Must be one of the values specified in FALLBACK_POLICIES.

#mutexObject (readonly, protected) Originally defined in module RubyExecutor

Returns the value of attribute mutex

#observersObject (protected) Originally defined in module Observable

Returns the value of attribute observers

#stop_eventObject (readonly, protected) Originally defined in module RubyExecutor

Returns the value of attribute stop_event

#stopped_eventObject (readonly, protected) Originally defined in module RubyExecutor

Returns the value of attribute stopped_event

#timeout_intervalFixnum

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.



272
273
274
275
276
277
# File 'lib/concurrent/timer_task.rb', line 272

def timeout_interval
  mutex.lock
  @timeout_interval
ensure
  mutex.unlock
end

Class Method Details

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

Note:

Calls Concurrent::Dereferenceable# set_deref_options passing opts. All options supported by Concurrent::Dereferenceable can be set during object initialization.

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 (Integer)

    number of seconds between task executions (default: EXECUTION_INTERVAL)

  • :timeout_interval (Integer)

    number of seconds a task can run before it is considered to have failed (default: TIMEOUT_INTERVAL)

  • :run_now (Boolean)

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

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:

Raises:

  • ArgumentError when no block is given.

See Also:

Since:

  • 0.6.0



239
240
241
# File 'lib/concurrent/timer_task.rb', line 239

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

Instance Method Details

#add_observer(*args, &block) ⇒ Object Originally defined in module Observable

Returns the added observer

Returns:

  • (Object)

    the added observer

#can_overflow?Boolean Originally defined in module Executor

Note:

Always returns false

Does the task queue have a maximum size?

Returns:

  • (Boolean)

    True if the task queue has a maximum size else false.

#count_observersInteger Originally defined in module Observable

Returns the observers count

Returns:

  • (Integer)

    the observers count

#delete_observer(*args) ⇒ Object Originally defined in module Observable

Returns the deleted observer

Returns:

  • (Object)

    the deleted observer

#delete_observersObservable Originally defined in module Observable

Returns self

Returns:

#executeTimerTask

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:

Since:

  • 0.6.0



220
221
222
223
224
225
226
227
228
# File 'lib/concurrent/timer_task.rb', line 220

def execute
  mutex.synchronize do
    if @running.false?
      @running.make_true
      schedule_next_task(@run_now ? 0 : @execution_interval)
    end
  end
  self
end

#init_executorObject (protected) Originally defined in module RubyExecutor

Initialize the executor by creating and initializing all the internal synchronization objects.

#init_mutexObject (protected) Originally defined in module Dereferenceable

Note:

This method must be called from within the constructor of the including class.

Initializes the internal Mutex.

See Also:

#killObject Originally defined in module RubyExecutor

Begin an immediate shutdown. In-progress tasks will be allowed to complete but enqueued tasks will be dismissed and no new tasks will be accepted. Has no additional effect if the thread pool is not running.

#log(level, progname, message = nil, &block) ⇒ Object Originally defined in module Logging

Logs through Configuration#logger, it can be overridden by setting @logger

Parameters:

  • level (Integer)

    one of Logger::Severity constants

  • progname (String)

    e.g. a path of an Actor

  • message (String, nil) (defaults to: nil)

    when nil block is used to generate the message

Yield Returns:

  • (String)

    a message

#running?Boolean

Is the executor running?

Returns:

  • (Boolean)

    true when running, false when shutting down or shutdown



201
202
203
# File 'lib/concurrent/timer_task.rb', line 201

def running?
  @running.true?
end

#serialized?Boolean Originally defined in module Executor

Note:

Always returns false

Does this executor guarantee serialization of its operations?

Returns:

  • (Boolean)

    True if the executor guarantees that all operations will be post in the order they are received and no two operations may occur simultaneously. Else false.

#set_deref_options(opts = {}) ⇒ Object (protected) Originally defined in module Dereferenceable

Note:

Most classes that include this module will call #set_deref_options

Set the options which define the operations #value performs before returning data to the caller (dereferencing).

from within the constructor, thus allowing these options to be set at object creation.

Parameters:

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

    the options defining dereference behavior.

Options Hash (opts):

  • :dup_on_deref (String) — default: false

    call #dup before returning the data

  • :freeze_on_deref (String) — default: false

    call #freeze before returning the data

  • :copy_on_deref (String) — default: nil

    call the given Proc passing the internal value and returning the value returned from the proc

#shutdownObject Originally defined in module RubyExecutor

Begin an orderly shutdown. Tasks already in the queue will be executed, but no new tasks will be accepted. Has no additional effect if the thread pool is not running.

#shutdown?Boolean Originally defined in module RubyExecutor

Is the executor shutdown?

Returns:

  • (Boolean)

    true when shutdown, false when shutting down or running

#shuttingdown?Boolean Originally defined in module RubyExecutor

Is the executor shuttingdown?

Returns:

  • (Boolean)

    true when not running and not shutdown, else false

#valueObject Also known as: deref Originally defined in module Dereferenceable

Return the value this object represents after applying the options specified by the #set_deref_options method.

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.

This method is thread-safe and synchronized with the internal #mutex.

Returns:

  • (Object)

    the current value of the object

#value=(val) ⇒ Object (protected) Originally defined in module Dereferenceable

Set the internal value of this object

Parameters:

  • val (Object)

    the new value

#wait_for_termination(timeout = nil) ⇒ Boolean Originally defined in module RubyExecutor

Note:

Does not initiate shutdown or termination. Either shutdown or kill must be called before this method (or on another thread).

Block until executor shutdown is complete or until timeout seconds have passed.

Parameters:

  • timeout (Integer) (defaults to: nil)

    the maximum number of seconds to wait for shutdown to complete

Returns:

  • (Boolean)

    true if shutdown complete or false on timeout

#with_observer(*args, &block) ⇒ Observable Originally defined in module Observable

as #add_observer but it can be used for chaining

Returns: