Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/Scheduler.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/Scheduler.java	(revision 970)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/Scheduler.java	(revision 1362)
@@ -136,5 +136,4 @@
 				break;
 			}
-			
 			// process the received event
 			processRequest(ev);
@@ -147,10 +146,7 @@
 
 		case GridSimTags.GRIDLET_SUBMIT:
-			processWorkloadUnitSubmit(ev, false);
-			break;
-
-		case GridSimTags.GRIDLET_SUBMIT_ACK:
-			processWorkloadUnitSubmit(ev, true);
-			break;
+			processWorkloadUnitSubmit(ev);
+			break;
+
 
 		case GridSimTags.GRIDLET_RETURN:
@@ -187,7 +183,7 @@
 	}
 
-	protected void processWorkloadUnitSubmit(Sim_event ev, boolean ack) {
+	protected void processWorkloadUnitSubmit(Sim_event ev) {
 		WorkloadUnit job = (WorkloadUnit) ev.get_data();
-		managementSystem.notifySubmittedWorkloadUnit(job, ack);
+		managementSystem.notifySubmittedWorkloadUnit(job);
 	}
 	
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/AbstractManagementSystem.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/AbstractManagementSystem.java	(revision 579)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/AbstractManagementSystem.java	(revision 1362)
@@ -212,5 +212,5 @@
 	public abstract WorkloadUnitHandler getWorkloadUnitHandler();
 	
-	public abstract void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack);
+	public abstract void notifySubmittedWorkloadUnit(WorkloadUnit wu);
 
 	public abstract void notifyReturnedWorkloadUnit(WorkloadUnit wu);
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/local/LocalManagementSystem.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/local/LocalManagementSystem.java	(revision 1357)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/local/LocalManagementSystem.java	(revision 1362)
@@ -1,18 +1,12 @@
 package schedframe.scheduling.policy.local;
-
-import dcworms.schedframe.scheduling.ExecTask;
-import dcworms.schedframe.scheduling.Executable;
-import eduni.simjava.Sim_event;
-import eduni.simjava.Sim_system;
-import gridsim.dcworms.DCWormsTags;
-import gridsim.dcworms.filter.ExecTaskFilter;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
-
-import org.apache.commons.lang.ArrayUtils;
+import java.util.Set;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -38,9 +32,9 @@
 import schedframe.resources.units.ResourceUnitName;
 import schedframe.resources.units.StandardResourceUnitName;
+import schedframe.scheduling.ExecutionHistoryItem;
 import schedframe.scheduling.ResourceHistoryItem;
 import schedframe.scheduling.Scheduler;
 import schedframe.scheduling.TaskList;
 import schedframe.scheduling.TaskListImpl;
-import schedframe.scheduling.UsedResourcesList;
 import schedframe.scheduling.WorkloadUnitHandler;
 import schedframe.scheduling.manager.resources.LocalResourceManager;
@@ -64,4 +58,10 @@
 import simulator.stats.GSSAccumulator;
 import simulator.utils.DoubleMath;
+import dcworms.schedframe.scheduling.ExecTask;
+import dcworms.schedframe.scheduling.Executable;
+import eduni.simjava.Sim_event;
+import eduni.simjava.Sim_system;
+import gridsim.dcworms.DCWormsTags;
+import gridsim.dcworms.filter.ExecTaskFilter;
 
 public class LocalManagementSystem extends AbstractManagementSystem {
@@ -127,6 +127,4 @@
 				finalizeExecutable(execTask);
 				sendFinishedWorkloadUnit(execTask);
-				log.debug(execTask.getJobId() + "_" + execTask.getId() + " finished execution on " + new DateTime());
-				log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad());
 				if (pluginSupportsEvent(tag)) {
 					SchedulingEvent event = new TaskFinishedEvent(execTask.getJobId(), execTask.getId());
@@ -164,12 +162,50 @@
 			break;
 			
+		case DCWormsTags.TASK_PAUSE:{
+			String[] ids = (String[]) ev.get_data();
+			execTask = jobRegistry.getTask(ids[0], ids[1]);
+			taskPause(execTask);
+			if (pluginSupportsEvent(tag)) {
+				SchedulingEvent event = new SchedulingEvent(SchedulingEventType.TASK_PAUSED);
+				SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
+						queues, getJobRegistry(), getResourceManager(), moduleList);
+				executeSchedulingPlan(decision);
+			}
+		}
+		break;
+		
+		case DCWormsTags.TASK_RESUME:{
+			String[] ids = (String[]) ev.get_data();
+			execTask = jobRegistry.getTask(ids[0], ids[1]);
+			taskResume(execTask, execTask.getAllocatedResources().getLast().getResourceUnits());
+			if (pluginSupportsEvent(tag)) {
+				SchedulingEvent event = new StartTaskExecutionEvent(ids[0], ids[1]);
+				SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
+						queues, getJobRegistry(), getResourceManager(), moduleList);
+				executeSchedulingPlan(decision);
+			}
+		}
+		break;
+			
+		case DCWormsTags.TASK_MOVE:{
+			Object[] data = (Object[]) ev.get_data();
+			execTask = jobRegistry.getTask((String)data[0], (String)data[1]);
+			taskMove(execTask, (Map<ResourceUnitName, ResourceUnit>)data[2]);
+			if (pluginSupportsEvent(tag)) {
+				SchedulingEvent event = new StartTaskExecutionEvent((String)data[0], (String)data[1]);
+				SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
+						queues, getJobRegistry(), getResourceManager(), moduleList);
+				executeSchedulingPlan(decision);
+			}
+		}
+		break;
 			
 		case DCWormsTags.TASK_EXECUTION_CHANGED:
 			execTask = (ExecTask) ev.get_data();
