% CJSCommunicatingJob Communicating job for CJS clusters
%   A CJSCommunicatingJob object is a type of parallel.CommunicatingJob
%   that is created by calling createCommunicatingJob(cluster) or findJob(cluster) 
%   on a cluster of type Local, LSF, Generic, HPCServer, PBSPro, or Torque.
%
%   parallel.job.CJSCommunicatingJob methods:
%      cancel                - Cancel a pending, queued, or running job
%      createTask            - Create a new task in a job
%      delete                - Remove a job object from its cluster and memory
%      diary                 - Display or save text of batch job
%      fetchOutputs          - Retrieve output arguments from all tasks in a job
%      findTask              - Find task objects belonging to a job
%      listAutoAttachedFiles - List the files that are automatically attached to the job
%      load                  - Load workspace variables from batch job
%      recreate              - Create a new job from an existing job
%      submit                - Submit job for execution in the cluster
%      wait                  - Wait for job execution to change state
%
%   parallel.job.CJSCommunicatingJob properties:
%      AdditionalPaths   - Folders to add to MATLAB search path of workers
%      AttachedFiles     - Files and folders that are sent to the workers
%      AutoAddClientPath - Determines whether user-added-entries on the client's path are added to each worker's path
%      AutoAttachFiles   - Determines whether dependent files will be automatically sent to the workers
%      CreateDateTime    - Date and time at which this job was created
%      FinishDateTime    - Date and time at which this job finished running
%      ID                - Job's numeric identifier
%      JobData           - Data made available to all workers for job's tasks
%      Name              - Name of this job
%      NumThreads        - Number of computational threads for workers
%      NumWorkersRange   - Specify limits for number of workers
%      Parent            - Cluster object containing this job
%      StartDateTime     - Date and time at which this job started running
%      State             - State of the job: pending, queued, running, finished, or failed
%      SubmitDateTime    - Date and time at which this job was submitted
%      Tag               - Label associated with this job
%      Tasks             - Array of task objects contained in this job
%      Type              - Job's type: Independent, Pool, or SPMD
%      UserData          - Data associated with a job object
%      Username          - Name of the user who created this job
%
%   See also parallel.CommunicatingJob, parallel.Job, 
%            parallel.Cluster/createCommunicatingJob, parallel.Cluster/findJob.

% Copyright 2011-2020 The MathWorks, Inc.

