Merge pull request #299 from ruby-concurrency/cancel-future

Future, TimerSet, ScheduleTask cancel, reset, and reschedule

Author: jdantonio

.rspec

@@ -1,4 +1,3 @@
 --require spec_helper
 --color
---backtrace
 --format documentation

.yardopts

@@ -1,4 +1,3 @@
---protected
 --no-private
 --embed-mixins
 --output-dir ./yardoc

lib/concurrent/collection/priority_queue.rb

@@ -21,7 +21,7 @@ module Concurrent
   #   When running under JRuby the class `PriorityQueue` extends `JavaPriorityQueue`.
   #   When running under all other interpreters it extends `MutexPriorityQueue`.
   #
-  #   @note This implementation is *not* thread safe and performs no blocking.
+  #   @note This implementation is *not* thread safe.
   #
   #   @see http://en.wikipedia.org/wiki/Priority_queue
   #   @see http://ruby-doc.org/stdlib-2.0.0/libdoc/thread/rdoc/Queue.html
@@ -292,14 +292,57 @@ def self.from_list(list, opts = {})
         queue
       end
     end
+  end
 
-    # @!macro priority_queue
-    class PriorityQueue < JavaPriorityQueue
-    end
-  else
+  PriorityQueueImplementation = case
+                                when Concurrent.on_jruby?
+                                  JavaPriorityQueue
+                                else
+                                  MutexPriorityQueue
+                                end
+  private_constant :PriorityQueueImplementation
 
-    # @!macro priority_queue
-    class PriorityQueue < MutexPriorityQueue
-    end
+  # @!macro priority_queue
+  class PriorityQueue < PriorityQueueImplementation
+
+    alias_method :has_priority?, :include?
+
+    alias_method :size, :length
+
+    alias_method :deq, :pop
+    alias_method :shift, :pop
+
+    alias_method :<<, :push
+    alias_method :enq, :push
+
+    # @!method initialize(opts = {})
+    #   @!macro priority_queue_method_initialize
+
+    # @!method clear
+    #   @!macro priority_queue_method_clear
+
+    # @!method delete(item)
+    #   @!macro priority_queue_method_delete
+
+    # @!method empty?
+    #   @!macro priority_queue_method_empty
+
+    # @!method include?(item)
+    #   @!macro priority_queue_method_include
+
+    # @!method length
+    #   @!macro priority_queue_method_length
+
+    # @!method peek
+    #   @!macro priority_queue_method_peek
+
+    # @!method pop
+    #   @!macro priority_queue_method_pop
+
+    # @!method push(item)
+    #   @!macro priority_queue_method_push
+
+    # @!method self.from_list(list, opts = {})
+    #   @!macro priority_queue_method_from_list
   end
 end

lib/concurrent/errors.rb

@@ -3,6 +3,9 @@ module Concurrent
   # Raised when errors occur during configuration.
   ConfigurationError = Class.new(StandardError)
 
+  # Raised when an asynchronous operation is cancelled before execution.
+  CancelledOperationError = Class.new(StandardError)
+
   # Raised when a lifecycle method (such as `stop`) is called in an improper
   # sequence or when the object is in an inappropriate state.
   LifecycleError = Class.new(StandardError)

lib/concurrent/executor/timer_set.rb

@@ -1,16 +1,17 @@
-require 'thread'
+require 'concurrent/scheduled_task'
 require 'concurrent/atomic/event'
 require 'concurrent/collection/priority_queue'
-require 'concurrent/executor/executor'
 require 'concurrent/executor/executor_service'
 require 'concurrent/executor/single_thread_executor'
-require 'concurrent/utility/monotonic_time'
 
 module Concurrent
 
   # Executes a collection of tasks, each after a given delay. A master task
   # 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.
+  # time. Tasks are run on the global thread pool or on the supplied executor.
+  # Each task is represented as a `ScheduledTask`.
+  #
+  # @see Concurrent::ScheduledTask
   #
   # @!macro monotonic_clock_warning
   class TimerSet < RubyExecutorService
@@ -32,32 +33,28 @@ def initialize(opts = {})
     # delay is less than 1/100th of a second the task will be immediately post
     # to the executor.
     #
-    # @param [Float] delay the number of seconds to wait for before executing the task
+    # @param [Float] delay the number of seconds to wait for before executing the task.
+    # @param [Array<Object>] args the arguments passed to the task on execution.
     #
-    # @yield the task to be performed
+    # @yield the task to be performed.
     #
-    # @return [Boolean] true if the message is post, false after shutdown
+    # @return [Concurrent::ScheduledTask, false] IVar representing the task if the post
+    #   is successful; false after shutdown.
     #
-    # @raise [ArgumentError] if the intended execution time is not in the future
-    # @raise [ArgumentError] if no block is given
+    # @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
     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?
-
-        if (delay) <= 0.01
-          @task_executor.post(*args, &task)
-        else
-          @queue.push(Task.new(Concurrent.monotonic_time + delay, args, task))
-          @timer_executor.post(&method(:process_tasks))
-        end
-        @condition.set
-      end
-      true
+      return false unless running?
+      opts = {
+        executor: @task_executor,
+        args: args,
+        timer_set: self
+      }
+      task = ScheduledTask.execute(delay, opts, &task) # may raise exception
+      task.unscheduled? ? false : task
     end
 
     # Begin an immediate shutdown. In-progress tasks will be allowed to
@@ -68,48 +65,14 @@ def kill
       shutdown
     end
 
