%CommunicatingJob - base class for all communicating job objects
%   A communicating job object contains all the tasks that define what each
%   worker does as part of complete job execution. Workers executing a
%   communicating job all run at the same time, and can communicate with each
%   other. The nature of the communication allowed is determined by the 'Type'
%   of communicating job.
%
%   All communicating jobs may have only a single task defined before
%   submission.
%
%   There are two types of communicating job: 'Pool' and 'SPMD'. These can
%   be selected by using the 'Type' argument to the createCommunicatingJob
%   method.
%
%   'Pool' jobs run the task function on a single worker, and that task function
%   can use the remaining workers as a parallel pool to run PARFOR loops and SPMD
%   blocks.
%
%   'SPMD' jobs run the task function on each worker simultaneously, as if the
%   entire task function was being executed within a single SPMD block.
%
%   parallel.CommunicatingJob 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.CommunicatingJob 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
%      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
%      NumWorkersRange   - Specify limits for number of workers
%      Parent            - Cluster object containing this job
%      RunningDuration   - The duration the job has been running for, if it has started
%      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
%
%   See also parallel.Job, parallel.Cluster/createCommunicatingJob.

% Copyright 2011-2021 The MathWorks, Inc.

classdef CommunicatingJob < parallel.Job

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Public / documented properties
    properties ( Dependent, PCTConstraint = 'workerlimits|presubmission' )
        %NumWorkersRange Specify limits for number of workers
        %   NumWorkersRange specifies an upper and lower bound for the number of
        %   workers to be used for communicating job execution. If a scalar
        %   value is specified, then this is used as both the lower and upper
        %   bound, and the job runs on exactly that number of workers.
        NumWorkersRange
    end
    methods
        function v = get.NumWorkersRange( job )
            v = job.hGetProperty( {'MinNumWorkers', 'MaxNumWorkers'} );
            v = [v{:}];
        end
        function set.NumWorkersRange( job, args )
            job.hSetPropertyNoCheck( {'MinNumWorkers', 'MaxNumWorkers'}, {args(1) args(end)} );
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Protected/Hidden methods and properties
    properties ( Access = protected )
        % Iff this is a PoolJob, this property is true if the currently executing
        % process is running as part of the pool (rather than the "lead task" in
        % a non-interactive PoolJob).
        IsPoolTask = false;
        
        % This will contain the appropriate representation of a pool. On a
        % client it will be a parallel.Pool, on a worker it will be a
        % SessionObject. Note that in a pool job if the pool is size one, then 
        % this will be a SessionObject.  
        Pool = parallel.Pool.empty();
        
        % Default PoolShutdownSuccessful to 'true' so that non-pool jobs don't
        % think something has gone wrong.
        PoolShutdownSuccessful = true;
    end
    
    properties ( Dependent, SetAccess = private, Hidden )
        MinNumWorkers;
    end
    
    methods
        function val = get.MinNumWorkers( obj )
            val = obj.NumWorkersRange(1);
        end
    end
    
    methods ( Access = protected )
        function obj = CommunicatingJob( parent, id, variant )
            obj@parallel.Job( parent, id, variant );
        end
    end

    methods ( Hidden )
        function [jobPropertyMap, propNames]  = hGetDisplayItems(obj, diFactory)
            jobPropertyMap = hGetDisplayItems@parallel.Job(obj, diFactory);
            
            % The order the properties appear in the propNames array is the
            % order in which they will be displayed.
            propNames = {...
                'ID', ...
                'Type', ...
                'Username', ...
                'State', ...
                'SubmitDateTime', ...
                'StartDateTime', ...
                'RunningDuration', ...
                'NumWorkersRange', ...
                'NumThreads', ...
                'Separator', ...
                'AutoAttachFiles', ...
                'Auto Attached Files', ...
                'AttachedFiles', ...
                'AutoAddClientPath', ...
                'AdditionalPaths' ...
                };
           
            % Do not display EnvironmentVariables for Local clusters as it
            % is a redundant property.
            if ~isa(obj.Parent, 'parallel.cluster.Local')
                propNames = [propNames, 'EnvironmentVariables'];
            end
            
            % Add Communicating job specific displayable items to map.
            jobPropertyMap('NumWorkersRange') = diFactory.createDefaultItem(obj.hGetDisplayPropertiesNoError('NumWorkersRange'));
        end
        
        function tf = hIsPoolInteractive( job )
        % Return true if this is a CommunicatingPoolJob to be used interactively.
            tf = ( job.hGetProperty( 'ExecutionMode' ) == 1 );
        end
        function hSetInteractive( job, tf )
        % Indicate that this job is to be used for interactive purposes
            validateattributes( tf, {'logical'}, {'scalar'} );
            mode = double(tf);
            job.hSetPropertyNoCheck( 'ExecutionMode', mode );
        end
    end
    
    methods ( Hidden )
        function hWarnIfForbiddenOutput( job, outputs )
            %hWarnIfForbiddenOutput Issue a warning if one of the outputs
            % is a distributed array or a Composite and the
            % CommunicatingJob is not "spmd"
            
            if ~strcmp(job.Type, "spmd")
                for k = 1:numel(outputs)
                    outArgk = outputs(k);
                    % If the output is a struct, we check for all
                    % the fields' values.
                    if isstruct(outputs{k})
                        outArgk = struct2cell(outputs{k});
                    end
                    if any(cellfun(@(x) isa(x, 'Composite') || isa(x, 'distributed'), outArgk))
                        warning(message('parallel:convenience:WarningInvalidOutputCommunicationJob'));
                        return
                    end
                end
            end
            
        end
        
    end
    
    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % API1 compatibility execution framework and properties
    methods ( Hidden ) % API1 compatibility layer
        function tf = pCurrentTaskIsPartOfAPool( job )
            tf = job.IsPoolTask;
        end
        function pInstantiatePool( job, ~, ~, ~, args )
            % Capture value of numlabs right now because pool instantiation
            % involves setting up the MPI communicators for pool execution, and
            % this means that on the 'client' lab, numlabs will become 1.
            nLabs = numlabs;
            [job.IsPoolTask, job.Pool] = parallel.internal.apishared.PoolFactory.instantiate( job.pIsInteractivePool(), ...
                labindex, nLabs, iGetLabBroadcastFunctor(), args );
        end
        function OK = pShutdownPool( job )
            try
                parallel.internal.mpi.ensureDeadlockDetectionOff();
                delete(job.Pool);
                job.PoolShutdownSuccessful = true;
            catch err
                job.PoolShutdownSuccessful = false;
                throw(err)
            end
            OK = job.PoolShutdownSuccessful;
        end
        function tf = pIsInteractivePool( job )
            tf = job.hIsPoolInteractive();
        end
    end
end
%#ok<*ATUNK> custom attributes

% Used during pool setup. The functor here simply defers to labBroadcast.
function functor = iGetLabBroadcastFunctor()
    functor = @(varargin) labBroadcast(1, varargin{:});
end
