source: DCWoRMS/branches/coolemall/src/test/testSOP/algorithms/multicloud/MetaSched.java @ 1606

package test.testSOP.algorithms.multicloud;

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import schedframe.events.scheduling.SchedulingEvent;
import schedframe.resources.StandardResourceType;
import schedframe.resources.computing.ComputingResource;
import schedframe.resources.computing.Node;
import schedframe.resources.computing.Processor;
import schedframe.scheduling.Scheduler;
import schedframe.scheduling.SchedulerDescription;
import schedframe.scheduling.manager.resources.ClusterResourceManager;
import schedframe.scheduling.manager.resources.ManagedComputingResources;
import schedframe.scheduling.manager.resources.ResourceManager;
import schedframe.scheduling.manager.tasks.JobRegistry;
import schedframe.scheduling.plan.SchedulingPlanInterface;
import schedframe.scheduling.plan.impl.Allocation;
import schedframe.scheduling.plan.impl.ScheduledTask;
import schedframe.scheduling.plan.impl.SchedulingPlan;
import schedframe.scheduling.plugin.Module;
import schedframe.scheduling.plugin.ModuleList;
import schedframe.scheduling.plugin.grid.ResourceDiscovery;
import schedframe.scheduling.queue.TaskQueue;
import schedframe.scheduling.queue.TaskQueueList;
import schedframe.scheduling.tasks.TaskInterface;
import schedframe.scheduling.tasks.WorkloadUnit;
import example.globalplugin.BaseGlobalPlugin;


public class MetaSched  extends BaseGlobalPlugin {
        Log log = LogFactory.getLog(MetaSched.class); 
        private LinkedList<String> lastUsedResources = new LinkedList<String>();
        
        private static ArrayList<String> cacheTaskId = new ArrayList<String>();
        private static ArrayList<String> cacheChosenCluster = new ArrayList<String>();
        private static ArrayList<Integer> cacheNbTries = new ArrayList<Integer>();
        
        private static int MAX_TRIES = 3;
        public SchedulingPlanInterface<?> schedule(SchedulingEvent event,
                        TaskQueueList queues, 
                        JobRegistry jobRegistry,
                        ResourceManager resourceManager, ModuleList modules)  {
        
                
                ResourceDiscovery resources = null;
                for(int i = 0; i < modules.size(); i++){
                        Module m = modules.get(i);
                        switch(m.getType()){
                                case RESOURCE_DISCOVERY: resources = (ResourceDiscovery) m;
                                        break;
                        }
                }
                
//              switch(event.getType()) {
//                      case TIMER:
//                              log.debug("Metasched called by timer");
//                              incrementNbTriesCache();
//                              removeOldCacheEntries();
//                              break;
//                      default:
//                              log.debug("Defau");
//                              break;
//              }
                SchedulingPlan plan = new SchedulingPlan();
                
                TaskQueue q = queues.get(0);
//              log.debug("Queues size:"+queues.size());
                int size = q.size();
                
                // order of the resources on this list is not determined
                List<SchedulerDescription> availableResources = resources.getResources();
                
//              log.debug("Available resource: "+availableResources.size());
//              log.debug(availableResources.get(0).getId());
//              log.debug(resourceManager.getSchedulers().get(0).getFullName());
//              log.debug(resourceManager.getSchedulers().get(0).getCompResources().get(0).getLoadInterface().getRecentUtilization().getValue());
//              log.debug(resourceManager.getSchedulers().get(0).getCompResources().get(0).getFullName());
//              log.debug(availableResources.get(0).getProvider().getPropertyKeys());
//              getY(resourceManager.getSchedulers().get(0).getCompResources());
                
//              log.debug("0 : "+availableResources.get(0).getProvider().getProviderId());
//              log.debug("1 : "+availableResources.get(1).getProvider().getProviderId());
                
                
                //For each task in the queue
                for(int i = 0; i < size; i++) {
                        WorkloadUnit job = q.remove(0);
                        TaskInterface<?> task = (TaskInterface<?>)job;
                        Scheduler sched = chooseCloud(resourceManager.getSchedulers(), task);
                        incrementNbTriesCache();
                        removeOldCacheEntries();
                        //Skip task if we dont have a cloud available
                        if (sched == null)
                                continue;
                        cacheTaskId.add(task.getJobId());
                        cacheChosenCluster.add(sched.getFullName());
                        cacheNbTries.add(1);
//                      log.debug("Choosen cloud : "+sched.getFullName());
                        //Round robin
                        SchedulerDescription sd = availableResources.get(availableResources.size() - 1);
                        for(SchedulerDescription schedDesc: availableResources){
                                if(!lastUsedResources.contains(schedDesc.getProvider().getProviderId())){
                                        if(lastUsedResources.size() + 1 >= availableResources.size()){
                                                lastUsedResources.poll();
                                        }
                                        lastUsedResources.add(schedDesc.getProvider().getProviderId());
                                        sd = schedDesc;
                                        break;
                                }
                        }
                        
                        Allocation allocation = new Allocation();
                        allocation.setProcessesCount(1);
                        allocation.setProviderName(sd.getProvider().getProviderId());
                        ScheduledTask scheduledTask = new ScheduledTask(task);
                        scheduledTask.setTaskId(task.getId());
                        scheduledTask.setJobId(task.getJobId());
                        scheduledTask.addAllocation(allocation);                
                        
                        plan.addTask(scheduledTask);
                        
                        //Metasched
//                      Allocation allocation = new Allocation();
//                      allocation.setProcessesCount(1);
//                      allocation.setProviderName(sched.get_name());
//                      ScheduledTask scheduledTask = new ScheduledTask(task);
//                      scheduledTask.setTaskId(task.getId());
//                      scheduledTask.setJobId(task.getJobId());
//                      scheduledTask.addAllocation(allocation);                
//                      
//                      plan.addTask(scheduledTask);
                }
                return plan;
        }

