Merge pull request #349 from pitr-ch/futures

Edge::Future updates

Author: Petr Chalupa

concurrent-ruby-edge.gemspec

@@ -10,7 +10,7 @@ Gem::Specification.new do |s|
   s.name             = 'concurrent-ruby-edge'
   s.version          = Concurrent::EDGE_VERSION
   s.platform         = Gem::Platform::RUBY
-  s.authors          = ["Jerry D'Antonio", 'The Ruby Concurrency Team']
+  s.authors          = ["Jerry D'Antonio", 'Petr Chalupa', 'The Ruby Concurrency Team']
   s.email            = ['[email protected]', '[email protected]']
   s.homepage         = 'http://www.concurrent-ruby.com'
   s.summary          = 'Edge features and additions to the concurrent-ruby gem.'

examples/edge_futures.in.rb

@@ -19,7 +19,7 @@
 
 ### Chaining
 
-head    = Concurrent.future { 1 } #
+head    = Concurrent.completed_future 1 #
 branch1 = head.then(&:succ) #
 branch2 = head.then(&:succ).then(&:succ) #
 branch1.zip(branch2).value!
@@ -29,10 +29,6 @@
 # pick only first completed
 (branch1 | branch2).value!
 
-# auto splat arrays for blocks
-Concurrent.future { [1, 2] }.then(&:+).value!
-
-
 ### Error handling
 
 Concurrent.future { Object.new }.then(&:succ).then(&:succ).rescue { |e| e.class }.value # error propagates
@@ -94,6 +90,7 @@
 
 future = Concurrent.future
 event  = Concurrent.event
