Changeset 490 for DCWoRMS/trunk/src/schedframe/scheduling/policy/local

Timestamp:

10/09/12 13:58:06 (13 years ago)

Author:

wojtekp

Message:

 

File:

• DCWoRMS/trunk/src/schedframe/scheduling/policy/local/LocalManagementSystem.java (modified) (28 diffs)

DCWoRMS/trunk/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;
@@ -7 +9 @@
 import gridsim.gssim.DCWormsTags;
 import gridsim.gssim.filter.ExecTaskFilter;
-import gssim.schedframe.scheduling.ExecTask;
-import gssim.schedframe.scheduling.Executable;
 
 import java.util.ArrayList;
@@ -41 +41 @@
 import schedframe.scheduling.ResourceHistoryItem;
 import schedframe.scheduling.Scheduler;
+import schedframe.scheduling.TaskList;
 import schedframe.scheduling.TaskListImpl;
-import schedframe.scheduling.UsedResourceList;
+import schedframe.scheduling.UsedResourcesList;
 import schedframe.scheduling.WorkloadUnitHandler;
 import schedframe.scheduling.manager.resources.LocalResourceManager;
@@ -78 +79 @@
 
                 super(providerId, entityName, execTimeEstimationPlugin, queues);
-
-                //schedulingPlugin = (LocalSchedulingPlugin) InstanceFactory.createInstance(schedulingPluginClassName, LocalSchedulingPlugin.class);
                 
                 if (schedPlugin == null) {
@@ -85 +84 @@
                 }
                 this.schedulingPlugin =  schedPlugin;
-                accTotalLoad = new Accumulator();
-                moduleList = new ModuleListImpl(1);
-
+                this.moduleList = new ModuleListImpl(1);
+                
+                this.accTotalLoad = new Accumulator();
         }
 
         public void init(Scheduler sched, ManagedResources managedResources) {
                 super.init(sched, managedResources);
-                //scheduler = sched;
-                //resourceManager = ResourceManagerFactory.createResourceManager(scheduler);
                 double load = 0;
                 accTotalLoad.add(load);
@@ -115 +112 @@
                         }
                         sendTimerEvent();
-
                         break;
 
                 case DCWormsTags.TASK_READY_FOR_EXECUTION:
                         
-                        ExecTask data = (ExecTask) ev.get_data();
-                        try {
-                                data.setStatus(DCWormsTags.READY);
+                        ExecTask execTask = (ExecTask) ev.get_data();
+                        try {
+                                execTask.setStatus(DCWormsTags.READY);
                                 if (pluginSupportsEvent(tag)) {
-                                        SchedulingEvent event = new StartTaskExecutionEvent(data.getJobId(), data.getId());
+                                        SchedulingEvent event = new StartTaskExecutionEvent(execTask.getJobId(), execTask.getId());
                                         SchedulingPlanInterface<?> decision =  schedulingPlugin.schedule(event,
                                                         queues,  getJobRegistry(), getResourceManager(), moduleList);
@@ -136 +132 @@
                 case DCWormsTags.TASK_EXECUTION_FINISHED:
                         obj = ev.get_data();
-                        ExecTask exec = (ExecTask) obj;
-                        if (exec.getStatus() == DCWormsTags.INEXEC) {
-                                finalizeExecutable(exec);
-
-                                sendFinishedWorkloadUnit(exec);
-                                //task.setGridletStatus(Gridlet.SUCCESS);
-                                //task.finalizeGridlet();
-                                log.debug(exec.getJobId() + "_" + exec.getId() + " finished execution on " + new DateTime());
+                        execTask = (ExecTask) obj;
+                        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(jobRegistry.getRunningTasks().size()));
-                                /*UsedResourceList<ResourceHistoryItem> lastUsedList = task.getUsedResources();
-                                Map<ResourceUnitName, AbstractResourceUnit> lastUsed = lastUsedList.getLast()
-                                                .getResourceUnits();
-                                getAllocationManager().freeResources(lastUsed);
-                                ProcessingElements pes = (ProcessingElements) lastUsed.get(StandardResourceUnitName.PROCESSINGELEMENTS);
-                                for (ComputingResource resource : pes) {
-                                        resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, task));
+                                if (pluginSupportsEvent(tag)) {
+                                        SchedulingEvent event = new TaskFinishedEvent(execTask.getJobId(), execTask.getId());
+                                        SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
+                                                        queues, getJobRegistry(), getResourceManager(), moduleList);
+                                        executeSchedulingPlan(decision);
                                 }
-                                SubTaskFilter filter = new SubTaskFilter(task.getGridletID(), GssimTags.TASK_REQUESTED_TIME_EXPIRED);
-                                scheduler.sim_cancel(filter, null);
-                                super.sendFinishJob((Executable) task.getGridlet());*/
-                        }
+                        }
+
+                        Job job = jobRegistry.getJob(execTask.getJobId());
+                        if(!job.isFinished()){
+                                getWorkloadUnitHandler().handleJob(job);
+                        }
+                        break;
+                        
+                case DCWormsTags.TASK_REQUESTED_TIME_EXPIRED:
+                        obj = ev.get_data();
+                        execTask = (Executable) obj;
                         if (pluginSupportsEvent(tag)) {
-                                SchedulingEvent event = new TaskFinishedEvent(exec.getJobId(), exec.getId());
+                                SchedulingEvent event = new TaskRequestedTimeExpiredEvent(execTask.getJobId(), execTask.getId());
                                 SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
                                                 queues, getJobRegistry(), getResourceManager(), moduleList);
                                 executeSchedulingPlan(decision);
                         }