-			updateTaskExecution(execTask, SchedulingEventType.RESOURCE_STATE_CHANGED);
+			updateTaskExecutionPhase(execTask, SchedulingEventType.RESOURCE_STATE_CHANGED);
 			break;
-			
+
 		case DCWormsTags.UPDATE_PROCESSING:
-			updateProcessingTimes(ev);
+			updateProcessingTimes();
 			break;
 			
@@ -177,6 +213,7 @@
 			if (pluginSupportsEvent(tag)) {
 				SchedulingEvent event = new SchedulingEvent(SchedulingEventType.POWER_LIMIT_EXCEEDED);
-				schedulingPlugin.schedule(event,
+				SchedulingPlanInterface<?> decision = schedulingPlugin.schedule(event,
 						queues,  getJobRegistry(), getResourceManager(), moduleList);
+				executeSchedulingPlan(decision);
 			}
 			break;
@@ -184,4 +221,63 @@
 	}
 
+
+	public void taskPause(ExecTask execTask) {
+		if (execTask == null) {
+			return;
+		} else {
+			try {
+				execTask.setStatus(DCWormsTags.PAUSED);
+				
+				Executable exec = (Executable) execTask;
+				Map<ResourceUnitName, ResourceUnit> lastUsed = exec.getAllocatedResources().getLast().getResourceUnits();
+				getAllocationManager().freeResources(lastUsed);
+				
+				saveExecutionHistory(exec, exec.getCompletionPercentage(), exec.getEstimatedDuration());
+				
+				ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), -1);
+				scheduler.sim_cancel(filter, null);
+				
+				PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE);
+				updateComputingResources(peUnit, EnergyEventType.TASK_FINISHED, exec);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+		}		
+	}
+	
+	public void taskResume(ExecTask execTask, Map<ResourceUnitName, ResourceUnit> resources) {
+		if (execTask == null) {
+			return;
+		} else if (execTask.getStatus() == DCWormsTags.PAUSED) {
+			try {
+				execTask.setStatus(DCWormsTags.RESUMED);
+				Executable exec = (Executable) execTask;
+
+				boolean status = allocateResources(exec, resources);
+				if(status == false){
+					TaskList newTasks = new TaskListImpl();
+					newTasks.add(exec);
+					schedulingPlugin.placeTasksInQueues(newTasks, queues, getResourceManager(), moduleList);
+					exec.setStatus(DCWormsTags.READY);
+				} else {
+					runTask(execTask);
+					
+					PEUnit peUnit = (PEUnit)resources.get(StandardResourceUnitName.PE);
+					updateComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
+				}
+
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+		}			
+	}
+	
+	public void taskMove(ExecTask execTask, Map<ResourceUnitName, ResourceUnit> map) {
+		taskPause(execTask);
+		taskResume(execTask, map);
+	}
+	
 	public void notifyReturnedWorkloadUnit(WorkloadUnit wu) {
 		if (pluginSupportsEvent(DCWormsTags.TASK_EXECUTION_FINISHED)) {
@@ -229,5 +325,5 @@
 
 		Executable exec = (Executable)task;
-		boolean allocationStatus = getAllocationManager().allocateResources(choosenResources);
+		boolean allocationStatus = allocateResources(exec, choosenResources);
 		if(allocationStatus == false){
 			log.info("Task " + task.getJobId() + "_" + task.getId() + " requires more resources than is available at this moment.");
@@ -239,25 +335,9 @@
 		log.debug(task.getJobId() + "_" + task.getId() + " starts executing on " + new DateTime());
 
-		//if (phaseDuration < 0.0)
-		//	return;
-
-		//exec.setEstimatedDuration(exec.getEstimatedDuration() + phaseDuration);
-		DateTime currentTime = new DateTime();
-		ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime);
-		resHistItem.setCompletionPercentage(0);
-		exec.addUsedResources(resHistItem);
-		
-		try {
-			exec.setStatus(DCWormsTags.INEXEC);
-		} catch (Exception e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-		}
+		runTask(exec);
 		
 		PEUnit peUnit = (PEUnit)choosenResources.get(StandardResourceUnitName.PE);
 		updateComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
 		
-		updateTaskExecution(exec, SchedulingEventType.START_TASK_EXECUTION);
-		//scheduler.sendInternal(time, DCWormsTags.TASK_EXECUTION_FINISHED, exec);
 
 		try {
@@ -269,8 +349,32 @@
 		}
 
-		log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad());
-		
-	}
-	
+		log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad());		
+	}
+	
+	private void runTask(ExecTask execTask){
+		Executable exec = (Executable) execTask;
+		Map<ResourceUnitName, ResourceUnit> resources = exec.getAllocatedResources().getLast().getResourceUnits();
+		
+		try {
+			exec.setStatus(DCWormsTags.INEXEC);
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+
+		int phaseDuration = Double.valueOf(execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.START_TASK_EXECUTION), 
+				execTask, resources, exec.getCompletionPercentage())).intValue();
+
+		saveExecutionHistory(exec, exec.getCompletionPercentage(), phaseDuration);
+		if(exec.getResourceConsumptionProfile().isLast()){
+			scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_FINISHED, execTask);;
+		} else {
+			scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_CHANGED, execTask);
+		}
+		
+		PEUnit peUnit = (PEUnit)resources.get(StandardResourceUnitName.PE);
+		updateComputingResources(peUnit, EnergyEventType.TASK_STARTED, exec);
+		
+	}
 	protected void finalizeExecutable(ExecTask execTask){
 		
@@ -304,12 +408,15 @@
 			}
 		}
