package experiments.e2dc2013.recs.plugins.scheduling.exp1; import experiments.e2dc2013.recs.plugins.scheduling.RecsSP; import gridsim.dcworms.DCWormsTags; import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.MissingResourceException; import schedframe.events.scheduling.SchedulingEvent; import schedframe.resources.ResourceStatus; import schedframe.resources.computing.Node; import schedframe.resources.computing.ComputingResource; import schedframe.resources.computing.Core; import schedframe.resources.computing.Processor; import schedframe.resources.computing.profiles.energy.airthroughput.StandardAirflowStateName; import schedframe.resources.computing.profiles.energy.airthroughput.CustomAirflowStateName; import schedframe.resources.computing.profiles.energy.power.StandardPowerStateName; import schedframe.resources.units.ProcessingElements; import schedframe.resources.units.ResourceUnit; import schedframe.resources.units.ResourceUnitName; import schedframe.resources.units.StandardResourceUnitName; import schedframe.scheduling.manager.resources.ClusterResourceManager; import schedframe.scheduling.manager.resources.ResourceManager; import schedframe.scheduling.manager.tasks.JobRegistry; import schedframe.scheduling.manager.tasks.JobRegistryImpl; import schedframe.scheduling.plan.SchedulingPlanInterface; import schedframe.scheduling.plan.impl.SchedulingPlan; import schedframe.scheduling.plugin.ModuleList; import schedframe.scheduling.queue.TaskQueue; import schedframe.scheduling.queue.TaskQueueList; import schedframe.scheduling.tasks.TaskInterface; public class RecsInOffSP extends RecsSP { private static int START_ID = 9; private static int END_ID = 17; public SchedulingPlanInterface schedule(SchedulingEvent event, TaskQueueList queues, JobRegistry jobRegistry, ResourceManager resManager, ModuleList modules) { ClusterResourceManager resourceManager = (ClusterResourceManager) resManager; SchedulingPlan plan = new SchedulingPlan(); // choose the events types to serve. // Different actions for different events are possible. switch (event.getType()) { case START_TASK_EXECUTION: case TASK_FINISHED: //case TIMER: // our tasks are placed only in first queue (see BaseLocalSchedulingPlugin.placeJobsInQueues() method) TaskQueue q = queues.get(0); List notSelectedNodes = resourceManager.getNodes(); // check all tasks in queue for (int i = 0; i < q.size(); i++) { TaskInterface task = q.get(i); // if status of the tasks in READY if (task.getStatus() == DCWormsTags.READY) { Node node = chooseProvider(resourceManager, task); if (node != null) { node.getAirflowInterface().setAirflowState(StandardAirflowStateName.ON); notSelectedNodes.remove(node); addToSchedulingPlan(plan, task, chooseResourcesForExecution(node, task)); } } } turnOffIdleNodes(resourceManager.getNodes()); adjustOtherFans(notSelectedNodes); break; } return plan; } private Node chooseProvider(ClusterResourceManager resourceManager, TaskInterface task) { List nodes = filterNodes(resourceManager.getNodes(), task); for(Node node: nodes){ Integer id = Integer.parseInt(node.getName().split("_")[1]); if((id >= START_ID && id <= END_ID)){ if(node.getStatus() == ResourceStatus.UNAVAILABLE) { node.getPowerInterface().setPowerState(StandardPowerStateName.ON); } return node; } } return null; } private Map chooseResourcesForExecution( Node node, TaskInterface task) { Map map = new HashMap(); int cpuRequest; try { cpuRequest = Double.valueOf(task.getCpuCntRequest()).intValue(); } catch (NoSuchFieldException e) { cpuRequest = 0; } if (cpuRequest != 0) { List cores = node.getProcessors().get(0).getCores(); List choosenResources = new ArrayList(); for (int i = 0; i < cores.size(); i++) { if (cores.get(i).getStatus() == ResourceStatus.FREE) { //choosenResources.add(cores.get(i)); cpuRequest--; } } if (cpuRequest > 0) { //return null; } choosenResources.add(node.getProcessors().get(0)); ProcessingElements pe = new ProcessingElements(); pe.addAll(choosenResources); map.put(StandardResourceUnitName.PE, pe); return map; } return null; } private List filterNodes(List nodes, TaskInterface task){ List filteredNodes = new ArrayList(); for (Node node : nodes) { int cpuRequest; try { cpuRequest = Double.valueOf(task.getCpuCntRequest()).intValue(); } catch (NoSuchFieldException e) { cpuRequest = 0; } if (cpuRequest != 0) { List cores = node.getProcessors().get(0).getCores(); if (cores.size() < cpuRequest) { continue; } int freeCores = 0; for(Core core: cores){ if(core.getStatus() == ResourceStatus.FREE || core.getStatus() == ResourceStatus.UNAVAILABLE) freeCores++; } if(freeCores != cores.size()) continue; try { if(!getExecutiveness(createExecutivenessQuery(task, node))) continue; } catch (FileNotFoundException e) { continue; } catch (IOException e) { continue; } catch (MissingResourceException e){ continue; } filteredNodes.add(node); } } return filteredNodes; } private void turnOffIdleNodes(List nodes){ for(Node node : nodes){ Processor proc = node.getProcessors().get(0); int freeCores = 0; for(Core core: proc.getCores()){ if(core.getStatus() == ResourceStatus.FREE) freeCores++; } if(freeCores == proc.getCores().size()) node.getPowerInterface().setPowerState(StandardPowerStateName.OFF); } } private void adjustOtherFans(List nodes){ for(Node node : nodes){ if(node.getFreeProcessorsNumber() == node.getProcessorsNumber()){ node.getAirflowInterface().setAirflowState(StandardAirflowStateName.OFF); } else { node.getAirflowInterface().setAirflowState(StandardAirflowStateName.ON); } } } }