-                        Job job = jobRegistry.getJob(exec.getJobId());
-                        if(!job.isFinished()){
-                                getWorkloadUnitHandler().handleJob(job);
-                        }
-
                         break;
-                case DCWormsTags.TASK_REQUESTED_TIME_EXPIRED:
-                        obj = ev.get_data();
-                        exec = (Executable) obj;
-                        if (pluginSupportsEvent(tag)) {
-                                SchedulingEvent event = new TaskRequestedTimeExpiredEvent(exec.getJobId(), exec.getId());
-                                SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
-                                                queues, getJobRegistry(), getResourceManager(), moduleList);
-                                executeSchedulingPlan(decision);
-                        }
-
-                        break;
+                        
                 case DCWormsTags.UPDATE:
                         updateProcessingTimes(ev);
@@ -185 +169 @@
                 }
         }
-        
 
         public void notifyReturnedWorkloadUnit(WorkloadUnit wu) {
@@ -223 +206 @@
                                 } else {
                                         ExecTask exec = (ExecTask) task;
-                                        executeTask(exec, chooseResourcesForExecution(allocation.getProviderName(), (ExecTask)task));
+                                        executeTask(exec, chooseResourcesForExecution(allocation.getProviderName(), exec));
                                 }
                         }
@@ -237 +220 @@
                         return;
                 removeFromQueue(task);
-                //double completionPercentage = (submittedTask.getLength() - submittedTask.getRemainingGridletLength())/submittedTask.getLength();
+
                 SchedulingEvent event = new SchedulingEvent(SchedulingEventType.START_TASK_EXECUTION);
                 int time = Double.valueOf(
@@ -251 +234 @@
                 ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime);
                 exec.addUsedResources(resHistItem);
-
-                scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED,
-                                exec);
+                try {
+                        exec.setStatus(DCWormsTags.INEXEC);
+                } catch (Exception e) {
+                        // TODO Auto-generated catch block
+                        e.printStackTrace();
+                }
+                scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, exec);
 
                 try {
@@ -262 +249 @@
                         scheduler.sendInternal(t, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec);
                 }
-                
-                try {
-                        exec.setStatus(DCWormsTags.INEXEC);
-                } catch (Exception e1) {
-                        // TODO Auto-generated catch block
-                        e1.printStackTrace();
-                }
+
                 log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size()));
                 
                 PEUnit peUnit = (PEUnit)choosenResources.get(StandardResourceUnitName.PE);
