Changeset 1362 for DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/local/LocalManagementSystem.java

Timestamp:

06/03/14 15:12:11 (11 years ago)

Author:

wojtekp

Message:

 

File:

• DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/local/LocalManagementSystem.java (modified) (18 diffs)

DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/local/LocalManagementSystem.java

@@ -1 +1 @@
 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.LinkedList;
 import java.util.List;
 import java.util.Map;
-
-import org.apache.commons.lang.ArrayUtils;
+import java.util.Set;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -38 +32 @@
 import schedframe.resources.units.ResourceUnitName;
 import schedframe.resources.units.StandardResourceUnitName;
+import schedframe.scheduling.ExecutionHistoryItem;
 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;
@@ -64 +58 @@
 import simulator.stats.GSSAccumulator;
 import simulator.utils.DoubleMath;
+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;
 
 public class LocalManagementSystem extends AbstractManagementSystem {
@@ -127 +127 @@
                                 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());
@@ -164 +162 @@
                         break;
                         
+                case DCWormsTags.TASK_PAUSE:{
+                        String[] ids = (String[]) ev.get_data();
+                        execTask = jobRegistry.getTask(ids[0], ids[1]);
+                        taskPause(execTask);
+                        if (pluginSupportsEvent(tag)) {
+                                SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TASK_PAUSED);
+                                SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
+                                                queues, getJobRegistry(), getResourceManager(), moduleList);
+                                executeSchedulingPlan(decision);
+                        }
+                }
+                break;
+                
+                case DCWormsTags.TASK_RESUME:{
+                        String[] ids = (String[]) ev.get_data();
+                        execTask = jobRegistry.getTask(ids[0], ids[1]);
+                        taskResume(execTask, execTask.getAllocatedResources().getLast().getResourceUnits());
+                        if (pluginSupportsEvent(tag)) {
+                                SchedulingEvent event = new StartTaskExecutionEvent(ids[0], ids[1]);
+                                SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
+                                                queues, getJobRegistry(), getResourceManager(), moduleList);
+                                executeSchedulingPlan(decision);
+                        }
+                }
+                break;
+                        
+                case DCWormsTags.TASK_MOVE:{
+                        Object[] data = (Object[]) ev.get_data();
+                        execTask = jobRegistry.getTask((String)data[0], (String)data[1]);
+                        taskMove(execTask, (Map<ResourceUnitName, ResourceUnit>)data[2]);
+                        if (pluginSupportsEvent(tag)) {
+                                SchedulingEvent event = new StartTaskExecutionEvent((String)data[0], (String)data[1]);
+                                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);
+                        updateTaskExecutionPhase(execTask, SchedulingEventType.RESOURCE_STATE_CHANGED);
                         break;
-                        
+
                 case DCWormsTags.UPDATE_PROCESSING:
-                        updateProcessingTimes(ev);
+                        updateProcessingTimes();
                         break;
                         
@@ -177 +213 @@
                         if (pluginSupportsEvent(tag)) {
                                 SchedulingEvent event = new SchedulingEvent(SchedulingEventType.POWER_LIMIT_EXCEEDED);
-                                schedulingPlugin.schedule(event,
+                                SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
                                                 queues,  getJobRegistry(), getResourceManager(), moduleList);
+                                executeSchedulingPlan(decision);
                         }
                         break;
@@ -184 +221 @@
         }
 
