Class: Async::Task
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
-
#defer_stop ⇒ Object
Defer the handling of stop.
- #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.
58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
# File 'lib/async/task.rb', line 58 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 @defer_stop = nil end |
Instance Attribute Details
#fiber ⇒ Object (readonly)
119 120 121 |
# File 'lib/async/task.rb', line 119 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.
205 206 207 |
# File 'lib/async/task.rb', line 205 def result @result end |
#status ⇒ Object (readonly)
156 157 158 |
# File 'lib/async/task.rb', line 156 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.
291 292 293 |
# File 'lib/async/task.rb', line 291 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.
297 298 299 |
# File 'lib/async/task.rb', line 297 def self.current? Thread.current[:async_task] end |
.yield ⇒ Object
With no replacement.
51 52 53 |
# File 'lib/async/task.rb', line 51 def self.yield Fiber.scheduler.transfer end |
Instance Method Details
#alive? ⇒ Boolean
Whether the internal fiber is alive, i.e. it
122 123 124 |
# File 'lib/async/task.rb', line 122 def alive? @fiber&.alive? end |
#annotate(annotation, &block) ⇒ Object
83 84 85 86 87 88 89 |
# File 'lib/async/task.rb', line 83 def annotate(annotation, &block) if @fiber @fiber.annotate(annotation, &block) else super end end |
#annotation ⇒ Object
91 92 93 94 95 96 97 |
# File 'lib/async/task.rb', line 91 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.
172 173 174 175 176 177 178 179 180 |
# File 'lib/async/task.rb', line 172 def async(*arguments, **, &block) raise FinishedError if self.finished? task = Task.new(self, **, &block) task.run(*arguments) return task end |
#backtrace(*arguments) ⇒ Object
79 80 81 |
# File 'lib/async/task.rb', line 79 def backtrace(*arguments) @fiber&.backtrace(*arguments) end |
#completed? ⇒ Boolean Also known as: complete?
The task has completed execution and generated a result.
149 150 151 |
# File 'lib/async/task.rb', line 149 def completed? @status == :completed end |
#current? ⇒ Boolean
301 302 303 |
# File 'lib/async/task.rb', line 301 def current? Fiber.current.equal?(@fiber) end |
#defer_stop ⇒ Object
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.
260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 |
# File 'lib/async/task.rb', line 260 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 raise Stop, "Stopping current task (was deferred)!" end end else # If we are deferring stop already, entering it again is a no-op. yield end end |
#failed? ⇒ Boolean
139 140 141 |
# File 'lib/async/task.rb', line 139 def failed? @status == :failed end |
#finished? ⇒ Boolean
Whether we can remove this node from the reactor graph.
128 129 130 131 |
# File 'lib/async/task.rb', line 128 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
75 76 77 |
# File 'lib/async/task.rb', line 75 def reactor self.root end |
#run(*arguments) ⇒ Object
Begin the execution of the task.
159 160 161 162 163 164 165 166 167 168 169 |
# File 'lib/async/task.rb', line 159 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.
135 136 137 |
# File 'lib/async/task.rb', line 135 def running? @status == :running end |
#sleep(duration = nil) ⇒ Object
Prefer Kernel#sleep except when compatibility with ‘stable-v1` is required.
104 105 106 |
# File 'lib/async/task.rb', line 104 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.
212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 |
# File 'lib/async/task.rb', line 212 def stop(later = false) if self.stopped? # If the task is already stopped, a `stop` state transition re-enters the same state which is a no-op. However, we will also attempt to stop any running children too. This can happen if the children did not stop correctly the first time around. Doing this should probably be considered a bug, but it's better to be safe than sorry. return stopped! 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 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
144 145 146 |
# File 'lib/async/task.rb', line 144 def stopped? @status == :stopped end |
#to_s ⇒ Object
99 100 101 |
# File 'lib/async/task.rb', line 99 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 task resulted in an unhandled error (derived from ‘StandardError`), this will be raised. If the task was stopped, this will return `nil`.
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).
188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 |
# File 'lib/async/task.rb', line 188 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 @status == :failed 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.
109 110 111 |
# File 'lib/async/task.rb', line 109 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.
114 115 116 |
# File 'lib/async/task.rb', line 114 def yield Fiber.scheduler.yield end |