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

DCWoRMS/trunk/build/classes/schedframe/scheduling/Cluster.java

r477	r539
13	13	import gridsim.GridSimTags;
14	14	import gridsim.IO_data;
15		import gridsim.~~gssim.~~WormsTags;
	15	import gridsim.dcworms.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/build/classes/schedframe/scheduling/GridResourceDiscovery.java

r477	r539
5	5	import gridsim.GridSimTags;
6	6	import gridsim.IO_data;
7		import gridsim.~~gssim.~~WormsTags;
	7	import gridsim.dcworms.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/build/classes/schedframe/scheduling/Scheduler.java

r477	r539
8	8	import gridsim.GridSimTags;
9	9	import gridsim.IO_data;
10		import gridsim.~~gssim.~~WormsTags;
	10	import gridsim.dcworms.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	}
…	…
133	133	// managemetnSystem_.setEndSimulation();
134	134	run = false;
135		~~/*Sim_stat stats = get_stat();~~
136		~~List<Object[]> measures = stats.get_measures();~~
137		~~for (Object[] info : measures) {~~
138		~~String measure = (String) info[0];~~
139		~~if (measure~~
140		~~.startsWith(GssimConstants.TASKS_QUEUE_LENGTH_MEASURE_NAME)) {~~
141		~~System.out.println("====="+this.get_name()+";"+stats.average(measure));~~
142		}
143		~~}*/~~
144	135	break;
145	136	}
…	…
155	146
156	147	case GridSimTags.GRIDLET_SUBMIT:
157		process~~GSSIMJob~~Submit(ev, false);
	148	processWorkloadUnitSubmit(ev, false);
158	149	break;
159	150
160	151	case GridSimTags.GRIDLET_SUBMIT_ACK:
161		process~~GSSIMJob~~Submit(ev, true);
	152	processWorkloadUnitSubmit(ev, true);
162	153	break;
163	154
164	155	case GridSimTags.GRIDLET_RETURN:
165		process~~GSSIMJob~~Return(ev);
	156	processWorkloadUnitReturn(ev);
