Changeset 539 for DCWoRMS/trunk/build/classes/schedframe/scheduling/policy/global

Timestamp:

10/31/12 13:52:06 (12 years ago)

Author:

wojtekp

Message:

 

Location:

DCWoRMS/trunk/build/classes/schedframe/scheduling/policy/global

Files:

• GlobalManagementSystem$GlobalWorkloadUnitHandler.class (modified) (previous)
• GlobalManagementSystem.class (modified) (previous)
• GlobalManagementSystem.java (modified) (13 diffs)
• GridBroker.class (modified) (previous)
• GridBroker.java (modified) (3 diffs)

DCWoRMS/trunk/build/classes/schedframe/scheduling/policy/global/GlobalManagementSystem.java

@@ -11 +11 @@
 import org.qcg.broker.schemas.schedulingplan.types.AllocationStatus;
 
+import dcworms.schedframe.scheduling.ExecTask;
+import dcworms.schedframe.scheduling.Executable;
+
 import qcg.shared.constants.BrokerConstants;
-
-import schedframe.events.scheduling.EventReason;
-import schedframe.events.scheduling.SchedulingEvent;
 import schedframe.events.scheduling.TaskArrivedEvent;
-import schedframe.events.scheduling.TaskCanceledEvent;
 import schedframe.events.scheduling.TimerEvent;
-import schedframe.scheduling.Scheduler;
+import schedframe.scheduling.TaskListImpl;
 import schedframe.scheduling.WorkloadUnitHandler;
-import schedframe.scheduling.WorkloadUnitListImpl;
 import schedframe.scheduling.plan.AllocationInterface;
 import schedframe.scheduling.plan.ScheduledTaskInterface;
@@ -31 +29 @@
 import schedframe.scheduling.tasks.Job;
 import schedframe.scheduling.tasks.JobInterface;
-import schedframe.scheduling.tasks.SubmittedTask;
 import schedframe.scheduling.tasks.Task;
 import schedframe.scheduling.tasks.TaskInterface;
 import schedframe.scheduling.tasks.WorkloadUnit;
-
 import eduni.simjava.Sim_event;
 import gridsim.GridSim;
 import gridsim.GridSimTags;
-import gridsim.Gridlet;
 import gridsim.IO_data;
