.

include/wfr/common/wfs/Workflow.h

@@ -66,6 +66,9 @@ private:
 	wfr::wfr_timestamp_sc_t startTime;
 	wfr::wfr_timestamp_sc_t endTime;
 
+	unsigned lastWepNodeCount = 0;
+	wfr::wfr_timestamp_sc_t wepCompletedTime;
+
 	#ifdef WFR_ENABLE_STATISTICS
 		std::mutex statMutex;
 
@@ -190,10 +193,13 @@ public:
 	const wfr::wfr_timestamp_sc_t& getStartTime() const noexcept { return startTime; }
 	const wfr::wfr_timestamp_sc_t& getEndTime() const noexcept { return endTime; }
 	wfr::wfr_duration_sc_t getWorkflowDuration() const noexcept { return endTime - startTime; };
+	wfr::wfr_duration_sc_t getTimeTillWepCompletion() const noexcept { return wepCompletedTime - startTime; };
 	wfr::wfr_duration_sc_t getMakespan() const noexcept { return endTime - createTime; };
 
 	#ifdef WFR_ENABLE_STATISTICS
 
+		void updateWepStatistics() noexcept;
+
 		wfr::wfr_duration_sc_t getEnactmentTime() const noexcept { return enactmentTime; }
 		void addToEnactmentTime(wfr::wfr_duration_sc_t time) {
 			std::lock_guard<std::mutex> lock(statMutex);

include/wfr/scheduling/scheduler/cpop/CPOPScheduler.h

@@ -0,0 +1,85 @@
+
+#pragma once
+
+
+#include <wfr/scheduling/AbstractScheduler.h>
+#include <wfr/prediction/PredictionManager.h>
+#include <wfr/common/resource/Resource.h>
+#include <wfr/common/resource/ResourceManager.h>
+#include <wfr/common/types/WfrTypes.h>
+#include <mutex>
+#include <thread>
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <iostream>
+#include <condition_variable>
+
+namespace wfr {
+
+namespace wfs {
+class AtomicActivityInstance;
+}
+namespace wep {
+class AtomicWepNode;
+class ActivityWepNode;
+class AbstractWepNode;
+}
+namespace resource {
+class ResourceManager;
+}
+namespace prediction {
+class PredictionManager;
+}
+}
+
+
+namespace wfr {
+namespace scheduling {
+
+
+class CPOPScheduler: public scheduling::SchedulerBase {
+protected:
+	resource::ResourceManager& resourceManager;
+	prediction::PredictionManager& predictionManager;
+
+	struct TaskRankEntry {
+		double rankU = 0;
+		double rankD = 0;
+		double priority = 0;
+		bool finalizedU = false;
+		bool finalizedD = false;
+
+		wfs::AtomicActivityImplementation** fastestImplementations;
+		prediction::ExecutionTimeResult* fastestExecutionTimes;
+		double averageExecutionTime = 0.0;
+
+		TaskRankEntry(wfr_size_t resourceCount) : fastestImplementations(new wfs::AtomicActivityImplementation*[resourceCount]),
+				fastestExecutionTimes(new prediction::ExecutionTimeResult[resourceCount]) {}
+	};
+
+	std::set<wep::AtomicWepNode*> readySet;
+
+	void calcTaskRankD(wep::ActivityWepNode& node, std::vector<resource::ComputingResource*>& resources);
+
+	void calcTaskRankU(wep::ActivityWepNode& node, std::vector<resource::ComputingResource*>& resources);
+
+	void scheduleReadySet(WepMapEntry& wepe, std::set<wep::ActivityWepNode*>& cpTasks, resource::ComputingResource* cpResource,
+			std::vector<resource::ComputingResource*>& resources, std::unique_lock<std::mutex>& schedulerLock);
+
+public:
+	CPOPScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager)
+			: resourceManager(resourceManager), predictionManager(predictionManager) {}
+
+	~CPOPScheduler() {}
+
+	virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) override;
+	virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) override;
+	virtual void _unregisterWEP(WepMapEntry& wepe, std::unique_lock<std::mutex>& schedulerLock) override;
+};
+
+} /* namespace scheduling */
+} /* namespace wfr */
+
+

include/wfr/scheduling/scheduler/hcpt/HCPTScheduler.h

@@ -0,0 +1,81 @@
+
+#pragma once
+
+
+#include <wfr/scheduling/AbstractScheduler.h>
+#include <wfr/prediction/PredictionManager.h>
+#include <wfr/common/resource/Resource.h>
+#include <wfr/common/resource/ResourceManager.h>
+#include <wfr/common/types/WfrTypes.h>
+#include <mutex>
+#include <thread>
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <iostream>
+#include <condition_variable>
+
+namespace wfr {
+
+namespace wfs {
+class AtomicActivityInstance;
+}
+namespace wep {
+class AtomicWepNode;
+class ActivityWepNode;
+class AbstractWepNode;
+}
+namespace resource {
+class ResourceManager;
+}
+namespace prediction {
+class PredictionManager;
+}
+}
+
+
+namespace wfr {
+namespace scheduling {
+
+
+class HCPTScheduler: public scheduling::SchedulerBase {
+protected:
+	resource::ResourceManager& resourceManager;
+	prediction::PredictionManager& predictionManager;
+
+	struct TaskRankEntry {
+		double aest = 0;
+		double alst = 0;
+		bool aestFinalized = false;
+		bool alstFinalized = false;
+
+		wfs::AtomicActivityImplementation** fastestImplementations;
+		prediction::ExecutionTimeResult* fastestExecutionTimes;
+		double averageExecutionTime = 0.0;
+
+		TaskRankEntry(wfr_size_t resourceCount) : fastestImplementations(new wfs::AtomicActivityImplementation*[resourceCount]),
+				fastestExecutionTimes(new prediction::ExecutionTimeResult[resourceCount]) {}
+	};
+
+	std::set<wep::AtomicWepNode*> readySet;
+
+	void calcTaskRanks(wep::ActivityWepNode& node, std::vector<resource::ComputingResource*>& resources);
+
+	void scheduleReadySet(WepMapEntry& wepe, std::vector<resource::ComputingResource*>& resources, std::unique_lock<std::mutex>& schedulerLock);
+
+public:
+	HCPTScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager)
+			: resourceManager(resourceManager), predictionManager(predictionManager) {}
+
+	~HCPTScheduler() {}
+
+	virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) override;
+	virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) override;
+	virtual void _unregisterWEP(WepMapEntry& wepe, std::unique_lock<std::mutex>& schedulerLock) override;
+};
+
+} /* namespace scheduling */
+} /* namespace wfr */
+
+

include/wfr/scheduling/scheduler/justintime2/JustInTime2Scheduler.h