        private void incrementNbTriesCache() {
                for (int i = 0 ; i < cacheNbTries.size(); i++) {
                        cacheNbTries.set(i, cacheNbTries.get(i)+1);
                }
        }
        
        private void removeOldCacheEntries() {
                for (int i = 0 ; i < cacheNbTries.size(); i++) {
                        if (cacheNbTries.get(i) > MAX_TRIES) {
                                log.debug("Cache entry :"+cacheTaskId.get(i)+"/"+cacheChosenCluster.get(i)+"/"+cacheNbTries.get(i)+" is too old");
                                cacheNbTries.remove(i);
                                cacheTaskId.remove(i);
                                cacheChosenCluster.remove(i);
                        }
                }
        }

        // return std::max((float)c.cpu_number_hosts_le_50p/(float)c.host_number, (float)c.mem_number_hosts_le_50p/(float)c.host_number);
        private float getX(ManagedComputingResources compRes) { 
                float hostNumber = compRes.size();
//              log.debug("Host number: "+hostNumber);
                float cpu_number_hosts_le_50p=0;
                float mem_number_hosts_le_50p=0;
                
                for (ComputingResource cr : compRes) {
                        Node node = (Node)cr;
//                      log.debug(node.getFullName());
//                      log.debug("Free Processors = "+node.getFreeProcessorsNumber()+"/"+node.getProcessorsNumber());
//                      log.debug(((Processor)node.getChildren().get(0)).);
                        if ((float)node.getFreeProcessorsNumber() >= 0.5 * (float)node.getProcessorsNumber()) {
                                cpu_number_hosts_le_50p++;
                        }
                        try {
                                if ((float)node.getFreeMemory() >= 0.5 *(float)node.getTotalMemory()) {
                                        mem_number_hosts_le_50p++;
                                }
                        } catch (NoSuchFieldException e) {
                                // TODO Auto-generated catch block
                                e.printStackTrace();
                        }
                }
                float ret = Math.max(cpu_number_hosts_le_50p/hostNumber,mem_number_hosts_le_50p/hostNumber);
//              log.debug("X=max("+cpu_number_hosts_le_50p+"/"+hostNumber+","+mem_number_hosts_le_50p+"/"+hostNumber+")="+ret);
                return ret;
        }
        
