Better immediete factory methods: (succeeded|failed|completed)_future

Author: pitr-ch

examples/benchmark_new_futures.rb

@@ -4,25 +4,6 @@
 require 'concurrent'
 require 'concurrent-edge'
 
-# require 'ruby-prof'
-#
-# result = RubyProf.profile do
-#   1000.times do
-#     head    = Concurrent.future { 1 }
-#     branch1 = head.then(&:succ)
-#     branch2 = head.then(&:succ).then(&:succ)
-#     branch3 = head.then(&:succ).then(&:succ).then(&:succ)
-#     Concurrent.join(branch1, branch2, branch3).then { |(a, b, c)| a + b + c }.value!
-#   end
-# end
-#
-# printer = RubyProf::FlatPrinter.new(result)
-# printer.print(STDOUT)
-#
-# printer = RubyProf::GraphPrinter.new(result)
-# printer.print(STDOUT, {})
-#
-# exit
 
 scale  = 1
 time   = 10 * scale
@@ -43,7 +24,7 @@
 
 Benchmark.ips(time, warmup) do |x|
   of = Concurrent::Promise.execute { 1 }
-  nf = Concurrent.future(:fast) { 1 }
+  nf = Concurrent.succeeded_future(1, :fast)
   x.report('value-old') { of.value! }
   x.report('value-new') { nf.value! }
   x.compare!
@@ -60,7 +41,7 @@
     head.value!
   end
   x.report('graph-new') do
-    head = Concurrent.future(:fast) { 1 }
+    head = Concurrent.succeeded_future(1, :fast)
     10.times do
       branch1 = head.then(&:succ)
       branch2 = head.then(&:succ).then(&:succ)
@@ -73,13 +54,13 @@
 
 Benchmark.ips(time, warmup) do |x|
   x.report('immediate-old') { Concurrent::Promise.execute { nil }.value! }
-  x.report('immediate-new') { Concurrent.future(:fast) { nil }.value! }
+  x.report('immediate-new') { Concurrent.succeeded_future(nil, :fast).value! }
   x.compare!
 end
 
 Benchmark.ips(time, warmup) do |x|
   of = Concurrent::Promise.execute { 1 }
-  nf = Concurrent.future(:fast) { 1 }
+  nf = Concurrent.succeeded_future(1, :fast)
   x.report('then-old') { 100.times.reduce(nf) { |nf, _| nf.then(&:succ) }.value! }
   x.report('then-new') { 100.times.reduce(nf) { |nf, _| nf.then(&:succ) }.value! }
   x.compare!

examples/edge_futures.in.rb

@@ -19,7 +19,7 @@
 
 ### Chaining
 
-head    = Concurrent.completed_future 1 #
+head    = Concurrent.succeeded_future 1 #
 branch1 = head.then(&:succ) #
 branch2 = head.then(&:succ).then(&:succ) #
 branch1.zip(branch2).value!
@@ -35,7 +35,7 @@
 Concurrent.future { Object.new }.then(&:succ).rescue { 1 }.then(&:succ).value
 Concurrent.future { 1 }.then(&:succ).rescue { |e| e.message }.then(&:succ).value
 
-failing_zip = Concurrent.completed_future(1) & Concurrent.future { raise 'boom' }
+failing_zip = Concurrent.succeeded_future(1) & Concurrent.failed_future(StandardError.new('boom'))
 failing_zip.result
 failing_zip.then { |v| 'never happens' }.result
 failing_zip.rescue { |a, b| (a || b).message }.value

examples/edge_futures.out.rb

@@ -1,7 +1,7 @@
 ### Simple asynchronous task
 
 future = Concurrent.future { sleep 0.1; 1 + 1 } # evaluation starts immediately
-    # => <#Concurrent::Edge::Future:0x7fc6218f2318 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc73208180 pending blocks:[]>
 future.completed?                                  # => false
 # block until evaluated
 future.value                                       # => 2
@@ -11,7 +11,7 @@
 ### Failing asynchronous task
 
 future = Concurrent.future { raise 'Boom' }
-    # => <#Concurrent::Edge::Future:0x7fc6218eae38 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc731fa0a8 pending blocks:[]>
 future.value                                       # => nil
 future.value! rescue $!                            # => #<RuntimeError: Boom>
 future.reason                                      # => #<RuntimeError: Boom>
