BareMetalComputeService.h

 
#ifndef WRENCH_BAREMETALCOMPUTESERVICE_H
#define WRENCH_BAREMETALCOMPUTESERVICE_H
 
 
#include <queue>
 
#include "wrench/services/compute/ComputeService.h"
#include "BareMetalComputeServiceProperty.h"
#include "BareMetalComputeServiceMessagePayload.h"
#include "wrench/services/helper_services/host_state_change_detector/HostStateChangeDetector.h"
 
 
 
namespace wrench {
 
    class Simulation;
    class StorageService;
    class FailureCause;
    class Alarm;
    class Action;
    class ActionExecutionService;
 
 
    class BareMetalComputeService : public ComputeService {
 
        friend class CloudComputeService;
        friend class BatchComputeService;
 
    private:
 
        WRENCH_PROPERTY_COLLECTION_TYPE default_property_values = {
                {BareMetalComputeServiceProperty::TASK_STARTUP_OVERHEAD,                          "0.0"},
                {BareMetalComputeServiceProperty::FAIL_ACTION_AFTER_ACTION_EXECUTOR_CRASH,        "true"},
                {BareMetalComputeServiceProperty::TERMINATE_WHENEVER_ALL_RESOURCES_ARE_DOWN,      "false"},
        };
 
WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE  default_messagepayload_values = {
                {BareMetalComputeServiceMessagePayload::STOP_DAEMON_MESSAGE_PAYLOAD,                    1024},
                {BareMetalComputeServiceMessagePayload::DAEMON_STOPPED_MESSAGE_PAYLOAD,                 1024},
                {BareMetalComputeServiceMessagePayload::JOB_TYPE_NOT_SUPPORTED_MESSAGE_PAYLOAD,         1024},
                {BareMetalComputeServiceMessagePayload::NOT_ENOUGH_CORES_MESSAGE_PAYLOAD,               1024},
                {BareMetalComputeServiceMessagePayload::SUBMIT_STANDARD_JOB_REQUEST_MESSAGE_PAYLOAD,    1024},
                {BareMetalComputeServiceMessagePayload::SUBMIT_STANDARD_JOB_ANSWER_MESSAGE_PAYLOAD,     1024},
                {BareMetalComputeServiceMessagePayload::STANDARD_JOB_DONE_MESSAGE_PAYLOAD,              1024},
                {BareMetalComputeServiceMessagePayload::STANDARD_JOB_FAILED_MESSAGE_PAYLOAD,            1024},
                {BareMetalComputeServiceMessagePayload::TERMINATE_STANDARD_JOB_REQUEST_MESSAGE_PAYLOAD, 1024},
                {BareMetalComputeServiceMessagePayload::TERMINATE_STANDARD_JOB_ANSWER_MESSAGE_PAYLOAD,  1024},
                {BareMetalComputeServiceMessagePayload::SUBMIT_PILOT_JOB_REQUEST_MESSAGE_PAYLOAD,       1024},
                {BareMetalComputeServiceMessagePayload::SUBMIT_PILOT_JOB_ANSWER_MESSAGE_PAYLOAD,        1024},
                {BareMetalComputeServiceMessagePayload::PILOT_JOB_STARTED_MESSAGE_PAYLOAD,              1024},
                {BareMetalComputeServiceMessagePayload::PILOT_JOB_EXPIRED_MESSAGE_PAYLOAD,              1024},
                {BareMetalComputeServiceMessagePayload::PILOT_JOB_FAILED_MESSAGE_PAYLOAD,               1024},
                {BareMetalComputeServiceMessagePayload::TERMINATE_PILOT_JOB_REQUEST_MESSAGE_PAYLOAD,    1024},
                {BareMetalComputeServiceMessagePayload::TERMINATE_PILOT_JOB_ANSWER_MESSAGE_PAYLOAD,     1024},
                {BareMetalComputeServiceMessagePayload::SUBMIT_COMPOUND_JOB_REQUEST_MESSAGE_PAYLOAD,    1024},
                {BareMetalComputeServiceMessagePayload::SUBMIT_COMPOUND_JOB_ANSWER_MESSAGE_PAYLOAD,     1024},
                {BareMetalComputeServiceMessagePayload::COMPOUND_JOB_DONE_MESSAGE_PAYLOAD,              1024},
                {BareMetalComputeServiceMessagePayload::COMPOUND_JOB_FAILED_MESSAGE_PAYLOAD,            1024},
                {BareMetalComputeServiceMessagePayload::TERMINATE_COMPOUND_JOB_REQUEST_MESSAGE_PAYLOAD, 1024},
                {BareMetalComputeServiceMessagePayload::TERMINATE_COMPOUND_JOB_ANSWER_MESSAGE_PAYLOAD,  1024},
                {BareMetalComputeServiceMessagePayload::RESOURCE_DESCRIPTION_REQUEST_MESSAGE_PAYLOAD,     1024},
                {BareMetalComputeServiceMessagePayload::RESOURCE_DESCRIPTION_ANSWER_MESSAGE_PAYLOAD,      1024},
                {BareMetalComputeServiceMessagePayload::IS_THERE_AT_LEAST_ONE_HOST_WITH_AVAILABLE_RESOURCES_REQUEST_MESSAGE_PAYLOAD, 1024},
                {BareMetalComputeServiceMessagePayload::IS_THERE_AT_LEAST_ONE_HOST_WITH_AVAILABLE_RESOURCES_ANSWER_MESSAGE_PAYLOAD, 1024},
        };
 
    public:
 
