require 'concurrent/synchronization' require 'concurrent/atomic/atomic_boolean' require 'concurrent/atomic/atomic_fixnum' require 'concurrent/collection/lock_free_stack' require 'concurrent/errors' require 'concurrent/re_include' module Concurrent # {include:file:docs-source/promises-main.md} module Promises # @!macro promises.param.default_executor # @param [Executor, :io, :fast] default_executor Instance of an executor or a name of the # global executor. Default executor propagates to chained futures unless overridden with # executor parameter or changed with {AbstractEventFuture#with_default_executor}. # # @!macro promises.param.executor # @param [Executor, :io, :fast] executor Instance of an executor or a name of the # global executor. The task is executed on it, default executor remains unchanged. # # @!macro promises.param.args # @param [Object] args arguments which are passed to the task when it's executed. # (It might be prepended with other arguments, see the @yeild section). # # @!macro promises.shortcut.on # Shortcut of {#$0_on} with default `:io` executor supplied. # @see #$0_on # # @!macro promises.shortcut.using # Shortcut of {#$0_using} with default `:io` executor supplied. # @see #$0_using # # @!macro promise.param.task-future # @yieldreturn will become result of the returned Future. # Its returned value becomes {Future#value} fulfilling it, # raised exception becomes {Future#reason} rejecting it. # # @!macro promise.param.callback # @yieldreturn is forgotten. # Container of all {Future}, {Event} factory methods. They are never constructed directly with # new. module FactoryMethods extend ReInclude extend self module Configuration # @return [Executor, :io, :fast] the executor which is used when none is supplied # to a factory method. The method can be overridden in the receivers of # `include FactoryMethod` def default_executor :io end end include Configuration # @!macro promises.shortcut.on # @return [ResolvableEvent] def resolvable_event resolvable_event_on default_executor end # Created resolvable event, user is responsible for resolving the event once by # {Promises::ResolvableEvent#resolve}. # # @!macro promises.param.default_executor # @return [ResolvableEvent] def resolvable_event_on(default_executor = self.default_executor) ResolvableEventPromise.new(default_executor).future end # @!macro promises.shortcut.on # @return [ResolvableFuture] def resolvable_future resolvable_future_on default_executor end # Creates resolvable future, user is responsible for resolving the future once by # {Promises::ResolvableFuture#resolve}, {Promises::ResolvableFuture#fulfill}, # or {Promises::ResolvableFuture#reject} # # @!macro promises.param.default_executor # @return [ResolvableFuture] def resolvable_future_on(default_executor = self.default_executor) ResolvableFuturePromise.new(default_executor).future end # @!macro promises.shortcut.on # @return [Future] def future(*args, &task) future_on(default_executor, *args, &task) end # Constructs new Future which will be resolved after block is evaluated on default executor. # Evaluation begins immediately. # # @!macro promises.param.default_executor # @!macro promises.param.args # @yield [*args] to the task. # @!macro promise.param.task-future # @return [Future] def future_on(default_executor, *args, &task) ImmediateEventPromise.new(default_executor).future.then(*args, &task) end # Creates resolved future with will be either fulfilled with the given value or rejection with # the given reason. # # @param [true, false] fulfilled # @param [Object] value # @param [Object] reason # @!macro promises.param.default_executor # @return [Future] def resolved_future(fulfilled, value, reason, default_executor = self.default_executor) ImmediateFuturePromise.new(default_executor, fulfilled, value, reason).future end # Creates resolved future with will be fulfilled with the given value. # # @!macro promises.param.default_executor # @param [Object] value # @return [Future] def fulfilled_future(value, default_executor = self.default_executor) resolved_future true, value, nil, default_executor end # Creates resolved future with will be rejected with the given reason. # # @!macro promises.param.default_executor # @param [Object] reason # @return [Future] def rejected_future(reason, default_executor = self.default_executor) resolved_future false, nil, reason, default_executor end # Creates resolved event. # # @!macro promises.param.default_executor # @return [Event] def resolved_event(default_executor = self.default_executor) ImmediateEventPromise.new(default_executor).event end # General constructor. Behaves differently based on the argument's type. It's provided for convenience # but it's better to be explicit. # # @see rejected_future, resolved_event, fulfilled_future # @!macro promises.param.default_executor # @return [Event, Future] # # @overload make_future(nil, default_executor = self.default_executor) # @param [nil] nil # @return [Event] resolved event. # # @overload make_future(a_future, default_executor = self.default_executor) # @param [Future] a_future # @return [Future] a future which will be resolved when a_future is. # # @overload make_future(an_event, default_executor = self.default_executor) # @param [Event] an_event # @return [Event] an event which will be resolved when an_event is. # # @overload make_future(exception, default_executor = self.default_executor) # @param [Exception] exception # @return [Future] a rejected future with the exception as its reason. # # @overload make_future(value, default_executor = self.default_executor) # @param [Object] value when none of the above overloads fits # @return [Future] a fulfilled future with the value. def make_future(argument = nil, default_executor = self.default_executor) case argument when AbstractEventFuture # returning wrapper would change nothing argument when Exception rejected_future argument, default_executor when nil resolved_event default_executor else fulfilled_future argument, default_executor end end # @!macro promises.shortcut.on # @return [Future, Event] def delay(*args, &task) delay_on default_executor, *args, &task end # Creates new event or future which is resolved only after it is touched, # see {Concurrent::AbstractEventFuture#touch}. # # @!macro promises.param.default_executor # @overload delay_on(default_executor, *args, &task) # If task is provided it returns a {Future} representing the result of the task. # @!macro promises.param.args # @yield [*args] to the task. # @!macro promise.param.task-future # @return [Future] # @overload delay_on(default_executor) # If no task is provided, it returns an {Event} # @return [Event] def delay_on(default_executor, *args, &task) event = DelayPromise.new(default_executor).event task ? event.chain(*args, &task) : event end # @!macro promises.shortcut.on # @return [Future, Event] def schedule(intended_time, *args, &task) schedule_on default_executor, intended_time, *args, &task end # Creates new event or future which is resolved in intended_time. # # @!macro promises.param.default_executor # @!macro promises.param.intended_time # @param [Numeric, Time] intended_time `Numeric` means to run in `intended_time` seconds. # `Time` means to run on `intended_time`. # @overload schedule_on(default_executor, intended_time, *args, &task) # If task is provided it returns a {Future} representing the result of the task. # @!macro promises.param.args # @yield [*args] to the task. # @!macro promise.param.task-future # @return [Future] # @overload schedule_on(default_executor, intended_time) # If no task is provided, it returns an {Event} # @return [Event] def schedule_on(default_executor, intended_time, *args, &task) event = ScheduledPromise.new(default_executor, intended_time).event task ? event.chain(*args, &task) : event end # @!macro promises.shortcut.on # @return [Future] def zip_futures(*futures_and_or_events) zip_futures_on default_executor, *futures_and_or_events end # Creates new future which is resolved after all futures_and_or_events are resolved. # Its value is array of zipped future values. Its reason is array of reasons for rejection. # If there is an error it rejects. # @!macro promises.event-conversion # If event is supplied, which does not have value and can be only resolved, it's # represented as `:fulfilled` with value `nil`. # # @!macro promises.param.default_executor # @param [AbstractEventFuture] futures_and_or_events # @return [Future] def zip_futures_on(default_executor, *futures_and_or_events) ZipFuturesPromise.new_blocked_by(futures_and_or_events, default_executor).future end alias_method :zip, :zip_futures # @!macro promises.shortcut.on # @return [Event] def zip_events(*futures_and_or_events) zip_events_on default_executor, *futures_and_or_events end # Creates new event which is resolved after all futures_and_or_events are resolved. # (Future is resolved when fulfilled or rejected.) # # @!macro promises.param.default_executor # @param [AbstractEventFuture] futures_and_or_events # @return [Event] def zip_events_on(default_executor, *futures_and_or_events) ZipEventsPromise.new_blocked_by(futures_and_or_events, default_executor).event end # @!macro promises.shortcut.on # @return [Future] def any_resolved_future(*futures_and_or_events) any_resolved_future_on default_executor, *futures_and_or_events end alias_method :any, :any_resolved_future # Creates new future which is resolved after first futures_and_or_events is resolved. # Its result equals result of the first resolved future. # @!macro promises.any-touch # If resolved it does not propagate {Concurrent::AbstractEventFuture#touch}, leaving delayed # futures un-executed if they are not required any more. # @!macro promises.event-conversion # # @!macro promises.param.default_executor # @param [AbstractEventFuture] futures_and_or_events # @return [Future] def any_resolved_future_on(default_executor, *futures_and_or_events) AnyResolvedFuturePromise.new_blocked_by(futures_and_or_events, default_executor).future end # @!macro promises.shortcut.on # @return [Future] def any_fulfilled_future(*futures_and_or_events) any_fulfilled_future_on default_executor, *futures_and_or_events end # Creates new future which is resolved after first of futures_and_or_events is fulfilled. # Its result equals result of the first resolved future or if all futures_and_or_events reject, # it has reason of the last resolved future. # @!macro promises.any-touch # @!macro promises.event-conversion # # @!macro promises.param.default_executor # @param [AbstractEventFuture] futures_and_or_events # @return [Future] def any_fulfilled_future_on(default_executor, *futures_and_or_events) AnyFulfilledFuturePromise.new_blocked_by(futures_and_or_events, default_executor).future end # @!macro promises.shortcut.on # @return [Future] def any_event(*futures_and_or_events) any_event_on default_executor, *futures_and_or_events end # Creates new event which becomes resolved after first of the futures_and_or_events resolves. # @!macro promises.any-touch # # @!macro promises.param.default_executor # @param [AbstractEventFuture] futures_and_or_events # @return [Event] def any_event_on(default_executor, *futures_and_or_events) AnyResolvedEventPromise.new_blocked_by(futures_and_or_events, default_executor).event end # TODO consider adding first(count, *futures) # TODO consider adding zip_by(slice, *futures) processing futures in slices # TODO or rather a generic aggregator taking a function end module InternalStates # @!visibility private class State def resolved? raise NotImplementedError end def to_sym raise NotImplementedError end end # @!visibility private class Pending < State def resolved? false end def to_sym :pending end end # @!visibility private class Reserved < Pending end # @!visibility private class ResolvedWithResult < State def resolved? true end def to_sym :resolved end def result [fulfilled?, value, reason] end def fulfilled? raise NotImplementedError end def value raise NotImplementedError end def reason raise NotImplementedError end def apply raise NotImplementedError end end # @!visibility private class Fulfilled < ResolvedWithResult def initialize(value) @Value = value end def fulfilled? true end def apply(args, block) block.call value, *args end def value @Value end def reason nil end def to_sym :fulfilled end end # @!visibility private class FulfilledArray < Fulfilled def apply(args, block) block.call(*value, *args) end end # @!visibility private class Rejected < ResolvedWithResult def initialize(reason) @Reason = reason end def fulfilled? false end def value nil end def reason @Reason end def to_sym :rejected end def apply(args, block) block.call reason, *args end end # @!visibility private class PartiallyRejected < ResolvedWithResult def initialize(value, reason) super() @Value = value @Reason = reason end def fulfilled? false end def to_sym :rejected end def value @Value end def reason @Reason end def apply(args, block) block.call(*reason, *args) end end # @!visibility private PENDING = Pending.new # @!visibility private RESERVED = Reserved.new # @!visibility private RESOLVED = Fulfilled.new(nil) def RESOLVED.to_sym :resolved end end private_constant :InternalStates # @!macro promises.shortcut.event-future # @see Event#$0 # @see Future#$0 # @!macro promises.param.timeout # @param [Numeric] timeout the maximum time in second to wait. # @!macro promises.warn.blocks # @note This function potentially blocks current thread until the Future is resolved. # Be careful it can deadlock. Try to chain instead. # Common ancestor of {Event} and {Future} classes, many shared methods are defined here. class AbstractEventFuture < Synchronization::Object safe_initialization! attr_atomic(:internal_state) private :internal_state=, :swap_internal_state, :compare_and_set_internal_state, :update_internal_state # @!method internal_state # @!visibility private include InternalStates def initialize(promise, default_executor) super() @Lock = Mutex.new @Condition = ConditionVariable.new @Promise = promise @DefaultExecutor = default_executor @Callbacks = LockFreeStack.new @Waiters = AtomicFixnum.new 0 self.internal_state = PENDING end private :initialize # Returns its state. # @return [Symbol] # # @overload an_event.state # @return [:pending, :resolved] # @overload a_future.state # Both :fulfilled, :rejected implies :resolved. # @return [:pending, :fulfilled, :rejected] def state internal_state.to_sym end # Is it in pending state? # @return [Boolean] def pending? !internal_state.resolved? end # Is it in resolved state? # @return [Boolean] def resolved? internal_state.resolved? end # Propagates touch. Requests all the delayed futures, which it depends on, to be # executed. This method is called by any other method requiring resolved state, like {#wait}. # @return [self] def touch @Promise.touch self end # @!macro promises.touches # Calls {Concurrent::AbstractEventFuture#touch}. # @!macro promises.method.wait # Wait (block the Thread) until receiver is {#resolved?}. # @!macro promises.touches # # @!macro promises.warn.blocks # @!macro promises.param.timeout # @return [self, true, false] self implies timeout was not used, true implies timeout was used # and it was resolved, false implies it was not resolved within timeout. def wait(timeout = nil) result = wait_until_resolved(timeout) timeout ? result : self end # Returns default executor. # @return [Executor] default executor # @see #with_default_executor # @see FactoryMethods#future_on # @see FactoryMethods#resolvable_future # @see FactoryMethods#any_fulfilled_future_on # @see similar def default_executor @DefaultExecutor end # @!macro promises.shortcut.on # @return [Future] def chain(*args, &task) chain_on @DefaultExecutor, *args, &task end # Chains the task to be executed asynchronously on executor after it is resolved. # # @!macro promises.param.executor # @!macro promises.param.args # @return [Future] # @!macro promise.param.task-future # # @overload an_event.chain_on(executor, *args, &task) # @yield [*args] to the task. # @overload a_future.chain_on(executor, *args, &task) # @yield [fulfilled, value, reason, *args] to the task. # @yieldparam [true, false] fulfilled # @yieldparam [Object] value # @yieldparam [Object] reason def chain_on(executor, *args, &task) ChainPromise.new_blocked_by1(self, @DefaultExecutor, executor, args, &task).future end # @return [String] Short string representation. def to_s format '%s %s>', super[0..-2], state end alias_method :inspect, :to_s # Resolves the resolvable when receiver is resolved. # # @param [Resolvable] resolvable # @return [self] def chain_resolvable(resolvable) on_resolution! { resolvable.resolve_with internal_state } end alias_method :tangle, :chain_resolvable # @!macro promises.shortcut.using # @return [self] def on_resolution(*args, &callback) on_resolution_using @DefaultExecutor, *args, &callback end # Stores the callback to be executed synchronously on resolving thread after it is # resolved. # # @!macro promises.param.args # @!macro promise.param.callback # @return [self] # # @overload an_event.on_resolution!(*args, &callback) # @yield [*args] to the callback. # @overload a_future.on_resolution!(*args, &callback) # @yield [fulfilled, value, reason, *args] to the callback. # @yieldparam [true, false] fulfilled # @yieldparam [Object] value # @yieldparam [Object] reason def on_resolution!(*args, &callback) add_callback :callback_on_resolution, args, callback end # Stores the callback to be executed asynchronously on executor after it is resolved. # # @!macro promises.param.executor # @!macro promises.param.args # @!macro promise.param.callback # @return [self] # # @overload an_event.on_resolution_using(executor, *args, &callback) # @yield [*args] to the callback. # @overload a_future.on_resolution_using(executor, *args, &callback) # @yield [fulfilled, value, reason, *args] to the callback. # @yieldparam [true, false] fulfilled # @yieldparam [Object] value # @yieldparam [Object] reason def on_resolution_using(executor, *args, &callback) add_callback :async_callback_on_resolution, executor, args, callback end # @!macro promises.method.with_default_executor # Crates new object with same class with the executor set as its new default executor. # Any futures depending on it will use the new default executor. # @!macro promises.shortcut.event-future # @abstract # @return [AbstractEventFuture] def with_default_executor(executor) raise NotImplementedError end # @!visibility private def resolve_with(state, raise_on_reassign = true, reserved = false) if compare_and_set_internal_state(reserved ? RESERVED : PENDING, state) # go to synchronized block only if there were waiting threads @Lock.synchronize { @Condition.broadcast } unless @Waiters.value == 0 call_callbacks state else return rejected_resolution(raise_on_reassign, state) end self end # For inspection. # @!visibility private # @return [Array] def blocks @Callbacks.each_with_object([]) do |(method, args), promises| promises.push(args[0]) if method == :callback_notify_blocked end end # For inspection. # @!visibility private def callbacks @Callbacks.each.to_a end # For inspection. # @!visibility private def promise @Promise end # For inspection. # @!visibility private def touched? promise.touched? end # For inspection. # @!visibility private def waiting_threads @Waiters.each.to_a end # @!visibility private def add_callback_notify_blocked(promise, index) add_callback :callback_notify_blocked, promise, index end # @!visibility private def add_callback_clear_delayed_node(node) add_callback(:callback_clear_delayed_node, node) end # @!visibility private def with_hidden_resolvable # TODO (pitr-ch 10-Dec-2018): documentation, better name if in edge self end private def add_callback(method, *args) state = internal_state if state.resolved? call_callback method, state, args else @Callbacks.push [method, args] state = internal_state # take back if it was resolved in the meanwhile call_callbacks state if state.resolved? end self end def callback_clear_delayed_node(state, node) node.value = nil end # @return [Boolean] def wait_until_resolved(timeout) return true if resolved? touch @Lock.synchronize do @Waiters.increment begin unless resolved? @Condition.wait @Lock, timeout end ensure # JRuby may raise ConcurrencyError @Waiters.decrement end end resolved? end def call_callback(method, state, args) self.send method, state, *args end def call_callbacks(state) method, args = @Callbacks.pop while method call_callback method, state, args method, args = @Callbacks.pop end end def with_async(executor, *args, &block) Concurrent.executor(executor).post(*args, &block) end def async_callback_on_resolution(state, executor, args, callback) with_async(executor, state, args, callback) do |st, ar, cb| callback_on_resolution st, ar, cb end end def callback_notify_blocked(state, promise, index) promise.on_blocker_resolution self, index end end # Represents an event which will happen in future (will be resolved). The event is either # pending or resolved. It should be always resolved. Use {Future} to communicate rejections and # cancellation. class Event < AbstractEventFuture alias_method :then, :chain # @!macro promises.method.zip # Creates a new event or a future which will be resolved when receiver and other are. # Returns an event if receiver and other are events, otherwise returns a future. # If just one of the parties is Future then the result # of the returned future is equal to the result of the supplied future. If both are futures # then the result is as described in {FactoryMethods#zip_futures_on}. # # @return [Future, Event] def zip(other) if other.is_a?(Future) ZipFutureEventPromise.new_blocked_by2(other, self, @DefaultExecutor).future else ZipEventEventPromise.new_blocked_by2(self, other, @DefaultExecutor).event end end alias_method :&, :zip # Creates a new event which will be resolved when the first of receiver, `event_or_future` # resolves. # # @return [Event] def any(event_or_future) AnyResolvedEventPromise.new_blocked_by2(self, event_or_future, @DefaultExecutor).event end alias_method :|, :any # Creates new event dependent on receiver which will not evaluate until touched, see {#touch}. # In other words, it inserts delay into the chain of Futures making rest of it lazy evaluated. # # @return [Event] def delay event = DelayPromise.new(@DefaultExecutor).event ZipEventEventPromise.new_blocked_by2(self, event, @DefaultExecutor).event end # @!macro promise.method.schedule # Creates new event dependent on receiver scheduled to execute on/in intended_time. # In time is interpreted from the moment the receiver is resolved, therefore it inserts # delay into the chain. # # @!macro promises.param.intended_time # @return [Event] def schedule(intended_time) chain do event = ScheduledPromise.new(@DefaultExecutor, intended_time).event ZipEventEventPromise.new_blocked_by2(self, event, @DefaultExecutor).event end.flat_event end # Converts event to a future. The future is fulfilled when the event is resolved, the future may never fail. # # @return [Future] def to_future future = Promises.resolvable_future ensure chain_resolvable(future) end # Returns self, since this is event # @return [Event] def to_event self end # @!macro promises.method.with_default_executor # @return [Event] def with_default_executor(executor) EventWrapperPromise.new_blocked_by1(self, executor).event end private def rejected_resolution(raise_on_reassign, state) Concurrent::MultipleAssignmentError.new('Event can be resolved only once') if raise_on_reassign return false end def callback_on_resolution(state, args, callback) callback.call(*args) end end # Represents a value which will become available in future. May reject with a reason instead, # e.g. when the tasks raises an exception. class Future < AbstractEventFuture # Is it in fulfilled state? # @return [Boolean] def fulfilled? state = internal_state state.resolved? && state.fulfilled? end # Is it in rejected state? # @return [Boolean] def rejected? state = internal_state state.resolved? && !state.fulfilled? end # @!macro promises.warn.nil # @note Make sure returned `nil` is not confused with timeout, no value when rejected, # no reason when fulfilled, etc. # Use more exact methods if needed, like {#wait}, {#value!}, {#result}, etc. # @!macro promises.method.value # Return value of the future. # @!macro promises.touches # # @!macro promises.warn.blocks # @!macro promises.warn.nil # @!macro promises.param.timeout # @!macro promises.param.timeout_value # @param [Object] timeout_value a value returned by the method when it times out # @return [Object, nil, timeout_value] the value of the Future when fulfilled, # timeout_value on timeout, # nil on rejection. def value(timeout = nil, timeout_value = nil) if wait_until_resolved timeout internal_state.value else timeout_value end end # Returns reason of future's rejection. # @!macro promises.touches # # @!macro promises.warn.blocks # @!macro promises.warn.nil # @!macro promises.param.timeout # @!macro promises.param.timeout_value # @return [Object, timeout_value] the reason, or timeout_value on timeout, or nil on fulfillment. def reason(timeout = nil, timeout_value = nil) if wait_until_resolved timeout internal_state.reason else timeout_value end end # Returns triplet fulfilled?, value, reason. # @!macro promises.touches # # @!macro promises.warn.blocks # @!macro promises.param.timeout # @return [Array(Boolean, Object, Object), nil] triplet of fulfilled?, value, reason, or nil # on timeout. def result(timeout = nil) internal_state.result if wait_until_resolved timeout end # @!macro promises.method.wait # @raise [Exception] {#reason} on rejection def wait!(timeout = nil) result = wait_until_resolved!(timeout) timeout ? result : self end # @!macro promises.method.value # @return [Object, nil, timeout_value] the value of the Future when fulfilled, # or nil on rejection, # or timeout_value on timeout. # @raise [Exception] {#reason} on rejection def value!(timeout = nil, timeout_value = nil) if wait_until_resolved! timeout internal_state.value else timeout_value end end # Allows rejected Future to be risen with `raise` method. # If the reason is not an exception `Runtime.new(reason)` is returned. # # @example # raise Promises.rejected_future(StandardError.new("boom")) # raise Promises.rejected_future("or just boom") # @raise [Concurrent::Error] when raising not rejected future # @return [Exception] def exception(*args) raise Concurrent::Error, 'it is not rejected' unless rejected? raise ArgumentError unless args.size <= 1 reason = Array(internal_state.reason).flatten.compact if reason.size > 1 ex = Concurrent::MultipleErrors.new reason ex.set_backtrace(caller) ex else ex = if reason[0].respond_to? :exception reason[0].exception(*args) else RuntimeError.new(reason[0]).exception(*args) end ex.set_backtrace Array(ex.backtrace) + caller ex end end # @!macro promises.shortcut.on # @return [Future] def then(*args, &task) then_on @DefaultExecutor, *args, &task end # Chains the task to be executed asynchronously on executor after it fulfills. Does not run # the task if it rejects. It will resolve though, triggering any dependent futures. # # @!macro promises.param.executor # @!macro promises.param.args # @!macro promise.param.task-future # @return [Future] # @yield [value, *args] to the task. def then_on(executor, *args, &task) ThenPromise.new_blocked_by1(self, @DefaultExecutor, executor, args, &task).future end # @!macro promises.shortcut.on # @return [Future] def rescue(*args, &task) rescue_on @DefaultExecutor, *args, &task end # Chains the task to be executed asynchronously on executor after it rejects. Does not run # the task if it fulfills. It will resolve though, triggering any dependent futures. # # @!macro promises.param.executor # @!macro promises.param.args # @!macro promise.param.task-future # @return [Future] # @yield [reason, *args] to the task. def rescue_on(executor, *args, &task) RescuePromise.new_blocked_by1(self, @DefaultExecutor, executor, args, &task).future end # @!macro promises.method.zip # @return [Future] def zip(other) if other.is_a?(Future) ZipFuturesPromise.new_blocked_by2(self, other, @DefaultExecutor).future else ZipFutureEventPromise.new_blocked_by2(self, other, @DefaultExecutor).future end end alias_method :&, :zip # Creates a new event which will be resolved when the first of receiver, `event_or_future` # resolves. Returning future will have value nil if event_or_future is event and resolves # first. # # @return [Future] def any(event_or_future) AnyResolvedFuturePromise.new_blocked_by2(self, event_or_future, @DefaultExecutor).future end alias_method :|, :any # Creates new future dependent on receiver which will not evaluate until touched, see {#touch}. # In other words, it inserts delay into the chain of Futures making rest of it lazy evaluated. # # @return [Future] def delay event = DelayPromise.new(@DefaultExecutor).event ZipFutureEventPromise.new_blocked_by2(self, event, @DefaultExecutor).future end # @!macro promise.method.schedule # @return [Future] def schedule(intended_time) chain do event = ScheduledPromise.new(@DefaultExecutor, intended_time).event ZipFutureEventPromise.new_blocked_by2(self, event, @DefaultExecutor).future end.flat end # @!macro promises.method.with_default_executor # @return [Future] def with_default_executor(executor) FutureWrapperPromise.new_blocked_by1(self, executor).future end # Creates new future which will have result of the future returned by receiver. If receiver # rejects it will have its rejection. # # @param [Integer] level how many levels of futures should flatten # @return [Future] def flat_future(level = 1) FlatFuturePromise.new_blocked_by1(self, level, @DefaultExecutor).future end alias_method :flat, :flat_future # Creates new event which will be resolved when the returned event by receiver is. # Be careful if the receiver rejects it will just resolve since Event does not hold reason. # # @return [Event] def flat_event FlatEventPromise.new_blocked_by1(self, @DefaultExecutor).event end # @!macro promises.shortcut.using # @return [self] def on_fulfillment(*args, &callback) on_fulfillment_using @DefaultExecutor, *args, &callback end # Stores the callback to be executed synchronously on resolving thread after it is # fulfilled. Does nothing on rejection. # # @!macro promises.param.args # @!macro promise.param.callback # @return [self] # @yield [value, *args] to the callback. def on_fulfillment!(*args, &callback) add_callback :callback_on_fulfillment, args, callback end # Stores the callback to be executed asynchronously on executor after it is # fulfilled. Does nothing on rejection. # # @!macro promises.param.executor # @!macro promises.param.args # @!macro promise.param.callback # @return [self] # @yield [value, *args] to the callback. def on_fulfillment_using(executor, *args, &callback) add_callback :async_callback_on_fulfillment, executor, args, callback end # @!macro promises.shortcut.using # @return [self] def on_rejection(*args, &callback) on_rejection_using @DefaultExecutor, *args, &callback end # Stores the callback to be executed synchronously on resolving thread after it is # rejected. Does nothing on fulfillment. # # @!macro promises.param.args # @!macro promise.param.callback # @return [self] # @yield [reason, *args] to the callback. def on_rejection!(*args, &callback) add_callback :callback_on_rejection, args, callback end # Stores the callback to be executed asynchronously on executor after it is # rejected. Does nothing on fulfillment. # # @!macro promises.param.executor # @!macro promises.param.args # @!macro promise.param.callback # @return [self] # @yield [reason, *args] to the callback. def on_rejection_using(executor, *args, &callback) add_callback :async_callback_on_rejection, executor, args, callback end # Allows to use futures as green threads. The receiver has to evaluate to a future which # represents what should be done next. It basically flattens indefinitely until non Future # values is returned which becomes result of the returned future. Any encountered exception # will become reason of the returned future. # # @return [Future] # @param [#call(value)] run_test # an object which when called returns either Future to keep running with # or nil, then the run completes with the value. # The run_test can be used to extract the Future from deeper structure, # or to distinguish Future which is a resulting value from a future # which is suppose to continue running. # @example # body = lambda do |v| # v += 1 # v < 5 ? Promises.future(v, &body) : v # end # Promises.future(0, &body).run.value! # => 5 def run(run_test = method(:run_test)) RunFuturePromise.new_blocked_by1(self, @DefaultExecutor, run_test).future end # @!visibility private def apply(args, block) internal_state.apply args, block end # Converts future to event which is resolved when future is resolved by fulfillment or rejection. # # @return [Event] def to_event event = Promises.resolvable_event ensure chain_resolvable(event) end # Returns self, since this is a future # @return [Future] def to_future self end # @return [String] Short string representation. def to_s if resolved? format '%s with %s>', super[0..-2], (fulfilled? ? value : reason).inspect else super end end alias_method :inspect, :to_s private def run_test(v) v if v.is_a?(Future) end def rejected_resolution(raise_on_reassign, state) if raise_on_reassign if internal_state == RESERVED raise Concurrent::MultipleAssignmentError.new( "Future can be resolved only once. It is already reserved.") else raise Concurrent::MultipleAssignmentError.new( "Future can be resolved only once. It's #{result}, trying to set #{state.result}.", current_result: result, new_result: state.result) end end return false end def wait_until_resolved!(timeout = nil) result = wait_until_resolved(timeout) raise self if rejected? result end def async_callback_on_fulfillment(state, executor, args, callback) with_async(executor, state, args, callback) do |st, ar, cb| callback_on_fulfillment st, ar, cb end end def async_callback_on_rejection(state, executor, args, callback) with_async(executor, state, args, callback) do |st, ar, cb| callback_on_rejection st, ar, cb end end def callback_on_fulfillment(state, args, callback) state.apply args, callback if state.fulfilled? end def callback_on_rejection(state, args, callback) state.apply args, callback unless state.fulfilled? end def callback_on_resolution(state, args, callback) callback.call(*state.result, *args) end end # Marker module of Future, Event resolved manually. module Resolvable include InternalStates end # A Event which can be resolved by user. class ResolvableEvent < Event include Resolvable # @!macro raise_on_reassign # @raise [MultipleAssignmentError] when already resolved and raise_on_reassign is true. # @!macro promise.param.raise_on_reassign # @param [Boolean] raise_on_reassign should method raise exception if already resolved # @return [self, false] false is returner when raise_on_reassign is false and the receiver # is already resolved. # # Makes the event resolved, which triggers all dependent futures. # # @!macro promise.param.raise_on_reassign # @!macro promise.param.reserved # @param [true, false] reserved # Set to true if the resolvable is {#reserve}d by you, # marks resolution of reserved resolvable events and futures explicitly. # Advanced feature, ignore unless you use {Resolvable#reserve} from edge. def resolve(raise_on_reassign = true, reserved = false) resolve_with RESOLVED, raise_on_reassign, reserved end # Creates new event wrapping receiver, effectively hiding the resolve method. # # @return [Event] def with_hidden_resolvable @with_hidden_resolvable ||= EventWrapperPromise.new_blocked_by1(self, @DefaultExecutor).event end # Behaves as {AbstractEventFuture#wait} but has one additional optional argument # resolve_on_timeout. # # @param [true, false] resolve_on_timeout # If it times out and the argument is true it will also resolve the event. # @return [self, true, false] # @see AbstractEventFuture#wait def wait(timeout = nil, resolve_on_timeout = false) super(timeout) or if resolve_on_timeout # if it fails to resolve it was resolved in the meantime # so return true as if there was no timeout !resolve(false) else false end end end # A Future which can be resolved by user. class ResolvableFuture < Future include Resolvable # Makes the future resolved with result of triplet `fulfilled?`, `value`, `reason`, # which triggers all dependent futures. # # @param [true, false] fulfilled # @param [Object] value # @param [Object] reason # @!macro promise.param.raise_on_reassign # @!macro promise.param.reserved def resolve(fulfilled = true, value = nil, reason = nil, raise_on_reassign = true, reserved = false) resolve_with(fulfilled ? Fulfilled.new(value) : Rejected.new(reason), raise_on_reassign, reserved) end # Makes the future fulfilled with `value`, # which triggers all dependent futures. # # @param [Object] value # @!macro promise.param.raise_on_reassign # @!macro promise.param.reserved def fulfill(value, raise_on_reassign = true, reserved = false) resolve_with Fulfilled.new(value), raise_on_reassign, reserved end # Makes the future rejected with `reason`, # which triggers all dependent futures. # # @param [Object] reason # @!macro promise.param.raise_on_reassign # @!macro promise.param.reserved def reject(reason, raise_on_reassign = true, reserved = false) resolve_with Rejected.new(reason), raise_on_reassign, reserved end # Evaluates the block and sets its result as future's value fulfilling, if the block raises # an exception the future rejects with it. # # @yield [*args] to the block. # @yieldreturn [Object] value # @return [self] def evaluate_to(*args, &block) promise.evaluate_to(*args, block) end # Evaluates the block and sets its result as future's value fulfilling, if the block raises # an exception the future rejects with it. # # @yield [*args] to the block. # @yieldreturn [Object] value # @return [self] # @raise [Exception] also raise reason on rejection. def evaluate_to!(*args, &block) promise.evaluate_to(*args, block).wait! end # @!macro promises.resolvable.resolve_on_timeout # @param [::Array(true, Object, nil), ::Array(false, nil, Exception), nil] resolve_on_timeout # If it times out and the argument is not nil it will also resolve the future # to the provided resolution. # Behaves as {AbstractEventFuture#wait} but has one additional optional argument # resolve_on_timeout. # # @!macro promises.resolvable.resolve_on_timeout # @return [self, true, false] # @see AbstractEventFuture#wait def wait(timeout = nil, resolve_on_timeout = nil) super(timeout) or if resolve_on_timeout # if it fails to resolve it was resolved in the meantime # so return true as if there was no timeout !resolve(*resolve_on_timeout, false) else false end end # Behaves as {Future#wait!} but has one additional optional argument # resolve_on_timeout. # # @!macro promises.resolvable.resolve_on_timeout # @return [self, true, false] # @raise [Exception] {#reason} on rejection # @see Future#wait! def wait!(timeout = nil, resolve_on_timeout = nil) super(timeout) or if resolve_on_timeout if resolve(*resolve_on_timeout, false) false else # if it fails to resolve it was resolved in the meantime # so return true as if there was no timeout raise self if rejected? true end else false end end # Behaves as {Future#value} but has one additional optional argument # resolve_on_timeout. # # @!macro promises.resolvable.resolve_on_timeout # @return [Object, timeout_value, nil] # @see Future#value def value(timeout = nil, timeout_value = nil, resolve_on_timeout = nil) if wait_until_resolved timeout internal_state.value else if resolve_on_timeout unless resolve(*resolve_on_timeout, false) # if it fails to resolve it was resolved in the meantime # so return value as if there was no timeout return internal_state.value end end timeout_value end end # Behaves as {Future#value!} but has one additional optional argument # resolve_on_timeout. # # @!macro promises.resolvable.resolve_on_timeout # @return [Object, timeout_value, nil] # @raise [Exception] {#reason} on rejection # @see Future#value! def value!(timeout = nil, timeout_value = nil, resolve_on_timeout = nil) if wait_until_resolved! timeout internal_state.value else if resolve_on_timeout unless resolve(*resolve_on_timeout, false) # if it fails to resolve it was resolved in the meantime # so return value as if there was no timeout raise self if rejected? return internal_state.value end end timeout_value end end # Behaves as {Future#reason} but has one additional optional argument # resolve_on_timeout. # # @!macro promises.resolvable.resolve_on_timeout # @return [Exception, timeout_value, nil] # @see Future#reason def reason(timeout = nil, timeout_value = nil, resolve_on_timeout = nil) if wait_until_resolved timeout internal_state.reason else if resolve_on_timeout unless resolve(*resolve_on_timeout, false) # if it fails to resolve it was resolved in the meantime # so return value as if there was no timeout return internal_state.reason end end timeout_value end end # Behaves as {Future#result} but has one additional optional argument # resolve_on_timeout. # # @!macro promises.resolvable.resolve_on_timeout # @return [::Array(Boolean, Object, Exception), nil] # @see Future#result def result(timeout = nil, resolve_on_timeout = nil) if wait_until_resolved timeout internal_state.result else if resolve_on_timeout unless resolve(*resolve_on_timeout, false) # if it fails to resolve it was resolved in the meantime # so return value as if there was no timeout internal_state.result end end # otherwise returns nil end end # Creates new future wrapping receiver, effectively hiding the resolve method and similar. # # @return [Future] def with_hidden_resolvable @with_hidden_resolvable ||= FutureWrapperPromise.new_blocked_by1(self, @DefaultExecutor).future end end # @abstract # @private class AbstractPromise < Synchronization::Object safe_initialization! include InternalStates def initialize(future) super() @Future = future end def future @Future end alias_method :event, :future def default_executor future.default_executor end def state future.state end def touch end def to_s format '%s %s>', super[0..-2], @Future end alias_method :inspect, :to_s def delayed_because nil end private def resolve_with(new_state, raise_on_reassign = true) @Future.resolve_with(new_state, raise_on_reassign) end # @return [Future] def evaluate_to(*args, block) resolve_with Fulfilled.new(block.call(*args)) rescue Exception => error resolve_with Rejected.new(error) raise error unless error.is_a?(StandardError) end end class ResolvableEventPromise < AbstractPromise def initialize(default_executor) super ResolvableEvent.new(self, default_executor) end end class ResolvableFuturePromise < AbstractPromise def initialize(default_executor) super ResolvableFuture.new(self, default_executor) end public :evaluate_to end # @abstract class InnerPromise < AbstractPromise end # @abstract class BlockedPromise < InnerPromise private_class_method :new def self.new_blocked_by1(blocker, *args, &block) blocker_delayed = blocker.promise.delayed_because promise = new(blocker_delayed, 1, *args, &block) blocker.add_callback_notify_blocked promise, 0 promise end def self.new_blocked_by2(blocker1, blocker2, *args, &block) blocker_delayed1 = blocker1.promise.delayed_because blocker_delayed2 = blocker2.promise.delayed_because delayed = if blocker_delayed1 && blocker_delayed2 # TODO (pitr-ch 23-Dec-2016): use arrays when we know it will not grow (only flat adds delay) LockFreeStack.of2(blocker_delayed1, blocker_delayed2) else blocker_delayed1 || blocker_delayed2 end promise = new(delayed, 2, *args, &block) blocker1.add_callback_notify_blocked promise, 0 blocker2.add_callback_notify_blocked promise, 1 promise end def self.new_blocked_by(blockers, *args, &block) delayed = blockers.reduce(nil) { |d, f| add_delayed d, f.promise.delayed_because } promise = new(delayed, blockers.size, *args, &block) blockers.each_with_index { |f, i| f.add_callback_notify_blocked promise, i } promise end def self.add_delayed(delayed1, delayed2) if delayed1 && delayed2 delayed1.push delayed2 delayed1 else delayed1 || delayed2 end end def initialize(delayed, blockers_count, future) super(future) @Delayed = delayed @Countdown = AtomicFixnum.new blockers_count end def on_blocker_resolution(future, index) countdown = process_on_blocker_resolution(future, index) resolvable = resolvable?(countdown, future, index) on_resolvable(future, index) if resolvable end def delayed_because @Delayed end def touch clear_and_propagate_touch end # for inspection only def blocked_by blocked_by = [] ObjectSpace.each_object(AbstractEventFuture) { |o| blocked_by.push o if o.blocks.include? self } blocked_by end private def clear_and_propagate_touch(stack_or_element = @Delayed) return if stack_or_element.nil? if stack_or_element.is_a? LockFreeStack stack_or_element.clear_each { |element| clear_and_propagate_touch element } else stack_or_element.touch unless stack_or_element.nil? # if still present end end # @return [true,false] if resolvable def resolvable?(countdown, future, index) countdown.zero? end def process_on_blocker_resolution(future, index) @Countdown.decrement end def on_resolvable(resolved_future, index) raise NotImplementedError end end # @abstract class BlockedTaskPromise < BlockedPromise def initialize(delayed, blockers_count, default_executor, executor, args, &task) raise ArgumentError, 'no block given' unless block_given? super delayed, 1, Future.new(self, default_executor) @Executor = executor @Task = task @Args = args end def executor @Executor end end class ThenPromise < BlockedTaskPromise private def initialize(delayed, blockers_count, default_executor, executor, args, &task) super delayed, blockers_count, default_executor, executor, args, &task end def on_resolvable(resolved_future, index) if resolved_future.fulfilled? Concurrent.executor(@Executor).post(resolved_future, @Args, @Task) do |future, args, task| evaluate_to lambda { future.apply args, task } end else resolve_with resolved_future.internal_state end end end class RescuePromise < BlockedTaskPromise private def initialize(delayed, blockers_count, default_executor, executor, args, &task) super delayed, blockers_count, default_executor, executor, args, &task end def on_resolvable(resolved_future, index) if resolved_future.rejected? Concurrent.executor(@Executor).post(resolved_future, @Args, @Task) do |future, args, task| evaluate_to lambda { future.apply args, task } end else resolve_with resolved_future.internal_state end end end class ChainPromise < BlockedTaskPromise private def on_resolvable(resolved_future, index) if Future === resolved_future Concurrent.executor(@Executor).post(resolved_future, @Args, @Task) do |future, args, task| evaluate_to(*future.result, *args, task) end else Concurrent.executor(@Executor).post(@Args, @Task) do |args, task| evaluate_to(*args, task) end end end end # will be immediately resolved class ImmediateEventPromise < InnerPromise def initialize(default_executor) super Event.new(self, default_executor).resolve_with(RESOLVED) end end class ImmediateFuturePromise < InnerPromise def initialize(default_executor, fulfilled, value, reason) super Future.new(self, default_executor). resolve_with(fulfilled ? Fulfilled.new(value) : Rejected.new(reason)) end end class AbstractFlatPromise < BlockedPromise def initialize(delayed_because, blockers_count, event_or_future) delayed = LockFreeStack.of1(self) super(delayed, blockers_count, event_or_future) # noinspection RubyArgCount @Touched = AtomicBoolean.new false @DelayedBecause = delayed_because || LockFreeStack.new event_or_future.add_callback_clear_delayed_node delayed.peek end def touch if @Touched.make_true clear_and_propagate_touch @DelayedBecause end end private def touched? @Touched.value end def on_resolvable(resolved_future, index) resolve_with resolved_future.internal_state end def resolvable?(countdown, future, index) !@Future.internal_state.resolved? && super(countdown, future, index) end def add_delayed_of(future) delayed = future.promise.delayed_because if touched? clear_and_propagate_touch delayed else BlockedPromise.add_delayed @DelayedBecause, delayed clear_and_propagate_touch @DelayedBecause if touched? end end end class FlatEventPromise < AbstractFlatPromise private def initialize(delayed, blockers_count, default_executor) super delayed, 2, Event.new(self, default_executor) end def process_on_blocker_resolution(future, index) countdown = super(future, index) if countdown.nonzero? internal_state = future.internal_state unless internal_state.fulfilled? resolve_with RESOLVED return countdown end value = internal_state.value case value when AbstractEventFuture add_delayed_of value value.add_callback_notify_blocked self, nil countdown else resolve_with RESOLVED end end countdown end end class FlatFuturePromise < AbstractFlatPromise private def initialize(delayed, blockers_count, levels, default_executor) raise ArgumentError, 'levels has to be higher than 0' if levels < 1 # flat promise may result to a future having delayed futures, therefore we have to have empty stack # to be able to add new delayed futures super delayed || LockFreeStack.new, 1 + levels, Future.new(self, default_executor) end def process_on_blocker_resolution(future, index) countdown = super(future, index) if countdown.nonzero? internal_state = future.internal_state unless internal_state.fulfilled? resolve_with internal_state return countdown end value = internal_state.value case value when AbstractEventFuture add_delayed_of value value.add_callback_notify_blocked self, nil countdown else evaluate_to(lambda { raise TypeError, "returned value #{value.inspect} is not a Future" }) end end countdown end end class RunFuturePromise < AbstractFlatPromise private def initialize(delayed, blockers_count, default_executor, run_test) super delayed, 1, Future.new(self, default_executor) @RunTest = run_test end def process_on_blocker_resolution(future, index) internal_state = future.internal_state unless internal_state.fulfilled? resolve_with internal_state return 0 end value = internal_state.value continuation_future = @RunTest.call value if continuation_future add_delayed_of continuation_future continuation_future.add_callback_notify_blocked self, nil else resolve_with internal_state end 1 end end class ZipEventEventPromise < BlockedPromise def initialize(delayed, blockers_count, default_executor) super delayed, 2, Event.new(self, default_executor) end private def on_resolvable(resolved_future, index) resolve_with RESOLVED end end class ZipFutureEventPromise < BlockedPromise def initialize(delayed, blockers_count, default_executor) super delayed, 2, Future.new(self, default_executor) @result = nil end private def process_on_blocker_resolution(future, index) # first blocking is future, take its result @result = future.internal_state if index == 0 # super has to be called after above to piggyback on volatile @Countdown super future, index end def on_resolvable(resolved_future, index) resolve_with @result end end class EventWrapperPromise < BlockedPromise def initialize(delayed, blockers_count, default_executor) super delayed, 1, Event.new(self, default_executor) end private def on_resolvable(resolved_future, index) resolve_with RESOLVED end end class FutureWrapperPromise < BlockedPromise def initialize(delayed, blockers_count, default_executor) super delayed, 1, Future.new(self, default_executor) end private def on_resolvable(resolved_future, index) resolve_with resolved_future.internal_state end end class ZipFuturesPromise < BlockedPromise private def initialize(delayed, blockers_count, default_executor) super(delayed, blockers_count, Future.new(self, default_executor)) @Resolutions = ::Array.new(blockers_count, nil) on_resolvable nil, nil if blockers_count == 0 end def process_on_blocker_resolution(future, index) # TODO (pitr-ch 18-Dec-2016): Can we assume that array will never break under parallel access when never re-sized? @Resolutions[index] = future.internal_state # has to be set before countdown in super super future, index end def on_resolvable(resolved_future, index) all_fulfilled = true values = ::Array.new(@Resolutions.size) reasons = ::Array.new(@Resolutions.size) @Resolutions.each_with_index do |internal_state, i| fulfilled, values[i], reasons[i] = internal_state.result all_fulfilled &&= fulfilled end if all_fulfilled resolve_with FulfilledArray.new(values) else resolve_with PartiallyRejected.new(values, reasons) end end end class ZipEventsPromise < BlockedPromise private def initialize(delayed, blockers_count, default_executor) super delayed, blockers_count, Event.new(self, default_executor) on_resolvable nil, nil if blockers_count == 0 end def on_resolvable(resolved_future, index) resolve_with RESOLVED end end # @abstract class AbstractAnyPromise < BlockedPromise end class AnyResolvedEventPromise < AbstractAnyPromise private def initialize(delayed, blockers_count, default_executor) super delayed, blockers_count, Event.new(self, default_executor) end def resolvable?(countdown, future, index) true end def on_resolvable(resolved_future, index) resolve_with RESOLVED, false end end class AnyResolvedFuturePromise < AbstractAnyPromise private def initialize(delayed, blockers_count, default_executor) super delayed, blockers_count, Future.new(self, default_executor) end def resolvable?(countdown, future, index) true end def on_resolvable(resolved_future, index) resolve_with resolved_future.internal_state, false end end class AnyFulfilledFuturePromise < AnyResolvedFuturePromise private def resolvable?(countdown, future, index) future.fulfilled? || # inlined super from BlockedPromise countdown.zero? end end class DelayPromise < InnerPromise def initialize(default_executor) event = Event.new(self, default_executor) @Delayed = LockFreeStack.of1(self) super event event.add_callback_clear_delayed_node @Delayed.peek end def touch @Future.resolve_with RESOLVED end def delayed_because @Delayed end end class ScheduledPromise < InnerPromise def intended_time @IntendedTime end def inspect "#{to_s[0..-2]} intended_time: #{@IntendedTime}>" end private def initialize(default_executor, intended_time) super Event.new(self, default_executor) @IntendedTime = intended_time in_seconds = begin now = Time.now schedule_time = if @IntendedTime.is_a? Time @IntendedTime else now + @IntendedTime end [0, schedule_time.to_f - now.to_f].max end Concurrent.global_timer_set.post(in_seconds) do @Future.resolve_with RESOLVED end end end extend FactoryMethods private_constant :AbstractPromise, :ResolvableEventPromise, :ResolvableFuturePromise, :InnerPromise, :BlockedPromise, :BlockedTaskPromise, :ThenPromise, :RescuePromise, :ChainPromise, :ImmediateEventPromise, :ImmediateFuturePromise, :AbstractFlatPromise, :FlatFuturePromise, :FlatEventPromise, :RunFuturePromise, :ZipEventEventPromise, :ZipFutureEventPromise, :EventWrapperPromise, :FutureWrapperPromise, :ZipFuturesPromise, :ZipEventsPromise, :AbstractAnyPromise, :AnyResolvedFuturePromise, :AnyFulfilledFuturePromise, :AnyResolvedEventPromise, :DelayPromise, :ScheduledPromise end end