TaskScheduler { public void scheduleTasks() { final List<JobContainerRequest> requests = session.getContainersRequests(); if (!isDAG(requests)) { LOG.error("TonY execution graph does not form a DAG, exiting."); session.setFinalStatus(FinalApplicationStatus.FAILED, "App failed due to it not being a DAG."); dependencyCheckPassed = false; return; } buildTaskDependencyGraph(requests); for (JobContainerRequest request : requests) { if (checkDependencySatisfied(request)) { scheduleJob(request); } } } TaskScheduler(TonySession session, AMRMClientAsync<AMRMClient.ContainerRequest> amRMClient, Map<String, LocalResource> localResources,
      FileSystem resourceFs, Configuration tonyConf, Map<String, Map<String, LocalResource>> jobTypeToContainerResources); void scheduleTasks();  }
@Test public void testScheduleTasks() { JobContainerRequest dbJob = new JobContainerRequest("db", 1, 0L, 1, 0, 1, "", new ArrayList<>()); JobContainerRequest workerJob = new JobContainerRequest("worker", 2, 0L, 1, 0, 1, "", Arrays.asList("db", "ps")); JobContainerRequest psJob = new JobContainerRequest("ps", 4, 0L, 1, 0, 1, "", Arrays.asList("db")); JobContainerRequest dbWriterJob = new JobContainerRequest("dbwriter", 8, 0L, 1, 0, 1, "", new ArrayList<>()); JobContainerRequest cleanupJob = new JobContainerRequest("cleanup", 16, 0L, 1, 0, 1, "", Arrays.asList("db", "worker")); List<JobContainerRequest> requests = Arrays.asList(dbJob, workerJob, psJob, dbWriterJob, cleanupJob); when(session.getContainersRequests()).thenReturn(requests); when(session.getContainerRequestForType("db")).thenReturn(dbJob); when(session.getContainerRequestForType("worker")).thenReturn(workerJob); when(session.getContainerRequestForType("ps")).thenReturn(psJob); when(session.getContainerRequestForType("dbwriter")).thenReturn(dbWriterJob); when(session.getContainerRequestForType("cleanup")).thenReturn(cleanupJob); taskScheduler.scheduleTasks(); assertTrue(taskScheduler.checkDependencySatisfied(dbJob)); assertTrue(taskScheduler.checkDependencySatisfied(dbWriterJob)); assertFalse(taskScheduler.checkDependencySatisfied(psJob)); assertFalse(taskScheduler.checkDependencySatisfied(cleanupJob)); assertFalse(taskScheduler.checkDependencySatisfied(workerJob)); for (int i = 0; i < dbWriterJob.getNumInstances(); i++) { taskScheduler.registerDependencyCompleted("dbwriter"); } assertTrue(taskScheduler.checkDependencySatisfied(dbJob)); assertTrue(taskScheduler.checkDependencySatisfied(dbWriterJob)); assertFalse(taskScheduler.checkDependencySatisfied(psJob)); assertFalse(taskScheduler.checkDependencySatisfied(cleanupJob)); assertFalse(taskScheduler.checkDependencySatisfied(workerJob)); for (int i = 0; i < dbJob.getNumInstances(); i++) { taskScheduler.registerDependencyCompleted("db"); } assertTrue(taskScheduler.checkDependencySatisfied(dbJob)); assertTrue(taskScheduler.checkDependencySatisfied(dbWriterJob)); assertTrue(taskScheduler.checkDependencySatisfied(psJob)); assertFalse(taskScheduler.checkDependencySatisfied(cleanupJob)); assertFalse(taskScheduler.checkDependencySatisfied(workerJob)); for (int i = 0; i < psJob.getNumInstances() - 1; i++) { taskScheduler.registerDependencyCompleted("ps"); } assertTrue(taskScheduler.checkDependencySatisfied(dbJob)); assertTrue(taskScheduler.checkDependencySatisfied(dbWriterJob)); assertTrue(taskScheduler.checkDependencySatisfied(psJob)); assertFalse(taskScheduler.checkDependencySatisfied(cleanupJob)); assertFalse(taskScheduler.checkDependencySatisfied(workerJob)); taskScheduler.registerDependencyCompleted("ps"); assertTrue(taskScheduler.checkDependencySatisfied(dbJob)); assertTrue(taskScheduler.checkDependencySatisfied(dbWriterJob)); assertTrue(taskScheduler.checkDependencySatisfied(psJob)); assertFalse(taskScheduler.checkDependencySatisfied(cleanupJob)); assertTrue(taskScheduler.checkDependencySatisfied(workerJob)); for (int i = 0; i < workerJob.getNumInstances(); i++) { taskScheduler.registerDependencyCompleted("worker"); } assertTrue(taskScheduler.checkDependencySatisfied(dbJob)); assertTrue(taskScheduler.checkDependencySatisfied(dbWriterJob)); assertTrue(taskScheduler.checkDependencySatisfied(psJob)); assertTrue(taskScheduler.checkDependencySatisfied(cleanupJob)); assertTrue(taskScheduler.checkDependencySatisfied(workerJob)); }