TaskManagerServices { @SuppressWarnings("deprecation") public static long calculateNetworkBufferMemory(long totalJavaMemorySize, Configuration config) { Preconditions.checkArgument(totalJavaMemorySize > 0); int segmentSize = config.getInteger(TaskManagerOptions.MEMORY_SEGMENT_SIZE); final long networkBufBytes; if (TaskManagerServicesConfiguration.hasNewNetworkBufConf(config)) { float networkBufFraction = config.getFloat(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_FRACTION); long networkBufMin = config.getLong(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MIN); long networkBufMax = config.getLong(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MAX); TaskManagerServicesConfiguration .checkNetworkBufferConfig(segmentSize, networkBufFraction, networkBufMin, networkBufMax); networkBufBytes = Math.min(networkBufMax, Math.max(networkBufMin, (long) (networkBufFraction * totalJavaMemorySize))); TaskManagerServicesConfiguration .checkConfigParameter(networkBufBytes < totalJavaMemorySize, "(" + networkBufFraction + ", " + networkBufMin + ", " + networkBufMax + ")", "(" + TaskManagerOptions.NETWORK_BUFFERS_MEMORY_FRACTION.key() + ", " + TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MIN.key() + ", " + TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MAX.key() + ")", "Network buffer memory size too large: " + networkBufBytes + " >= " + totalJavaMemorySize + " (total JVM memory size)"); } else { int numNetworkBuffers = config.getInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS); networkBufBytes = (long) numNetworkBuffers * (long) segmentSize; TaskManagerServicesConfiguration.checkNetworkConfigOld(numNetworkBuffers); TaskManagerServicesConfiguration .checkConfigParameter(networkBufBytes < totalJavaMemorySize, networkBufBytes, TaskManagerOptions.NETWORK_NUM_BUFFERS.key(), "Network buffer memory size too large: " + networkBufBytes + " >= " + totalJavaMemorySize + " (total JVM memory size)"); } return networkBufBytes; } private  TaskManagerServices(
		TaskManagerLocation taskManagerLocation,
		MemoryManager memoryManager,
		IOManager ioManager,
		NetworkEnvironment networkEnvironment,
		BroadcastVariableManager broadcastVariableManager,
		FileCache fileCache,
		TaskSlotTable taskSlotTable,
		JobManagerTable jobManagerTable,
		JobLeaderService jobLeaderService); MemoryManager getMemoryManager(); IOManager getIOManager(); NetworkEnvironment getNetworkEnvironment(); TaskManagerLocation getTaskManagerLocation(); BroadcastVariableManager getBroadcastVariableManager(); FileCache getFileCache(); TaskSlotTable getTaskSlotTable(); JobManagerTable getJobManagerTable(); JobLeaderService getJobLeaderService(); static TaskManagerServices fromConfiguration(
			TaskManagerServicesConfiguration taskManagerServicesConfiguration,
			ResourceID resourceID); @SuppressWarnings("deprecation") static long calculateNetworkBufferMemory(long totalJavaMemorySize, Configuration config); static long calculateNetworkBufferMemory(TaskManagerServicesConfiguration tmConfig); static long calculateHeapSizeMB(long totalJavaMemorySizeMB, Configuration config);  }
