WFR-lib/include/wfr/scheduling/AbstractScheduler.h

#pragma once


#include <wfr/common/resource/Resource.h>
#include <wfr/common/types/WfrTypes.h>
#include <mutex>
#include <thread>
#include <memory>
#include <vector>
#include <queue>
#include <map>
#include <set>
#include <condition_variable>

namespace wfr {

// Forward declarations
namespace wep {
class WorkflowExecutionPlan;
class AtomicWepNode;
}
namespace enactment {
class EnactmentEngine;
}
namespace execution {
class ExecutionManager;
}
namespace prediction {
class PredictionManager;
}
namespace resource {
class ResourceManager;
}


// Main namespace
namespace scheduling {

// Forward declaration
class SchedulingDecision;


/*
 * Represents the interface every scheduler needs to implement.
 *
 * Workflow execution sequence:
 *   1) registerWEP() is called by enactment engine to register new workflow instance
 *   2) notifyWEPUpdate() is called by enactment engine to notify scheduler of workflow progress
 *   3) The scheduler creates scheduling decisions objects and passes them to the execution layer
 *        - uses convenience function SchedulerBase::executeSchedulingDecision, or
 *        - directly calls ExecutionManager::executeSchedulingDecision
 *   4) The execution layer calls notifySchedulingDecisionStatusUpdate to notify scheduler about executed (or failed) activities.
 *        - the scheduler needs then to call EnactmentEngine::notifyActivityInstanceStateUpdate() to notify the enactment engine.
 *        - SchedulerBase already automatically notifies the enactment engine.
 *   5) Goto 2) till unregisterWEP() is called.
 */
class AbstractScheduler {
public:
	virtual ~AbstractScheduler() noexcept {}

	/*
	 * Is called by the enactment engine to let the scheduler know of a new workflow instance.
	 *
	 * @param wep workflow execution plan object representing the workflow instance
	 * @param ee reference to enactment engine
	 */
	virtual void registerWEP(wep::WorkflowExecutionPlan& wep, enactment::EnactmentEngine& ee) noexcept = 0;

	/*
	 * Is called by the enactment engine to let the scheduler know that a workflow instance has finished
	 *
	 * @param wep workflow execution plan object representing the workflow instance
	 */
	virtual void unregisterWEP(wep::WorkflowExecutionPlan& wep) noexcept = 0;

	/*
	 * Is called by the enactment engine to let the scheduler know about changes to the workflow execution plan.
	 *
	 * @param wep workflow execution plan object representing the workflow instance
	 * @param readytoExecute set of new ready-to-execute activities (for convenience)
	 */
	virtual void notifyWEPUpdate(wep::WorkflowExecutionPlan& wep, std::set<wep::AtomicWepNode*> readyToExecute) noexcept = 0;

	/*
	 * Called by the execution layer to notify the scheduler that an activity finished (or failed) execution.
	 *
	 * @param sd reference to scheduling decision object.
	 */
	virtual void notifySchedulingDecisionStatusUpdate(std::shared_ptr<SchedulingDecision> sd) noexcept = 0;

	virtual double getCummulativePredictionTime() noexcept = 0;
};



/*
 * Scheduler base class implementing common scheduler functionality.
 */
class SchedulerBase : public AbstractScheduler {
public:
	struct WepMapEntry {
		WepMapEntry(enactment::EnactmentEngine* engine, wep::WorkflowExecutionPlan* wep)
				: engine(engine), wep(wep) {}
		enactment::EnactmentEngine* engine;
		wep::WorkflowExecutionPlan* wep;
		std::set<std::shared_ptr<SchedulingDecision>> decisions;
		void* userData = nullptr;
	};
protected:
	typedef std::map<wep::WorkflowExecutionPlan*, WepMapEntry> wepMap_t;
	std::mutex schedulerMutex;
	wepMap_t wepMap;
	bool simulationMode;
	double cummulativePredictionTime = 0;
	std::map<resource::ComputingResource*, std::vector<std::shared_ptr<SchedulingDecision>>> scheduledTasks;

protected:
	// Hook functions for AbstractScheduler functions below to allow customisation of common functionality
	virtual void _registerWEP(WepMapEntry& wepe, std::unique_lock<std::mutex>& schedulerLock) {};
	virtual void _unregisterWEP(WepMapEntry& wepe, std::unique_lock<std::mutex>& schedulerLock) {}
	virtual void _notifyWEPUpdate(WepMapEntry& wepe, std::set<wep::AtomicWepNode*> readyToExecute, std::unique_lock<std::mutex>& schedulerLock) {}
	virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) {}

	virtual double _getResourceEstimatedAvailableTime(resource::ComputingResource* resource);

	// Convenience functions for passing a scheduling decision to the execution layer.
	bool executeSchedulingDecision(WepMapEntry& wepe, std::shared_ptr<SchedulingDecision> sd, std::unique_lock<std::mutex>& schedulerLock) noexcept;

public:
	SchedulerBase();
	virtual ~SchedulerBase() noexcept;

	// From AbstractScheduler class
	virtual void registerWEP(wep::WorkflowExecutionPlan& wep, enactment::EnactmentEngine& ee) noexcept override;
	virtual void unregisterWEP(wep::WorkflowExecutionPlan& wep) noexcept override;
	virtual void notifyWEPUpdate(wep::WorkflowExecutionPlan& wep, std::set<wep::AtomicWepNode*> readyToExecute) noexcept override;
	virtual void notifySchedulingDecisionStatusUpdate(std::shared_ptr<SchedulingDecision> wep) noexcept override;
	virtual double getCummulativePredictionTime() noexcept override { return cummulativePredictionTime; }
};


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