WorkflowOnManycores/WFR-lib
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Improved HEFT scheduler and fixed some bugs.

Browse Source
This commit is contained in:
Matthias
2017-07-25 17:13:22 +02:00
parent 3d4bb80dae
commit b93d8565a0
5 changed files with 222 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/wfs/ActivityInstance.cpp
+2 -2
View File
@@ -143,12 +143,12 @@ void ActivityInstance::setErrorState(std::exception_ptr exception) noexcept {
void ActivityInstance::executeSimMode() noexcept {
if (WorkflowRuntime::getGlobalConfig<bool>("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));
src/scheduling/SchedulerManager.cpp
+1
View File
@@ -9,6 +9,7 @@
#include <wfr/common/resource/ResourceManager.h>
#include <wfr/scheduling/scheduler/dummy/DummyScheduler.h>
#include <wfr/scheduling/scheduler/heft/HEFTScheduler.h>
#include <wfr/scheduling/scheduler/heft/HEFTScheduler.h>
#include <wfr/scheduling/scheduler/random/RandomScheduler.h>
#include <wfr/scheduling/scheduler/justintime/JustInTimeScheduler.h>
#include <wfr/scheduling/scheduler/minmin/MinMinScheduler.h>
src/scheduling/scheduler/heft/HEFTScheduler.cpp
+197 -165
View File
@@ -2,214 +2,246 @@
#include <wfr/prediction/PredictionManager.h>
#include <wfr/common/resource/ResourceManager.h>
#include <wfr/scheduling/scheduler/heft/HEFTScheduler.h>
#include <wfr/common/wep/WorkflowExecutionPlan.h>
#include <wfr/common/wep/AbstractWepNode.h>
#include <wfr/common/wep/activities/AtomicWepNode.h>
#include <wfr/common/wep/ActivityWepNode.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/ActivityInstance.h>
#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
#include <wfr/common/wfs/AtomicActivityImplementation.h>
#include <wfr/common/wfs/templates/AtomicActivityTemplate.h>
#include <wfr/enactment/EnactmentEngine.h>
#include <wfr/common/scheduling/SchedulingDecision.h>
#include <wfr/execution/ExecutionManager.h>
#include <wfr/wep/utils/WepSearch.h>
#include <wfr/common/utils/print.h>
#include <iostream>
#include <set>
#include <boost/format.hpp>
#include <wfr/scheduling/scheduler/heft/HEFTScheduler.h>
#include <queue>
#include <memory>
#include <cstring>
namespace wfr {
namespace scheduling {
void HEFTScheduler::calcTaskRanks(wep::ActivityWepNode& node, std::vector<resource::ComputingResource*>& resources) {
/*std::cout << "calcTaskRanks: " << node.activity.baseActivity.baseTemplate.activityName << " (" << node.activity.instanceID << ")"
<< " [" << print::toString(node.activity.baseActivity.activityType()) << "]"
<< " [" << (int)node.nodeType << "]" << std::endl;*/
NumericalMeasurement HEFTScheduler::getAverageExecutionCost(wfs::AtomicActivityInstance& ai, InstanceUserData* instanceData, const std::vector<resource::ComputingResourceType*>& 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<wfs::AtomicActivityTemplate>();
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<resource::ComputingResourceType*>& resourceTypes, const std::vector<resource::ComputingResource*>& 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<InstanceUserData>());
succData = (InstanceUserData*)succ->getUserData().get();
}
auto succRank = getRank(succ->as<wfs::AtomicActivityInstance>(), 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<wfs::AtomicActivityTemplate>().atomicActivityType->getImplementations()) {
auto pred = predictionManager.predictExecutionTime(node.activity, *impl, res->resourceType, 1);
if (pred.isValid()) {
if (!fastestPred.isValid() || pred < fastestPred) {
fastestPred = pred;
fastestImpl = impl;
}
}
}
entry->fastestImplementations[ri] = fastestImpl;
entry->fastestExecutionTimes[ri] = fastestPred;
if (fastestPred.isValid()) {
sumExecutionTimeCount++;
sumExecutionTimes += fastestPred.executionTime.getValue();
}
ri++;
}
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<void>(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<wep::ActivityWepNode>();
/*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<resource::ComputingResourceType*>& resourceTypes, const std::vector<resource::ComputingResource*>& 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<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& 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<wep::AtomicWepNode>().activity.as<wfs::AtomicActivityInstance>();
auto instanceDataPtr = ai.getUserData();
InstanceUserData* instanceData;
if (instanceDataPtr) {
instanceData = (InstanceUserData*)instanceDataPtr.get();
} else {
ai.setUserData(std::make_shared<InstanceUserData>());
instanceData = (InstanceUserData*)ai.getUserData().get();
}
calcTaskRanks(wepe.wep->getStartNode()->as<wep::ActivityWepNode>(), 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<SchedulingDecision> sd = std::make_shared<SchedulingDecision>(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<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) {
//auto resources = resourceManager.getComputingResources();
//scheduleReadySet(wepe, resources, schedulerLock);
}
void HEFTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector<resource::ComputingResource*>& resources, std::unique_lock<std::mutex>& schedulerLock) {
struct _schedulingPriorityQueueComparison {
bool operator()(wep::AtomicWepNode* e1, wep::AtomicWepNode* e2) {
auto entry1 = (TaskRankEntry*)e1->getUserData().get();
auto entry2 = (TaskRankEntry*)e2->getUserData().get();
if (entry1 && !entry2) {
return false;
} else if (!entry1 && entry2) {
return true;
} else if (entry1 && entry2) {
return entry1->rank < entry2->rank;
} else {
return false;
}
}
};
std::priority_queue<wep::AtomicWepNode*, std::vector<wep::AtomicWepNode*>, _schedulingPriorityQueueComparison> schedulingPriorityQueue;
// Put elements from ready set into priority queue thereby sorting them by rank
for (auto n : readySet) {
schedulingPriorityQueue.push(n);
}
while (!schedulingPriorityQueue.empty()) {
auto task = schedulingPriorityQueue.top();
schedulingPriorityQueue.pop();
auto entry = (TaskRankEntry*)task->getUserData().get();
int ri = 0;
prediction::ExecutionTimeResult fastestPred;
wfs::AtomicActivityImplementation* fastestImpl = nullptr;
resource::ComputingResource* fastestResource = nullptr;
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<SchedulingDecision> sd = std::make_shared<SchedulingDecision>(&task->activity, fastestImpl, wepe.wep, reservation);
sd->setPrediction(fastestPred);
task->activity.setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled);
executeSchedulingDecision(wepe, sd, schedulerLock);
readySet.erase(task);
}
}
}
}
void HEFTScheduler::_unregisterWEP(WepMapEntry& wepe, std::unique_lock<std::mutex>& schedulerLock) {
/*std::cout << "Task Ranks: " << std::endl;
std::set<wep::AbstractWepNode*> visitedNodes;
auto visitFunc = [&](wep::AbstractWepNode* node) -> bool {
if (visitedNodes.find(node) == visitedNodes.end()) {
auto entry = (TaskRankEntry*)node->getUserData().get();
if (entry) {
std::cout << "\t" << node->as<wep::ActivityWepNode>().activity.baseActivity.baseTemplate.activityName
<< "[" << node->as<wep::ActivityWepNode>().activity.instanceID << "]: " << entry->rank << std::endl;
} else {
std::cout << "\t" << node->as<wep::ActivityWepNode>().activity.baseActivity.baseTemplate.activityName
<< "[" << node->as<wep::ActivityWepNode>().activity.instanceID << "]: No Entry" << std::endl;
}
visitedNodes.insert(node);
}
return true;
};
wfr::utils::wep::wepBreadthFirstSearch(wepe.wep->getStartNode(), visitFunc);*/
}
} /* namespace scheduler */
} /* namespace wfr */