CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable,
    ResourceAllocationCommitter, MutableConfScheduler { static void schedule(CapacityScheduler cs) throws InterruptedException{ int current = 0; Collection<FiCaSchedulerNode> nodes = cs.nodeTracker.getAllNodes(); int nodeSize = nodes.size(); if(nodeSize == 0) { return; } int start = random.nextInt(nodeSize); boolean printSkipedNodeLogging = false; if (Time.monotonicNow() / 1000 % 10 == 0) { printSkipedNodeLogging = (!printedVerboseLoggingForAsyncScheduling); } else { printedVerboseLoggingForAsyncScheduling = false; } for (FiCaSchedulerNode node : nodes) { if (current++ >= start) { if (shouldSkipNodeSchedule(node, cs, printSkipedNodeLogging)) { continue; } cs.allocateContainersToNode(node.getNodeID(), false); } } current = 0; for (FiCaSchedulerNode node : nodes) { if (current++ > start) { break; } if (shouldSkipNodeSchedule(node, cs, printSkipedNodeLogging)) { continue; } cs.allocateContainersToNode(node.getNodeID(), false); } if (printSkipedNodeLogging) { printedVerboseLoggingForAsyncScheduling = true; } Thread.sleep(cs.getAsyncScheduleInterval()); } CapacityScheduler(); @Override void setConf(Configuration conf); @Override Configuration getConf(); @Override QueueMetrics getRootQueueMetrics(); CSQueue getRootQueue(); @Override CapacitySchedulerConfiguration getConfiguration(); @Override RMContainerTokenSecretManager getContainerTokenSecretManager(); @Override ResourceCalculator getResourceCalculator(); @VisibleForTesting void setResourceCalculator(ResourceCalculator rc); @Override int getNumClusterNodes(); @Override RMContext getRMContext(); @Override void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); void reinitialize(Configuration newConf, RMContext rmContext,
         boolean validation); @Override void reinitialize(Configuration newConf, RMContext rmContext); @VisibleForTesting PlacementRule getUserGroupMappingPlacementRule(); PlacementRule getAppNameMappingPlacementRule(); @VisibleForTesting void updatePlacementRules(); @Override CSQueue getQueue(String queueName); String normalizeQueueName(String name); boolean isAmbiguous(String queueName); @Override @Lock(Lock.NoLock.class) Allocation allocate(ApplicationAttemptId applicationAttemptId,
      List<ResourceRequest> ask, List<SchedulingRequest> schedulingRequests,
      List<ContainerId> release, List<String> blacklistAdditions,
      List<String> blacklistRemovals, ContainerUpdates updateRequests); @Override @Lock(Lock.NoLock.class) QueueInfo getQueueInfo(String queueName,
      boolean includeChildQueues, boolean recursive); @Override @Lock(Lock.NoLock.class) List<QueueUserACLInfo> getQueueUserAclInfo(); @Override void handle(SchedulerEvent event); @Lock(Lock.NoLock.class) @VisibleForTesting @Override FiCaSchedulerApp getApplicationAttempt(
      ApplicationAttemptId applicationAttemptId); @Lock(Lock.NoLock.class) FiCaSchedulerNode getNode(NodeId nodeId); @Lock(Lock.NoLock.class) List<FiCaSchedulerNode> getAllNodes(); @Override @Lock(Lock.NoLock.class) void recover(RMState state); @Override void killReservedContainer(RMContainer container); @Override void markContainerForPreemption(ApplicationAttemptId aid,
      RMContainer cont); @VisibleForTesting @Override void killContainer(RMContainer container); void markContainerForKillable(
      RMContainer killableContainer); @Override boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); boolean isSystemAppsLimitReached(); @Override void removeQueue(String queueName); @Override void addQueue(Queue queue); @Override void setEntitlement(String inQueue, QueueEntitlement entitlement); @Override String moveApplication(ApplicationId appId,
      String targetQueueName); @Override void preValidateMoveApplication(ApplicationId appId,
      String newQueue); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Resource getMaximumResourceCapability(String queueName); @Override Set<String> getPlanQueues(); @Override Priority checkAndGetApplicationPriority(
          Priority priorityRequestedByApp, UserGroupInformation user,
          String queuePath, ApplicationId applicationId); @Override Priority updateApplicationPriority(Priority newPriority,
      ApplicationId applicationId, SettableFuture<Object> future,
      UserGroupInformation user); @Override PreemptionManager getPreemptionManager(); @Override ResourceUsage getClusterResourceUsage(); @VisibleForTesting void submitResourceCommitRequest(Resource cluster,
      CSAssignment csAssignment); @Override boolean attemptAllocationOnNode(SchedulerApplicationAttempt appAttempt,
      SchedulingRequest schedulingRequest, SchedulerNode schedulerNode); @VisibleForTesting ResourceCommitRequest<FiCaSchedulerApp, FiCaSchedulerNode> createResourceCommitRequest(CSAssignment csAssignment); @Override boolean tryCommit(Resource cluster, ResourceCommitRequest r,
      boolean updatePending); int getAsyncSchedulingPendingBacklogs(); @Override CapacitySchedulerQueueManager getCapacitySchedulerQueueManager(); WorkflowPriorityMappingsManager getWorkflowPriorityMappingsManager(); boolean moveReservedContainer(RMContainer toBeMovedContainer,
      FiCaSchedulerNode targetNode); @Override long checkAndGetApplicationLifetime(String queueName,
      long lifetimeRequestedByApp); @Override long getMaximumApplicationLifetime(String queueName); @Override boolean isConfigurationMutable(); @Override MutableConfigurationProvider getMutableConfProvider(); @Override void resetSchedulerMetrics(); boolean isMultiNodePlacementEnabled(); int getNumAsyncSchedulerThreads(); @VisibleForTesting void setMaxRunningAppsEnforcer(CSMaxRunningAppsEnforcer enforcer); @Override boolean placementConstraintEnabled(); @VisibleForTesting void setActivitiesManagerEnabled(boolean enabled);  }
