Changeset 883 for DCWoRMS/branches/coolemall/src/simulator/workload

Timestamp:

02/26/13 08:41:50 (12 years ago)

Author:

wojtekp

Message:

 

Location:

DCWoRMS/branches/coolemall

Files:

• . (copied) (copied from DCWoRMS/trunk)
• src/simulator/workload/WorkloadLoader.java (modified) (10 diffs)
• src/simulator/workload/reader/archive/AbstractWAParser.java (modified) (4 diffs)
• src/simulator/workload/reader/archive/QcgWAJobReader.java (modified) (5 diffs)
• src/simulator/workload/reader/archive/WAFields.java (copied) (copied from DCWoRMS/trunk/src/simulator/workload/reader/archive/WAFields.java)
• src/simulator/workload/reader/archive/WAParser.java (modified) (1 diff)
• src/simulator/workload/reader/archive/swf/QcgSWFJobReader.java (modified) (9 diffs)
• src/simulator/workload/writer/swf/QcgSWFJobWriter.java (modified) (2 diffs)

DCWoRMS/branches/coolemall/src/simulator/workload/WorkloadLoader.java

@@ -2 +2 @@
 
 import org.qcg.broker.schemas.jobdesc.QcgJob;
+import org.qcg.broker.schemas.jobdesc.ResourceConsumptionProfileType;
+import org.qcg.broker.schemas.jobdesc.Task;
 
 import java.io.IOException;
@@ -27 +29 @@
 
 
+import schedframe.scheduling.tasks.Job;
 import simulator.utils.XsltTransformations;
 import simulator.workload.exceptons.NoSuchCommentException;
 import simulator.workload.reader.archive.AbstractWAParser;
+import simulator.workload.reader.archive.QcgWAJobReader;
 import simulator.workload.reader.archive.WAFields;
 import simulator.workload.reader.archive.WAReader;
@@ -58 +62 @@
         
         protected Map<String, JobDescription> jobGridletsMap;
-        
+        protected Map<String, JobDescription> jobProfilesMap;
+
         public WorkloadLoader(XMLJobReader<QcgJob> xmlReader, WAReader<QcgJob> swfReader){
                 if(swfReader == null){
-                        throw new RuntimeException("Swf reader is required to build proper gridlerts.");
+                        throw new RuntimeException("Swf reader is required to build proper tasks.");
                 }
                 this.xmlJobReader = xmlReader;
@@ -68 +73 @@
                 this.generatedTasksCnt = 0;
                 this.jobGridletsMap = new TreeMap<String, JobDescription>();
+                this.jobProfilesMap = new TreeMap<String, JobDescription>();
                 try {
                         this.xsltTransformation = new XsltTransformations();
@@ -90 +96 @@
                 if(jobDescription.size() == 0){
                         if(log.isWarnEnabled())
-                                log.warn("Omiting job gridlet creation for job "+jobDescription.getDescription().getAppId()+". This job contains no tasks.");
+                                log.warn("Omitting tasks creation for job "+jobDescription.getDescription().getAppId()+". This job contains no tasks.");
                         return null;
                 }
@@ -103 +109 @@
                         
                         String waTaskDesc[] = localWAParser.readTask(jobId, taskDesc.getTaskId());
+                        
+                        try{
+                                Integer.parseInt(taskDesc.getTaskId());
+                        } catch (NumberFormatException e){
+                                return jobDescription;
+                        }
                         if(waTaskDesc != null) {
                                 
@@ -125 +137 @@
                                         timeValue = timeValue * puSpeed * allocProcesors;
                                         taskDesc.setTaskLength(timeValue);
+                                        
+                                        //if(taskDesc.getDescription().getExecution()!= n
 
                                 } else if(parserType == 1){
@@ -177 +191 @@
                         while((jobDesc = this.xmlJobReader.readRaw()) != null){
                                 jobDesc = this.xsltTransformation.extendJobDescription(jobDesc);
+                                int prevJobCnt = generatedJobsCnt;
                                 job = createJobDescription(jobDesc, puSpeed);
-                                if(job != null)
+                                if(job != null && prevJobCnt != generatedJobsCnt)
                                         this.jobGridletsMap.put(job.getJobId(), job);
+                                else if (prevJobCnt == generatedJobsCnt){
+                                        Task taskPattern = job.getDescription().getTask(0);
+                                        this.jobProfilesMap.put(taskPattern.getTaskId(), job);
+                                }
                         }
                 } else {                                                // use swf job reader. Job created by this reader does not require xslt transformation
@@ -188 +207 @@
                         }
                 }
+                
+                if(generatedJobsCnt == 0){
+                        while((jobDesc = this.waReader.readRaw()) != null){
+                                QcgWAJobReader qcgReader = (QcgWAJobReader)waReader;
+                                jobDesc = qcgReader.mergeSwfAndXmlProfile(jobProfilesMap, jobDesc);
+                                job = createJobDescription(jobDesc, puSpeed);
+                                if(job != null)
+                                        this.jobGridletsMap.put(job.getJobId(), job);
+                        }
+                }
 
                 this.localWAParser.close();
@@ -249 +278 @@
                 return simStartTimeDefined;
         }