166	157	break;
167	158
…	…
174	165	protected void processOtherRequest(Sim_event ev) {
175	166	switch (ev.get_tag()) {
176		case WormsTags.QUERY_RESOURCE_DESC:
	167	case DCWormsTags.QUERY_RESOURCE_DESC:
177	168	SchedulerDescription desc = new SchedulerDescription(new LocalSystem(get_name(), null, null));
178	169	Map<ResourceUnitName, List<ResourceUnit>> units = managementSystem.getResourceManager().getSharedResourceUnits();
…	…
181	172
182	173	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);
	174	send(ev.get_src(), GridSimTags.SCHEDULE_NOW, DCWormsTags.QUERY_RESOURCE_DESC_RESULT, data);
184	175	break;
185	176
…	…
189	180	}
190	181	}
191
192		/*public boolean processOtherEvent(Sim_event ev) {
193		return false;
194		}*/
195
196		protected void processGSSIMJobReturn(Sim_event ev) {
197		WorkloadUnit<?> job = (WorkloadUnit<?>) ev.get_data();
	182
	183	protected void processWorkloadUnitReturn(Sim_event ev) {
	184	WorkloadUnit job = (WorkloadUnit) ev.get_data();
198	185	managementSystem.notifyReturnedWorkloadUnit(job);
199	186	}
200	187
201		protected void process~~GSSIMJob~~Submit(Sim_event ev, boolean ack) {
202		WorkloadUnit~~<?> job = (WorkloadUnit<?>~~) ev.get_data();
	188	protected void processWorkloadUnitSubmit(Sim_event ev, boolean ack) {
	189	WorkloadUnit job = (WorkloadUnit) ev.get_data();
203	190	managementSystem.notifySubmittedWorkloadUnit(job, ack);
204	191	}
…	…
223	210	this.send(this.get_id(), delay, tag, data);
224	211	}
225
226		~~/*protected void send(int dest, int tag, int transaction, Object obj){~~
227		~~IO_data data = new Transaction_IO_data(transaction, obj, 8, dest);~~
228		~~super.send(dest, GridSimTags.SCHEDULE_NOW, tag, data);~~
229		~~}*/~~
230	212
231	213	public Sim_port getOutputPort() {
…	…
271	253	public List<QueueDescription> getQueuesDescription(){
272	254	List<QueueDescription> queues = new ArrayList<QueueDescription>();
273		for(TaskQueue queue: managementSystem.get~~Access~~Queues()){
	255	for(TaskQueue queue: managementSystem.getQueues()){
274	256	QueueDescription qd;
275	257	try {

DCWoRMS/trunk/build/classes/schedframe/scheduling/WorkloadUnitHandler.java

r477	r539
1	1	package schedframe.scheduling;
2	2
3		import ~~schedframe.scheduling.tasks.Job~~;
4		import schedframe.scheduling.tasks.~~SubmittedTask~~;
	3	import dcworms.schedframe.scheduling.ExecTask;
	4	import schedframe.scheduling.tasks.JobInterface;
5	5	import schedframe.scheduling.tasks.TaskInterface;
6		~~import gssim.schedframe.scheduling.ExecTask;~~
7	6
8		public interface WorkloadUnitHandler{
	7	public interface WorkloadUnitHandler{
9	8
10		public void handleJob(Job job);
	9	public void handleJob(JobInterface<?> job);
11	10
12	11	public void handleTask(TaskInterface<?> task);
13	12
14	13	public void handleExecutable(ExecTask task);
15
16		public void handleSubmittedTask(SubmittedTask task);
	14
17	15	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/manager/resources/LocalResourceManager.java

r477	r539
13	13	import java.util.Properties;
14	14	import java.util.Set;
15		~~import java.util.Stack;~~
16	15
17	16	import schedframe.exceptions.ResourceException;
…	…
175	174	return resourceUnit;
176	175	}
177		~~/*public List<AbstractResourceUnit> getAvailableResourceUnits(String resourceName) throws Exception {~~
178		~~ComputingResource resource = getResourceByName(resourceName);~~
179		~~List<AbstractResourceUnit> resourceUnits = new ArrayList<AbstractResourceUnit>();~~
180		~~while(resource != null){~~
181		~~for(List<AbstractResourceUnit> resUnits: resource.getResourceCharacteristic().getResourceUnits().values())~~
182		~~resUnits.addAll(resourceUnits);~~
183		~~resource = resource.getParent();~~
184		}
185
186		~~return resourceUnits;~~
187		~~}*/~~
188	176
189	177	public List<? extends ComputingResource> filterResources(Properties properties) {
…	…
242	230	computingResources = (List<ComputingResource>) getResourcesOfType(StandardResourceType.Core);
243	231	} catch (ResourceException e) {
244		throw new RuntimeException("~~GSSIM~~ internal error");
	232	throw new RuntimeException("DCWorms internal error");
245	233	}
246	234	PEUnit peUnit = new ProcessingElements(computingResources);
…	…
253	241	computingResources = (List<ComputingResource>) getResourcesOfType(StandardResourceType.Processor);
254	242	} catch (ResourceException e) {
255		throw new RuntimeException("~~GSSIM~~ internal error");
	243	throw new RuntimeException("DCWorms internal error");
256	244	}
257	245	PEUnit peUnit = new ProcessingElements(computingResources);
…	…
318	306	return false;
319	307	}
320		~~/*ResourceUnit peUnit = resources.get(StandardResourceUnitName.PE);~~
321
322		~~if (peUnit != null) {~~
323		~~if (peUnit instanceof ProcessingElements) {~~
324		~~ProcessingElements choosenProcessors = (ProcessingElements) peUnit;~~
325
326		~~for (int i = 0; i < choosenProcessors.size(); i++) {~~
327		~~choosenProcessors.get(i).setStatus(ResourceStatus.BUSY);~~
328		}
329		}
330		~~}*/~~
331
332		~~/*Memory m = (Memory) resources.get(StandardResourceUnitName.MEMORY);~~
333		~~if (m != null) {~~
334		~~m.getProvisioner().setState(ResourceUnitState.BUSY);~~
335		~~}*/~~
336	308
337	309	for(ResourceUnitName resUnitName: resources.keySet()){
…	…
345	317	public void freeResources(Map<ResourceUnitName, ResourceUnit> resources) {
346	318
347		~~/*ResourceUnit peUnit = resources.get(StandardResourceUnitName.PE);~~
348
349		~~if (peUnit instanceof ProcessingElements) {~~
350		~~ProcessingElements processingElements = (ProcessingElements) peUnit;~~
351
352		~~for (int i = 0; i < processingElements.size(); i++) {~~
353		~~processingElements.get(i).setStatus(ResourceStatus.FREE);~~
354		}
355		~~}*/~~
356
357		~~/*Memory m = (Memory) resources.get(StandardResourceUnitName.MEMORY);~~
358		~~if (m != null) {~~
359		~~m.getProvisioner().setState(ResourceUnitState.FREE);~~
360		~~}*/~~
361
362	319	for(ResourceUnitName resUnitName: resources.keySet()){
363	320	ResourceUnit resUnit = resources.get(resUnitName);
…	…
366	323	}
367	324
368
369	325	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/manager/tasks/AbstractJobRegistry.java

r477	r539
2	2
3	3
4		~~import java.util.Map;~~
5	4	import java.util.concurrent.ConcurrentHashMap;
6	5
…	…
14	13
15	14
16		public abstract class AbstractJobRegistry /extends ConcurrentHashMap<String, Job> /implements JobRegistry, Cloneable{
	15	public abstract class AbstractJobRegistry /extends ConcurrentHashMap<String, Job>/ implements JobRegistry, Cloneable{
17	16
18	17	private static final long serialVersionUID = 8409060063583755824L;
	18
19	19
20		private static Log log = LogFactory.getLog(AbstractJobRegistry.class);
21
22		protected static final ConcurrentHashMap<String, Job> jobs = new ConcurrentHashMap<String, Job>();
	20	protected static final ConcurrentHashMap<String, JobInterface<?>> jobs = new ConcurrentHashMap<String, JobInterface<?>>();
23	21
24	22	protected AbstractJobRegistry(){
25		~~//log.warn("Methods from JobRegistry interface are not implemented.");~~
26	23	}
27	24
28	25	public boolean addJob(JobInterface<?> job) {
29		try {
30		jobs.put(job.getId(), (Job) job);
31		} catch (NoSuchFieldException e) {
32		log.error(e.getMessage());
33		return false;
34		}
	26	jobs.put(job.getId(), job);
35	27	return true;
36	28	}
…	…
38	30	public boolean addTask(TaskInterface<?> task) {
39	31	if(jobs.containsKey(task.getJobId())){
40		~~jobs.get~~(task.getJobId()).add((Task)task);
	32	getJob(task.getJobId()).add((Task)task);
41	33	return true;
42	34	} else {
…	…
45	37	}
46	38
47		public Job ~~get(String jobId)~~{
	39	public JobInterface<?> getJobInfo(String jobId) {
48	40	return jobs.get(jobId);
49	41	}
50
51		~~public JobInterface<?> getJobInfo(String jobID) {~~
52		~~return jobs.get(jobID);~~
53		}
54	42
55		public TaskInterface<?> getTaskInfo(String jobID, String taskId) {
	43	public TaskInterface<?> getTaskInfo(String jobId, String taskId) {
56	44	Task task = null;
57		Job job = ~~jobs.get(jobID~~);
	45	Job job = getJob(jobId);
58	46
59	47	if(job == null)
…	…
63	51	task = job.getTask(taskId);
64	52	} catch (NoSuchFieldException e) {
65		~~log.error(e.getMessage());~~
66	53	}
67	54	return task;
68	55	}
69	56
70		/*public List<JobInterface<?>> getActiveJobs() {
71		log.error("getActiveJobs() not implemented.");
72		return null;
	57	public Job getJob(String jobId){
	58	return (Job)jobs.get(jobId);
73	59	}
74	60
75		~~public List<TaskInterface<?>> getActiveTasks() {~~
76		~~log.error("getActiveTasks() not implemented.");~~
77		~~return null;~~
78		~~}*/~~
79
80
81
82
83	61	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/manager/tasks/JobRegistry.java

r477	r539
1	1	package schedframe.scheduling.manager.tasks;
2	2
3		~~import gssim.schedframe.scheduling.ExecTask;~~
4	3
5	4	import java.util.List;
6	5
	6	import dcworms.schedframe.scheduling.ExecTask;
	7
	8	import schedframe.ExecutablesList;
7	9	import schedframe.scheduling.tasks.JobInterface;
8	10	import schedframe.scheduling.tasks.TaskInterface;
…	…
11	13	public interface JobRegistry {
12	14
13		~~//public List<JobInterface<?>> getActiveJobs();~~
14
15		~~//public List<TaskInterface<?>> getActiveTasks();~~
16
17	15	public JobInterface<?> getJobInfo(String jobId);
18	16
…	…
31	29
32	30
33		~~//public List<SubmittedTask> getSubmittedTasks();~~
	31	public ExecutablesList getExecutableTasks();
34	32
35		public ExecTask get~~SubmittedTask~~(String jobId, String taskId);
	33	public ExecTask getExecutable(String jobId, String taskId);
36	34
37	35
38		public List<? extends TaskInterface<?>> get~~Ready~~Tasks(List<JobInterface<?>> jobsList);
	36	public List<? extends TaskInterface<?>> getAvailableTasks(List<JobInterface<?>> jobsList);
39	37
40	38	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/manager/tasks/JobRegistryImpl.java

r477	r539
1	1	package schedframe.scheduling.manager.tasks;
2	2
3		import gridsim.Gridlet;
4		import gssim.schedframe.scheduling.ExecTask;
5		import gssim.schedframe.scheduling.Executable;
	3	import gridsim.dcworms.DCWormsTags;
6	4
7	5	import java.util.ArrayList;
8		~~import java.util.Collections;~~
9		~~import java.util.HashMap;~~
10	6	import java.util.List;
11		~~import java.util.Map;~~
12	7
13	8	import org.apache.commons.lang.ArrayUtils;
14	9	import org.apache.commons.logging.Log;
15	10	import org.apache.commons.logging.LogFactory;
16		~~import org.joda.time.DateTime;~~
17	11	import org.qcg.broker.schemas.resreqs.ParentType;
18	12	import org.qcg.broker.schemas.resreqs.types.TaskStatesName;
19	13
	14	import dcworms.schedframe.scheduling.ExecTask;
	15
20	16	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;
	17	import schedframe.ExecutablesList;
26	18	import schedframe.scheduling.tasks.JobInterface;
27		~~import schedframe.scheduling.tasks.SubmittedTask;~~
28	19	import schedframe.scheduling.tasks.Task;
29		~~import simulator.WormsConstants;~~
30	20
31	21	public class JobRegistryImpl extends AbstractJobRegistry {
…	…
37	27	private String context;
38	28
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>();
	29	//TO DO - consider data structure
	30	protected static final ExecutablesList executables = new ExecutablesList();
	31	//protected static final List<ExecTask> executables = Collections.synchronizedList(new ArrayList<ExecTask>());;
	32	//protected static final List<ExecTaskInterface> executables = new CopyOnWriteArrayList<ExecTaskInterface>();
42	33
43		public JobRegistryImpl(String context_) {
44		~~context = context_~~;
	34	public JobRegistryImpl(String context) {
	35	this.context = context;
45	36	}
46	37
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);
	38	public boolean addExecTask(ExecTask newTask) {
	39	if(getExecutable(newTask.getJobId(), newTask.getId()) == null) {
	40	synchronized (executables) {
	41	executables.add(newTask);
56	42	}
57	43	return true;
…	…
60	46	}
61	47
	48	public ExecutablesList getExecutableTasks() {
	49	return executables;
	50	}
62	51	public List<ExecTask> getTasks(int status) {
63	52	List<ExecTask> taskList = new ArrayList<ExecTask>();
64		synchronized (~~submittedTask~~s) {
65		for (ExecTask task: ~~submittedTask~~s) {
	53	synchronized (executables) {
	54	for (ExecTask task: executables) {
66	55	if (task.getStatus() == status) {
67		~~//SubmittedTask subTask = (SubmittedTask) task;~~
68	56	List<String> visitedResource = task.getVisitedResources();
69		if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)){
	57	if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)) {
70	58	taskList.add(task);
71	59	}
72		~~/*if(subTask.getVisitedResources().contains(context)){~~
73		~~taskList.add(subTask);~~
74		~~}*/~~
75	60	}
76	61	}
…	…
80	65
81	66	public List<ExecTask> getQueuedTasks() {
82		return getTasks(~~Gridlet~~.QUEUED);
	67	return getTasks(DCWormsTags.QUEUED);
83	68	}
84	69
85	70	public List<ExecTask> getRunningTasks() {
86		return getTasks(~~Gridlet~~.INEXEC);
	71	return getTasks(DCWormsTags.INEXEC);
87	72	}
88	73
89	74	public List<ExecTask> getReadyTasks() {
90		return getTasks(~~Gridlet~~.READY);
	75	return getTasks(DCWormsTags.READY);
91	76	}
92	77
93	78	public List<ExecTask> getFinishedTasks() {
94		return getTasks(~~Gridlet~~.SUCCESS);
	79	return getTasks(DCWormsTags.SUCCESS);
95	80	}
96	81
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) {
126		if (task.getJobId().compareTo(jobId) == 0 && task.getId().compareTo(taskId)==0) {
	82	public ExecTask getExecutable(String jobId, String taskId){
	83	synchronized (executables) {
	84	for (ExecTask task : executables) {
	85	if (task.getJobId().compareTo(jobId) == 0 && task.getId().compareTo(taskId) == 0) {
127	86	return task;
128	87	}
…	…
131	90	return null;
132	91	}
133
134	92
135	93	@SuppressWarnings("unchecked")
136		public List<Task> get~~Ready~~Tasks(List<JobInterface<?>> wuList) {
137		List<Task> ~~ready~~Tasks = new ArrayList<Task>();
	94	public List<Task> getAvailableTasks(List<JobInterface<?>> wuList) {
	95	List<Task> availableTasks = new ArrayList<Task>();
138	96	List<Task> waitingTasks = new ArrayList<Task>();
139	97
…	…
143	101	}
144	102
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;
	103	availableTasks.addAll(getPrecedenceConstrainedAvailableTasks(waitingTasks));
	104	return availableTasks;
283	105	}
284	106
285	107
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){
	108	private List<Task> getPrecedenceConstrainedAvailableTasks(List<Task> tasks){
294	109
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){
	110	List<Task> availableTasks = new ArrayList<Task>();
	111	int size = tasks.size();
329	112
330		~~List<Task> readyTasks = new ArrayList<Task>();~~
331
332		~~int size = tasks.size();~~
333	113	for(int i = 0; i < size; i++){
334	114	int parCnt;
335		int previousTask~~Ready~~Cnt = 0;
	115	int previousTaskSucceedCnt = 0;
336	116	Task task = tasks.get(i);
337	117	if(task.getStatus() != (int)BrokerConstants.TASK_STATUS_UNSUBMITTED)
…	…
341	121	} catch(Exception e){
342	122	parCnt = 0;
343		~~//e.printStackTrace();~~
344	123	}
345		if(parCnt == 0)
346		{
347		readyTasks.add(task);
	124	if(parCnt == 0){
	125	availableTasks.add(task);
348	126	}
349		else
350		{
	127	else {
351	128	for(int j = 0; j < parCnt; j++){
352	129	ParentType par = task.getDescription().getWorkflow().getParent(j);
…	…
356	133	}
357	134	}
358		previousTask~~Ready~~Cnt++;
	135	previousTaskSucceedCnt++;
359	136	}
360	137
361		if(previousTask~~Ready~~Cnt == parCnt && task.getDescription().getWorkflow().getAnd() != null)
362		~~ready~~Tasks.add(task);
363		else if(previousTask~~Ready~~Cnt > 0 && task.getDescription().getWorkflow().getOr() != null)
364		~~ready~~Tasks.add(task);
365		else if (previousTask~~Ready~~Cnt == parCnt)
366		~~ready~~Tasks.add(task);
	138	if(previousTaskSucceedCnt == parCnt && task.getDescription().getWorkflow().getAnd() != null)
	139	availableTasks.add(task);
	140	else if(previousTaskSucceedCnt > 0 && task.getDescription().getWorkflow().getOr() != null)
	141	availableTasks.add(task);
	142	else if (previousTaskSucceedCnt == parCnt)
	143	availableTasks.add(task);
367	144	}
368	145	}
369		return ~~ready~~Tasks;
	146	return availableTasks;
370	147	}
371	148
…	…
380	157	return false;
381	158	}
	159
382	160	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/plan/ScheduledTaskInterface.java

r477	r539
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 ~~WorkloadUnit~~<?> getTask();
	194	public TaskInterface<?> getTask();
195	195
196	196	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/plan/SchedulingPlanInterface.java

r477	r539
99	99
100	100
101		public ArrayList<ScheduledTaskInterface> getTasks();
	101	public ArrayList<ScheduledTaskInterface<?>> getTasks();
102	102	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/plan/impl/ScheduledTask.java

r477	r539
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 ~~WorkloadUnit~~<?> task;
181		protected ArrayList<AllocationInterface> allocationList;
	178	protected TaskInterface<?> task;
	179	protected ArrayList<AllocationInterface<?>> allocationList;
182	180
183		public ScheduledTask(~~WorkloadUnit~~<?> task){
	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 ~~WorkloadUnit~~<?> getTask(){
	190	public TaskInterface<?> getTask(){
193	191	return this.task;
194	192	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/plan/impl/SchedulingPlan.java

r477	r539
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/build/classes/schedframe/scheduling/plugin/SchedulingPlugin.java

r477	r539
3	3	import schedframe.Plugin;
4	4	import schedframe.events.scheduling.SchedulingEvent;
5		import schedframe.scheduling.WorkloadUnitListImpl;
	5	import schedframe.scheduling.TaskList;
	6	import schedframe.scheduling.TaskListImpl;
6	7	import schedframe.scheduling.manager.resources.ResourceManager;
7	8	import schedframe.scheduling.manager.tasks.JobRegistry;
…	…
12	13	public interface SchedulingPlugin extends Plugin{
13	14
14		public int place~~JobsInQueues(WorkloadUnitListImpl newJob~~s,
	15	public int placeTasksInQueues(TaskList newTasks,
15	16	TaskQueueList queues,
16	17	ResourceManager resourceManager, ModuleList modules);

DCWoRMS/trunk/build/classes/schedframe/scheduling/plugin/estimation/ExecutionTimeEstimationPlugin.java

r477	r539
1	1	package schedframe.scheduling.plugin.estimation;
2	2
3		~~import gssim.schedframe.scheduling.ExecTask;~~
4	3
5		~~import java.util.List;~~
6	4	import java.util.Map;
	5
	6	import dcworms.schedframe.scheduling.ExecTask;
7	7
8	8	import schedframe.Plugin;
…	…
32	32	* @return estimated execution time of a task of specified length
33	33	*/
34		public double execTimeEstimation(SchedulingEvent event, Map<ResourceUnitName, ResourceUnit> allocatedResources,
35		ExecTask task,
	34	public double execTimeEstimation(SchedulingEvent event, ExecTask task, Map<ResourceUnitName, ResourceUnit> allocatedResources,
36	35	double completionPercentage);
37	36

DCWoRMS/trunk/build/classes/schedframe/scheduling/policy/AbstractManagementSystem.java

r477	r539
8	8	import org.joda.time.DateTimeUtilsExt;
9	9
	10	import dcworms.schedframe.scheduling.ExecTask;
	11	import dcworms.schedframe.scheduling.Executable;
	12	import dcworms.schedframe.scheduling.queues.AbstractStatsSupportingQueue;
	13
10	14	import schedframe.PluginConfiguration;
11	15	import schedframe.events.scheduling.SchedulingEventType;
12		~~import schedframe.resources.units.StandardResourceUnitName;~~
13	16	import schedframe.scheduling.Scheduler;
14	17	import schedframe.scheduling.WorkloadUnitHandler;
…	…
16	19	import schedframe.scheduling.manager.resources.ResourceManager;
17	20	import schedframe.scheduling.manager.resources.ResourceManagerFactory;
	21	import schedframe.scheduling.manager.tasks.JobRegistry;
18	22	import schedframe.scheduling.manager.tasks.JobRegistryImpl;
19		~~import schedframe.scheduling.manager.tasks.JobRegistry;~~
20	23	import schedframe.scheduling.plan.AllocationInterface;
21	24	import schedframe.scheduling.plan.SchedulingPlanInterface;
…	…
27	30	import schedframe.scheduling.queue.TaskQueueList;
28	31	import schedframe.scheduling.tasks.Job;
	32	import schedframe.scheduling.tasks.TaskInterface;
29	33	import schedframe.scheduling.tasks.WorkloadUnit;
30		import simulator.WormsConstants;
	34	import simulator.DCWormsConstants;
31	35	import eduni.simjava.Sim_event;
32	36	import gridsim.GridSim;
33	37	import gridsim.GridSimTags;
34		~~import gridsim.Gridlet;~~
35	38	import gridsim.IO_data;
36		import gridsim.gssim.WormsTags;
37		import gssim.schedframe.scheduling.ExecTask;
38		import gssim.schedframe.scheduling.Executable;
39		import gssim.schedframe.scheduling.queues.AbstractStatsSupportingQueue;
	39	import gridsim.dcworms.DCWormsTags;
40	40
41	41	public abstract class AbstractManagementSystem {
…	…
45	45	protected String name;
46	46
	47	protected TaskQueueList queues;
47	48	protected ResourceManager resourceManager;
48
49		protected ~~TaskQueueList queues~~;
50
	49	protected JobRegistryImpl jobRegistry;
	50	protected ModuleList moduleList;
	51
51	52	protected SchedulingPlugin schedulingPlugin;
52
53	53	protected ExecutionTimeEstimationPlugin execTimeEstimationPlugin;
54	54
55		protected ModuleList moduleList;
56
57		protected JobRegistryImpl jobRegistry;
	55	protected Scheduler scheduler;
58	56
59	57
…	…
67	65	}
68	66
	67	public void init(Scheduler sched, ManagedResources managedResources) {
	68	scheduler = sched;
	69	resourceManager = ResourceManagerFactory.createResourceManager(scheduler, managedResources);
	70	scheduler.set_stat(DCWormsConstants.getResourcesStatisticsObject(queues.size()));
	71	for(int i = 0; i < queues.size(); i++){
	72	TaskQueue q = queues.get(i);
	73	if(q instanceof AbstractStatsSupportingQueue<?>){
	74	AbstractStatsSupportingQueue<?> queue = (AbstractStatsSupportingQueue<?>) q;
	75	queue.setStats(scheduler.get_stat(), DCWormsConstants.TASKS_QUEUE_LENGTH_MEASURE_NAME + "_" + Integer.toString(i));
	76	}
	77	}
	78	}
	79
69	80	public void processEvent(Sim_event ev) {
70	81	processOtherEvent(ev);
…	…
81	92	}
82	93
	94
83	95	public String getName() {
84	96	return name;
85		}
86
87		~~public PluginConfiguration getSchedulingPluginConfiguration() {~~
88		~~return schedulingPlugin.getConfiguration();~~
89	97	}
90	98
…	…
106	114	return jobRegistry;
107	115	}
	116
	117	public Scheduler getScheduler() {
	118	return scheduler;
	119	}
	120
	121	public PluginConfiguration getSchedulingPluginConfiguration() {
	122	return schedulingPlugin.getConfiguration();
	123	}
108	124
109	125	public boolean pluginSupportsEvent(int eventType){
…	…
111	127	}
112	128
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
	129	public TaskQueueList getQueues(){
	130	return queues;
	131	}
	132
	133	public Map<String, Integer> getQueuesSize() {
	134	Map<String, Integer> queue_size = new HashMap<String, Integer>();
	135	for (TaskQueue queue : queues) {
	136	queue_size.put(queue.getName(), queue.size());
	137	}
	138	return queue_size;
	139	}
	140
123	141	//POPRAWIC (ale co? bo teraz chyba jest ok)
124		protected void submit~~WorkloadUnit(WorkloadUnit<?> wu, AllocationInterface~~ allocation) {
	142	protected void submitTask(TaskInterface<?> task, AllocationInterface<?> allocation) {
125	143	String providerName = allocation.getProviderName();
126	144	if (providerName == null) {
127	145	return;
128	146	}
129		//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;
164		}
165
166		protected boolean sendFinishedWorkloadUnit(WorkloadUnit<?> wu) {
	147	removeFromQueue(task);
	148	scheduler.send(providerName, GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, task);
	149	}
	150
	151	protected boolean sendFinishedWorkloadUnit(WorkloadUnit wu) {
167	152
168	153	Executable exec = (Executable) wu;
169	154	if(scheduler.getParent() == null)
170	155	{
171		Job job = jobRegistry.get(exec.getJobId());
	156	Job job = jobRegistry.getJob(exec.getJobId());
172	157
173	158	if(job.isFinished()){
174	159	scheduler.send(job.getSenderId(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_RETURN, job);
175	160	return true;
176		}else return true;
	161	}
	162	else return true;
177	163	}
178	164
…	…
183	169
184	170	protected void sendExecutableReadyEvent(ExecTask exec) {
185
186		~~/*if (wu instanceof JobInterface) {~~
187		~~scheduler.sendInternal(Long.valueOf(0).doubleValue(), GssimTags.TASK_READY_FOR_EXECUTION,~~
188		~~wu);~~
189		~~return;~~
190		~~}*/~~
191	171
192	172	long delay = 0;
…	…
201	181	}
202	182
203		scheduler.sendInternal(Long.valueOf(delay).doubleValue(), WormsTags.TASK_READY_FOR_EXECUTION,
	183	scheduler.sendInternal(Long.valueOf(delay).doubleValue(), DCWormsTags.TASK_READY_FOR_EXECUTION,
204	184	exec);
205	185	}
…	…
213	193	if (obj != null) {
214	194	int delay = (Integer) obj;
215		scheduler.sendInternal(delay, WormsTags.TIMER, null);
	195	scheduler.sendInternal(delay, DCWormsTags.TIMER, null);
216	196	}
217	197	}
…	…
219	199	}
220	200
221		protected boolean removeFromQueue(~~WorkloadUnit<?> wu~~) {
	201	protected boolean removeFromQueue(TaskInterface<?> task) {
222	202	for(TaskQueue queue : queues){
223		if(queue.contains(wu)){
224		queue.remove(wu);
	203	if(queue.contains(task)){
	204	queue.remove(task);
225	205	return true;
226	206	}
…	…
228	208	return false;
229	209	}
230
231		~~public TaskQueueList getAccessQueues(){~~
232		~~return queues;~~
233		}
234
235		~~public Map<String, Integer> getQueuesSize() {~~
236		~~Map<String, Integer> queue_size = new HashMap<String, Integer>();~~
237		~~for (TaskQueue queue : queues) {~~
238		~~queue_size.put(queue.getName(), queue.size());~~
239		}
240		~~return queue_size;~~
241		}
242
243		~~public void init(Scheduler sched, ManagedResources managedResources) {~~
244		~~scheduler = sched;~~
245		~~resourceManager = ResourceManagerFactory.createResourceManager(scheduler, managedResources);~~
246		~~scheduler.set_stat(WormsConstants.getResourcesStatisticsObject(queues.size()));~~
247		~~for(int i = 0; i < queues.size(); i++){~~
248		~~TaskQueue q = queues.get(i);~~
249		~~if(q instanceof AbstractStatsSupportingQueue<?>){~~
250		~~AbstractStatsSupportingQueue<?> queue = (AbstractStatsSupportingQueue<?>) q;~~
251		~~queue.setStats(scheduler.get_stat(), WormsConstants.TASKS_QUEUE_LENGTH_MEASURE_NAME + "_" + Integer.toString(i));~~
252		}
253		}
254		}
255
256		~~protected Scheduler scheduler;~~
257
258		~~public Scheduler getScheduler() {~~
259		~~return scheduler;~~
260		}
261	210
262	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);
263	218
264	219	}

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

r477	r539
11	11	import org.qcg.broker.schemas.schedulingplan.types.AllocationStatus;
12	12
	13	import dcworms.schedframe.scheduling.ExecTask;
	14	import dcworms.schedframe.scheduling.Executable;
	15
13	16	import qcg.shared.constants.BrokerConstants;
14
15		~~import schedframe.events.scheduling.EventReason;~~
16		~~import schedframe.events.scheduling.SchedulingEvent;~~
17	17	import schedframe.events.scheduling.TaskArrivedEvent;
18		~~import schedframe.events.scheduling.TaskCanceledEvent;~~
19	18	import schedframe.events.scheduling.TimerEvent;
20		import schedframe.scheduling.~~Scheduler~~;
	19	import schedframe.scheduling.TaskListImpl;
21	20	import schedframe.scheduling.WorkloadUnitHandler;
22		~~import schedframe.scheduling.WorkloadUnitListImpl;~~
23	21	import schedframe.scheduling.plan.AllocationInterface;
24	22	import schedframe.scheduling.plan.ScheduledTaskInterface;
…	…
31	29	import schedframe.scheduling.tasks.Job;
32	30	import schedframe.scheduling.tasks.JobInterface;
33		~~import schedframe.scheduling.tasks.SubmittedTask;~~
34	31	import schedframe.scheduling.tasks.Task;
35	32	import schedframe.scheduling.tasks.TaskInterface;
36	33	import schedframe.scheduling.tasks.WorkloadUnit;
37
38	34	import eduni.simjava.Sim_event;
39	35	import gridsim.GridSim;
40	36	import gridsim.GridSimTags;
41		~~import gridsim.Gridlet;~~
42	37	import gridsim.IO_data;
43		import gridsim.gssim.WormsTags;
44		import gssim.schedframe.scheduling.ExecTask;
45		import gssim.schedframe.scheduling.Executable;
	38	import gridsim.dcworms.DCWormsTags;
46	39
47	40	public class GlobalManagementSystem extends AbstractManagementSystem {
…	…
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);
…	…
81	71	}
82	72
83		public void notifySubmittedWorkloadUnit(WorkloadUnit~~<?>~~ wu, boolean ack) {
	73	public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) {
84	74	if (!pluginSupportsEvent(GridSimTags.GRIDLET_SUBMIT)) {
85	75	log.error("Plugin " + schedulingPlugin.getClass()
…	…
90	80
91	81	registerWorkloadUnit(wu);
92		/*Job job = (Job) wu;
93		jobRegistry.addJob(job);
94
95		if (log.isInfoEnabled())
96		log.info("Received job " + job.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis()));
97
	82
	83	}
	84
	85	private void registerWorkloadUnit(WorkloadUnit wu){
	86	if(!wu.isRegistered()){
	87	wu.register(jobRegistry);
	88	}
	89	wu.accept(getWorkloadUnitHandler());
	90	}
	91
	92
	93	protected void schedule(JobInterface<?> job){
98	94	List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>();
99	95	jobsList.add(job);
100		WorkloadUnitList readyWorkloadUnits = new WorkloadUnitList();
101		readyWorkloadUnits.addAll(jobRegistry.getReadyTasks(jobsList));
102		schedulingPlugin.placeJobsInQueues(readyWorkloadUnits, queues, getResourceManager(), moduleList);
103
104		schedule(new TaskArrivedEvent());*/
105
106		}
107
108		private void registerWorkloadUnit(WorkloadUnit<?> wu){
109		if(!wu.isRegistered()){
110		wu.register(jobRegistry);
111		}
112		wu.accept(getWorkloadUnitHandler());
113		}
114
115
116		protected void scheduleReadyTasks(Job job){
117		List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>();
118		jobsList.add(job);
119		WorkloadUnitListImpl readyWorkloadUnits = new WorkloadUnitListImpl();
120		readyWorkloadUnits.addAll(jobRegistry.getReadyTasks(jobsList));
121		schedulingPlugin.placeJobsInQueues(readyWorkloadUnits, queues, getResourceManager(), moduleList);
122
123		schedule(new TaskArrivedEvent());
124		}
125
126		protected void schedule(SchedulingEvent schedulingEvent) {
127
128		try {
129		SchedulingPlanInterface decision = schedulingPlugin.schedule(
130		schedulingEvent, queues, getJobRegistry(), getResourceManager(), moduleList);
131		if (decision == null)
132		return;
133
	96	TaskListImpl readyTasks = new TaskListImpl();
	97	readyTasks.addAll(jobRegistry.getAvailableTasks(jobsList));
	98
	99	schedulingPlugin.placeTasksInQueues(readyTasks, queues, getResourceManager(), moduleList);
	100	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(
	101	new TaskArrivedEvent(), queues, getJobRegistry(), getResourceManager(), moduleList);
	102	if (decision != null)
134	103	executeSchedulingPlan(decision);
135
136		} catch (Exception e) {
137		e.printStackTrace();
138		}
139		}
140
141		public void notifyReturnedWorkloadUnit(WorkloadUnit<?> wu) {
	104	}
	105
	106	public void notifyReturnedWorkloadUnit(WorkloadUnit wu) {
142	107	Executable exec = (Executable) wu;
143	108
…	…
149	114
150	115	try {
151		Job job = jobRegistry.get(exec.getJobId());
152		/*Task task = job.getTask(exec.getTaskId());
	116	Job job = jobRegistry.getJob(exec.getJobId());
	117	Task task = job.getTask(exec.getTaskId());
153	118	if(exec.getProcessesId() == null){
154	119	try {
155	120	task.setStatus(exec.getStatus());
156	121	} catch (Exception e) {
157		// TODO Auto-generated catch block
158		e.printStackTrace();
	122
159	123	}
160	124	} else {
…	…
167	131	}
168	132	}
169		}*/
	133	}
170	134
171	135	if(job.isFinished()){
…	…
173	137	}
174	138	else {
175		scheduleReadyTasks(job);
176		/*List<JobInterface<?>> jobs = new ArrayList<JobInterface<?>>();
177		jobs.add(jobRegistry.getJobInfo(job.getId()));
178		WorkloadUnitList readyWorkloadUnits = new WorkloadUnitList();
179		readyWorkloadUnits.addAll(jobRegistry.getReadyTasks(jobs));
180		schedulingPlugin.placeJobsInQueues(readyWorkloadUnits, queues,
181		getResourceManager(), moduleList);
182		schedule(new TaskArrivedEvent());*/
	139	schedule(job);
183	140	}
184	141
…	…
189	146	}
190	147
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();
	148	protected void executeSchedulingPlan(SchedulingPlanInterface<?> decision) {
	149
	150	ArrayList<ScheduledTaskInterface<?>> taskSchedulingDecisions = decision.getTasks();
218	151	for (int i = 0; i < taskSchedulingDecisions.size(); i++) {
219	152
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.get(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) {
	153	ScheduledTaskInterface<?> taskDecision = taskSchedulingDecisions.get(i);
	154
	155	//log.info(decision.getDocument());
	156
	157	String jobID = taskDecision.getJobId();
	158	String taskID = taskDecision.getTaskId();
	159
	160	// Task allocations that were rejected because of lack of resources or which were canceled and
	161	// not scheduled again are returned to the user.
	162	if(taskDecision.getStatus() == AllocationStatus.REJECTED){
	163	Job job = jobRegistry.getJob(jobID);
	164	scheduler.send(job.getSenderId(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_RETURN, job);
	165	continue;
	166	}
	167
	168	Task task = (Task) jobRegistry.getTaskInfo(jobID, taskID);
	169
	170	ArrayList<AllocationInterface<?>> allocations = taskDecision.getAllocations();
	171	for (int j = 0; j < allocations.size(); j++) {
	172
	173	AllocationInterface<?> allocation = allocations.get(j);
	174	Executable exec = createExecutable(task, allocation);
	175	submitTask(exec, allocation);
	176	task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED);
	177	}
	178	}
	179	}
	180
	181	private Executable createExecutable(Task task, AllocationInterface<?> allocation) {
	182
	183	String refersTo = allocation.getProcessGroupId(); // null;//allocation.getRefersTo();
	184	if(refersTo == null)
	185	refersTo = task.getId();
	186
	187	Executable exec = null;
	188
	189	if(refersTo.equals(task.getId())){
	190	exec = new Executable(task);
	191	} else {
	192	List<AbstractProcesses> processes = task.getProcesses();
	193	if(processes == null) {
	194	try {
	195	log.error("Allocation: " + allocation.getDocument() + "\nrefers to unknown task or processes set." +
	196	" Set correct value (task id or prcesses set id) for allocation refersTo attribute.");
	197	} catch (Exception e) {
	198	e.printStackTrace();
	199	}
	200	}
	201	boolean found = false;
	202	for(int j = 0; j < processes.size() && !found; j++){
	203	AbstractProcesses procesesSet = processes.get(j);
	204	if(refersTo.equals(procesesSet.getId())){
	205	exec = new Executable(task, procesesSet);
	206	found = true;
	207	}
	208	}
	209	if(!found){
	210	log.error("Allocation refers to unknown proceses set.");
	211	}
	212	}
	213
	214	exec.setReservationId(allocation.getReservationId());
	215
	216	/*HostInterface<?> host = allocation.getHost();
	217	ComputingResourceTypeInterface<?> crt = host.getMachineParameters();
	218	if(crt != null){
	219	ComputingResourceTypeItemInterface<?> crti = crt.getComputingResourceTypeItem(0);
	220	if(crti != null){
	221	ParameterPropertyInterface<?> properties[] = crti.getHostParameter().getProperty();
	222	for(int p = 0; p < properties.length; p++){
	223	ParameterPropertyInterface<?> property = properties[p];
	224	if("chosenCPUs".equals(property.getName())){
	225	Object cpuNames = property.getValue();
	226	exec.addSpecificResource(ResourceParameterName.FREECPUS, cpuNames);
	227	}
	228	}
	229	}
	230	}*/
	231	return exec;
	232	}
	233
	234	protected void submitTask(TaskInterface<?> task, AllocationInterface<?> allocation) {
254	235
255	236	String providerName = allocation.getProviderName();
…	…
257	238	return;
258	239	}
259		~~TaskInterface<?> task = (TaskInterface<?>) job;~~
260	240	removeFromQueue(task);
261	241
262	242	int resID = GridSim.getEntityId(providerName);
263		IO_data data = new IO_data(~~job~~, 0, resID);
	243	IO_data data = new IO_data(task, 0, resID);
264	244	scheduler.send(scheduler.getOutputPort(), GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, data);
265
266		~~//scheduler.send(providerName, GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, job);~~
	245	//scheduler.send(providerName, GridSimTags.SCHEDULE_NOW, GridSimTags.GRIDLET_SUBMIT, job);
	246
267	247	if(log.isDebugEnabled())
268		try {
269		log.debug("Submitted job " + job.getId() + " to " + providerName);
270		} catch (NoSuchFieldException e) {
271		// TODO Auto-generated catch block
272		e.printStackTrace();
273		}
	248	log.debug("Submitted job " + task.getId() + " to " + providerName);
	249
274	250	}
275	251
276	252	class GlobalWorkloadUnitHandler implements WorkloadUnitHandler{
277	253
278		public void handleJob(Job job){
279
280		jobRegistry.addJob(job);
	254	public void handleJob(JobInterface<?> job){
281	255	if (log.isInfoEnabled())
282	256	log.info("Received job " + job.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis()));
283	257
284		scheduleReadyTasks(job);
	258	jobRegistry.addJob(job);
	259	schedule(job);
285	260	}
286	261
…	…
292	267	throw new RuntimeException("Not implemented since it isn't expected that tasks are send directly to the global scheduler.");
293	268	}
294
295		~~public void handleSubmittedTask(SubmittedTask task) {~~
296		~~throw new RuntimeException("Not implemented since it isn't expected that tasks are send directly to the global scheduler.");~~
297		}
298	269	}
299	270
…	…
301	272	return new GlobalWorkloadUnitHandler();
302	273	}
303
304	274
305	275

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

r477	r539
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/build/classes/schedframe/scheduling/policy/local/LocalManagementSystem.java

r477	r539
1	1	package schedframe.scheduling.policy.local;
2	2
	3	import dcworms.schedframe.scheduling.ExecTask;
	4	import dcworms.schedframe.scheduling.Executable;
3	5	import eduni.simjava.Sim_event;
4	6	import eduni.simjava.Sim_system;
5	7	import gridsim.Accumulator;
6	8	import gridsim.GridSimTags;
7		~~import gridsim.Gridlet;~~
8	9	import gridsim.ResourceCalendar;
9		import gridsim.gssim.WormsTags;
10		import gridsim.gssim.filter.SubTaskFilter;
11		import gssim.schedframe.scheduling.ExecTask;
12		import gssim.schedframe.scheduling.Executable;
	10	import gridsim.dcworms.DCWormsTags;
	11	import gridsim.dcworms.filter.ExecTaskFilter;
13	12
14	13	import java.util.ArrayList;
…	…
27	26	import qcg.shared.constants.BrokerConstants;
28	27	import schedframe.ResourceController;
29		~~import schedframe.events.scheduling.EventReason;~~
30	28	import schedframe.events.scheduling.SchedulingEvent;
31	29	import schedframe.events.scheduling.SchedulingEventType;
32	30	import schedframe.events.scheduling.StartTaskExecutionEvent;
33		~~import schedframe.events.scheduling.TaskCanceledEvent;~~
34	31	import schedframe.events.scheduling.TaskFinishedEvent;
35	32	import schedframe.events.scheduling.TaskRequestedTimeExpiredEvent;
…	…
45	42	import schedframe.scheduling.ResourceHistoryItem;
46	43	import schedframe.scheduling.Scheduler;
47		import schedframe.scheduling.UsedResourceList;
	44	import schedframe.scheduling.TaskList;
	45	import schedframe.scheduling.TaskListImpl;
	46	import schedframe.scheduling.UsedResourcesList;
48	47	import schedframe.scheduling.WorkloadUnitHandler;
49		~~import schedframe.scheduling.WorkloadUnitListImpl;~~
50	48	import schedframe.scheduling.manager.resources.LocalResourceManager;
51	49	import schedframe.scheduling.manager.resources.ManagedResources;
…	…
63	61	import schedframe.scheduling.tasks.Job;
64	62	import schedframe.scheduling.tasks.JobInterface;
65		~~import schedframe.scheduling.tasks.SubmittedTask;~~
66	63	import schedframe.scheduling.tasks.Task;
67	64	import schedframe.scheduling.tasks.TaskInterface;
68	65	import schedframe.scheduling.tasks.WorkloadUnit;
69		import simulator.WormsConstants;
	66	import simulator.DCWormsConstants;
70	67	import simulator.utils.DoubleMath;
71	68
…	…
83	80
84	81	super(providerId, entityName, execTimeEstimationPlugin, queues);
85
86		~~//schedulingPlugin = (LocalSchedulingPlugin) InstanceFactory.createInstance(schedulingPluginClassName, LocalSchedulingPlugin.class);~~
87	82
88	83	if (schedPlugin == null) {
…	…
90	85	}
91	86	this.schedulingPlugin = schedPlugin;
92		~~accTotalLoad = new Accumulator(~~);
93		~~moduleList = new ModuleListImpl(1);~~
94
	87	this.moduleList = new ModuleListImpl(1);
	88
	89	this.accTotalLoad = new Accumulator();
95	90	}
96	91
97	92	public void init(Scheduler sched, ManagedResources managedResources) {
98	93	super.init(sched, managedResources);
99		~~//scheduler = sched;~~
100		~~//resourceManager = ResourceManagerFactory.createResourceManager(scheduler);~~
101	94	double load = 0;
102	95	accTotalLoad.add(load);
…	…
112	105	switch (tag) {
113	106
114		case WormsTags.TIMER:
	107	case DCWormsTags.TIMER:
115	108	if (pluginSupportsEvent(tag)) {
116	109	SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TIMER);
117		SchedulingPlanInterface decision = schedulingPlugin.schedule(event,
	110	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
118	111	queues, getJobRegistry(), getResourceManager(), moduleList);
119	112	executeSchedulingPlan(decision);
120	113	}
121	114	sendTimerEvent();
122
123	115	break;
124	116
125		case WormsTags.TASK_READY_FOR_EXECUTION:
126
127		ExecTask data = (ExecTask) ev.get_data();
128		try {
129		data.setStatus(Gridlet.READY);
	117	case DCWormsTags.TASK_READY_FOR_EXECUTION:
	118	ExecTask execTask = (ExecTask) ev.get_data();
	119	try {
	120	execTask.setStatus(DCWormsTags.READY);
130	121	if (pluginSupportsEvent(tag)) {
131		SchedulingEvent event = new StartTaskExecutionEvent(~~data.getJobId(), data~~.getId());
132		SchedulingPlanInterface decision = schedulingPlugin.schedule(event,
	122	SchedulingEvent event = new StartTaskExecutionEvent(execTask.getJobId(), execTask.getId());
	123	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
133	124	queues, getJobRegistry(), getResourceManager(), moduleList);
134	125	executeSchedulingPlan(decision);
…	…
139	130	break;
140	131
141		case WormsTags.TASK_EXECUTION_FINISHED:
	132	case DCWormsTags.TASK_EXECUTION_FINISHED:
142	133	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());
148		//task.setGridletStatus(Gridlet.SUCCESS);
149		//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()));
152		/*UsedResourceList<ResourceHistoryItem> lastUsedList = task.getUsedResources();
153		Map<ResourceUnitName, AbstractResourceUnit> lastUsed = lastUsedList.getLast()
154		.getResourceUnits();
155		getAllocationManager().freeResources(lastUsed);
156		ProcessingElements pes = (ProcessingElements) lastUsed.get(StandardResourceUnitName.PROCESSINGELEMENTS);
157		for (ComputingResource resource : pes) {
158		resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, task));
	134	execTask = (ExecTask) obj;
	135	if (execTask.getStatus() == DCWormsTags.INEXEC) {
	136
	137	finalizeExecutable(execTask);
	138	sendFinishedWorkloadUnit(execTask);
	139	log.debug(execTask.getJobId() + "_" + execTask.getId() + " finished execution on " + new DateTime());
	140	log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size()));
	141	if (pluginSupportsEvent(tag)) {
	142	SchedulingEvent event = new TaskFinishedEvent(execTask.getJobId(), execTask.getId());
	143	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
	144	queues, getJobRegistry(), getResourceManager(), moduleList);
	145	executeSchedulingPlan(decision);
159	146	}
160		SubTaskFilter filter = new SubTaskFilter(task.getGridletID(), GssimTags.TASK_REQUESTED_TIME_EXPIRED);
161		scheduler.sim_cancel(filter, null);
162		super.sendFinishJob((Executable) task.getGridlet());*/
163		}
	147	}
	148
	149	Job job = jobRegistry.getJob(execTask.getJobId());
	150	if(!job.isFinished()){
	151	getWorkloadUnitHandler().handleJob(job);
	152	}
	153	break;
	154
	155	case DCWormsTags.TASK_REQUESTED_TIME_EXPIRED:
	156	obj = ev.get_data();
	157	execTask = (Executable) obj;
164	158	if (pluginSupportsEvent(tag)) {
165		SchedulingEvent event = new Task~~FinishedEvent(task.getJobId(), t~~ask.getId());
166		SchedulingPlanInterface decision = schedulingPlugin.schedule(event,
	159	SchedulingEvent event = new TaskRequestedTimeExpiredEvent(execTask.getJobId(), execTask.getId());
	160	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
167	161	queues, getJobRegistry(), getResourceManager(), moduleList);
168	162	executeSchedulingPlan(decision);
169	163	}
170		~~Job job = jobRegistry.get(task.getJobId());~~
171		~~if(!job.isFinished()){~~
172		~~getWorkloadUnitHandler().handleJob(job);~~
173		}
174
175	164	break;
176		case WormsTags.TASK_REQUESTED_TIME_EXPIRED:
177		obj = ev.get_data();
178		task = (SubmittedTask) obj;
179		if (pluginSupportsEvent(tag)) {
180		SchedulingEvent event = new TaskRequestedTimeExpiredEvent(task.getJobId(), task.getId());
181		SchedulingPlanInterface decision = schedulingPlugin.schedule(event,
182		queues, getJobRegistry(), getResourceManager(), moduleList);
183		executeSchedulingPlan(decision);
184		}
185
186		break;
187		case WormsTags.UPDATE:
	165
	166	case DCWormsTags.UPDATE:
188	167	updateProcessingTimes(ev);
189	168	break;
190	169	}
191	170	}
192
193
194		public void notifyReturnedWorkloadUnit(WorkloadUnit<?> wu) {
195		if (pluginSupportsEvent(WormsTags.TASK_EXECUTION_FINISHED)) {
	171
	172	public void notifyReturnedWorkloadUnit(WorkloadUnit wu) {
	173	if (pluginSupportsEvent(DCWormsTags.TASK_EXECUTION_FINISHED)) {
196	174	SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TASK_FINISHED);
197		SchedulingPlanInterface decision = schedulingPlugin.schedule(event,
	175	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
198	176	queues, getJobRegistry(), getResourceManager(), moduleList);
199	177	executeSchedulingPlan(decision);
…	…
203	181	//}
204	182	}
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();
	183
	184	protected void executeSchedulingPlan(SchedulingPlanInterface<?> decision) {
	185
	186	ArrayList<ScheduledTaskInterface<?>> taskSchedulingDecisions = decision.getTasks();
238	187	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;
	188	ScheduledTaskInterface<?> taskDecision = taskSchedulingDecisions.get(i);
	189
	190	if (taskDecision.getStatus() == AllocationStatus.REJECTED) {
	191	continue;
	192	}
	193
	194	ArrayList<AllocationInterface<?>> allocations = taskDecision.getAllocations();
	195
	196	TaskInterface<?> task = taskDecision.getTask();
	197	for (int j = 0; j < allocations.size(); j++) {
	198
	199	AllocationInterface<?> allocation = allocations.get(j);
	200	if (allocation.isProcessing()) {
	201	ExecTask exec = (ExecTask) task;
	202	executeTask(exec, allocation.getRequestedResources());
	203	} else if(resourceManager.getSchedulerName(allocation.getProviderName()) != null){
	204	allocation.setProviderName(resourceManager.getSchedulerName(allocation.getProviderName()));
	205	submitTask(task, allocation);
	206	} else {
	207	ExecTask exec = (ExecTask) task;
	208	executeTask(exec, chooseResourcesForExecution(allocation.getProviderName(), exec));
245	209	}
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		}
	210	}
	211
297	212	}
298	213	}
299	214
300	215	protected void executeTask(ExecTask task, Map<ResourceUnitName, ResourceUnit> choosenResources) {
301		// Executable exec = (Executable) task;
302
303		SubmittedTask submittedTask = (SubmittedTask)task;
	216
	217	Executable exec = (Executable)task;
304	218	boolean allocationStatus = getAllocationManager().allocateResources(choosenResources);
305	219	if(allocationStatus == false)
306	220	return;
307	221	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();
	222
315	223	SchedulingEvent event = new SchedulingEvent(SchedulingEventType.START_TASK_EXECUTION);
316	224	int time = Double.valueOf(
317		execTimeEstimationPlugin.execTimeEstimation(event, ~~choosenResources, task, completionPercentage~~)).intValue();
	225	execTimeEstimationPlugin.execTimeEstimation(event, task, choosenResources, exec.getCompletionPercentage())).intValue();
318	226	log.debug(task.getJobId() + "_" + task.getId() + " starts executing on " + new DateTime()
319	227	+ " will finish after " + time);
…	…
322	230	return;
323	231
324		~~submittedTask~~.setEstimatedDuration(time);
	232	exec.setEstimatedDuration(time);
325	233	DateTime currentTime = new DateTime();
326	234	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);
334
	235	exec.addUsedResources(resHistItem);
335	236	try {
336		long expectedDuration = submittedTask.getExpectedDuration().getMillis() / 1000;
337		scheduler.sendInternal(expectedDuration, WormsTags.TASK_REQUESTED_TIME_EXPIRED, submittedTask);
	237	exec.setStatus(DCWormsTags.INEXEC);
	238	} catch (Exception e) {
	239	// TODO Auto-generated catch block
	240	e.printStackTrace();
	241	}
	242	scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, exec);
	243
	244	try {
	245	long expectedDuration = exec.getExpectedDuration().getMillis() / 1000;
	246	scheduler.sendInternal(expectedDuration, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec);
338	247	} 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()));
	248	double t = exec.getEstimatedDuration();
	249	scheduler.sendInternal(t, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec);
	250	}
	251
	252	log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size()));
345	253
346	254	PEUnit peUnit = (PEUnit)choosenResources.get(StandardResourceUnitName.PE);
347		if(peUnit instanceof ProcessingElements){
	255
	256	notifyComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
	257
	258	/*if(peUnit instanceof ProcessingElements){
348	259	ProcessingElements pes = (ProcessingElements) peUnit;
349	260	for (ComputingResource resource : pes) {
350		resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_~~FINISHED, submittedTask~~));
	261	resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, exec));
351	262	}
352	263	} else {
…	…
355	266	resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
356	267	} catch (ResourceException e) {
357
358		}
359		resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, submittedTask));
360		}
361		/*ProcessingElements pes = (ProcessingElements) choosenResources.get(StandardResourceUnitName.PE);
362		for (ComputingResource resource : pes) {
363		resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, submittedTask));
364		}*/
	268	return;
	269	}
	270	resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, exec));
	271	}
	272	*/
365	273
366	274	/*for(ExecTaskInterface etask : jobRegistry.getRunningTasks()){
…	…
371	279	}
372	280
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();
379		Map<ResourceUnitName, ResourceUnit> lastUsed = lastUsedList.getLast()
380		.getResourceUnits();
381		getAllocationManager().freeResources(lastUsed);
382
383		PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE);
384		if(peUnit instanceof ProcessingElements){
385		ProcessingElements pes = (ProcessingElements) peUnit;
386		for (ComputingResource resource : pes) {
387		resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask));
388		}
389		} else {
390		ComputingResource resource = null;
391		try {
392		resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
393		} catch (ResourceException e) {
394
395		}
396		resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask));
397		}
398		/*ProcessingElements pes = (ProcessingElements) lastUsed.get(StandardResourceUnitName.PE);
399		for (ComputingResource resource : pes) {
400		resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, subTask));
401		}*/
402		SubTaskFilter filter = new SubTaskFilter(subTask.getGridletID(), WormsTags.TASK_REQUESTED_TIME_EXPIRED);
	281	public void finalizeExecutable(ExecTask execTask){
	282
	283	Executable exec = (Executable)execTask;
	284	exec.finalizeExecutable();
	285
	286	ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_REQUESTED_TIME_EXPIRED);
403	287	scheduler.sim_cancel(filter, null);
404	288
405		Executable executable = (Executable) subTask.getGridlet();
406		Job job = jobRegistry.get(executable.getJobId());
407
408		Task task = null;
	289	Task task;
	290	Job job = jobRegistry.getJob(exec.getJobId());
409	291	try {
410		task = job.getTask(exec~~utable~~.getTaskId());
	292	task = job.getTask(exec.getTaskId());
411	293	} catch (NoSuchFieldException e) {
412		~~e.printStackTrace()~~;
413		}
414		if(exec~~utable~~.getProcessesId() == null){
415		try {
416		task.setStatus(exec~~utable~~.getStatus());
	294	return;
	295	}
	296	if(exec.getProcessesId() == null){
	297	try {
	298	task.setStatus(exec.getStatus());
417	299	} catch (Exception e) {
418	300	e.printStackTrace();
…	…
422	304	for(int i = 0; i < processesList.size(); i++){
423	305	AbstractProcesses processes = processesList.get(i);
424		if(processes.getId().equals(exec~~utable~~.getProcessesId())){
425		processes.setStatus(exec~~utable~~.getStatus());
	306	if(processes.getId().equals(exec.getProcessesId())){
	307	processes.setStatus(exec.getStatus());
426	308	break;
427	309	}
428	310	}
429	311	}
	312
	313	UsedResourcesList lastUsedList = exec.getUsedResources();
	314	Map<ResourceUnitName, ResourceUnit> lastUsed = lastUsedList.getLast()
	315	.getResourceUnits();
	316	getAllocationManager().freeResources(lastUsed);
	317
	318	PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE);
	319	notifyComputingResources(peUnit, EnergyEventType.TASK_FINISHED, exec);
	320	/*if(peUnit instanceof ProcessingElements){
	321	ProcessingElements pes = (ProcessingElements) peUnit;
	322	for (ComputingResource resource : pes) {
	323	resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec));
	324	}
	325	} else {
	326	ComputingResource resource = null;
	327	try {
	328	resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
	329	} catch (ResourceException e) {
	330	return;
	331	}
	332	resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_FINISHED, exec));
	333	}*/
	334
430	335	//sendFinishedWorkloadUnit(executable);
431	336	}
…	…
441	346	while (iter.hasNext()) {
442	347	ExecTask task = iter.next();
443		SubmittedTask subTask = (SubmittedTask)task;
444		UsedResourceList<ResourceHistoryItem> usedResourcesList = subTask.getUsedResources();
445		ResourceUnit unit = usedResourcesList.getLast().getResourceUnits()
	348	Executable exec = (Executable)task;
	349	exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * timeSpan/exec.getEstimatedDuration());
	350
	351	UsedResourcesList usedResourcesList = exec.getUsedResources();
	352	PEUnit peUnit = (PEUnit)usedResourcesList.getLast().getResourceUnits()
446	353	.get(StandardResourceUnitName.PE);
447
448		double load = getMIShare(timeSpan, (PEUnit) unit);
449		subTask.updateGridletFinishedSoFar(load);
	354	double load = getMIShare(timeSpan, peUnit);
450	355	addTotalLoad(load);
451	356	}
452	357	}
453
	358	private void notifyComputingResources(PEUnit peUnit, EnergyEventType eventType, Object obj){
	359
	360	if(peUnit instanceof ProcessingElements){
	361	ProcessingElements pes = (ProcessingElements) peUnit;
	362	for (ComputingResource resource : pes) {
	363	resource.handleEvent(new EnergyEvent(eventType, obj));
	364	}
	365	} else {
	366	ComputingResource resource = null;
	367	try {
	368	resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
	369	} catch (ResourceException e) {
	370	return;
	371	}
	372	resource.handleEvent(new EnergyEvent(eventType, obj));
	373	}
	374	}
	375
454	376	private double getMIShare(double timeSpan, PEUnit pes) {
455	377	double localLoad;
…	…
470	392	protected void updateProcessingTimes(Sim_event ev) {
471	393	updateProcessingProgress();
472		for (ExecTask task : jobRegistry.getRunningTasks()) {
473		~~SubmittedTask subTask = (SubmittedTask)t~~ask;
474		List<String> visitedResource = ~~subTask~~.getVisitedResources();
	394	for (ExecTask execTask : jobRegistry.getRunningTasks()) {
	395	Executable exec = (Executable)execTask;
	396	List<String> visitedResource = exec.getVisitedResources();
475	397	String originResource = ev.get_data().toString();
476	398	if(!ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), originResource)){
…	…
478	400	}
479	401
480		Map<ResourceUnitName, ResourceUnit> choosenResources = ~~subTask~~.getUsedResources().getLast().getResourceUnits();
481		~~double completionPercentage = (task.getLength() - subTask.getRemainingGridletLength())/task.getLength();~~
482		~~double time = execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.RESOURCE_STATE_CHANGED),~~
483		choosenResources, task, completionPercentage);
484
485		~~/*if(!subTask.getVisitedResources().contains(ev.get_data().toString()))~~ {
	402	Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getUsedResources().getLast().getResourceUnits();
	403	int time = Double.valueOf(execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.RESOURCE_STATE_CHANGED),
	404	execTask, choosenResources, exec.getCompletionPercentage())).intValue();
	405
	406	//check if the new estimated end time is equal to the previous one; if yes the continue without update
	407	if( DoubleMath.subtract((exec.getExecStartTime() + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + time)) == 0.0){
486	408	continue;
487		}*/
488		//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){
490		continue;
491		}
492		SubTaskFilter filter = new SubTaskFilter(subTask.getGridletID(), WormsTags.TASK_EXECUTION_FINISHED);
	409	}
	410	exec.setEstimatedDuration(time);
	411	ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED);
493	412	scheduler.sim_cancel(filter, null);
494		scheduler.sendInternal(time, WormsTags.TASK_EXECUTION_FINISHED, task);
495
	413	scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, execTask);
496	414	}
497	415	}
…	…
511	429	numberOfPE = numberOfPE + resUnit.getAmount();
512	430	}
513		~~//numberOfPE = getResourceManager().getPE().size();~~
514	431	} catch (Exception e) {
515	432	numberOfPE = 1;
…	…
532	449	ExecTask task) {
533	450
534		ResourceManager resourceManager = this.resourceManager;
	451	Map<ResourceUnitName, ResourceUnit> map = new HashMap<ResourceUnitName, ResourceUnit>();
	452	LocalResourceManager resourceManager = getResourceManager();
535	453	if(resourceName != null){
536	454	ComputingResource resource = null;
…	…
540	458	return null;
541	459	}
542
543	460	resourceManager = new LocalResourceManager(resource);
544	461	}
545		~~Map<ResourceUnitName, ResourceUnit> map = new HashMap<ResourceUnitName, ResourceUnit>();~~
546
547	462
548	463	int cpuRequest;
…	…
553	468	}
554	469
555		~~//PEUnit processingUnits = null;~~
556	470	if (cpuRequest != 0) {
557	471
558	472	List<ResourceUnit> availableUnits = null;
559	473	try {
560		availableUnits = ~~getResourceManager()~~.getPE();
	474	availableUnits = resourceManager.getPE();
561	475	} catch (ResourceException e) {
562	476	return null;
563	477	}
	478
564	479	List<ResourceUnit> choosenPEUnits = new ArrayList<ResourceUnit>();
565
566	480	for (int i = 0; i < availableUnits.size() && cpuRequest > 0; i++) {
567	481	PEUnit peUnit = (PEUnit) availableUnits .get(i);
…	…
576	490	return null;
577	491	}
578
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	492	map.put(StandardResourceUnitName.PE, choosenPEUnits.get(0));
607	493	}
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		}*/
	494
