Updated ScheduledTask to use new monotinic clock.

Author: jdantonio

lib/concurrent/executor/timer_set.rb

@@ -8,7 +8,7 @@
 
 module Concurrent
 
-  # Executes a collection of tasks at the specified times. A master thread
+  # Executes a collection of tasks, each after a given delay. A master thread
   # monitors the set and schedules each task for execution at the appropriate
   # time. Tasks are run on the global task pool or on the supplied executor.
   #
@@ -32,12 +32,11 @@ def initialize(opts = {})
       init_executor
     end
 
-    # Post a task to be execute at the specified time. The given time may be either
-    # a `Time` object or the number of seconds to wait. If the intended execution
-    # time is within 1/100th of a second of the current time the task will be
-    # immediately post to the executor.
+    # Post a task to be execute run after a given delay (in seconds). If the
+    # delay is less than 1/100th of a second the task will be immediately post
+    # to the executor.
     #
-    # @param [Object] intended_time the time to schedule the task for execution
+    # @param [Float] delay the number of seconds to wait for before executing the task
     #
     # @yield the task to be performed
     #
@@ -46,24 +45,10 @@ def initialize(opts = {})
     # @raise [ArgumentError] if the intended execution time is not in the future
     # @raise [ArgumentError] if no block is given
     #
-    # @!macro [attach] convert_time_to_interval_warning
-    #
-    #   @note Clock times are susceptible to changes in the system clock that
-    #     occur while the application is running. Timers based on intervals are
-    #     much more accurate because they can be set based on a monotonic clock.
-    #     Subsequently, execution intervals based on clock times will be
-    #     immediately converted to intervals based on a monotonic clock. Under
-    #     most scenarios this will make no difference. Should the system clock
-    #     change *after* the interval has been calculated, the interval will *not*
-    #     change. This is the intended behavior. This timer is not intended for
-    #     use in realtime operations or as a replacement for `cron` or similar
-    #     services. This level of accuracy is sufficient for the use cases this
-    #     timer was intended to solve.
-    #
-    #   @!macro monotonic_clock_warning
-    def post(intended_time, *args, &task)
+    # @!macro deprecated_scheduling_by_clock_time
+    def post(delay, *args, &task)
       raise ArgumentError.new('no block given') unless block_given?
-      interval = calculate_interval(intended_time)
+      interval = TimerSet.calculate_interval(delay)
 
       mutex.synchronize do
         return false unless running?
@@ -80,40 +65,45 @@ def post(intended_time, *args, &task)
       true
     end
 
+    # @!visibility private
+    def <<(task)
+      post(0.0, &task)
+      self
+    end
+
     # For a timer, #kill is like an orderly shutdown, except we need to manually
     # (and destructively) clear the queue first
     def kill
       mutex.synchronize { @queue.clear }
       shutdown
     end
 
-    private
-
-    # Calculate a time interval with milliseconds at which to execute a
-    # task. If the given time is a `Time` object it will be converted
-    # accordingly. If the time is a floating point value greater than
-    # zero it will be understood as a number of seconds in the future.
+    # Schedule a task to be execute run after a given delay (in seconds).
     #
-    # @param [Time, Float] intended_time the time to schedule the task for
-    #   execution, expressed as a `Time` object or a floating point number
-    #   representing a number of seconds
+    # @param [Float] delay the number of seconds to wait for before executing the task
     #
-    # @return [Float] the intended time interval as seconds/millis
+    # @return [Float] the number of seconds to delay
     #
     # @raise [ArgumentError] if the intended execution time is not in the future
+    # @raise [ArgumentError] if no block is given
+    #
+    # @!macro deprecated_scheduling_by_clock_time
     #
-    # @!macro convert_time_to_interval_warning
-    def calculate_interval(intended_time)
-      if intended_time.is_a?(Time)
+    # @!visibility private
+    def self.calculate_interval(delay)
+      if delay.is_a?(Time)
+        warn '[DEPRECATED] Use an interval not a clock time, schedule is now based on a monotonic clock'
         now = Time.now
-        raise ArgumentError.new('schedule time must be in the future') if intended_time <= now
-        intended_time.to_f - now.to_f
+        raise ArgumentError.new('schedule time must be in the future') if delay <= now
+        delay.to_f - now.to_f
       else
-        raise ArgumentError.new('seconds must be greater than zero') if intended_time.to_f < 0.0
-        intended_time.to_f
+        raise ArgumentError.new('seconds must be greater than zero') if delay.to_f < 0.0
+        delay.to_f
       end
     end
 
