Context Navigation

← Previous Revision
Next Revision →
Normal
Revision Log

LocalManagementSystem.java @ 539

Revision 539, 19.6 KB checked in by wojtekp, 12 years ago (diff)

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

Rev	Line
[477]	1	package schedframe.scheduling.policy.local;
	2
[539]	3	import dcworms.schedframe.scheduling.ExecTask;
	4	import dcworms.schedframe.scheduling.Executable;
[477]	5	import eduni.simjava.Sim_event;
	6	import eduni.simjava.Sim_system;
	7	import gridsim.Accumulator;
	8	import gridsim.GridSimTags;
	9	import gridsim.ResourceCalendar;
[539]	10	import gridsim.dcworms.DCWormsTags;
	11	import gridsim.dcworms.filter.ExecTaskFilter;
[477]	12
	13	import java.util.ArrayList;
	14	import java.util.HashMap;
	15	import java.util.Iterator;
	16	import java.util.List;
	17	import java.util.Map;
	18
	19	import org.apache.commons.lang.ArrayUtils;
	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	import org.qcg.broker.schemas.schedulingplan.types.AllocationStatus;
	25
	26	import qcg.shared.constants.BrokerConstants;
	27	import schedframe.ResourceController;
	28	import schedframe.events.scheduling.SchedulingEvent;
	29	import schedframe.events.scheduling.SchedulingEventType;
	30	import schedframe.events.scheduling.StartTaskExecutionEvent;
	31	import schedframe.events.scheduling.TaskFinishedEvent;
	32	import schedframe.events.scheduling.TaskRequestedTimeExpiredEvent;
	33	import schedframe.exceptions.ResourceException;
	34	import schedframe.resources.computing.ComputingResource;
	35	import schedframe.resources.computing.profiles.energy.EnergyEvent;
	36	import schedframe.resources.computing.profiles.energy.EnergyEventType;
	37	import schedframe.resources.units.PEUnit;
	38	import schedframe.resources.units.ProcessingElements;
	39	import schedframe.resources.units.ResourceUnit;
	40	import schedframe.resources.units.ResourceUnitName;
	41	import schedframe.resources.units.StandardResourceUnitName;
	42	import schedframe.scheduling.ResourceHistoryItem;
	43	import schedframe.scheduling.Scheduler;
[539]	44	import schedframe.scheduling.TaskList;
	45	import schedframe.scheduling.TaskListImpl;
	46	import schedframe.scheduling.UsedResourcesList;
[477]	47	import schedframe.scheduling.WorkloadUnitHandler;
	48	import schedframe.scheduling.manager.resources.LocalResourceManager;
	49	import schedframe.scheduling.manager.resources.ManagedResources;
	50	import schedframe.scheduling.manager.resources.ResourceManager;
	51	import schedframe.scheduling.plan.AllocationInterface;
	52	import schedframe.scheduling.plan.ScheduledTaskInterface;
	53	import schedframe.scheduling.plan.SchedulingPlanInterface;
	54	import schedframe.scheduling.plugin.SchedulingPlugin;
	55	import schedframe.scheduling.plugin.estimation.ExecutionTimeEstimationPlugin;
	56	import schedframe.scheduling.plugin.grid.ModuleListImpl;
	57	import schedframe.scheduling.plugin.grid.ModuleType;
	58	import schedframe.scheduling.policy.AbstractManagementSystem;
	59	import schedframe.scheduling.queue.TaskQueueList;
	60	import schedframe.scheduling.tasks.AbstractProcesses;
	61	import schedframe.scheduling.tasks.Job;
	62	import schedframe.scheduling.tasks.JobInterface;
	63	import schedframe.scheduling.tasks.Task;
	64	import schedframe.scheduling.tasks.TaskInterface;
	65	import schedframe.scheduling.tasks.WorkloadUnit;
