reswqa commented on code in PR #22084:
URL: https://github.com/apache/flink/pull/22084#discussion_r1128947802
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java:
##########
@@ -242,23 +243,206 @@ public void testRecycleAfterDestroy() {
localBufferPool.lazyDestroy();
// All buffers have been requested, but can not be returned yet.
- assertEquals(numBuffers, getNumRequestedFromMemorySegmentPool());
+
assertThat(getNumRequestedFromMemorySegmentPool()).isEqualTo(numBuffers);
// Recycle should return buffers to memory segment pool
for (Buffer buffer : requests) {
buffer.recycleBuffer();
}
}
+ @Test
+ void testDecreasePoolSize() throws Exception {
+ final int maxMemorySegments = 10;
+ final int requiredMemorySegments = 4;
+ final int maxOverdraftBuffers = 2;
+ final int largePoolSize = 5;
+ final int smallPoolSize = 4;
+ LocalBufferPool bufferPool =
+ new LocalBufferPool(
+ networkBufferPool,
+ requiredMemorySegments,
+ maxMemorySegments,
+ 0,
+ Integer.MAX_VALUE,
+ maxOverdraftBuffers);
+ Queue<MemorySegment> buffers = new LinkedList<>();
+
+ // set a larger pool size.
+ bufferPool.setNumBuffers(largePoolSize);
+ assertThat(bufferPool.getNumBuffers()).isEqualTo(largePoolSize);
+
+ // request all buffer.
+ for (int i = 0; i < largePoolSize; i++) {
+ buffers.add(bufferPool.requestMemorySegmentBlocking());
+ }
+ assertThat(bufferPool.isAvailable()).isFalse();
+
+ // request 1 overdraft buffers.
+ buffers.add(bufferPool.requestMemorySegmentBlocking());
+
assertThat(bufferPool.getNumberOfRequestedOverdraftMemorySegments()).isOne();
+ assertThat(bufferPool.isAvailable()).isFalse();
+
+ // set a small pool size.
+ bufferPool.setNumBuffers(smallPoolSize);
+ assertThat(bufferPool.getNumBuffers()).isEqualTo(smallPoolSize);
+ assertThat(bufferPool.getNumberOfAvailableMemorySegments()).isZero();
+
assertThat(bufferPool.getNumberOfRequestedOverdraftMemorySegments()).isOne();
+ assertThat(bufferPool.isAvailable()).isFalse();
+ buffers.add(bufferPool.requestMemorySegmentBlocking());
+
assertThat(bufferPool.getNumberOfRequestedOverdraftMemorySegments()).isEqualTo(2);
+ assertThat(bufferPool.isAvailable()).isFalse();
+
+ // return all overdraft buffers.
+ bufferPool.recycle(buffers.poll());
+
assertThat(bufferPool.getNumberOfRequestedOverdraftMemorySegments()).isOne();
+ assertThat(bufferPool.isAvailable()).isFalse();
+ bufferPool.recycle(buffers.poll());
+
assertThat(bufferPool.getNumberOfRequestedOverdraftMemorySegments()).isZero();
+ assertThat(bufferPool.isAvailable()).isFalse();
+
+ // return the excess buffer.
+ bufferPool.recycle(buffers.poll());
+ assertThat(bufferPool.isAvailable()).isFalse();
+ // return non-excess buffers.
+ bufferPool.recycle(buffers.poll());
+ assertThat(bufferPool.getNumberOfAvailableMemorySegments()).isOne();
+ assertThat(bufferPool.isAvailable()).isTrue();
+
+ while (!buffers.isEmpty()) {
+ bufferPool.recycle(buffers.poll());
+ }
+ bufferPool.lazyDestroy();
+ }
+
+ @Test
+ void testIncreasePoolSize() throws Exception {
Review Comment:
Agree with you, fixed.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
