Improvements

- :io is now default executor
- Concurrent.future without block returns CompletableFuture
- Concurrent.event returns CompletableEvent

Author: pitr-ch

lib/concurrent/edge/future.rb

@@ -10,36 +10,39 @@ module Edge
     module FutureShortcuts
       # TODO to construct event to be set later to trigger rest of the tree
 
-      def event(default_executor = :fast)
-        CompletableEvent.new(default_executor)
-      end
-
-      # Constructs new Future which will be completed after block is evaluated on executor. Evaluation begins immediately.
-      # @return [Future]
-      # @note TODO allow to pass in variables as Thread.new(args) {|args| _ } does
-      def future(default_executor = :fast, &task)
-        ImmediatePromise.new(default_executor).event.chain(&task)
+      # User is responsible for completing the event once.
+      # @return [CompletableEvent]
+      def event(default_executor = :io)
+        CompletableEventPromise.new(default_executor).future
+      end
+
+      # @overload future(default_executor = :io, &task)
+      #   Constructs new Future which will be completed after block is evaluated on executor. Evaluation begins immediately.
+      #   @return [Future]
+      #   @note FIXME allow to pass in variables as Thread.new(args) {|args| _ } does
+      # @overload future(default_executor = :io)
+      #   User is responsible for completing the future once.
+      #   @return [CompletableFuture]
+      def future(default_executor = :io, &task)
+        if task
+          ImmediatePromise.new(default_executor).event.chain(&task)
+        else
+          CompletableFuturePromise.new(default_executor).future
+        end
       end
 
       alias_method :async, :future
 
       # Constructs new Future which will be completed after block is evaluated on executor. Evaluation is delays until
       # requested by {Future#wait} method, {Future#value} and {Future#value!} methods are calling {Future#wait} internally.
       # @return [Delay]
-      def delay(default_executor = :fast, &task)
+      def delay(default_executor = :io, &task)
         Delay.new(default_executor).event.chain(&task)
       end
 
-      # Constructs {Promise} which helds its {Future} in {AbstractPromise#future} method. Intended for completion by user.
-      # User is responsible not to complete the Promise twice.
-      # @return [AbstractPromise] in this case instance of {OuterPromise}
-      def promise(default_executor = :fast)
-        CompletablePromise.new(default_executor)
-      end
-
       # Schedules the block to be executed on executor in given intended_time.
       # @return [Future]
-      def schedule(intended_time, default_executor = :fast, &task)
+      def schedule(intended_time, default_executor = :io, &task)
         ScheduledPromise.new(intended_time, default_executor).future.chain(&task)
       end
 
@@ -82,7 +85,7 @@ class Event < Synchronization::Object
       extend FutureShortcuts
 
       # @api private
-      def initialize(promise, default_executor = :fast)
+      def initialize(promise, default_executor = :io)
         super()
         synchronize { ns_initialize(promise, default_executor) }
       end
@@ -196,7 +199,7 @@ def with_default_executor(executor = default_executor)
 
       private
 
-      def ns_initialize(promise, default_executor = :fast)
+      def ns_initialize(promise, default_executor = :io)
         @promise          = promise
         @state            = :pending
         @callbacks        = []
@@ -375,13 +378,14 @@ def wait!(timeout = nil)
       # @raise [Exception] when #failed? it raises #reason
       # @return [Object] see Dereferenceable#deref
       def value!(timeout = nil)
+        touch
         synchronize { ns_value! timeout }
       end
 
-      # @example allows Obligation to be risen
-      #   failed_ivar = Ivar.new.fail
-      #   raise failed_ivar
+      # @example allows failed Future to be risen
+      #   raise Concurrent.future.fail
       def exception(*args)
+        touch
         synchronize { ns_exception(*args) }
       end
 
@@ -447,7 +451,7 @@ def ns_add_callback(method, *args)
 
       private
 
-      def ns_initialize(promise, default_executor = :fast)
+      def ns_initialize(promise, default_executor = :io)
         super(promise, default_executor)
         @value  = nil
         @reason = nil
@@ -528,8 +532,6 @@ def ns_complete(success, value, reason, raise = true)
         callbacks
       end
 
-      private
-
       def ns_complete_state(success, value, reason)
         if success
           @value = value
@@ -573,6 +575,46 @@ def pr_async_callback_on_completion(success, value, reason, executor, callback)
       end
     end
 
