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::shared_ptr<FileLocation>  > file_locations);
 
        std::shared_ptr<StandardJob>  createStandardJob(WorkflowTask *task,
                                       std::map<WorkflowFile *,
                                               std::shared_ptr<FileLocation>  > file_locations);
 
        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