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, &block) ⇒ Object
The preferred method to invoke asynchronous behavior at the top level.
Instance Method Summary collapse
-
#<<(fiber) ⇒ Object
Schedule a fiber (or equivalent object) to be resumed on the next loop through the reactor.
-
#async(*args) {|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: NIO::Selector.new) ⇒ Reactor
constructor
A new instance of Reactor.
- #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: NIO::Selector.new) ⇒ Reactor
Returns a new instance of Reactor.
61 62 63 64 65 66 67 68 69 70 71 |
# File 'lib/async/reactor.rb', line 61 def initialize(parent = nil, selector: NIO::Selector.new) super(parent) @selector = selector @timers = Timers::Group.new @ready = [] @running = [] @stopped = true end |
Instance Attribute Details
#stopped ⇒ Object (readonly)
78 79 80 |
# File 'lib/async/reactor.rb', line 78 def stopped @stopped end |
Class Method Details
.run(*args, &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 |
Instance Method Details
#<<(fiber) ⇒ Object
Schedule a fiber (or equivalent object) to be resumed on the next loop through the reactor.
128 129 130 |
# File 'lib/async/reactor.rb', line 128 def << fiber @ready << fiber end |
#async(*args) {|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.
94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 |
# File 'lib/async/reactor.rb', line 94 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) # Async.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.
203 204 205 206 207 208 209 |
# File 'lib/async/reactor.rb', line 203 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.
213 214 215 |
# File 'lib/async/reactor.rb', line 213 def closed? @selector.nil? end |
#finished? ⇒ Boolean
139 140 141 |
# File 'lib/async/reactor.rb', line 139 def finished? super && @ready.empty? && @running.empty? end |
#register(io, interest, value = Fiber.current) ⇒ Object
110 111 112 113 114 115 |
# File 'lib/async/reactor.rb', line 110 def register(io, interest, value = Fiber.current) monitor = @selector.register(io, interest) monitor.value = value return monitor end |
#run(*args, &block) ⇒ Object
145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 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 |
# File 'lib/async/reactor.rb', line 145 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| # running used to correctly answer on `finished?`, and to reuse Array object. @running, @ready = @ready, @running if @running.any? @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 # - nil: no timers # - -ve: timers expired already # - 0: timers ready to fire # - +ve: timers waiting to fire if interval && interval < 0 interval = 0 end # Async.logger.debug(self) {"Updating #{@children.count} children..."} # As timeouts may have been updated, and caused fibers to complete, we should check this. # If there is nothing to do, then finish: if !interval && self.finished? return initial_task end # Async.logger.debug(self) {"Selecting with #{@children.count} fibers interval = #{interval.inspect}..."} if monitors = @selector.select(interval) monitors.each do |monitor| monitor.value.resume end end end until @stopped return initial_task ensure Async.logger.debug(self) {"Exiting run-loop because #{$! ? $!.inspect : 'finished'}."} @stopped = true end |
#sleep(duration) ⇒ Object
Put the calling fiber to sleep for a given ammount of time.
219 220 221 222 223 224 225 226 227 228 229 230 231 |
# File 'lib/async/reactor.rb', line 219 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.
119 120 121 122 123 124 |
# File 'lib/async/reactor.rb', line 119 def stop unless @stopped @stopped = true @selector.wakeup end end |
#stopped? ⇒ Boolean
80 81 82 |
# File 'lib/async/reactor.rb', line 80 def stopped? @stopped end |
#timeout(*args, &block) ⇒ Object
TODO remove
252 253 254 255 256 |
# File 'lib/async/reactor.rb', line 252 def timeout(*args, &block) warn "#{self.class}\#timeout(...) is deprecated, use #{self.class}\#with_timeout(...) instead." with_timeout(*args, &block) end |
#to_s ⇒ Object
73 74 75 |
# File 'lib/async/reactor.rb', line 73 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.
236 237 238 239 240 241 242 243 244 245 246 247 248 249 |
# File 'lib/async/reactor.rb', line 236 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.
133 134 135 136 137 |
# File 'lib/async/reactor.rb', line 133 def yield(fiber = Fiber.current) @ready << fiber Fiber.yield end |