lib/async/task.rb in async-1.32.1 vs lib/async/task.rb in async-2.0.0
- old
+ new
@@ -19,11 +19,10 @@
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
require 'fiber'
-require 'forwardable'
require_relative 'node'
require_relative 'condition'
module Async
@@ -36,61 +35,49 @@
def alive?
true
end
- def resume
+ def transfer
@task.stop
end
end
end
- # A task represents the state associated with the execution of an asynchronous
- # block.
+ # Raised if a timeout occurs on a specific Fiber. Handled gracefully by `Task`.
+ # @public Since `stable-v1`.
+ class TimeoutError < StandardError
+ def initialize(message = "execution expired")
+ super
+ end
+ end
+
+ # Encapsulates the state of a running task and it's result.
+ # @public Since `stable-v1`.
class Task < Node
- extend Forwardable
-
- # Yield the unerlying `result` for the task. If the result
- # is an Exception, then that result will be raised an its
- # exception.
- # @return [Object] result of the task
- # @raise [Exception] if the result is an exception
- # @yield [result] result of the task if a block if given.
+ # @deprecated With no replacement.
def self.yield
- if block_given?
- result = yield
- else
- result = Fiber.yield
- end
-
- if result.is_a? Exception
- raise result
- else
- return result
- end
+ Fiber.scheduler.transfer
end
# Create a new task.
- # @param reactor [Async::Reactor] the reactor this task will run within.
- # @param parent [Async::Task] the parent task.
- def initialize(reactor, parent = Task.current?, logger: nil, finished: nil, **options, &block)
- super(parent || reactor, **options)
+ # @parameter reactor [Reactor] the reactor this task will run within.
+ # @parameter parent [Task] the parent task.
+ def initialize(parent = Task.current?, finished: nil, **options, &block)
+ super(parent, **options)
- @reactor = reactor
-
@status = :initialized
@result = nil
@finished = finished
- @logger = logger || @parent.logger
-
- @fiber = make_fiber(&block)
-
- @defer_stop = nil
+ @block = block
+ @fiber = nil
end
- attr :logger
+ def reactor
+ self.root
+ end
if Fiber.current.respond_to?(:backtrace)
def backtrace(*arguments)
@fiber&.backtrace(*arguments)
end
@@ -98,18 +85,23 @@
def to_s
"\#<#{self.description} (#{@status})>"
end
- # @attr ios [Reactor] The reactor the task was created within.
- attr :reactor
+ # @deprecated Prefer {Kernel#sleep} except when compatibility with `stable-v1` is required.
+ def sleep(duration = nil)
+ super
+ end
- def_delegators :@reactor, :with_timeout, :sleep
+ # @deprecated Replaced by {Scheduler#timeout_after}.
+ def with_timeout(timeout, exception = TimeoutError, message = "execution expired", &block)
+ Fiber.scheduler.timeout_after(timeout, exception, message, &block)
+ end
# Yield back to the reactor and allow other fibers to execute.
def yield
- Task.yield{reactor.yield}
+ Fiber.scheduler.yield
end
# @attr fiber [Fiber] The fiber which is being used for the execution of this task.
attr :fiber
@@ -123,136 +115,98 @@
# Begin the execution of the task.
def run(*arguments)
if @status == :initialized
@status = :running
- @fiber.resume(*arguments)
+ schedule(arguments)
else
raise RuntimeError, "Task already running!"
end
end
def async(*arguments, **options, &block)
- task = Task.new(@reactor, self, **options, &block)
+ task = Task.new(self, **options, &block)
task.run(*arguments)
return task
end
# Retrieve the current result of the task. Will cause the caller to wait until result is available.
- # @raise [RuntimeError] if the task's fiber is the current fiber.
- # @return [Object] the final expression/result of the task's block.
+ # @raises[RuntimeError] If the task's fiber is the current fiber.
+ # @returns [Object] The final expression/result of the task's block.
def wait
- raise RuntimeError, "Cannot wait on own fiber" if Fiber.current.equal?(@fiber)
+ raise "Cannot wait on own fiber" if Fiber.current.equal?(@fiber)
if running?
@finished ||= Condition.new
@finished.wait
+ end
+
+ case @result
+ when Exception
+ raise @result
else
- Task.yield{@result}
+ return @result
end
end
- # Deprecated.
- alias result wait
- # Soon to become attr :result
+ # Access the result of the task without waiting. May be nil if the task is not completed.
+ attr :result
# Stop the task and all of its children.
def stop(later = false)
if self.stopped?
# If we already stopped this task... don't try to stop it again:
return
end
- # If we are deferring stop...
- if @defer_stop == false
- # Don't stop now... but update the state so we know we need to stop later.
- @defer_stop = true
- return false
- end
-
if self.running?
if self.current?
if later
- @reactor << Stop::Later.new(self)
+ Fiber.scheduler.push Stop::Later.new(self)
else
raise Stop, "Stopping current task!"
end
elsif @fiber&.alive?
begin
- @fiber.resume(Stop.new)
+ Fiber.scheduler.raise(@fiber, Stop)
rescue FiberError
- @reactor << Stop::Later.new(self)
+ Fiber.scheduler.push Stop::Later.new(self)
end
end
else
# We are not running, but children might be, so transition directly into stopped state:
stop!
end
end
- # Defer the handling of stop. During the execution of the given block, if a stop is requested, it will be deferred until the block exits. This is useful for ensuring graceful shutdown of servers and other long-running tasks. You should wrap the response handling code in a defer_stop block to ensure that the task is stopped when the response is complete but not before.
- #
- # You can nest calls to defer_stop, but the stop will only be deferred until the outermost block exits.
- #
- # If stop is invoked a second time, it will be immediately executed.
- #
- # @yields {} The block of code to execute.
- def defer_stop
- # Tri-state variable for controlling stop:
- # - nil: defer_stop has not been called.
- # - false: defer_stop has been called and we are not stopping.
- # - true: defer_stop has been called and we will stop when exiting the block.
- if @defer_stop.nil?
- # If we are not deferring stop already, we can defer it now:
- @defer_stop = false
-
- begin
- yield
- rescue Stop
- # If we are exiting due to a stop, we shouldn't try to invoke stop again:
- @defer_stop = nil
- raise
- ensure
- # If we were asked to stop, we should do so now:
- if @defer_stop
- @defer_stop = nil
- self.stop
- end
- end
- else
- # If we are deferring stop already, entering it again is a no-op.
- yield
- end
- end
-
# Lookup the {Task} for the current fiber. Raise `RuntimeError` if none is available.
- # @return [Async::Task]
- # @raise [RuntimeError] if task was not {set!} for the current fiber.
+ # @returns [Task]
+ # @raises[RuntimeError] If task was not {set!} for the current fiber.
def self.current
Thread.current[:async_task] or raise RuntimeError, "No async task available!"
end
# Check if there is a task defined for the current fiber.
- # @return [Async::Task, nil]
+ # @returns [Task | Nil]
def self.current?
Thread.current[:async_task]
end
def current?
self.equal?(Thread.current[:async_task])
end
# Check if the task is running.
- # @return [Boolean]
+ # @returns [Boolean]
def running?
@status == :running
end
# Whether we can remove this node from the reactor graph.
- # @return [Boolean]
+ # @returns [Boolean]
def finished?
super && @status != :running
end
def failed?
@@ -272,52 +226,53 @@
end
private
# This is a very tricky aspect of tasks to get right. I've modelled it after `Thread` but it's slightly different in that the exception can propagate back up through the reactor. If the user writes code which raises an exception, that exception should always be visible, i.e. cause a failure. If it's not visible, such code fails silently and can be very difficult to debug.
- def fail!(exception = false, propagate = true)
+ # As an explcit choice, the user can start a task which doesn't propagate exceptions. This only applies to `StandardError` and derived tasks. This allows tasks to internally capture their error state which is raised when invoking `Task#result` similar to how `Thread#join` works. This mode makes {ruby Async::Task} behave more like a promise, and you would need to ensure that someone calls `Task#result` otherwise you might miss important errors.
+ def fail!(exception = nil, propagate = true)
@status = :failed
@result = exception
- if exception
- if propagate
- raise exception
- elsif @finished.nil?
- # If no one has called wait, we log this as a warning:
- Console.logger.warn(self, "Task may have ended with unhandled exception.", exception)
- else
- Console.logger.debug(self, exception)
- end
+ if propagate
+ raise
+ elsif @finished.nil?
+ # If no one has called wait, we log this as an error:
+ Console.logger.error(self) {$!}
+ else
+ Console.logger.debug(self) {$!}
end
end
def stop!
- # logger.debug(self) {"Task was stopped with #{@children&.size.inspect} children!"}
+ # Console.logger.info(self, self.annotation) {"Task was stopped with #{@children&.size.inspect} children!"}
@status = :stopped
stop_children(true)
end
- def make_fiber(&block)
- Fiber.new do |*arguments|
+ def schedule(arguments)
+ @fiber = Fiber.new do
set!
begin
- @result = yield(self, *arguments)
+ @result = @block.call(self, *arguments)
@status = :complete
# Console.logger.debug(self) {"Task was completed with #{@children.size} children!"}
rescue Stop
stop!
rescue StandardError => error
fail!(error, false)
rescue Exception => exception
fail!(exception, true)
ensure
- # Console.logger.debug(self) {"Task ensure $!=#{$!} with #{@children.size} children!"}
+ # Console.logger.info(self) {"Task ensure $! = #{$!} with #{@children&.size.inspect} children!"}
finish!
end
end
+
+ self.root.resume(@fiber)
end
# Finish the current task, and all bound bound IO objects.
def finish!
# Allow the fiber to be recycled.
@@ -334,9 +289,8 @@
# Set the current fiber's `:async_task` to this task.
def set!
# This is actually fiber-local:
Thread.current[:async_task] = self
- Console.logger = @logger if @logger
end
end
end