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

.

Browse Source
This commit is contained in:
Matthias
2017-07-24 16:27:50 +02:00
parent ec80b339e2
commit 3d4bb80dae
46 changed files with 1388 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/resource/Resource.cpp
+4 -2
View File
@@ -111,8 +111,8 @@ std::string getResourceCapabilityTypeString(const ResourceCapabilityType& type)
}
bool ComputingResource::reserveSlots(const wfr_size_t& slots) noexcept {
if (slots <= getFreeSlotCount()) {
bool ComputingResource::reserveSlots(const wfr_size_t& slots, bool requireFreeSlots) noexcept {
if (!requireFreeSlots || slots <= getFreeSlotCount()) {
reservedSlots += slots;
return true;
} else {
@@ -151,5 +151,7 @@ bool ComputingResource::releaseBusySlots(const wfr_size_t& slots) noexcept {
} /* namespace resource */
} /* namespace wfr */
src/common/resource/ResourceManager.cpp
+39 -8
View File
@@ -15,6 +15,11 @@
#include <boost/format.hpp>
#include <wfr/common/resource/Resource.h>
#include <wfr/common/resource/ResourceReservation.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/ActivityTemplate.h>
#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
#include <wfr/common/wfs/templates/AtomicActivityTemplate.h>
#include <wfr/common/wfs/AtomicActivityImplementation.h>
namespace wfr {
namespace resource {
@@ -22,6 +27,10 @@ namespace resource {
ResourceManager* ResourceManager::singleton = nullptr;
std::atomic<wfr_id_t> ResourceManager::_nextResourceId = ATOMIC_VAR_INIT(0);
void ResourceManager::init(int argc, char** argv) {
std::string resourceConfigFile;
bool cpuAutoConfig = true;
@@ -80,7 +89,7 @@ ResourceManager::ResourceManager(std::string resourceConfigFile, bool cpuAutoCon
ComputingResourceType* resType = new ComputingResourceType("CPU", ComputingDeviceClass::CPU, threadCount, {{ResourceCapabilityType::GenericCPU, 0, 0}});
std::string tmp(resType->resourceTypeName);
computingResourceTypes.insert({tmp, resType});
ComputingResource* dev = new ComputingResource("cpu0", *resType);
ComputingResource* dev = new ComputingResource("cpu0", getNextResourceId(), *resType);
computingResources.insert({dev->resourceIdentifier, dev});
}
if (!resourceConfigFile.empty()) {
@@ -220,7 +229,7 @@ void ResourceManager::readConfigFile(std::string resourceConfigFile) {
throw wfr::exception::wfr_fatal_error((boost::format("Resource config file: unknown device type \"%1%\".") % devTypeName).str());
}
ComputingResource* device = new ComputingResource(devName, *(*devType).second);
ComputingResource* device = new ComputingResource(devName, getNextResourceId(), *(*devType).second);
computingResources.insert({device->resourceIdentifier, device});
}
@@ -268,13 +277,11 @@ std::vector<ComputingResource*> ResourceManager::filterComputingResources(std::v
}
ComputingResourceReservation ResourceManager::reserveComputingResource(ComputingResource* device, const wfr_size_t& slots) noexcept {
if (device->getFreeSlotCount() >= slots) {
if (device->reserveSlots(slots)) {
return ComputingResourceReservation(device, slots);
}
std::shared_ptr<ComputingResourceReservation> ResourceManager::reserveComputingResource(ComputingResource* device, const wfr_size_t& slots, bool requireFreeSlots) noexcept {
if (device->reserveSlots(slots, requireFreeSlots)) {
return std::make_shared<ComputingResourceReservation>(device, slots);
}
return ComputingResourceReservation();
return std::shared_ptr<ComputingResourceReservation>();
}
@@ -288,6 +295,29 @@ std::vector<ComputingResourceType*> ResourceManager::getComputingResourceTypes()
}
std::vector<ComputingResourceType*> ResourceManager::getCompatibleComputingResourceTypes(wfs::AtomicActivityInstance* activity) const noexcept {
std::vector<ComputingResourceType*> retval;
for (auto t : computingResourceTypes) {
auto implementations = activity->baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>().atomicActivityType->getImplementations();
for (auto i : implementations) {
bool breakImplementationLoop = false;
for (auto c : i->getRequiredCapabilities()) {
if (t.second->hasCapability(c)) {
retval.push_back(t.second);
breakImplementationLoop = true;
break;
}
}
if (breakImplementationLoop) {
break;
}
}
}
return retval;
}
ComputingResourceType* ResourceManager::getComputingResourceType(std::string name) const noexcept {
return computingResourceTypes.at(name);
}
@@ -363,6 +393,7 @@ void ResourceManager::printResourceTable(std::ostream& out) const noexcept {
out << "======= Computing Resource Table ========" << std::endl;
for (auto& res : computingResources) {
out << "Resource: \"" << res.second->resourceIdentifier << "\":" << std::endl;
out << "\tResourceId: " << res.second->resourceId << std::endl;
out << "\tResourceType: \"" << res.second->resourceType.resourceTypeName << "\"" << std::endl;
}
out << "===== Computing Resource Table End ======" << std::endl;
src/common/wfs/ActivityInstance.cpp
+1
View File
@@ -136,6 +136,7 @@ void ActivityInstance::setInstanceStatus(const ActivityInstanceStatus& status) {
void ActivityInstance::setErrorState(std::exception_ptr exception) noexcept {
errorCause = exception;
setInstanceStatus(ActivityInstanceStatus::Error);
std::rethrow_exception(exception);
}
src/common/wfs/instances/ConditionalActivityInstance.cpp
+16 -1
View File
@@ -2,6 +2,7 @@
#include <wfr/common/wfs/instances/ConditionalActivityInstance.h>
#include <wfr/common/wfs/templates/ConditionalActivityTemplate.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/Workflow.h>
namespace wfr {
@@ -20,7 +21,9 @@ bool ConditionalActivityInstance::executePreExecutionHooks(bool earlyEvaluation)
allInputsAvailable = true;
}
if (templ.branchFuncHook != nullptr) {
retval = templ.branchFuncHook(*this) || retval;
if (earlyEvaluation || allInputsAvailable) {
retval = templ.branchFuncHook(*this) || retval;
}
} else {
setErrorState(std::make_exception_ptr(wfr::exception::wfr_fatal_error("BranchFuncHook is null.")));
}
@@ -41,6 +44,18 @@ bool ConditionalActivityInstance::executePostExecutionHooks(bool earlyEvaluation
return retval;
}
void ConditionalActivityInstance::notifyChildHasFinished(ActivityInstance& ai) noexcept {
finishedChildrenCount++;
if (finishedChildrenCount >= baseActivity.getChildrenCount(activeExecutionBranchID)) { // see if all children are finished
setInstanceStatus(ActivityInstanceStatus::Finished);
if (parentActivity != nullptr) { // propagate to parent
parentActivity->as<CompositeActivityInstance>().notifyChildHasFinished(*this);
} else { // if there is no parent then it's the toplevel activity
baseActivity.workflow.setFinishFlag(true);
}
}
}
} /* namespace wfs */
} /* namespace wfr */
src/common/wfs/instances/ContainerActivityInstance.cpp
+3 -1
View File
@@ -15,7 +15,9 @@ bool ContainerActivityInstance::executePreExecutionHooks(bool earlyEvaluation) {
}
auto& templ = baseActivity.baseTemplate.as<ContainerActivityTemplate>();
if (templ.inputFuncHook != nullptr) {
retval = templ.inputFuncHook(*this);
if (earlyEvaluation || allInputsAvailable) {
retval = templ.inputFuncHook(*this);
}
} else {
allInputsAvailable = true;
setInstanceStatus(ActivityInstanceStatus::ReadyForExecution);
src/common/wfs/instances/ForLoopActivityInstance.cpp
+3 -1
View File
@@ -28,7 +28,9 @@ bool ForLoopActivityInstance::executePreExecutionHooks(bool earlyEvaluation) {
allInputsAvailable = true;
}
if (templ.counterFuncHook != nullptr) {
retval = templ.counterFuncHook(*this) || retval;
if (earlyEvaluation || allInputsAvailable) {
retval = templ.counterFuncHook(*this) || retval;
}
} else {
setErrorState(std::make_exception_ptr(wfr::exception::wfr_fatal_error("IteratorFuncHook is null.")));
}
src/common/wfs/instances/WhileLoopActivityInstance.cpp
+1 -1
View File
@@ -21,7 +21,7 @@ bool WhileLoopActivityInstance::executePreExecutionHooks(bool earlyEvaluation) {
allInputsAvailable = true;
}
if (templ.conditionFuncHook != nullptr) {
if (iterationCount <= 0 || finishedIterations[iterationCount-1]) {
if ((earlyEvaluation || allInputsAvailable) && (iterationCount <= 0 || finishedIterations[iterationCount-1])) {
retval = templ.conditionFuncHook(*this) || retval;
}
} else {
src/enactment/EnactmentEngine.cpp
+29
View File
@@ -6,11 +6,13 @@
#include <wfr/scheduling/AbstractScheduler.h>
#include <wfr/common/wfs/Workflow.h>
#include <wfr/wep/utils/WepDotFileWriter.h>
#include <wfr/wep/utils/WepProgressionWriter.h>
#include <wfr/common/eventsystem/Event.h>
#include <wfr/common/eventsystem/EventSystem.h>
#include <wfr/common/wfs/ActivityInstance.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/ActivityTemplate.h>
#include <wfr/common/wep/activities/AtomicWepNode.h>
#include <wfr/WorkflowRuntime.h>
#include <wfr/config.h>
#include <boost/program_options.hpp>
@@ -29,6 +31,7 @@ void EnactmentEngine::init(int argc, char** argv) {
opts.add_options()
("help", "Print help message.")
("disable-wepdotwriter", "Disable WEP to .dot writer.")
("enable-wepprogressionwriter", "Write WEP progression to disk.")
("early-evaluation", "Enables early evaluation.")
("late-evaluation", "Enables late evaluation [Default].");
boost::program_options::variables_map vm;
@@ -39,6 +42,7 @@ void EnactmentEngine::init(int argc, char** argv) {
boost::program_options::options_description show_opts("EnactmentEngine Options");
show_opts.add_options()
("disable-wepdotwriter", "Disable WEP to .dot writer.")
("enable-wepprogressionwriter", "Write WEP progression to disk.")
("early-evaluation", "Enables early evaluation [Default].")
("late-evaluation", "Enables late evaluation.");
std::cerr << show_opts << std::endl;
@@ -53,6 +57,14 @@ void EnactmentEngine::init(int argc, char** argv) {
std::cerr << "WEP-to-.dot writer is enabled." << std::endl;
}
if (vm.count("enable-wepprogressionwriter")) {
WorkflowRuntime::setGlobalConfig("wepProgressionWriterEnabled", true);
std::cerr << "WEP progression writer is enabled" << std::endl;
} else {
WorkflowRuntime::setGlobalConfig("wepProgressionWriterEnabled", false);
std::cerr << "WEP progression writer is disabled" << std::endl;
}
if (vm.count("early-evaluation") && vm.count("late-evaluation")) {
std::cerr << "Cannot have early and late evaluation enabled at the same time." << std::endl;
std::abort();
@@ -72,6 +84,7 @@ void EnactmentEngine::finish() {
EnactmentEngine::EnactmentEngine(wfs::Workflow& workflow, scheduling::AbstractScheduler& scheduler) noexcept
: enableWepDotFileWriter(WorkflowRuntime::getGlobalConfig<bool>("dotWriterEnabled")),
enableWepProgressionWriter(WorkflowRuntime::getGlobalConfig<bool>("wepProgressionWriterEnabled")),
enableEarlyEvaluation(WorkflowRuntime::getGlobalConfig<bool>("earlyEvaluationEnabled")),
workflow(workflow), scheduler(scheduler) {}
@@ -97,6 +110,13 @@ void EnactmentEngine::workflowExecutionLoop(lwd::_wfr_lwd* input, lwd::_wfr_lwd*
}
#endif
std::shared_ptr<wep::utils::WepProgressionWriter> progressionWriter;
if (enableWepProgressionWriter) {
progressionWriter = std::make_shared<wep::utils::WepProgressionWriter>(wep);
progressionWriter->openFile();
}
scheduler.registerWEP(*wep, *this);
auto ren = wepBuilder.getReadyToExecuteAtomicNodes();
scheduler.notifyWEPUpdate(*wep, std::set<wep::AtomicWepNode*>(ren.begin(), ren.end()));
@@ -136,6 +156,11 @@ void EnactmentEngine::workflowExecutionLoop(lwd::_wfr_lwd* input, lwd::_wfr_lwd*
dotWriter->printToFile();
}
#endif
if (enableWepProgressionWriter) {
progressionWriter->writeProgressToFile();
}
if (wep->workflow.getWorkflowStatus() == wfr::wfs::WorkflowStatus::Error
|| wep->workflow.getWorkflowStatus() == wfr::wfs::WorkflowStatus::Finished) {
breakExecutionLoop = true;
@@ -154,6 +179,10 @@ void EnactmentEngine::workflowExecutionLoop(lwd::_wfr_lwd* input, lwd::_wfr_lwd*
eventBusy.reset();
scheduler.unregisterWEP(*wep.get());
if (enableWepProgressionWriter) {
progressionWriter->closeFile();
}
// Reset thread priority
setpriority(PRIO_PROCESS, myPID, oldPriority);
}
src/enactment/WepBuilder.cpp
+15 -12
View File
@@ -19,6 +19,7 @@
#include <wfr/common/wep/WorkflowExecutionPlan.h>
#include <wfr/common/utils/print.h>
#include <wfr/wep/utils/WepSearch.h>
#include <iostream>
namespace wfr {
@@ -80,7 +81,7 @@ std::shared_ptr<wep::WorkflowExecutionPlan> WepBuilder::createWEP(wfr::wfs::Work
activityInstanceMap_t instanceMap;
// Visit children
for (auto& child : children) {
visitActivity(*child, instanceMap, wepNodeMap);
createActivityInstance(*child, instanceMap, wepNodeMap);
}
iterationIdStack.pop();
parentWepStack.pop();
@@ -93,7 +94,7 @@ std::shared_ptr<wep::WorkflowExecutionPlan> WepBuilder::createWEP(wfr::wfs::Work
activityInstanceMap_t instanceMap;
// Visit children
for (auto& child : children) {
visitActivity(*child, instanceMap, wepNodeMap);
createActivityInstance(*child, instanceMap, wepNodeMap);
}
iterationIdStack.pop();
}
@@ -204,9 +205,7 @@ WepBuilder::wepNodeMap_t::mapped_type WepBuilder::createWEPnode(wfs::Activity& a
void WepBuilder::visitActivity(wfr::wfs::Activity& element, activityInstanceMap_t& instanceMap, wepNodeMap_t& wepNodeMap) {
//wep->workflow.addToWepVisitationCount();
void WepBuilder::createActivityInstance(wfr::wfs::Activity& element, activityInstanceMap_t& instanceMap, wepNodeMap_t& wepNodeMap) {
wepVisitationCount++;
// Create activity instance
@@ -291,7 +290,7 @@ void WepBuilder::visitActivity(wfr::wfs::Activity& element, activityInstanceMap_
parentStack.push(ti);
parentWepStack.push(wepNode);
// Handle Children
handleChildren(composite, wepNode);
createChildInstances(composite, wepNode);
// Execute post-execution hooks (can we skip this?)
ti->executePostExecutionHooks(earlyEvaluationMode);
// Pop parent from stack
@@ -316,7 +315,7 @@ void WepBuilder::visitActivity(wfr::wfs::Activity& element, activityInstanceMap_
}
void WepBuilder::handleChildren(wfs::CompositeActivityInstance& composite, wepNodeMap_t::mapped_type& parent, wepNodeMap_t::mapped_type* prevIter) {
void WepBuilder::createChildInstances(wfs::CompositeActivityInstance& composite, wepNodeMap_t::mapped_type& parent, wepNodeMap_t::mapped_type* prevIter) {
// Fetch child activities
std::vector<wfr::wfs::Activity*> children;
if (composite.isConditional()) {
@@ -336,7 +335,7 @@ void WepBuilder::handleChildren(wfs::CompositeActivityInstance& composite, wepNo
activityInstanceMap_t instanceMap;
// Visit children
for (auto& child : children) {
visitActivity(*child, instanceMap, wepNodeMap);
createActivityInstance(*child, instanceMap, wepNodeMap);
}
iterationIdStack.pop();
parentWepStack.pop();
@@ -347,7 +346,7 @@ void WepBuilder::handleChildren(wfs::CompositeActivityInstance& composite, wepNo
activityInstanceMap_t instanceMap;
// Visit children
for (auto& child : children) {
visitActivity(*child, instanceMap, wepNodeMap);
createActivityInstance(*child, instanceMap, wepNodeMap);
}
iterationIdStack.pop();
}
@@ -376,6 +375,7 @@ bool WepBuilder::updateExecutionPlan(std::queue<wfs::ActivityInstance*>& updateI
<< node.activity.baseActivity.baseTemplate.activityName
<< " (" << node.nodeID << ")"
<< " [" << print::toString(node.activity.activityType()) << "]: "
<< " [WepNodeType " << (int)node.nodeType << "]: "
<< node.activity.baseActivity.baseTemplate.activityName << " - " << node.activity.baseActivity.activityID << "-" << node.activity.instanceID
<< std::endl;*/
@@ -386,7 +386,10 @@ bool WepBuilder::updateExecutionPlan(std::queue<wfs::ActivityInstance*>& updateI
}
// Execute pre-execution hooks
if (!node.activity.areAllInputsAvailable() && node.activity.getInstanceStatus() < wfr::wfs::ActivityInstanceStatus::Finished) {
if (node.nodeType != wep::WepNodeType::CompositeEnd
&& !node.activity.areAllInputsAvailable() &&
((node.activity.getInstanceStatus() < wfr::wfs::ActivityInstanceStatus::ReadyForExecution && !earlyEvaluationMode)
|| (node.activity.getInstanceStatus() < wfr::wfs::ActivityInstanceStatus::Finished && earlyEvaluationMode))) {
node.activity.executePreExecutionHooks(earlyEvaluationMode);
}
@@ -444,7 +447,7 @@ bool WepBuilder::updateExecutionPlan(std::queue<wfs::ActivityInstance*>& updateI
wnmi->second.second->getPredecessors().clear();
parentWepStack.push((*wnmi).second);
// Handle Children
handleChildren(node.activity.as<wfs::CompositeActivityInstance>(), wnmi->second);
createChildInstances(node.activity.as<wfs::CompositeActivityInstance>(), wnmi->second);
parentWepStack.pop();
} else {
throw wfr::exception::wfr_fatal_error("Could not find wepNodeMap entry.");
@@ -474,7 +477,7 @@ bool WepBuilder::updateExecutionPlan(std::queue<wfs::ActivityInstance*>& updateI
if (loop.getHandledIterationCount() < loop.getIterationCount()) {
this->parentStack.push(&composite);
auto wnmi = wepNodeMap.find(&composite);
handleChildren(composite, wnmi->second);
createChildInstances(composite, wnmi->second);
this->parentStack.pop();
}
}
src/eventsystem/Event.cpp
+2 -1
View File
@@ -185,6 +185,7 @@ std::string ComputingResourceStatusEvent::toString() const noexcept {
std::shared_ptr<ComputingResourceStatusEvent> ComputingResourceStatusEvent::fireEvent( wfr::wfs::Workflow* wf,
wfr::wfs::ActivityImplementationType deviceType,
wfr_id_t deviceID,
wfr_id_t subDeviceId,
wfr::wfs::ComputingResourceStatus oldStatus,
wfr::wfs::ComputingResourceStatus newStatus,
wfr::wfs::ActivityInstance* executedTask,
@@ -192,7 +193,7 @@ std::shared_ptr<ComputingResourceStatusEvent> ComputingResourceStatusEvent::fire
wfr_size_t deviceSlotCount,
wfr_timestamp_sc_t timestamp,
bool noLocking ) noexcept {
auto event = std::make_shared<ComputingResourceStatusEvent>(wf, deviceType, deviceID, oldStatus, newStatus, executedTask, executedImpl, deviceSlotCount, timestamp);
auto event = std::make_shared<ComputingResourceStatusEvent>(wf, deviceType, deviceID, subDeviceId, oldStatus, newStatus, executedTask, executedImpl, deviceSlotCount, timestamp);
EventSystem::getInstance().fireEvent(event, noLocking);
return event;
}
src/eventsystem/SQLiteLogger.cpp
+51 -50
View File
@@ -13,6 +13,7 @@
#include <wfr/common/wfs/ActivityInstance.h>
#include <wfr/common/wfs/ActivityTemplate.h>
#include <wfr/WorkflowRuntime.h>
#include <wfr/common/utils/print.h>
#include <ctime>
#include <sstream>
#include <thread>
@@ -127,7 +128,7 @@ void SQLiteLogger::dbwriterfunc(SQLiteLogger* logger) {
trans = new odb::transaction(logger->dbHandle->begin());
statsWrittenSnapshot = logger->statWrittenEvents;
}
/*switch (event->eventType) {
switch (event->eventType) {
case EventType::WorkflowStatusEvent:
logger->handleWorkflowStatusEvent((WorkflowStatusEvent&)*event);
logger->statWrittenEvents++;
@@ -150,7 +151,7 @@ void SQLiteLogger::dbwriterfunc(SQLiteLogger* logger) {
break;
default:
break;
}*/
}
if (trans != nullptr && logger->statWrittenEvents - statsWrittenSnapshot > 100) {
trans->commit();
delete trans;
@@ -184,10 +185,10 @@ void SQLiteLogger::notifyEvent(const std::shared_ptr<Event>& event) noexcept {
}
/*
std::shared_ptr<odbmodel::Workflow> SQLiteLogger::getWorkflowFromWfsMap(wfs::Workflow& wf) {
std::shared_ptr<odbmodel::Workflow> ret;
auto iter = workflowMap.find(wf.getWorkflowID());
auto iter = workflowMap.find(wf.workflowID);
if (iter != workflowMap.end()) {
ret = (*iter).second;
}
@@ -196,12 +197,12 @@ std::shared_ptr<odbmodel::Workflow> SQLiteLogger::getWorkflowFromWfsMap(wfs::Wor
void SQLiteLogger::persistWorkflow(std::shared_ptr<odbmodel::Workflow>& pWf, wfs::Workflow& wf) {
dbHandle->persist<odbmodel::Workflow>(pWf);
workflowMap.insert(workflowMap_t::value_type(wf.getWorkflowID(), pWf));
workflowMap.insert(workflowMap_t::value_type(wf.workflowID, pWf));
}
std::shared_ptr<odbmodel::Activity> SQLiteLogger::getActivityFromWfsMap(wfs::Activity& act) {
std::shared_ptr<odbmodel::Activity> SQLiteLogger::getActivityFromWfsMap(const wfs::Activity& act) {
std::shared_ptr<odbmodel::Activity> ret;
auto iter = activityMap.find(act.getActivityID());
auto iter = activityMap.find(act.activityID);
if (iter != activityMap.end()) {
ret = (*iter).second;
}
@@ -210,12 +211,12 @@ std::shared_ptr<odbmodel::Activity> SQLiteLogger::getActivityFromWfsMap(wfs::Act
void SQLiteLogger::persistActivity(std::shared_ptr<odbmodel::Activity>& ap, wfs::Activity& a) {
dbHandle->persist<odbmodel::Activity>(ap);
activityMap.insert(activityMap_t::value_type(a.getActivityID(), ap));
activityMap.insert(activityMap_t::value_type(a.activityID, ap));
}
std::shared_ptr<odbmodel::ActivityInstance> SQLiteLogger::getActivityInstanceFromWfsMap(wfs::ActivityInstance& act) {
std::shared_ptr<odbmodel::ActivityInstance> ret;
auto iter = activityInstanceMap.find(act.getInstanceID());
auto iter = activityInstanceMap.find(act.instanceID);
if (iter != activityInstanceMap.end()) {
ret = (*iter).second;
}
@@ -224,7 +225,7 @@ std::shared_ptr<odbmodel::ActivityInstance> SQLiteLogger::getActivityInstanceFro
void SQLiteLogger::persistActivityInstance(std::shared_ptr<odbmodel::ActivityInstance>& ap, wfs::ActivityInstance& a) {
dbHandle->persist<odbmodel::ActivityInstance>(ap);
activityInstanceMap.insert(activityInstanceMap_t::value_type(a.getInstanceID(), ap));
activityInstanceMap.insert(activityInstanceMap_t::value_type(a.instanceID, ap));
}
@@ -236,8 +237,8 @@ void SQLiteLogger::handleWorkflowStatusEvent(WorkflowStatusEvent &event) {
boost::uuids::uuid uuid = boost::uuids::random_generator()();
const std::string tmp = boost::lexical_cast<std::string>(uuid);
std::shared_ptr<odbmodel::Workflow> pWf(new odbmodel::Workflow(wf.getWorkflowID(), tmp));
pWf->setName(wf.getBaseTemplate().getWorkflowName());
std::shared_ptr<odbmodel::Workflow> pWf(new odbmodel::Workflow(wf.workflowID, tmp));
pWf->setName(wf.baseTemplate.workflowName);
persistWorkflow(pWf, wf);
writeWorkflowStatus(pWf, event);
}
@@ -250,7 +251,7 @@ void SQLiteLogger::handleWorkflowStatusEvent(WorkflowStatusEvent &event) {
dbHandle->update(wfp);
writeWorkflowStatus(wfp, event);
} else {
WorkflowRuntime::logMessage("Workflow or Activity not found in wfsMaps", true);
WorkflowRuntime::logMessage("Workflow or Activity not found in wfsMaps", std::chrono::system_clock::now(), true);
}
}
break;
@@ -267,11 +268,11 @@ void SQLiteLogger::writeWorkflowStatus(std::shared_ptr<odbmodel::Workflow>& pWf,
std::shared_ptr<odbmodel::WorkflowStatusEvent> pWSE(new odbmodel::WorkflowStatusEvent());
pWSE->setParentWorkflow(pWf);
pWSE->setOldStatus(EventMemberUtils::toString(event.getOldStatus()));
pWSE->setNewStatus(EventMemberUtils::toString(event.getNewStatus()));
pWSE->setOldStatus(wfr::print::toString(event.getOldStatus()));
pWSE->setNewStatus(wfr::print::toString(event.getNewStatus()));
typedef std::chrono::time_point<std::chrono::system_clock, std::chrono::nanoseconds> ns_t;
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.getTimestamp());
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.timestamp);
pWSE->setTimestamp(ns.time_since_epoch().count());
dbHandle->persist<odbmodel::WorkflowStatusEvent>(pWSE);
@@ -280,19 +281,19 @@ void SQLiteLogger::writeWorkflowStatus(std::shared_ptr<odbmodel::Workflow>& pWf,
void SQLiteLogger::handleActicityStatusEvent(ActivityStatusEvent &event) {
auto& a = event.getActivity();
auto& wf = a.getWorkflow();
auto& wf = a.workflow;
switch (event.getNewStatus()) {
case wfs::ActivityStatus::Created: {
auto wfp = getWorkflowFromWfsMap(wf);
if (wfp != nullptr) {
std::shared_ptr<odbmodel::Activity> pA(new odbmodel::Activity(a.getActivityID(), a.getBaseTemplate().getBranchID()));
pA->setName(a.getBaseTemplate().getActivityName());
pA->setType(EventMemberUtils::toString(a.getActivityType()));
std::shared_ptr<odbmodel::Activity> pA(new odbmodel::Activity(a.activityID, a.baseTemplate.branchID));
pA->setName(a.baseTemplate.activityName);
pA->setType(wfr::print::toString(a.activityType()));
pA->setParentWorkflow(wfp);
persistActivity(pA, a);
writeActivityStatus(pA, event);
} else {
WorkflowRuntime::logMessage("Workflow not found in wfsMaps", true);
WorkflowRuntime::logMessage("Workflow not found in wfsMaps", std::chrono::system_clock::now(), true);
}
}
break;
@@ -301,13 +302,13 @@ void SQLiteLogger::handleActicityStatusEvent(ActivityStatusEvent &event) {
auto ap = getActivityFromWfsMap(a);
if (wfp != nullptr && ap != nullptr) {
for (wfr_size_t t = 0; t < a.getSuccessorCount(); t++) {
auto sp = getActivityFromWfsMap(*a.getSuccessorList()[t]);
auto sp = getActivityFromWfsMap(*a.getSuccessors()[t]);
if (sp != nullptr) {
ap->getSuccessors().push_back(sp);
}
}
for (wfr_size_t t = 0; t < a.getPredecessorCount(); t++) {
auto pp = getActivityFromWfsMap(*a.getPredecessorList()[t]);
auto pp = getActivityFromWfsMap(*a.getPredecessors()[t]);
if (pp != nullptr) {
ap->getPredecessors().push_back(pp);
}
@@ -320,7 +321,7 @@ void SQLiteLogger::handleActicityStatusEvent(ActivityStatusEvent &event) {
dbHandle->update(ap);
writeActivityStatus(ap, event);
} else {
WorkflowRuntime::logMessage("Workflow or Activity not found in wfsMaps", true);
WorkflowRuntime::logMessage("Workflow or Activity not found in wfsMaps", std::chrono::system_clock::now(), true);
}
}
break;
@@ -337,11 +338,11 @@ void SQLiteLogger::writeActivityStatus(std::shared_ptr<odbmodel::Activity>& pA,
std::shared_ptr<odbmodel::ActivityStatusEvent> pASE(new odbmodel::ActivityStatusEvent());
pASE->setParentActivity(pA);
pASE->setOldStatus(EventMemberUtils::toString(event.getOldStatus()));
pASE->setNewStatus(EventMemberUtils::toString(event.getNewStatus()));
pASE->setOldStatus(wfr::print::toString(event.getOldStatus()));
pASE->setNewStatus(wfr::print::toString(event.getNewStatus()));
typedef std::chrono::time_point<std::chrono::system_clock, std::chrono::nanoseconds> ns_t;
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.getTimestamp());
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.timestamp);
pASE->setTimestamp(ns.time_since_epoch().count());
dbHandle->persist<odbmodel::ActivityStatusEvent>(pASE);
@@ -350,21 +351,21 @@ void SQLiteLogger::writeActivityStatus(std::shared_ptr<odbmodel::Activity>& pA,
void SQLiteLogger::handleActicityInstanceStatusEvent(ActivityInstanceStatusEvent &event) {
auto& ai = event.getActivityInstance();
auto pai = ai.getParent();
auto& a = ai.getBaseActivity();
auto& wf = a.getWorkflow();
auto pai = ai.getParentActivity();
auto& a = ai.baseActivity;
auto& wf = a.workflow;
switch (event.getNewStatus()) {
case wfs::ActivityInstanceStatus::Created: {
auto wfp = getWorkflowFromWfsMap(wf);
auto ap = getActivityFromWfsMap(a);
if (wfp != nullptr && ap != nullptr) {
std::shared_ptr<odbmodel::ActivityInstance> pAI(new odbmodel::ActivityInstance(ai.getInstanceID(), ai.getIterationID()));
std::shared_ptr<odbmodel::ActivityInstance> pAI(new odbmodel::ActivityInstance(ai.instanceID, ai.iterationID));
pAI->setBaseActivity(ap);
pAI->setParentWorkflow(wfp);
persistActivityInstance(pAI, ai);
writeActivityInstanceStatus(pAI, event);
} else {
WorkflowRuntime::logMessage("Workflow or Activity not found in wfsMaps", true);
WorkflowRuntime::logMessage("Workflow or Activity not found in wfsMaps", std::chrono::system_clock::now(), true);
}
}
break;
@@ -382,7 +383,7 @@ void SQLiteLogger::handleActicityInstanceStatusEvent(ActivityInstanceStatusEvent
dbHandle->update(pAi);
writeActivityInstanceStatus(pAi, event);
} else {
WorkflowRuntime::logMessage("Workflow or Activity not found in wfsMaps", true);
WorkflowRuntime::logMessage("Workflow or Activity not found in wfsMaps", std::chrono::system_clock::now(), true);
}
}
break;
@@ -400,18 +401,18 @@ void SQLiteLogger::writeActivityInstanceStatus(std::shared_ptr<odbmodel::Activit
std::shared_ptr<odbmodel::ActivityInstanceStatusEvent> pASE(new odbmodel::ActivityInstanceStatusEvent());
pASE->setParentInstance(pA);
pASE->setOldStatus(EventMemberUtils::toString(event.getOldStatus()));
pASE->setNewStatus(EventMemberUtils::toString(event.getNewStatus()));
pASE->setOldStatus(wfr::print::toString(event.getOldStatus()));
pASE->setNewStatus(wfr::print::toString(event.getNewStatus()));
typedef std::chrono::time_point<std::chrono::system_clock, std::chrono::nanoseconds> ns_t;
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.getTimestamp());
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.timestamp);
pASE->setTimestamp(ns.time_since_epoch().count());
dbHandle->persist<odbmodel::ActivityInstanceStatusEvent>(pASE);
}
void SQLiteLogger::handleDeviceInstanceStatusEvent(DeviceInstanceStatusEvent &event) {
void SQLiteLogger::handleComputingResourceStatusEvent(ComputingResourceStatusEvent &event) {
auto ai = event.getExecutedTask();
auto wf = event.getWorkflow();
std::shared_ptr<odbmodel::ActivityInstance> pai;
@@ -422,29 +423,29 @@ void SQLiteLogger::handleDeviceInstanceStatusEvent(DeviceInstanceStatusEvent &ev
if (wf != nullptr) {
pwf = getWorkflowFromWfsMap(*wf);
}
writeDeviceInstanceStatus(pwf, pai, event);
writeComputingResourceStatus(pwf, pai, event);
}
void SQLiteLogger::writeDeviceInstanceStatus(std::shared_ptr<odbmodel::Workflow>& pwf, std::shared_ptr<odbmodel::ActivityInstance>& pai, DeviceInstanceStatusEvent &event) {
void SQLiteLogger::writeComputingResourceStatus(std::shared_ptr<odbmodel::Workflow>& pwf, std::shared_ptr<odbmodel::ActivityInstance>& pai, ComputingResourceStatusEvent &event) {
std::shared_ptr<odbmodel::DeviceInstanceStatusEvent> pDSE(new odbmodel::DeviceInstanceStatusEvent());
pDSE->setParentWorkflow(pwf);
pDSE->setDeviceSlotCount(event.getDeviceSlotCount());
pDSE->setDeviceId(event.getDeviceID());
pDSE->setDeviceType(EventMemberUtils::toString(event.getDeviceType()));
pDSE->setDeviceSubId(event.getDeviceSubID());
pDSE->setDeviceType(wfr::print::toString(event.getDeviceType()));
if (pai != nullptr) {
pDSE->setExecutedInstance(pai);
}
if (event.getExecutedImpl() != nullptr) {
pDSE->setExecutedImplId(event.getExecutedImpl()->getImplementationId());
pDSE->setExecutedImplId(event.getExecutedImpl()->implementationID);
}
pDSE->setOldStatus(EventMemberUtils::toString(event.getOldStatus()));
pDSE->setNewStatus(EventMemberUtils::toString(event.getNewStatus()));
pDSE->setOldStatus(wfr::print::toString(event.getOldStatus()));
pDSE->setNewStatus(wfr::print::toString(event.getNewStatus()));
typedef std::chrono::time_point<std::chrono::system_clock, std::chrono::nanoseconds> ns_t;
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.getTimestamp());
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.timestamp);
pDSE->setTimestamp(ns.time_since_epoch().count());
dbHandle->persist<odbmodel::DeviceInstanceStatusEvent>(pDSE);
@@ -466,18 +467,18 @@ void SQLiteLogger::writeComponentStatus(std::shared_ptr<odbmodel::Workflow>& pwf
std::shared_ptr<odbmodel::ComponentStatusEvent> pCSE(new odbmodel::ComponentStatusEvent());
pCSE->setWorkflow(pwf);
pCSE->setComponentType(EventMemberUtils::toString(event.getComponentType()));
pCSE->setComponentType(wfr::print::toString(event.getComponentType()));
pCSE->setOldStatus(EventMemberUtils::toString(event.getOldStatus()));
pCSE->setNewStatus(EventMemberUtils::toString(event.getNewStatus()));
pCSE->setOldStatus(wfr::print::toString(event.getOldStatus()));
pCSE->setNewStatus(wfr::print::toString(event.getNewStatus()));
typedef std::chrono::time_point<std::chrono::system_clock, std::chrono::nanoseconds> ns_t;
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.getTimestamp());
ns_t ns = std::chrono::time_point_cast<std::chrono::nanoseconds>(event.timestamp);
pCSE->setTimestamp(ns.time_since_epoch().count());
dbHandle->persist<odbmodel::ComponentStatusEvent>(pCSE);
}
*/
} /* namespace eventsystem */
} /* namespace wfr */
src/execution/ExecutionManager.cpp
+2 -1
View File
@@ -55,7 +55,8 @@ void ExecutionManager::init(int argc, char** argv) {
{
std::stringstream ss;
ss << "ExecutionManager: PthreadBackend: using " << singleton.pThreadBackend->getTotalSlotCount() << " pthread threads.";
ss << "ExecutionManager: PthreadBackend: having " << singleton.pThreadBackend->getDeviceCount()
<< " devices with a total of " << singleton.pThreadBackend->getTotalSlotCount() << " pthread threads.";
wfr::WorkflowRuntime::logMessage(ss.str());
}
}
src/execution/backends/pthreads/PThreadBackend.cpp
+102 -24
View File
@@ -14,14 +14,11 @@
namespace wfr {
namespace execution {
PThreadBackend::PThreadBackend(ExecutionManager& manager, const std::set<resource::ComputingResource*>& resources)
: manager(manager), resources(resources), maxWorkerCount(0) {
for (auto dev : resources) {
maxWorkerCount += dev->getMaxSlotCount();
}
PThreadDevice::PThreadDevice(PThreadBackend& backend, resource::ComputingResource* resource)
: backend(backend), resource(resource), maxWorkerCount(resource->getMaxSlotCount()) {
}
PThreadBackend::~PThreadBackend() {
PThreadDevice::~PThreadDevice() {
std::unique_lock<std::mutex> objLock(objMutex);
while (currentBusyWorkerCount > 0 || !taskQueue.empty()) {
objLock.unlock();
@@ -42,12 +39,39 @@ PThreadBackend::~PThreadBackend() {
idleWorkers.clear();
}
void PThreadBackend::notifyExecutionHandleUpdate(ExecutionHandle& handle, PThreadWorker& worker) {
manager.notifyExecutionHandleUpdate(handle);
wfr_id_t PThreadDevice::getDeviceId() {
return resource->resourceId;
}
ExecutionHandle* PThreadBackend::popTaskExecution(PThreadWorker& worker) {
void PThreadDevice::notifyExecutionHandleUpdate(ExecutionHandle& handle, PThreadWorker& worker) {
backend.notifyExecutionHandleUpdate(handle, worker);
}
bool PThreadDevice::queueTaskExecution(ExecutionHandle& handle) {
bool retval = true;
std::lock_guard<std::mutex> objLock(objMutex);
taskQueue.push(&handle);
slotsInQueue += handle.schedulingDecision->slotCount;
handle.schedulingDecision->activityInstance->setInstanceStatus(wfs::ActivityInstanceStatus::Queued);
if (slotsInQueue > idleWorkers.size()) {
if (maxWorkerCount == 0 || currentTotalWorkerCount < maxWorkerCount) {
PThreadWorker::createNewThread(*this);
currentTotalWorkerCount++;
//#ifdef WFR_ENABLE_STATISTICS
// statistics::WfrStatistics::setMaxObservedCurrentPthreadCount(currentTotalWorkerCount);
//#endif
} else {
handle.schedulingDecision->activityInstance->setInstanceStatus(wfs::ActivityInstanceStatus::WaitingForFreeResources);
}
}
workerCondVar.notify_all();
return retval;
}
ExecutionHandle* PThreadDevice::popTaskExecution(PThreadWorker& worker) {
std::unique_lock<std::mutex> objLock(objMutex);
if (worker.isBusy()) {
occupiedSlotCount-=worker.getUsedSlotCount();
@@ -82,9 +106,9 @@ ExecutionHandle* PThreadBackend::popTaskExecution(PThreadWorker& worker) {
worker.setBusy(true);
worker.setUsedSlotCount(handle->schedulingDecision->slotCount);
auto sd = handle->schedulingDecision;
auto reservation = sd->reservationHandle;
if (reservation.isValid()) {
reservation.resource->markReservedSlotsBusy(reservation.slotCount);
auto reservation = sd->getComputingResourceReservation();
if (reservation && reservation->isValid()) {
reservation->resource->markReservedSlotsBusy(reservation->slotCount);
} else if (sd->resource) {
sd->resource->markFreeSlotsBusy(sd->slotCount);
}
@@ -99,26 +123,80 @@ ExecutionHandle* PThreadBackend::popTaskExecution(PThreadWorker& worker) {
}
}
PThreadBackend::PThreadBackend(ExecutionManager& manager, const std::set<resource::ComputingResource*>& resources)
: manager(manager) {
for (auto dev : resources) {
auto device = std::make_shared<PThreadDevice>(*this, dev);
devices.insert({dev, device});
}
}
PThreadBackend::~PThreadBackend() {}
void PThreadBackend::notifyExecutionHandleUpdate(ExecutionHandle& handle, PThreadWorker& worker) {
manager.notifyExecutionHandleUpdate(handle);
}
bool PThreadBackend::queueTaskExecution(ExecutionHandle& handle) {
bool retval = true;
std::lock_guard<std::mutex> objLock(objMutex);
taskQueue.push(&handle);
slotsInQueue += handle.schedulingDecision->slotCount;
handle.schedulingDecision->activityInstance->setInstanceStatus(wfs::ActivityInstanceStatus::Queued);
if (slotsInQueue > idleWorkers.size()) {
if (maxWorkerCount == 0 || currentTotalWorkerCount < maxWorkerCount) {
PThreadWorker::createNewThread(*this);
currentTotalWorkerCount++;
//#ifdef WFR_ENABLE_STATISTICS
// statistics::WfrStatistics::setMaxObservedCurrentPthreadCount(currentTotalWorkerCount);
//#endif
auto res = handle.schedulingDecision->resource;
if (res) {
auto device = devices.find(res);
if (device != devices.end()) {
device->second->queueTaskExecution(handle);
} else {
handle.schedulingDecision->activityInstance->setInstanceStatus(wfs::ActivityInstanceStatus::WaitingForFreeResources);
throw wfr::exception::wfr_not_found(std::string("Could not find pthread resource ") + std::string(res->resourceIdentifier));
}
} else {
wfr_size_t devBusySlots = 9999;
std::shared_ptr<PThreadDevice> dev = nullptr;
bool devFound = false;
for (auto device : devices) {
if (device.second->getFreeSlotCount() >= handle.schedulingDecision->slotCount) {
device.second->queueTaskExecution(handle);
devFound = true;
break;
} else {
if (device.second->getBusyWorkerCount() <= devBusySlots) {
devBusySlots = device.second->getBusyWorkerCount();
dev = device.second;
}
}
}
if (!devFound) {
dev->queueTaskExecution(handle);
}
}
workerCondVar.notify_all();
return retval;
}
wfr_size_t PThreadBackend::getDeviceCount() {
return devices.size();
}
wfr_size_t PThreadBackend::getBusyWorkerCount() {
wfr_size_t count = 0;
for (auto device : devices) {
count += device.second->getBusyWorkerCount();
}
return count;
}
wfr_size_t PThreadBackend::getTotalSlotCount() {
wfr_size_t count = 0;
for (auto device : devices) {
count += device.second->getTotalSlotCount();
}
return count;
}
wfr_size_t PThreadBackend::getFreeSlotCount() {
return getTotalSlotCount() - getBusyWorkerCount();
}
} /* namespace execution */
} /* namespace wfr */
src/execution/backends/pthreads/PThreadWorker.cpp
+12 -13
View File
@@ -23,12 +23,11 @@ wfr_id_t PThreadWorker::getNextThreadID() {
return nextThreadID.fetch_add(1);
}
PThreadWorker::PThreadWorker(PThreadBackend& backend)
: threadID(getNextThreadID()), stopThread(false), threadRunning(false), threadBusy(false), backend(backend), thread(nullptr),
PThreadWorker::PThreadWorker(PThreadDevice& device)
: threadID(getNextThreadID()), stopThread(false), threadRunning(false), threadBusy(false), device(device), thread(nullptr),
usedSlotCount(0), status(wfs::ComputingResourceStatus::Created) {
thread = new std::thread(workerFunc, this);
wfr::eventsystem::ComputingResourceStatusEvent::fireEvent(nullptr, wfr::wfs::ActivityImplementationType::PThread, threadID,
wfr::eventsystem::ComputingResourceStatusEvent::fireEvent(nullptr, wfr::wfs::ActivityImplementationType::PThread, device.getDeviceId(), threadID,
wfr::wfs::ComputingResourceStatus::Unknown, wfr::wfs::ComputingResourceStatus::Created);
}
@@ -47,7 +46,7 @@ void PThreadWorker::workerFunc(PThreadWorker* threadObj) {
threadObj->threadRunning = true;
while (!threadObj->stopThread) {
ExecutionHandle* handle;
while ((handle = threadObj->backend.popTaskExecution(*threadObj)) != nullptr) {
while ((handle = threadObj->device.popTaskExecution(*threadObj)) != nullptr) {
auto sd = handle->schedulingDecision;
auto execFunc = sd->activityImplementation->as<wfs::PThreadActivityImplementation>().executionFuncHook;
sd->activityInstance->setInstanceStatus(wfr::wfs::ActivityInstanceStatus::Running);
@@ -55,10 +54,10 @@ void PThreadWorker::workerFunc(PThreadWorker* threadObj) {
handle->setState(ExecutionHandleState::Running);
try {
execFunc(*sd->activityInstance);
auto reservation = sd->reservationHandle;
if (reservation.isValid()) {
reservation.resource->releaseBusySlots(reservation.slotCount);
reservation.invalidateReservation();
auto reservation = sd->getComputingResourceReservation();
if (reservation && reservation->isValid()) {
reservation->resource->releaseBusySlots(reservation->slotCount);
reservation->invalidateReservation();
} else if (sd->resource) {
sd->resource->releaseBusySlots(sd->slotCount);
}
@@ -74,7 +73,7 @@ void PThreadWorker::workerFunc(PThreadWorker* threadObj) {
sd.getActivityInstance().setInstanceStatus(wfr::wfs::ActivityInstanceStatus::Error);
*/
}
threadObj->backend.notifyExecutionHandleUpdate(*handle, *threadObj);
threadObj->device.notifyExecutionHandleUpdate(*handle, *threadObj);
}
}
threadObj->threadRunning = false;
@@ -93,13 +92,13 @@ void PThreadWorker::join() {
void PThreadWorker::setStatus(wfs::ComputingResourceStatus status, std::shared_ptr<wfr::scheduling::SchedulingDecision> sd) {
wfr::eventsystem::ComputingResourceStatusEvent::fireEvent( (sd.get() == nullptr) ? nullptr : &sd->activityInstance->baseActivity.workflow,
wfr::wfs::ActivityImplementationType::PThread, threadID, this->status, status, (sd.get() == nullptr) ? nullptr : sd->activityInstance,
wfr::wfs::ActivityImplementationType::PThread, device.getDeviceId(), threadID, this->status, status, (sd.get() == nullptr) ? nullptr : sd->activityInstance,
(sd.get() == nullptr) ? nullptr : sd->activityImplementation, (sd.get() == nullptr) ? 0 : sd->slotCount);
this->status = status;
}
PThreadWorker* PThreadWorker::createNewThread(PThreadBackend& backend) {
auto thread = new PThreadWorker(backend);
PThreadWorker* PThreadWorker::createNewThread(PThreadDevice& device) {
auto thread = new PThreadWorker(device);
return thread;
}
src/prediction/services/oracle/OraclePredictionService.cpp
+1 -1
View File
@@ -27,7 +27,7 @@ ExecutionTimeResult OraclePredictionService::predictExecutionTime(wfs::ActivityI
auto devname = prop.first.substr(9);
if (devname.compare(resourceType.resourceTypeName) == 0) {
double exectime = boost::lexical_cast<double>(prop.second);
std::cout << "Exectime: " << devname << " = " << exectime << std::endl;
//std::cout << "Exectime: " << devname << " = " << exectime << std::endl;
return ExecutionTimeResult(&instance, &implementation, &resourceType, NumericalMeasurement(exectime, 0));
}
}
src/scheduling/AbstractScheduler.cpp
+2
View File
@@ -7,6 +7,8 @@
#include <wfr/common/wep/AbstractWepNode.h>
#include <wfr/common/wep/activities/AtomicWepNode.h>
#include <wfr/common/wfs/ActivityInstance.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/ActivityTemplate.h>
#include <wfr/common/wfs/AtomicActivityImplementation.h>
#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
#include <wfr/enactment/EnactmentEngine.h>
src/scheduling/SchedulerManager.cpp
+16
View File
@@ -11,6 +11,10 @@
#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>
#include <wfr/scheduling/scheduler/maxmin/MaxMinScheduler.h>
#include <wfr/scheduling/scheduler/sufferage/SufferageScheduler.h>
#include <wfr/scheduling/scheduler/peft/PEFTScheduler.h>
namespace wfr {
namespace scheduling {
@@ -79,6 +83,18 @@ 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("minmin", []()->AbstractScheduler* {
return new MinMinScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
}));
schedulerFactories.insert(factoryMap_t::value_type("maxmin", []()->AbstractScheduler* {
return new MaxMinScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
}));
schedulerFactories.insert(factoryMap_t::value_type("sufferage", []()->AbstractScheduler* {
return new SufferageScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
}));
schedulerFactories.insert(factoryMap_t::value_type("peft", []()->AbstractScheduler* {
return new PEFTScheduler(resource::ResourceManager::getInstance(), prediction::PredictionManager::getInstance());
}));
if (schedulerFactories.find(defaultScheduler) == schedulerFactories.end()) {
throw wfr::exception::wfr_fatal_error((boost::format("Default scheduler \"%1%\" does not exist.") % defaultScheduler).str());
}
src/scheduling/cmake.inc
+4
View File
@@ -6,4 +6,8 @@ set(wfr_cpp_files "${wfr_cpp_files}"
src/scheduling/scheduler/heft/HEFTScheduler.cpp
src/scheduling/scheduler/random/RandomScheduler.cpp
src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp
src/scheduling/scheduler/minmin/MinMinScheduler.cpp
src/scheduling/scheduler/maxmin/MaxMinScheduler.cpp
src/scheduling/scheduler/sufferage/SufferageScheduler.cpp
src/scheduling/scheduler/peft/PEFTScheduler.cpp
)
src/scheduling/scheduler/heft/HEFTScheduler.cpp
+9 -4
View File
@@ -153,16 +153,20 @@ void HEFTScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<
wfr::utils::wep::wepBreadthFirstReverseSearch(wepe.wep->getEndNode(), visitFunc);
//std::cout << "HEFT Iteration" << std::endl << "======================" << std::endl;
// Go through scheduling queue and schedule activities;
/*while (!schedulingQueue.empty()) {
auto e = schedulingQueue.front();*/
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);
@@ -179,8 +183,9 @@ void HEFTScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<
}
}
if (fastestImpl) {
auto reservation = resourceManager.reserveComputingResource(fastestResource, 1);
if (reservation.isValid()) {
//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);
src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp
+1 -1
View File
@@ -58,7 +58,7 @@ void JustInTimeScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std
}
}
if (fastestImpl) {
auto reservation = resourceManager.reserveComputingResource(fastestRes, 1);
auto reservation = resourceManager.reserveComputingResource(fastestRes, 1, false);
auto sd = std::make_shared<SchedulingDecision>(&ai, fastestImpl, wepe.wep, reservation);
sd->setPrediction(fastestPred);
ai.setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled);
src/scheduling/scheduler/maxmin/MaxMinScheduler.cpp
+50
View File
@@ -0,0 +1,50 @@
#include <wfr/scheduling/scheduler/maxmin/MaxMinScheduler.h>
#include <wfr/common/wep/WorkflowExecutionPlan.h>
#include <wfr/common/wep/AbstractWepNode.h>
#include <wfr/common/wep/activities/AtomicWepNode.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/ActivityInstance.h>
#include <wfr/common/wfs/AtomicActivityImplementation.h>
#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
#include <wfr/common/wfs/templates/AtomicActivityTemplate.h>
#include <wfr/common/resource/ResourceManager.h>
#include <wfr/enactment/EnactmentEngine.h>
#include <wfr/common/scheduling/SchedulingDecision.h>
#include <wfr/execution/ExecutionManager.h>
#include <wfr/WorkflowRuntime.h>
#include <wfr/wep/utils/WepSearch.h>
#include <iostream>
#include <set>
#include <queue>
namespace wfr {
namespace scheduling {
std::pair<wep::AtomicWepNode*, MinMinScheduler::ECTEntry> MaxMinScheduler::_getNextTaskToSchedule(std::map<wep::AtomicWepNode*, ECTEntry>& estimatedTaskCompletionTimes) {
wep::AtomicWepNode* selectedNode = nullptr;
ECTEntry selectedEntry;
bool firstIteration = true;
for (auto n : estimatedTaskCompletionTimes) {
if (firstIteration) {
selectedNode = n.first;
selectedEntry = n.second;
firstIteration = false;
} else {
// the only difference to min-min is that we select the task with the maximum earliest completion time.
if (std::get<2>(n.second) > std::get<2>(selectedEntry)) {
selectedNode = n.first;
selectedEntry = n.second;
}
}
}
return std::make_pair(selectedNode, selectedEntry);
}
} /* namespace scheduling */
} /* namespace wfr */
src/scheduling/scheduler/minmin/MinMinScheduler.cpp
+164
View File
@@ -0,0 +1,164 @@
#include <wfr/scheduling/scheduler/minmin/MinMinScheduler.h>
#include <wfr/common/wep/WorkflowExecutionPlan.h>
#include <wfr/common/wep/AbstractWepNode.h>
#include <wfr/common/wep/activities/AtomicWepNode.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/ActivityInstance.h>
#include <wfr/common/wfs/AtomicActivityImplementation.h>
#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
#include <wfr/common/wfs/templates/AtomicActivityTemplate.h>
#include <wfr/common/resource/ResourceManager.h>
#include <wfr/enactment/EnactmentEngine.h>
#include <wfr/common/scheduling/SchedulingDecision.h>
#include <wfr/execution/ExecutionManager.h>
#include <wfr/WorkflowRuntime.h>
#include <wfr/wep/utils/WepSearch.h>
#include <iostream>
#include <set>
#include <queue>
namespace wfr {
namespace scheduling {
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<wep::AtomicWepNode*, MinMinScheduler::ECTEntry> MinMinScheduler::_getNextTaskToSchedule(std::map<wep::AtomicWepNode*, ECTEntry>& estimatedTaskCompletionTimes) {
wep::AtomicWepNode* selectedNode = nullptr;
ECTEntry selectedEntry;
bool firstIteration = true;
for (auto n : estimatedTaskCompletionTimes) {
if (firstIteration) {
selectedNode = n.first;
selectedEntry = n.second;
firstIteration = false;
} else {
if (std::get<2>(n.second) < std::get<2>(selectedEntry)) {
selectedNode = n.first;
selectedEntry = n.second;
}
}
}
return std::make_pair(selectedNode, selectedEntry);
}
void MinMinScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) {
while (!readyToExecute.empty()) {
std::map<wep::AtomicWepNode*, ECTEntry> estimatedTaskCompletionTimes;
auto resourceTypes = resourceManager.getComputingResourceTypes();
// Iterate over all tasks
for (auto n : readyToExecute) {
resource::ComputingResource* selectedResource = nullptr;
double selectedEarliestEstimatedCompletionTime = -1;
wfs::AtomicActivityImplementation* selectedImpl = nullptr;
prediction::ExecutionTimeResult selectedPred;
std::map<resource::ComputingResource*, double> estimatedCompletionTimes;
// Iterate over all resource types
for (auto restype : resourceTypes) {
// Get fastest implementation and predicted time for given resource type
wfs::AtomicActivityImplementation* fastestImpl = nullptr;
prediction::ExecutionTimeResult fastestPred;
// Check cache first
bool cacheHit = false;
auto aaType = n->activity.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>().atomicActivityType.get();
auto& innerMap = implementationCache[aaType];
auto cacheEntry = innerMap.find(restype);
if (cacheEntry != innerMap.end()) {
fastestImpl = cacheEntry->second.first;
fastestPred = cacheEntry->second.second;
cacheHit = true;
}
if (!cacheHit) {
// Cache miss
auto& aat = n->activity.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>();
for (auto i : aat.atomicActivityType->getImplementations()) {
auto execPred = predictionManager.predictExecutionTime(n->activity, *i, *restype);
if (execPred.isValid() && (!fastestPred.isValid() || execPred < fastestPred)) {
fastestPred = execPred;
fastestImpl = i;
}
}
// store into cache
if (fastestPred.isValid()) {
innerMap.insert({restype, {fastestImpl, fastestPred}});
}
}
if (fastestPred.isValid()) {
// Iterate over all resources and calculate earliest completion time
for (auto res : restype->getResources()) {
auto ect = _getResourceEstimatedAvailableTime(res) + fastestPred.executionTime.getValue();
if (selectedEarliestEstimatedCompletionTime < 0 || ect < selectedEarliestEstimatedCompletionTime) {
selectedEarliestEstimatedCompletionTime = ect;
selectedResource = res;
selectedImpl = fastestImpl;
selectedPred = fastestPred;
}
} // End: Iterate over all resources
}
} // End: Iterate over all resource types
if (selectedResource && selectedEarliestEstimatedCompletionTime >= 0) {
estimatedTaskCompletionTimes.insert({n, std::make_tuple(selectedResource, selectedImpl, selectedEarliestEstimatedCompletionTime, selectedPred)});
}
} // End: Iterate over all tasks
// Select next task to schedule
auto selectedTask = _getNextTaskToSchedule(estimatedTaskCompletionTimes);
// Schedule task
if (selectedTask.first) {
auto reservation = resourceManager.reserveComputingResource(std::get<0>(selectedTask.second), 1, false);
auto sd = std::make_shared<SchedulingDecision>(&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<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) {
auto& innerMap = scheduledTasks[sd->resource];
for (auto i = innerMap.begin(); i != innerMap.end(); i++) {
if (*i == sd) {
innerMap.erase(i);
break;
}
}
}
} /* namespace scheduling */
} /* namespace wfr */
src/scheduling/scheduler/peft/PEFTScheduler.cpp
Executable
+225
View File
@@ -0,0 +1,225 @@
#include <wfr/prediction/PredictionManager.h>
#include <wfr/common/resource/ResourceManager.h>
#include <wfr/scheduling/scheduler/peft/PEFTScheduler.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 <queue>
#include <memory>
#include <cstring>
namespace wfr {
namespace scheduling {
void PEFTScheduler::calcOptimisticCosts(wep::ActivityWepNode& node, std::vector<resource::ComputingResource*>& resources) {
/*std::cout << "calcOptimisticCosts: " << node.activity.baseActivity.baseTemplate.activityName << " (" << node.activity.instanceID << ")"
<< " [" << print::toString(node.activity.baseActivity.activityType()) << "]"
<< " [" << (int)node.nodeType << "]" << std::endl;*/
auto entry = (OptimisticCostTableEntry*)node.getUserData().get();
if (!entry) {
entry = new OptimisticCostTableEntry(resources.size());
if (node.nodeType == wep::WepNodeType::Atomic) {
// get fastest implementation for each resource
unsigned ri = 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 (!fastestPred.isValid() || pred < fastestPred) {
fastestPred = pred;
fastestImpl = impl;
}
}
entry->fastestImplementations[ri] = fastestImpl;
entry->fastestExecutionTimes[ri] = fastestPred;
ri++;
}
}
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" << std::endl;
for (unsigned i = 0; i < resources.size(); i++) {
entry->values[i] = 0.0;
}
entry->finalized = true;
} else {
unsigned ri1 = 0;
bool succsFinalized = true;
for (auto res1: resources) {
double maxOct = 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 = (OptimisticCostTableEntry*)succ.getUserData().get();
if (!succEntry || !succEntry->finalized) {
// recursively calculate optimistic costs
//std::cout << "recursive call start" << std::endl;
calcOptimisticCosts(succ, resources);
//std::cout << "recursive call end" << std::endl;
if (!succEntry) {
succEntry = (OptimisticCostTableEntry*)succ.getUserData().get();
}
succsFinalized = succsFinalized && succEntry->finalized;
}
unsigned ri2 = 0;
double minOct = -1;
for (auto res2 : resources) {
double execTime = 0;
double commTime = 0;
if (succ.nodeType == wep::WepNodeType::Atomic) {
// get fastest implementation
prediction::ExecutionTimeResult fastestPred = succEntry->fastestExecutionTimes[ri2];
if (fastestPred.isValid()) {
execTime = fastestPred.executionTime.getValue();
} else {
execTime = 999999;
}
if (res1 != res2) {
auto commPred = predictionManager.predictCommunicationTime(node.activity, *res1, succ.activity, *res2);
if (commPred.isValid) {
commTime = commPred.communicationTime.getValue();
}
}
}
double oct = succEntry->values[ri2] + execTime + commTime;
if (minOct < 0 || oct < minOct) {
minOct = oct;
}
ri2++;
}
//std::cout << "\tminOCT: " << minOct << std::endl;
if (minOct > maxOct) {
maxOct = minOct;
}
}
//std::cout << "\tmaxOCT: " << maxOct << std::endl;
entry->values[ri1] = maxOct;
ri1++;
}
// Calculate rank
double sum = 0;
for (unsigned i = 0; i < entry->resourceCount; i++) {
sum += entry->values[i];
}
entry->rank = sum / entry->resourceCount;
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 PEFTScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) {
auto resourceTypes = resourceManager.getComputingResourceTypes();
auto resources = resourceManager.getComputingResources();
calcOptimisticCosts(wepe.wep->getStartNode()->as<wep::ActivityWepNode>(), resources);
readySet.insert(readyToExecute.begin(), readyToExecute.end());
scheduleReadySet(wepe, resources, schedulerLock);
}
void PEFTScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) {
//auto resources = resourceManager.getComputingResources();
//scheduleReadySet(wepe, resources, schedulerLock);
}
void PEFTScheduler::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 = (OptimisticCostTableEntry*)e1->getUserData().get();
auto entry2 = (OptimisticCostTableEntry*)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 = (OptimisticCostTableEntry*)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 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);
}
}
}
}
} /* namespace scheduler */
} /* namespace wfr */
src/scheduling/scheduler/random/RandomScheduler.cpp
+1 -1
View File
@@ -60,7 +60,7 @@ void RandomScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::se
auto impl = std::get<1>(selected);
auto res = std::get<2>(selected);
auto pred = std::get<3>(selected);
auto reservation = resourceManager.reserveComputingResource(res, 1);
auto reservation = resourceManager.reserveComputingResource(res, 1, false);
auto sd = std::make_shared<SchedulingDecision>(instance, impl, wepe.wep, reservation);
sd->setPrediction(pred);
instance->setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled);
src/scheduling/scheduler/sufferage/SufferageScheduler.cpp
+176
View File
@@ -0,0 +1,176 @@
#include <wfr/scheduling/scheduler/sufferage/SufferageScheduler.h>
#include <wfr/common/wep/WorkflowExecutionPlan.h>
#include <wfr/common/wep/AbstractWepNode.h>
#include <wfr/common/wep/activities/AtomicWepNode.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/ActivityInstance.h>
#include <wfr/common/wfs/AtomicActivityImplementation.h>
#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
#include <wfr/common/wfs/templates/AtomicActivityTemplate.h>
#include <wfr/common/resource/ResourceManager.h>
#include <wfr/enactment/EnactmentEngine.h>
#include <wfr/common/scheduling/SchedulingDecision.h>
#include <wfr/execution/ExecutionManager.h>
#include <wfr/WorkflowRuntime.h>
#include <wfr/wep/utils/WepSearch.h>
#include <iostream>
#include <set>
#include <queue>
namespace wfr {
namespace scheduling {
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<wep::AtomicWepNode*, SufferageScheduler::ECTEntry> SufferageScheduler::_getNextTaskToSchedule(std::map<wep::AtomicWepNode*, ECTEntry>& estimatedTaskCompletionTimes) {
wep::AtomicWepNode* selectedNode = nullptr;
ECTEntry selectedEntry;
bool firstIteration = true;
for (auto n : estimatedTaskCompletionTimes) {
if (firstIteration) {
selectedNode = n.first;
selectedEntry = n.second;
firstIteration = false;
} else {
if (std::get<2>(n.second) > std::get<2>(selectedEntry)) {
selectedNode = n.first;
selectedEntry = n.second;
}
}
}
return std::make_pair(selectedNode, selectedEntry);
}
void SufferageScheduler::_notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) {
while (!readyToExecute.empty()) {
std::map<wep::AtomicWepNode*, ECTEntry> estimatedTaskCompletionTimes;
auto resourceTypes = resourceManager.getComputingResourceTypes();
// Iterate over all tasks
for (auto n : readyToExecute) {
resource::ComputingResource* selectedResource = nullptr;
double selectedEarliestEstimatedCompletionTime = -1;
wfs::AtomicActivityImplementation* selectedImpl = nullptr;
prediction::ExecutionTimeResult selectedPred;
double selectedEarliestEstimatedCompletionTime2 = -1;
std::map<resource::ComputingResource*, double> estimatedCompletionTimes;
// Iterate over all resource types
for (auto restype : resourceTypes) {
// Get fastest implementation and predicted time for given resource type
wfs::AtomicActivityImplementation* fastestImpl = nullptr;
prediction::ExecutionTimeResult fastestPred;
// Check cache first
bool cacheHit = false;
auto aaType = n->activity.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>().atomicActivityType.get();
auto& innerMap = implementationCache[aaType];
auto cacheEntry = innerMap.find(restype);
if (cacheEntry != innerMap.end()) {
fastestImpl = cacheEntry->second.first;
fastestPred = cacheEntry->second.second;
cacheHit = true;
}
if (!cacheHit) {
// Cache miss
auto& aat = n->activity.baseActivity.baseTemplate.as<wfs::AtomicActivityTemplate>();
for (auto i : aat.atomicActivityType->getImplementations()) {
auto execPred = predictionManager.predictExecutionTime(n->activity, *i, *restype);
if (execPred.isValid() && (!fastestPred.isValid() || execPred < fastestPred)) {
fastestPred = execPred;
fastestImpl = i;
}
}
// store into cache
if (fastestPred.isValid()) {
innerMap.insert({restype, {fastestImpl, fastestPred}});
}
}
if (fastestPred.isValid()) {
// Iterate over all resources and calculate earliest completion time
for (auto res : restype->getResources()) {
auto ect = _getResourceEstimatedAvailableTime(res) + fastestPred.executionTime.getValue();
if (selectedEarliestEstimatedCompletionTime < 0 || ect < selectedEarliestEstimatedCompletionTime) {
selectedEarliestEstimatedCompletionTime2 = selectedEarliestEstimatedCompletionTime;
selectedEarliestEstimatedCompletionTime = ect;
selectedResource = res;
selectedImpl = fastestImpl;
selectedPred = fastestPred;
} else if (selectedEarliestEstimatedCompletionTime2 < 0 || ect < selectedEarliestEstimatedCompletionTime2) {
selectedEarliestEstimatedCompletionTime2 = ect;
}
} // End: Iterate over all resources
}
} // End: Iterate over all resource types
// Calculate sufferage
double sufferage = 0;
if (selectedEarliestEstimatedCompletionTime > 0 && selectedEarliestEstimatedCompletionTime2 > 0) {
sufferage = selectedEarliestEstimatedCompletionTime2 - selectedEarliestEstimatedCompletionTime;
}
//std::cout << "selectedEarliestEstimatedCompletionTime " << n->activity.baseActivity.baseTemplate.activityName << ": " << selectedEarliestEstimatedCompletionTime << std::endl;
//std::cout << "selectedEarliestEstimatedCompletionTime2 " << n->activity.baseActivity.baseTemplate.activityName << ": " << selectedEarliestEstimatedCompletionTime2 << std::endl;
//std::cout << "Sufferage " << n->activity.baseActivity.baseTemplate.activityName << ": " << sufferage << std::endl;
if (selectedResource) {
estimatedTaskCompletionTimes.insert({n, std::make_tuple(selectedResource, selectedImpl, sufferage, selectedPred)});
}
} // End: Iterate over all tasks
// Select next task to schedule
auto selectedTask = _getNextTaskToSchedule(estimatedTaskCompletionTimes);
// Schedule task
if (selectedTask.first) {
auto reservation = resourceManager.reserveComputingResource(std::get<0>(selectedTask.second), 1, false);
auto sd = std::make_shared<SchedulingDecision>(&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<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) {
auto& innerMap = scheduledTasks[sd->resource];
for (auto i = innerMap.begin(); i != innerMap.end(); i++) {
if (*i == sd) {
innerMap.erase(i);
break;
}
}
}
} /* namespace scheduling */
} /* namespace wfr */
src/wep/cmake.inc
+1
View File
@@ -1,4 +1,5 @@
set(wfr_cpp_files "${wfr_cpp_files}"
src/wep/utils/WepDotFileWriter.cpp
src/wep/utils/WepProgressionWriter.cpp
)
src/wep/utils/WepDotFileWriter.cpp
+5 -2
View File
@@ -114,13 +114,16 @@ std::string WepDotFileWriter::getNodeLabel(AbstractWepNode* node) {
break;
case wep::WepNodeType::CompositeStart:
ss << getNodeType(node)
<< " \\n" << node->as<ActivityWepNode>().activity.baseActivity.baseTemplate.activityName
<< " (IID: " << node->as<ActivityWepNode>().activity.instanceID << ")"
<< "\\n<" << getNodeStatus(node) << ">";
break;
case wep::WepNodeType::CompositeEnd:
ss << getNodeType(node);
ss << getNodeType(node)
<< " \\n" << node->as<ActivityWepNode>().activity.baseActivity.baseTemplate.activityName
<< " (IID: " << node->as<ActivityWepNode>().activity.instanceID << ")";
break;
case wep::WepNodeType::IterationStart:
ss << "Iteration " << node->as<IterationStartWepNode>().iterationID
<< "\\n<" << getNodeStatus(node) << ">";
break;
src/wep/utils/WepProgressionWriter.cpp
+75
View File
@@ -0,0 +1,75 @@
#include <wfr/wep/utils/WepProgressionWriter.h>
#include <wfr/common/wep/AbstractWepNode.h>
#include <wfr/common/wep/ActivityWepNode.h>
#include <wfr/common/wep/activities/IterationStartWepNode.h>
#include <wfr/common/wep/activities/IterationEndWepNode.h>
#include <wfr/common/wep/WorkflowExecutionPlan.h>
#include <wfr/common/wfs/Workflow.h>
#include <wfr/common/wfs/WorkflowTemplate.h>
#include <wfr/common/wfs/Activity.h>
#include <wfr/common/wfs/ActivityTemplate.h>
#include <wfr/common/wfs/ActivityInstance.h>
#include <wfr/WorkflowRuntime.h>
#include <boost/format.hpp>
#include <iostream>
#include <cassert>
#include <sstream>
#include <cerrno>
#include <cstring>
namespace wfr {
namespace wep {
namespace utils {
WepProgressionWriter::WepProgressionWriter(const std::shared_ptr<WorkflowExecutionPlan>& wep) noexcept
: wep(wep), workflowName(wep->workflow.baseTemplate.workflowName) {
assert(wep);
}
void WepProgressionWriter::openFile(const std::string filename) {
if (filename.empty()) {
outfile.open(workflowName + ".progress");
} else {
outfile.open(filename);
}
if (!outfile.good()) {
std::cerr << "Could not open progress file." << std::endl;
}
epoch = wep->workflow.getStartTime().time_since_epoch().count();
auto reltime = std::chrono::system_clock::now().time_since_epoch().count() - epoch;
outfile << wep->getVersion() << "; " << reltime << "; Start; "
<< wep->workflow.getCreatedAtomicInstanceCount() + wep->workflow.getCreatedCompositeInstanceCount() << "; "
<< wep->workflow.getCreatedAtomicInstanceCount() << "; " << wep->workflow.getCreatedCompositeInstanceCount()
<< std::endl;
}
void WepProgressionWriter::writeProgressToFile() {
if (outfile.is_open()) {
auto reltime = std::chrono::system_clock::now().time_since_epoch().count() - epoch;
outfile << wep->getVersion() << "; " << reltime << "; Update; "
<< wep->workflow.getCreatedAtomicInstanceCount() + wep->workflow.getCreatedCompositeInstanceCount() << "; "
<< wep->workflow.getCreatedAtomicInstanceCount() << "; " << wep->workflow.getCreatedCompositeInstanceCount()
<< std::endl;
}
}
void WepProgressionWriter::closeFile() {
if (outfile.is_open()) {
auto reltime = std::chrono::system_clock::now().time_since_epoch().count() - epoch;
outfile << wep->getVersion() << "; " << reltime << "; Finished; "
<< wep->workflow.getCreatedAtomicInstanceCount() + wep->workflow.getCreatedCompositeInstanceCount() << "; "
<< wep->workflow.getCreatedAtomicInstanceCount() << "; " << wep->workflow.getCreatedCompositeInstanceCount()
<< std::endl;
outfile.flush();
outfile.close();
}
}
} /* namespace utils */
} /* namespace wep */
} /* namespace wfr */