+# Don't forget to keep the reference, `Concurrent.future.then { |v| v }` is incompletable
 
 # will be blocked until completed
 t1     = Thread.new { future.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:0x7fcc038ca588 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a956d70 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:0x7fcc038c05b0 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a946c18 failed blocks:[]>
 future.value                                       # => nil
 future.value! rescue $!                            # => #<RuntimeError: Boom>
 future.reason                                      # => #<RuntimeError: Boom>
@@ -21,7 +21,7 @@
 
 ### Chaining
 
-head    = Concurrent.future { 1 } 
+head    = Concurrent.completed_future 1 
 branch1 = head.then(&:succ) 
 branch2 = head.then(&:succ).then(&:succ) 
 branch1.zip(branch2).value!                        # => [2, 3]
@@ -32,10 +32,6 @@
 # pick only first completed
 (branch1 | branch2).value!                         # => 2
 
-# auto splat arrays for blocks
-Concurrent.future { [1, 2] }.then(&:+).value!      # => 3
-
-
 ### Error handling
 
 Concurrent.future { Object.new }.then(&:succ).then(&:succ).rescue { |e| e.class }.value # error propagates
@@ -50,28 +46,28 @@
 
 # will not evaluate until asked by #value or other method requiring completion
 future = Concurrent.delay { 'lazy' }
-    # => <#Concurrent::Edge::Future:0x7fcc022afaa0 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a8f6ec0 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:0x7fcc022a5640 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a8ee1a8 pending blocks:[]>
 branch1 = head.then(&:succ)
-    # => <#Concurrent::Edge::Future:0x7fcc0229c478 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a8ed438 pending blocks:[]>
 branch2 = head.delay.then(&:succ)
-    # => <#Concurrent::Edge::Future:0x7fcc0228f318 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a8e7588 pending blocks:[]>
 join    = branch1 & branch2
-    # => <#Concurrent::Edge::Future:0x7fcc0228de78 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a8e4ec8 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
-    # => 0
+    # => 1
 [head, branch1, branch2, join].map(&:completed?)   # => [true, true, false, false]
 
 join.value                                         # => [2, 2]
@@ -92,14 +88,14 @@
 ### Schedule
 
 scheduled = Concurrent.schedule(0.1) { 1 }
-    # => <#Concurrent::Edge::Future:0x7fcc0385b368 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a8a4468 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:0x7fcc03843948 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a895828 pending blocks:[]>
 # will not be scheduled until value is requested
 sleep 0.1 
 scheduled.value # returns after another 0.1sec     # => 2
@@ -108,35 +104,36 @@
 ### Completable Future and Event
 
 future = Concurrent.future
-    # => <#Concurrent::Edge::CompletableFuture:0x7fcc0304fdb8 pending blocks:[]>
+    # => <#Concurrent::Edge::CompletableFuture:0x7fd41a87c648 pending blocks:[]>
 event  = Concurrent.event
-    # => <#Concurrent::Edge::CompletableEvent:0x7fcc0304e210 pending blocks:[]>
+    # => <#Concurrent::Edge::CompletableEvent:0x7fd41a8770d0 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:0x7fcc0304fdb8 success blocks:[]>
+    # => <#Concurrent::Edge::CompletableFuture:0x7fd41a87c648 success blocks:[]>
 future.success 1 rescue $!
-    # => #<Concurrent::MultipleAssignmentError: multiple assignment>
+    # => #<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:0x7fcc0304e210 completed blocks:[]>
+    # => <#Concurrent::Edge::CompletableEvent:0x7fd41a8770d0 completed blocks:[]>
 
 [t1, t2].each &:join 
 
 
 ### Callbacks
 
-queue  = Queue.new                                 # => #<Thread::Queue:0x007fcc03101720>
+queue  = Queue.new                                 # => #<Thread::Queue:0x007fd41b0543e8>
 future = Concurrent.delay { 1 + 1 }
-    # => <#Concurrent::Edge::Future:0x7fcc030fbaf0 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41b04f028 pending blocks:[]>
 
 future.on_success { queue << 1 } # evaluated asynchronously
-    # => <#Concurrent::Edge::Future:0x7fcc030fbaf0 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41b04f028 pending blocks:[]>
 future.on_success! { queue << 2 } # evaluated on completing thread
-    # => <#Concurrent::Edge::Future:0x7fcc030fbaf0 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41b04f028 pending blocks:[]>
 
 queue.empty?                                       # => true
 future.value                                       # => 2
@@ -147,15 +144,15 @@
 ### Thread-pools
 
 Concurrent.future(:fast) { 2 }.then(:io) { File.read __FILE__ }.wait
-    # => <#Concurrent::Edge::Future:0x7fcc030e8bd0 success blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a857550 success blocks:[]>
 
 
 ### Interoperability with actors
 
 actor = Concurrent::Actor::Utils::AdHoc.spawn :square do
   -> v { v ** 2 }
 end
-    # => #<Concurrent::Actor::Reference:0x7fcc0223f020 /square (Concurrent::Actor::Utils::AdHoc)>
+    # => #<Concurrent::Actor::Reference:0x7fd41c03b5b8 /square (Concurrent::Actor::Utils::AdHoc)>
 
 Concurrent.
     future { 2 }.
@@ -168,32 +165,32 @@
 
 ### Interoperability with channels
 
-ch1 = Concurrent::Edge::Channel.new                # => #<Concurrent::Edge::Channel:0x007fcc021ec6b8>
-ch2 = Concurrent::Edge::Channel.new                # => #<Concurrent::Edge::Channel:0x007fcc021e7b18>
+ch1 = Concurrent::Edge::Channel.new                # => #<Concurrent::Edge::Channel:0x007fd41a297458>
+ch2 = Concurrent::Edge::Channel.new                # => #<Concurrent::Edge::Channel:0x007fd41a296940>
 
 result = Concurrent.select(ch1, ch2)
-    # => <#Concurrent::Edge::CompletableFuture:0x7fcc021e6060 pending blocks:[]>
+    # => <#Concurrent::Edge::CompletableFuture:0x7fd41a295c98 pending blocks:[]>
 ch1.push 1                                         # => nil
 result.value!
-    # => [1, #<Concurrent::Edge::Channel:0x007fcc021ec6b8>]
+    # => [1, #<Concurrent::Edge::Channel:0x007fd41a297458>]
 
 Concurrent.
     future { 1+1 }.
     then_push(ch1)
-    # => <#Concurrent::Edge::Future:0x7fcc021dc3d0 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a284010 pending blocks:[]>
 result = Concurrent.
     future { '%02d' }.
     then_select(ch1, ch2).
     then { |format, (value, channel)| format format, value }
-    # => <#Concurrent::Edge::Future:0x7fcc021cd9e8 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a25d938 pending blocks:[]>
 result.value!                                      # => "02"
 
 
 ### Common use-cases Examples
 
 # simple background processing
 Concurrent.future { do_stuff }
-    # => <#Concurrent::Edge::Future:0x7fcc021b7530 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a23d520 pending blocks:[]>
 
 # parallel background processing
 jobs = 10.times.map { |i| Concurrent.future { i } } 
@@ -210,7 +207,7 @@ def schedule_job
 end                                                # => :schedule_job
 
 schedule_job
-    # => <#Concurrent::Edge::Future:0x7fcc0218faf8 pending blocks:[]>
+    # => <#Concurrent::Edge::Future:0x7fd41a2245e8 pending blocks:[]>
 @end = true                                        # => true
 
 
@@ -223,7 +220,7 @@ def schedule_job
     data[message]
   end
 end
-    # => #<Concurrent::Actor::Reference:0x7fcc0214f458 /db (Concurrent::Actor::Utils::AdHoc)>
+    # => #<Concurrent::Actor::Reference:0x7fd41a206228 /db (Concurrent::Actor::Utils::AdHoc)>
 
 concurrent_jobs = 11.times.map do |v|
   Concurrent.
@@ -253,7 +250,7 @@ def schedule_job
     end
   end
 end
-    # => #<Concurrent::Actor::Reference:0x7fcc02930398 /DB-pool (Concurrent::Actor::Utils::Pool)>
+    # => #<Concurrent::Actor::Reference:0x7fd41a0530c0 /DB-pool (Concurrent::Actor::Utils::Pool)>
 
 concurrent_jobs = 11.times.map do |v|
   Concurrent.

lib/concurrent/edge.rb

@@ -21,10 +21,10 @@ module Concurrent
   #
   # @!macro [attach] edge_warning
   #   @api Edge
-  #   @note **Edge Feature:** Edge features are under active development and may
-  #     change frequently. They are not expected to keep backward compatibility.
-  #     They may also lack tests and documentation. Use with caution.
+  #   @note **Edge Feature:** Edge features are under active development and may change frequently. They are expected not to
+  #     keep backward compatibility (there may also lack tests and documentation). Semantic versions will
+  #     be obeyed though. Features developed in `concurrent-ruby-edge` are expected to move
+  #     to `concurrent-ruby` when final.
   module Edge
-
   end
 end