Cooperative cancellation for promises and any other async processing

Author: pitr-ch

lib/concurrent/edge/promises.rb

@@ -70,4 +70,82 @@ def each_body(value, &block)
 
     end
   end
+
+  # inspired by https://msdn.microsoft.com/en-us/library/dd537607(v=vs.110).aspx
+  class Cancellation < Synchronization::Object
+    safe_initialization!
+
+    def self.create
+      [(i = new), i.token]
+    end
+
+    private_class_method :new
+
+    def initialize
+      @Cancel = Promises.completable_event
+      @Token  = Token.new @Cancel.with_hidden_completable
+    end
+
+    def token
+      @Token
+    end
+
+    def cancel
+      try_cancel or raise MultipleAssignmentError, 'cannot cancel twice'
+    end
+
+    def try_cancel
+      !!@Cancel.complete(false)
+    end
+
+    def canceled?
+      @Cancel.complete?
+    end
+
+    class Token < Synchronization::Object
+      safe_initialization!
+
+      def initialize(cancel)
+        @Cancel = cancel
+      end
+
+      def event
+        @Cancel
+      end
+
+      def on_cancellation(*args, &block)
+        @Cancel.on_completion *args, &block
+      end
+
+      def then(*args, &block)
+        @Cancel.chain *args, &block
+      end
+
+      def canceled?
+        @Cancel.complete?
+      end
+
+      def loop_until_canceled(&block)
+        until canceled?
+          result = block.call
+        end
+        result
+      end
+
+      def raise_if_canceled
+        raise CancelledOperationError if canceled?
+        self
+      end
+
+      def join(*tokens)
+        Token.new Promises.any_event(@Cancel, *tokens.map(&:event))
+      end
+
+    end
+
+    private_constant :Token
+
+    # TODO (pitr-ch 27-Mar-2016): cooperation with mutex, select etc?
+    # TODO (pitr-ch 27-Mar-2016): examples (scheduled to be cancelled in 10 sec)
+  end
 end

lib/concurrent/promises.rb

@@ -1425,7 +1425,5 @@ def initialize(default_executor, intended_time)
   end
 end
 
-# TODO cancelable Futures, will cancel the future but the task will finish anyway
-# TODO task interrupts, how to support?
 # TODO when value is requested the current thread may evaluate the tasks to get the value for performance reasons it may not evaluate :io though
 # TODO try work stealing pool, each thread has it's own queue

spec/concurrent/promises_spec.rb

@@ -434,4 +434,17 @@ def behaves_as_delay(delay, value)
     end
   end
 
+  describe 'Cancellation', edge: true do
+    specify do
+      source, token = Concurrent::Cancellation.create
+
+      futures = Array.new(2) { future(token) { |t| t.loop_until_canceled { Thread.pass }; :done } }
+
+      source.cancel
+      futures.each do |future|
+        expect(future.value!).to eq :done
+      end
+    end
+  end
+
 end