BatchComputeService.h

 
#ifndef WRENCH_BATCH_SERVICE_H
#define WRENCH_BATCH_SERVICE_H
 
#include "wrench/services/compute/ComputeService.h"
#include "wrench/services/compute/batch/BatchJob.h"
#include "wrench/services/compute/batch/BatschedNetworkListener.h"
#include "wrench/services/compute/batch/BatchComputeServiceProperty.h"
#include "wrench/services/compute/batch/BatchComputeServiceMessagePayload.h"
#include "wrench/services/helper_services/alarm/Alarm.h"
#include "wrench/job/CompoundJob.h"
#include "wrench/job/Job.h"
#include "wrench/services/compute/batch/batch_schedulers/BatchScheduler.h"
 
#include <deque>
#include <queue>
#include <set>
#include <tuple>
 
namespace wrench {
 
    class WorkloadTraceFileReplayer;
    class BareMetalComputeServiceOneShot;
 
    class BatchComputeService : public ComputeService {
 
    private:
 
        WRENCH_PROPERTY_COLLECTION_TYPE default_property_values = {
                {BatchComputeServiceProperty::TASK_STARTUP_OVERHEAD,                     "0"},
                {BatchComputeServiceProperty::HOST_SELECTION_ALGORITHM,                    "FIRSTFIT"},
                {BatchComputeServiceProperty::TASK_SELECTION_ALGORITHM,                    "maximum_flops"},
#ifdef ENABLE_BATSCHED
                {BatchComputeServiceProperty::BATCH_SCHEDULING_ALGORITHM,                  "conservative_bf"},
//                {BatchComputeServiceProperty::BATCH_SCHEDULING_ALGORITHM,                  "easy_bf"},
//                {BatchComputeServiceProperty::BATCH_SCHEDULING_ALGORITHM,                  "easy_bf_fast"},
 
                {BatchComputeServiceProperty::BATCH_QUEUE_ORDERING_ALGORITHM,              "fcfs"},
#else
                {BatchComputeServiceProperty::BATCH_SCHEDULING_ALGORITHM,            "fcfs"},
#endif
                {BatchComputeServiceProperty::BATCH_RJMS_PADDING_DELAY, "5"},
                {BatchComputeServiceProperty::SIMULATED_WORKLOAD_TRACE_FILE,               ""},
                {BatchComputeServiceProperty::USE_REAL_RUNTIMES_AS_REQUESTED_RUNTIMES_IN_WORKLOAD_TRACE_FILE,     "false"},
                {BatchComputeServiceProperty::IGNORE_INVALID_JOBS_IN_WORKLOAD_TRACE_FILE,     "false"},
                {BatchComputeServiceProperty::SUBMIT_TIME_OF_FIRST_JOB_IN_WORKLOAD_TRACE_FILE,                          "-1"},
                {BatchComputeServiceProperty::OUTPUT_CSV_JOB_LOG,                          ""},
                {BatchComputeServiceProperty::SIMULATE_COMPUTATION_AS_SLEEP,               "false"},
                {BatchComputeServiceProperty::BATSCHED_LOGGING_MUTED,                      "true"},
                {BatchComputeServiceProperty::BATSCHED_CONTIGUOUS_ALLOCATION,              "false"}
        };
 
WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE  default_messagepayload_values = {
                {BatchComputeServiceMessagePayload::STOP_DAEMON_MESSAGE_PAYLOAD,                 1024},
                {BatchComputeServiceMessagePayload::RESOURCE_DESCRIPTION_REQUEST_MESSAGE_PAYLOAD,1024},
                {BatchComputeServiceMessagePayload::RESOURCE_DESCRIPTION_ANSWER_MESSAGE_PAYLOAD, 1024},
                {BatchComputeServiceMessagePayload::DAEMON_STOPPED_MESSAGE_PAYLOAD,              1024},
                {BatchComputeServiceMessagePayload::STANDARD_JOB_DONE_MESSAGE_PAYLOAD,           1024},
                {BatchComputeServiceMessagePayload::SUBMIT_STANDARD_JOB_REQUEST_MESSAGE_PAYLOAD, 1024},
                {BatchComputeServiceMessagePayload::SUBMIT_STANDARD_JOB_ANSWER_MESSAGE_PAYLOAD,  1024},
                {BatchComputeServiceMessagePayload::TERMINATE_STANDARD_JOB_REQUEST_MESSAGE_PAYLOAD,  1024},
                {BatchComputeServiceMessagePayload::TERMINATE_STANDARD_JOB_ANSWER_MESSAGE_PAYLOAD,  1024},
                {BatchComputeServiceMessagePayload::SUBMIT_PILOT_JOB_REQUEST_MESSAGE_PAYLOAD,    1024},
                {BatchComputeServiceMessagePayload::SUBMIT_PILOT_JOB_ANSWER_MESSAGE_PAYLOAD,     1024},
                {BatchComputeServiceMessagePayload::STANDARD_JOB_FAILED_MESSAGE_PAYLOAD,         1024},
                {BatchComputeServiceMessagePayload::PILOT_JOB_STARTED_MESSAGE_PAYLOAD,           1024},
                {BatchComputeServiceMessagePayload::PILOT_JOB_EXPIRED_MESSAGE_PAYLOAD,           1024},
                {BatchComputeServiceMessagePayload::TERMINATE_PILOT_JOB_ANSWER_MESSAGE_PAYLOAD,  1024},
                {BatchComputeServiceMessagePayload::TERMINATE_PILOT_JOB_REQUEST_MESSAGE_PAYLOAD, 1024},
                {BatchComputeServiceMessagePayload::SUBMIT_COMPOUND_JOB_REQUEST_MESSAGE_PAYLOAD,    1024},
                {BatchComputeServiceMessagePayload::SUBMIT_COMPOUND_JOB_ANSWER_MESSAGE_PAYLOAD,     1024},
                {BatchComputeServiceMessagePayload::COMPOUND_JOB_DONE_MESSAGE_PAYLOAD,              1024},
                {BatchComputeServiceMessagePayload::COMPOUND_JOB_FAILED_MESSAGE_PAYLOAD,            1024},
                {BatchComputeServiceMessagePayload::TERMINATE_COMPOUND_JOB_REQUEST_MESSAGE_PAYLOAD, 1024},
                {BatchComputeServiceMessagePayload::TERMINATE_COMPOUND_JOB_ANSWER_MESSAGE_PAYLOAD,  1024},
                {BatchComputeServiceMessagePayload::IS_THERE_AT_LEAST_ONE_HOST_WITH_AVAILABLE_RESOURCES_REQUEST_MESSAGE_PAYLOAD, 1024},
                {BatchComputeServiceMessagePayload::IS_THERE_AT_LEAST_ONE_HOST_WITH_AVAILABLE_RESOURCES_ANSWER_MESSAGE_PAYLOAD, 1024},
        };
 