+    class CompletableEvent < Event
+      # Complete the event
+      # @api public
+      def complete(raise = true)
+        super raise
+      end
+    end
+
+    class CompletableFuture < Future
+      # Complete the future
+      # @api public
+      def complete(success, value, reason, raise = true)
+        super success, value, reason, raise
+      end
+
+      def success(value)
+        promise.success(value)
+      end
+
+      def try_success(value)
+        promise.try_success(value)
+      end
+
+      def fail(reason = StandardError.new)
+        promise.fail(reason)
+      end
+
+      def try_fail(reason = StandardError.new)
+        promise.try_fail(reason)
+      end
+
+      def evaluate_to(*args, &block)
+        promise.evaluate_to(*args, &block)
+      end
+
+      def evaluate_to!(*args, &block)
+        promise.evaluate_to!(*args, &block)
+      end
+    end
+
     # TODO modularize blocked_by and notify blocked
 
     # @abstract
@@ -638,21 +680,21 @@ def pr_evaluate_to(future, *args, &block)
       end
     end
 
-    class CompletableEvent < AbstractPromise
+    class CompletableEventPromise < AbstractPromise
       public :complete
 
       private
 
-      def ns_initialize(default_executor = :fast)
-        super Event.new(self, default_executor)
+      def ns_initialize(default_executor = :io)
+        super CompletableEvent.new(self, default_executor)
       end
     end
 
     # @note Be careful not to fullfill the promise twice
     # @example initialization
     #   Concurrent.promise
     # @note TODO consider to allow being blocked_by
-    class CompletablePromise < CompletableEvent
+    class CompletableFuturePromise < AbstractPromise
       # Set the `IVar` to a value and wake or notify all threads waiting on it.
       #
       # @param [Object] value the value to store in the `IVar`
@@ -679,6 +721,7 @@ def try_fail(reason = StandardError.new)
         !!complete(false, nil, reason, false)
       end
 
+      public :complete
       public :evaluate_to
 
       # @return [Future]
@@ -688,8 +731,8 @@ def evaluate_to!(*args, &block)
 
       private
 
-      def ns_initialize(default_executor = :fast)
-        super Future.new(self, default_executor)
+      def ns_initialize(default_executor = :io)
+        super CompletableFuture.new(self, default_executor)
       end
     end
 
@@ -765,7 +808,7 @@ def executor
 
       private
 
-      def ns_initialize(blocked_by_future, default_executor = :fast, executor = default_executor, &task)
+      def ns_initialize(blocked_by_future, default_executor = :io, executor = default_executor, &task)
         raise ArgumentError, 'no block given' unless block_given?
         super Future.new(self, default_executor), [blocked_by_future]
         @task     = task
@@ -796,7 +839,7 @@ def pr_completable(_, _, _, _, _)
     class ThenPromise < BlockedTaskPromise
       private
 
-      def ns_initialize(blocked_by_future, default_executor = :fast, executor = default_executor, &task)
+      def ns_initialize(blocked_by_future, default_executor = :io, executor = default_executor, &task)
         blocked_by_future.is_a? Future or
             raise ArgumentError, 'only Future can be appended with then'
         super(blocked_by_future, default_executor, executor, &task)
@@ -814,7 +857,7 @@ def pr_completable(done_future, _, future, executor, task)
     class RescuePromise < BlockedTaskPromise
       private
 
-      def ns_initialize(blocked_by_future, default_executor = :fast, executor = default_executor, &task)
+      def ns_initialize(blocked_by_future, default_executor = :io, executor = default_executor, &task)
         blocked_by_future.is_a? Future or
             raise ArgumentError, 'only Future can be rescued'
         super(blocked_by_future, default_executor, executor, &task)
@@ -851,7 +894,7 @@ def self.new(*args)
 
       private
 
-      def ns_initialize(default_executor = :fast)
+      def ns_initialize(default_executor = :io)
         super Event.new(self, default_executor)
       end
     end
@@ -877,7 +920,7 @@ def ns_done(future)
         super future
       end
 
-      def ns_initialize(blocked_by_future, levels = 1, default_executor = :fast)
+      def ns_initialize(blocked_by_future, levels = 1, default_executor = :io)
         blocked_by_future.is_a? Future or
             raise ArgumentError, 'only Future can be flatten'
         super(Future.new(self, default_executor), [blocked_by_future])
@@ -893,8 +936,9 @@ def pr_completable(_, blocked_by, future)
     class AllPromise < BlockedPromise
       private
 