classdef ( ConstructOnLoad = true ) CJSCommunicatingJob < parallel.CommunicatingJob & ...
        parallel.internal.cluster.CJSJobMixin

    properties ( PCTGetSet, Transient, Hidden )
        JobSchedulerData
        EnvironmentCredentials
    end

    methods ( Hidden )
        function obj = CJSCommunicatingJob( parent, jobid, pvar, jobsid, support )
            parallel.Job.loadCheck( nargin );
            %{
            validateattributes( parent, {'parallel.cluster.CJSCluster'}, ...
                                {'scalar'} );
            validateattributes( support, {'parallel.internal.cluster.CJSSupport' }, ...
                                {'scalar'} );
            %}
            obj@parallel.CommunicatingJob( parent, jobid, pvar );
            obj@parallel.internal.cluster.CJSJobMixin( support, jobsid );
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % abstract job methods
    methods ( Access = protected )
        function t = createTaskOnOneJob( job, varargin )
            import parallel.internal.cluster.CJSJobMethods

            if isempty( job.Tasks )
                allowMultipleTasks = false;
                t = CJSJobMethods.createTask( job, job.Support, job.SupportID, ...
                                              allowMultipleTasks, varargin{:} );
            else
                error(message('parallel:job:CommunicatingJobOneTask'));
            end
        end
        function submitOneJob( job )
            job.submitOneJobCommon(@hSubmitCommunicatingJob);
        end
        
        function newJob = recreateOneJob( oldJob, varargin )
            cluster = oldJob.Parent;
            jobPVPairs = extractPVPairsFromExistingObject( oldJob, [ {'Type', 'NumWorkersRange'}, getCommonJobPropsForJobRecreation() ] );
            newJob = cluster.createCommunicatingJob( jobPVPairs{:} );
            jobHiddenPVPairs = extractPVPairsFromExistingObject( oldJob, getCommonHiddenJobPropsForRecreation() ); 
            for idx = 1:2:numel(jobHiddenPVPairs)
                newJob.hSetProperty( jobHiddenPVPairs{idx}, jobHiddenPVPairs{idx+1} );
            end
            %Communicating jobs only have one task
            taskPVPairs = extractCreationArgsFromExistingTask( oldJob.Tasks(1), getCommonTaskPropsForJobRecreation() );
            newJob.createTask( taskPVPairs{:});
        end
        function destroyJob( jobOrJobs )
            import parallel.internal.cluster.CJSJobMethods;
            arrayfun( @(job) CJSJobMethods.destroyOneJob( job.Parent, job, ...
                job.Support, job.SupportID ), jobOrJobs );
        end
        function cancelJob( jobOrJobs, cancelException )
            import parallel.internal.cluster.CJSJobMethods;
            arrayfun( @(job, ce) CJSJobMethods.cancelOneJob( job.Parent, job, ...
                job.Support, ce ), jobOrJobs, cancelException );
        end
        function varargout = getTasksByState( obj )
            [varargout{1:nargout}] = obj.getTasksByStateImpl();
        end
        function attachFileDataToJob( obj, filedata )
            obj.Support.attachFileDataToJob( obj.SupportID, obj.Variant, filedata );
        end
    end

    methods ( Access = private )
        function trimExcessTasks( job )
        % Called prior to execution on lab 1 to remove any excess tasks that
        % might have been created.
            try
                tasks = job.Tasks;
                for ii = numlabs+1:numel(tasks)
                    % We cannot use the full 'delete' here, as this causes the
                    % task (and hence job) to be cancelled. So, we use a secret
                    % method to rid ourselves of the troublesome tasks.
                    hRemoveOneExcessTask( tasks(ii) );
                end
            catch E %#ok<NASGU> swallow errors in this method
            end
        end
        function tf = canModifyJobOnWorker( ~ )
        % On the worker only, can we modify the job?
            tf = ( labindex == 1 );
        end
    end

    % API1 compatibility
    methods ( Hidden )
        function pPreJobEvaluate( job, ~ )
            canModifyJob  = ( labindex == 1 );
            idForRandInit = labindex;
            if canModifyJob
                % The following line of code is not permitted to throw:
                job.trimExcessTasks();
            end
            % Only setup GPU allocations during preCJSJob if we're running an SPMD job. All
            % pool jobs get the allocations set up during parallel.Pool
            % initialisation.
            setupGPUAllocations = (job.Variant == parallel.internal.types.Variant.CommunicatingSPMDJob);
            preCJSJob( job, canModifyJob, idForRandInit, setupGPUAllocations );
        end
        function pPostJobEvaluate( job )
            task           = getCurrentTask();
            taskHasError   = ~isempty( task.Error );
            iPostJobEvaluate( job, taskHasError, ...
                              job.PoolShutdownSuccessful, job.Variant );
        end
    end

    methods ( Hidden )
        function tf = hWorkerCrashedDuringExecution( job )
        % See if we've got any crash dump files.
            tf = ~isempty( job.hGetCrashDumpFiles() );
        end
    end

end
%#ok<*ATUNK> custom attributes

% --------------------------------------------------------------------------
% iPostJobEvaluate - a close mimic of simpleparalleljob/pPostJobEvaluate
function iPostJobEvaluate( job, taskHasError, ...
                           poolShutdownSuccessful, jobVariant )
    import parallel.internal.types.Variant
    
    % Abort the job if the pool failed to shut down for a pool job, or if 
    % anything went wrong in the SPMD job.
    isPoolJob = jobVariant == Variant.CommunicatingPoolJob;
    if ( isPoolJob && ~ poolShutdownSuccessful ) || ...
        ( jobVariant == Variant.CommunicatingSPMDJob && taskHasError )
        iAbortJob( );
    end
    
    if isPoolJob
        labBarrier();
    end

    % Get the parent of this job and make sure it's a CJSCluster
    cluster = job.Parent;
    if ~isa( cluster, 'parallel.cluster.CJSCluster' )
        return
    end

    % Ensure all tasks get here, with error detection
    mpigateway( 'setidle' );
    mpigateway( 'setrunning' );
    labBarrier;

    % This is the end of a parallel session
    mpiParallelSessionEnding;
    mpiFinalize;
end

% --------------------------------------------------------------------------
% iAbortJob - abort execution of a job
function iAbortJob( )
    import parallel.internal.types.States

    % This quits MATLAB hard on all the workers, providing workers were started
    % using mpiexec.
    dctSchedulerMessage( 0, 'About to invoke MPI_Abort' );
    mpigateway( 'abort' );
end
