source: DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/JobRegistryImpl.java @ 1362

package schedframe.scheduling.manager.tasks;

import gridsim.dcworms.DCWormsTags;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.qcg.broker.schemas.resreqs.ParentType;
import org.qcg.broker.schemas.resreqs.Workflow;
import org.qcg.broker.schemas.resreqs.types.TaskStatesName;

import dcworms.schedframe.scheduling.ExecTask;

import qcg.shared.constants.BrokerConstants;
import schedframe.ExecutablesList;
import schedframe.scheduling.tasks.JobInterface;
import schedframe.scheduling.tasks.Task;

public class JobRegistryImpl extends AbstractJobRegistry {

        private static Log log = LogFactory.getLog(JobRegistryImpl.class);

        private String context;

        //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) {
                this.context = context;
        }

        public boolean addExecTask(ExecTask newTask) {
                if(getTask(newTask.getJobId(), newTask.getId()) == null) {
                        synchronized (executables)  {
                                executables.add(newTask);
                        }
                        return true;
                }
                return false;
        }

        public ExecutablesList getTasks() {
                return executables;
        }
        public List<ExecTask> getTasks(int status) {
                List<ExecTask> taskList = new ArrayList<ExecTask>();
                synchronized (executables)  {
                        for (ExecTask task: executables) {
                                if (task.getStatus() == status) {
                                        Set<String> visitedResource = task.getAllocatedResources().getLast().getVisitedResources();
                                        for(String res: visitedResource){
                                                if(res.contains(context)){
                                                        taskList.add(task);
                                                        break;
                                                }
                                        }
                                        if(task.getSchedulerName().contains(context)) {
                                                taskList.add(task);
                                        }
                                }
                        }
                }
                return taskList;
        }

        public List<ExecTask> getQueuedTasks() {
                return getTasks(DCWormsTags.QUEUED);
        }
        
        public List<ExecTask> getRunningTasks() {
                return getTasks(DCWormsTags.INEXEC);
        }
        
        public List<ExecTask> getReadyTasks() {
                return getTasks(DCWormsTags.READY);
        }
        
        public List<ExecTask> getFinishedTasks() {
                return getTasks(DCWormsTags.SUCCESS);
        }
        
        public ExecTask getTask(String jobId, String taskId){
                synchronized (executables)  {
                        for (ExecTask task : executables) {
                                if (task.getJobId().compareTo(jobId) == 0 && task.getId().compareTo(taskId) == 0) {
                                        return task;
                                }
                        }
                }
                return null;
        }

        @SuppressWarnings("unchecked")
        public List<Task> getAvailableTasks(List<JobInterface<?>> jobList) {
                List<Task> availableTasks = new ArrayList<Task>();
                List<Task> waitingTasks = new ArrayList<Task>();
                
                for(int i = 0; i < jobList.size(); i++){
                        JobInterface<?> job = (JobInterface<?>)jobList.get(i);
                        waitingTasks.addAll((List<Task>)job.getTask());
                }

                availableTasks.addAll(getPrecedenceConstrainedAvailableTasks(waitingTasks));
                return availableTasks;
        }

        private List<Task> getPrecedenceConstrainedAvailableTasks(List<Task> tasks){
                
                List<Task> availableTasks = new ArrayList<Task>();
                int size = tasks.size();
                
                for(int i = 0; i < size; i++){
                        int parCnt;
                        int previousTaskSucceedCnt = 0;
                        Task task = tasks.get(i);
                        if(task.getStatus() != (int)BrokerConstants.TASK_STATUS_UNSUBMITTED)
                                continue;
                        //the following procedure supports only one nested structure
                        Workflow w = task.getDescription().getWorkflow();
                        if (w == null){
                                availableTasks.add(task);
                                continue;
                        }
                        if(w.getAnd() != null) {
                                parCnt = w.getAnd().getParentOpTypeItemCount();
                                if(parCnt == 0)
                                {
                                        availableTasks.add(task);
                                }
                                else
                                {
                                        for(int j = 0; j < parCnt; j++){
                                                ParentType par = w.getAnd().getParentOpTypeItem(j).getParent();
                                                if(par.getTriggerState().compareTo(TaskStatesName.FINISHED) == 0){
                                                        if(!checkTaskCompletion(task.getJobId(), par.getContent())){
                                                                break;
                                                        }
                                                }
                                                previousTaskSucceedCnt++;
                                        }

                                        if(previousTaskSucceedCnt == parCnt)
                                                availableTasks.add(task);
                                }
                        } 
                        else if(w.getOr() != null) {
                                parCnt = w.getOr().getParentOpTypeItemCount();
                                if(parCnt == 0)
                                {
                                        availableTasks.add(task);
                                }
                                else
                                {
                                        for(int j = 0; j < parCnt; j++){
                                                ParentType par = w.getOr().getParentOpTypeItem(j).getParent();
                                                if(par.getTriggerState().compareTo(TaskStatesName.FINISHED) == 0){
                                                        if(!checkTaskCompletion(task.getJobId(), par.getContent())){
                                                                continue;
                                                        }
                                                }
                                                previousTaskSucceedCnt++;
                                        }

                                        if(previousTaskSucceedCnt > 0)
                                                availableTasks.add(task);
                                }
                        }
                        else {
                                parCnt = w.getParentCount();
                                if(parCnt == 0)
                                {
                                        availableTasks.add(task);
                                }
                                else
                                {
                                        for(int j = 0; j < parCnt; j++){
                                                ParentType par = w.getParent(j);
                                                if(par.getTriggerState().compareTo(TaskStatesName.FINISHED) == 0){
                                                        if(par.getContent().contains("_")){
                                                                if(!checkTaskCompletion(par.getContent().split("_")[0], par.getContent().split("_")[1])){
                                                                        continue;
                                                                }
                                                        }
                                                        else if(!checkTaskCompletion(task.getJobId(), par.getContent())){
                                                                continue;
                                                        }
                                                }
                                                previousTaskSucceedCnt++;
                                        }

                                        if(previousTaskSucceedCnt == parCnt)
                                                availableTasks.add(task);
                                }
                        }
                        
                        /*try{          
                                parCnt = task.getDescription().getWorkflow().getParentCount();
                        } catch(Exception e){
                                parCnt = 0;
                        }
                        if(parCnt == 0){
                                availableTasks.add(task);
                        }
                        else {
                                for(int j = 0; j < parCnt; j++){
                                        ParentType par = task.getDescription().getWorkflow().getParent(j);
                                        if(par.getTriggerState().compareTo(TaskStatesName.FINISHED) == 0){
                                                if(!checkTaskCompletion(task.getJobId(), par.getContent())){
                                                        break;
                                                }
                                        }
                                        previousTaskSucceedCnt++;
                                }
                                
                                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 availableTasks;
        }
        
        private boolean checkTaskCompletion(String jobID, String taskID){
                JobInterface<?> job = getJobInfo(jobID);
                try {
                        if(job.getTask(taskID).isFinished())
                                return true;
                } catch (NoSuchFieldException e) {
                        //e.printStackTrace();
                }
                return false;
        }

}