[539]	66	import simulator.DCWormsConstants;
[477]	67	import simulator.utils.DoubleMath;
	68
	69	public class LocalManagementSystem extends AbstractManagementSystem {
	70
	71	private Log log = LogFactory.getLog(LocalManagementSystem.class);
	72
	73	protected double lastUpdateTime;
	74
	75	protected Accumulator accTotalLoad;
	76
	77	public LocalManagementSystem(String providerId, String entityName, SchedulingPlugin schedPlugin,
	78	ExecutionTimeEstimationPlugin execTimeEstimationPlugin, TaskQueueList queues)
	79	throws Exception {
	80
	81	super(providerId, entityName, execTimeEstimationPlugin, queues);
	82
	83	if (schedPlugin == null) {
	84	throw new Exception("Can not create local scheduling plugin instance.");
	85	}
	86	this.schedulingPlugin = schedPlugin;
[539]	87	this.moduleList = new ModuleListImpl(1);
	88
	89	this.accTotalLoad = new Accumulator();
[477]	90	}
	91
	92	public void init(Scheduler sched, ManagedResources managedResources) {
	93	super.init(sched, managedResources);
	94	double load = 0;
	95	accTotalLoad.add(load);
	96	}
	97
	98	public void processEvent(Sim_event ev) {
	99
	100	updateProcessingProgress();
	101
	102	int tag = ev.get_tag();
	103	Object obj;
	104
	105	switch (tag) {
	106
[539]	107	case DCWormsTags.TIMER:
[477]	108	if (pluginSupportsEvent(tag)) {
	109	SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TIMER);
[539]	110	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
[477]	111	queues, getJobRegistry(), getResourceManager(), moduleList);
	112	executeSchedulingPlan(decision);
	113	}
	114	sendTimerEvent();
	115	break;
	116
[539]	117	case DCWormsTags.TASK_READY_FOR_EXECUTION:
	118	ExecTask execTask = (ExecTask) ev.get_data();
[477]	119	try {
[539]	120	execTask.setStatus(DCWormsTags.READY);
[477]	121	if (pluginSupportsEvent(tag)) {
[539]	122	SchedulingEvent event = new StartTaskExecutionEvent(execTask.getJobId(), execTask.getId());
	123	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
[477]	124	queues, getJobRegistry(), getResourceManager(), moduleList);
	125	executeSchedulingPlan(decision);
	126	}
	127	} catch (Exception e) {
	128	e.printStackTrace();
	129	}
	130	break;
	131
[539]	132	case DCWormsTags.TASK_EXECUTION_FINISHED:
[477]	133	obj = ev.get_data();
[539]	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);
[477]	146	}
	147	}
[539]	148
	149	Job job = jobRegistry.getJob(execTask.getJobId());
[477]	150	if(!job.isFinished()){
	151	getWorkloadUnitHandler().handleJob(job);
	152	}
	153	break;
[539]	154
	155	case DCWormsTags.TASK_REQUESTED_TIME_EXPIRED:
[477]	156	obj = ev.get_data();
[539]	157	execTask = (Executable) obj;
[477]	158	if (pluginSupportsEvent(tag)) {
[539]	159	SchedulingEvent event = new TaskRequestedTimeExpiredEvent(execTask.getJobId(), execTask.getId());
	160	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
[477]	161	queues, getJobRegistry(), getResourceManager(), moduleList);
	162	executeSchedulingPlan(decision);
	163	}
	164	break;
[539]	165
	166	case DCWormsTags.UPDATE:
[477]	167	updateProcessingTimes(ev);
	168	break;
	169	}
	170	}
	171
[539]	172	public void notifyReturnedWorkloadUnit(WorkloadUnit wu) {
	173	if (pluginSupportsEvent(DCWormsTags.TASK_EXECUTION_FINISHED)) {
[477]	174	SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TASK_FINISHED);
[539]	175	SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
[477]	176	queues, getJobRegistry(), getResourceManager(), moduleList);
	177	executeSchedulingPlan(decision);
	178	}
	179	//if(scheduler.getParent() != null){
	180	sendFinishedWorkloadUnit(wu);
	181	//}
	182	}
	183
