@Test public void testUpdateMaxAllocationUsesTotal() throws IOException { final int configuredMaxVCores = 20; final int configuredMaxMemory = 10 * 1024; Resource configuredMaximumResource = Resource.newInstance (configuredMaxMemory, configuredMaxVCores, configuredMaxVCores); configureScheduler(); YarnConfiguration conf = getConf(); conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES, configuredMaxVCores); conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB, configuredMaxMemory); conf.setLong( YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_SCHEDULING_WAIT_MS, 0); MockRM rm = new MockRM(conf); try { rm.start(); AbstractYarnScheduler scheduler = (AbstractYarnScheduler) rm .getResourceScheduler(); Resource emptyResource = Resource.newInstance(0, 0, 0); Resource fullResource1 = Resource.newInstance(1024, 5, 5); Resource fullResource2 = Resource.newInstance(2048, 10, 10); SchedulerNode mockNode1 = mock(SchedulerNode.class); when(mockNode1.getNodeID()).thenReturn(NodeId.newInstance("foo", 8080)); when(mockNode1.getAvailableResource()).thenReturn(emptyResource); when(mockNode1.getTotalResource()).thenReturn(fullResource1); SchedulerNode mockNode2 = mock(SchedulerNode.class); when(mockNode1.getNodeID()).thenReturn(NodeId.newInstance("bar", 8081)); when(mockNode2.getAvailableResource()).thenReturn(emptyResource); when(mockNode2.getTotalResource()).thenReturn(fullResource2); verifyMaximumResourceCapability(configuredMaximumResource, scheduler); scheduler.nodes = new HashMap<NodeId, SchedulerNode>(); scheduler.nodes.put(mockNode1.getNodeID(), mockNode1); scheduler.updateMaximumAllocation(mockNode1, true); verifyMaximumResourceCapability(fullResource1, scheduler); scheduler.nodes.put(mockNode2.getNodeID(), mockNode2); scheduler.updateMaximumAllocation(mockNode2, true); verifyMaximumResourceCapability(fullResource2, scheduler); scheduler.nodes.remove(mockNode2.getNodeID()); scheduler.updateMaximumAllocation(mockNode2, false); verifyMaximumResourceCapability(fullResource1, scheduler); scheduler.nodes.remove(mockNode1.getNodeID()); scheduler.updateMaximumAllocation(mockNode1, false); verifyMaximumResourceCapability(configuredMaximumResource, scheduler); } finally { rm.stop(); } }
protected void updateMaximumAllocation(SchedulerNode node, boolean add) { Resource totalResource = node.getTotalResource(); maxAllocWriteLock.lock(); try { if (add) { int nodeMemory = totalResource.getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; maximumAllocation.setMemory(Math.min( configuredMaximumAllocation.getMemory(), maxNodeMemory)); } int nodeVCores = totalResource.getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; maximumAllocation.setVirtualCores(Math.min( configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } int nodeGCores = totalResource.getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; maximumAllocation.setGpuCores(Math.min( configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } else { if (maxNodeMemory == totalResource.getMemory()) { maxNodeMemory = -1; } if (maxNodeVCores == totalResource.getVirtualCores()) { maxNodeVCores = -1; } if (maxNodeGCores == totalResource.getGpuCores()) { maxNodeGCores = -1; } if (maxNodeMemory == -1 || maxNodeVCores == -1 || maxNodeGCores == -1) { for (Map.Entry<NodeId, N> nodeEntry : nodes.entrySet()) { int nodeMemory = nodeEntry.getValue().getTotalResource().getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; } int nodeVCores = nodeEntry.getValue().getTotalResource().getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; } int nodeGCores = nodeEntry.getValue().getTotalResource().getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; } } if (maxNodeMemory == -1) { maximumAllocation.setMemory(configuredMaximumAllocation.getMemory()); } else { maximumAllocation.setMemory( Math.min(configuredMaximumAllocation.getMemory(), maxNodeMemory)); } if (maxNodeVCores == -1) { maximumAllocation.setVirtualCores(configuredMaximumAllocation.getVirtualCores()); } else { maximumAllocation.setVirtualCores( Math.min(configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } if (maxNodeGCores == -1) { maximumAllocation.setGpuCores(configuredMaximumAllocation.getGpuCores()); } else { maximumAllocation.setGpuCores( Math.min(configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } } } finally { maxAllocWriteLock.unlock(); } }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { protected void updateMaximumAllocation(SchedulerNode node, boolean add) { Resource totalResource = node.getTotalResource(); maxAllocWriteLock.lock(); try { if (add) { int nodeMemory = totalResource.getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; maximumAllocation.setMemory(Math.min( configuredMaximumAllocation.getMemory(), maxNodeMemory)); } int nodeVCores = totalResource.getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; maximumAllocation.setVirtualCores(Math.min( configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } int nodeGCores = totalResource.getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; maximumAllocation.setGpuCores(Math.min( configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } else { if (maxNodeMemory == totalResource.getMemory()) { maxNodeMemory = -1; } if (maxNodeVCores == totalResource.getVirtualCores()) { maxNodeVCores = -1; } if (maxNodeGCores == totalResource.getGpuCores()) { maxNodeGCores = -1; } if (maxNodeMemory == -1 || maxNodeVCores == -1 || maxNodeGCores == -1) { for (Map.Entry<NodeId, N> nodeEntry : nodes.entrySet()) { int nodeMemory = nodeEntry.getValue().getTotalResource().getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; } int nodeVCores = nodeEntry.getValue().getTotalResource().getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; } int nodeGCores = nodeEntry.getValue().getTotalResource().getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; } } if (maxNodeMemory == -1) { maximumAllocation.setMemory(configuredMaximumAllocation.getMemory()); } else { maximumAllocation.setMemory( Math.min(configuredMaximumAllocation.getMemory(), maxNodeMemory)); } if (maxNodeVCores == -1) { maximumAllocation.setVirtualCores(configuredMaximumAllocation.getVirtualCores()); } else { maximumAllocation.setVirtualCores( Math.min(configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } if (maxNodeGCores == -1) { maximumAllocation.setGpuCores(configuredMaximumAllocation.getGpuCores()); } else { maximumAllocation.setGpuCores( Math.min(configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } } } finally { maxAllocWriteLock.unlock(); } } }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { protected void updateMaximumAllocation(SchedulerNode node, boolean add) { Resource totalResource = node.getTotalResource(); maxAllocWriteLock.lock(); try { if (add) { int nodeMemory = totalResource.getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; maximumAllocation.setMemory(Math.min( configuredMaximumAllocation.getMemory(), maxNodeMemory)); } int nodeVCores = totalResource.getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; maximumAllocation.setVirtualCores(Math.min( configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } int nodeGCores = totalResource.getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; maximumAllocation.setGpuCores(Math.min( configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } else { if (maxNodeMemory == totalResource.getMemory()) { maxNodeMemory = -1; } if (maxNodeVCores == totalResource.getVirtualCores()) { maxNodeVCores = -1; } if (maxNodeGCores == totalResource.getGpuCores()) { maxNodeGCores = -1; } if (maxNodeMemory == -1 || maxNodeVCores == -1 || maxNodeGCores == -1) { for (Map.Entry<NodeId, N> nodeEntry : nodes.entrySet()) { int nodeMemory = nodeEntry.getValue().getTotalResource().getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; } int nodeVCores = nodeEntry.getValue().getTotalResource().getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; } int nodeGCores = nodeEntry.getValue().getTotalResource().getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; } } if (maxNodeMemory == -1) { maximumAllocation.setMemory(configuredMaximumAllocation.getMemory()); } else { maximumAllocation.setMemory( Math.min(configuredMaximumAllocation.getMemory(), maxNodeMemory)); } if (maxNodeVCores == -1) { maximumAllocation.setVirtualCores(configuredMaximumAllocation.getVirtualCores()); } else { maximumAllocation.setVirtualCores( Math.min(configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } if (maxNodeGCores == -1) { maximumAllocation.setGpuCores(configuredMaximumAllocation.getGpuCores()); } else { maximumAllocation.setGpuCores( Math.min(configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } } } finally { maxAllocWriteLock.unlock(); } } AbstractYarnScheduler(String name); }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { protected void updateMaximumAllocation(SchedulerNode node, boolean add) { Resource totalResource = node.getTotalResource(); maxAllocWriteLock.lock(); try { if (add) { int nodeMemory = totalResource.getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; maximumAllocation.setMemory(Math.min( configuredMaximumAllocation.getMemory(), maxNodeMemory)); } int nodeVCores = totalResource.getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; maximumAllocation.setVirtualCores(Math.min( configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } int nodeGCores = totalResource.getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; maximumAllocation.setGpuCores(Math.min( configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } else { if (maxNodeMemory == totalResource.getMemory()) { maxNodeMemory = -1; } if (maxNodeVCores == totalResource.getVirtualCores()) { maxNodeVCores = -1; } if (maxNodeGCores == totalResource.getGpuCores()) { maxNodeGCores = -1; } if (maxNodeMemory == -1 || maxNodeVCores == -1 || maxNodeGCores == -1) { for (Map.Entry<NodeId, N> nodeEntry : nodes.entrySet()) { int nodeMemory = nodeEntry.getValue().getTotalResource().getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; } int nodeVCores = nodeEntry.getValue().getTotalResource().getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; } int nodeGCores = nodeEntry.getValue().getTotalResource().getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; } } if (maxNodeMemory == -1) { maximumAllocation.setMemory(configuredMaximumAllocation.getMemory()); } else { maximumAllocation.setMemory( Math.min(configuredMaximumAllocation.getMemory(), maxNodeMemory)); } if (maxNodeVCores == -1) { maximumAllocation.setVirtualCores(configuredMaximumAllocation.getVirtualCores()); } else { maximumAllocation.setVirtualCores( Math.min(configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } if (maxNodeGCores == -1) { maximumAllocation.setGpuCores(configuredMaximumAllocation.getGpuCores()); } else { maximumAllocation.setGpuCores( Math.min(configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } } } finally { maxAllocWriteLock.unlock(); } } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); synchronized List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); @Override Resource getMaximumResourceCapability(String queueName); T getApplicationAttempt(ApplicationAttemptId applicationAttemptId); @Override SchedulerAppReport getSchedulerAppInfo(
      ApplicationAttemptId appAttemptId); @Override ApplicationResourceUsageReport getAppResourceUsageReport(
      ApplicationAttemptId appAttemptId); T getCurrentAttemptForContainer(ContainerId containerId); @Override RMContainer getRMContainer(ContainerId containerId); @Override SchedulerNodeReport getNodeReport(NodeId nodeId); @Override String moveApplication(ApplicationId appId, String newQueue); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); synchronized void recoverContainersOnNode(
      List<NMContainerStatus> containerReports, RMNode nm); SchedulerNode getSchedulerNode(NodeId nodeId); @Override synchronized void moveAllApps(String sourceQueue, String destQueue); @Override synchronized void killAllAppsInQueue(String queueName); synchronized void updateNodeResource(RMNode nm, 
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); List<ResourceRequest> getPendingResourceRequestsForAttempt(
      ApplicationAttemptId attemptId); }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { protected void updateMaximumAllocation(SchedulerNode node, boolean add) { Resource totalResource = node.getTotalResource(); maxAllocWriteLock.lock(); try { if (add) { int nodeMemory = totalResource.getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; maximumAllocation.setMemory(Math.min( configuredMaximumAllocation.getMemory(), maxNodeMemory)); } int nodeVCores = totalResource.getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; maximumAllocation.setVirtualCores(Math.min( configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } int nodeGCores = totalResource.getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; maximumAllocation.setGpuCores(Math.min( configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } else { if (maxNodeMemory == totalResource.getMemory()) { maxNodeMemory = -1; } if (maxNodeVCores == totalResource.getVirtualCores()) { maxNodeVCores = -1; } if (maxNodeGCores == totalResource.getGpuCores()) { maxNodeGCores = -1; } if (maxNodeMemory == -1 || maxNodeVCores == -1 || maxNodeGCores == -1) { for (Map.Entry<NodeId, N> nodeEntry : nodes.entrySet()) { int nodeMemory = nodeEntry.getValue().getTotalResource().getMemory(); if (nodeMemory > maxNodeMemory) { maxNodeMemory = nodeMemory; } int nodeVCores = nodeEntry.getValue().getTotalResource().getVirtualCores(); if (nodeVCores > maxNodeVCores) { maxNodeVCores = nodeVCores; } int nodeGCores = nodeEntry.getValue().getTotalResource().getGpuCores(); if (nodeGCores > maxNodeGCores) { maxNodeGCores = nodeGCores; } } if (maxNodeMemory == -1) { maximumAllocation.setMemory(configuredMaximumAllocation.getMemory()); } else { maximumAllocation.setMemory( Math.min(configuredMaximumAllocation.getMemory(), maxNodeMemory)); } if (maxNodeVCores == -1) { maximumAllocation.setVirtualCores(configuredMaximumAllocation.getVirtualCores()); } else { maximumAllocation.setVirtualCores( Math.min(configuredMaximumAllocation.getVirtualCores(), maxNodeVCores)); } if (maxNodeGCores == -1) { maximumAllocation.setGpuCores(configuredMaximumAllocation.getGpuCores()); } else { maximumAllocation.setGpuCores( Math.min(configuredMaximumAllocation.getGpuCores(), maxNodeGCores)); } } } } finally { maxAllocWriteLock.unlock(); } } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); synchronized List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); @Override Resource getMaximumResourceCapability(String queueName); T getApplicationAttempt(ApplicationAttemptId applicationAttemptId); @Override SchedulerAppReport getSchedulerAppInfo(
      ApplicationAttemptId appAttemptId); @Override ApplicationResourceUsageReport getAppResourceUsageReport(
      ApplicationAttemptId appAttemptId); T getCurrentAttemptForContainer(ContainerId containerId); @Override RMContainer getRMContainer(ContainerId containerId); @Override SchedulerNodeReport getNodeReport(NodeId nodeId); @Override String moveApplication(ApplicationId appId, String newQueue); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); synchronized void recoverContainersOnNode(
      List<NMContainerStatus> containerReports, RMNode nm); SchedulerNode getSchedulerNode(NodeId nodeId); @Override synchronized void moveAllApps(String sourceQueue, String destQueue); @Override synchronized void killAllAppsInQueue(String queueName); synchronized void updateNodeResource(RMNode nm, 
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); List<ResourceRequest> getPendingResourceRequestsForAttempt(
      ApplicationAttemptId attemptId);  }