TaskScheduler { static boolean isDAG(final List<JobContainerRequest> containersRequests) { Set<JobContainerRequest> visited = new HashSet<>(); for (JobContainerRequest containerRequest : containersRequests) { if (!visited.contains(containerRequest) && !isSubgraphDAG(containerRequest, new ArrayList<>(), containersRequests, visited)) { return false; } } return true; } 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 testIsDAG() { JobContainerRequest dbJob = new JobContainerRequest("db", 1, 0L, 1, 0, 1, "", new ArrayList<>()); JobContainerRequest workerJob = new JobContainerRequest("worker", 1, 0L, 1, 0, 1, "", Arrays.asList("db", "ps")); JobContainerRequest psJob = new JobContainerRequest("ps", 1, 0L, 1, 0, 1, "", Arrays.asList("db")); JobContainerRequest dbWriterJob = new JobContainerRequest("dbwriter", 1, 0L, 1, 0, 1, "", new ArrayList<>()); JobContainerRequest cleanupJob = new JobContainerRequest("cleanup", 1, 0L, 1, 0, 1, "", Arrays.asList("db", "worker")); List<JobContainerRequest> requests = Arrays.asList(dbJob, workerJob, psJob, dbWriterJob, cleanupJob); boolean actualIsDAG = TaskScheduler.isDAG(requests); assertTrue(actualIsDAG); }
@Test public void testNotDAG() { JobContainerRequest dbJob = new JobContainerRequest("db", 1, 0L, 1, 0, 1, "", Arrays.asList("cleanup")); JobContainerRequest workerJob = new JobContainerRequest("worker", 1, 0L, 1, 0, 1, "", Arrays.asList("db", "ps")); JobContainerRequest psJob = new JobContainerRequest("ps", 1, 0L, 1, 0, 1, "", Arrays.asList("db")); JobContainerRequest dbWriterJob = new JobContainerRequest("dbwriter", 1, 0L, 1, 0, 1, "", new ArrayList<>()); JobContainerRequest cleanupJob = new JobContainerRequest("cleanup", 1, 0L, 1, 0, 1, "", Arrays.asList("db", "worker")); List<JobContainerRequest> requests = Arrays.asList(dbJob, workerJob, psJob, dbWriterJob, cleanupJob); boolean actualIsDAG = TaskScheduler.isDAG(requests); assertFalse(actualIsDAG); }