%Cluster Base class for all Cluster objects
%   parallel.Cluster is the base class for all cluster objects. Specific types
%   of cluster can be constructed either by creating a profile and using the
%   PARCLUSTER function, or by calling the subclass constructor directly.
%
%   All parallel.Cluster subclass constructors accept property-value pairs as
%   arguments. Supported properties depend on the specific subclass.
%
%   parallel.Cluster methods:
%      batch                  - Run MATLAB script or function as batch job
%      createCommunicatingJob - Create a new communicating job
%      createJob              - Create a new independent job
%      findJob                - Find job objects belonging to a cluster object
%      isequal                - True if clusters have same property values
%      parpool                - Create a parallel pool of workers on a cluster and return a pool object
%      saveAsProfile          - Save modified properties to a profile
%      saveProfile            - Save modified properties to the current profile
%
%   parallel.Cluster properties:
%      ClusterMatlabRoot                - Specifies the path to MATLAB for workers to use
%      Host                             - Host of the cluster's headnode
%      JobStorageLocation               - Location where cluster stores job and task information
%      Jobs                             - List of jobs contained in this cluster
%      LicenseNumber                    - License number to use with online licensing
%      Modified                         - True if any properties in this cluster have been modified
%      NumThreads                       - Number of computational threads for workers
%      NumWorkers                       - Number of workers available for this cluster
%      OperatingSystem                  - Operating system of the nodes used by the cluster
%      Profile                          - Profile used to build this cluster
%      RequiresOnlineLicensing          - Cluster uses online licensing
%      Type                             - Type of this cluster
%      UserData                         - Data associated with a cluster object in this client session
%
%   parallel.Cluster types:
%      parallel.cluster.Generic         - Interact with a Generic cluster type
%      parallel.cluster.HPCServer       - Interact with a Microsoft HPC Server Cluster
%      parallel.cluster.Local           - Interact with a Local cluster
%      parallel.cluster.LSF             - Interact with a Platform LSF cluster
%      parallel.cluster.MJS             - Interact with an MJS cluster
%      parallel.cluster.MJSComputeCloud - Interact with an MJS cluster that was started using on the cloud
%      parallel.cluster.PBSPro          - Interact with a PBS Pro cluster
%      parallel.cluster.Slurm           - Interact with a Slurm cluster
%      parallel.cluster.Torque          - Interact with a Torque cluster
%
%   See also PARCLUSTER, BATCH, PARPOOL.

% Copyright 2011-2020 The MathWorks, Inc.