    public:
        BatchComputeService(const std::string &hostname,
                            std::vector<std::string> compute_hosts,
                            std::string scratch_space_mount_point,
                            WRENCH_PROPERTY_COLLECTION_TYPE property_list = {},
                            WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE messagepayload_list = {}
        );
 
        bool supportsStandardJobs() override;
        bool supportsCompoundJobs() override;
        bool supportsPilotJobs() override;
 
        /***********************/
        /***********************/
        std::map<std::string,double> getStartTimeEstimates(std::set<std::tuple<std::string,unsigned long,unsigned long, double>> resources);
 
        std::vector<std::tuple<std::string, std::string, int, int, int, double, double>> getQueue();
 
        /***********************/
        /***********************/
 
        /***********************/
        /***********************/
        ~BatchComputeService() override;
        // helper function
        static unsigned long parseUnsignedLongServiceSpecificArgument(std::string key, const std::map<std::string, std::string> &args);
 
        void validateServiceSpecificArguments(std::shared_ptr<CompoundJob> compound_job,
                                              std::map<std::string, std::string> &service_specific_args) override;
 
        /***********************/
        /***********************/
 
    private:
        friend class WorkloadTraceFileReplayer;
        friend class FCFSBatchScheduler;
        friend class ConservativeBackfillingBatchScheduler;
        friend class ConservativeBackfillingBatchSchedulerCoreLevel;
 
        friend class BatschedBatchScheduler;
 
        BatchComputeService(const std::string hostname,
                            std::vector<std::string> compute_hosts,
                            unsigned long cores_per_host,
                            double ram_per_host,
                            std::string scratch_space_mount_point,
                            WRENCH_PROPERTY_COLLECTION_TYPE property_list,
                            WRENCH_MESSAGE_PAYLOADCOLLECTION_TYPE messagepayload_list,
                            std::string suffix
        );
 
        //submits a standard job
        void submitCompoundJob(std::shared_ptr<CompoundJob> job, const std::map<std::string, std::string> &batch_job_args) override;
 
        // terminate a standard job
        void terminateCompoundJob(std::shared_ptr<CompoundJob> job) override;
 
        std::vector<std::tuple<std::string, double, double, double, double, unsigned int, std::string>> workload_trace;
        std::shared_ptr<WorkloadTraceFileReplayer> workload_trace_replayer;
 
        bool clean_exit = false;
 
        //create alarms for compound jobs
        std::map<std::shared_ptr<CompoundJob>,std::shared_ptr<Alarm>> compound_job_alarms;
 
        /* Resources information in batch_standard_and_pilot_jobs */
        unsigned long total_num_of_nodes;
        unsigned long num_cores_per_node;
        std::map<std::string, unsigned long> nodes_to_cores_map;
        std::vector<double> timeslots;
        std::map<std::string, unsigned long> available_nodes_to_cores;
        std::map<unsigned long, std::string> host_id_to_names;
        std::vector<std::string> compute_hosts;
        /* End Resources information in batch_standard_and_pilot_jobs */
 
        // Vector of one-shot bare-metal compute services
        std::map<std::shared_ptr<CompoundJob>, std::shared_ptr<BareMetalComputeServiceOneShot>> running_bare_metal_one_shot_compute_services;
 