        // Public Constructor
        BareMetalComputeService(const std::string &hostname,
                                const std::map<std::string, std::tuple<unsigned long, double>> compute_resources,
                                std::string scratch_space_mount_point,
                                WRENCH_PROPERTY_COLLECTION_TYPE property_list = {},
                                WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE messagepayload_list = {}
        );
 
        // Public Constructor
        BareMetalComputeService(const std::string &hostname,
                                const std::vector<std::string> compute_hosts,
                                std::string scratch_space_mount_point,
                                WRENCH_PROPERTY_COLLECTION_TYPE property_list = {},
                                WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE messagepayload_list = {}
        );
 
        virtual bool supportsStandardJobs() override;
        virtual bool supportsCompoundJobs() override;
        virtual bool supportsPilotJobs() override;
 
        /***********************/
        /***********************/
 
        void submitCompoundJob(std::shared_ptr<CompoundJob> job, const std::map<std::string, std::string> &service_specific_args) override;
 
        void terminateCompoundJob(std::shared_ptr<CompoundJob> job) override;
 
        ~BareMetalComputeService();
 
 
    protected:
        friend class JobManager;
 
        void validateServiceSpecificArguments(std::shared_ptr<CompoundJob> job,
                                                      std::map<std::string, std::string> &service_specific_args) override;
 
 
        BareMetalComputeService(const std::string &hostname,
                                std::map<std::string, std::tuple<unsigned long, double>> compute_resources,
                                WRENCH_PROPERTY_COLLECTION_TYPE property_list,
                                WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE messagepayload_list,
                                double ttl,
                                std::shared_ptr<PilotJob> pj, std::string suffix,
                                std::shared_ptr<StorageService> scratch_space); // reference to upper level scratch space
 
        BareMetalComputeService(const std::string &hostname,
                                std::map<std::string, std::tuple<unsigned long, double>> compute_resources,
                                WRENCH_PROPERTY_COLLECTION_TYPE property_list,
                                WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE messagepayload_list,
                                std::shared_ptr<StorageService> scratch_space);
 
        void validateProperties();
 
 
        // Low-level constructor helper method
        void initiateInstance(const std::string &hostname,
                              std::map<std::string, std::tuple<unsigned long, double>> compute_resources,
                              WRENCH_PROPERTY_COLLECTION_TYPE property_list,
                              WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE messagepayload_list,
                              double ttl,
                              std::shared_ptr<PilotJob> pj);
 
 
    protected:
 
        std::shared_ptr<Alarm> death_alarm = nullptr;
        std::shared_ptr<PilotJob> containing_pilot_job; // In case this service is in fact a pilot job
 
 
        std::unordered_map<std::shared_ptr<StandardJob> , std::set<std::shared_ptr<DataFile>>> files_in_scratch;
 
        std::set<std::shared_ptr<CompoundJob>> current_jobs;
 
        std::set<std::shared_ptr<Action>> not_ready_actions;
        std::vector<std::shared_ptr<Action>> ready_actions;
        std::set<std::shared_ptr<Action>> dispatched_actions;
        std::unordered_map<std::shared_ptr<CompoundJob>, int> num_dispatched_actions_for_cjob;
 
        double ttl;
        double death_date;
        bool has_ttl;
 
 
        // Add the scratch files of one standardjob to the list of all the scratch files of all the standard jobs inside the pilot job
//        void storeFilesStoredInScratch(std::set<std::shared_ptr<DataFile>> scratch_files);
 
        // Cleanup the scratch if I am a pilot job
        void cleanUpScratch();
 
        int main() override;
 
        // Helper functions to make main() a bit more palatable
 
        void terminate(bool send_failure_notifications, ComputeService::TerminationCause termination_cause);
 
        void processActionDone(std::shared_ptr<Action> action);
 
        void processCompoundJobTerminationRequest(std::shared_ptr<CompoundJob> job, simgrid::s4u::Mailbox *answer_mailbox);
 
        bool processNextMessage();
 
        void dispatchReadyActions();
 
 
        void terminateCurrentCompoundJob(std::shared_ptr<CompoundJob> job, ComputeService::TerminationCause termination_cause);
 
        void processGetResourceInformation(simgrid::s4u::Mailbox *answer_mailbox, const std::string &key);
 
//        void processSubmitPilotJob(const std::string &answer_mailbox, std::shared_ptr<PilotJob> job, std::map<std::string, std::string> service_specific_args);
 
        void processSubmitCompoundJob(simgrid::s4u::Mailbox *answer_mailbox, std::shared_ptr<CompoundJob> job,
                                      std::map<std::string, std::string> &service_specific_arguments);
 
        void processIsThereAtLeastOneHostWithAvailableResources(
                simgrid::s4u::Mailbox *answer_mailbox, unsigned long num_cores, double ram);
 
//        std::tuple<std::string, unsigned long> pickAllocation(std::shared_ptr<WorkflowTask>task,
//                                                              std::string required_host, unsigned long required_num_cores, double required_ram,
//                                                              std::set<std::string> &hosts_to_avoid);
 
//        bool jobCanRun(std::shared_ptr<StandardJob> job, std::map<std::string, std::string> &service_specific_arguments);
//
//        bool isThereAtLeastOneHostWithResources(unsigned long num_cores, double ram);
 
        void cleanup(bool has_terminated_cleanly, int return_value) override;
 
//        bool areAllComputeResourcesDownWithNoWUERunning();
 
        static std::tuple<std::string, unsigned long> parseResourceSpec(const std::string &spec);
 
 
        int exit_code = 0;
 
        std::shared_ptr<HostStateChangeDetector> host_state_change_monitor;
 
        std::shared_ptr<ActionExecutionService> action_execution_service;
 
        /***********************/
        /***********************/
 
    };
};
 
 
#endif //WRENCH_BAREMETALCOMPUTESERVICE_H