@@ -21,7 +21,7 @@
 
 ### Chaining
 
-head    = Concurrent.completed_future 1 
+head    = Concurrent.succeeded_future 1 
 branch1 = head.then(&:succ) 
 branch2 = head.then(&:succ).then(&:succ) 
 branch1.zip(branch2).value!                        # => [2, 3]
@@ -41,10 +41,10 @@
 Concurrent.future { 1 }.then(&:succ).rescue { |e| e.message }.then(&:succ).value
     # => 3
 
-failing_zip = Concurrent.completed_future(1) & Concurrent.future { raise 'boom' }
-    # => <#Concurrent::Edge::Future:0x7fc6218b0f08 pending blocks:[]>
-failing_zip.result                                 # => [false, [1, nil], [nil, #<RuntimeError: boom>]]
-failing_zip.then { |v| 'never happens' }.result    # => [false, [1, nil], [nil, #<RuntimeError: boom>]]
+failing_zip = Concurrent.succeeded_future(1) & Concurrent.failed_future(StandardError.new('boom'))
+    # => <#Concurrent::Edge::Future:0x7fcc731c00b0 failed blocks:[]>
+failing_zip.result                                 # => [false, [1, nil], [nil, #<StandardError: boom>]]
+failing_zip.then { |v| 'never happens' }.result    # => [false, [1, nil], [nil, #<StandardError: boom>]]
 failing_zip.rescue { |a, b| (a || b).message }.value
     # => "boom"
 failing_zip.chain { |success, values, reasons| [success, values.compact, reasons.compactß] }.value
@@ -54,28 +54,28 @@
 
 # will not evaluate until asked by #value or other method requiring completion
 future = Concurrent.delay { 'lazy' }
-    # => <#Concurrent::Edge::Future:0x7fc6218a37e0 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc731a1840 pending blocks:[]>
 sleep 0.1 
 future.completed?                                  # => false
 future.value                                       # => "lazy"
 
 # propagates trough chain allowing whole or partial lazy chains
 
 head    = Concurrent.delay { 1 }
-    # => <#Concurrent::Edge::Future:0x7fc6218a0720 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc73193b28 pending blocks:[]>
 branch1 = head.then(&:succ)
-    # => <#Concurrent::Edge::Future:0x7fc6212c7b50 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc73190900 pending blocks:[]>
 branch2 = head.delay.then(&:succ)
-    # => <#Concurrent::Edge::Future:0x7fc6212c6098 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc7318b400 pending blocks:[]>
 join    = branch1 & branch2
-    # => <#Concurrent::Edge::Future:0x7fc6212c4f40 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc73180af0 pending blocks:[]>
 
 sleep 0.1 # nothing will complete                  # => 0
 [head, branch1, branch2, join].map(&:completed?)   # => [false, false, false, false]
 
 branch1.value                                      # => 2
 sleep 0.1 # forces only head to complete, branch 2 stays incomplete
-    # => 1
+    # => 0
 [head, branch1, branch2, join].map(&:completed?)   # => [true, true, false, false]
 
 join.value                                         # => [2, 2]
@@ -96,14 +96,14 @@
 ### Schedule
 
 scheduled = Concurrent.schedule(0.1) { 1 }
-    # => <#Concurrent::Edge::Future:0x7fc62128c550 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc73143e48 pending blocks:[]>
 
 scheduled.completed?                               # => false
 scheduled.value # available after 0.1sec           # => 1
 
 # and in chain
 scheduled = Concurrent.delay { 1 }.schedule(0.1).then(&:succ)
-    # => <#Concurrent::Edge::Future:0x7fc6228bcdc0 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc7313a758 pending blocks:[]>
 # will not be scheduled until value is requested
 sleep 0.1 
 scheduled.value # returns after another 0.1sec     # => 2
@@ -112,36 +112,36 @@
 ### Completable Future and Event
 
 future = Concurrent.future
-    # => <#Concurrent::Edge::CompletableFuture:0x7fc623083720 pending blocks:[]>
+    # => <#Concurrent::Edge::CompletableFuture:0x7fcc731286e8 pending blocks:[]>
 event  = Concurrent.event