+
+        protected JobDescription createJobDescription(String jobDesc, long puSpeed, Map<String, JobDescription> jobProfilesMap) throws IOException{
+                
+                JobDescription jobDescription = null;
+                try {
+                        jobDescription = this.xsltTransformation.splitJobToTasks(jobDesc);
+                } catch (Exception e) {
+                        throw new IOException(e.getMessage());
+                }
+                
+                // look for any 
+                
+                if(jobDescription.size() == 0){
+                        if(log.isWarnEnabled())
+                                log.warn("Omitting tasks creation for job "+jobDescription.getDescription().getAppId()+". This job contains no tasks.");
+                        return null;
+                }
+
+                String jobId = jobDescription.getJobId();
+                int parserType = localWAParser.getType();
+                
+                for(int i = 0; i < jobDescription.size(); i++){
+                        TaskDescription taskDesc = jobDescription.get(i);
+                        
+                        String waTaskDesc[] = localWAParser.readTask(jobId, taskDesc.getTaskId());
+                        
+                        if(waTaskDesc != null) {
+                                
+                                if(parserType == 0){
+                                
+                                        taskDesc.setUserDn(waTaskDesc[SWFFields.DATA_USER_ID]);
+                                        
+                                        long timeValue = Long.parseLong(waTaskDesc[SWFFields.DATA_SUBMIT_TIME]);
+                                        taskDesc.setSubmissionTime(timeValue);
+                                        
+                                        long waitTime = Long.parseLong(waTaskDesc[SWFFields.DATA_WAIT_TIME]);
+                                        taskDesc.setWorkloadLogWaitTime(waitTime);
+                                        
+                                        timeValue = Long.parseLong(waTaskDesc[SWFFields.DATA_RUN_TIME]);
+                                        if(timeValue <= 0)
+                                                return null;
+                                        
+                                        long allocProcesors = Long.parseLong(waTaskDesc[SWFFields.DATA_NUMBER_OF_ALLOCATED_PROCESSORS]);
+                                        if(allocProcesors <= 0)
+                                                return null;
+                                        
+                                        timeValue = timeValue * puSpeed * allocProcesors;
+                                        taskDesc.setTaskLength(timeValue);
+
+                                } else if(parserType == 1){
+                                
+                                        taskDesc.setUserDn(waTaskDesc[GWFFields.DATA_USER_ID]);
+                                        
+                                        long timeValue = Long.parseLong(waTaskDesc[GWFFields.DATA_SUBMIT_TIME]);
+                                        taskDesc.setSubmissionTime(timeValue);
+                                        
+                                        long waitTime = Long.parseLong(waTaskDesc[GWFFields.DATA_WAIT_TIME]);
+                                        taskDesc.setWorkloadLogWaitTime(waitTime);
+                                        
+                                        timeValue = Long.parseLong(waTaskDesc[GWFFields.DATA_RUN_TIME]);
+                                        if(timeValue <= 0)
+                                                return null;
+                                        
+                                        long allocProcesors = Long.parseLong(waTaskDesc[GWFFields.DATA_NPROCS]);
+                                        if(allocProcesors <= 0)
+                                                return null;
+                                        
+                                        timeValue = timeValue * puSpeed * allocProcesors;
+                                        taskDesc.setTaskLength(timeValue);
+                                }
+                        }
+                        /*JobDescription xmlJobDescription = jobProfilesMap.get(taskDesc.getDescription().getExecution().getExecutable().getApplication().getName());
+                        if(xmlJobDescription != null){
+                                Task patternTask = xmlJobDescription.getDescription().getTask(0);
+                                if(patternTask != null){
+                                        ResourceConsumptionProfileType rcp = patternTask.getExecution().getResourceConsumptionProfile();
+                                        taskDesc.getDescription().getExecution().setResourceConsumptionProfile(rcp);
+                                }
+                        }*/
+                        this.generatedTasksCnt++;
+                        
+                }
+                this.generatedJobsCnt++;
+                
+                return jobDescription;
+        }
         
 }

