WFR-lib/src/scheduling/AbstractScheduler.cpp

#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/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>
#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);
	scheduledTasks[sd->resource].push_back(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()) {
			auto& innerMap = scheduledTasks[sd->resource];
			for (auto i = innerMap.begin(); i != innerMap.end(); i++) {
				if (*i == sd) {
					innerMap.erase(i);
					break;
				}
			}
			_notifySchedulingDecisionStatusUpdate(wepe->second, sd, objLock);
			// Notify enactment engine
			auto ee = wepe->second.engine;
			ee->notifyActivityInstanceStateUpdate(*sd->activityInstance, *this);
			wepe->second.decisions.erase(sd);
			sd->activityInstance->setSchedulingDecision();
		}
	}
}

double SchedulerBase::_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;
	}
}

} /* namespace scheduling */
} /* namespace wfr */