% CJSTask Task for CJS clusters
%   A CJSTask object is a type of parallel.Task that is created by calling
%   createTask(job) or findTask(job) on a job that was created from a 
%   cluster of type Local, LSF, Generic, HPCServer, PBSPro, or Torque.
%
%   parallel.task.CJSTask methods:
%      cancel                - Cancel a pending or running task
%      delete                - Remove a task object from its job and memory
%      listAutoAttachedFiles - List the files that are automatically attached for this task
%      wait                  - Wait for task object to change state
%
%   parallel.task.CJSTask properties:
%      CaptureDiary       - Specify whether to return diary output
%      CreateDateTime     - Date and time at which this task was created
%      Diary              - Text produced by execution of task object's function
%      Error              - Task error information
%      ErrorIdentifier    - Task error identifier
%      ErrorMessage       - Message from task error
%      FinishDateTime     - Date and time at which this task finished running
%      Function           - Function called when evaluating task
%      ID                 - Task's numeric identifier
%      InputArguments     - Input arguments to task function
%      Name               - Name of this task
%      NumOutputArguments - Number of arguments returned by task function
%      OutputArguments    - Output arguments from running Function on a worker
%      Parent             - Job containing this task
%      RunningDuration    - The duration the job has been running for, if it has started
%      SchedulerID        - The ID given to this task by the scheduler
%      StartDateTime      - Date and time at which this task started running
%      State              - Current state of task
%      UserData           - Data associated with a task object
%      Warnings           - Task warning information
%      Worker             - Object representing the worker for this task
%
%   See also parallel.Task, parallel.Job, parallel.Job/createTask, 
%            parallel.Job/findTask.

% Copyright 2011-2022 The MathWorks, Inc.

