Skip to content

Commit af85ecc

Browse files
Fix reload reconciliation of keyed children.
Assisted-By: devx/904563b8-dbee-48b0-9726-f036df3ed96d
1 parent 4db57a0 commit af85ecc

6 files changed

Lines changed: 141 additions & 15 deletions

File tree

lib/async/container/channel.rb

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ class Channel
1313
def initialize(timeout: 1.0)
1414
@in, @out = ::IO.pipe
1515
@in.timeout = timeout
16+
@stopping = false
1617
end
1718

1819
# The input end of the pipe.
@@ -23,6 +24,17 @@ def initialize(timeout: 1.0)
2324
# @attribute [IO]
2425
attr :out
2526

27+
# Whether this child is being deliberately stopped and should not be restarted.
28+
# @returns [Boolean]
29+
def stopping?
30+
@stopping
31+
end
32+
33+
# Mark this child as being deliberately stopped, so its supervising fiber will not restart it.
34+
def stopping!
35+
@stopping = true
36+
end
37+
2638
# Close the input end of the pipe.
2739
def close_read
2840
@in.close

lib/async/container/generic.rb

Lines changed: 25 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -144,8 +144,7 @@ def wait_until_ready
144144
end
145145
end
146146

147-
self.sleep
148-
147+
# Check readiness before sleeping: after a reload that only reuses or removes children, no new child sends a readiness message, so `sleep` would otherwise block indefinitely even though everything is already ready.
149148
if self.status?(:ready)
150149
Console.debug(self) do |buffer|
151150
buffer.puts "All ready:"
@@ -156,6 +155,8 @@ def wait_until_ready
156155

157156
return true
158157
end
158+
159+
self.sleep
159160
end
160161
end
161162