[539]	184	protected void executeSchedulingPlan(SchedulingPlanInterface<?> decision) {
[477]	185
[539]	186	ArrayList<ScheduledTaskInterface<?>> taskSchedulingDecisions = decision.getTasks();
[477]	187	for (int i = 0; i < taskSchedulingDecisions.size(); i++) {
[539]	188	ScheduledTaskInterface<?> taskDecision = taskSchedulingDecisions.get(i);
[477]	189
[539]	190	if (taskDecision.getStatus() == AllocationStatus.REJECTED) {
	191	continue;
	192	}
[477]	193
[539]	194	ArrayList<AllocationInterface<?>> allocations = taskDecision.getAllocations();
[477]	195
[539]	196	TaskInterface<?> task = taskDecision.getTask();
	197	for (int j = 0; j < allocations.size(); j++) {
[477]	198
[539]	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));
[477]	209	}
[539]	210	}
[477]	211
	212	}
	213	}
	214
	215	protected void executeTask(ExecTask task, Map<ResourceUnitName, ResourceUnit> choosenResources) {
[539]	216
	217	Executable exec = (Executable)task;
[477]	218	boolean allocationStatus = getAllocationManager().allocateResources(choosenResources);
	219	if(allocationStatus == false)
	220	return;
	221	removeFromQueue(task);
[539]	222
[477]	223	SchedulingEvent event = new SchedulingEvent(SchedulingEventType.START_TASK_EXECUTION);
	224	int time = Double.valueOf(
[539]	225	execTimeEstimationPlugin.execTimeEstimation(event, task, choosenResources, exec.getCompletionPercentage())).intValue();
[477]	226	log.debug(task.getJobId() + "_" + task.getId() + " starts executing on " + new DateTime()
	227	+ " will finish after " + time);
	228
	229	if (time < 0.0)
	230	return;
	231
[539]	232	exec.setEstimatedDuration(time);
[477]	233	DateTime currentTime = new DateTime();
	234	ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime);
[539]	235	exec.addUsedResources(resHistItem);
	236	try {
	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);
[477]	243
	244	try {
[539]	245	long expectedDuration = exec.getExpectedDuration().getMillis() / 1000;
	246	scheduler.sendInternal(expectedDuration, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec);
[477]	247	} catch (NoSuchFieldException e) {
[539]	248	double t = exec.getEstimatedDuration();
	249	scheduler.sendInternal(t, DCWormsTags.TASK_REQUESTED_TIME_EXPIRED, exec);
[477]	250	}
[539]	251
	252	log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad(jobRegistry.getRunningTasks().size()));
[477]	253
[539]	254	PEUnit peUnit = (PEUnit)choosenResources.get(StandardResourceUnitName.PE);
[477]	255
[539]	256	notifyComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
	257
	258	/*if(peUnit instanceof ProcessingElements){
[477]	259	ProcessingElements pes = (ProcessingElements) peUnit;
	260	for (ComputingResource resource : pes) {
[539]	261	resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, exec));
[477]	262	}
	263	} else {
	264	ComputingResource resource = null;
	265	try {
	266	resource = ResourceController.getComputingResourceByName(peUnit.getResourceId());
	267	} catch (ResourceException e) {
[539]	268	return;
[477]	269	}
[539]	270	resource.handleEvent(new EnergyEvent(EnergyEventType.TASK_STARTED, exec));
[477]	271	}
[539]	272	*/
[477]	273
	274	/*for(ExecTaskInterface etask : jobRegistry.getRunningTasks()){
	275	System.out.println(etask.getJobId());
	276	for(String taskId: etask.getVisitedResources())
	277	System.out.println("====="+taskId);
	278	}*/
	279	}
	280
[539]	281	public void finalizeExecutable(ExecTask execTask){
[477]	282
[539]	283	Executable exec = (Executable)execTask;
	284	exec.finalizeExecutable();
[477]	285
[539]	286	ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_REQUESTED_TIME_EXPIRED);
[477]	287	scheduler.sim_cancel(filter, null);
	288
[539]	289	Task task;
	290	Job job = jobRegistry.getJob(exec.getJobId());
[477]	291	try {
[539]	292	task = job.getTask(exec.getTaskId());
[477]	293	} catch (NoSuchFieldException e) {
[539]	294	return;
[477]	295	}
[539]	296	if(exec.getProcessesId() == null){
[477]	297	try {
[539]	298	task.setStatus(exec.getStatus());
[477]	299	} catch (Exception e) {
	300	e.printStackTrace();
	301	}
	302	} else {
	303	List<AbstractProcesses> processesList = task.getProcesses();
	304	for(int i = 0; i < processesList.size(); i++){
	305	AbstractProcesses processes = processesList.get(i);
[539]	306	if(processes.getId().equals(exec.getProcessesId())){
	307	processes.setStatus(exec.getStatus());
[477]	308	break;
	309	}
	310	}
	311	}