classdef ( ConstructOnLoad = true ) CJSTask < parallel.Task

    properties ( Transient, SetAccess = immutable, GetAccess = private )
        Support   % Support object for property access
        SupportID % ID of this Task as far as Support is concerned
    end

    properties ( Transient, PCTGetSet, PCTConstraint='string' )
        %SchedulerID The ID given to the task by the scheduler
        %   (read-only after job submission)
        SchedulerID = '';
    end

    methods ( Hidden )
        function obj = CJSTask( parent, taskid, tasksid, support )
            parallel.Task.loadCheck( nargin );
            obj@parallel.Task( parent, taskid );
            obj.Support   = support;
            obj.SupportID = tasksid;
        end
        
        function stateEnum = hGetStateEnum(tasks)
            stateEnum = arrayfun( @hGetStateEnumOneTask, tasks);
        end

        function stateEnum = hGetStateEnumOneTask(obj)
            stateEnum = obj.Parent.Parent.hGetTaskStateEnum(obj);
        end
    end

    methods ( Hidden, Access = { ?parallel.job.CJSCommunicatingJob, ...
                            ?parallel.job.CJSConcurrentJob } )
        function hRemoveOneExcessTask( task )
        % This method is called by the task with spmdIndex == 1 when a given task is
        % known to be surplus to requirements
        % Ask the support to destroy the task.
            job = task.Parent;
            task.Support.destroyTasks( task );
            task.Support.notifyDeleteTask( job );
        end
    end

    methods ( Access = protected )
        function destroyTasks( taskOrTasks )
            jobs = [taskOrTasks.Parent];
            [uniqueJobs, ~, I] = unique( jobs );
            for ii = 1:numel( uniqueJobs )
                job = uniqueJobs(ii);
                jobState = job.hGetProperty( 'StateEnum' );
                if jobState ~= parallel.internal.types.States.Pending
                    % This is a bit weird, but it's what API1 does. Why not
                    % check the task state?
                    cluster = job.Parent;
                    tasksForThisJob = taskOrTasks(I == ii);
                    for jj = 1:numel( tasksForThisJob )
                        taskForThisJob = tasksForThisJob(jj);
                        try
                            cluster.hDestroyTask( taskForThisJob );
                        catch E
                            warning(message('parallel:task:TaskDestructionFailed', E.message));
                        end
                    end
                end
            end
            
            % Group the tasks by their support and ask each support to
            % destroy their tasks
            supports = [taskOrTasks.Support];
            [uniqueSupports, ~, I] = unique( supports );
            for ii = 1:numel( uniqueSupports )
                support = uniqueSupports(ii);
                support.destroyTasks( taskOrTasks(I == ii) );
            end
            
            % Notify all jobs that some of their tasks have been deleted
            for ii = 1:numel( uniqueJobs )
                job = uniqueJobs(ii);
                support = job.hGetSupport();
                support.notifyDeleteTask( job );
            end
        end

        function cancelOneTask( task, cancelException )
            import parallel.internal.types.States

            % Early return for the task-already-finished case
            taskState = task.StateEnum;
            if taskState == States.Finished
                return
            end
            jobState = task.Parent.StateEnum;
            if jobState == States.Pending
                okToWriteCancellation = true;
            else
                try
                    cluster = task.Parent.Parent;
                    okToWriteCancellation = cluster.hCancelTask( task );
                catch E
                    warning(message('parallel:task:TaskCancellationFailed', E.message));
                    okToWriteCancellation = false;
                end
            end
            if okToWriteCancellation
                task.hWriteCancellation( cancelException );
                task.Support.notifyCancelTask( task.Parent );
            end
        end

        function w = getWorker( task )
            w = task.hGetProperty( 'Worker' );
        end
        
        function supportVec = checkVectorizedSetSupport( ~ )
            % Override checkVectorizedSetSupport method inherited from
            % CustomGetSet to indicate that vectorized set calls are
            % supported. This is done to minimise filesystem calls.
            supportVec = true;
        end
    end

    methods ( Hidden )
        function hSetPropertyNoCheck( obj, propName, val )
            % hSetPropertyNoCheck supports vectorized calls for arrays of
            % obj and cell arrays of propName
            
            numTasks = numel( obj );
            
            % Quick exit if there is only one task
            if numTasks == 1
                obj.Support.setTaskProperties( obj.SupportID, propName, val );
                return
            end
            
            % Otherwise ensure that propName and val are cell arrays so
            % that we can treat the single and multiple property sets in
            % the code below uniformly
            if ~iscell( propName )
                propName = {propName};
                val = {val};
            end
            
            % Group the tasks by their support
            supports = [obj.Support];
            [uniqueSupports, ~, J] = unique( supports );
            
            for ii = 1:numel( uniqueSupports )
                support = uniqueSupports(ii);
                
                indicesForThisSupport = J == ii;
                tasksForThisSupport = obj( indicesForThisSupport );
                taskSIds = {tasksForThisSupport.SupportID};
                valuesForThisSupport = cellfun( @(v) v(indicesForThisSupport), val, 'UniformOutput', false );
                
                try
                    support.setTaskProperties( taskSIds, propName, valuesForThisSupport );
                catch err
                   throwAsCaller( err );
                end
            end
        end
        
        function v = hGetProperty( obj, propName )
            % hGetProperty supports vectorized calls for arrays of obj and 
            % cell arrays of propName
            
            numTasks = numel( obj );
            
            % Warn if the diary is incomplete for any of the tasks
            if any( strcmpi( propName, 'Diary' ) )
                for jj = 1:numTasks
                    obj(jj).warnIfDiaryIncomplete();
                end
            end
            
            % Quick exit if there is only one task
            if numTasks == 1
                v = obj.Support.getTaskProperties( obj.SupportID, propName );
                return
            end
            
            % Otherwise pre-allocate the return value, which will always be
            % a cell array since numTasks > 1. For convenience, convert
            % propName to a cell array (if necessary) to ensure
            % getTaskProperties() always returns a cell array.
            if ~iscell( propName )
                propName = {propName};
            end
            numProps = numel( propName );
            v = cell( numTasks, numProps );
            
            % Group the tasks by their support
            supports = [obj.Support];
            [uniqueSupports, ~, J] = unique( supports );
            
            for ii = 1:numel( uniqueSupports )
                support = uniqueSupports(ii);
                
                indicesForThisSupport = J == ii;
                tasksForThisSupport = obj( indicesForThisSupport );
                taskSIds = {tasksForThisSupport.SupportID};
                
                try
                    v(indicesForThisSupport, :) = support.getTaskProperties( taskSIds, propName );
                catch err
                   throwAsCaller( err );
                end
            end
        end
        function sid = hGetSupportID( task )
        % Public access needed when submitting Jobs. May refactor this later.
            sid = task.SupportID;
        end
        function support = hGetSupport( obj )
        % Only used by parallel.internal.cluster.hasSameSupport
            support = obj.Support;
        end
        function diaryFile = hGetDiaryFile( obj, taskLoc )
            diaryFile = obj.Support.getDiaryFileForTask( taskLoc );
        end
        function hWriteCancellation( task, cancelException )
        % Called either from CJSTask.cancelOneTask or CJS*Job.cancelJob via
        % CJSJobMethods.
            finishDateTime = datetime( 'now', 'TimeZone', 'local' );
            finishState    = parallel.internal.types.States.Finished;
            namesValues    = { 'ErrorIdentifier' , cancelException.identifier;
                               'ErrorMessage'    , cancelException.message;
                               'Error'           , cancelException;
                               'FinishDateTime'  , finishDateTime;
                               'StateEnum'       , finishState };
            task.hSetPropertyNoCheck( namesValues(:,1), namesValues(:,2) );
        end
        function [taskPropertyMap, propNames]  = hGetDisplayItems(obj, diFactory)
            taskPropertyMap = hGetDisplayItems@parallel.Task(obj, diFactory);
           
            % The order the properties appear in the propNames array is the
            % order in which they will be displayed.
            propNames = {...
                'ID', ...
                'State', ...
                'Function', ...
                'Parent', ...
                'SchedulerID', ...
                'StartDateTime', ...
                'RunningDuration', ...
                'Separator', ...
                'Error'...
                'Warnings'
                };
            % don't actually want to display Scheduler ID on local clusters
            if isa(obj(1).Parent.Parent, 'parallel.cluster.Local')
                propNames = propNames([1:4,6:end]);
            end
          
            % CJSTask specific displayable items to map. 'Function' is 
            % present in all task displays, but must be gotten carefully 
            % for MJS Tasks, so it is added to the map here and not in the 
            % base class.
            functionName = parallel.internal.cluster.generateTaskFunctionString(obj.hGetDisplayPropertiesNoError('FunctionName'));
            taskPropertyMap('Function') = diFactory.createDefaultItem(functionName); 
            taskPropertyMap('SchedulerID') = diFactory.createDefaultItem(obj.hGetDisplayPropertiesNoError('SchedulerID'));
        end
        
        function hSetMatlabLogDir( task )
        % Call utRegisterLogDirectory with the task storage directory.
            dirName = task.Support.getJobDirForOs( task.Parent.hGetSupportID(), ...
                                                   task.Parent.Parent.OperatingSystem );
            distcompSetMatlabLogDir( dirName );
        end
        
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % API1 compatibility
    methods ( Hidden )
        function pPreTaskEvaluate( task )
            import parallel.internal.types.States
            worker = getCurrentWorker();
            task.hSetProperty( {'StartDateTime', 'StateEnum', 'Worker'}, ...
                               {datetime('now', 'TimeZone', 'local'), States.Running, worker} );
        end
        function pPostTaskEvaluate( task, output, mexception, diaryOut, warnings )
            namesValues = { 'OutputArguments'   , output;
                            'Error'             , mexception;
                            'FinishDateTime'    , datetime('now', 'TimeZone', 'local');
                            'DiagnosticWarnings', warnings;
                            'StateEnum'         , parallel.internal.types.States.Finished };
            if ~isempty( mexception )
                namesValues = [ namesValues ;
                                { 'ErrorMessage'    , mexception.message;
                                  'ErrorIdentifier' , mexception.identifier } ];
            end

            if ~isempty( diaryOut )
                % This will only happen if we are using the EvalcEvaluator, which
                % is only used for taskFinish functions of a parallel pool job.
                namesValues = [ namesValues ; { 'Diary', diaryOut } ];
            end
            try
                task.hSetPropertyNoCheck( namesValues(:,1), namesValues(:,2) );
            catch E
                % Get here if there's a problem saving the result. Try one more
                % time to save the result, discarding the OutputArguments as
                % that's where the problem is most likely to exist.
                if strcmp( E.identifier, 'parallel:cluster:FileSerializerDataTooLarge' )
                    E = addCause( MException( message( ...
                        'parallel:task:TaskResultTooLarge' ) ), E );
                end
                namesValues{1, 2} = [];
                namesValues{2, 2} = E;
                task.hSetPropertyNoCheck( namesValues( :,1 ), namesValues( :,2 ) );
            end
        end
    end
end