+    private
+
     # A struct for encapsulating a task and its intended execution time.
     # It facilitates proper prioritization by overriding the comparison
     # (spaceship) operator as a comparison of the intended execution

lib/concurrent/scheduled_task.rb

@@ -4,37 +4,172 @@
 
 module Concurrent
 
-  # {include:file:doc/scheduled_task.md}
+  # `ScheduledTask` is a close relative of `Concurrent::Future` but with one
+  # important difference. A `Future` is set to execute as soon as possible
+  # whereas a `ScheduledTask` is set to execute at a specific time. This
+  # implementation is loosely based on Java's
+  # [ScheduledExecutorService](http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ScheduledExecutorService.html). 
+  # 
+  # The *intended* schedule time of task execution is set on object construction
+  # with first argument, `delay`. The delay is a numeric (floating point or integer)
+  # representing a number of seconds in the future. Any other value or a numeric
+  # equal to or less than zero will result in an exception. The *actual* schedule
+  # time of task execution is set when the `execute` method is called.
+  #  
+  # The constructor can also be given zero or more processing options. Currently
+  # the only supported options are those recognized by the
+  # [Dereferenceable](Dereferenceable) module. 
+  # 
+  # The final constructor argument is a block representing the task to be performed.
+  # If no block is given an `ArgumentError` will be raised.
+  # 
+  # **States**
+  # 
+  # `ScheduledTask` mixes in the  [Obligation](Obligation) module thus giving it
+  # "future" behavior. This includes the expected lifecycle states. `ScheduledTask`
+  # has one additional state, however. While the task (block) is being executed the
+  # state of the object will be `:in_progress`. This additional state is necessary
+  # because it has implications for task cancellation. 
+  # 
+  # **Cancellation**
+  # 
+  # A `:pending` task can be cancelled using the `#cancel` method. A task in any
+  # other state, including `:in_progress`, cannot be cancelled. The `#cancel`
+  # method returns a boolean indicating the success of the cancellation attempt.
+  # A cancelled `ScheduledTask` cannot be restarted. It is immutable. 
+  # 
+  # **Obligation and Observation**
+  # 
+  # The result of a `ScheduledTask` can be obtained either synchronously or
+  # asynchronously. `ScheduledTask` mixes in both the [Obligation](Obligation)
+  # module and the
+  # [Observable](http://ruby-doc.org/stdlib-2.0/libdoc/observer/rdoc/Observable.html)
+  # module from the Ruby standard library. With one exception `ScheduledTask`
+  # behaves identically to [Future](Observable) with regard to these modules. 
+  #
+  # @example Basic usage
+  #
+  #   require 'concurrent'
+  #   require 'thread'   # for Queue
+  #   require 'open-uri' # for open(uri)
+  #   
+  #   class Ticker
+  #     def get_year_end_closing(symbol, year)
+  #       uri = "http://ichart.finance.yahoo.com/table.csv?s=#{symbol}&a=11&b=01&c=#{year}&d=11&e=31&f=#{year}&g=m"
+  #       data = open(uri) {|f| f.collect{|line| line.strip } }
+  #       data[1].split(',')[4].to_f
+  #     end
+  #   end
+  #   
+  #   # Future
+  #   price = Concurrent::Future.execute{ Ticker.new.get_year_end_closing('TWTR', 2013) }
+  #   price.state #=> :pending
+  #   sleep(1)    # do other stuff
+  #   price.value #=> 63.65
+  #   price.state #=> :fulfilled
+  #   
+  #   # ScheduledTask
+  #   task = Concurrent::ScheduledTask.execute(2){ Ticker.new.get_year_end_closing('INTC', 2013) }
+  #   task.state #=> :pending
+  #   sleep(3)   # do other stuff
+  #   task.value #=> 25.96
+  # 
+  # @example Successful task execution
+  #   
+  #   task = Concurrent::ScheduledTask.new(2){ 'What does the fox say?' }
+  #   task.state         #=> :unscheduled
+  #   task.execute
+  #   task.state         #=> pending
+  #   
+  #   # wait for it...
+  #   sleep(3)
+  #   
+  #   task.unscheduled? #=> false
+  #   task.pending?     #=> false
+  #   task.fulfilled?   #=> true
+  #   task.rejected?    #=> false
+  #   task.value        #=> 'What does the fox say?'
+  # 
+  # @example One line creation and execution
+  # 
+  #   task = Concurrent::ScheduledTask.new(2){ 'What does the fox say?' }.execute
+  #   task.state         #=> pending
+  # 
+  # @example Failed task execution
+  # 
+  #   task = Concurrent::ScheduledTask.execute(2){ raise StandardError.new('Call me maybe?') }
+  #   task.pending?      #=> true
+  #   
+  #   # wait for it...
+  #   sleep(3)
+  #   
+  #   task.unscheduled? #=> false
+  #   task.pending?     #=> false
+  #   task.fulfilled?   #=> false
+  #   task.rejected?    #=> true
+  #   task.value        #=> nil
+  #   task.reason       #=> #<StandardError: Call me maybe?> 
+  # 
+  # @example Task execution with observation
+  # 
+  #   observer = Class.new{
+  #     def update(time, value, reason)
+  #       puts "The task completed at #{time} with value '#{value}'"
+  #     end
+  #   }.new
+  #   
+  #   task = Concurrent::ScheduledTask.new(2){ 'What does the fox say?' }
+  #   task.add_observer(observer)
+  #   task.execute
+  #   task.pending?      #=> true
+  #   
+  #   # wait for it...
+  #   sleep(3)
+  #   
+  #   #>> The task completed at 2013-11-07 12:26:09 -0500 with value 'What does the fox say?'
+  #
+  # @!macro monotonic_clock_warning
+  #
+  # @!macro [attach] deprecated_scheduling_by_clock_time
+  #
+  #   @note Scheduling is now based on a monotonic clock. This makes the timer much
+  #     more accurate, but only when scheduling by passing a delay in seconds.
+  #     Scheduling a task based on a clock time is deprecated. It will still work
+  #     but will not be supported in the 1.0 release.
   class ScheduledTask < IVar
 
