ScheduledTask is now a TimerSet::Task.

Author: jdantonio

lib/concurrent/executor/timer_set.rb

@@ -20,13 +20,14 @@ class TimerSet < RubyExecutorService
     class Task < Concurrent::IVar
       include Comparable
 
-      def initialize(parent, delay, args, task)
-        super()
+      def initialize(parent, delay, args, task, opts = {})
+        super(IVar::NO_VALUE, opts, &nil)
         synchronize do
           ns_set_delay_and_time!(delay)
           @parent = parent
           @args = args
           @task = task
+          self.observers = CopyOnNotifyObserverSet.new
         end
       end
 
@@ -56,8 +57,13 @@ def processing?
         synchronize { ns_check_state?(:processing) }
       end
 
+      # Cancel this task and prevent it from executing. A task can only be
+      # cancelled if it is pending or unscheduled.
+      #
+      # @return [Boolean] true if task execution is successfully cancelled
+      #   else false
       def cancel
-        if compare_and_set_state(:cancelled, :pending)
+        if compare_and_set_state(:cancelled, :pending, :unscheduled)
           complete(false, nil, CancelledOperationError.new)
           # To avoid deadlocks this call must occur outside of #synchronize
           # Changing the state above should prevent redundant calls
@@ -76,11 +82,11 @@ def reschedule(delay)
       end
 
       # @!visibility private
-      def execute
+      def process_task
         safe_execute(@task, @args)
       end
 
-      protected :set, :try_set
+      protected :set, :try_set, :fail, :complete
 
       protected
 
@@ -89,11 +95,12 @@ def ns_set_delay_and_time!(delay)
         @time = Concurrent.monotonic_time + @delay
       end
 
-      def ns_reschedule(delay)
+      def ns_reschedule(delay, fail_if_cannot_remove = true)
         return false unless ns_check_state?(:pending)
         ns_set_delay_and_time!(delay)
-        return false unless @parent.send(:remove_task, self)
-        @parent.send(:ns_post_task, self)
+        removed = @parent.send(:remove_task, self)
+        return false if fail_if_cannot_remove && !removed
+        @parent.send(:post_task, self)
       end
     end
 
@@ -177,15 +184,19 @@ def ns_initialize(opts)
       self.auto_terminate = opts.fetch(:auto_terminate, true)
     end
 
+    def post_task(task)
+      synchronize{ ns_post_task(task) }
+    end
+
     # @!visibility private
     def ns_post_task(task)
       return false unless ns_running?
       if (task.original_delay) <= 0.01
-        @task_executor.post{ task.execute }
+        @task_executor.post{ task.process_task }
       else
         @queue.push(task)
         # only post the process method when the queue is empty
-        @timer_executor.post(&method(:process_tasks)) if @queue.size == 1
+        @timer_executor.post(&method(:process_tasks)) if @queue.length == 1
         @condition.set
       end
       true
@@ -202,15 +213,6 @@ def remove_task(task)
       synchronize{ @queue.delete(task) }
     end
 
-    # @note This is intended as a callback method from Task only.
-    #   It is not intended to be used directly. Cancel a task by
-    #   using the `Task#cancel` method.
-    #
-    # @!visibility private
-    def reschedule_task(task, delay)
-
-    end
-
     # @!visibility private
     def shutdown_execution
       @queue.clear
@@ -246,7 +248,7 @@ def process_tasks
           # queue now must have the same pop time, or a closer one, as
           # when we peeked).
           task = synchronize { @queue.pop }
-          @task_executor.post{ task.execute }
+          @task_executor.post{ task.process_task }
         else
           @condition.wait([diff, 60].min)
         end

lib/concurrent/obligation.rb

@@ -164,7 +164,7 @@ def set_state(success, value, reason)
 
     # @!visibility private
     def state=(value)
-      mutex.synchronize { @state = value }
+      mutex.synchronize { ns_set_state(value) }
     end
 
     # Atomic compare and set operation
@@ -176,9 +176,9 @@ def state=(value)
     # @return [Boolean] true is state is changed, false otherwise
     #
     # @!visibility private
-    def compare_and_set_state(next_state, expected_current)
+    def compare_and_set_state(next_state, *expected_current)
       mutex.synchronize do
-        if @state == expected_current
+        if expected_current.include? @state
           @state = next_state
           true
         else
@@ -215,5 +215,10 @@ def if_state(*expected_states)
     def ns_check_state?(expected)
       @state == expected
     end
+
+    # @!visibility private
+    def ns_set_state(value)
+      @state = value
+    end
   end
 end

lib/concurrent/scheduled_task.rb

@@ -1,7 +1,5 @@
-require 'concurrent/ivar'
-require 'concurrent/utility/timer'
-require 'concurrent/executor/executor'
-require 'concurrent/executor/safe_task_executor'
+require 'concurrent/configuration'
+require 'concurrent/executor/timer_set'
 
 module Concurrent
 
@@ -133,22 +131,15 @@ module Concurrent
   #   #>> The task completed at 2013-11-07 12:26:09 -0500 with value 'What does the fox say?'
   #
   # @!macro monotonic_clock_warning
-  class ScheduledTask < IVar
-
-    attr_reader :delay
+  class ScheduledTask < TimerSet::Task
 
     # Schedule a task for execution at a specified future time.
     #
     # @yield the task to be performed
     #
     # @param [Float] delay the number of seconds to wait for before executing the task
     #
