Changeset 481 for DCWoRMS/trunk/src/schedframe/scheduling
- Timestamp:
- 10/08/12 10:23:45 (13 years ago)
- Location:
- DCWoRMS/trunk/src/schedframe/scheduling
- Files:
-
- 1 deleted
- 22 edited
- 2 copied
Legend:
- Unmodified
- Added
- Removed
-
DCWoRMS/trunk/src/schedframe/scheduling/Cluster.java
r477 r481 13 13 import gridsim.GridSimTags; 14 14 import gridsim.IO_data; 15 import gridsim.gssim. WormsTags;15 import gridsim.gssim.DCWormsTags; 16 16 17 17 public class Cluster extends Scheduler{ … … 23 23 protected void processOtherRequest(Sim_event ev) { 24 24 switch (ev.get_tag()) { 25 case WormsTags.QUERY_RESOURCE_DESC:25 case DCWormsTags.QUERY_RESOURCE_DESC: 26 26 SchedulerDescription desc = new SchedulerDescription(new LocalSystem(get_name(), null, null)); 27 27 Map<ResourceUnitName, List<ResourceUnit>> units = managementSystem.getResourceManager().getSharedResourceUnits(); … … 30 30 31 31 IO_data data = new IO_data(desc, 0, ev.get_src()); 32 send(ev.get_src(), GridSimTags.SCHEDULE_NOW, WormsTags.QUERY_RESOURCE_DESC_RESULT, data);32 send(ev.get_src(), GridSimTags.SCHEDULE_NOW, DCWormsTags.QUERY_RESOURCE_DESC_RESULT, data); 33 33 break; 34 34 -
DCWoRMS/trunk/src/schedframe/scheduling/GridResourceDiscovery.java
r477 r481 5 5 import gridsim.GridSimTags; 6 6 import gridsim.IO_data; 7 import gridsim.gssim. WormsTags;7 import gridsim.gssim.DCWormsTags; 8 8 9 9 import java.util.ArrayList; … … 58 58 for(int i = 0; i < resourceList.size(); i++){ 59 59 int resourceId = resourceList.get(i).get_id(); 60 gridBroker.send(resourceId, GridSimTags.SCHEDULE_NOW, WormsTags.QUERY_RESOURCE_DESC, null);60 gridBroker.send(resourceId, GridSimTags.SCHEDULE_NOW, DCWormsTags.QUERY_RESOURCE_DESC, null); 61 61 } 62 62 63 63 //filter only the query response messages 64 Sim_type_p pred = new Sim_type_p( WormsTags.QUERY_RESOURCE_DESC_RESULT);64 Sim_type_p pred = new Sim_type_p(DCWormsTags.QUERY_RESOURCE_DESC_RESULT); 65 65 Sim_event ev = new Sim_event(); 66 66 … … 80 80 for(int i = 0; i < gridBroker.getChildren().size(); i++){ 81 81 int resourceId = gridBroker.getChildren().get(i).get_id(); 82 gridBroker.send(resourceId, GridSimTags.SCHEDULE_NOW, WormsTags.QUERY_RESOURCE_DESC, null);82 gridBroker.send(resourceId, GridSimTags.SCHEDULE_NOW, DCWormsTags.QUERY_RESOURCE_DESC, null); 83 83 } 84 84 85 85 //filter only the query response messages 86 Sim_type_p pred = new Sim_type_p( WormsTags.QUERY_RESOURCE_DESC_RESULT);86 Sim_type_p pred = new Sim_type_p(DCWormsTags.QUERY_RESOURCE_DESC_RESULT); 87 87 Sim_event ev = new Sim_event(); 88 88 -
DCWoRMS/trunk/src/schedframe/scheduling/Scheduler.java
r478 r481 8 8 import gridsim.GridSimTags; 9 9 import gridsim.IO_data; 10 import gridsim.gssim. WormsTags;10 import gridsim.gssim.DCWormsTags; 11 11 12 12 import java.util.ArrayList; … … 28 28 import schedframe.scheduling.manager.resources.ManagedResources; 29 29 import schedframe.scheduling.policy.AbstractManagementSystem; 30 import schedframe.scheduling.queue.QueueDescription; 30 31 import schedframe.scheduling.queue.TaskQueue; 31 import schedframe.scheduling.queue.QueueDescription;32 32 import schedframe.scheduling.tasks.WorkloadUnit; 33 33 … … 117 117 if (obj != null) { 118 118 int delay = (Integer) obj; 119 send(this.get_id(), delay, WormsTags.TIMER);119 send(this.get_id(), delay, DCWormsTags.TIMER); 120 120 } 121 121 } … … 174 174 protected void processOtherRequest(Sim_event ev) { 175 175 switch (ev.get_tag()) { 176 case WormsTags.QUERY_RESOURCE_DESC:176 case DCWormsTags.QUERY_RESOURCE_DESC: 177 177 SchedulerDescription desc = new SchedulerDescription(new LocalSystem(get_name(), null, null)); 178 178 Map<ResourceUnitName, List<ResourceUnit>> units = managementSystem.getResourceManager().getSharedResourceUnits(); … … 181 181 182 182 IO_data data = new IO_data(desc, 0, ev.get_src()); 183 send(ev.get_src(), GridSimTags.SCHEDULE_NOW, WormsTags.QUERY_RESOURCE_DESC_RESULT, data);183 send(ev.get_src(), GridSimTags.SCHEDULE_NOW, DCWormsTags.QUERY_RESOURCE_DESC_RESULT, data); 184 184 break; 185 185 … … 271 271 public List<QueueDescription> getQueuesDescription(){ 272 272 List<QueueDescription> queues = new ArrayList<QueueDescription>(); 273 for(TaskQueue queue: managementSystem.get AccessQueues()){273 for(TaskQueue queue: managementSystem.getQueues()){ 274 274 QueueDescription qd; 275 275 try { -
DCWoRMS/trunk/src/schedframe/scheduling/TaskList.java
r478 r481 3 3 import java.util.List; 4 4 5 import schedframe.scheduling.tasks. WorkloadUnit;5 import schedframe.scheduling.tasks.TaskInterface; 6 6 7 7 8 public interface WorkloadUnitList extends List<WorkloadUnit>{8 public interface TaskList extends List<TaskInterface<?>>{ 9 9 10 10 } -
DCWoRMS/trunk/src/schedframe/scheduling/TaskListImpl.java
r478 r481 3 3 import java.util.ArrayList; 4 4 5 import schedframe.scheduling.tasks. WorkloadUnit;5 import schedframe.scheduling.tasks.TaskInterface; 6 6 7 7 8 public class WorkloadUnitListImpl extends ArrayList<WorkloadUnit> implements WorkloadUnitList {8 public class TaskListImpl extends ArrayList<TaskInterface<?>> implements TaskList { 9 9 10 10 private static final long serialVersionUID = -3824600938144742457L; 11 11 12 public WorkloadUnitListImpl(){12 public TaskListImpl(){ 13 13 super(); 14 14 } 15 15 16 public WorkloadUnitListImpl(int initialSize){16 public TaskListImpl(int initialSize){ 17 17 super(initialSize); 18 18 } -
DCWoRMS/trunk/src/schedframe/scheduling/WorkloadUnitHandler.java
r477 r481 2 2 3 3 import schedframe.scheduling.tasks.Job; 4 import schedframe.scheduling.tasks.SubmittedTask; 4 5 5 import schedframe.scheduling.tasks.TaskInterface; 6 6 import gssim.schedframe.scheduling.ExecTask; … … 13 13 14 14 public void handleExecutable(ExecTask task); 15 16 public void handleSubmittedTask(SubmittedTask task); 15 17 16 } -
DCWoRMS/trunk/src/schedframe/scheduling/manager/tasks/AbstractJobRegistry.java
r480 r481 13 13 14 14 15 public abstract class AbstractJobRegistry /*extends ConcurrentHashMap<String, Job> */implements JobRegistry, Cloneable{15 public abstract class AbstractJobRegistry /*extends ConcurrentHashMap<String, Job>*/ implements JobRegistry, Cloneable{ 16 16 17 17 private static final long serialVersionUID = 8409060063583755824L; 18 18 19 19 private static Log log = LogFactory.getLog(AbstractJobRegistry.class); 20 21 protected static final ConcurrentHashMap<String, Job> jobs = new ConcurrentHashMap<String, Job>(); 20 protected static final ConcurrentHashMap<String, JobInterface<?>> jobs = new ConcurrentHashMap<String, JobInterface<?>>(); 22 21 23 22 protected AbstractJobRegistry(){ 24 //log.warn("Methods from JobRegistry interface are not implemented.");25 23 } 26 24 27 25 public boolean addJob(JobInterface<?> job) { 28 jobs.put(job.getId(), (Job)job);26 jobs.put(job.getId(), job); 29 27 return true; 30 28 } … … 32 30 public boolean addTask(TaskInterface<?> task) { 33 31 if(jobs.containsKey(task.getJobId())){ 34 jobs.get(task.getJobId()).add((Task)task);32 getJob(task.getJobId()).add((Task)task); 35 33 return true; 36 34 } else { … … 39 37 } 40 38 41 public Job getJob(String jobId){39 public JobInterface<?> getJobInfo(String jobId) { 42 40 return jobs.get(jobId); 43 41 } 44 45 public JobInterface<?> getJobInfo(String jobID) {46 return jobs.get(jobID);47 }48 42 49 public TaskInterface<?> getTaskInfo(String jobI D, String taskId) {43 public TaskInterface<?> getTaskInfo(String jobId, String taskId) { 50 44 Task task = null; 51 Job job = jobs.get(jobID);45 Job job = getJob(jobId); 52 46 53 47 if(job == null) … … 57 51 task = job.getTask(taskId); 58 52 } catch (NoSuchFieldException e) { 59 log.error(e.getMessage());60 53 } 61 54 return task; 62 55 } 63 56 64 /*public List<JobInterface<?>> getActiveJobs() { 65 log.error("getActiveJobs() not implemented."); 66 return null; 57 public Job getJob(String jobId){ 58 return (Job)jobs.get(jobId); 67 59 } 68 60 69 public List<TaskInterface<?>> getActiveTasks() {70 log.error("getActiveTasks() not implemented.");71 return null;72 }*/73 74 61 } -
DCWoRMS/trunk/src/schedframe/scheduling/manager/tasks/JobRegistry.java
r477 r481 5 5 import java.util.List; 6 6 7 import schedframe.ExecutablesList; 7 8 import schedframe.scheduling.tasks.JobInterface; 8 9 import schedframe.scheduling.tasks.TaskInterface; … … 11 12 public interface JobRegistry { 12 13 13 //public List<JobInterface<?>> getActiveJobs();14 15 //public List<TaskInterface<?>> getActiveTasks();16 17 14 public JobInterface<?> getJobInfo(String jobId); 18 15 … … 31 28 32 29 33 //public List<SubmittedTask> getSubmittedTasks();30 public ExecutablesList getExecutableTasks(); 34 31 35 public ExecTask get SubmittedTask(String jobId, String taskId);32 public ExecTask getExecutable(String jobId, String taskId); 36 33 37 34 38 public List<? extends TaskInterface<?>> get ReadyTasks(List<JobInterface<?>> jobsList);35 public List<? extends TaskInterface<?>> getAvailableTasks(List<JobInterface<?>> jobsList); 39 36 40 37 } -
DCWoRMS/trunk/src/schedframe/scheduling/manager/tasks/JobRegistryImpl.java
r477 r481 1 1 package schedframe.scheduling.manager.tasks; 2 2 3 import gridsim. Gridlet;3 import gridsim.gssim.DCWormsTags; 4 4 import gssim.schedframe.scheduling.ExecTask; 5 import gssim.schedframe.scheduling.Executable;6 5 7 6 import java.util.ArrayList; 8 import java.util.Collections;9 import java.util.HashMap;10 7 import java.util.List; 11 import java.util.Map;12 8 13 9 import org.apache.commons.lang.ArrayUtils; 14 10 import org.apache.commons.logging.Log; 15 11 import org.apache.commons.logging.LogFactory; 16 import org.joda.time.DateTime;17 12 import org.qcg.broker.schemas.resreqs.ParentType; 18 13 import org.qcg.broker.schemas.resreqs.types.TaskStatesName; 19 14 20 15 import qcg.shared.constants.BrokerConstants; 21 import schedframe.resources.units.ResourceUnit; 22 import schedframe.resources.units.ResourceUnitName; 23 import schedframe.scheduling.ResourceHistoryItem; 24 import schedframe.scheduling.plan.AllocationInterface; 25 import schedframe.scheduling.tasks.AbstractProcesses; 16 import schedframe.ExecutablesList; 26 17 import schedframe.scheduling.tasks.JobInterface; 27 import schedframe.scheduling.tasks.SubmittedTask;28 18 import schedframe.scheduling.tasks.Task; 29 import simulator.WormsConstants;30 19 31 20 public class JobRegistryImpl extends AbstractJobRegistry { … … 37 26 private String context; 38 27 39 //TO DO - change data structure 40 protected static final List<ExecTask> submittedTasks = Collections.synchronizedList(new ArrayList<ExecTask>());; 41 //protected static final List<ExecTaskInterface> submittedTasks = new CopyOnWriteArrayList<ExecTaskInterface>(); 28 //TO DO - consider data structure 29 protected static final ExecutablesList executables = new ExecutablesList(); 30 //protected static final List<ExecTask> executables = Collections.synchronizedList(new ArrayList<ExecTask>());; 31 //protected static final List<ExecTaskInterface> executables = new CopyOnWriteArrayList<ExecTaskInterface>(); 42 32 43 public JobRegistryImpl(String context _) {44 context = context_;33 public JobRegistryImpl(String context) { 34 this.context = context; 45 35 } 46 36 47 /*protected void setContext(String context_) { 48 context = context_; 49 }*/ 50 51 public boolean addTask(ExecTask newTask) { 52 if(getSubmittedTask(newTask.getJobId(), newTask.getId()) == null) 53 { 54 synchronized (submittedTasks) { 55 submittedTasks.add(newTask); 37 public boolean addExecTask(ExecTask newTask) { 38 if(getExecutable(newTask.getJobId(), newTask.getId()) == null) { 39 synchronized (executables) { 40 executables.add(newTask); 56 41 } 57 42 return true; … … 60 45 } 61 46 47 public ExecutablesList getExecutableTasks() { 48 return executables; 49 } 62 50 public List<ExecTask> getTasks(int status) { 63 51 List<ExecTask> taskList = new ArrayList<ExecTask>(); 64 synchronized ( submittedTasks) {65 for (ExecTask task: submittedTasks) {52 synchronized (executables) { 53 for (ExecTask task: executables) { 66 54 if (task.getStatus() == status) { 67 //SubmittedTask subTask = (SubmittedTask) task;68 55 List<String> visitedResource = task.getVisitedResources(); 69 if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)) {56 if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)) { 70 57 taskList.add(task); 71 58 } 72 /*if(subTask.getVisitedResources().contains(context)){73 taskList.add(subTask);74 }*/75 59 } 76 60 } … … 80 64 81 65 public List<ExecTask> getQueuedTasks() { 82 return getTasks( Gridlet.QUEUED);66 return getTasks(DCWormsTags.QUEUED); 83 67 } 84 68 85 69 public List<ExecTask> getRunningTasks() { 86 return getTasks( Gridlet.INEXEC);70 return getTasks(DCWormsTags.INEXEC); 87 71 } 88 72 89 73 public List<ExecTask> getReadyTasks() { 90 return getTasks( Gridlet.READY);74 return getTasks(DCWormsTags.READY); 91 75 } 92 76 93 77 public List<ExecTask> getFinishedTasks() { 94 return getTasks( Gridlet.SUCCESS);78 return getTasks(DCWormsTags.SUCCESS); 95 79 } 96 80 97 98 public List<ExecTask> getAllSubmittedTasks() { 99 List<ExecTask> taskList; 100 synchronized (submittedTasks) { 101 taskList = new ArrayList<ExecTask>(submittedTasks); 102 } 103 return taskList; 104 } 105 106 public List<SubmittedTask> getSubmittedTasks() { 107 List<SubmittedTask> taskList = new ArrayList<SubmittedTask>(); 108 synchronized (submittedTasks) { 109 for (ExecTask task : submittedTasks) { 110 SubmittedTask subTask = (SubmittedTask) task; 111 List<String> visitedResource = subTask.getVisitedResources(); 112 if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)){ 113 taskList.add(subTask); 114 } 115 /*if(subTask.getVisitedResources().contains(context)){ 116 taskList.add(subTask); 117 }*/ 118 } 119 } 120 return taskList; 121 } 122 123 public ExecTask getSubmittedTask(String jobId, String taskId){ 124 synchronized (submittedTasks) { 125 for (ExecTask task : submittedTasks) { 81 public ExecTask getExecutable(String jobId, String taskId){ 82 synchronized (executables) { 83 for (ExecTask task : executables) { 126 84 if (task.getJobId().compareTo(jobId) == 0 && task.getId().compareTo(taskId)==0) { 127 85 return task; … … 131 89 return null; 132 90 } 133 134 91 135 92 @SuppressWarnings("unchecked") 136 public List<Task> get ReadyTasks(List<JobInterface<?>> wuList) {137 List<Task> readyTasks = new ArrayList<Task>();93 public List<Task> getAvailableTasks(List<JobInterface<?>> wuList) { 94 List<Task> availableTasks = new ArrayList<Task>(); 138 95 List<Task> waitingTasks = new ArrayList<Task>(); 139 96 … … 143 100 } 144 101 145 readyTasks.addAll(getPrecedenceConstrainedReadyTasks(waitingTasks)); 146 return readyTasks; 147 } 148 149 public Executable getTaskExecutable(Integer executableId){ 150 synchronized (submittedTasks) { 151 for (ExecTask task : submittedTasks) { 152 SubmittedTask subTask = (SubmittedTask) task; 153 Executable exec = (Executable)subTask.getGridlet(); 154 if (exec.getGridletID() == executableId) { 155 return exec; 156 } 157 } 158 } 159 return null; 160 } 161 162 public List<Executable> getJobExecutables(String jobId){ 163 List<Executable> list = new ArrayList<Executable>(); 164 synchronized (submittedTasks) { 165 for(int i = 0; i < submittedTasks.size(); i++){ 166 SubmittedTask subTask = (SubmittedTask) submittedTasks.get(i); 167 Executable exec = (Executable)subTask.getGridlet(); 168 169 if(exec.getJobId().equals(jobId)) 170 list.add(exec); 171 } 172 } 173 return list; 174 } 175 176 public JobRegistryImpl clone() { 177 JobRegistryImpl jr = null; 178 try { 179 jr = (JobRegistryImpl) super.clone(); 180 } catch (CloneNotSupportedException e) { 181 // TODO Auto-generated catch block 182 e.printStackTrace(); 183 } 184 185 return jr; 186 } 187 188 /*public AbstractExecutable getTaskExecutabls(String jobId, String taskId){ 189 List<AbstractExecutable> list = new ArrayList<AbstractExecutable>(); 190 synchronized (submittedTasks) { 191 for(int i = 0; i < size(); i++){ 192 SubmittedTask subTask = (SubmittedTask) submittedTasks.get(i); 193 AbstractExecutable exec = (AbstractExecutable)subTask.getGridlet(); 194 195 if(exec.getJobId().equals(jobId) && exec.getId().equals(taskId)) 196 return exec; 197 } 198 } 199 return null; 200 }*/ 201 202 203 public Executable createExecutable(Task task, AllocationInterface allocation) { 204 205 String refersTo = allocation.getProcessGroupId(); // null;//allocation.getRefersTo(); 206 if(refersTo == null) 207 refersTo = task.getId(); 208 209 Executable exec = null; 210 211 if(refersTo.equals(task.getId())){ 212 exec = new Executable(task); 213 } else { 214 List<AbstractProcesses> processes = task.getProcesses(); 215 if(processes == null) { 216 try { 217 log.error("Allocation: " + allocation.getDocument() + "\nrefers to unknown task or processes set." + 218 " Set correct value (task id or prcesses set id) for allocation refersTo attribute."); 219 } catch (Exception e) { 220 e.printStackTrace(); 221 } 222 } 223 boolean found = false; 224 for(int j = 0; j < processes.size() && !found; j++){ 225 AbstractProcesses procesesSet = processes.get(j); 226 if(refersTo.equals(procesesSet.getId())){ 227 exec = new Executable(task, procesesSet); 228 found = true; 229 } 230 } 231 if(!found){ 232 log.error("Allocation refers to unknown proceses set."); 233 } 234 } 235 236 // exec.setUserID(task.getSenderId()); 237 exec.setLength(task.getLength()); 238 exec.setReservationId(allocation.getReservationId()); 239 240 /*HostInterface<?> host = allocation.getHost(); 241 ComputingResourceTypeInterface<?> crt = host.getMachineParameters(); 242 if(crt != null){ 243 ComputingResourceTypeItemInterface<?> crti = crt.getComputingResourceTypeItem(0); 244 if(crti != null){ 245 ParameterPropertyInterface<?> properties[] = crti.getHostParameter().getProperty(); 246 for(int p = 0; p < properties.length; p++){ 247 ParameterPropertyInterface<?> property = properties[p]; 248 if("chosenCPUs".equals(property.getName())){ 249 Object cpuNames = property.getValue(); 250 exec.addSpecificResource(ResourceParameterName.FREECPUS, cpuNames); 251 } 252 } 253 } 254 }*/ 255 return exec; 256 } 257 258 259 public List<Executable> createExecutables(Task task) { 260 261 List<AbstractProcesses> processes = task.getProcesses(); 262 263 List<Executable> executables = new ArrayList<Executable>(); 264 265 if(processes == null || processes.size()==0){ 266 Executable exec = new Executable(task); 267 exec.setUserID(task.getSenderId()); 268 exec.setLength(task.getLength()); 269 executables.add(exec); 270 } else { 271 272 boolean found = false; 273 for(int j = 0; j < processes.size() && !found; j++){ 274 AbstractProcesses procesesSet = processes.get(j); 275 Executable exec = new Executable(task, procesesSet); 276 exec.setUserID(task.getSenderId()); 277 exec.setLength(task.getLength()); 278 executables.add(exec); 279 } 280 } 281 282 return executables; 102 availableTasks.addAll(getPrecedenceConstrainedAvailableTasks(waitingTasks)); 103 return availableTasks; 283 104 } 284 105 285 106 286 /**************************************/ 287 protected static Map<Integer, Map<String, Object>> history = new HashMap<Integer, Map<String,Object>>(); 288 289 public static Map<Integer, Map<String, Object>> getAllocationHistory(){ 290 return history; 291 } 292 293 public void saveHistory (SubmittedTask submittedTask, int estimatedTime, Map<ResourceUnitName, ResourceUnit> choosenResources){ 107 private List<Task> getPrecedenceConstrainedAvailableTasks(List<Task> tasks){ 294 108 295 /*submittedTask.setEstimatedDuration(estimatedTime); 296 297 DateTime currentTime = new DateTime(); 298 ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime); 299 submittedTask.addUsedResources(resHistItem);*/ 300 301 ResourceHistoryItem resHistItem = submittedTask.getUsedResources().getLast(); 302 DateTime currentTime = new DateTime(); 303 Map<String, Object> historyItem = new HashMap<String, Object>(); 304 List<ResourceHistoryItem> list = new ArrayList<ResourceHistoryItem>(1); 305 list.add(resHistItem); 306 historyItem.put(WormsConstants.RESOURCES, list); 307 historyItem.put(WormsConstants.START_TIME, currentTime); 308 currentTime = currentTime.plusSeconds(estimatedTime); 309 historyItem.put(WormsConstants.END_TIME, currentTime); 310 311 history.put(Integer.valueOf(submittedTask.getGridletID()), historyItem); 312 /*ProcessingElements pes = (ProcessingElements) choosenResources.get(ResourceParameterName.PROCESSINGELEMENTS); 313 for (ComputingResource resource : pes) { 314 //submittedTask.addToResPath(resource.getName()); 315 submittedTask.visitResource(resource.getName()); 316 ComputingResource parent = resource.getParent(); 317 while (parent != null && !submittedTask.getResPath().contains(parent.getName() + "_")) { 318 submittedTask.addToResPath(parent.getName()); 319 parent = parent.getParent(); 320 } 321 while (parent != null && !submittedTask.getVisitedResources().contains(parent.getName() + "_")) { 322 submittedTask.visitResource(parent.getName()); 323 parent = parent.getParent(); 324 } 325 }*/ 326 } 327 328 private List<Task> getPrecedenceConstrainedReadyTasks(List<Task> tasks){ 109 List<Task> availableTasks = new ArrayList<Task>(); 110 int size = tasks.size(); 329 111 330 List<Task> readyTasks = new ArrayList<Task>();331 332 int size = tasks.size();333 112 for(int i = 0; i < size; i++){ 334 113 int parCnt; 335 int previousTask ReadyCnt = 0;114 int previousTaskSucceedCnt = 0; 336 115 Task task = tasks.get(i); 337 116 if(task.getStatus() != (int)BrokerConstants.TASK_STATUS_UNSUBMITTED) … … 341 120 } catch(Exception e){ 342 121 parCnt = 0; 343 //e.printStackTrace();344 122 } 345 if(parCnt == 0) 346 { 347 readyTasks.add(task); 123 if(parCnt == 0){ 124 availableTasks.add(task); 348 125 } 349 else 350 { 126 else { 351 127 for(int j = 0; j < parCnt; j++){ 352 128 ParentType par = task.getDescription().getWorkflow().getParent(j); … … 356 132 } 357 133 } 358 previousTask ReadyCnt++;134 previousTaskSucceedCnt++; 359 135 } 360 136 361 if(previousTask ReadyCnt == parCnt && task.getDescription().getWorkflow().getAnd() != null)362 readyTasks.add(task);363 else if(previousTask ReadyCnt > 0 && task.getDescription().getWorkflow().getOr() != null)364 readyTasks.add(task);365 else if (previousTask ReadyCnt == parCnt)366 readyTasks.add(task);137 if(previousTaskSucceedCnt == parCnt && task.getDescription().getWorkflow().getAnd() != null) 138 availableTasks.add(task); 139 else if(previousTaskSucceedCnt > 0 && task.getDescription().getWorkflow().getOr() != null) 140 availableTasks.add(task); 141 else if (previousTaskSucceedCnt == parCnt) 142 availableTasks.add(task); 367 143 } 368 144 } 369 return readyTasks;145 return availableTasks; 370 146 } 371 147 … … 380 156 return false; 381 157 } 158 382 159 } -
DCWoRMS/trunk/src/schedframe/scheduling/plan/ScheduledTaskInterface.java
r478 r481 4 4 5 5 import schedframe.DescriptionContainer; 6 import schedframe.scheduling.tasks. WorkloadUnit;6 import schedframe.scheduling.tasks.TaskInterface; 7 7 8 8 public interface ScheduledTaskInterface<T> extends DescriptionContainer<T> { … … 190 190 191 191 192 public ArrayList<AllocationInterface > getAllocations();192 public ArrayList<AllocationInterface<?>> getAllocations(); 193 193 194 public WorkloadUnitgetTask();194 public TaskInterface<?> getTask(); 195 195 196 196 } -
DCWoRMS/trunk/src/schedframe/scheduling/plan/SchedulingPlanInterface.java
r477 r481 99 99 100 100 101 public ArrayList<ScheduledTaskInterface > getTasks();101 public ArrayList<ScheduledTaskInterface<?>> getTasks(); 102 102 } -
DCWoRMS/trunk/src/schedframe/scheduling/plan/impl/ScheduledTask.java
r478 r481 6 6 import org.exolab.castor.xml.MarshalException; 7 7 import org.exolab.castor.xml.ValidationException; 8 import org.qcg.broker.schemas.schedulingplan.types.AllocationStatus; 8 9 9 10 import schedframe.scheduling.plan.AllocationInterface; 10 11 import schedframe.scheduling.plan.ScheduledTaskInterface; 11 12 import schedframe.scheduling.plan.ScheduledTimeInterface; 12 import schedframe.scheduling.tasks.WorkloadUnit; 13 14 15 import org.qcg.broker.schemas.schedulingplan.types.AllocationStatus; 13 import schedframe.scheduling.tasks.TaskInterface; 16 14 17 15 public class ScheduledTask implements ScheduledTaskInterface<org.qcg.broker.schemas.schedulingplan.Task> { … … 22 20 public ScheduledTask(){ 23 21 t = new org.qcg.broker.schemas.schedulingplan.Task(); 24 allocationList = new ArrayList<AllocationInterface >();22 allocationList = new ArrayList<AllocationInterface<?>>(); 25 23 } 26 24 27 25 public ScheduledTask(org.qcg.broker.schemas.schedulingplan.Task value){ 28 26 t = value; 29 allocationList = new ArrayList<AllocationInterface >();27 allocationList = new ArrayList<AllocationInterface<?>>(); 30 28 } 31 29 … … 178 176 179 177 180 protected WorkloadUnittask;181 protected ArrayList<AllocationInterface > allocationList;178 protected TaskInterface<?> task; 179 protected ArrayList<AllocationInterface<?>> allocationList; 182 180 183 public ScheduledTask( WorkloadUnittask){181 public ScheduledTask(TaskInterface<?> task){ 184 182 this(); 185 183 this.task = task; 186 184 } 187 185 188 public ArrayList<AllocationInterface > getAllocations() {186 public ArrayList<AllocationInterface<?>> getAllocations() { 189 187 return this.allocationList; 190 188 } 191 189 192 public WorkloadUnitgetTask(){190 public TaskInterface<?> getTask(){ 193 191 return this.task; 194 192 } -
DCWoRMS/trunk/src/schedframe/scheduling/plan/impl/SchedulingPlan.java
r477 r481 18 18 public SchedulingPlan(){ 19 19 sp = new org.qcg.broker.schemas.schedulingplan.SchedulingPlan(); 20 taskList = new ArrayList<ScheduledTaskInterface >();20 taskList = new ArrayList<ScheduledTaskInterface<?>>(); 21 21 } 22 22 23 23 public SchedulingPlan(org.qcg.broker.schemas.schedulingplan.SchedulingPlan value){ 24 24 sp = value; 25 taskList = new ArrayList<ScheduledTaskInterface >();25 taskList = new ArrayList<ScheduledTaskInterface<?>>(); 26 26 } 27 27 … … 106 106 107 107 108 protected ArrayList<ScheduledTaskInterface > taskList;108 protected ArrayList<ScheduledTaskInterface<?>> taskList; 109 109 110 public ArrayList<ScheduledTaskInterface > getTasks() {110 public ArrayList<ScheduledTaskInterface<?>> getTasks() { 111 111 112 112 return this.taskList; -
DCWoRMS/trunk/src/schedframe/scheduling/plugin/SchedulingPlugin.java
r477 r481 3 3 import schedframe.Plugin; 4 4 import schedframe.events.scheduling.SchedulingEvent; 5 import schedframe.scheduling. WorkloadUnitListImpl;5 import schedframe.scheduling.TaskListImpl; 6 6 import schedframe.scheduling.manager.resources.ResourceManager; 7 7 import schedframe.scheduling.manager.tasks.JobRegistry; … … 12 12 public interface SchedulingPlugin extends Plugin{ 13 13 14 public int place JobsInQueues(WorkloadUnitListImpl newJobs,14 public int placeTasksInQueues(TaskListImpl newTasks, 15 15 TaskQueueList queues, 16 16 ResourceManager resourceManager, ModuleList modules); -
DCWoRMS/trunk/src/schedframe/scheduling/policy/AbstractManagementSystem.java
r480 r481 10 10 import schedframe.PluginConfiguration; 11 11 import schedframe.events.scheduling.SchedulingEventType; 12 import schedframe.resources.units.StandardResourceUnitName;13 12 import schedframe.scheduling.Scheduler; 14 13 import schedframe.scheduling.WorkloadUnitHandler; … … 16 15 import schedframe.scheduling.manager.resources.ResourceManager; 17 16 import schedframe.scheduling.manager.resources.ResourceManagerFactory; 17 import schedframe.scheduling.manager.tasks.JobRegistry; 18 18 import schedframe.scheduling.manager.tasks.JobRegistryImpl; 19 import schedframe.scheduling.manager.tasks.JobRegistry;20 19 import schedframe.scheduling.plan.AllocationInterface; 21 20 import schedframe.scheduling.plan.SchedulingPlanInterface; … … 27 26 import schedframe.scheduling.queue.TaskQueueList; 28 27 import schedframe.scheduling.tasks.Job; 28 import schedframe.scheduling.tasks.TaskInterface; 29 29 import schedframe.scheduling.tasks.WorkloadUnit; 30 import simulator. WormsConstants;30 import simulator.DCWormsConstants; 31 31 import eduni.simjava.Sim_event; 32 32 import gridsim.GridSim; 33 33 import gridsim.GridSimTags; 34 import gridsim.Gridlet;35 34 import gridsim.IO_data; 36 import gridsim.gssim. WormsTags;35 import gridsim.gssim.DCWormsTags; 37 36 import gssim.schedframe.scheduling.ExecTask; 38 37 import gssim.schedframe.scheduling.Executable; … … 45 44 protected String name; 46 45 46 protected TaskQueueList queues; 47 47 protected ResourceManager resourceManager; 48 49 protected TaskQueueList queues;50 48 protected JobRegistryImpl jobRegistry; 49 protected ModuleList moduleList; 50 51 51 protected SchedulingPlugin schedulingPlugin; 52 53 52 protected ExecutionTimeEstimationPlugin execTimeEstimationPlugin; 54 53 55 protected ModuleList moduleList; 56 57 protected JobRegistryImpl jobRegistry; 54 protected Scheduler scheduler; 58 55 59 56 … … 67 64 } 68 65 66 public void init(Scheduler sched, ManagedResources managedResources) { 67 scheduler = sched; 68 resourceManager = ResourceManagerFactory.createResourceManager(scheduler, managedResources); 69 scheduler.set_stat(DCWormsConstants.getResourcesStatisticsObject(queues.size())); 70 for(int i = 0; i < queues.size(); i++){ 71 TaskQueue q = queues.get(i); 72 if(q instanceof AbstractStatsSupportingQueue<?>){ 73 AbstractStatsSupportingQueue<?> queue = (AbstractStatsSupportingQueue<?>) q; 74 queue.setStats(scheduler.get_stat(), DCWormsConstants.TASKS_QUEUE_LENGTH_MEASURE_NAME + "_" + Integer.toString(i)); 75 } 76 } 77 } 78 69 79 public void processEvent(Sim_event ev) { 70 80 processOtherEvent(ev); … … 81 91 } 82 92 93 83 94 public String getName() { 84 95 return name; 85 }86 87 public PluginConfiguration getSchedulingPluginConfiguration() {88 return schedulingPlugin.getConfiguration();89 96 } 90 97 … … 106 113 return jobRegistry; 107 114 } 115 116 public Scheduler getScheduler() { 117 return scheduler; 118 } 119 120 public PluginConfiguration getSchedulingPluginConfiguration() { 121 return schedulingPlugin.getConfiguration(); 122 } 108 123 109 124 public boolean pluginSupportsEvent(int eventType){ … … 111 126 } 112 127 113 public abstract void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack); 114 115 public abstract void notifyCanceledWorkloadUnit(WorkloadUnit wu); 116 117 public abstract void notifyReturnedWorkloadUnit(WorkloadUnit wu); 118 119 protected abstract void executeSchedulingPlan(SchedulingPlanInterface decision); 120 121 122 128 public TaskQueueList getQueues(){ 129 return queues; 130 } 131 132 public Map<String, Integer> getQueuesSize() { 133 Map<String, Integer> queue_size = new HashMap<String, Integer>(); 134 for (TaskQueue queue : queues) { 135 queue_size.put(queue.getName(), queue.size()); 136 } 137 return queue_size; 138 } 139 123 140 //POPRAWIC (ale co? bo teraz chyba jest ok) 124 protected void submit WorkloadUnit(WorkloadUnit wu, AllocationInterfaceallocation) {141 protected void submitTask(TaskInterface<?> task, AllocationInterface<?> allocation) { 125 142 String providerName = allocation.getProviderName(); 126 143 if (providerName == null) { … … 128 145 } 129 146 //Executable exec = (Executable) wu; 130 removeFromQueue(wu); 131 scheduler.send(providerName, GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, wu); 132 } 133 134 protected boolean sendCanceledWorkloadUnit(int tag, Executable task, int executableId, int destId) { 135 136 if (tag != GridSimTags.GRIDLET_CANCEL) { 137 return false; 138 } 139 140 long taskSize = 0; 141 if (task != null) { 142 taskSize = task.getGridletOutputSize(); 143 } 144 145 // if no Gridlet found, then create a new Gridlet but set its status 146 // to FAILED. Then, most importantly, set the resource parameters 147 // because the user will search/filter based on a resource ID. 148 else if (task == null) { 149 try { 150 taskSize = 100; 151 task = jobRegistry.getTaskExecutable(executableId); 152 task.setGridletStatus(Gridlet.FAILED); 153 int cost = resourceManager.getSharedResourceUnits().get(StandardResourceUnitName.COST) != null ? resourceManager 154 .getSharedResourceUnits().get(StandardResourceUnitName.COST).get(0).getAmount() 155 : 1; 156 task.setResourceParameter(scheduler.get_id(), cost); 157 } catch (Exception e) { 158 // empty ... 159 } 160 } 161 scheduler.send(scheduler.getOutputPort(), GridSimTags.SCHEDULE_NOW, tag, new IO_data(task, taskSize, destId)); 162 163 return true; 147 removeFromQueue(task); 148 scheduler.send(providerName, GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, task); 164 149 } 165 150 … … 184 169 185 170 protected void sendExecutableReadyEvent(ExecTask exec) { 186 187 /*if (wu instanceof JobInterface) {188 scheduler.sendInternal(Long.valueOf(0).doubleValue(), GssimTags.TASK_READY_FOR_EXECUTION,189 wu);190 return;191 }*/192 171 193 172 long delay = 0; … … 202 181 } 203 182 204 scheduler.sendInternal(Long.valueOf(delay).doubleValue(), WormsTags.TASK_READY_FOR_EXECUTION,183 scheduler.sendInternal(Long.valueOf(delay).doubleValue(), DCWormsTags.TASK_READY_FOR_EXECUTION, 205 184 exec); 206 185 } … … 214 193 if (obj != null) { 215 194 int delay = (Integer) obj; 216 scheduler.sendInternal(delay, WormsTags.TIMER, null);195 scheduler.sendInternal(delay, DCWormsTags.TIMER, null); 217 196 } 218 197 } … … 229 208 return false; 230 209 } 231 232 public TaskQueueList getAccessQueues(){233 return queues;234 }235 236 public Map<String, Integer> getQueuesSize() {237 Map<String, Integer> queue_size = new HashMap<String, Integer>();238 for (TaskQueue queue : queues) {239 queue_size.put(queue.getName(), queue.size());240 }241 return queue_size;242 }243 244 public void init(Scheduler sched, ManagedResources managedResources) {245 scheduler = sched;246 resourceManager = ResourceManagerFactory.createResourceManager(scheduler, managedResources);247 scheduler.set_stat(WormsConstants.getResourcesStatisticsObject(queues.size()));248 for(int i = 0; i < queues.size(); i++){249 TaskQueue q = queues.get(i);250 if(q instanceof AbstractStatsSupportingQueue<?>){251 AbstractStatsSupportingQueue<?> queue = (AbstractStatsSupportingQueue<?>) q;252 queue.setStats(scheduler.get_stat(), WormsConstants.TASKS_QUEUE_LENGTH_MEASURE_NAME + "_" + Integer.toString(i));253 }254 }255 }256 257 protected Scheduler scheduler;258 259 public Scheduler getScheduler() {260 return scheduler;261 }262 210 263 211 public abstract WorkloadUnitHandler getWorkloadUnitHandler(); 212 213 public abstract void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack); 214 215 public abstract void notifyReturnedWorkloadUnit(WorkloadUnit wu); 216 217 protected abstract void executeSchedulingPlan(SchedulingPlanInterface<?> decision); 264 218 265 219 } -
DCWoRMS/trunk/src/schedframe/scheduling/policy/global/GlobalManagementSystem.java
r480 r481 12 12 13 13 import qcg.shared.constants.BrokerConstants; 14 15 import schedframe.events.scheduling.EventReason;16 14 import schedframe.events.scheduling.SchedulingEvent; 17 15 import schedframe.events.scheduling.TaskArrivedEvent; 18 import schedframe.events.scheduling.TaskCanceledEvent;19 16 import schedframe.events.scheduling.TimerEvent; 20 import schedframe.scheduling.Scheduler;21 17 import schedframe.scheduling.WorkloadUnitHandler; 22 import schedframe.scheduling. WorkloadUnitListImpl;18 import schedframe.scheduling.TaskListImpl; 23 19 import schedframe.scheduling.plan.AllocationInterface; 24 20 import schedframe.scheduling.plan.ScheduledTaskInterface; … … 31 27 import schedframe.scheduling.tasks.Job; 32 28 import schedframe.scheduling.tasks.JobInterface; 33 import schedframe.scheduling.tasks.SubmittedTask;34 29 import schedframe.scheduling.tasks.Task; 35 30 import schedframe.scheduling.tasks.TaskInterface; 36 31 import schedframe.scheduling.tasks.WorkloadUnit; 37 38 32 import eduni.simjava.Sim_event; 39 33 import gridsim.GridSim; 40 34 import gridsim.GridSimTags; 41 import gridsim.Gridlet;42 35 import gridsim.IO_data; 43 import gridsim.gssim. WormsTags;36 import gridsim.gssim.DCWormsTags; 44 37 import gssim.schedframe.scheduling.ExecTask; 45 38 import gssim.schedframe.scheduling.Executable; … … 54 47 super(providerId, entityName, execTimeEstimationPlugin, queues); 55 48 56 /*schedulingPlugin = (GlobalSchedulingPlugin) InstanceFactory.createInstance(57 schedulingPluginClassName,58 GlobalSchedulingPlugin.class);*/59 49 if(schedPlugin == null){ 60 50 throw new Exception("Can not create global scheduling plugin instance"); … … 69 59 switch (tag) { 70 60 71 case WormsTags.TIMER:72 if (pluginSupportsEvent( WormsTags.TIMER)) {61 case DCWormsTags.TIMER: 62 if (pluginSupportsEvent(DCWormsTags.TIMER)) { 73 63 TimerEvent event = new TimerEvent(); 74 SchedulingPlanInterface decision = schedulingPlugin.schedule(event,64 SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event, 75 65 queues, getJobRegistry(), getResourceManager(), moduleList); 76 66 executeSchedulingPlan(decision); … … 114 104 115 105 116 protected void schedule ReadyTasks(Job job){106 protected void scheduleAvaialbleTasks(Job job){ 117 107 List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>(); 118 108 jobsList.add(job); 119 WorkloadUnitListImpl readyWorkloadUnits = new WorkloadUnitListImpl();120 ready WorkloadUnits.addAll(jobRegistry.getReadyTasks(jobsList));121 schedulingPlugin.placeJobsInQueues(readyWorkloadUnits, queues, getResourceManager(), moduleList);122 109 TaskListImpl readyTasks = new TaskListImpl(); 110 readyTasks.addAll(jobRegistry.getAvailableTasks(jobsList)); 111 112 schedulingPlugin.placeTasksInQueues(readyTasks, queues, getResourceManager(), moduleList); 123 113 schedule(new TaskArrivedEvent()); 124 114 } … … 126 116 protected void schedule(SchedulingEvent schedulingEvent) { 127 117 128 try { 129 SchedulingPlanInterface decision = schedulingPlugin.schedule( 130 schedulingEvent, queues, getJobRegistry(), getResourceManager(), moduleList); 131 if (decision == null) 132 return; 133 118 SchedulingPlanInterface<?> decision = schedulingPlugin.schedule( 119 schedulingEvent, queues, getJobRegistry(), getResourceManager(), moduleList); 120 if (decision != null) 134 121 executeSchedulingPlan(decision); 135 136 } catch (Exception e) {137 e.printStackTrace();138 }139 122 } 140 123 … … 173 156 } 174 157 else { 175 schedule ReadyTasks(job);158 scheduleAvaialbleTasks(job); 176 159 /*List<JobInterface<?>> jobs = new ArrayList<JobInterface<?>>(); 177 160 jobs.add(jobRegistry.getJobInfo(job.getId())); … … 189 172 } 190 173 191 public void notifyCanceledWorkloadUnit(WorkloadUnit wu){; 192 193 Executable task = (Executable) wu; 194 String jobID = task.getJobId(); 195 String taskID = task.getId(); 196 197 if(log.isDebugEnabled()) 198 log.debug("Received canceled job" + jobID + "_" + taskID); 199 200 TaskInterface<?> ti = jobRegistry.getTaskInfo(jobID, taskID) ; 201 try { 202 203 ti.setStatus((int)BrokerConstants.JOB_STATUS_CANCELED); 204 205 TaskCanceledEvent event = new TaskCanceledEvent(jobID, taskID); 206 event.setReason(EventReason.RESERVATION_EXCEEDED); 207 schedule(event); 208 209 } catch (Exception e) { 210 log.error("Exception during scheduling. " + e.getMessage()); 211 e.printStackTrace(); 212 } 213 } 214 215 protected void executeSchedulingPlan(SchedulingPlanInterface decision) { 216 217 ArrayList<ScheduledTaskInterface> taskSchedulingDecisions = decision.getTasks(); 174 protected void executeSchedulingPlan(SchedulingPlanInterface<?> decision) { 175 176 ArrayList<ScheduledTaskInterface<?>> taskSchedulingDecisions = decision.getTasks(); 218 177 for (int i = 0; i < taskSchedulingDecisions.size(); i++) { 219 178 220 try { 221 ScheduledTaskInterface taskDecision = taskSchedulingDecisions.get(i); 222 223 //log.info(decision.getDocument()); 224 225 String jobID = taskDecision.getJobId(); 226 String taskID = taskDecision.getTaskId(); 227 228 // Task allocations that were rejected because of lack of resources or which were canceled and 229 // not scheduled again are returned to the user. 230 if(taskDecision.getStatus() == AllocationStatus.REJECTED){ 231 Job job = jobRegistry.getJob(jobID); 232 scheduler.send(job.getSenderId(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_RETURN, job); 233 continue; 234 } 235 236 ArrayList<AllocationInterface> allocations = taskDecision.getAllocations(); 237 238 Task task = (Task) jobRegistry.getTaskInfo(jobID, taskID); 239 for (int j = 0; j < allocations.size(); j++) { 240 241 AllocationInterface allocation = allocations.get(j); 242 Executable exec = jobRegistry.createExecutable(task, allocation); 243 submitWorkloadUnit(exec, allocation); 244 task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED); 245 } 246 247 }catch (Exception e){ 248 e.printStackTrace(); 249 } 250 } 251 } 252 253 protected void submitWorkloadUnit(WorkloadUnit job, AllocationInterface allocation) { 179 ScheduledTaskInterface<?> taskDecision = taskSchedulingDecisions.get(i); 180 181 //log.info(decision.getDocument()); 182 183 String jobID = taskDecision.getJobId(); 184 String taskID = taskDecision.getTaskId(); 185 186 // Task allocations that were rejected because of lack of resources or which were canceled and 187 // not scheduled again are returned to the user. 188 if(taskDecision.getStatus() == AllocationStatus.REJECTED){ 189 Job job = jobRegistry.getJob(jobID); 190 scheduler.send(job.getSenderId(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_RETURN, job); 191 continue; 192 } 193 194 Task task = (Task) jobRegistry.getTaskInfo(jobID, taskID); 195 196 ArrayList<AllocationInterface<?>> allocations = taskDecision.getAllocations(); 197 for (int j = 0; j < allocations.size(); j++) { 198 199 AllocationInterface<?> allocation = allocations.get(j); 200 Executable exec = createExecutable(task, allocation); 201 submitTask(exec, allocation); 202 task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED); 203 } 204 } 205 } 206 207 private Executable createExecutable(Task task, AllocationInterface<?> allocation) { 208 209 String refersTo = allocation.getProcessGroupId(); // null;//allocation.getRefersTo(); 210 if(refersTo == null) 211 refersTo = task.getId(); 212 213 Executable exec = null; 214 215 if(refersTo.equals(task.getId())){ 216 exec = new Executable(task); 217 } else { 218 List<AbstractProcesses> processes = task.getProcesses(); 219 if(processes == null) { 220 try { 221 log.error("Allocation: " + allocation.getDocument() + "\nrefers to unknown task or processes set." + 222 " Set correct value (task id or prcesses set id) for allocation refersTo attribute."); 223 } catch (Exception e) { 224 e.printStackTrace(); 225 } 226 } 227 boolean found = false; 228 for(int j = 0; j < processes.size() && !found; j++){ 229 AbstractProcesses procesesSet = processes.get(j); 230 if(refersTo.equals(procesesSet.getId())){ 231 exec = new Executable(task, procesesSet); 232 found = true; 233 } 234 } 235 if(!found){ 236 log.error("Allocation refers to unknown proceses set."); 237 } 238 } 239 240 exec.setReservationId(allocation.getReservationId()); 241 242 /*HostInterface<?> host = allocation.getHost(); 243 ComputingResourceTypeInterface<?> crt = host.getMachineParameters(); 244 if(crt != null){ 245 ComputingResourceTypeItemInterface<?> crti = crt.getComputingResourceTypeItem(0); 246 if(crti != null){ 247 ParameterPropertyInterface<?> properties[] = crti.getHostParameter().getProperty(); 248 for(int p = 0; p < properties.length; p++){ 249 ParameterPropertyInterface<?> property = properties[p]; 250 if("chosenCPUs".equals(property.getName())){ 251 Object cpuNames = property.getValue(); 252 exec.addSpecificResource(ResourceParameterName.FREECPUS, cpuNames); 253 } 254 } 255 } 256 }*/ 257 return exec; 258 } 259 260 protected void submitTask(TaskInterface<?> task, AllocationInterface<?> allocation) { 254 261 255 262 String providerName = allocation.getProviderName(); … … 257 264 return; 258 265 } 259 TaskInterface<?> task = (TaskInterface<?>) job;260 266 removeFromQueue(task); 261 267 262 268 int resID = GridSim.getEntityId(providerName); 263 IO_data data = new IO_data( job, 0, resID);269 IO_data data = new IO_data(task, 0, resID); 264 270 scheduler.send(scheduler.getOutputPort(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, data); 265 271 266 272 //scheduler.send(providerName, GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, job); 267 273 if(log.isDebugEnabled()) 268 log.debug("Submitted job " + job.getId() + " to " + providerName);274 log.debug("Submitted job " + task.getId() + " to " + providerName); 269 275 270 276 } … … 274 280 public void handleJob(Job job){ 275 281 276 jobRegistry.addJob(job);277 282 if (log.isInfoEnabled()) 278 283 log.info("Received job " + job.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis())); 279 284 280 scheduleReadyTasks(job); 285 jobRegistry.addJob(job); 286 scheduleAvaialbleTasks(job); 281 287 } 282 288 … … 289 295 } 290 296 291 public void handleSubmittedTask(SubmittedTask task) {292 throw new RuntimeException("Not implemented since it isn't expected that tasks are send directly to the global scheduler.");293 }294 297 } 295 298 … … 297 300 return new GlobalWorkloadUnitHandler(); 298 301 } 299 300 302 301 303 -
DCWoRMS/trunk/src/schedframe/scheduling/policy/global/GridBroker.java
r477 r481 19 19 public class GridBroker extends GlobalManagementSystem { 20 20 21 22 21 private static Log log = LogFactory.getLog(GridBroker.class); 23 22 … … 27 26 public GridBroker(String name, SchedulingPlugin schedulingPlugin, ExecutionTimeEstimationPlugin execTimeEstimationPlugin, TaskQueueList queues) throws Exception { 28 27 super(name, "BROKER", schedulingPlugin, execTimeEstimationPlugin, queues); 29 30 //make use of plug-in interface 31 32 //Properties prop = new Properties(); 33 //prop.put("plugin.name", name); 34 //prop.put("plugin.utils.timeoperations", "gssim.scheduling.plugin.local.GssimTimeOperations"); 35 //schedulingPlugin.init(prop); 28 36 29 otherGridSchedulersIds = new HashSet<Integer>(); 37 38 30 moduleList = new ModuleListImpl(2); 39 //this.moduleList.add(new GridResourceDiscovery(this.getScheduler()));40 //moduleList.add(new GridReservationManagerNew(this));41 42 if(log.isDebugEnabled())43 log.debug(name + ": Creating a broker interface object");44 31 } 45 32 46 33 public void init(Scheduler scheduler, ManagedResources managedResources) { 47 34 super.init(scheduler, managedResources); 48 //this.scheduler = scheduler;49 //this.resourceManager = ResourceManagerFactory.createResourceManager(scheduler, managedResources);50 35 this.moduleList.add((GridResourceDiscovery)resourceManager); 51 36 } … … 57 42 } 58 43 return providerIds; 59 //return GridSim.getGridResourceList();60 44 } 61 45 -
DCWoRMS/trunk/src/schedframe/scheduling/policy/local/LocalManagementSystem.java
r480 r481 4 4 import eduni.simjava.Sim_system; 5 5 import gridsim.Accumulator; 6 import gridsim.GridSimTags;7 import gridsim.Gridlet;8 6 import gridsim.ResourceCalendar; 9 import gridsim.gssim. WormsTags;10 import gridsim.gssim.filter. SubTaskFilter;7 import gridsim.gssim.DCWormsTags; 8 import gridsim.gssim.filter.ExecTaskFilter; 11 9 import gssim.schedframe.scheduling.ExecTask; 12 10 import gssim.schedframe.scheduling.Executable; … … 27 25 import qcg.shared.constants.BrokerConstants; 28 26 import schedframe.ResourceController; 29 import schedframe.events.scheduling.EventReason;30 27 import schedframe.events.scheduling.SchedulingEvent; 31 28 import schedframe.events.scheduling.SchedulingEventType; 32 29 import schedframe.events.scheduling.StartTaskExecutionEvent; 33 import schedframe.events.scheduling.TaskCanceledEvent;34 30 import schedframe.events.scheduling.TaskFinishedEvent; 35 31 import schedframe.events.scheduling.TaskRequestedTimeExpiredEvent; … … 45 41 import schedframe.scheduling.ResourceHistoryItem; 46 42 import schedframe.scheduling.Scheduler; 43 import schedframe.scheduling.TaskListImpl; 47 44 import schedframe.scheduling.UsedResourceList; 48 45 import schedframe.scheduling.WorkloadUnitHandler; 49 import schedframe.scheduling.WorkloadUnitListImpl;50 46 import schedframe.scheduling.manager.resources.LocalResourceManager; 51 47 import schedframe.scheduling.manager.resources.ManagedResources; … … 63 59 import schedframe.scheduling.tasks.Job; 64 60 import schedframe.scheduling.tasks.JobInterface; 65 import schedframe.scheduling.tasks.SubmittedTask;66 61 import schedframe.scheduling.tasks.Task; 67 62 import schedframe.scheduling.tasks.TaskInterface; 68 63 import schedframe.scheduling.tasks.WorkloadUnit; 69 import simulator. WormsConstants;64 import simulator.DCWormsConstants; 70 65 import simulator.utils.DoubleMath; 71 66 … … 112 107 switch (tag) { 113 108 114 case WormsTags.TIMER:109 case DCWormsTags.TIMER: 115 110 if (pluginSupportsEvent(tag)) { 116 111 SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TIMER); 117 SchedulingPlanInterface decision = schedulingPlugin.schedule(event,112 SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event, 118 113 queues, getJobRegistry(), getResourceManager(), moduleList); 119 114 executeSchedulingPlan(decision); … … 123 118 break; 124 119 125 case WormsTags.TASK_READY_FOR_EXECUTION:120 case DCWormsTags.TASK_READY_FOR_EXECUTION: 126 121 127 122 ExecTask data = (ExecTask) ev.get_data(); 128 123 try { 129 data.setStatus( Gridlet.READY);124 data.setStatus(DCWormsTags.READY); 130 125 if (pluginSupportsEvent(tag)) { 131 126 SchedulingEvent event = new StartTaskExecutionEvent(data.getJobId(), data.getId()); 132 SchedulingPlanInterface decision = schedulingPlugin.schedule(event,127 SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event, 133 128 queues, getJobRegistry(), getResourceManager(), moduleList); 134 129 executeSchedulingPlan(decision); … … 139 134 break; 140 135 141 case WormsTags.TASK_EXECUTION_FINISHED:136 case DCWormsTags.TASK_EXECUTION_FINISHED: 142 137 obj = ev.get_data(); 143 ExecTask task= (ExecTask) obj;144 if ( task.getStatus() == Gridlet.INEXEC) {145 finalizeExecutable( task);146 SubmittedTask subTask = (SubmittedTask)task; 147 sendFinishedWorkloadUnit( (ExecTask)subTask.getGridlet());138 ExecTask exec = (ExecTask) obj; 139 if (exec.getStatus() == DCWormsTags.INEXEC) { 140 finalizeExecutable(exec); 141 142 sendFinishedWorkloadUnit(exec); 148 143 //task.setGridletStatus(Gridlet.SUCCESS); 149 144 //task.finalizeGridlet(); 150 log.debug( task.getJobId() + "_" + task.getId() + " finished execution on " + new DateTime());151 log.info( WormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size()));145 log.debug(exec.getJobId() + "_" + exec.getId() + " finished execution on " + new DateTime()); 146 log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size())); 152 147 /*UsedResourceList<ResourceHistoryItem> lastUsedList = task.getUsedResources(); 153 148 Map<ResourceUnitName, AbstractResourceUnit> lastUsed = lastUsedList.getLast() … … 163 158 } 164 159 if (pluginSupportsEvent(tag)) { 165 SchedulingEvent event = new TaskFinishedEvent( task.getJobId(), task.getId());166 SchedulingPlanInterface decision = schedulingPlugin.schedule(event,160 SchedulingEvent event = new TaskFinishedEvent(exec.getJobId(), exec.getId()); 161 SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event, 167 162 queues, getJobRegistry(), getResourceManager(), moduleList); 168 163 executeSchedulingPlan(decision); 169 164 } 170 Job job = jobRegistry.getJob( task.getJobId());165 Job job = jobRegistry.getJob(exec.getJobId()); 171 166 if(!job.isFinished()){ 172 167 getWorkloadUnitHandler().handleJob(job); … … 174 169 175 170 break; 176 case WormsTags.TASK_REQUESTED_TIME_EXPIRED:171 case DCWormsTags.TASK_REQUESTED_TIME_EXPIRED: 177 172 obj = ev.get_data(); 178 task = (SubmittedTask) obj;173 exec = (Executable) obj; 179 174 if (pluginSupportsEvent(tag)) { 180 SchedulingEvent event = new TaskRequestedTimeExpiredEvent( task.getJobId(), task.getId());181 SchedulingPlanInterface decision = schedulingPlugin.schedule(event,175 SchedulingEvent event = new TaskRequestedTimeExpiredEvent(exec.getJobId(), exec.getId()); 176 SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event, 182 177 queues, getJobRegistry(), getResourceManager(), moduleList); 183 178 executeSchedulingPlan(decision); … … 185 180 186 181 break; 187 case WormsTags.UPDATE:182 case DCWormsTags.UPDATE: 188 183 updateProcessingTimes(ev); 189 184 break; … … 193 188 194 189 public void notifyReturnedWorkloadUnit(WorkloadUnit wu) { 195 if (pluginSupportsEvent( WormsTags.TASK_EXECUTION_FINISHED)) {190 if (pluginSupportsEvent(DCWormsTags.TASK_EXECUTION_FINISHED)) { 196 191 SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TASK_FINISHED); 197 SchedulingPlanInterface decision = schedulingPlugin.schedule(event,192 SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event, 198 193 queues, getJobRegistry(), getResourceManager(), moduleList); 199 194 executeSchedulingPlan(decision); … … 203 198 //} 204 199 } 205 206 public void notifyCanceledWorkloadUnit(WorkloadUnit job) { 207 208 if (!pluginSupportsEvent(GridSimTags.GRIDLET_CANCEL)) 209 return; 210 211 Executable executable = (Executable) job; 212 String jobID = executable.getJobId(); 213 214 SchedulingPlanInterface decision = null; 215 216 try { 217 executable.setStatus((int) BrokerConstants.JOB_STATUS_CANCELED); 218 219 TaskCanceledEvent event = new TaskCanceledEvent(executable.getJobId(), executable.getTaskId()); 220 event.setReason(EventReason.RESERVATION_EXCEEDED); 221 decision = schedulingPlugin 222 .schedule(event, queues, getJobRegistry(), getResourceManager(), moduleList); 223 224 if (decision == null) 225 return; 226 227 executeSchedulingPlan(decision); 228 229 } catch (Exception e) { 230 log.error("Exception during scheduling. " + e.getMessage()); 231 e.printStackTrace(); 232 } 233 } 234 235 protected void executeSchedulingPlan(SchedulingPlanInterface decision) { 236 237 ArrayList<ScheduledTaskInterface> taskSchedulingDecisions = decision.getTasks(); 200 201 protected void executeSchedulingPlan(SchedulingPlanInterface<?> decision) { 202 203 ArrayList<ScheduledTaskInterface<?>> taskSchedulingDecisions = decision.getTasks(); 238 204 for (int i = 0; i < taskSchedulingDecisions.size(); i++) { 239 try { 240 ScheduledTaskInterface taskDecision = taskSchedulingDecisions.get(i); 241 242 // not scheduled again are returned to the user. 243 if (taskDecision.getStatus() == AllocationStatus.REJECTED) { 244 continue; 205 ScheduledTaskInterface<?> taskDecision = taskSchedulingDecisions.get(i); 206 207 if (taskDecision.getStatus() == AllocationStatus.REJECTED) { 208 continue; 209 } 210 211 ArrayList<AllocationInterface<?>> allocations = taskDecision.getAllocations(); 212 213 TaskInterface<?> task = taskDecision.getTask(); 214 for (int j = 0; j < allocations.size(); j++) { 215 216 AllocationInterface<?> allocation = allocations.get(j); 217 if (allocation.isProcessing()) { 218 ExecTask exec = (ExecTask) task; 219 executeTask(exec, allocation.getRequestedResources()); 220 } else if(resourceManager.getSchedulerName(allocation.getProviderName()) != null){ 221 allocation.setProviderName(resourceManager.getSchedulerName(allocation.getProviderName())); 222 submitTask(task, allocation); 223 } else { 224 ExecTask exec = (ExecTask) task; 225 executeTask(exec, chooseResourcesForExecution(allocation.getProviderName(), (ExecTask)task)); 245 226 } 246 247 ArrayList<AllocationInterface> allocations = taskDecision.getAllocations(); 248 249 WorkloadUnit task = taskDecision.getTask(); 250 for (int j = 0; j < allocations.size(); j++) { 251 252 AllocationInterface allocation = allocations.get(j); 253 if (allocation.isProcessing()) { 254 255 ExecTask exec = (ExecTask) task; 256 257 258 //Executable e = (Executable)task; 259 /*SubmittedTask submittedTask = jobRegistry.getSubmittedTask(e.getJobId(), e.getId()); 260 if(submittedTask == null) 261 { submittedTask = new SubmittedTask(e); 262 jobRegistry.addTask(submittedTask); 263 }*/ 264 265 /*e.visitResource(scheduler.get_name()); 266 Scheduler parentScheduler = scheduler.getParent(); 267 while (parentScheduler != null && !e.getVisitedResources().contains(parentScheduler.get_name())) { 268 e.visitResource(parentScheduler.get_name()); 269 parentScheduler = parentScheduler.getParent(); 270 }*/ 271 272 273 executeTask(exec, allocation.getRequestedResources()); 274 //} else if(GridSim.getEntityId(allocation.getProviderName()) != -1 || scheduler.getScheduler(allocation.getProviderName())!=null){ 275 } else if(resourceManager.getSchedulerName(allocation.getProviderName()) != null){ 276 allocation.setProviderName(resourceManager.getSchedulerName(allocation.getProviderName())); 277 submitWorkloadUnit(task, allocation); 278 } else { 279 280 ExecTask exec = (ExecTask) task; 281 282 //Executable exec = jobRegistry.createExecutable(t, allocation); 283 //exec.setResourceParameter(scheduler.get_id(), 1); 284 /*e.visitResource(scheduler.get_name()); 285 Scheduler parentScheduler = scheduler.getParent(); 286 while (parentScheduler != null && !e.getVisitedResources().contains(parentScheduler.get_name())) { 287 e.visitResource(parentScheduler.get_name()); 288 parentScheduler = parentScheduler.getParent(); 289 }*/ 290 executeTask(exec, chooseResourcesForExecution(allocation.getProviderName(), (ExecTask)task)); 291 } 292 } 293 294 } catch (Exception e) { 295 e.printStackTrace(); 296 } 227 } 228 297 229 } 298 230 } 299 231 300 232 protected void executeTask(ExecTask task, Map<ResourceUnitName, ResourceUnit> choosenResources) { 301 // Executable exec = (Executable) task; 302 303 SubmittedTask submittedTask = (SubmittedTask)task; 233 234 Executable exec = (Executable)task; 304 235 boolean allocationStatus = getAllocationManager().allocateResources(choosenResources); 305 236 if(allocationStatus == false) 306 237 return; 307 238 removeFromQueue(task); 308 //SubmittedTask submittedTask = (SubmittedTask)task; 309 /* submittedTask = jobRegistry.getSubmittedTask(exec.getJobId(), exec.getId()); 310 if(submittedTask == null) 311 { submittedTask = new SubmittedTask(exec); 312 jobRegistry.addTask(submittedTask); 313 }*/ 314 double completionPercentage = (submittedTask.getLength() - submittedTask.getRemainingGridletLength())/submittedTask.getLength(); 239 //double completionPercentage = (submittedTask.getLength() - submittedTask.getRemainingGridletLength())/submittedTask.getLength(); 315 240 SchedulingEvent event = new SchedulingEvent(SchedulingEventType.START_TASK_EXECUTION); 316 241 int time = Double.valueOf( 317 execTimeEstimationPlugin.execTimeEstimation(event, task, choosenResources, completionPercentage)).intValue();242 execTimeEstimationPlugin.execTimeEstimation(event, task, choosenResources, exec.getCompletionPercentage())).intValue(); 318 243 log.debug(task.getJobId() + "_" + task.getId() + " starts executing on " + new DateTime() 319 244 + " will finish after " + time); … … 322 247 return; 323 248 324 submittedTask.setEstimatedDuration(time);249 exec.setEstimatedDuration(time); 325 250 DateTime currentTime = new DateTime(); 326 251 ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime); 327 submittedTask.addUsedResources(resHistItem); 328 submittedTask.setFinishTime(currentTime.getMillis() / 1000); 329 330 jobRegistry.saveHistory(submittedTask, time, choosenResources); 331 332 scheduler.sendInternal(time, WormsTags.TASK_EXECUTION_FINISHED, 333 submittedTask); 252 exec.addUsedResources(resHistItem); 253 254 scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, 255 exec); 334 256 335 257 try { 336 long expectedDuration = submittedTask.getExpectedDuration().getMillis() / 1000;337 scheduler.sendInternal(expectedDuration, WormsTags.TASK_REQUESTED_TIME_EXPIRED, submittedTask);258 long expectedDuration = exec.getExpectedDuration().getMillis() / 1000; 259 scheduler.sendInternal(expectedDuration, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec); 338 260 } catch (NoSuchFieldException e) { 339 double t = submittedTask.getEstimatedDuration(); 340 scheduler.sendInternal(t, WormsTags.TASK_REQUESTED_TIME_EXPIRED, submittedTask); 341 } 342 343 submittedTask.setGridletStatus(Gridlet.INEXEC); 344 log.info(WormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size())); 261 double t = exec.getEstimatedDuration(); 262 scheduler.sendInternal(t, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec); 263 } 264 265 try { 266 exec.setStatus(DCWormsTags.INEXEC); 267 } catch (Exception e1) { 268 // TODO Auto-generated catch block 269 e1.printStackTrace(); 270 } 271 log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size())); 345 272 346 273 PEUnit peUnit = (PEUnit)choosenResources.get(StandardResourceUnitName.PE); … … 348 275 ProcessingElements pes = (ProcessingElements) peUnit; 349 276 for (ComputingResource resource : pes) { 350 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, submittedTask));277 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec)); 351 278 } 352 279 } else { … … 357 284 358 285 } 359 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, submittedTask));286 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec)); 360 287 } 361 288 /*ProcessingElements pes = (ProcessingElements) choosenResources.get(StandardResourceUnitName.PE); … … 371 298 } 372 299 373 public void finalizeExecutable(ExecTask exec){ 374 375 SubmittedTask subTask = (SubmittedTask)exec; 376 subTask.setGridletStatus(Gridlet.SUCCESS); 377 subTask.finalizeGridlet(); 378 UsedResourceList<ResourceHistoryItem> lastUsedList = subTask.getUsedResources(); 300 public void finalizeExecutable(ExecTask execTask){ 301 302 Executable exec = (Executable)execTask; 303 try { 304 exec.setStatus(DCWormsTags.SUCCESS); 305 } catch (Exception e1) { 306 // TODO Auto-generated catch block 307 e1.printStackTrace(); 308 } 309 exec.finalizeExecutable(); 310 UsedResourceList<ResourceHistoryItem> lastUsedList = exec.getUsedResources(); 379 311 Map<ResourceUnitName, ResourceUnit> lastUsed = lastUsedList.getLast() 380 312 .getResourceUnits(); … … 385 317 ProcessingElements pes = (ProcessingElements) peUnit; 386 318 for (ComputingResource resource : pes) { 387 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask));319 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec)); 388 320 } 389 321 } else { … … 394 326 395 327 } 396 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask));328 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec)); 397 329 } 398 330 /*ProcessingElements pes = (ProcessingElements) lastUsed.get(StandardResourceUnitName.PE); … … 400 332 resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask)); 401 333 }*/ 402 SubTaskFilter filter = new SubTaskFilter(subTask.getGridletID(),WormsTags.TASK_REQUESTED_TIME_EXPIRED);334 ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_REQUESTED_TIME_EXPIRED); 403 335 scheduler.sim_cancel(filter, null); 404 336 405 Executable executable = (Executable) subTask.getGridlet(); 406 Job job = jobRegistry.getJob(exec utable.getJobId());337 338 Job job = jobRegistry.getJob(exec.getJobId()); 407 339 408 340 Task task = null; 409 341 try { 410 task = job.getTask(exec utable.getTaskId());342 task = job.getTask(exec.getTaskId()); 411 343 } catch (NoSuchFieldException e) { 412 344 e.printStackTrace(); 413 345 } 414 if(exec utable.getProcessesId() == null){346 if(exec.getProcessesId() == null){ 415 347 try { 416 task.setStatus(exec utable.getStatus());348 task.setStatus(exec.getStatus()); 417 349 } catch (Exception e) { 418 350 e.printStackTrace(); … … 422 354 for(int i = 0; i < processesList.size(); i++){ 423 355 AbstractProcesses processes = processesList.get(i); 424 if(processes.getId().equals(exec utable.getProcessesId())){425 processes.setStatus(exec utable.getStatus());356 if(processes.getId().equals(exec.getProcessesId())){ 357 processes.setStatus(exec.getStatus()); 426 358 break; 427 359 } … … 441 373 while (iter.hasNext()) { 442 374 ExecTask task = iter.next(); 443 SubmittedTask subTask = (SubmittedTask)task;444 UsedResourceList<ResourceHistoryItem> usedResourcesList = subTask.getUsedResources();375 Executable exec = (Executable)task; 376 UsedResourceList<ResourceHistoryItem> usedResourcesList = exec.getUsedResources(); 445 377 ResourceUnit unit = usedResourcesList.getLast().getResourceUnits() 446 378 .get(StandardResourceUnitName.PE); 447 379 448 380 double load = getMIShare(timeSpan, (PEUnit) unit); 449 subTask.updateGridletFinishedSoFar(load);381 exec.setCompletionPercentage(100 * timeSpan/exec.getEstimatedDuration()); 450 382 addTotalLoad(load); 451 383 } … … 471 403 updateProcessingProgress(); 472 404 for (ExecTask task : jobRegistry.getRunningTasks()) { 473 SubmittedTask subTask = (SubmittedTask)task;474 List<String> visitedResource = subTask.getVisitedResources();405 Executable exec = (Executable)task; 406 List<String> visitedResource = exec.getVisitedResources(); 475 407 String originResource = ev.get_data().toString(); 476 408 if(!ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), originResource)){ … … 478 410 } 479 411 480 Map<ResourceUnitName, ResourceUnit> choosenResources = subTask.getUsedResources().getLast().getResourceUnits();481 double completionPercentage = (task.getLength() - subTask.getRemainingGridletLength())/task.getLength();412 Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getUsedResources().getLast().getResourceUnits(); 413 //double completionPercentage = (task.getLength() - subTask.getRemainingGridletLength())/task.getLength(); 482 414 double time = execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.RESOURCE_STATE_CHANGED), 483 task, choosenResources, completionPercentage);415 task, choosenResources, exec.getCompletionPercentage()); 484 416 485 417 /*if(!subTask.getVisitedResources().contains(ev.get_data().toString())) { … … 487 419 }*/ 488 420 //check if the new estimated end time is equal to the previous one; if yes the continue without update 489 if( DoubleMath.subtract(( subTask.getExecStartTime() + subTask.getEstimatedDuration()), (new DateTime().getMillis()/1000 + time)) == 0.0){421 if( DoubleMath.subtract((exec.getExecStartTime() + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + time)) == 0.0){ 490 422 continue; 491 423 } 492 SubTaskFilter filter = new SubTaskFilter(subTask.getGridletID(),WormsTags.TASK_EXECUTION_FINISHED);424 ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED); 493 425 scheduler.sim_cancel(filter, null); 494 scheduler.sendInternal(time, WormsTags.TASK_EXECUTION_FINISHED, task);426 scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, task); 495 427 496 428 } … … 577 509 } 578 510 579 /*try {580 List<? extends ComputingResource> processingElements = resourceManager.getResourcesOfType(StandardResourceType.Processor);581 List<ComputingResource> choosenResources = new ArrayList<ComputingResource>();582 int peSize = processingElements.size();583 for (int i = 0; i < peSize && cpuRequest > 0; i++) {584 if (processingElements.get(i).getStatus() == ResourceStatus.FREE) {585 choosenResources.add(processingElements.get(i));586 cpuRequest--;587 }588 }589 if (cpuRequest > 0)590 {591 return null;592 }593 processingUnits = new ProcessingElements(choosenResources);594 } catch (Exception e) {595 596 List<ResourceUnit> procResUnit = resourceManager.getDistributedResourceUnits(StandardResourceUnitName.PE);597 598 for(ResourceUnit resUnit: procResUnit){599 if (resUnit.getFreeAmount() >= cpuRequest)600 {601 processingUnits = new PEUnit(resUnit.getResourceId(), cpuRequest, cpuRequest);602 break;603 }604 }605 }*/606 511 map.put(StandardResourceUnitName.PE, choosenPEUnits.get(0)); 607 512 } 608 /*int memoryRequest; 609 try { 610 memoryRequest = Double.valueOf(task.getMemoryRequest()).intValue(); 611 } catch (NoSuchFieldException e) { 612 memoryRequest = 0; 613 } 614 if (memoryRequest != 0) { 615 List<ResourceUnit> resUnit = resourceManager.getSharedResourceUnits().get(StandardResourceUnitName.MEMORY); 616 617 Memory memory = null; 618 for (ResourceUnit memUnit : resUnit) { 619 Memory m = (Memory) memUnit; 620 621 if (m.getFreeAmount() >= memoryRequest) { 622 System.out.println(m.getResourceId()+ ";"+m.getAmount()+";"+m.getFreeAmount()); 623 memory = new Memory(m, memoryRequest, memoryRequest); 624 } 625 } 626 if(memory == null) 627 return null; 628 map.put(StandardResourceUnitName.MEMORY, memory); 629 }*/ 513 630 514 return map; 631 515 } 632 633 634 635 public void notifySubmittedWorkloadUnit(WorkloadUnit job, boolean ack) { 516 517 public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) { 636 518 updateProcessingProgress(); 637 registerWorkloadUnit( job);519 registerWorkloadUnit(wu); 638 520 } 639 521 … … 654 536 List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>(); 655 537 jobsList.add(job); 656 WorkloadUnitListImpl readyTasks = new WorkloadUnitListImpl();657 for(Task task: jobRegistry.get ReadyTasks(jobsList)){538 TaskListImpl readyTasks = new TaskListImpl(); 539 for(Task task: jobRegistry.getAvailableTasks(jobsList)){ 658 540 task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED); 659 541 readyTasks.add(task); … … 667 549 public void handleTask(TaskInterface<?> ti){ 668 550 Task task = (Task)ti; 669 670 if(task.getProcesses() == null){ 551 List<AbstractProcesses> processes = task.getProcesses(); 552 553 if(processes == null || processes.size() == 0){ 671 554 Executable exec = new Executable(task); 672 exec.setUserID(task.getSenderId());673 exec.setLength(task.getLength());674 555 registerWorkloadUnit(exec); 675 556 } else { 676 List<AbstractProcesses> processesList = task.getProcesses(); 677 for(int i = 0; i < processesList.size(); i++){ 678 AbstractProcesses processes = processesList.get(i); 679 Executable exec = new Executable(task, processes); 680 exec.setUserID(task.getSenderId()); 681 exec.setLength(task.getLength()); 557 for(int j = 0; j < processes.size(); j++){ 558 AbstractProcesses procesesSet = processes.get(j); 559 Executable exec = new Executable(task, procesesSet); 682 560 registerWorkloadUnit(exec); 683 561 } … … 701 579 } 702 580 703 exec.setResourceParameter(scheduler.get_id(), 1); 704 SubmittedTask subTask = new SubmittedTask(exec); 705 jobRegistry.addTask(subTask); 706 WorkloadUnitListImpl newTasks = new WorkloadUnitListImpl(); 707 newTasks.add(subTask); 708 709 schedulingPlugin.placeJobsInQueues(newTasks, queues, getResourceManager(), moduleList); 710 711 if (subTask.getStatus() == Gridlet.QUEUED) { 581 exec.setSchedulerName(scheduler.get_id()); 582 jobRegistry.addExecTask(exec); 583 TaskListImpl newTasks = new TaskListImpl(); 584 newTasks.add(exec); 585 586 schedulingPlugin.placeTasksInQueues(newTasks, queues, getResourceManager(), moduleList); 587 588 if (exec.getStatus() == DCWormsTags.QUEUED) { 712 589 sendExecutableReadyEvent(exec); 713 }714 }715 716 public void handleSubmittedTask(SubmittedTask task){717 718 task.visitResource(scheduler.get_name());719 Scheduler parentScheduler = scheduler.getParent();720 while (parentScheduler != null && !task.getVisitedResources().contains(parentScheduler.get_name())) {721 task.visitResource(parentScheduler.get_name());722 parentScheduler = parentScheduler.getParent();723 }724 725 jobRegistry.addTask(task);726 WorkloadUnitListImpl newTasks = new WorkloadUnitListImpl();727 newTasks.add(task);728 729 schedulingPlugin.placeJobsInQueues(newTasks, queues, getResourceManager(), moduleList);730 731 if (task.getStatus() == Gridlet.QUEUED) {732 sendExecutableReadyEvent(task);733 590 } 734 591 } -
DCWoRMS/trunk/src/schedframe/scheduling/queue/TaskQueue.java
r478 r481 1 1 package schedframe.scheduling.queue; 2 2 3 import gridsim. Gridlet;3 import gridsim.gssim.DCWormsTags; 4 4 import gssim.schedframe.scheduling.queues.AbstractStatsSupportingQueue; 5 5 6 6 import org.joda.time.DateTime; 7 7 8 import schedframe.scheduling.tasks. WorkloadUnit;8 import schedframe.scheduling.tasks.TaskInterface; 9 9 10 public class TaskQueue extends AbstractStatsSupportingQueue<WorkloadUnit> implements Queue<WorkloadUnit>{ 11 10 public class TaskQueue extends AbstractStatsSupportingQueue<TaskInterface<?>> implements Queue<TaskInterface<?>>{ 12 11 13 12 private static final long serialVersionUID = 6576299222910508209L; … … 17 16 protected boolean supportReservation; 18 17 19 20 18 public TaskQueue (boolean supportReservation){ 21 name = "Queue";22 priority = 0;19 this.name = "Queue"; 20 this.priority = 0; 23 21 this.supportReservation = supportReservation; 24 25 22 } 26 23 27 public boolean add( WorkloadUnit wu){24 public boolean add(TaskInterface<?> task){ 28 25 try { 29 wu.setStatus(Gridlet.QUEUED);26 task.setStatus(DCWormsTags.QUEUED); 30 27 } catch(Exception e){ 31 28 throw new RuntimeException(e); 32 29 } 33 //updateStats(); 34 return super.add(wu); 30 return super.add(task); 35 31 } 36 32 37 public void add(int pos, WorkloadUnit wu){33 public void add(int pos, TaskInterface<?> task){ 38 34 try { 39 wu.setStatus(Gridlet.QUEUED);35 task.setStatus(DCWormsTags.QUEUED); 40 36 } catch(Exception e){ 41 37 throw new RuntimeException(e); 42 38 } 43 //updateStats(); 44 super.add(pos, wu); 39 super.add(pos, task); 45 40 } 46 41 47 42 public DateTime getArrivalTime(int pos) throws IndexOutOfBoundsException { 48 //return get(pos).getSubmissionTimeToBroker(); 49 return null; 43 return get(pos).getSubmissionTimeToBroker(); 50 44 } 51 45 … … 69 63 return supportReservation; 70 64 } 71 72 73 /*public boolean contains (WorkloadUnitInterface<?> wu) {74 for(int i = 0; i< size();i++){75 try {76 if(get(i).getId().equals(wu.getId()))77 return true;78 } catch (NoSuchFieldException e) {79 return false;80 }81 }82 return false;83 }84 85 public boolean remove (WorkloadUnitInterface<?> wu) {86 boolean found = false;87 int index = 0;88 for(int i = 0; i< size() && !found;i++){89 try {90 if(get(i).getId().equals(wu.getId()))91 {92 found = true;93 index = i;94 }95 96 } catch (NoSuchFieldException e) {97 return false;98 }99 }100 remove(index);101 return true;102 }*/103 65 } -
DCWoRMS/trunk/src/schedframe/scheduling/tasks/Job.java
r480 r481 178 178 } 179 179 180 public int getUserI D(){180 public int getUserId(){ 181 181 return this.senderId; 182 182 } -
DCWoRMS/trunk/src/schedframe/scheduling/tasks/Processes.java
r480 r481 199 199 200 200 @Override 201 public int getUserI D() {201 public int getUserId() { 202 202 // TODO Auto-generated method stub 203 203 return 0; -
DCWoRMS/trunk/src/schedframe/scheduling/tasks/Task.java
r480 r481 471 471 472 472 @Override 473 public int getUserI D() {473 public int getUserId() { 474 474 // TODO Auto-generated method stub 475 475 return 0; -
DCWoRMS/trunk/src/schedframe/scheduling/tasks/TaskInterface.java
r480 r481 88 88 */ 89 89 public long getLength(); 90 91 /** 92 * 93 * @param length measured in instructions. 94 */ 95 public void setLength(long length); 96 90 97 91 /** 98 92 * -
DCWoRMS/trunk/src/schedframe/scheduling/tasks/WorkloadUnit.java
r480 r481 8 8 public String getId(); 9 9 10 public int getUserI D();10 public int getUserId(); 11 11 12 12 public int getStatus();
Note: See TracChangeset
for help on using the changeset viewer.