[539]	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
[477]	335	//sendFinishedWorkloadUnit(executable);
	336	}
	337
	338	protected void updateProcessingProgress() {
	339	double timeSpan = DoubleMath.subtract(Sim_system.clock(), lastUpdateTime);
	340	if (timeSpan <= 0.0) {
	341	// don't update when nothing changed
	342	return;
	343	}
	344	lastUpdateTime = Sim_system.clock();
	345	Iterator<ExecTask> iter = jobRegistry.getRunningTasks().iterator();
	346	while (iter.hasNext()) {
	347	ExecTask task = iter.next();
[539]	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()
[477]	353	.get(StandardResourceUnitName.PE);
[539]	354	double load = getMIShare(timeSpan, peUnit);
[477]	355	addTotalLoad(load);
	356	}
	357	}
[539]	358	private void notifyComputingResources(PEUnit peUnit, EnergyEventType eventType, Object obj){
[477]	359
[539]	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
[477]	376	private double getMIShare(double timeSpan, PEUnit pes) {
	377	double localLoad;
	378	ResourceCalendar resCalendar = (ResourceCalendar) moduleList.getModule(ModuleType.RESOURCE_CALENDAR);
	379	if (resCalendar == null)
	380	localLoad = 0;
	381	else
	382	// 1 - localLoad_ = available MI share percentage
	383	localLoad = resCalendar.getCurrentLoad();
	384
	385	int speed = pes.getSpeed();
	386	int cnt = pes.getAmount();
	387
	388	double totalMI = speed * cnt * timeSpan * (1 - localLoad);
	389	return totalMI;
	390	}
	391
	392	protected void updateProcessingTimes(Sim_event ev) {
	393	updateProcessingProgress();
[539]	394	for (ExecTask execTask : jobRegistry.getRunningTasks()) {
	395	Executable exec = (Executable)execTask;
	396	List<String> visitedResource = exec.getVisitedResources();
[477]	397	String originResource = ev.get_data().toString();
	398	if(!ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), originResource)){
	399	continue;
	400	}
	401
[539]	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();
[477]	405
	406	//check if the new estimated end time is equal to the previous one; if yes the continue without update
[539]	407	if( DoubleMath.subtract((exec.getExecStartTime() + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + time)) == 0.0){
[477]	408	continue;
	409	}
[539]	410	exec.setEstimatedDuration(time);
	411	ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED);
[477]	412	scheduler.sim_cancel(filter, null);
[539]	413	scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, execTask);
[477]	414	}
	415	}
	416
	417	public double calculateTotalLoad(int size) {
	418	// background load, defined during initialization
	419	double load;
	420	ResourceCalendar resCalendar = (ResourceCalendar) moduleList.getModule(ModuleType.RESOURCE_CALENDAR);
	421	if (resCalendar == null)
	422	load = 0;
	423	else
	424	load = resCalendar.getCurrentLoad();
	425
	426	int numberOfPE = 0;
	427	try {
	428	for(ResourceUnit resUnit : getResourceManager().getPE()){
	429	numberOfPE = numberOfPE + resUnit.getAmount();
	430	}
	431	} catch (Exception e) {
	432	numberOfPE = 1;
	433	}
	434	double tasksPerPE = (double) size / numberOfPE;
	435	load += Math.min(1.0 - load, tasksPerPE);
	436
	437	return load;
	438	}
	439
	440	public Accumulator getTotalLoad() {
	441	return accTotalLoad;
	442	}
	443
	444	protected void addTotalLoad(double load) {
	445	accTotalLoad.add(load);
	446	}
	447
	448	private Map<ResourceUnitName, ResourceUnit> chooseResourcesForExecution(String resourceName,
	449	ExecTask task) {
	450
[539]	451	Map<ResourceUnitName, ResourceUnit> map = new HashMap<ResourceUnitName, ResourceUnit>();
	452	LocalResourceManager resourceManager = getResourceManager();
[477]	453	if(resourceName != null){
	454	ComputingResource resource = null;
	455	try {
	456	resource = resourceManager.getResourceByName(resourceName);
	457	} catch (ResourceException e) {
	458	return null;
	459	}
	460	resourceManager = new LocalResourceManager(resource);
	461	}
	462
	463	int cpuRequest;
	464	try {
	465	cpuRequest = Double.valueOf(task.getCpuCntRequest()).intValue();
	466	} catch (NoSuchFieldException e) {
	467	cpuRequest = 1;
	468	}
	469
	470	if (cpuRequest != 0) {
	471
	472	List<ResourceUnit> availableUnits = null;
	473	try {
[539]	474	availableUnits = resourceManager.getPE();
[477]	475	} catch (ResourceException e) {
	476	return null;
	477	}
[539]	478
[477]	479	List<ResourceUnit> choosenPEUnits = new ArrayList<ResourceUnit>();
	480	for (int i = 0; i < availableUnits.size() && cpuRequest > 0; i++) {
	481	PEUnit peUnit = (PEUnit) availableUnits .get(i);
	482	if(peUnit.getFreeAmount() > 0){
	483	int allocPE = Math.min(peUnit.getFreeAmount(), cpuRequest);
	484	cpuRequest = cpuRequest - allocPE;
	485	choosenPEUnits.add(peUnit.replicate(allocPE));
	486	}
	487	}
	488
	489	if(cpuRequest > 0){
	490	return null;
	491	}
	492	map.put(StandardResourceUnitName.PE, choosenPEUnits.get(0));
	493	}
	494
	495	return map;
	496	}
	497