@@ -0,0 +1,34 @@
+
+#pragma once
+
+
+#include <wfr/scheduling/AbstractScheduler.h>
+
+
+namespace wfr {
+namespace wep {
+class AtomicWepNode;
+class WEPnode;
+}
+}
+
+
+namespace wfr {
+namespace scheduling {
+
+class JustInTime2Scheduler: public wfr::scheduling::SchedulerBase {
+private:
+	resource::ResourceManager& resourceManager;
+	prediction::PredictionManager& predictionManager;
+	std::set<wep::AtomicWepNode*> nodeQueue;
+
+public:
+	JustInTime2Scheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager)
+			: resourceManager(resourceManager), predictionManager(predictionManager) {}
+
+	virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock);
+};
+
+} /* namespace scheduling */
+} /* namespace wfr */
+

include/wfr/wep/utils/WepSearch.h

@@ -17,15 +17,23 @@ namespace wep {
 
 template<typename VisitType>
 void wepDepthFirstSearch(wfr::wep::AbstractWepNode* startNode, const VisitType& visitFunction,
-		std::set<wfr::wep::AbstractWepNode*>& visitedNodes = std::set<wfr::wep::AbstractWepNode*>()) {
+		std::set<wfr::wep::AbstractWepNode*>* visitedNodes = nullptr) {
-	visitedNodes.insert(startNode);
+	bool deleteVisitedNodeSet = false;
+	if (visitedNodes == nullptr) {
+		visitedNodes = new std::set<wfr::wep::AbstractWepNode*>();
+		deleteVisitedNodeSet = true;
+	}
+	visitedNodes->insert(startNode);
 	if (visitFunction(startNode)) {
 		for (auto c : startNode->getSuccessors()) {
-			if (visitedNodes.find(c) == visitedNodes.end()) {
+			if (visitedNodes->find(c) == visitedNodes->end()) {
 				wepDepthFirstSearch(c, visitFunction, visitedNodes);
 			}
 		}
 	}
+	if (deleteVisitedNodeSet) {
+		delete visitedNodes;
+	}
 }
 
 

src/common/wfs/Workflow.cpp

@@ -115,6 +115,14 @@ WorkflowStatus Workflow::executeSync( lwd::_wfr_lwd* input, lwd::_wfr_lwd* outpu
 
 #ifdef WFR_ENABLE_STATISTICS
 
+	void Workflow::updateWepStatistics() noexcept {
+		unsigned wepNodeCount = createdCompositeInstanceCount+createdAtomicInstanceCount;
+		if (lastWepNodeCount != wepNodeCount) {
+			lastWepNodeCount = wepNodeCount;
+			wepCompletedTime = std::chrono::system_clock::now();
+		}
+	}
+
 	void Workflow::printStatistics(std::ostream& sout) {
 		sout << "WFR-Statistics: Created Instances: " << (createdCompositeInstanceCount+createdAtomicInstanceCount) << std::endl;
 		sout << "WFR-Statistics: Finished Instances: " << (finishedCompositeInstanceCount+finishedAtomicInstanceCount) << std::endl;
@@ -141,6 +149,8 @@ WorkflowStatus Workflow::executeSync( lwd::_wfr_lwd* input, lwd::_wfr_lwd* outpu
 
 		sout << "WFR-Statistics: Enactment Time: "
 			 << ((double)std::chrono::duration_cast<std::chrono::milliseconds>(enactmentTime).count())/1000.0 << " sec" << std::endl;
+		sout << "WFR-Statistics: Time till WEP Completion: "
+			 << ((double)std::chrono::duration_cast<std::chrono::milliseconds>(getTimeTillWepCompletion()).count())/1000.0 << " sec" << std::endl;
 		sout << "WFR-Statistics: Workflow Duration: "
 			 << ((double)std::chrono::duration_cast<std::chrono::milliseconds>(getWorkflowDuration()).count())/1000.0 << " sec" << std::endl;
 		sout << "WFR-Statistics: Makespan: "

src/enactment/EnactmentEngine.cpp

@@ -30,7 +30,7 @@ void EnactmentEngine::init(int argc, char** argv) {
     boost::program_options::options_description opts("EnactmentEngine Options");
     opts.add_options()
 			("help", "Print help message.")
-    		("disable-wepdotwriter", "Disable WEP to .dot writer.")
+    		("enable-wepdotwriter", "Disable WEP to .dot writer.")
     		("enable-wepprogressionwriter", "Write WEP progression to disk.")
     		("early-evaluation", "Enables early evaluation.")
     		("late-evaluation", "Enables late evaluation [Default].");
@@ -41,7 +41,7 @@ void EnactmentEngine::init(int argc, char** argv) {
     if (vm.count("help")) {
         boost::program_options::options_description show_opts("EnactmentEngine Options");
         show_opts.add_options()
-        		("disable-wepdotwriter", "Disable WEP to .dot writer.")
+        		("enable-wepdotwriter", "Enable WEP to .dot writer.")
 	    		("enable-wepprogressionwriter", "Write WEP progression to disk.")
 	    		("early-evaluation", "Enables early evaluation [Default].")
 	    		("late-evaluation", "Enables late evaluation.");
@@ -49,12 +49,12 @@ void EnactmentEngine::init(int argc, char** argv) {
     	return;
     }
 
-    if (vm.count("disable-wepdotwriter")) {
+    if (vm.count("enable-wepdotwriter")) {
-    	WorkflowRuntime::setGlobalConfig("dotWriterEnabled", false);
-    	std::cerr << "WEP-to-.dot writer is disabled." << std::endl;
-    } else {
     	WorkflowRuntime::setGlobalConfig("dotWriterEnabled", true);
     	std::cerr << "WEP-to-.dot writer is enabled." << std::endl;
+    } else {
+    	WorkflowRuntime::setGlobalConfig("dotWriterEnabled", false);
+    	std::cerr << "WEP-to-.dot writer is disabled." << std::endl;
     }
 
     if (vm.count("enable-wepprogressionwriter")) {
@@ -157,9 +157,12 @@ void EnactmentEngine::workflowExecutionLoop(lwd::_wfr_lwd* input, lwd::_wfr_lwd*
 				}
 			#endif
 
+#ifdef WFR_ENABLE_STATISTICS
+			workflow.updateWepStatistics();
 			if (enableWepProgressionWriter) {
 				progressionWriter->writeProgressToFile();
 			}
+#endif
 
 			if (wep->workflow.getWorkflowStatus() == wfr::wfs::WorkflowStatus::Error
 					|| wep->workflow.getWorkflowStatus() == wfr::wfs::WorkflowStatus::Finished) {

src/scheduling/SchedulerManager.cpp

@@ -12,10 +12,12 @@
 #include <wfr/scheduling/scheduler/heft/HEFTScheduler.h>
 #include <wfr/scheduling/scheduler/random/RandomScheduler.h>
 #include <wfr/scheduling/scheduler/justintime/JustInTimeScheduler.h>
+#include <wfr/scheduling/scheduler/justintime2/JustInTime2Scheduler.h>
 #include <wfr/scheduling/scheduler/minmin/MinMinScheduler.h>
 #include <wfr/scheduling/scheduler/maxmin/MaxMinScheduler.h>
 #include <wfr/scheduling/scheduler/sufferage/SufferageScheduler.h>
 #include <wfr/scheduling/scheduler/peft/PEFTScheduler.h>
+#include <wfr/scheduling/scheduler/cpop/CPOPScheduler.h>
 #include <wfr/scheduling/scheduler/lookahead/LookaheadScheduler.h>
 
 namespace wfr {
@@ -79,6 +81,9 @@ SchedulerManager::SchedulerManager(std::string defaultScheduler) : defaultSchedu
     schedulerFactories.insert(factoryMap_t::value_type("heft", []()->AbstractScheduler* {
     	return new HEFTScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
     }));
+    schedulerFactories.insert(factoryMap_t::value_type("cpop", []()->AbstractScheduler* {
+    	return new CPOPScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
+    }));
     schedulerFactories.insert(factoryMap_t::value_type("lookahead", []()->AbstractScheduler* {
     	return new LookaheadScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
     }));
@@ -88,6 +93,9 @@ SchedulerManager::SchedulerManager(std::string defaultScheduler) : defaultSchedu
     schedulerFactories.insert(factoryMap_t::value_type("jit", []()->AbstractScheduler* {
     	return new JustInTimeScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
     }));
+    schedulerFactories.insert(factoryMap_t::value_type("jit2", []()->AbstractScheduler* {
+    	return new JustInTime2Scheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
+    }));
     schedulerFactories.insert(factoryMap_t::value_type("minmin", []()->AbstractScheduler* {
     	return new MinMinScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
     }));

src/scheduling/cmake.inc

@@ -4,8 +4,10 @@ set(wfr_cpp_files "${wfr_cpp_files}"
 	src/scheduling/SchedulerManager.cpp
 	src/scheduling/scheduler/dummy/DummyScheduler.cpp
 	src/scheduling/scheduler/heft/HEFTScheduler.cpp
+	src/scheduling/scheduler/cpop/CPOPScheduler.cpp
 	src/scheduling/scheduler/random/RandomScheduler.cpp
 	src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp
+	src/scheduling/scheduler/justintime2/JustInTime2Scheduler.cpp
 	src/scheduling/scheduler/minmin/MinMinScheduler.cpp
 	src/scheduling/scheduler/maxmin/MaxMinScheduler.cpp
 	src/scheduling/scheduler/sufferage/SufferageScheduler.cpp

src/scheduling/scheduler/cpop/CPOPScheduler.cpp

@@ -0,0 +1,390 @@
+
+
+#include <wfr/prediction/PredictionManager.h>
+#include <wfr/common/resource/ResourceManager.h>
+#include <wfr/common/wep/WorkflowExecutionPlan.h>
+#include <wfr/common/wep/AbstractWepNode.h>
+#include <wfr/common/wep/activities/AtomicWepNode.h>
+#include <wfr/common/wep/ActivityWepNode.h>
+#include <wfr/common/wfs/Activity.h>
+#include <wfr/common/wfs/ActivityInstance.h>
+#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
+#include <wfr/common/wfs/AtomicActivityImplementation.h>
+#include <wfr/common/wfs/templates/AtomicActivityTemplate.h>
+#include <wfr/enactment/EnactmentEngine.h>
+#include <wfr/common/scheduling/SchedulingDecision.h>
+#include <wfr/execution/ExecutionManager.h>
+#include <wfr/wep/utils/WepSearch.h>
+#include <wfr/common/utils/print.h>
+#include <iostream>
+#include <set>
+#include <boost/format.hpp>
+#include <wfr/scheduling/scheduler/cpop/CPOPScheduler.h>
+#include <queue>
+#include <memory>
+#include <cstring>
+
+
+namespace wfr {
+namespace scheduling {
+
+
+void CPOPScheduler::calcTaskRankD(wep::ActivityWepNode& node, std::vector<resource::ComputingResource*>& resources) {
+	auto entry = (TaskRankEntry*)node.getUserData().get();
+	if (!entry) {
+		entry = new TaskRankEntry(resources.size());
+		if (node.nodeType == wep::WepNodeType::Atomic) {
+			// get fastest implementation for each resource
+			unsigned ri = 0;
+			double sumExecutionTimes = 0.0;
+			unsigned sumExecutionTimeCount = 0;
+			for (auto res : resources) {
+				prediction::ExecutionTimeResult fastestPred;
+				wfs::AtomicActivityImplementation* fastestImpl = nullptr;
+				for (auto impl : node.activity.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>().atomicActivityType->getImplementations()) {
+					auto pred = predictionManager.predictExecutionTime(node.activity, *impl, res->resourceType, 1);
+					if (pred.isValid()) {
+						if (!fastestPred.isValid() || pred < fastestPred) {
+							fastestPred = pred;
+							fastestImpl = impl;
+						}
+					}
+				}
+				entry->fastestImplementations[ri] = fastestImpl;
+				entry->fastestExecutionTimes[ri] = fastestPred;
+				if (fastestPred.isValid()) {
+					sumExecutionTimeCount++;
+					sumExecutionTimes += fastestPred.executionTime.getValue();
+				}
+				ri++;
+			}
+			entry->averageExecutionTime = sumExecutionTimes / (double)sumExecutionTimeCount;
+		}
+		node.setUserData(std::shared_ptr<void>(entry));
+	}
+
+	if (!entry->finalizedD) {
+		auto predecessors = node.getPredecessors();
+		if (predecessors.size() == 0) {
+			// entry task
+			entry->rankD = 0;
+			entry->finalizedD = true;
+		} else {
+			bool predsFinalized = true;
+			double maxPredRank = 0.0;
+			for (auto p : predecessors) {
+				auto& pred = p->as<wep::ActivityWepNode>();
+				auto predEntry = (TaskRankEntry*)pred.getUserData().get();
+				if (!predEntry || !predEntry->finalizedD) {
+					// recursively calculate ranks
+					calcTaskRankD(pred, resources);
+					if (!predEntry) {
+						predEntry = (TaskRankEntry*)pred.getUserData().get();
+					}
+					predsFinalized = predsFinalized && predEntry->finalizedD;
+				}
+				// Get average communication time
+				double averaveCommTime = 0.0;
+				if (node.nodeType == wep::WepNodeType::Atomic && pred.nodeType == wep::WepNodeType::Atomic) {
+					double sumCommTime = 0.0;
+					unsigned sumCommTimeCount = 0;
+					for (auto res1 : resources) {
+						for (auto res2 : resources) {
+							if (res1 != res2) {
+								auto prediction = predictionManager.predictCommunicationTime(node.activity, *res1, pred.activity, *res2);
+								if (prediction.isValid) {
+									sumCommTime += prediction.communicationTime.getValue();
+									sumCommTimeCount++;
+								}
+							}
+						}
+					}
+					averaveCommTime = sumCommTime / (double)sumCommTimeCount;
+				}
+				double avgExecTime = 0;
+				if (pred.nodeType == wep::WepNodeType::Atomic) {
+					avgExecTime = predEntry->averageExecutionTime;
+				}
+				auto predRank = avgExecTime + averaveCommTime + predEntry->rankD;
+				if (predRank > maxPredRank) {
+					maxPredRank = predRank;
+				}
+			}
+			entry->rankD = maxPredRank;
+			if (predsFinalized) {
+				if (node.nodeType == wep::WepNodeType::CompositeEnd || node.nodeType == wep::WepNodeType::IterationEnd) {
+					if (node.getLastKnownInstanceStatus() >= wfs::ActivityInstanceStatus::ReadyForExecution) {
+						entry->finalizedD = true;
+					}
+				} else {
+					entry->finalizedD = true;
+				}
+			}
+		}
+	}
+}
+
+void CPOPScheduler::calcTaskRankU(wep::ActivityWepNode& node, std::vector<resource::ComputingResource*>& resources) {
+
+	auto entry = (TaskRankEntry*)node.getUserData().get();
+	if (!entry) {
+		entry = new TaskRankEntry(resources.size());
+		if (node.nodeType == wep::WepNodeType::Atomic) {
+			// get fastest implementation for each resource
+			unsigned ri = 0;
+			double sumExecutionTimes = 0.0;
+			unsigned sumExecutionTimeCount = 0;
+			for (auto res : resources) {
+				prediction::ExecutionTimeResult fastestPred;
+				wfs::AtomicActivityImplementation* fastestImpl = nullptr;
+				for (auto impl : node.activity.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>().atomicActivityType->getImplementations()) {
+					auto pred = predictionManager.predictExecutionTime(node.activity, *impl, res->resourceType, 1);
+					if (pred.isValid()) {
+						if (!fastestPred.isValid() || pred < fastestPred) {
+							fastestPred = pred;
+							fastestImpl = impl;
+						}
+					}
+				}
+				entry->fastestImplementations[ri] = fastestImpl;
+				entry->fastestExecutionTimes[ri] = fastestPred;
+				if (fastestPred.isValid()) {
+					sumExecutionTimeCount++;
+					sumExecutionTimes += fastestPred.executionTime.getValue();
+				}
+				ri++;
+			}
+			entry->averageExecutionTime = sumExecutionTimes / (double)sumExecutionTimeCount;
+		}
+		node.setUserData(std::shared_ptr<void>(entry));
+	}
+
+	if (!entry->finalizedU) {
+		auto successors = node.getSuccessors();
+		if (successors.size() == 0) {
+			// exit task
+			entry->rankU = entry->averageExecutionTime;
+			entry->finalizedU = true;
+		} else {
+			bool succsFinalized = true;
+			double maxSuccRank = 0.0;
+			for (auto s : successors) {
+				auto& succ = s->as<wep::ActivityWepNode>();
+				auto succEntry = (TaskRankEntry*)succ.getUserData().get();
+				if (!succEntry || !succEntry->finalizedU) {
+					// recursively calculate ranks
+					calcTaskRankU(succ, resources);
+					if (!succEntry) {
+						succEntry = (TaskRankEntry*)succ.getUserData().get();
+					}
+					succsFinalized = succsFinalized && succEntry->finalizedU;
+				}
+				// Get average communication time
+				double averaveCommTime = 0.0;
+				if (node.nodeType == wep::WepNodeType::Atomic && succ.nodeType == wep::WepNodeType::Atomic) {
+					double sumCommTime = 0.0;
+					unsigned sumCommTimeCount = 0;
+					for (auto res1 : resources) {
+						for (auto res2 : resources) {
+							if (res1 != res2) {
+								auto pred = predictionManager.predictCommunicationTime(node.activity, *res1, succ.activity, *res2);
+								if (pred.isValid) {
+									sumCommTime += pred.communicationTime.getValue();
+									sumCommTimeCount++;
+								}
+							}
+						}
+					}
+					averaveCommTime = sumCommTime / (double)sumCommTimeCount;
+				}
+				auto succRank = averaveCommTime + succEntry->rankU;
+				if (succRank > maxSuccRank) {
+					maxSuccRank = succRank;
+				}
+			}
+			if (node.nodeType == wep::WepNodeType::Atomic) {
+				entry->rankU = entry->averageExecutionTime + maxSuccRank;
+			} else {
+				entry->rankU = maxSuccRank;
+			}
+			if (succsFinalized) {
+				if (node.nodeType == wep::WepNodeType::CompositeStart || node.nodeType == wep::WepNodeType::IterationStart) {
+					if (node.getLastKnownInstanceStatus() >= wfs::ActivityInstanceStatus::ReadyForExecution) {
+						entry->finalizedU = true;
+					}
+				} else {
+					entry->finalizedU = true;
+				}
+			}
+		}
+	}
+
+	entry->priority = entry->rankU + entry->rankD;
+}
+
+
+void CPOPScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) {
+	auto resourceTypes = resourceManager.getComputingResourceTypes();
+	auto resources = resourceManager.getComputingResources();
+
+	calcTaskRankD(wepe.wep->getEndNode()->as<wep::ActivityWepNode>(), resources);
+	calcTaskRankU(wepe.wep->getStartNode()->as<wep::ActivityWepNode>(), resources);
+
+	readySet.insert(readyToExecute.begin(), readyToExecute.end());
+
+	std::set<wep::ActivityWepNode*> criticalPathNodes;
+	double cpPrio = ((TaskRankEntry*)wepe.wep->getStartNode()->as<wep::ActivityWepNode>().getUserData().get())->priority;
+	utils::wep::wepDepthFirstSearch(wepe.wep->getStartNode(),
+			[&criticalPathNodes, cpPrio](wep::AbstractWepNode* node)->bool {
+		if (node->nodeType == wep::WepNodeType::Atomic) {
+			auto t = node->as<wep::ActivityWepNode>();
+			double prio = ((TaskRankEntry*)t.getUserData().get())->priority;
+			/*std::cout << "\tTask-Prio: " << t.activity.baseActivity.baseTemplate.activityName << " (" << t.activity.instanceID << ")"
+					<< " [" << print::toString(t.activity.baseActivity.activityType()) << "]"
+					<< " [" << (int)t.nodeType << "]: " << prio << std::endl;*/
+			if (abs(prio - cpPrio) < 0.00001) { // because of rounding errors we cannot use "=="
+				criticalPathNodes.insert(&node->as<wep::ActivityWepNode>());
+			}
+		}
+		return true;
+	});
+
+	/*std::cout << std::endl <<"CriticalPathPriority: " << cpPrio << std::endl;
+	std::cout <<"CriticalPathPriority2: " << ((TaskRankEntry*)wepe.wep->getEndNode()->as<wep::ActivityWepNode>().getUserData().get())->priority << std::endl;
+	std::cout << "CriticalPathTasks: " << std::endl;
+	for (auto t : criticalPathNodes) {
+		std::cout << "\tTask: " << t->activity.baseActivity.baseTemplate.activityName << " (" << t->activity.instanceID << ")"
+				<< " [" << print::toString(t->activity.baseActivity.activityType()) << "]"
+				<< " [" << (int)t->nodeType << "]" << std::endl;
+	}*/
+
+	resource::ComputingResource* cpResource = nullptr;
+	double cpAvgExecTime = -1;
+	unsigned ri = 0;
+	for (auto res : resources) {
+		double sumExecTime = 0;
+		unsigned sumExecTimeCount = 0;
+		for (auto t : criticalPathNodes) {
+			auto entry = (TaskRankEntry*)t->getUserData().get();
+			sumExecTime += entry->fastestExecutionTimes[ri].executionTime.getValue();
+			sumExecTimeCount++;
+		}
+		double avgExecTime = sumExecTime / (double)sumExecTimeCount;
+		if (cpAvgExecTime < 0 || avgExecTime < cpAvgExecTime) {
+			cpAvgExecTime = avgExecTime;
+			cpResource = res;
+		}
+		ri++;
+	}
+	//std::cout << "CriticalPathProcessor: " << cpResource->resourceIdentifier << std::endl << std::endl;
+
+	scheduleReadySet(wepe, criticalPathNodes, cpResource, resources, schedulerLock);
+}
+
+
+void CPOPScheduler::scheduleReadySet(WepMapEntry& wepe, std::set<wep::ActivityWepNode*>& cpTasks, resource::ComputingResource* cpResource,
+		std::vector<resource::ComputingResource*>& resources, std::unique_lock<std::mutex>& schedulerLock) {
+	struct _schedulingPriorityQueueComparison {
+		bool operator()(wep::AtomicWepNode* e1, wep::AtomicWepNode* e2) {
+			auto entry1 = (TaskRankEntry*)e1->getUserData().get();
+			auto entry2 = (TaskRankEntry*)e2->getUserData().get();
+			if (entry1 && !entry2) {
+				return false;
+			} else if (!entry1 && entry2) {
+				return true;
+			} else if (entry1 && entry2) {
+				return entry1->priority < entry2->priority;
+			} else {
+				return false;
+			}
+		}
+	};
+	std::priority_queue<wep::AtomicWepNode*, std::vector<wep::AtomicWepNode*>, _schedulingPriorityQueueComparison> schedulingPriorityQueue;
+	// Put elements from ready set into priority queue thereby sorting them by rank
+	for (auto n : readySet) {
+		schedulingPriorityQueue.push(n);
+	}
+
+	while (!schedulingPriorityQueue.empty()) {
+		auto task = schedulingPriorityQueue.top();
+		schedulingPriorityQueue.pop();
+		auto entry = (TaskRankEntry*)task->getUserData().get();
+		prediction::ExecutionTimeResult fastestPred;
+		wfs::AtomicActivityImplementation* fastestImpl = nullptr;
+		resource::ComputingResource* fastestResource = nullptr;
+		if (cpTasks.find(task) != cpTasks.end()) {
+			unsigned ri = 0;
+			for (auto res : resources) {
+				if (res == cpResource) {
+					break;
+				}
+				ri++;
+			}
+			fastestImpl = entry->fastestImplementations[ri];
+			fastestPred = entry->fastestExecutionTimes[ri];
+			fastestResource = cpResource;
+		} else {
+			int ri = 0;
+			double earliestEstimatedFinishTime = -1;
+			for (auto res : resources) {
+				if (res != cpResource && entry->fastestImplementations[ri]) {
+					auto estimatedFinishTime = _getResourceEstimatedAvailableTime(res) + entry->fastestExecutionTimes[ri].executionTime.getValue();
+					if (earliestEstimatedFinishTime < 0.0 || estimatedFinishTime < earliestEstimatedFinishTime) {
+						earliestEstimatedFinishTime = estimatedFinishTime;
+						fastestImpl = entry->fastestImplementations[ri];
+						fastestPred = entry->fastestExecutionTimes[ri];
+						fastestResource = res;
+					}
+				}
+				ri++;
+			}
+		}
+		if (fastestResource) {
+
+			/*auto entry = (TaskRankEntry*)task->getUserData().get();
+			std::cout << task->activity.baseActivity.baseTemplate.activityName << "[" << task->activity.instanceID << "]: "
+					<< entry->rank << std::endl;
+			std::cout << "\tFastest Resource: " << fastestResource->resourceIdentifier << ": prediction = " << fastestPred.getExecutionTime().getValue() << std::endl;*/
+
+			auto reservation = resourceManager.reserveComputingResource(fastestResource, 1, true);
+			if (reservation) {
+				std::shared_ptr<SchedulingDecision> sd = std::make_shared<SchedulingDecision>(&task->activity, fastestImpl, wepe.wep, reservation);
+				sd->setPrediction(fastestPred);
+				task->activity.setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled);
+				executeSchedulingDecision(wepe, sd, schedulerLock);
+				readySet.erase(task);
+			}
+		}
+	}
+}
+
+void CPOPScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) {
+	//auto resources = resourceManager.getComputingResources();
+	//scheduleReadySet(wepe, resources, schedulerLock);
+}
+
+
+void CPOPScheduler::_unregisterWEP(WepMapEntry& wepe, std::unique_lock<std::mutex>& schedulerLock) {
+	/*std::cout << "Task Ranks: " << std::endl;
+	std::set<wep::AbstractWepNode*> visitedNodes;
+	auto visitFunc = [&](wep::AbstractWepNode* node) -> bool {
+		if (visitedNodes.find(node) == visitedNodes.end()) {
+			auto entry = (TaskRankEntry*)node->getUserData().get();
+			if (entry) {
+				std::cout << "\t" << node->as<wep::ActivityWepNode>().activity.baseActivity.baseTemplate.activityName
+						<< "[" << node->as<wep::ActivityWepNode>().activity.instanceID << "]: " << entry->rank << std::endl;
+			} else {
+				std::cout << "\t" << node->as<wep::ActivityWepNode>().activity.baseActivity.baseTemplate.activityName
+						<< "[" << node->as<wep::ActivityWepNode>().activity.instanceID << "]: No Entry" << std::endl;
+			}
+			visitedNodes.insert(node);
+		}
+		return true;
+	};
+	wfr::utils::wep::wepBreadthFirstSearch(wepe.wep->getStartNode(), visitFunc);*/
+}
+
+} /* namespace scheduler */
+} /* namespace wfr */
+
+

src/scheduling/scheduler/hcpt/HCPTScheduler.cpp

@@ -0,0 +1,259 @@
+
+
+#include <wfr/prediction/PredictionManager.h>
+#include <wfr/common/resource/ResourceManager.h>
+#include <wfr/common/wep/WorkflowExecutionPlan.h>
+#include <wfr/common/wep/AbstractWepNode.h>
+#include <wfr/common/wep/activities/AtomicWepNode.h>
+#include <wfr/common/wep/ActivityWepNode.h>
+#include <wfr/common/wfs/Activity.h>
+#include <wfr/common/wfs/ActivityInstance.h>
+#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
+#include <wfr/common/wfs/AtomicActivityImplementation.h>
+#include <wfr/common/wfs/templates/AtomicActivityTemplate.h>
+#include <wfr/enactment/EnactmentEngine.h>
+#include <wfr/common/scheduling/SchedulingDecision.h>
+#include <wfr/execution/ExecutionManager.h>
+#include <wfr/wep/utils/WepSearch.h>
+#include <wfr/common/utils/print.h>
+#include <iostream>
+#include <set>
+#include <boost/format.hpp>
+#include <wfr/scheduling/scheduler/hcpt/HCPTScheduler.h>
+#include <queue>
+#include <memory>
+#include <cstring>
+
+
+namespace wfr {
+namespace scheduling {
+
+
+
+void HCPTScheduler::calcTaskRanks(wep::ActivityWepNode& node, std::vector<resource::ComputingResource*>& resources) {
+	/*std::cout << "calcTaskRanks: " << node.activity.baseActivity.baseTemplate.activityName << " (" << node.activity.instanceID << ")"
+			<< " [" << print::toString(node.activity.baseActivity.activityType()) << "]"
+			<< " [" << (int)node.nodeType << "]" << std::endl;*/
+
+	auto entry = (TaskRankEntry*)node.getUserData().get();
+	if (!entry) {
+		entry = new TaskRankEntry(resources.size());
+		if (node.nodeType == wep::WepNodeType::Atomic) {
+			// get fastest implementation for each resource
+			unsigned ri = 0;
+			double sumExecutionTimes = 0.0;
+			unsigned sumExecutionTimeCount = 0;
+			for (auto res : resources) {
+				prediction::ExecutionTimeResult fastestPred;
+				wfs::AtomicActivityImplementation* fastestImpl = nullptr;
+				for (auto impl : node.activity.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>().atomicActivityType->getImplementations()) {
+					auto pred = predictionManager.predictExecutionTime(node.activity, *impl, res->resourceType, 1);
+					if (pred.isValid()) {
+						if (!fastestPred.isValid() || pred < fastestPred) {
+							fastestPred = pred;
+							fastestImpl = impl;
+						}
+					}
+				}
+				entry->fastestImplementations[ri] = fastestImpl;
+				entry->fastestExecutionTimes[ri] = fastestPred;
+				if (fastestPred.isValid()) {
+					sumExecutionTimeCount++;
+					sumExecutionTimes += fastestPred.executionTime.getValue();
+				}
+				ri++;
+			}
+			entry->averageExecutionTime = sumExecutionTimes / (double)sumExecutionTimeCount;
+		}
+		node.setUserData(std::shared_ptr<void>(entry));
+	}
+
+	if (!entry->aestFinalized) {
+		auto predecessors = node.getPredecessors();
+		if (predecessors.size() == 0) {
+			//entry task
+			//std::cout << "\tNo predecessors => entry task: " << std::endl;
+			entry->aest = 0;
+		} else {
+			double maxPAest = -1;
+			bool allPredsFinalized = true;
+			for (auto p : predecessors) {
+				auto& pred = p->as<wep::ActivityWepNode>();
+				auto predEntry = (TaskRankEntry*)pred.getUserData().get();
+				double avgExecTime = 0;
+				if (pred.nodeType == wep::WepNodeType::Atomic) {
+					avgExecTime = predEntry->averageExecutionTime;
+				}
+				double pAest = predEntry->aest + avgExecTime;
+				if (maxPAest < 0 || pAest > maxPAest) {
+					maxPAest = pAest;
+				}
+				if (!predEntry->aestFinalized) {
+					allPredsFinalized = true;
+				}
+			}
+			entry->aest = maxPAest;
+			if (allPredsFinalized) {
+				entry->aest = true;
+			} else {
+				entry->aest = false;
+			}
+		}
+	}
+
+	if (!entry->alstFinalized) {
+		auto successors = node.getSuccessors();
+		if (successors.size() == 0) {
+			// exit task
+			//std::cout << "\tNo successors => exit task: " << entry->aest << std::endl;
+			entry->alst = entry->aest;
+			entry->alstFinalized = entry->aestFinalized;
+		} else {
+			double minPAlst = -1;
+			bool allSuccsFinalized = true;
+			for (auto s : successors) {
+				auto& succ = s->as<wep::ActivityWepNode>();
+				auto succEntry = (TaskRankEntry*)succ.getUserData().get();
+				if (!succEntry || !succEntry->alstFinalized) {
+					// recursively calculate alst
+					//std::cout << "recursive call start" << std::endl;
+					calcTaskRanks(succ, resources);
+					//std::cout << "recursive call end" << std::endl;
+					if (!succEntry) {
+						succEntry = (TaskRankEntry*)succ.getUserData().get();
+					}
+					allSuccsFinalized = allSuccsFinalized && succEntry->alstFinalized;
+				}
+				double sAlst = succEntry->alst;
+				if (minPAlst < 0 || sAlst < minPAlst) {
+					minPAlst = sAlst;
+				}
+			}
+			double avgExecTime = 0;
+			if (node.nodeType == wep::WepNodeType::Atomic) {
+				avgExecTime = entry->averageExecutionTime;
+			}
+			entry->alst = minPAlst - avgExecTime;
+			if (allSuccsFinalized) {
+				if (node.nodeType == wep::WepNodeType::CompositeStart || node.nodeType == wep::WepNodeType::IterationStart) {
+					if (node.getLastKnownInstanceStatus() >= wfs::ActivityInstanceStatus::ReadyForExecution) {
+						entry->alstFinalized = true;
+					}
+				} else {
+					entry->alstFinalized = true;
+				}
+			}
+		}
+	}
+}
+
+
+void HCPTScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) {
+	auto resourceTypes = resourceManager.getComputingResourceTypes();
+	auto resources = resourceManager.getComputingResources();
+
+	calcTaskRanks(wepe.wep->getStartNode()->as<wep::ActivityWepNode>(), resources);
+
+	readySet.insert(readyToExecute.begin(), readyToExecute.end());
+
+	scheduleReadySet(wepe, resources, schedulerLock);
+}
+
+
+void HCPTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector<resource::ComputingResource*>& resources, std::unique_lock<std::mutex>& schedulerLock) {
+
+
+
+
+	struct _schedulingPriorityQueueComparison {
+		bool operator()(wep::AtomicWepNode* e1, wep::AtomicWepNode* e2) {
+			auto entry1 = (TaskRankEntry*)e1->getUserData().get();
+			auto entry2 = (TaskRankEntry*)e2->getUserData().get();
+			if (entry1 && !entry2) {
+				return false;
+			} else if (!entry1 && entry2) {
+				return true;
+			} else if (entry1 && entry2) {
+				return entry1->rank < entry2->rank;
+			} else {
+				return false;
+			}
+		}
+	};
+	std::priority_queue<wep::AtomicWepNode*, std::vector<wep::AtomicWepNode*>, _schedulingPriorityQueueComparison> schedulingPriorityQueue;
+	// Put elements from ready set into priority queue thereby sorting them by rank
+	for (auto n : readySet) {
+		schedulingPriorityQueue.push(n);
+	}
+
+	while (!schedulingPriorityQueue.empty()) {
+		auto task = schedulingPriorityQueue.top();
+		schedulingPriorityQueue.pop();
+		auto entry = (TaskRankEntry*)task->getUserData().get();
+		int ri = 0;
+
+		prediction::ExecutionTimeResult fastestPred;
+		wfs::AtomicActivityImplementation* fastestImpl = nullptr;
+		resource::ComputingResource* fastestResource = nullptr;
+		double earliestEstimatedFinishTime = -1;
+		for (auto res : resources) {
+			if (entry->fastestImplementations[ri]) {
+				auto estimatedFinishTime = _getResourceEstimatedAvailableTime(res) + entry->fastestExecutionTimes[ri].executionTime.getValue();
+				if (earliestEstimatedFinishTime < 0.0 || estimatedFinishTime < earliestEstimatedFinishTime) {
+					earliestEstimatedFinishTime = estimatedFinishTime;
+					fastestImpl = entry->fastestImplementations[ri];
+					fastestPred = entry->fastestExecutionTimes[ri];
+					fastestResource = res;
+				}
+			}
+			ri++;
+		}
+		if (fastestResource) {
+
+			/*auto entry = (TaskRankEntry*)task->getUserData().get();
+			std::cout << task->activity.baseActivity.baseTemplate.activityName << "[" << task->activity.instanceID << "]: "
+					<< entry->rank << std::endl;
+			std::cout << "\tFastest Resource: " << fastestResource->resourceIdentifier << ": prediction = " << fastestPred.getExecutionTime().getValue() << std::endl;*/
+
+			auto reservation = resourceManager.reserveComputingResource(fastestResource, 1, true);
+			if (reservation) {
+				std::shared_ptr<SchedulingDecision> sd = std::make_shared<SchedulingDecision>(&task->activity, fastestImpl, wepe.wep, reservation);
+				sd->setPrediction(fastestPred);
+				task->activity.setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled);
+				executeSchedulingDecision(wepe, sd, schedulerLock);
+				readySet.erase(task);
+			}
+		}
+	}
+}
+
+void HCPTScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) {
+	//auto resources = resourceManager.getComputingResources();
+	//scheduleReadySet(wepe, resources, schedulerLock);
+}
+
+
+void HCPTScheduler::_unregisterWEP(WepMapEntry& wepe, std::unique_lock<std::mutex>& schedulerLock) {
+	/*std::cout << "Task Ranks: " << std::endl;
+	std::set<wep::AbstractWepNode*> visitedNodes;
+	auto visitFunc = [&](wep::AbstractWepNode* node) -> bool {
+		if (visitedNodes.find(node) == visitedNodes.end()) {
+			auto entry = (TaskRankEntry*)node->getUserData().get();
+			if (entry) {
+				std::cout << "\t" << node->as<wep::ActivityWepNode>().activity.baseActivity.baseTemplate.activityName
+						<< "[" << node->as<wep::ActivityWepNode>().activity.instanceID << "]: " << entry->rank << std::endl;
+			} else {
+				std::cout << "\t" << node->as<wep::ActivityWepNode>().activity.baseActivity.baseTemplate.activityName
+						<< "[" << node->as<wep::ActivityWepNode>().activity.instanceID << "]: No Entry" << std::endl;
+			}
+			visitedNodes.insert(node);
+		}
+		return true;
+	};
+	wfr::utils::wep::wepBreadthFirstSearch(wepe.wep->getStartNode(), visitFunc);*/
+}
+
+} /* namespace scheduler */
+} /* namespace wfr */
+
+

src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp

@@ -35,28 +35,32 @@ void JustInTimeScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std
 		wfs::AtomicActivityImplementation* fastestImpl = nullptr;
 		resource::ComputingResource* fastestRes = nullptr;
 		prediction::ExecutionTimeResult fastestPred;
+		double minEFT = -1;
+
 		for (auto restype : resourceTypes) {
-			resource::ComputingResource* freeRes = nullptr;
+			wfs::AtomicActivityImplementation* bestImpl = nullptr;
-			for (auto res : restype->getResources()) {
+			prediction::ExecutionTimeResult bestPred;
-				if (res->getFreeSlotCount() > 0) {
+			auto& aat = ai.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>();
-					freeRes = res;
+			for (auto impl : aat.atomicActivityType->getImplementations()) {
-					break;
+				auto execPred = predictionManager.predictExecutionTime(ai, *impl, *restype);
-				}
+				if (execPred.isValid()) {
-			}
+					if (!bestImpl || !bestPred.isValid() || execPred.executionTime.getValue() < bestPred.executionTime.getValue()) {
-			if (freeRes) {
+						bestImpl = impl;
-				auto& aat = ai.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>();
+						bestPred = execPred;
-				for (auto impl : aat.atomicActivityType->getImplementations()) {
-					auto execPred = predictionManager.predictExecutionTime(ai, *impl, *restype);
-					if (execPred.isValid()) {
-						if (!fastestImpl || !fastestPred.isValid() || execPred.executionTime.getValue() < fastestPred.executionTime.getValue()) {
-							fastestImpl = impl;
-							fastestRes = freeRes;
-							fastestPred = execPred;
-						}
 					}
 				}
 			}
+			for (auto res : restype->getResources()) {
+				auto eft = _getResourceEstimatedAvailableTime(res) + bestPred.executionTime.getValue();
+				if (minEFT < 0 || eft < minEFT) {
+					minEFT = eft;
+					fastestImpl = bestImpl;
+					fastestRes = res;
+					fastestPred = bestPred;
+				}
+			}
 		}
+
 		if (fastestImpl) {
 			auto reservation = resourceManager.reserveComputingResource(fastestRes, 1, false);
 			auto sd = std::make_shared<SchedulingDecision>(&ai, fastestImpl, wepe.wep, reservation);

src/scheduling/scheduler/justintime2/JustInTime2Scheduler.cpp

@@ -0,0 +1,82 @@
+
+#include <wfr/scheduling/scheduler/justintime2/JustInTime2Scheduler.h>
+#include <wfr/common/wep/WorkflowExecutionPlan.h>
+#include <wfr/common/wep/AbstractWepNode.h>
+#include <wfr/common/wep/activities/AtomicWepNode.h>
+#include <wfr/common/wfs/Activity.h>
+#include <wfr/common/wfs/ActivityInstance.h>
+#include <wfr/common/wfs/AtomicActivityImplementation.h>
+#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
+#include <wfr/common/wfs/templates/AtomicActivityTemplate.h>
+#include <wfr/prediction/PredictionManager.h>
+#include <wfr/common/resource/ResourceManager.h>
+#include <wfr/enactment/EnactmentEngine.h>
+#include <wfr/common/scheduling/SchedulingDecision.h>
+#include <wfr/execution/ExecutionManager.h>
+#include <wfr/WorkflowRuntime.h>
+#include <wfr/wep/utils/WepSearch.h>
+#include <iostream>
+#include <set>
+#include <queue>
+
+
+namespace wfr {
+namespace scheduling {
+
+
+void JustInTime2Scheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) {
+	for (auto n : readyToExecute) {
+		nodeQueue.insert(n);
+	}
+	auto resourceTypes = resourceManager.getComputingResourceTypes();
+	std::vector<wep::AtomicWepNode*> scheduledQueue;
+	for (auto n : nodeQueue) {
+		auto& ai = n->activity.as<wfs::AtomicActivityInstance>();
+		wfs::AtomicActivityImplementation* fastestImpl = nullptr;
+		resource::ComputingResource* fastestRes = nullptr;
+		prediction::ExecutionTimeResult fastestPred;
+
+		for (auto restype : resourceTypes) {
+			wfs::AtomicActivityImplementation* bestImpl = nullptr;
+			prediction::ExecutionTimeResult bestPred;
+			auto& aat = ai.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>();
+			for (auto impl : aat.atomicActivityType->getImplementations()) {
+				auto execPred = predictionManager.predictExecutionTime(ai, *impl, *restype);
+				if (execPred.isValid()) {
+					if (!bestImpl || !bestPred.isValid() || execPred.executionTime.getValue() < bestPred.executionTime.getValue()) {
+						bestImpl = impl;
+						bestPred = execPred;
+					}
+				}
+			}
+			if (!fastestPred.isValid() || bestPred.executionTime.getValue() < fastestPred.executionTime.getValue()) {
+				for (auto res : restype->getResources()) {
+					if (res->getFreeSlotCount() > 0 ) {
+						fastestImpl = bestImpl;
+						fastestRes = res;
+						fastestPred = bestPred;
+						break;
+					}
+				}
+			}
+		}
+
+		if (fastestImpl) {
+			auto reservation = resourceManager.reserveComputingResource(fastestRes, 1, false);
+			auto sd = std::make_shared<SchedulingDecision>(&ai, fastestImpl, wepe.wep, reservation);
+			sd->setPrediction(fastestPred);
+			ai.setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled);
+			scheduledQueue.push_back(n);
+			executeSchedulingDecision(wepe, sd, schedulerLock);
+		} else {
+			//throw exception::wfr_fatal_error("JustInTimeScheduler could not find a suitable implementation.");
+		}
+	}
+	for (auto n : scheduledQueue) {
+		nodeQueue.erase(n);
+	}
+}
+
+
+} /* namespace scheduling */
+} /* namespace wfr */

src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp

@@ -66,9 +66,6 @@ void LookaheadScheduler::calcTaskRanks(wep::ActivityWepNode& node, std::vector<r
 			entry->averageExecutionTime = sumExecutionTimes / (double)sumExecutionTimeCount;
 		}
 		node.setUserData(std::shared_ptr<void>(entry));
-		if (node.nodeType == wep::WepNodeType::Atomic) {
-			node.activity.setUserData(std::shared_ptr<void>(entry));
-		}
 	}
 	if (!entry->finalized) {
 		auto successors = node.getSuccessors();
@@ -194,16 +191,29 @@ void LookaheadScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector<resourc
 		resource::ComputingResource* fastestResource = nullptr;
 		double earliestEstimatedFinishTime = -1;
 		double minSuccEFT = -1;
-		auto succs = task->activity.getSuccessors();
+
+		// get atomic activity successors
+		std::set<wfr::wep::AbstractWepNode*> successors;
+		auto visitSuccsFunc = [&successors, task](wfr::wep::AbstractWepNode* node)->bool {
+			if (node == task) {
+				return true;
+			} else if (node->nodeType == wep::WepNodeType::Atomic) {
+				successors.insert(node);
+				return false;
+			} else {
+				return true;
+			}
+		};
+		utils::wep::wepDepthFirstSearch(task, visitSuccsFunc);
+
 		for (auto res : resources) {
 			if (entry->fastestImplementations[ri]) {
 				auto estimatedFinishTime = _getResourceEstimatedAvailableTime(res) + entry->fastestExecutionTimes[ri].executionTime.getValue();
 
-				auto maxSuccEFT = -1;
+				if (successors.size() > 0) {
-				if (succs.size() > 0) {
 					// do lookahead
 					struct _succSchedulingQueueComparison {
-						bool operator()(wfs::ActivityInstance* e1, wfs::ActivityInstance* e2) {
+						bool operator()(wfr::wep::AbstractWepNode* e1, wfr::wep::AbstractWepNode* e2) {
 							auto entry1 = (TaskRankEntry*)e1->getUserData().get();
 							auto entry2 = (TaskRankEntry*)e2->getUserData().get();
 							if (entry1 && !entry2) {
@@ -217,35 +227,39 @@ void LookaheadScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector<resourc
 							}
 						}
 					};
-					std::priority_queue<wfs::ActivityInstance*, std::vector<wfs::ActivityInstance*>, _succSchedulingQueueComparison> succSchedulingQueue;
+					std::priority_queue<wfr::wep::AbstractWepNode*, std::vector<wfr::wep::AbstractWepNode*>, _succSchedulingQueueComparison> succSchedulingQueue;
+					for (auto succ : successors) {
+						succSchedulingQueue.push(succ);
+					}
 					std::map<resource::ComputingResource*, double> resourceEATMap;
 					resourceEATMap[res] = estimatedFinishTime;
-					for (auto succ : succs) {
-						if (succ->activityType() == wfs::ActivityType::Atomic) {
-							succSchedulingQueue.push(succ);
-						}
-					}
 					while (!succSchedulingQueue.empty()) {
 						auto succ = succSchedulingQueue.top();
 						succSchedulingQueue.pop();
 						auto succEntry = (TaskRankEntry*)succ->getUserData().get();
 						unsigned ri2 = 0;
-						unsigned minEFT = -1;
+						double minEFT = -1;
 						resource::ComputingResource* minEFTResource = nullptr;
 						for (auto res2 : resources) {
 							double succEFT = -1;
 							if (entry->fastestImplementations[ri2]) {
-								succEFT = resourceEATMap[res2] + succEntry->fastestExecutionTimes[ri2].executionTime.getValue();
+								auto resourceEAT = resourceEATMap[res2];
+								if (resourceEAT <= 0) {
+									resourceEAT =  _getResourceEstimatedAvailableTime(res2);
+								}
+								succEFT = resourceEAT + succEntry->fastestExecutionTimes[ri2].executionTime.getValue();
 							}
 							if (succEFT > 0 && (minEFT < 0 || succEFT < minEFT)) {
 								minEFT = succEFT;
 								minEFTResource = res2;
 							}
+							ri2++;
 						}
 						if (minEFT > 0) {
 							resourceEATMap[minEFTResource] = minEFT;
 						}
 					}
+					double maxSuccEFT = -1;
 					for (auto e : resourceEATMap) {
 						if (maxSuccEFT < 0 || e.second > maxSuccEFT) {
 							maxSuccEFT = e.second;