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 in 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 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 task object is passed into the block as a block argument 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 threes 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.run!

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

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

task.stop #=> 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.run!

#=> 'Boom!'

Last ‘#value` and `Dereferenceable` mixin

task = Concurrent::TimerTask.new(
  dup_on_deref: true,
  execution_interval: 5
){ Time.now }

task.run!
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.stop
  end
end

timer_task.run # 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.run!

#=> (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.stop

task = Concurrent::TimerTask.new(execution_interval: 1, timeout_interval: 1){ sleep }
task.add_observer(TaskObserver.new)
task.run!

#=> (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.stop

task = Concurrent::TimerTask.new(execution_interval: 1){ raise StandardError }
task.add_observer(TaskObserver.new)
task.run!

#=> (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.stop

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

Methods included from Observable

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

Methods included from RubyExecutor

#<<, #kill, #post, #shutdown, #shutdown?, #shuttingdown?, #wait_for_termination

Methods included from Logging

#log

Methods included from Executor

#can_overflow?

Methods included from Dereferenceable

#value

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:



186
187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/concurrent/timer_task.rb', line 186

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



249
250
251
252
253
254
 
# File 'lib/concurrent/timer_task.rb', line 249

def execution_interval
  mutex.lock
  @execution_interval
ensure
  mutex.unlock
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.



275
276
277
278
279
280
 
# File 'lib/concurrent/timer_task.rb', line 275

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



242
243
244
 
# File 'lib/concurrent/timer_task.rb', line 242

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

.run!(*args, &block) ⇒ Object

Deprecated.

Updated to use ‘Executor` instead of `Runnable`



311
312
313
314
 
# File 'lib/concurrent/timer_task.rb', line 311

def self.run!(*args, &block)
  warn "[DEPRECATED] `run!` is deprecated, please use `execute` instead."
  Concurrent::Runnable::Context.new(TimerTask.new(*args, &block))
end

Instance Method Details

#cancel(*args) ⇒ Object

Deprecated.

Updated to use ‘Executor` instead of `Runnable`



305
 
# File 'lib/concurrent/timer_task.rb', line 305

def cancel(*args) deprecated(:cancel, :shutdown, *args); end

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

Since:

  • 0.6.0



223
224
225
226
227
228
229
230
231
 
# File 'lib/concurrent/timer_task.rb', line 223

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

#run ⇒ Object

Deprecated.

Updated to use ‘Executor` instead of `Runnable`

Raises:

  • (Concurrent::Runnable::LifecycleError) 


317
318
319
320
321
322
 
# File 'lib/concurrent/timer_task.rb', line 317

def run
  raise Concurrent::Runnable::LifecycleError.new('already running') if @running.true?
  self.execute
  self.wait_for_termination
  true
end

#run!(*args) ⇒ Object

Deprecated.

Updated to use ‘Executor` instead of `Runnable`



308
 
# File 'lib/concurrent/timer_task.rb', line 308

def run!(*args) deprecated(:run!, :execute); end

#running? ⇒ Boolean

Is the executor running?

Returns:

  • (Boolean) —

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



204
205
206
 
# File 'lib/concurrent/timer_task.rb', line 204

def running?
  @running.true?
end

#stop(*args) ⇒ Object

Deprecated.

Updated to use ‘Executor` instead of `Runnable`



302
 
# File 'lib/concurrent/timer_task.rb', line 302

def stop(*args) deprecated(:stop, :shutdown, *args); end

#terminate(*args) ⇒ Object

Deprecated.

Updated to use ‘Executor` instead of `Runnable`



299
 
# File 'lib/concurrent/timer_task.rb', line 299

def terminate(*args) deprecated(:terminate, :kill, *args); end