Context Navigation

← Previous Revision
Latest Revision
Next Revision →
Normal
Revision Log

LocalARManagementSystem.java @ 104

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

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

Rev	Line
[104]	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