-		
-		UsedResourcesList lastUsedList = exec.getUsedResources();
-		Map<ResourceUnitName, ResourceUnit> lastUsed = lastUsedList.getLast()
-				.getResourceUnits();
+
+		Map<ResourceUnitName, ResourceUnit> lastUsed = exec.getAllocatedResources().getLast().getResourceUnits();
 		getAllocationManager().freeResources(lastUsed);
+		
+		saveExecutionHistory(exec, 100,0);
 		
 		PEUnit peUnit = (PEUnit)lastUsed.get(StandardResourceUnitName.PE);
 		updateComputingResources(peUnit, EnergyEventType.TASK_FINISHED, exec);
+		
+		log.debug(execTask.getJobId() + "_" + execTask.getId() + " finished execution on " + new DateTime());
+		log.info(DCWormsConstants.USAGE_MEASURE_NAME + ": " + calculateTotalLoad());
 	}
 	
@@ -326,17 +433,10 @@
 			ExecTask task = iter.next();
 			Executable exec = (Executable)task;
-
-			if(exec.getUsedResources().size() > 1){
-				//System.out.println("--- upadteProgressX: " + Sim_system.sim_clock() );
-				//System.out.println("taskId: " + exec.getId() + "; completion percentage: " + exec.getCompletionPercentage() + "; timespan: " + timeSpan + "; estimatedDuration: " +  exec.getEstimatedDuration());
-				exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * (timeSpan / (exec.getEstimatedDuration()) * (1.0 - exec.getUsedResources().get(exec.getUsedResources().size()-1).getCompletionPercentage()/100.0)));
-				//System.out.println("newProgress: " + exec.getCompletionPercentage()  );	
-			} 
-			else {
-				//System.out.println("--- upadteProgress2: " + Sim_system.sim_clock() );
-				//System.out.println("taskId: " + exec.getId() + "; completion percentage: " + exec.getCompletionPercentage() + "; timespan: " + timeSpan + "; estimatedDuration: " +  exec.getEstimatedDuration());
-				exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * (timeSpan / exec.getEstimatedDuration()));
-				//System.out.println("newProgress: " + exec.getCompletionPercentage()  );	
-			}
+			ExecutionHistoryItem execHistItem = exec.getExecHistory().getLast();
+			//System.out.println("--- upadteProgressX: " + Sim_system.sim_clock() );
+			//System.out.println("taskId: " + exec.getId() + "; completion percentage: " + exec.getCompletionPercentage() + "; timespan: " + timeSpan + "; estimatedDuration: " +  execHistItem.getCompletionPercentage());
+			exec.setCompletionPercentage(exec.getCompletionPercentage() + 100 * (timeSpan / (execHistItem.getEstimatedDuration()) * (1.0 - execHistItem.getCompletionPercentage()/100.0)));
+			//System.out.println("newProgress: " + exec.getCompletionPercentage()  );	
+
 		}
 	}
