From 19d695d4d8e642cea8bab542a039c7925c86446f Mon Sep 17 00:00:00 2001 From: Matthias Date: Wed, 26 Jul 2017 18:56:44 +0200 Subject: [PATCH] Added lookahead scheduler. Most scheduler now use estimated finish time instead of predicted execution time. --- include/wfr/scheduling/AbstractScheduler.h | 3 + .../scheduling/scheduler/heft/HEFTScheduler.h | 2 - .../scheduler/lookahead/LookaheadScheduler.h | 79 +++++ .../scheduler/minmin/MinMinScheduler.h | 4 - .../scheduler/sufferage/SufferageScheduler.h | 4 - src/scheduling/AbstractScheduler.cpp | 38 +++ src/scheduling/SchedulerManager.cpp | 4 + src/scheduling/cmake.inc | 1 + .../scheduler/heft/HEFTScheduler.cpp | 18 +- .../lookahead/LookaheadScheduler.cpp | 319 ++++++++++++++++++ .../scheduler/minmin/MinMinScheduler.cpp | 44 +-- .../scheduler/peft/PEFTScheduler.cpp | 18 +- .../sufferage/SufferageScheduler.cpp | 44 +-- 13 files changed, 468 insertions(+), 110 deletions(-) create mode 100755 include/wfr/scheduling/scheduler/lookahead/LookaheadScheduler.h create mode 100755 src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp diff --git a/include/wfr/scheduling/AbstractScheduler.h b/include/wfr/scheduling/AbstractScheduler.h index a6bb7e2..1f7e189 100755 --- a/include/wfr/scheduling/AbstractScheduler.h +++ b/include/wfr/scheduling/AbstractScheduler.h @@ -113,6 +113,7 @@ protected: wepMap_t wepMap; bool simulationMode; double cummulativePredictionTime = 0; + std::map>> scheduledTasks; protected: // Hook functions for AbstractScheduler functions below to allow customisation of common functionality @@ -121,6 +122,8 @@ protected: virtual void _notifyWEPUpdate(WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock) {} virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) {} + virtual double _getResourceEstimatedAvailableTime(resource::ComputingResource* resource); + // Convenience functions for passing a scheduling decision to the execution layer. bool executeSchedulingDecision(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) noexcept; diff --git a/include/wfr/scheduling/scheduler/heft/HEFTScheduler.h b/include/wfr/scheduling/scheduler/heft/HEFTScheduler.h index 143f1a2..ed1f918 100755 --- a/include/wfr/scheduling/scheduler/heft/HEFTScheduler.h +++ b/include/wfr/scheduling/scheduler/heft/HEFTScheduler.h @@ -56,8 +56,6 @@ protected: fastestExecutionTimes(new prediction::ExecutionTimeResult[resourceCount]) {} }; - //std::map> optimisticCostTable; - std::set readySet; void calcTaskRanks(wep::ActivityWepNode& node, std::vector& resources); diff --git a/include/wfr/scheduling/scheduler/lookahead/LookaheadScheduler.h b/include/wfr/scheduling/scheduler/lookahead/LookaheadScheduler.h new file mode 100755 index 0000000..5905339 --- /dev/null +++ b/include/wfr/scheduling/scheduler/lookahead/LookaheadScheduler.h @@ -0,0 +1,79 @@ + +#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 LookaheadScheduler: public scheduling::SchedulerBase { +protected: + resource::ResourceManager& resourceManager; + prediction::PredictionManager& predictionManager; + + struct TaskRankEntry { + double rank = 0; + bool finalized = 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: + LookaheadScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) + : resourceManager(resourceManager), predictionManager(predictionManager) {} + + ~LookaheadScheduler() {} + + 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/minmin/MinMinScheduler.h b/include/wfr/scheduling/scheduler/minmin/MinMinScheduler.h index 285cee1..7e9006d 100644 --- a/include/wfr/scheduling/scheduler/minmin/MinMinScheduler.h +++ b/include/wfr/scheduling/scheduler/minmin/MinMinScheduler.h @@ -25,13 +25,9 @@ protected: resource::ResourceManager& resourceManager; prediction::PredictionManager& predictionManager; - std::map>> scheduledTasks; - typedef std::map> ImplementationCacheInnerMap; std::map implementationCache; - double _getResourceEstimatedAvailableTime(resource::ComputingResource* resource); - typedef std::tuple ECTEntry; virtual std::pair _getNextTaskToSchedule(std::map& estimatedTaskCompletionTimes); diff --git a/include/wfr/scheduling/scheduler/sufferage/SufferageScheduler.h b/include/wfr/scheduling/scheduler/sufferage/SufferageScheduler.h index 04e4b65..8ddf26a 100644 --- a/include/wfr/scheduling/scheduler/sufferage/SufferageScheduler.h +++ b/include/wfr/scheduling/scheduler/sufferage/SufferageScheduler.h @@ -25,13 +25,9 @@ protected: resource::ResourceManager& resourceManager; prediction::PredictionManager& predictionManager; - std::map>> scheduledTasks; - typedef std::map> ImplementationCacheInnerMap; std::map implementationCache; - double _getResourceEstimatedAvailableTime(resource::ComputingResource* resource); - typedef std::tuple ECTEntry; std::pair _getNextTaskToSchedule(std::map& estimatedTaskCompletionTimes); diff --git a/src/scheduling/AbstractScheduler.cpp b/src/scheduling/AbstractScheduler.cpp index 10d89cc..1e118e3 100755 --- a/src/scheduling/AbstractScheduler.cpp +++ b/src/scheduling/AbstractScheduler.cpp @@ -67,6 +67,7 @@ void SchedulerBase::notifyWEPUpdate(wep::WorkflowExecutionPlan& wep, std::set sd, std::unique_lock& schedulerLock) noexcept { wepe.decisions.insert(sd); + scheduledTasks[sd->resource].push_back(sd); schedulerLock.unlock(); sd->activityInstance->setSchedulingDecision(sd); if (sd->activityInstance->activityType() == wfs::ActivityType::Atomic) { @@ -92,6 +93,13 @@ void SchedulerBase::notifySchedulingDecisionStatusUpdate(std::shared_ptr objLock(schedulerMutex); auto wepe = wepMap.find(sd->wep); if (wepe != wepMap.end()) { + auto& innerMap = scheduledTasks[sd->resource]; + for (auto i = innerMap.begin(); i != innerMap.end(); i++) { + if (*i == sd) { + innerMap.erase(i); + break; + } + } _notifySchedulingDecisionStatusUpdate(wepe->second, sd, objLock); // Notify enactment engine auto ee = wepe->second.engine; @@ -102,6 +110,36 @@ void SchedulerBase::notifySchedulingDecisionStatusUpdate(std::shared_ptrgetFreeSlotCount() > 0) { + return 0.0; + } else { + // Calculating the estimated available time for resources with more than 1 slot is not trivial + // Here we are employing a heuristic that partitions the scheduled tasks into chunks the size of the maximal slot count, + // takes the minimum time of each chunk and adds them together. + unsigned i = 0; + double eat = 0; + double minTime = -1; + auto& tasks = scheduledTasks[resource]; + for (auto t : tasks) { + i++; + if (minTime < 0) { + minTime = t->getPrediction().getExecutionTime().getValue(); + } else { + if (i > resource->getMaxSlotCount()) { + eat += minTime; + i = 0; + minTime = t->getPrediction().getExecutionTime().getValue(); + } else { + if (t->getPrediction().getExecutionTime().getValue() < minTime) { + minTime = t->getPrediction().getExecutionTime().getValue(); + } + } + } + } + return eat; + } +} } /* namespace scheduling */ } /* namespace wfr */ diff --git a/src/scheduling/SchedulerManager.cpp b/src/scheduling/SchedulerManager.cpp index 613f15d..4e50fc6 100755 --- a/src/scheduling/SchedulerManager.cpp +++ b/src/scheduling/SchedulerManager.cpp @@ -16,6 +16,7 @@ #include #include #include +#include namespace wfr { namespace scheduling { @@ -78,6 +79,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("lookahead", []()->AbstractScheduler* { + return new LookaheadScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance()); + })); schedulerFactories.insert(factoryMap_t::value_type("random", []()->AbstractScheduler* { return new RandomScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance()); })); diff --git a/src/scheduling/cmake.inc b/src/scheduling/cmake.inc index 90a4bf5..975fd60 100755 --- a/src/scheduling/cmake.inc +++ b/src/scheduling/cmake.inc @@ -10,4 +10,5 @@ set(wfr_cpp_files "${wfr_cpp_files}" src/scheduling/scheduler/maxmin/MaxMinScheduler.cpp src/scheduling/scheduler/sufferage/SufferageScheduler.cpp src/scheduling/scheduler/peft/PEFTScheduler.cpp + src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp ) diff --git a/src/scheduling/scheduler/heft/HEFTScheduler.cpp b/src/scheduling/scheduler/heft/HEFTScheduler.cpp index 46d2710..844d31a 100755 --- a/src/scheduling/scheduler/heft/HEFTScheduler.cpp +++ b/src/scheduling/scheduler/heft/HEFTScheduler.cpp @@ -29,6 +29,7 @@ namespace wfr { namespace scheduling { + void HEFTScheduler::calcTaskRanks(wep::ActivityWepNode& node, std::vector& resources) { /*std::cout << "calcTaskRanks: " << node.activity.baseActivity.baseTemplate.activityName << " (" << node.activity.instanceID << ")" << " [" << print::toString(node.activity.baseActivity.activityType()) << "]" @@ -156,11 +157,6 @@ void HEFTScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set< scheduleReadySet(wepe, resources, schedulerLock); } -void HEFTScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) { - //auto resources = resourceManager.getComputingResources(); - //scheduleReadySet(wepe, resources, schedulerLock); -} - void HEFTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector& resources, std::unique_lock& schedulerLock) { struct _schedulingPriorityQueueComparison { @@ -193,9 +189,12 @@ void HEFTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vectorfastestImplementations[ri] && res->getFreeSlotCount() > 0) { - if (!fastestPred.isValid() || entry->fastestExecutionTimes[ri].executionTime.getValue() < fastestPred.executionTime.getValue()) { + 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; @@ -222,6 +221,11 @@ void HEFTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector sd, std::unique_lock& schedulerLock) { + //auto resources = resourceManager.getComputingResources(); + //scheduleReadySet(wepe, resources, schedulerLock); +} + void HEFTScheduler::_unregisterWEP(WepMapEntry& wepe, std::unique_lock& schedulerLock) { /*std::cout << "Task Ranks: " << std::endl; diff --git a/src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp b/src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp new file mode 100755 index 0000000..17bc5fd --- /dev/null +++ b/src/scheduling/scheduler/lookahead/LookaheadScheduler.cpp @@ -0,0 +1,319 @@ + + +#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 LookaheadScheduler::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 (node.nodeType == wep::WepNodeType::Atomic) { + node.activity.setUserData(std::shared_ptr(entry)); + } + } + if (!entry->finalized) { + auto successors = node.getSuccessors(); + if (successors.size() == 0) { + // exit task + //std::cout << "\tNo successors => exit task: " << entry->averageExecutionTime << std::endl; + entry->rank = entry->averageExecutionTime; + entry->finalized = true; + } else { + bool succsFinalized = true; + double maxSuccRank = 0.0; + for (auto s : successors) { + auto& succ = s->as(); + /*std::cout << "\tSuccessor: " << succ.activity.baseActivity.baseTemplate.activityName << " (" << succ.activity.instanceID << ")" + << " [" << print::toString(succ.activity.baseActivity.activityType()) << "]" + << "[" << (int)succ.nodeType << "]" << std::endl;*/ + auto succEntry = (TaskRankEntry*)succ.getUserData().get(); + if (!succEntry || !succEntry->finalized) { + // recursively calculate ranks + //std::cout << "recursive call start" << std::endl; + calcTaskRanks(succ, resources); + //std::cout << "recursive call end" << std::endl; + if (!succEntry) { + succEntry = (TaskRankEntry*)succ.getUserData().get(); + } + succsFinalized = succsFinalized && succEntry->finalized; + } + // 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->rank; + if (succRank > maxSuccRank) { + maxSuccRank = succRank; + } + } + if (node.nodeType == wep::WepNodeType::Atomic) { + /*std::cout << "AverageExecTime: " << node.activity.baseActivity.baseTemplate.activityName << " (" << node.activity.instanceID << ")" + << " [" << print::toString(node.activity.baseActivity.activityType()) << "]" + << " [" << (int)node.nodeType << "]: " << entry->averageExecutionTime << std::endl;*/ + entry->rank = entry->averageExecutionTime + maxSuccRank; + } else { + entry->rank = maxSuccRank; + } + /*std::cout << "Rank: " << node.activity.baseActivity.baseTemplate.activityName << " (" << node.activity.instanceID << ")" + << " [" << print::toString(node.activity.baseActivity.activityType()) << "]" + << " [" << (int)node.nodeType << "]: " << entry->rank << std::endl;*/ + if (succsFinalized) { + if (node.nodeType == wep::WepNodeType::CompositeStart || node.nodeType == wep::WepNodeType::IterationStart) { + if (node.getLastKnownInstanceStatus() >= wfs::ActivityInstanceStatus::ReadyForExecution) { + entry->finalized = true; + } + } else { + entry->finalized = true; + } + /*if (entry->finalized) { + std::cout << "Finalizing: " << node.activity.baseActivity.baseTemplate.activityName << " (" << node.activity.instanceID << ")" + << " [" << print::toString(node.activity.baseActivity.activityType()) << "]" + << " [" << (int)node.nodeType << "]" << std::endl; + }*/ + } + } + } +} + + +void LookaheadScheduler::_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 LookaheadScheduler::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; + double minSuccEFT = -1; + auto succs = task->activity.getSuccessors(); + for (auto res : resources) { + if (entry->fastestImplementations[ri]) { + auto estimatedFinishTime = _getResourceEstimatedAvailableTime(res) + entry->fastestExecutionTimes[ri].executionTime.getValue(); + + auto maxSuccEFT = -1; + if (succs.size() > 0) { + // do lookahead + struct _succSchedulingQueueComparison { + bool operator()(wfs::ActivityInstance* e1, wfs::ActivityInstance* 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, _succSchedulingQueueComparison> succSchedulingQueue; + 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; + resource::ComputingResource* minEFTResource = nullptr; + for (auto res2 : resources) { + double succEFT = -1; + if (entry->fastestImplementations[ri2]) { + succEFT = resourceEATMap[res2] + succEntry->fastestExecutionTimes[ri2].executionTime.getValue(); + } + if (succEFT > 0 && (minEFT < 0 || succEFT < minEFT)) { + minEFT = succEFT; + minEFTResource = res2; + } + } + if (minEFT > 0) { + resourceEATMap[minEFTResource] = minEFT; + } + } + for (auto e : resourceEATMap) { + if (maxSuccEFT < 0 || e.second > maxSuccEFT) { + maxSuccEFT = e.second; + } + } + if (minSuccEFT < 0 || maxSuccEFT < minSuccEFT) { + minSuccEFT = maxSuccEFT; + earliestEstimatedFinishTime = estimatedFinishTime; + fastestImpl = entry->fastestImplementations[ri]; + fastestPred = entry->fastestExecutionTimes[ri]; + fastestResource = res; + } + // no successors + } else 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 LookaheadScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) { + //auto resources = resourceManager.getComputingResources(); + //scheduleReadySet(wepe, resources, schedulerLock); +} + + +void LookaheadScheduler::_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/minmin/MinMinScheduler.cpp b/src/scheduling/scheduler/minmin/MinMinScheduler.cpp index 7e1ac19..63c253b 100644 --- a/src/scheduling/scheduler/minmin/MinMinScheduler.cpp +++ b/src/scheduling/scheduler/minmin/MinMinScheduler.cpp @@ -24,39 +24,6 @@ namespace scheduling { - -double MinMinScheduler::_getResourceEstimatedAvailableTime(resource::ComputingResource* resource) { - if (resource->getFreeSlotCount() > 0) { - return 0.0; - } else { - // Calculating the estimated available time for resources with more than 1 slot is not trivial - // Here we are employing a heuristic that partitions the scheduled tasks into chunks the size of the maximal slot count, - // takes the minimum time of each chunk and adds them together. - unsigned i = 0; - double eat = 0; - double minTime = -1; - auto& tasks = scheduledTasks[resource]; - for (auto t : tasks) { - i++; - if (minTime < 0) { - minTime = t->getPrediction().getExecutionTime().getValue(); - } else { - if (i > resource->getMaxSlotCount()) { - eat += minTime; - i = 0; - minTime = t->getPrediction().getExecutionTime().getValue(); - } else { - if (t->getPrediction().getExecutionTime().getValue() < minTime) { - minTime = t->getPrediction().getExecutionTime().getValue(); - } - } - } - } - return eat; - } -} - - std::pair MinMinScheduler::_getNextTaskToSchedule(std::map& estimatedTaskCompletionTimes) { wep::AtomicWepNode* selectedNode = nullptr; ECTEntry selectedEntry; @@ -143,22 +110,13 @@ void MinMinScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::se auto sd = std::make_shared(&selectedTask.first->activity, std::get<1>(selectedTask.second), wepe.wep, reservation); sd->setPrediction(std::get<3>(selectedTask.second)); selectedTask.first->activity.setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled); - scheduledTasks[sd->resource].push_back(sd); executeSchedulingDecision(wepe, sd, schedulerLock); readyToExecute.erase(selectedTask.first); } } } -void MinMinScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) { - auto& innerMap = scheduledTasks[sd->resource]; - for (auto i = innerMap.begin(); i != innerMap.end(); i++) { - if (*i == sd) { - innerMap.erase(i); - break; - } - } -} +void MinMinScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) {} } /* namespace scheduling */ } /* namespace wfr */ diff --git a/src/scheduling/scheduler/peft/PEFTScheduler.cpp b/src/scheduling/scheduler/peft/PEFTScheduler.cpp index 425ee0d..eda9125 100755 --- a/src/scheduling/scheduler/peft/PEFTScheduler.cpp +++ b/src/scheduling/scheduler/peft/PEFTScheduler.cpp @@ -29,6 +29,7 @@ namespace wfr { namespace scheduling { + void PEFTScheduler::calcOptimisticCosts(wep::ActivityWepNode& node, std::vector& resources) { /*std::cout << "calcOptimisticCosts: " << node.activity.baseActivity.baseTemplate.activityName << " (" << node.activity.instanceID << ")" << " [" << print::toString(node.activity.baseActivity.activityType()) << "]" @@ -157,11 +158,6 @@ void PEFTScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set< scheduleReadySet(wepe, resources, schedulerLock); } -void PEFTScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) { - //auto resources = resourceManager.getComputingResources(); - //scheduleReadySet(wepe, resources, schedulerLock); -} - void PEFTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector& resources, std::unique_lock& schedulerLock) { struct _schedulingPriorityQueueComparison { @@ -194,9 +190,12 @@ void PEFTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vectorfastestImplementations[ri] && res->getFreeSlotCount() > 0) { - if (!fastestPred.isValid() || entry->fastestExecutionTimes[ri].executionTime.getValue() < fastestPred.executionTime.getValue()) { + 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; @@ -217,6 +216,11 @@ void PEFTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector sd, std::unique_lock& schedulerLock) { + //auto resources = resourceManager.getComputingResources(); + //scheduleReadySet(wepe, resources, schedulerLock); +} + } /* namespace scheduler */ diff --git a/src/scheduling/scheduler/sufferage/SufferageScheduler.cpp b/src/scheduling/scheduler/sufferage/SufferageScheduler.cpp index 46b6f28..97c5763 100644 --- a/src/scheduling/scheduler/sufferage/SufferageScheduler.cpp +++ b/src/scheduling/scheduler/sufferage/SufferageScheduler.cpp @@ -24,39 +24,6 @@ namespace scheduling { - -double SufferageScheduler::_getResourceEstimatedAvailableTime(resource::ComputingResource* resource) { - if (resource->getFreeSlotCount() > 0) { - return 0.0; - } else { - // Calculating the estimated available time for resources with more than 1 slot is not trivial - // Here we are employing a heuristic that partitions the scheduled tasks into chunks the size of the maximal slot count, - // takes the minimum time of each chunk and adds them together. - unsigned i = 0; - double eat = 0; - double minTime = -1; - auto& tasks = scheduledTasks[resource]; - for (auto t : tasks) { - i++; - if (minTime < 0) { - minTime = t->getPrediction().getExecutionTime().getValue(); - } else { - if (i > resource->getMaxSlotCount()) { - eat += minTime; - i = 0; - minTime = t->getPrediction().getExecutionTime().getValue(); - } else { - if (t->getPrediction().getExecutionTime().getValue() < minTime) { - minTime = t->getPrediction().getExecutionTime().getValue(); - } - } - } - } - return eat; - } -} - - std::pair SufferageScheduler::_getNextTaskToSchedule(std::map& estimatedTaskCompletionTimes) { wep::AtomicWepNode* selectedNode = nullptr; ECTEntry selectedEntry; @@ -155,22 +122,13 @@ void SufferageScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std: auto sd = std::make_shared(&selectedTask.first->activity, std::get<1>(selectedTask.second), wepe.wep, reservation); sd->setPrediction(std::get<3>(selectedTask.second)); selectedTask.first->activity.setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled); - scheduledTasks[sd->resource].push_back(sd); executeSchedulingDecision(wepe, sd, schedulerLock); readyToExecute.erase(selectedTask.first); } } } -void SufferageScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) { - auto& innerMap = scheduledTasks[sd->resource]; - for (auto i = innerMap.begin(); i != innerMap.end(); i++) { - if (*i == sd) { - innerMap.erase(i); - break; - } - } -} +void SufferageScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) {} } /* namespace scheduling */ } /* namespace wfr */