Initial commit

src/scheduling/AbstractScheduler.cpp

@@ -0,0 +1,108 @@
+
+#include "wfr/scheduling/AbstractScheduler.h"
+#include <wfr/prediction/PredictionManager.h>
+#include <wfr/common/resource/ResourceManager.h>
+#include <wfr/WorkflowRuntime.h>
+#include <wfr/common/wep/WorkflowExecutionPlan.h>
+#include <wfr/common/wep/AbstractWepNode.h>
+#include <wfr/common/wep/activities/AtomicWepNode.h>
+#include <wfr/common/wfs/ActivityInstance.h>
+#include <wfr/common/wfs/AtomicActivityImplementation.h>
+#include <wfr/common/wfs/instances/AtomicActivityInstance.h>
+#include <wfr/enactment/EnactmentEngine.h>
+#include <wfr/common/scheduling/SchedulingDecision.h>
+#include <wfr/execution/ExecutionManager.h>
+#include <wfr/wep/utils/WepSearch.h>
+#include <iostream>
+#include <set>
+#include <queue>
+
+
+namespace wfr {
+namespace scheduling {
+
+SchedulerBase::SchedulerBase() : simulationMode(WorkflowRuntime::getGlobalConfig<bool>("sim-mode")) {}
+
+
+SchedulerBase::~SchedulerBase() noexcept {
+	wepMap.clear();
+}
+
+
+void SchedulerBase::registerWEP(wep::WorkflowExecutionPlan& wep, enactment::EnactmentEngine& ee) noexcept {
+	{
+		std::unique_lock<std::mutex> objLock(schedulerMutex);
+		wepMap.emplace(std::make_pair(&wep, WepMapEntry(&ee, &wep)));
+		_registerWEP(wepMap.find(&wep)->second, objLock);
+	}
+}
+
+
+void SchedulerBase::unregisterWEP(wep::WorkflowExecutionPlan& wep) noexcept {
+	std::unique_lock<std::mutex> objLock(schedulerMutex);
+	auto wepe = wepMap.find(&wep);
+	if (wepe != wepMap.end()) {
+		// Wait till all pending scheduling decisions have been handled
+		while (!wepe->second.decisions.empty()) {
+			objLock.unlock();
+			std::this_thread::yield();
+			objLock.lock();
+		}
+		_unregisterWEP(wepe->second, objLock);
+		wepMap.erase(wepe);
+	}
+}
+
+
+void SchedulerBase::notifyWEPUpdate(wep::WorkflowExecutionPlan& wep, std::set<wep::AtomicWepNode*> readyToExecute) noexcept {
+	std::unique_lock<std::mutex> objLock(schedulerMutex);
+	auto wepe = wepMap.find(&wep);
+	if (wepe != wepMap.end()) {
+		_notifyWEPUpdate(wepe->second, readyToExecute, objLock);
+	}
+}
+
+
+bool SchedulerBase::executeSchedulingDecision(WepMapEntry& wepe, std::shared_ptr<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) noexcept {
+	wepe.decisions.insert(sd);
+	schedulerLock.unlock();
+	sd->activityInstance->setSchedulingDecision(sd);
+	if (sd->activityInstance->activityType() == wfs::ActivityType::Atomic) {
+		sd->activityInstance->as<wfs::AtomicActivityInstance>().setOwnedByExecutionLayer(true);
+	}
+	if (sd->getPrediction().isValid()) {
+		cummulativePredictionTime += sd->getPrediction().executionTime.getValue();
+	}
+	bool retval = execution::ExecutionManager::getInstance().executeSchedulingDecision(sd, *this);
+	if (!retval) {
+		// Some error happened (Notify enactment engine)
+		notifySchedulingDecisionStatusUpdate(sd);
+		sd->activityInstance->setSchedulingDecision();
+		wepe.decisions.erase(sd);
+	}
+	schedulerLock.lock();
+	return retval;
+}
+
+
+void SchedulerBase::notifySchedulingDecisionStatusUpdate(std::shared_ptr<SchedulingDecision> sd) noexcept {
+	{
+		std::unique_lock<std::mutex> objLock(schedulerMutex);
+		auto wepe = wepMap.find(sd->wep);
+		if (wepe != wepMap.end()) {
+			_notifySchedulingDecisionStatusUpdate(wepe->second, sd, objLock);
+			// Notify enactment engine
+			auto ee = wepe->second.engine;
+			ee->notifyActivityInstanceStateUpdate(*sd->activityInstance, *this);
+			wepe->second.decisions.erase(sd);
+			sd->activityInstance->setSchedulingDecision();
+		}
+	}
+}
+
+
+} /* namespace scheduling */
+} /* namespace wfr */
+
+
+