This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9bd05c557c7 [FLINK-40524][network] Skip the recoveredBuffers monitor
on the getNextBuffer hot path for channels that never recover
9bd05c557c7 is described below
commit 9bd05c557c74c21e68faa6f5a11e47c6c513c025
Author: Rui Fan <[email protected]>
AuthorDate: Mon Aug 31 18:57:52 2026 +0200
[FLINK-40524][network] Skip the recoveredBuffers monitor on the
getNextBuffer hot path for channels that never recover
---
.../partition/consumer/LocalInputChannel.java | 45 +++++++++++-----------
1 file changed, 23 insertions(+), 22 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
index fb5870a3471..05b4874cb4a 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
@@ -539,31 +539,32 @@ public class LocalInputChannel extends InputChannel
public Optional<BufferAndAvailability> getNextBuffer() throws IOException {
checkError();
- // Read inRecovery and poll the recovered buffer under a single lock
acquisition to avoid
- // grabbing the monitor twice on the hot path.
- boolean inRecovery;
- Buffer recoveredBuf = null;
- synchronized (recoveredBuffers) {
- inRecovery = this.inRecovery;
- if (inRecovery && !hasPendingPriorityEvent &&
!recoveredBuffers.isEmpty()) {
- recoveredBuf = recoveredBuffers.poll();
+ if (needsRecovery) {
+ // Read inRecovery and poll the recovered buffer under a single
lock acquisition to
+ // avoid grabbing the monitor twice on the hot path.
+ boolean inRecovery;
+ Buffer recoveredBuf = null;
+ synchronized (recoveredBuffers) {
+ inRecovery = this.inRecovery;
+ if (inRecovery && !hasPendingPriorityEvent &&
!recoveredBuffers.isEmpty()) {
+ recoveredBuf = recoveredBuffers.poll();
+ }
}
- }
- if (inRecovery) {
- // Always return an already-polled recovered buffer first:
hasPendingPriorityEvent may
- // be flipped to true by a concurrent notifyPriorityEvent() after
the poll, and
- // re-reading
- // it here would otherwise drop this buffer. A pending priority
event is served on the
- // next getNextBuffer() call instead.
- if (recoveredBuf != null) {
- return wrapRecoveredBufferAsAvailability(recoveredBuf);
- }
- if (hasPendingPriorityEvent) {
- return pullPriorityFromSubpartitionView();
+ if (inRecovery) {
+ // Always return an already-polled recovered buffer first:
hasPendingPriorityEvent
+ // may be flipped to true by a concurrent
notifyPriorityEvent() after the poll, and
+ // re-reading it here would otherwise drop this buffer. A
pending priority event is
+ // served on the next getNextBuffer() call instead.
+ if (recoveredBuf != null) {
+ return wrapRecoveredBufferAsAvailability(recoveredBuf);
+ }
+ if (hasPendingPriorityEvent) {
+ return pullPriorityFromSubpartitionView();
+ }
+ // Drain not finished yet; block normal upstream data until
delivery completes.
+ return Optional.empty();
}
- // Drain not finished yet; block normal upstream data until
delivery completes.
- return Optional.empty();
}
if (!toBeConsumedBuffers.isEmpty()) {
