Class: Async::Reactor
- Extended by:
- Forwardable
- Defined in:
- lib/async/reactor.rb
Overview
An asynchronous, cooperatively scheduled event reactor.
Instance Attribute Summary collapse
- #stopped ⇒ Object readonly
Attributes inherited from Node
#annotation, #children, #parent
Class Method Summary collapse
-
.run(*args, **options, &block) ⇒ Object
The preferred method to invoke asynchronous behavior at the top level.
- .selector ⇒ Object
Instance Method Summary collapse
-
#<<(fiber) ⇒ Object
Schedule a fiber (or equivalent object) to be resumed on the next loop through the reactor.
-
#async(*args, **options) {|Task| ... } ⇒ Task
Start an asynchronous task within the specified reactor.
-
#close ⇒ void
Stop each of the children tasks and close the selector.
-
#closed? ⇒ Boolean
Check if the selector has been closed.
- #finished? ⇒ Boolean
-
#initialize(parent = nil, selector: self.class.selector, logger: nil) ⇒ Reactor
constructor
A new instance of Reactor.
- #logger ⇒ Object
- #register(io, interest, value = Fiber.current) ⇒ Object
-
#run(*args, &block) ⇒ Object
Run the reactor until either all tasks complete or #stop is invoked.
-
#sleep(duration) ⇒ Object
Put the calling fiber to sleep for a given ammount of time.
-
#stop ⇒ void
Stop the reactor at the earliest convenience.
- #stopped? ⇒ Boolean
-
#timeout(*args, &block) ⇒ Object
TODO remove.
- #to_s ⇒ Object
-
#with_timeout(timeout, exception = TimeoutError) ⇒ Object
Invoke the block, but after the specified timeout, raise TimeoutError in any currenly blocking operation.
-
#yield(fiber = Fiber.current) ⇒ Object
Yield the current fiber and resume it on the next iteration of the event loop.
Methods inherited from Node
#annotate, #consume, #description, #print_hierarchy, #reap, #traverse
Constructor Details
#initialize(parent = nil, selector: self.class.selector, logger: nil) ⇒ Reactor
Returns a new instance of Reactor.
73 74 75 76 77 78 79 80 81 82 83 84 |
# File 'lib/async/reactor.rb', line 73 def initialize(parent = nil, selector: self.class.selector, logger: nil) super(parent) @selector = selector @timers = Timers::Group.new @logger = logger @ready = [] @running = [] @stopped = true end |
Instance Attribute Details
#stopped ⇒ Object (readonly)
95 96 97 |
# File 'lib/async/reactor.rb', line 95 def stopped @stopped end |
Class Method Details
.run(*args, **options, &block) ⇒ Object
The preferred method to invoke asynchronous behavior at the top level.
- When invoked within an existing reactor task, it will run the given block asynchronously. Will return the task once it has been scheduled.
- When invoked at the top level, will create and run a reactor, and invoke the block as an asynchronous task. Will block until the reactor finishes running.
45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
# File 'lib/async/reactor.rb', line 45 def self.run(*args, **, &block) if current = Task.current? reactor = current.reactor return reactor.async(*args, **, &block) else reactor = self.new(**) begin return reactor.run(*args, &block) ensure reactor.close end end end |
.selector ⇒ Object
61 62 63 64 65 66 67 68 69 70 71 |
# File 'lib/async/reactor.rb', line 61 def self.selector if backend = ENV['ASYNC_BACKEND']&.to_sym if NIO::Selector.backends.include?(backend) return NIO::Selector.new(backend) else warn "Could not find ASYNC_BACKEND=#{backend}!" end end return NIO::Selector.new end |
Instance Method Details
#<<(fiber) ⇒ Object
Schedule a fiber (or equivalent object) to be resumed on the next loop through the reactor.
146 147 148 |
# File 'lib/async/reactor.rb', line 146 def << fiber @ready << fiber end |
#async(*args, **options) {|Task| ... } ⇒ Task
Start an asynchronous task within the specified reactor. The task will be executed until the first blocking call, at which point it will yield and and this method will return.
This is the main entry point for scheduling asynchronus tasks.
112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 |
# File 'lib/async/reactor.rb', line 112 def async(*args, **, &block) task = Task.new(self, **, &block) # I want to take a moment to explain the logic of this. # When calling an async block, we deterministically execute it until the # first blocking operation. We don't *have* to do this - we could schedule # it for later execution, but it's useful to: # - Fail at the point of call where possible. # - Execute determinstically where possible. # - Avoid overhead if no blocking operation is performed. task.run(*args) # logger.debug "Initial execution of task #{fiber} complete (#{result} -> #{fiber.alive?})..." return task end |
#close ⇒ void
This method returns an undefined value.
Stop each of the children tasks and close the selector.
222 223 224 225 226 227 228 |
# File 'lib/async/reactor.rb', line 222 def close @children&.each(&:stop) # TODO Should we also clear all timers? @selector.close @selector = nil end |
#closed? ⇒ Boolean
Check if the selector has been closed.
232 233 234 |
# File 'lib/async/reactor.rb', line 232 def closed? @selector.nil? end |
#finished? ⇒ Boolean
157 158 159 160 |
# File 'lib/async/reactor.rb', line 157 def finished? # I'm not sure if checking `@running.empty?` is really required. super && @ready.empty? && @running.empty? end |
#logger ⇒ Object
86 87 88 |
# File 'lib/async/reactor.rb', line 86 def logger @logger ||= Console.logger end |
#register(io, interest, value = Fiber.current) ⇒ Object
128 129 130 131 132 133 |
# File 'lib/async/reactor.rb', line 128 def register(io, interest, value = Fiber.current) monitor = @selector.register(io, interest) monitor.value = value return monitor end |
#run(*args, &block) ⇒ Object
164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 |
# File 'lib/async/reactor.rb', line 164 def run(*args, &block) raise RuntimeError, 'Reactor has been closed' if @selector.nil? @stopped = false initial_task = self.async(*args, &block) if block_given? @timers.wait do |interval| # logger.debug(self) {"@ready = #{@ready} @running = #{@running}"} if @ready.any? # running used to correctly answer on `finished?`, and to reuse Array object. @running, @ready = @ready, @running @running.each do |fiber| fiber.resume if fiber.alive? end @running.clear # if there are tasks ready to execute, don't sleep. if @ready.any? interval = 0 else # The above tasks may schedule, cancel or affect timers in some way. We need to compute a new wait interval for the blocking selector call below: interval = @timers.wait_interval end end # As timeouts may have been updated, and caused fibers to complete, we should check this. if interval.nil? if self.finished? # If there is nothing to do, then finish: return initial_task end elsif interval < 0 # We have timers ready to fire, don't sleep in the selctor: interval = 0 end # logger.debug(self) {"Selecting with #{@children&.size} children with interval = #{interval.inspect}..."} if monitors = @selector.select(interval) monitors.each do |monitor| monitor.value.resume end end end until @stopped return initial_task ensure logger.debug(self) {"Exiting run-loop because #{$! ? $! : 'finished'}."} @stopped = true end |
#sleep(duration) ⇒ Object
Put the calling fiber to sleep for a given ammount of time.
238 239 240 241 242 243 244 245 246 247 248 249 250 |
# File 'lib/async/reactor.rb', line 238 def sleep(duration) fiber = Fiber.current timer = self.after(duration) do if fiber.alive? fiber.resume end end Task.yield ensure timer.cancel if timer end |
#stop ⇒ void
This method returns an undefined value.
Stop the reactor at the earliest convenience. Can be called from a different thread safely.
137 138 139 140 141 142 |
# File 'lib/async/reactor.rb', line 137 def stop unless @stopped @stopped = true @selector.wakeup end end |
#stopped? ⇒ Boolean
97 98 99 |
# File 'lib/async/reactor.rb', line 97 def stopped? @stopped end |
#timeout(*args, &block) ⇒ Object
TODO remove
271 272 273 274 275 |
# File 'lib/async/reactor.rb', line 271 def timeout(*args, &block) warn "#{self.class}\#timeout(...) is deprecated, use #{self.class}\#with_timeout(...) instead." with_timeout(*args, &block) end |
#to_s ⇒ Object
90 91 92 |
# File 'lib/async/reactor.rb', line 90 def to_s "<#{self.description} stopped=#{@stopped}>" end |
#with_timeout(timeout, exception = TimeoutError) ⇒ Object
Invoke the block, but after the specified timeout, raise TimeoutError in any currenly blocking operation. If the block runs to completion before the timeout occurs or there are no non-blocking operations after the timeout expires, the code will complete without any exception.
255 256 257 258 259 260 261 262 263 264 265 266 267 268 |
# File 'lib/async/reactor.rb', line 255 def with_timeout(timeout, exception = TimeoutError) fiber = Fiber.current timer = self.after(timeout) do if fiber.alive? error = exception.new("execution expired") fiber.resume error end end yield timer ensure timer.cancel if timer end |
#yield(fiber = Fiber.current) ⇒ Object
Yield the current fiber and resume it on the next iteration of the event loop.
151 152 153 154 155 |
# File 'lib/async/reactor.rb', line 151 def yield(fiber = Fiber.current) @ready << fiber Fiber.yield end |