123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::TimerTask

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

Constant Summary

Concern::Logging - Included

SEV_LABEL

AbstractExecutorService - Inherited

FALLBACK_POLICIES

Class Method Summary

RubyExecutorService - Inherited

.new

AbstractExecutorService - Inherited

.new

Create a new thread pool.

Instance Attribute Summary

Concern::Dereferenceable - Included

#value

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

RubyExecutorService - Inherited

#fallback_policy, #ns_running?, #ns_shutdown?, #ns_shuttingdown?

AbstractExecutorService - Inherited

#auto_terminate=

Set the auto-terminate behavior for this executor.
#auto_terminate?

Is the executor auto-terminate when the application exits?
#fallback_policy, #name,
#running?

Is the executor running?
#shutdown

Begin an orderly shutdown.
#shutdown?

Is the executor shutdown?
#shuttingdown?

Is the executor shuttingdown?
#ns_auto_terminate?

ExecutorService - Included

#can_overflow?

Does the task queue have a maximum size?
#serialized?

Does this executor guarantee serialization of its operations?

Instance Method Summary

Concern::Observable - Included

#add_observer

Adds an observer to this set.
#count_observers

Return the number of observers associated with this object.
#delete_observer

Remove observer as an observer on this object so that it will no longer receive notifications.
#delete_observers

Remove all observers associated with this object.
#with_observer

As #add_observer but can be used for chaining.

Concern::Dereferenceable - Included

#deref

Alias for Concern::Dereferenceable#value.
#apply_deref_options,
#ns_set_deref_options

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

RubyExecutorService - Inherited

#<<

Submit a task to the executor for asynchronous processing.
#auto_terminate=

Set the auto-terminate behavior for this executor.
#auto_terminate?

Is the executor auto-terminate when the application exits?
#can_overflow?

Does the task queue have a maximum size?
#kill

Begin an immediate shutdown.
#post

Submit a task to the executor for asynchronous processing.
#running?

Is the executor running?
#serialized?

Does this executor guarantee serialization of its operations?
#shutdown

Begin an orderly shutdown.
#shutdown?

Is the executor shutdown?
#shuttingdown?

Is the executor shuttingdown?
#wait_for_termination

Block until executor shutdown is complete or until timeout seconds have passed.
#ns_shutdown_execution, #stop_event, #stopped_event

AbstractExecutorService - Inherited

#<<

Submit a task to the executor for asynchronous processing.
#can_overflow?

Does the task queue have a maximum size?
#kill

Begin an immediate shutdown.
#post

Submit a task to the executor for asynchronous processing.
#serialized?

Does this executor guarantee serialization of its operations?
#to_s,
#wait_for_termination

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

Returns an action which executes the fallback_policy once the queue size reaches max_queue.
#ns_execute,
#ns_kill_execution

Callback method called when the executor has been killed.
#ns_shutdown_execution

Callback method called when an orderly shutdown has completed.

Concern::Deprecation - Included

#deprecated, #deprecated_method

ExecutorService - Included

#<<

Submit a task to the executor for asynchronous processing.
#post

Submit a task to the executor for asynchronous processing.

Concern::Logging - Included

#log

Logs through global_logger, it can be overridden by setting @logger.

Synchronization::LockableObject - Inherited

#new_condition

Constructor Details

.new(opts = {}, &task) ⇒ TimerTask

Create a new TimerTask with the given task and configuration.

Raises:

  • (ArgumentError)
[ GitHub ] 

  


# 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


  







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

            
    • 
        
  • 
          :dup_on_deref
          (Boolean)
           — default:
              false          
            — 
    

    Call #dup before returning the data from #value
    

            
    • 
        
  • 
          :freeze_on_deref
          (Boolean)
           — default:
              false          
            — 
    

    Call #freeze before returning the data from #value
    

            
    • 
        
  • 
          :copy_on_deref
          (Proc)
           — default:
              nil          
            — 
    

    When calling the #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.
    

    
  
    • 





  [ GitHub ]


  

  


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



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


  







Instance Attribute Details



  

    #execution_interval  ⇒ Fixnum  (rw)
  



  

    
  





  
Returns:


    
  
  • 
    (Fixnum)
—    
    

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

    
  
    • 





  [ GitHub ]


  

  


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



def execution_interval
  synchronize { @execution_interval }
end


  








  

    #execution_interval=(value)   (rw)
  



  

    

See additional method definition at line 259.


  



  [ GitHub ]


  

  


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



def execution_interval
  synchronize { @execution_interval }
end


  








  

    #interval_type  ⇒ Symbol  (readonly)
  



  

    
  





  
Returns:


    
  
  • 
    (Symbol)
—    
    

    method to calculate the interval between executions
    

    
  
    • 





  [ GitHub ]


  

  


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



attr_reader :interval_type


  








  

    #running?  ⇒ Boolean  (readonly)
  



  

    

Is the executor running?


  





  
Returns:


    
  
  • 
    (Boolean)
—    
    

    true when running, false when shutting down or shutdown
    

    
  
    • 





  [ GitHub ]


  

  


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



def running?
  @running.true?
end


  








  

    #timeout_interval  ⇒ Fixnum  (rw)
  



  

    
  





  
Returns:


    
  
  • 
    (Fixnum)
—    
    

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

    
  
    • 





  [ GitHub ]


  

  


# 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


  








  

    #timeout_interval=(value)   (rw)
  



  

    

See additional method definition at line 281.


  



  [ GitHub ]


  

  


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



def timeout_interval
  warn 'TimerTask timeouts are now ignored as these were not able to be implemented correctly'
end


  







Instance Method Details



  

    #calculate_next_interval(start_time)   (private)
  

  [ GitHub ]


  

  


# 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


  








  

    #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
    

    
  
    • 





  [ GitHub ]


  

  


# 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_task(completion)   (private)
  

  [ GitHub ]


  

  


# 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


  








  

    #ns_initialize(opts, &task)   (private)
  

  [ GitHub ]


  

  


# 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


  








  

    #ns_kill_execution   (private)
  

  [ GitHub ]


  

  


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



def ns_kill_execution
  @running.make_false
  super
end


  








  

    #ns_shutdown_execution   (private)
  

  [ GitHub ]


  

  


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



def ns_shutdown_execution
  @running.make_false
  super
end


  








  

    #schedule_next_task(interval = execution_interval)   (private)
  

  [ GitHub ]


  

  


# 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