-    # Schedule a task to be executed after a given delay (in seconds).
-    #
-    # @param [Float] delay the number of seconds to wait for before executing the task
-    #
-    # @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
-    #
-    # @!visibility private
-    def self.calculate_delay!(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 delay <= now
-        delay.to_f - now.to_f
-      else
-        raise ArgumentError.new('seconds must be greater than zero') if delay.to_f < 0.0
-        delay.to_f
-      end
-    end
+    private :<<
 
     protected
 
-    # 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
-    # times.
+    # Initialize the object.
     #
+    # @param [Hash] opts the options to create the object with.
     # @!visibility private
-    Task = Struct.new(:time, :args, :op) do
-      include Comparable
-
-      def <=>(other)
-        self.time <=> other.time
-      end
-    end
-
-    private_constant :Task
-
     def ns_initialize(opts)
       @queue          = PriorityQueue.new(order: :min)
       @task_executor  = Executor.executor_from_options(opts) || Concurrent.global_io_executor
@@ -118,6 +81,44 @@ def ns_initialize(opts)
       self.auto_terminate = opts.fetch(:auto_terminate, true)
     end
 
+    # Post the task to the internal queue.
+    #
+    # @note This is intended as a callback method from ScheduledTask
+    #   only. It is not intended to be used directly. Post a task
+    #   by using the `SchedulesTask#execute` method.
+    #
+    # @!visibility private
+    def post_task(task)
+      synchronize{ ns_post_task(task) }
+    end
+
+    # @!visibility private
+    def ns_post_task(task)
+      return false unless ns_running?
+      if (task.initial_delay) <= 0.01
+        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.length == 1
+        @condition.set
+      end
+      true
+    end
+
+    # Remove the given task from the queue.
+    #
+    # @note This is intended as a callback method from `ScheduledTask`
+    #   only. It is not intended to be used directly. Cancel a task
+    #   by using the `ScheduledTask#cancel` method.
+    #
+    # @!visibility private
+    def remove_task(task)
+      synchronize{ @queue.delete(task) }
+    end
+
+    # `ExecutorServic` callback called during shutdown.
+    #
     # @!visibility private
     def shutdown_execution
       @queue.clear
@@ -137,7 +138,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
@@ -153,18 +154,11 @@ 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.args, &task.op)
+          task.executor.post{ task.process_task }
         else
           @condition.wait([diff, 60].min)
         end
       end
     end
-
-    private
-
-    def <<(task)
-      post(0.0, &task)
-      self
-    end
   end
 end

lib/concurrent/future.rb

@@ -45,7 +45,7 @@ def initialize(opts = {}, &block)
     #   future.state #=> :pending
     def execute
       if compare_and_set_state(:pending, :unscheduled)
-        @executor.post(@args){ work }
+        @executor.post{ safe_execute(@task, @args) }
         self
       end
     end
@@ -84,6 +84,43 @@ def set(value = IVar::NO_VALUE, &block)
       execute
     end
 
+    # Attempt to cancel the operation if it has not already processed.
+    # The operation can only be cancelled while still `pending`. It cannot
+    # be cancelled once it has begun processing or has completed.
+    #
+    # @return [Boolean] was the operation successfully cancelled.
+    def cancel
+      if compare_and_set_state(:cancelled, :pending)
+        complete(false, nil, CancelledOperationError.new)
+        true
+      else
+        false
+      end
+    end
+
+    # Has the operation been successfully cancelled?
+    #
+    # @return [Boolean]
+    def cancelled?
+      state == :cancelled
+    end
+
+    # Wait the given number of seconds for the operation to complete.
+    # On timeout attempt to cancel the operation.
+    #
+    # @param [Numeric] timeout the maximum time in seconds to wait.
+    # @return [Boolean] true if the operation completed before the timeout
+    #   else false
+    def wait_or_cancel(timeout)
+      wait(timeout)
+      if complete?
+        true
+      else
+        cancel
+        false
+      end
+    end
+
     protected
 
     def ns_initialize(value, opts)
@@ -93,13 +130,5 @@ def ns_initialize(value, opts)
       @executor = Executor.executor_from_options(opts) || Concurrent.global_io_executor
       @args = get_arguments_from(opts)
     end
-
-    private
-
-    # @!visibility private
-    def work # :nodoc:
-      success, val, reason = SafeTaskExecutor.new(@task, rescue_exception: true).execute(*@args)
-      complete(success, val, reason)
-    end
   end
 end

lib/concurrent/ivar.rb

@@ -148,6 +148,7 @@ def try_set(value = NO_VALUE, &block)
 
     protected
 
+    # @!visibility private
     def ns_initialize(value, opts)
       init_obligation(self)
       self.observers = CopyOnWriteObserverSet.new
@@ -160,21 +161,34 @@ def ns_initialize(value, opts)
       end
     end
 
+    # @!visibility private
+    def safe_execute(task, args = [])
+      if compare_and_set_state(:processing, :pending)
+        success, val, reason = SafeTaskExecutor.new(task, rescue_exception: true).execute(*@args)
+        complete(success, val, reason)
+        yield(success, val, reason) if block_given?
+      end
+    end
+
+    # @!visibility private
     def complete(success, value, reason)
       complete_without_notification(success, value, reason)
       notify_observers(self.value, reason)
       self
     end
 
+    # @!visibility private
     def complete_without_notification(success, value, reason)
       synchronize { ns_complete_without_notification(success, value, reason) }
       self
     end
 
+    # @!visibility private
     def notify_observers(value, reason)
       observers.notify_and_delete_observers{ [Time.now, value, reason] }
     end
 
+    # @!visibility private
     def ns_complete_without_notification(success, value, reason)
       raise MultipleAssignmentError if [:fulfilled, :rejected].include? @state
       set_state(success, value, reason)