package schedframe.scheduling.policy.local; import eduni.simjava.Sim_event; import eduni.simjava.Sim_system; import gridsim.Accumulator; import gridsim.GridSimTags; import gridsim.Gridlet; import gridsim.ResourceCalendar; import gridsim.gssim.WormsTags; import gridsim.gssim.filter.SubTaskFilter; import gssim.schedframe.scheduling.ExecTask; import gssim.schedframe.scheduling.Executable; 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.ResourceController; import schedframe.events.scheduling.EventReason; import schedframe.events.scheduling.SchedulingEvent; import schedframe.events.scheduling.SchedulingEventType; import schedframe.events.scheduling.StartTaskExecutionEvent; import schedframe.events.scheduling.TaskCanceledEvent; import schedframe.events.scheduling.TaskFinishedEvent; import schedframe.events.scheduling.TaskRequestedTimeExpiredEvent; import schedframe.exceptions.ResourceException; 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.UsedResourceList; import schedframe.scheduling.WorkloadUnitHandler; import schedframe.scheduling.WorkloadUnitListImpl; 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.plugin.grid.ModuleType; 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.SubmittedTask; import schedframe.scheduling.tasks.Task; import schedframe.scheduling.tasks.TaskInterface; import schedframe.scheduling.tasks.WorkloadUnit; import simulator.WormsConstants; import simulator.utils.DoubleMath; public class LocalManagementSystem extends AbstractManagementSystem { private Log log = LogFactory.getLog(LocalManagementSystem.class); protected double lastUpdateTime; protected Accumulator accTotalLoad; public LocalManagementSystem(String providerId, String entityName, SchedulingPlugin schedPlugin, ExecutionTimeEstimationPlugin execTimeEstimationPlugin, TaskQueueList queues) throws Exception { super(providerId, entityName, execTimeEstimationPlugin, queues); //schedulingPlugin = (LocalSchedulingPlugin) InstanceFactory.createInstance(schedulingPluginClassName, LocalSchedulingPlugin.class); if (schedPlugin == null) { throw new Exception("Can not create local scheduling plugin instance."); } this.schedulingPlugin = schedPlugin; accTotalLoad = new Accumulator(); moduleList = new ModuleListImpl(1); } public void init(Scheduler sched, ManagedResources managedResources) { super.init(sched, managedResources); //scheduler = sched; //resourceManager = ResourceManagerFactory.createResourceManager(scheduler); double load = 0; accTotalLoad.add(load); } public void processEvent(Sim_event ev) { updateProcessingProgress(); int tag = ev.get_tag(); Object obj; switch (tag) { case WormsTags.TIMER: if (pluginSupportsEvent(tag)) { SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TIMER); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } sendTimerEvent(); break; case WormsTags.TASK_READY_FOR_EXECUTION: ExecTask data = (ExecTask) ev.get_data(); try { data.setStatus(Gridlet.READY); if (pluginSupportsEvent(tag)) { SchedulingEvent event = new StartTaskExecutionEvent(data.getJobId(), data.getId()); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } } catch (Exception e) { e.printStackTrace(); } break; case WormsTags.TASK_EXECUTION_FINISHED: obj = ev.get_data(); ExecTask task = (ExecTask) obj; if (task.getStatus() == Gridlet.INEXEC) { finalizeExecutable(task); SubmittedTask subTask = (SubmittedTask)task; sendFinishedWorkloadUnit((ExecTask)subTask.getGridlet()); //task.setGridletStatus(Gridlet.SUCCESS); //task.finalizeGridlet(); log.debug(task.getJobId() + "_" + task.getId() + " finished execution on " + new DateTime()); log.info(WormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size())); /*UsedResourceList lastUsedList = task.getUsedResources(); Map 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)); } SubTaskFilter filter = new SubTaskFilter(task.getGridletID(), GssimTags.TASK_REQUESTED_TIME_EXPIRED); scheduler.sim_cancel(filter, null); super.sendFinishJob((Executable) task.getGridlet());*/ } if (pluginSupportsEvent(tag)) { SchedulingEvent event = new TaskFinishedEvent(task.getJobId(), task.getId()); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } Job job = jobRegistry.get(task.getJobId()); if(!job.isFinished()){ getWorkloadUnitHandler().handleJob(job); } break; case WormsTags.TASK_REQUESTED_TIME_EXPIRED: obj = ev.get_data(); task = (SubmittedTask) obj; if (pluginSupportsEvent(tag)) { SchedulingEvent event = new TaskRequestedTimeExpiredEvent(task.getJobId(), task.getId()); SchedulingPlanInterface decision = schedulingPlugin.schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); executeSchedulingPlan(decision); } break; case WormsTags.UPDATE: updateProcessingTimes(ev); break; } } public void notifyReturnedWorkloadUnit(WorkloadUnit wu) { if (pluginSupportsEvent(WormsTags.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); //} } public void notifyCanceledWorkloadUnit(WorkloadUnit job) { if (!pluginSupportsEvent(GridSimTags.GRIDLET_CANCEL)) return; Executable executable = (Executable) job; String jobID = executable.getJobId(); SchedulingPlanInterface decision = null; try { executable.setStatus((int) BrokerConstants.JOB_STATUS_CANCELED); TaskCanceledEvent event = new TaskCanceledEvent(executable.getJobId(), executable.getTaskId()); event.setReason(EventReason.RESERVATION_EXCEEDED); decision = schedulingPlugin .schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); if (decision == null) return; executeSchedulingPlan(decision); } catch (Exception e) { log.error("Exception during scheduling. " + e.getMessage()); e.printStackTrace(); } } protected void executeSchedulingPlan(SchedulingPlanInterface decision) { ArrayList taskSchedulingDecisions = decision.getTasks(); for (int i = 0; i < taskSchedulingDecisions.size(); i++) { try { ScheduledTaskInterface taskDecision = taskSchedulingDecisions.get(i); // not scheduled again are returned to the user. if (taskDecision.getStatus() == AllocationStatus.REJECTED) { continue; } ArrayList allocations = taskDecision.getAllocations(); WorkloadUnit task = taskDecision.getTask(); for (int j = 0; j < allocations.size(); j++) { AllocationInterface allocation = allocations.get(j); if (allocation.isProcessing()) { ExecTask exec = (ExecTask) task; //Executable e = (Executable)task; /*SubmittedTask submittedTask = jobRegistry.getSubmittedTask(e.getJobId(), e.getId()); if(submittedTask == null) { submittedTask = new SubmittedTask(e); jobRegistry.addTask(submittedTask); }*/ /*e.visitResource(scheduler.get_name()); Scheduler parentScheduler = scheduler.getParent(); while (parentScheduler != null && !e.getVisitedResources().contains(parentScheduler.get_name())) { e.visitResource(parentScheduler.get_name()); parentScheduler = parentScheduler.getParent(); }*/ executeTask(exec, allocation.getRequestedResources()); //} else if(GridSim.getEntityId(allocation.getProviderName()) != -1 || scheduler.getScheduler(allocation.getProviderName())!=null){ } else if(resourceManager.getSchedulerName(allocation.getProviderName()) != null){ allocation.setProviderName(resourceManager.getSchedulerName(allocation.getProviderName())); submitWorkloadUnit(task, allocation); } else { ExecTask exec = (ExecTask) task; //Executable exec = jobRegistry.createExecutable(t, allocation); //exec.setResourceParameter(scheduler.get_id(), 1); /*e.visitResource(scheduler.get_name()); Scheduler parentScheduler = scheduler.getParent(); while (parentScheduler != null && !e.getVisitedResources().contains(parentScheduler.get_name())) { e.visitResource(parentScheduler.get_name()); parentScheduler = parentScheduler.getParent(); }*/ executeTask(exec, chooseResourcesForExecution(allocation.getProviderName(), (ExecTask)task)); } } } catch (Exception e) { e.printStackTrace(); } } } protected void executeTask(ExecTask task, Map choosenResources) { // Executable exec = (Executable) task; SubmittedTask submittedTask = (SubmittedTask)task; boolean allocationStatus = getAllocationManager().allocateResources(choosenResources); if(allocationStatus == false) return; removeFromQueue(task); //SubmittedTask submittedTask = (SubmittedTask)task; /* submittedTask = jobRegistry.getSubmittedTask(exec.getJobId(), exec.getId()); if(submittedTask == null) { submittedTask = new SubmittedTask(exec); jobRegistry.addTask(submittedTask); }*/ double completionPercentage = (submittedTask.getLength() - submittedTask.getRemainingGridletLength())/submittedTask.getLength(); SchedulingEvent event = new SchedulingEvent(SchedulingEventType.START_TASK_EXECUTION); int time = Double.valueOf( execTimeEstimationPlugin.execTimeEstimation(event, choosenResources, task, completionPercentage)).intValue(); log.debug(task.getJobId() + "_" + task.getId() + " starts executing on " + new DateTime() + " will finish after " + time); if (time < 0.0) return; submittedTask.setEstimatedDuration(time); DateTime currentTime = new DateTime(); ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime); submittedTask.addUsedResources(resHistItem); submittedTask.setFinishTime(currentTime.getMillis() / 1000); jobRegistry.saveHistory(submittedTask, time, choosenResources); scheduler.sendInternal(time, WormsTags.TASK_EXECUTION_FINISHED, submittedTask); try { long expectedDuration = submittedTask.getExpectedDuration().getMillis() / 1000; scheduler.sendInternal(expectedDuration, WormsTags.TASK_REQUESTED_TIME_EXPIRED, submittedTask); } catch (NoSuchFieldException e) { double t = submittedTask.getEstimatedDuration(); scheduler.sendInternal(t, WormsTags.TASK_REQUESTED_TIME_EXPIRED, submittedTask); } submittedTask.setGridletStatus(Gridlet.INEXEC); log.info(WormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size())); PEUnit peUnit = (PEUnit)choosenResources.get(StandardResourceUnitName.PE); if(peUnit instanceof ProcessingElements){ ProcessingElements pes = (ProcessingElements) peUnit; for (ComputingResource resource : pes) { resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, submittedTask)); } } else { ComputingResource resource = null; try { resource = ResourceController.getComputingResourceByName(peUnit.getResourceId()); } catch (ResourceException e) { } resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, submittedTask)); } /*ProcessingElements pes = (ProcessingElements) choosenResources.get(StandardResourceUnitName.PE); for (ComputingResource resource : pes) { resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, submittedTask)); }*/ /*for(ExecTaskInterface etask : jobRegistry.getRunningTasks()){ System.out.println(etask.getJobId()); for(String taskId: etask.getVisitedResources()) System.out.println("====="+taskId); }*/ } public void finalizeExecutable(ExecTask exec){ SubmittedTask subTask = (SubmittedTask)exec; subTask.setGridletStatus(Gridlet.SUCCESS); subTask.finalizeGridlet(); UsedResourceList lastUsedList = subTask.getUsedResources(); Map 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, subTask)); } } else { ComputingResource resource = null; try { resource = ResourceController.getComputingResourceByName(peUnit.getResourceId()); } catch (ResourceException e) { } resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask)); } /*ProcessingElements pes = (ProcessingElements) lastUsed.get(StandardResourceUnitName.PE); for (ComputingResource resource : pes) { resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask)); }*/ SubTaskFilter filter = new SubTaskFilter(subTask.getGridletID(), WormsTags.TASK_REQUESTED_TIME_EXPIRED); scheduler.sim_cancel(filter, null); Executable executable = (Executable) subTask.getGridlet(); Job job = jobRegistry.get(executable.getJobId()); Task task = null; try { task = job.getTask(executable.getTaskId()); } catch (NoSuchFieldException e) { e.printStackTrace(); } if(executable.getProcessesId() == null){ try { task.setStatus(executable.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(executable.getProcessesId())){ processes.setStatus(executable.getStatus()); break; } } } //sendFinishedWorkloadUnit(executable); } 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(); SubmittedTask subTask = (SubmittedTask)task; UsedResourceList usedResourcesList = subTask.getUsedResources(); ResourceUnit unit = usedResourcesList.getLast().getResourceUnits() .get(StandardResourceUnitName.PE); double load = getMIShare(timeSpan, (PEUnit) unit); subTask.updateGridletFinishedSoFar(load); addTotalLoad(load); } } private double getMIShare(double timeSpan, PEUnit pes) { double localLoad; ResourceCalendar resCalendar = (ResourceCalendar) moduleList.getModule(ModuleType.RESOURCE_CALENDAR); if (resCalendar == null) localLoad = 0; else // 1 - localLoad_ = available MI share percentage localLoad = resCalendar.getCurrentLoad(); int speed = pes.getSpeed(); int cnt = pes.getAmount(); double totalMI = speed * cnt * timeSpan * (1 - localLoad); return totalMI; } protected void updateProcessingTimes(Sim_event ev) { updateProcessingProgress(); for (ExecTask task : jobRegistry.getRunningTasks()) { SubmittedTask subTask = (SubmittedTask)task; List visitedResource = subTask.getVisitedResources(); String originResource = ev.get_data().toString(); if(!ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), originResource)){ continue; } Map choosenResources = subTask.getUsedResources().getLast().getResourceUnits(); double completionPercentage = (task.getLength() - subTask.getRemainingGridletLength())/task.getLength(); double time = execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.RESOURCE_STATE_CHANGED), choosenResources, task, completionPercentage); /*if(!subTask.getVisitedResources().contains(ev.get_data().toString())) { continue; }*/ //check if the new estimated end time is equal to the previous one; if yes the continue without update if( DoubleMath.subtract((subTask.getExecStartTime() + subTask.getEstimatedDuration()), (new DateTime().getMillis()/1000 + time)) == 0.0){ continue; } SubTaskFilter filter = new SubTaskFilter(subTask.getGridletID(), WormsTags.TASK_EXECUTION_FINISHED); scheduler.sim_cancel(filter, null); scheduler.sendInternal(time, WormsTags.TASK_EXECUTION_FINISHED, task); } } public double calculateTotalLoad(int size) { // background load, defined during initialization double load; ResourceCalendar resCalendar = (ResourceCalendar) moduleList.getModule(ModuleType.RESOURCE_CALENDAR); if (resCalendar == null) load = 0; else load = resCalendar.getCurrentLoad(); int numberOfPE = 0; try { for(ResourceUnit resUnit : getResourceManager().getPE()){ numberOfPE = numberOfPE + resUnit.getAmount(); } //numberOfPE = getResourceManager().getPE().size(); } catch (Exception e) { numberOfPE = 1; } double tasksPerPE = (double) size / numberOfPE; load += Math.min(1.0 - load, tasksPerPE); return load; } public Accumulator getTotalLoad() { return accTotalLoad; } protected void addTotalLoad(double load) { accTotalLoad.add(load); } private Map chooseResourcesForExecution(String resourceName, ExecTask task) { ResourceManager resourceManager = this.resourceManager; if(resourceName != null){ ComputingResource resource = null; try { resource = resourceManager.getResourceByName(resourceName); } catch (ResourceException e) { return null; } resourceManager = new LocalResourceManager(resource); } Map map = new HashMap(); int cpuRequest; try { cpuRequest = Double.valueOf(task.getCpuCntRequest()).intValue(); } catch (NoSuchFieldException e) { cpuRequest = 1; } //PEUnit processingUnits = null; if (cpuRequest != 0) { List availableUnits = null; try { availableUnits = getResourceManager().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; } /*try { List processingElements = resourceManager.getResourcesOfType(StandardResourceType.Processor); List choosenResources = new ArrayList(); int peSize = processingElements.size(); for (int i = 0; i < peSize && cpuRequest > 0; i++) { if (processingElements.get(i).getStatus() == ResourceStatus.FREE) { choosenResources.add(processingElements.get(i)); cpuRequest--; } } if (cpuRequest > 0) { return null; } processingUnits = new ProcessingElements(choosenResources); } catch (Exception e) { List procResUnit = resourceManager.getDistributedResourceUnits(StandardResourceUnitName.PE); for(ResourceUnit resUnit: procResUnit){ if (resUnit.getFreeAmount() >= cpuRequest) { processingUnits = new PEUnit(resUnit.getResourceId(), cpuRequest, cpuRequest); break; } } }*/ map.put(StandardResourceUnitName.PE, choosenPEUnits.get(0)); } /*int memoryRequest; try { memoryRequest = Double.valueOf(task.getMemoryRequest()).intValue(); } catch (NoSuchFieldException e) { memoryRequest = 0; } if (memoryRequest != 0) { List resUnit = resourceManager.getSharedResourceUnits().get(StandardResourceUnitName.MEMORY); Memory memory = null; for (ResourceUnit memUnit : resUnit) { Memory m = (Memory) memUnit; if (m.getFreeAmount() >= memoryRequest) { System.out.println(m.getResourceId()+ ";"+m.getAmount()+";"+m.getFreeAmount()); memory = new Memory(m, memoryRequest, memoryRequest); } } if(memory == null) return null; map.put(StandardResourceUnitName.MEMORY, memory); }*/ return map; } public void notifySubmittedWorkloadUnit(WorkloadUnit job, boolean ack) { updateProcessingProgress(); registerWorkloadUnit(job); } private void registerWorkloadUnit(WorkloadUnit wu){ if(!wu.isRegistered()){ wu.register(jobRegistry); } wu.accept(getWorkloadUnitHandler()); } class LocalWorkloadUnitHandler implements WorkloadUnitHandler{ public void handleJob(Job job){ if (log.isInfoEnabled()) log.info("Received job " + job.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis())); List> jobsList = new ArrayList>(); jobsList.add(job); WorkloadUnitListImpl readyTasks = new WorkloadUnitListImpl(); for(Task task: jobRegistry.getReadyTasks(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; if(task.getProcesses() == null){ Executable exec = new Executable(task); exec.setUserID(task.getSenderId()); exec.setLength(task.getLength()); registerWorkloadUnit(exec); } else { List processesList = task.getProcesses(); for(int i = 0; i < processesList.size(); i++){ AbstractProcesses processes = processesList.get(i); Executable exec = new Executable(task, processes); exec.setUserID(task.getSenderId()); exec.setLength(task.getLength()); registerWorkloadUnit(exec); } } } 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()); Scheduler parentScheduler = scheduler.getParent(); while (parentScheduler != null && !exec.getVisitedResources().contains(parentScheduler.get_name())) { exec.visitResource(parentScheduler.get_name()); parentScheduler = parentScheduler.getParent(); } exec.setResourceParameter(scheduler.get_id(), 1); SubmittedTask subTask = new SubmittedTask(exec); jobRegistry.addTask(subTask); WorkloadUnitListImpl newTasks = new WorkloadUnitListImpl(); newTasks.add(subTask); schedulingPlugin.placeJobsInQueues(newTasks, queues, getResourceManager(), moduleList); if (subTask.getStatus() == Gridlet.QUEUED) { sendExecutableReadyEvent(exec); } } public void handleSubmittedTask(SubmittedTask task){ task.visitResource(scheduler.get_name()); Scheduler parentScheduler = scheduler.getParent(); while (parentScheduler != null && !task.getVisitedResources().contains(parentScheduler.get_name())) { task.visitResource(parentScheduler.get_name()); parentScheduler = parentScheduler.getParent(); } jobRegistry.addTask(task); WorkloadUnitListImpl newTasks = new WorkloadUnitListImpl(); newTasks.add(task); schedulingPlugin.placeJobsInQueues(newTasks, queues, getResourceManager(), moduleList); if (task.getStatus() == Gridlet.QUEUED) { sendExecutableReadyEvent(task); } } } public WorkloadUnitHandler getWorkloadUnitHandler() { return new LocalWorkloadUnitHandler(); } public LocalResourceManager getResourceManager() { if (resourceManager instanceof ResourceManager) return (LocalResourceManager) resourceManager; else return null; } }