You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2022/05/31 07:34:04 UTC
[flink] branch master updated: [FLINK-27522][network] Ignore max buffers per channel when allocate buffer
This is an automated email from the ASF dual-hosted git repository.
pnowojski 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 91462206889 [FLINK-27522][network] Ignore max buffers per channel when allocate buffer
91462206889 is described below
commit 914622068892ee2c5bd60532f3cabec1f581262d
Author: fanrui <19...@gmail.com>
AuthorDate: Sat May 28 15:40:07 2022 +0800
[FLINK-27522][network] Ignore max buffers per channel when allocate buffer
---
.../flink/runtime/io/network/buffer/LocalBufferPool.java | 6 ------
.../flink/runtime/io/network/buffer/LocalBufferPoolTest.java | 10 ++++++++--
2 files changed, 8 insertions(+), 8 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
index 4936670e09a..0d5068c02f3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
@@ -372,12 +372,6 @@ class LocalBufferPool implements BufferPool {
synchronized (availableMemorySegments) {
checkDestroyed();
- // target channel over quota; do not return a segment
- if (targetChannel != UNKNOWN_CHANNEL
- && subpartitionBuffersCount[targetChannel] >= maxBuffersPerChannel) {
- return null;
- }
-
segment = availableMemorySegments.poll();
if (segment == null) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
index 56e1befddda..7401b5c6e9e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
@@ -453,7 +453,11 @@ public class LocalBufferPoolTest extends TestLogger {
final BufferBuilder bufferBuilder02 = localBufferPool.requestBufferBuilderBlocking(0);
assertFalse(localBufferPool.getAvailableFuture().isDone());
- assertNull(localBufferPool.requestBufferBuilder(0));
+ // When this channel reaches maxBuffersPerChannel, LocalBufferPool will be unavailable. But
+ // when a memory segment is requested from LocalBufferPool and availableMemorySegments isn't
+ // empty, maxBuffersPerChannel will be ignored and buffers will continue to be allocated.
+ // Check FLINK-27522 for more information.
+ final BufferBuilder bufferBuilder03 = localBufferPool.requestBufferBuilderBlocking(0);
final BufferBuilder bufferBuilder21 = localBufferPool.requestBufferBuilderBlocking(2);
final BufferBuilder bufferBuilder22 = localBufferPool.requestBufferBuilderBlocking(2);
assertFalse(localBufferPool.getAvailableFuture().isDone());
@@ -464,9 +468,11 @@ public class LocalBufferPoolTest extends TestLogger {
bufferBuilder21.close();
assertFalse(localBufferPool.getAvailableFuture().isDone());
bufferBuilder02.close();
- assertTrue(localBufferPool.getAvailableFuture().isDone());
+ assertFalse(localBufferPool.getAvailableFuture().isDone());
bufferBuilder01.close();
assertTrue(localBufferPool.getAvailableFuture().isDone());
+ bufferBuilder03.close();
+ assertTrue(localBufferPool.getAvailableFuture().isDone());
bufferBuilder22.close();
assertTrue(localBufferPool.getAvailableFuture().isDone());
}