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
- #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.
56 57 58 59 60 61 62 63 64 65 66 67 68 69 |
# File 'lib/async/task.rb', line 56 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)
115 116 117 |
# File 'lib/async/task.rb', line 115 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.
201 202 203 |
# File 'lib/async/task.rb', line 201 def result @result end |
#status ⇒ Object (readonly)
152 153 154 |
# File 'lib/async/task.rb', line 152 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.
244 245 246 |
# File 'lib/async/task.rb', line 244 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.
250 251 252 |
# File 'lib/async/task.rb', line 250 def self.current? Thread.current[:async_task] end |
.yield ⇒ Object
With no replacement.
49 50 51 |
# File 'lib/async/task.rb', line 49 def self.yield Fiber.scheduler.transfer end |
Instance Method Details
#alive? ⇒ Boolean
Whether the internal fiber is alive, i.e. it
118 119 120 |
# File 'lib/async/task.rb', line 118 def alive? @fiber&.alive? end |
#annotate(annotation, &block) ⇒ Object
79 80 81 82 83 84 85 |
# File 'lib/async/task.rb', line 79 def annotate(annotation, &block) if @fiber @fiber.annotate(annotation, &block) else super end end |
#annotation ⇒ Object
87 88 89 90 91 92 93 |
# File 'lib/async/task.rb', line 87 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.
168 169 170 171 172 173 174 175 176 |
# File 'lib/async/task.rb', line 168 def async(*arguments, **, &block) raise FinishedError if self.finished? task = Task.new(self, **, &block) task.run(*arguments) return task end |
#backtrace(*arguments) ⇒ Object
75 76 77 |
# File 'lib/async/task.rb', line 75 def backtrace(*arguments) @fiber&.backtrace(*arguments) end |
#completed? ⇒ Boolean Also known as: complete?
The task has completed execution and generated a result.
145 146 147 |
# File 'lib/async/task.rb', line 145 def completed? @status == :completed end |
#current? ⇒ Boolean
254 255 256 |
# File 'lib/async/task.rb', line 254 def current? Fiber.current.equal?(@fiber) end |
#failed? ⇒ Boolean
135 136 137 |
# File 'lib/async/task.rb', line 135 def failed? @status == :failed end |
#finished? ⇒ Boolean
Whether we can remove this node from the reactor graph.
124 125 126 127 |
# File 'lib/async/task.rb', line 124 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
71 72 73 |
# File 'lib/async/task.rb', line 71 def reactor self.root end |
#run(*arguments) ⇒ Object
Begin the execution of the task.
155 156 157 158 159 160 161 162 163 164 165 |
# File 'lib/async/task.rb', line 155 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.
131 132 133 |
# File 'lib/async/task.rb', line 131 def running? @status == :running end |
#sleep(duration = nil) ⇒ Object
Prefer Kernel#sleep except when compatibility with ‘stable-v1` is required.
100 101 102 |
# File 'lib/async/task.rb', line 100 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.
208 209 210 211 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 |
# File 'lib/async/task.rb', line 208 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
140 141 142 |
# File 'lib/async/task.rb', line 140 def stopped? @status == :stopped end |
#to_s ⇒ Object
95 96 97 |
# File 'lib/async/task.rb', line 95 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).
184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 |
# File 'lib/async/task.rb', line 184 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.
105 106 107 |
# File 'lib/async/task.rb', line 105 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.
110 111 112 |
# File 'lib/async/task.rb', line 110 def yield Fiber.scheduler.yield end |