@@ -294,7 +295,7 @@ def spawn(name: nil, restart: false, key: nil, health_check_timeout: nil, startu
294295
Console.error(self, "Policy error in child_exit!", exception: error)
295296
end
296297

297-
if restart && !@stopping
298+
if restart && !@stopping && !child.stopping?
298299
@statistics.restart!
299300
else
300301
break
@@ -327,20 +328,39 @@ def async(**options, &block)
327328
end
328329

329330
# Reload the container's keyed instances.
331+
#
332+
# Any keyed child which is not re-marked during {yield} is considered obsolete and is
333+
# stopped. Its supervising fiber removes it from {@keyed} once it has exited.
334+
#
335+
# @returns [Boolean] Whether any keyed instances were stopped.
330336
def reload
331337
@keyed.each_value(&:clear!)
332338

333339
yield
334340

335341
dirty = false
336342

337-
@keyed.delete_if do |key, value|
338-
value.stop? && (dirty = true)
343+
# Snapshot the values so we can stop obsolete children without mutating `@keyed` mid-iteration (the supervising fiber deletes the entry when the child exits):
344+
@keyed.values.each do |keyed|
345+
unless keyed.marked?
346+
stop_child(keyed.value)
347+
dirty = true
348+
end
339349
end
340350

341351
return dirty
342352
end
343353

354+
# Stop a single child instance and prevent it from being restarted.
355+
# @parameter child [Channel] The child instance to stop.
356+
# @parameter graceful [Boolean | Numeric] Whether to stop the child gracefully.
357+
def stop_child(child, graceful = true)
358+
# Prevent the supervising fiber from restarting the child once it exits:
359+
child.stopping!
360+
361+
@group.stop_child(child, graceful)
362+
end
363+
344364
# Mark the container's keyed instance which ensures that it won't be discarded.
345365
def mark?(key)
346366
if key

lib/async/container/group.rb

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -180,6 +180,48 @@ def stop(graceful = true)
180180
end
181181
end
182182

183+
# Stop a single running child, gracefully if possible, then forcefully.
184+
#
185+
# Mirrors the multi-phase sequence of {stop}, but scoped to one child: send SIGINT and
186+
# wait up to `graceful` seconds, then send SIGKILL and wait for it to exit.
187+
#
188+
# @parameter channel [Channel] The channel of the child to stop.
189+
# @parameter graceful [Boolean | Numeric] Whether to interrupt first, or a specific timeout.
190+
def stop_child(channel, graceful = true)
191+
io = channel.in
192+
fiber = @running[io]
193+
194+
return unless fiber
195+
196+
if graceful
197+
# Send SIGINT to the child:
198+
fiber.resume(Interrupt)
199+
200+
if graceful == true
201+
graceful = DEFAULT_GRACEFUL_TIMEOUT
202+
end
203+
204+
clock = Clock.start
205+
206+
# Wait for the child to exit:
207+
while @running.key?(io)
208+
duration = graceful - clock.total
209+
break if duration < 0
210+
211+
wait_for_children(duration)
212+
end
213+
end
214+
ensure
215+
# Force kill if it's still running:
216+
if fiber && @running.key?(io)
217+
fiber.resume(Kill)
218+
219+
while @running.key?(io)
220+
wait_for_children(nil)
221+
end
222+
end
223+
end
224+
183225
# Wait for a message in the specified {Channel}.
184226
def wait_for(channel)
185227
io = channel.in

lib/async/container/keyed.rb

Lines changed: 0 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -38,16 +38,6 @@ def mark!
3838
def clear!
3939
@marked = false
4040
end
41-
42-
# Stop the instance if it was not marked.
43-
#
44-
# @returns [Boolean] True if the instance was stopped.
45-
def stop?
46-
unless @marked
47-
@value.stop
48-
return true
49-
end
50-
end
5141
end
5242
end
5343
end

releases.md

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
# Releases
22

3+
## Unreleased
4+
5+
- **Fixed**: Reloading a container now correctly stops keyed children whose keys are no longer configured. Previously this raised `NoMethodError` (the child has no `stop` method) and would not have stopped `restart: true` children even if it had. `Generic#reload` now stops obsolete children via a new `Group#stop_child`, and a per-child stopping flag prevents them from being respawned.
6+
37
## v0.37.0
48

59
- Rename `ASYNC_CONTAINER_GRACEFUL_TIMEOUT` to `ASYNC_CONTAINER_GRACEFUL_STOP` and apply it at the controller level as `GRACEFUL_STOP`. `Group#stop` now only applies the shutdown policy it is given.

test/async/container/controller.rb

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,64 @@ def controller.setup(container)
8282

8383
controller.wait
8484
end
85+
86+
it "spawns a newly configured keyed child on reload" do
87+
keys = ["a"]
88+
89+
controller.define_singleton_method(:setup) do |container|
90+
container.reload do
91+
keys.each do |key|
92+
container.spawn(key: key) do |instance|
93+
instance.ready!
94+
sleep
95+
end
96+
end
97+
end
98+
end
99+
100+
controller.start
101+
102+
expect(controller.container["a"]).not.to be_nil
103+
expect(controller.container["b"]).to be_nil
104+
105+
# The configuration now includes an additional keyed child:
106+
keys << "b"
107+
controller.reload
108+
109+
expect(controller.container["a"]).not.to be_nil
110+
expect(controller.container["b"]).not.to be_nil
111+
ensure
112+
controller.stop(false)
113+
end
114+
115+
it "stops a keyed child that is no longer configured on reload" do
116+
keys = ["a", "b"]
117+
118+
controller.define_singleton_method(:setup) do |container|
119+
container.reload do
120+
keys.each do |key|
121+
container.spawn(key: key) do |instance|
122+
instance.ready!
123+
sleep
124+
end
125+
end
126+
end
127+
end
128+
129+
controller.start
130+
131+
expect(controller.container["a"]).not.to be_nil
132+
expect(controller.container["b"]).not.to be_nil
133+
134+
# The configuration no longer includes "b", so reload should stop it:
135+
keys.delete("b")
136+
controller.reload
137+
138+
expect(controller.container["a"]).not.to be_nil
139+
expect(controller.container["b"]).to be_nil
140+
ensure
141+
controller.stop(false)
142+
end
85143
end
86144

87145
with "notify" do

0 commit comments

Comments
 (0)