source: DCWoRMS/branches/coolemall/src/dcworms/schedframe/scheduling/Executable.java @ 1315

package dcworms.schedframe.scheduling;


import gridsim.GridSim;
import gridsim.dcworms.DCWormsTags;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.ArrayUtils;
import org.joda.time.DateTime;
import org.joda.time.DateTimeUtilsExt;
import org.joda.time.ReadableDuration;
import org.qcg.broker.schemas.resreqs.ResourceConsumptionProfileType;
import org.qcg.broker.schemas.resreqs.ResourceConsumptionType;
import org.qcg.broker.schemas.resreqs.StringParameterType;

import qcg.shared.constants.BrokerConstants;

import schedframe.resources.computing.ComputingResource;
import schedframe.resources.units.PEUnit;
import schedframe.resources.units.ProcessingElements;
import schedframe.resources.units.StandardResourceUnitName;
import schedframe.scheduling.ResourceHistoryItem;
import schedframe.scheduling.UsedResourcesList;
import schedframe.scheduling.WorkloadUnitHandler;
import schedframe.scheduling.manager.tasks.JobRegistryImpl;
import schedframe.scheduling.tasks.AbstractProcesses;
import schedframe.scheduling.tasks.AbstractProcessesGroup;
import schedframe.scheduling.tasks.Task;
import schedframe.scheduling.tasks.phases.ResourceConsumption;
import schedframe.scheduling.tasks.phases.ResourceConsumptionProfile;
import schedframe.scheduling.tasks.requirements.ResourceParameterName;

/**
 * 
 * @author Marcin Krystek
 *
 */
public class Executable implements ExecTask{

        protected Task task;
        protected String processesSetId;
        
        protected int status; 
        protected Map<ResourceParameterName, Object> specificResources;
        
        protected String reservationId;
        protected double completionPercentage;
        
        protected int estimatedDuration;
        //TO DO remove and benefit from visitedResources
        protected String schedName;
        protected UsedResourcesList usedResources;
        //TO DO consider removing
        protected List<String> visitedResources;

        protected double submissionTime;
    protected double arrivalTime;
        protected double execStartTime ;
    protected double totalCompletionTime; 
        protected double finishTime; 
        
        private ResourceConsumptionProfile resourceConsumptionProfile;
        
        public Executable(Task t){
                this.task = t;
                this.status = DCWormsTags.CREATED;
                
                this.usedResources = new UsedResourcesList();
                this.visitedResources = new ArrayList<String>();
                init();
        }
        
        public Executable(Task t, AbstractProcesses procesesSet){
                this.task = t;
                this.status = DCWormsTags.CREATED;
                this.processesSetId = procesesSet.getId(); 
                
                this.usedResources = new UsedResourcesList();
                this.visitedResources = new ArrayList<String>();
                init();
        }
        
        protected void init() {
                double currentTime = DateTimeUtilsExt.currentTimeMillis() / 1000;
                this.submissionTime =  currentTime;
        this.totalCompletionTime = 0.0;
        preparePhases(null);
         }
        
        public int getUserId() {
                return task.getSenderId();
        }
        public String getUserDN() {
                return task.getUserDN();
        }

        public String getJobId() {
                return task.getJobId();
        }
        
        public String getTaskId(){
                return this.task.getId();
        }
        
        public String getId() {
                if(processesSetId == null)
                        return task.getId();
                else 
                        return task.getId() + "_" + processesSetId;
        }
        
        public String getProcessesId(){
                return this.processesSetId;
        }
        
    public int getUniqueId(){
        if(processesSetId == null){
                return (task.getJobId() + "_" + task.getId()).hashCode();
        } else {
                return (task.getJobId() + "_" + task.getId() + "_" + processesSetId).hashCode();
        }
    }
    
        public List<AbstractProcesses> getProcesses() {
                return task.getProcesses();
        }

        public List<AbstractProcesses> getProcesses(
                        AbstractProcessesGroup processGroup) {
                return task.getProcesses(processGroup);
        }

        public List<AbstractProcessesGroup> getProcessesGroups() {
                return task.getProcessesGroups();
        }

