% MJSJobMixin Mixin class to collect common hidden/controlled access methods for
% MJS*Job classes.

%   Copyright 2013-2020 The MathWorks, Inc.

classdef MJSJobMixin < handle
    
    properties ( Constant )
        ClientRelease = char(pctGetMATLABRelease());
    end
    
    properties ( Transient, SetAccess = immutable, GetAccess = protected )
        Support
        SupportID
    end
    
    properties ( Access = private )
        MATLABRelease = '';
    end
    
    methods ( Access = protected )
        function obj = MJSJobMixin( support, jobsid )
            obj.Support = support;
            obj.SupportID = jobsid;
        end

        function tl = getTasks( job )
            try
                if parallel.internal.cluster.hasSameSupport( job )
                    tl = job(1).Support.getTasks( job, job.hGetSupportID() );
                else
                    tl = arrayfun( @(x) x.Support.getTasks( ...
                            x, x.hGetSupportID() ), job(:), ...
                            'UniformOutput', false );
                end
            catch E
                dctSchedulerMessage(1, 'Error during MJSJobMixin.getTasks', E);
                throw( distcomp.handleJavaException( job, E ) );
            end
        end
        
        function tc = getTaskCount( job )
            try
                if parallel.internal.cluster.hasSameSupport( job )
                    tc = job(1).Support.getTaskCount( job.hGetSupportID() );
                else
                    tc = arrayfun( @(x) x.Support.getTaskCount( ...
                            x.hGetSupportID() ), job(:) );
                end
            catch E
                dctSchedulerMessage(1, 'Error during MJSJobMixin.getTaskCount', E);
                throw( distcomp.handleJavaException( job, E ) );
            end
        end
        
        function jobStatusData = getStatusData( jobOrJobs )
            try
                if parallel.internal.cluster.hasSameSupport( jobOrJobs )
                    jobStatusData = jobOrJobs(1).Support.getJobStatusData( jobOrJobs.hGetSupportID() );
                else
                    jobStatusData = arrayfun( @(x) x.Support.getJobStatusData( ...
                            x.hGetSupportID() ), jobOrJobs(:));
                end
            catch E
                throw( distcomp.handleJavaException( jobOrJobs, E ) );
            end
        end
        
        function tasks = doCreateTaskOnOneJob( job, allowMultipleTasks, fcn, nout, varargin )
        % This method engenders the knowledge that only jobs in states >=
        % 'pending' but not 'finished' may have tasks added to them.
            import parallel.internal.apishared.TaskCreation
            import parallel.internal.cluster.DefaultProperties
            import parallel.internal.cluster.ConstructorArgsHelper
            import parallel.internal.types.States

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

                job.hErrorIfReleaseMismatch( 'parallel:job:TaskCreationVersionMismatch' );

                jobState = job.StateEnum;
                if jobState < States.Pending || jobState >= States.Finished
                    error( message( 'parallel:job:MJSJobAddTaskState' ) );
                end

                numTasks = numel( taskFcns );
                if ~allowMultipleTasks && numTasks > 1
                    error( message( 'parallel:job:BadNumberOfTasksToCreate', job.Type ) );
                end

                % Pass on the 'Profile' from the Cluster, prepend this.
                if ~isempty( job.Parent.Profile )
                    setArgs = [ 'Profile', job.Parent.Profile, setArgs ];
                end
                [ names, values ] = ConstructorArgsHelper.resolveTaskBuildArguments(...
                    ?parallel.task.MJSTask, setArgs{:} );
                containsCallbacks = iContainsCallbacks( names );
                setArgs = reshape( [ names; values ], size( names, 1 ), [] ); % {names{1}, values{1}, names{2}, values{2}, ...}

                % Use the defaults defined in DefaultProperties when possible
                defaultTaskStruct = DefaultProperties.getNewTaskPropStruct();

                % Parse the task arguments
                p = inputParser();
                p.KeepUnmatched = true;
                p.addParameter( 'Name', '' ); % implies: use default name
                p.addParameter( 'Timeout', intmax( 'int64' ) ); % No default
                p.addParameter( 'CaptureDiary', defaultTaskStruct.CaptureDiary );
                p.addParameter( 'MaximumRetries', 1 ); % No default
                p.addParameter( 'MaxNumRetries', 1 ); % No default
                p.parse( setArgs{:} );
                results = p.Results;

                % Have 'MaxNumRetries' override 'MaximumRetries' if specified
                if ~any( ismember( p.UsingDefaults, 'MaxNumRetries' ) )
                    results.MaximumRetries = results.MaxNumRetries;
                end

                taskInfoStructs = struct(...
                    'Function', taskFcns, ...
                    'NumOutputArguments', num2cell( numArgsOut ), ...
                    'InputArguments', argsIn, ...
                    'Name', results.Name, ...
                    'Timeout', results.Timeout, ...
                    'CaptureDiary', results.CaptureDiary, ...
                    'MaximumRetries', results.MaximumRetries, ...
                    'ContainsCallbacks', containsCallbacks );

                sameFcn = ~iscell( fcn ); % Multiple functions specified with cell array
                tasks = job.Support.buildTasks( job, job.SupportID, sameFcn, taskInfoStructs );

                % Set task fields that have not been handled above
                setNames = fields( p.Unmatched );
                for ii = 1:numTasks
                    try
                        for jj = 1:numel( setNames )
                            tasks(ii).(setNames{jj}) = p.Unmatched.(setNames{jj});
                        end
                    catch E
                        % It's somewhat tricky to get here since the
                        % resolveTaskBuildArguments stage has already checked
                        % constraints.
                        delete( tasks(ii:end) );
                        throwAsCaller( E );
                    end
                end
            catch E
                throw( distcomp.handleJavaException( job, E ) );
            end
        end
    end

    methods ( Hidden )
        function hDelete( obj )
            try
                if parallel.internal.cluster.hasSameSupport( obj )
                    obj(1).Support.destroyJob( obj.hGetSupportID() );
                else
                    arrayfun( @(x) x.Support.destroyJob( ...
                            x.hGetSupportID() ), obj);
                end
                arrayfun(@(job) job.Support.notifyDeleteJob( job ), obj);
            catch E
                throw( distcomp.handleJavaException( obj, E ) );
            end            
        end
        
        function hCancel( obj, cancelException )
            try
                if parallel.internal.cluster.hasSameSupport( obj )
                    obj(1).Support.cancelJob( obj.hGetSupportID(), cancelException );
                else
                    arrayfun( @(x, ce) x.Support.cancelJob( ...
                            x.hGetSupportID(), ce ), obj, ce);
                end
                arrayfun(@(job) job.Support.notifyCancelJob( job ), obj);
            catch E
                throw( distcomp.handleJavaException( obj, E ) );
            end
        end
        
        function v = hGetProperty( obj, propName )
            try
                % Currently we only vectorise the calls to get job properties when all jobs
                % have the same support object. We could do this with hetfun in the future.
                if parallel.internal.cluster.hasSameSupport( obj )
                    v = obj(1).Support.getJobProperties( obj.hGetSupportID(), propName );
                else
                    if ~iscell( propName )
                        propName = {propName};
                    end
                    
                    v = arrayfun( @(x) x.Support.getJobProperties( ...
                            x.hGetSupportID(), propName ), obj(:) );
                end
            catch E
                throw( distcomp.handleJavaException( obj, E ) );
            end
        end
        
        function sid = hGetSupportID( obj )
            sid = vertcat( obj.SupportID );
        end
        
        function support = hGetSupport( obj )
            support = obj.Support;
        end
        
        function s = hGetStateEnum( obj )
            s = parallel.internal.cluster.getWorkUnitStateEnum( obj );
        end
        
        function userName = hGetClusterUsername( obj )
            userName = obj.Parent.Username; %#ok<MCNPN> The property 'Parent' is defined in parallel.Job
        end
        
        function tf = hIsClusterAvailable( obj )
            tf = obj.Parent.hIsClusterAvailable(); %#ok<MCNPN> The property 'Parent' is defined in parallel.Job
        end
        
        function hSetPropertyNoCheck( obj, propName, val )
            % hSetPropertyNoCheck supports vectorized calls for arrays of obj and 
            % cell arrays of propName, so that remote calls can be minimized with MJS.
            try
                hErrorIfReleaseMismatch( obj, 'parallel:job:PropertySetVersionMismatch' );
                
                if parallel.internal.cluster.hasSameSupport( obj )
                    obj(1).Support.setJobProperties( obj.hGetSupportID(), propName, val );
                else
                    for idx = 1:numel( obj )
                        if iscell( propName )
                            values = cellfun( @(x) x{idx}, val, 'UniformOutput', false );
                        else
                            values = val{idx};
                        end
                        obj(idx).Support.setJobProperties( obj(idx).hGetSupportID(), propName, values );
                    end
                end
            catch E
                throw( distcomp.handleJavaException( obj, E ) );
            end
        end
        
        function hRegisterStateListeners( obj )
            obj.Support.registerStateListeners( obj );
        end
        
        function jobIdentifier = hCreateJobUpdateIdentifier( job )
            jobIdentifier = job.Support.hCreateJobUpdateIdentifier( job.SupportID, job );
        end
    
        function hErrorIfReleaseMismatch( obj, errorId )
            % Retrieve MATLAB releases for all jobs that don't have it
            % cached already
            jobReleases = arrayfun( @(job) job.MATLABRelease, obj, 'UniformOutput', false );
            uncached = cellfun( @isempty, jobReleases );
            if any( uncached )
                jobsToRetrieve = obj(uncached);
                
                if parallel.internal.cluster.hasSameSupport( jobsToRetrieve )
                    values = jobsToRetrieve(1).Support.getJobProperties( jobsToRetrieve.hGetSupportID(), {'MATLABRelease'} );
                else
                    values = arrayfun( @(job) job.Support.getJobProperties( job.SupportID, 'MATLABRelease' ), jobsToRetrieve, 'UniformOutput', false );
                end
                
                jobReleases(uncached) = values;
                for ii = 1:numel(jobsToRetrieve)
                    jobsToRetrieve(ii).MATLABRelease = values{ii};
                end
            end
            
            clientRelease = parallel.internal.cluster.MJSJobMixin.ClientRelease;
            matches = strcmp( jobReleases, clientRelease );
            if ~all( matches )
                error( message( errorId, jobReleases{find( ~matches, 1 )} ) );
            end
        end
    end
end

function hasCallbacks = iContainsCallbacks(names)
    callbackNames = {'RunningFcn', 'FinishedFcn'};
    hasCallbacks = false;
    if any(ismember(callbackNames, names))
        hasCallbacks = true;
    end
end