Context Navigation

LocalARManagementSystem.java @ 104

Revision 104, 19.9 KB checked in by wojtekp, 13 years ago (diff)

Property svn:mime-type set to `text/plain`

Line
1	package test.rewolucja.scheduling.implementation;
2
3	import eduni.simjava.Sim_event;
4	import gridsim.GridSimTags;
5	import gridsim.Gridlet;
6	import gridsim.gssim.GssimConstants;
7	import gridsim.gssim.GssimTags;
8	import gridsim.gssim.ResourceHistoryItem;
9	import gridsim.gssim.SubmittedTask;
10	import gridsim.gssim.policy.ARAllocationPolicy;
11	import gssim.schedframe.scheduling.AbstractExecutable;
12	import gssim.schedframe.scheduling.ExecTaskInterface;
13	import gssim.schedframe.scheduling.Executable;
14	import gssim.schedframe.scheduling.plugin.local.GssimTimeOperations;
15
16	import java.util.Collection;
17	import java.util.List;
18	import java.util.Map;
19
20	import org.apache.commons.logging.Log;
21	import org.apache.commons.logging.LogFactory;
22	import org.joda.time.DateTime;
23	import org.joda.time.DateTimeUtilsExt;
24
25	import schedframe.exceptions.ReservationException;
26	import schedframe.resources.units.ResourceUnit;
27	import schedframe.scheduling.AbstractResourceRequirements;
28	import schedframe.scheduling.AbstractTimeRequirements;
29	import schedframe.scheduling.Offer;
30	import schedframe.scheduling.Queue;
31	import schedframe.scheduling.Reservation;
32	import schedframe.scheduling.ResourceUsage;
33	import schedframe.scheduling.TaskInterface;
34	import schedframe.scheduling.TimeResourceAllocation;
35	import schedframe.scheduling.events.ReservationActiveEvent;
36	import schedframe.scheduling.events.SchedulingEvent;
37	import schedframe.scheduling.events.SchedulingEventType;
38	import schedframe.scheduling.events.SchedulingResponseType;
39	import schedframe.scheduling.events.StartTaskExecutionEvent;
40	import schedframe.scheduling.events.TaskRequestedTimeExpiredEvent;
41	import schedframe.scheduling.plugin.SchedulingPluginConfiguration;
42	import schedframe.scheduling.plugin.estimation.ExecTimeEstimationPlugin;
43	import schedframe.scheduling.plugin.grid.Module;
44	import schedframe.scheduling.plugin.local.LocalSchedulingARPlugin;
45	import schedframe.scheduling.utils.ResourceParameterName;
46	import test.rewolucja.reservation.LocalReservationManagerNew;
47	import test.rewolucja.reservation.ReservationNew;
48	import test.rewolucja.resources.description.ExecResourceDescription;
49	import test.rewolucja.scheduling.plan.SchedulingPlanInterfaceNew;
50	import test.rewolucja.scheduling.queue.GSSIMQueue;
51
52	public class LocalARManagementSystem extends LocalManagementSystem {
53
54	private Log log = LogFactory.getLog(ARAllocationPolicy.class);
55
56	protected LocalSchedulingARPlugin arSchedulingPlugin;
57
58	protected LocalReservationManagerNew reservationManager;
59
60
61	public LocalARManagementSystem(String providerID, String entityName,
62	String schedulingPluginClassName,
63	ExecTimeEstimationPlugin execTimeEstimationPlugin,
64	ExecResourceDescription resourceDescription) throws Exception {
65	super(providerID, entityName, schedulingPluginClassName,
66	execTimeEstimationPlugin, resourceDescription);
67	this.arSchedulingPlugin = (LocalSchedulingARPlugin) this.schedulingPlugin;
68	this.moduleList.add( new LocalReservationManagerNew(0, new GssimTimeOperations(), getName()));
69	// this.reservationManager = new LocalReservationManagerNew(0, new GssimTimeOperations(), this.resName);
70
71	for(int i = 0; i < moduleList.size();i++){
72	Module m = moduleList.get(i);
73	switch(m.getType()){
74	case LOCAL_RESERVATION_MANAGER: reservationManager = (LocalReservationManagerNew) m;
75	break;
76
77	}
78	}
79	}
80
81
82
83	public List<Offer> getOffer(AbstractTimeRequirements<?> timeRequirements,
84	AbstractResourceRequirements<?> resourceRequirements)
85	throws ReservationException {
86	updateProcessingProgress();
87	List<Offer> result = null;
88
89	LocalReservationManagerNew reservationManager = null;
90	for(int i = 0; i < moduleList.size();i++){
91	Module m = moduleList.get(i);
92	switch(m.getType()){
93	case LOCAL_RESERVATION_MANAGER: reservationManager = (LocalReservationManagerNew) m;
94	break;
95
96	}
97	}
98	result = arSchedulingPlugin.getOffers(timeRequirements,
99	resourceRequirements, jobRegistry, queues,
100	getResourceManager(), reservationManager);
101	return result;
102	}
103
104	public ReservationNew createReservation(
105	AbstractTimeRequirements<?> timeRequirements,
106	AbstractResourceRequirements<?> resourceRequirements)
107	throws ReservationException {
108	updateProcessingProgress();
109	ReservationNew result = null;
110
111	result = arSchedulingPlugin.createReservation(timeRequirements,
112	resourceRequirements, jobRegistry, queues,
113	getResourceManager(), reservationManager);
114
115	return result;
116
117	}
118
119	public List<ReservationNew> createReservation(ResourceUsage resourceUsage)
120	throws ReservationException {
121	updateProcessingProgress();
122
123	List<ReservationNew> result = null;
124
125	result = arSchedulingPlugin.createReservation(resourceUsage,
126	jobRegistry, queues, getResourceManager(),
127	reservationManager);
128
129	return result;
130	}
131
132	public ReservationNew commitReservation(ReservationNew reservation)
133	throws ReservationException {
134	updateProcessingProgress();
135
136	ReservationNew result = null;
137
138	result = arSchedulingPlugin.commitReservation(reservation, jobRegistry,
139	queues, getResourceManager(), reservationManager);
140
141	/*
142	* if("23".equals(reservation.getJobId())){ int a = 1; } Processors
143	* processors; try { processors = (Processors)
144	* reservation.getAllocatedResource().
145	* getResourceUnit(ResourceParameterName.CPUCOUNT); if(processors !=
146	* null){ Processors choosenResources =
147	* getResourceUnitsManager().chooseProcessors(processors);
148	* getResourceUnitsManager().createResourceReservation(reservation,
149	* choosenResources); } else { log.error("SOMETHING IS WRONG HERE. " +
150	* "THERE SHOULD BE SOME RESOURCE REQUIREMENT IN THIS RESERVATION.\n" +
151	* reservation); }
152	*
153	* } catch (NoSuchFieldException e) { e.printStackTrace(); }
154	*/
155	return result;
156
157	}
158
159	public ReservationNew commitReservation(ReservationNew reservation,
160	TimeResourceAllocation resourceUsage) throws ReservationException {
161	updateProcessingProgress();
162
163	ReservationNew result = null;
164
165	result = arSchedulingPlugin.commitReservation(reservation,
166	resourceUsage, jobRegistry, queues, getResourceManager(),
167	reservationManager);
168
169	return result;
170	}
171
172	public void modifyReservation(ReservationNew reservation,
173	TimeResourceAllocation resourceUsage) throws ReservationException {
174
175	updateProcessingProgress();
176
177	arSchedulingPlugin.modifyReservation(reservation, resourceUsage,
178	jobRegistry, queues, getResourceManager(),
179	reservationManager);
180
181	}
182
183	public void cancelReservation(ReservationNew reservation)
184	throws ReservationException {
185	updateProcessingProgress();
186
187	arSchedulingPlugin.cancelReservation(reservation, jobRegistry, queues,
188	reservationManager);
189	}
190
191	public ReservationNew.Status checkStatus(ReservationNew reservation)
192	throws ReservationException {
193	updateProcessingProgress();
194
195	ReservationNew.Status status = arSchedulingPlugin.getStatus(reservation,
196	reservationManager);
197
198	return status;
199	}
200
201	public Collection<ReservationNew> getReservations(){
202	return this.reservationManager.getReservations();
203	}
204
205	protected ExecTaskInterface findTask(Reservation r, List<Queue<? extends TaskInterface<?>>> queues){
206	ExecTaskInterface task = null;
207
208	for(int i = 0; i < queues.size() && task == null; i++){
209	Queue<? extends TaskInterface<?>> queue = queues.get(i);
210	for(int j = 0; j < queue.size() && task == null; j++){
211	TaskInterface<?> t = queue.get(j);
212	if(t.getJobId().equals(r.getJobId()) &&
213	t.getId().equals(r.getTaskId())){
214	task = (ExecTaskInterface)t;
215	}
216	}
217	}
218
219	return task;
220	}
221
222	private boolean checkFreeSlot(TaskInterface<?> task){
223
224	SubmittedTask submittedTask = (SubmittedTask) task;
225	if(submittedTask.getReservationID() != -1)
226	return true;
227
228	long expectedRuntime = getExpectedRuntime(task);
229	if(expectedRuntime < 0)
230	return false;
231
232	int cpuCnt = 0;
233	try {
234	cpuCnt = Double.valueOf(submittedTask.getCpuCntRequest()).intValue();
235	} catch (NoSuchFieldException e) {
236	cpuCnt = 1;
237	}
238
239	//local reservation manager sprawdza tylko zalozoene rezrwacje wiec potrzebuje poki co tylko czasow a nie ilosci cpu
240	ResourceComponents processors = new ResourceComponents(this.getName());
241	//Processors processors = new Processors(this.resName, this.getResourceUnitsManager().getResourcesOfType(ResourceType.CPU), cpuCnt);
242	DateTime startTime = new DateTime();
243	DateTime endTime = startTime.plusSeconds(Long.valueOf(expectedRuntime).intValue());
244
245	List<TimeResourceAllocation> usageList = null;
246
247	try {
248	usageList = reservationManager.resourceUsageList(processors, jobRegistry.getRunningTasks(), startTime, endTime);
249	long reqDuration = (long)expectedRuntime * 1000;
250	long duration = reqDuration;
251	int reqCpu = cpuCnt;
252	DateTime start = null;
253	for(int i = 0; i < usageList.size() && duration > 0; i++){
254	TimeResourceAllocation allocation = usageList.get(i);
255	ResourceUnit unit = allocation.getAllocatedResource().
256	getResourceUnit(ResourceParameterName.CPUCOUNT);
257	if(unit.getFreeAmount() >= reqCpu){
258	duration = duration - (allocation.getEndMillis() - allocation.getStartMillis());
259	if(start == null)
260	start = allocation.getStart();
261	} else {
262	duration = reqDuration;
263	start = null;
264	}
265	}
266	if(duration > 0){
267	return false;
268	}
269	} catch (NoSuchFieldException e) {
270	return false;
271	}
272
273	return true;
274	}
275
276	public long getExpectedRuntime(TaskInterface<?> task){
277
278	SubmittedTask submittedTask = (SubmittedTask) task;
279	long expectedDuration;
280	try {
281	expectedDuration = submittedTask.getExpectedDuration().getMillis()/1000;
282	} catch (NoSuchFieldException e) {
283	//log.warn(e.getMessage());
284	// assume that task without defined requested time will not be performed;
285	return -1;
286	//expectedDuration = execTimeEstimation(task);
287	//if executing time for a given task is not defined then assume that it
288	//can be performed anyway ant set expected duration to 0
289	//expectedDuration = 0;
290	}
291	long expectedRuntime = -1;
292	if(task.getStatus() == Gridlet.READY)
293	expectedRuntime = expectedDuration;
294	else if(task.getStatus() == Gridlet.INEXEC)
295	{
296	double execStartTime = submittedTask.getExecStartTime();
297	long currentTimeMillis = new DateTime().getMillis()/1000;
298	expectedRuntime = (Double.valueOf(execStartTime).longValue() + expectedDuration - currentTimeMillis);
299	}
300	return expectedRuntime;
301	}
302
303	public void processOtherEvent(Sim_event ev){
304	updateProcessingProgress();
305
306	int tag = ev.get_tag();
307	Object obj;
308
309	switch(tag){
310
311	case GssimTags.TIMER:
312
313	if(pluginSupportsEvent(tag)){
314	SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TIMER);
315	arSchedulingPlugin.schedule(event,
316	jobRegistry,
317	queues,
318	getResourceManager(),
319	this.moduleList);
320
321	arSchedulingPlugin.schedule(event,
322	jobRegistry,
323	queues,
324	getResourceManager(),
325	this.moduleList);
326	}
327
328	SchedulingPluginConfiguration config = (SchedulingPluginConfiguration)schedulingPlugin.getConfiguration();
329	if(config != null){
330	Map<SchedulingEventType, Object> events = config.getServedEvents();
331	if(events != null){
332	int delay = (Integer) events.get(SchedulingEventType.TIMER);
333
334	getLogicalResource().sendInternal( delay, GssimTags.TIMER, null);
335	//this.timerHandler.notify(delay, null);
336	}
337	}
338	break;
339
340	case GssimTags.TASK_READY_FOR_EXECUTION:
341	Executable data = (Executable) ev.get_data();
342	System.out.println("oooooooooooo"+data.getJobId());
343	try {
344	data.setGridletStatus(Gridlet.READY);
345
346	/* Reservation.Status r_stat = this.reservationManager.getReservStatus(data.getReservationId());
347	if(r_stat == Reservation.Status.ACTIVE){
348	ResourceUnitsManagerImpl manager = getResourceUnitsManager();
349	if(!manager.areResourcesReserved(data)){
350	Map<ResourceParameterName, ResourceUnit> resources = manager.chooseResourcesFor(data);
351	if(resources != null)
352	manager.createResourceReservation(data, resources);
353	}
354	}
355	*/
356	if(pluginSupportsEvent(tag)){
357	SchedulingEvent event = new StartTaskExecutionEvent(data.getJobId(), data.getId());
358	SchedulingPlanInterfaceNew decision= arSchedulingPlugin.schedule(event,
359	jobRegistry,
360	queues,
361	getResourceManager(),
362	this.moduleList);
363
364	arSchedulingPlugin.schedule(event,
365	jobRegistry,
366	queues,
367	getResourceManager(),
368	this.moduleList);
369
370	executeSchedulingPlan(decision);
371
372	}
373	} catch (Exception e) {
374	e.printStackTrace();
375	}
376
377	break;
378
379	case GssimTags.TASK_EXECUTION_FINISHED:
380	obj = ev.get_data();
381	SubmittedTask t = (SubmittedTask) obj;
382
383	if(t.getStatus() == Gridlet.INEXEC){
384	t.setGridletStatus(Gridlet.SUCCESS);
385	t.finalizeGridlet();
386	System.out.println("finished"+t.getJobId());
387	// this.finishedTasks.put(t.getJobId()+"_"+t.getId(), t);
388
389	super.sendFinishJob((AbstractExecutable)t.getGridlet());
390	if(pluginSupportsEvent(tag)) {
391	SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TASK_FINISHED);
392	SchedulingPlanInterfaceNew decision=arSchedulingPlugin.schedule(event,
393	jobRegistry,
394	queues,
395	getResourceManager(),
396	this.moduleList);
397	executeSchedulingPlan(decision);
398
399	}
400	}
401
402	break;
403
404	case GridSimTags.AR_STATUS_ACTIVE:
405	obj = ev.get_data();
406	String resId = (String) obj;
407	List<ReservationNew> list = reservationManager.getReservations();
408	for(int i = 0; i < list.size(); i++){
409	ReservationNew r = list.get(i);
410	if(r.getId().equals(resId)){
411	r.setStatus(ReservationNew.Status.ACTIVE);
412	long delay = (r.getEndMillis() - DateTimeUtilsExt.currentTimeMillis()) / 1000;
413	log.info("reservation for " + r.getJobId() + " ends after: " + delay);
414	// ExecTaskInterface<?> task = findTask(r, queues);
415
416	// task may not be submitted to the resource yet. This may happen when task submit time
417	// and reservation start time are equal
418	// if(task != null){
419	// Map<ResourceParameterName, ResourceUnit> choosenResources = getResourceUnitsManager().chooseResourcesFor(task);
420	// getResourceUnitsManager().createResourceReservation(task, choosenResources);
421	// }
422
423	getLogicalResource().sendInternal(Long.valueOf(delay).doubleValue(), GssimTags.AR_STATUS_COMPLETED, r);
424	if(pluginSupportsEvent(tag)){
425	SchedulingPlanInterfaceNew decision= arSchedulingPlugin.schedule(new ReservationActiveEvent(r),
426	jobRegistry,
427	queues,
428	getResourceManager(),
429	this.moduleList);
430	executeSchedulingPlan(decision);
431	}
432	break;
433	}
434	}
435
436	break;
437
438	case GridSimTags.AR_STATUS_EXPIRED:
439	log.error("Implement reservation expiration handler");
440	break;
441
442	case GridSimTags.AR_STATUS_COMPLETED:
443	ReservationNew reservation = (ReservationNew) ev.get_data();
444	reservation.setStatus(ReservationNew.Status.FINISHED);
445	log.debug("finishing reservation: " + reservation);
446	String jt = reservation.getJobId() + "_" + reservation.getTaskId();
447	/* SubmittedTask task = this.finishedTasks.get(jt);
448
449	if(task != null){
450	List<Map<ResourceParameterName, ResourceUnit>> lastUsed = task.getUsedResources();
451	getResourceUnitsManager().removeResourceReservation(lastUsed.get(lastUsed.size() - 1));
452	this.finishedTasks.remove(jt);
453	return;
454	}
455	*/
456	int index;
457	for(index = 0; index < jobRegistry.getRunningTasks().size(); index++){
458	SubmittedTask st = (SubmittedTask) jobRegistry.getRunningTasks().get(index);
459
460	if(st.getReservationID() == Integer.valueOf(reservation.getId()).intValue()){
461
462	Integer gridletId = Integer.valueOf(st.getGridletID());
463	Map<String, Object> gridletHistory = history.get(gridletId);
464	DateTime expectedEndTime = (DateTime) gridletHistory.get(GssimConstants.END_TIME);
465	DateTime currentTime = new DateTime();
466	if(expectedEndTime.isAfter(currentTime)){
467	st.setGridletStatus(Gridlet.CANCELED);
468	st.finalizeGridlet();
469	gridletHistory.put(GssimConstants.END_TIME, currentTime);
470	log.debug("stop task execution: " + jt);
471	super.sendCancelJob(GridSimTags.GRIDLET_CANCEL, (AbstractExecutable)st.getGridlet(), gridletId, st.getUserID());
472	List<ResourceHistoryItem> lastUsed = st.getUsedResources();
473	//getResourceUnitsManager().removeResourceReservation(lastUsed.get(lastUsed.size() - 1).getResourceUnits());
474	return;
475	}
476	}
477	}
478
479
480
481	GSSIMQueue queue = queues.get(0);
482	String jobId = reservation.getJobId();
483	String taskId = reservation.getTaskId();
484
485	for(int i = 0; i < queue.size(); i++){
486	ExecTaskInterface queuedTask = (ExecTaskInterface) queue.get(i);
487	System.out.println(jobId+";"+taskId);
488	if(jobId.equals(queuedTask.getJobId()) && taskId.equals(queuedTask.getId())){
489	System.out.println(jobId+";"+taskId+ "failed");
490	SubmittedTask st = (SubmittedTask) queuedTask;
491	st.setGridletStatus(Gridlet.FAILED);
492	st.finalizeGridlet();
493	queue.remove(i);
494	log.debug("task didnt event start. remove from queue");
495	super.sendCancelJob(GridSimTags.GRIDLET_CANCEL, (AbstractExecutable)st.getGridlet(), st.getGridletID(), st.getUserID());
496	break;
497	}
498	}
499
500	break;
501	case GssimTags.TASK_REQUESTED_TIME_EXPIRED:
502	obj = ev.get_data();
503	t = (SubmittedTask) obj;
504	if(pluginSupportsEvent(tag)) {
505	SchedulingEvent event = new TaskRequestedTimeExpiredEvent(t.getJobId(), t.getId());
506	SchedulingResponseType responseEvent = arSchedulingPlugin.handleResourceAllocationViolation(event,
507	jobRegistry,
508	queues,
509	getResourceManager(),
510	this.moduleList);
511	if(responseEvent == SchedulingResponseType.TERMINATE_TASK){
512	if(t.getStatus() == Gridlet.INEXEC){
513	t.setGridletStatus(Gridlet.SUCCESS);
514	t.finalizeGridlet();
515	Integer gridletId = Integer.valueOf(t.getGridletID());
516	Map<String, Object> gridletHistory = history.get(gridletId);
517	DateTime currentTime = new DateTime();
518	gridletHistory.put(GssimConstants.END_TIME, currentTime);
519	super.sendFinishJob((AbstractExecutable)t.getGridlet());
520	}
521	}
522	else if(responseEvent == SchedulingResponseType.KILL_TASK){
523	if(t.getStatus() == Gridlet.INEXEC){
524	t.setGridletStatus(Gridlet.CANCELED);
525	t.finalizeGridlet();
526	Integer gridletId = Integer.valueOf(t.getGridletID());
527	Map<String, Object> gridletHistory = history.get(gridletId);
528	DateTime currentTime = new DateTime();
529	gridletHistory.put(GssimConstants.END_TIME, currentTime);
530	log.debug("stop task execution: " + t.getJobId() + "_" + t.getId());
531	super.sendCancelJob(GridSimTags.GRIDLET_CANCEL, (AbstractExecutable)t.getGridlet(), gridletId, t.getUserID());
532	}
533	}
534	else if(responseEvent == SchedulingResponseType.STOP_AND_RESUME_FROM_CHECKPOINT){
535	if(t.getStatus() == Gridlet.INEXEC){
536	jobPause(t.getGridletID(), t.getUserID(), false);
537	jobResume(t.getGridletID(), t.getUserID(), false);
538	this.communicationInterface.cancel(GssimTags.TASK_REQUESTED_TIME_EXPIRED, t);
539
540	}
541	}
542	else if(responseEvent == SchedulingResponseType.ONE_HOUR_GRACE_PERIOD){
543	if(t.getStatus() == Gridlet.INEXEC){
544	this.getLogicalResource().sendInternal(Long.valueOf(3600).doubleValue(), GssimTags.TASK_REQUESTED_TIME_EXPIRED, t);
545	return;
546	}
547	}
548	else if(responseEvent == SchedulingResponseType.EXECUTE_ANYWAY){
549	if(t.getStatus() == Gridlet.INEXEC){
550	return;
551	}
552	}
553	if(pluginSupportsEvent(GssimTags.TASK_EXECUTION_FINISHED)){
554	SchedulingEvent newEvent = new SchedulingEvent(SchedulingEventType.TASK_FINISHED);
555	arSchedulingPlugin.schedule(newEvent,
556	jobRegistry,
557	queues,
558	getResourceManager(),
559	this.moduleList);
560	arSchedulingPlugin.schedule(newEvent,
561	jobRegistry,
562	queues,
563	getResourceManager(),
564	this.moduleList);
565
566	}
567
568	}
569
570	break;
571	}
572	}
573	}

Note: See TracBrowser for help on using the repository browser.

Download in other formats:

Original Format