You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2024/06/24 07:42:13 UTC
(flink) 03/06: Revert "[FLINK-33668][runtime] Make the SortBuffer be able to return unused memory segments"
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit af1c8605d4cc294fb01ad13f13d662417b8a0893
Author: Yuxin Tan <ta...@gmail.com>
AuthorDate: Fri Jun 21 13:47:14 2024 +0800
Revert "[FLINK-33668][runtime] Make the SortBuffer be able to return unused memory segments"
This reverts commit ac2ec3496400b594ffe5c711bc1b6343999a8479.
---
.../runtime/io/network/partition/SortBuffer.java | 23 +-----------------
.../tiered/storage/SortBufferAccumulator.java | 16 +------------
.../storage/TieredStorageSortBufferTest.java | 28 ----------------------
3 files changed, 2 insertions(+), 65 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortBuffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortBuffer.java
index d10383cc0d6..371484696a6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortBuffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortBuffer.java
@@ -76,7 +76,7 @@ public abstract class SortBuffer implements DataBuffer {
private final int bufferSize;
/** Number of guaranteed buffers can be allocated from the buffer pool for data sort. */
- private int numGuaranteedBuffers;
+ private final int numGuaranteedBuffers;
// ---------------------------------------------------------------------------------------------
// Statistics and states
@@ -184,27 +184,6 @@ public abstract class SortBuffer implements DataBuffer {
return false;
}
- /**
- * Try to release some unused memory segments.
- *
- * <p>Note that this class is not thread safe, so please make sure to call {@link
- * #append(ByteBuffer source, int targetSubpartition, Buffer.DataType dataType)} and this method
- * with lock acquired.
- *
- * @param numFreeSegments the number of segments to be released.
- * @return true if released successfully, otherwise false.
- */
- public boolean returnFreeSegments(int numFreeSegments) {
- if (numFreeSegments < numGuaranteedBuffers - segments.size()) {
- for (int i = 0; i < numFreeSegments; i++) {
- bufferRecycler.recycle(freeSegments.poll());
- }
- numGuaranteedBuffers -= numFreeSegments;
- return true;
- }
- return false;
- }
-
private void writeIndex(int subpartitionIndex, int numRecordBytes, Buffer.DataType dataType) {
MemorySegment segment = segments.get(writeSegmentIndex);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SortBufferAccumulator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SortBufferAccumulator.java
index 2f5c469e33a..8dc6eb2d69f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SortBufferAccumulator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SortBufferAccumulator.java
@@ -151,9 +151,7 @@ public class SortBufferAccumulator implements BufferAccumulator {
int newSize = this.memoryManager.getBufferPoolSize();
int oldSize = poolSize.getAndSet(newSize);
if (oldSize > newSize) {
- if (!returnFreeSegments(oldSize - newSize)) {
- flushCurrentDataBuffer();
- }
+ flushCurrentDataBuffer();
}
},
poolSizeCheckInterval,
@@ -298,18 +296,6 @@ public class SortBufferAccumulator implements BufferAccumulator {
}
}
- private boolean returnFreeSegments(int numSegments) {
- synchronized (lock) {
- if (currentDataBuffer == null
- || currentDataBuffer.isReleased()
- || !currentDataBuffer.hasRemaining()) {
- return false;
- } else {
- return currentDataBuffer.returnFreeSegments(numSegments);
- }
- }
- }
-
private void writeLargeRecord(ByteBuffer record, int subpartitionId, Buffer.DataType dataType) {
checkState(dataType != Buffer.DataType.EVENT_BUFFER);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageSortBufferTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageSortBufferTest.java
index 2e14969812d..b0ab84f86b0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageSortBufferTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageSortBufferTest.java
@@ -29,7 +29,6 @@ import org.apache.flink.runtime.io.network.partition.SortBuffer;
import org.junit.jupiter.api.Test;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.ArrayList;
@@ -195,33 +194,6 @@ class TieredStorageSortBufferTest {
assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(numBuffersForSort);
}
- @Test
- void testReturnFreeSegments() throws IOException, InterruptedException {
- final int bufferPoolSize = 512;
- final int numBuffersForSort = 20;
- NetworkBufferPool globalPool = new NetworkBufferPool(bufferPoolSize, BUFFER_SIZE_BYTES);
- BufferPool bufferPool =
- globalPool.createBufferPool(bufferPoolSize, bufferPoolSize, bufferPoolSize);
-
- LinkedList<MemorySegment> segments = new LinkedList<>();
- for (int i = 0; i < numBuffersForSort; ++i) {
- segments.add(bufferPool.requestMemorySegmentBlocking());
- }
- TieredStorageSortBuffer sortBuffer =
- new TieredStorageSortBuffer(
- segments, bufferPool, 1, BUFFER_SIZE_BYTES, numBuffersForSort, true);
-
- for (int i = 0; i < 5; i++) {
- byte[] bytes = new byte[BUFFER_SIZE_BYTES];
- ByteBuffer record = ByteBuffer.wrap(bytes);
- sortBuffer.append(record, 0, Buffer.DataType.DATA_BUFFER);
- }
- assertThat(sortBuffer.returnFreeSegments(10)).isTrue();
- assertThat(sortBuffer.returnFreeSegments(10)).isFalse();
-
- sortBuffer.finish();
- }
-
private static BufferWithSubpartition copyIntoSegment(SortBuffer dataBuffer) {
MemorySegment segment = MemorySegmentFactory.allocateUnpooledSegment(BUFFER_SIZE_BYTES);
return dataBuffer.getNextBuffer(segment);