package schedframe.scheduling.policy.local; import dcworms.schedframe.scheduling.ExecTask; import dcworms.schedframe.scheduling.Executable; import eduni.simjava.Sim_event; import eduni.simjava.Sim_system; import gridsim.dcworms.DCWormsTags; import gridsim.dcworms.filter.ExecTaskFilter; import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import org.apache.commons.lang.ArrayUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.joda.time.DateTime; import org.joda.time.DateTimeUtilsExt; import org.qcg.broker.schemas.schedulingplan.types.AllocationStatus; import qcg.shared.constants.BrokerConstants; import schedframe.SimulatedEnvironment; import schedframe.events.scheduling.SchedulingEvent; import schedframe.events.scheduling.SchedulingEventType; import schedframe.events.scheduling.StartTaskExecutionEvent; import schedframe.events.scheduling.TaskFinishedEvent; import schedframe.events.scheduling.TaskRequestedTimeExpiredEvent; import schedframe.exceptions.ResourceException; import schedframe.resources.StandardResourceType; import schedframe.resources.computing.ComputingResource; import schedframe.resources.computing.profiles.energy.EnergyEvent; import schedframe.resources.computing.profiles.energy.EnergyEventType; import schedframe.resources.units.PEUnit; import schedframe.resources.units.ProcessingElements; import schedframe.resources.units.ResourceUnit; import schedframe.resources.units.ResourceUnitName; import schedframe.resources.units.StandardResourceUnitName; import schedframe.scheduling.ResourceHistoryItem; import schedframe.scheduling.Scheduler; import schedframe.scheduling.TaskList; import schedframe.scheduling.TaskListImpl; import schedframe.scheduling.UsedResourcesList; import schedframe.scheduling.WorkloadUnitHandler; import schedframe.scheduling.manager.resources.LocalResourceManager; import schedframe.scheduling.manager.resources.ManagedResources; import schedframe.scheduling.manager.resources.ResourceManager; import schedframe.scheduling.plan.AllocationInterface; import schedframe.scheduling.plan.ScheduledTaskInterface; import schedframe.scheduling.plan.SchedulingPlanInterface; import schedframe.scheduling.plugin.SchedulingPlugin; import schedframe.scheduling.plugin.estimation.ExecutionTimeEstimationPlugin; import schedframe.scheduling.plugin.grid.ModuleListImpl; import schedframe.scheduling.policy.AbstractManagementSystem; import schedframe.scheduling.queue.TaskQueueList; import schedframe.scheduling.tasks.AbstractProcesses; import schedframe.scheduling.tasks.Job; import schedframe.scheduling.tasks.JobInterface; import schedframe.scheduling.tasks.Task; import schedframe.scheduling.tasks.TaskInterface; import schedframe.scheduling.tasks.WorkloadUnit; import simulator.DCWormsConstants; import simulator.stats.GSSAccumulator; import simulator.utils.DoubleMath; public class LocalManagementSystem extends AbstractManagementSystem { private Log log = LogFactory.getLog(LocalManagementSystem.class); protected double lastUpdateTime; public LocalManagementSystem(String providerId, String entityName, SchedulingPlugin schedPlugin, ExecutionTimeEstimationPlugin execTimeEstimationPlugin, TaskQueueList queues) throws Exception { super(providerId, entityName, execTimeEstimationPlugin, queues); if (schedPlugin == null) { throw new Exception("Can not create local scheduling plugin instance."); } this.schedulingPlugin = schedPlugin; this.moduleList = new ModuleListImpl(1); } public void init(Scheduler sched, ManagedResources managedResources) { super.init(sched, managedResources); } public void processEvent(Sim_event ev) { updateProcessingProgress(); int tag = ev.get_tag(); switch (tag) { case DCWormsTags.TIMER: if (pluginSupportsEvent(tag)) { SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TIMER); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } sendTimerEvent(); break; case DCWormsTags.TASK_READY_FOR_EXECUTION: ExecTask execTask = (ExecTask) ev.get_data(); try { execTask.setStatus(DCWormsTags.READY); if (pluginSupportsEvent(tag)) { SchedulingEvent event = new StartTaskExecutionEvent(execTask.getJobId(), execTask.getId()); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } } catch (Exception e) { e.printStackTrace(); } break; case DCWormsTags.TASK_EXECUTION_FINISHED: execTask = (ExecTask) ev.get_data(); if (execTask.getStatus() == DCWormsTags.INEXEC) { finalizeExecutable(execTask); sendFinishedWorkloadUnit(execTask); log.debug(execTask.getJobId() + "_" + execTask.getId() + " finished execution on " + new DateTime()); log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad()); if (pluginSupportsEvent(tag)) { SchedulingEvent event = new TaskFinishedEvent(execTask.getJobId(), execTask.getId()); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } } Job job = jobRegistry.getJob(execTask.getJobId()); if(!job.isFinished()){ getWorkloadUnitHandler().handleJob(job); } //SUPPORTS PRECEDING CONSTRAINST COMING BOTH FROM SWF FILES /*else { try { job.setStatus((int)BrokerConstants.JOB_STATUS_FINISHED); } catch (Exception e) { } for(JobInterface j: jobRegistry.getJobs((int)BrokerConstants.JOB_STATUS_SUBMITTED)){ getWorkloadUnitHandler().handleJob(j); } }*/ break; case DCWormsTags.TASK_REQUESTED_TIME_EXPIRED: execTask = (Executable) ev.get_data(); if (pluginSupportsEvent(tag)) { SchedulingEvent event = new TaskRequestedTimeExpiredEvent(execTask.getJobId(), execTask.getId()); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } break; case DCWormsTags.TASK_EXECUTION_CHANGED: execTask = (ExecTask) ev.get_data(); updateTaskExecution(execTask, SchedulingEventType.RESOURCE_STATE_CHANGED); break; case DCWormsTags.UPDATE_PROCESSING: updateProcessingTimes(ev); break; case DCWormsTags.POWER_LIMIT_EXCEEDED: if (pluginSupportsEvent(tag)) { SchedulingEvent event = new SchedulingEvent(SchedulingEventType.POWER_LIMIT_EXCEEDED); schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); } break; } } public void notifyReturnedWorkloadUnit(WorkloadUnit wu) { if (pluginSupportsEvent(DCWormsTags.TASK_EXECUTION_FINISHED)) { SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TASK_FINISHED); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } //if(scheduler.getParent() != null){ sendFinishedWorkloadUnit(wu); //} } protected void executeSchedulingPlan(SchedulingPlanInterface decision) { ArrayList> taskSchedulingDecisions = decision.getTasks(); for (int i = 0; i < taskSchedulingDecisions.size(); i++) { ScheduledTaskInterface taskDecision = taskSchedulingDecisions.get(i); if (taskDecision.getStatus() == AllocationStatus.REJECTED) { continue; } ArrayList> allocations = taskDecision.getAllocations(); TaskInterface task = taskDecision.getTask(); for (int j = 0; j < allocations.size(); j++) { AllocationInterface allocation = allocations.get(j); if (allocation.getRequestedResources() == null || allocation.getRequestedResources().size() > 0) { ExecTask exec = (ExecTask) task; executeTask(exec, allocation.getRequestedResources()); } else if(resourceManager.getSchedulerName(allocation.getProviderName()) != null){ allocation.setProviderName(resourceManager.getSchedulerName(allocation.getProviderName())); submitTask(task, allocation); } else { ExecTask exec = (ExecTask) task; executeTask(exec, chooseResourcesForExecution(allocation.getProviderName(), exec)); } } } } protected void executeTask(ExecTask task, Map choosenResources) { Executable exec = (Executable)task; boolean allocationStatus = getAllocationManager().allocateResources(choosenResources); if(allocationStatus == false){ log.info("Task " + task.getJobId() + "_" + task.getId() + " requires more resources than is available at this moment."); return; } removeFromQueue(task); log.debug(task.getJobId() + "_" + task.getId() + " starts executing on " + new DateTime()); //if (phaseDuration < 0.0) // return; //exec.setEstimatedDuration(exec.getEstimatedDuration() + phaseDuration); DateTime currentTime = new DateTime(); ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime); resHistItem.setCompletionPercentage(0); exec.addUsedResources(resHistItem); try { exec.setStatus(DCWormsTags.INEXEC); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } PEUnit peUnit = (PEUnit)choosenResources.get(StandardResourceUnitName.PE); updateComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec); updateTaskExecution(exec, SchedulingEventType.START_TASK_EXECUTION); //scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, exec); try { long expectedDuration = exec.getExpectedDuration().getMillis() / 1000; scheduler.sendInternal(expectedDuration, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec); } catch (NoSuchFieldException e) { //double t = exec.getEstimatedDuration(); //scheduler.sendInternal(t, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec); } log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad()); } protected void finalizeExecutable(ExecTask execTask){ Executable exec = (Executable)execTask; exec.finalizeExecutable(); ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_REQUESTED_TIME_EXPIRED); scheduler.sim_cancel(filter, null); Task task; Job job = jobRegistry.getJob(exec.getJobId()); try { task = job.getTask(exec.getTaskId()); } catch (NoSuchFieldException e) { return; } if(exec.getProcessesId() == null){ try { task.setStatus(exec.getStatus()); } catch (Exception e) { e.printStackTrace(); } } else { List processesList = task.getProcesses(); for(int i = 0; i < processesList.size(); i++){ AbstractProcesses processes = processesList.get(i); if(processes.getId().equals(exec.getProcessesId())){ processes.setStatus(exec.getStatus()); break; } } } UsedResourcesList lastUsedList = exec.getUsedResources(); Map lastUsed = lastUsedList.getLast() .getResourceUnits(); getAllocationManager().freeResources(lastUsed); PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE); updateComputingResources(peUnit, EnergyEventType.TASK_FINISHED, exec); } protected void updateProcessingProgress() { double timeSpan = DoubleMath.subtract(Sim_system.clock(), lastUpdateTime); if (timeSpan <= 0.0) { // don't update when nothing changed return; } lastUpdateTime = Sim_system.clock(); Iterator iter = jobRegistry.getRunningTasks().iterator(); while (iter.hasNext()) { ExecTask task = iter.next(); Executable exec = (Executable)task; if(exec.getUsedResources().size() > 1){ //System.out.println("--- upadteProgressX: " + Sim_system.sim_clock() ); //System.out.println("taskId: " + exec.getId() + "; completion percentage: " + exec.getCompletionPercentage() + "; timespan: " + timeSpan + "; estimatedDuration: " + exec.getEstimatedDuration()); exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * (timeSpan / (exec.getEstimatedDuration()) * (1.0 - exec.getUsedResources().get(exec.getUsedResources().size()-1).getCompletionPercentage()/100.0))); //System.out.println("newProgress: " + exec.getCompletionPercentage() ); } else { //System.out.println("--- upadteProgress2: " + Sim_system.sim_clock() ); //System.out.println("taskId: " + exec.getId() + "; completion percentage: " + exec.getCompletionPercentage() + "; timespan: " + timeSpan + "; estimatedDuration: " + exec.getEstimatedDuration()); exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * (timeSpan / exec.getEstimatedDuration())); //System.out.println("newProgress: " + exec.getCompletionPercentage() ); } } } private void updateComputingResources(PEUnit peUnit, EnergyEventType eventType, Object obj){ if(peUnit instanceof ProcessingElements){ ProcessingElements pes = (ProcessingElements) peUnit; for (ComputingResource resource : pes) { resource.handleEvent(new EnergyEvent(eventType, obj, scheduler.getFullName())); //DataCenterWorkloadSimulator.getEventManager().sendToResources(resource.getType(), 0, new EnergyEvent(eventType, obj)); } /*try { for (ComputingResource resource : resourceManager.getResourcesOfType(pes.get(0).getType())) { resource.handleEvent(new EnergyEvent(eventType, obj)); } } catch (ResourceException e) { // TODO Auto-generated catch block e.printStackTrace(); }*/ } else { ComputingResource resource = null; try { resource = SimulatedEnvironment.getComputingResourceByName(peUnit.getResourceId()); } catch (ResourceException e) { return; } resource.handleEvent(new EnergyEvent(eventType, obj, scheduler.getFullName())); } } protected void updateProcessingTimes(Sim_event ev) { for (ExecTask execTask : jobRegistry.getRunningTasks()) { Executable exec = (Executable)execTask; Map choosenResources = exec.getUsedResources().getLast().getResourceUnits(); double lastTimeStamp = exec.getUsedResources().getLast().getTimeStamp().getMillis()/1000; int phaseDuration = Double.valueOf(execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.RESOURCE_STATE_CHANGED), execTask, choosenResources, exec.getCompletionPercentage())).intValue(); if(DoubleMath.subtract((lastTimeStamp + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + phaseDuration)) == 0.0){ continue; } //System.out.println("=== upadteTIme: " + Sim_system.sim_clock() ); //System.out.println("execId: " + exec.getId() + "; estimatedDuration " + exec.getEstimatedDuration()); //System.out.println("execId: " + exec.getId() + "; difference " + DoubleMath.subtract((lastTimeStamp + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + phaseDuration))); //System.out.println("completionPercantage: " + exec.getCompletionPercentage() + "; basic duration: " +exec.getResourceConsumptionProfile().getCurrentResourceConsumption().getDuration() + "; phaseDuration: " + phaseDuration); exec.setEstimatedDuration(phaseDuration); DateTime currentTime = new DateTime(); ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime); resHistItem.setCompletionPercentage(exec.getCompletionPercentage()); exec.addUsedResources(resHistItem); if(exec.getResourceConsumptionProfile().getCurrentResourceConsumption() == exec.getResourceConsumptionProfile().getResourceConsumptionList().getLast()){ ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED); scheduler.sim_cancel(filter, null); scheduler.sendInternal(phaseDuration , DCWormsTags.TASK_EXECUTION_FINISHED, execTask); //PEUnit peUnit = (PEUnit)exec.getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE); //notifyComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec); } else{ ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_CHANGED); scheduler.sim_cancel(filter, null); scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_CHANGED, execTask); //PEUnit peUnit = (PEUnit)exec.getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE); //notifyComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec); } } } protected void updateTaskExecution(ExecTask execTask, SchedulingEventType schedEvType) { if (execTask.getStatus() == DCWormsTags.INEXEC) { Executable exec = (Executable)execTask; try { exec.setStatus(DCWormsTags.NEW_EXEC_PHASE); } catch (Exception e) { } Map choosenResources = exec.getUsedResources().getLast().getResourceUnits(); int phaseDuration = Double.valueOf(execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(schedEvType), execTask, choosenResources, exec.getCompletionPercentage())).intValue(); exec.setEstimatedDuration(phaseDuration); if(exec.getResourceConsumptionProfile().getCurrentResourceConsumption() == exec.getResourceConsumptionProfile().getResourceConsumptionList().getLast()){ scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_FINISHED, execTask); PEUnit peUnit = (PEUnit)exec.getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE); updateComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec); } else { scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_CHANGED, execTask); PEUnit peUnit = (PEUnit)exec.getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE); updateComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec); } } } public double calculateTotalLoad() { GSSAccumulator loadAcc = new GSSAccumulator(); try { for(ComputingResource compRes: getResourceManager().getResourcesOfType(StandardResourceType.Node)){ loadAcc.add(compRes.getLoadInterface().getRecentUtilization().getValue()); } } catch (ResourceException e) { // TODO Auto-generated catch block e.printStackTrace(); } return loadAcc.getMean(); } private Map chooseResourcesForExecution(String resourceName, ExecTask task) { Map map = new HashMap(); LocalResourceManager resourceManager = getResourceManager(); if(resourceName != null){ ComputingResource resource = null; try { resource = resourceManager.getResourceByName(resourceName); } catch (ResourceException e) { return null; } resourceManager = new LocalResourceManager(resource); } int cpuRequest; try { cpuRequest = Double.valueOf(task.getCpuCntRequest()).intValue(); } catch (NoSuchFieldException e) { cpuRequest = 1; } if (cpuRequest != 0) { List availableUnits = null; try { availableUnits = resourceManager.getPE(); } catch (ResourceException e) { return null; } List choosenPEUnits = new ArrayList(); for (int i = 0; i < availableUnits.size() && cpuRequest > 0; i++) { PEUnit peUnit = (PEUnit) availableUnits.get(i); if(peUnit.getFreeAmount() > 0){ int allocPE = Math.min(peUnit.getFreeAmount(), cpuRequest); cpuRequest = cpuRequest - allocPE; choosenPEUnits.add(peUnit.replicate(allocPE)); } } if(cpuRequest > 0){ return null; } map.put(StandardResourceUnitName.PE, choosenPEUnits.get(0)); } return map; } public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) { //250314 //updateProcessingProgress(); if (log.isInfoEnabled()) log.info("Received job " + wu.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis())); registerWorkloadUnit(wu); } private void registerWorkloadUnit(WorkloadUnit wu){ if(!wu.isRegistered()){ wu.register(jobRegistry); } wu.accept(getWorkloadUnitHandler()); } class LocalWorkloadUnitHandler implements WorkloadUnitHandler{ public void handleJob(JobInterface job){ if (log.isInfoEnabled()) log.info("Handle " + job.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis())); try { if(job.getStatus() == BrokerConstants.JOB_STATUS_UNSUBMITTED) job.setStatus((int)BrokerConstants.JOB_STATUS_SUBMITTED); } catch (Exception e) { e.printStackTrace(); } List> jobsList = new ArrayList>(); jobsList.add(job); TaskListImpl availableTasks = new TaskListImpl(); for(Task task: jobRegistry.getAvailableTasks(jobsList)){ task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED); availableTasks.add(task); } for(TaskInterface task: availableTasks){ registerWorkloadUnit(task); } } public void handleTask(TaskInterface t){ Task task = (Task)t; List processes = task.getProcesses(); if(processes == null || processes.size() == 0){ Executable exec = new Executable(task); registerWorkloadUnit(exec); } else { for(int j = 0; j < processes.size(); j++){ AbstractProcesses procesesSet = processes.get(j); Executable exec = new Executable(task, procesesSet); registerWorkloadUnit(exec); } } } public void handleExecutable(ExecTask task){ Executable exec = (Executable) task; jobRegistry.addExecTask(exec); exec.trackResource(scheduler.get_name()); Scheduler parentScheduler = scheduler.getParent(); List visitedResource = exec.getVisitedResources(); String [] visitedResourcesArray = visitedResource.toArray(new String[visitedResource.size()]); while (parentScheduler != null && !ArrayUtils.contains(visitedResourcesArray, parentScheduler.get_name())) { exec.trackResource(parentScheduler.get_name()); parentScheduler = parentScheduler.getParent(); } exec.setSchedulerName(scheduler.get_id()); TaskList newTasks = new TaskListImpl(); newTasks.add(exec); schedulingPlugin.placeTasksInQueues(newTasks, queues, getResourceManager(), moduleList); if (exec.getStatus() == DCWormsTags.QUEUED) { sendExecutableReadyEvent(exec); } } } public WorkloadUnitHandler getWorkloadUnitHandler() { return new LocalWorkloadUnitHandler(); } public LocalResourceManager getResourceManager() { if (resourceManager instanceof ResourceManager) return (LocalResourceManager) resourceManager; else return null; } }