        //return (float)(c.highest_mean_resource_le_50_p / 100.0);
        private float getY(ManagedComputingResources compRes) {
                float mean_cpu_le_50p = 0;
                float nbCPU=0;
                float mean_mem_le_50p = 0;
                float nbMEM=0;
                for (ComputingResource cr : compRes) {
                        Node node = (Node)cr;
                        //node is loaded < 50%
                        if ((float)node.getFreeProcessorsNumber() >= 0.5 * (float)node.getProcessorsNumber()) {
                                mean_cpu_le_50p+=((float)node.getProcessorsNumber()-(float)node.getFreeProcessorsNumber())/(float)node.getProcessorsNumber();
                                nbCPU++;
//                              log.debug("CPU Inf 50p");
//                              log.debug("(float)node.getProcessorsNumber() = "+(float)node.getProcessorsNumber());
//                              log.debug("(float)node.getFreeProcessorsNumber() = "+(float)node.getFreeProcessorsNumber());
                        }
                        try {
                                if ((float)node.getFreeMemory() >= 0.5 *(float)node.getTotalMemory()) {
                                        mean_mem_le_50p+=((float)node.getTotalMemory() - (float)node.getFreeMemory()) / (float)node.getTotalMemory();
                                        nbMEM++;
//                                      log.debug(((float)node.getTotalMemory() - (float)node.getFreeMemory()));
//                                      log.debug((float)node.getFreeMemory());
//                                      log.debug("mem Inf 50p");
                                }
                        } catch (NoSuchFieldException e) {
                                // TODO Auto-generated catch block
                                e.printStackTrace();
                        }
                }
                if (nbCPU != 0) {
//                      log.debug("cpu: "+mean_cpu_le_50p+"/"+nbCPU);
                        mean_cpu_le_50p = mean_cpu_le_50p / nbCPU;                              
                }
                else {
                        mean_cpu_le_50p = (float) 0.5;
                }
                if (nbMEM != 0) {
//                      log.debug("mem: "+mean_mem_le_50p+"/"+nbMEM);
                        mean_mem_le_50p = mean_mem_le_50p / nbMEM;
                }
                else {
                        mean_mem_le_50p = (float) 0.5;
                }
//              log.debug("Y = max("+mean_cpu_le_50p+","+mean_mem_le_50p+")");
                return Math.max(mean_cpu_le_50p, mean_mem_le_50p);
        }
        
        private Scheduler chooseCloud(List<Scheduler> schedulers, TaskInterface<?> task) {
                Scheduler chosenCloud = null;
                float value = Float.NEGATIVE_INFINITY;
                Collections.sort(schedulers, new ComparatorRankingMetasched());
                for (int i = 0 ; i < schedulers.size() ; i++) {
                        //If scheduler has already been chosen, continue
                        if (cloudAlreadyChosen(schedulers.get(i), task)) {
                                log.debug("Task "+task.getJobId()+" already chose cloud "+schedulers.get(i).getFullName());
                                continue;
                        }
                                
//                      log.debug("Cloud : "+schedulers.get(i).get_name());
                        float x = getX(schedulers.get(i).getCompResources());
                        float y = getY(schedulers.get(i).getCompResources());
                        float z = i;
                        float newValue = x * ((float)schedulers.get(i).getCompResources().size() / 2)
                                                - y * (float)schedulers.get(i).getCompResources().size()
                                                - z;
                        if (newValue > value) {
                                chosenCloud = schedulers.get(i);
                                value = newValue;
                        }
                }
                if (chosenCloud == null) {
                        log.debug("No cloud was found !");
                }
                return chosenCloud;
        }


        private boolean cloudAlreadyChosen(Scheduler sched, TaskInterface<?> task) {
                for (int i = 0 ; i < cacheTaskId.size() ; i++) {
                        if (task.getJobId().equals(cacheTaskId.get(i)) && sched.getFullName().equals(cacheChosenCluster.get(i))) {
                                return true;
                        }
                }
                return false;
        }
}