Changeset 481 for DCWoRMS/trunk/src/schedframe/scheduling/manager

Timestamp:

10/08/12 10:23:45 (13 years ago)

Author:

wojtekp

Message:

 

Location:

DCWoRMS/trunk/src/schedframe/scheduling/manager/tasks

Files:

• AbstractJobRegistry.java (modified) (4 diffs)
• JobRegistry.java (modified) (3 diffs)
• JobRegistryImpl.java (modified) (9 diffs)

DCWoRMS/trunk/src/schedframe/scheduling/manager/tasks/AbstractJobRegistry.java

@@ -13 +13 @@
 
 
-public abstract class AbstractJobRegistry /*extends ConcurrentHashMap<String, Job> */implements JobRegistry, Cloneable{
+public abstract class AbstractJobRegistry /*extends ConcurrentHashMap<String, Job>*/ implements JobRegistry, Cloneable{
         
         private static final long serialVersionUID = 8409060063583755824L;
+
         
-        private static Log log = LogFactory.getLog(AbstractJobRegistry.class);
-        
-        protected static final ConcurrentHashMap<String, Job> jobs = new ConcurrentHashMap<String, Job>();
+        protected static final ConcurrentHashMap<String, JobInterface<?>> jobs = new ConcurrentHashMap<String, JobInterface<?>>();
         
         protected AbstractJobRegistry(){
-                //log.warn("Methods from JobRegistry interface are not implemented.");
         }
         
         public boolean addJob(JobInterface<?> job) {
-                jobs.put(job.getId(), (Job) job);
+                jobs.put(job.getId(),  job);
                 return true;
         }
@@ -32 +30 @@
         public boolean addTask(TaskInterface<?> task) {
                 if(jobs.containsKey(task.getJobId())){
-                        jobs.get(task.getJobId()).add((Task)task);
+                        getJob(task.getJobId()).add((Task)task);
                         return true;
                 } else {
@@ -39 +37 @@
         }
 
-        public Job getJob(String jobId){
+        public JobInterface<?> getJobInfo(String jobId) {
                 return jobs.get(jobId);
         }
-        
-        public JobInterface<?> getJobInfo(String jobID) {
-                return jobs.get(jobID);
-        }
 
-        public TaskInterface<?> getTaskInfo(String jobID, String taskId) {
+        public TaskInterface<?> getTaskInfo(String jobId, String taskId) {
                 Task task = null;
-                Job job = jobs.get(jobID);
+                Job job = getJob(jobId);
                 
                 if(job == null)
@@ -57 +51 @@
                         task = job.getTask(taskId);
                 } catch (NoSuchFieldException e) {
-                        log.error(e.getMessage());
                 }
                 return task;
         }
         
-        /*public List<JobInterface<?>> getActiveJobs() {
-                log.error("getActiveJobs() not implemented.");
-                return null;
+        public Job getJob(String jobId){
+                return (Job)jobs.get(jobId);
         }
 
-        public List<TaskInterface<?>> getActiveTasks() {
-                log.error("getActiveTasks() not implemented.");
-                return null;
-        }*/
-
 }

DCWoRMS/trunk/src/schedframe/scheduling/manager/tasks/JobRegistry.java

@@ -5 +5 @@
 import java.util.List;
 
+import schedframe.ExecutablesList;
 import schedframe.scheduling.tasks.JobInterface;
 import schedframe.scheduling.tasks.TaskInterface;
@@ -11 +12 @@
 public interface JobRegistry {
 
-        //public List<JobInterface<?>> getActiveJobs();
-
-        //public List<TaskInterface<?>> getActiveTasks();
-        
         public JobInterface<?> getJobInfo(String jobId);
 
@@ -31 +28 @@
 
 
-        //public List<SubmittedTask> getSubmittedTasks();
+        public ExecutablesList getExecutableTasks(); 
         
-        public ExecTask getSubmittedTask(String jobId, String taskId);
+        public ExecTask getExecutable(String jobId, String taskId);
 
         
-        public List<? extends TaskInterface<?>> getReadyTasks(List<JobInterface<?>> jobsList);
+        public List<? extends TaskInterface<?>> getAvailableTasks(List<JobInterface<?>> jobsList);
 
 }

DCWoRMS/trunk/src/schedframe/scheduling/manager/tasks/JobRegistryImpl.java

@@ -1 +1 @@
 package schedframe.scheduling.manager.tasks;
 
-import gridsim.Gridlet;
+import gridsim.gssim.DCWormsTags;
 import gssim.schedframe.scheduling.ExecTask;
-import gssim.schedframe.scheduling.Executable;
 
 import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 import org.apache.commons.lang.ArrayUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.joda.time.DateTime;
 import org.qcg.broker.schemas.resreqs.ParentType;
 import org.qcg.broker.schemas.resreqs.types.TaskStatesName;
 
 import qcg.shared.constants.BrokerConstants;
-import schedframe.resources.units.ResourceUnit;
-import schedframe.resources.units.ResourceUnitName;
-import schedframe.scheduling.ResourceHistoryItem;
-import schedframe.scheduling.plan.AllocationInterface;
-import schedframe.scheduling.tasks.AbstractProcesses;
+import schedframe.ExecutablesList;
 import schedframe.scheduling.tasks.JobInterface;
-import schedframe.scheduling.tasks.SubmittedTask;
 import schedframe.scheduling.tasks.Task;
-import simulator.WormsConstants;
 
 public class JobRegistryImpl extends AbstractJobRegistry {
@@ -37 +26 @@
         private String context;
 
-        //TO DO - change data structure
-        protected static final List<ExecTask> submittedTasks = Collections.synchronizedList(new ArrayList<ExecTask>());;
-        //protected static final List<ExecTaskInterface> submittedTasks = new CopyOnWriteArrayList<ExecTaskInterface>();
+        //TO DO - consider data structure
+        protected static final ExecutablesList executables = new ExecutablesList();
+        //protected static final List<ExecTask> executables = Collections.synchronizedList(new ArrayList<ExecTask>());;
+        //protected static final List<ExecTaskInterface> executables = new CopyOnWriteArrayList<ExecTaskInterface>();
 
-        public JobRegistryImpl(String context_) {
-                context = context_;
+        public JobRegistryImpl(String context) {
+                this.context = context;
         }
 
-        /*protected void setContext(String context_) {
-                context = context_;
-        }*/
-
-        public boolean addTask(ExecTask newTask) {
-                if(getSubmittedTask(newTask.getJobId(), newTask.getId()) == null)
-                {
-                        synchronized (submittedTasks)  {
-                                submittedTasks.add(newTask);
+        public boolean addExecTask(ExecTask newTask) {
+                if(getExecutable(newTask.getJobId(), newTask.getId()) == null) {
+                        synchronized (executables)  {
+                                executables.add(newTask);
                         }
                         return true;
@@ -60 +45 @@
         }
 
+        public ExecutablesList getExecutableTasks() {
+                return executables;
+        }
         public List<ExecTask> getTasks(int status) {
                 List<ExecTask> taskList = new ArrayList<ExecTask>();
-                synchronized (submittedTasks)  {
-                        for (ExecTask task: submittedTasks) {
+                synchronized (executables)  {
+                        for (ExecTask task: executables) {
                                 if (task.getStatus() == status) {
-                                        //SubmittedTask subTask = (SubmittedTask) task;
                                         List<String> visitedResource = task.getVisitedResources();
-                                        if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)){
+                                        if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)) {
                                                 taskList.add(task);
                                         }
-                                        /*if(subTask.getVisitedResources().contains(context)){
-                                                taskList.add(subTask);
-                                        }*/
                                 }
                         }
@@ -80 +64 @@
 
         public List<ExecTask> getQueuedTasks() {
-                return getTasks(Gridlet.QUEUED);
+                return getTasks(DCWormsTags.QUEUED);
         }
         
         public List<ExecTask> getRunningTasks() {
-                return getTasks(Gridlet.INEXEC);
+                return getTasks(DCWormsTags.INEXEC);
         }
         
         public List<ExecTask> getReadyTasks() {
-                return getTasks(Gridlet.READY);
+                return getTasks(DCWormsTags.READY);
         }
         
         public List<ExecTask> getFinishedTasks() {
-                return getTasks(Gridlet.SUCCESS);
+                return getTasks(DCWormsTags.SUCCESS);
         }
         
-        
-        public List<ExecTask> getAllSubmittedTasks() {
-                List<ExecTask> taskList;
-                synchronized (submittedTasks)  {
-                        taskList = new ArrayList<ExecTask>(submittedTasks);
-                }
-                return taskList;
-        }
-
-        public List<SubmittedTask> getSubmittedTasks() {
-                List<SubmittedTask> taskList = new ArrayList<SubmittedTask>();
-                synchronized (submittedTasks)  {
-                        for (ExecTask task : submittedTasks) {
-                                SubmittedTask subTask = (SubmittedTask) task;
-                                List<String> visitedResource = subTask.getVisitedResources();
-                                if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)){
-                                        taskList.add(subTask);
-                                }
-                                /*if(subTask.getVisitedResources().contains(context)){
-                                        taskList.add(subTask);
-                                }*/
-                        }
-                }
-                return taskList;
-        }
-        
-        public ExecTask getSubmittedTask(String jobId, String taskId){
-                synchronized (submittedTasks)  {
-                        for (ExecTask task : submittedTasks) {
+        public ExecTask getExecutable(String jobId, String taskId){
+                synchronized (executables)  {
+                        for (ExecTask task : executables) {
                                 if (task.getJobId().compareTo(jobId) == 0 && task.getId().compareTo(taskId)==0) {
                                         return task;
@@ -131 +89 @@
                 return null;
         }
-        
 
         @SuppressWarnings("unchecked")
-        public List<Task> getReadyTasks(List<JobInterface<?>> wuList) {
-                List<Task> readyTasks = new ArrayList<Task>();
+        public List<Task> getAvailableTasks(List<JobInterface<?>> wuList) {
+                List<Task> availableTasks = new ArrayList<Task>();
                 List<Task> waitingTasks = new ArrayList<Task>();
                 
@@ -143 +100 @@
                 }
 
-                readyTasks.addAll(getPrecedenceConstrainedReadyTasks(waitingTasks));
-                return readyTasks;
-        }
-        
-        public Executable getTaskExecutable(Integer executableId){
-                synchronized (submittedTasks)  {
-                        for (ExecTask task : submittedTasks) {
-                                SubmittedTask subTask = (SubmittedTask) task;
-                                Executable exec = (Executable)subTask.getGridlet();
-                                if (exec.getGridletID() == executableId) {
-                                        return exec;
-                                }
-                        }
-                }
-                return null;
-        }
-        
-        public List<Executable> getJobExecutables(String jobId){
-                List<Executable> list = new ArrayList<Executable>();
-                synchronized (submittedTasks)  {
-                        for(int i = 0; i < submittedTasks.size(); i++){
-                                SubmittedTask subTask = (SubmittedTask) submittedTasks.get(i);
-                                Executable exec = (Executable)subTask.getGridlet();
-                                
-                                if(exec.getJobId().equals(jobId))
-                                        list.add(exec);
-                        }
-                }
-                return list;
-        }
-        
-        public JobRegistryImpl clone() {
-                JobRegistryImpl jr = null;
-                try {
-                        jr = (JobRegistryImpl) super.clone();
-                } catch (CloneNotSupportedException e) {
-                        // TODO Auto-generated catch block
-                        e.printStackTrace();
-                }
-
-                return jr;
-        }
-
-        /*public AbstractExecutable getTaskExecutabls(String jobId, String taskId){
-                List<AbstractExecutable> list = new ArrayList<AbstractExecutable>();
-                synchronized (submittedTasks)  {
-                        for(int i = 0; i < size(); i++){
-                                SubmittedTask subTask = (SubmittedTask) submittedTasks.get(i);
-                                AbstractExecutable exec = (AbstractExecutable)subTask.getGridlet();
-                                
-                                if(exec.getJobId().equals(jobId) && exec.getId().equals(taskId))
-                                        return exec;
-                        }
-                }
-                return null;
-        }*/
-        
-        
-        public Executable createExecutable(Task task, AllocationInterface allocation) {
-
-                String refersTo = allocation.getProcessGroupId(); // null;//allocation.getRefersTo();
-                if(refersTo == null)
-                        refersTo = task.getId();
-                        
-                Executable exec = null;
-
-                if(refersTo.equals(task.getId())){
-                        exec = new Executable(task);
-                } else {
-                        List<AbstractProcesses> processes = task.getProcesses();
-                        if(processes == null) {
-                                try {
-                                        log.error("Allocation: " + allocation.getDocument() + "\nrefers to unknown task or processes set." +
-                                                        " Set correct value (task id or prcesses set id) for allocation refersTo attribute.");
-                                } catch (Exception e) {
-                                        e.printStackTrace();
-                                }
-                        }
-                        boolean found = false;
-                        for(int j = 0; j < processes.size() && !found; j++){
-                                AbstractProcesses procesesSet = processes.get(j);
-                                if(refersTo.equals(procesesSet.getId())){
-                                        exec = new Executable(task, procesesSet);
-                                        found = true;
-                                }
-                        }
-                        if(!found){
-                                log.error("Allocation refers to unknown proceses set.");
-                        }
-                }
-                
-        //      exec.setUserID(task.getSenderId());
-                exec.setLength(task.getLength());
-                exec.setReservationId(allocation.getReservationId());
-                        
-                /*HostInterface<?> host = allocation.getHost();
-                ComputingResourceTypeInterface<?> crt = host.getMachineParameters();
-                if(crt != null){
-                        ComputingResourceTypeItemInterface<?> crti = crt.getComputingResourceTypeItem(0);
-                        if(crti != null){
-                                ParameterPropertyInterface<?> properties[] = crti.getHostParameter().getProperty();
-                                for(int p = 0; p < properties.length; p++){
-                                        ParameterPropertyInterface<?> property = properties[p];
-                                        if("chosenCPUs".equals(property.getName())){
-                                                Object cpuNames = property.getValue();
-                                                exec.addSpecificResource(ResourceParameterName.FREECPUS, cpuNames);
-                                        }
-                                }
-                        }
-                }*/
-                return exec;
-        }
-        
-
-        public List<Executable> createExecutables(Task task) {
-
-                List<AbstractProcesses> processes = task.getProcesses();
-                
-                List<Executable> executables = new ArrayList<Executable>();
-
-                if(processes == null || processes.size()==0){
-                        Executable exec = new Executable(task);
-                        exec.setUserID(task.getSenderId());
-                        exec.setLength(task.getLength());
-                        executables.add(exec);
-                } else {
-
-                        boolean found = false;
-                        for(int j = 0; j < processes.size() && !found; j++){
-                                AbstractProcesses procesesSet = processes.get(j);
-                                Executable exec = new Executable(task, procesesSet);
-                                exec.setUserID(task.getSenderId());
-                                exec.setLength(task.getLength());
-                                executables.add(exec);
-                        }
-                }
-
-                return  executables;
+                availableTasks.addAll(getPrecedenceConstrainedAvailableTasks(waitingTasks));
+                return availableTasks;
         }
 
 
-        /**************************************/
-        protected static Map<Integer, Map<String, Object>> history = new HashMap<Integer, Map<String,Object>>();
-        
-        public static Map<Integer, Map<String, Object>> getAllocationHistory(){
-                return history;
-        }
-        
-        public void saveHistory (SubmittedTask submittedTask, int estimatedTime, Map<ResourceUnitName, ResourceUnit> choosenResources){
+        private List<Task> getPrecedenceConstrainedAvailableTasks(List<Task> tasks){
                 
-                /*submittedTask.setEstimatedDuration(estimatedTime);
-
-                DateTime currentTime = new DateTime();
-                ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime);
-                submittedTask.addUsedResources(resHistItem);*/
-
-                ResourceHistoryItem resHistItem = submittedTask.getUsedResources().getLast();
-                DateTime currentTime = new DateTime();
-                Map<String, Object> historyItem = new HashMap<String, Object>();
-                List<ResourceHistoryItem> list = new ArrayList<ResourceHistoryItem>(1);
-                list.add(resHistItem);
-                historyItem.put(WormsConstants.RESOURCES, list);
-                historyItem.put(WormsConstants.START_TIME, currentTime);
-                currentTime = currentTime.plusSeconds(estimatedTime);
-                historyItem.put(WormsConstants.END_TIME, currentTime);
-
-                history.put(Integer.valueOf(submittedTask.getGridletID()), historyItem);
-                /*ProcessingElements pes = (ProcessingElements) choosenResources.get(ResourceParameterName.PROCESSINGELEMENTS);
-                for (ComputingResource resource : pes) {
-                        //submittedTask.addToResPath(resource.getName());
-                        submittedTask.visitResource(resource.getName());
-                        ComputingResource parent = resource.getParent();
-                        while (parent != null && !submittedTask.getResPath().contains(parent.getName() + "_")) {
-                                submittedTask.addToResPath(parent.getName());
-                                parent = parent.getParent();
-                        }
-                        while (parent != null && !submittedTask.getVisitedResources().contains(parent.getName() + "_")) {
-                                submittedTask.visitResource(parent.getName());
-                                parent = parent.getParent();
-                        }
-                }*/
-        }
-
-        private List<Task> getPrecedenceConstrainedReadyTasks(List<Task> tasks){
+                List<Task> availableTasks = new ArrayList<Task>();
+                int size = tasks.size();
                 
-                List<Task> readyTasks = new ArrayList<Task>();
-
-                int size = tasks.size();
                 for(int i = 0; i < size; i++){
                         int parCnt;
-                        int previousTaskReadyCnt = 0;
+                        int previousTaskSucceedCnt = 0;
                         Task task = tasks.get(i);
                         if(task.getStatus() != (int)BrokerConstants.TASK_STATUS_UNSUBMITTED)
@@ -341 +120 @@
                         } catch(Exception e){
                                 parCnt = 0;
-                                //e.printStackTrace();
                         }
-                        if(parCnt == 0)
-                        {
-                                readyTasks.add(task);
+                        if(parCnt == 0){
+                                availableTasks.add(task);
                         }
-                        else
-                        {
+                        else {
                                 for(int j = 0; j < parCnt; j++){
                                         ParentType par = task.getDescription().getWorkflow().getParent(j);
@@ -356 +132 @@
                                                 }
                                         }
-                                        previousTaskReadyCnt++;
+                                        previousTaskSucceedCnt++;
                                 }
                                 
-                                if(previousTaskReadyCnt == parCnt && task.getDescription().getWorkflow().getAnd() != null)
-                                        readyTasks.add(task);
-                                else if(previousTaskReadyCnt > 0 && task.getDescription().getWorkflow().getOr() != null)
-                                        readyTasks.add(task);
-                                else if (previousTaskReadyCnt == parCnt)
-                                        readyTasks.add(task);
+                                if(previousTaskSucceedCnt == parCnt && task.getDescription().getWorkflow().getAnd() != null)
+                                        availableTasks.add(task);
+                                else if(previousTaskSucceedCnt > 0 && task.getDescription().getWorkflow().getOr() != null)
+                                        availableTasks.add(task);
+                                else if (previousTaskSucceedCnt == parCnt)
+                                        availableTasks.add(task);
                         }
                 }               
-                return readyTasks;
+                return availableTasks;
         }
         
@@ -380 +156 @@
                 return false;
         }
+
 }