Future can be cancelled before processing begins.

Author: jdantonio

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/future.rb

@@ -84,6 +84,38 @@ 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
+      compare_and_set_state(:cancelled, :pending)
+    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)
@@ -97,9 +129,13 @@ def ns_initialize(value, opts)
     private
 
     # @!visibility private
-    def work # :nodoc:
-      success, val, reason = SafeTaskExecutor.new(@task, rescue_exception: true).execute(*@args)
-      complete(success, val, reason)
+    def work
+      if compare_and_set_state(:processing, :pending)
+        success, val, reason = SafeTaskExecutor.new(@task, rescue_exception: true).execute(*@args)
+        complete(success, val, reason)
+      else
+        complete(false, nil, CancelledOperationError.new)
+      end
     end
   end
 end

spec/concurrent/future_spec.rb

@@ -18,7 +18,9 @@ module Concurrent
     let!(:rejected_reason) { StandardError.new('mojo jojo') }
 
     let(:pending_subject) do
-      Future.new(executor: executor){ sleep(0.1); fulfilled_value }.execute
+      executor = Concurrent::SingleThreadExecutor.new
+      executor.post{ sleep(5) }
+      Future.execute(executor: executor){ fulfilled_value }
     end
 
     let(:fulfilled_subject) do
@@ -110,8 +112,11 @@ def get_ivar_from_args(opts)
       end
 
       it 'sets the state to :pending' do
-        latch = Concurrent::CountDownLatch.new(1)
-        future = Future.new(executor: executor){ latch.wait(10) }
+        latch = Concurrent::CountDownLatch.new
+        executor = Concurrent::SingleThreadExecutor.new
+        executor.post{ latch.wait(2) }
+
+        future = Future.new(executor: executor){ 42 }
         future.execute
         expect(future).to be_pending
         latch.count_down
@@ -150,10 +155,29 @@ def get_ivar_from_args(opts)
 
       let(:executor) { ImmediateExecutor.new }
 
+      it 'sets the state to :processing while the task is executing' do
+        start_latch = Concurrent::CountDownLatch.new
+        continue_latch = Concurrent::CountDownLatch.new
+        executor = Concurrent::SingleThreadExecutor.new
+
+        future = Future.execute(executor: executor) do
+          start_latch.count_down
+          continue_latch.wait(2)
+          42
+        end
+
+        start_latch.wait(2)
+        state = future.state
+        continue_latch.count_down
+        future.value
+
+        expect(state).to eq :processing
+      end
+
       it 'passes all arguments to handler' do
-        @expected = false
-        Future.new(executor: executor){ @expected = true }.execute
-        expect(@expected).to be_truthy
+        expected = false
+        Future.new(executor: executor){ expected = true }.execute
+        expect(expected).to be_truthy
       end
 
       it 'sets the value to the result of the handler' do
@@ -198,6 +222,82 @@ def get_ivar_from_args(opts)
       end
     end
 
+    context 'cancellation' do
+
+      context '#cancel' do
+
+        it 'fails to cancel the task once processing has begun' do
+          start_latch = Concurrent::CountDownLatch.new
+          continue_latch = Concurrent::CountDownLatch.new
+          f = Future.execute do
+            start_latch.count_down
+            continue_latch.wait(2)
+            42
+          end
+
+          start_latch.wait(2)
+          cancelled = f.cancel
+          continue_latch.count_down
+
+          expect(cancelled).to be false
+          expect(f.value).to eq 42
+          expect(f).to be_fulfilled
+        end
+
+        it 'fails to cancel the task once processing is complete' do
+          f = Future.execute{ 42 }
+          f.wait
+          cancelled = f.cancel
+
+          expect(cancelled).to be false
+          expect(f.value).to eq 42
+          expect(f).to be_fulfilled
+        end
+
+        it 'cancels a pending task' do
+          executor = Concurrent::SingleThreadExecutor.new
+          latch = Concurrent::CountDownLatch.new
+          executor.post{ latch.wait(2) }
+
+          f = Future.execute(executor: executor){ 42 }
+          cancelled = f.cancel
+          latch.count_down
+
+          expect(cancelled).to be true
+          expect(f.value).to be_nil
+          expect(f).to be_rejected
+          expect(f.reason).to be_a Concurrent::CancelledOperationError
+        end
+      end
+
+      context '#wait_or_cancel' do
+
+        it 'returns true if the operation completes before timeout' do
+          f = Future.execute{ 42 }
+          success = f.wait_or_cancel(1)
+
+          expect(success).to be true
+          expect(f.value).to eq 42
+          expect(f).to be_fulfilled
+        end
+
+        it 'cancels the task on timeout' do
+          latch = Concurrent::CountDownLatch.new
+          executor = Concurrent::SingleThreadExecutor.new
+          executor.post{ latch.wait(2) }
+
+          f = Future.execute(executor: executor){ 42 }
+          success = f.wait_or_cancel(0.1)
+          latch.count_down
+
+          expect(success).to be false
+          expect(f.value).to be_nil
+          expect(f).to be_rejected
+          expect(f.reason).to be_a Concurrent::CancelledOperationError
+        end
+      end
+    end
+
     context 'observation' do
 
       let(:executor) { ImmediateExecutor.new }