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 2023/03/02 15:33:03 UTC
[flink] branch master updated (b4d43b47c99 -> 382148e1229)
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from b4d43b47c99 [FLINK-30989][table] Fix some options don't take effect in batch mode
new 70c2ab65ade [FLINK-31288][runtime] Disable overdraft buffer for non pipelined result partition.
new 382148e1229 [FLINK-31288][doc] Update doc ralated to overdraft buffer.
The 2 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:
.../docs/deployment/memory/network_mem_tuning.md | 4 ++++
docs/content/docs/deployment/memory/network_mem_tuning.md | 4 ++++
.../generated/all_taskmanager_network_section.html | 2 +-
.../netty_shuffle_environment_configuration.html | 2 +-
.../configuration/NettyShuffleEnvironmentOptions.java | 3 ++-
.../io/network/partition/ResultPartitionFactory.java | 15 ++++++++++++++-
6 files changed, 26 insertions(+), 4 deletions(-)
[flink] 02/02: [FLINK-31288][doc] Update doc ralated to overdraft buffer.
Posted by gu...@apache.org.
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 382148e1229901ab54503c8d9af6a18ea4c078dc
Author: Weijie Guo <re...@163.com>
AuthorDate: Thu Mar 2 18:08:37 2023 +0800
[FLINK-31288][doc] Update doc ralated to overdraft buffer.
---
docs/content.zh/docs/deployment/memory/network_mem_tuning.md | 4 ++++
docs/content/docs/deployment/memory/network_mem_tuning.md | 4 ++++
2 files changed, 8 insertions(+)
diff --git a/docs/content.zh/docs/deployment/memory/network_mem_tuning.md b/docs/content.zh/docs/deployment/memory/network_mem_tuning.md
index 65ab04b60ee..5f38a245e6c 100644
--- a/docs/content.zh/docs/deployment/memory/network_mem_tuning.md
+++ b/docs/content.zh/docs/deployment/memory/network_mem_tuning.md
@@ -133,6 +133,10 @@ Flink 有多个本地缓冲区池 —— 每个输出和输入流对应一个。
为了缓解这种情况，增加了透支缓冲区的概念。这些透支缓冲区是可选的，Flink 可以仅仅使用常规的缓冲区逐渐取得进展，也就是
说 `0` 是 `taskmanager.network.memory.max-overdraft-buffers-per-gate` 可以接受的配置值。
+{{< hint warning >}}
+该特性只对`Pipelined Shuffle`生效。
+{{< /hint >}}
+
## 缓冲区的数量
独占缓冲区和流动缓冲区的默认配置应该足以应对最大吞吐。如果想要最小化缓冲数据量，那么可以将独占缓冲区设置为 `0`，同时减小内存段的大小。
diff --git a/docs/content/docs/deployment/memory/network_mem_tuning.md b/docs/content/docs/deployment/memory/network_mem_tuning.md
index c1af259f57d..bd07da88e8e 100644
--- a/docs/content/docs/deployment/memory/network_mem_tuning.md
+++ b/docs/content/docs/deployment/memory/network_mem_tuning.md
@@ -142,6 +142,10 @@ buffers concept has been added. Those overdraft buffers are strictly optional an
gradually make progress using only regular buffers, which means `0` is an acceptable configuration
for the `taskmanager.network.memory.max-overdraft-buffers-per-gate`.
+{{< hint warning >}}
+This feature only takes effect for `Pipelined Shuffle`.
+{{< /hint >}}
+
## The number of in-flight buffers
The default settings for exclusive buffers and floating buffers should be sufficient for the maximum throughput. If the minimum of in-flight data needs to be set, the exclusive buffers can be set to `0` and the memory segment size can be decreased.
[flink] 01/02: [FLINK-31288][runtime] Disable overdraft buffer for non pipelined result partition.
Posted by gu...@apache.org.
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 70c2ab65adeaa03af69b647ba2087ac6cc09e4b1
Author: Weijie Guo <re...@163.com>
AuthorDate: Thu Mar 2 15:29:25 2023 +0800
[FLINK-31288][runtime] Disable overdraft buffer for non pipelined result partition.
---
.../generated/all_taskmanager_network_section.html | 2 +-
.../netty_shuffle_environment_configuration.html | 2 +-
.../configuration/NettyShuffleEnvironmentOptions.java | 3 ++-
.../io/network/partition/ResultPartitionFactory.java | 15 ++++++++++++++-
4 files changed, 18 insertions(+), 4 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html b/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html
index 202b574a0e1..7d80b82ada8 100644
--- a/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html
+++ b/docs/layouts/shortcodes/generated/all_taskmanager_network_section.html
@@ -102,7 +102,7 @@
<td><h5>taskmanager.network.memory.max-overdraft-buffers-per-gate</h5></td>
<td style="word-wrap: break-word;">5</td>
<td>Integer</td>
- <td>Number of max overdraft network buffers to use for each ResultPartition. The overdraft buffers will be used when the subtask cannot apply to the normal buffers due to back pressure, while subtask is performing an action that can not be interrupted in the middle, like serializing a large record, flatMap operator producing multiple records for one single input record or processing time timer producing large output. In situations like that system will allow subtask to requ [...]
+ <td>Number of max overdraft network buffers to use for each ResultPartition. The overdraft buffers will be used when the subtask cannot apply to the normal buffers due to back pressure, while subtask is performing an action that can not be interrupted in the middle, like serializing a large record, flatMap operator producing multiple records for one single input record or processing time timer producing large output. In situations like that system will allow subtask to requ [...]
</tr>
<tr>
<td><h5>taskmanager.network.memory.read-buffer.required-per-gate.max</h5></td>
diff --git a/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html b/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html
index 6a6aa6913b9..cbea22beb26 100644
--- a/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html
+++ b/docs/layouts/shortcodes/generated/netty_shuffle_environment_configuration.html
@@ -90,7 +90,7 @@
<td><h5>taskmanager.network.memory.max-overdraft-buffers-per-gate</h5></td>
<td style="word-wrap: break-word;">5</td>
<td>Integer</td>
- <td>Number of max overdraft network buffers to use for each ResultPartition. The overdraft buffers will be used when the subtask cannot apply to the normal buffers due to back pressure, while subtask is performing an action that can not be interrupted in the middle, like serializing a large record, flatMap operator producing multiple records for one single input record or processing time timer producing large output. In situations like that system will allow subtask to requ [...]
+ <td>Number of max overdraft network buffers to use for each ResultPartition. The overdraft buffers will be used when the subtask cannot apply to the normal buffers due to back pressure, while subtask is performing an action that can not be interrupted in the middle, like serializing a large record, flatMap operator producing multiple records for one single input record or processing time timer producing large output. In situations like that system will allow subtask to requ [...]
</tr>
<tr>
<td><h5>taskmanager.network.memory.read-buffer.required-per-gate.max</h5></td>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
index b0bd65292e6..cab12fe1972 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
@@ -363,7 +363,8 @@ public class NettyShuffleEnvironmentOptions {
+ " such uninterruptible action, without blocking unaligned checkpoints for long period of"
+ " time. Overdraft buffers are provided on best effort basis only if the system has some"
+ " unused buffers available. Subtask that has used overdraft buffers won't be allowed to"
- + " process any more records until the overdraft buffers are returned to the pool.");
+ + " process any more records until the overdraft buffers are returned to the pool."
+ + " It should be noted that this config option only takes effect for Pipelined Shuffle.");
/** The timeout for requesting exclusive buffers for each channel. */
@Documentation.ExcludeFromDocumentation(
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 3d8c27bc09d..069536e8664 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
@@ -301,6 +301,19 @@ public class ResultPartitionFactory {
}
}
+ /** Return whether this result partition need overdraft buffer. */
+ private static boolean isOverdraftBufferNeeded(ResultPartitionType resultPartitionType) {
+ // Only pipelined / pipelined-bounded partition needs overdraft buffer. More
+ // specifically, there is no reason to request more buffers for non-pipelined (i.e.
+ // batch) shuffle. The reasons are as follows:
+ // 1. For BoundedBlockingShuffle, each full buffer will be directly released.
+ // 2. For SortMergeShuffle, the maximum capacity of buffer pool is 4 * numSubpartitions. It
+ // is efficient enough to spill this part of memory to disk.
+ // 3. For Hybrid Shuffle, the buffer pool is unbounded. If it can't get a normal buffer, it
+ // also can't get an overdraft buffer.
+ return resultPartitionType.isPipelinedOrPipelinedBoundedResultPartition();
+ }
+
/**
* The minimum pool size should be <code>numberOfSubpartitions + 1</code> for two
* considerations:
@@ -330,7 +343,7 @@ public class ResultPartitionFactory {
pair.getRight(),
numberOfSubpartitions,
maxBuffersPerChannel,
- maxOverdraftBuffersPerGate);
+ isOverdraftBufferNeeded(type) ? maxOverdraftBuffersPerGate : 0);
};
}