@@ -368,28 +468,27 @@
 	}
 
-	protected void updateProcessingTimes(Sim_event ev) {
+	protected void updateProcessingTimes() {
 		for (ExecTask execTask : jobRegistry.getRunningTasks()) {
 			Executable exec = (Executable)execTask;
 
-			Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getUsedResources().getLast().getResourceUnits();
-			double lastTimeStamp = exec.getUsedResources().getLast().getTimeStamp().getMillis()/1000;
+
+			Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getAllocatedResources().getLast().getResourceUnits();
+
 			int phaseDuration = Double.valueOf(execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(SchedulingEventType.RESOURCE_STATE_CHANGED), 
 					execTask, choosenResources, exec.getCompletionPercentage())).intValue();
 
-			if(DoubleMath.subtract((lastTimeStamp + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + phaseDuration)) == 0.0){
+			ExecutionHistoryItem execHistItem = exec.getExecHistory().getLast();
+			double lastTimeStamp = execHistItem.getTimeStamp().getMillis()/1000;
+			if(DoubleMath.subtract((lastTimeStamp + execHistItem.getEstimatedDuration()), (new DateTime().getMillis()/1000 + phaseDuration)) == 0.0){
 				continue;
 			}
 			//System.out.println("=== upadteTIme: " + Sim_system.sim_clock() );
-			//System.out.println("execId: " + exec.getId() +  "; estimatedDuration " + exec.getEstimatedDuration());
-			//System.out.println("execId: " + exec.getId() + "; difference " + DoubleMath.subtract((lastTimeStamp + exec.getEstimatedDuration()), (new DateTime().getMillis()/1000 + phaseDuration)));
+			//System.out.println("execId: " + exec.getId() +  "; estimatedDuration " + execHistItem.getEstimatedDuration());
+			//System.out.println("execId: " + exec.getId() + "; difference " + DoubleMath.subtract((lastTimeStamp + execHistItem.getEstimatedDuration()), (new DateTime().getMillis()/1000 + phaseDuration)));
 			//System.out.println("completionPercantage: " + exec.getCompletionPercentage() + "; basic duration: " +exec.getResourceConsumptionProfile().getCurrentResourceConsumption().getDuration() +   "; phaseDuration: " +  phaseDuration);
 
-			exec.setEstimatedDuration(phaseDuration);
-			DateTime currentTime = new DateTime();
-			ResourceHistoryItem resHistItem = new ResourceHistoryItem(choosenResources, currentTime);
-			resHistItem.setCompletionPercentage(exec.getCompletionPercentage());
-			exec.addUsedResources(resHistItem);
-			
-			if(exec.getResourceConsumptionProfile().getCurrentResourceConsumption() == exec.getResourceConsumptionProfile().getResourceConsumptionList().getLast()){
+			saveExecutionHistory(exec, exec.getCompletionPercentage(), phaseDuration);
+			
+			if(exec.getResourceConsumptionProfile().isLast()){
 				ExecTaskFilter filter = new ExecTaskFilter(exec.getUniqueId(), DCWormsTags.TASK_EXECUTION_FINISHED);
 				scheduler.sim_cancel(filter, null);
@@ -407,5 +506,5 @@
 	}	
 
-	protected void updateTaskExecution(ExecTask execTask, SchedulingEventType schedEvType) {
+	protected void updateTaskExecutionPhase(ExecTask execTask, SchedulingEventType schedEvType) {
 
 		if (execTask.getStatus() == DCWormsTags.INEXEC) {
@@ -415,22 +514,22 @@
 				exec.setStatus(DCWormsTags.NEW_EXEC_PHASE);
 			} catch (Exception e) {
-			}
-			
-			Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getUsedResources().getLast().getResourceUnits();
+				
+			}
+
+			Map<ResourceUnitName, ResourceUnit> choosenResources = exec.getAllocatedResources().getLast().getResourceUnits();
 
 			int phaseDuration = Double.valueOf(execTimeEstimationPlugin.execTimeEstimation(new SchedulingEvent(schedEvType), 
 					execTask, choosenResources, exec.getCompletionPercentage())).intValue();
-
-			exec.setEstimatedDuration(phaseDuration);
-			
-			if(exec.getResourceConsumptionProfile().getCurrentResourceConsumption() == exec.getResourceConsumptionProfile().getResourceConsumptionList().getLast()){
+			
+			saveExecutionHistory(exec, exec.getCompletionPercentage(), phaseDuration);
+			
+			if(exec.getResourceConsumptionProfile().isLast()){
 				scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_FINISHED, execTask);
-				PEUnit peUnit = (PEUnit)exec.getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
-				updateComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec);
 			} else {
 				scheduler.sendInternal(phaseDuration, DCWormsTags.TASK_EXECUTION_CHANGED, execTask);
-				PEUnit peUnit = (PEUnit)exec.getUsedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
-				updateComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec);
-			}
+			}
+			
+			PEUnit peUnit = (PEUnit)exec.getAllocatedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
+			updateComputingResources(peUnit, EnergyEventType.RESOURCE_UTILIZATION_CHANGED, exec);
 		}
 	}	