DCWoRMS/branches/coolemall/src/simulator/workload/reader/archive/AbstractWAParser.java

@@ -32 +32 @@
         protected HashMap<String, String> reverseIdMapping; // key - xmlJobId_xmlTaskId, value - swf job id
         protected HashMap<String, Long> jobIndex;
+        protected HashMap<String, String> appMapping;
         protected String fields[];
         protected int fieldsNo;
@@ -44 +45 @@
                 this.reverseIdMapping = new HashMap<String, String>();
                 this.jobIndex = new HashMap<String, Long>();
+                this.appMapping = new HashMap<String, String>();
                 this.headerLoaded = false;
                 this.buildIndex = true;
@@ -144 +146 @@
                         } else if(label.equals(WAFields.COMMENT_IDMAPPING)){
                                 loadIDMapping();
+                                continue;
+                        } else if(label.equals(WAFields.COMMENT_APPLICATION)){
+                                String [] valueData = value.split(" ");
+                                String appId = valueData[0];
+                                String appName = valueData[1];
+                                appMapping.put(appId, appName);
                                 continue;
                         }
@@ -285 +293 @@
                 return null;
         }
+        
+        
+        public String getAppMapping(String appId) {
+                String appName = appMapping.get(appId);
+                return appName;
+        }
+        
 }

DCWoRMS/branches/coolemall/src/simulator/workload/reader/archive/QcgWAJobReader.java

@@ -2 +2 @@
 
 import java.io.IOException;
+import java.io.StringReader;
 import java.io.StringWriter;
 import java.io.Writer;
+import java.util.Map;
 
 import org.exolab.castor.xml.MarshalException;
 import org.exolab.castor.xml.Marshaller;
 import org.exolab.castor.xml.ResolverException;
+import org.exolab.castor.xml.Unmarshaller;
 import org.exolab.castor.xml.ValidationException;
 import org.exolab.castor.xml.XMLContext;
@@ -13 +16 @@
 import simulator.workload.reader.archive.swf.SWFFields;
 import org.qcg.broker.schemas.jobdesc.QcgJob;
+import org.qcg.broker.schemas.jobdesc.ResourceConsumptionProfileType;
 import org.qcg.broker.schemas.jobdesc.Task;
