1996fanrui commented on code in PR #27783:
URL: https://github.com/apache/flink/pull/27783#discussion_r3008450979
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java:
##########
@@ -281,7 +286,18 @@ public Buffer requestBufferBlocking() throws
InterruptedException, IOException {
bufferManager.requestExclusiveBuffers(networkBuffersPerChannel);
exclusiveBuffersAssigned = true;
}
- return bufferManager.requestBufferBlocking();
+ if (!inputGate.isCheckpointingDuringRecoveryEnabled()) {
+ return bufferManager.requestBufferBlocking();
+ }
+ // When checkpointing during recovery is enabled, use a heap buffer as
fallback
+ // to avoid blocking if the buffer pool is insufficient during
recovery.
+ Buffer buffer = bufferManager.requestBuffer();
+ if (buffer != null) {
+ return buffer;
+ }
+ MemorySegment memorySegment =
+
MemorySegmentFactory.allocateUnpooledSegment(MemoryManager.DEFAULT_PAGE_SIZE);
+ return new NetworkBuffer(memorySegment,
FreeingBufferRecycler.INSTANCE);
Review Comment:
`allocateUnpooledSegment` requests a heap buffer, so it is unlimited buffer
until OOM, that is why it could avoid deadlock.
I have improved the comment and added a todo
> TODO(FLINK-38544): replace heap fallback with disk spilling to bound
memory usage.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java:
##########
@@ -125,7 +147,9 @@ public void recover(
}
}
} finally {
- buffer.recycleBuffer();
+ if (!buffer.isRecycled()) {
+ buffer.recycleBuffer();
+ }
Review Comment:
https://github.com/apache/flink/pull/27783/changes#r3008317692
Hey, the recover method comment mentioned it. This approach unified all code
paths.
> If so, this I think asks for a bug, because if you pass a buffer with two
reference counts to this method, that code path would recycle buffer twice
inside this method.
You are right, it has potential bug for general codebase. In theory, it
won't happen since recovery codebase only has one reference.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
