[FLINK-4545] replace the network buffers parameter #3721

NicoK · 2017-04-13T12:49:54Z

(based on #3708 and #3713)

Instead, allow the configuration with the following three new (more flexible) parameters:

taskmanager.network.memory.fraction: fraction of JVM memory to use for network buffers (default: 0.1)
taskmanager.network.memory.min: minimum memory size for network buffers (default: 64 MB)
taskmanager.network.memory.max: maximum memory size for network buffers (default: 1 GB)

Note that I needed to adapt two unit tests which would have been killed on Travis CI because these defaults result in ~150MB memory being used for network buffers which apparently was too much there.

zentol · 2017-04-18T11:48:27Z

I've merged the 2 PR's that this one build upon; could you rebase this one?

Instead, allow the configuration with the following three new (more flexible) parameters: * "taskmanager.network.memory.fraction": fraction of JVM memory to use for network buffers (default: 0.1) * "taskmanager.network.memory.min": minimum memory size for network buffers (default: 64 MB) * "taskmanager.network.memory.max": maximum memory size for network buffers (default: 1 GB) # Please enter the commit message for your changes. Lines starting

… account

…ry config

…eters and methods

These verify that the results are the same as in the calculation done by Java.

Also update the descriptions of taskmanager.memory.fraction not being relative to the full size of taskmanager.heap.mb but that network buffer memory is subtracted before!

Due to the increased defaults for network buffer memory use, some builds on Travis CI fail with unit tests being killed. This affects * RocksDbBackendEventTimeWindowCheckpointingITCase and * HBaseConnectorITCase We fix this by limiting the maximum amount of network buffer memory to 80MB (current defaults would yield 150MB, previously 64MB were used).

zentol · 2017-04-18T14:23:47Z

docs/setup/config.md

 ### Configuring the Network Buffers

-If you ever see the Exception `java.io.IOException: Insufficient number of network buffers`, please use the following formula to adjust the number of network buffers:
+If you ever see the Exception `java.io.IOException: Insufficient number of network buffers`, you


We usually don't do manual line breaks in the documentation; otherwise if you resize the window funky things start to happen.

afaik those things only happen if you leave an empty space at the end of the line or so - here, the lines are properly joined inside HTML

zentol

What error message occurred (if any) when the tests failed on travis?

zentol · 2017-04-18T14:52:46Z

flink-dist/src/main/flink-bin/bin/config.sh

+        network_buffers_bytes=`awk "BEGIN { x = lshift(${FLINK_TM_HEAP},20) * ${FLINK_TM_NET_BUF_FRACTION}; netbuf = x > ${FLINK_TM_NET_BUF_MAX} ? ${FLINK_TM_NET_BUF_MAX} : x < ${FLINK_TM_NET_BUF_MIN} ? ${FLINK_TM_NET_BUF_MIN} : x; printf \"%.0f\n\", netbuf }"`
+    fi
+
+    # recalculate the JVM heap memory by taking the network buffers into account


This is more of a verification isn't it?

what do you mean?

To me, "recalculate" implied that it would change some configuration value, but that's not happening. It's only verifying that the memory for network buffers is less than the heap memory.

no, actually, the user may give the FLINK_TM_HEAP environment variable or configure the "flink heap size" via taskmanager.heap.mb but this is not the real "heap" size - rather the overall memory size used by flink (including off-heap). So this function removes the off-heap part and returns the real heap sizes to use with -Xmx and -Xms

zentol · 2017-04-18T15:23:00Z

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java

 	public static long calculateNetworkBuf(long totalJavaMemorySize, Configuration config) {
+		assert totalJavaMemorySize > 0;


use Preconditions instead?

zentol · 2017-04-18T15:56:14Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java

+
+			long javaMem = Math.max(max + 1, ran.nextLong());
+
+			final long networkBufMem = TaskManagerServices.calculateNetworkBuf(javaMem, config);


What we definitely need here is a catch block that prints the used parameters if any assertion fails. It is a bit odd to use random parameters in the first place though :/

Random testing is quite powerful to identify unforeseen errors, especially since these inputs are user-configurable. I'll keep that but add better error messages as suggested.

zentol · 2017-04-18T16:01:25Z

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java

+	 *
+	 * @return memory to use for network buffers (in bytes)
+	 */
+	public static long calculateNetworkBuf(long totalJavaMemorySize, Configuration config) {


how about a slightly longer method name?

zentol · 2017-04-18T16:13:54Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java

+		// old + 1 new parameter = new:
+		Configuration config1 = config.clone();
+		config1.setFloat(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f);
+		assertEquals(Math.min(defaultMax, Math.max(defaultMin, (long) (0.1f * (10L << 20)))),


This seems overly complicated. If we would use a set of well-defined values instead of the defaults we could get rid of the min/max calls, no? This applies to the whole test.

yes, if we define our own test-defaults, your are right - however, we would decouple the test from the defaults that are set for real-world applications and I wanted to keep them as close as possible including any future change in the default values

zentol · 2017-04-18T16:25:37Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java

+
+		config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, -1); // use fraction of given memory
+		config.setFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION, 0.1f); // 10%
+		assertEquals(810, TaskManagerServices.calculateHeapSizeMB(1000, config));


Now that i see this test it got me thinking: the managedMemory and networkBuffersmemory frations do not work on the same base value; i.e if both are set at 0.5 then one (the network i think) gets 0.5 of the total memory, while the managedMemory gets 0.25. I'm wondering how intuitive this is; they are similar when used alone, but when both are used 0.5 doesn't equal 0.5 in a way.

yes, unfortunately, that is the case and it has always been that way in the past (with the network buffers not being a fraction but a fixed amount of memory) but it is also properly documented (now).

With the possibility to specify min and max values for the network buffer memory size, the actual fraction may be different than the given one and we don't really want to fail jobs because we can't ensure the given fraction for the managed memory in that case, do we?
As a side note, this is also the safest way to ensure that the invariants hold, especially for the 0.5 vs. 0.5 example: inside Java, we will always allocate the network buffer memory first and then identify the remaining free heap space (if on-heap) from which we will use the given fraction.

zentol · 2017-04-18T16:26:23Z

flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java

@@ -450,7 +450,7 @@ public FileBaseStatistics getStatistics(BaseStatistics cachedStats) throws IOExc
 		}
 		catch (Throwable t) {
 			if (LOG.isErrorEnabled()) {
-				LOG.error("Unexpected problen while getting the file statistics for file '" + this.filePath + "': "
+				LOG.error("Unexpected problem while getting the file statistics for file '" + this.filePath + "': "


Let's see whether i can remember to not squash this commit :)

sorry, should I create a separate PR? (a separate JIRA is definitely overkill for this)

You can leave it as it is.

zentol · 2017-04-18T16:29:59Z

flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java

+
+	@Before
+	public void checkOperatingSystem() {
+		Assume.assumeTrue("This test checks shell scripts not available on Windows.",


missing "that are"?

zentol · 2017-04-18T16:36:12Z

flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java

+		Random ran = new Random();
+		for (int i = 0; i < NUM_RANDOM_TESTS; ++i) {
+			// tolerate that values differ by 1% (due to different floating point precisions)
+			compareNetworkBufJavaVsScript(getRandomConfig(ran), 0.01f);


As with the other randomized tests we should print the configured used for the test in case of failure.

oh, here, I actually do already print the configuration in the error message

zentol · 2017-04-18T16:37:27Z

flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java

+
+		float frac = Math.max(ran.nextFloat(), Float.MIN_VALUE);
+
+//		long min = Math.max(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue(), ran.nextLong());


these lines can be removed i guess

NicoK · 2017-04-19T15:33:41Z

For the two tests that failed on Travis CI: they were simply killed and a "Killed" appeared in their logs which is usually an indicator that memory ran out and the kernel killed a process

StephanEwen · 2017-05-04T13:28:33Z

I am checking this PR out now...

StephanEwen · 2017-05-04T19:46:59Z

Code is good in general and well tested (including the shell scripts, which is great!)

I would do some on-the fly polishing while merging.
The main thing I want to adjust is that configuration parameters are specified in Megabytes, not Bytes. All other memory related parameters are in Megabytes, so that one should be as well, for consistency. Also, I think we don't need a finer granularity these days.

StephanEwen · 2017-05-05T12:01:14Z

Okay, taking a step back. Looking through the code some more, the internal arithmetric should certainly stay in bytes. However, bytes are tedious to configure.

I suggest to add support to the configuration to interpret memory units, so that we can configure values via

512m
10 kb
...

I have started some utility here: https://github.com/StephanEwen/incubator-flink/tree/mem_size

That means that we would keep the PR in this form and add memory configuration parsing as a followup.

StephanEwen · 2017-05-06T13:30:05Z

Merging this.
I filed a follow-up JIRA to address the "configuration with units" to make sure all memory-related parameters behave the same way, without loss of byte precision where needed: https://issues.apache.org/jira/browse/FLINK-6469

Instead, allow the configuration with the following three new (more flexible) parameters: * "taskmanager.network.memory.fraction": fraction of JVM memory to use for network buffers (default: 0.1) * "taskmanager.network.memory.min": minimum memory size for network buffers (default: 64 MB) * "taskmanager.network.memory.max": maximum memory size for network buffers (default: 1 GB) This closes apache#3721

Nico Kruber added 9 commits April 18, 2017 13:54

[hotfix] fix typo in error message

207a104

[hotfix] fix typo in taskmanager.sh usage string

c64716c

[FLINK-4545] adapt taskmanager.sh to take network buffers memory into…

a7f22ce

… account

[FLINK-4545] add configuration checks for the new network buffer memo…

f3497f3

…ry config

[FLINK-4545] add unit tests using the new network configuration param…

d2bb1a4

…eters and methods

[FLINK-4545] add unit tests for heap size calculation in shell scripts

e7c3395

These verify that the results are the same as in the calculation done by Java.

[FLINK-4545] update the docs with the changed network buffer parameter

a068570

Also update the descriptions of taskmanager.memory.fraction not being relative to the full size of taskmanager.heap.mb but that network buffer memory is subtracted before!

NicoK force-pushed the flink-4545 branch from c48beb0 to 10835c2 Compare April 18, 2017 11:55

zentol reviewed Apr 18, 2017

View reviewed changes

[FLINK-4545] address some PR comments by @zentol

448e95c

asfgit closed this in 0bb49e5 May 6, 2017

rmetzger added the component=Runtime/Network label Mar 14, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-4545] replace the network buffers parameter #3721

[FLINK-4545] replace the network buffers parameter #3721

NicoK commented Apr 13, 2017

zentol commented Apr 18, 2017

zentol Apr 18, 2017

NicoK Apr 18, 2017

zentol left a comment

zentol Apr 18, 2017

NicoK Apr 19, 2017

zentol Apr 19, 2017

NicoK Apr 19, 2017

zentol Apr 18, 2017

zentol Apr 18, 2017

NicoK Apr 19, 2017

zentol Apr 18, 2017

zentol Apr 18, 2017

NicoK Apr 19, 2017

zentol Apr 18, 2017

NicoK Apr 19, 2017

zentol Apr 18, 2017

NicoK Apr 19, 2017

zentol Apr 19, 2017

zentol Apr 18, 2017

zentol Apr 18, 2017

NicoK Apr 19, 2017

zentol Apr 18, 2017

NicoK commented Apr 19, 2017

StephanEwen commented May 4, 2017

StephanEwen commented May 4, 2017 •

edited

Loading

StephanEwen commented May 5, 2017

StephanEwen commented May 6, 2017

		public static long calculateNetworkBuf(long totalJavaMemorySize, Configuration config) {
		assert totalJavaMemorySize > 0;


		long javaMem = Math.max(max + 1, ran.nextLong());

		final long networkBufMem = TaskManagerServices.calculateNetworkBuf(javaMem, config);


		float frac = Math.max(ran.nextFloat(), Float.MIN_VALUE);

		// long min = Math.max(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue(), ran.nextLong());

[FLINK-4545] replace the network buffers parameter #3721

[FLINK-4545] replace the network buffers parameter #3721

Conversation

NicoK commented Apr 13, 2017

zentol commented Apr 18, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zentol left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

NicoK commented Apr 19, 2017

StephanEwen commented May 4, 2017

StephanEwen commented May 4, 2017 • edited Loading

StephanEwen commented May 5, 2017

StephanEwen commented May 6, 2017

StephanEwen commented May 4, 2017 •

edited

Loading