Class: Async::Task
Overview
Encapsulates the state of a running task and it’s result.
“‘mermaid stateDiagram-v2
- *
-
–> Initialized
Initialized –> Running : Run
Running –> Completed : Return Value Running –> Failed : Exception
Completed –> [*] Failed –> [*]
Running –> Stopped : Stop Stopped –> [*] Completed –> Stopped : Stop Failed –> Stopped : Stop Initialized –> Stopped : Stop “‘
Defined Under Namespace
Classes: FinishedError
Instance Attribute Summary collapse
- #fiber ⇒ Object readonly
-
#result ⇒ Object
readonly
Access the result of the task without waiting.
- #status ⇒ Object readonly
Attributes inherited from Node
#children, #head, #parent, #tail
Class Method Summary collapse
-
.current ⇒ Object
Lookup the Task for the current fiber.
-
.current? ⇒ Boolean
Check if there is a task defined for the current fiber.
-
.yield ⇒ Object
deprecated
Deprecated.
With no replacement.
Instance Method Summary collapse
-
#alive? ⇒ Boolean
Whether the internal fiber is alive, i.e.
- #annotate(annotation, &block) ⇒ Object
- #annotation ⇒ Object
-
#async(*arguments, **options, &block) ⇒ Object
Run an asynchronous task as a child of the current task.
- #backtrace(*arguments) ⇒ Object
-
#completed? ⇒ Boolean
(also: #complete?)
The task has completed execution and generated a result.
- #current? ⇒ Boolean
- #failed? ⇒ Boolean
-
#finished? ⇒ Boolean
Whether we can remove this node from the reactor graph.
-
#initialize(parent = Task.current?, finished: nil, **options, &block) ⇒ Task
constructor
Create a new task.
- #reactor ⇒ Object
-
#run(*arguments) ⇒ Object
Begin the execution of the task.
-
#running? ⇒ Boolean
Whether the task is running.
-
#sleep(duration = nil) ⇒ Object
deprecated
Deprecated.
Prefer Kernel#sleep except when compatibility with ‘stable-v1` is required.
-
#stop(later = false) ⇒ Object
Stop the task and all of its children.
-
#stopped? ⇒ Boolean
The task has been stopped.
- #to_s ⇒ Object
-
#wait ⇒ Object
Retrieve the current result of the task.
-
#with_timeout(duration, exception = TimeoutError, message = "execution expired", &block) ⇒ Object
Execute the given block of code, raising the specified exception if it exceeds the given duration during a non-blocking operation.
-
#yield ⇒ Object
Yield back to the reactor and allow other fibers to execute.
Methods inherited from Node
#The parent node.=, #children?, #consume, #description, #print_hierarchy, #root, #terminate, #transient?, #traverse
Constructor Details
#initialize(parent = Task.current?, finished: nil, **options, &block) ⇒ Task
Create a new task.
76 77 78 79 80 81 82 83 84 85 86 87 88 89 |
# File 'lib/async/task.rb', line 76 def initialize(parent = Task.current?, finished: nil, **, &block) super(parent, **) # These instance variables are critical to the state of the task. # In the initialized state, the @block should be set, but the @fiber should be nil. # In the running state, the @fiber should be set. # In a finished state, the @block should be nil, and the @fiber should be nil. @block = block @fiber = nil @status = :initialized @result = nil @finished = finished end |
Instance Attribute Details
#fiber ⇒ Object (readonly)
135 136 137 |
# File 'lib/async/task.rb', line 135 def fiber @fiber end |
#result ⇒ Object (readonly)
Access the result of the task without waiting. May be nil if the task is not completed. Does not raise exceptions.
221 222 223 |
# File 'lib/async/task.rb', line 221 def result @result end |
#status ⇒ Object (readonly)
172 173 174 |
# File 'lib/async/task.rb', line 172 def status @status end |
Class Method Details
.current ⇒ Object
Lookup the Async::Task for the current fiber. Raise ‘RuntimeError` if none is available. @raises If task was not #set! for the current fiber.
264 265 266 |
# File 'lib/async/task.rb', line 264 def self.current Thread.current[:async_task] or raise RuntimeError, "No async task available!" end |
.current? ⇒ Boolean
Check if there is a task defined for the current fiber.
270 271 272 |
# File 'lib/async/task.rb', line 270 def self.current? Thread.current[:async_task] end |
.yield ⇒ Object
With no replacement.
69 70 71 |
# File 'lib/async/task.rb', line 69 def self.yield Fiber.scheduler.transfer end |
Instance Method Details
#alive? ⇒ Boolean
Whether the internal fiber is alive, i.e. it
138 139 140 |
# File 'lib/async/task.rb', line 138 def alive? @fiber&.alive? end |
#annotate(annotation, &block) ⇒ Object
99 100 101 102 103 104 105 |
# File 'lib/async/task.rb', line 99 def annotate(annotation, &block) if @fiber @fiber.annotate(annotation, &block) else super end end |
#annotation ⇒ Object
107 108 109 110 111 112 113 |
# File 'lib/async/task.rb', line 107 def annotation if @fiber @fiber.annotation else super end end |
#async(*arguments, **options, &block) ⇒ Object
Run an asynchronous task as a child of the current task.
188 189 190 191 192 193 194 195 196 |
# File 'lib/async/task.rb', line 188 def async(*arguments, **, &block) raise FinishedError if self.finished? task = Task.new(self, **, &block) task.run(*arguments) return task end |
#backtrace(*arguments) ⇒ Object
95 96 97 |
# File 'lib/async/task.rb', line 95 def backtrace(*arguments) @fiber&.backtrace(*arguments) end |
#completed? ⇒ Boolean Also known as: complete?
The task has completed execution and generated a result.
165 166 167 |
# File 'lib/async/task.rb', line 165 def completed? @status == :completed end |
#current? ⇒ Boolean
274 275 276 |
# File 'lib/async/task.rb', line 274 def current? Fiber.current.equal?(@fiber) end |
#failed? ⇒ Boolean
155 156 157 |
# File 'lib/async/task.rb', line 155 def failed? @status == :failed end |
#finished? ⇒ Boolean
Whether we can remove this node from the reactor graph.
144 145 146 147 |
# File 'lib/async/task.rb', line 144 def finished? # If the block is nil and the fiber is nil, it means the task has finished execution. This becomes true after `finish!` is called. super && @block.nil? && @fiber.nil? end |
#reactor ⇒ Object
91 92 93 |
# File 'lib/async/task.rb', line 91 def reactor self.root end |
#run(*arguments) ⇒ Object
Begin the execution of the task.
175 176 177 178 179 180 181 182 183 184 185 |
# File 'lib/async/task.rb', line 175 def run(*arguments) if @status == :initialized @status = :running schedule do @block.call(self, *arguments) end else raise RuntimeError, "Task already running!" end end |
#running? ⇒ Boolean
Whether the task is running.
151 152 153 |
# File 'lib/async/task.rb', line 151 def running? @status == :running end |
#sleep(duration = nil) ⇒ Object
Prefer Kernel#sleep except when compatibility with ‘stable-v1` is required.
120 121 122 |
# File 'lib/async/task.rb', line 120 def sleep(duration = nil) super end |
#stop(later = false) ⇒ Object
Stop the task and all of its children.
If ‘later` is false, it means that `stop` has been invoked directly. When `later` is true, it means that `stop` is invoked by `stop_children` or some other indirect mechanism. In that case, if we encounter the “current” fiber, we can’t stop it right away, as it’s currently performing ‘#stop`. Stopping it immediately would interrupt the current stop traversal, so we need to schedule the stop to occur later.
228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 |
# File 'lib/async/task.rb', line 228 def stop(later = false) if self.stopped? # If we already stopped this task... don't try to stop it again: return end # If the fiber is alive, we need to stop it: if @fiber&.alive? if self.current? # If the fiber is current, and later is `true`, we need to schedule the fiber to be stopped later, as it's currently invoking `stop`: if later # If the fiber is the current fiber and we want to stop it later, schedule it: Fiber.scheduler.push(Stop::Later.new(self)) else # Otherwise, raise the exception directly: raise Stop, "Stopping current task!" end else # If the fiber is not curent, we can raise the exception directly: begin # There is a chance that this will stop the fiber that originally called stop. If that happens, the exception handling in `#stopped` will rescue the exception and re-raise it later. Fiber.scheduler.raise(@fiber, Stop) rescue FiberError # In some cases, this can cause a FiberError (it might be resumed already), so we schedule it to be stopped later: 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 |
#stopped? ⇒ Boolean
The task has been stopped
160 161 162 |
# File 'lib/async/task.rb', line 160 def stopped? @status == :stopped end |
#to_s ⇒ Object
115 116 117 |
# File 'lib/async/task.rb', line 115 def to_s "\#<#{self.description} (#{@status})>" end |
#wait ⇒ Object
Retrieve the current result of the task. Will cause the caller to wait until result is available. If the result was an exception, raise that exception.
Conceptually speaking, waiting on a task should return a result, and if it throws an exception, this is certainly an exceptional case that should represent a failure in your program, not an expected outcome. In other words, you should not design your programs to expect exceptions from ‘#wait` as a normal flow control, and prefer to catch known exceptions within the task itself and return a result that captures the intention of the failure, e.g. a `TimeoutError` might simply return `nil` or `false` to indicate that the operation did not generate a valid result (as a timeout was an expected outcome of the internal operation in this case).
204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 |
# File 'lib/async/task.rb', line 204 def wait raise "Cannot wait on own fiber!" if Fiber.current.equal?(@fiber) # `finish!` will set both of these to nil before signaling the condition: if @block || @fiber @finished ||= Condition.new @finished.wait end if @result.is_a?(Exception) raise @result else return @result end end |
#with_timeout(duration, exception = TimeoutError, message = "execution expired", &block) ⇒ Object
Execute the given block of code, raising the specified exception if it exceeds the given duration during a non-blocking operation.
125 126 127 |
# File 'lib/async/task.rb', line 125 def with_timeout(duration, exception = TimeoutError, = "execution expired", &block) Fiber.scheduler.with_timeout(duration, exception, , &block) end |
#yield ⇒ Object
Yield back to the reactor and allow other fibers to execute.
130 131 132 |
# File 'lib/async/task.rb', line 130 def yield Fiber.scheduler.yield end |