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
commit 6265c89af5dda97832ede37be618b124934f1d74 Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 00:21:59 2026 +0200 [FLINK-39521][network] BufferManager: gate credit notification behind notifyInitiallyEnabled Add a notifyInitiallyEnabled constructor parameter to BufferManager and a new enableNotify() method: while notification is disabled, requested and recycled floating buffers are queued but never announced to the producer as credit; enableNotify() opens the gate and announces the queued buffers atomically with respect to concurrent recycle/floating-buffer callbacks. All existing callers (RecoveredInputChannel, RemoteInputChannel) pass true, so behavior is unchanged. The disabled mode is used by a later commit of this PR, where a remote channel created in recovery state withholds credit until its recovered state has been consumed. --- .../network/partition/consumer/BufferManager.java | 44 ++++++++++++++++++---- .../partition/consumer/RecoveredInputChannel.java | 2 +- .../partition/consumer/RemoteInputChannel.java | 3 +- 3 files changed, 39 insertions(+), 10 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java index db38025def9..1eed2a0284f 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/BufferManager.java @@ -70,13 +70,24 @@ public class BufferManager implements BufferListener, BufferRecycler { @GuardedBy("bufferQueue") private int numRequiredBuffers; + /** + * Gates credit announcements while a recovery drain borrows this channel's buffers. Kept under + * {@code bufferQueue} to avoid inverting the queue/recovered-buffer lock order. + */ + @GuardedBy("bufferQueue") + private boolean notifyAvailable; + public BufferManager( - MemorySegmentProvider globalPool, InputChannel inputChannel, int numRequiredBuffers) { + MemorySegmentProvider globalPool, + InputChannel inputChannel, + int numRequiredBuffers, + boolean notifyInitiallyEnabled) { this.globalPool = checkNotNull(globalPool); this.inputChannel = checkNotNull(inputChannel); checkArgument(numRequiredBuffers >= 0); this.numRequiredBuffers = numRequiredBuffers; + this.notifyAvailable = notifyInitiallyEnabled; } // ------------------------------------------------------------------------ @@ -158,23 +169,22 @@ public class BufferManager implements BufferListener, BufferRecycler { /** * Requests floating buffers from the buffer pool based on the given required amount, and - * returns the actual requested amount. If the required amount is not fully satisfied, it will - * register as a listener. + * returns the number of buffers that may be announced to the producer as credit. During + * recovery, requested buffers are queued but announced only by {@link #enableNotify()}. */ int requestFloatingBuffers(int numRequired) { - int numRequestedBuffers = 0; synchronized (bufferQueue) { // Similar to notifyBufferAvailable(), make sure that we never add a buffer after // channel // released all buffers via releaseAllResources(). if (inputChannel.isReleased()) { - return numRequestedBuffers; + return 0; } numRequiredBuffers = numRequired; - numRequestedBuffers = tryRequestBuffers(); + int numRequestedBuffers = tryRequestBuffers(); + return notifyAvailable ? numRequestedBuffers : 0; } - return numRequestedBuffers; } private int tryRequestBuffers() { @@ -209,6 +219,7 @@ public class BufferManager implements BufferListener, BufferRecycler { @Override public void recycle(MemorySegment segment) { @Nullable Buffer releasedFloatingBuffer = null; + boolean announceCredit = false; synchronized (bufferQueue) { try { // Similar to notifyBufferAvailable(), make sure that we never add a buffer @@ -226,11 +237,12 @@ public class BufferManager implements BufferListener, BufferRecycler { } finally { bufferQueue.notifyAll(); } + announceCredit = releasedFloatingBuffer == null && notifyAvailable; } if (releasedFloatingBuffer != null) { releasedFloatingBuffer.recycleBuffer(); - } else { + } else if (announceCredit) { try { inputChannel.notifyBufferAvailable(1); } catch (Throwable t) { @@ -344,6 +356,9 @@ public class BufferManager implements BufferListener, BufferRecycler { isBufferUsed = true; numBuffers += 1 + tryRequestBuffers(); bufferQueue.notifyAll(); + if (!notifyAvailable) { + numBuffers = 0; + } } inputChannel.notifyBufferAvailable(numBuffers); @@ -359,6 +374,19 @@ public class BufferManager implements BufferListener, BufferRecycler { // Nothing to do actually. } + /** + * Opens the recovery credit gate and announces the queued buffers atomically with respect to + * concurrent recycle/floating-buffer callbacks. + */ + void enableNotify() throws IOException { + int available; + synchronized (bufferQueue) { + notifyAvailable = true; + available = bufferQueue.getAvailableBufferSize(); + } + inputChannel.notifyBufferAvailable(available); + } + // ------------------------------------------------------------------------ // Getter properties // ------------------------------------------------------------------------ diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java index b68ae24c8c5..ed3bd0822a5 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java @@ -105,7 +105,7 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan numBytesIn, numBuffersIn); - bufferManager = new BufferManager(inputGate.getMemorySegmentProvider(), this, 0); + bufferManager = new BufferManager(inputGate.getMemorySegmentProvider(), this, 0, true); this.networkBuffersPerChannel = networkBuffersPerChannel; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java index 645446120d0..227bc44ca05 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java @@ -156,7 +156,8 @@ public class RemoteInputChannel extends InputChannel { this.initialCredit = networkBuffersPerChannel; this.connectionId = checkNotNull(connectionId); this.connectionManager = checkNotNull(connectionManager); - this.bufferManager = new BufferManager(inputGate.getMemorySegmentProvider(), this, 0); + this.bufferManager = + new BufferManager(inputGate.getMemorySegmentProvider(), this, 0, true); this.channelStatePersister = new ChannelStatePersister(stateWriter, getChannelInfo()); // Migrate recovered buffers from RecoveredInputChannel if provided.
