-
Notifications
You must be signed in to change notification settings - Fork 804
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
pubsub: Ensure batcher flushes on shutdown, even if min batch size isn't met #3386
base: master
Are you sure you want to change the base?
Conversation
…n't met This PR ensures that the batcher flushes on shutdown, even if the pending length is less than the min batch size specified. Sending events is preferred to dropping, even if limits are not obeyed.
The tests are failing with a data race: WARNING: DATA RACE Previous read at 0x00c00044e3f0 by goroutine 979: Goroutine 978 (running) created at: |
Ah, thanks. I ran without -race 🤦. Fixing! |
@vangent fixed, thanks for the notification. |
|
||
// On shutdown, ensure that we attempt to flush any pending items | ||
// if there's a minimum batch size. | ||
if atomic.LoadInt32(&b.nHandlers) < int32(b.opts.MaxHandlers) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It shouldn't be necessary to use atomic
to check/modify nHandlers. There's a mutex on the struct to protect it. IIUC, the problem is that you added a read on this line outside of the lock. I think if you move the b.mu.Unlock a few lines up to below this new stanza, it will be fine.
|
||
// On shutdown, ensure that we attempt to flush any pending items | ||
// if there's a minimum batch size. | ||
if atomic.LoadInt32(&b.nHandlers) < int32(b.opts.MaxHandlers) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What happens if nHandlers == MaxHandlers? Won't we drop some messages then?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
At this point, nextBatch
in the handlers call will return the remaining items as shutdown
is set to true, so everything will be processed as expected.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If nHandlers == MaxHandlers, nextBatch
won't even be called...?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's not obvious to me that we should be checking nHandlers
here at all. Seems like during shutdown we need to choose between possibly creating more than MaxHandlers handlers, or dropping messages.
This PR ensures that the batcher flushes on shutdown, even if the pending length is less than the min batch size specified. Sending events is preferred to dropping, even if limits are not obeyed.
Related to #3383, but doesn't necessarily replace.