% CJSJobMixin Mixin class providing hidden/controlled access methods for CJS*Job
% classes.

% Copyright 2011-2020 The MathWorks, Inc.

classdef CJSJobMixin < handle
    properties (Transient, SetAccess = immutable, GetAccess = protected)
        Support
        SupportID
    end

    properties ( Access = private, Hidden )
        % Cached StateEnum which allows detection of changes in job state. 
        CachedStateEnum
        NotifyStateChanges = false;
    end
    
    methods ( Access = protected )
        function obj = CJSJobMixin( support, jobsid )
            obj.Support   = support;
            obj.SupportID = jobsid;
        end

        function tc = getTaskCount( obj )
            if numel( obj ) > 1
                tc = arrayfun( @(x) x.Support.getTaskCount( ...
                        x.hGetSupportID() ), obj );
            else
                tc = obj.Support.getTaskCount( obj.hGetSupportID() );
            end
        end
        
        function tl = getTasks( obj )
            if numel( obj ) > 1
                tl = arrayfun( @(x) x.Support.getTasks( ...
                        x, x.hGetSupportID() ), obj, ...
                        'UniformOutput', false );
            else
                tl = obj.Support.getTasks( obj, obj.hGetSupportID() );
            end
        end
        
        function jobStatusData = getStatusData( jobOrJobs )
            try
                jobStatusData = parallel.internal.cluster.getJobStatusData(jobOrJobs);
            catch err
                throwAsCaller(err);
            end
        end
        
        function submitOneJobCommon( job, clusterSubmitFcn )
        % Common submission wrapper for all CJS job types. clusterSubmitFcn is a
        % method on the CJS cluster object that performs the submission.
            job.Parent.hJobPreSubmitCheck( job );
            try
                clusterSubmitFcn( job.Parent, job, job.Support, job.SupportID );
            catch E
                % move back to pending state if clusterSubmitFcn fails.
                state = parallel.internal.types.States.Pending;
                job.hSetPropertyNoCheck( 'StateEnum', state );
                rethrow( E );
            end
            job.Support.notifySubmitJob( job );
        end
        
        function varargout = getTasksByStateImpl( obj )
            import parallel.internal.types.States;

            statesStr = { States.Pending.Name, ...
                          States.Running.Name, ...
                          States.Finished.Name, ...
                          States.Failed.Name };

            if numel( obj ) > 1
                prff = arrayfun( @(x) x.Support.getTasksByState( ...
                        x, x.hGetSupportID(), statesStr ), obj, ...
                        'UniformOutput', false );
                prff = vertcat( prff{:} );
            else
                prff = obj.Support.getTasksByState( obj, obj.hGetSupportID(), statesStr );
            end
            % Need to concatenate Finished/Failed
            assert( nargout == 3, 'We assume getTasksByStateImpl is always called with 3 output arguments' );
            if nargout == 3
                % To be consistent with the API and to allow concatenation in calling functions
                % each cell of the varargout must be a 1x1 cell containing a row vector of tasks.
                % prff is an Nx4 cell where N == numel(obj).
                varargout = { {[prff{:,1}]}, {[prff{:,2}]}, {[prff{:,3:end}]} };
            else
                varargout = { [ prff(:,1:2), {[prff{:,3:end}]} ] };
            end
        end
    end

    methods ( Hidden, Access = ?parallel.internal.cluster.CJSSupport )
        function t = hBuildTask( job, taskid, tasksid )
            t = parallel.task.CJSTask( job, taskid, tasksid, job.Support );
        end
    end

    methods ( Hidden, Access = { ...
        ?parallel.internal.cluster.JobStateHelper, ...
        ?parallel.cluster.CJSCluster, ...
        ?parallel.job.CJSIndependentJob, ...
        ?parallel.job.CJSConcurrentJob } )
        function s = hGetStateFromTasks( job )
            import parallel.internal.cluster.CJSJobMethods
            s = CJSJobMethods.getJobStateFromTasks( job );
        end
    end

    methods ( Hidden )
        function hSetPropertyNoCheck( obj, propName, val )
            obj.Support.setJobProperties( obj.SupportID, obj.Variant, propName, val );
        end
        function v = hGetProperty( obj, propName )
            % There's no point vectorizing hGetProperty for CJS jobs, as
            % the properties for each job are stored in separate file, so
            % each file has to be read in turn regardless (this is unlike
            % MJS where a single database query can be made for all jobs).
            % This will change if we switch to using a database for CJS
            % jobs and tasks.
            
            try
                % We still need to fulfill the contract for vectorized
                % calls to hGetProperty even though we're not actually
                % vectorizing.
                if numel( obj ) > 1
                    v = arrayfun( @(x) x.Support.getJobProperties( ...
                            x.hGetSupportID, x.Variant, propName ), ...
                            obj, 'UniformOutput', false );
                    % We want a cell array with one row per job and one
                    % column per property.
                    v = vertcat(v{:});
                else
                    v = obj.Support.getJobProperties( obj.hGetSupportID, obj.Variant, propName );
                end
            catch err
                throwAsCaller(err);
            end
        end
        function support = hGetSupport( obj )
            support = [obj.Support]';
        end
        function sid = hGetSupportID( obj )
            if numel( obj ) > 1
                sid = {obj.SupportID}';
            else
                % Maintain backwards compatibility with non-vectorized code.
                sid = obj.SupportID;
            end
        end
        function userName = hGetClusterUsername( ~ )
            userName = distcomp.pGetDefaultUsername();
        end
        function hRegisterStateListeners( job )
            job.NotifyStateChanges = true;
        end

        function files = hGetCrashDumpFiles( job )
        % Get a list of file names corresponding to crash dumps.
            files = job.Support.getCrashDumpsForJob( job.hGetSupportID() );
        end
        
        function jobIdentifier = hCreateJobUpdateIdentifier( job )
            jobIdentifier = job.Support.hCreateJobUpdateIdentifier( job.SupportID, job );
        end
        
        function s = hGetStateEnum( job )
            numberOfJobs = numel( job );
            s = cell( numberOfJobs, 1 );
            for ii = 1:numberOfJobs
                thisJob = job(ii);
                cluster = thisJob.Parent;
                s{ii} = cluster.hGetJobState( thisJob, thisJob.hGetProperty( 'StateEnum' ) );
                thisJob.checkForChangesInStateEnum( s{ii} );
            end
            % For backwards compatibility when we are getting the state
            % enum for one job, we need to return a scalar, not a
            % cell array containing a scalar.
            if numberOfJobs == 1
                s = s{1};
            end
        end
    end
    
    methods
        function SchedulerIDs = getTaskSchedulerIDs( obj )
            %getTaskSchedulerIDs returns a cell array containing the
            %   SchedulerID of each task on this job.
            
            SchedulerIDs = get( obj.Tasks, {'SchedulerID'} ); %#ok<MCNPN> CJSJobs always have Tasks
            SchedulerIDs = unique( SchedulerIDs( ~cellfun(@isempty, SchedulerIDs) ), 'stable' );
            SchedulerIDs = obj.Parent.hSimplifySchedulerIDs( obj, SchedulerIDs ); %#ok<MCNPN> CJSJobs always have a Parent
        end
    end

    methods ( Hidden, Access = ?parallel.cluster.CJSCluster )
        function hSetTerminalStateFromCluster( obj, state )
            import parallel.internal.cluster.CJSJobMethods

            CJSJobMethods.setJobTerminalStateFromCluster( obj, state );
            % If we get here, we are being informed that the job is terminal, so
            % we should maybe-update our tasks with finish time information.
            if state == parallel.internal.types.States.Finished
                iMaybeUpdateTaskFinishTimes( obj.Tasks, ...
                                             obj.hGetProperty( 'FinishDateTime' ) );
            end
        end
    end
    
    methods ( Access = private )
        function checkForChangesInStateEnum( job, newState )
            if job.NotifyStateChanges && ...
                    ~isempty(job.CachedStateEnum) && newState ~= job.CachedStateEnum
                job.CachedStateEnum = newState;
                job.Support.notifyStateChange( job, newState );
            end
            job.CachedStateEnum = newState;
        end
    end
end

% --------------------------------------------------------------------------
% If we find out that a job has finished from the Cluster, we should ensure that
% the task finish times are set.
function iMaybeUpdateTaskFinishTimes( tasks, jobFinishDateTime )
    import parallel.internal.types.States

    for ii = 1:length( tasks )
        thisTask = tasks(ii);
        if thisTask.hGetProperty( 'StateEnum' ) == States.Finished
            try
                if isempty( thisTask.hGetProperty( 'FinishDateTime' ) )
                    thisTask.hSetPropertyNoCheck( 'FinishDateTime', jobFinishDateTime );
                end
            catch E
                % Failed to load the FinishDateTime. It is likely the
                % output file is corrupted, perhaps due to the task
                % failing.
                dctSchedulerMessage(0, ['Failed to get the FinishDateTime for task with TaskID %d ', ...
                    'due to the following error:\n%s'], ...
                    thisTask.ID, E.message);
            end
        end
    end
end
