Class: SidekiqUniqueJobs::TimerTask

Inherits:

Concurrent::RubyExecutorService

Object
Concurrent::RubyExecutorService
SidekiqUniqueJobs::TimerTask

show all

Includes:: Concurrent::Concern::Dereferenceable, Concurrent::Concern::Observable

Defined in:: lib/sidekiq_unique_jobs/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.

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

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!'

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

Constant Summary collapse

EXECUTION_INTERVAL = Default :execution_interval in seconds.

TIMEOUT_INTERVAL = Default :timeout_interval in seconds.

Instance Attribute Summary collapse

#execution_interval ⇒ Fixnum
Number of seconds after the task completes before the task is performed again.

Class Method Summary collapse

.execute(opts = {}) {|task| ... } ⇒ TimerTask
Create and execute a new TimerTask.

Instance Method Summary collapse

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

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

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/sidekiq_unique_jobs/timer_task.rb', line 181

def initialize(opts = {}, &task)
  raise ArgumentError, "no block given" unless task

  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.



232
233
234

# File 'lib/sidekiq_unique_jobs/timer_task.rb', line 232

def execution_interval
  synchronize { @execution_interval }
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 (Integer) —
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)

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.



225
226
227

# File 'lib/sidekiq_unique_jobs/timer_task.rb', line 225

def self.execute(opts = {}, &)
  TimerTask.new(opts, &).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/sidekiq_unique_jobs/timer_task.rb', line 208

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



191
192
193

# File 'lib/sidekiq_unique_jobs/timer_task.rb', line 191

def running?
  @running.true?
end

Class: SidekiqUniqueJobs::TimerTask

Overview

Examples:

Basic usage

Configuring :execution_interval

Immediate execution with :run_now

Last #value and Dereferenceable mixin

Controlling execution from within the block

Observation

Constant Summary collapse

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

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

Instance Attribute Details

#execution_interval ⇒ Fixnum

Class Method Details

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

Examples:

Instance Method Details

#execute ⇒ TimerTask

Examples:

Instance and execute in separate steps

Instance and execute in one line

#running? ⇒ Boolean

Configuring `:execution_interval`

Immediate execution with `:run_now`

Last `#value` and `Dereferenceable` mixin

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

#execution_interval ⇒ `Fixnum`

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

#execute ⇒ `TimerTask`

#running? ⇒ `Boolean`