        public org.qcg.broker.schemas.resreqs.Task getDescription() {
                return task.getDescription();
        }

        public String getDocument() throws Exception {
                return task.getDocument();
        }
        
        public long getLength() {
                return task.getLength();
        }

        public int getStatus() {
                return status;
        }
        
        public boolean isFinished()
    {
                return task.isFinished();
    }
        
        public void setStatus(int newStatus) throws Exception {
                int prevStatus = status;
            
        if (status == newStatus) {
            return;
        }

        if (newStatus < DCWormsTags.CREATED || newStatus > DCWormsTags.NEW_EXEC_PHASE) {
            throw new Exception("Executable.setStatuts() : Error - " +
                    "Invalid integer range for Execiutable status.");
        }

        status = newStatus;
        
        if(status == DCWormsTags.INEXEC){
                loadResourceConsumptionProfile();
                task.setStatus((int) BrokerConstants.TASK_STATUS_RUNNING);
        } else if(status == DCWormsTags.QUEUED){
                task.setStatus((int) BrokerConstants.TASK_STATUS_QUEUED);
        }
        
        double currentTime = DateTimeUtilsExt.currentTimeMillis() / 1000; // time in seconds 
    
                
                if (newStatus == DCWormsTags.SUCCESS || newStatus == DCWormsTags.CANCELED) {
            finishTime = DateTimeUtilsExt.currentTimeMillis() / 1000;
        }
                
                if(newStatus == DCWormsTags.SUBMITTED){
                         arrivalTime = GridSim.clock();
                }

        if (prevStatus == DCWormsTags.INEXEC) {
            if (status == DCWormsTags.CANCELED || status == DCWormsTags.PAUSED ||
                status == DCWormsTags.SUCCESS) {
                totalCompletionTime += (currentTime -  execStartTime);
            }
        }

        if (prevStatus == DCWormsTags.RESUMED && status == DCWormsTags.SUCCESS) {
            totalCompletionTime += (currentTime -  execStartTime);
        }

        if (status == DCWormsTags.INEXEC ||
            (prevStatus == DCWormsTags.PAUSED && status == DCWormsTags.RESUMED) ) {
                execStartTime = currentTime;
                
                ProcessingElements pes = (ProcessingElements) getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
                for (ComputingResource resource : pes) {

                                LinkedList<ComputingResource> toExamine = new LinkedList<ComputingResource>();
                                toExamine.push(resource);

                                while (!toExamine.isEmpty()) {
                                        ComputingResource compResource = toExamine.pop();
                                        List<ComputingResource> resources = compResource.getChildren();
                                        int numberOfRes = resources.size();
                                        for (int i = 0; i < numberOfRes; i++) {
                                                ComputingResource resourceChild = resources.get(i);
                                                trackResource(resourceChild.getFullName());
                                                toExamine.addLast(resourceChild);
                                        }
                                }

                        
                        trackResource(resource.getFullName());
                        
                        ComputingResource parent = resource.getParent();
                                List<String> visitedResource = getVisitedResources();
                                String [] visitedResourcesArray = visitedResource.toArray(new String[visitedResource.size()]);
                        while (parent != null && !ArrayUtils.contains(visitedResourcesArray, parent.getFullName())) {
                                trackResource(parent.getFullName());
                                parent = parent.getParent();
                        }
                }
        }
        
        if(status == DCWormsTags.NEW_EXEC_PHASE){
                if(prevStatus == DCWormsTags.INEXEC){
                        status = DCWormsTags.INEXEC;
                        setCompletionPercentage(0);
                        resourceConsumptionProfile.setCurrentPhase(resourceConsumptionProfile.getCurrentPhase() + 1);
                        
                        DateTime currentDateTime = new DateTime();
                        
                        if(getUsedResources().getLast().getTimeStamp().getMillis() == currentDateTime.getMillis()){
                                return;
                        }
                        ResourceHistoryItem resHistItem = new ResourceHistoryItem(getUsedResources().getLast().getResourceUnits(), currentDateTime);
                        resHistItem.setCompletionPercentage(0);
                        addUsedResources(resHistItem);
                }
        }
        }
        