-import gridsim.gssim.WormsTags;
-import gssim.schedframe.scheduling.ExecTask;
-import gssim.schedframe.scheduling.Executable;
+import gridsim.dcworms.DCWormsTags;
 
 public class GlobalManagementSystem extends AbstractManagementSystem {
@@ -54 +47 @@
                 super(providerId, entityName,  execTimeEstimationPlugin, queues);
 
-                /*schedulingPlugin = (GlobalSchedulingPlugin) InstanceFactory.createInstance(
-                                schedulingPluginClassName,
-                                GlobalSchedulingPlugin.class);*/
                 if(schedPlugin == null){
                         throw new Exception("Can not create global scheduling plugin instance");
@@ -69 +59 @@
                 switch (tag) {
 
-                case WormsTags.TIMER:
-                        if (pluginSupportsEvent(WormsTags.TIMER)) {
+                case DCWormsTags.TIMER:
+                        if (pluginSupportsEvent(DCWormsTags.TIMER)) {
                                 TimerEvent event = new  TimerEvent();
-                                SchedulingPlanInterface decision =  schedulingPlugin.schedule(event, 
+                                SchedulingPlanInterface<?> decision =  schedulingPlugin.schedule(event, 
                                                 queues,  getJobRegistry(), getResourceManager(), moduleList);
                                 executeSchedulingPlan(decision);
@@ -81 +71 @@
         }
         
-        public void notifySubmittedWorkloadUnit(WorkloadUnit<?> wu, boolean ack) {
+        public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) {
                 if (!pluginSupportsEvent(GridSimTags.GRIDLET_SUBMIT)) {
                         log.error("Plugin " + schedulingPlugin.getClass()
@@ -90 +80 @@
                 
                 registerWorkloadUnit(wu);
-                /*Job job = (Job) wu;
-                jobRegistry.addJob(job);
-
-                if (log.isInfoEnabled())
-                        log.info("Received job " + job.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis()));
-
+
+        }
+        
+        private void registerWorkloadUnit(WorkloadUnit wu){
+                if(!wu.isRegistered()){
+                        wu.register(jobRegistry);
+                }
+                wu.accept(getWorkloadUnitHandler());
+        }
+        
+
+        protected void schedule(JobInterface<?> job){
                 List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>();
                 jobsList.add(job);
-                WorkloadUnitList readyWorkloadUnits = new WorkloadUnitList();
-                readyWorkloadUnits.addAll(jobRegistry.getReadyTasks(jobsList));
-                schedulingPlugin.placeJobsInQueues(readyWorkloadUnits, queues, getResourceManager(), moduleList);
-
-                schedule(new TaskArrivedEvent());*/
-
-        }
-        
-        private void registerWorkloadUnit(WorkloadUnit<?> wu){
-                if(!wu.isRegistered()){
-                        wu.register(jobRegistry);
-                }
-                wu.accept(getWorkloadUnitHandler());
-        }
-        
-
-        protected void scheduleReadyTasks(Job job){
-                List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>();
-                jobsList.add(job);
-                WorkloadUnitListImpl readyWorkloadUnits = new WorkloadUnitListImpl();
-                readyWorkloadUnits.addAll(jobRegistry.getReadyTasks(jobsList));
-                schedulingPlugin.placeJobsInQueues(readyWorkloadUnits, queues, getResourceManager(), moduleList);
-
-                schedule(new TaskArrivedEvent());
-        }
-        
-        protected void schedule(SchedulingEvent schedulingEvent) {
-
-                try {
-                        SchedulingPlanInterface decision = schedulingPlugin.schedule(
-                                        schedulingEvent, queues, getJobRegistry(),  getResourceManager(), moduleList);
-                        if (decision == null)
-                                return;
-
+                TaskListImpl readyTasks = new TaskListImpl();
+                readyTasks.addAll(jobRegistry.getAvailableTasks(jobsList));
+                
+                schedulingPlugin.placeTasksInQueues(readyTasks, queues, getResourceManager(), moduleList);
+                SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(
+                                new TaskArrivedEvent(), queues, getJobRegistry(),  getResourceManager(), moduleList);
+                if (decision != null)
                         executeSchedulingPlan(decision);
-
-                } catch (Exception e) {
-                        e.printStackTrace();
-                }
-        }
-        
-        public void notifyReturnedWorkloadUnit(WorkloadUnit<?> wu) {
+        }
+
+        public void notifyReturnedWorkloadUnit(WorkloadUnit wu) {
                 Executable exec = (Executable) wu;
 
@@ -149 +114 @@
                 
                 try {
-                        Job job = jobRegistry.get(exec.getJobId());
-                        /*Task task = job.getTask(exec.getTaskId());
+                        Job job = jobRegistry.getJob(exec.getJobId());
+                        Task task = job.getTask(exec.getTaskId());
                         if(exec.getProcessesId() == null){
                                 try {
                                         task.setStatus(exec.getStatus());
                                 } catch (Exception e) {
-                                        // TODO Auto-generated catch block
-                                        e.printStackTrace();
+
                                 }
                         } else {
@@ -167 +131 @@
                                         }
                                 }
-                        }*/
+                        }
                         
                         if(job.isFinished()){
@@ -173 +137 @@
                         }
                         else {
-                                scheduleReadyTasks(job);
-                                /*List<JobInterface<?>> jobs = new ArrayList<JobInterface<?>>();
-                                jobs.add(jobRegistry.getJobInfo(job.getId()));
-                                WorkloadUnitList readyWorkloadUnits = new WorkloadUnitList();
-                                readyWorkloadUnits.addAll(jobRegistry.getReadyTasks(jobs));
-                                schedulingPlugin.placeJobsInQueues(readyWorkloadUnits, queues,
-                                                getResourceManager(), moduleList);
-                                schedule(new TaskArrivedEvent());*/
+                                schedule(job);
                         }
                         
@@ -189 +146 @@
         }
 
-        public void notifyCanceledWorkloadUnit(WorkloadUnit<?> wu){;
-
-                Executable task = (Executable) wu;
-                String jobID = task.getJobId();
-                String taskID = task.getId();
-                
-                if(log.isDebugEnabled())
-                        log.debug("Received canceled job" + jobID + "_" + taskID);
-                
-                TaskInterface<?> ti = jobRegistry.getTaskInfo(jobID, taskID) ;
-                try {
-
-                        ti.setStatus((int)BrokerConstants.JOB_STATUS_CANCELED);
-                        
-                        TaskCanceledEvent event = new TaskCanceledEvent(jobID, taskID);
-                        event.setReason(EventReason.RESERVATION_EXCEEDED);
-                        schedule(event);
-                        
-                } catch (Exception e) {
-                        log.error("Exception during scheduling. " + e.getMessage());
-                        e.printStackTrace();
-                }
-        }
-        
-        protected void executeSchedulingPlan(SchedulingPlanInterface decision) {
-
-                ArrayList<ScheduledTaskInterface> taskSchedulingDecisions = decision.getTasks();
+        protected void executeSchedulingPlan(SchedulingPlanInterface<?> decision) {
+
+                ArrayList<ScheduledTaskInterface<?>> taskSchedulingDecisions = decision.getTasks();
                 for (int i = 0; i < taskSchedulingDecisions.size(); i++) {
 
-                        try {
-                                ScheduledTaskInterface taskDecision = taskSchedulingDecisions.get(i);
-        
-                                //log.info(decision.getDocument());
-        
-                                String jobID = taskDecision.getJobId();
-                                String taskID = taskDecision.getTaskId();
-                                
-                                // Task allocations that were rejected because of lack of resources or which were canceled and 
-                                // not scheduled again are returned to the user.
-                                if(taskDecision.getStatus() == AllocationStatus.REJECTED){
-                                        Job job = jobRegistry.get(jobID);
-                                        scheduler.send(job.getSenderId(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_RETURN, job);
-                                        continue;
-                                }
-                                
-                                ArrayList<AllocationInterface> allocations = taskDecision.getAllocations();
-
-                                Task task = (Task) jobRegistry.getTaskInfo(jobID, taskID);
-                                for (int j = 0; j < allocations.size(); j++) {
-
-                                        AllocationInterface allocation = allocations.get(j);
-                                        Executable exec = jobRegistry.createExecutable(task, allocation);
-                                        submitWorkloadUnit(exec, allocation);
-                                        task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED);
-                                }                                                               
-
-                        }catch (Exception e){
-                                e.printStackTrace();
-                        }
-                }
-        }
-
-        protected void submitWorkloadUnit(WorkloadUnit<?> job, AllocationInterface allocation) {
+                        ScheduledTaskInterface<?> taskDecision = taskSchedulingDecisions.get(i);
+
+                        //log.info(decision.getDocument());
+
+                        String jobID = taskDecision.getJobId();
+                        String taskID = taskDecision.getTaskId();
+                        
+                        // Task allocations that were rejected because of lack of resources or which were canceled and 
+                        // not scheduled again are returned to the user.
+                        if(taskDecision.getStatus() == AllocationStatus.REJECTED){
+                                Job job = jobRegistry.getJob(jobID);
+                                scheduler.send(job.getSenderId(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_RETURN, job);
+                                continue;
+                        }
+                        
+                        Task task = (Task) jobRegistry.getTaskInfo(jobID, taskID);
+                        
+                        ArrayList<AllocationInterface<?>> allocations = taskDecision.getAllocations();
+                        for (int j = 0; j < allocations.size(); j++) {
+
+                                AllocationInterface<?> allocation = allocations.get(j);
+                                Executable exec = createExecutable(task, allocation);
+                                submitTask(exec, allocation);
+                                task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED);
+                        }                                                               
+                }
+        }
+
+        private 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.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;
+        }
+        
+        protected void submitTask(TaskInterface<?> task, AllocationInterface<?> allocation) {
 
                 String providerName = allocation.getProviderName();
@@ -257 +238 @@
                         return;
                 }
-                TaskInterface<?> task = (TaskInterface<?>) job;
                 removeFromQueue(task);
                 
                 int resID = GridSim.getEntityId(providerName);
-                IO_data data = new IO_data(job, 0, resID);
+                IO_data data = new IO_data(task, 0, resID);
                 scheduler.send(scheduler.getOutputPort(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, data);  
-                
-                //scheduler.send(providerName, GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, job);                      
+                //scheduler.send(providerName, GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, job);      
+                
                 if(log.isDebugEnabled())
-                        try {
-                                log.debug("Submitted job " + job.getId() + " to " + providerName);
-                        } catch (NoSuchFieldException e) {
-                                // TODO Auto-generated catch block
-                                e.printStackTrace();
-                        }
+                        log.debug("Submitted job " + task.getId() + " to " + providerName);
+
         }
 
         class GlobalWorkloadUnitHandler implements  WorkloadUnitHandler{
 
-                public void handleJob(Job job){
-
-                        jobRegistry.addJob(job);
+                public void handleJob(JobInterface<?> job){
                         if (log.isInfoEnabled())
                                 log.info("Received job " + job.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis()));
 
-                        scheduleReadyTasks(job);
+                        jobRegistry.addJob(job);
+                        schedule(job);
                 }
                 
@@ -292 +267 @@
                         throw new RuntimeException("Not implemented since it isn't expected that tasks are send directly to the global scheduler.");
                 }
-
-                public void handleSubmittedTask(SubmittedTask task) {
-                        throw new RuntimeException("Not implemented since it isn't expected that tasks are send directly to the global scheduler.");
-                }
         }
 
@@ -301 +272 @@
                 return new GlobalWorkloadUnitHandler();
         }
-        
 
 

DCWoRMS/trunk/build/classes/schedframe/scheduling/policy/global/GridBroker.java

@@ -19 +19 @@
 public class GridBroker extends GlobalManagementSystem { 
 
-        
         private static Log log = LogFactory.getLog(GridBroker.class);
 
@@ -27 +26 @@
         public GridBroker(String name, SchedulingPlugin schedulingPlugin, ExecutionTimeEstimationPlugin execTimeEstimationPlugin, TaskQueueList queues) throws Exception {
                 super(name, "BROKER",  schedulingPlugin, execTimeEstimationPlugin, queues);
-                
-                //make use of plug-in interface
-                
-                //Properties prop = new Properties();
-                //prop.put("plugin.name", name);
-                //prop.put("plugin.utils.timeoperations", "gssim.scheduling.plugin.local.GssimTimeOperations");
-                //schedulingPlugin.init(prop);
+
                 otherGridSchedulersIds = new HashSet<Integer>();
-                
                 moduleList = new ModuleListImpl(2);
-                //this.moduleList.add(new GridResourceDiscovery(this.getScheduler()));
-                //moduleList.add(new GridReservationManagerNew(this));
-                
-                if(log.isDebugEnabled())
-                        log.debug(name + ": Creating a broker interface object");
         }
 
         public void init(Scheduler scheduler, ManagedResources managedResources) {
                 super.init(scheduler, managedResources);
-                //this.scheduler = scheduler;
-                //this.resourceManager = ResourceManagerFactory.createResourceManager(scheduler, managedResources);
                 this.moduleList.add((GridResourceDiscovery)resourceManager);
         }
@@ -57 +42 @@
                 }
                 return providerIds;
-                //return GridSim.getGridResourceList();
         }