package experiments.e2dc2013.recs.plugins.scheduling.exp2; 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.Collections; import java.util.Comparator; 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.profiles.energy.airthroughput.StandardAirflowStateName; 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.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 RecsIn2OutSP extends RecsSP { private static int START_ID = 0; 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)); } } } adjustOtherFans(notSelectedNodes); break; } return plan; } private Node chooseProvider(ClusterResourceManager resourceManager, TaskInterface task) { List nodes = filterNodes(resourceManager.getNodes(), task); Collections.sort(nodes, new ResourceIdComparator()); for(Node node: nodes){ Integer id = Integer.parseInt(node.getName().split("_")[1]); if((id >= START_ID && id <= END_ID)){ 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) 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 adjustOtherFans(List nodes){ for(Node node : nodes){ if(node.getFreeProcessorsNumber() == node.getProcessorsNumber()){ //node.getAirThroughputInterface().setAirThroughputState(StandardAirThroughputStateName.FAN_OFF); } else { node.getAirflowInterface().setAirflowState(StandardAirflowStateName.ON); } } } class ResourceIdComparator implements Comparator { public int compare(Node node1, Node node2) { Integer id1 = Integer.parseInt(node1.getName().split("_")[1]); Integer id2 = Integer.parseInt(node2.getName().split("_")[1]); if(id1 >=9 && id1 <=17 && id2 >=0 && id2<= 8) return -1; else if (id2 >=9 && id2 <=17 && id1 >=0 && id1<= 8) return 1; else if (id1 >=9 && id1 <=17 && id2 >=9 && id2<= 17){ if(id1 > id2){ return 1; } else if (id1 < id2) return -1; else return 0; } else if (id1 >=0 && id1 <=8 && id2 >=0 && id2<= 8){ if(id1 > id2){ return 1; } else if (id1 < id2) return -1; else return 0; }else return 0; } } }