JobManager.h

 
#ifndef WRENCH_PILOTJOBMANAGER_H
#define WRENCH_PILOTJOBMANAGER_H
 
#include <vector>
#include <set>
 
#include "wrench/services/Service.h"
 
namespace wrench {
 
    class WMS;
 
    class Workflow;
 
    class WorkflowTask;
 
    class WorkflowFile;
 
    class WorkflowJob;
 
    class PilotJob;
 
    class StandardJob;
 
    class ComputeService;
 
    class StorageService;
 
    /***********************/
    /***********************/
 
 
    class JobManager : public Service {
 
    public:
 
 
        void stop() override;
 
        void kill();
 
 
        std::shared_ptr<StandardJob> createStandardJob(std::vector<WorkflowTask *> tasks,
                                                       std::map<WorkflowFile *, std::shared_ptr<FileLocation> > file_locations,
                                                       std::vector<std::tuple<WorkflowFile *, std::shared_ptr<FileLocation>, std::shared_ptr<FileLocation>  >> pre_file_copies,
                                                       std::vector<std::tuple<WorkflowFile *, std::shared_ptr<FileLocation>, std::shared_ptr<FileLocation>  >> post_file_copies,
                                                       std::vector<std::tuple<WorkflowFile *, std::shared_ptr<FileLocation>  >> cleanup_file_deletions);
 
        std::shared_ptr<StandardJob> createStandardJob(std::vector<WorkflowTask *> tasks,
                                                       std::map<WorkflowFile *, std::vector<std::shared_ptr<FileLocation>>> file_locations,
                                                       std::vector<std::tuple<WorkflowFile *, std::shared_ptr<FileLocation>, std::shared_ptr<FileLocation>  >> pre_file_copies,
                                                       std::vector<std::tuple<WorkflowFile *, std::shared_ptr<FileLocation>, std::shared_ptr<FileLocation>  >> post_file_copies,
                                                       std::vector<std::tuple<WorkflowFile *, std::shared_ptr<FileLocation>  >> cleanup_file_deletions);
 
 
        std::shared_ptr<StandardJob> createStandardJob(std::vector<WorkflowTask *> tasks,
                                                       std::map<WorkflowFile *, std::shared_ptr<FileLocation>> file_locations);
 
        std::shared_ptr<StandardJob> createStandardJob(std::vector<WorkflowTask *> tasks,
                                                       std::map<WorkflowFile *, std::vector<std::shared_ptr<FileLocation>>> file_locations);
 
        std::shared_ptr<StandardJob> createStandardJob(WorkflowTask *task,
                                                       std::map<WorkflowFile *, std::shared_ptr<FileLocation>> file_locations);
 
        std::shared_ptr<StandardJob> createStandardJob(WorkflowTask *task,
                                                       std::map<WorkflowFile *, std::vector<std::shared_ptr<FileLocation>>> file_locations);
 
        std::shared_ptr<StandardJob> createStandardJob(std::vector<WorkflowTask *> tasks);
 
        std::shared_ptr<StandardJob> createStandardJob(WorkflowTask *task);
 
        std::shared_ptr<PilotJob> createPilotJob();
 
        void submitJob(std::shared_ptr<WorkflowJob> job, std::shared_ptr<ComputeService> compute_service,
                       std::map<std::string, std::string> service_specific_args = {});
 
        void terminateJob(std::shared_ptr<WorkflowJob> job);
 
//        void forgetJob(WorkflowJob *job);
 
        std::set<std::shared_ptr<PilotJob>> getPendingPilotJobs();
 
        std::set<std::shared_ptr<PilotJob>> getRunningPilotJobs();
 
        /***********************/
        /***********************/
 
        ~JobManager() override;
 
    protected:
 
        friend class WMS;
 
        explicit JobManager(std::shared_ptr<WMS> wms);
 
        /***********************/
        /***********************/
 
    private:
 
        int main() override;
 
        bool processNextMessage();
 
        void
        processStandardJobCompletion(std::shared_ptr<StandardJob> job, std::shared_ptr<ComputeService> compute_service);
 
        void
        processStandardJobFailure(std::shared_ptr<StandardJob> job, std::shared_ptr<ComputeService> compute_service,
                                  std::shared_ptr<FailureCause> cause);
 
        void processPilotJobStart(std::shared_ptr<PilotJob> job, std::shared_ptr<ComputeService> compute_service);
 
        void processPilotJobExpiration(std::shared_ptr<PilotJob> job, std::shared_ptr<ComputeService> compute_service);
 
 
        // Relevant WMS
        std::shared_ptr<WMS> wms;
 
        // Job lists
        std::set<std::shared_ptr<StandardJob>> new_standard_jobs;
        std::set<std::shared_ptr<StandardJob>> pending_standard_jobs;
        std::set<std::shared_ptr<StandardJob>> running_standard_jobs;
        std::set<std::shared_ptr<StandardJob>> completed_standard_jobs;
        std::set<std::shared_ptr<StandardJob>> failed_standard_jobs;
 
        std::set<std::shared_ptr<PilotJob>> new_pilot_jobs;
        std::set<std::shared_ptr<PilotJob>> pending_pilot_jobs;
        std::set<std::shared_ptr<PilotJob>> running_pilot_jobs;
        std::set<std::shared_ptr<PilotJob>> completed_pilot_jobs;
 
    };
 
    /***********************/
    /***********************/
 
};
 
#endif //WRENCH_PILOTJOBMANAGER_H