+
+        public void taskPause(ExecTask execTask) {
+                if (execTask == null) {
+                        return;
+                } else {
+                        try {
+                                execTask.setStatus(DCWormsTags.PAUSED);
+                                
+                                Executable exec = (Executable) execTask;
+                                Map<ResourceUnitName, ResourceUnit> lastUsed = exec.getAllocatedResources().getLast().getResourceUnits();
+                                getAllocationManager().freeResources(lastUsed);
+                                
+                                saveExecutionHistory(exec, exec.getCompletionPercentage(), exec.getEstimatedDuration());
+                                
+                                ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), -1);
+                                scheduler.sim_cancel(filter, null);
+                                
+                                PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE);
+                                updateComputingResources(peUnit, EnergyEventType.TASK_FINISHED, exec);
+                        } catch (Exception e) {
+                                // TODO Auto-generated catch block
+                                e.printStackTrace();
+                        }
+                }               
+        }
+        
+        public void taskResume(ExecTask execTask, Map<ResourceUnitName, ResourceUnit> resources) {
+                if (execTask == null) {
+                        return;
+                } else if (execTask.getStatus() == DCWormsTags.PAUSED) {
+                        try {
+                                execTask.setStatus(DCWormsTags.RESUMED);
+                                Executable exec = (Executable) execTask;
+
+                                boolean status = allocateResources(exec, resources);
+                                if(status == false){
+                                        TaskList newTasks = new TaskListImpl();
+                                        newTasks.add(exec);
+                                        schedulingPlugin.placeTasksInQueues(newTasks, queues, getResourceManager(), moduleList);
+                                        exec.setStatus(DCWormsTags.READY);
+                                } else {
+                                        runTask(execTask);
+                                        
+                                        PEUnit peUnit = (PEUnit)resources.get(StandardResourceUnitName.PE);
+                                        updateComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
+                                }
+
+                        } catch (Exception e) {
+                                // TODO Auto-generated catch block
+                                e.printStackTrace();
+                        }
+                }                       
+        }
+        
+        public void taskMove(ExecTask execTask, Map<ResourceUnitName, ResourceUnit> map) {
+                taskPause(execTask);
+                taskResume(execTask, map);
+        }
+        
         public void notifyReturnedWorkloadUnit(WorkloadUnit wu) {
                 if (pluginSupportsEvent(DCWormsTags.TASK_EXECUTION_FINISHED)) {
@@ -229 +325 @@
 
                 Executable exec = (Executable)task;
-                boolean allocationStatus = getAllocationManager().allocateResources(choosenResources);
+                boolean allocationStatus = allocateResources(exec, choosenResources);
                 if(allocationStatus == false){
                         log.info("Task " + task.getJobId() + "_" + task.getId() + " requires more resources than is available at this moment.");
@@ -239 +335 @@
                 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();
-                }
+                runTask(exec);
                 
                 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 {
@@ -269 +349 @@
                 }
 
-                log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad());
-                
-        }
-        
+                log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad());            
+        }
+        
+        private void runTask(ExecTask execTask){
+                Executable exec = (Executable) execTask;
+                Map<ResourceUnitName, ResourceUnit> resources = exec.getAllocatedResources().getLast().getResourceUnits();
+                
+                try {
+                        exec.setStatus(DCWormsTags.INEXEC);
+                } catch (Exception e) {
+                        // TODO Auto-generated catch block
+                        e.printStackTrace();
+                }
+
+                int phaseDuration = Double.valueOf(execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.START_TASK_EXECUTION), 
+                                execTask, resources, exec.getCompletionPercentage())).intValue();
+
+                saveExecutionHistory(exec, exec.getCompletionPercentage(), phaseDuration);
+                if(exec.getResourceConsumptionProfile().isLast()){
+                        scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_FINISHED, execTask);;
+                } else {
+                        scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_CHANGED, execTask);
+                }
+                
+                PEUnit peUnit = (PEUnit)resources.get(StandardResourceUnitName.PE);
+                updateComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
+                
+        }
         protected void finalizeExecutable(ExecTask execTask){
                 
@@ -304 +408 @@
                         }
                 }
-                
-                UsedResourcesList lastUsedList = exec.getUsedResources();
-                Map<ResourceUnitName, ResourceUnit> lastUsed = lastUsedList.getLast()
-                                .getResourceUnits();
+
+                Map<ResourceUnitName, ResourceUnit> lastUsed = exec.getAllocatedResources().getLast().getResourceUnits();
                 getAllocationManager().freeResources(lastUsed);
+                
+                saveExecutionHistory(exec, 100,0);
                 
                 PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE);
                 updateComputingResources(peUnit, EnergyEventType.TASK_FINISHED, exec);
+                
+                log.debug(execTask.getJobId() + "_" + execTask.getId() + " finished execution on " + new DateTime());
+                log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad());
         }
         
@@ -326 +433 @@
                         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()  );       
-                        }
+                        ExecutionHistoryItem execHistItem = exec.getExecHistory().getLast();
+                        //System.out.println("--- upadteProgressX: " + Sim_system.sim_clock() );
+                        //System.out.println("taskId: " + exec.getId() + "; completion percentage: " + exec.getCompletionPercentage() + "; timespan: " + timeSpan + "; estimatedDuration: " +  execHistItem.getCompletionPercentage());
+                        exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * (timeSpan / (execHistItem.getEstimatedDuration()) * (1.0 - execHistItem.getCompletionPercentage()/100.0)));
+                        //System.out.println("newProgress: " + exec.getCompletionPercentage()  );       
+
                 }
         }
@@ -368 +468 @@
         }
 
