lib/concurrent/futures.rb

   1 #
   2 # concurrent/futures - futures and lazy evaluation for Ruby
   3 #
   4 # Copyright (C) 2007  MenTaLguY <mental@rydia.net>
   5 #
   6 # This file is made available under the same terms as Ruby.
   7 #
   8
   9 require 'thread'
  10 begin
  11   require 'fastthread'
  12 rescue LoadError
  13 end
  14 require 'thwait'
  15
  16 module Concurrent
  17 module Futures
  18
  19 module Future
  20   extend Futures
  21 end
  22
  23 class AsyncError < RuntimeError
  24   attr_reader :reason
  25
  26   def initialize( reason, desc=nil )
  27     super( desc )
  28     @reason = reason
  29   end
  30 end
  31
  32 require 'concurrent/futures.so' # provides Thunk
  33
  34 def async( &block )
  35   Thunk.new( Thread.new( &block ) )
  36 end
  37 alias spawn async
  38 alias future async
  39
  40 def await( future )
  41   Thunk.value future
  42 end
  43
  44 def await_any( *futures )
  45   threads = futures.map { |future| Thread.new { Futures::await future } }
  46   begin
  47     threads.index ThreadsWait.new( *threads ).wait_next
  48   ensure
  49     threads.each { |thread| thread.raise RuntimeError }
  50   end
  51 end
  52
  53 AlreadyFulfilledError = Class.new StandardError
  54
  55 class Promise
  56   def initialize
  57     @lock = Mutex.new
  58     @ready = ConditionVariable.new
  59   end
  60
  61   def future
  62     @future ||= Thunk.new self
  63   end
  64
  65   def fulfill( value )
  66     @lock.synchronize do
  67       if defined? @value
  68         raise AlreadyFulfilledError, "promise already fulfilled"
  69       end
  70       @value = value
  71       @ready.broadcast
  72     end
  73     self
  74   end
  75
  76   def fulfilled?
  77     @lock.synchronize { defined? @value }
  78   end
  79
  80   def value
  81     @lock.synchronize do
  82       @ready.wait @lock until defined? @value
  83       @value
  84     end
  85   end
  86
  87   def fail( ex )
  88     @lock.synchronize do
  89       @value = Thunk.new( Thread.new { raise ex } )
  90     end
  91     self
  92   end
  93 end
  94
  95 class Lazy
  96   def initialize( &block )
  97     @lock = Mutex.new
  98   end
  99
 100   def value
 101     @lock.synchronize do
 102       if @block
 103         @value = Thunk.new Thread.new( &block )
 104         @block = nil
 105       end
 106       @value
 107     end
 108   end
 109
 110   def inspect
 111     @lock.synchronize do
 112       if @block
 113         "#<Lazy pending #{ @block.inspect }>"
 114       else
 115         "#<Lazy requested #{ @value.inspect }>"
 116       end
 117     end
 118   end
 119 end
 120
 121 def lazy( &block )
 122   Thunk.new( Lazy.new( &block ) )
 123 end
 124
 125 class Ref
 126   def initialize( value=nil )
 127     @lock = Mutex.new
 128     @value = value
 129   end
 130
 131   def exchange( value )
 132     @lock.synchronize do
 133       result = @value
 134       @value = value
 135       result
 136     end
 137   end
 138
 139   def modify( &block )
 140     @lock.synchronize do
 141       value = @value
 142       @value = Future.async { block.call value }
 143     end
 144   end
 145 end
 146
 147 end
 148 end
 149