@SuppressWarnings("deprecation") @Test public void calculateNetworkBufOld() throws Exception { Configuration config = new Configuration(); config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 1); assertEquals(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue().longValue(), TaskManagerServices.calculateNetworkBufferMemory(10L << 20, config)); assertEquals(TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue().longValue(), TaskManagerServices.calculateNetworkBufferMemory(64L << 20, config)); int numBuffers = (int) ((2L << 32) / TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue()); config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, numBuffers); assertEquals(2L << 32, TaskManagerServices.calculateNetworkBufferMemory(2L << 33, config)); }
@Test public void calculateNetworkBufNew() throws Exception { Configuration config = new Configuration(); final Float defaultFrac = TaskManagerOptions.NETWORK_BUFFERS_MEMORY_FRACTION.defaultValue(); final Long defaultMin = TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue(); final Long defaultMax = TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MAX.defaultValue(); assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), defaultMin, defaultMax), TaskManagerServices.calculateNetworkBufferMemory((64L << 20 + 1), config)); assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), defaultMin, defaultMax), TaskManagerServices.calculateNetworkBufferMemory((10L << 30), config)); calculateNetworkBufNew(config); }
@SuppressWarnings("deprecation") @Test public void calculateNetworkBufMixed() throws Exception { Configuration config = new Configuration(); config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 1); final Float defaultFrac = TaskManagerOptions.NETWORK_BUFFERS_MEMORY_FRACTION.defaultValue(); final Long defaultMin = TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue(); final Long defaultMax = TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MAX.defaultValue(); Configuration config1 = config.clone(); config1.setFloat(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_FRACTION, 0.1f); assertEquals(enforceBounds((long) (0.1f * (10L << 20)), defaultMin, defaultMax), TaskManagerServices.calculateNetworkBufferMemory((64L << 20 + 1), config1)); assertEquals(enforceBounds((long) (0.1f * (10L << 30)), defaultMin, defaultMax), TaskManagerServices.calculateNetworkBufferMemory((10L << 30), config1)); config1 = config.clone(); long newMin = TaskManagerOptions.MEMORY_SEGMENT_SIZE.defaultValue(); config1.setLong(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MIN, newMin); assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), newMin, defaultMax), TaskManagerServices.calculateNetworkBufferMemory((10L << 20), config1)); assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), newMin, defaultMax), TaskManagerServices.calculateNetworkBufferMemory((10L << 30), config1)); config1 = config.clone(); long newMax = Math.max(64L << 20 + 1, TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MIN.defaultValue()); config1.setLong(TaskManagerOptions.NETWORK_BUFFERS_MEMORY_MAX, newMax); assertEquals(enforceBounds((long) (defaultFrac * (10L << 20)), defaultMin, newMax), TaskManagerServices.calculateNetworkBufferMemory((64L << 20 + 1), config1)); assertEquals(enforceBounds((long) (defaultFrac * (10L << 30)), defaultMin, newMax), TaskManagerServices.calculateNetworkBufferMemory((10L << 30), config1)); assertTrue(TaskManagerServicesConfiguration.hasNewNetworkBufConf(config1)); calculateNetworkBufNew(config); }
@Test public void calculateNetworkBufFromHeapSize() throws Exception { PowerMockito.mockStatic(EnvironmentInformation.class); when(EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag()).thenReturn(1000L << 20); when(EnvironmentInformation.getMaxJvmHeapMemory()).thenReturn(1000L << 20); TaskManagerServicesConfiguration tmConfig; tmConfig = getTmConfig(TaskManagerOptions.MANAGED_MEMORY_SIZE.defaultValue(), TaskManagerOptions.MANAGED_MEMORY_FRACTION.defaultValue(), 0.1f, 60L << 20, 1L << 30, MemoryType.HEAP); when(EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag()).thenReturn(1000L << 20); assertEquals(100L << 20, TaskManagerServices.calculateNetworkBufferMemory(tmConfig)); tmConfig = getTmConfig(10, TaskManagerOptions.MANAGED_MEMORY_FRACTION.defaultValue(), 0.1f, 60L << 20, 1L << 30, MemoryType.OFF_HEAP); when(EnvironmentInformation.getMaxJvmHeapMemory()).thenReturn(890L << 20); assertEquals((100L << 20) + 1 , TaskManagerServices.calculateNetworkBufferMemory(tmConfig)); tmConfig = getTmConfig(-1, 0.1f, 0.1f, 60L << 20, 1L << 30, MemoryType.OFF_HEAP); when(EnvironmentInformation.getMaxJvmHeapMemory()).thenReturn(810L << 20); assertEquals((100L << 20) + 1 , TaskManagerServices.calculateNetworkBufferMemory(tmConfig)); }