package test.article.recs.plugins.scheduling; import gridsim.dcworms.DCWormsTags; import java.io.FileInputStream; 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 java.util.PropertyResourceBundle; import java.util.ResourceBundle; import schedframe.events.scheduling.SchedulingEvent; import schedframe.resources.ResourceStatus; import schedframe.resources.computing.ComputingNode; import schedframe.resources.computing.ComputingResource; import schedframe.resources.computing.Core; import schedframe.resources.computing.Processor; 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.grid.ModuleList; import schedframe.scheduling.queue.TaskQueue; import schedframe.scheduling.queue.TaskQueueList; import schedframe.scheduling.tasks.TaskInterface; import test.article.recs.utils.AppType; public class RecsExclusivenessEnOptSP extends RecsSP { private static String TIME_DATA_FILE_NAME = "src/test/article/recs/data/time_data.properties"; private static String POWER_DATA_FILE_NAME = "src/test/article/recs/data/power_data.properties"; private ResourceBundle powBundle; private ResourceBundle timeBundle; 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); // check all tasks in queue for (int i = 0; i < q.size(); i++) { TaskInterface task = q.get(i); initApplicationType(task); // if status of the tasks in READY if (task.getStatus() == DCWormsTags.READY) { Map choosenResources = chooseResourcesForExecution(resourceManager, task); if (choosenResources != null) { addToSchedulingPlan(plan, task, choosenResources); } } } break; } return plan; } private Map chooseResourcesForExecution( ClusterResourceManager resourceManager, TaskInterface task) { Map map; List nodes = resourceManager.getComputingNodes(); Collections.sort(nodes, new EnergyComparator(task)); for (ComputingNode node : nodes) { int cpuRequest; try { cpuRequest = Double.valueOf(task.getCpuCntRequest()).intValue(); } catch (NoSuchFieldException e) { cpuRequest = 0; } if (cpuRequest != 0) { /*Properties properties = new Properties(); properties.setProperty("type", StandardResourceType.Core.getName()); properties.setProperty("status", ResourceStatus.FREE.toString()); */ List cores = node.getProcessors().get(0).getCores(); if (cores.size() < cpuRequest/* || node.getProcessors().get(0).filterDescendants(properties).size() != cores.size()*/) { continue; } int freeCores = 0; for(Core core: cores){ if(core.getStatus() == ResourceStatus.FREE) freeCores++; } if(freeCores != cores.size()) continue; List choosenResources = new ArrayList(); for (int i = 0; i < cores.size() && cpuRequest > 0; i++) { if (cores.get(i).getStatus() == ResourceStatus.FREE) { choosenResources.add(cores.get(i)); cpuRequest--; } } if (cpuRequest > 0) { continue; } map = new HashMap(); ProcessingElements pe = new ProcessingElements(); pe.addAll(choosenResources); map.put(StandardResourceUnitName.PE, pe); return map; } } return null; } protected String createQuery(TaskInterface task, ComputingNode node) { String query; query = getApplicationType(task) + "." + getNodeCategory(node) + "." + getFrequency(node) + "." + getCoreCnt(task); return query; } private String getApplicationType(TaskInterface task){ AppType appType = taskToApp.getAppType(task); return appType.toString(); } private String getNodeCategory(ComputingNode node){ return node.getCategory(); } private int getFrequency(ComputingNode node){ Processor proc = (Processor) node.getProcessors().get(0); double freq = proc.getPowerInterface().getFrequency(); return Double.valueOf(freq).intValue(); } private int getCoreCnt(TaskInterface task){ double cpuReq; try { cpuReq = task.getCpuCntRequest(); } catch (NoSuchFieldException e) { cpuReq = 1; } return Double.valueOf(cpuReq).intValue(); } protected double getMeasuredPower(String query) throws FileNotFoundException, IOException{ ResourceBundle powBundle = getPowBundle(); return Double.valueOf(powBundle.getString(query)).doubleValue(); } private ResourceBundle getPowBundle() throws FileNotFoundException, IOException{ if(powBundle == null){ powBundle = new PropertyResourceBundle(new FileInputStream(POWER_DATA_FILE_NAME)); } return powBundle; } protected double getMeasuredTime(String query) throws FileNotFoundException, IOException{ ResourceBundle timeBundle = getTimeBundle(); return Double.valueOf(timeBundle.getString(query)).doubleValue(); } private ResourceBundle getTimeBundle() throws FileNotFoundException, IOException{ if(timeBundle == null){ timeBundle = new PropertyResourceBundle(new FileInputStream(TIME_DATA_FILE_NAME)); } return timeBundle; } class EnergyComparator implements Comparator{ private TaskInterface task; public EnergyComparator(TaskInterface task){ this.task = task; } public int compare(ComputingNode node1, ComputingNode node2){ double node1EU = Double.MAX_VALUE; double node2EU = Double.MAX_VALUE; try { node1EU = getMeasuredTime(createQuery(task, node1)) * getMeasuredPower(createQuery(task, node1)); node2EU = getMeasuredTime(createQuery(task, node2)) * getMeasuredPower(createQuery(task, node2)); } catch (FileNotFoundException e) { } catch (IOException e) { } catch (MissingResourceException e){ } if(node1EU > node2EU) return 1; else if (node1EU < node2EU) return -1; else return 0; } } }