        public void addSpecificResource(ResourceParameterName resourceName, Object value){
                if(this.specificResources == null)
                        this.specificResources = new HashMap<ResourceParameterName, Object>();
                this.specificResources.put(resourceName, value);
        }

        public boolean expectSpecificResource(ResourceParameterName resourceName){
                if(this.specificResources == null)
                        return false;
                return this.specificResources.containsKey(resourceName);
        }
        
        public Object getExpectedSpecificResource(ResourceParameterName resourceName){
                if(this.specificResources == null)
                        return null;
                
                return this.specificResources.get(resourceName);
        }
        
        public void setReservationId(String reservationId){
                this.reservationId = reservationId;
        }
        
        public boolean requireReservation(){
                return (reservationId != null);
        }
        
        public String getReservationId(){
                return this.reservationId;
        }
        
        public double getCompletionPercentage() {
                return completionPercentage;
        }

        public void setCompletionPercentage(double completionPercentage) {
                this.completionPercentage = completionPercentage;
        }

        public void addUsedResources(ResourceHistoryItem usedResources){
                this.usedResources.add(usedResources);
        }
        
        public UsedResourcesList getUsedResources(){
                return this.usedResources;
        }
        
    public void setSchedulerName(int resourceId){
        this.schedName = GridSim.getEntityName(resourceId);
    }

    public String getSchedulerName(){
        return schedName;
    }
        
        public int getEstimatedDuration(){
                return this.estimatedDuration;
        }
        
        public void setEstimatedDuration(int value){
                this.estimatedDuration = value;
        }

        public boolean isRegistered() {
                return task.isRegistered();
        }

        public void register(JobRegistryImpl jobRegistry) {
                task.register(jobRegistry);
        }
        
        public void trackResource(String resName){
                visitedResources.add(resName);
        }
        
        public List<String> getVisitedResources(){
                return visitedResources;
        }
        
        public ReadableDuration getExpectedDuration() throws NoSuchFieldException {
                return task.getExpectedDuration();
        }

        public double getParameterDoubleValue(ResourceParameterName parameterName)
                        throws NoSuchFieldException, IllegalArgumentException {
                return task.getParameterDoubleValue(parameterName);
        }

        public String getParameterStringValue(ResourceParameterName parameterName)
                        throws NoSuchFieldException, IllegalArgumentException {
                return task.getParameterStringValue(parameterName);
        }
        
        public double getCpuCntRequest() throws NoSuchFieldException{
                return getParameterDoubleValue(ResourceParameterName.CPUCOUNT);
        }
        
        public double getMemoryRequest() throws NoSuchFieldException{
                return getParameterDoubleValue(ResourceParameterName.MEMORY);
        }
        
        public DateTime getExecutionEndTime() throws NoSuchFieldException {
                return task.getExecutionEndTime();
        }

        public DateTime getExecutionStartTime() throws NoSuchFieldException {
                return task.getExecutionStartTime();
        }

        public DateTime getSubmissionTimeToBroker() {
                return task.getSubmissionTimeToBroker();
        }

        public long getWorkloadLogWaitTime() {
                return task.getWorkloadLogWaitTime();
        }

    public double getExecStartTime() {
        return execStartTime;
    }

    public double getFinishTime() {
        return finishTime;
    }

    public double getSubmissionTime() {
        return submissionTime;
    }
    
    public double getWaitingTime() {
        return execStartTime - submissionTime;
    }
    
