% %Job Base class for all Job objects
%   A job object contains all the tasks that define what each worker does as
%   part of complete job execution.
%
%   Job objects are created by calling either createJob(cluster)
%   which creates an instance of parallel.IndependentJob, or by calling
%   createCommunicatingJob(cluster) which creates an instance of
%   parallel.CommunicatingJob.
%
%   parallel.Job 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 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
%      EnvironmentVariables - Names of environment variables that are sent to the workers
%      FileStore            - Get the store of files for this job.
%      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
%      Parent               - Cluster object containing this job
%      RunningDuration      - How long the job has been running for
%      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
%      ValueStore           - Get the store of data for this job.
%
%   See also parcluster, parallel.Cluster, parallel.Task.

% Copyright 2011-2023 The MathWorks, Inc.

classdef ( ConstructOnLoad = true ) Job < ...
        handle & ...                                   % Jobs are handle objects
        matlab.mixin.Heterogeneous & ...               % Different job subclasses may be concatenated
        parallel.internal.cluster.CustomPropDisp & ... % Use our display methods
        parallel.internal.customattr.CustomGetSet      % Use custom attributes

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % All public properties

    % Mutable at any time:
    properties ( PCTGetSet, Transient, PCTConstraint = 'string' )
        %Tag Label associated with this job
        Tag    = ''

        %Name Name of this job
        Name   = ''
    end
    properties ( PCTGetSet, Transient )
        %JobData Data made available to all workers for job's tasks
        JobData  = []
    end

    % Mutable only before submission:
    properties ( PCTGetSet, Transient, PCTConstraint = 'nonemptycellstr|presubmission' )
        %AttachedFiles Files and folders that are sent to the workers
        %   AttachedFiles can be modified only before the job is submitted.
        %   (read-only after submission)
        AttachedFiles   = {}

        %AdditionalPaths Folders to add to MATLAB search path of workers
        %   AdditionalPaths can be modified only before the job is submitted.
        %   (read-only after submission)
        AdditionalPaths = {}
        
        %EnvironmentVariables Names of environment variables that are sent to the workers
        %   EnvironmentVariables defines the names of environment variables
        %   which will be copied from the client session to the workers.
        %
        %   (read-only after submission)
        EnvironmentVariables
    end
    
    properties ( Hidden, PCTGetSet, Transient, PCTConstraint = 'positiveintscalar|presubmission' )
        %NumPathsToAppend Indicates the number of path entries in AdditionalPaths to be appended to the worker's paths.
        %   I.e. AdditionalPaths(1:end-NumPathsToAppend) are to be prepended,
        %        AdditionalPaths(end - NumPathsToAppend + 1:end) are to be appended
        NumPathsToAppend = 0;
    end

    properties ( PCTGetSet, Transient, PCTConstraint = 'logicalscalar|presubmission' )
        %AutoAttachFiles Determines whether dependent files will be automatically sent to the workers
        %   If true the files required to run the job's task functions will be calculated
        %   and sent to the workers at job submission time.
        %   See also parallel.Job/listAutoAttachedFiles, parallel.Job/AttachedFiles,
        %   parallel.Job/AdditionalPaths.
        AutoAttachFiles = true

        %AutoAddClientPath Determines whether user-added-entries on the client's path are added to each worker's path
        %   A logical value that controls whether user-added-entries on the
        %   client's path will be added to each worker's path at the start
        %   of the job.  For pool and batch jobs, the default value is
        %   true.  For all other jobs the default value is false.
        AutoAddClientPath = false;
    end

    % read-only:
    properties ( PCTGetSet, Transient, SetAccess = private )
        %CreateDateTime Date and time at which this job was created
        %   (read-only)
        CreateDateTime = datetime.empty()

        %SubmitDateTime Date and time at which this job was submitted
        %   If this job has not been submitted, SubmitDateTime is empty.
        %   (read-only)
        SubmitDateTime = datetime.empty()

        %StartDateTime Date and time at which this job started running
        %   If this job has not started running, StartDateTime is empty.
        %   (read-only)
        StartDateTime  = datetime.empty()

        %FinishDateTime Date and time at which this job finished running
        %   If this job has not finished running, FinishDateTime is empty.
        %   (read-only)
        FinishDateTime = datetime.empty()

        %Username Name of the user who created this job
        %   (read-only)
        Username   = ''
        
        %NumThreads Number of computational threads for workers 
        %   NumThreads specifies the maximum number of computational 
        %   threads used by workers.  This value is inherited from the
        %   cluster from which this job was created.
        %   (read-only)
        NumThreads = 1
    end

    % non-stored properties:
    properties ( Transient )
        %UserData Data associated with a job object
        %   UserData can be used to store any data associated with a job
        %   object. This data is stored in the client MATLAB session, and is
        %   not available on the workers. UserData is not stored in a profile.
        UserData = []
    end

    % Derived properties
    properties ( Dependent, SetAccess = private )
        %Tasks Array of task objects contained in this job
        %   (read-only)
        %   See also parallel.Job/createTask.
        Tasks

        %State State of the job: pending, queued, running, finished, or failed
        %   (read-only)
        State

        %Type Job's type: Independent, Pool, or SPMD
        %   The type is determined at the time the job is created.
        %   (read-only)
        %
        %   See also parallel.Cluster/createJob,
        %            parallel.Cluster/createCommunicatingJob.
        Type
        
        %RunningDuration How long the job has been running for
        %    (read-only)
        RunningDuration
    end

    % Constant properties
    properties ( Transient, SetAccess = immutable )
        %Parent Cluster object containing this job
        %   (read-only)
        Parent = []

        %ID Job's numeric identifier
        %   (read-only)
        ID     = 0
    end

    properties (Transient, Dependent, SetAccess = private)
        % ValueStore Get the store of data for this job.
        %   Entries added to ValueStore are available in both MATLAB clients
        %   and workers. ValueStore entries are stored in job storage and do
        %   not contribute to process memory use. Use the ValueStore to send 
        %   data from workers back to clients during the execution of a job.
        %   Access the ValueStore on workers using getCurrentValueStore.
        %   (read-only)
        %
        %   See also parallel.ValueStore.
        ValueStore

        % FileStore Get the store of files for this job.
        %   Files added to FileStore are available in both MATLAB clients
        %   and workers. FileStore entries are stored in job storage and do
        %   not contribute to process memory use. Use the FileStore to send 
        %   files from workers back to clients during the execution of a job.
        %   Access the FileStore on workers using getCurrentFileStore.
        %   (read-only)
        %
        %   See also parallel.FileStore.
        FileStore
    end

    properties (Transient, Hidden, Access = private )
        % Lazily create ValueStore when first accessed.
        ValueStoreCache = [];
        % Lazily create FileStore when first accessed.
        FileStoreCache = [];
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Hidden and/or protected properties:
    properties ( PCTGetSet, Transient, Hidden, SetAccess = private )
        CreateTime = '' % Old char vector for job create time
        SubmitTime = '' % Old char vector for job submit time
        StartTime  = '' % Old char vector for job start time
        FinishTime = '' % Old char vector for job finish time
    end

    properties ( PCTGetSet, Transient, SetAccess = protected, Hidden )
        AttachedFileData    = [] % Zipped AttachedFiles data
        DependentFiles      = {} % List of files that the attached files are dependent on
        ExecutionMode       = 0  % Interactive or not
    end
    properties ( PCTGetSet, Transient, Hidden )
        ApiTag              = '' % Hidden 'tag' to indicate which method was used to construct the Job
        AttachedFilePaths   = [] % Map of client paths to AttachedFiles in the zip-archive
        MATLABDrivePathMap  = [] % Map of client paths to relativePaths in MATLAB Drive folder
    end
    properties ( Dependent, SetAccess = private, Hidden )
        StateEnum % Backing for 'State'
    end
    properties ( Hidden, SetAccess = immutable )
        Variant = parallel.internal.types.Variant.IndependentJob % Backing for 'Type'
    end
    properties ( Abstract, SetAccess = private, Hidden )
        MinNumWorkers
    end

    properties ( Transient, Hidden )
        BatchVecData = [] % Hidden cache of pre-computed data from hBatchVec()
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % get/set methods for Dependent properties, typically deferring to
    % abstract/protected methods.
    methods
        function tl = get.Tasks( obj )
            try
                tl = obj.getTasks();
                % Ensure we always return a column.
                tl = reshape( tl, numel(tl), 1 );
            catch E
                throw( E );
            end
        end
        function s = get.State( obj )
            s = obj.StateEnum.Name;
        end
        function s = get.Type( obj )
            s = obj.Variant.Name;
        end
        function se = get.StateEnum( obj )
            % Getting the state of a Job can be somewhat tricky so defer to an
            % abstract method.
            se = obj.hGetStateEnum();
        end
        
        function d = get.RunningDuration( obj )
            d = parallel.internal.cluster.computeRunningDuration( ...
                obj.StartDateTime, obj.FinishDateTime);
        end

        function store = get.ValueStore(job)
            if isempty(job.ValueStoreCache) || ~isvalid(job.ValueStoreCache)
                job.ValueStoreCache = job.createValueStore();
            end
            store = job.ValueStoreCache;
        end

        function store = get.FileStore(job)
            if isempty(job.FileStoreCache) || ~isvalid(job.FileStoreCache)
                job.FileStoreCache = job.createFileStore();
            end
            store = job.FileStoreCache;
        end
    end

    methods ( Sealed )
        function t = createTask( job, varargin )
        %createTask Create a new task in a job
        %    t = createTask(j, F, N, {x1,..., xn}) creates a new task object in job j,
        %    and returns a reference, t, to the added task object. This task evaluates
        %    the function specified by a function handle or function name, F, with the
        %    given arguments, x1,...,xn, returning N output arguments.
        %
        %    t = createTask(j, F, N, {C1,...,Cm}) uses a cell array of m cell arrays
        %    to create m task objects in job j, and returns a vector of references, t,
        %    to the new task objects.  Each task evaluates the function specified by a
        %    function handle or function name F.  The cell array C1 provides the input
        %    arguments to the first task, C2 to the second task, and so on, so that
        %    there is one task per cell array.  Each task returns N output arguments.
        %    If F is a cell array, each element of F specifies a function for each
        %    task in the vector; it must have m elements.  If N is an array of doubles,
        %    each element specifies the number of output arguments for each task in the
        %    vector.  Multidimensional matrices of inputs F, N and {C1,...,Cm}  are
        %    supported; if a cell array is used for F, or a double array for N, its
        %    dimensions must match those of the input arguments cell array of cell
        %    arrays. The output t will be a vector with the same number of elements as
        %    {C1,...,Cm}.
        %
        %    t = createTask(..., 'p1',v1,'p2',v2, ...) creates a task object with the
        %    specified property values. If an invalid property name or property value
        %    is specified, the object will not be created and an error will be thrown.
        %
        %    t = createTask(..., 'Profile', 'profileName', ...) creates a task
        %    object with the property values specified in the profile
        %    'profileName'. If not specified and the cluster has a 'Profile'
        %    specified, then the cluster's profile is automatically applied.
        %
        %    Examples:
        %    Create and submit two jobs.  The first job has a task that generates a
        %    10-by-10 random matrix, the second has a task to execute the function
        %    [y1, y2, y3] = foo(6.98, 7.65) on a worker and return the three output
        %    arguments in the task property OutputArguments.
        %    % Create the first job object.
        %    myCluster = parcluster; % create a cluster object from default profile
        %    j = createJob(myCluster);
        %
        %    % Add a task object that generates a 10-by-10 random matrix.
        %    t = createTask(j, @rand, 1, {10,10});
        %
        %    % Run the job.
        %    submit(j);
        %
        %    % Get the output from the task evaluation.
        %    taskOutput = fetchOutputs(j);
        %
        %    % Show the 10-by-10 random matrix.
        %    disp(taskOutput{1});
        %
        %    % Delete the job object.
        %    delete(j)
        %
        %    % Create the second job and its task.
        %    j = createJob(myCluster);
        %    createTask(j, @foo, 3, {6.98, 7.65})
        %    submit(j);
        %
        %    Create a job with 3 tasks, each of which creates a 10-by-10 random
        %    matrix.
        %    myCluster = parcluster;
        %    j = createJob(myCluster);
        %    t = createTask(j, @rand, 1, {{10,10} {10,10} {10,10}});
        %
        %    See also parallel.Cluster/createJob,
        %             parallel.Cluster/createCommunicatingJob,
        %             parallel.Job/submit, parallel.Job/wait,
        %             parallel.Job/fetchOutputs.
            validateattributes( job, {'parallel.Job'}, {'scalar'}, 'createTask', 'job', 1 );
            job.errorIfInvalid();
            try
                t = job.createTaskOnOneJob( varargin{:} );
                % Ensure we always return a column
                t = reshape( t, numel(t), 1 );
                dctSchedulerMessage(2, 'Created %s', parallel.internal.logging.LogHelper.getTasksString(t));
            catch E
                throw( E );
            end
        end

        function submit( job )
        %SUBMIT Submit job for execution in the cluster
        %    SUBMIT(j) queues the job object, j, in the cluster queue.  The
        %    cluster used for this job was determined when the job was
        %    created.
        %
        %    Example:
        %    myCluster = parcluster; % create a cluster object from default profile
        %
        %    % Create a job object.
        %    j = createJob(myCluster);
        %
        %    % Add a task object to be evaluated for the job.
        %    t = createTask(j, @myFunction, 1, {10, 10});
        %
        %    % Queue the job object in the cluster.
        %    submit(j);
        %
        %    See also parallel.Job/createTask, parallel.Job/wait,
        %             parallel.Job/fetchOutputs.
            validateattributes( job, {'parallel.Job'}, {'nonempty', 'scalar'}, 'submit', 'job', 1 );
            job.errorIfInvalid();
            job.errorIfNotPending( 'submit' );
            job.preSubmit();
            try
                job.submitOneJob();
                dctSchedulerMessage(2, 'Submitted %s', parallel.internal.logging.LogHelper.getJobsString(job));
            catch E
                throw( E );
            end
        end
        
        function newJob = recreate( oldJob, varargin )
        %recreate Create a new job from an existing job
        %    job = recreate(oldJob) creates a job object which is the same as oldJob. 
        %    This means that it contains the same tasks and options as oldJob.
        %    If oldJob was created using batch the recreated job will
        %    automatically be submitted, otherwise the recreated job will
        %    be in the pending state.
        %
        %    job = recreate(oldIndependentJob, 'Tasks', tasksToRecreate) creates a job object
        %    with tasks that correspond to tasksToRecreate. tasksToRecreate
        %    must be one of the following:
        %       - An array of parallel.Task objects belonging to oldIndependentJob
        %       - A 1 x N logical array, where N is the size of
        %       oldIndependentJob.Tasks. Only the tasks indexed by this
        %       array are recreated.
        %       - A function handle which accepts oldIndependentJob.Tasks
        %       as an input argument, and returns a 1 x N logical array, 
        %       where N is the size of oldIndependentJob.Tasks. Only the
        %       tasks indexed by this array are recreated.
        %    For example, use this syntax to rerun tasks containing errors
        %    or warnings.
        %    The Tasks option is supported only for independent jobs.
        %
        %    job = recreate(oldJob, 'TaskState', states) creates a job object
        %    with tasks that correspond to the tasks with States specified
        %    by states. states is a string or cell array of strings
        %    representing the State of the required tasks from oldJob.
        %    The TaskState option is supported only for independent jobs.
        %    Valid states are 'pending', 'running', 'finished' and 'failed'.
        %
        %    job = recreate(oldIndependentJob, 'TaskID', ids) creates a job object
        %    with tasks that correspond to the tasks with IDs specified by ids.
        %    ids is a vector of integers representing the required tasks from 
        %    oldIndependentJob. The TaskID option is supported only for independent jobs.
        %
        %    Example:
        %    % Recreate the entire job, myJob.
        %    newJob = recreate(myJob)
        %
        %    % Recreate a job which has only tasks in myJob which have an
        %    error.
        %    newJob = recreate(myJob, 'Tasks', @hasError);
        %    % Recreate a job which has only tasks in myJob which have a
        %    warning.
        %    newJob = recreate(myJob, 'Tasks', @hasWarnings);
        %
        %    % Recreate a job which has only pending tasks in myJob.
        %    newJob = recreate(myJob, 'TaskState', 'pending');
        %
        %    % Recreate an independent job which has only the tasks with IDs 1 and 2 in 
        %    % oldIndependentJob.
        %    newJob = recreate(oldIndependentJob, 'TaskID', [1, 2]);
        %
        %    See also parallel.Job/createTask, parallel.Job/submit,
        %             parallel.Cluster/createJob,
        %             parallel.Cluster/createCommunicatingJob. 
            narginchk( 1, 3 );
            validateattributes( oldJob, {'parallel.Job'}, {'nonempty', 'scalar'}, 'recreate', 'job', 1 );
            oldJob.errorIfInvalid();
            
            % Error is original job was created using parpool
            import parallel.internal.cluster.ParpoolJobConstants
            import parallel.internal.types.States;
            
            invalidJobTags = {ParpoolJobConstants.ParpoolJobTag};
            if any(strcmp(invalidJobTags, oldJob.ApiTag))
                error(message('parallel:job:CannotRecreateInteractiveJob'));
            end
            
            % Convert any string inputs to character vectors
            [varargin{:}] = parallel.internal.general.convertPVPairsToChars(varargin{:});
                
            % Ensure PV pairs are indeed given in pairs
            numberOfInputs = numel(varargin);
            if mod(numberOfInputs,2) ~= 0
                error(message('parallel:cluster:PropertyValueInPairs'));
            end
            
            if numberOfInputs == 2
                parameter = varargin{1};
                value = varargin{2};
                switch parameter
                    case 'Tasks'
                        % Only Independent jobs support the Tasks PV pair
                        iErrorIfNotIndependentJob(oldJob, message('parallel:job:TasksPVPairDisallowed'));
                        validateattributes(value, {'parallel.Task', 'logical', 'function_handle'}, ...
                            {'vector'}, 'recreate');
                    case 'TaskID'
                        % Only Independent jobs support the TaskID PV pair 
                        iErrorIfNotIndependentJob(oldJob, message('parallel:job:TaskIDPVPairDisallowed'));
                        validateattributes(value, {'numeric'}, {'vector', 'integer', 'positive'}, 'recreate'); 
                    case 'TaskState'
                        % Only Independent jobs support the TaskState PV pair 
                        iErrorIfNotIndependentJob(oldJob, message('parallel:job:TaskStatePVPairDisallowed'));
                               
                        validateattributes(value, {'cell', 'char'}, {}, 'recreate');
                        if ~iscell(value)
                            value = {value};
                        end
                        value = unique(value);
                        validTaskStates = { States.Pending.Name, States.Running.Name, ...
                            States.Finished.Name, States.Failed.Name };
                        okTaskStates  =  ismember(value, validTaskStates);
                        if ~all(okTaskStates)
                            okNamesStr = strjoin(validTaskStates, ', ');
                            error(message('parallel:job:InvalidStateToRecreate', okNamesStr));
                        end                        
                    otherwise
                        error(message('parallel:job:RecreateJobInvalidInput', parameter));
                end
            end
                  
            try
                newJob = oldJob.recreateOneJob( varargin{:} );
                dctSchedulerMessage(2, 'Recreated %s from %s', ...
                    parallel.internal.logging.LogHelper.getJobsString(newJob), ...
                    parallel.internal.logging.LogHelper.getJobsString(oldJob));
                % If the old job was created using batch, automatically
                % submit the recreated job
                if parallel.internal.apishared.BatchJobMethods.isBatchJob(newJob)
                    submit(newJob);
                end
            catch E
                throw( E );
            end 
        end

        function listAutoAttachedFiles( job )
        %listAutoAttachedFiles List the files that are automatically attached to the job
        %   listAutoAttachedFiles(j) lists the files of the job object, j, that will be
        %   or have already been attached to the job, following a dependency analysis
        %   performed on the tasks of the job.
        %
        %   See also parallel.Task/listAutoAttachedFiles, parallel.Job/AttachedFiles,
        %   parallel.Job/AutoAttachFiles.
            validateattributes( job, {'parallel.Job'}, {'nonempty', 'scalar'}, 'listAutoAttachedFiles', 'job', 1 );
            
            attachedFiles = job.hGetAutoAttachedFiles();

            % Do the display of the calculated attached files.
            if isempty( attachedFiles )
                m = message( 'parallel:job:NoAutoAttachedFiles', job.ID );
            else
                formattedFiles = parallel.internal.apishared.AttachedFiles.formatAttachedFilesForDisplay( attachedFiles );
                m = message( 'parallel:job:AutoAttachedFiles', job.ID, formattedFiles );
            end
            disp( m.getString() );
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % protected abstract methods:
    methods ( Access = protected, Abstract )
        % getTasks must return a list of parallel.Task objects
        tl = getTasks( obj )
        
        % getTaskCount only returns the total number of tasks for this job
        tc = getTaskCount( obj )

        % cancelJob must contact the cluster and cancel job execution,
        % writing the cancelException provided where appropriate
        cancelJob( job, cancelException )

        % destroyJob must not only terminate job execution, it must also
        % remove all trace of the job from disk/storage/...
        destroyJob( job )

        % Submit a scalar job
        submitOneJob( job )
        
        % Recreate a scalar job
        newJob = recreateOneJob( oldJob, varargin )
        
        % Attach the job's zipped file data.
        attachFileDataToJob( job, filedata )

        % Create task on a single job
        t = createTaskOnOneJob( job, varargin )
        
         % Return job status data for Job Monitor updates
        jobStatusData = getStatusData( jobOrJobs )

        % Fetches job status data for Job Monitor updates in an asynchronous fashion (if supported)
        % The callback will receive two arguments:
        % * An array of `parallel.internal.cluster.JobStatusData` objects (which might be empty
        %   if there's an error).
        % * An MException (which will be empty if there was no error)
        % n.b. The callback might be called multiple times with different arrays of homogeneous jobs
        % of the same type.
        getStatusDataAsync( jobOrJobs, callback )

        % Create the value store for this job
        createValueStore( job );

        % Create the file store for this job
        createFileStore( job );
    end
    
    methods ( Hidden, Abstract )
        % Return a job identifier for Job Monitor updates
        jobIdentifier = hCreateJobUpdateIdentifier( job );
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Protected methods
    methods ( Access = protected )
        function obj = Job( parent, id, variant )
        % The Job constructor requires the cluster and job ID as these are immutable
        % properties.
        %{
            validateattributes( parent, {'parallel.Cluster'}, {'scalar'} );
            validateattributes( id, {'numeric'}, ...
                                {'scalar', 'real', 'finite', 'integer', 'positive'} );
            validateattributes( variant, {'parallel.internal.types.Variant'}, ...
                                {'scalar'} );
            %}
            obj.Parent  = parent;
            obj.ID      = double(id);
            obj.Variant = variant; 
        end

        function varargout = getTasksByState( job )
            % Override this if you can do this more efficiently. Return a 3-cell of
            % tasks, or 3 output arguments if you'd prefer.
            import parallel.internal.types.States;

            allTasks   = job.getTasks();
            % In the vectorized case a cell-array is returned from getTasks.
            if iscell( allTasks )
                allTasks = [ allTasks{:} ];
            end
            taskStates = get( allTasks, {'StateEnum'} );
            taskStates = [ taskStates{:} ];
            states     = [ States.Pending, States.Running, States.Finished, States.Failed ];
            prff       = arrayfun( @(state) allTasks(taskStates == state), ...
                                   states, 'UniformOutput', false );

            % Need to concatenate Finished/Failed
            if nargout == 3
                varargout = { prff(1), prff(2), {[prff{3:end}]} };
            else
                varargout = { [ prff(1:2), {[prff{3:end}]} ] };
            end
        end

        function ok = doWait( job, varargin )
            % Default implementation of job.wait() - override if you have a more
            % efficient means.
            ok = parallel.internal.cluster.wait( job, varargin{:} );
        end

        function preSubmit( job )
            % Override this if you need to do some cluster specific pre-submission
            % actions.
            if job.AutoAddClientPath
                job.addClientPath();
            end
            job.attachRequiredFiles();
            job.createMATLABDrivePathMap();
            job.setJobEnvironmentCredentials();
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Sealed protected methods
    methods ( Access = protected, Sealed )
        function errorIfInvalid( jobOrJobs )
            if any( ~isvalid( jobOrJobs ) )
                throwAsCaller( MException(message('parallel:job:JobInvalidated')) );
            end
        end
        function errorIfNotPending( jobOrJobs, method )
        %errorIfNotPending Throws an error if jobOrJobs not all pending.

            import parallel.internal.types.States

            errorIfInvalid( jobOrJobs );
            states = hGetStateEnum( jobOrJobs );
            for ii = 1:numel( jobOrJobs )
                if states(ii) ~= States.Pending
                    throwAsCaller( MException(message('parallel:job:OperationOnlyValidWhenPending', method)) );
                end
            end
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Static, sealed protected methods
    methods (Access = protected, Static, Sealed)
        function obj = getDefaultScalarElement() %#ok<STOUT>
            % getDefaultScalarElement is required for matlab.mixin.Heterogeneous
            % to ensure that gaps in arrays are filled in with valid objects.
            error(message('parallel:cluster:ArrayWithGapsNotAllowed', 'parallel.Job'));
        end
    end
    
    methods ( Hidden )
        function hSetProperty( obj, propName, val )
            % hSetProperty has the opportunity to ensure consistency of the
            % to-be-set properties with those properties that have already been
            % set. However, there's no job to do, so simply defer to
            % hSetPropertyNoCheck.
            try
                hSetPropertyNoCheck( obj, propName, val );
            catch E
                dctSchedulerMessage( 0, 'Caught an error and throwing as caller. Original error:\n%s', E.getReport() );
                throwAsCaller( E );
            end
        end
        function hCheckTimingForSet( jobs, timing, propName )
        % Job defines the State, so it is in a position to check
        % the timing.
            jobsStates = get(jobs, 'StateEnum');
            if ~iscell(jobsStates)
                jobsStates = {jobsStates};
            end
            cellfun(@(state) parallel.internal.cluster.checkTimingForSet(state, ...
                timing, propName), jobsStates);
        end
        function tasks = hGetTasks( obj )
            % We need a public way to get the tasks for a vector of jobs 
            % efficiently for display.
            tasks = obj.getTasks();
            
            % getTasks returns a cell array in the vectorized case, and an
            % array of tasks in the non-vectorized case. We make all outputs
            % a cell here to make indexing easier in JobDisplayer.
            if ~iscell( tasks )
                tasks = { tasks };
            end
            % Reshape the task lists individually to be columns.
            tasks = cellfun( @(t) reshape( t, numel( t ), 1 ), tasks, ...
                'UniformOutput', false );
        end
        function count = hGetTaskCount( obj )
            % Public way to get task count for a vector of jobs for
            % efficient vectorized display
            count = obj.getTaskCount();
        end
        function [jobPropertyMap, propNames] = hGetDisplayItems(obj, diFactory)
            % Create an empty map to hold displayable job properties
            jobPropertyMap = containers.Map('KeyType', 'char', 'ValueType', 'any');
            
            % The job base class is never displayed, so it does not
            % have a specified property display order to put into propNames
            propNames = {};
            
            % Add common and custom job displayable items to map
            names = {...
                'ID', ...
                'Type', ...
                'Username', ...
                'State', ...
                'AutoAttachFiles', ...
                'AutoAddClientPath', ...
                'NumThreads' ...
                'RunningDuration' ...
            };
            timeNames = {...
                'SubmitDateTime', ...
                'StartDateTime', ...
                'FinishDateTime'};
            specialNames = {...
                'AttachedFiles', ...
                'AdditionalPaths', ...
                'EnvironmentVariables'};
            [customNames, diFactoryCreators] = obj.hGetCustomDisplayPropNames();
            
            % Get all property values in a single call.
            % hGetDisplayPropertiesNoError should only be called once.
            propsToGet = [names timeNames specialNames customNames];
            propValues = obj.hGetDisplayPropertiesNoError(propsToGet);
            propValueMap = containers.Map(propsToGet, propValues);
            
            values = diFactory.makeMultipleItems(@createDefaultItem, ...
                cellfun(@(name) propValueMap(name), names, 'UniformOutput', false));
            jobPropertyMap = [jobPropertyMap; containers.Map(names, values)];
            
            % Add datetime items to map
            timeValues = diFactory.makeMultipleItems(@createDateTimeItem, ...
                cellfun(@(name) propValueMap(name), timeNames, 'UniformOutput', false));
            jobPropertyMap = [jobPropertyMap; containers.Map(timeNames, timeValues)];
            
            jobPropertyMap('Auto Attached Files') = diFactory.createAutoAttachedFilesItem( obj );

            jobPropertyMap('AttachedFiles') = diFactory.createAttachedFilesItem( ...
                propValueMap('AttachedFiles'), obj );

            jobPropertyMap('AdditionalPaths') = diFactory.createAdditionalPathsItem( ...
                propValueMap('AdditionalPaths'), obj );

            % Add non-default displayable items to map
            jobPropertyMap('RunningDuration') =  diFactory.createDurationItem(...
               propValueMap('RunningDuration'));
            % 'Separator' is not a typical name-value object. However, it
            % needs to be added to the propNames list in between property
            % groups, so it is put into the map with the other
            % displayable items.
            jobPropertyMap('Separator') = diFactory.createSeparator();
            jobPropertyMap('FileStore') = diFactory.createDefaultItem( parallel.FileStore.createDummyForDisplay() );
            jobPropertyMap('ValueStore') = diFactory.createDefaultItem( parallel.ValueStore.createDummyForDisplay() );

            jobPropertyMap('EnvironmentVariables') = diFactory.createEnvironmentVariablesItem(...
                propValueMap('EnvironmentVariables'), obj);

            % Add custom displayable items to map
            for ii = 1:numel( customNames )
                customName = customNames{ii};
                diFactoryCreator = diFactoryCreators{ii};
                jobPropertyMap(customName) = diFactoryCreator( diFactory, propValueMap(customName) );
            end
        end

        function [propNames, diFactoryCreators] = hGetCustomDisplayPropNames( ~ )
            % Subclasses can override this to return a cellstr of
            % additional properties to display and their creators
            propNames = {};
            diFactoryCreators = {};
        end

        function hSetDependentFiles( job, dependentFiles )
            job.DependentFiles = dependentFiles;
        end

        function tf = hWorkerCrashedDuringExecution( ~ )
        % Called by parallel pool infrastructure to see if this job should be preserved
        % because there is crash dump information. Subclasses may wish to
        % override this if they know that preserving the job would preserve
        % important stuff like crash dumps.
            tf = false;
        end
        
    end
    
    methods( Hidden, Sealed )
        function tasksToAdd = hGetTasksForJobRecreation(oldJob, varargin)
            if numel(varargin) == 2
                parameter = varargin{1};
                value = varargin{2};
                switch parameter
                    case 'Tasks'
                        if isa(value, 'parallel.Task')
                            tasksToAdd = value;
                            % Verify that all tasks come from original job.
                            taskParents = unique([tasksToAdd.Parent]);
                            if numel(taskParents) ~= 1 || ~isequal(taskParents, oldJob)
                                error(message('parallel:job:RecreatedTasksNotFromOriginalJob'));
                            end
                        elseif isa(value, 'logical')
                            % Verify that logical array is the same size as
                            % task array
                            oldTasks = oldJob.Tasks;
                            if numel(value) ~= numel(oldTasks)
                                error(message('parallel:job:RecreatedTasksLogicalArraySizeMismatch', ...
                                    numel(value), numel(oldTasks)));
                            end
                            tasksToAdd = oldTasks(value);
                        elseif isa(value, 'function_handle')
                            tasksToAdd = parallel.internal.customattr.WorkUnitFinder.find(oldJob.Tasks, value);
                        else
                            % Should never get here
                            assert(false, 'Unknown ''Tasks'' value %s for retrieving tasks', class(value));
                        end
                    case 'TaskID'
                        cellOfTasksToAdd = arrayfun(@(x) oldJob.findTask('ID', x), value, 'UniformOutput', false);
                        tasksToAdd = [cellOfTasksToAdd{:}];
                    case 'TaskState'
                        if ~iscell(value)
                            value = {value};
                        end
                        value = unique(value);
                        % Take a snapshot of task states to prevent tasks
                        % changing state and being missed completely.
                        taskStateSnapShot = {oldJob.Tasks.State};
                        tasksToAddIndex = ismember(taskStateSnapShot, value);
                        tasksToAdd = oldJob.Tasks(tasksToAddIndex);
                    otherwise
                        % Should never get here
                        assert(false, 'Unknown parameter %s for retrieving tasks', parameter);
                end
            else
                tasksToAdd = oldJob.findTask();
            end
        end        
        
        function link = hGetLink(obj, displayValue, clusterMemento)
            if nargin < 3
                serializedMemento = serialize(parallel.internal.display.JobMemento(obj));
            else
                serializedMemento = serialize(parallel.internal.display.JobMemento(obj, clusterMemento));
            end
            
            matlabFunction = sprintf('%s(''%s'')', ...
                parallel.internal.display.Displayer.DisplayObjectFunction, serializedMemento);
            link = parallel.internal.display.HTMLDisplayType(displayValue, matlabFunction);
        end
        
        function attachedFiles = hGetAutoAttachedFilesNoError( job )
            attachedFiles = {};
            try
                % Not only do we not want to error, we also do not want to warn,
                % which can happen if we are not able to analyze the files.
                warningState = warning('off', 'parallel:cluster:DepfunError');
                warningCleanup = onCleanup( @() warning( warningState ) );
                attachedFiles = job.hGetAutoAttachedFiles();
            catch err %#ok<NASGU>
            end
        end
        
        function jobStatusData = hGetJobStatusData( jobOrJobs )
            statusFcn = @(x) getStatusData(x);
            jobStatusData = parallel.internal.cluster.hetfun(statusFcn, jobOrJobs, 'stable');
        end

        function hGetJobStatusDataAsync( jobOrJobs, callback )
            statusFcn = @(x) getStatusDataAsync(x, callback);
            % This will invoke callbacks for each homogeneous chunk of jobs of the same type.
            % We don't need the 'stable' flag as jobs can come in any order (plus, 'stable'
            % is ignored if no output arguments are present). We will take care of sorting
            % the jobs in the presentation layer.
            parallel.internal.cluster.hetfun(statusFcn, jobOrJobs);
        end
    end
    
    methods ( Hidden, Access = ?parallel.Task )
        function doCalculation = hShouldCalculateAttachedFiles( job )
            doCalculation = job.AutoAttachFiles && ... % We are auto-attaching files
                            job.StateEnum == parallel.internal.types.States.Pending; % The job is not yet submitted
        end
    end
        
    methods ( Hidden, Sealed, Access = ?parallel.Pool )
        function attachedFiles = hGetAutoAttachedFiles( job )
            if job.hShouldCalculateAttachedFiles()
                % If we haven't cached any auto-attached files and the job has
                % not yet been submitted, we need to calculate them for each
                % task function associated with the job.
                attachedFiles = job.Tasks.hCalculateDependentFiles();
                jobDependencies = iGetDependencies( job.AttachedFiles );
                if ~isempty( jobDependencies )
                    jobDependencies = [ jobDependencies{:} ];
                end
                attachedFiles = [ vertcat(attachedFiles{:}); jobDependencies(:) ];
            else
                % Once the job is submitted the auto-attached files are cached in DependentFiles.
                % In that case we just list the cached files.
                attachedFiles = job.DependentFiles;
            end
            % We only want to list the unique attached files from all the tasks.
            attachedFiles = unique( attachedFiles, 'stable' );
            % If we've calculated we need files that are already attached, remove
            % these from the auto-attached list.
            attachedFiles = setdiff( attachedFiles, job.AttachedFiles, 'stable' );
        end
    end

    methods ( Hidden, Sealed, Access = ?parallel.internal.pool.FileManager )
        function attachedFiles = hGetAllAttachedFiles( job )
        % Get all attached files, both those that were specified on the job and the ones
        % that were attached as a result of dependency analysis.
            attachedFiles = unique([job.AttachedFiles; job.hGetAutoAttachedFiles()], 'stable');
        end
    end

    methods ( Hidden )
        %hWarnIfForbiddenOutput Issue a warning in case of forbidden output
        % type
        function hWarnIfForbiddenOutput( ~, ~ )
            % Default is no forbidden output types
        end
    end

    methods ( Sealed, Access = private )
        % Helper for batch load/diary methods
        function errorIfNotBatch( job )
            E = [];
            if numel( job ) > 1
                E = MException(message('parallel:convenience:BatchMethodScalarJob'));
            elseif ~isvalid( job )
                E = MException(message('parallel:job:JobInvalidated'));
            elseif ~ parallel.internal.apishared.BatchJobMethods.isBatchJob( job )
                msg = message( 'parallel:convenience:NotBatchJob' );
                E   = MException( msg.Identifier, '%s', msg.getString() );
            end
            if ~isempty( E )
                throwAsCaller( E );
            end
        end

        function attachRequiredFiles( job )
        % Template method that attaches all required files to the job, automatically
        % calculating dependencies for all the job's tasks if that feature is
        % enabled. The method then delegates to its subclasses to actually attach the
        % list of files and the raw bytes of the zip archive.
            if job.AutoAttachFiles
                if isempty(job.BatchVecData)
                    taskFunctionCache = parallel.internal.cluster.TaskFunctionCache;
                    % We want to cache the analyzed task functions per job submission.
                    % We do not want to cache between submissions, as the code may have
                    % changed in that time.
                    attachedFiles = job.Tasks.hCalculateAndCacheDependentFiles( taskFunctionCache );
                else
                    attachedFiles = {job.BatchVecData.AutoAttachedFiles};
                end
                % We also want to calculate the dependencies of the manually
                % attached files of the job.
                jobDependencies = iGetDependencies( job.AttachedFiles );
                if ~isempty( jobDependencies )
                    jobDependencies = [ jobDependencies{:} ];
                end
                attachedFiles = [ vertcat( attachedFiles{:} ); jobDependencies(:) ];
                
                % These are the dependent files per task, so we need to
                % dedupe any files found in more than one task which is
                % likely. See g2352356
                attachedFiles = unique( attachedFiles, 'stable' );
                
                % We are including the tasks' dependent files here so that getting all
                % the dependent files of a job does not require looping over all the
                % tasks. See g978825.
                job.DependentFiles = attachedFiles;
            else
                % For batch the batch functions for all the tasks are stored on the
                % job's dependent files.
                attachedFiles = job.DependentFiles;
            end

            % Join the manually attached files and their dependencies with the
            % automatically attached files.
            attachedFiles = [ attachedFiles; job.AttachedFiles ];

            % And now make sure the list of attached files is unique.
            attachedFiles = unique( attachedFiles, 'stable' );

            % Toolbox functions and built-ins already exist on the worker, so
            % remove these from the list. It is also not possible to copy
            % built-ins, so these must be filtered here.
            attachedFiles = ...
                parallel.internal.apishared.AttachedFiles.filterToolboxAndBuiltin( attachedFiles );

            % If we are deployed we need to add the relevant .auth files.
            attachedFiles = ...
                parallel.internal.apishared.AttachedFiles.addAuthFilesIfDeployed( attachedFiles );
            
            % (g1026876): We've got to remove files in attached directories from this list.
            attachedFiles = ...
                parallel.internal.apishared.AttachedFiles.removeFilesInAttachedDirs( attachedFiles );
            
            % Transform the AttachedFile paths to relative paths, so we can
            % maintain their structure in the zip file.
            pathMap = parallel.internal.apishared.AttachedFileMap();
            [relativePaths, filePathMap] = ...
                parallel.internal.apishared.AttachedFiles.generateTempDirectoryStructure( ...
                        attachedFiles, ...
                        pathMap );

            % If workers do not support Packages, but Packages were attached, then stop submission
            % here. The simplest way to check is from the pathMap.
            % TODO g2702048 remove this condition when Packages are enabled everywhere.
            if ~job.Parent.hWorkersSupportPackages() && ~isempty(pathMap.getPackageClientDirsAndIds())
                error(message('parallel:job:WorkerDoesNotSupportPackages'));
            end

            % Copy the AttachedFiles to a temporary location.
            [tempDir, tempCleanup] = ...
                parallel.internal.apishared.AttachedFiles.copyAllFilesToTempDir( ...
                    attachedFiles, relativePaths ); %#ok<ASGLU>

            % zip() will only maintain relative paths, files specified by
            % absolute paths are stored in the root of the archive, which
            % we don't want. We need to pass the tempdir to zipFiles, so
            % that the relative paths can be resolved.
            filedata = ...
                parallel.internal.apishared.AttachedFiles.zipFiles( relativePaths, tempDir );

            % Set the relative file paths on the job so that the
            % worker can find the attached files at the other end.
            job.AttachedFilePaths = filePathMap;

            % Now zip up the files and attach them to the job.
            % CJS and MJS do this differently, so call a protected method.
            job.attachFileDataToJob( filedata );
        end
        
        function createMATLABDrivePathMap(job)
            job.MATLABDrivePathMap = parallel.internal.apishared.MATLABDrivePathMap( job.AdditionalPaths );
        end

        function addClientPath(job)
            oldPaths = job.AdditionalPaths;
            % Here we need only path entries that do not correspond to Modular Packages,
            % as the path entries for those will be handled separately.
            [newPaths, numPathsToAppend] = parallel.internal.apishared.JobPathHelper.getUserAddedNonPackagePathEntries();
            job.AdditionalPaths = [ oldPaths(:); newPaths(:) ];
            job.NumPathsToAppend = numPathsToAppend;
        end
        
        function setJobEnvironmentCredentials(job)
            import parallel.internal.apishared.EnvVars
            
            % Check if parent cluster filters out special variables
            filteredEnvironmentVariables = job.Parent.hFilterEnvironmentVariables(job.EnvironmentVariables);
            % Get automatically included variables for parent cluster
            autoVariables = job.Parent.hGetAutoEnvironmentVariables();
            filteredEnvironmentVariables = [filteredEnvironmentVariables autoVariables];
            % Only copy over variables which have been defined.
            defined = EnvVars.checkIfEnvironmentVariableDefined(filteredEnvironmentVariables{:});
            environmentMap = EnvVars.getEnvironmentVariablesMap(filteredEnvironmentVariables(defined));
            if ~isempty(environmentMap)
                job.hSetPropertyNoCheck('EnvironmentCredentials', environmentMap);
            end
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % protected static methods
    methods ( Access = protected, Static )
        function loadCheck( ctorNargin )
        %loadCheck if the constructor is being called with zero arguments,
        %assumption is that the job is being loaded. Therefore error.
            if ctorNargin == 0
                error( message( 'parallel:job:CannotLoadFromMatFile' ) );
            end
        end
    end

    methods ( Sealed )
        % Note that several methods are defined in this 'Sealed' block so that they
        % can be dispatched using an array of Job, but we do not actually allow that
        % to work. (The error returned by attempting to call a non-Sealed method is
        % a little unhelpful).

        % Defined externally:
        out = load( job, varargin );
        diary( job, optionalFilename );

        function varargout = findTask( job, varargin )
        %findTask  Find task objects belonging to a job
        %    tasks = findTask(J) finds all tasks for job J. J can be an array of
        %    job objects.
        %
        %    tasks = findTask(J, 'p1', v1, 'p2', v2, ...) gets a N x 1 array of
        %    task objects belonging to job object J. The returned task objects
        %    will be only those having the specified property-value pairs.  When
        %    a property value is specified, it must use the same format that the
        %    property access returns. For example, if the Name property is
        %    'MyTask', then findTask will not find that object while searching
        %    for a Name property value of myTask as the match is case sensitive.
        %
        %    tasks = findTask(J, taskFcn) gets a N x 1 array of
        %    task objects belonging to job object J. Only the tasks passing
        %    the filter function, taskFcn, are returned. taskFcn is a 
        %    function handle which accepts J.Tasks and returns a logical
        %    array used to index J.Tasks.
        %
        %    [pending, running, finished] = findTask(J) returns the tasks from the job
        %    object in three output variables, grouped by their current state.
        %
        %    If J is contained in a remote service, findTask will result in a call to
        %    the remote service. This could result in findTask taking a long time to
        %    complete, depending on the number of tasks retrieved and the network
        %    speed. Also, if the remote service is no longer available, an error will
        %    be thrown.
        %
        %    See also parallel.Job/createTask.
            import parallel.internal.customattr.WorkUnitFinder;
            
            % Convert any string inputs to character vectors
            [varargin{:}] = parallel.internal.general.convertPVPairsToChars(varargin{:});

            validateattributes( job, {'parallel.Job'}, {'nonempty'}, ...
                'findTask', 'job', 1 );
            errorIfInvalid( job );

            if ~any( nargout == [ 0 1 3 ] )
                error(message('parallel:job:FindJobNargout'));
            end
            try
                % Make sure the jobs are in a row vector, so we can assume
                % this for concatenation.
                job = job(:)';
                if nargout == 3
                    [tp, tr, tf] = parallel.internal.cluster.hetfun( @getTasksByState, job );
                    % each output is a cell of cells, need to unpick first to
                    % straight cell array
                    tp = [tp{:}]; tr = [tr{:}]; tf = [tf{:}];
                    % And then reconstitute into a 3-cell containing columns.
                    tasksCell = { [tp{:}].', [tr{:}].', [tf{:}].' };
                else
                    % Use getTasks so that it can be vectorized for MJS to
                    % minimize the number of remote calls. Support heterogeneous
                    % job arrays using hetfun.
                    tl = parallel.internal.cluster.hetfun( @hGetTasks, job );
                    
                    % Concatenate all the tasks from the heterogeneous array.
                    tl = [ tl{:} ];
                    if ~isempty( tl )
                        % Now concatenate all the tasks into one array.
                        tasksCell = { vertcat( tl{:} ) };
                    else
                        tasksCell = {};
                    end
                end
                varargout = cellfun( @(x) WorkUnitFinder.find(x, varargin), ...
                    tasksCell, 'UniformOutput', false );
            catch E
                throw( E );
            end
        end

        function okout = wait( job, state, timeout )
        %WAIT Wait for job execution to change state
        %    WAIT(J) blocks execution in the client MATLAB session until the
        %    single job identified by the scalar job object J reaches the
        %    'finished' state. This occurs when all its tasks are finished
        %    processing on remote workers.
        %
        %    WAIT(J, STATE) blocks execution in the client session until the
        %    job object J changes state to STATE. For a job object the
        %    valid states are 'queued', 'running' and 'finished'.
        %
        %    OK = WAIT(J, STATE, TIMEOUT) blocks execution in the client session
        %    until the job object J changes state to STATE or until TIMEOUT
        %    seconds elapsed, whichever happens first. OK is TRUE if state has
        %    been reached or FALSE in case of a timeout.
        %
        %    If a job has previously been in state STATE, then wait will
        %    return immediately. For example, if a job in the 'finished' state is asked
        %    to WAIT(JOB, 'queued'), then the call will return immediately.
        %
        %    Example:
        %    % Create a job object.
        %    myCluster = parcluster; % create a cluster object from default profile
        %    j = createJob(myCluster);
        %    % Add a task object that generates a 10x10 random matrix.
        %    t = createTask(j, @rand, 1, {10,10});
        %    % Run the job.
        %    submit(j);
        %    % Wait until the job is finished.
        %    wait(j, 'finished');
        %
        %    See also parallel.Job/submit, parallel.Job/fetchOutputs,
        %             parallel.Job/delete.

            import parallel.internal.types.States;

            validateattributes( job, {'parallel.Job'}, {'nonempty', 'scalar'}, ...
                                'wait', 'job', 1 );
            errorIfInvalid( job );

            if nargin < 2
                state = States.Finished;
            else
                try
                    % Convert any string inputs to character vectors
                    state = convertStringsToChars(state);
                    state = States.fromName( state );
                catch E
                    ME = MException(message('parallel:cluster:FailedToInterpretState'));
                    ME = addCause( ME, E );
                    throw( ME );
                end
                okWaitStates = [ States.Queued, States.Running, States.Finished ];
                okWaitState  = any( state == okWaitStates );
                if ~okWaitState
                    okNames = arrayfun( @(x) x.Name, okWaitStates, 'UniformOutput', false );
                    okNamesStr = strtrim( sprintf( '%s ', okNames{:} ) );
                    error(message('parallel:cluster:InvalidStateToWait', state.Name, okNamesStr));
                end
            end

            if nargin < 3
                timeout = Inf;
            end

            try
                dctSchedulerMessage(2, 'Waiting for %s to reach %s', ...
                    parallel.internal.logging.LogHelper.getJobsString(job), state.Name);
                ok = job.doWait( state, timeout );
                dctSchedulerMessage(2, 'Finished waiting for %s to reach %s', ...
                    parallel.internal.logging.LogHelper.getJobsString(job), state.Name);
            catch E
                throw( E );
            end
            % Only set ans if we ask for output, or one is documented
            if nargout > 0 || nargin == 3
                okout = ok;
            end
        end

        function outputCell = fetchOutputs( job )
        %fetchOutputs  Retrieve output arguments from all tasks in a job
        %    data = fetchOutputs(job) returns data, the output arguments
        %    contained in the tasks of a finished job. If the scalar job has M
        %    tasks, each row of the M-by-N cell array data contains the output
        %    arguments for the corresponding task in the job. Each row has N
        %    elements, where N is the greatest number of output arguments from
        %    any one task in the job. The N elements of a row are arrays
        %    containing the output arguments from that task. If a task has less
        %    than N output arguments, the excess arrays in the row for that task
        %    are empty. The order of the rows in data will be the same as the
        %    order of the tasks contained in the job's 'Tasks' property.
        %
        %    Note that issuing a call to fetchOutputs will not remove the
        %    output data from the location where it is stored. To remove the output
        %    data, use the delete function to remove individual tasks or their
        %    parent job object.
        %
        %    fetchOutputs will report an error if the job is not in the
        %    'finished' state, or if one of the Tasks encountered an error
        %    during execution. If some tasks completed successfully, then the
        %    output arguments can be accessed directly from the OutputArguments
        %    property of the task.
        %
        %    Example:
        %    % Create a job to generate a random matrix.
        %    myCluster = parcluster; % create a cluster object from default profile
        %    j = createJob(myCluster, 'Name', 'myJob');
        %    t = createTask(j, @rand, 1, {10});
        %    submit(j);
        %    % Wait until the job is finished.
        %    wait(j);
        %    % Get the random matrix.
        %    data = fetchOutputs(j);
        %
        %    See also parallel.Job/wait, parallel.Job/delete.

            validateattributes( job, {'parallel.Job'}, {'nonempty', 'scalar'}, ...
                                'fetchOutputs', 'job', 1 );
            errorIfInvalid( job );

            jobState = job.StateEnum;
            if jobState == parallel.internal.types.States.Unavailable
                error(message('parallel:job:JobUnavailable'));
            elseif jobState == parallel.internal.types.States.Failed
                job.Parent.hIssueFailedJobError(job);
            elseif jobState ~= parallel.internal.types.States.Finished
                error(message('parallel:job:JobNotYetFinished'));
            end
            tasks      = job.Tasks;
            if job.Variant == parallel.internal.types.Variant.CommunicatingPoolJob
                tasks  = tasks(1);
            end
            dctSchedulerMessage(4, 'Fetch outputs for %s', ...
                    parallel.internal.logging.LogHelper.getJobsString(job));
            % Collect all task properties required in a single vectorized
            % call
            tasksProps = get(tasks, {'NumOutputArguments', 'Error', 'ErrorMessage', ...
                'OutputArguments', 'ErrorIdentifier', 'Warnings'});
            numOutputArguments = tasksProps(:,1);
            errors = tasksProps(:,2);
            errorMessages = tasksProps(:,3);
            outputArguments = tasksProps(:,4);
            errorIdentifiers = tasksProps(:,5);
            warnings = tasksProps(:,6);

            % Warning if one of the output of a CommunicatingJob is a
            % Composite or a distributed array.
            if ~isempty(outputArguments)
                job.hWarnIfForbiddenOutput(outputArguments{1});
            end

            outputCell = cell( numel( tasks ), max( [numOutputArguments{:}] ) );
            for ii = 1:numel( tasks )
                thisTask   = tasks(ii);
                thisTaskID = thisTask.ID;

                % If there were warnings issued we want to list them.
                % We issue warnings before throwing any error to make sure they're not missed.
                parallel.Task.hWarnAboutWarnings(warnings{ii}, 'parallel:job:WarningsDuringTaskExecution', {thisTaskID});

                if isempty( errors{ii} ) && isempty( errorMessages{ii} )
                    thisTaskOutputs = outputArguments{ii};
                    numExpectedOutputs = numOutputArguments{ii};
                    numActualOutputs = numel( thisTaskOutputs );
                    if ( numActualOutputs ~= numExpectedOutputs )
                        error(message('parallel:job:OutputArgumentsMismatch', thisTaskID, ...
                            numActualOutputs, numExpectedOutputs));
                    end
                    outputCell(ii, 1:numOutputArguments{ii}) = thisTaskOutputs(:);
                else
                    % Throw the task error.
                    E2 = MException(message('parallel:job:ErrorDuringTaskExecution', thisTaskID));
                    if ~isempty( errors{ii} )
                        E2 = addCause( E2, errors{ii} );
                    else
                        % Should never get here.
                        E2 = addCause( E2, MException( errorIdentifiers{ii}, '%s', errorMessages{ii} ) );
                    end
                    throw( E2 );
                end
            end
        end

        function delete( jobOrJobs, varargin )
        %DELETE  Remove a job object from its cluster and memory
        %    DELETE(j) removes the job object, j, from the local session, and
        %    removes the job from the cluster's JobStorageLocation.  When the
        %    job is deleted, references to it become invalid. Invalid objects
        %    should be removed from the workspace with the CLEAR command. If
        %    multiple references to an object exist in the workspace, deleting
        %    one reference to that object invalidates the remaining references
        %    to it.  These remaining references should be cleared from the
        %    workspace with the CLEAR command.
        %
        %    The task objects contained in a job will also be deleted when a job
        %    object is deleted. Any references to those task objects will also be
        %    invalid, and you should clear them from the workspace.
        %
        %    If j is an array of job objects and one of the objects cannot be
        %    deleted, the other objects in the array will be deleted and a
        %    warning will be returned.
        %
        %    Example:
        %    % Create a job object.
        %    myCluster = parcluster; % create a cluster object from default profile
        %    j = createJob(myCluster, 'Name', 'myJob');
        %    t = createTask(j, @rand, 1, {10});
        %    % Delete the job object.
        %    delete(j);
        %    clear j t
        %
        %    See also parallel.Job/wait, parallel.Job/fetchOutputs, parallel.Task/delete.

            % varargin is specified to ensure we get dispatched correctly; but,
            % we don't support any additional arguments other than the magic
            % string 'invalidateOnly' which allows us to invalidate the handle
            % but not destroy the underlying data.
            narginchk( 1, 2 );
            try
                if nargin == 2
                    if ~isequal( varargin{1}, 'invalidateOnly' )
                        error(message('parallel:job:InvalidDeleteSyntax'));
                    end
                    delete@parallel.internal.customattr.CustomGetSet( jobOrJobs );
                    return
                end
                
                dctSchedulerMessage(2, 'Deleting %s', ...
                    parallel.internal.logging.LogHelper.getJobsString(jobOrJobs));
                parallel.internal.cluster.hetfun( @iDeleteJobs, jobOrJobs );           
                for ii = 1:numel( jobOrJobs )
                    % Use the built-in delete method to invalidate the objects
                    delete@parallel.internal.customattr.CustomGetSet( jobOrJobs(ii) );
                end
            catch E
                % Catching and rethrowing to remove stack trace
                throw(E)
            end
        end

        function cancel( jobOrJobs, optMessage )
        %CANCEL  Cancel a pending, queued, or running job
        %    CANCEL(J) stops the job object, J, that is pending, queued or running.
        %    The job's State property is set to finished, and a cancel is executed
        %    on all tasks in the job that are not in the finished state.  Any results
        %    that have been computed for the job object are saved and may be accessed
        %    normally.  A job object that has been canceled cannot be started again.
        %
        %    CANCEL(J,'MESSAGE') cancels the job with an additional
        %    user-specified message. This message will be added to the default
        %    cancellation message. The cancellation message is used to build the
        %    tasks' 'Error' property.
        %
        %    If the job is running in a cluster, any worker sessions that are
        %    evaluating tasks belonging to the job object will be restarted.
        %
        %    If the job is already in the finished state, no action is taken.
        %
        %    Example:
        %    myCluster = parcluster; % create a cluster object from default profile
        %    j = createJob(myCluster);
        %    cancel(j);
        %
        %    See also parallel.Job/submit, parallel.Task/cancel.

        
            import parallel.internal.cluster.CancelException

            validateattributes( jobOrJobs, {'parallel.Job'}, {}, 'cancel', 'job', 1 );
            errorIfInvalid( jobOrJobs );
            
            if nargin < 2
                exceptionBuilder = @(username) CancelException.getDefaultForJob( username );
            else
                % Convert any string inputs to character vectors
                optMessage = convertStringsToChars(optMessage);

                validateattributes( optMessage, {'char'}, {}, ...
                    'cancel', 'message', 2 );
                exceptionBuilder = @(username) CancelException.getWithUserMessageForJob( username, optMessage );
            end

            try
                dctSchedulerMessage(2, 'Cancelling %s', ...
                    parallel.internal.logging.LogHelper.getJobsString(jobOrJobs));
                parallel.internal.cluster.hetfun( @(jobs) iCancelJobs( jobs, exceptionBuilder ), jobOrJobs );
            catch E
                % Catching and rethrowing to remove stack trace
                throw(E)
            end
        end
    end

    methods ( Sealed, Hidden )
        function hInvalidate( jobOrJobs )
        %hInvalidate Invalidate job objects without destroying underlying data.
        % Note that this will ONLY invalidate this set of objects - there
        % is no invalidation of tasks or other objects associated with this
        % job.
            delete( jobOrJobs, 'invalidateOnly' );
        end
    end
    % --------------------------------------------------------------------------
    % API1 property/method compatibility. Properties needed by execution
    % framework.
    properties ( Hidden, Dependent )
        PathDependencies
        FileDependencies
    end

    % Name-changed compatibility
    properties ( Hidden, Dependent )
        Id
    end

    methods
        function pd = get.PathDependencies( job )
            pd = job.AdditionalPaths;
        end
        function pd = get.FileDependencies( job )
            pd = job.AttachedFiles;
        end
        function Id = get.Id( job )
            Id = job.ID;
        end
    end

    % --------------------------------------------------------------------------
    % API1 property/method compatibility. Methods needed to provide backwards
    % compatibility for batch
    methods (Hidden)
        
        function destroy( jobOrJobs )
            try
                delete( jobOrJobs );
            catch err
                throw(err);
            end
        end
        function outputCell = getAllOutputArguments( job )
            try
                outputCell = job.fetchOutputs();
            catch err
                throw(err);
            end
        end
      
    end
    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Execution framework - compatibility layer
    methods ( Hidden, Abstract )
        tf = pCurrentTaskIsPartOfAPool( job )
        pPreJobEvaluate( job, task )
        pPostJobEvaluate( job )
        
        % hGetStateEnum must return a parallel.internal.types.States for this job
        s = hGetStateEnum( obj )
    end
    methods ( Hidden, Sealed )
        function varargout = pReturnDependencyData( job )
            try
                % The hidden AttachedFilePaths contain all the paths for the AttachedFiles 
                % and the DependentFiles.
                attachedProps = job.hGetProperty( {'AttachedFilePaths', 'AttachedFileData'} );

                attachedFilePaths = attachedProps{1};
                attachedFileData = attachedProps{2};
                varargout = { attachedFilePaths, attachedFileData };
            catch E
                E2 = MException(message('parallel:job:CorruptData'));
                E2 = addCause( E2, E );
                throw( E2 );
            end
        end
    end
    methods ( Hidden )
        function tf = pIsMatlabPoolJob( job )
            tf = (job.Variant == parallel.internal.types.Variant.CommunicatingPoolJob);
        end
        
        function hWarnIfNotSubmitted( obj )
            if strcmp(obj.State, 'pending')
                parallel.internal.warningNoBackTrace(message('parallel:job:CurrentStatePending', obj.Name));
            end
        end

        function s = hGetStateEnumForDisplay( obj )
            s = obj.hGetStateEnum();
            % hGetStateEnum returns a cell array for vector inputs and a
            % scalar for a single input. The display code always requires
            % cell arrays, therefore we need to wrap the state in a cell
            % array.
            if numel(s) == 1
                s = {s};
            end
        end
    end

end
%#ok<*ATUNK> custom attributes

function dependencies = iGetDependencies( paths )
% Given a set of paths to attached files (which may include folders) return
% two cell arrays of cell arrays having the same size of paths:
% - dependencies lists all the dependencies for each element in paths

if isempty( paths )
    dependencies = {};
    return
end

% Do not call 'getCodeAndSubfolders' on Package directories, we wish to pass those through
% unmodified.
allFiles = paths;
isPkgFolder = parallel.internal.packages.isPackageDir(paths);
allFiles(~isPkgFolder) = ...
    parallel.internal.apishared.AttachedFiles.getCodeAndSubfolders(paths(~isPkgFolder));

dependencies = cellfun( ...
    @parallel.internal.apishared.AttachedFiles.calculateAttachedFiles, allFiles, ...
    'UniformOutput', false );
end

function iErrorIfNotIndependentJob(job, errorMessage)
if ~isa(job, 'parallel.IndependentJob')
    error(errorMessage);
end
end

% Method called by hetfun to delete jobs
function iDeleteJobs(jobs)
    % destroyJob is responsible for removing tasks too.
    try
        destroyJob( jobs );
    catch E
        throw( E );
    end
end

% Method called by hetfun to cancel jobs
function iCancelJobs(jobs, exceptionBuilder)
    cancelException = arrayfun(@(job) exceptionBuilder( job.hGetClusterUsername() ), jobs);
    cancelJob( jobs, cancelException );
end