630	495	return map;
631	496	}
632
633
634
635		public void notifySubmittedWorkloadUnit(WorkloadUnit<?> job, boolean ack) {
	497
	498	public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) {
636	499	updateProcessingProgress();
637		registerWorkloadUnit(~~job~~);
638		}
639
640		private void registerWorkloadUnit(WorkloadUnit~~<?>~~ wu){
	500	registerWorkloadUnit(wu);
	501	}
	502
	503	private void registerWorkloadUnit(WorkloadUnit wu){
641	504	if(!wu.isRegistered()){
642	505	wu.register(jobRegistry);
…	…
647	510	class LocalWorkloadUnitHandler implements WorkloadUnitHandler{
648	511
649		public void handleJob(Job job){
	512	public void handleJob(JobInterface<?> job){
650	513
651	514	if (log.isInfoEnabled())
…	…
654	517	List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>();
655	518	jobsList.add(job);
656		~~WorkloadUnitListImpl readyTasks = new WorkloadUnit~~ListImpl();
657		for(Task task: jobRegistry.get~~Ready~~Tasks(jobsList)){
	519	TaskListImpl availableTasks = new TaskListImpl();
	520	for(Task task: jobRegistry.getAvailableTasks(jobsList)){
658	521	task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED);
659		readyTasks.add(task);
660		}
661
662		for(WorkloadUnit<?> e:readyTasks){
663		registerWorkloadUnit(e);
664		}
665		}
666
667		public void handleTask(TaskInterface<?> ti){
668		Task task = (Task)ti;
669
670		if(task.getProcesses() == null){
	522	availableTasks.add(task);
	523	}
	524
	525	for(TaskInterface<?> task: availableTasks){
	526	registerWorkloadUnit(task);
	527	}
	528	}
	529
	530	public void handleTask(TaskInterface<?> t){
	531	Task task = (Task)t;
	532	List<AbstractProcesses> processes = task.getProcesses();
	533
	534	if(processes == null \|\| processes.size() == 0){
671	535	Executable exec = new Executable(task);
672		~~exec.setUserID(task.getSenderId());~~
673		~~exec.setLength(task.getLength());~~
674	536	registerWorkloadUnit(exec);
675	537	} 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());
	538	for(int j = 0; j < processes.size(); j++){
	539	AbstractProcesses procesesSet = processes.get(j);
	540	Executable exec = new Executable(task, procesesSet);
682	541	registerWorkloadUnit(exec);
683	542	}
…	…
686	545
687	546	public void handleExecutable(ExecTask task){
	547
688	548	Executable exec = (Executable) task;
689
690		// int cost =
691		// this.resourceManager.getResourceCharacteristic().getResUnits() !=
692		// null ?
693		// this.resourceManager.getResourceCharacteristic().getResUnits().get(ResourceParameterName.COST).getAmount()
694		// : 1;
695
696		exec.visitResource(scheduler.get_name());
	549	jobRegistry.addExecTask(exec);
	550
	551	exec.trackResource(scheduler.get_name());
697	552	Scheduler parentScheduler = scheduler.getParent();
698		while (parentScheduler != null && !exec.getVisitedResources().contains(parentScheduler.get_name())) {
699		exec.visitResource(parentScheduler.get_name());
	553	List<String> visitedResource = exec.getVisitedResources();
	554	String [] visitedResourcesArray = visitedResource.toArray(new String[visitedResource.size()]);
	555	while (parentScheduler != null && !ArrayUtils.contains(visitedResourcesArray, parentScheduler.get_name())) {
	556	exec.trackResource(parentScheduler.get_name());
700	557	parentScheduler = parentScheduler.getParent();
701	558	}
702
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) {
	559	exec.setSchedulerName(scheduler.get_id());
	560
	561	TaskList newTasks = new TaskListImpl();
	562	newTasks.add(exec);
	563
	564	schedulingPlugin.placeTasksInQueues(newTasks, queues, getResourceManager(), moduleList);
	565
	566	if (exec.getStatus() == DCWormsTags.QUEUED) {
712	567	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	568	}
734	569	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/queue/TaskQueue.java

r477	r539
1	1	package schedframe.scheduling.queue;
2	2
3		import gridsim.Gridlet;
4		import gssim.schedframe.scheduling.queues.AbstractStatsSupportingQueue;
	3	import gridsim.dcworms.DCWormsTags;
5	4
6	5	import org.joda.time.DateTime;
7	6
8		import ~~schedframe.scheduling.tasks.WorkloadUnit~~;
	7	import dcworms.schedframe.scheduling.queues.AbstractStatsSupportingQueue;
9	8
10		public class TaskQueue extends AbstractStatsSupportingQueue<WorkloadUnit<?>> implements Queue<WorkloadUnit<?>>{
	9	import schedframe.scheduling.tasks.TaskInterface;
11	10
	11	public class TaskQueue extends AbstractStatsSupportingQueue<TaskInterface<?>> implements Queue<TaskInterface<?>>{
12	12
13	13	private static final long serialVersionUID = 6576299222910508209L;
…	…
17	17	protected boolean supportReservation;
18	18
19
20	19	public TaskQueue (boolean supportReservation){
21		name = "Queue";
22		priority = 0;
	20	this.name = "Queue";
	21	this.priority = 0;
23	22	this.supportReservation = supportReservation;
24
25	23	}
26	24
27		public boolean add(~~WorkloadUnit<?> wu~~){
	25	public boolean add(TaskInterface<?> task){
28	26	try {
29		~~wu.setStatus(Gridlet~~.QUEUED);
	27	task.setStatus(DCWormsTags.QUEUED);
30	28	} catch(Exception e){
31	29	throw new RuntimeException(e);
32	30	}
33		//updateStats();
34		return super.add(wu);
	31	return super.add(task);
35	32	}
36	33
37		public void add(int pos, ~~WorkloadUnit<?> wu~~){
	34	public void add(int pos, TaskInterface<?> task){
38	35	try {
39		~~wu.setStatus(Gridlet~~.QUEUED);
	36	task.setStatus(DCWormsTags.QUEUED);
40	37	} catch(Exception e){
41	38	throw new RuntimeException(e);
42	39	}
43		//updateStats();
44		super.add(pos, wu);
	40	super.add(pos, task);
45	41	}
46	42
47	43	public DateTime getArrivalTime(int pos) throws IndexOutOfBoundsException {
48		//return get(pos).getSubmissionTimeToBroker();
49		return null;
	44	return get(pos).getSubmissionTimeToBroker();
50	45	}
51	46
…	…
69	64	return supportReservation;
70	65	}
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	66	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/AbstractProcesses.java

r477	r539
3	3	import schedframe.scheduling.tasks.requirements.ResourceParameterName;
4	4
5		public abstract class AbstractProcesses~~<T> implements WorkloadUnit<T>~~{
	5	public abstract class AbstractProcesses implements WorkloadUnit{
6	6
7	7

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/Job.java

r477	r539
18	18	import schedframe.scheduling.WorkloadUnitHandler;
19	19	import schedframe.scheduling.manager.tasks.JobRegistryImpl;
20		~~import schedframe.scheduling.policy.AbstractManagementSystem;~~
21	20
22	21
…	…
179	178	}
180	179
181		public int getUserID(){
	180	public int getUserId(){
182	181	return this.senderId;
183	182	}
…	…
204	203
205	204	List<Task> readyTasks = new ArrayList<Task>();
206
207	205	int size = tasks.size();
	206
208	207	for(int i = 0; i < size; i++){
209	208	int parCnt;
…	…
216	215	} catch(Exception e){
217	216	parCnt = 0;
218		//e.printStackTrace();
219		}
220		if(parCnt == 0)
221		{
	217	}
	218	if(parCnt == 0) {
222	219	readyTasks.add(task);
223	220	}
224		else
225		{
	221	else {
226	222	for(int j = 0; j < parCnt; j++){
227	223	ParentType par = task.getDescription().getWorkflow().getParent(j);

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/JobInterface.java

r477	r539
11	11	*
12	12	*/
13		public interface JobInterface<T> extends WorkloadUnit<T> {
	13	public interface JobInterface<T> extends WorkloadUnit, DescriptionContainer<T> {
14	14
15	15	/**
…	…
18	18	* @throws NoSuchFieldException if there is no tasks for this job, and job id can not be obtained
19	19	*/
20		public abstract String getId() throws NoSuchFieldException;
	20	//public abstract String getId() throws NoSuchFieldException;
21	21
22	22	/**
…	…
44	44	* @return constant which represents current status of this job
45	45	*/
46		public int getStatus();
	46	//public int getStatus();
47	47
48		public int getUserID();
	48	//public int getUserID();
49	49	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/Processes.java

r477	r539
1	1	package schedframe.scheduling.tasks;
2
3		~~import java.util.List;~~
4	2
5	3	import org.qcg.broker.schemas.resreqs.ComputingResourceBaseTypeItem;
…	…
196	194	}
197	195
198		public org.qcg.broker.schemas.resreqs.Processes getDescription(){
199		return this.pr;
200		}
201
202		@Override
203		public List getTask() {
204		// TODO Auto-generated method stub
205		return null;
206		}
207
208		@Override
209		public TaskInterface getTask(String taskId) throws NoSuchFieldException {
210		// TODO Auto-generated method stub
211		return null;
212		}
213
214		@Override
215		public int getTaskCount() {
	196	public int getStatus() {
	197	return this.status;
	198	}
	199
	200	@Override
	201	public int getUserId() {
216	202	// TODO Auto-generated method stub
217	203	return 0;
218	204	}
219	205
220		@Override
221		public int getStatus() {
222		// TODO Auto-generated method stub
223		return 0;
224		}
225
226		@Override
227		public int getUserID() {
228		// TODO Auto-generated method stub
229		return 0;
230		}
231
232		@Override
233		public String getDocument() throws Exception {
234		// TODO Auto-generated method stub
235		return null;
236		}
	206
237	207
238	208	@Override

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/Task.java

r477	r539
30	30	import schedframe.scheduling.WorkloadUnitHandler;
31	31	import schedframe.scheduling.manager.tasks.JobRegistryImpl;
32		~~import schedframe.scheduling.policy.AbstractManagementSystem;~~
33	32	import schedframe.scheduling.tasks.requirements.ResourceParameterName;
34	33
…	…
38	37	*
39	38	*/
40		public class Task ~~/extends AbstractTask/~~ implements TaskInterface<org.qcg.broker.schemas.resreqs.Task> {
	39	public class Task implements TaskInterface<org.qcg.broker.schemas.resreqs.Task> {
41	40
42	41	protected static Unmarshaller unmarshaller;
…	…
72	71	private int senderId;
73	72	private long workloadLogWaitTime;
74		//String resPathHistory;
75
	73
76	74	public Task(org.qcg.broker.schemas.resreqs.Task task){
77	75	this.task = task;
…	…
80	78	this.brokerSubmitTime = null;
81	79	this.duration = null;
82		~~// this.gridletID_ = (getJobId() + "_" + getId()).hashCode();~~
83	80	prepareTopology();
84	81	}
…	…
91	88	this.brokerSubmitTime = null;
92	89	this.duration = null;
93		~~// this.gridletID_ = (getJobId() + getId()).hashCode();~~
94	90	prepareTopology();
95	91	}
…	…
269	265	}
270	266
271		public String getUserDn() {
	267	public String getUserDN() {
272	268	return this.task.getUserDN();
273	269	}
…	…
471	467	}*/
472	468
	469
473	470	@Override
474		public List <Task> getTask() {
475		List<Task> tasks = new ArrayList<Task>();
476		tasks.add(this);
477		return tasks;
478		}
479
480		@Override
481		public Task getTask(String taskId) throws NoSuchFieldException {
482		// TODO Auto-generated method stub
483		return null;
484		}
485
486		@Override
487		public int getTaskCount() {
	471	public int getUserId() {
488	472	// TODO Auto-generated method stub
489	473	return 0;
490	474	}
491	475
492		~~@Override~~
493		~~public int getUserID() {~~
494		~~// TODO Auto-generated method stub~~
495		~~return 0;~~
496		}
497
498	476	public boolean isRegistered() {
499	477	return isRegistered;

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/TaskInterface.java

r477	r539
6	6	import org.joda.time.ReadableDuration;
7	7
	8	import schedframe.DescriptionContainer;
8	9	import schedframe.scheduling.tasks.requirements.ResourceParameterName;
9	10
…	…
14	15	*
15	16	*/
16		public interface TaskInterface<T> extends WorkloadUnit<T> {
	17	public interface TaskInterface<T> extends WorkloadUnit, DescriptionContainer<T> {
17	18
18	19	/**
…	…
20	21	* @return task identifier
21	22	*/
22		public abstract String getId();
	23	//public abstract String getId();
23	24
24	25	/**
…	…
33	34	* submitted this task.
34	35	*/
35		public abstract String getUserDn();
	36	public abstract String getUserDN();
36	37
37	38	/**
…	…
87	88	*/
88	89	public long getLength();
89
90		/**
91		*
92		* @param length measured in instructions.
93		*/
94		public void setLength(long length);
95
	90
96	91	/**
97	92	*
98	93	* @return constant which represent current task status
99	94	*/
100		public int getStatus();
	95	//public int getStatus();
101	96
102	97
…	…
107	102	public List<AbstractProcesses> getProcesses(AbstractProcessesGroup processGroup);
108	103
109		public void setStatus(int status) throws Exception;
	104	//public void setStatus(int status) throws Exception;
110	105
111	106	public double getCpuCntRequest() throws NoSuchFieldException;
…	…
114	109
115	110	public long getWorkloadLogWaitTime();
116
117		//public void addToResPath(String resName);
118
119		//public String getResPath();
	111
120	112
121	113	}

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/WorkloadUnit.java

r477	r539
1	1	package schedframe.scheduling.tasks;
2	2
3		~~import java.util.List;~~
4
5		~~import schedframe.DescriptionContainer;~~
6	3	import schedframe.scheduling.WorkloadUnitHandler;
7	4	import schedframe.scheduling.manager.tasks.JobRegistryImpl;
8		~~import schedframe.scheduling.policy.AbstractManagementSystem;~~
9	5
10		public interface WorkloadUnit~~<T> extends DescriptionContainer<T>~~ {
	6	public interface WorkloadUnit {
11	7
12		/**
13		*
14		* @return job identifier
15		* @throws NoSuchFieldException if there is no tasks for this job, and job id can not be obtained
16		*/
17		public abstract String getId() throws NoSuchFieldException;
	8	public String getId();
18	9
19		/**
20		*
21		* @return list of tasks which belongs to this job
22		*/
23		public abstract List<? extends TaskInterface<?>> getTask();
	10	public int getUserId();
24	11
25		/**
26		*
27		* @param taskId
28		* @return task with specified taskId
29		* @throws NoSuchFieldException if task with taskId does not exist in this job
30		*/
31		~~public abstract TaskInterface<?> getTask(String taskId) throws NoSuchFieldException;~~
32
33		/**
34		*
35		* @return number of tasks in this job
36		*/
37		~~public abstract int getTaskCount();~~
38
39		/**
40		*
41		* @return constant which represents current status of this job
42		*/
43	12	public int getStatus();
44	13
45	14	public void setStatus(int status) throws Exception;
46
	15
47	16	public boolean isFinished();
48
49		public int getUserID();
50
	17
51	18	public boolean isRegistered();
52	19

Context Navigation

Legend:

DCWoRMS/trunk/build/classes/schedframe/scheduling/Cluster.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/GridResourceDiscovery.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/Scheduler.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/WorkloadUnitHandler.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/manager/resources/LocalResourceManager.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/manager/tasks/AbstractJobRegistry.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/manager/tasks/JobRegistry.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/manager/tasks/JobRegistryImpl.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/plan/ScheduledTaskInterface.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/plan/SchedulingPlanInterface.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/plan/impl/ScheduledTask.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/plan/impl/SchedulingPlan.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/plugin/SchedulingPlugin.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/plugin/estimation/ExecutionTimeEstimationPlugin.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/policy/AbstractManagementSystem.java

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

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

DCWoRMS/trunk/build/classes/schedframe/scheduling/policy/local/LocalManagementSystem.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/queue/TaskQueue.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/AbstractProcesses.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/Job.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/JobInterface.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/Processes.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/Task.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/TaskInterface.java

DCWoRMS/trunk/build/classes/schedframe/scheduling/tasks/WorkloadUnit.java

Download in other formats: