pnowojski commented on code in PR #27783:
URL: https://github.com/apache/flink/pull/27783#discussion_r3008196005
##########
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:
Why/how buffer can be already recycled here? 🤔 Is there a valid code path
that recycles this buffer before we get here? 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.
So I think still the ownership model is not 100% clear here.
##########
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:
Can you explain how does it solve a potential deadlock and how much extra
buffers does it require and maybe add a todo about spilling?
Especially, that you have two places on the input requesting buffers. Here
and `recoverWithFiltering`. If `recoverWithFiltering` is downstream, but here
you can use unpooled memory here, that would still lead to a deadlock: you
exhaust the pool downstream, but here you keep creating unpooled buffers that
you try to feed to blocked downstream. I'm not sure in which order buffers are
requested 🤔
Do you maybe have a test that checks what happens when buffer pool is
exhausted? A test that previously without this unpooled allocation was
deadlocking, and now it's working?
--
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]
