Merge pull request #314 from pitr-ch/futures

Futures: passing args on construction, interoperability with channel, documentation

Author: Petr Chalupa

README.md

@@ -106,6 +106,13 @@ be obeyed though. Features developed in `concurrent-ruby-edge` are expected to m
   Communicating Sequential Processes (CSP).
 * [Exchanger](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Exchanger.html)
 * [LazyRegister](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/LazyRegister.html)
+* [New Future Promise Framework](http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Edge.html) - new 
+  unified implementation of Futures and Promises which combines Features of previous `Future`,
+  `Promise`, `IVar`, `Probe`, `dataflow`, `Delay`, `TimerTask` into single framework. It uses extensively
+  new synchronization layer to make all the paths lock-free with exception of blocking threads on `#wait`.
+  It offers better performance and does not block threads (exception being `#wait` and similar methods where it's
+  intended).
+
 
 ## Usage
 
@@ -149,6 +156,8 @@ require 'concurrent/actor'          # Concurrent::Actor and supporting code
 require 'concurrent/channel '       # Concurrent::Channel and supporting code
 ```
 
+If the library does not behave as expected, `Concurrent.use_stdlib_logger(Logger::DEBUG)` could help to revel the problem.
+
 ## Installation
 
 ```shell

examples/edge_futures.in.rb

@@ -22,10 +22,15 @@
 head    = Concurrent.future { 1 } #
 branch1 = head.then(&:succ) #
 branch2 = head.then(&:succ).then(&:succ) #
-branch1.zip(branch2).value
-(branch1 & branch2).then { |(a, b)| a + b }.value
+branch1.zip(branch2).value!
+(branch1 & branch2).then { |a, b| a + b }.value!
+(branch1 & branch2).then(&:+).value!
+Concurrent.zip(branch1, branch2, branch1).then { |*values| values.reduce &:+ }.value!
 # pick only first completed
-(branch1 | branch2).value
+(branch1 | branch2).value!
+
+# auto splat arrays for blocks
+Concurrent.future { [1, 2] }.then(&:+).value!
 
 
 ### Error handling
@@ -38,7 +43,7 @@
 ### Delay
 
 # will not evaluate until asked by #value or other method requiring completion
-scheduledfuture = Concurrent.delay { 'lazy' }
+future = Concurrent.delay { 'lazy' }
 sleep 0.1 #
 future.completed?
 future.value
@@ -136,6 +141,25 @@
 actor.ask(2).then(&:succ).value
 
 
+### Interoperability with channels
+
+ch1 = Concurrent::Edge::Channel.new
+ch2 = Concurrent::Edge::Channel.new
+
+result = Concurrent.select(ch1, ch2)
+ch1.push 1
+result.value!
+
+Concurrent.
+    future { 1+1 }.
+    then_push(ch1)
+result = Concurrent.
+    future { '%02d' }.
+    then_select(ch1, ch2).
+    then { |format, (value, channel)| format format, value }
+result.value!
+
+
 ### Common use-cases Examples
 
 # simple background processing
@@ -147,12 +171,62 @@
 
 
 # periodic task
+@end = false
+
 def schedule_job
   Concurrent.schedule(1) { do_stuff }.
-      rescue { |e| report_error e }.
-      then { schedule_job }
+      rescue { |e| StandardError === e ? report_error(e) : raise(e) }.
+      then { schedule_job unless @end }
 end
 
 schedule_job
+@end = true
+
+
+# How to limit processing where there are limited resources?
+# By creating an actor managing the resource
+DB   = Concurrent::Actor::Utils::AdHoc.spawn :db do
+  data = Array.new(10) { |i| '*' * i }
+  lambda do |message|
+    # pretending that this queries a DB
+    data[message]
+  end
+end
+
+concurrent_jobs = 11.times.map do |v|
+  Concurrent.
+      future { v }.
+      # ask the DB with the `v`, only one at the time, rest is parallel
+      then_ask(DB).
+      # get size of the string, fails for 11
+      then(&:size).
+      rescue { |reason| reason.message } # translate error to value (exception, message)
+end #
+
+Concurrent.zip(*concurrent_jobs).value!
+
+
+# In reality there is often a pool though:
+data      = Array.new(10) { |i| '*' * i }
+pool_size = 5
+
+DB_POOL = Concurrent::Actor::Utils::Pool.spawn!('DB-pool', pool_size) do |index|
+  # DB connection constructor
+  Concurrent::Actor::Utils::AdHoc.spawn(name: "worker-#{index}", args: [data]) do |data|
+    lambda do |message|
+      # pretending that this queries a DB
+      data[message]
+    end
+  end
+end
 
+concurrent_jobs = 11.times.map do |v|
+  Concurrent.
+      future { v }.
+      # ask the DB_POOL with the `v`, only 5 at the time, rest is parallel
+      then_ask(DB_POOL).
+      then(&:size).
+      rescue { |reason| reason.message }
+end #
 
+Concurrent.zip(*concurrent_jobs).value!