-        protected void updateProcessingTimes(Sim_event ev) {
+        protected void updateProcessingTimes() {
                 for (ExecTask execTask : jobRegistry.getRunningTasks()) {
                         Executable exec = (Executable)execTask;
 
-                        Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getUsedResources().getLast().getResourceUnits();
-                        double lastTimeStamp = exec.getUsedResources().getLast().getTimeStamp().getMillis()/1000;
+
+                        Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getAllocatedResources().getLast().getResourceUnits();
+
                         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){
+                        ExecutionHistoryItem execHistItem = exec.getExecHistory().getLast();
+                        double lastTimeStamp = execHistItem.getTimeStamp().getMillis()/1000;
+                        if(DoubleMath.subtract((lastTimeStamp + execHistItem.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("execId: " + exec.getId() +  "; estimatedDuration " + execHistItem.getEstimatedDuration());
+                        //System.out.println("execId: " + exec.getId() + "; difference " + DoubleMath.subtract((lastTimeStamp + execHistItem.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()){
+                        saveExecutionHistory(exec, exec.getCompletionPercentage(), phaseDuration);
+                        
+                        if(exec.getResourceConsumptionProfile().isLast()){
                                 ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED);
                                 scheduler.sim_cancel(filter, null);
@@ -407 +506 @@
         }       
 
-        protected void updateTaskExecution(ExecTask execTask, SchedulingEventType schedEvType) {
+        protected void updateTaskExecutionPhase(ExecTask execTask, SchedulingEventType schedEvType) {
 
                 if (execTask.getStatus() == DCWormsTags.INEXEC) {
@@ -415 +514 @@
                                 exec.setStatus(DCWormsTags.NEW_EXEC_PHASE);
                         } catch (Exception e) {
-                        }
-                        
-                        Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getUsedResources().getLast().getResourceUnits();
+                                
+                        }
+
+                        Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getAllocatedResources().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()){
+                        
+                        saveExecutionHistory(exec, exec.getCompletionPercentage(), phaseDuration);
+                        
+                        if(exec.getResourceConsumptionProfile().isLast()){
                                 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);
-                        }
+                        }
+                        
+                        PEUnit peUnit = (PEUnit)exec.getAllocatedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
+                        updateComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec);
                 }
         }       
@@ -501 +600 @@
         }
         
-        public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) {
+        public void notifySubmittedWorkloadUnit(WorkloadUnit wu) {
                 //250314
                 //updateProcessingProgress();
@@ -562 +661 @@
                         Executable exec = (Executable) task;
                         jobRegistry.addExecTask(exec);
-                        
-                        exec.trackResource(scheduler.get_name());
-                        Scheduler parentScheduler = scheduler.getParent();
-                        List<String> 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());
+
+                        exec.setSchedulerName(scheduler.getFullName());
                         
                         TaskList newTasks = new TaskListImpl();
@@ -596 +687 @@
         }
         
+        
+        public void saveExecutionHistory(Executable exec, double completionPercentage, double estimatedDuration){
+
+                ExecutionHistoryItem execHistoryItem = new ExecutionHistoryItem(new DateTime());
+                execHistoryItem.setCompletionPercentage(completionPercentage);
+                execHistoryItem.setEstimatedDuration(estimatedDuration);
+                execHistoryItem.setResIndex(exec.getAllocatedResources().size() -1);
+                execHistoryItem.setStatus(exec.getStatus());
+                exec.addExecHistory(execHistoryItem);
+                
+                ProcessingElements pes = (ProcessingElements) exec.getAllocatedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
+                for (ComputingResource resource : pes) {
+
+                        LinkedList<ComputingResource> toExamine = new LinkedList<ComputingResource>();
+                        toExamine.push(resource);
+
+                        while (!toExamine.isEmpty()) {
+                                ComputingResource compResource = toExamine.pop();
+                                List<ComputingResource> resources = compResource.getChildren();
+                                if(resources.isEmpty()){
+                                        Set<String> visitedResources = exec.getAllocatedResources().getLast().getVisitedResources();
+                                        if(!visitedResources.contains(compResource.getFullName())){
+                                                exec.getAllocatedResources().getLast().trackResource(compResource.getFullName());
+                                        }
+                                } else {
+                                        for (int i = 0; i < resources.size(); i++) {
+                                                ComputingResource resourceChild = resources.get(i);
+                                                toExamine.addLast(resourceChild);
+                                        }
+                                }
+                        }
+                }
+        }
+        
+        public boolean allocateResources(Executable exec, Map<ResourceUnitName, ResourceUnit> choosenResources){
+                boolean allocationStatus = getAllocationManager().allocateResources(choosenResources);
+                if(allocationStatus){
+                        ResourceHistoryItem resourceHistoryItem = new ResourceHistoryItem(choosenResources);
+                        exec.addAllocatedResources(resourceHistoryItem);
+                        return true;
+                }
+                return false;
+        }
 }