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 2024/10/25 14:31:10 UTC
(flink) branch master updated (aaa8ad48f8e -> d488b7a989d)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from aaa8ad48f8e [FLINK-36593] Update io.airlift:aircompressor to 0.27. This closes #25567
new 36d9edc4222 [FLINK-36556] Add starting buffer size config option
new 63cf67e433d [hotfix] Move BufferDebloatConfiguration's checkState's into constructor
new d488b7a989d [FLINK-36555] Guarantee buffer grows even with very small alpha
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../generated/all_taskmanager_section.html | 6 ++++
.../task_manager_memory_configuration.html | 6 ++++
.../flink/configuration/TaskManagerOptions.java | 10 ++++++
.../io/network/NettyShuffleServiceFactory.java | 1 +
.../PipelinedApproximateSubpartition.java | 7 ++--
.../network/partition/PipelinedSubpartition.java | 8 +++--
.../network/partition/ResultPartitionFactory.java | 14 ++++++--
.../partition/consumer/SingleInputGateFactory.java | 1 +
.../NettyShuffleEnvironmentConfiguration.java | 18 ++++++++++
.../throughput/BufferDebloatConfiguration.java | 28 +++++++++++----
.../flink/runtime/throughput/BufferDebloater.java | 26 ++++++++++++--
.../flink/runtime/throughput/BufferSizeEMA.java | 16 ++++++---
.../io/network/NettyShuffleEnvironmentBuilder.java | 8 +++++
.../io/network/NettyShuffleEnvironmentTest.java | 5 ++-
...nedApproximateSubpartitionWithReadViewTest.java | 3 +-
.../partition/PipelinedSubpartitionTest.java | 20 +++++++++--
.../PipelinedSubpartitionWithReadViewTest.java | 2 +-
.../network/partition/ResultPartitionBuilder.java | 1 +
.../partition/ResultPartitionFactoryTest.java | 1 +
.../runtime/throughput/BufferDebloaterTest.java | 2 +-
.../runtime/throughput/BufferSizeEMATest.java | 41 ++++++++++++++++------
21 files changed, 186 insertions(+), 38 deletions(-)
(flink) 03/03: [FLINK-36555] Guarantee buffer grows even with very small alpha
Posted by pn...@apache.org.
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
commit d488b7a989d2e812ab86f568060c399e63a425ac
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Aug 30 14:42:01 2024 +0200
[FLINK-36555] Guarantee buffer grows even with very small alpha
With small buffer sizes and/or small EMA's alpha, previously
it could happen buffer size won't grow at all, because for example:
- current buffer size is 16b
- max new desired size is only capped at 2 * 16b = 32b
- so with alpha ~= 0.03 (or smaller) the new buffer size would be still 16b
---
.../flink/runtime/throughput/BufferSizeEMA.java | 9 ++---
.../runtime/throughput/BufferDebloaterTest.java | 2 +-
.../runtime/throughput/BufferSizeEMATest.java | 41 ++++++++++++++++------
3 files changed, 36 insertions(+), 16 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferSizeEMA.java b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferSizeEMA.java
index b4bdbe47db2..f1a6a9e1bf2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferSizeEMA.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferSizeEMA.java
@@ -27,7 +27,7 @@ public class BufferSizeEMA {
/** EMA algorithm specific constant which responsible for speed of reaction. */
private final double alpha;
- private int lastBufferSize;
+ private double lastBufferSize;
public BufferSizeEMA(int maxBufferSize, int minBufferSize, long numberOfSamples) {
this(maxBufferSize, maxBufferSize, minBufferSize, numberOfSamples);
@@ -64,10 +64,11 @@ public class BufferSizeEMA {
// Example of change speed:
// growing = 32768, 29647, 26823, 24268, 21956, 19864
// shrinking = 19864, 21755, 23826, 26095, 28580, 31301, 32768
- long desirableBufferSize =
- Math.min(totalBufferSizeInBytes / totalBuffers, 2L * lastBufferSize);
+ double desirableBufferSize =
+ Math.min(((double) totalBufferSizeInBytes) / totalBuffers, 2L * lastBufferSize);
lastBufferSize += alpha * (desirableBufferSize - lastBufferSize);
- return lastBufferSize = Math.max(minBufferSize, Math.min(lastBufferSize, maxBufferSize));
+ lastBufferSize = Math.max(minBufferSize, Math.min(lastBufferSize, maxBufferSize));
+ return (int) Math.round(lastBufferSize);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferDebloaterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferDebloaterTest.java
index 0f49af792dc..8cd15f44bcd 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferDebloaterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferDebloaterTest.java
@@ -46,7 +46,7 @@ class BufferDebloaterTest {
.withBufferSize(50, 1100)
.withNumberOfBuffersInUse(16)
.withThroughput(3333)
- .expectBufferSize(249);
+ .expectBufferSize(250);
}
@Test
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferSizeEMATest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferSizeEMATest.java
index 365448e1155..cccf300e0b6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferSizeEMATest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/throughput/BufferSizeEMATest.java
@@ -32,21 +32,40 @@ class BufferSizeEMATest {
// The result value seeks to the bottom limit but it will take a while until it reaches it.
assertThat(calculator.calculateBufferSize(111, 13)).isEqualTo(104);
- assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(59);
- assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(37);
+ assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(60);
+ assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(38);
assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(26);
- assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(20);
- assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(17);
- assertThat(calculator.calculateBufferSize(107, 13)).isEqualTo(12);
+ assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(21);
+ assertThat(calculator.calculateBufferSize(107, 7)).isEqualTo(18);
+ assertThat(calculator.calculateBufferSize(107, 13)).isEqualTo(13);
+ assertThat(calculator.calculateBufferSize(107, 13)).isEqualTo(11);
assertThat(calculator.calculateBufferSize(107, 13)).isEqualTo(10);
// Upgrade
assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(15);
- assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(22);
- assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(33);
- assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(49);
- assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(73);
- assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(109);
+ assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(23);
+ assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(34);
+ assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(51);
+ assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(76);
+ assertThat(calculator.calculateBufferSize(333, 1)).isEqualTo(114);
+ }
+
+ /**
+ * Check that even with very small alpha, buffer size doesn't get stuck and can not eventually
+ * grow.
+ */
+ @Test
+ public void testSmallBufferSizeChanges() {
+ int numberOfSamples = 10000;
+ int startingBufferSize = 16;
+ BufferSizeEMA calculator =
+ new BufferSizeEMA(startingBufferSize, 32 * 1024, 16, numberOfSamples);
+
+ for (int i = 0; i < numberOfSamples; i++) {
+ calculator.calculateBufferSize(20000, 10);
+ }
+
+ assertThat(calculator.calculateBufferSize(10000, 10)).isGreaterThan(startingBufferSize);
}
@Test
@@ -70,7 +89,7 @@ class BufferSizeEMATest {
assertThat(calculator.calculateBufferSize(0, 1)).isEqualTo(100);
assertThat(calculator.calculateBufferSize(0, 1)).isEqualTo(50);
assertThat(calculator.calculateBufferSize(0, 1)).isEqualTo(25);
- assertThat(calculator.calculateBufferSize(0, 1)).isEqualTo(12);
+ assertThat(calculator.calculateBufferSize(0, 1)).isEqualTo(13);
assertThat(calculator.calculateBufferSize(0, 1)).isEqualTo(10);
assertThat(calculator.calculateBufferSize(0, 1)).isEqualTo(10);
}
(flink) 02/03: [hotfix] Move BufferDebloatConfiguration's checkState's into constructor
Posted by pn...@apache.org.
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
commit 63cf67e433da46c57282ce78a0a9fd2dc93fa4ae
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Sep 2 18:25:45 2024 +0200
[hotfix] Move BufferDebloatConfiguration's checkState's into constructor
---
.../throughput/BufferDebloatConfiguration.java | 19 ++++++++++---------
1 file changed, 10 insertions(+), 9 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloatConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloatConfiguration.java
index 35b6c545586..2a8203aed65 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloatConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloatConfiguration.java
@@ -47,6 +47,16 @@ public final class BufferDebloatConfiguration {
int minBufferSize,
int bufferDebloatThresholdPercentages,
int numberOfSamples) {
+ // Right now the buffer size can not be grater than integer max value according to
+ // MemorySegment and buffer implementation.
+ checkArgument(maxBufferSize > 0);
+ checkArgument(minBufferSize > 0);
+ checkArgument(numberOfSamples > 0);
+ checkArgument(maxBufferSize >= minBufferSize);
+ checkArgument(targetTotalTime.toMillis() > 0.0);
+ checkArgument(maxBufferSize >= startingBufferSize);
+ checkArgument(minBufferSize <= startingBufferSize);
+
this.targetTotalTime = checkNotNull(targetTotalTime);
this.startingBufferSize = startingBufferSize;
this.maxBufferSize = maxBufferSize;
@@ -97,15 +107,6 @@ public final class BufferDebloatConfiguration {
int bufferDebloatThresholdPercentages = config.get(BUFFER_DEBLOAT_THRESHOLD_PERCENTAGES);
final int numberOfSamples = config.get(BUFFER_DEBLOAT_SAMPLES);
- // Right now the buffer size can not be grater than integer max value according to
- // MemorySegment and buffer implementation.
- checkArgument(maxBufferSize > 0);
- checkArgument(minBufferSize > 0);
- checkArgument(numberOfSamples > 0);
- checkArgument(maxBufferSize >= minBufferSize);
- checkArgument(targetTotalTime.toMillis() > 0.0);
- checkArgument(maxBufferSize >= startingBufferSize);
- checkArgument(minBufferSize <= startingBufferSize);
return new BufferDebloatConfiguration(
config.get(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED),
targetTotalTime,
(flink) 01/03: [FLINK-36556] Add starting buffer size config option
Posted by pn...@apache.org.
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
commit 36d9edc42227fde680f047b05a034913f99e04a3
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Aug 30 13:51:07 2024 +0200
[FLINK-36556] Add starting buffer size config option
Starting buffer size of 32KB (previous behaviour) could cause
large amount of in-flight data to accumulate before BufferDebloating
could even kick in. Now this starting size is smaller by default
and is allowed to be configured.
---
.../generated/all_taskmanager_section.html | 6 +++++
.../task_manager_memory_configuration.html | 6 +++++
.../flink/configuration/TaskManagerOptions.java | 10 +++++++++
.../io/network/NettyShuffleServiceFactory.java | 1 +
.../PipelinedApproximateSubpartition.java | 7 ++++--
.../network/partition/PipelinedSubpartition.java | 8 +++++--
.../network/partition/ResultPartitionFactory.java | 14 ++++++++++--
.../partition/consumer/SingleInputGateFactory.java | 1 +
.../NettyShuffleEnvironmentConfiguration.java | 18 +++++++++++++++
.../throughput/BufferDebloatConfiguration.java | 13 +++++++++++
.../flink/runtime/throughput/BufferDebloater.java | 26 ++++++++++++++++++++--
.../flink/runtime/throughput/BufferSizeEMA.java | 7 +++++-
.../io/network/NettyShuffleEnvironmentBuilder.java | 8 +++++++
.../io/network/NettyShuffleEnvironmentTest.java | 5 ++++-
...nedApproximateSubpartitionWithReadViewTest.java | 3 ++-
.../partition/PipelinedSubpartitionTest.java | 20 ++++++++++++++---
.../PipelinedSubpartitionWithReadViewTest.java | 2 +-
.../network/partition/ResultPartitionBuilder.java | 1 +
.../partition/ResultPartitionFactoryTest.java | 1 +
19 files changed, 142 insertions(+), 15 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/all_taskmanager_section.html b/docs/layouts/shortcodes/generated/all_taskmanager_section.html
index 44022f050e5..9a5271e314b 100644
--- a/docs/layouts/shortcodes/generated/all_taskmanager_section.html
+++ b/docs/layouts/shortcodes/generated/all_taskmanager_section.html
@@ -110,6 +110,12 @@
<td>MemorySize</td>
<td>Size of memory buffers used by the network stack and the memory manager.</td>
</tr>
+ <tr>
+ <td><h5>taskmanager.memory.starting-segment-size</h5></td>
+ <td style="word-wrap: break-word;">1 kb</td>
+ <td>MemorySize</td>
+ <td>Starting size of memory buffers used by the network stack and the memory manager, when using automatic buffer size adjustment.</td>
+ </tr>
<tr>
<td><h5>taskmanager.network.bind-policy</h5></td>
<td style="word-wrap: break-word;">"ip"</td>
diff --git a/docs/layouts/shortcodes/generated/task_manager_memory_configuration.html b/docs/layouts/shortcodes/generated/task_manager_memory_configuration.html
index fec8a688ef5..dda54a15131 100644
--- a/docs/layouts/shortcodes/generated/task_manager_memory_configuration.html
+++ b/docs/layouts/shortcodes/generated/task_manager_memory_configuration.html
@@ -110,6 +110,12 @@
<td>MemorySize</td>
<td>Size of memory buffers used by the network stack and the memory manager.</td>
</tr>
+ <tr>
+ <td><h5>taskmanager.memory.starting-segment-size</h5></td>
+ <td style="word-wrap: break-word;">1 kb</td>
+ <td>MemorySize</td>
+ <td>Starting size of memory buffers used by the network stack and the memory manager, when using automatic buffer size adjustment.</td>
+ </tr>
<tr>
<td><h5>taskmanager.memory.task.heap.size</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
index e9cecce4bdf..b6ea5c23c66 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
@@ -215,6 +215,16 @@ public class TaskManagerOptions {
"Minimum possible size of memory buffers used by the network stack and the memory manager. "
+ "ex. can be used for automatic buffer size adjustment.");
+ /** Starting size of memory buffers used by the network stack and the memory manager. */
+ @Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER)
+ public static final ConfigOption<MemorySize> STARTING_MEMORY_SEGMENT_SIZE =
+ key("taskmanager.memory.starting-segment-size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("1024"))
+ .withDescription(
+ "Starting size of memory buffers used by the network stack and the memory manager, "
+ + "when using automatic buffer size adjustment.");
+
/**
* The config parameter for automatically defining the TaskManager's binding address, if {@link
* #HOST} configuration option is not set.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleServiceFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleServiceFactory.java
index 71cdfd2177f..ecaac632758 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleServiceFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NettyShuffleServiceFactory.java
@@ -226,6 +226,7 @@ public class NettyShuffleServiceFactory
config.networkBuffersPerChannel(),
config.floatingNetworkBuffersPerGate(),
config.networkBufferSize(),
+ config.startingBufferSize(),
config.isBatchShuffleCompressionEnabled(),
config.getCompressionCodec(),
config.getMaxBuffersPerChannel(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedApproximateSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedApproximateSubpartition.java
index 460b6a9d58d..b5247d7bbe3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedApproximateSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedApproximateSubpartition.java
@@ -42,8 +42,11 @@ public class PipelinedApproximateSubpartition extends PipelinedSubpartition {
private boolean isPartialBufferCleanupRequired = false;
PipelinedApproximateSubpartition(
- int index, int receiverExclusiveBuffersPerChannel, ResultPartition parent) {
- super(index, receiverExclusiveBuffersPerChannel, parent);
+ int index,
+ int receiverExclusiveBuffersPerChannel,
+ int startingBufferSize,
+ ResultPartition parent) {
+ super(index, receiverExclusiveBuffersPerChannel, startingBufferSize, parent);
}
/**
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index a093ab9313a..06ee355fc28 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -107,7 +107,7 @@ public class PipelinedSubpartition extends ResultSubpartition implements Channel
/** Writes in-flight data. */
private ChannelStateWriter channelStateWriter;
- private int bufferSize = Integer.MAX_VALUE;
+ private int bufferSize;
/** The channelState Future of unaligned checkpoint. */
@GuardedBy("buffers")
@@ -132,13 +132,17 @@ public class PipelinedSubpartition extends ResultSubpartition implements Channel
// ------------------------------------------------------------------------
PipelinedSubpartition(
- int index, int receiverExclusiveBuffersPerChannel, ResultPartition parent) {
+ int index,
+ int receiverExclusiveBuffersPerChannel,
+ int startingBufferSize,
+ ResultPartition parent) {
super(index, parent);
checkArgument(
receiverExclusiveBuffersPerChannel >= 0,
"Buffers per channel must be non-negative.");
this.receiverExclusiveBuffersPerChannel = receiverExclusiveBuffersPerChannel;
+ this.bufferSize = startingBufferSize;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
index 8c1fe90a156..503c46fbfca 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactory.java
@@ -72,6 +72,8 @@ public class ResultPartitionFactory {
private final int networkBufferSize;
+ private final int startingBufferSize;
+
private final boolean batchShuffleCompressionEnabled;
private final CompressionCodec compressionCodec;
@@ -98,6 +100,7 @@ public class ResultPartitionFactory {
int configuredNetworkBuffersPerChannel,
int floatingNetworkBuffersPerGate,
int networkBufferSize,
+ int startingBufferSize,
boolean batchShuffleCompressionEnabled,
CompressionCodec compressionCodec,
int maxBuffersPerChannel,
@@ -116,6 +119,7 @@ public class ResultPartitionFactory {
this.batchShuffleReadIOExecutor = batchShuffleReadIOExecutor;
this.blockingSubpartitionType = blockingSubpartitionType;
this.networkBufferSize = networkBufferSize;
+ this.startingBufferSize = startingBufferSize;
this.batchShuffleCompressionEnabled = batchShuffleCompressionEnabled;
this.compressionCodec = compressionCodec;
this.maxBuffersPerChannel = maxBuffersPerChannel;
@@ -189,11 +193,17 @@ public class ResultPartitionFactory {
if (type == ResultPartitionType.PIPELINED_APPROXIMATE) {
subpartitions[i] =
new PipelinedApproximateSubpartition(
- i, configuredNetworkBuffersPerChannel, pipelinedPartition);
+ i,
+ configuredNetworkBuffersPerChannel,
+ startingBufferSize,
+ pipelinedPartition);
} else {
subpartitions[i] =
new PipelinedSubpartition(
- i, configuredNetworkBuffersPerChannel, pipelinedPartition);
+ i,
+ configuredNetworkBuffersPerChannel,
+ startingBufferSize,
+ pipelinedPartition);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
index e3f562c237b..9904d976f8b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java
@@ -224,6 +224,7 @@ public class SingleInputGateFactory {
owningTaskName,
gateIndex,
debloatConfiguration.getTargetTotalTime().toMillis(),
+ debloatConfiguration.getStartingBufferSize(),
debloatConfiguration.getMaxBufferSize(),
debloatConfiguration.getMinBufferSize(),
debloatConfiguration.getBufferDebloatThresholdPercentages(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
index 70f4e84009f..c33cd9b219d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
@@ -60,6 +60,8 @@ public class NettyShuffleEnvironmentConfiguration {
private final int networkBufferSize;
+ private final int startingBufferSize;
+
private final int partitionRequestInitialBackoff;
private final int partitionRequestMaxBackoff;
@@ -114,6 +116,7 @@ public class NettyShuffleEnvironmentConfiguration {
public NettyShuffleEnvironmentConfiguration(
int numNetworkBuffers,
int networkBufferSize,
+ int startingBufferSize,
int partitionRequestInitialBackoff,
int partitionRequestMaxBackoff,
int partitionRequestListenerTimeout,
@@ -138,6 +141,7 @@ public class NettyShuffleEnvironmentConfiguration {
this.numNetworkBuffers = numNetworkBuffers;
this.networkBufferSize = networkBufferSize;
+ this.startingBufferSize = startingBufferSize;
this.partitionRequestInitialBackoff = partitionRequestInitialBackoff;
this.partitionRequestMaxBackoff = partitionRequestMaxBackoff;
this.partitionRequestListenerTimeout = partitionRequestListenerTimeout;
@@ -171,6 +175,10 @@ public class NettyShuffleEnvironmentConfiguration {
return networkBufferSize;
}
+ public int startingBufferSize() {
+ return startingBufferSize;
+ }
+
public int partitionRequestInitialBackoff() {
return partitionRequestInitialBackoff;
}
@@ -282,6 +290,15 @@ public class NettyShuffleEnvironmentConfiguration {
final int pageSize = ConfigurationParserUtils.getPageSize(configuration);
+ int startingBufferSize = Integer.MAX_VALUE;
+ if (configuration.get(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED)) {
+ startingBufferSize =
+ (int)
+ configuration
+ .get(TaskManagerOptions.STARTING_MEMORY_SEGMENT_SIZE)
+ .getBytes();
+ }
+
final NettyConfig nettyConfig =
createNettyConfig(
configuration,
@@ -360,6 +377,7 @@ public class NettyShuffleEnvironmentConfiguration {
return new NettyShuffleEnvironmentConfiguration(
numberOfNetworkBuffers,
pageSize,
+ startingBufferSize,
initialRequestBackoff,
maxRequestBackoff,
listenerTimeout,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloatConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloatConfiguration.java
index a99877fc7ea..35b6c545586 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloatConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloatConfiguration.java
@@ -32,6 +32,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** Configuration for {@link BufferDebloater}. */
public final class BufferDebloatConfiguration {
private final Duration targetTotalTime;
+ private final int startingBufferSize;
private final int maxBufferSize;
private final int minBufferSize;
private final int bufferDebloatThresholdPercentages;
@@ -41,11 +42,13 @@ public final class BufferDebloatConfiguration {
private BufferDebloatConfiguration(
boolean enabled,
Duration targetTotalTime,
+ int startingBufferSize,
int maxBufferSize,
int minBufferSize,
int bufferDebloatThresholdPercentages,
int numberOfSamples) {
this.targetTotalTime = checkNotNull(targetTotalTime);
+ this.startingBufferSize = startingBufferSize;
this.maxBufferSize = maxBufferSize;
this.minBufferSize = minBufferSize;
this.bufferDebloatThresholdPercentages = bufferDebloatThresholdPercentages;
@@ -61,6 +64,10 @@ public final class BufferDebloatConfiguration {
return targetTotalTime;
}
+ public int getStartingBufferSize() {
+ return startingBufferSize;
+ }
+
public int getMaxBufferSize() {
return maxBufferSize;
}
@@ -83,6 +90,9 @@ public final class BufferDebloatConfiguration {
Math.toIntExact(config.get(TaskManagerOptions.MEMORY_SEGMENT_SIZE).getBytes());
int minBufferSize =
Math.toIntExact(config.get(TaskManagerOptions.MIN_MEMORY_SEGMENT_SIZE).getBytes());
+ int startingBufferSize =
+ Math.toIntExact(
+ config.get(TaskManagerOptions.STARTING_MEMORY_SEGMENT_SIZE).getBytes());
int bufferDebloatThresholdPercentages = config.get(BUFFER_DEBLOAT_THRESHOLD_PERCENTAGES);
final int numberOfSamples = config.get(BUFFER_DEBLOAT_SAMPLES);
@@ -94,9 +104,12 @@ public final class BufferDebloatConfiguration {
checkArgument(numberOfSamples > 0);
checkArgument(maxBufferSize >= minBufferSize);
checkArgument(targetTotalTime.toMillis() > 0.0);
+ checkArgument(maxBufferSize >= startingBufferSize);
+ checkArgument(minBufferSize <= startingBufferSize);
return new BufferDebloatConfiguration(
config.get(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED),
targetTotalTime,
+ startingBufferSize,
maxBufferSize,
minBufferSize,
bufferDebloatThresholdPercentages,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloater.java b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloater.java
index c056a04d60d..b098d9421d5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloater.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferDebloater.java
@@ -53,6 +53,26 @@ public class BufferDebloater {
int minBufferSize,
int bufferDebloatThresholdPercentages,
long numberOfSamples) {
+ this(
+ owningTaskName,
+ gateIndex,
+ targetTotalTime,
+ maxBufferSize,
+ maxBufferSize,
+ minBufferSize,
+ bufferDebloatThresholdPercentages,
+ numberOfSamples);
+ }
+
+ public BufferDebloater(
+ String owningTaskName,
+ int gateIndex,
+ long targetTotalTime,
+ int startingBufferSize,
+ int maxBufferSize,
+ int minBufferSize,
+ int bufferDebloatThresholdPercentages,
+ long numberOfSamples) {
this.owningTaskName = owningTaskName;
this.gateIndex = gateIndex;
this.targetTotalTime = targetTotalTime;
@@ -60,8 +80,10 @@ public class BufferDebloater {
this.minBufferSize = minBufferSize;
this.bufferDebloatThresholdFactor = bufferDebloatThresholdPercentages / 100.0;
- this.lastBufferSize = maxBufferSize;
- bufferSizeEMA = new BufferSizeEMA(maxBufferSize, minBufferSize, numberOfSamples);
+ this.lastBufferSize = startingBufferSize;
+ bufferSizeEMA =
+ new BufferSizeEMA(
+ startingBufferSize, maxBufferSize, minBufferSize, numberOfSamples);
LOG.debug(
"{}: Buffer debloater init settings: gateIndex={}, targetTotalTime={}, maxBufferSize={}, minBufferSize={}, bufferDebloatThresholdPercentages={}, numberOfSamples={}",
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferSizeEMA.java b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferSizeEMA.java
index 73ca183edaa..b4bdbe47db2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferSizeEMA.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/throughput/BufferSizeEMA.java
@@ -30,10 +30,15 @@ public class BufferSizeEMA {
private int lastBufferSize;
public BufferSizeEMA(int maxBufferSize, int minBufferSize, long numberOfSamples) {
+ this(maxBufferSize, maxBufferSize, minBufferSize, numberOfSamples);
+ }
+
+ public BufferSizeEMA(
+ int startingBufferSize, int maxBufferSize, int minBufferSize, long numberOfSamples) {
this.maxBufferSize = maxBufferSize;
this.minBufferSize = minBufferSize;
alpha = 2.0 / (numberOfSamples + 1);
- this.lastBufferSize = maxBufferSize;
+ this.lastBufferSize = startingBufferSize;
}
/**
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
index 654ea221945..13f9d9ceab5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentBuilder.java
@@ -51,6 +51,8 @@ public class NettyShuffleEnvironmentBuilder {
private int bufferSize = DEFAULT_NETWORK_BUFFER_SIZE;
+ private int startingBufferSize = Integer.MAX_VALUE;
+
private int numNetworkBuffers = DEFAULT_NUM_NETWORK_BUFFERS;
private int partitionRequestInitialBackoff;
@@ -116,6 +118,11 @@ public class NettyShuffleEnvironmentBuilder {
return this;
}
+ public NettyShuffleEnvironmentBuilder startingBufferSize(int startingBufferSize) {
+ this.startingBufferSize = startingBufferSize;
+ return this;
+ }
+
public NettyShuffleEnvironmentBuilder setNumNetworkBuffers(int numNetworkBuffers) {
this.numNetworkBuffers = numNetworkBuffers;
return this;
@@ -257,6 +264,7 @@ public class NettyShuffleEnvironmentBuilder {
new NettyShuffleEnvironmentConfiguration(
numNetworkBuffers,
bufferSize,
+ startingBufferSize,
partitionRequestInitialBackoff,
partitionRequestMaxBackoff,
partitionRequestTimeout,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
index c626a2c7336..8fe179a84f4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
@@ -184,7 +184,10 @@ class NettyShuffleEnvironmentTest {
getDebloatingMetric(
metrics, i, MetricNames.DEBLOATED_BUFFER_SIZE))
.getValue())
- .isEqualTo(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue().getBytes());
+ .isEqualTo(
+ TaskManagerOptions.STARTING_MEMORY_SEGMENT_SIZE
+ .defaultValue()
+ .getBytes());
assertThat(
((Gauge<Long>)
getDebloatingMetric(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedApproximateSubpartitionWithReadViewTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedApproximateSubpartitionWithReadViewTest.java
index 457f11a6265..dd9b298ca6b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedApproximateSubpartitionWithReadViewTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedApproximateSubpartitionWithReadViewTest.java
@@ -38,7 +38,8 @@ class PipelinedApproximateSubpartitionWithReadViewTest
@Override
void before() throws IOException {
setup(ResultPartitionType.PIPELINED_APPROXIMATE);
- subpartition = new PipelinedApproximateSubpartition(0, 2, resultPartition);
+ subpartition =
+ new PipelinedApproximateSubpartition(0, 2, Integer.MAX_VALUE, resultPartition);
availablityListener = new AwaitableBufferAvailablityListener();
readView = subpartition.createReadView(availablityListener);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
index 16c606b1fdd..2e8e34d4001 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
@@ -59,6 +59,7 @@ import static org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils.
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assumptions.assumeThat;
+import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -307,6 +308,13 @@ public class PipelinedSubpartitionTest extends SubpartitionTestBase {
assertThat(subpartition.getNumberOfQueuedBuffers()).isOne();
}
+ @TestTemplate
+ public void testStartingBufferSize() throws Exception {
+ int startingBufferSize = 1234;
+ final PipelinedSubpartition subpartition = createPipelinedSubpartition(startingBufferSize);
+ assertEquals(startingBufferSize, subpartition.add(createFilledFinishedBufferConsumer(4)));
+ }
+
@TestTemplate
void testNewBufferSize() throws Exception {
// given: Buffer size equal to integer max value by default.
@@ -513,21 +521,27 @@ public class PipelinedSubpartitionTest extends SubpartitionTestBase {
assertThat(view.isReleased()).isTrue();
}
+ public static PipelinedSubpartition createPipelinedSubpartition(int startingBufferSize) {
+ final ResultPartition parent = PartitionTestUtils.createPartition();
+
+ return new PipelinedSubpartition(0, 2, startingBufferSize, parent);
+ }
+
public static PipelinedSubpartition createPipelinedSubpartition() {
final ResultPartition parent = PartitionTestUtils.createPartition();
- return new PipelinedSubpartition(0, 2, parent);
+ return new PipelinedSubpartition(0, 2, Integer.MAX_VALUE, parent);
}
public static PipelinedSubpartition createPipelinedSubpartition(ResultPartition parent) {
- return new PipelinedSubpartition(0, 2, parent);
+ return new PipelinedSubpartition(0, 2, Integer.MAX_VALUE, parent);
}
private static class FailurePipelinedSubpartition extends PipelinedSubpartition {
FailurePipelinedSubpartition(
int index, int receiverExclusiveBuffersPerChannel, ResultPartition parent) {
- super(index, receiverExclusiveBuffersPerChannel, parent);
+ super(index, receiverExclusiveBuffersPerChannel, Integer.MAX_VALUE, parent);
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionWithReadViewTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionWithReadViewTest.java
index e562e73c672..1eebe25273a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionWithReadViewTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionWithReadViewTest.java
@@ -81,7 +81,7 @@ class PipelinedSubpartitionWithReadViewTest {
@BeforeEach
void before() throws IOException {
setup(ResultPartitionType.PIPELINED);
- subpartition = new PipelinedSubpartition(0, 2, resultPartition);
+ subpartition = new PipelinedSubpartition(0, 2, Integer.MAX_VALUE, resultPartition);
availablityListener = new AwaitableBufferAvailablityListener();
readView = subpartition.createReadView(availablityListener);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
index 4be92a45800..a9fc2bbf5cf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionBuilder.java
@@ -235,6 +235,7 @@ public class ResultPartitionBuilder {
networkBuffersPerChannel,
floatingNetworkBuffersPerGate,
networkBufferSize,
+ Integer.MAX_VALUE,
blockingShuffleCompressionEnabled,
compressionCodec,
maxBuffersPerChannel,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
index eac943859b8..bd8156634df 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
@@ -176,6 +176,7 @@ class ResultPartitionFactoryTest {
1,
1,
SEGMENT_SIZE,
+ Integer.MAX_VALUE,
false,
CompressionCodec.LZ4,
Integer.MAX_VALUE,