        // Master List of batch_standard_and_pilot_jobs jobs
        std::map<std::shared_ptr<CompoundJob>, std::shared_ptr<BatchJob>> all_jobs;
 
        //A set of running batch_standard_and_pilot_jobs jobs
        std::map<std::shared_ptr<CompoundJob>, std::shared_ptr<BatchJob>> running_jobs;
 
        // The batch_standard_and_pilot_jobs queue
        std::deque<std::shared_ptr<BatchJob>> batch_queue;
 
        // A set of "waiting" batch_standard_and_pilot_jobs jobs, i.e., jobs that are waiting to be sent to
        //  the  scheduler (useful for batsched only)
        std::set<std::shared_ptr<BatchJob>> waiting_jobs;
 
        // Scheduler
        std::unique_ptr<BatchScheduler> scheduler;
 
 
#ifdef ENABLE_BATSCHED
 
        std::set<std::string> scheduling_algorithms = {"conservative_bf", "crasher", "easy_bf", "easy_bf_fast",
                                                       "easy_bf_plot_liquid_load_horizon",
                                                       "energy_bf", "energy_bf_dicho", "energy_bf_idle_sleeper",
                                                       "energy_bf_monitoring",
                                                       "energy_bf_monitoring_inertial", "energy_bf_subpart_sleeper",
                                                       "energy_watcher", "fcfs_fast", "fast_conservative_bf",
                                                       "filler", "killer", "killer2", "random", "rejecter",
                                                       "sequencer", "sleeper", "submitter", "waiting_time_estimator"
        };
 
        std::set<std::string> queue_ordering_options = {"fcfs", "lcfs", "desc_bounded_slowdown", "desc_slowdown",
                                                        "asc_size", "desc_size", "asc_walltime", "desc_walltime"
 
        };
#else
        std::set<std::string> scheduling_algorithms = {"fcfs", "conservative_bf", "conservative_bf_core_level"
        };
 
        //Batch queue ordering options
        std::set<std::string> queue_ordering_options = {
        };
 
#endif
 
        unsigned long generateUniqueJobID();
 
        void removeJobFromRunningList(std::shared_ptr<BatchJob> job);
 
        void removeJobFromBatchQueue(std::shared_ptr<BatchJob> job);
 
        void removeBatchJobFromJobsList(std::shared_ptr<BatchJob> job);
 
        int main() override;
 
        bool processNextMessage();
 
        void startBackgroundWorkloadProcess();
 
        void processGetResourceInformation(simgrid::s4u::Mailbox *answer_mailbox, const std::string &key);
 
        void processCompoundJobCompletion(std::shared_ptr<BareMetalComputeServiceOneShot> executor, std::shared_ptr<CompoundJob> job);
 
        void processCompoundJobFailure(std::shared_ptr<BareMetalComputeServiceOneShot> executor,
                                       std::shared_ptr<CompoundJob> job,
                                       std::shared_ptr<FailureCause> cause);
 
        void terminateRunningCompoundJob(std::shared_ptr<CompoundJob> job, ComputeService::TerminationCause termination_cause);
 
 
        //Terminate the batch_standard_and_pilot_jobs service (this is usually for pilot jobs when they act as a batch_standard_and_pilot_jobs service)
        void cleanup(bool has_returned_from_main, int return_value) override;
 
        // Terminate
        void terminate(bool send_failure_notifications, ComputeService::TerminationCause termination_cause);
 
        //Process standard job timeout
        void processCompoundJobTimeout(std::shared_ptr<CompoundJob> job);
 
        //process standard job termination request
        void processCompoundJobTerminationRequest(std::shared_ptr<CompoundJob> job, simgrid::s4u::Mailbox *answer_mailbox);
 
        // process a batch_standard_and_pilot_jobs bach_job tiemout event
        void processAlarmJobTimeout(std::shared_ptr<BatchJob>bach_job);
 
        //free up resources
        void freeUpResources(std::map<std::string, std::tuple<unsigned long, double>> resources);
 
        //send call back to the pilot job submitters
        void sendPilotJobExpirationNotification(std::shared_ptr<PilotJob> job);
 
        //send call back to the standard job submitters
        void sendCompoundJobFailureNotification(std::shared_ptr<CompoundJob> job, std::string job_id, std::shared_ptr<FailureCause> cause);
 
        // process a job submission
        void processJobSubmission(std::shared_ptr<BatchJob>job, simgrid::s4u::Mailbox *answer_mailbox);
 
        //start a job
        void startJob(std::map<std::string, std::tuple<unsigned long, double>>, std::shared_ptr<CompoundJob> ,
                      std::shared_ptr<BatchJob>, unsigned long, unsigned long, unsigned long);
 
 
        void processExecuteJobFromBatSched(std::string bat_sched_reply);
 
        void processIsThereAtLeastOneHostWithAvailableResources(simgrid::s4u::Mailbox *answer_mailbox, unsigned long num_cores, double ram);
 
    };
}
 
 
#endif //WRENCH_BATCH_SERVICE_H