kgo: fix deadlock in Produce when using MaxBufferedBytes #787
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
The select has three cases; two use drainBuffered which properly added -1 to the blocked count. The first case didn't. We now do. Practically, this meant that if you ever were blocked while lingering, you would never linger again. For #726.
twmb
force-pushed
the
777
branch
5 times, most recently
from
66ce331 to
ff7ebfb
Compare
|
I've loop tested this locally,
So far I am unable to recreate what fails in CI at all. |
Copying from the issue, """ 1) Produce() record A (100 bytes) 2) Produce() record B (50 bytes), waiting for buffer to free 3) Produce() record C (50 bytes), waiting for buffer to free 4) Record A is produced, finishRecordPromise() gets called, detects it was over the limit so publish 1 message to waitBuffer 5) Record B is unlocked, finishRecordPromise() gets called, does not detect it was over the limit (only 50 bytes), so record C is never unblocked and will wait indefinitely on waitBuffer """ The fix requires adding a lock while producing. This reuses the existing lock on the `producer` type. This can lead to a few more spurious wakeups in other functions that use this same mutex, but that's fine. The prior algorithm counted anything to produce immediately into the buffered records and bytes fields; the fix for #777 could not really be possible unless we avoid counting the "buffered" aspect right away. Specifically, we need to have a goroutine looping with a sync.Cond that checks *IF* we add the record, will we still be blocked? This allows us to wake up all blocked goroutines always (unlike one at a time, the problem this issue points out), and each goroutine can check under a lock if they still do not fit. This also fixes an unreported bug where, if a record WOULD be blocked but fails early due to no topic / not in a transaction while in a transactional client, the serial promise finishing goroutine would deadlock. Closes #777.
|
Turns out the code was correct. The test needed to use I did find a small "optimization" at least, in all this looking. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Copying from the issue,
"""
"""
The fix requires adding a lock while producing. This reuses the existing
lock on the
producertype. This can lead to a few more spuriouswakeups in other functions that use this same mutex, but that's fine.
The prior algorithm counted anything to produce immediately into the
buffered records and bytes fields; the fix for #777 could not really be
possible unless we avoid counting the "buffered" aspect right away.
Specifically, we need to have a goroutine looping with a sync.Cond that
checks IF we add the record, will we still be blocked? This allows us
to wake up all blocked goroutines always (unlike one at a time, the
problem this issue points out), and each goroutine can check under a
lock if they still do not fit.
This also fixes an unreported bug where, if a record WOULD be blocked
but fails early due to no topic / not in a transaction while in a
transactional client, the serial promise finishing goroutine would
deadlock.
Closes #777.