pq: reduce read contention when caught up #17765
Conversation
By keeping track of the lock-holding reader's demand (a quantity of events and a deadline), we can avoid waking up the reader for each event that is written to the queue, significantly reducing the time that the lock-holding _writer_ is blocked by synchronous read operations. The lock-holding reader is awoken when its demand is met, has expired, or when the writer has rotated pages. | batch size | unpatched | this patch | | ---------- | --------- | ---------- | | 125 | 260k | 498k | | 2000 | 219k | 516k |
🤖 GitHub commentsExpand to view the GitHub comments
Just comment with:
|
|
This pull request does not have a backport label. Could you fix it @yaauie? 🙏
|
|
💚 Build Succeeded
|
| @@ -64,6 +66,9 @@ public final class Queue implements Closeable { | |||
|
|
|||
| protected volatile long unreadCount; | |||
|
|
|||
| // the readDemand is a record of the currently-waiting-reader's demand and expiry | |||
| private volatile ReadDemand readDemand; | |||
There was a problem hiding this comment.
The ReadDemand is operated inside the lock.lock() either in writer or readers. The value is promised to be up-to-date, so I think volatile can be removed.
| this.readDemand = new ReadDemand(deadlineMillis, elementsNeeded); | ||
|
|
||
| boolean unElapsed = this.notEmpty.awaitUntil(new Date(deadlineMillis)); | ||
| this.readDemand = null; |
There was a problem hiding this comment.
I don't have big concern but wanna clarify the behaviour.
Reader releases the lock when notEmpty.awaitUntil() is called and do the waiting. Potentially, other readers could acquire the lock and update this.readDemand to a further delay deadlineMillis. Updating this.readDemand is not thread-safe and I believe it is not required to be. The worst case is the deadline extends to (timeout 50ms x number of threads).
Is my understanding correct? If so, can we add a comment to remind that the action is not atomic?
There was a problem hiding this comment.
Other readers cannot acquire the lock because Queue#readSerializedBatch and Queue#nonBlockReadBatch (the only places where a reader can acquire the queue's lock) are synchronized.
[EDIT: I believe that Queue#nonBlockReadBatch is only referenced from tests; I cannot find other references to this, or to non_block_read_batch that could be called from the ruby-side.]
Release notes
What does this PR do?
Eliminates per-event overhead from workers consuming from a persisted queue.
By keeping track of the lock-holding reader's demand (a quantity of events and a deadline), we can avoid waking up the reader for each event that is written to the queue, significantly reducing the time that the lock-holding writer is blocked by synchronous read operations. The lock-holding reader is awoken when its demand is met, has expired, or when the writer has rotated pages.
Why is it important/What is the impact to the user?
Significantly improves PQ throughput.
From my local tests (ARM / M3 Pro 12 CPU / SSD) with
input { java_generator {} } output { sink {} }:Checklist
[ ] I have made corresponding changes to the documentation[ ] I have made corresponding change to the default configuration files (and/or docker env variables)[ ] I have added tests that prove my fix is effective or that my feature worksAuthor's Checklist
How to test this PR locally
Related issues