TimerSet tasks can now be rescheduled.

Author: jdantonio

lib/concurrent/executor/timer_set.rb

@@ -20,23 +20,26 @@ class TimerSet < RubyExecutorService
     class Task < Concurrent::IVar
       include Comparable
 
-      def initialize(parent, time, args, task)
+      def initialize(parent, delay, args, task)
         super()
         synchronize do
+          ns_set_delay_and_time!(delay)
           @parent = parent
-          @time = time
           @args = args
           @task = task
         end
       end
 
-      # @!visibility private
-      def time
+      def original_delay
+        synchronize { @delay }
+      end
+
+      def schedule_time
         synchronize { @time }
       end
 
       def <=>(other)
-        self.time <=> other.time
+        self.schedule_time <=> other.schedule_time
       end
 
       def cancelled?
@@ -49,18 +52,37 @@ def cancel
           # To avoid deadlocks this call must occur outside of #synchronize
           # Changing the state above should prevent redundant calls
           @parent.send(:remove_task, self)
-          true
         else
           false
         end
       end
 
+      #def reset
+      #  reschedule(synchronize{ @delay })
+      #end
+
+      def reschedule(delay)
+        synchronize do
+          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)
+        end
+      end
+
       # @!visibility private
       def execute
         safe_execute(@task, @args)
       end
 
       protected :set, :try_set
+
+      protected
+
+      def ns_set_delay_and_time!(delay)
+        @delay = TimerSet.calculate_delay!(delay)
+        @time = Concurrent.monotonic_time + @delay
+      end
     end
 
     # Create a new set of timed tasks.
@@ -93,26 +115,9 @@ def initialize(opts = {})
     # @!macro deprecated_scheduling_by_clock_time
     def post(delay, *args, &task)
       raise ArgumentError.new('no block given') unless block_given?
-      delay = TimerSet.calculate_delay!(delay) # raises exceptions
-
-      synchronize do
-        return false unless running?
-
-        time = Concurrent.monotonic_time + delay
-        task = Task.new(self, time, args, task)
-
-        if (delay) <= 0.01
-          @task_executor.post{ task.execute }
-        else
-          @queue.push(task)
-          # `process_tasks` method will run until queue is empty
-          # only post the process method when the queue is empty
-          @timer_executor.post(&method(:process_tasks)) if @queue.size == 1
-        end
-
-        @condition.set
-        task
-      end
+      task = Task.new(self, delay, args, task) # may raise exception
+      ok = synchronize{ ns_post_task(task) }
+      ok ? task : false
     end
 
     # Begin an immediate shutdown. In-progress tasks will be allowed to
@@ -147,8 +152,33 @@ def self.calculate_delay!(delay)
       end
     end
 
+    private :<<
+
     protected
 
+    # @!visibility private
+    def ns_initialize(opts)
+      @queue          = PriorityQueue.new(order: :min)
+      @task_executor  = Executor.executor_from_options(opts) || Concurrent.global_io_executor
+      @timer_executor = SingleThreadExecutor.new
+      @condition      = Event.new
+      self.auto_terminate = opts.fetch(:auto_terminate, true)
+    end
+
+    # @!visibility private
+    def ns_post_task(task)
+      return false unless ns_running?
+      if (task.original_delay) <= 0.01
+        @task_executor.post{ task.execute }
+      else
+        @queue.push(task)
+        # only post the process method when the queue is empty
+        @timer_executor.post(&method(:process_tasks)) if @queue.size == 1
+        @condition.set
+      end
+      true
+    end
+
     # Remove the given task from the queue.
     #
     # @note This is intended as a callback method from Task only.
@@ -160,13 +190,13 @@ 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 ns_initialize(opts)
-      @queue          = PriorityQueue.new(order: :min)
-      @task_executor  = Executor.executor_from_options(opts) || Concurrent.global_io_executor
-      @timer_executor = SingleThreadExecutor.new
-      @condition      = Event.new
-      self.auto_terminate = opts.fetch(:auto_terminate, true)
+    def reschedule_task(task, delay)
+
     end
 
     # @!visibility private
@@ -188,7 +218,7 @@ def process_tasks
         break unless task
 
         now = Concurrent.monotonic_time
-        diff = task.time - now
+        diff = task.schedule_time - now
 
         if diff <= 0
           # We need to remove the task from the queue before passing
@@ -210,12 +240,5 @@ def process_tasks
         end
       end
     end
-
-    private
-
-    # @!visibility private
-    def <<(task)
-      raise NotImplementedError.new
-    end
   end
 end

lib/concurrent/obligation.rb

@@ -187,7 +187,7 @@ def compare_and_set_state(next_state, expected_current)
       end
     end
 
-    # executes the block within mutex if current state is included in expected_states
+    # Executes the block within mutex if current state is included in expected_states
     #
     # @return block value if executed, false otherwise
     #
@@ -203,5 +203,17 @@ def if_state(*expected_states)
         end
       end
     end
