Class: Async::Container::Group
- Inherits:
-
Object
- Object
- Async::Container::Group
- Defined in:
- lib/async/container/group.rb
Overview
Manages a group of running processes.
Instance Attribute Summary collapse
-
#running ⇒ Object
readonly
Returns the value of attribute running.
- #the running tasks, indexed by IO.(runningtasks, indexedbyIO.) ⇒ Object readonly
Instance Method Summary collapse
-
#any? ⇒ Boolean
Whether the group contains any running processes.
-
#empty? ⇒ Boolean
Whether the group is empty.
-
#initialize ⇒ Group
constructor
Initialize an empty group.
-
#interrupt ⇒ Object
Interrupt all running processes.
-
#running? ⇒ Boolean
Whether the group contains any running processes.
-
#sleep(duration) ⇒ Object
Sleep for at most the specified duration until some state change occurs.
-
#stop(timeout = 1) ⇒ Object
Stop all child processes using #terminate.
-
#terminate ⇒ Object
Terminate all running processes.
-
#wait ⇒ Object
Begin any outstanding queued processes and wait for them indefinitely.
-
#wait_for(channel) ⇒ Object
Wait for a message in the specified Channel.
Constructor Details
#initialize ⇒ Group
Initialize an empty group.
33 34 35 36 37 38 |
# File 'lib/async/container/group.rb', line 33 def initialize @running = {} # This queue allows us to wait for processes to complete, without spawning new processes as a result. @queue = nil end |
Instance Attribute Details
#running ⇒ Object (readonly)
Returns the value of attribute running.
41 42 43 |
# File 'lib/async/container/group.rb', line 41 def running @running end |
#the running tasks, indexed by IO.(runningtasks, indexedbyIO.) ⇒ Object (readonly)
41 |
# File 'lib/async/container/group.rb', line 41 attr :running |
Instance Method Details
#any? ⇒ Boolean
Whether the group contains any running processes.
51 52 53 |
# File 'lib/async/container/group.rb', line 51 def any? @running.any? end |
#empty? ⇒ Boolean
Whether the group is empty.
57 58 59 |
# File 'lib/async/container/group.rb', line 57 def empty? @running.empty? end |
#interrupt ⇒ Object
Interrupt all running processes. This resumes the controlling fiber with an instance of Interrupt.
80 81 82 83 84 85 |
# File 'lib/async/container/group.rb', line 80 def interrupt Console.logger.debug(self, "Sending interrupt to #{@running.size} running processes...") @running.each_value do |fiber| fiber.resume(Interrupt) end end |
#running? ⇒ Boolean
Whether the group contains any running processes.
45 46 47 |
# File 'lib/async/container/group.rb', line 45 def running? @running.any? end |
#sleep(duration) ⇒ Object
Sleep for at most the specified duration until some state change occurs.
62 63 64 65 66 67 |
# File 'lib/async/container/group.rb', line 62 def sleep(duration) self.resume self.suspend self.wait_for_children(duration) end |
#stop(timeout = 1) ⇒ Object
Stop all child processes using #terminate.
98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 |
# File 'lib/async/container/group.rb', line 98 def stop(timeout = 1) # Use a default timeout if not specified: timeout = 1 if timeout == true if timeout start_time = Async::Clock.now self.interrupt while self.any? duration = Async::Clock.now - start_time remaining = timeout - duration if remaining >= 0 self.wait_for_children(duration) else self.wait_for_children(0) break end end end # Terminate all children: self.terminate # Wait for all children to exit: self.wait end |
#terminate ⇒ Object
Terminate all running processes. This resumes the controlling fiber with an instance of Terminate.
89 90 91 92 93 94 |
# File 'lib/async/container/group.rb', line 89 def terminate Console.logger.debug(self, "Sending terminate to #{@running.size} running processes...") @running.each_value do |fiber| fiber.resume(Terminate) end end |
#wait ⇒ Object
Begin any outstanding queued processes and wait for them indefinitely.
70 71 72 73 74 75 76 |
# File 'lib/async/container/group.rb', line 70 def wait self.resume while self.running? self.wait_for_children end end |
#wait_for(channel) ⇒ Object
Wait for a message in the specified Channel.
128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 |
# File 'lib/async/container/group.rb', line 128 def wait_for(channel) io = channel.in @running[io] = Fiber.current while @running.key?(io) result = Fiber.yield if result == Interrupt channel.interrupt! elsif result == Terminate channel.terminate! elsif = channel.receive yield else return channel.wait end end ensure @running.delete(io) end |