From 8d43a1c3c2525916c020e71ef9c49bb77310337e Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 27 Jul 2017 19:31:54 +0200 Subject: [PATCH] . --- include/wfr/common/wfs/Workflow.h | 6 + .../scheduling/scheduler/cpop/CPOPScheduler.h | 85 ++++ .../scheduling/scheduler/hcpt/HCPTScheduler.h | 81 ++++ .../justintime2/JustInTime2Scheduler.h | 34 ++ include/wfr/wep/utils/WepSearch.h | 14 +- src/common/wfs/Workflow.cpp | 10 + src/enactment/EnactmentEngine.cpp | 15 +- src/scheduling/SchedulerManager.cpp | 8 + src/scheduling/cmake.inc | 2 + .../scheduler/cpop/CPOPScheduler.cpp | 390 ++++++++++++++++++ .../scheduler/hcpt/HCPTScheduler.cpp | 259 ++++++++++++ .../justintime/JustInTimeScheduler.cpp | 38 +- .../justintime2/JustInTime2Scheduler.cpp | 82 ++++ .../lookahead/LookaheadScheduler.cpp | 44 +- 14 files changed, 1027 insertions(+), 41 deletions(-) create mode 100755 include/wfr/scheduling/scheduler/cpop/CPOPScheduler.h create mode 100755 include/wfr/scheduling/scheduler/hcpt/HCPTScheduler.h create mode 100755 include/wfr/scheduling/scheduler/justintime2/JustInTime2Scheduler.h create mode 100755 src/scheduling/scheduler/cpop/CPOPScheduler.cpp create mode 100755 src/scheduling/scheduler/hcpt/HCPTScheduler.cpp create mode 100755 src/scheduling/scheduler/justintime2/JustInTime2Scheduler.cpp diff --git a/include/wfr/common/wfs/Workflow.h b/include/wfr/common/wfs/Workflow.h index 7fe568e..8012fab 100755 --- a/include/wfr/common/wfs/Workflow.h +++ b/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 lock(statMutex); diff --git a/include/wfr/scheduling/scheduler/cpop/CPOPScheduler.h b/include/wfr/scheduling/scheduler/cpop/CPOPScheduler.h new file mode 100755 index 0000000..c0f6120 --- /dev/null +++ b/include/wfr/scheduling/scheduler/cpop/CPOPScheduler.h @@ -0,0 +1,85 @@ + +#pragma once + + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +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 readySet; + + void calcTaskRankD(wep::ActivityWepNode& node, std::vector& resources); + + void calcTaskRankU(wep::ActivityWepNode& node, std::vector& resources); + + void scheduleReadySet(WepMapEntry& wepe, std::set& cpTasks, resource::ComputingResource* cpResource, + std::vector& resources, std::unique_lock& schedulerLock); + +public: + CPOPScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) + : resourceManager(resourceManager), predictionManager(predictionManager) {} + + ~CPOPScheduler() {} + + virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock) override; + virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) override; + virtual void _unregisterWEP(WepMapEntry& wepe, std::unique_lock& schedulerLock) override; +}; + +} /* namespace scheduling */ +} /* namespace wfr */ + + diff --git a/include/wfr/scheduling/scheduler/hcpt/HCPTScheduler.h b/include/wfr/scheduling/scheduler/hcpt/HCPTScheduler.h new file mode 100755 index 0000000..eba2d4a --- /dev/null +++ b/include/wfr/scheduling/scheduler/hcpt/HCPTScheduler.h @@ -0,0 +1,81 @@ + +#pragma once + + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +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 readySet; + + void calcTaskRanks(wep::ActivityWepNode& node, std::vector& resources); + + void scheduleReadySet(WepMapEntry& wepe, std::vector& resources, std::unique_lock& schedulerLock); + +public: + HCPTScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) + : resourceManager(resourceManager), predictionManager(predictionManager) {} + + ~HCPTScheduler() {} + + virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock) override; + virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) override; + virtual void _unregisterWEP(WepMapEntry& wepe, std::unique_lock& schedulerLock) override; +}; + +} /* namespace scheduling */ +} /* namespace wfr */ + + diff --git a/include/wfr/scheduling/scheduler/justintime2/JustInTime2Scheduler.h b/include/wfr/scheduling/scheduler/justintime2/JustInTime2Scheduler.h new file mode 100755 index 0000000..7b39e86 --- /dev/null +++ b/include/wfr/scheduling/scheduler/justintime2/JustInTime2Scheduler.h @@ -0,0 +1,34 @@ + +#pragma once + + +#include + + +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 nodeQueue; + +public: + JustInTime2Scheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) + : resourceManager(resourceManager), predictionManager(predictionManager) {} + + virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock); +}; + +} /* namespace scheduling */ +} /* namespace wfr */ + diff --git a/include/wfr/wep/utils/WepSearch.h b/include/wfr/wep/utils/WepSearch.h index 3f67ac0..317b21e 100755 --- a/include/wfr/wep/utils/WepSearch.h +++ b/include/wfr/wep/utils/WepSearch.h @@ -17,15 +17,23 @@ namespace wep { template void wepDepthFirstSearch(wfr::wep::AbstractWepNode* startNode, const VisitType& visitFunction, - std::set& visitedNodes = std::set()) { - visitedNodes.insert(startNode); + std::set* visitedNodes = nullptr) { + bool deleteVisitedNodeSet = false; + if (visitedNodes == nullptr) { + visitedNodes = new std::set(); + 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; + } } diff --git a/src/common/wfs/Workflow.cpp b/src/common/wfs/Workflow.cpp index 8a01831..7baed18 100755 --- a/src/common/wfs/Workflow.cpp +++ b/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(enactmentTime).count())/1000.0 << " sec" << std::endl; + sout << "WFR-Statistics: Time till WEP Completion: " + << ((double)std::chrono::duration_cast(getTimeTillWepCompletion()).count())/1000.0 << " sec" << std::endl; sout << "WFR-Statistics: Workflow Duration: " << ((double)std::chrono::duration_cast(getWorkflowDuration()).count())/1000.0 << " sec" << std::endl; sout << "WFR-Statistics: Makespan: " diff --git a/src/enactment/EnactmentEngine.cpp b/src/enactment/EnactmentEngine.cpp index 897e5d4..03fe393 100755 --- a/src/enactment/EnactmentEngine.cpp +++ b/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")) { - WorkflowRuntime::setGlobalConfig("dotWriterEnabled", false); - std::cerr << "WEP-to-.dot writer is disabled." << std::endl; - } else { + if (vm.count("enable-wepdotwriter")) { 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) { diff --git a/src/scheduling/SchedulerManager.cpp b/src/scheduling/SchedulerManager.cpp index 4e50fc6..10bf087 100755 --- a/src/scheduling/SchedulerManager.cpp +++ b/src/scheduling/SchedulerManager.cpp @@ -12,10 +12,12 @@ #include #include #include +#include #include #include #include #include +#include #include 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()); })); diff --git a/src/scheduling/cmake.inc b/src/scheduling/cmake.inc index 975fd60..e35095b 100755 --- a/src/scheduling/cmake.inc +++ b/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 diff --git a/src/scheduling/scheduler/cpop/CPOPScheduler.cpp b/src/scheduling/scheduler/cpop/CPOPScheduler.cpp new file mode 100755 index 0000000..f6ab949 --- /dev/null +++ b/src/scheduling/scheduler/cpop/CPOPScheduler.cpp @@ -0,0 +1,390 @@ + + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace wfr { +namespace scheduling { + + +void CPOPScheduler::calcTaskRankD(wep::ActivityWepNode& node, std::vector& 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().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(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(); + 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& 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().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(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(); + 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 readyToExecute, std::unique_lock& schedulerLock) { + auto resourceTypes = resourceManager.getComputingResourceTypes(); + auto resources = resourceManager.getComputingResources(); + + calcTaskRankD(wepe.wep->getEndNode()->as(), resources); + calcTaskRankU(wepe.wep->getStartNode()->as(), resources); + + readySet.insert(readyToExecute.begin(), readyToExecute.end()); + + std::set criticalPathNodes; + double cpPrio = ((TaskRankEntry*)wepe.wep->getStartNode()->as().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(); + 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()); + } + } + return true; + }); + + /*std::cout << std::endl <<"CriticalPathPriority: " << cpPrio << std::endl; + std::cout <<"CriticalPathPriority2: " << ((TaskRankEntry*)wepe.wep->getEndNode()->as().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& cpTasks, resource::ComputingResource* cpResource, + std::vector& resources, std::unique_lock& 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, _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 sd = std::make_shared(&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 sd, std::unique_lock& schedulerLock) { + //auto resources = resourceManager.getComputingResources(); + //scheduleReadySet(wepe, resources, schedulerLock); +} + + +void CPOPScheduler::_unregisterWEP(WepMapEntry& wepe, std::unique_lock& schedulerLock) { + /*std::cout << "Task Ranks: " << std::endl; + std::set 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().activity.baseActivity.baseTemplate.activityName + << "[" << node->as().activity.instanceID << "]: " << entry->rank << std::endl; + } else { + std::cout << "\t" << node->as().activity.baseActivity.baseTemplate.activityName + << "[" << node->as().activity.instanceID << "]: No Entry" << std::endl; + } + visitedNodes.insert(node); + } + return true; + }; + wfr::utils::wep::wepBreadthFirstSearch(wepe.wep->getStartNode(), visitFunc);*/ +} + +} /* namespace scheduler */ +} /* namespace wfr */ + + diff --git a/src/scheduling/scheduler/hcpt/HCPTScheduler.cpp b/src/scheduling/scheduler/hcpt/HCPTScheduler.cpp new file mode 100755 index 0000000..a11eea3 --- /dev/null +++ b/src/scheduling/scheduler/hcpt/HCPTScheduler.cpp @@ -0,0 +1,259 @@ + + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace wfr { +namespace scheduling { + + + +void HCPTScheduler::calcTaskRanks(wep::ActivityWepNode& node, std::vector& 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().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(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(); + 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(); + 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 readyToExecute, std::unique_lock& schedulerLock) { + auto resourceTypes = resourceManager.getComputingResourceTypes(); + auto resources = resourceManager.getComputingResources(); + + calcTaskRanks(wepe.wep->getStartNode()->as(), resources); + + readySet.insert(readyToExecute.begin(), readyToExecute.end()); + + scheduleReadySet(wepe, resources, schedulerLock); +} + + +void HCPTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector& resources, std::unique_lock& 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, _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 sd = std::make_shared(&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 sd, std::unique_lock& schedulerLock) { + //auto resources = resourceManager.getComputingResources(); + //scheduleReadySet(wepe, resources, schedulerLock); +} + + +void HCPTScheduler::_unregisterWEP(WepMapEntry& wepe, std::unique_lock& schedulerLock) { + /*std::cout << "Task Ranks: " << std::endl; + std::set 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().activity.baseActivity.baseTemplate.activityName + << "[" << node->as().activity.instanceID << "]: " << entry->rank << std::endl; + } else { + std::cout << "\t" << node->as().activity.baseActivity.baseTemplate.activityName + << "[" << node->as().activity.instanceID << "]: No Entry" << std::endl; + } + visitedNodes.insert(node); + } + return true; + }; + wfr::utils::wep::wepBreadthFirstSearch(wepe.wep->getStartNode(), visitFunc);*/ +} + +} /* namespace scheduler */ +} /* namespace wfr */ + + diff --git a/src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp b/src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp index a178f2b..ec9224f 100755 --- a/src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp +++ b/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; - for (auto res : restype->getResources()) { - if (res->getFreeSlotCount() > 0) { - freeRes = res; - break; - } - } - if (freeRes) { - auto& aat = ai.baseActivity.baseTemplate.as(); - 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; - } + wfs::AtomicActivityImplementation* bestImpl = nullptr; + prediction::ExecutionTimeResult bestPred; + auto& aat = ai.baseActivity.baseTemplate.as(); + 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; } } } + 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(&ai, fastestImpl, wepe.wep, reservation); diff --git a/src/scheduling/scheduler/justintime2/JustInTime2Scheduler.cpp b/src/scheduling/scheduler/justintime2/JustInTime2Scheduler.cpp new file mode 100755 index 0000000..df6e7d5 --- /dev/null +++ b/src/scheduling/scheduler/justintime2/JustInTime2Scheduler.cpp @@ -0,0 +1,82 @@ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace wfr { +namespace scheduling { + + +void JustInTime2Scheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock) { + for (auto n : readyToExecute) { + nodeQueue.insert(n); + } + auto resourceTypes = resourceManager.getComputingResourceTypes(); + std::vector scheduledQueue; + for (auto n : nodeQueue) { + auto& ai = n->activity.as(); + 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(); + 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(&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 */ diff --git a/src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp b/src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp index 17bc5fd..82add38 100755 --- a/src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp +++ b/src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp @@ -66,9 +66,6 @@ void LookaheadScheduler::calcTaskRanks(wep::ActivityWepNode& node, std::vectoraverageExecutionTime = sumExecutionTimes / (double)sumExecutionTimeCount; } node.setUserData(std::shared_ptr(entry)); - if (node.nodeType == wep::WepNodeType::Atomic) { - node.activity.setUserData(std::shared_ptr(entry)); - } } if (!entry->finalized) { auto successors = node.getSuccessors(); @@ -194,16 +191,29 @@ void LookaheadScheduler::scheduleReadySet(WepMapEntry& wepe, std::vectoractivity.getSuccessors(); + + // get atomic activity successors + std::set 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 (succs.size() > 0) { + if (successors.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, _succSchedulingQueueComparison> succSchedulingQueue; + std::priority_queue, _succSchedulingQueueComparison> succSchedulingQueue; + for (auto succ : successors) { + succSchedulingQueue.push(succ); + } std::map 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;