-                if(peUnit instanceof ProcessingElements){
+                
+                notifyComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
+                
+                /*if(peUnit instanceof ProcessingElements){
                         ProcessingElements pes = (ProcessingElements) peUnit;
                         for (ComputingResource resource : pes) {
-                                resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec));
+                                resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, exec));
                         }
                 } else {
@@ -282 +266 @@
                                 resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
                         } catch (ResourceException e) {
-                                
-                        }
-                        resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec));
-                }
-                /*ProcessingElements pes = (ProcessingElements) choosenResources.get(StandardResourceUnitName.PE);
-                for (ComputingResource resource : pes) {
-                        resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, submittedTask));
-                }*/
+                                return;
+                        }
+                        resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, exec));
+                }
+*/
 
                 /*for(ExecTaskInterface etask : jobRegistry.getRunningTasks()){
@@ -301 +282 @@
                 
                 Executable exec = (Executable)execTask;
-                try {
-                        exec.setStatus(DCWormsTags.SUCCESS);
-                } catch (Exception e1) {
-                        // TODO Auto-generated catch block
-                        e1.printStackTrace();
-                }
                 exec.finalizeExecutable();
-                UsedResourceList<ResourceHistoryItem> lastUsedList = exec.getUsedResources();
-                Map<ResourceUnitName, ResourceUnit> lastUsed = lastUsedList.getLast()
-                                .getResourceUnits();
-                getAllocationManager().freeResources(lastUsed);
-                
-                PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE);
-                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) {
-                                
-                        }
-                        resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec));
-                }
-                /*ProcessingElements pes = (ProcessingElements) lastUsed.get(StandardResourceUnitName.PE);
-                for (ComputingResource resource : pes) {
-                        resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask));
-                }*/
+                
                 ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_REQUESTED_TIME_EXPIRED);
                 scheduler.sim_cancel(filter, null);
                 
-
+                Task task;
                 Job job = jobRegistry.getJob(exec.getJobId());
-
-                Task task = null;
                 try {
                         task = job.getTask(exec.getTaskId());
                 } catch (NoSuchFieldException e) {
-                        e.printStackTrace();
+                        return;
                 }
                 if(exec.getProcessesId() == null){
@@ -360 +310 @@
                         }
                 }
+                
+                UsedResourcesList lastUsedList = exec.getUsedResources();
+                Map<ResourceUnitName, ResourceUnit> lastUsed = lastUsedList.getLast()
+                                .getResourceUnits();
+                getAllocationManager().freeResources(lastUsed);
+                
+                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);
         }
@@ -374 +347 @@
                         ExecTask task = iter.next();
                         Executable exec = (Executable)task;
-                        UsedResourceList<ResourceHistoryItem> usedResourcesList = exec.getUsedResources();
-                        ResourceUnit unit = usedResourcesList.getLast().getResourceUnits()
+                        UsedResourcesList usedResourcesList = exec.getUsedResources();
+                        PEUnit peUnit = (PEUnit)usedResourcesList.getLast().getResourceUnits()
                                         .get(StandardResourceUnitName.PE);
 
-                        double load = getMIShare(timeSpan, (PEUnit) unit);
+                        double load = getMIShare(timeSpan, peUnit);
                         exec.setCompletionPercentage(100 * timeSpan/exec.getEstimatedDuration());
                         addTotalLoad(load);
                 }
         }
-
+        private void notifyComputingResources(PEUnit peUnit, EnergyEventType eventType, Object obj){
+
+                if(peUnit instanceof ProcessingElements){
+                        ProcessingElements pes = (ProcessingElements) peUnit;
+                        for (ComputingResource resource : pes) {
+                                resource.handleEvent(new EnergyEvent(eventType, obj));
+                        }
+                } else {
+                        ComputingResource resource = null;
+                        try {
+                                resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
+                        } catch (ResourceException e) {
+                                return;
+                        }
+                        resource.handleEvent(new EnergyEvent(eventType, obj));
+                }
+        }
+        
         private double getMIShare(double timeSpan, PEUnit pes) {
                 double localLoad;
@@ -402 +392 @@
         protected void updateProcessingTimes(Sim_event ev) {
                 updateProcessingProgress();
-                for (ExecTask task : jobRegistry.getRunningTasks()) {
-                        Executable exec = (Executable)task;
+                for (ExecTask execTask : jobRegistry.getRunningTasks()) {
+                        Executable exec = (Executable)execTask;
                         List<String> visitedResource = exec.getVisitedResources();
                         String originResource = ev.get_data().toString();
@@ -411 +401 @@
                         
                         Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getUsedResources().getLast().getResourceUnits();
-                        //double completionPercentage = (task.getLength() - subTask.getRemainingGridletLength())/task.getLength();
                         double time = execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.RESOURCE_STATE_CHANGED), 
-                                        task, choosenResources, exec.getCompletionPercentage());
-
-                        /*if(!subTask.getVisitedResources().contains(ev.get_data().toString())) {
-                                continue;
-                        }*/
+                                        execTask, choosenResources, exec.getCompletionPercentage());
+
                         //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){
@@ -424 +410 @@
                         ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED);
                         scheduler.sim_cancel(filter, null);
