diff --git a/include/wfr/common/eventsystem/Event.h b/include/wfr/common/eventsystem/Event.h index 6a02c14..857adc9 100755 --- a/include/wfr/common/eventsystem/Event.h +++ b/include/wfr/common/eventsystem/Event.h @@ -224,6 +224,7 @@ protected: wfr::wfs::Workflow* workflow; wfr::wfs::ActivityImplementationType deviceType; wfr_id_t deviceID; + wfr_id_t subDeviceId; wfr::wfs::ComputingResourceStatus oldStatus; wfr::wfs::ComputingResourceStatus newStatus; wfr::wfs::ActivityInstance* executedTask; @@ -231,13 +232,13 @@ protected: wfr_size_t deviceSlotCount; public: - ComputingResourceStatusEvent(wfr::wfs::Workflow* wf, wfr::wfs::ActivityImplementationType deviceType, wfr_id_t deviceID, + ComputingResourceStatusEvent(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 = nullptr, wfr::wfs::AtomicActivityImplementation* executedImpl = nullptr, wfr_size_t deviceSlotCount = 0, wfr_timestamp_sc_t timestamp = std::chrono::system_clock::now()) noexcept : Event(EventType::ComputingResourceStatusEvent, timestamp), workflow(wf), deviceType(deviceType), - deviceID(deviceID), oldStatus(oldStatus), newStatus(newStatus), executedTask(executedTask), + deviceID(deviceID), subDeviceId(subDeviceId), oldStatus(oldStatus), newStatus(newStatus), executedTask(executedTask), executedImpl(executedImpl), deviceSlotCount(deviceSlotCount) {} wfr::wfs::ActivityImplementationType getDeviceType() const noexcept { return deviceType; } @@ -246,6 +247,8 @@ public: wfr_id_t getDeviceID() const noexcept { return deviceID; } + wfr_id_t getDeviceSubID() const noexcept { return subDeviceId; } + wfr::wfs::ComputingResourceStatus getNewStatus() const noexcept { return newStatus; } wfr::wfs::ComputingResourceStatus getOldStatus() const noexcept { return oldStatus; } @@ -261,6 +264,7 @@ public: static std::shared_ptr 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 = nullptr, diff --git a/include/wfr/common/resource/Resource.h b/include/wfr/common/resource/Resource.h index a502d57..037ebc7 100755 --- a/include/wfr/common/resource/Resource.h +++ b/include/wfr/common/resource/Resource.h @@ -25,11 +25,12 @@ class AbstractResource { public: const ResourceClass resourceClass; const std::string resourceIdentifier; + const wfr_id_t resourceId; // Public Functions public: - AbstractResource( ResourceClass resourceClass, const std::string& resourceIdentifier ) noexcept - : resourceClass(resourceClass), resourceIdentifier(resourceIdentifier) {} + AbstractResource( ResourceClass resourceClass, const std::string& resourceIdentifier, const wfr_id_t resourceId ) noexcept + : resourceClass(resourceClass), resourceIdentifier(resourceIdentifier), resourceId(resourceId) {} virtual ~AbstractResource() {} virtual bool hasCapability(const RequiredResourceCapability& type) const noexcept = 0; @@ -100,8 +101,8 @@ public: // Public Functions public: - ResourceBase(ResourceClass resourceClass, const std::string& resourceIdentifier, RType& resourceType ) noexcept - : AbstractResource(resourceClass, resourceIdentifier), resourceType(resourceType) {} + ResourceBase(ResourceClass resourceClass, const std::string& resourceIdentifier, const wfr_id_t resourceId, RType& resourceType ) noexcept + : AbstractResource(resourceClass, resourceIdentifier, resourceId), resourceType(resourceType) {} virtual bool hasCapability(const RequiredResourceCapability& type) const noexcept { return resourceType.hasCapability(type); } virtual bool hasCapability(wfr::resource::ResourceCapabilityType type) const noexcept { return resourceType.hasCapability(type); } @@ -138,17 +139,20 @@ protected: // Public Functions public: - ComputingResource( const std::string& resourceIdentifier, ComputingResourceType& resourceType ) noexcept - : ResourceBase(ResourceClass::ComputingResource, resourceIdentifier, resourceType) { + ComputingResource( const std::string& resourceIdentifier, const wfr_id_t resourceId, ComputingResourceType& resourceType ) noexcept + : ResourceBase(ResourceClass::ComputingResource, resourceIdentifier, resourceId, resourceType) { resourceType.addResource(this); } wfr_size_t getMaxSlotCount() const noexcept { return resourceType.maxSlots; } wfr_size_t getReservedSlotCount() const noexcept { return reservedSlots; } wfr_size_t getBusySlotCount() const noexcept { return busySlots; } - wfr_size_t getFreeSlotCount() const noexcept { return resourceType.maxSlots - (busySlots + reservedSlots); } + wfr_size_t getFreeSlotCount() const noexcept { + auto nonFreeSlots = busySlots + reservedSlots; + return nonFreeSlots >= resourceType.maxSlots ? 0 : resourceType.maxSlots - nonFreeSlots; + } - bool reserveSlots(const wfr_size_t& slots) noexcept; + bool reserveSlots(const wfr_size_t& slots, bool requireFreeSlots = false) noexcept; bool markReservedSlotsBusy(const wfr_size_t& slots) noexcept; bool markFreeSlotsBusy(const wfr_size_t& slots) noexcept; bool releaseBusySlots(const wfr_size_t& slots) noexcept; diff --git a/include/wfr/common/resource/ResourceManager.h b/include/wfr/common/resource/ResourceManager.h index f8afb68..6a042d1 100755 --- a/include/wfr/common/resource/ResourceManager.h +++ b/include/wfr/common/resource/ResourceManager.h @@ -13,7 +13,13 @@ #include + namespace wfr { + +namespace wfs { +class AtomicActivityInstance; +} + namespace resource { class AbstractResource; @@ -32,6 +38,13 @@ private: std::map computingResourceTypes; std::map computingResources; +private: + static std::atomic _nextResourceId; + + static wfr_id_t getNextResourceId() { + return _nextResourceId.fetch_add(1); + } + // Non-public functions private: ResourceManager(std::string resourceConfigFile, bool cpuAutoConfig, bool printResourceTable); @@ -53,11 +66,12 @@ public: ComputingResource* getComputingResource(std::string identifier) const noexcept; std::vector filterComputingResources(std::vector filter) const noexcept; - ComputingResourceReservation reserveComputingResource(ComputingResource* device, const wfr_size_t& slots = 1) noexcept; + std::shared_ptr reserveComputingResource(ComputingResource* device, const wfr_size_t& slots, bool requireFreeSlots = false) noexcept; bool releaseComputingResources(ComputingResourceReservation* handle) noexcept; wfr_size_t getComputingResourceTypeCount() const noexcept { return computingResourceTypes.size(); } std::vector getComputingResourceTypes() const noexcept; + std::vector getCompatibleComputingResourceTypes(wfs::AtomicActivityInstance* activity) const noexcept; ComputingResourceType* getComputingResourceType(std::string name) const noexcept; void printResourceTable(std::ostream& out) const noexcept; diff --git a/include/wfr/common/scheduling/SchedulingDecision.h b/include/wfr/common/scheduling/SchedulingDecision.h index ecf4e9e..5495cdf 100755 --- a/include/wfr/common/scheduling/SchedulingDecision.h +++ b/include/wfr/common/scheduling/SchedulingDecision.h @@ -25,20 +25,20 @@ public: wfs::AtomicActivityImplementation* const activityImplementation = nullptr; wep::WorkflowExecutionPlan* const wep = nullptr; resource::ComputingResource* const resource = nullptr; - resource::ComputingResourceReservation reservationHandle; wfr_size_t slotCount = 0; // Mutable properties that need to be accessed via getter-/setter-functions. // Some of them are invariant, but this class needs to be copyable, so no const properties :( private: prediction::ExecutionTimeResult prediction; + std::shared_ptr reservationHandle; // Public functions public: SchedulingDecision(wfs::ActivityInstance* activityInstance, wfs::AtomicActivityImplementation* activityImplementation, - wep::WorkflowExecutionPlan* wep, resource::ComputingResourceReservation reservationHandle) noexcept + wep::WorkflowExecutionPlan* wep, std::shared_ptr reservationHandle) noexcept : activityInstance(activityInstance), activityImplementation(activityImplementation), wep(wep), - resource(reservationHandle.resource), reservationHandle(reservationHandle), slotCount(reservationHandle.slotCount) {} + resource(reservationHandle->resource), slotCount(reservationHandle->slotCount), reservationHandle(reservationHandle) {} SchedulingDecision(wfs::ActivityInstance* activityInstance, wfs::AtomicActivityImplementation* activityImplementation, wep::WorkflowExecutionPlan* wep, resource::ComputingResource* resource = nullptr, const wfr_size_t& slotCount = 1) noexcept : activityInstance(activityInstance), activityImplementation(activityImplementation), wep(wep), @@ -47,6 +47,7 @@ public: const prediction::ExecutionTimeResult& getPrediction() const noexcept { return prediction; } void setPrediction(const prediction::ExecutionTimeResult& prediction) noexcept { this->prediction = prediction; } + std::shared_ptr getComputingResourceReservation() noexcept { return reservationHandle; } }; } /* namespace scheduling */ diff --git a/include/wfr/common/wfs/ActivityInstance.h b/include/wfr/common/wfs/ActivityInstance.h index 562873c..9f72d06 100755 --- a/include/wfr/common/wfs/ActivityInstance.h +++ b/include/wfr/common/wfs/ActivityInstance.h @@ -216,7 +216,10 @@ public: bool areAllInputsAvailable() const noexcept { return allInputsAvailable; } - void setAllInputsAvailable(const bool& areAvailable = true) noexcept { allInputsAvailable = areAvailable; } + void setAllInputsAvailable(const bool& areAvailable = true) noexcept { + allInputsAvailable = areAvailable; + dirtyStateFlag = false; + } bool areAllOutputsAvailable() const noexcept { return allOutputsAvailable; } diff --git a/include/wfr/common/wfs/instances/ConditionalActivityInstance.h b/include/wfr/common/wfs/instances/ConditionalActivityInstance.h index e0d852b..470ebd3 100755 --- a/include/wfr/common/wfs/instances/ConditionalActivityInstance.h +++ b/include/wfr/common/wfs/instances/ConditionalActivityInstance.h @@ -28,6 +28,8 @@ public: virtual bool executePreExecutionHooks(bool earlyEvaluation) override; virtual bool executePostExecutionHooks(bool earlyEvaluation) override; + + virtual void notifyChildHasFinished(ActivityInstance& ai) noexcept override; }; } /* namespace wfs */ diff --git a/include/wfr/common/wfs/instances/LoopActivityInstance.h b/include/wfr/common/wfs/instances/LoopActivityInstance.h index a6348ad..a2818d9 100755 --- a/include/wfr/common/wfs/instances/LoopActivityInstance.h +++ b/include/wfr/common/wfs/instances/LoopActivityInstance.h @@ -51,10 +51,12 @@ public: void setFinalised(bool v) noexcept { finalisedFlag = v; } bool hasIterationFinished(wfr_id_t iterationID) const noexcept { - if (iterationID < finishedIterations.size()) - return finishedIterations[iterationID]; - else + if (iterationID < finishedIterations.size()) { + bool retval = finishedIterations[iterationID]; + return retval; + } else { return false; + } } bool isIterationRunning(wfr_id_t iterationID)const noexcept { diff --git a/include/wfr/enactment/EnactmentEngine.h b/include/wfr/enactment/EnactmentEngine.h index 80c0034..f871a78 100755 --- a/include/wfr/enactment/EnactmentEngine.h +++ b/include/wfr/enactment/EnactmentEngine.h @@ -39,6 +39,7 @@ private: std::condition_variable updateCondition; bool enableWepDotFileWriter; + bool enableWepProgressionWriter; bool enableEarlyEvaluation; // Invariant properties that can be accessed directly. diff --git a/include/wfr/enactment/WepBuilder.h b/include/wfr/enactment/WepBuilder.h index 9f0a895..20e043a 100755 --- a/include/wfr/enactment/WepBuilder.h +++ b/include/wfr/enactment/WepBuilder.h @@ -60,11 +60,11 @@ private: private: wepNodeMap_t::mapped_type createWEPnode(wfs::Activity& a, wfs::ActivityInstance& ai, wepNodeMap_t& wepNodeMap); - void visitActivity(wfs::Activity& element, activityInstanceMap_t& instanceMap, wepNodeMap_t& wepNodeMap); + void createActivityInstance(wfs::Activity& element, activityInstanceMap_t& instanceMap, wepNodeMap_t& wepNodeMap); wepNodeMap_t::mapped_type createIterationWepNodes(wfr::wfs::ActivityInstance& ai, wepNodeMap_t::mapped_type& parent, wfr_id_t iterationID); - void handleChildren(wfs::CompositeActivityInstance& composite, wepNodeMap_t::mapped_type& parent, wepNodeMap_t::mapped_type* prevIter = nullptr); + void createChildInstances(wfs::CompositeActivityInstance& composite, wepNodeMap_t::mapped_type& parent, wepNodeMap_t::mapped_type* prevIter = nullptr); // Public functions public: diff --git a/include/wfr/eventsystem/SQLiteLogger.h b/include/wfr/eventsystem/SQLiteLogger.h index b1a5d1a..002fe47 100755 --- a/include/wfr/eventsystem/SQLiteLogger.h +++ b/include/wfr/eventsystem/SQLiteLogger.h @@ -64,15 +64,15 @@ protected: int statProcessedEvents; int statWrittenEvents; - /*typedef std::map> workflowMap_t; + typedef std::map> workflowMap_t; typedef std::map> activityMap_t; typedef std::map> activityInstanceMap_t; workflowMap_t workflowMap; activityMap_t activityMap; - activityInstanceMap_t activityInstanceMap;*/ + activityInstanceMap_t activityInstanceMap; - /*void handleWorkflowStatusEvent(WorkflowStatusEvent &event); + void handleWorkflowStatusEvent(WorkflowStatusEvent &event); void handleActicityStatusEvent(ActivityStatusEvent &event); void handleActicityInstanceStatusEvent(ActivityInstanceStatusEvent &event); void handleComputingResourceStatusEvent(ComputingResourceStatusEvent &event); @@ -80,7 +80,7 @@ protected: std::shared_ptr getWorkflowFromWfsMap(wfs::Workflow& wf); void persistWorkflow(std::shared_ptr& wfp, wfs::Workflow& wf); - std::shared_ptr getActivityFromWfsMap(wfs::Activity& act); + std::shared_ptr getActivityFromWfsMap(const wfs::Activity& act); void persistActivity(std::shared_ptr& ap, wfs::Activity& a); std::shared_ptr getActivityInstanceFromWfsMap(wfs::ActivityInstance& act); void persistActivityInstance(std::shared_ptr& ap, wfs::ActivityInstance& a); @@ -89,7 +89,7 @@ protected: void writeActivityStatus(std::shared_ptr& pA, ActivityStatusEvent &event); void writeActivityInstanceStatus(std::shared_ptr& pA, ActivityInstanceStatusEvent &event); void writeComputingResourceStatus(std::shared_ptr& pwf, std::shared_ptr& pai, ComputingResourceStatusEvent &event); - void writeComponentStatus(std::shared_ptr& pwf, ComponentStatusEvent &event);*/ + void writeComponentStatus(std::shared_ptr& pwf, ComponentStatusEvent &event); public: SQLiteLogger(std::string filename); diff --git a/include/wfr/eventsystem/odb-model/SimpleDBModel.h b/include/wfr/eventsystem/odb-model/SimpleDBModel.h index 6e06b2a..03de563 100755 --- a/include/wfr/eventsystem/odb-model/SimpleDBModel.h +++ b/include/wfr/eventsystem/odb-model/SimpleDBModel.h @@ -477,6 +477,8 @@ class DeviceInstanceStatusEvent { wfr_id_t deviceID; + wfr_id_t deviceSubID; + std::string newStatus; std::string oldStatus; @@ -490,7 +492,7 @@ class DeviceInstanceStatusEvent { wfr_size_t deviceSlotCount; public: - DeviceInstanceStatusEvent() : oid(0), deviceID(0), timestamp(0), executedImplID(0), deviceSlotCount(0) {} + DeviceInstanceStatusEvent() : oid(0), deviceID(0), deviceSubID(0), timestamp(0), executedImplID(0), deviceSlotCount(0) {} const std::string& getNewStatus() { return newStatus; @@ -532,10 +534,18 @@ class DeviceInstanceStatusEvent { return deviceID; } + wfr_id_t getDeviceSubId() { + return deviceSubID; + } + void setDeviceId(wfr_id_t deviceId) { deviceID = deviceId; } + void setDeviceSubId(wfr_id_t deviceSubId) { + deviceSubID = deviceSubId; + } + wfr_size_t getDeviceSlotCount() { return deviceSlotCount; } diff --git a/include/wfr/execution/backends/pthreads/PThreadBackend.h b/include/wfr/execution/backends/pthreads/PThreadBackend.h index e01cf74..5f00bf8 100755 --- a/include/wfr/execution/backends/pthreads/PThreadBackend.h +++ b/include/wfr/execution/backends/pthreads/PThreadBackend.h @@ -5,6 +5,8 @@ #include #include #include +#include +#include #include namespace wfr { @@ -18,16 +20,18 @@ namespace execution { class ExecutionManager; class ExecutionHandle; +class PThreadBackend; class PThreadWorker; -class PThreadBackend { + +class PThreadDevice { private: std::mutex objMutex; std::condition_variable workerCondVar; - ExecutionManager& manager; + PThreadBackend& backend; std::set workers; std::set idleWorkers; - std::set resources; + resource::ComputingResource* resource; std::queue taskQueue; wfr_size_t maxWorkerCount; wfr_size_t currentTotalWorkerCount = 0; @@ -37,19 +41,47 @@ private: bool isShutdown = false; public: - PThreadBackend(ExecutionManager& manager, const std::set& resources); - virtual ~PThreadBackend(); + PThreadDevice(PThreadBackend& backend, resource::ComputingResource* resource); + virtual ~PThreadDevice(); void notifyExecutionHandleUpdate(ExecutionHandle& handle, PThreadWorker& worker); bool queueTaskExecution(ExecutionHandle& handle); + wfr_id_t getDeviceId(); + ExecutionHandle* popTaskExecution(PThreadWorker& worker); wfr_size_t getBusyWorkerCount() { return currentBusyWorkerCount; } wfr_size_t getTotalSlotCount() { return maxWorkerCount; } + wfr_size_t getFreeSlotCount() { return maxWorkerCount - currentBusyWorkerCount; } +}; + + + +class PThreadBackend { +private: + std::mutex objMutex; + ExecutionManager& manager; + std::map> devices; + +public: + PThreadBackend(ExecutionManager& manager, const std::set& resources); + virtual ~PThreadBackend(); + + void notifyExecutionHandleUpdate(ExecutionHandle& handle, PThreadWorker& worker); + + bool queueTaskExecution(ExecutionHandle& handle); + + wfr_size_t getDeviceCount(); + + wfr_size_t getBusyWorkerCount(); + + wfr_size_t getTotalSlotCount(); + + wfr_size_t getFreeSlotCount(); }; } /* namespace execution */ diff --git a/include/wfr/execution/backends/pthreads/PThreadWorker.h b/include/wfr/execution/backends/pthreads/PThreadWorker.h index b6ca2bc..019f98a 100755 --- a/include/wfr/execution/backends/pthreads/PThreadWorker.h +++ b/include/wfr/execution/backends/pthreads/PThreadWorker.h @@ -21,6 +21,7 @@ class SchedulingDecision; namespace execution { class PThreadBackend; +class PThreadDevice; class ExecutionHandle; class PThreadWorker { @@ -44,13 +45,13 @@ private: bool stopThread; bool threadRunning; bool threadBusy; - PThreadBackend& backend; + PThreadDevice& device; std::mutex threadMutex; std::thread* thread; wfr_size_t usedSlotCount; wfs::ComputingResourceStatus status; - PThreadWorker(PThreadBackend& backend); + PThreadWorker(PThreadDevice& device); public: virtual ~PThreadWorker(); @@ -73,7 +74,7 @@ public: void setStatus(wfs::ComputingResourceStatus status, std::shared_ptr sd = nullptr); - static PThreadWorker* createNewThread(PThreadBackend& backend); + static PThreadWorker* createNewThread(PThreadDevice& device); }; } /* namespace execution */ diff --git a/include/wfr/scheduling/scheduler/maxmin/MaxMinScheduler.h b/include/wfr/scheduling/scheduler/maxmin/MaxMinScheduler.h new file mode 100644 index 0000000..84d0468 --- /dev/null +++ b/include/wfr/scheduling/scheduler/maxmin/MaxMinScheduler.h @@ -0,0 +1,38 @@ + +#pragma once + + +#include +#include +#include + + +namespace wfr { +namespace wep { +class AtomicWepNode; +class WEPnode; +} +namespace wfs { +class AtomicActivityType; +} +} + + +namespace wfr { +namespace scheduling { + +class MaxMinScheduler: public MinMinScheduler { +protected: + virtual std::pair _getNextTaskToSchedule(std::map& estimatedTaskCompletionTimes) override; + + +public: + MaxMinScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) + : MinMinScheduler(resourceManager, predictionManager) {} + + ~MaxMinScheduler() {} +}; + +} /* namespace scheduling */ +} /* namespace wfr */ + diff --git a/include/wfr/scheduling/scheduler/minmin/MinMinScheduler.h b/include/wfr/scheduling/scheduler/minmin/MinMinScheduler.h new file mode 100644 index 0000000..285cee1 --- /dev/null +++ b/include/wfr/scheduling/scheduler/minmin/MinMinScheduler.h @@ -0,0 +1,51 @@ + +#pragma once + + +#include +#include + + +namespace wfr { +namespace wep { +class AtomicWepNode; +class WEPnode; +} +namespace wfs { +class AtomicActivityType; +} +} + + +namespace wfr { +namespace scheduling { + +class MinMinScheduler: public wfr::scheduling::SchedulerBase { +protected: + resource::ResourceManager& resourceManager; + prediction::PredictionManager& predictionManager; + + std::map>> scheduledTasks; + + typedef std::map> ImplementationCacheInnerMap; + std::map implementationCache; + + double _getResourceEstimatedAvailableTime(resource::ComputingResource* resource); + + typedef std::tuple ECTEntry; + virtual std::pair _getNextTaskToSchedule(std::map& estimatedTaskCompletionTimes); + + +public: + MinMinScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) + : resourceManager(resourceManager), predictionManager(predictionManager) {} + + ~MinMinScheduler() {} + + virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock) override; + virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) override; +}; + +} /* namespace scheduling */ +} /* namespace wfr */ + diff --git a/include/wfr/scheduling/scheduler/peft/PEFTScheduler.h b/include/wfr/scheduling/scheduler/peft/PEFTScheduler.h new file mode 100755 index 0000000..547e1c4 --- /dev/null +++ b/include/wfr/scheduling/scheduler/peft/PEFTScheduler.h @@ -0,0 +1,85 @@ + +#pragma once + + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace wfr { + +namespace wfs { +class AtomicActivityInstance; +} +namespace wep { +class AtomicWepNode; +class ActivityWepNode; +class AbstractWepNode; +} +namespace resource { +class ResourceManager; +} +namespace prediction { +class PredictionManager; +} +} + + +namespace wfr { +namespace scheduling { + + +class PEFTScheduler: public scheduling::SchedulerBase { +protected: + resource::ResourceManager& resourceManager; + prediction::PredictionManager& predictionManager; + + struct OptimisticCostTableEntry { + wfr_size_t resourceCount; + double* values; + double rank = 0; + bool finalized = false; + + wfs::AtomicActivityImplementation** fastestImplementations; + prediction::ExecutionTimeResult* fastestExecutionTimes; + + OptimisticCostTableEntry(wfr_size_t resourceCount) : resourceCount(resourceCount), + values(new double[resourceCount]), fastestImplementations(new wfs::AtomicActivityImplementation*[resourceCount]), + fastestExecutionTimes(new prediction::ExecutionTimeResult[resourceCount]) {} + ~OptimisticCostTableEntry() { + delete[] values; + } + }; + + //std::map> optimisticCostTable; + + std::set readySet; + + void calcOptimisticCosts(wep::ActivityWepNode& node, std::vector& resources); + + void scheduleReadySet(WepMapEntry& wepe, std::vector& resources, std::unique_lock& schedulerLock); + +public: + PEFTScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) + : resourceManager(resourceManager), predictionManager(predictionManager) {} + + ~PEFTScheduler() {} + + virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock) override; + virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) override; +}; + +} /* namespace scheduling */ +} /* namespace wfr */ + + diff --git a/include/wfr/scheduling/scheduler/sufferage/SufferageScheduler.h b/include/wfr/scheduling/scheduler/sufferage/SufferageScheduler.h new file mode 100644 index 0000000..04e4b65 --- /dev/null +++ b/include/wfr/scheduling/scheduler/sufferage/SufferageScheduler.h @@ -0,0 +1,51 @@ + +#pragma once + + +#include +#include + + +namespace wfr { +namespace wep { +class AtomicWepNode; +class WEPnode; +} +namespace wfs { +class AtomicActivityType; +} +} + + +namespace wfr { +namespace scheduling { + +class SufferageScheduler: public wfr::scheduling::SchedulerBase { +protected: + resource::ResourceManager& resourceManager; + prediction::PredictionManager& predictionManager; + + std::map>> scheduledTasks; + + typedef std::map> ImplementationCacheInnerMap; + std::map implementationCache; + + double _getResourceEstimatedAvailableTime(resource::ComputingResource* resource); + + typedef std::tuple ECTEntry; + std::pair _getNextTaskToSchedule(std::map& estimatedTaskCompletionTimes); + + +public: + SufferageScheduler(resource::ResourceManager& resourceManager, prediction::PredictionManager& predictionManager) + : resourceManager(resourceManager), predictionManager(predictionManager) {} + + ~SufferageScheduler() {} + + virtual void _notifyWEPUpdate(SchedulerBase::WepMapEntry& wepe, std::set readyToExecute, std::unique_lock& schedulerLock) override; + virtual void _notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) override; +}; + +} /* namespace scheduling */ +} /* namespace wfr */ + diff --git a/include/wfr/wep/utils/WepProgressionWriter.h b/include/wfr/wep/utils/WepProgressionWriter.h new file mode 100644 index 0000000..e211802 --- /dev/null +++ b/include/wfr/wep/utils/WepProgressionWriter.h @@ -0,0 +1,45 @@ + +#pragma once + + +#include +#include +#include +#include + + +namespace wfr { +namespace wep { + +// Forward declarations +class WorkflowExecutionPlan; +class AbstractWepNode; + +// main namespace +namespace utils { + + +class WepProgressionWriter { + +// Invariant properties that can be accessed directly. +public: + const std::shared_ptr wep; + const std::string workflowName; + +private: + std::ofstream outfile; + int64_t epoch = 0; + +public: + WepProgressionWriter(const std::shared_ptr& wep) noexcept; + + void openFile(const std::string filename = std::string()); + void writeProgressToFile(); + void closeFile(); + +}; + + +} /* namespace utils */ +} /* namespace wep */ +} /* namespace wfr */ diff --git a/src/common/resource/Resource.cpp b/src/common/resource/Resource.cpp index 6b3d53e..1fb4192 100755 --- a/src/common/resource/Resource.cpp +++ b/src/common/resource/Resource.cpp @@ -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 */ diff --git a/src/common/resource/ResourceManager.cpp b/src/common/resource/ResourceManager.cpp index 9bbfc16..a2ad68f 100755 --- a/src/common/resource/ResourceManager.cpp +++ b/src/common/resource/ResourceManager.cpp @@ -15,6 +15,11 @@ #include #include #include +#include +#include +#include +#include +#include namespace wfr { namespace resource { @@ -22,6 +27,10 @@ namespace resource { ResourceManager* ResourceManager::singleton = nullptr; + +std::atomic 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 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 ResourceManager::reserveComputingResource(ComputingResource* device, const wfr_size_t& slots, bool requireFreeSlots) noexcept { + if (device->reserveSlots(slots, requireFreeSlots)) { + return std::make_shared(device, slots); } - return ComputingResourceReservation(); + return std::shared_ptr(); } @@ -288,6 +295,29 @@ std::vector ResourceManager::getComputingResourceTypes() } + +std::vector ResourceManager::getCompatibleComputingResourceTypes(wfs::AtomicActivityInstance* activity) const noexcept { + std::vector retval; + for (auto t : computingResourceTypes) { + auto implementations = activity->baseActivity.baseTemplate.as().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; diff --git a/src/common/wfs/ActivityInstance.cpp b/src/common/wfs/ActivityInstance.cpp index 52ca7b1..9398830 100755 --- a/src/common/wfs/ActivityInstance.cpp +++ b/src/common/wfs/ActivityInstance.cpp @@ -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); } diff --git a/src/common/wfs/instances/ConditionalActivityInstance.cpp b/src/common/wfs/instances/ConditionalActivityInstance.cpp index 5c359f5..e4399ac 100755 --- a/src/common/wfs/instances/ConditionalActivityInstance.cpp +++ b/src/common/wfs/instances/ConditionalActivityInstance.cpp @@ -2,6 +2,7 @@ #include #include #include +#include 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().notifyChildHasFinished(*this); + } else { // if there is no parent then it's the toplevel activity + baseActivity.workflow.setFinishFlag(true); + } + } +} + } /* namespace wfs */ } /* namespace wfr */ diff --git a/src/common/wfs/instances/ContainerActivityInstance.cpp b/src/common/wfs/instances/ContainerActivityInstance.cpp index 3fe0aa7..e7a4ee3 100755 --- a/src/common/wfs/instances/ContainerActivityInstance.cpp +++ b/src/common/wfs/instances/ContainerActivityInstance.cpp @@ -15,7 +15,9 @@ bool ContainerActivityInstance::executePreExecutionHooks(bool earlyEvaluation) { } auto& templ = baseActivity.baseTemplate.as(); if (templ.inputFuncHook != nullptr) { - retval = templ.inputFuncHook(*this); + if (earlyEvaluation || allInputsAvailable) { + retval = templ.inputFuncHook(*this); + } } else { allInputsAvailable = true; setInstanceStatus(ActivityInstanceStatus::ReadyForExecution); diff --git a/src/common/wfs/instances/ForLoopActivityInstance.cpp b/src/common/wfs/instances/ForLoopActivityInstance.cpp index 4dff4a5..0b6ffc9 100755 --- a/src/common/wfs/instances/ForLoopActivityInstance.cpp +++ b/src/common/wfs/instances/ForLoopActivityInstance.cpp @@ -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."))); } diff --git a/src/common/wfs/instances/WhileLoopActivityInstance.cpp b/src/common/wfs/instances/WhileLoopActivityInstance.cpp index 4ff6f15..b238386 100755 --- a/src/common/wfs/instances/WhileLoopActivityInstance.cpp +++ b/src/common/wfs/instances/WhileLoopActivityInstance.cpp @@ -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 { diff --git a/src/enactment/EnactmentEngine.cpp b/src/enactment/EnactmentEngine.cpp index 37855e1..897e5d4 100755 --- a/src/enactment/EnactmentEngine.cpp +++ b/src/enactment/EnactmentEngine.cpp @@ -6,11 +6,13 @@ #include #include #include +#include #include #include #include #include #include +#include #include #include #include @@ -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("dotWriterEnabled")), + enableWepProgressionWriter(WorkflowRuntime::getGlobalConfig("wepProgressionWriterEnabled")), enableEarlyEvaluation(WorkflowRuntime::getGlobalConfig("earlyEvaluationEnabled")), workflow(workflow), scheduler(scheduler) {} @@ -97,6 +110,13 @@ void EnactmentEngine::workflowExecutionLoop(lwd::_wfr_lwd* input, lwd::_wfr_lwd* } #endif + + std::shared_ptr progressionWriter; + if (enableWepProgressionWriter) { + progressionWriter = std::make_shared(wep); + progressionWriter->openFile(); + } + scheduler.registerWEP(*wep, *this); auto ren = wepBuilder.getReadyToExecuteAtomicNodes(); scheduler.notifyWEPUpdate(*wep, std::set(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); } diff --git a/src/enactment/WepBuilder.cpp b/src/enactment/WepBuilder.cpp index cb0aab9..4170e1f 100755 --- a/src/enactment/WepBuilder.cpp +++ b/src/enactment/WepBuilder.cpp @@ -19,6 +19,7 @@ #include #include #include +#include namespace wfr { @@ -80,7 +81,7 @@ std::shared_ptr 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 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 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& 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& 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& updateI wnmi->second.second->getPredecessors().clear(); parentWepStack.push((*wnmi).second); // Handle Children - handleChildren(node.activity.as(), wnmi->second); + createChildInstances(node.activity.as(), 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& 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(); } } diff --git a/src/eventsystem/Event.cpp b/src/eventsystem/Event.cpp index 46c41a9..e1da8ad 100755 --- a/src/eventsystem/Event.cpp +++ b/src/eventsystem/Event.cpp @@ -185,6 +185,7 @@ std::string ComputingResourceStatusEvent::toString() const noexcept { std::shared_ptr 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::fire wfr_size_t deviceSlotCount, wfr_timestamp_sc_t timestamp, bool noLocking ) noexcept { - auto event = std::make_shared(wf, deviceType, deviceID, oldStatus, newStatus, executedTask, executedImpl, deviceSlotCount, timestamp); + auto event = std::make_shared(wf, deviceType, deviceID, subDeviceId, oldStatus, newStatus, executedTask, executedImpl, deviceSlotCount, timestamp); EventSystem::getInstance().fireEvent(event, noLocking); return event; } diff --git a/src/eventsystem/SQLiteLogger.cpp b/src/eventsystem/SQLiteLogger.cpp index d7a2d87..c03695d 100755 --- a/src/eventsystem/SQLiteLogger.cpp +++ b/src/eventsystem/SQLiteLogger.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -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) noexcept { } -/* + std::shared_ptr SQLiteLogger::getWorkflowFromWfsMap(wfs::Workflow& wf) { std::shared_ptr 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 SQLiteLogger::getWorkflowFromWfsMap(wfs::Wor void SQLiteLogger::persistWorkflow(std::shared_ptr& pWf, wfs::Workflow& wf) { dbHandle->persist(pWf); - workflowMap.insert(workflowMap_t::value_type(wf.getWorkflowID(), pWf)); + workflowMap.insert(workflowMap_t::value_type(wf.workflowID, pWf)); } -std::shared_ptr SQLiteLogger::getActivityFromWfsMap(wfs::Activity& act) { +std::shared_ptr SQLiteLogger::getActivityFromWfsMap(const wfs::Activity& act) { std::shared_ptr 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 SQLiteLogger::getActivityFromWfsMap(wfs::Act void SQLiteLogger::persistActivity(std::shared_ptr& ap, wfs::Activity& a) { dbHandle->persist(ap); - activityMap.insert(activityMap_t::value_type(a.getActivityID(), ap)); + activityMap.insert(activityMap_t::value_type(a.activityID, ap)); } std::shared_ptr SQLiteLogger::getActivityInstanceFromWfsMap(wfs::ActivityInstance& act) { std::shared_ptr 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 SQLiteLogger::getActivityInstanceFro void SQLiteLogger::persistActivityInstance(std::shared_ptr& ap, wfs::ActivityInstance& a) { dbHandle->persist(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(uuid); - std::shared_ptr pWf(new odbmodel::Workflow(wf.getWorkflowID(), tmp)); - pWf->setName(wf.getBaseTemplate().getWorkflowName()); + std::shared_ptr 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& pWf, std::shared_ptr 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 ns_t; - ns_t ns = std::chrono::time_point_cast(event.getTimestamp()); + ns_t ns = std::chrono::time_point_cast(event.timestamp); pWSE->setTimestamp(ns.time_since_epoch().count()); dbHandle->persist(pWSE); @@ -280,19 +281,19 @@ void SQLiteLogger::writeWorkflowStatus(std::shared_ptr& 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 pA(new odbmodel::Activity(a.getActivityID(), a.getBaseTemplate().getBranchID())); - pA->setName(a.getBaseTemplate().getActivityName()); - pA->setType(EventMemberUtils::toString(a.getActivityType())); + std::shared_ptr 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& pA, std::shared_ptr 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 ns_t; - ns_t ns = std::chrono::time_point_cast(event.getTimestamp()); + ns_t ns = std::chrono::time_point_cast(event.timestamp); pASE->setTimestamp(ns.time_since_epoch().count()); dbHandle->persist(pASE); @@ -350,21 +351,21 @@ void SQLiteLogger::writeActivityStatus(std::shared_ptr& 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 pAI(new odbmodel::ActivityInstance(ai.getInstanceID(), ai.getIterationID())); + std::shared_ptr 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 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 ns_t; - ns_t ns = std::chrono::time_point_cast(event.getTimestamp()); + ns_t ns = std::chrono::time_point_cast(event.timestamp); pASE->setTimestamp(ns.time_since_epoch().count()); dbHandle->persist(pASE); } -void SQLiteLogger::handleDeviceInstanceStatusEvent(DeviceInstanceStatusEvent &event) { +void SQLiteLogger::handleComputingResourceStatusEvent(ComputingResourceStatusEvent &event) { auto ai = event.getExecutedTask(); auto wf = event.getWorkflow(); std::shared_ptr 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& pwf, std::shared_ptr& pai, DeviceInstanceStatusEvent &event) { +void SQLiteLogger::writeComputingResourceStatus(std::shared_ptr& pwf, std::shared_ptr& pai, ComputingResourceStatusEvent &event) { std::shared_ptr 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 ns_t; - ns_t ns = std::chrono::time_point_cast(event.getTimestamp()); + ns_t ns = std::chrono::time_point_cast(event.timestamp); pDSE->setTimestamp(ns.time_since_epoch().count()); dbHandle->persist(pDSE); @@ -466,18 +467,18 @@ void SQLiteLogger::writeComponentStatus(std::shared_ptr& pwf std::shared_ptr 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 ns_t; - ns_t ns = std::chrono::time_point_cast(event.getTimestamp()); + ns_t ns = std::chrono::time_point_cast(event.timestamp); pCSE->setTimestamp(ns.time_since_epoch().count()); dbHandle->persist(pCSE); } -*/ + } /* namespace eventsystem */ } /* namespace wfr */ diff --git a/src/execution/ExecutionManager.cpp b/src/execution/ExecutionManager.cpp index 9aeeb6e..260f1bb 100755 --- a/src/execution/ExecutionManager.cpp +++ b/src/execution/ExecutionManager.cpp @@ -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()); } } diff --git a/src/execution/backends/pthreads/PThreadBackend.cpp b/src/execution/backends/pthreads/PThreadBackend.cpp index 1f29bc2..4d245ae 100755 --- a/src/execution/backends/pthreads/PThreadBackend.cpp +++ b/src/execution/backends/pthreads/PThreadBackend.cpp @@ -14,14 +14,11 @@ namespace wfr { namespace execution { -PThreadBackend::PThreadBackend(ExecutionManager& manager, const std::set& 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 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 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 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& resources) + : manager(manager) { + for (auto dev : resources) { + auto device = std::make_shared(*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 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 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 */ diff --git a/src/execution/backends/pthreads/PThreadWorker.cpp b/src/execution/backends/pthreads/PThreadWorker.cpp index 97c3c77..2119e13 100755 --- a/src/execution/backends/pthreads/PThreadWorker.cpp +++ b/src/execution/backends/pthreads/PThreadWorker.cpp @@ -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().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 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; } diff --git a/src/prediction/services/oracle/OraclePredictionService.cpp b/src/prediction/services/oracle/OraclePredictionService.cpp index 0c933b3..0f57a9a 100755 --- a/src/prediction/services/oracle/OraclePredictionService.cpp +++ b/src/prediction/services/oracle/OraclePredictionService.cpp @@ -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(prop.second); - std::cout << "Exectime: " << devname << " = " << exectime << std::endl; + //std::cout << "Exectime: " << devname << " = " << exectime << std::endl; return ExecutionTimeResult(&instance, &implementation, &resourceType, NumericalMeasurement(exectime, 0)); } } diff --git a/src/scheduling/AbstractScheduler.cpp b/src/scheduling/AbstractScheduler.cpp index cea4a5f..10d89cc 100755 --- a/src/scheduling/AbstractScheduler.cpp +++ b/src/scheduling/AbstractScheduler.cpp @@ -7,6 +7,8 @@ #include #include #include +#include +#include #include #include #include diff --git a/src/scheduling/SchedulerManager.cpp b/src/scheduling/SchedulerManager.cpp index de89c4a..5408f78 100755 --- a/src/scheduling/SchedulerManager.cpp +++ b/src/scheduling/SchedulerManager.cpp @@ -11,6 +11,10 @@ #include #include #include +#include +#include +#include +#include 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()); } diff --git a/src/scheduling/cmake.inc b/src/scheduling/cmake.inc index 5e3738c..90a4bf5 100755 --- a/src/scheduling/cmake.inc +++ b/src/scheduling/cmake.inc @@ -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 ) diff --git a/src/scheduling/scheduler/heft/HEFTScheduler.cpp b/src/scheduling/scheduler/heft/HEFTScheduler.cpp index 323d39f..4fa2b4a 100755 --- a/src/scheduling/scheduler/heft/HEFTScheduler.cpp +++ b/src/scheduling/scheduler/heft/HEFTScheduler.cpp @@ -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 sd = std::make_shared(e.instance, fastestImpl, wepe.wep, reservation); sd->setPrediction(prediction); e.instance->setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled); diff --git a/src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp b/src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp index cbdc9f1..a178f2b 100755 --- a/src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp +++ b/src/scheduling/scheduler/justintime/JustInTimeScheduler.cpp @@ -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(&ai, fastestImpl, wepe.wep, reservation); sd->setPrediction(fastestPred); ai.setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled); diff --git a/src/scheduling/scheduler/maxmin/MaxMinScheduler.cpp b/src/scheduling/scheduler/maxmin/MaxMinScheduler.cpp new file mode 100644 index 0000000..c75408e --- /dev/null +++ b/src/scheduling/scheduler/maxmin/MaxMinScheduler.cpp @@ -0,0 +1,50 @@ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace wfr { +namespace scheduling { + + + + +std::pair MaxMinScheduler::_getNextTaskToSchedule(std::map& 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 */ diff --git a/src/scheduling/scheduler/minmin/MinMinScheduler.cpp b/src/scheduling/scheduler/minmin/MinMinScheduler.cpp new file mode 100644 index 0000000..7e1ac19 --- /dev/null +++ b/src/scheduling/scheduler/minmin/MinMinScheduler.cpp @@ -0,0 +1,164 @@ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +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 MinMinScheduler::_getNextTaskToSchedule(std::map& 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 readyToExecute, std::unique_lock& schedulerLock) { + while (!readyToExecute.empty()) { + std::map 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 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().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(); + 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(&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 sd, std::unique_lock& 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 */ diff --git a/src/scheduling/scheduler/peft/PEFTScheduler.cpp b/src/scheduling/scheduler/peft/PEFTScheduler.cpp new file mode 100755 index 0000000..425ee0d --- /dev/null +++ b/src/scheduling/scheduler/peft/PEFTScheduler.cpp @@ -0,0 +1,225 @@ + + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace wfr { +namespace scheduling { + + +void PEFTScheduler::calcOptimisticCosts(wep::ActivityWepNode& node, std::vector& 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().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(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(); + /*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 readyToExecute, std::unique_lock& schedulerLock) { + auto resourceTypes = resourceManager.getComputingResourceTypes(); + auto resources = resourceManager.getComputingResources(); + + calcOptimisticCosts(wepe.wep->getStartNode()->as(), resources); + + readySet.insert(readyToExecute.begin(), readyToExecute.end()); + + scheduleReadySet(wepe, resources, schedulerLock); +} + +void PEFTScheduler::_notifySchedulingDecisionStatusUpdate(WepMapEntry& wepe, std::shared_ptr sd, std::unique_lock& schedulerLock) { + //auto resources = resourceManager.getComputingResources(); + //scheduleReadySet(wepe, resources, schedulerLock); +} + + +void PEFTScheduler::scheduleReadySet(WepMapEntry& wepe, std::vector& resources, std::unique_lock& 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, _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 sd = std::make_shared(&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 */ + + diff --git a/src/scheduling/scheduler/random/RandomScheduler.cpp b/src/scheduling/scheduler/random/RandomScheduler.cpp index 2edbe9b..921e12b 100755 --- a/src/scheduling/scheduler/random/RandomScheduler.cpp +++ b/src/scheduling/scheduler/random/RandomScheduler.cpp @@ -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(instance, impl, wepe.wep, reservation); sd->setPrediction(pred); instance->setInstanceStatus(wfs::ActivityInstanceStatus::Scheduled); diff --git a/src/scheduling/scheduler/sufferage/SufferageScheduler.cpp b/src/scheduling/scheduler/sufferage/SufferageScheduler.cpp new file mode 100644 index 0000000..46b6f28 --- /dev/null +++ b/src/scheduling/scheduler/sufferage/SufferageScheduler.cpp @@ -0,0 +1,176 @@ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +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 SufferageScheduler::_getNextTaskToSchedule(std::map& 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 readyToExecute, std::unique_lock& schedulerLock) { + while (!readyToExecute.empty()) { + std::map 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 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().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(); + 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(&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 sd, std::unique_lock& 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 */ diff --git a/src/wep/cmake.inc b/src/wep/cmake.inc index 07ee24e..db44e30 100755 --- a/src/wep/cmake.inc +++ b/src/wep/cmake.inc @@ -1,4 +1,5 @@ set(wfr_cpp_files "${wfr_cpp_files}" src/wep/utils/WepDotFileWriter.cpp + src/wep/utils/WepProgressionWriter.cpp ) diff --git a/src/wep/utils/WepDotFileWriter.cpp b/src/wep/utils/WepDotFileWriter.cpp index 7a9572e..155f85f 100755 --- a/src/wep/utils/WepDotFileWriter.cpp +++ b/src/wep/utils/WepDotFileWriter.cpp @@ -114,13 +114,16 @@ std::string WepDotFileWriter::getNodeLabel(AbstractWepNode* node) { break; case wep::WepNodeType::CompositeStart: ss << getNodeType(node) + << " \\n" << node->as().activity.baseActivity.baseTemplate.activityName + << " (IID: " << node->as().activity.instanceID << ")" << "\\n<" << getNodeStatus(node) << ">"; break; case wep::WepNodeType::CompositeEnd: - ss << getNodeType(node); + ss << getNodeType(node) + << " \\n" << node->as().activity.baseActivity.baseTemplate.activityName + << " (IID: " << node->as().activity.instanceID << ")"; break; case wep::WepNodeType::IterationStart: - ss << "Iteration " << node->as().iterationID << "\\n<" << getNodeStatus(node) << ">"; break; diff --git a/src/wep/utils/WepProgressionWriter.cpp b/src/wep/utils/WepProgressionWriter.cpp new file mode 100644 index 0000000..cf1d184 --- /dev/null +++ b/src/wep/utils/WepProgressionWriter.cpp @@ -0,0 +1,75 @@ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +namespace wfr { +namespace wep { +namespace utils { + + +WepProgressionWriter::WepProgressionWriter(const std::shared_ptr& 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 */