-      def ns_initialize(blocked_by_futures, default_executor = :fast)
+      def ns_initialize(blocked_by_futures, default_executor = :io)
         klass = blocked_by_futures.any? { |f| f.is_a?(Future) } ? Future : Event
+        # noinspection RubyArgCount
         super(klass.new(self, default_executor), blocked_by_futures)
       end
 
@@ -918,7 +962,7 @@ class AnyPromise < BlockedPromise
 
       private
 
-      def ns_initialize(blocked_by_futures, default_executor = :fast)
+      def ns_initialize(blocked_by_futures, default_executor = :io)
         blocked_by_futures.all? { |f| f.is_a? Future } or
             raise ArgumentError, 'accepts only Futures not Events'
         super(Future.new(self, default_executor), blocked_by_futures)
@@ -940,7 +984,7 @@ def touch
 
       private
 
-      def ns_initialize(default_executor = :fast)
+      def ns_initialize(default_executor = :io)
         super Event.new(self, default_executor)
       end
     end
@@ -957,7 +1001,7 @@ def inspect
 
       private
 
-      def ns_initialize(intended_time, default_executor = :fast)
+      def ns_initialize(intended_time, default_executor = :io)
         super Event.new(self, default_executor)
         in_seconds = begin
           @intended_time = intended_time

spec/concurrent/edge/future_spec.rb

@@ -65,15 +65,28 @@
   describe '.event' do
     specify do
       completable_event = Concurrent.event
-      one               = completable_event.event.chain { 1 }
-      join              = Concurrent.join(completable_event.event).chain { 1 }
+      one               = completable_event.chain { 1 }
+      join              = Concurrent.join(completable_event).chain { 1 }
       expect(one.completed?).to be false
       completable_event.complete
       expect(one.value).to eq 1
       expect(join.wait.completed?).to be true
     end
   end
 
+  describe '.future without block' do
+    specify do
+      completable_future = Concurrent.future
+      one                = completable_future.then(&:succ)
+      join               = Concurrent.join(completable_future).then { |v| v }
+      expect(one.completed?).to be false
+      completable_future.success 0
+      expect(one.value).to eq 1
+      expect(join.wait!.completed?).to be true
+      expect(join.value!).to eq 0
+    end
+  end
+
   describe '.any' do
     it 'continues on first result' do
       queue = Queue.new
@@ -107,11 +120,11 @@
 
     it 'chains' do
       future0 = Concurrent.future { 1 }.then { |v| v + 2 } # both executed on default FAST_EXECUTOR
-      future1 = future0.then(:io) { raise 'boo' } # executed on IO_EXECUTOR
+      future1 = future0.then(:fast) { raise 'boo' } # executed on IO_EXECUTOR
       future2 = future1.then { |v| v + 1 } # will fail with 'boo' error, executed on default FAST_EXECUTOR
       future3 = future1.rescue { |err| err.message } # executed on default FAST_EXECUTOR
       future4 = future0.chain { |success, value, reason| success } # executed on default FAST_EXECUTOR
-      future5 = future3.with_default_executor(:io) # connects new future with different executor, the new future is completed when future3 is
+      future5 = future3.with_default_executor(:fast) # connects new future with different executor, the new future is completed when future3 is
       future6 = future5.then(&:capitalize) # executes on IO_EXECUTOR because default was set to :io on future5
       future7 = Concurrent.join(future0, future3)
       future8 = future0.rescue { raise 'never happens' } # future0 succeeds so future8'll have same value as future 0
@@ -127,15 +140,15 @@
 
       expect(table.join("\n")).to eq <<-TABLE.gsub(/^\s+\|/, '').strip
         |index success      value reason pool d.pool
-        |    0    true          3        fast   fast
-        |    1   false               boo   io   fast
-        |    2   false               boo fast   fast
-        |    3    true        boo        fast   fast
-        |    4    true       true        fast   fast
-        |    5    true        boo                 io
-        |    6    true        Boo          io     io
-        |    7    true [3, "boo"]               fast
-        |    8    true          3        fast   fast
+        |    0    true          3          io     io
+        |    1   false               boo fast     io
+        |    2   false               boo   io     io
+        |    3    true        boo          io     io
+        |    4    true       true          io     io
+        |    5    true        boo               fast
+        |    6    true        Boo        fast   fast
+        |    7    true [3, "boo"]                 io
+        |    8    true          3          io     io
       TABLE
     end