classdef Cluster < handle ...
        & matlab.mixin.Heterogeneous ...               % We can concatenate disparate Clusters
        & parallel.internal.cluster.CustomPropDisp ... % For display
        & parallel.internal.customattr.CustomGetSet    % For PCT* property attributes

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Implementing properties for Cluster subclasses:
    % * All Profile properties must be PCTGetSet and Transient
    % * All such properties specified are liable to be saved into a resulting Profile
    % * All other implementation properties must not be PCTGetSet
    % * All Profile properties must be permissible P-V pairs during construction
    % * The constructed object must faithfully reflect the values specified
    %   during construction
    % * The constructor should use ConstructorArgsHelper.interpretClusterCtorArgs as
    %   this will perform constraint checks as well as checking for bad properties
    % * All constructors should invoke the Cluster constructor and then, for each
    %   PCTGetSet property defined in that class, call obj.PropStorage.addFactoryValue()
    % * Any P-V pairs requiring special handling should then handle those, updating
    %   the PropStorage if necessary to indicate the source of the value
    % * The constructor should then invoke obj.applyConstructorArgs( propsFromArgs )
    %   which will call hSetProperty for each property remaining, as well as
    %   indicating correctly to the PropStorage where the arg value came from.
    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Common parallel.Cluster properties

    % Several properties of all Clusters are selectable only at construction
    % time. Therefore, they also must be specified in the Cluster constructor.
    properties ( SetAccess = immutable )
        %Type Type of this cluster
        %   The type of a cluster is the last part of the cluster class name,
        %   without the 'parallel.cluster' prefix.
        Type
    end

    properties ( Transient, Dependent, SetAccess = private )
        %Profile Profile used to build this cluster
        %   This property contains the name used to build this cluster, or it
        %   is empty if none was used. This is the profile to which values
        %   will be saved when calling 'saveProfile'. Properties defined by
        %   this profile will be applied to all job and task objects
        %   constructed by this cluster.
        %
        %   See also parallel.Cluster.saveProfile, 
        %            parallel.Cluster.saveAsProfile, 
        %            PARCLUSTER.
        Profile
    end

    % Common public properties, declared Abstract as these might be PCTGetSet in
    % subclasses
    properties ( Abstract )
        %NumWorkers Number of workers available for this cluster
        %   NumWorkers indicates the number of workers available to the
        %   cluster for running your jobs. NumWorkers must fall within the
        %   range defined by a job's NumWorkersRange.
        %   (read-only)
        %
        %   See also parallel.CommunicatingJob.NumWorkersRange
        NumWorkers

        %ClusterMatlabRoot Specifies the path to MATLAB for workers to use
        %   ClusterMatlabRoot specifies the path name to MATLAB for the
        %   cluster to use for starting MATLAB worker processes. 
        %
        %   See also parallel.Cluster.JobStorageLocation
        ClusterMatlabRoot

        %OperatingSystem Operating system of the nodes used by the cluster
        %   OperatingSystem specifies the operating system of the nodes in
        %   the cluster. Valid values are 'windows', 'unix', and 'mixed'.
        %
        %   See also parallel.Cluster.Host
        OperatingSystem

        %JobStorageLocation Location where cluster stores job and task information
        %
        %   See also parallel.Cluster.ClusterMatlabRoot
        JobStorageLocation
        
        %LicenseNumber License number to use with online licensing
        %   License number specifies the license number to use on the workers
        %   when the cluster is using online licensing.  This string
        %   must be the same as one of the licenses listed in your MathWorks
        %   account.
        %
        %   LicenseNumber cannot be defined if RequiresOnlineLicensing is
        %   false.
        %    
        %   See also parallel.Cluster.RequiresOnlineLicensing
        LicenseNumber
        
        %RequiresOnlineLicensing Cluster uses online licensing
        %   RequiresOnlineLicensing indicates whether the cluster uses
        %   online licensing.  You must provide your MathWorks account
        %   login information if this value is true.  If you have more than one
        %   MATLAB Parallel Server license associated with your
        %   MathWorks account you will also need to supply the LicenseNumber to
        %   use.
        %
        %   See also parallel.Cluster.LicenseNumber
        RequiresOnlineLicensing
    end
    
    properties ( Abstract, SetAccess = immutable )
        %Host Host of the cluster's headnode
        %   You can use the host of the cluster in a profile to find a
        %   desired cluster when creating the cluster object with
        %   parcluster or the cluster constructor.
        %
        %   See also PARCLUSTER
        Host
    end

    properties ( PCTGetSet, Transient, PCTConstraint = 'positivenonzerofiniteintscalar' )
        %NumThreads Number of computational threads for workers 
        %   NumThreads specifies the maximum number of computational 
        %   threads used by workers. The default value number of 
        %   computational threads is 1.  The number of threads
        %   must be a finite value greater than 0.
        NumThreads = 1
    end

    properties ( Transient )
        %UserData Data associated with a cluster object in this client session
        %   UserData can be used to store any data associated with a cluster
        %   object. This data is stored in the client MATLAB session, and is
        %   not available on the workers. UserData is not stored in a profile.
        %
        %   Commands such as 
        %       clear all 
        %       clear functions 
        %   will clear an object in the local session, permanently removing
        %   the data in the UserData property.
        %   
        %   The default value is an empty vector.
        %   
        %   See also CLEAR
        UserData = []
    end

    properties ( Dependent, SetAccess = private )
        %Jobs List of jobs contained in this cluster
        %   An array of all the job objects in the cluster. Job objects
        %   will be in the order indicated by their ID property, consistent
        %   with the sequence in which they were created, regardless of
        %   their State.
        %
        %   See also parallel.Cluster.createJob,
        %            parallel.Cluster.createCommunicatingJob,
        %            parallel.Cluster.findJob,
        %            parallel.Job.submit, parallel.Job.delete.
        Jobs

        %Modified True if any properties in this cluster have been modified
        %   A cluster is 'Modified' if properties have been changed since it was
        %   created using a profile. If no profile was used, then the cluster
        %   will be modified if any properties are not in their default
        %   state. Use 'saveProfile' to modify the underlying profile.
        %
        %   See also parallel.Cluster.saveProfile,
        %            parallel.Cluster.Profile.
        Modified
    end
    
    properties ( Transient, Hidden, SetAccess = protected )
        % Whether or not this scheduler is considered to be shared.  
        % All schedulers are considered to be shared apart from MathWorks 
        % Hosted Cloud.
        IsSharedCluster = true
    end
    
    properties ( Transient, Access = private )
        % The actual profile object that corresponds to this
        % cluster.
        ProfileObject
        ProfileDeletedListener
        MatlabLogoutListener
    end

    methods % Get/Set for Dependent
        function jobs = get.Jobs( obj )
        % Use the abstract method to retrieve the Jobs list.
            try
                jobs = obj.getJobs();
                jobs = reshape( jobs, numel(jobs), 1 );
            catch E
                throw( E );
            end
        end
        function tf = get.Modified( obj )
            tf = obj.PropStorage.UserModified;
        end
        
        function name = get.Profile( obj )
            if isempty(obj.ProfileObject)
                name = '';
            else
                name = obj.ProfileObject.Name;
            end
        end
        
        function set.Profile( obj, profileName )
            if isempty( profileName )
                profileObj = parallel.settings.Profile.empty();
            else
                p = parallel.Settings;
                profileObj = p.findProfile('Name', profileName);
            end
            obj.setProfileObjectAndListeners( profileObj );
        end
    end

    properties ( Transient, GetAccess = protected, SetAccess = immutable )
        PropStorage % Used for PCTGetSet properties, and tracking properties
                    % set from Profiles or during construction.
    end
    
    properties ( Transient, Access = protected )
        LicenseEntitlement = parallel.internal.mwa.Entitlement.empty();
    end

    properties ( Constant, GetAccess = private )
        EnforceInitClient = initclient();

        %MinimalMementoProperties A string array that defines the
        % properties needed for a minimal cluster memento (i.e. one that
        % allows a cluster object to be constructed that can find the
        % original cluster's jobs). A property should only be included if
        % it is required in the memento for every sub-class of
        % parallel.Cluster.
        %
        % A sub-class can define additional properties to be included in
        % the memento in a constant, private property of the same name. A
        % sub-class should not include properties that are included by any
        % of its super-classes.  A super-class should only define a
        % property if it should be included in the memento of each of its
        % sub-classes.
        %
        % The full list of properties required by any class can be gathered
        % using hDetermineMementoProperties.
        MinimalMementoProperties = string.empty;
        
        %FullMementoProperties A string array that defines the properties
        % needed, in addition to those listed in MinimalMementoProperties,
        % for a full cluster memento (i.e. one that allows a cluster object
        % to be constructed that can find the original cluster's jobs). A
        % property should only be included if it is required in the memento
        % for every sub-class of parallel.Cluster.
        %
        % A sub-class can define additional properties to be included in
        % the memento in a constant, private property of the same name. A
        % sub-class should not include properties that are included by any
        % of its super-classes.  A super-class should only define a
        % property if it should be included in the memento of each of its
        % sub-classes.
        %
        % The full list of properties required by any class can be gathered
        % using hDetermineMementoProperties.
        FullMementoProperties = "NumThreads";
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Cluster abstract methods

    % These methods must be provided by cluster implementations:
    methods ( Abstract, Access = protected )
        % Return a list of parallel.Job objects:
        jobs  = getJobs( obj )

        % parallel.Cluster will use this method when a new
        % parallel.IndependentJob is to be built.
        job  = buildIndependentJob( obj )

        % parallel.Cluster will use this method when a new
        % parallel.CommunicatingJob is to be built. variant will be either Pool
        % or Spmd.
        job  = buildCommunicatingJob( obj, variant )

        job  = buildConcurrentJob( obj )

        % build a ...pmode.poolmessaging.ConnectionManager
        connMgr = buildConnectionManager( obj )

        % Implementation behind hGetNumConcurrentTasks
        num = getNumConcurrentTasks( obj, job )
        
        % Get all the properties needed for the ClusterMemento
        props = getAllMementoProperties( obj, clusterMementoType )
    end
    
    methods ( Abstract, Hidden )
        % Return a provider which Job Monitor can use to pull updates
        provider = hCreateJobStatusProvider( obj );
    end
    
    methods ( Hidden, Sealed, Access = ?parallel.ConcurrentJob )
        function num = hGetNumConcurrentTasks( obj, job )
        % On a worker, provide the job with information about how many tasks were
        % actually launched. Forward to the protected method.
            num = getNumConcurrentTasks( obj, job );
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % helper methods
    methods ( Access = private )

        function applyJobProperties( obj, job, varargin )
            % This method is used when constructing a new job. This method
            % is used to set all the additional properties specified during
            % a call to create*Job. If anything goes wrong setting the
            % properties, it will throwAsCaller the error.
            try
                [names, values] = obj.parseJobArguments( metaclass( job ), varargin{:} );
                
                % Check that the user-specified properties exist and are
                % publicly settable on the job object.  This will error if
                % are not.
                job.checkValidUserSettablePropNames( names );
                
                % We also need to set the job's NumThreads value to that of
                % cluster object.  Since NumThreads isn't a public-settable
                % property, we need to use the Hidden 'hSetAllowNonPublic'
                % of parallel.Job.
                names = [names, 'NumThreads'];
                values = [values, obj.NumThreads];
                
                job.hSetAllowNonPublic( names, values );
            catch E
                try
                    job.delete();
                catch E2 %#ok<NASGU>
                    % We're about to throw the original error.
                end
                % Caller is top-level entry-point
                throwAsCaller( E );
            end
        end

        function [n, v] = getModifiedProps( obj )
        % Return a names and values of all the modified properties on this
        % Cluster.
            obj.updateDerivedProfileProperties();

            [n, v] = obj.PropStorage.getProfileProperties();
        end

        function savePropsToSchedulerComponent( obj, schedComp )
        % Save the cluster's properties to the supplied scheduler component.
            import parallel.internal.types.SettingsLevel;
            obj.updateDerivedProfileProperties();

            % Start by unsetting all user values - this ensures all props
            % will be in the "factory" state.  (This gives the same end 
            % result as only unsetting the PropStorage Factory values.)
            % Then set all user values and only set profile values if
            % they are not already set.
            [uN, uV] = obj.PropStorage.getUserProperties();
            [pN, pV] = obj.PropStorage.getProfileProperties();
            
            userLevel = SettingsLevel.User;
            restorePoint = schedComp.hGetRestorePoint(userLevel);
            try
                schedComp.unsetAll( userLevel.Name );
                schedComp.hSetFromCluster( uN, uV );
                schedComp.hSetFromClusterIfNotAlreadySet( pN, pV );
            catch err
                schedComp.hRestore(restorePoint);
                rethrow(err);
            end
        end
        
        function setClusterToUnmodified( obj )
            obj.PropStorage.setNonFactoryAsProfile();
        end
        
        function setProfileObjectAndListeners( obj, profileObj )
            obj.ProfileObject = profileObj;
            if isempty( profileObj )
                obj.ProfileDeletedListener = [];
            else
                obj.ProfileDeletedListener = event.listener( profileObj, ...
                    'ObjectBeingDestroyed', ...
                    @(~, ~) obj.handleProfileDeleted() );
            end
        end
        
        function handleProfileDeleted( obj )
            obj.ProfileObject = parallel.settings.Profile.empty();
            obj.ProfileDeletedListener = [];
        end
    end
        
    methods ( Static, Access = private )
        function [prof, schedComp, cleanupOnErrorFcn] = createProfile( type, profileName )
            import parallel.Cluster
            schedComp = Cluster.createSchedulerComponent( type, profileName );
            p = parallel.Settings;
            try
                prof = p.createProfile( profileName, schedComp.Name );
            catch err
                schedComp.delete();
                rethrow( err );
            end

            cleanupOnErrorFcn = @nDeleteProfileAndSchedComp;

            function nDeleteProfileAndSchedComp()
                prof.delete();
                schedComp.delete();
            end
        end

        function schedComp = createSchedulerComponent( type, profileName )
            p = parallel.Settings; 
            proposedName = sprintf( '%sSchedulerComponent', profileName );
            scName = p.hGetUnusedSchedulerComponentName( proposedName );
            schedComp = p.createSchedulerComponent( type, scName );
        end

    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Protected access methods.
    methods ( Access = protected )
        function [ names, values] = parseJobArguments(obj, jobMetaClass, varargin)
            % This method should be called during job construction to build up those
            % properties which 'append' during construction.
            import parallel.internal.customattr.Reflection
            import parallel.internal.cluster.ConstructorArgsHelper

            % Convert any string inputs to character vectors
            [varargin{:}] = parallel.internal.general.convertPVPairsToChars(varargin{:});

            % If the Cluster has a 'Profile', we prepend that to the job "set" args.
            if ~isempty(obj.Profile)
                jobArgs = ['Profile', obj.Profile, varargin];
            else
                jobArgs = varargin;
            end
            
            propsWhichAppend = {'AttachedFiles', 'AdditionalPaths', 'EnvironmentVariables'};

            [names, values] = ConstructorArgsHelper.resolveJobBuildArguments( ...
                jobMetaClass, propsWhichAppend, jobArgs{:});
        end

        function varargout = getJobsByState( obj )
        % Default implementation of getting jobs partitioned by their
        % State. This implementation is the obvious, simple
        % implementation. Cluster subclasses may be able to override this
        % implementation with something more efficient.

        % Note that we bundle 'failed' jobs in with 'finished'. 'unavailable'
        % jobs are dropped.
            import parallel.internal.types.States;

            allJobs   = obj.getJobs();
            jobStates = parallel.internal.cluster.hetfun( @(x) x.StateEnum, allJobs );
            jobStates = [ jobStates{:} ];
            states    = [ States.Pending, ...
                          States.Queued, ...
                          States.Running, ...
                          States.Finished, ...
                          States.Failed ];
            pqrff     = arrayfun( @(state) allJobs( jobStates == state ), states, ...
                                  'UniformOutput', false );
            % Need to concatenate Finished/Failed
            if nargout == 4
                varargout = { pqrff(1), pqrff(2), pqrff(3), {[pqrff{4:end}]} };
            else
                varargout = { [ pqrff(1:3), {[pqrff{4:end}]} ] };
            end
        end

        function E = buildCannotModify( obj, propertyName )
        % buildCannotModify use this if your Cluster subclass does not support
        % modification of a property.
            validateattributes( propertyName, {'char'}, {'row'} );
            m = message( 'MATLAB:class:SetProhibited', propertyName, ...
                         class( obj ) );
            E = MException( m.Identifier, '%s', m.getString() );
        end
        
        function tf = isNotCurrentValue( obj, name, value )
            validateattributes( name, {'char'}, {'row'} );
            tf = ~isequal( obj.(name), value );
        end

        function applyConstructorArgs( obj, propsFromArgs, applyHidden )
        % NB that even though very nearly all propsFromArgs are properties to be
        % stored in Profiles, UserData may appear here. So we need to be a
        % little more careful.
            import parallel.internal.customattr.Reflection

            names          = propsFromArgs.keys();
            valuesAndTypes = propsFromArgs.values();
            if nargin < 3
                % By default do not apply constructor args for Hidden
                % properties.
                applyHidden = false;
            end
            if applyHidden
                [props, useGetSet] = Reflection.getAllPublicProperties( obj );
            else
                [props, useGetSet] = Reflection.getPublicProperties( obj );
            end
            % Reorder the properties if necessary.
            [names, valuesAndTypes] = obj.reorderForSet( names, valuesAndTypes );

            try
                for ii = 1:length( names )
                    name         = names{ii};
                    idx          = find( strcmp( name, props ) );
                    if length( idx ) ~= 1
                        % Never get here, invalid props should already be stripped.
                        error(message('parallel:cluster:ConstructorArgInvalid', name));
                    end
                    thisGetSet   = useGetSet(idx);
                    valueAndType = valuesAndTypes{ii};
                    if thisGetSet
                        % use 'hSetProperty' here to give hSetProperty a chance to
                        % object. e.g. Local and NumWorkers.
                        obj.hSetProperty( name, valueAndType{1} );
                        obj.PropStorage.setValueType( name, valueAndType{2} );
                    else
                        obj.(name) = valueAndType{1};
                    end
                end
            catch E
                throwAsCaller( E );
            end
        end

        function updateDerivedProfileProperties( ~ )
        % updateDerivedProfileProperties this is to allow Clusters to have
        % properties that can be specified in a Profile, but also might be
        % 'derived' in the sense that they may be modified behind the scenes.
        % The motivating example here is the Local cluster which has
        % 'NumWorkers' which may be both specified by a Profile, but the true
        % value must be retrieved from the underlying scheduler. Subclasses
        % overriding this method should use the
        % ClusterProps.setValueAtSameLevel() method so that the value-type is
        % not modified, but the value is.
        end
        
        function props = getPublicPropertiesForComparison( obj )
            import parallel.internal.customattr.Reflection
            props = Reflection.getAllPublicProperties( obj );
        end

        function tf = scalarIsequal( obj1, obj2 )
        % Compare two scalar Cluster objects for 'isequal' behavior.            

            if isequal( class( obj1 ), class( obj2 ) )
                % Same class, compare properties *except* 'Jobs'.
                allProps = obj1.getPublicPropertiesForComparison();
                allPropsExceptJobs = setdiff( allProps, 'Jobs' );

                vals1  = get( obj1, allPropsExceptJobs );
                vals2  = get( obj2, allPropsExceptJobs );

                tf = isequal( vals1, vals2 );
            else
                tf = false;
            end
        end

        function tf = isEquivalentWorkers( obj1, obj2 )
        % isEquivalentWorkers returns TRUE if two clusters are equivalent
        % - in the sense that they submit to the same pool of workers.  obj1 and
        % obj2 are guaranteed to be of the same class. Clusters should override
        % this if they know a better way than to use the ISEQUAL test.
            tf = isequal( obj1, obj2 );
        end

        function tf = isEquivalentStorage( obj1, obj2 )
        % isEquivalentStorage takes two Clusters of the same class
        % and returns TRUE iff they store jobs in the same place.
            tf = isequal( obj1, obj2 );
        end

        function obj = Cluster( type, cProf )
        % Cluster superclass constructor for all Cluster types. This is Hidden
        % because there is no customer-visible behavior here.
            
            try
                parallel.internal.cluster.initializeClient();
            catch err
                throw(err);
            end
        
            % This will error in a deployed application if
            % this is the second MCR component in the application.
            parallel.internal.apishared.mcrShutdownHandler( 'initialize' );
            
            validateattributes( type, {'char'}, {'row'} );
            validateattributes( cProf, {'char'}, {} ); % Could be empty

            % simply set immutable properties.
            obj.Type        = type;
            obj.Profile     = cProf;
            obj.PropStorage = parallel.internal.settings.ClusterProps();
            obj.PropStorage.addFactoryValue( 'NumThreads', 1 );
        end

        function entitlement = checkLicenseNumber( obj, licenseNumber )
            % Empty license numbers are allowed, no need to check.
            % If the cluster does not even need online licensing, no need to check either.
            if isempty( licenseNumber ) || ~obj.RequiresOnlineLicensing
                entitlement = parallel.internal.mwa.Entitlement.empty();
                return
            end

            % Get the license category for this cluster type
            schedulerType = parallel.internal.types.SchedulerType.fromName( obj.Type );
            licenseCategory = schedulerType.LicenseCategory;
            
            % Get a list of all the current user's entitlements
            currentUser = parallel.internal.mwa.ensureLoggedIn();
            allEntitlements = currentUser.getAllEntitlements( licenseCategory );
            if isempty( allEntitlements )
                % If there are no entitlements then we have no MDCS
                % licenses and we should error
                error( message( 'parallel:cluster:NoMDCSLicense' ) );
            end
            
            % Check if the given license number matches any valid ones
            validLicenseNumbers = {allEntitlements.LicenseNumber};
            matches = strcmp( licenseNumber, validLicenseNumbers );
            if any( matches )
                matchIndex = find( matches, 1, 'first' );
                entitlement = allEntitlements(matchIndex);
            else
                % If there are valid license numbers but the user didn't
                % pick one, we should error and let them know which license
                % numbers they can choose.
                if length( validLicenseNumbers ) > 1
                    validLicenseNumbers = [ sprintf( '  %s\n', validLicenseNumbers{1:end-1} ), sprintf( '  %s', validLicenseNumbers{end} ) ];
                else
                    validLicenseNumbers = sprintf( '  %s', validLicenseNumbers{1} );
                end
                error( message( 'parallel:cluster:InvalidLicenseNumber', licenseNumber, validLicenseNumbers ) );
            end
        end
        
        function ensureLicenseNumberIsSetOrError( obj )
            % There is one optional input argument - the error id to use when
            % there are no licenses for the given user.
            narginchk(1, 2);
            if ~obj.RequiresOnlineLicensing
                return
            end

            % Ensure that a user is logged in and registered
            schedulerType = parallel.internal.types.SchedulerType.fromName( obj.Type );
            licenseCategory = schedulerType.LicenseCategory;
            currentUser = parallel.internal.mwa.ensureLoggedInAndRegistered( licenseCategory );
            
            % If the license number has been set, use that one
            if ~isempty( obj.LicenseNumber )
                return
            end

            % Get the license number to use.
            % NB We'll always get a non-empty entitlement back
            obj.LicenseEntitlement = currentUser.getEntitlement( licenseCategory );
            obj.hSetPropertyNoCheck( 'LicenseNumber', obj.LicenseEntitlement.LicenseNumber );
        end
        
        function [names, valuesAndTypes] = reorderForSet( ~, names, valuesAndTypes )
        % [needsReorder, reorderIdx] = getNewPropOrderForSet( obj, names )
        % reorderPropsForSetIfRequired will determine if the supplied property names 
        % need re-ordering for the purposes of the set (e.g. because there are props that
        % are dependent on one another).  There must be only 1 instance of each
        % property in the names (PropSet.amalgamate must already have been called).
        %
        % Returns an array of indices such that
        %     reorderedNames = names(reorderIdx)
        %
        % Intended to be called from applyConstructorArgs.  
        % Subclasses can override this if required.
        
            if numel( unique( names ) ) ~= numel( names )
                error( message( 'parallel:internal:cluster:ReorderPropertiesForSetNotUnique' ) );
            end
            newOrderIdx  = 1:numel( names );
            % Must set LicenseNumber after RequiresOnlineLicensing because
            % the LicenseNumber can only be validated if RequiresOnlineLicensing
            % is true.
            propsInOrder = {'RequiresOnlineLicensing', 'LicenseNumber'};

            % If the names don't contain all the props that require re-ordering,
            % then just return early.
            [~, ~, namesIdx] = intersect( propsInOrder, names , 'stable' );
            if numel( namesIdx ) < numel( propsInOrder )
                return
            end
            
            % Pull out the propsInOrder and move them to the end of the list
            % in the correct order.
            newOrderIdx(namesIdx) = [];
            newOrderIdx           = [newOrderIdx(:); namesIdx(:)];
            names = names(newOrderIdx);
            valuesAndTypes = valuesAndTypes(newOrderIdx);
        end
        
        function OK = awaitWorkerCompletion(obj, job, timeoutSeconds) %#ok<INUSD> defining API
            % Waits for the workers to be totally free before destroying the job.
            % Default implementation does nothing.
            OK = true;
        end
        
        function listenForMatlabLogout( obj )
            % Register a listener for MATLAB logout events if one has not
            % already been registered for this cluster. This method is used
            % for lazily initializing the listener and must be called by
            % subclasses whenever RequiresOnlineLicensing is set
            % to true, including in the constructor.
            if isempty( obj.MatlabLogoutListener )
                confirmLogout = true;
                obj.MatlabLogoutListener = parallel.internal.mwa.addLogoutListener( ...
                    'InvalidatingClusters', @obj.onInvalidatingClusters, confirmLogout );
            end
        end
        
        function onInvalidatingClusters( obj )
            % A method that will be executed when a MATLAB logout occurs if
            % listenForMatlabLogout() has been called on this cluster.
            % Default implementation is to call the cluster's private
            % delete method if it is using online licensing.
            %
            % Subclasses should override this method if they need to
            % perform additional work when a MATLAB logout occurs, such as
            % invalidating the cluster's children.
            if obj.RequiresOnlineLicensing
                obj.delete();
            end
        end

        function preCreateJob( obj, variant, varargin ) %#ok<INUSD> defining API
            % Provides a hook for sub-classes to perform cluster-specific
            % checks before creating a job. Default does nothing.
        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.Cluster'));
        end
        
        % This method gathers a list of properties needed for a cluster
        % memento. It looks for Constant private properties called
        % MinimalMementoProperties and FullMementoProperties on the input
        % metaclass that respectively define the properties needed for
        % minimal and full mementos for that class. The function recurses
        % through each of the super-classes of the input metaclass and
        % accumulates all the properties required. The properties listed in
        % both MinimalMementoProperties and FullMementoProperties will be
        % accumulated for a full memento.
        function props = hDetermineMementoProperties(clazz, clusterMementoType)
            
            import parallel.internal.types.ClusterMementoType
            
            props = string.empty;
            if isempty(clazz)
                return
            end
            
            % Gather the minimum properties needed for the memento so that
            % a cluster can be constructed and its jobs can be found.
            metaProperty = findobj(clazz.PropertyList, 'Name', 'MinimalMementoProperties');
            if ~isempty(metaProperty) && strcmp(metaProperty.GetAccess, 'private') && metaProperty.Constant
                props = [props, metaProperty.DefaultValue];
            end
            
            if clusterMementoType == ClusterMementoType.Full
                % Gather the properties needed for the memento so that a
                % new cluster can be constructed that displays the same as
                % the original.
                metaProperty = findobj(clazz.PropertyList, 'Name', 'FullMementoProperties');
                if ~isempty(metaProperty) && strcmp(metaProperty.GetAccess, 'private') && metaProperty.Constant
                    props = [props, metaProperty.DefaultValue];
                end
            end
            
            % If clazz is parallel.Cluster, we don't need to recurse
            % through any more superclasses.
            if strcmp(clazz.Name, 'parallel.Cluster')
                return
            end
            
            % Recurse to get any properties defined in the super-classes.
            superClasses = clazz.SuperclassList;
            for ii = 1:numel( superClasses )
                props = [props, parallel.Cluster.hDetermineMementoProperties(superClasses(ii), clusterMementoType)]; %#ok<AGROW>
            end
        end
    end
    
    methods ( Hidden, Sealed )
        function v = hGetPropertyNoCheck( obj, name )
        % Provides direct access to properties, this cannot be
        % overridden by subclasses, override hGetProperty instead.
            obj.updateDerivedProfileProperties();
            if ~iscell( name )
                v = obj.PropStorage.getValue( name );
            else
                v = cellfun( @(x) obj.PropStorage.getValue( x ), ...
                             name, 'UniformOutput', false );
            end
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Undocumented/Hidden public parallel.Cluster API
    methods ( Hidden )
        function hJobPreSubmitCheck( obj, job )
            % Note that this method is overridden in parallel.cluster.CJSCluster.
            parallel.internal.cluster.SubmissionChecks.warnIfPoolOpenOnThisCluster( ...
                obj, job.MinNumWorkers );
        end
        function jobs = hGetJobs( obj )
            jobs = obj.getJobs();
            if ~iscell( jobs )
                jobs = { jobs };
            end
        end
        function v = hGetProperty( obj, name )
        % Return one or more properties of the CJSCluster.
            v = obj.hGetPropertyNoCheck( name );
        end
        function hSetPropertyNoCheck( obj, names, values )
        % Simply apply the property at the "User" level.
            if ~iscell( names )
                names  = { names };
                values = { values };
            end
            cellfun( @(n, v) obj.PropStorage.setUserValue( n, v ), ...
                     names, values );
        end
        function tf = hIsVariantSupported(~, variant)
        % Retain because some clusters may not support concurrent jobs. (Required for
        % validation)
            validateattributes(variant, {'parallel.internal.types.Variant'}, {'scalar'});
            % Default to true, except for Concurrent.
            tf = ( variant ~= parallel.internal.types.Variant.ConcurrentJob );
        end
        function tf = hIsExecutionModeSupported(~, ~)
            % By default all clusters support all execution modes.
            % Clusters can override this if needed.
            tf = true;
        end
        function tf = hEquivalentWorkers( obj1, obj2 )
        %hEquivalentWorkers TRUE if clusters are equivalent
        % in the sense that they submit to the same pool of workers.
        % tf = hEquivalentWorkers(obj1, obj2) requires two scalar Cluster instances.
        % Uses isEquivalentWorkers if obj1 and obj2 are of the same class.
            tf = isequal( class( obj1 ), class( obj2 ) ) && ...
                 isEquivalentWorkers( obj1, obj2 );
        end
        function tf = hEquivalentStorage( obj1, obj2 )
        %hEquivalentStorage TRUE if clusters store jobs/tasks in the same place
        % tf = hEquivalentStorage(obj1, obj2) requires two scalar Cluster instances.
        % Uses isEquivalentStorage if obj1 and obj2 are of the same class.
            tf = isequal( class( obj1 ), class( obj2 ) ) && ...
                 isEquivalentStorage( obj1, obj2 );
        end
        
        function hSetLicenseEntitlementNoCheck( obj, entitlement )
        %hSetLicenseEntitlement set the license number based on the entitlement.
        % Called by workers in distcomp_evaluate_*task to set the license entitlement
        % and the license number.  Does NOT do any checking that the supplied entitlement 
        % is valid (i.e. no web service calls).
            obj.LicenseEntitlement = entitlement;
            obj.hSetPropertyNoCheck( 'LicenseNumber', entitlement.LicenseNumber );
        end
        
        function val = hGetLicenseEntitlement( obj )
            val = obj.LicenseEntitlement;
        end
        
        % Add Cluster Properties to Display
        function [clusterPropertyMap, propNames] = hGetDisplayItems(obj, diFactory)
            % Create an empty map to hold displayable cluster properties
            clusterPropertyMap = containers.Map('KeyType', 'char', 'ValueType', 'any');
           
            % The cluster base class is never displayed, so it does not
            % have a specified property display order to put into propNames
            propNames = {};
            
            % Add default displayable items to map.
            names = {...
                'Profile', ...
                'Modified', ...
                'Host', ...
                'NumWorkers', ...
                'NumThreads', ...
                'RequiresOnlineLicensing', ...
                'LicenseNumber', ...
                'ClusterMatlabRoot', ...
                'OperatingSystem'};
            values = diFactory.makeMultipleItems(@createDefaultItem, obj.hGetDisplayPropertiesNoError(names));
            clusterPropertyMap = [clusterPropertyMap; containers.Map(names, values)];
            
            % Add non-default display items to map 
            clusterPropertyMap('JobStorageLocation') = diFactory.createJobStorageStructItem(obj.hGetDisplayPropertiesNoError('JobStorageLocation')); 
            % '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.  
            clusterPropertyMap('Separator') = diFactory.createSeparator();  
        end
        
        % Get the data necessary to create a new cluster that displays the
        % same as this instance. The output must be a struct of fieldnames
        % and values which correspond to PV pairs that will go into the
        % cluster constructor.
        function data = hGetMementoData(obj, clusterMementoType)
            import parallel.internal.types.ClusterMementoType
            props = cellstr(obj.getAllMementoProperties(clusterMementoType));
            
            if clusterMementoType == ClusterMementoType.Full && ~isempty(obj.ProfileObject)
                % If Profile isn't empty, it must be the first field in the
                % struct so that it is the first argument to the
                % constructor when a new cluster is created from the
                % memento. This ensures that any properties with values
                % different to those defined in the profile are set
                % correctly on the new cluster object. Setting Profile as
                % the first element in props will achieve this.
                props = ['Profile' props];
            end
            
            values = obj.get(props);
            data = cell2struct(values, props, 2);
        end
        
        function tf = hIsClusterAvailable(~)
            % Specifies whether the cluster is available.
            tf = true;
        end
        
        function hDeleteOneJob(obj, job, timeoutInSeconds)
            % Attempts to delete a single job, canceling it first and
            % waiting for its completion, depending on the cluster.
            
            if ~obj.hIsClusterAvailable()
                % If the cluster is no longer available, bail out
                return
            end
            
            try
                cancel(job);
            catch E
                dctSchedulerMessage(4, sprintf('Failed to cancel job %d:', job.ID), E);
            end
            
            % wait for the job to terminate
            if timeoutInSeconds > 0
                t = tic;
                OK = obj.awaitWorkerCompletion(job, timeoutInSeconds);
                elapsed = toc(t);
                dctSchedulerMessage(5, ['hAwaitWorkerCompletion on job %d ', ...
                    'completed with result %d in %.03f seconds.'], ...
                    job.ID, OK, elapsed);
            end
            
            try 
                delete(job);
            catch E
                if isequal('parallel:cluster:storage:WorkUnitNotFound', E.identifier)
                    % Workunit has already been deleted. This may happen if
                    % using multiple clients. See g1767991.
                    dctSchedulerMessage(1, sprintf('Failed to delete job %d:', job.ID), E);
                else
                    rethrow(E);
                end
            end
        end
        
        % Throws a warning if the cluster type cannot disable SPMD for a
        % parallel pool.  The default implementation is a no-op.  Override
        % as appropriate in sub-classes.
        function hWarnIfCannotDisableSPMD(~)
        end
        
        function hSetIsSharedCluster( obj, val )
        % hSetIsSharedCluster - sets the IsSharedCluster property to 
        % the specified value.  Used in tests.
            obj.IsSharedCluster = val;
        end
        
        function numWorkersRange = hGetDefaultPoolNumWorkersRange( obj, preferredNumWorkers )
            % Can be overridden in leaf classes
            % If the cluster is for personal use only, we'll always try to create
            % the pool using the full cluster size and ignore the PreferredNumWorkers
            % value.
            if nargin < 2
                maxWorkers = obj.NumWorkers;
            else
                if obj.IsSharedCluster
                    maxWorkers = min(obj.NumWorkers, preferredNumWorkers);
                else
                    maxWorkers = obj.NumWorkers;
                end
            end
            numWorkersRange = [1, maxWorkers];
        end
        
        function hErrorIfNumPoolWorkersInvalid( obj, numWorkers )
            % Can be overridden in leaf classes
            % Accessing element 1 in case numWorkers has 2 elements
            minWorkers = numWorkers(1);
            clusterNumWorkers = obj.NumWorkers;
            if minWorkers > clusterNumWorkers
                if isempty( obj.Profile )
                    error(message('parallel:cluster:NumWorkersExceeded', minWorkers, clusterNumWorkers));
                else
                    error(message('parallel:cluster:ProfileNumWorkersExceeded', minWorkers, obj.Profile, clusterNumWorkers));
                end
            end
        end
        
        function name = hGetSchedulerComponentName( obj )
            % Specifies the name of the scheduler component associated with
            % the cluster. By default the name of the scheduler component
            % class matches the cluster class. In cases where this is not
            % true this method can be overridden in leaf classes.
            name = obj.Type;
        end
        
        function filteredVariables = hFilterEnvironmentVariables( ~, variables )
            % Overwrite this method in child cluster classes to restrict
            % which client environment variables may be copied to workers.
            filteredVariables = variables;
        end
        
        function autoVariables = hGetAutoEnvironmentVariables( ~ )
            % Overwrite this method in child cluster classes to list known
            % environment variables which we always want to copy to workers
            % before executing user code
            autoVariables = {};
        end
        
        function maxNumWorkers = hGetMaxNumWorkers( obj )
            % Override this method for auto-resizable clusters to report
            % the maximum number of workers this cluster can obtain
            maxNumWorkers = obj.NumWorkers;
        end
    end
    
    methods ( Hidden, Sealed )
        function link = hGetLink( obj, displayString )
            % A FullClusterMemento is required so that clicking the link
            % displays a cluster the same as the original cluster object.
            serializedMemento = serialize( parallel.internal.display.FullClusterMemento( obj ) );
            matlabFunction = sprintf( '%s(''%s'')', ...
                parallel.internal.display.Displayer.DisplayObjectFunction, serializedMemento );
            link = parallel.internal.display.HTMLDisplayType( displayString, matlabFunction );
        end
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Documented, public parallel.Cluster API. Sealed to prevent unfortunate messages.
    methods ( Sealed )
        job = batch( obj, scriptName, varargin )
        pool = parpool(obj,varargin)
        varargout = findJob( obj, varargin )
        job = createJob( obj, varargin )
        job = createCommunicatingJob( obj, varargin )
        saveProfile( obj, varargin )
    end
    methods ( Hidden, Sealed )        
        
        job = createIndependentJob( obj, varargin )
        
        function job = createConcurrentJob( obj, varargin )
            import parallel.internal.types.Variant
            import parallel.internal.customattr.PropSet

            validateattributes( obj, {'parallel.Cluster'}, {'scalar'}, ...
                                'createConcurrentJob', 'cluster', 1 );
            obj.preCreateJob( Variant.ConcurrentJob, varargin{:} );
            job = obj.buildConcurrentJob();
            obj.applyJobProperties( job, varargin{:} );
        end
        
        job = hBatchVec( obj, taskFcn, numArgsOut, argsIn, varargin )
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Documented, public parallel.Cluster API. Not sealed since vectorization is not 
    % supported
    methods
        saveAsProfile( obj, profileName, varargin )
    end

    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Simple public method implementations
    methods ( Sealed )
        function tf = isequal( varargin )
        %ISEQUAL True if clusters have same property values
        %    ISEQUAL(C1,C2) returns logical 1 (TRUE) if clusters C1 and C2
        %    have the same property values, and logical 0 (FALSE) otherwise.
        %
        %    ISEQUAL(C1,C2,C3,...) returns TRUE if all clusters are equal.
        %
        %    ISEQUAL can operate on arrays of clusters. In that case, each
        %    element of the arrays are compared.
        %
        %    When comparing clusters, ISEQUAL does not compare the contents of
        %    the 'Jobs' property.
        %
        %    See also ISEQUAL, PARCLUSTER.

            narginchk(1, Inf);
            szsCell  = cellfun( @size, varargin, 'UniformOutput', false );
            szsMatch = all( cellfun( @(x) isequal(szsCell{1}, x), szsCell(2:end) ) );
            if ~szsMatch
                tf = false;
                return
            end
            allClusters = all( cellfun( @(x) isa(x, 'parallel.Cluster'), varargin ) );
            if ~allClusters
                tf = false;
                return
            end

            % Actually need to compare
            tf = true;
            obj1 = varargin{1};
            argToCompare = 2;
            try
                % Outer loop over argument list
                while tf && argToCompare <= nargin
                    obj2 = varargin{argToCompare};
                    elementToCompare = 1;
                    % Inner loop over elements of argument
                    while tf && elementToCompare <= numel( obj1 )
                        tf = tf && ...
                             scalarIsequal( obj1(elementToCompare), obj2(elementToCompare) );
                        elementToCompare = elementToCompare + 1;
                    end
                    argToCompare = argToCompare + 1;
                end
            catch E
                throw( E );
            end
        end
    end
    
    methods ( Access = private )
        function delete( obj )
            delete@handle( obj );
        end
    end
    
    % --------------------------------------------------------------------------
    % Public/Hidden API-1 compatibility
    methods ( Hidden, Sealed )
        function [cm, strategy] = pCreateConnectionManager( obj )
            [cm, strategy] = obj.buildConnectionManager();
        end
    end
end
%#ok<*ATUNK> custom attributes