@@ -501,5 +600,5 @@
 	}
 	
-	public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) {
+	public void notifySubmittedWorkloadUnit(WorkloadUnit wu) {
 		//250314
 		//updateProcessingProgress();
@@ -562,14 +661,6 @@
 			Executable exec = (Executable) task;
 			jobRegistry.addExecTask(exec);
-			
-			exec.trackResource(scheduler.get_name());
-			Scheduler parentScheduler = scheduler.getParent();
-			List<String> visitedResource = exec.getVisitedResources();
-			String [] visitedResourcesArray = visitedResource.toArray(new String[visitedResource.size()]);
-			while (parentScheduler != null && !ArrayUtils.contains(visitedResourcesArray, parentScheduler.get_name())) {
-				exec.trackResource(parentScheduler.get_name());
-				parentScheduler = parentScheduler.getParent();
-			}
-			exec.setSchedulerName(scheduler.get_id());
+
+			exec.setSchedulerName(scheduler.getFullName());
 			
 			TaskList newTasks = new TaskListImpl();
@@ -596,3 +687,46 @@
 	}
 	
+	
+	public void saveExecutionHistory(Executable exec, double completionPercentage, double estimatedDuration){
+
+		ExecutionHistoryItem execHistoryItem = new ExecutionHistoryItem(new DateTime());
+		execHistoryItem.setCompletionPercentage(completionPercentage);
+		execHistoryItem.setEstimatedDuration(estimatedDuration);
+		execHistoryItem.setResIndex(exec.getAllocatedResources().size() -1);
+		execHistoryItem.setStatus(exec.getStatus());
+		exec.addExecHistory(execHistoryItem);
+		
+		ProcessingElements pes = (ProcessingElements) exec.getAllocatedResources().getLast().getResourceUnits().get(StandardResourceUnitName.PE);
+		for (ComputingResource resource : pes) {
+
+			LinkedList<ComputingResource> toExamine = new LinkedList<ComputingResource>();
+			toExamine.push(resource);
+
+			while (!toExamine.isEmpty()) {
+				ComputingResource compResource = toExamine.pop();
+				List<ComputingResource> resources = compResource.getChildren();
+				if(resources.isEmpty()){
+					Set<String> visitedResources = exec.getAllocatedResources().getLast().getVisitedResources();
+					if(!visitedResources.contains(compResource.getFullName())){
+						exec.getAllocatedResources().getLast().trackResource(compResource.getFullName());
+					}
+				} else {
+					for (int i = 0; i < resources.size(); i++) {
+						ComputingResource resourceChild = resources.get(i);
+						toExamine.addLast(resourceChild);
+					}
+				}
+			}
+		}
+	}
+	
+	public boolean allocateResources(Executable exec, Map<ResourceUnitName, ResourceUnit> choosenResources){
+		boolean allocationStatus = getAllocationManager().allocateResources(choosenResources);
+		if(allocationStatus){
+			ResourceHistoryItem resourceHistoryItem = new ResourceHistoryItem(choosenResources);
+			exec.addAllocatedResources(resourceHistoryItem);
+			return true;
+		}
+		return false;
+	}
 }
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/global/GlobalManagementSystem.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/global/GlobalManagementSystem.java	(revision 1357)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/policy/global/GlobalManagementSystem.java	(revision 1362)
@@ -71,5 +71,5 @@
 	}
 	
