Fix buffer_drain_concurrency not doing anything.
#14545
Conversation
Signed-off-by: Arthur Schreiber <[email protected]>
Review ChecklistHello reviewers! 👋 Please follow this checklist when reviewing this Pull Request. General
Tests
Documentation
New flags
If a workflow is added or modified:
Backward compatibility
|
vitess-bot
bot
added
NeedsDescriptionUpdate
arthurschreiber
requested review from
harshit-gangal,
systay,
frouioui and
GuptaManan100
as code owners
arthurschreiber
added
Component: Query Serving
Type: Bug
and removed
NeedsIssue
| entryChan := make(chan *entry, len(q)) | ||
|
|
||
| parallelism := min(bufferDrainConcurrency, len(q)) | ||
|
|
||
| var wg sync.WaitGroup | ||
| wg.Add(parallelism) | ||
|
|
||
| for i := 0; i < parallelism; i++ { | ||
| go func() { | ||
| for _, e := range q { | ||
| sb.unblockAndWait(e, err, true /* releaseSlot */, true /* blockingWait */) | ||
| } | ||
|
|
||
| wg.Done() | ||
| }() | ||
| } | ||
|
|
||
| for _, e := range q { | ||
| sb.unblockAndWait(e, err, true /* releaseSlot */, true /* blockingWait */) | ||
| entryChan <- e | ||
| } | ||
|
|
||
| close(entryChan) | ||
| wg.Wait() |
There was a problem hiding this comment.
Would it make sense to only pump the entries through the channel if bufferDrainConcurrency is set to a value higher than 1? Or is the overhead negligible?
There was a problem hiding this comment.
No, I think this is an important optimization. We need a legacy codepath for when bufferDrainConcurrency <= 1.
| // TODO(mberlin): Parallelize the drain by pumping the data through a channel. | ||
|
|
||
| // Parallelize the drain by pumping the data through a channel. | ||
| entryChan := make(chan *entry, len(q)) |
There was a problem hiding this comment.
Do I need to make the channel buffered? 🤔 I think it should be fine to not have it buffered, but I'm not 100% sure.
There was a problem hiding this comment.
I think this should not be buffered, definitely not buffered as len(q) because that's a lot of underlying memory being allocated!
|
@vmg Do you mind taking a look? 🙇♂️ |
| // TODO(mberlin): Parallelize the drain by pumping the data through a channel. | ||
|
|
||
| // Parallelize the drain by pumping the data through a channel. | ||
| entryChan := make(chan *entry, len(q)) |
There was a problem hiding this comment.
I think this should not be buffered, definitely not buffered as len(q) because that's a lot of underlying memory being allocated!
| entryChan := make(chan *entry, len(q)) | ||
|
|
||
| parallelism := min(bufferDrainConcurrency, len(q)) | ||
|
|
||
| var wg sync.WaitGroup | ||
| wg.Add(parallelism) | ||
|
|
||
| for i := 0; i < parallelism; i++ { | ||
| go func() { | ||
| for _, e := range q { | ||
| sb.unblockAndWait(e, err, true /* releaseSlot */, true /* blockingWait */) | ||
| } | ||
|
|
||
| wg.Done() | ||
| }() | ||
| } | ||
|
|
||
| for _, e := range q { | ||
| sb.unblockAndWait(e, err, true /* releaseSlot */, true /* blockingWait */) | ||
| entryChan <- e | ||
| } | ||
|
|
||
| close(entryChan) | ||
| wg.Wait() |
There was a problem hiding this comment.
No, I think this is an important optimization. We need a legacy codepath for when bufferDrainConcurrency <= 1.
| sb.unblockAndWait(e, err, true /* releaseSlot */, true /* blockingWait */) | ||
| } | ||
|
|
||
| wg.Done() |
There was a problem hiding this comment.
defer wg.Done() at the start of the goroutine to prevent a deadlock on panic.
| for _, e := range q { | ||
| sb.unblockAndWait(e, err, true /* releaseSlot */, true /* blockingWait */) | ||
| entryChan <- e |
There was a problem hiding this comment.
To elaborate: entryChan doesn't need to be buffered because the sending goroutine (i.e. this one) doesn't have anything else to do besides sending the entries through the channel. If it's not blocked on the channel send, it'll block on wg.Wait() a couple lines below. So buffering is not an optimization here.
|
This PR is being marked as stale because it has been open for 30 days with no activity. To rectify, you may do any of the following:
If no action is taken within 7 days, this PR will be closed. |
github-actions
bot
added
the
Stale
|
This PR was closed because it has been stale for 7 days with no activity. |
Description
As described in #11684, the
--buffer_drain_concurrencyCLI argument tovtgatedoes not actually do anything.This pull request implements the logic required to make this flag actually do something. 😬 When the buffer is drained, we now spawn as many goroutines as specified by
--buffer_drain_concurrencyto drain the buffer in parallel.I don't think introducing a new flag as described in #11684 actually makes sense - instead I propose we mention this in the v19 changelog that this flag is now doing something, and don't backport the change to any earlier releases.
Related Issue(s)
#11684
Checklist
Deployment Notes