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.
-
#initialize ⇒ Reactor
constructor
A new instance of Reactor.
- #register(*args) ⇒ 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.
-
#timeout(duration) ⇒ Object
Invoke the block, but after the timeout, raise TimeoutError in any currenly blocking operation.
- #to_s ⇒ Object
-
#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, #finished?, #print_hierarchy, #reap, #traverse
Constructor Details
#initialize ⇒ Reactor
Returns a new instance of Reactor.
61 62 63 64 65 66 67 68 69 70 |
# File 'lib/async/reactor.rb', line 61 def initialize super @selector = NIO::Selector.new @timers = Timers::Group.new @ready = [] @stopped = true end |
Instance Attribute Details
#stopped ⇒ Object (readonly)
77 78 79 |
# File 'lib/async/reactor.rb', line 77 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.
123 124 125 |
# File 'lib/async/reactor.rb', line 123 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.
89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
# File 'lib/async/reactor.rb', line 89 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.
181 182 183 184 185 186 187 188 |
# File 'lib/async/reactor.rb', line 181 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.
192 193 194 |
# File 'lib/async/reactor.rb', line 192 def closed? @selector.nil? end |
#register(*args) ⇒ Object
105 106 107 108 109 110 111 |
# File 'lib/async/reactor.rb', line 105 def register(*args) monitor = @selector.register(*args) monitor.value = Fiber.current return monitor end |
#run(*args, &block) ⇒ Object
136 137 138 139 140 141 142 143 144 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 |
# File 'lib/async/reactor.rb', line 136 def run(*args, &block) raise RuntimeError, 'Reactor has been closed' if @selector.nil? @stopped = false # Allow the user to kick of the initial async tasks. initial_task = async(*args, &block) if block_given? @timers.wait do |interval| # - nil: no timers # - -ve: timers expired already # - 0: timers ready to fire # - +ve: timers waiting to fire interval = 0 if interval && interval < 0 # Async.logger.debug{"[#{self} Pre] Updating #{@children.count} children..."} # As timeouts may have been updated, and caused fibers to complete, we should check this. @ready.each do |fiber| fiber.resume if fiber.alive? end; @ready.clear # If there is nothing to do, then finish: # Async.logger.debug{"[#{self}] @children.empty? = #{@children.empty?} && interval #{interval.inspect}"} return initial_task if @children.empty? && interval.nil? # Async.logger.debug{"Selecting with #{@children.count} fibers interval = #{interval.inspect}..."} if monitors = @selector.select(interval) monitors.each do |monitor| if fiber = monitor.value fiber.resume # if fiber.alive? end end end end until @stopped return initial_task ensure Async.logger.debug{"[#{self} Ensure] Exiting run-loop (stopped: #{@stopped} exception: #{$!.inspect})..."} @stopped = true end |
#sleep(duration) ⇒ Object
Put the calling fiber to sleep for a given ammount of time.
198 199 200 201 202 203 204 205 206 207 208 209 210 |
# File 'lib/async/reactor.rb', line 198 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.
115 116 117 118 119 120 |
# File 'lib/async/reactor.rb', line 115 def stop unless @stopped @stopped = true @selector.wakeup end end |
#timeout(duration) ⇒ Object
Invoke the block, but after the timeout, raise TimeoutError in any currenly blocking operation.
216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 |
# File 'lib/async/reactor.rb', line 216 def timeout(duration) backtrace = caller task = Fiber.current timer = self.after(duration) do if task.alive? error = TimeoutError.new("execution expired") error.set_backtrace backtrace task.resume error end end yield ensure timer.cancel if timer end |
#to_s ⇒ Object
72 73 74 |
# File 'lib/async/reactor.rb', line 72 def to_s "<#{self.description} stopped=#{@stopped}>" end |
#yield(fiber = Fiber.current) ⇒ Object
Yield the current fiber and resume it on the next iteration of the event loop.
128 129 130 131 132 |
# File 'lib/async/reactor.rb', line 128 def yield(fiber = Fiber.current) @ready << fiber Fiber.yield end |