    public void finalizeExecutable(){
                try {
                        setStatus(DCWormsTags.SUCCESS);
                } catch (Exception e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
    }
    
        public void accept(WorkloadUnitHandler wuh) {
                wuh.handleExecutable(this);
        }
    
        /*public boolean equals(Object obj){
                if(obj instanceof Executable){
                        Executable t = (Executable) obj;
                        return getId().equals(t.getId()) && getJobId().equals(t.getJobId());
                }
                return false;
        }*/
        
        public ResourceConsumptionProfile getResourceConsumptionProfile(){
                return resourceConsumptionProfile;
        }

        public ResourceConsumption getCurrentResourceConsumption(){
                return resourceConsumptionProfile.getCurrentResourceConsumption();
        }
        
        public String getApplicationName(){
                return task.getApplicationName();
        }
        
        private void preparePhases(String resourceType) {
                LinkedList<ResourceConsumption> resourceConsumptionList = new LinkedList<ResourceConsumption>();
                
                long usefulWork = -1;
                
                if(task.getDescription().getExecution() == null || task.getDescription().getExecution().getResourceConsumptionProfile() == null
                                || task.getDescription().getExecution().getResourceConsumptionProfile().length == 0 ){
                        ResourceConsumption resConsumption = null;
                        try {
                                resConsumption = new ResourceConsumption(getLength(), task.getComputingResourceRequirements());
                        } catch (NoSuchFieldException e) {
                                // TODO Auto-generated catch block
                                e.printStackTrace();
                        }
                        resourceConsumptionList.add(resConsumption);
                }else{
                        boolean found = false;
                        if(resourceType != null){
                                for(ResourceConsumptionProfileType resConsumptioProfile: task.getDescription().getExecution().getResourceConsumptionProfile()){
                                        if(resConsumptioProfile.getType() != null && resConsumptioProfile.getType().equals(resourceType)){
                                                for(ResourceConsumptionType resConsumption: resConsumptioProfile.getResourceConsumption()){
                                                        ResourceConsumption resourceConsumption = new ResourceConsumption(resConsumption);
                                                        resourceConsumptionList.add(resourceConsumption);
                                                }
                                                for(StringParameterType prop: resConsumptioProfile.getProperties()){
                                                        if(prop.getName().equals("usefulWork")){
                                                                usefulWork = Double.valueOf(prop.getContent()).longValue();
                                                                break;
                                                        }
                                                }
                                                found = true;
                                                break;
                                        }
                                }       
                        }                       
                        if(!found){
                                for(ResourceConsumptionType resConsumption: task.getDescription().getExecution().getResourceConsumptionProfile()[0].getResourceConsumption()){
                                        ResourceConsumption resourceConsumption = new ResourceConsumption(resConsumption);
                                        resourceConsumptionList.add(resourceConsumption);
                                }
                                
                                for(StringParameterType prop: task.getDescription().getExecution().getResourceConsumptionProfile()[0].getProperties()){
                                        if(prop.getName().equals("usefulWork")){
                                                usefulWork = Double.valueOf(prop.getContent()).longValue();
                                                break;
                                        }
                                }
                        }
                }
                
                usefulWork = (usefulWork != -1) ? usefulWork : this.getLength();
                this.resourceConsumptionProfile = new ResourceConsumptionProfile(resourceConsumptionList);
                this.resourceConsumptionProfile.setUsefulWork(usefulWork);

        }

        private boolean loadResourceConsumptionProfile(){

                PEUnit peUnit = (PEUnit)getUsedResources().getLast().getResourceUnits()
                                .get(StandardResourceUnitName.PE);
                if(peUnit instanceof ProcessingElements){
                        ProcessingElements pes = (ProcessingElements) peUnit;
                        for (ComputingResource resource : pes) {
                                while(resource != null && resource.getResourceCharacteristic().getParameters().get("product") == null){
                                        resource = resource.getParent();
                                }
                                if(resource != null) {
                                        String productName = resource.getResourceCharacteristic().getParameters().get("product").get(0).getContent();
                                        if(productName.equals("Fusion G - T40N"))
                                                preparePhases("Fusion G - T40N");
                                        else if(productName.equals("Atom - D510"))
                                                preparePhases("Atom - D510");
                                        else if(productName.equals("Atom - N2600"))
                                                preparePhases("Atom - N2600");
                                        else if(productName.equals("Core i7 - 2715QE"))
                                                preparePhases("Core i7 - 2715QE");
                                        else if(productName.equals("Core i7 - 3615QE"))
                                                preparePhases("Core i7 - 3615QE");
                                        else if(productName.equals("Xeon E5-2630"))
                                                preparePhases("Xeon E5-2630");
                                        else if(productName.equals("Xeon E5-2603"))
                                                preparePhases("Xeon E5-2603");
                                        else if(productName.equals("Xeon L5310"))
                                                preparePhases("Xeon L5310");
                                }
                                return true;
                        }
                }
                return false;
                
        }
}