-    attr_reader :schedule_time
+    attr_reader :delay
 
-    def initialize(intended_time, opts = {}, &block)
+    # @!macro deprecated_scheduling_by_clock_time
+    def initialize(delay, opts = {}, &block)
       raise ArgumentError.new('no block given') unless block_given?
-      TimerSet.calculate_schedule_time(intended_time) # raises exceptons
+      @delay = TimerSet.calculate_interval(delay)
 
       super(NO_VALUE, opts)
 
       self.observers = CopyOnNotifyObserverSet.new
-      @intended_time = intended_time
       @state         = :unscheduled
       @task          = block
       @executor      = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_operation_pool
     end
 
-    # @since 0.5.0
     def execute
       if compare_and_set_state(:pending, :unscheduled)
-        now = Time.now
-        @schedule_time = TimerSet.calculate_schedule_time(@intended_time, now)
-        Concurrent::timer(@schedule_time.to_f - now.to_f) { @executor.post(&method(:process_task)) }
+        @schedule_time = Time.now + @delay
+        Concurrent::timer(@delay) { @executor.post(&method(:process_task)) }
         self
       end
     end
 
-    # @since 0.5.0
-    def self.execute(intended_time, opts = {}, &block)
-      return ScheduledTask.new(intended_time, opts, &block).execute
+    # @!macro deprecated_scheduling_by_clock_time
+    def self.execute(delay, opts = {}, &block)
+      return ScheduledTask.new(delay, opts, &block).execute
+    end
+
+    # @deprecated
+    def schedule_time
+      warn '[DEPRECATED] time is now based on a monotonic clock'
+      @schedule_time
     end
 
     def cancelled?
@@ -54,12 +189,6 @@ def cancel
     end
     alias_method :stop, :cancel
 
-    def add_observer(*args, &block)
-      if_state(:unscheduled, :pending, :in_progress) do
-        observers.add_observer(*args, &block)
-      end
-    end
-
     protected :set, :fail, :complete
 
     private

spec/concurrent/executor/timer_set_spec.rb

@@ -22,6 +22,7 @@ module Concurrent
     end
 
     it 'executes a given task when given a Time' do
+      warn 'deprecated syntax'
       latch = CountDownLatch.new(1)
       subject.post(Time.now + 0.1){ latch.count_down }
       expect(latch.wait(0.2)).to be_truthy
@@ -70,6 +71,7 @@ module Concurrent
     end
 
     it 'raises an exception when given a task with a past Time value' do
+      warn 'deprecated syntax'
       expect {
         subject.post(Time.now - 10){ nil }
       }.to raise_error(ArgumentError)