[539]	498	public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) {
[477]	499	updateProcessingProgress();
[539]	500	registerWorkloadUnit(wu);
[477]	501	}
	502
[539]	503	private void registerWorkloadUnit(WorkloadUnit wu){
[477]	504	if(!wu.isRegistered()){
	505	wu.register(jobRegistry);
	506	}
	507	wu.accept(getWorkloadUnitHandler());
	508	}
	509
	510	class LocalWorkloadUnitHandler implements WorkloadUnitHandler{
	511
[539]	512	public void handleJob(JobInterface<?> job){
[477]	513
	514	if (log.isInfoEnabled())
	515	log.info("Received job " + job.getId() + " at " + new DateTime(DateTimeUtilsExt.currentTimeMillis()));
	516
	517	List<JobInterface<?>> jobsList = new ArrayList<JobInterface<?>>();
	518	jobsList.add(job);
[539]	519	TaskListImpl availableTasks = new TaskListImpl();
	520	for(Task task: jobRegistry.getAvailableTasks(jobsList)){
[477]	521	task.setStatus((int)BrokerConstants.TASK_STATUS_QUEUED);
[539]	522	availableTasks.add(task);
[477]	523	}
	524
[539]	525	for(TaskInterface<?> task: availableTasks){
	526	registerWorkloadUnit(task);
[477]	527	}
	528	}
	529
[539]	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){
[477]	535	Executable exec = new Executable(task);
	536	registerWorkloadUnit(exec);
	537	} else {
[539]	538	for(int j = 0; j < processes.size(); j++){
	539	AbstractProcesses procesesSet = processes.get(j);
	540	Executable exec = new Executable(task, procesesSet);
[477]	541	registerWorkloadUnit(exec);
	542	}
	543	}
	544	}
	545
	546	public void handleExecutable(ExecTask task){
[539]	547
[477]	548	Executable exec = (Executable) task;
[539]	549	jobRegistry.addExecTask(exec);
	550
	551	exec.trackResource(scheduler.get_name());
[477]	552	Scheduler parentScheduler = scheduler.getParent();
[539]	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());
[477]	557	parentScheduler = parentScheduler.getParent();
	558	}
[539]	559	exec.setSchedulerName(scheduler.get_id());
[477]	560
[539]	561	TaskList newTasks = new TaskListImpl();
	562	newTasks.add(exec);
[477]	563
[539]	564	schedulingPlugin.placeTasksInQueues(newTasks, queues, getResourceManager(), moduleList);
[477]	565
[539]	566	if (exec.getStatus() == DCWormsTags.QUEUED) {
[477]	567	sendExecutableReadyEvent(exec);
	568	}
	569	}
	570	}
	571
	572	public WorkloadUnitHandler getWorkloadUnitHandler() {
	573	return new LocalWorkloadUnitHandler();
	574	}
	575
	576
	577	public LocalResourceManager getResourceManager() {
	578	if (resourceManager instanceof ResourceManager)
	579	return (LocalResourceManager) resourceManager;
	580	else
	581	return null;
	582	}
	583
	584	}

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

Download in other formats:

Original Format