You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by yi...@apache.org on 2023/03/10 14:32:24 UTC
[flink] branch release-1.17 updated: [FLINK-31386][network] Fix the potential deadlock issue of blocking shuffle
This is an automated email from the ASF dual-hosted git repository.
yingjie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new b90bf7acde9 [FLINK-31386][network] Fix the potential deadlock issue of blocking shuffle
b90bf7acde9 is described below
commit b90bf7acde9632a399137c20be36f12706bd75f2
Author: kevin.cyj <ke...@alibaba-inc.com>
AuthorDate: Thu Mar 9 11:52:21 2023 +0800
[FLINK-31386][network] Fix the potential deadlock issue of blocking shuffle
Currently, the SortMergeResultPartition may allocate more network buffers than the guaranteed size of the LocalBufferPool. As a result, some result partitions may need to wait other result partitions to release the over-allocated network buffers to continue. However, the result partitions which have allocated more than guaranteed buffers relies on the processing of input data to trigger data spilling and buffer recycling. The input data further relies on batch reading buffers used by [...]
This closes #22148.
---
.../io/disk/BatchShuffleReadBufferPool.java | 2 +-
.../partition/SortMergeResultPartition.java | 12 ++-
.../partition/SortMergeResultPartitionTest.java | 108 ++++++++++++++++++++-
3 files changed, 117 insertions(+), 5 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/BatchShuffleReadBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/BatchShuffleReadBufferPool.java
index fc9aa1b0ef2..863c0429400 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/BatchShuffleReadBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/BatchShuffleReadBufferPool.java
@@ -54,7 +54,7 @@ public class BatchShuffleReadBufferPool {
* Memory size in bytes can be allocated from this buffer pool for a single request (4M is for
* better sequential read).
*/
- private static final int NUM_BYTES_PER_REQUEST = 4 * 1024 * 1024;
+ public static final int NUM_BYTES_PER_REQUEST = 4 * 1024 * 1024;
/**
* Wait for at most 2 seconds before return if there is no enough available buffers currently.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java
index def80f40574..ba84f8b6e0b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartition.java
@@ -186,6 +186,11 @@ public class SortMergeResultPartition extends ResultPartition {
}
}
+ // reserve the "guaranteed" buffers for this buffer pool to avoid the case that those
+ // buffers are taken by other result partitions and can not be released, which may cause
+ // deadlock
+ requestGuaranteedBuffers();
+
// initialize the buffer pool eagerly to avoid reporting errors such as OOM too late
readBufferPool.initialize();
LOG.info("Sort-merge partition {} initialized.", getPartitionId());
@@ -325,7 +330,7 @@ public class SortMergeResultPartition extends ResultPartition {
}
}
- private void requestNetworkBuffers() throws IOException {
+ private void requestGuaranteedBuffers() throws IOException {
int numRequiredBuffer = bufferPool.getNumberOfRequiredMemorySegments();
if (numRequiredBuffer < 2) {
throw new IOException(
@@ -339,8 +344,13 @@ public class SortMergeResultPartition extends ResultPartition {
freeSegments.add(checkNotNull(bufferPool.requestMemorySegmentBlocking()));
}
} catch (InterruptedException exception) {
+ freeSegments.forEach(bufferPool::recycle);
throw new IOException("Failed to allocate buffers for result partition.", exception);
}
+ }
+
+ private void requestNetworkBuffers() throws IOException {
+ requestGuaranteedBuffers();
// avoid taking too many buffers in one result partition
while (freeSegments.size() < bufferPool.getMaxNumberOfMemorySegments()) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionTest.java
index 6f01823c3fd..c202669e9f3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SortMergeResultPartitionTest.java
@@ -50,6 +50,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Queue;
import java.util.Random;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
@@ -325,7 +326,7 @@ public class SortMergeResultPartitionTest {
BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
SortMergeResultPartition partition = createSortMergedPartition(10, bufferPool);
- assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(0);
+ assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(numBuffers);
partition.emitRecord(ByteBuffer.allocate(bufferSize * (numBuffers - 1)), 0);
partition.emitRecord(ByteBuffer.allocate(bufferSize * (numBuffers - 1)), 1);
@@ -348,7 +349,7 @@ public class SortMergeResultPartitionTest {
BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
SortMergeResultPartition partition = createSortMergedPartition(10, bufferPool);
- assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(0);
+ assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(numBuffers);
partition.emitRecord(ByteBuffer.allocate(bufferSize * (numBuffers - 1)), 0);
partition.emitRecord(ByteBuffer.allocate(bufferSize * (numBuffers - 1)), 1);
@@ -381,7 +382,7 @@ public class SortMergeResultPartitionTest {
BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
SortMergeResultPartition partition = createSortMergedPartition(10, bufferPool);
- assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(0);
+ assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(numBuffers);
partition.emitRecord(ByteBuffer.allocate(bufferSize * (numBuffers - 1)), 5);
assertThat(bufferPool.bestEffortGetNumOfUsedBuffers())
@@ -423,6 +424,107 @@ public class SortMergeResultPartitionTest {
testResultPartitionBytesCounter(true);
}
+ @TestTemplate
+ void testNetworkBufferReservation() throws IOException {
+ int numBuffers = 10;
+
+ BufferPool bufferPool = globalPool.createBufferPool(numBuffers, 2 * numBuffers);
+ SortMergeResultPartition partition = createSortMergedPartition(1, bufferPool);
+ assertThat(bufferPool.bestEffortGetNumOfUsedBuffers()).isEqualTo(numBuffers);
+
+ partition.finish();
+ partition.close();
+ }
+
+ @TestTemplate
+ void testNoDeadlockOnSpecificConsumptionOrder() throws Exception {
+ // see https://issues.apache.org/jira/browse/FLINK-31386 for more information
+ int numNetworkBuffers = 2 * BatchShuffleReadBufferPool.NUM_BYTES_PER_REQUEST / bufferSize;
+ NetworkBufferPool networkBufferPool = new NetworkBufferPool(numNetworkBuffers, bufferSize);
+ BatchShuffleReadBufferPool readBufferPool =
+ new BatchShuffleReadBufferPool(
+ BatchShuffleReadBufferPool.NUM_BYTES_PER_REQUEST, bufferSize);
+
+ BufferPool bufferPool =
+ networkBufferPool.createBufferPool(numNetworkBuffers, numNetworkBuffers);
+ SortMergeResultPartition partition =
+ createSortMergedPartition(1, bufferPool, readBufferPool);
+ for (int i = 0; i < numNetworkBuffers; ++i) {
+ partition.emitRecord(ByteBuffer.allocate(bufferSize), 0);
+ }
+ partition.finish();
+ partition.close();
+
+ CountDownLatch condition1 = new CountDownLatch(1);
+ CountDownLatch condition2 = new CountDownLatch(1);
+
+ Runnable task1 =
+ () -> {
+ try {
+ ResultSubpartitionView view = partition.createSubpartitionView(0, listener);
+ BufferPool bufferPool1 =
+ networkBufferPool.createBufferPool(
+ numNetworkBuffers / 2, numNetworkBuffers);
+ SortMergeResultPartition partition1 =
+ createSortMergedPartition(1, bufferPool1);
+ readAndEmitData(view, partition1);
+
+ condition1.countDown();
+ condition2.await();
+ readAndEmitAllData(view, partition1);
+ } catch (Exception ignored) {
+ }
+ };
+ Thread consumer1 = new Thread(task1);
+ consumer1.start();
+
+ Runnable task2 =
+ () -> {
+ try {
+ condition1.await();
+ BufferPool bufferPool2 =
+ networkBufferPool.createBufferPool(
+ numNetworkBuffers / 2, numNetworkBuffers);
+ condition2.countDown();
+
+ SortMergeResultPartition partition2 =
+ createSortMergedPartition(1, bufferPool2);
+ ResultSubpartitionView view = partition.createSubpartitionView(0, listener);
+ readAndEmitAllData(view, partition2);
+ } catch (Exception ignored) {
+ }
+ };
+ Thread consumer2 = new Thread(task2);
+ consumer2.start();
+
+ consumer1.join();
+ consumer2.join();
+ }
+
+ private boolean readAndEmitData(ResultSubpartitionView view, SortMergeResultPartition partition)
+ throws Exception {
+ MemorySegment segment = MemorySegmentFactory.allocateUnpooledSegment(bufferSize);
+ ResultSubpartition.BufferAndBacklog buffer;
+ do {
+ buffer = view.getNextBuffer();
+ if (buffer != null) {
+ Buffer data = ((CompositeBuffer) buffer.buffer()).getFullBufferData(segment);
+ partition.emitRecord(data.getNioBufferReadable(), 0);
+ if (!data.isRecycled()) {
+ data.recycleBuffer();
+ }
+ return buffer.buffer().isBuffer();
+ }
+ } while (true);
+ }
+
+ private void readAndEmitAllData(ResultSubpartitionView view, SortMergeResultPartition partition)
+ throws Exception {
+ while (readAndEmitData(view, partition)) {}
+ partition.finish();
+ partition.close();
+ }
+
private void testResultPartitionBytesCounter(boolean isBroadcast) throws IOException {
int numBuffers = useHashDataBuffer ? 100 : 15;
int numSubpartitions = 2;