+
+    protected
+
+    # Am I in the current state?
+    #
+    # @param [Symbol] expected The state to check against
+    # @return [Boolean] true if in the expected state else false
+    #
+    # @!visibility private
+    def ns_check_state?(expected)
+      @state == expected
+    end
   end
 end

spec/concurrent/executor/timer_set_spec.rb

@@ -4,7 +4,8 @@ module Concurrent
 
   describe TimerSet do
 
-    subject{ TimerSet.new(executor: :immediate) }
+    let(:executor){ Concurrent::SingleThreadExecutor.new }
+    subject{ TimerSet.new(executor: executor) }
 
     after(:each){ subject.kill }
 
@@ -48,7 +49,8 @@ module Concurrent
       end
 
       it 'immediately posts a task when the delay is zero' do
-        expect(Thread).not_to receive(:new).with(any_args)
+        timer = subject.instance_variable_get(:@timer_executor)
+        expect(timer).not_to receive(:post).with(any_args)
         subject.post(0){ true }
       end
     end
@@ -192,7 +194,7 @@ module Concurrent
         start_latch.wait(2)
         success = job.cancel
         continue_latch.count_down
-        
+
         expect(success).to be false
         expect(job.value).to eq 42
         expect(job.reason).to be_nil
@@ -203,7 +205,7 @@ module Concurrent
 
         job.wait(2)
         success = job.cancel
-        
+
         expect(success).to be false
         expect(job.value).to eq 42
         expect(job.reason).to be_nil
@@ -219,6 +221,103 @@ module Concurrent
         expect(job.value(0)).to be_nil
         expect(job.reason).to be_a CancelledOperationError
       end
+
+      it 'returns false when not running' do
+        task = subject.post(10){ nil }
+        subject.shutdown
+        subject.wait_for_termination(2)
+        expect(task.cancel).to be false
+      end
+    end
+
+    context 'task rescheduling' do
+
+      let(:queue) { subject.instance_variable_get(:@queue) }
+
+      it 'raises an exception when given an invalid time' do
+        expect(queue).to receive(:push).once.with(any_args).and_call_original
+        task = subject.post(10){ nil }
+        expect{ task.reschedule(-1) }.to raise_error(ArgumentError)
+      end
+
+      it 'does not change the current schedule when given an invalid time' do
+        expect(queue).to receive(:push).once.with(any_args).and_call_original
+        task = subject.post(10){ nil }
+        expected = task.schedule_time
+        begin
+          task.reschedule(-1)
+        rescue
+        end
+        expect(task.schedule_time).to eq expected
+      end
+
+      it 'reschdules a pending and unpost task when given a valid time' do
+        original_delay = 10
+        rescheduled_delay = 20
+        expect(queue).to receive(:push).twice.with(any_args).and_call_original
+        task = subject.post(original_delay){ nil }
+        original_schedule = task.schedule_time
+        success = task.reschedule(rescheduled_delay)
+        expect(success).to be true
+        expect(task.original_delay).to be_within(0.01).of(rescheduled_delay)
+        expect(task.schedule_time).to be > original_schedule
+      end
+
+      it 'returns false once the task has been post to the executor' do
+        expect(queue).to receive(:push).once.with(any_args).and_call_original
+        start_latch = Concurrent::CountDownLatch.new
+        continue_latch = Concurrent::CountDownLatch.new
+
+        task = subject.post(0.1) do
+          start_latch.count_down
+          continue_latch.wait(2)
+        end
+        start_latch.wait(2)
+
+        expected = task.schedule_time
+        success = task.reschedule(10)
+        continue_latch.count_down
+        expect(success).to be false
+        expect(task.schedule_time).to eq expected
+      end
+
+      it 'returns false once the task is processing' do
+        expect(queue).to receive(:push).once.with(any_args).and_call_original
+        start_latch = Concurrent::CountDownLatch.new
+        continue_latch = Concurrent::CountDownLatch.new
+        task = subject.post(0.1) do
+          start_latch.count_down
+          continue_latch.wait(2)
+        end
+        start_latch.wait(2)
+
+        expected = task.schedule_time
+        success = task.reschedule(10)
+        continue_latch.count_down
+        expect(success).to be false
+        expect(task.schedule_time).to eq expected
+      end
+
+      it 'returns false once the task has is complete' do
+        expect(queue).to receive(:push).once.with(any_args).and_call_original
+        task = subject.post(0.1){ nil }
+        task.value(2)
+        expected = task.schedule_time
+        success = task.reschedule(10)
+        expect(success).to be false
+        expect(task.schedule_time).to eq expected
+      end
+
+      it 'returns false when not running' do
+        expect(queue).to receive(:push).once.with(any_args).and_call_original
+        task = subject.post(10){ nil }
+        subject.shutdown
+        subject.wait_for_termination(2)
+        expected = task.schedule_time
+        success = task.reschedule(10)
+        expect(success).to be false
+        expect(task.schedule_time).to be_within(0.01).of(expected)
+      end
     end
 
     context 'termination' do