@Test public void testResourceOverCommit() throws Exception { Configuration conf = new Configuration(); conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class); MockRM rm = new MockRM(conf); rm.start(); ResourceScheduler scheduler = rm.getResourceScheduler(); MockNM nm = rm.registerNode("127.0.0.1:1234", 4 * GB); NodeId nmId = nm.getNodeId(); RMApp app = MockRMAppSubmitter.submitWithMemory(2048, rm); nm.nodeHeartbeat(true); RMAppAttempt attempt1 = app.getCurrentAppAttempt(); MockAM am = rm.sendAMLaunched(attempt1.getAppAttemptId()); am.registerAppAttempt(); assertMemory(scheduler, nmId, 2 * GB, 2 * GB); am.addRequests(new String[] {"127.0.0.1", "127.0.0.2"}, 2 * GB, 1, 1); AllocateResponse alloc1Response = am.schedule(); nm.nodeHeartbeat(true); while (alloc1Response.getAllocatedContainers().isEmpty()) { LOG.info("Waiting for containers to be created for app 1..."); Thread.sleep(100); alloc1Response = am.schedule(); } List<Container> allocated1 = alloc1Response.getAllocatedContainers(); assertEquals(1, allocated1.size()); Container c1 = allocated1.get(0); assertEquals(2 * GB, c1.getResource().getMemorySize()); assertEquals(nmId, c1.getNodeId()); assertMemory(scheduler, nmId, 4 * GB, 0); nm.nodeHeartbeat(true); assertEquals(4 * GB, nm.getCapability().getMemorySize()); updateNodeResource(rm, nmId, 2 * GB, 2, -1); waitMemory(scheduler, nmId, 4 * GB, -2 * GB, 200, 5 * 1000); assertNoPreemption(am.schedule().getPreemptionMessage()); nm.nodeHeartbeat(true); assertEquals(2 * GB, nm.getCapability().getMemorySize()); ContainerStatus containerStatus = BuilderUtils.newContainerStatus( c1.getId(), ContainerState.COMPLETE, "", 0, c1.getResource()); nm.containerStatus(containerStatus); LOG.info("Waiting for containers to be finished for app 1..."); GenericTestUtils.waitFor( () -> attempt1.getJustFinishedContainers().size() == 1, 100, 2000); assertEquals(1, am.schedule().getCompletedContainersStatuses().size()); assertMemory(scheduler, nmId, 2 * GB, 0); am.addRequests(new String[] {"127.0.0.1", "127.0.0.2"}, 3 * GB, 1, 1); AllocateResponse allocResponse2 = am.schedule(); assertTrue("Shouldn't have enough resource to allocate containers", allocResponse2.getAllocatedContainers().isEmpty()); for (int i = 0; i < 10; i++) { Thread.sleep(100); allocResponse2 = am.schedule(); assertTrue("Shouldn't have enough resource to allocate containers", allocResponse2.getAllocatedContainers().isEmpty()); } updateNodeResource(rm, nmId, 5 * GB, 2, -1); waitMemory(scheduler, nmId, 2 * GB, 3 * GB, 100, 5 * 1000); nm.nodeHeartbeat(true); while (allocResponse2.getAllocatedContainers().isEmpty()) { LOG.info("Waiting for containers to be created for app 1..."); Thread.sleep(100); allocResponse2 = am.schedule(); } assertEquals(1, allocResponse2.getAllocatedContainers().size()); Container c2 = allocResponse2.getAllocatedContainers().get(0); assertEquals(3 * GB, c2.getResource().getMemorySize()); assertEquals(nmId, c2.getNodeId()); assertMemory(scheduler, nmId, 5 * GB, 0); updateNodeResource(rm, nmId, 3 * GB, 2, 2 * 1000); waitMemory(scheduler, nmId, 5 * GB, -2 * GB, 200, 5 * 1000); PreemptionMessage preemptMsg = am.schedule().getPreemptionMessage(); assertPreemption(c2.getId(), preemptMsg); updateNodeResource(rm, nmId, 5 * GB, 2, -1); waitMemory(scheduler, nmId, 5 * GB, 0, 200, 5 * 1000); Thread.sleep(3 * 1000); assertMemory(scheduler, nmId, 5 * GB, 0); long t0 = Time.now(); updateNodeResource(rm, nmId, 3 * GB, 2, 2 * 1000); waitMemory(scheduler, nmId, 5 * GB, -2 * GB, 200, 5 * 1000); preemptMsg = am.schedule().getPreemptionMessage(); assertPreemption(c2.getId(), preemptMsg); GenericTestUtils.waitFor(() -> { try { nm.nodeHeartbeat(true); } catch (Exception e) { LOG.error("Cannot heartbeat", e); } SchedulerNodeReport report = scheduler.getNodeReport(nmId); return report.getAvailableResource().getMemorySize() > 0; }, 200, 5 * 1000); assertMemory(scheduler, nmId, 2 * GB, 1 * GB); List<ContainerStatus> completedContainers = am.schedule().getCompletedContainersStatuses(); assertEquals(1, completedContainers.size()); ContainerStatus c2status = completedContainers.get(0); assertContainerKilled(c2.getId(), c2status); assertTime(2000, Time.now() - t0); rm.stop(); }
@Test(timeout = 30000) public void testAppReservationWithDominantResourceCalculator() throws Exception { CapacitySchedulerConfiguration csconf = new CapacitySchedulerConfiguration(); csconf.setResourceComparator(DominantResourceCalculator.class); YarnConfiguration conf = new YarnConfiguration(csconf); conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class); MockRM rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 10 * GB, 1); MockNM nm2 = rm.registerNode("127.0.0.1:1235", 10 * GB, 4); rm.registerNode("127.0.0.1:1236", 10 * GB, 4); RMApp app1 = MockRMAppSubmitter.submitWithMemory(1024, rm); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); SchedulerNodeReport report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); Assert.assertEquals(1 * GB, report_nm1.getUsedResource().getMemorySize()); Assert.assertEquals(9 * GB, report_nm1.getAvailableResource().getMemorySize()); am1.addRequests(new String[] { "127.0.0.1", "127.0.0.2" }, 1 * GB, 1, 1); am1.schedule(); nm1.nodeHeartbeat(true); Thread.sleep(1000); AllocateResponse allocResponse = am1.schedule(); ApplicationResourceUsageReport report = rm.getResourceScheduler().getAppResourceUsageReport( attempt1.getAppAttemptId()); Assert.assertEquals(0, allocResponse.getAllocatedContainers().size()); Assert.assertEquals(0, report.getNumReservedContainers()); nm2.nodeHeartbeat(true); while (allocResponse.getAllocatedContainers().size() == 0) { Thread.sleep(100); allocResponse = am1.schedule(); } report = rm.getResourceScheduler().getAppResourceUsageReport( attempt1.getAppAttemptId()); Assert.assertEquals(1, allocResponse.getAllocatedContainers().size()); Assert.assertEquals(0, report.getNumReservedContainers()); rm.stop(); }