-
Notifications
You must be signed in to change notification settings - Fork 833
pubsub: Ensure batcher flushes on shutdown, even if min batch size isn't met#3386
pubsub: Ensure batcher flushes on shutdown, even if min batch size isn't met#3386tonyhb wants to merge 2 commits intogoogle:masterfrom
Conversation
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! |
|
|
||
| // 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.
|
Ping? |