-    # => <#Concurrent::Edge::CompletableEvent:0x7fc623081100 pending blocks:[]>
+    # => <#Concurrent::Edge::CompletableEvent:0x7fcc73123058 pending blocks:[]>
 # Don't forget to keep the reference, `Concurrent.future.then { |v| v }` is incompletable
 
 # will be blocked until completed
 t1     = Thread.new { future.value } 
 t2     = Thread.new { event.wait } 
 
 future.success 1
-    # => <#Concurrent::Edge::CompletableFuture:0x7fc623083720 success blocks:[]>
+    # => <#Concurrent::Edge::CompletableFuture:0x7fcc731286e8 success blocks:[]>
 future.success 1 rescue $!
     # => #<Concurrent::MultipleAssignmentError: Future can be completed only once. Current result is [true, 1, nil], trying to set [true, 1, nil]>
 future.try_success 2                               # => false
 event.complete
-    # => <#Concurrent::Edge::CompletableEvent:0x7fc623081100 completed blocks:[]>
+    # => <#Concurrent::Edge::CompletableEvent:0x7fcc73123058 completed blocks:[]>
 
 [t1, t2].each &:join 
 
 
 ### Callbacks
 
-queue  = Queue.new                                 # => #<Thread::Queue:0x007fc62127df00>
+queue  = Queue.new                                 # => #<Thread::Queue:0x007fcc73110638>
 future = Concurrent.delay { 1 + 1 }
-    # => <#Concurrent::Edge::Future:0x7fc62127c060 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc7310ab98 pending blocks:[]>
 
 future.on_success { queue << 1 } # evaluated asynchronously
-    # => <#Concurrent::Edge::Future:0x7fc62127c060 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc7310ab98 pending blocks:[]>
 future.on_success! { queue << 2 } # evaluated on completing thread
-    # => <#Concurrent::Edge::Future:0x7fc62127c060 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc7310ab98 pending blocks:[]>
 
 queue.empty?                                       # => true
 future.value                                       # => 2
@@ -152,15 +152,15 @@
 ### Thread-pools
 
 Concurrent.future(:fast) { 2 }.then(:io) { File.read __FILE__ }.wait
-    # => <#Concurrent::Edge::Future:0x7fc62125d9a8 success blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc730f98e8 success blocks:[]>
 
 
 ### Interoperability with actors
 
 actor = Concurrent::Actor::Utils::AdHoc.spawn :square do
   -> v { v ** 2 }
 end
-    # => #<Concurrent::Actor::Reference:0x7fc621234a30 /square (Concurrent::Actor::Utils::AdHoc)>
+    # => #<Concurrent::Actor::Reference:0x7fcc730c36f8 /square (Concurrent::Actor::Utils::AdHoc)>
 
 Concurrent.
     future { 2 }.
@@ -173,32 +173,32 @@
 
 ### Interoperability with channels
 
-ch1 = Concurrent::Edge::Channel.new                # => #<Concurrent::Edge::Channel:0x007fc621205460>
-ch2 = Concurrent::Edge::Channel.new                # => #<Concurrent::Edge::Channel:0x007fc6212041c8>
+ch1 = Concurrent::Edge::Channel.new                # => #<Concurrent::Edge::Channel:0x007fcc73043188>
+ch2 = Concurrent::Edge::Channel.new                # => #<Concurrent::Edge::Channel:0x007fcc730425f8>
 
 result = Concurrent.select(ch1, ch2)
-    # => <#Concurrent::Edge::CompletableFuture:0x7fc6211fe5c0 pending blocks:[]>
+    # => <#Concurrent::Edge::CompletableFuture:0x7fcc730411a8 pending blocks:[]>
 ch1.push 1                                         # => nil
 result.value!
-    # => [1, #<Concurrent::Edge::Channel:0x007fc621205460>]
+    # => [1, #<Concurrent::Edge::Channel:0x007fcc73043188>]
 
 Concurrent.
     future { 1+1 }.
     then_push(ch1)
-    # => <#Concurrent::Edge::Future:0x7fc6211f7338 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc73032c98 pending blocks:[]>
 result = Concurrent.
     future { '%02d' }.
     then_select(ch1, ch2).
     then { |format, (value, channel)| format format, value }
-    # => <#Concurrent::Edge::Future:0x7fc6211ec668 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc7302a4f8 pending blocks:[]>
 result.value!                                      # => "02"
 
 
 ### Common use-cases Examples
 
 # simple background processing
 Concurrent.future { do_stuff }
