You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2023/01/10 09:11:36 UTC
[GitHub] [flink] xintongsong commented on a diff in pull request #21620: [FLINK-30473][network] Optimize the InputGate network memory management for TaskManager
xintongsong commented on code in PR #21620:
URL: https://github.com/apache/flink/pull/21620#discussion_r1065498134
##########
flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleUtils.java:
##########
@@ -119,6 +142,95 @@ public static int computeNetworkBuffersForAnnouncing(
return requirementForInputs + requirementForOutputs;
}
+ public static int maxRequiredBuffersPerGate(
+ ResultPartitionType partitionType,
+ boolean isGateRequiredMaxBuffersConfigured,
+ int requiredMaxBuffersPerGate) {
+ int requiredMaxBuffers;
+ if (isGateRequiredMaxBuffersConfigured) {
+ requiredMaxBuffers = requiredMaxBuffersPerGate;
+ } else {
+ requiredMaxBuffers =
+ partitionType.isBlockingOrBlockingPersistentResultPartition()
+ ? DEFAULT_MAX_BUFFERS_PER_GATE_FOR_BLOCKING
+ : DEFAULT_MAX_BUFFERS_PER_GATE_FOR_STREAM;
Review Comment:
What about hybrid shuffle? Hybrid shuffle is also used for batch workloads, which according to the config option description should use the batch threashold.
##########
flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java:
##########
@@ -213,6 +214,7 @@ public class NettyShuffleEnvironmentOptions {
* Number of extra network buffers to use for each outgoing/incoming gate (result
* partition/input gate).
*/
+ @Deprecated
Review Comment:
I think what we decided in the FLIP is not to deprecate these two options immediately.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateFactory.java:
##########
@@ -293,20 +324,74 @@ protected InputChannel createKnownInputChannel(
connectionManager,
partitionRequestInitialBackoff,
partitionRequestMaxBackoff,
- networkBuffersPerChannel,
+ buffersPerChannel,
metrics);
}
}
@VisibleForTesting
static SupplierWithException<BufferPool, IOException> createBufferPoolFactory(
- BufferPoolFactory bufferPoolFactory, int floatingNetworkBuffersPerGate) {
- Pair<Integer, Integer> pair =
- NettyShuffleUtils.getMinMaxFloatingBuffersPerInputGate(
- floatingNetworkBuffersPerGate);
+ BufferPoolFactory bufferPoolFactory,
+ int minFloatingBuffersPerGate,
+ int maxFloatingBuffersPerGate) {
+ Pair<Integer, Integer> pair = Pair.of(minFloatingBuffersPerGate, maxFloatingBuffersPerGate);
return () -> bufferPoolFactory.createBufferPool(pair.getLeft(), pair.getRight());
}
+ /**
+ * Based on whether the used exclusive buffers exceed the threshold, decide whether all buffers
+ * in the gate use Floating Buffers.
+ *
+ * <p>The threshold is configured by {@link
+ * NettyShuffleEnvironmentOptions#NETWORK_READ_BUFFERS_REQUIRED_PER_GATE_MAX}. If the option is
+ * not configured, the threshold for Blocking jobs is {@link
+ * NettyShuffleUtils#DEFAULT_MAX_BUFFERS_PER_GATE_FOR_BLOCKING} and the threshold for Streaming
+ * jobs is {#link NettyShuffleUtils#DEFAULT_MAX_BUFFERS_PER_GATE_FOR_STREAMING}.
+ */
+ protected class GateBuffersNumCalculator {
+
+ private final int minFloatingBuffers;
+
+ private final int maxFloatingBuffers;
+
+ private final int exclusiveBuffersPerChannel;
+
+ GateBuffersNumCalculator(ResultPartitionType partitionType, int numInputChannels) {
+ int maxGateBuffersThreshold =
+ maxRequiredBuffersPerGate(
+ partitionType,
+ isGateRequiredMaxBuffersConfigured,
+ requiredMaxBuffersPerGate);
+
+ int adjustedBuffersPerChannel =
+ adjustExclusiveBuffersPerChannel(
+ networkBuffersPerChannel, numInputChannels, maxGateBuffersThreshold);
+ boolean useFloatingBuffer = adjustedBuffersPerChannel < 0;
+
+ this.minFloatingBuffers = useFloatingBuffer ? maxGateBuffersThreshold : 1;
Review Comment:
This is incorrect.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleUtils.java:
##########
@@ -119,6 +142,95 @@ public static int computeNetworkBuffersForAnnouncing(
return requirementForInputs + requirementForOutputs;
}
+ public static int maxRequiredBuffersPerGate(
+ ResultPartitionType partitionType,
+ boolean isGateRequiredMaxBuffersConfigured,
+ int requiredMaxBuffersPerGate) {
+ int requiredMaxBuffers;
+ if (isGateRequiredMaxBuffersConfigured) {
+ requiredMaxBuffers = requiredMaxBuffersPerGate;
+ } else {
+ requiredMaxBuffers =
+ partitionType.isBlockingOrBlockingPersistentResultPartition()
+ ? DEFAULT_MAX_BUFFERS_PER_GATE_FOR_BLOCKING
+ : DEFAULT_MAX_BUFFERS_PER_GATE_FOR_STREAM;
+ }
+ checkState(requiredMaxBuffers >= 0, "Max required buffers per gate must be non-negative.");
+ return requiredMaxBuffers;
+ }
+
+ public static int getMaxFloatingBuffersInGate(
+ int numInputChannels, int networkBuffersPerChannel, int floatingNetworkBuffersPerGate) {
+ return getExclusiveBuffersInGate(numInputChannels, networkBuffersPerChannel)
+ + floatingNetworkBuffersPerGate;
+ }
+
+ private static int getNumBuffersToAnnounceForInputGates(
+ ResultPartitionType type,
+ int numBuffersPerChannel,
+ int numFloatingBuffersPerGate,
+ boolean isGateRequiredMaxBuffersConfigured,
+ int requiredMaxBuffersPerGate,
+ int numInputChannels) {
+ int maxGateBuffersThreshold =
+ maxRequiredBuffersPerGate(
+ type, isGateRequiredMaxBuffersConfigured, requiredMaxBuffersPerGate);
+
+ int adjustedBuffersPerChannel =
+ adjustExclusiveBuffersPerChannel(
+ numBuffersPerChannel, numInputChannels, maxGateBuffersThreshold);
+ boolean useFloatingBuffer = adjustedBuffersPerChannel < 0;
+
+ int maxFloatingBuffers =
+ useFloatingBuffer
+ ? getMaxFloatingBuffersInGate(
+ numInputChannels, numBuffersPerChannel, numFloatingBuffersPerGate)
+ : numFloatingBuffersPerGate;
+ int exclusiveBuffersPerChannel = useFloatingBuffer ? 0 : adjustedBuffersPerChannel;
+
+ return exclusiveBuffersPerChannel * numInputChannels + maxFloatingBuffers;
+ }
+
+ /**
+ * Adjusting the exclusive network buffers based on whether the total exclusive buffers in one
+ * gate has exceeded the gate buffer threshold.
+ *
+ * @return Adjusted buffers or -1. Return -1 if and only if the total exclusive buffers will
+ * exceed the gate buffer threshold though the exclusive network buffers per channel is 1.
+ * Returning -1 indicates all reading buffers in gate should use floating buffers.
+ */
+ public static int adjustExclusiveBuffersPerChannel(
+ int numBuffersPerChannel, int numInputChannels, int maxGateBuffersThreshold) {
+ if (numBuffersPerChannel == 0) {
+ return 0;
+ }
+
+ int adjustedBuffersPerChannel = -1;
+ for (int i = numBuffersPerChannel; i > 0; i--) {
+ if (getExclusiveBuffersInGate(numInputChannels, i) <= maxGateBuffersThreshold) {
+ adjustedBuffersPerChannel = i;
+ break;
+ }
+ }
+ return adjustedBuffersPerChannel;
+ }
Review Comment:
The whole method can be simplified as:
```
return Math.min(numBuffersPerChannel, maxGateBuffersThreshold / numInputChannels);
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java:
##########
@@ -282,6 +300,17 @@ public static NettyShuffleEnvironmentConfiguration fromConfiguration(
configuration.getInteger(
NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE);
+ boolean isRequiredMaxBuffersConfigured =
+ configuration.contains(
+ NettyShuffleEnvironmentOptions.NETWORK_READ_BUFFERS_REQUIRED_PER_GATE_MAX);
+
+ int requiredMaxBuffersPerGate =
+ isRequiredMaxBuffersConfigured
+ ? configuration.getInteger(
+ NettyShuffleEnvironmentOptions
+ .NETWORK_READ_BUFFERS_REQUIRED_PER_GATE_MAX)
+ : -1;
Review Comment:
Use `configuration.getOptional(NettyShuffleEnvironmentOptions.NETWORK_READ_BUFFERS_REQUIRED_PER_GATE_MAX)`, and you won't need two fields for this.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleUtils.java:
##########
@@ -119,6 +142,95 @@ public static int computeNetworkBuffersForAnnouncing(
return requirementForInputs + requirementForOutputs;
}
+ public static int maxRequiredBuffersPerGate(
+ ResultPartitionType partitionType,
+ boolean isGateRequiredMaxBuffersConfigured,
+ int requiredMaxBuffersPerGate) {
+ int requiredMaxBuffers;
+ if (isGateRequiredMaxBuffersConfigured) {
+ requiredMaxBuffers = requiredMaxBuffersPerGate;
+ } else {
+ requiredMaxBuffers =
+ partitionType.isBlockingOrBlockingPersistentResultPartition()
+ ? DEFAULT_MAX_BUFFERS_PER_GATE_FOR_BLOCKING
+ : DEFAULT_MAX_BUFFERS_PER_GATE_FOR_STREAM;
+ }
+ checkState(requiredMaxBuffers >= 0, "Max required buffers per gate must be non-negative.");
+ return requiredMaxBuffers;
+ }
+
+ public static int getMaxFloatingBuffersInGate(
+ int numInputChannels, int networkBuffersPerChannel, int floatingNetworkBuffersPerGate) {
+ return getExclusiveBuffersInGate(numInputChannels, networkBuffersPerChannel)
+ + floatingNetworkBuffersPerGate;
+ }
Review Comment:
Incorrect.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org