You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2019/07/08 13:34:07 UTC
[flink] 04/10: [hotfix] [runtime] Split heap/managed memory
computation into smaller reusable methods
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 92fe83ffa88659d310a18666a57d1562fcba8fcb
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Sun Jul 7 18:34:47 2019 +0200
[hotfix] [runtime] Split heap/managed memory computation into smaller reusable methods
---
...TaskManagerHeapSizeCalculationJavaBashTest.java | 21 ++---
.../runtime/taskexecutor/TaskManagerServices.java | 98 ++++++++++++++--------
.../ContaineredTaskManagerParametersTest.java | 22 +++--
3 files changed, 88 insertions(+), 53 deletions(-)
diff --git a/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java b/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
index 87cf194..3e9ec37 100644
--- a/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
+++ b/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
@@ -279,18 +279,15 @@ public class TaskManagerHeapSizeCalculationJavaBashTest extends TestLogger {
String.valueOf(config.getFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION))};
String scriptOutput = executeScript(command);
- long absoluteTolerance = (long) (javaHeapSizeMB * tolerance);
- if (absoluteTolerance < 1) {
- assertEquals("Different heap sizes with configuration: " + config.toString(),
- String.valueOf(javaHeapSizeMB), scriptOutput);
- } else {
- Long scriptHeapSizeMB = Long.valueOf(scriptOutput);
- assertThat(
- "Different heap sizes (Java: " + javaHeapSizeMB + ", Script: " + scriptHeapSizeMB +
- ") with configuration: " + config.toString(), scriptHeapSizeMB,
- allOf(greaterThanOrEqualTo(javaHeapSizeMB - absoluteTolerance),
- lessThanOrEqualTo(javaHeapSizeMB + absoluteTolerance)));
- }
+ // we need a tolerance of at least one, to compensate for MB/byte conversion rounding errors
+ long absoluteTolerance = Math.max(1L, (long) (javaHeapSizeMB * tolerance));
+
+ Long scriptHeapSizeMB = Long.valueOf(scriptOutput);
+ assertThat(
+ "Different heap sizes (Java: " + javaHeapSizeMB + ", Script: " + scriptHeapSizeMB +
+ ") with configuration: " + config.toString(), scriptHeapSizeMB,
+ allOf(greaterThanOrEqualTo(javaHeapSizeMB - absoluteTolerance),
+ lessThanOrEqualTo(javaHeapSizeMB + absoluteTolerance)));
}
/**
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
old mode 100644
new mode 100755
index ee00dde..e211fc0
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
@@ -423,50 +423,68 @@ public class TaskManagerServices {
public static long calculateHeapSizeMB(long totalJavaMemorySizeMB, Configuration config) {
Preconditions.checkArgument(totalJavaMemorySizeMB > 0);
- // subtract the Java memory used for network buffers (always off-heap)
- final long networkBufMB = NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(
- totalJavaMemorySizeMB << 20, // megabytes to bytes
- config) >> 20; // bytes to megabytes
- final long remainingJavaMemorySizeMB = totalJavaMemorySizeMB - networkBufMB;
-
- // split the available Java memory between heap and off-heap
-
- final boolean useOffHeap = config.getBoolean(TaskManagerOptions.MEMORY_OFF_HEAP);
-
- final long heapSizeMB;
- if (useOffHeap) {
-
- long offHeapSize;
- String managedMemorySizeDefaultVal = TaskManagerOptions.MANAGED_MEMORY_SIZE.defaultValue();
- if (!config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE).equals(managedMemorySizeDefaultVal)) {
- try {
- offHeapSize = MemorySize.parse(config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE), MEGA_BYTES).getMebiBytes();
- } catch (IllegalArgumentException e) {
- throw new IllegalConfigurationException(
- "Could not read " + TaskManagerOptions.MANAGED_MEMORY_SIZE.key(), e);
- }
- } else {
- offHeapSize = Long.valueOf(managedMemorySizeDefaultVal);
- }
+ // all values below here are in bytes
- if (offHeapSize <= 0) {
- // calculate off-heap section via fraction
- double fraction = config.getFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION);
- offHeapSize = (long) (fraction * remainingJavaMemorySizeMB);
- }
+ final long totalProcessMemory = megabytesToBytes(totalJavaMemorySizeMB);
+ final long networkReservedMemory = getReservedNetworkMemory(config, totalProcessMemory);
+ final long heapAndManagedMemory = totalProcessMemory - networkReservedMemory;
+
+ if (config.getBoolean(TaskManagerOptions.MEMORY_OFF_HEAP)) {
+ final long managedMemorySize = getManagedMemoryFromHeapAndManaged(config, heapAndManagedMemory);
- ConfigurationParserUtils.checkConfigParameter(offHeapSize < remainingJavaMemorySizeMB, offHeapSize,
+ ConfigurationParserUtils.checkConfigParameter(managedMemorySize < heapAndManagedMemory, managedMemorySize,
TaskManagerOptions.MANAGED_MEMORY_SIZE.key(),
- "Managed memory size too large for " + networkBufMB +
+ "Managed memory size too large for " + (networkReservedMemory >> 20) +
" MB network buffer memory and a total of " + totalJavaMemorySizeMB +
" MB JVM memory");
- heapSizeMB = remainingJavaMemorySizeMB - offHeapSize;
- } else {
- heapSizeMB = remainingJavaMemorySizeMB;
+ return bytesToMegabytes(heapAndManagedMemory - managedMemorySize);
+ }
+ else {
+ return bytesToMegabytes(heapAndManagedMemory);
+ }
+ }
+
+ /**
+ * Gets the size of managed memory from the JVM process size, which at that point includes
+ * network buffer memory, managed memory, and non-flink-managed heap memory.
+ * All values are in bytes.
+ */
+ public static long getManagedMemoryFromProcessMemory(Configuration config, long totalProcessMemory) {
+ final long heapAndManagedMemory = totalProcessMemory - getReservedNetworkMemory(config, totalProcessMemory);
+ return getManagedMemoryFromHeapAndManaged(config, heapAndManagedMemory);
+ }
+
+ /**
+ * Gets the size of managed memory from the heap size after subtracting network buffer memory.
+ * All values are in bytes.
+ */
+ public static long getManagedMemoryFromHeapAndManaged(Configuration config, long heapAndManagedMemory) {
+ if (config.contains(TaskManagerOptions.MANAGED_MEMORY_SIZE)) {
+ // take the configured absolute value
+ final String sizeValue = config.getString(TaskManagerOptions.MANAGED_MEMORY_SIZE);
+ try {
+ return MemorySize.parse(sizeValue, MEGA_BYTES).getBytes();
+ }
+ catch (IllegalArgumentException e) {
+ throw new IllegalConfigurationException(
+ "Could not read " + TaskManagerOptions.MANAGED_MEMORY_SIZE.key(), e);
+ }
+ }
+ else {
+ // calculate managed memory size via fraction
+ final float fraction = config.getFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION);
+ return (long) (fraction * heapAndManagedMemory);
}
+ }
- return heapSizeMB;
+ /**
+ * Gets the amount of memory reserved for networking, given the total JVM memory.
+ * All values are in bytes.
+ */
+ public static long getReservedNetworkMemory(Configuration config, long totalProcessMemory) {
+ // subtract the Java memory used for network buffers (always off-heap)
+ return NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(totalProcessMemory, config);
}
/**
@@ -519,4 +537,12 @@ public class TaskManagerServices {
managedMemoryPerSlot,
Collections.emptyMap());
}
+
+ private static long bytesToMegabytes(long bytes) {
+ return bytes >> 20;
+ }
+
+ private static long megabytesToBytes(long megabytes) {
+ return megabytes << 20;
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
old mode 100644
new mode 100755
index c49f8e6..a1f4cad
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
@@ -21,13 +21,16 @@ package org.apache.flink.runtime.clusterframework;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ResourceManagerOptions;
-import org.apache.flink.runtime.taskmanager.NettyShuffleEnvironmentConfiguration;
+import org.apache.flink.runtime.taskexecutor.TaskManagerServices;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
import static org.apache.flink.configuration.TaskManagerOptions.MEMORY_OFF_HEAP;
+import static org.hamcrest.Matchers.anyOf;
+import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -52,10 +55,19 @@ public class ContaineredTaskManagerParametersTest extends TestLogger {
ConfigConstants.DEFAULT_YARN_HEAP_CUTOFF);
long cutoff = Math.max((long) (CONTAINER_MEMORY * memoryCutoffRatio), minCutoff);
- final long networkBufMB = NettyShuffleEnvironmentConfiguration.calculateNetworkBufferMemory(
- (CONTAINER_MEMORY - cutoff) << 20, // megabytes to bytes
- conf) >> 20; // bytes to megabytes
- assertEquals(networkBufMB + cutoff, params.taskManagerDirectMemoryLimitMB());
+ final long networkBufMB = TaskManagerServices.getReservedNetworkMemory(
+ conf,
+ (CONTAINER_MEMORY - cutoff) << 20 // megabytes to bytes
+ ) >> 20; // bytes to megabytes
+
+ // this is unfortunately necessary due to rounding errors that happen due to back and
+ // forth conversion between bytes and megabytes
+ // ideally all logic calculates precisely with bytes and we use coarser units only un
+ // user-facing configuration and parametrization classes
+ assertThat(networkBufMB + cutoff,
+ anyOf(
+ equalTo(params.taskManagerDirectMemoryLimitMB()),
+ equalTo(params.taskManagerDirectMemoryLimitMB() - 1)));
}
/**