-    # @param [Hash] opts the options controlling how the future will be processed
-    # @option opts [Boolean] :operation (false) when `true` will execute the future on the global
-    #   operation pool (for long-running operations), when `false` will execute the future on the
-    #   global task pool (for short-running tasks)
-    # @option opts [object] :executor when provided will run all operations on
-    #   this executor rather than the global thread pool (overrides :operation)
+    # @!macro executor_and_deref_options
     #
     # @!macro [attach] deprecated_scheduling_by_clock_time
     #
@@ -158,16 +149,12 @@ class ScheduledTask < IVar
     #     but will not be supported in the 1.0 release.
     def initialize(delay, opts = {}, &block)
       raise ArgumentError.new('no block given') unless block_given?
-      @delay = TimerSet.calculate_delay!(delay)
-      super(IVar::NO_VALUE, opts.merge(__task_from_block__: block), &nil)
-    end
-
-    def ns_initialize(value, opts)
-      super
-      self.observers = CopyOnNotifyObserverSet.new
-      @state         = :unscheduled
-      @task          = opts[:__task_from_block__]
-      @executor      = Executor.executor_from_options(opts) || Concurrent.global_io_executor
+      super(Concurrent.global_timer_set, delay, [], block, &nil)
+      synchronize do
+        ns_set_state(:unscheduled)
+        @__original_delay__ = delay
+        @time = nil
+      end
     end
 
     # Execute an `:unscheduled` `ScheduledTask`. Immediately sets the state to `:pending`
@@ -177,51 +164,25 @@ def ns_initialize(value, opts)
     # @return [ScheduledTask] a reference to `self`
     def execute
       if compare_and_set_state(:pending, :unscheduled)
-        @schedule_time = Time.now + @delay
-        Concurrent::timer(@delay) { @executor.post(&method(:process_task)) }
-        self
+        synchronize{ ns_reschedule(@__original_delay__, false) }
       end
+      self
     end
 
     # Create a new `ScheduledTask` object with the given block, execute it, and return the
     # `:pending` object.
     #
     # @param [Float] delay the number of seconds to wait for before executing the task
     #
-    # @param [Hash] opts the options controlling how the future will be processed
-    # @option opts [Boolean] :operation (false) when `true` will execute the future on the global
-    #   operation pool (for long-running operations), when `false` will execute the future on the
-    #   global task pool (for short-running tasks)
-    # @option opts [object] :executor when provided will run all operations on
-    #   this executor rather than the global thread pool (overrides :operation)
+    # @!macro executor_and_deref_options
     #
     # @return [ScheduledTask] the newly created `ScheduledTask` in the `:pending` state
     #
     # @raise [ArgumentError] if no block is given
     #
     # @!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
-
-    # Has the task been cancelled?
-    #
-    # @return [Boolean] true if the task is in the given state else false
-    def cancelled?
-      state == :cancelled
-    end
-
-    # In the task execution in progress?
-    #
-    # @return [Boolean] true if the task is in the given state else false
-    def processing?
-      state == :processing
+      new(delay, &block).execute
     end
 
     # In the task execution in progress?
@@ -239,26 +200,17 @@ def in_progress?
     #
     # @return [Boolean] true if task execution is successfully cancelled
     #   else false
-    def cancel
-      if_state(:unscheduled, :pending) do
-        @state = :cancelled
-        event.set
-        true
-      end
+    #
+    # @deprecated
+    def stop
+      warn '[DEPRECATED] use #processing? instead'
+      cancel
     end
-    alias_method :stop, :cancel
-
-    protected :set, :fail, :complete
 
-    private
-
-    # @!visibility private
-    def process_task
-      safe_execute(@task) do |success, val, reason|
-        event.set
-        time = Time.now
-        observers.notify_and_delete_observers { [time, self.value, reason] }
-      end
+    # @deprecated
+    def delay
+      warn '[DEPRECATED] use #original_delay instead'
+      original_delay
     end
   end
 end

spec/concurrent/scheduled_task_spec.rb

@@ -5,6 +5,7 @@
 module Concurrent
 
   describe ScheduledTask do
+
     context 'behavior' do
 
       # obligation
@@ -57,7 +58,7 @@ def trigger_observable(observable)
         Timecop.freeze do
           now = Time.now
           task = ScheduledTask.new(expected){ nil }.execute
-          expect(task.delay).to be_within(0.1).of(expected)
+          expect(task.original_delay).to be_within(0.1).of(expected)
         end
       end
 
@@ -66,7 +67,7 @@ def trigger_observable(observable)
         expected = 60 * 10
         schedule = Time.now + expected
         task = ScheduledTask.new(schedule){ nil }.execute
-        expect(task.delay).to be_within(0.1).of(expected)
+        expect(task.original_delay).to be_within(0.1).of(expected)
       end
 
       it 'raises an exception when seconds is less than zero' do
@@ -185,11 +186,12 @@ def trigger_observable(observable)
         expect(task.cancel).to be_truthy
       end
 
-      it 'sets the state to :cancelled when cancelled' do
+      it 'sets the reason to CancelledOperationError when cancelled' do
         task = ScheduledTask.new(10){ 42 }.execute
         sleep(0.1)
         task.cancel
-        expect(task).to be_cancelled
+        expect(task).to be_rejected
+        expect(task.reason).to be_a CancelledOperationError
       end
     end