Tighten queue cancel - #1188
Tighten queue cancel#1188
Conversation
| // Ensure timer is stopped once ctx is done and we return. | ||
| defer timer.Stop() |
There was a problem hiding this comment.
nit: This seems to be a no-op since Go 1.23.
There was a problem hiding this comment.
Yeah, you're right - I was worried about a spurious tick but we have to have come out of here for the defer to work anyway.
Removed!
| case q.inputs <- qi: | ||
| case <-ctx.Done(): | ||
| return func() (tessera.Index, error) { | ||
| return tessera.Index{}, ctx.Err() | ||
| } | ||
| case <-q.queueCtx.Done(): | ||
| return func() (tessera.Index, error) { | ||
| return tessera.Index{}, q.queueCtx.Err() | ||
| } |
There was a problem hiding this comment.
If q.queueCtx is cancelled but the batching goroutine is still active, it's still possible to get an entry into the shutting-down queue.
There was a problem hiding this comment.
That's true, but only if the batching goroutine is in <-q.inputs on L111 (inputs is not buffered), so either that's impossible because it already detected the ctx cancelled, or as soon as it processes the recv and goes around the loop it'll detect the cancelled context and call cancelItems().
| { | ||
| name: "worker blocked in flush and batches full", | ||
| maxSize: 2, | ||
| numItems: 2 + 2 + 2 + 1, // 2 in doFlush (flushFunc below), 2 in batches chan, 2 blocked in flush() (on batches<-), 1 blocked in Add() (on inputs<-). |
There was a problem hiding this comment.
The test here doesn't fail (not flaky) but it is not always the same as what it describes.
The a.Add on L202 may not finish adding the entry when cancel() on L212 is called.
addWg.Go(func() {
futures[i] = q.Add(t.Context(), tessera.NewEntry(fmt.Appendf(nil, "item %d", i)))
})There was a problem hiding this comment.
Gnash, gnash. Ok should be better now.
1dfde7c to
5db83a6
Compare
2 participants
This PR tightens up the queue semantics when it's being closed down.
This should help avoid any deadlocks or ambiguity around whether leaves sent via
Add()were accepted or errored.