Utils { public static Map<String, JobContainerRequest> parseContainerRequests(Configuration conf) { Set<String> jobNames = getAllJobTypes(conf); Set<String> untrackedJobTypes = Arrays.stream(getUntrackedJobTypes(conf)).collect(Collectors.toSet()); Map<String, JobContainerRequest> containerRequests = new HashMap<>(); int priority = 0; List<String> prepareStageTasks = new ArrayList<>(conf.getTrimmedStringCollection(TonyConfigurationKeys.APPLICATION_PREPARE_STAGE)); List<String> trainingStageTasks = new ArrayList<>(conf.getTrimmedStringCollection(TonyConfigurationKeys.APPLICATION_TRAINING_STAGE)); ensureStagedTasksIntegrity(prepareStageTasks, trainingStageTasks, jobNames); List<String> tasksToDependOn = prepareStageTasks.stream().filter(x -> !untrackedJobTypes.contains(x)).collect(Collectors.toList()); for (String jobName : jobNames) { int numInstances = conf.getInt(TonyConfigurationKeys.getInstancesKey(jobName), 0); String memoryString = conf.get(TonyConfigurationKeys.getResourceKey(jobName, Constants.MEMORY), TonyConfigurationKeys.DEFAULT_MEMORY); long memory = Long.parseLong(parseMemoryString(memoryString)); int vCores = conf.getInt(TonyConfigurationKeys.getResourceKey(jobName, Constants.VCORES), TonyConfigurationKeys.DEFAULT_VCORES); int gpus = conf.getInt(TonyConfigurationKeys.getResourceKey(jobName, Constants.GPUS), TonyConfigurationKeys.DEFAULT_GPUS); String nodeLabel = conf.get(TonyConfigurationKeys.getNodeLabelKey(jobName)); List<String> dependsOn = new ArrayList<>(); if (trainingStageTasks.contains(jobName)) { dependsOn.addAll(tasksToDependOn); } if (numInstances > 0) { containerRequests.put(jobName, new JobContainerRequest(jobName, numInstances, memory, vCores, gpus, priority, nodeLabel, dependsOn)); priority++; } } return containerRequests; } private  Utils(); static boolean poll(Callable<Boolean> func, int interval, int timeout); static T pollTillNonNull(Callable<T> func, int interval, int timeout); static String parseMemoryString(String memory); static void zipFolder(java.nio.file.Path sourceFolderPath, java.nio.file.Path zipPath); static void unzipArchive(String src, String dst); static void setCapabilityGPU(Resource resource, int gpuCount); static String constructUrl(String urlString); static String constructContainerUrl(Container container); static String constructContainerUrl(String nodeAddress, ContainerId containerId); static void printTaskUrl(TaskInfo taskInfo, Log log); static void printTonyPortalUrl(String portalUrl, String appId, Log log); static String getTonySrcZipName(String appId); static Map<String, String> parseKeyValue(String[] keyValues); static Options getCommonOptions(); static int executeShell(String taskCommand, long timeout, Map<String, String> env); static String getCurrentHostName(); static String getHostNameOrIpFromTokenConf(Configuration conf); static void addEnvironmentForResource(LocalResource resource, FileSystem fs, String envPrefix,
                                               Map<String, String> env); static Map<String, JobContainerRequest> parseContainerRequests(Configuration conf); static Set<String> getAllJobTypes(Configuration conf); static int getNumTotalTasks(Configuration conf); static String getTaskType(String confKey); static boolean isArchive(String path); static boolean renameFile(String oldName, String newName); static String constructTFConfig(String clusterSpec, String jobName, int taskIndex); static String getClientResourcesPath(String appId, String fileName); static void cleanupHDFSPath(Configuration hdfsConf, Path path); static void addResources(String[] resources, Map<String, LocalResource> resourcesMap, FileSystem fs); static void addResource(String path, Map<String, LocalResource> resourcesMap, FileSystem fs); static String buildRMUrl(Configuration yarnConf, String appId); static void printCompletedTrackedTasks(int completedTrackedTasks, int totalTrackedTasks); static String parseClusterSpecForPytorch(String clusterSpec); static String[] parseClusterSpecForMXNet(String clusterSpec); static String resolveNameToIpAddress(String hostname); static void createDirIfNotExists(FileSystem fs, Path dir, FsPermission permission); static String[] getUntrackedJobTypes(Configuration conf); static boolean isJobTypeTracked(String taskName, Configuration tonyConf); static String[] getStopOnFailureJobTypes(Configuration conf); static void uploadFileAndSetConfResources(Path hdfsPath, Path filePath, String fileName,
                                                   Configuration tonyConf, FileSystem fs,
                                                   LocalResourceType resourceType, String resourceKey); static String[] splitAddressPort(String hostname); static void appendConfResources(String key, String resource, Configuration tonyConf); static void initYarnConf(Configuration yarnConf); static void initHdfsConf(Configuration hdfsConf); static void extractResources(String appId); static Map<String, String> getContainerEnvForDocker(Configuration tonyConf, String jobType); static Map<String, String> linksToBeDisplayedOnPage(String jobId);  }
@Test public void testParseContainerRequests() { Configuration conf = new Configuration(); conf.addResource("tony-default.xml"); conf.setInt("tony.worker.instances", 3); conf.setInt("tony.evaluator.instances", 1); conf.setInt("tony.worker.gpus", 1); conf.setInt("tony.evaluator.vcores", 2); conf.setInt("tony.chief.gpus", 1); conf.setInt("tony.db.instances", 1); conf.setInt("tony.dbwriter.instances", 1); conf.setStrings("tony.application.prepare-stage", "dbwriter, db"); conf.setStrings("tony.application.untracked.jobtypes", "db"); Map<String, JobContainerRequest> requests = Utils.parseContainerRequests(conf); assertEquals(requests.get("worker").getNumInstances(), 3); assertEquals(requests.get("evaluator").getNumInstances(), 1); assertEquals(requests.get("worker").getGPU(), 1); assertEquals(requests.get("evaluator").getVCores(), 2); assertEquals(requests.get("worker").getMemory(), 2048); assertFalse(requests.containsKey("chief")); assertEquals(requests.get("worker").getDependsOn(), new ArrayList<>(Arrays.asList("dbwriter"))); assertEquals(requests.get("evaluator").getDependsOn(), new ArrayList<>(Arrays.asList("dbwriter"))); assertEquals(requests.get("db").getDependsOn(), new ArrayList<>()); assertEquals(requests.get("dbwriter").getDependsOn(), new ArrayList<>()); }