-                        scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, task);
-
+                        scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, execTask);
                 }
         }       
@@ -443 +428 @@
                                 numberOfPE = numberOfPE + resUnit.getAmount();
                         }
-                        //numberOfPE = getResourceManager().getPE().size();
                 } catch (Exception e) {
                         numberOfPE = 1;
@@ -464 +448 @@
                         ExecTask task) {
 
+                Map<ResourceUnitName, ResourceUnit> map = new HashMap<ResourceUnitName, ResourceUnit>();
                 ResourceManager resourceManager = this.resourceManager;
                 if(resourceName != null){
@@ -472 +457 @@
                                 return null;
                         }
-
                         resourceManager = new LocalResourceManager(resource);
                 }
-                Map<ResourceUnitName, ResourceUnit> map = new HashMap<ResourceUnitName, ResourceUnit>();
-
 
                 int cpuRequest;
@@ -485 +467 @@
                 }
 
-                //PEUnit processingUnits = null;
                 if (cpuRequest != 0) {
                         
@@ -494 +475 @@
                                 return null;
                         }
+                        
                         List<ResourceUnit> choosenPEUnits = new ArrayList<ResourceUnit>();
-
                         for (int i = 0; i < availableUnits.size() && cpuRequest > 0; i++) {
                                 PEUnit peUnit = (PEUnit) availableUnits .get(i);
@@ -508 +489 @@
                                 return null;
                         }
-
                         map.put(StandardResourceUnitName.PE, choosenPEUnits.get(0));
                 }
@@ -529 +509 @@
         class LocalWorkloadUnitHandler implements WorkloadUnitHandler{
                 
-                public void handleJob(Job job){
+                public void handleJob(JobInterface<?> job){
 
                         if (log.isInfoEnabled())
@@ -536 +516 @@
                         List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>();
                         jobsList.add(job);
-                        TaskListImpl readyTasks = new TaskListImpl();
+                        TaskListImpl availableTasks = new TaskListImpl();
                         for(Task task: jobRegistry.getAvailableTasks(jobsList)){
                                 task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED);
-                                readyTasks.add(task);
-                        }
-
-                        for(WorkloadUnit e:readyTasks){ 
-                                registerWorkloadUnit(e);
-                        }
-                }
-                
-                public void handleTask(TaskInterface<?> ti){
-                        Task task = (Task)ti;
+                                availableTasks.add(task);
+                        }
+
+                        for(TaskInterface<?> task: availableTasks){     
+                                registerWorkloadUnit(task);
+                        }
+                }
+                
+                public void handleTask(TaskInterface<?> t){
+                        Task task = (Task)t;
                         List<AbstractProcesses> processes = task.getProcesses();
 
@@ -564 +544 @@
                 
                 public void handleExecutable(ExecTask task){
+                        
                         Executable exec = (Executable) task;
-
-                        // int cost =
-                        // this.resourceManager.getResourceCharacteristic().getResUnits() !=
-                        // null ?
-                        // this.resourceManager.getResourceCharacteristic().getResUnits().get(ResourceParameterName.COST).getAmount()
-                        // : 1;
-
-                        exec.visitResource(scheduler.get_name());
+                        jobRegistry.addExecTask(exec);
+                        
+                        exec.trackResource(scheduler.get_name());
                         Scheduler parentScheduler = scheduler.getParent();
-                        while (parentScheduler != null && !exec.getVisitedResources().contains(parentScheduler.get_name())) {
-                                exec.visitResource(parentScheduler.get_name());
+                        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());
-                        jobRegistry.addExecTask(exec);
-                        TaskListImpl newTasks = new TaskListImpl();
+                        
+                        TaskList newTasks = new TaskListImpl();
                         newTasks.add(exec);