-	public void notifySubmittedWorkloadUnit(WorkloadUnit wu, boolean ack) {
+	public void notifySubmittedWorkloadUnit(WorkloadUnit wu) {
 		if (!pluginSupportsEvent(GridSimTags.GRIDLET_SUBMIT)) {
 			log.error("Plugin " + schedulingPlugin.getClass()
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/ResourceHistoryItem.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/ResourceHistoryItem.java	(revision 1317)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/ResourceHistoryItem.java	(revision 1362)
@@ -1,26 +1,19 @@
 package schedframe.scheduling;
 
+import java.util.HashSet;
 import java.util.Map;
-
-import org.joda.time.DateTime;
+import java.util.Set;
 
 import schedframe.resources.units.ResourceUnit;
 import schedframe.resources.units.ResourceUnitName;
 
-//TODO - change name
 public class ResourceHistoryItem {
 
 	protected Map<ResourceUnitName, ResourceUnit> usedResources;
-	protected DateTime timeStamp;
-	protected double completionPercentage;
-	
-	/**
-	 * 
-	 * @param map hash map of resource units, which should be remembered
-	 * @param time the moment in time when this resource configuration was created  
-	 */
-	public ResourceHistoryItem(Map<ResourceUnitName, ResourceUnit> map, DateTime time){
-		this.usedResources = map;
-		this.timeStamp = time;
+	protected Set<String> visitedResources;
+
+	public ResourceHistoryItem(Map<ResourceUnitName, ResourceUnit> usedResources){
+		this.usedResources = usedResources;
+		this.visitedResources = new HashSet<String>();
 	}
 	
@@ -29,14 +22,10 @@
 	}
 	
-	public DateTime getTimeStamp(){
-		return timeStamp;
+	public void trackResource(String resName){
+		visitedResources.add(resName);
 	}
-
-	public double getCompletionPercentage() {
-		return completionPercentage;
-	}
-
-	public void setCompletionPercentage(double completionPercentage) {
-		this.completionPercentage = completionPercentage;
+	
+	public Set<String> getVisitedResources(){
+		return visitedResources;
 	}
 	
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/JobRegistry.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/JobRegistry.java	(revision 1151)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/JobRegistry.java	(revision 1362)
@@ -3,8 +3,11 @@
 
 import java.util.List;
+import java.util.Map;
 
 import dcworms.schedframe.scheduling.ExecTask;
 
 import schedframe.ExecutablesList;
+import schedframe.resources.units.ResourceUnit;
+import schedframe.resources.units.ResourceUnitName;
 import schedframe.scheduling.tasks.JobInterface;
 import schedframe.scheduling.tasks.TaskInterface;
@@ -27,11 +30,11 @@
 
 	
-	public List<ExecTask> getTasks(int status) ;
+	public List<ExecTask> getTasks(int status);
 
-	public List<ExecTask> getQueuedTasks() ;
+	public List<ExecTask> getQueuedTasks();
 	
-	public List<ExecTask> getRunningTasks() ;
+	public List<ExecTask> getRunningTasks();
 	
-	public List<ExecTask> getReadyTasks() ;
+	public List<ExecTask> getReadyTasks();
 	
 	public List<ExecTask> getFinishedTasks();
@@ -39,4 +42,14 @@
 	
 	public List<? extends TaskInterface<?>> getAvailableTasks(List<JobInterface<?>> jobsList);
+	
+	
+	public boolean pauseTask(String jobId, String taskId);
+	
+	public boolean resumeTask(String jobId, String taskId);
+	
+	public boolean migrateTask(String jobId, String taskId, Map<ResourceUnitName, ResourceUnit> choosenResources);
+	
+	public boolean migrateTask(String jobId, String taskId, String nodeName);
+	
 
 }
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/JobRegistryImpl.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/JobRegistryImpl.java	(revision 1151)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/JobRegistryImpl.java	(revision 1362)
@@ -5,6 +5,6 @@
 import java.util.ArrayList;
 import java.util.List;
-
-import org.apache.commons.lang.ArrayUtils;
+import java.util.Set;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -21,6 +21,4 @@
 
 public class JobRegistryImpl extends AbstractJobRegistry {
-	
-	private static final long serialVersionUID = 8030555906990767342L;
 
 	private static Log log = LogFactory.getLog(JobRegistryImpl.class);
@@ -55,6 +53,12 @@
 			for (ExecTask task: executables) {
 				if (task.getStatus() == status) {
-					List<String> visitedResource = task.getVisitedResources();
-					if(ArrayUtils.contains(visitedResource.toArray(new String[visitedResource.size()]), context)) {
+					Set<String> visitedResource = task.getAllocatedResources().getLast().getVisitedResources();
+					for(String res: visitedResource){
+						if(res.contains(context)){
+							taskList.add(task);
+							break;
+						}
+					}
+					if(task.getSchedulerName().contains(context)) {
 						taskList.add(task);
 					}
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/AbstractJobRegistry.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/AbstractJobRegistry.java	(revision 1151)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/manager/tasks/AbstractJobRegistry.java	(revision 1362)
@@ -2,18 +2,23 @@
 
 
+import gridsim.dcworms.DCWormsTags;
+
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+import schedframe.resources.units.ResourceUnit;
+import schedframe.resources.units.ResourceUnitName;
 import schedframe.scheduling.tasks.Job;
 import schedframe.scheduling.tasks.JobInterface;
 import schedframe.scheduling.tasks.Task;
 import schedframe.scheduling.tasks.TaskInterface;
+import simulator.DataCenterWorkloadSimulator;
+import dcworms.schedframe.scheduling.ExecTask;
 
 
 public abstract class AbstractJobRegistry /*extends ConcurrentHashMap<String, Job>*/ implements JobRegistry, Cloneable{
 	
-	private static final long serialVersionUID = 8409060063583755824L;
-
 	
 	protected static final ConcurrentHashMap<String, JobInterface<?>> jobs = new ConcurrentHashMap<String, JobInterface<?>>();
@@ -79,3 +84,49 @@
 	}
 
+	@Override
+	public boolean pauseTask(String jobId, String taskId) {
+		ExecTask execTask = getTask(jobId, taskId);
+		if (execTask == null) {
+			return false;
+		}else{
+			String[] ids = new String[2];
+			ids[0] = new String(jobId);
+			ids[1] = new String(taskId);
+			DataCenterWorkloadSimulator.getEventManager().sendToAllSchedulers(0, DCWormsTags.TASK_PAUSE, ids);
+			return true;
+		}
+	}
+
+	public boolean resumeTask(String jobId, String taskId) {
+		ExecTask execTask = getTask(jobId, taskId);
+		if (execTask == null) {
+			return false;
+		}else{
+			String[] ids = new String[2];
+			ids[0] = new String(jobId);
+			ids[1] = new String(taskId);
+			DataCenterWorkloadSimulator.getEventManager().sendToAllSchedulers(0, DCWormsTags.TASK_RESUME, ids);
+			return true;	
+		}
+	}
+
+	public boolean migrateTask(String jobId, String taskId, Map<ResourceUnitName, ResourceUnit> choosenResources) {
+		ExecTask execTask = getTask(jobId, taskId);
+		if (execTask == null) {
+			return false;
+		}else{
+			Object[] data = new Object[3];
+			data[0] = new String(jobId);
+			data[1] = new String(taskId);
+			data[2] = choosenResources;
+			DataCenterWorkloadSimulator.getEventManager().sendToAllSchedulers(0, DCWormsTags.TASK_MOVE, data);
+			return true;	
+		}
+	}
+
+	public boolean migrateTask(String jobId, String taskId, String nodeName) {
+		// TODO Auto-generated method stub
+		return false;
+	}
+	
 }
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/ExecutionHistoryItem.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/ExecutionHistoryItem.java	(revision 1362)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/ExecutionHistoryItem.java	(revision 1362)
@@ -0,0 +1,54 @@
+package schedframe.scheduling;
+
+import org.joda.time.DateTime;
+
+public class ExecutionHistoryItem {
+
+	protected DateTime timeStamp;
+	protected double completionPercentage;
+	protected double estimatedDuration;
+	protected int status;
+	protected int resUsageIndex;
+
+	public ExecutionHistoryItem(DateTime time){
+		this.timeStamp = time;
+	}
+
+	public DateTime getTimeStamp(){
+		return timeStamp;
+	}
+
+	public double getCompletionPercentage() {
+		return completionPercentage;
+	}
+
+	public void setCompletionPercentage(double completionPercentage) {
+		this.completionPercentage = completionPercentage;
+	}
+
+	public double getEstimatedDuration() {
+		return estimatedDuration;
+	}
+
+	public void setEstimatedDuration(double estimatedDuration) {
+		this.estimatedDuration = estimatedDuration;
+	}
+
+	public int getResIndex() {
+		return resUsageIndex;
+	}
+
+	public void setResIndex(int resUsageIndex) {
+		this.resUsageIndex = resUsageIndex;
+	}
+
+	public int getStatus() {
+		return status;
+	}
+
+	public void setStatus(int status) {
+		this.status = status;
+	}
+
+	
+}
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/tasks/Task.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/tasks/Task.java	(revision 1207)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/tasks/Task.java	(revision 1362)
@@ -1,3 +1,5 @@
 package schedframe.scheduling.tasks;
+
+import gridsim.dcworms.DCWormsTags;
 
 import java.io.StringReader;
@@ -439,5 +441,5 @@
 	public boolean isFinished(){
 		if(processes == null)
-			return (status > 3 && status <= 6);
+			return (status >= DCWormsTags.SUCCESS && status <= DCWormsTags.FAILED);
 		
 		for(int i = 0; i < processes.size(); i++){
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/tasks/phases/ExecutionProfile.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/tasks/phases/ExecutionProfile.java	(revision 1362)
+++ DCWoRMS/branches/coolemall/src/schedframe/scheduling/tasks/phases/ExecutionProfile.java	(revision 1362)
@@ -0,0 +1,46 @@
+package schedframe.scheduling.tasks.phases;
+
+import java.util.LinkedList;
+
+public class ExecutionProfile {
+	
+	protected LinkedList<ResourceConsumption> resourceConsumptionList;
+	protected long usefulWork;
+	private int currentPhase;
+
+	public ExecutionProfile(LinkedList<ResourceConsumption> resourceConsumptionList) {
+		this.resourceConsumptionList = resourceConsumptionList;
+		this.currentPhase = 0;
+	}
+	
+	public LinkedList<ResourceConsumption> getResourceConsumptionList(){
+		return resourceConsumptionList;
+	}
+	
+	public ResourceConsumption getCurrentResourceConsumption(){
+		return resourceConsumptionList.get(currentPhase);
+	}
+
+	public int getCurrentPhase() {
+		return currentPhase;
+	}
+
+	public void setCurrentPhase(int currentPhase) {
+		this.currentPhase = currentPhase;
+	}
+
+	public long getUsefulWork() {
+		return usefulWork;
+	}
+
+	public void setUsefulWork(long usefulWork) {
+		this.usefulWork = usefulWork;
+	}
+	
+	public boolean isLast(){
+		if(currentPhase == resourceConsumptionList.size() - 1){
+			return true;
+		}
+		return false;
+	}
+}
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/tasks/phases/ResourceConsumptionProfile.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/tasks/phases/ResourceConsumptionProfile.java	(revision 1207)
+++ 	(revision )
@@ -1,47 +1,0 @@
-package schedframe.scheduling.tasks.phases;
-
-import java.util.LinkedList;
-
-public class ResourceConsumptionProfile {
-	
-	protected LinkedList<ResourceConsumption> resourceConsumptionList;
-	protected long usefulWork;
-	private int currentPhase;
-	
-	public ResourceConsumptionProfile() {
-		this.resourceConsumptionList = new LinkedList<ResourceConsumption>();
-		this.currentPhase = -1;
-	}
-	
-	public ResourceConsumptionProfile(LinkedList<ResourceConsumption> resourceConsumptionList) {
-		this.resourceConsumptionList = resourceConsumptionList;
-		this.currentPhase = -1;
-	}
-	
-	public LinkedList<ResourceConsumption> getResourceConsumptionList(){
-		return resourceConsumptionList;
-	}
-	
-	public ResourceConsumption getCurrentResourceConsumption(){
-		if(currentPhase < resourceConsumptionList.size())
-			return resourceConsumptionList.get(currentPhase);
-		else 
-			return null;
-	}
-
-	public int getCurrentPhase() {
-		return currentPhase;
-	}
-
-	public void setCurrentPhase(int currentPhase) {
-		this.currentPhase = currentPhase;
-	}
-
-	public long getUsefulWork() {
-		return usefulWork;
-	}
-
-	public void setUsefulWork(long usefulWork) {
-		this.usefulWork = usefulWork;
-	}
-}
Index: DCWoRMS/branches/coolemall/src/schedframe/scheduling/UsedResourcesList.java
===================================================================
--- DCWoRMS/branches/coolemall/src/schedframe/scheduling/UsedResourcesList.java	(revision 490)
+++ 	(revision )
@@ -1,12 +1,0 @@
-package schedframe.scheduling;
-
-import java.util.LinkedList;
-
-public class UsedResourcesList extends LinkedList<ResourceHistoryItem>{
-
-	/**
-	 * 
-	 */
-	private static final long serialVersionUID = 1L;
-	
-}
