From b93d8565a0f66c4711dbdce64e79f59d037d996e Mon Sep 17 00:00:00 2001 From: Matthias Date: Tue, 25 Jul 2017 17:13:22 +0200 Subject: [PATCH] Improved HEFT scheduler and fixed some bugs. --- .../scheduling/scheduler/heft/HEFTScheduler.h | 65 +--- include/wfr/wep/utils/WepSearch.h | 1 + src/common/wfs/ActivityInstance.cpp | 4 +- src/scheduling/SchedulerManager.cpp | 1 + .../scheduler/heft/HEFTScheduler.cpp | 362 ++++++++++-------- 5 files changed, 222 insertions(+), 211 deletions(-) diff --git a/include/wfr/scheduling/scheduler/heft/HEFTScheduler.h b/include/wfr/scheduling/scheduler/heft/HEFTScheduler.h index 3a588c2..143f1a2 100755 --- a/include/wfr/scheduling/scheduler/heft/HEFTScheduler.h +++ b/include/wfr/scheduling/scheduler/heft/HEFTScheduler.h @@ -21,9 +21,9 @@ namespace wfr { namespace wfs { class AtomicActivityInstance; } - namespace wep { class AtomicWepNode; +class ActivityWepNode; class AbstractWepNode; } namespace resource { @@ -39,63 +39,40 @@ namespace wfr { namespace scheduling { - - class HEFTScheduler: public scheduling::SchedulerBase { protected: - struct InstanceUserData { - bool isComplete = false; - bool isScheduled = false; - NumericalMeasurement rank; - NumericalMeasurement averageComputingCost; - // resourceName => (fastest implementation, prediction); - std::map> predictions; - }; - struct SchedulingQueueEntry { - InstanceUserData* instanceData = nullptr; - wfs::AtomicActivityInstance* instance = nullptr; - wep::AbstractWepNode* node = nullptr; - }; - struct _SchedulingQueueEntryComparison { - bool operator()(const SchedulingQueueEntry& e1, const SchedulingQueueEntry& e2) { - if (e1.instanceData && !e2.instanceData) { - return false; - } else if (!e1.instanceData && e2.instanceData) { - return true; - } else if (e1.instanceData && e2.instanceData) { - return e1.instanceData->rank.getValue() < e2.instanceData->rank.getValue(); - } else { - return false; - } - } - }; - std::priority_queue, _SchedulingQueueEntryComparison> schedulingPriorityQueue; - - - resource::ResourceManager& resourceManager; prediction::PredictionManager& predictionManager; - std::vector debugLines; - wfr_size_t debugIteration = 0; + 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]) {} + }; - NumericalMeasurement getRank(wfs::AtomicActivityInstance& ai, InstanceUserData* instanceData, const std::vector& resourceTypes, const std::vector& resources); - NumericalMeasurement getAverageExecutionCost(wfs::AtomicActivityInstance& ai, InstanceUserData* instanceData, const std::vector& resources); - NumericalMeasurement getMaxSuccValue(wfs::AtomicActivityInstance& ai, InstanceUserData* instanceData, const std::vector& resourceTypes, const std::vector& resources); + //std::map> optimisticCostTable; + + std::set readySet; + + void calcTaskRanks(wep::ActivityWepNode& node, std::vector& resources); + + void scheduleReadySet(WepMapEntry& wepe, std::vector& resources, std::unique_lock& schedulerLock); public: HEFTScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) : resourceManager(resourceManager), predictionManager(predictionManager) {} - ~HEFTScheduler() { - /*for (auto s : debugLines) { - std::cout << s << std::endl; - }*/ - } + ~HEFTScheduler() {} - virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock); + 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 */ diff --git a/include/wfr/wep/utils/WepSearch.h b/include/wfr/wep/utils/WepSearch.h index 736c728..3f67ac0 100755 --- a/include/wfr/wep/utils/WepSearch.h +++ b/include/wfr/wep/utils/WepSearch.h @@ -3,6 +3,7 @@ #include +#include namespace wfr { namespace wep { diff --git a/src/common/wfs/ActivityInstance.cpp b/src/common/wfs/ActivityInstance.cpp index 9398830..0899e65 100755 --- a/src/common/wfs/ActivityInstance.cpp +++ b/src/common/wfs/ActivityInstance.cpp @@ -143,12 +143,12 @@ void ActivityInstance::setErrorState(std::exception_ptr exception) noexcept { void ActivityInstance::executeSimMode() noexcept { if (WorkflowRuntime::getGlobalConfig("sim-mode")) { if (schedulingDecision) { - std::cout << "Simulating execution ... " << std::endl; + //std::cout << "Simulating execution ... " << std::endl; auto prediction = schedulingDecision->getPrediction(); if (prediction) { auto& predTime = prediction.getExecutionTime(); double time = WorkflowRuntime::getRandomNumber(predTime.getValue() - predTime.getDeviation(), predTime.getValue() + predTime.getDeviation()); - std::cout << "Sleeping for " << time << " seconds." << std::endl; + //std::cout << "Sleeping for " << time << " seconds." << std::endl; std::this_thread::sleep_for(std::chrono::milliseconds((long int)(time * 1000))); /*std::cout << "Doing stuff for " << time << " seconds." << std::endl; std::chrono::system_clock::time_point endtime = std::chrono::system_clock::now() + std::chrono::milliseconds((long int)(time * 1000)); diff --git a/src/scheduling/SchedulerManager.cpp b/src/scheduling/SchedulerManager.cpp index 5408f78..613f15d 100755 --- a/src/scheduling/SchedulerManager.cpp +++ b/src/scheduling/SchedulerManager.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include diff --git a/src/scheduling/scheduler/heft/HEFTScheduler.cpp b/src/scheduling/scheduler/heft/HEFTScheduler.cpp index 4fa2b4a..46d2710 100755 --- a/src/scheduling/scheduler/heft/HEFTScheduler.cpp +++ b/src/scheduling/scheduler/heft/HEFTScheduler.cpp @@ -2,214 +2,246 @@ #include #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 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()) << "]" + << " [" << (int)node.nodeType << "]" << std::endl;*/ -NumericalMeasurement HEFTScheduler::getAverageExecutionCost(wfs::AtomicActivityInstance& ai, InstanceUserData* instanceData, const std::vector& resources) { - // Iterate over resources and calculate average execution time - NumericalMeasurement average; - wfr_size_t _count = 0; - for (auto dev : resources) { - NumericalMeasurement resTime; - // Get fastest implementation (better said its execution time) for given resource - auto& aat = ai.baseActivity.baseTemplate.as(); - if (aat.atomicActivityType->getImplementationCount() > 0) { - for (auto i : aat.atomicActivityType->getImplementations()) { - auto execPred = predictionManager.predictExecutionTime(ai, *i, *dev); - if (execPred.isValid() && (!resTime.isValid() || execPred.getExecutionTime() < resTime)) { - resTime = execPred.getExecutionTime(); - instanceData->predictions[dev] = std::make_tuple(i, execPred); - } - } - if (resTime.isValid()) { - _count++; - average += resTime; - } - } - } - if (_count > 0) { - average /= _count; - } - return average; -} - - -NumericalMeasurement HEFTScheduler::getMaxSuccValue(wfs::AtomicActivityInstance& ai, InstanceUserData* instanceData, const std::vector& resourceTypes, const std::vector& resources) { - NumericalMeasurement retval(0, 0, true); - - // iterate over successors - for(wfr_size_t i = 0; i < ai.getSuccessorCount(); i++) { - auto succ = ai.getSuccessor(i); - - auto succDataPtr = succ->getUserData(); - InstanceUserData* succData; - if (succDataPtr) { - succData = (InstanceUserData*)succDataPtr.get(); - } else { - succ->setUserData(std::make_shared()); - succData = (InstanceUserData*)succ->getUserData().get(); - } - auto succRank = getRank(succ->as(), instanceData, resourceTypes, resources); - if (succRank.isValid()) { - - // Calculate average communication costs - NumericalMeasurement average(0, 0, true); - wfr_size_t _count = 0; - for (auto& dev1 : resources) { - for (auto& dev2 : resources) { - if (dev1 != dev2) { - auto commTime = predictionManager.predictCommunicationTime(ai, *dev1, *succ, *dev2); - if (commTime.isValid) { - _count++; - average += commTime.communicationTime; + 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++; } - if (_count > 0) { - average /= _count; - } - NumericalMeasurement tmp = succData->rank + average; - if (tmp.isValid() && (!retval.isValid() || retval < tmp)) { - retval = tmp; - } + entry->averageExecutionTime = sumExecutionTimes / (double)sumExecutionTimeCount; + } + node.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 { - throw exception::wfr_fatal_error("Invalid successor rank!"); + 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; + }*/ + } } } - return retval; -} - - -NumericalMeasurement HEFTScheduler::getRank(wfs::AtomicActivityInstance& ai, InstanceUserData* instanceData, const std::vector& resourceTypes, const std::vector& resources) { - - if (!instanceData->isComplete) { - if (!instanceData->averageComputingCost.isValid()) { - instanceData->averageComputingCost = getAverageExecutionCost(ai, instanceData, resourceTypes); - } - NumericalMeasurement maxSuccRank = getMaxSuccValue(ai, instanceData, resourceTypes, resources); - if (instanceData->averageComputingCost.isValid() && maxSuccRank.isValid()) { - instanceData->rank = instanceData->averageComputingCost + maxSuccRank; - instanceData->isComplete = true; - } - } - - return instanceData->rank; } void HEFTScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock) { - auto resourceTypes = resourceManager.getComputingResourceTypes(); auto resources = resourceManager.getComputingResources(); - // Visitor function for breadth first search. - auto visitFunc = [&](wep::AbstractWepNode* node) -> bool { - if (node->nodeType == wep::WepNodeType::Atomic) { - auto& ai = node->as().activity.as(); - auto instanceDataPtr = ai.getUserData(); - InstanceUserData* instanceData; - if (instanceDataPtr) { - instanceData = (InstanceUserData*)instanceDataPtr.get(); - } else { - ai.setUserData(std::make_shared()); - instanceData = (InstanceUserData*)ai.getUserData().get(); - } + calcTaskRanks(wepe.wep->getStartNode()->as(), resources); - getRank(ai, instanceData, resourceTypes, resources); - if (instanceData->isComplete && !instanceData->isScheduled && ai.getInstanceStatus() == wfs::ActivityInstanceStatus::ReadyForExecution) { - SchedulingQueueEntry tmp; - tmp.instance = &ai; - tmp.instanceData = instanceData; - tmp.node = node; - schedulingPriorityQueue.push(tmp); - instanceData->isScheduled = true; - } - } - return true; - }; - // Visit all nodes of the WEP and calculate the rank for each atomic activity. - wfr::utils::wep::wepBreadthFirstReverseSearch(wepe.wep->getEndNode(), visitFunc); + readySet.insert(readyToExecute.begin(), readyToExecute.end()); + scheduleReadySet(wepe, resources, schedulerLock); +} - //std::cout << "HEFT Iteration" << std::endl << "======================" << std::endl; - // Go through scheduling queue and schedule activities; - while (!schedulingPriorityQueue.empty()) { - auto e = schedulingPriorityQueue.top(); - - std::cout << e.instance->baseActivity.baseTemplate.activityName << "[" << e.instance->instanceID << "]: " - << e.instanceData->rank.getValue() << std::endl; - - wfs::AtomicActivityImplementation* fastestImpl = nullptr; - resource::ComputingResource* fastestResource = nullptr; - prediction::ExecutionTimeResult prediction; - wfr_size_t busyResourceCount = 0; - for (auto res : resources) { - //std::cout << "\tResource " << res->resourceIdentifier << ": freeSlots = " << res->getFreeSlotCount() << std::endl; - if (res->getFreeSlotCount() > 0) { - auto& pe = e.instanceData->predictions[&res->resourceType]; - wfs::AtomicActivityImplementation* impl = std::get<0>(pe); - if (impl) { - auto& pred = std::get<1>(pe); - if (pred.isValid() && (!prediction.isValid() || pred.getExecutionTime() < prediction.getExecutionTime())) { - fastestImpl = std::get<0>(pe); - fastestResource = res; - prediction = pred; - } - } - } else { - busyResourceCount++; - } - } - if (fastestImpl) { - //std::cout << "\tFastest Resource: " << fastestResource->resourceIdentifier << ": prediction = " << prediction.getExecutionTime().getValue() << std::endl; - auto reservation = resourceManager.reserveComputingResource(fastestResource, 1, true); - if (reservation) { - std::shared_ptr sd = std::make_shared(e.instance, fastestImpl, wepe.wep, reservation); - sd->setPrediction(prediction); - e.instance->setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled); - executeSchedulingDecision(wepe, sd, schedulerLock); - //schedulingQueue.pop(); - schedulingPriorityQueue.pop(); - - /*std::string d1 = (boost::format("%4%: %1%: Rank %2%, Prediction %3%") - % e.instance->baseActivity.baseTemplate.activityName - % e.instanceData->rank.getValue() - % prediction.executionTime.getValue() - % debugIteration).str(); - debugLines.push_back(d1);*/ - - } - } else if (busyResourceCount == resources.size()) { - // If all resources are busy we can break the loop - break; - } - } - debugIteration++; +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 { + 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; + for (auto res : resources) { + if (entry->fastestImplementations[ri] && res->getFreeSlotCount() > 0) { + if (!fastestPred.isValid() || entry->fastestExecutionTimes[ri].executionTime.getValue() < fastestPred.executionTime.getValue()) { + 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 HEFTScheduler::_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 */