public interface HiwayDBI {

	public static final String KEY_FILE_TIME_STAGEIN = "file-time-stagein";
	public static final String KEY_FILE_TIME_STAGEOUT = "file-time-stageout";
	public static final String KEY_HIWAY_EVENT = "hiway-event";
	public static final String KEY_INVOC_HOST = "invoc-host";
	public static final String KEY_INVOC_TIME_SCHED = "invoc-time-sched";
	public static final String KEY_INVOC_TIME_STAGEIN = "invoc-time-stagein";
	public static final String KEY_INVOC_TIME_STAGEOUT = "invoc-time-stageout";
	public static final String KEY_WF_NAME = "wf-name";
	public static final String KEY_WF_TIME = "wf-time";
	public static final String KEY_WF_OUTPUT = "wf-output";

	public Set<String> getHostNames();

	public Set<Long> getTaskIdsForWorkflow(String workflowName);

	public String getTaskName(long taskId);

	public Collection<InvocStat> getLogEntriesForTasks(Set<Long> taskIds);

	public Collection<InvocStat> getLogEntriesForTaskOnHostSince(long taskId, String hostName, long timestamp);

	public void logToDB(JsonReportEntry entry);

} public class Data implements Comparable<Data> {
	private static FileSystem hdfs;
	private static Path hdfsApplicationDirectory;

	private static Path hdfsBaseDirectory;

	private static FileSystem localFs = new LocalFileSystem();

	public static void setHdfs(FileSystem hdfs) {
		Data.hdfs = hdfs;
	}

	public static void setHdfsApplicationDirectory(Path hdfsApplicationDirectory) {
		Data.hdfsApplicationDirectory = hdfsApplicationDirectory;
	}

	public static void setHdfsBaseDirectory(Path hdfsBaseDirectory) {
		Data.hdfsBaseDirectory = hdfsBaseDirectory;
	}

	private String containerId;

	private String fileName;

	private Path localDirectory;

	// is the file output of the workflow
	private boolean output;

	public Data(Path localPath) {
		this(localPath, null);
	}

	public Data(Path localPath, String containerId) {
		this.output = false;

		this.localDirectory = localPath.getParent();
		this.fileName = localPath.getName();
		this.containerId = containerId;
	}

	public Data(String localPathString) {
		this(new Path(localPathString), null);
	}

	public Data(String localPathString, String containerId) {
		this(new Path(localPathString), containerId);
	}

	public void addToLocalResourceMap(Map<String, LocalResource> localResources) throws IOException {
		Path dest = getHdfsPath();

		LocalResource rsrc = Records.newRecord(LocalResource.class);
		rsrc.setType(LocalResourceType.FILE);
		rsrc.setVisibility(LocalResourceVisibility.APPLICATION);
		rsrc.setResource(ConverterUtils.getYarnUrlFromPath(dest));

		FileStatus status = hdfs.getFileStatus(dest);
		rsrc.setTimestamp(status.getModificationTime());
		rsrc.setSize(status.getLen());

		localResources.put(getLocalPath().toString(), rsrc);
	}

	@Override
	public int compareTo(Data other) {
		return this.getLocalPath().compareTo(other.getLocalPath());
	}

	public long countAvailableLocalData(Container container) throws IOException {
		BlockLocation[] blockLocations = null;

		Path hdfsLocation = getHdfsPath();
		while (blockLocations == null) {
			FileStatus fileStatus = hdfs.getFileStatus(hdfsLocation);
			blockLocations = hdfs.getFileBlockLocations(hdfsLocation, 0, fileStatus.getLen());
		}

		long sum = 0;
		for (BlockLocation blockLocation : blockLocations) {
			for (String host : blockLocation.getHosts()) {
				if (container.getNodeId().getHost().equals(host)) {
					sum += blockLocation.getLength();
					break;
				}
			}
		}
		return sum;
	}

	public long countAvailableTotalData() throws IOException {
		Path hdfsLocation = getHdfsPath();
		FileStatus fileStatus = hdfs.getFileStatus(hdfsLocation);
		return fileStatus.getLen();
	}

	@Override
	public boolean equals(Object obj) {
		return obj instanceof Data ? this.getLocalPath().equals(((Data) obj).getLocalPath()) : false;
	}

	public String getContainerId() {
		return containerId;
	}

	public Path getHdfsPath() {
		// if a container id has been created, this file is intermediate and will be found in the container's folder
		if (containerId != null) {
			return completeHdfsPath(new Path(hdfsApplicationDirectory, containerId));
		}

		// else, we should check if the file is an input file; if so, it can be found directly in the hdfs base directory
		Path basePath = completeHdfsPath(hdfsBaseDirectory);
		try {
			if (hdfs.exists(basePath)) {
				return basePath;
			}
		} catch (IOException e) {
			e.printStackTrace(System.out);
		}

		// otherwise, it is fair to assume that the file will be found in the application's folder
		return completeHdfsPath(hdfsApplicationDirectory);
	}
	
	private Path completeHdfsPath(Path pathPrefix) {
		Path directory = pathPrefix;
		if (!localDirectory.isUriPathAbsolute())
			directory = new Path(directory, localDirectory);
		return new Path(directory, fileName);
	}

	public Path getLocalPath() {
		return new Path(localDirectory, fileName);
	}

	public String getName() {
		return fileName;
	}

	@Override
	public int hashCode() {
		return this.getLocalPath().hashCode();
	}

	public boolean isOutput() {
		return output;
	}

	private void mkHdfsDir(Path dir) throws IOException {
		if (dir == null || hdfs.isDirectory(dir))
			return;
		mkHdfsDir(dir.getParent());
		hdfs.mkdirs(dir);
		hdfs.setPermission(dir, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
	}

	public void setContainerId(String containerId) {
		this.containerId = containerId;
	}

	public void setOutput(boolean output) {
		this.output = output;
	}

	public void stageIn() throws IOException {
		Path hdfsPath = getHdfsPath();
		Path localPath = getLocalPath();
		if (localDirectory.depth() > 0) {
			localFs.mkdirs(localDirectory);
		}
		hdfs.copyToLocalFile(false, hdfsPath, localPath);
	}

	public void stageOut() throws IOException {
		Path localPath = getLocalPath();
		Path hdfsDirectory = getHdfsPath().getParent();
		Path hdfsPath = getHdfsPath();
		if (hdfsDirectory.depth() > 0) {
			mkHdfsDir(hdfsDirectory);
		}
		hdfs.copyFromLocalFile(false, true, localPath, hdfsPath);
	}

	@Override
	public String toString() {
		return getLocalPath().toString();
	}

} public class HiWayConfiguration extends YarnConfiguration {

	public static enum HIWAY_DB_TYPE_OPTS {
		local, NoSQL, SQL
	}

	public static enum HIWAY_SCHEDULER_OPTS {
		era, dataAware, greedy, heft, roundRobin, memoryAware
	}

	public static enum HIWAY_WORKFLOW_LANGUAGE_OPTS {
		cuneiformE, cuneiformJ, dax, galaxy, log
	}

	public static final String HIWAY_AM_APPLICATION_TYPE = "hiway.am.application.type";

	public static final String HIWAY_AM_APPLICATION_TYPE_DEFAULT = "Hi-WAY";
	public static final String HIWAY_AM_DIRECTORY_BASE = "hiway.am.directory.base";

	public static final String HIWAY_AM_DIRECTORY_BASE_DEFAULT = "/";
	public static final String HIWAY_AM_DIRECTORY_CACHE = "hiway.am.directory.cache";
	public static final String HIWAY_AM_DIRECTORY_CACHE_DEFAULT = "hiway";
	public static final String HIWAY_AM_MEMORY = "hiway.am.memory";
	public static final int HIWAY_AM_MEMORY_DEFAULT = 1024;
	public static final String HIWAY_AM_PRIORITY = "hiway.am.priority";
	public static final int HIWAY_AM_PRIORITY_DEFAULT = 0;
	public static final String HIWAY_AM_QUEUE = "hiway.am.queue";
	public static final String HIWAY_AM_QUEUE_DEFAULT = "default";
	public static final String HIWAY_AM_TASK_RETRIES = "hiway.am.task.retries";
	public static final int HIWAY_AM_TASK_RETRIES_DEFAULT = 1;
	public static final String HIWAY_AM_TIMEOUT = "hiway.am.timeout";
	public static final int HIWAY_AM_TIMEOUT_DEFAULT = 604800;
	public static final String HIWAY_AM_VCORES = "hiway.am.vcores";
	public static final int HIWAY_AM_VCORES_DEFAULT = 1;
	public static final String HIWAY_DB_NOSQL_BUCKET = "hiway.db.nosql.bucket";
	public static final String HIWAY_DB_NOSQL_PASSWORD = "hiway.db.nosql.password";
	public static final String HIWAY_DB_NOSQL_URLS = "hiway.db.nosql.urls";

	public static final String HIWAY_DB_SQL_PASSWORD = "hiway.db.sql.password";
	public static final String HIWAY_DB_SQL_URL = "hiway.db.sql.url";
	public static final String HIWAY_DB_SQL_USER = "hiway.db.sql.user";
	public static final String HIWAY_DB_TYPE = "hiway.db.type";
	public static final HIWAY_DB_TYPE_OPTS HIWAY_DB_TYPE_DEFAULT = HIWAY_DB_TYPE_OPTS.local;
	public static final String HIWAY_GALAXY_PATH = "hiway.galaxy.path";
	public static final String HIWAY_SCHEDULER = "hiway.scheduler";
	public static final HIWAY_SCHEDULER_OPTS HIWAY_SCHEDULER_DEFAULT = HIWAY_SCHEDULER_OPTS.era;
	
	public static final String HIWAY_SCHEDULER_ERA_ALPHA = "hiway.scheduler.era.alpha";
	public static final double HIWAY_SCHEDULER_ERA_ALPHA_DEFAULT = 0.2;
	public static final String HIWAY_SCHEDULER_ERA_RHO = "hiway.scheduler.era.rho";
	public static final int HIWAY_SCHEDULER_ERA_RHO_DEFAULT = 1;
	
	public static final String HIWAY_SITE_XML = "hiway-site.xml";

	public static final String HIWAY_WORKER_MEMORY = "hiway.worker.memory";

	public static final int HIWAY_WORKER_MEMORY_DEFAULT = 1024;

	public static final String HIWAY_WORKER_PRIORITY = "hiway.worker.priority";
	public static final int HIWAY_WORKER_PRIORITY_DEFAULT = 0;
	public static final String HIWAY_WORKER_SHELL_ENV = "hiway.worker.shell.env";
	public static final String HIWAY_WORKER_SHELL_ENV_DEFAULT = "";
	public static final String HIWAY_WORKER_VCORES = "hiway.worker.vcores";
	public static final int HIWAY_WORKER_VCORES_DEFAULT = 1;
	public static final String HIWAY_WORKFLOW_LANGUAGE_CUNEIFORME_AM_CLASS = "de.huberlin.wbi.hiway.am.cuneiforme.CuneiformEApplicationMaster";
	public static final String HIWAY_WORKFLOW_LANGUAGE_CUNEIFORME_WORKER_CLASS = "de.huberlin.wbi.hiway.am.cuneiforme.CuneiformEWorker";
	public static final String HIWAY_WORKFLOW_LANGUAGE_CUNEIFORME_SERVER_IP = "hiway.cuneiform.server.ip";
	public static final String HIWAY_WORKFLOW_LANGUAGE_CUNEIFORMJ_AM_CLASS = "de.huberlin.wbi.hiway.am.cuneiformj.CuneiformJApplicationMaster";
	public static final String HIWAY_WORKFLOW_LANGUAGE_CUNEIFORMJ_WORKER_CLASS = "de.huberlin.wbi.hiway.common.Worker";

	public static final String HIWAY_WORKFLOW_LANGUAGE_DAX_AM_CLASS = "de.huberlin.wbi.hiway.am.dax.DaxApplicationMaster";
	public static final String HIWAY_WORKFLOW_LANGUAGE_DAX_WORKER_CLASS = "de.huberlin.wbi.hiway.common.Worker";
	public static final Map<String, HIWAY_WORKFLOW_LANGUAGE_OPTS> HIWAY_WORKFLOW_LANGUAGE_EXTS;
	public static final String HIWAY_WORKFLOW_LANGUAGE_GALAXY_AM_CLASS = "de.huberlin.wbi.hiway.am.galaxy.GalaxyApplicationMaster";

	public static final String HIWAY_WORKFLOW_LANGUAGE_GALAXY_WORKER_CLASS = "de.huberlin.wbi.hiway.common.Worker";
	public static final String HIWAY_WORKFLOW_LANGUAGE_LOG_AM_CLASS = "de.huberlin.wbi.hiway.am.log.LogApplicationMaster";
	public static final String HIWAY_WORKFLOW_LANGUAGE_LOG_WORKER_CLASS = "de.huberlin.wbi.hiway.common.Worker";

	public static boolean debug = false;

	public static boolean verbose = false;
	static {
		HIWAY_WORKFLOW_LANGUAGE_EXTS = new HashMap<>();
		HIWAY_WORKFLOW_LANGUAGE_EXTS.put("cf", HIWAY_WORKFLOW_LANGUAGE_OPTS.cuneiformJ);
		HIWAY_WORKFLOW_LANGUAGE_EXTS.put("xml", HIWAY_WORKFLOW_LANGUAGE_OPTS.dax);
		HIWAY_WORKFLOW_LANGUAGE_EXTS.put("ga", HIWAY_WORKFLOW_LANGUAGE_OPTS.galaxy);
		HIWAY_WORKFLOW_LANGUAGE_EXTS.put("log", HIWAY_WORKFLOW_LANGUAGE_OPTS.log);
	}

	static {
		addDefaultResource(HIWAY_SITE_XML);
	}

	public HiWayConfiguration() {
		super();
	}

	public HiWayConfiguration(Configuration conf) {
		super(conf);
	}

} public class TaskInstance implements Comparable<TaskInstance> {

	public static class Comparators {

		public static Comparator<TaskInstance> DEPTH = new Comparator<TaskInstance>() {
			@Override
			public int compare(TaskInstance task1, TaskInstance task2) {
				try {
					return Integer.compare(task1.getDepth(), task2.getDepth());
				} catch (WorkflowStructureUnknownException e) {
					e.printStackTrace(System.out);
					System.exit(1);
					throw new RuntimeException(e);
				}
			}
		};

		public static Comparator<TaskInstance> UPWARDSRANK = new Comparator<TaskInstance>() {
			@Override
			public int compare(TaskInstance task1, TaskInstance task2) {
				try {
					return -Double.compare(task1.getUpwardRank(), task2.getUpwardRank());
				} catch (WorkflowStructureUnknownException e) {
					e.printStackTrace(System.out);
					System.exit(1);
					throw new RuntimeException(e);
				}
			}
		};

	}

	protected static int runningId = 1;

	protected Set<TaskInstance> childTasks;
	// the command to be executed
	protected String command;
	// whether this task is completed yet
	protected boolean completed;
	protected int depth = 0;
	// this task instance's id
	protected final long id;
	// input and output data
	protected Set<Data> inputData;
	private String invocScript = "";
	// the programming language of this task (default: bash)
	protected String languageLabel;
	protected Set<Data> outputData;
	// parent and child tasks (denotes the workflow structure)
	protected Set<TaskInstance> parentTasks;
	protected Set<JsonReportEntry> report;
	protected long taskId;
	// the name and (internal) id of the task's executable (e.g. tar)
	protected String taskName;
	// the number of times this task has been attempted
	protected int tries = 0;
	// the upward rank of tasks in the workflow
	protected double upwardRank = 0d;
	// the id of the workflow this task instance belongs to
	protected UUID workflowId;

	public TaskInstance(long id, UUID workflowId, String taskName, long taskId, String languageLabel) {
		this.id = id;
		this.workflowId = workflowId;
		this.taskName = taskName;
		this.taskId = taskId;
		this.languageLabel = languageLabel;

		this.completed = false;
		this.inputData = new HashSet<>();
		this.outputData = new HashSet<>();
		this.report = new HashSet<>();
		this.parentTasks = new HashSet<>();
		this.childTasks = new HashSet<>();
	}

	public TaskInstance(UUID workflowId, String taskName, long taskId) {
		this(workflowId, taskName, taskId, ForeignLambdaExpr.LANGID_BASH);
	}

	public TaskInstance(UUID workflowId, String taskName, long taskId, String languageLabel) {
		this(runningId++, workflowId, taskName, taskId, languageLabel);
	}

	public void addChildTask(TaskInstance childTask) {
		childTasks.add(childTask);
	}

	public void addInputData(Data data) {
		inputData.add(data);
	}

	public void addOutputData(Data data) {
		outputData.add(data);
	}

	public void addParentTask(TaskInstance parentTask) {
		parentTasks.add(parentTask);
		this.setDepth(parentTask.getDepth() + 1);
	}

	public Map<String, LocalResource> buildScriptsAndSetResources(Container container) {
		Map<String, LocalResource> localResources = new HashMap<>();
		try {
			String containerId = container.getId().toString();
			File script = new File(String.valueOf(getId()));
			try (BufferedWriter scriptWriter = new BufferedWriter(new FileWriter(script))) {
				scriptWriter.write(getCommand());
			} catch (IOException e) {
				e.printStackTrace(System.out);
				System.exit(-1);
			}
			Data scriptData = new Data(script.getPath(), containerId);
			try {
				scriptData.stageOut();
			} catch (IOException e) {
				e.printStackTrace(System.out);
				System.exit(-1);
			}
			scriptData.addToLocalResourceMap(localResources);
		} catch (IOException e) {
			e.printStackTrace(System.out);
			System.exit(-1);
		}
		return localResources;
	}

	@Override
	public int compareTo(TaskInstance other) {
		return Long.compare(this.getId(), other.getId());
	}

	public long countAvailableLocalData(Container container) throws IOException {
		long sum = 0;
		for (Data input : getInputData()) {
			sum += input.countAvailableLocalData(container);
		}
		return sum;
	}

	public long countAvailableTotalData() throws IOException {
		long sum = 0;
		for (Data input : getInputData()) {
			sum += input.countAvailableTotalData();
		}
		return sum;
	}

	public Set<TaskInstance> getChildTasks() throws WorkflowStructureUnknownException {
		return childTasks;
	}

	public String getCommand() {
		return command;
	}

	public int getDepth() throws WorkflowStructureUnknownException {
		return depth;
	}

	public long getId() {
		return id;
	}

	public Set<Data> getInputData() {
		return inputData;
	}

	public String getInvocScript() {
		return invocScript;
	}

	public String getLanguageLabel() {
		return languageLabel;
	}

	public Set<Data> getOutputData() {
		return outputData;
	}

	public Set<TaskInstance> getParentTasks() throws WorkflowStructureUnknownException {
		return parentTasks;
	}

	public Set<JsonReportEntry> getReport() {
		return report;
	}

	public long getTaskId() {
		return taskId;
	}

	public String getTaskName() {
		return taskName;
	}

	public int getTries() {
		return tries;
	}

	public double getUpwardRank() throws WorkflowStructureUnknownException {
		return upwardRank;
	}

	public UUID getWorkflowId() {
		return workflowId;
	}

	public void incTries() {
		tries++;
	}

	public boolean isCompleted() {
		return completed;
	}

	public boolean readyToExecute() {
		for (TaskInstance parentTask : parentTasks) {
			if (!parentTask.isCompleted())
				return false;
		}
		return true;
	}

	public boolean retry(int maxRetries) {
		return tries <= maxRetries;
	}

	public void setCommand(String command) {
		this.command = command;
	}

	public void setCompleted() {
		completed = true;
	}

	public void setDepth(int depth) throws WorkflowStructureUnknownException {
		if (this.depth < depth) {
			this.depth = depth;
			for (TaskInstance child : this.getChildTasks()) {
				child.setDepth(depth + 1);
			}
		}
	}

	public void setInvocScript(String invocScript) {
		this.invocScript = invocScript;
	}

	public void setUpwardRank(double upwardRank) throws WorkflowStructureUnknownException {
		this.upwardRank = upwardRank;
	}

	@Override
	public String toString() {
		return id + " [" + taskName + "]";
	}

} public class WorkflowStructureUnknownException extends RuntimeException {

	private static final long serialVersionUID = 157794148736647758L;

	public WorkflowStructureUnknownException(HIWAY_WORKFLOW_LANGUAGE_OPTS language) {
		super(language.toString() + " workflows are not static. Static scheduling is not supported.");
	}

} public abstract class WorkflowScheduler {

	protected HiWayConfiguration conf;
	protected HiwayDBI dbInterface;
	protected FileSystem hdfs;
	protected int maxRetries = 0;
	protected Map<String, Long> maxTimestampPerHost;
	protected int numberOfFinishedTasks = 0;
	protected int numberOfPreviousRunTasks = 0;
	protected int numberOfRemainingTasks = 0;
	protected int numberOfRunningTasks = 0;
	protected boolean relaxLocality = true;
	protected Map<String, Map<Long, WienerProcessModel>> runtimePerBotPerVm;
	protected Set<Long> taskIds;
	// a queue of nodes on which containers are to be requested
	protected Queue<ContainerRequest> unissuedContainerRequests;
	protected String workflowName;
	protected int containerMemory;
	protected Map<String, Integer> customMemoryMap;
	protected int containerCores;
	protected int requestPriority;

	public WorkflowScheduler(String workflowName) {
		this.workflowName = workflowName;

		unissuedContainerRequests = new LinkedList<>();

		taskIds = new HashSet<>();
		runtimePerBotPerVm = new HashMap<>();
		maxTimestampPerHost = new HashMap<>();
	}

	public void init(HiWayConfiguration conf_, FileSystem hdfs_, int containerMemory_, Map<String, Integer> customMemoryMap_, int containerCores_,
	    int requestPriority_) {
		this.conf = conf_;
		this.hdfs = hdfs_;
		this.containerMemory = containerMemory_;
		this.customMemoryMap = customMemoryMap_;
		this.containerCores = containerCores_;
		this.requestPriority = requestPriority_;
	}

	protected ContainerRequest setupContainerAskForRM(String[] nodes, int memory) {
		// set the priority for the request
		Priority pri = Priority.newInstance(requestPriority);
		// pri.setPriority(requestPriority);

		// set up resource type requirements
		Resource capability = Resource.newInstance(memory, containerCores);
		// capability.setMemory(containerMemory);
		// capability.setVirtualCores(containerCores);

		return new ContainerRequest(capability, nodes, null, pri, relaxLocality());
	}

	public void addEntryToDB(JsonReportEntry entry) {
		// WorkflowDriver.writeToStdout("HiwayDB: Adding entry " + entry + " to database.");
		dbInterface.logToDB(entry);
		// WorkflowDriver.writeToStdout("HiwayDB: Added entry to database.");
	}

	protected abstract void addTask(TaskInstance task);

	public void addTasks(Collection<TaskInstance> tasks) {
		for (TaskInstance task : tasks) {
			addTask(task);
		}
	}

	public abstract void addTaskToQueue(TaskInstance task);

	public HiwayDBI getDbInterface() {
		return dbInterface;
	}

	public ContainerRequest getNextNodeRequest() {
		return unissuedContainerRequests.remove();
	}

	public abstract TaskInstance getTask(Container container);

	protected Set<String> getNodeIds() {
		return new HashSet<>(runtimePerBotPerVm.keySet());
	}

	public int getNumberOfFinishedTasks() {
		return numberOfFinishedTasks - numberOfPreviousRunTasks;
	}

	public abstract int getNumberOfReadyTasks();

	public int getNumberOfRunningTasks() {
		return numberOfRunningTasks;
	}

	public int getNumberOfTotalTasks() {
		int fin = getNumberOfFinishedTasks();
		int run = getNumberOfRunningTasks();
		int rem = numberOfRemainingTasks;
		
		return fin + run + rem;
	}

	protected Set<Long> getTaskIds() {
		return new HashSet<>(taskIds);
	}

	public boolean hasNextNodeRequest() {
		return !unissuedContainerRequests.isEmpty();
	}

	public void initializeProvenanceManager() {
		maxRetries = conf.getInt(HiWayConfiguration.HIWAY_AM_TASK_RETRIES, HiWayConfiguration.HIWAY_AM_TASK_RETRIES_DEFAULT);

		HiWayConfiguration.HIWAY_DB_TYPE_OPTS dbType = HiWayConfiguration.HIWAY_DB_TYPE_OPTS.valueOf(conf.get(HiWayConfiguration.HIWAY_DB_TYPE,
		    HiWayConfiguration.HIWAY_DB_TYPE_DEFAULT.toString()));
		switch (dbType) {
		case SQL:
			String sqlUser = conf.get(HiWayConfiguration.HIWAY_DB_SQL_USER);
			if (sqlUser == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_SQL_USER + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			String sqlPassword = conf.get(HiWayConfiguration.HIWAY_DB_SQL_PASSWORD);
			if (sqlPassword == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_SQL_PASSWORD + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			String sqlURL = conf.get(HiWayConfiguration.HIWAY_DB_SQL_URL);
			if (sqlURL == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_SQL_URL + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			dbInterface = new HiwayDB(sqlUser, sqlPassword, sqlURL);
			break;
		case NoSQL:
			sqlUser = conf.get(HiWayConfiguration.HIWAY_DB_SQL_USER);
			if (sqlUser == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_SQL_USER + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			sqlPassword = conf.get(HiWayConfiguration.HIWAY_DB_SQL_PASSWORD);
			if (sqlPassword == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_SQL_PASSWORD + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			sqlURL = conf.get(HiWayConfiguration.HIWAY_DB_SQL_URL);
			if (sqlURL == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_SQL_URL + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			String noSqlBucket = conf.get(HiWayConfiguration.HIWAY_DB_NOSQL_BUCKET);
			if (noSqlBucket == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_NOSQL_BUCKET + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			String noSqlPassword = conf.get(HiWayConfiguration.HIWAY_DB_NOSQL_PASSWORD);
			if (noSqlPassword == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_NOSQL_PASSWORD + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			String noSqlURIs = conf.get(HiWayConfiguration.HIWAY_DB_NOSQL_URLS);
			if (noSqlURIs == null) {
				WorkflowDriver.writeToStdout(HiWayConfiguration.HIWAY_DB_NOSQL_URLS + " not set in  " + HiWayConfiguration.HIWAY_SITE_XML);
				throw new RuntimeException();
			}
			List<URI> noSqlURIList = new ArrayList<>();
			for (String uri : noSqlURIs.split(",")) {
				noSqlURIList.add(URI.create(uri));
			}
			dbInterface = new HiwayDBNoSQL(noSqlBucket, noSqlPassword, noSqlURIList, sqlUser, sqlPassword, sqlURL);

			break;
		default:
			dbInterface = new ProvenanceManager();
			parseLogs();
		}
	}

	protected void newHost(String nodeId) {
		Map<Long, WienerProcessModel> runtimeEstimates = new HashMap<>();
		for (long taskId : getTaskIds()) {
			runtimeEstimates.put(taskId, new WienerProcessModel(taskId, nodeId, false));
		}
		runtimePerBotPerVm.put(nodeId, runtimeEstimates);
		maxTimestampPerHost.put(nodeId, 0l);
	}

	protected void newTask(long taskId) {
		taskIds.add(taskId);
		for (Entry<String, Map<Long, WienerProcessModel>> runtimeEstimates : runtimePerBotPerVm.entrySet()) {
			runtimeEstimates.getValue().put(taskId, new WienerProcessModel(taskId, runtimeEstimates.getKey(), false));
		}
	}

	public boolean nothingToSchedule() {
		return getNumberOfReadyTasks() == 0;
	}

	protected void parseLogs() {
		String hdfsBaseDirectoryName = conf.get(HiWayConfiguration.HIWAY_AM_DIRECTORY_BASE, HiWayConfiguration.HIWAY_AM_DIRECTORY_BASE_DEFAULT);
		String hdfsSandboxDirectoryName = conf.get(HiWayConfiguration.HIWAY_AM_DIRECTORY_CACHE, HiWayConfiguration.HIWAY_AM_DIRECTORY_CACHE_DEFAULT);
		Path hdfsBaseDirectory = new Path(new Path(hdfs.getUri()), hdfsBaseDirectoryName);
		Path hdfsSandboxDirectory = new Path(hdfsBaseDirectory, hdfsSandboxDirectoryName);
		try {
			for (FileStatus appDirStatus : hdfs.listStatus(hdfsSandboxDirectory)) {
				if (appDirStatus.isDirectory()) {
					Path appDir = appDirStatus.getPath();
					for (FileStatus srcStatus : hdfs.listStatus(appDir)) {
						Path src = srcStatus.getPath();
						String srcName = src.getName();
						if (srcName.endsWith(".log")) {
							Path dest = new Path(appDir.getName());
							WorkflowDriver.writeToStdout("Parsing log " + dest.toString());
							hdfs.copyToLocalFile(false, src, dest);

							try (BufferedReader reader = new BufferedReader(new FileReader(new File(dest.toString())))) {
								String line;
								while ((line = reader.readLine()) != null) {
									if (line.length() == 0)
										continue;
									JsonReportEntry entry = new JsonReportEntry(line);
									addEntryToDB(entry);
								}
							}
						}
					}
				}
			}
		} catch (IOException | JSONException e) {
			e.printStackTrace(System.out);
			System.exit(-1);
		}
	}

	public boolean relaxLocality() {
		return relaxLocality;
	}

	public Collection<ContainerId> taskCompleted(TaskInstance task, ContainerStatus containerStatus, long runtimeInMs) {

		numberOfRunningTasks--;
		numberOfFinishedTasks++;

		WorkflowDriver.writeToStdout("Task " + task + " on container " + containerStatus.getContainerId() + " completed successfully after " + runtimeInMs + " ms");

		return new ArrayList<>();
	}

	public Collection<ContainerId> taskFailed(TaskInstance task, ContainerStatus containerStatus) {
		numberOfRunningTasks--;

		WorkflowDriver.writeToStdout("Task " + task + " on container " + containerStatus.getContainerId() + " failed");
		if (task.retry(maxRetries)) {
			WorkflowDriver.writeToStdout("Retrying task " + task + ".");
			addTask(task);
		} else {
			WorkflowDriver.writeToStdout("Task " + task + " has exceeded maximum number of allowed retries. Aborting workflow.");
			System.exit(-1);
		}

		return new ArrayList<>();
	}

	protected void updateRuntimeEstimate(InvocStat stat) {
		WienerProcessModel re = runtimePerBotPerVm.get(stat.getHostName()).get(stat.getTaskId());
		re.addRuntime(stat.getTimestamp(), stat.getRealTime());
	}

	public void updateRuntimeEstimates(String runId) {
		if (HiWayConfiguration.verbose)
			WorkflowDriver.writeToStdout("Updating Runtime Estimates.");

		// WorkflowDriver.writeToStdout("HiwayDB: Querying Host Names from database.");
		Collection<String> newHostIds = dbInterface.getHostNames();
		// WorkflowDriver.writeToStdout("HiwayDB: Retrieved Host Names " + newHostIds.toString() + " from database.");
		newHostIds.removeAll(getNodeIds());
		for (String newHostId : newHostIds) {
			newHost(newHostId);
		}
		// WorkflowDriver.writeToStdout("HiwayDB: Querying Task Ids for workflow " + workflowName + " from database.");
		Collection<Long> newTaskIds = dbInterface.getTaskIdsForWorkflow(workflowName);
		// WorkflowDriver.writeToStdout("HiwayDB: Retrieved Task Ids " + newTaskIds.toString() + " from database.");

		newTaskIds.removeAll(getTaskIds());
		for (long newTaskId : newTaskIds) {
			newTask(newTaskId);
		}

		for (String hostName : getNodeIds()) {
			long oldMaxTimestamp = maxTimestampPerHost.get(hostName);
			long newMaxTimestamp = oldMaxTimestamp;
			for (long taskId : getTaskIds()) {
				// WorkflowDriver.writeToStdout("HiwayDB: Querying InvocStats for task id " + taskId + " on host " + hostName + " since timestamp " + oldMaxTimestamp
				// + " from database.");
				Collection<InvocStat> invocStats = dbInterface.getLogEntriesForTaskOnHostSince(taskId, hostName, oldMaxTimestamp);
				// WorkflowDriver.writeToStdout("HiwayDB: Retrieved InvocStats " + invocStats.toString() + " from database.");
				for (InvocStat stat : invocStats) {
					newMaxTimestamp = Math.max(newMaxTimestamp, stat.getTimestamp());
					updateRuntimeEstimate(stat);
					if (!runId.equals(stat.getRunId())) {
						numberOfPreviousRunTasks++;
						numberOfFinishedTasks++;
					}
				}
			}
			maxTimestampPerHost.put(hostName, newMaxTimestamp);
		}
	}
} public class ERA extends WorkflowScheduler {

	public static double alpha = 0.2;
	public static int rho = 1;

	protected Map<Long, Queue<TaskInstance>> readyTasksPerBot;
	protected Map<Long, Set<TaskInstance>> runningTasksPerBot;
	// protected Map<String, Map<Long, WienerProcessModel>> runtimePerBotPerVm;

	protected List<TaskInstance> replicas;
	protected Set<TaskInstance> todo;

	protected Map<TaskInstance, List<Container>> taskToContainers;

	protected boolean init;

	public ERA(String workflowName, HiWayConfiguration conf) {
		super(workflowName);
		readyTasksPerBot = new HashMap<>();
		runningTasksPerBot = new HashMap<>();
		replicas = new LinkedList<>();
		todo = new HashSet<>();
		taskToContainers = new HashMap<>();
		init = true;

		alpha = conf.getDouble(HiWayConfiguration.HIWAY_SCHEDULER_ERA_ALPHA, HiWayConfiguration.HIWAY_SCHEDULER_ERA_ALPHA_DEFAULT);
		rho = conf.getInt(HiWayConfiguration.HIWAY_SCHEDULER_ERA_RHO, HiWayConfiguration.HIWAY_SCHEDULER_ERA_RHO_DEFAULT);
	}

	@Override
	protected void addTask(TaskInstance task) {
		todo.add(task);
		long taskId = task.getTaskId();
		String taskName = task.getTaskName();

		WorkflowDriver.writeToStdout("Adding task of id " + taskId + " and name " + taskName + ".");
		numberOfRemainingTasks++;

		if (!getTaskIds().contains(taskId)) {
			newTask(taskId);
			WorkflowDriver.writeToStdout("Detected new BoT " + taskId + " (" + taskName + ").");
		}

		if (task.readyToExecute())
			addTaskToQueue(task);
	}

	@Override
	public void addTaskToQueue(TaskInstance task) {
		if (init) {
			for (int i = 0; i < rho; i++) {
				unissuedContainerRequests.add(setupContainerAskForRM(new String[0], containerMemory));
			}
			init = false;
		}
		unissuedContainerRequests.add(setupContainerAskForRM(new String[0], containerMemory));
		if (!readyTasksPerBot.containsKey(task.getTaskId())) {
			Queue<TaskInstance> q = new LinkedList<>();
			readyTasksPerBot.put(task.getTaskId(), q);
		}
		readyTasksPerBot.get(task.getTaskId()).add(task);
		WorkflowDriver.writeToStdout("Added task " + task + " to queue " + task.getTaskName());
	}

	@Override
	public TaskInstance getTask(Container container) {
		numberOfRemainingTasks--;
		numberOfRunningTasks++;

		String nodeId = container.getNodeId().getHost();
		if (!runtimePerBotPerVm.containsKey(nodeId)) {
			newHost(nodeId);
		}

		boolean replicate = false;

		Map<Long, Queue<TaskInstance>> b_ready = readyTasksPerBot;
		Map<Long, Set<TaskInstance>> b_run = runningTasksPerBot;

		Map<Long, Queue<TaskInstance>> b_select = b_ready;
		if (b_ready.isEmpty()) {
			b_select = new HashMap<>();
			for (Entry<Long, Set<TaskInstance>> e : b_run.entrySet()) {
				Queue<TaskInstance> tasks = new LinkedList<>(e.getValue());
				b_select.put(e.getKey(), tasks);
			}
			replicate = true;
		}

		Set<String> m = runtimePerBotPerVm.keySet();
		Queue<TaskInstance> b_min = null;
		double s_min = Double.MAX_VALUE;
		for (Entry<Long, Queue<TaskInstance>> b_i : b_select.entrySet()) {
			double e_j = runtimePerBotPerVm.get(nodeId).get(b_i.getKey()).getEstimate(System.currentTimeMillis(), replicate ? 0.5 : alpha);
			if (e_j == 0) {
				if (!replicate) {
					b_min = b_i.getValue();
					break;
				}
				e_j = Double.MAX_VALUE;
			}

			double e_min = Double.MAX_VALUE;
			double e_sum = e_j;
			int e_num = 1;
			for (String k : m) {
				double e_k = runtimePerBotPerVm.get(k).get(b_i.getKey()).getEstimate(System.currentTimeMillis(), replicate ? 0.5 : alpha);
				if (k.equals(nodeId) || e_k == 0)
					continue;
				if (e_k < e_min) {
					e_min = e_k;
				}
				e_sum += e_k;
				e_num++;
			}

			double s = (e_j - e_min) / (e_sum / e_num);
			if (s < s_min) {
				s_min = s;
				b_min = b_i.getValue();
			}
		}

		if (b_min != null && !b_min.isEmpty()) {
			TaskInstance task = b_min.remove();
			double estimate = runtimePerBotPerVm.get(nodeId).get(task.getTaskId()).getEstimate(System.currentTimeMillis(), replicate ? 0.5 : alpha);

			if (replicate) {
				replicas.add(task);
			} else {
				if (!runningTasksPerBot.containsKey(task.getTaskId())) {
					Set<TaskInstance> s = new HashSet<>();
					runningTasksPerBot.put(task.getTaskId(), s);
				}
				runningTasksPerBot.get(task.getTaskId()).add(task);

				if (readyTasksPerBot.get(task.getTaskId()).isEmpty()) {
					readyTasksPerBot.remove(task.getTaskId());
				}

			}

			if (!taskToContainers.containsKey(task)) {
				taskToContainers.put(task, new ArrayList<Container>());
			}
			taskToContainers.get(task).add(container);

			if (replicate) {
				WorkflowDriver
				    .writeToStdout("Assigned speculative copy of task " + task + " to container " + container.getId() + "@" + container.getNodeId().getHost());
			} else {
				WorkflowDriver.writeToStdout("Assigned task " + task + " to container " + container.getId() + "@" + container.getNodeId().getHost());
			}

			WorkflowDriver.writeToStdout("(estimate: " + estimate + "; sufferage: " + s_min + ")");
			task.incTries();

			return task;
		}

		return null;

	}

	@Override
	public int getNumberOfReadyTasks() {
		int nReadyTasks = 0;
		for (Queue<TaskInstance> queue : readyTasksPerBot.values())
			nReadyTasks += queue.size();
		return nReadyTasks;
	}

	@Override
	public int getNumberOfRunningTasks() {
		return taskToContainers.size();
	}

	@Override
	public int getNumberOfTotalTasks() {
		return getNumberOfFinishedTasks() + todo.size();
	}

	@Override
	public boolean nothingToSchedule() {
		return (todo.isEmpty() || (readyTasksPerBot.isEmpty() && !(replicas.size() < rho)));
	}

	@Override
	public Collection<ContainerId> taskCompleted(TaskInstance task, ContainerStatus containerStatus, long runtimeInMs) {
		super.taskCompleted(task, containerStatus, runtimeInMs);

		todo.remove(task);
		taskFinished(task);

		updateRuntimeEstimates(task.getWorkflowId().toString());

		// kill speculative copies
		Collection<ContainerId> toBeReleasedContainers = new ArrayList<>();
		for (Container container : taskToContainers.get(task)) {
			if (!container.getId().equals(containerStatus.getContainerId())) {
				toBeReleasedContainers.add(container.getId());
				unissuedContainerRequests.add(setupContainerAskForRM(new String[0], containerMemory));
			}
		}
		taskToContainers.remove(task);

		return toBeReleasedContainers;
	}

	@Override
	public Collection<ContainerId> taskFailed(TaskInstance task, ContainerStatus containerStatus) {
		super.taskFailed(task, containerStatus);

		taskFinished(task);

		Collection<ContainerId> toBeReleasedContainers = new ArrayList<>();
		if (!task.retry(maxRetries)) {
			for (Container container : taskToContainers.get(task)) {
				if (!container.getId().equals(containerStatus.getContainerId())) {
					toBeReleasedContainers.add(container.getId());
				}
			}
			taskToContainers.remove(task);
		}

		return toBeReleasedContainers;
	}

	private void taskFinished(TaskInstance task) {
		while (replicas.contains(task))
			replicas.remove(task);
		if (runningTasksPerBot.containsKey(task.getTaskId())) {
			runningTasksPerBot.get(task.getTaskId()).remove(task);
			if (runningTasksPerBot.get(task.getTaskId()).isEmpty())
				runningTasksPerBot.remove(task.getTaskId());
		}
	}

} public class GreedyQueue extends WorkflowScheduler {

	private Queue<TaskInstance> queue;

	public GreedyQueue(String workflowName) {
		super(workflowName);
		queue = new LinkedList<>();
	}

	@Override
	protected void addTask(TaskInstance task) {
		numberOfRemainingTasks++;
		if (task.readyToExecute())
			addTaskToQueue(task);
	}

	@Override
	public void addTaskToQueue(TaskInstance task) {
		unissuedContainerRequests.add(setupContainerAskForRM(new String[0], containerMemory));
		queue.add(task);
		WorkflowDriver.writeToStdout("Added task " + task + " to queue");
	}

	@Override
	public TaskInstance getTask(Container container) {
		numberOfRemainingTasks--;
		numberOfRunningTasks++;
		TaskInstance task = queue.remove();

		WorkflowDriver.writeToStdout("Assigned task " + task + " to container " + container.getId() + "@" + container.getNodeId().getHost());
		task.incTries();
		return task;
	}

	@Override
	public int getNumberOfReadyTasks() {
		return queue.size();
	}

} public class HEFT extends StaticScheduler {

	private int containers;

	private Map<String, ArrayList<Map<Double, Double>>> freeTimeSlotLengthsPerNode;
	private Map<String, ArrayList<TreeSet<Double>>> freeTimeSlotStartsPerNode;
	private Map<TaskInstance, Double> readyTimePerTask;

	private Map<String, ArrayList<TreeMap<Double, TaskInstance>>> taskOnsetsPerNode;

	public HEFT(String workflowName, int containers) {
		super(workflowName);

		this.containers = containers;
		WorkflowDriver.writeToStdout("HEFT has detected " + containers + " containers per worker node.");

		readyTimePerTask = new HashMap<>();
		freeTimeSlotStartsPerNode = new HashMap<>();
		freeTimeSlotLengthsPerNode = new HashMap<>();
		taskOnsetsPerNode = new HashMap<>();
	}

	@Override
	protected void addTask(TaskInstance task) {
		numberOfRemainingTasks++;
		Collection<String> nodes = runtimePerBotPerVm.keySet();

		// the readytime of a task is the time from workflow onset after which this task will be ready (according to the schedule)
		// if the task is an input task, its ready time will still be 0
		// otherwise, its readytime will have been set already as all predecessor tasks have a higher upward rank will have been scheduled already (during which
		// this task's readytime was updated)
		double readyTime = readyTimePerTask.get(task);

		List<String> bestNodes = new ArrayList<>();
		List<Integer> bestIs = new ArrayList<>();
		List<Double> bestNodeFreeTimeSlotActualStarts = new ArrayList<>();
		double bestFinish = Double.MAX_VALUE;

		// List<String> bestNode = null;
		// List<Integer> bestI = -1;
		// List<Double> bestNodeFreeTimeSlotActualStart = Double.MAX_VALUE;
		// List<Double> bestFinish = Double.MAX_VALUE;

		// compute the finish time of executing this task on each node; chose the earliest finish time
		for (String node : nodes) {
			ArrayList<TreeSet<Double>> freeTimeSlotStartsPerContainer = freeTimeSlotStartsPerNode.get(node);
			ArrayList<Map<Double, Double>> freeTimeSlotLengthsPerContainer = freeTimeSlotLengthsPerNode.get(node);

			// note that the weight is 1 ms if no runtime estimate is available yet
			double computationCost = runtimePerBotPerVm.get(node).get(task.getTaskId()).getEstimate();

			for (int i = 0; i < containers; i++) {
				// note that at the beginning, all nodes have a free timeslot of length MAX_VALUE starting at time 0
				TreeSet<Double> freeTimeSlotStarts = freeTimeSlotStartsPerContainer.get(i);
				Map<Double, Double> freeTimeSlotLengths = freeTimeSlotLengthsPerContainer.get(i);

				// freeTimeSlotStartsAfterReadyTime is the (sorted) set of timeslots available once this task is ready
				SortedSet<Double> freeTimeSlotStartsAfterReadyTime = (freeTimeSlotStarts.floor(readyTime) != null) ? freeTimeSlotStarts
						.tailSet(freeTimeSlotStarts.floor(readyTime)) : freeTimeSlotStarts.tailSet(freeTimeSlotStarts.ceiling(readyTime));

				// iterate through timeslots
				for (double freeTimeSlotStart : freeTimeSlotStartsAfterReadyTime) {

					// task onset could be later than timeslot onset
					double freeTimeSlotActualStart = Math.max(readyTime, freeTimeSlotStart);

					// if finishtime of task on this node is worse than the current optimum, we don't have to look at this node any more
					if (freeTimeSlotActualStart + computationCost > bestFinish)
						break;

					// otherwise, we found a new best node, but only if the computation cost does not exceed the length of the timeslot
					double freeTimeSlotLength = freeTimeSlotLengths.get(freeTimeSlotStart);
					if (freeTimeSlotActualStart > freeTimeSlotStart)
						freeTimeSlotLength -= freeTimeSlotActualStart - freeTimeSlotStart;
					if (computationCost < freeTimeSlotLength) {
						if (freeTimeSlotActualStart + computationCost < bestFinish) {
							bestNodes = new ArrayList<>();
							bestIs = new ArrayList<>();
							bestNodeFreeTimeSlotActualStarts = new ArrayList<>();
						}
						bestNodes.add(node);
						bestIs.add(i);
						bestNodeFreeTimeSlotActualStarts.add(freeTimeSlotActualStart);
						bestFinish = freeTimeSlotActualStart + computationCost;
					}
				}
			}
		}

		// if there is more than one potential node, sample
		int i = (int) (bestNodes.size() * Math.random());
		
		String bestNode = bestNodes.get(i);
		int bestI = bestIs.get(i);
		double bestNodeFreeTimeSlotActualStart = bestNodeFreeTimeSlotActualStarts.get(i);

		// assign task to node
		schedule.put(task, bestNode);
		taskOnsetsPerNode.get(bestNode).get(bestI).put(bestNodeFreeTimeSlotActualStart, task);
		if (HiWayConfiguration.verbose)
			WorkflowDriver.writeToStdout("Task " + task + " scheduled on node " + bestNode);
		if (task.readyToExecute()) {
			addTaskToQueue(task);
		}

		// update readytime of all successor tasks
		try {
			for (TaskInstance child : task.getChildTasks()) {
				if (bestFinish > readyTimePerTask.get(child)) {
					readyTimePerTask.put(child, bestFinish);
				}
			}
		} catch (WorkflowStructureUnknownException e) {
			e.printStackTrace(System.out);
			System.exit(-1);
		}

		// once the task has been scheduled on a node, update that node's free timeslots
		double timeslotStart = freeTimeSlotStartsPerNode.get(bestNode).get(bestI).floor(bestNodeFreeTimeSlotActualStart);
		double timeslotLength = freeTimeSlotLengthsPerNode.get(bestNode).get(bestI).get(timeslotStart);
		double diff = bestNodeFreeTimeSlotActualStart - timeslotStart;

		// add time slot before
		if (bestNodeFreeTimeSlotActualStart > timeslotStart) {
			freeTimeSlotLengthsPerNode.get(bestNode).get(bestI).put(timeslotStart, diff);
		} else {
			freeTimeSlotStartsPerNode.get(bestNode).get(bestI).remove(timeslotStart);
			freeTimeSlotLengthsPerNode.get(bestNode).get(bestI).remove(timeslotStart);
		}

		// add time slot after
		double computationCost = bestFinish - bestNodeFreeTimeSlotActualStart;
		double actualTimeSlotLength = timeslotLength - diff;
		if (computationCost < actualTimeSlotLength) {
			freeTimeSlotStartsPerNode.get(bestNode).get(bestI).add(bestFinish);
			freeTimeSlotLengthsPerNode.get(bestNode).get(bestI).put(bestFinish, actualTimeSlotLength - computationCost);
		}
	}

	@Override
	public void addTasks(Collection<TaskInstance> tasks) {
		if (queues.size() == 0) {
			WorkflowDriver.writeToStdout("No provenance data available for static scheduling. Aborting.");
			System.exit(-1);
		}

		// obtain task list and sort by depth
		List<TaskInstance> taskList = new LinkedList<>(tasks);
		Collections.sort(taskList, new DepthComparator());

		Collection<String> nodes = runtimePerBotPerVm.keySet();

		// compute upward ranks of all tasks
		for (int i = taskList.size() - 1; i >= 0; i--) {
			TaskInstance task = taskList.get(i);
			readyTimePerTask.put(task, 0d);

			// maxSuccessorRank is the maximum upward rank of the current task's children
			double maxSuccessorRank = 0;
			try {
				for (TaskInstance child : task.getChildTasks()) {
					if (child.getUpwardRank() > maxSuccessorRank) {
						maxSuccessorRank = child.getUpwardRank();
					}
				}
			} catch (WorkflowStructureUnknownException e) {
				e.printStackTrace(System.out);
				System.exit(-1);
			}

			// averageComputationCost is the average cost for executing this task on any node
			double averageComputationCost = 0;
			for (String node : nodes) {
				averageComputationCost += runtimePerBotPerVm.get(node).get(task.getTaskId()).getEstimate();
			}
			averageComputationCost /= nodes.size();

			// set upward rank of this task
			// note that the upward rank of a task will always be greater than that of its successors
			try {
				task.setUpwardRank(averageComputationCost + maxSuccessorRank);
			} catch (WorkflowStructureUnknownException e) {
				e.printStackTrace(System.out);
				System.exit(-1);
			}
		}

		// Phase 1: Task Prioritizing (sort by decreasing order of rank)
		Collections.sort(taskList, new UpwardsRankComparator());

		// Phase 2: Processor Selection; schedule tasks w/ higher rank (critical tasks at the beginning of the wf) first
		for (TaskInstance task : taskList) {
			addTask(task);
		}

		printSchedule();
	}

	@Override
	protected void newHost(String nodeId) {
		super.newHost(nodeId);

		ArrayList<TreeSet<Double>> freeTimeSlotStartsPerContainer = new ArrayList<>();
		freeTimeSlotStartsPerNode.put(nodeId, freeTimeSlotStartsPerContainer);
		for (int i = 0; i < containers; i++) {
			TreeSet<Double> freeTimeSlotStarts = new TreeSet<>();
			freeTimeSlotStarts.add(0d);
			freeTimeSlotStartsPerContainer.add(freeTimeSlotStarts);
		}

		ArrayList<Map<Double, Double>> freeTimeSlotLengthsPerContainer = new ArrayList<>();
		freeTimeSlotLengthsPerNode.put(nodeId, freeTimeSlotLengthsPerContainer);
		for (int i = 0; i < containers; i++) {
			Map<Double, Double> freeTimeSlotLengths = new HashMap<>();
			freeTimeSlotLengths.put(0d, Double.MAX_VALUE);
			freeTimeSlotLengthsPerContainer.add(freeTimeSlotLengths);
		}

		ArrayList<TreeMap<Double, TaskInstance>> taskOnsetsPerContainer = new ArrayList<>();
		taskOnsetsPerNode.put(nodeId, taskOnsetsPerContainer);
		for (int i = 0; i < containers; i++) {
			TreeMap<Double, TaskInstance> taskOnsets = new TreeMap<>();
			taskOnsetsPerContainer.add(taskOnsets);
		}
	}

	public void printSchedule() {
		StringBuilder sb = new StringBuilder("HEFT Schedule:\n");
		for (String node : taskOnsetsPerNode.keySet()) {
			ArrayList<TreeMap<Double, TaskInstance>> taskOnsetsPerContainer = taskOnsetsPerNode.get(node);
			for (int i = 0; i < containers; i++) {
				TreeMap<Double, TaskInstance> taskOnsets = taskOnsetsPerContainer.get(i);
				sb.append(node);
				sb.append("-");
				sb.append(i + 1);
				sb.append(": ");
				for (Double onset : taskOnsets.keySet()) {
					sb.append(onset);
					sb.append("-");
					TaskInstance task = taskOnsets.get(onset);
					sb.append(task.getTaskName());
					sb.append("-");
					sb.append(onset + runtimePerBotPerVm.get(node).get(task.getTaskId()).getEstimate());
					sb.append(" ");
				}
				sb.append("\n");
			}
		}
		System.out.print(sb.toString());
	}

} public class MemoryAware extends WorkflowScheduler {

	private Map<Integer, Queue<TaskInstance>> queuePerMem;

	@SuppressWarnings("rawtypes")
	private AMRMClientAsync amRMClient;
	private int maxMem;
	private int maxCores;

	@SuppressWarnings("rawtypes")
	public MemoryAware(String workflowName, AMRMClientAsync amRMClient) {
		super(workflowName);
		queuePerMem = new HashMap<>();
		this.amRMClient = amRMClient;
	}

	@Override
	public void init(HiWayConfiguration conf_, FileSystem hdfs_, int containerMemory_, Map<String, Integer> customMemoryMap_, int containerCores_,
			int requestPriority_) {
		super.init(conf_, hdfs_, containerMemory_, customMemoryMap_, containerCores_, requestPriority_);
		maxMem = conf.getInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB, YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB);
		maxCores = conf.getInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES, YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES);
	}

	@Override
	protected void addTask(TaskInstance task) {
		numberOfRemainingTasks++;
		if (task.readyToExecute())
			addTaskToQueue(task);
	}

	@Override
	public void addTaskToQueue(TaskInstance task) {
		int memory = customMemoryMap.containsKey(task.getTaskName()) ? customMemoryMap.get(task.getTaskName()) : containerMemory;
		WorkflowDriver.writeToStdout("Adding task " + task + " to queue " + memory);
		unissuedContainerRequests.add(setupContainerAskForRM(new String[0], memory));

		if (!queuePerMem.containsKey(memory)) {
			queuePerMem.put(memory, new LinkedList<TaskInstance>());
		}
		queuePerMem.get(memory).add(task);
	}

	@Override
	public TaskInstance getTask(Container container) {
		numberOfRemainingTasks--;
		numberOfRunningTasks++;

		int memory = container.getResource().getMemory();
		Queue<TaskInstance> queue = queuePerMem.get(memory);
		TaskInstance task = queue.remove();

		WorkflowDriver.writeToStdout("Assigned task " + task + " to container " + container.getId() + "@" + container.getNodeId().getHost() + ":"
				+ container.getResource().getVirtualCores() + ":" + container.getResource().getMemory());

		task.incTries();
		return task;
	}

	@Override
	public int getNumberOfReadyTasks() {
		int nReadyTasks = 0;
		for (Queue<TaskInstance> queue : queuePerMem.values())
			nReadyTasks += queue.size();
		return nReadyTasks;
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean hasNextNodeRequest() {
		boolean hasNextNodeRequest = super.hasNextNodeRequest();
		// Hadoop has a weird bug where open container requests are sometimes "forgotten" (i.e., not fulfilled despite available resources).
		// Running into this bug can be prevented by occasionally re-issuing ContainerRequests.
		if (!hasNextNodeRequest) {
			List<? extends Collection<ContainerRequest>> requestCollections = amRMClient.getMatchingRequests(Priority.newInstance(requestPriority),
					ResourceRequest.ANY, Resource.newInstance(maxMem, maxCores));
			for (Collection<ContainerRequest> requestCollection : requestCollections) {
				ContainerRequest first = requestCollection.iterator().next();
				amRMClient.removeContainerRequest(first);
				amRMClient.addContainerRequest(first);
			}
		}
		return hasNextNodeRequest;
	}

} public class RoundRobin extends StaticScheduler {

	private Iterator<String> nodeIterator;

	public RoundRobin(String workflowName) {
		super(workflowName);
	}

	@Override
	protected void addTask(TaskInstance task) {
		numberOfRemainingTasks++;
		if (!nodeIterator.hasNext()) {
			nodeIterator = queues.keySet().iterator();
		}
		String node = nodeIterator.next();
		schedule.put(task, node);
		WorkflowDriver.writeToStdout("Task " + task + " scheduled on node " + node);
		if (task.readyToExecute()) {
			addTaskToQueue(task);
		}
	}

	@Override
	public void addTasks(Collection<TaskInstance> tasks) {
		if (nodeIterator == null) {
			nodeIterator = queues.keySet().iterator();
		}
		List<TaskInstance> taskList = new LinkedList<>(tasks);
		Collections.sort(taskList, new DepthComparator());
		super.addTasks(taskList);
	}

}
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.StringReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
import org.apache.hadoop.yarn.client.api.async.NMClientAsync;
import org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.json.JSONException;
import org.json.JSONObject;
import de.huberlin.hiwaydb.useDB.HiwayDBI;
import de.huberlin.wbi.cuneiform.core.invoc.Invocation;
import de.huberlin.wbi.cuneiform.core.semanticmodel.JsonReportEntry;
import de.huberlin.wbi.hiway.common.Data;
import de.huberlin.wbi.hiway.common.HiWayConfiguration;
import de.huberlin.wbi.hiway.common.TaskInstance;
import de.huberlin.wbi.hiway.common.WorkflowStructureUnknownException;
import de.huberlin.wbi.hiway.scheduler.WorkflowScheduler;
import de.huberlin.wbi.hiway.scheduler.era.ERA;
import de.huberlin.wbi.hiway.scheduler.gq.GreedyQueue;
import de.huberlin.wbi.hiway.scheduler.heft.HEFT;
import de.huberlin.wbi.hiway.scheduler.ma.MemoryAware;
import de.huberlin.wbi.hiway.scheduler.rr.RoundRobin;
				workflowPath = localPath;
				workflowFile = new Data(workflowPath);
				workflowFile.stageOut();
			} else {
				workflowFile = new Data(workflowPath);
				workflowFile.stageIn();
			}

			// Register self with ResourceManager. This will start heartbeating to the RM.
			appMasterHostname = NetUtils.getHostname();
			RegisterApplicationMasterResponse response = amRMClient.registerApplicationMaster(appMasterHostname, appMasterRpcPort, appMasterTrackingUrl);

			switch (schedulerName) {
			case roundRobin:
			case heft:
				int workerMemory = conf.getInt(YarnConfiguration.NM_PMEM_MB, YarnConfiguration.DEFAULT_NM_PMEM_MB);
				scheduler = schedulerName.equals(HiWayConfiguration.HIWAY_SCHEDULER_OPTS.roundRobin) ? new RoundRobin(getWorkflowName()) : new HEFT(getWorkflowName(),
				    workerMemory / containerMemory);
				break;
			case greedy:
				scheduler = new GreedyQueue(getWorkflowName());
				break;
			case memoryAware:
				scheduler = new MemoryAware(getWorkflowName(), amRMClient);
				break;
			default:
				scheduler = new ERA(getWorkflowName(), conf);
			}
			scheduler.init(conf, hdfs, containerMemory, customMemoryMap, containerCores, requestPriority);

			scheduler.initializeProvenanceManager();
			writeEntryToLog(new JsonReportEntry(getRunId(), null, null, null, null, null, HiwayDBI.KEY_WF_NAME, getWorkflowName()));
			federatedReport = new Data(appId + ".log");

			// parse workflow, obtain ready tasks
			Collection<TaskInstance> readyTasks = parseWorkflow();

			// scheduler updates runtime estimates for all tasks comprising the workflow
			scheduler.updateRuntimeEstimates(getRunId().toString());

			scheduler.addTasks(readyTasks);

			// Dump out information about cluster capability as seen by the resource manager
			maxMem = response.getMaximumResourceCapability().getMemory();
			maxCores = response.getMaximumResourceCapability().getVirtualCores();
			WorkflowDriver.writeToStdout("Max mem capabililty of resources in this cluster " + maxMem);

			// A resource ask cannot exceed the max.
			if (containerMemory > maxMem) {
				WorkflowDriver.writeToStdout("Container memory specified above max threshold of cluster." + " Using max value." + ", specified=" + containerMemory
				    + ", max=" + maxMem);
				containerMemory = maxMem;
			}
			if (containerCores > maxCores) {
				WorkflowDriver.writeToStdout("Container vcores specified above max threshold of cluster." + " Using max value." + ", specified=" + containerCores
				    + ", max=" + maxCores);
				containerCores = maxCores;
			}

			while (!isDone()) {
				loop();
			}
			finish();
		} catch (Exception e) {
			e.printStackTrace(System.out);
			System.exit(-1);
		}
		return success;
	}

	public void setDetermineFileSizes() {
		determineFileSizes = true;
	}

	public void setDone() {
		this.done = true;
	}

	@SuppressWarnings("static-method")
	public void taskFailure(TaskInstance task, ContainerId containerId) {
		String line;

		try {
			WorkflowDriver.writeToStdout("[script]");
			try (BufferedReader reader = new BufferedReader(new StringReader(task.getCommand()))) {
				int i = 0;
				while ((line = reader.readLine()) != null)
					WorkflowDriver.writeToStdout(String.format("%02d  %s", Integer.valueOf(++i), line));
			}

			Data stdoutFile = new Data(task.getId() + "_" + Invocation.STDOUT_FILENAME, containerId.toString());
			stdoutFile.stageIn();

			WorkflowDriver.writeToStdout("[out]");
			try (BufferedReader reader = new BufferedReader(new FileReader(stdoutFile.getLocalPath().toString()))) {
				while ((line = reader.readLine()) != null)
					WorkflowDriver.writeToStdout(line);
			}

			Data stderrFile = new Data(task.getId() + "_" + Invocation.STDERR_FILENAME, containerId.toString());
			stderrFile.stageIn();

			WorkflowDriver.writeToStdout("[err]");
			try (BufferedReader reader = new BufferedReader(new FileReader(stderrFile.getLocalPath().toString()))) {
				while ((line = reader.readLine()) != null)
					WorkflowDriver.writeToStdout(line);
			}
		} catch (IOException e) {
			e.printStackTrace(System.out);
		}

		WorkflowDriver.writeToStdout("[end]");
	}

	public void taskSuccess(TaskInstance task, ContainerId containerId) {
		try {
			for (TaskInstance childTask : task.getChildTasks()) {
				if (childTask.readyToExecute())
					scheduler.addTaskToQueue(childTask);
			}
		} catch (WorkflowStructureUnknownException e) {