WMS.h

 
#ifndef WRENCH_WMS_H
#define WRENCH_WMS_H
 
#include <wrench/services/metering/EnergyMeterService.h>
#include <wrench/services/metering/BandwidthMeterService.h>
#include "wrench/services/Service.h"
#include "wrench/wms/DynamicOptimization.h"
#include "wrench/wms/StaticOptimization.h"
#include "wrench/wms/scheduler/PilotJobScheduler.h"
#include "wrench/wms/scheduler/StandardJobScheduler.h"
#include "wrench/services/compute/cloud/CloudComputeService.h"
#include "wrench/workflow/execution_events/StandardJobCompletedEvent.h"
#include "wrench/workflow/execution_events/StandardJobFailedEvent.h"
#include "wrench/workflow/execution_events/PilotJobStartedEvent.h"
#include "wrench/workflow/execution_events/PilotJobExpiredEvent.h"
#include "wrench/workflow/execution_events/FileCopyCompletedEvent.h"
#include "wrench/workflow/execution_events/FileCopyFailedEvent.h"
#include "wrench/workflow/execution_events/TimerEvent.h"
#include "wrench/workflow/Workflow.h"
 
namespace wrench {
 
    class Simulation;
    class ComputeService;
    class CloudComputeService;
    class VirtualizedClusterComputeService;
    class StorageService;
    class NetworkProximityService;
    class FileRegistryService;
 
    class WMS : public Service {
 
    public:
        void addWorkflow(Workflow *workflow, double start_time = 0);
        Workflow *getWorkflow();
        PilotJobScheduler *getPilotJobScheduler();
        StandardJobScheduler *getStandardJobScheduler();
 
        void addStaticOptimization(std::unique_ptr<StaticOptimization>);
 
        void addDynamicOptimization(std::unique_ptr<DynamicOptimization>);
 
 
    protected:
 
        /***********************/
        /***********************/
 
        WMS(std::unique_ptr<StandardJobScheduler> standard_job_scheduler,
            std::unique_ptr<PilotJobScheduler> pilot_job_scheduler,
            const std::set<std::shared_ptr<ComputeService>> &compute_services,
            const std::set<std::shared_ptr<StorageService>> &storage_services,
            const std::set<std::shared_ptr<NetworkProximityService>> &network_proximity_services,
            std::shared_ptr<FileRegistryService> file_registry_service,
            const std::string &hostname,
            const std::string suffix);
 
 
        void checkDeferredStart();
 
        void setTimer(double date, std::string message);
 
        std::shared_ptr<JobManager> createJobManager();
        std::shared_ptr<DataMovementManager> createDataMovementManager();
        std::shared_ptr<EnergyMeterService> createEnergyMeter(const std::map<std::string, double> &measurement_periods);
        std::shared_ptr<EnergyMeterService> createEnergyMeter(const std::vector<std::string> &hostnames, double measurement_period);
        std::shared_ptr<BandwidthMeterService> createBandwidthMeter(const std::map<std::string, double> &measurement_periods);
        std::shared_ptr<BandwidthMeterService> createBandwidthMeter(const std::vector<std::string> &linknames, double measurement_period);
 
        void runDynamicOptimizations();
 
        void runStaticOptimizations();
 
        template <class T>
        std::set<std::shared_ptr<T>> getAvailableComputeServices() {
            bool is_cloud = (std::type_index(typeid(T)) == std::type_index(typeid(CloudComputeService)));
            std::set<std::shared_ptr<T>> to_return;
            for (auto const &h : this->compute_services) {
                if (not is_cloud) {
                    auto shared_ptr = std::dynamic_pointer_cast<T>(h);
                    if (shared_ptr) {
                        to_return.insert(shared_ptr);
                    }
                } else {
                    auto shared_ptr_cloud = std::dynamic_pointer_cast<T>(h);
                    auto shared_ptr_vc = std::dynamic_pointer_cast<VirtualizedClusterComputeService>(h);
                    if (shared_ptr_cloud and (not shared_ptr_vc)) {
                        to_return.insert(shared_ptr_cloud);
                    }
                }
            }
            return to_return;
        }
 
        // The template specialization below does not compile with gcc, hence the above method!
        
//        /**
//         * @brief Obtain the list of compute services available to the WMS
//         * @tparam CloudComputeService
//         * @return a set of compute services
//         * @return
//         */
//        template <>
//        std::set<std::shared_ptr<CloudComputeService>> getAvailableComputeServices<CloudComputeService>() {
//            std::set<std::shared_ptr<CloudComputeService>> to_return;
//            for (auto const &h : this->compute_services) {
//                auto shared_ptr_cloud = std::dynamic_pointer_cast<CloudComputeService>(h);
//                auto shared_ptr_vc = std::dynamic_pointer_cast<VirtualizedClusterComputeService>(h);
//                if (shared_ptr_cloud and (not shared_ptr_vc)) {
//                    to_return.insert(shared_ptr_cloud);
//                }
//            }
//            return to_return;
//        }
 
 
        std::set<std::shared_ptr<StorageService>> getAvailableStorageServices();
        std::set<std::shared_ptr<NetworkProximityService>> getAvailableNetworkProximityServices();
        std::shared_ptr<FileRegistryService> getAvailableFileRegistryService();
 
        void waitForAndProcessNextEvent();
        bool waitForAndProcessNextEvent(double timeout);
        std::shared_ptr<WorkflowExecutionEvent>  waitForNextEvent();
        std::shared_ptr<WorkflowExecutionEvent>  waitForNextEvent(double timeout);
        virtual void processEventStandardJobCompletion(std::shared_ptr<StandardJobCompletedEvent>);
 
        virtual void processEventStandardJobFailure(std::shared_ptr<StandardJobFailedEvent>);
 
        virtual void processEventPilotJobStart(std::shared_ptr<PilotJobStartedEvent>);
 
        virtual void processEventPilotJobExpiration(std::shared_ptr<PilotJobExpiredEvent>);
 
        virtual void processEventFileCopyCompletion(std::shared_ptr<FileCopyCompletedEvent>);
 
        virtual void processEventFileCopyFailure(std::shared_ptr<FileCopyFailedEvent>);
 
        virtual void processEventTimer(std::shared_ptr<TimerEvent>);
 
        /***********************/
        /***********************/
 
        /***********************/
        /***********************/
 
    private:
        friend class Simulation;
        friend class DataMovementManager;
        friend class JobManager;
 
        Workflow *workflow;
        double start_time;
        std::set<std::shared_ptr<ComputeService>> compute_services;
        std::set<std::shared_ptr<StorageService>> storage_services;
        std::set<std::shared_ptr<NetworkProximityService>> network_proximity_services;
        std::shared_ptr<FileRegistryService> file_registry_service;
 
        std::shared_ptr<StandardJobScheduler> standard_job_scheduler;
        std::shared_ptr<PilotJobScheduler> pilot_job_scheduler;
 
        std::vector<std::unique_ptr<DynamicOptimization>> dynamic_optimizations;
        std::vector<std::unique_ptr<StaticOptimization>> static_optimizations;
 
        /***********************/
        /***********************/
 
    private:
        virtual int main() = 0;
 
    };
 
};
 
 
#endif //WRENCH_WMS_H