+
+import dcworms.schedframe.scheduling.utils.JobDescription;
 
 /**
@@ -25 +31 @@
         protected String currntJobID;
         protected Marshaller marshaller;
+        
+        protected Unmarshaller unmarshaller;
         
         public QcgWAJobReader(String fileName) throws NullPointerException,
@@ -42 +50 @@
                         context.addPackage("org.qcg.broker.schemas.jobdesc.QcgJob");
                         this.marshaller = context.createMarshaller();
+                        unmarshaller = context.createUnmarshaller();
                 } catch (ResolverException e) {
                         e.printStackTrace();
                         this.marshaller = null;
+                        unmarshaller = null;
                 }
         }
@@ -100 +110 @@
         }
 
+        public String mergeSwfAndXmlProfile( Map<String, JobDescription> jobProfilesMap, String swfJobDesc) throws IOException{
+                
+                StringReader reader = new StringReader(swfJobDesc);
+                QcgJob job = null;
+                try {
+                        job = (QcgJob)unmarshaller.unmarshal(reader);
+                } catch (MarshalException e) {
+                        // TODO Auto-generated catch block
+                        e.printStackTrace();
+                } catch (ValidationException e) {
+                        // TODO Auto-generated catch block
+                        e.printStackTrace();
+                }
+                
+                JobDescription xmlJobDescription = jobProfilesMap.get(job.getTask(0).getExecution().getExecutable().getApplication().getName());
+                if(xmlJobDescription != null){
+                        Task patternTask = xmlJobDescription.getDescription().getTask(0);
+                        if(patternTask != null){
+                                ResourceConsumptionProfileType rcp = patternTask.getExecution().getResourceConsumptionProfile();
+                                job.getTask(0).getExecution().setResourceConsumptionProfile(rcp);
+                        }
+                }
+                
+                Writer w = new StringWriter();
+                try {
+                        this.marshaller.setWriter(w);
+                        this.marshaller.marshal(job);
+                } catch (MarshalException e) {
+                        new IOException(e.getMessage());
+                } catch (ValidationException e) {
+                        new IOException(e.getMessage());
+                }
+                return w.toString();
+        }
 }

DCWoRMS/branches/coolemall/src/simulator/workload/reader/archive/WAParser.java

@@ -43 +43 @@
         
         public int getType();
+        
+        public String getAppMapping(String appId);
 }

DCWoRMS/branches/coolemall/src/simulator/workload/reader/archive/swf/QcgSWFJobReader.java

@@ -1 +1 @@
 package simulator.workload.reader.archive.swf;
 
+import org.qcg.broker.schemas.jobdesc.Application;
 import org.qcg.broker.schemas.jobdesc.ComputingResourceBaseTypeChoice;
 import org.qcg.broker.schemas.jobdesc.ComputingResourceType;
 import org.qcg.broker.schemas.jobdesc.ComputingResourceBaseTypeChoiceItem;
 import org.qcg.broker.schemas.jobdesc.ComputingResourceParameterType;
+import org.qcg.broker.schemas.jobdesc.Executable;
 import org.qcg.broker.schemas.jobdesc.ExecutionTimeType;
+import org.qcg.broker.schemas.jobdesc.ExecutionType;
 import org.qcg.broker.schemas.jobdesc.ParameterTypeChoice;
 import org.qcg.broker.schemas.jobdesc.ParameterTypeChoiceItem;
@@ -13 +16 @@
 import org.qcg.broker.schemas.jobdesc.types.ComputingParameterName;
 import org.qcg.broker.schemas.jobdesc.Task;
+
 import java.io.IOException;
 
@@ -20 +24 @@
 
 import simulator.workload.reader.archive.QcgWAJobReader;
+import test.workloadandapp.AppTypeGenerator;
 
 
@@ -34 +39 @@
         protected String currntJobID;
         
+        protected AppTypeGenerator appTypeGen;
+        
         public QcgSWFJobReader(String fileName) throws NullPointerException,
                         IOException {
                 super(fileName);
                 this.waParser.loadHeader();
+                this.appTypeGen = new AppTypeGenerator();
         }
         
@@ -75 +83 @@
                                                 if(value <= 0){
                                                         if(log.isWarnEnabled())
-                                                                log.warn("Task "+data[0] + " is omited. Number of allocated processors should be grather then 0.");
+                                                                log.warn("Task "+data[0] + " is omitted. Number of allocated processors should be greater than 0.");
                                                         return null;
                                                 }
@@ -82 +90 @@
                                                 if(value <= 0){
                                                         if(log.isWarnEnabled())
-                                                                log.warn("Task "+data[0] + " is omited. Number of requested processors should be grather then 0.");
+                                                                log.warn("Task "+data[0] + " is omitted. Number of requested processors should be greater than 0.");
                                                         return null;
                                                 }
@@ -89 +97 @@
                                                 if(value <= 0){
                                                         if(log.isWarnEnabled())
-                                                                log.warn("Task "+data[0] + " is omited. Task runtime should be grather then 0.");
+                                                                log.warn("Task "+data[0] + " is omitted. Task runtime should be greater than 0.");
                                                         return null;
                                                 }
@@ -95 +103 @@
                                 }
                                 
-                                if (value == -1) continue;
+                                if (value == -1 && i != SWFFields.DATA_EXECUTABLE_NUMBER) continue;
                                 
                         } catch (NumberFormatException e) {
@@ -207 +215 @@
                                         
                                 case SWFFields.DATA_EXECUTABLE_NUMBER:
+                                        String appName = waParser.getAppMapping(String.valueOf(value));
+                                        if(appName == null){
+                                                if(value != -1){
+                                                        appName = String.valueOf(value);
+                                                } else {
+                                                        appName = appTypeGen.randomAppType();
+                                                }
+                                        }
+                                        ExecutionType execType = new ExecutionType();
+                                        Executable executable = new Executable();
+                                        Application application = new Application();
+                                        application.setName(appName);
+                                        executable.setApplication(application);
+                                        execType.setExecutable(executable);
+                                        task.setExecution(execType);
                                         break;
                                         

DCWoRMS/branches/coolemall/src/simulator/workload/writer/swf/QcgSWFJobWriter.java

@@ -13 +13 @@
 import org.qcg.broker.schemas.jobdesc.Task;
 import org.qcg.broker.schemas.jobdesc.wrapper.TaskRequirements;
-import org.qcg.broker.schemas.jobdesc.wrapper.impl.TaskRequirementsImpl;
+import test.jobschema.DCWoRMSTaskRequirementsImpl;
 
 /**
@@ -37 +37 @@
                 String idMapping;
                 String taskId;
-                TaskRequirements taskReqWrapper = new TaskRequirementsImpl();
+                TaskRequirements taskReqWrapper = new DCWoRMSTaskRequirementsImpl();
                 
                 for(int i = 0; i < job.getTaskCount(); i++){