-    # => <#Concurrent::Edge::Future:0x7fc6211df170 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc72123c48 pending blocks:[]>
 
 # parallel background processing
 jobs = 10.times.map { |i| Concurrent.future { i } } 
@@ -215,7 +215,7 @@ def schedule_job
 end                                                # => :schedule_job
 
 schedule_job
-    # => <#Concurrent::Edge::Future:0x7fc62119c140 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fcc75011370 pending blocks:[]>
 @end = true                                        # => true
 
 
@@ -228,7 +228,7 @@ def schedule_job
     data[message]
   end
 end
-    # => #<Concurrent::Actor::Reference:0x7fc62117f568 /db (Concurrent::Actor::Utils::AdHoc)>
+    # => #<Concurrent::Actor::Reference:0x7fcc71832a08 /db (Concurrent::Actor::Utils::AdHoc)>
 
 concurrent_jobs = 11.times.map do |v|
   Concurrent.
@@ -258,7 +258,7 @@ def schedule_job
     end
   end
 end
-    # => #<Concurrent::Actor::Reference:0x7fc6218969f0 /DB-pool (Concurrent::Actor::Utils::Pool)>
+    # => #<Concurrent::Actor::Reference:0x7fcc72320118 /DB-pool (Concurrent::Actor::Utils::Pool)>
 
 concurrent_jobs = 11.times.map do |v|
   Concurrent.

lib/concurrent/edge/future.rb

@@ -39,9 +39,19 @@ def future(default_executor = :io, &task)
         end
       end
 
-      # @return [Future] which is already completed with value
-      def completed_future(value, default_executor = :io)
-        ImmediateFuturePromise.new(default_executor, value).future
+      # @return [Future] which is already completed
+      def completed_future(success, value, reason, default_executor = :io)
+        ImmediateFuturePromise.new(default_executor, success, value, reason).future
+      end
+
+      # @return [Future] which is already completed in success state with value
+      def succeeded_future(value, default_executor = :io)
+        completed_future true, value, nil, default_executor
+      end
+
+      # @return [Future] which is already completed in failed state with reason
+      def failed_future(reason, default_executor = :io)
+        completed_future false, nil, reason, default_executor
       end
 
       # @return [Event] which is already completed
@@ -678,10 +688,10 @@ def zip(other)
 
       alias_method :|, :any
 
-      # only proof of concept
       # @note may block
+      # @note only proof of concept
       def then_push(channel)
-        on_success { |value| channel.push value }
+        on_success(:io) { |value| channel.push value }
       end
 
       # @yield [value] executed async on `executor` when success
@@ -1104,8 +1114,9 @@ def initialize(default_executor)
 
     # @!visibility private
     class ImmediateFuturePromise < InnerPromise
-      def initialize(default_executor, value)
-        super Future.new(self, default_executor).complete_with(Future::Success.new(value))
+      def initialize(default_executor, success, value, reason)
+        super Future.new(self, default_executor).
+                  complete_with(success ? Future::Success.new(value) : Future::Failed.new(reason))
       end
     end
 
@@ -1360,7 +1371,7 @@ def initialize(default_executor, intended_time)
       end
     end
 
-    # proof of concept
+    # @note proof of concept
     class Channel < Synchronization::Object
       # TODO make lock free
       def initialize

spec/concurrent/edge/future_spec.rb

@@ -36,7 +36,7 @@
       future = Concurrent.future { 1 + 1 }
       expect(future.value!).to eq 2
 
-      future = Concurrent.completed_future(1).then { |v| v + 1 }
+      future = Concurrent.succeeded_future(1).then { |v| v + 1 }
       expect(future.value!).to eq 2
     end
   end
@@ -47,7 +47,7 @@
       expect(delay.completed?).to eq false
       expect(delay.value!).to eq 2
 
-      delay = Concurrent.completed_future(1).delay.then { |v| v + 1 }
+      delay = Concurrent.succeeded_future(1).delay.then { |v| v + 1 }
       expect(delay.completed?).to eq false
       expect(delay.value!).to eq 2
     end
@@ -66,7 +66,7 @@
       start  = Time.now.to_f
       queue  = Queue.new
       future = Concurrent.
-          completed_future(1).
+          succeeded_future(1).
           schedule(0.1).
           then { |v| v + 1 }.
           then { |v| queue.push(v); queue.push(Time.now.to_f - start); queue }