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

Timestamp:

03/04/13 08:25:34 (12 years ago)

Author:

wojtekp

Message:

 

File:

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

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

@@ -164 +164 @@
                         break;
                         
-                case DCWormsTags.PHASE_CHANGED:
+                case DCWormsTags.TASK_EXECUTION_CHANGED:
                         execTask = (ExecTask) ev.get_data();
-                        //updatePhases(execTask);
+                        updateTaskExecution(execTask, SchedulingEventType.RESOURCE_STATE_CHANGED);
                         break;
                 }
@@ -224 +224 @@
                 removeFromQueue(task);
 
-                SchedulingEvent event = new SchedulingEvent(SchedulingEventType.START_TASK_EXECUTION);
-                int time = Double.valueOf(
-                                execTimeEstimationPlugin.execTimeEstimation(event, task, choosenResources, exec.getCompletionPercentage())).intValue();
-                log.debug(task.getJobId() + "_" + task.getId() + " starts executing on " + new DateTime()
-                                + " will finish after " + time);
-
-                if (time < 0.0)
-                        return;
-
-                exec.setEstimatedDuration(time);
+                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);
                 exec.addUsedResources(resHistItem);
+                
                 try {
                         exec.setStatus(DCWormsTags.INEXEC);
@@ -244 +241 @@
                 }
                 
-                //updatePhases(exec);
-                scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, exec);
+                updateTaskExecution(exec, SchedulingEventType.START_TASK_EXECUTION);
+                //scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, exec);
 
                 try {
@@ -251 +248 @@
                         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);
+                        //double t = exec.getEstimatedDuration();
+                        //scheduler.sendInternal(t, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec);
                 }
 
@@ -261 +258 @@
                 notifyComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
                 
-                /*if(peUnit instanceof ProcessingElements){
-                        ProcessingElements pes = (ProcessingElements) peUnit;
-                        for (ComputingResource resource : pes) {
-                                resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, exec));
-                        }
-                } else {
-                        ComputingResource resource = null;
-                        try {
-                                resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
-                        } catch (ResourceException e) {
-                                return;
-                        }
-                        resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, exec));
-                }
-*/
-
-                /*for(ExecTaskInterface etask : jobRegistry.getRunningTasks()){
-                        System.out.println(etask.getJobId());
-                        for(String taskId: etask.getVisitedResources())
-                                System.out.println("====="+taskId);
-                }*/
         }
         
@@ -323 +299 @@
                 PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE);
                 notifyComputingResources(peUnit, EnergyEventType.TASK_FINISHED, exec);
-                /*if(peUnit instanceof ProcessingElements){
-                        ProcessingElements pes = (ProcessingElements) peUnit;
-                        for (ComputingResource resource : pes) {
-                                resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec));
-                        }
-                } else {
-                        ComputingResource resource = null;
-                        try {
-                                resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
-                        } catch (ResourceException e) {
-                                return;
-                        }
-                        resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec));
-                }*/
-
-                //sendFinishedWorkloadUnit(executable);
         }
         
@@ -353 +313 @@
                         Executable exec = (Executable)task;
                         //exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * timeSpan/exec.getEstimatedDuration());
-                        exec.setCompletionPercentage(exec.getCompletionPercentage() + (100 - exec.getCompletionPercentage()) * timeSpan/(exec.getEstimatedDuration() - new DateTime().getMillis()/1000 + exec.getExecStartTime() + timeSpan));
-                        
+                        exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * (timeSpan / exec.getEstimatedDuration()));
                         UsedResourcesList usedResourcesList = exec.getUsedResources();
                         PEUnit peUnit = (PEUnit)usedResourcesList.getLast().getResourceUnits()
@@ -364 +323 @@
         
         private void notifyComputingResources(PEUnit peUnit, EnergyEventType eventType, Object obj){
-
                 if(peUnit instanceof ProcessingElements){
                         ProcessingElements pes = (ProcessingElements) peUnit;
@@ -409 +367 @@
                 for (ExecTask execTask : jobRegistry.getRunningTasks()) {
                         Executable exec = (Executable)execTask;
-                        List<String> visitedResource = exec.getVisitedResources();
-                        String originResource = ev.get_data().toString();
-                        if(!ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), originResource)){
-                                continue;
-                        }
-                        
+
                         Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getUsedResources().getLast().getResourceUnits();
-                        int time = Double.valueOf(execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.RESOURCE_STATE_CHANGED), 
+                        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();
 
-                        //check if the new estimated end time is equal to the previous one; if yes the continue without update
-                        if(DoubleMath.subtract((exec.getExecStartTime() + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + time)) == 0.0){
-                                continue;
-                        }
-                        //exec.setEstimatedDuration(time);
-                        exec.setEstimatedDuration(Long.valueOf(new DateTime().getMillis()/1000).intValue() - Double.valueOf(exec.getExecStartTime()).intValue() + time);
-                        ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED);
-                        scheduler.sim_cancel(filter, null);
-                        scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, execTask);
+                        if(DoubleMath.subtract((lastTimeStamp + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + phaseDuration)) == 0.0){
+                                return;
+                        }
+
+                        exec.setEstimatedDuration(phaseDuration);
+                        DateTime currentTime = new DateTime();
+                        ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime);
+                        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 updatePhases(ExecTask execTask) {
-                updateProcessingProgress();
+
+        protected void updateTaskExecution(ExecTask execTask, SchedulingEventType schedEvType) {
 
                 if (execTask.getStatus() == DCWormsTags.INEXEC) {
                         Executable exec = (Executable)execTask;
-
-                        double phaseLength = 0;
-                        try{
-                                phaseLength = exec.getResourceConsumptionProfile().getResourceConsumption(exec.getCurrentPhase()).getResourceConsumptionTypeChoice().getPercentage()/100;
-                        } catch(Exception e){
-                                ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED);
-                                scheduler.sim_cancel(filter, null);
-                                double t = DoubleMath.subtract((exec.getExecStartTime() + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000));
-                                scheduler.sendInternal(t, DCWormsTags.TASK_EXECUTION_FINISHED, execTask);
-                                PEUnit peUnit = (PEUnit)exec.getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
-                                notifyComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec);
-                        }
-                        if(phaseLength != 0){                   
-                                exec.setCurrentPhase(exec.getCurrentPhase() + 1);
-                                ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.PHASE_CHANGED);
-                                scheduler.sim_cancel(filter, null);
-                                scheduler.sendInternal(phaseLength * exec.getEstimatedDuration(), DCWormsTags.PHASE_CHANGED, execTask);
-                                PEUnit peUnit = (PEUnit)exec.getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
-                                notifyComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec);
-                        } 
-                        System.out.println("===" + exec.getJobId() + ":" + phaseLength);
-                }
-        }       */
+                        
+                        try {
+                                exec.setStatus(DCWormsTags.NEW_EXEC_PHASE);
+                        } catch (Exception e) {
+                        }
+                        
+                        Map<ResourceUnitName, ResourceUnit> 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);
+                        } else {
+                                scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_CHANGED, execTask);
+                        }
+                }
+        }       
 
         public double calculateTotalLoad(int size) {