% Methods to help creation of tasks and destruction of jobs.

% Copyright 2011-2020 The MathWorks, Inc.

classdef ( Hidden, Sealed ) CJSJobMethods
    methods ( Hidden, Static )
        function cancelOneJob( cluster, job, jobSupport, cancelException )
        % Provides the implementation of CJS job.cancelJob

        % Algorithm:

        % shouldCancel = ( any tasks not finished ) || ( job has no
        %                          tasks && job not finished )
        % if shouldCancel
        %     if ( job is pending )
        %         write cancellation info to job and all tasks
        %     else
        %         ask the scheduler to cancel job execution
        %         if ( the scheduler thought it did stuff )
        %              write cancellation info to job and all unfinished tasks
        %         end
        %     end
        % end
            import parallel.internal.types.States

            try
                jobStateFromDisk = job.hGetProperty( 'StateEnum' );
                allTasks         = job.Tasks;
                if numel( allTasks ) > 0
                    taskStatesC    = arrayfun( @(t) hGetProperty( t, 'StateEnum' ), ...
                                               allTasks, 'UniformOutput', false );
                    taskStates     = [ taskStatesC{:} ];
                    taskIsFinished = ( taskStates == States.Finished );
                    shouldCancel   = any( ~taskIsFinished );
                else
                    taskIsFinished = false( size( allTasks ) );
                    shouldCancel   = ( jobStateFromDisk ~= States.Finished );
                end
            catch E
                % Corrupt data on disk
                ME = MException(message('parallel:job:JobDataCorrupted'));
                ME = addCause( ME, E );
                throw( ME );
            end
            if shouldCancel
                if jobStateFromDisk == States.Pending
                    okToWriteCancellation = true;
                else
                    % Actually ask the cluster to actually cancel execution
                    try
                        okToWriteCancellation = cluster.hCancelJob( job );
                    catch E
                        warning(message('parallel:cluster:FailedToCancelJob', E.message));
                        okToWriteCancellation = false;
                    end
                end
                if okToWriteCancellation
                    iWriteCancellationInformation( ...
                        job, allTasks( ~taskIsFinished ), cancelException );
                    jobSupport.notifyCancelJob( job )
                end
            end
        end

        function destroyOneJob( cluster, job, jobSupport, jobSupportID )
        % Helper for CJS job types - provides implementation of
        % Job.destroyJob().
            jobStateFromDisk = job.hGetProperty( 'StateEnum' );
            if jobStateFromDisk ~= parallel.internal.types.States.Pending
                try
                    cluster.hDestroyJob( job );
                catch E
                    % The cluster failed to do its job of destroying the job,
                    % but we must still continue and let the support destroy the
                    % job from disk.
                    if jobStateFromDisk == parallel.internal.types.States.Unavailable
                        warning(message('parallel:job:DestructionOfUnavailableJob')); 
                    else
                        warning(message('parallel:job:ClusterErrorDuringDestruction', E.message));
                    end
                end
            end
            jobSupport.destroyJob( cluster.hGetSupportID(), jobSupportID );
            jobSupport.notifyDeleteJob( job );
        end

        function tasks = createTask( job, jobSupport, jobSID, ...
                                     allowMultipleTasks, fcn, nout, varargin )
        % Provides the implementation of createTask for CJS*Jobs.

        % This method engenders the knowledge that only jobs in state 'pending'
        % may have tasks added to them.
            import parallel.internal.apishared.TaskCreation
            import parallel.internal.cluster.ConstructorArgsHelper
            import parallel.internal.customattr.PropSet

            [taskFcns, numArgsOut, argsIn, setArgs] = ...
                TaskCreation.createTaskArgCheck(numel(job), fcn, nout, varargin{:});

            if job.hGetProperty( 'StateEnum' ) ~= parallel.internal.types.States.Pending
                error(message('parallel:job:BadStateForCreateTask'));
            end

            if ~allowMultipleTasks && numel( taskFcns ) > 1
                error(message('parallel:job:BadNumberOfTasksToCreate', job.Type));
            end
            % Pass on the 'Profile' from the Cluster.
            if ~isempty( job.Parent.Profile )
                setArgs = [ 'Profile', job.Parent.Profile, setArgs ];
            end
            [names, values] = ConstructorArgsHelper.resolveTaskBuildArguments( ...
                ?parallel.task.CJSTask, setArgs{:} );
            
            % Extract the 'CaptureDiary' name-value pair, if present, so
            % that it can be processed in a vectorized way in buildTasks()
            % rather than being processed sequentially in the more general
            % setArgs code
            captureDiaryIndex = find(strcmp(names, 'CaptureDiary'));
            if isempty(captureDiaryIndex)
                captureDiary = false;
            else
                captureDiary = values{captureDiaryIndex};
                names(captureDiaryIndex) = [];
                values(captureDiaryIndex) = [];
            end
            
            tasks = jobSupport.buildTasks( job, jobSID, taskFcns, numArgsOut, argsIn, captureDiary );
            
            setArgs = PropSet.namesValuesToPv( names, values );
            if ~isempty( setArgs )
                for ti = 1:numel( taskFcns )
                    try
                        % Apply the set args - this might fail if there are invalid
                        % props in which case we bail out and delete the task.
                        set( tasks(ti), setArgs{:} );
                    catch E
                        % Delete remaining tasks
                        delete(tasks(ti:end));
                        % Assume we're being called directly from job.createTask().
                        throwAsCaller( E );
                    end
                end
            end
        end

        function state = getJobStateFromTasks( job )
        % If necessary, delve through the job's Tasks to derive the job state.
        % - job: a scalar CJS job
        % returns:
        % - state: a States representing the state of the job.

            import parallel.internal.types.States

            validateattributes( job, {'parallel.Job'}, {'scalar'} );
            state = job.hGetProperty( 'StateEnum' );

            if state == States.Pending || ...
                      state == States.Finished || ...
                      state == States.Failed
                return;
            end

            try
                stateFromTasks = iDeriveJobStateFromTasks( job.Tasks );
                if stateFromTasks ~= States.Unavailable
                    state = stateFromTasks;
                end
            catch E %#ok<NASGU> swallow and return the "disk" state.
            end
        end

        function setJobTerminalStateFromCluster( job, state )
        % Set job and all tasks to final state.  The job's state is always
        % set to state that we got from the scheduler.  Each task's state is
        % set to the state from the scheduler only if it wasn't already in a
        % terminal state.  (i.e. if the scheduler state is failed, but the
        % task completed successfully and is in state finished, then we
        % won't change the task state to failed).

            import parallel.internal.types.States

            assert( state.isTerminal() );

            tasks = job.Tasks;
            for ii = 1:numel( tasks )
                taskState = tasks(ii).hGetProperty( 'StateEnum' );
                if ~taskState.isTerminal()
                    tasks(ii).hSetPropertyNoCheck( 'StateEnum', state );
                end
            end
            job.hSetPropertyNoCheck( 'StateEnum', state );

            % If the job does not have a StartDateTime, derive it from the
            % tasks.
            if isempty( job.hGetProperty( 'StartDateTime' ) )
                jobStartDateTime = iDeriveStartDateTimeFromTasks( tasks );
                job.hSetPropertyNoCheck( 'StartDateTime', jobStartDateTime );
            end
            
            % Only set finish times if the job state is actually finished.
            % Failed jobs will not have a finish time
            if state == States.Finished && ...
                      isempty( job.hGetProperty( 'FinishDateTime' ) )
                % Ensure that the job finish time is set appropriately (it may
                % not be set if the task that was supposed to set it failed
                % before it was able to set the job finish time.

                % All the tasks are independent, so do nothing about task finish
                % times.
                jobFinishDateTime = iDeriveFinishDateTimeFromTasks( tasks );
                job.hSetPropertyNoCheck( 'FinishDateTime', jobFinishDateTime );
            end
        end

        function duplicateTasksOfSimultaneousJob( job, jobSupport, jobSId, totalNumTasks )
        % Call this prior to submission of any communicating or concurrent job
        % to set up the 2:N tasks. We are told by the scheduler how many tasks
        % in total will be required.
            import parallel.internal.types.Variant

            validateattributes( job, {'parallel.job.CJSCommunicatingJob', ...
                                'parallel.job.CJSConcurrentJob'}, {'scalar'} );

            numToCreate = totalNumTasks - 1;

            % If numToCreate is still not finite, then the Cluster probably
            % specifies NumWorkers == Inf (and job.NumWorkersRange is probably
            % [1 Inf]).
            if isinf( numToCreate )
                error(message('parallel:job:CommunicatingJobFiniteNumWorkers'));
            end

            leadTask = job.Tasks(1);
            if job.Variant == Variant.CommunicatingPoolJob && ...
                    ~ job.hIsPoolInteractive()
                % Note that it's only the non-interactive pool job where we
                % overwrite the TaskFunction - in that case, Task1 contains the
                % user function. In the interactive pool case, the full task
                % must be copied because the input arguments to @parallel.internal.pool.poolWorkerFcn
                % are secretly used to hook up the pool.
                taskFcn = @parallel.internal.pool.poolWorkerFcn;
                numOut  = 0;
                argsIn  = {};
            else
                taskFcn = leadTask.Function;
                numOut  = leadTask.NumOutputArguments;
                argsIn  = leadTask.InputArguments;
            end
            captDiary = leadTask.CaptureDiary;
            jobSupport.buildTasks( job, jobSId, ...
                                   repmat( {taskFcn}, 1, numToCreate ), ...
                                   repmat( numOut, 1, numToCreate ), ...
                                   repmat( {argsIn}, 1, numToCreate ), ...
                                   captDiary );
        end
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iDeriveFinishDateTimeFromTasks - actually grovel through the tasks to see which
% was the last FinishDateTime.
function finishDateTime = iDeriveFinishDateTimeFromTasks( tasks )
    % If an error is thrown while we are getting the FinishDateTime just leave
    % the finish time empty
    finishDateTimes = arrayfun( @(t) t.hGetProperty( 'FinishDateTime' ), tasks,...
                            'UniformOutput', false,...
                            'ErrorHandler', @iLogErrorReturnEmptyDateTime );
    validFinishDateTimes = ~cellfun( @isempty, finishDateTimes );
    validDateTimes = finishDateTimes( validFinishDateTimes );
    finishDateTime = max( [validDateTimes{:}] );
    
    % default to 'now' - in case there are no tasks or no valid finish datetimes.
    if isempty( finishDateTime )
        finishDateTime = datetime( 'now', 'TimeZone', 'local' );
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iDeriveStartDateTimeFromTasks - actually grovel through the tasks to see which
% was the first StartDateTime.
function startDateTime = iDeriveStartDateTimeFromTasks( tasks )
    % If an error is thrown while we are getting the StartDateTime just leave
    % the start time empty
    startDateTimes = arrayfun( @(t) t.hGetProperty( 'StartDateTime' ), tasks,...
                            'UniformOutput', false,...
                            'ErrorHandler', @iLogErrorReturnEmptyDateTime );
    validStartDateTimes = ~cellfun( @isempty, startDateTimes );
    validDateTimes = startDateTimes( validStartDateTimes );
    startDateTime = min( [validDateTimes{:}] );
    
    % default to datetime.empty - in case there are no tasks or no valid start datetimes.
    if isempty( startDateTime )
        startDateTime = datetime.empty();
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iLogErrorReturnEmptyDateTime - Used in ErrorHandler of arrayfun in iDeriveFinishDateTimeFromTasks
function emptyDateTime = iLogErrorReturnEmptyDateTime(errorStruct, task)
dctSchedulerMessage(0, ['Failed to get the FinishDateTime for task with TaskID %d ', ...
                        'due to the following error:\n%s'], ...
                        task.ID, errorStruct.message);
emptyDateTime = datetime.empty();
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iDeriveJobStateFromTasks - actually grovel through the tasks to see what state
% the Job ought to be in.
%
% This function tries to derive the state based on a best guess from the task 
% states without using any knowledge about the job state on disk or the state of 
% job according to the scheduler.  However, to get here, the job state on disk 
% is most likely Queued or Running although it might be Unavailable, in which
% case our answer here is the best we can do.  
%
% In which case:
% - unavailable tasks are not used to derive the state
% - if there are no tasks in an available state, then the
%   derived state is also unavailable
% - if any task state is Running, the job is still running.
% - if all task states are "terminal", the job is terminal (Finished)
% - if all tasks are pending, then the job is queued
% - if there are some tasks finished and some tasks still pending,
%   the job is running
function state = iDeriveJobStateFromTasks( tasks )
    import parallel.internal.types.States
    numTasks         = numel( tasks );
    allTaskStates    = cell(size(tasks));

    for ii = numTasks:-1:1
        thisTaskState = tasks(ii).hGetProperty( 'StateEnum' );
        if thisTaskState == States.Running
            % job *must* be running, so return early
            state = States.Running;
            return
        end
        allTaskStates{ii} = thisTaskState;
    end
    
    % Ignore any tasks that are in the unavailable state
    allTaskStates = [allTaskStates{:}];
    availableTasksIdx = arrayfun(@States.isValidState, allTaskStates);
    availableTaskStates = allTaskStates(availableTasksIdx);

    if numel(availableTaskStates) > 0
        % If we got here, none of the tasks were running, but the
        % job could still be running if some tasks are finished
        % but some are pending.  If all tasks are finished, then
        % the job must be finished, and if all tasks are pending
        % then it's likely that the job is queued.
        isTaskTerminal = arrayfun(@(x) x.isTerminal(), availableTaskStates);
        isTaskPending = arrayfun(@(x) isequal(x, States.Pending), availableTaskStates);
        if all(isTaskTerminal)
            state = States.Finished;
        elseif all(isTaskPending)
            state = States.Queued;
        else
            state = States.Running;
        end
    else
        state = States.Unavailable;
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iWriteCancellationInformation - write the information to a job
function iWriteCancellationInformation( job, someTasks, cancelException )
% Write finish time and state for a cancelled job.
    finishDateTime = datetime( 'now', 'TimeZone', 'local' );
    finishState = parallel.internal.types.States.Finished;
    job.hSetPropertyNoCheck( {'FinishDateTime', 'StateEnum'}, ...
                             {finishDateTime, finishState} );

    % Loop over tasks writing cancellation information.
    arrayfun( @(t) hWriteCancellation( t, cancelException ), someTasks );
end
