%ConcurrentJob - Job type for non-SPMD parallel pool implementation.

% Copyright 2012-2021 The MathWorks, Inc.

classdef ( Hidden ) ConcurrentJob < 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

    properties ( Access = protected )   
        % Always true for a ConcurrentJob as these are only to support interactive
        % pools.
        IsPoolTask = true; 
        
        % 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 = ConcurrentJob( parent, id )
            obj@parallel.Job( parent, id, ...
                              parallel.internal.types.Variant.ConcurrentJob );
        end
        function functor = getBroadcastFunctor( ~, ~ )
        % Most concurrent jobs cannot broadcast port proxying information so they simply
        % throw an error here. Subclasses may override this if they can
        % broadcast information.
            functor = @(varargin) ...
                      error(message('parallel:job:ConcurrentJobUnsupportedOperation'));
        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', ...
                'Separator',...
                'AutoAttachFiles', ...
                'Auto Attached Files', ...
                'AutoAddClientPath', ...
                'AttachedFiles', ...
                'AdditionalPaths' ...
                        };

            % 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

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % API1 compatibility execution framework and properties
    methods ( Hidden ) % API1 compatibility layer
        function tf = pIsMatlabPoolJob( job )
        % By default, ConcurrentJobs are always parallel pool jobs. However, to support
        % testing outside of a parallel pool, allow override via the ApiTag.
            tf = ~isequal( job.ApiTag, 'qeNonPoolConcurrent' );
        end
        function tf = pCurrentTaskIsPartOfAPool( job )
            tf = pIsMatlabPoolJob( job ) && job.IsPoolTask;
        end
        function pInstantiatePool( job, task, ~, ~, args )
            labIdx         = task.ID;
            nLabs          = job.Parent.hGetNumConcurrentTasks( job );
            [job.IsPoolTask, job.Pool] = parallel.internal.apishared.PoolFactory.instantiate( job.pIsInteractivePool(), ...
                labIdx, nLabs, getBroadcastFunctor(job, task), ...
                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
