% CJSCluster - base class for Clusters other than MJS.
% Manages:
% - Support, SupportID - the CJSSupport object and ID
% - HasSharedFilesystem - simple user-visible flag
% - JobStorageLocation

% Copyright 2011-2023 The MathWorks, Inc.

classdef ( Hidden ) CJSCluster < parallel.Cluster

    % --------------------------------------------------------------------------
    % Property definitions
    properties ( Transient, GetAccess = protected, SetAccess = private )
        Support     % Our CJSSupport
        SupportID   % Our ID for the CJSSupport
    end

    properties ( PCTGetSet, Transient, PCTConstraint = 'logicalscalar' )
        %HasSharedFilesystem Specify whether the client and cluster nodes share JobStorageLocation
        %   HasSharedFilesystem indicates whether the location identified by
        %   the JobStorageLocation property can be accessed from the client
        %   machine and all nodes in the cluster.
        %
        %   See also parallel.cluster.CJSCluster.JobStorageLocation
        HasSharedFilesystem = true;

        %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.CJSCluster.LicenseNumber
        RequiresOnlineLicensing
    end

    % Everything to do with JobStorageLocation - we stash the actual struct in props.
    properties ( PCTGetSet, Transient, PCTConstraint = 'datalocation' )
        %JobStorageLocation Folder where cluster stores job and task information
        %   JobStorageLocation is a string or structure specifying a folder
        %   where the cluster stores job information. In a shared file
        %   system, the client and all cluster nodes, must have access to
        %   this folder. In a nonshared file system, only the MATLAB client
        %   accesses job information in this folder.
        %
        %   If JobStorageLocation is not set, the default folder for job
        %   information is the current working directory of the MATLAB
        %   client the first time you create the cluster, either using
        %   parcluster or the cluster construction.
        %
        %   Use a structure to specify the JobStorageLocation in an
        %   environment of mixed platforms. The fields for the structure
        %   are named windows and unix. Each node then uses the field
        %   appropriate for its platform.
        %
        %   The job information stored in the folder identified by
        %   JobStorageLocation might not be compatible between different
        %   versions of MATLAB Parallel Server and different
        %   cluster types. Therefore, JobStorageLocation should not be
        %   shared by different cluster types nor by parallel computing
        %   products running different versions; each version on your
        %   cluster should have its own JobStorageLocation.
        %
        %   See also parallel.cluster.OperatingSystem,
        %            parallel.cluster.Type.
        JobStorageLocation = '';
    end

    properties ( PCTGetSet, Transient, PCTConstraint = 'string' )
        %ClusterMatlabRoot Path to MATLAB on the cluster
        %   ClusterMatlabRoot specifies the path to MATLAB for the cluster
        %   to use for starting MATLAB worker processes. The path must be
        %   accessible to all cluster nodes on which worker sessions will
        %   run.
        %
        %   ClusterMatlabRoot is a string; it must be structured
        %   appropriately for the file system of the cluster nodes. If the
        %   value is empty, the MATLAB executable must be on the path of
        %   the cluster nodes.
        ClusterMatlabRoot = matlabroot;

        %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 is ignored if RequiresOnlineLicensing is false.
        %
        %   See also parallel.cluster.CJSCluster.RequiresOnlineLicensing
        LicenseNumber
    end
    properties ( PCTGetSet, Transient, ...
                 PCTConstraint = 'enum:parallel.internal.types.OperatingSystem' )
        %OperatingSystem Operating System of 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
    end
    properties ( PCTGetSet, Transient, PCTConstraint = 'positiveintscalar' )
        %NumWorkers Number of workers in the cluster
        %   NumWorkers specifies the maximum number of workers that this
        %   cluster can start for running a job. NumWorkers must fall
        %   within the range defined by a communicating job's
        %   NumWorkersRange.
        NumWorkers = Inf;
    end
    properties ( Constant, GetAccess = private )
        %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 this class and every sub-class.
        %
        % 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 = "JobStorageLocation";
        
        %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 this class and every sub-class.
        %
        % 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 = [ ...
            "NumWorkers", ...
            "RequiresOnlineLicensing" ];
    end
    % --------------------------------------------------------------------------
    % Public/Hidden/Abstract methods: This class defines all the properties and
    % methods that CJSSupport and CJS*Job can rely on. These are the methods
    % that subclasses must override
    methods ( Hidden, Abstract )

        % hDestroyJob - contact the underlying scheduler to remove the job
        hDestroyJob( cluster, job )

        % hDestroyTask - contact the underlying scheduler to remove the task
        hDestroyTask( cluster, task )

        % hCancelJob - contact the underlying scheduler to cancel execution of a
        % job. Return true if execution was cancelled.
        didCancel = hCancelJob( cluster, job )

        % hCancelTask - contact the underlying scheduler to cancel execution of
        % a task. Return true if execution was cancelled.
        didCancel = hCancelTask( cluster, task )

        % hSubmit*Job - contact the underlying scheduler to schedule execution
        % of the job.
        hSubmitIndependentJob( cluster, job, jobSupport, jobSupportID )
        hSubmitCommunicatingJob( cluster, job, jobSupport, jobSupportID )
    end
    methods ( Hidden )
        % Return the state of a task using information from the scheduler
        % if possible.  Override in leaf classes if there is a more
        % efficient implementation.
        function stateEnum = hGetTaskStateEnum( obj, task )
            % Get the job's state first via hGetJobState (as opposed to
            % getJobState) as this will update the task's state files on
            % disk if the job is in a terminal state.
            job = task.Parent;
            obj.hGetJobState( job, job.hGetProperty( 'StateEnum' ) );
            stateEnum = task.hGetProperty( 'StateEnum' );
        end
        
        function hJobPreSubmitCheck( ~, ~ )
            % The definition of these method in the base class simply calls
            % parallel.internal.cluster.SubmissionChecks.warnIfPoolOpenOnThisCluster.
            % However, we do not want to issue the warning for
            % third-party clusters because the NumWorkers property of the
            % cluster object may not match the actual number of
            % licenses/execution slots available on the cluster. So we
            % override the method to be a no-op here and then override the
            % method again in parallel.cluster.Local to ensure the warning
            % is issued for Local clusters.
        end

        function state = hGetJobState(cluster, job, jobStateInStorage)
        % hGetJobState uses the states
        % - in the job state file,
        % - derived from the tasks and
        % - obtained from the scheduler
        % to deduce a sensible new job state.

            import parallel.internal.types.States

            if ~ismember(jobStateInStorage, [States.Queued, States.Running, States.Unavailable])
                % job is not queued/running/unavailable, so don't bother asking the cluster
                % object.
                state = jobStateInStorage;
                return
            end

            % Deduce a job state by looking at the states of its tasks
            stateFromTasks = job.hGetStateFromTasks();
            
            if jobStateInStorage == States.Unavailable && isvalid(job)
                % We've got a valid job that we do not know the state of. If we got a useful
                % state from looking at the tasks, let's write it to disk.
                try
                    if stateFromTasks.isTerminal()
                        job.hSetTerminalStateFromCluster(stateFromTasks);
                    else
                        job.hSetPropertyNoCheck('StateEnum', stateFromTasks);
                    end
                    jobStateInStorage = stateFromTasks;
                catch E
                    dctSchedulerMessage(2, 'Failed to write new state %s for job %d because: %s.', ...
                                        stateFromTasks.Name, job.ID, E.message);
                end
            end

            isOurJob = false;
            try
                isOurJob = cluster.wasSubmittedByThisType(job);
            catch err
                % This might error if storage is in a bad state, in which
                % case we have to assume that it isn't our job.
                dctSchedulerMessage(4, 'Failed to determine if job %d belongs to this cluster because: %s.', ...
                                    job.ID, err.message);
            end
            
            if ~isOurJob
                % Don't attempt to deduce the state of jobs submitted by other
                % cluster types
                state = jobStateInStorage;
                return
            end

            % If we get here, we've got a Queued/Running/Unavailable job
            % that we know how to deal with, so defer to our protected
            % method to deal with talking to the mechanisms behind the
            % cluster.
            clusterState = States.Unknown;
            try
                clusterState = cluster.getJobState(job, jobStateInStorage);
            catch err
                % This might error if storage is in a bad state, in which
                % case we have to fall back to the Unknown state
                dctSchedulerMessage(4, 'Failed to get the state for job %d from the cluster because: %s.', ...
                                    job.ID, err.message );
            end
            
            % Nobody knows anything about the job at this point, so bail.
            if clusterState == States.Unknown && jobStateInStorage == States.Unknown
                state = States.Unknown;
                return
            end
            
            % The state from the cluster isn't always trustworthy. Compare with
            % the states on disk to find a sensible state to use.
            clusterState = iSanitiseClusterJobState(job, clusterState, jobStateInStorage, stateFromTasks);

            % If the cluster says that the job has failed and the task
            % states on disk say the job is terminal, we'll trust the task
            % state as MATLAB can sometimes exit with a non-zero exit
            % status and still be OK. Otherwise use the state that is
            % furthest ahead.
            shouldIgnoreClusterFailed = numel(job.Tasks) > 0 && ...
                clusterState == States.Failed && ...
                stateFromTasks.isTerminal();

            if shouldIgnoreClusterFailed || stateFromTasks > clusterState
                state = stateFromTasks;
            else
                state = clusterState;
            end

            if state.isTerminal()
                % If the state is terminal, we should ensure that the job and tasks
                % have the right state.
                job.hSetTerminalStateFromCluster(state);
            elseif stateFromTasks > jobStateInStorage
                % If the state isn't terminal, but the tasks indicate that
                % the job state in storage is out of date, update the job
                % state in storage.
                job.hSetPropertyNoCheck('StateEnum', stateFromTasks);
            end
        end
    end


    % --------------------------------------------------------------------------
    % Protected/Abstract methods that must be provided by CJSCluster
    % implementations.
    methods ( Abstract, Access = protected )
        % Use the underlying cluster mechanisms to see if there's a better value
        % for the jobState given the state in the storage
        jobState = getJobState( cluster, job, jobStateInStorage )

        % This check simply tests whether a job was submitted by this Cluster
        % type.
        tf       = wasSubmittedByThisType( cluster, job )
    end
    
    % --------------------------------------------------------------------------
    % Debug Log methods - can be overridden by subclasses if necessary.
    methods ( Access = protected )
        function logText = getDebugLogForCommunicatingJob( obj, cJob )
        % Return CommunicatingJob (or ConcurrentJob) debug log
            jobSID       = cJob.hGetSupportID();
            taskSIDs     = arrayfun( @hGetSupportID, cJob.Tasks, ...
                                     'UniformOutput', false );
            taskIDs      = arrayfun( @(x) x.Id, cJob.Tasks, ...
                                     'UniformOutput', false );
            [logText{1}, logExists] = obj.Support.readCommJobLogFile( ...
                jobSID, cJob.Variant );
            logText{2}   = obj.Support.readTaskLogsIfPresent( ...
                jobSID, taskSIDs, taskIDs );
            logText{3}   = obj.getSpecificDebugLogForCommunicatingJob( cJob, logExists );
            logText{4}   = obj.Support.readCrashDumpsIfPresent( ...
                jobSID );
            logText      = sprintf( '%s', logText{:} );
        end

        function logText = getSpecificDebugLogForCommunicatingJob( varargin )
            logText = '';
        end

        function logText = getDebugLogForIndependentTask( obj, iJob, task )
        % Return IndependentJob.Task debug log
            taskSID      = task.hGetSupportID();
            jobSID       = iJob.hGetSupportID();
            [logText{1}, logExists] = obj.Support.readIndJobTaskLogFile( ...
                jobSID, iJob.Variant, taskSID );
            logText{2}   = obj.Support.readTaskLogsIfPresent( ...
                jobSID, {taskSID}, {task.Id} );
            logText{3}   = obj.getSpecificDebugLogForIndependentTask( iJob, task, logExists );
            logText      = sprintf( '%s', logText{:} );
        end

        function logText = getSpecificDebugLogForIndependentTask( varargin )
            logText = '';
        end

    end
    methods ( Access = protected )
        function num = getNumConcurrentTasks( obj, job )
            if mpiInitialized
                num = spmdSize;
            else
                num = numel( obj.Support.getTasks( job, job.hGetSupportID() ) );
            end
        end
    end

    % --------------------------------------------------------------------------
    % Public/Hidden methods implemented once and for all for CJSClusters.
    methods ( Sealed, Hidden )

        function support = hGetSupport( obj )
        % Return the underlying CJSSupport.
            support = obj.Support;
        end
        function sid = hGetSupportID( obj )
        % Return the identifier to be used with the support.
            if numel( obj ) > 1
                sid = {obj.SupportID}';
            else
                % Maintain backwards compatibility with non-vectorized code.
                sid = obj.SupportID;
            end
        end
        
        function [taskIDs, taskSIds] = hBuildTasksVec( obj, varargin )
            [taskIDs, taskSIds] = obj.Support.buildTasksVec( varargin{:} );
        end

        function newjob = hBuildJob( obj, variant, jobid, jobsid )
        % hBuildJob is called by the Support to actually construct the MCOS
        % Job representation from information on disk.
            import parallel.internal.types.Variant;
            switch variant
              case Variant.IndependentJob
                newjob = parallel.job.CJSIndependentJob( ...
                    obj, jobid, jobsid, obj.Support );
              case { Variant.CommunicatingSPMDJob, ...
                     Variant.CommunicatingPoolJob }
                newjob = parallel.job.CJSCommunicatingJob( ...
                    obj, jobid, variant, jobsid, obj.Support );
              otherwise
                assert( false );
            end
        end

        function [job, task] = hGetJobAndTaskFromLocations( obj, jobLoc, taskLoc )
        % Used by distcomp_evaluate_filetask to bring up the job and task
        % Note that 'location' is basically the supportId.
            job = obj.Support.getJobFromLocation( obj, jobLoc );
            task = obj.Support.getTaskFromLocation( job, taskLoc );
        end

        function filename = hGetTaskJavaLogIfPossible( obj, task )
        % hGetTaskJavaLogIfPossible - return a java logfile location, or ''.
        % Called from distcomp_evaluate_filetask to initiate java logging.
        % Returns job3/Task7.java.log - call site might choose to append ".0", ".1" etc.
            taskSID  = task.hGetSupportID();
            filename = obj.Support.getTaskJavaLogBaseIfPossible( taskSID );
        end
        
        function filename = hGetTaskCppLogIfPossible( obj, task )
        % hGetTaskCppLogIfPossible - return a cpp logfile location, or ''.
        % Called from distcomp_evaluate_filetask to initiate C++ logging.
        % Returns job3/Task7.cpp.log - call site might choose to append ".0", ".1" etc.
            taskSID  = task.hGetSupportID();
            filename = obj.Support.getTaskCppLogBaseIfPossible( taskSID );
        end

        function provider = hCreateJobStatusProvider( obj )
        % hCreateJobStatusProvider - return an update provider for Job Monitor
        % to pull updates.
            provider = obj.Support.hCreateJobStatusProvider( obj );
        end
    end
    methods ( Hidden )
        function hSetProperty( obj, names, values )
            if ~iscell( names )
                names  = { names };
                values = { values };
            end
            for ii = 1:length( names )
                if isequal( names{ii}, 'JobStorageLocation' )
                    try
                    	stor = parallel.internal.cluster.FileStorage.createFileStorage( values{ii} );
                    catch E
                        throwAsCaller( E );
                    end

                    obj.Support.invalidateAllChildren();
                    obj.Support   = parallel.internal.cluster.CJSSupport( stor, values{ii} );
                    obj.SupportID = stor.getRootLocation();
                end

                if isequal( names{ii}, 'RequiresOnlineLicensing' )
                    if isdeployed && values{ii}
                        % Can't use MHLM workers from a deployed application
                        error(message('parallel:cluster:CannotUseMHLMFromDeployed'));
                    end
                    
                    % Only do something if we are actually changing the value
                    if obj.isNotCurrentValue('RequiresOnlineLicensing', values{ii})
                        if values{ii}
                            % If changing to true, make sure the cluster is
                            % listening for MATLAB logout events
                            obj.listenForMatlabLogout();
                        else
                            % If changing to false, we must clear out the
                            % license number as it no longer makes sense
                            if ~isempty( obj.LicenseNumber )
                                obj.hSetPropertyNoCheck( 'LicenseNumber', '' );
                            end
                            
                            % Don't bother unregistering the listener,
                            % though, since it is very unlikely for
                            % RequiresOnlineLicensing to change
                            % from true to false and, if it does, we don't
                            % want to repeatedly re-register listeners
                        end
                    end
                end

                if isequal( names{ii}, 'LicenseNumber' ) && ...
                    obj.isNotCurrentValue('LicenseNumber', values{ii})
                    % If the license number is different to the currently
                    % stored number then check its validity before setting
                    % and make sure we stash the corresponding entitlement.
                    try
                        obj.LicenseEntitlement = obj.checkLicenseNumber( values{ii} );
                    catch E
                        throwAsCaller( E )
                    end
                end
                
                obj.hSetPropertyNoCheck( names{ii}, values{ii} );
            end
        end
        
        % Issues a warning if the cluster type cannot disable SPMD for a
        % parallel pool.
        function out = hResolveSpmdEnabledEnum(~, in)
            import parallel.internal.cluster.types.SpmdEnabled
            
            % Local clusters can disable SPMD, but third
            % party clusters cannot. So we issue a warning in this
            % implementation to cover the third party case and then
            % override the method in parallel.cluster.Local to return the input.
            if in == SpmdEnabled.False
                parallel.internal.warningNoBackTrace(message('parallel:lang:pool:CannotDisableSPMD'));
            end
            out = SpmdEnabled.True;
        end

        function factory = hCreatePoolEndpointFactory( ~ )
            % For CJS clusters, default is to use optimized broadcast.
            optimizedBroadcastDefault = true;

            % For CJS clusters, default is *not* to use secure communication. 
            % Users may override this using pctconfig('alwayssecurepools',
            % true).
            useSecureMatlabPool = false;

            % For CJS clusters, default is to use server sockets in the cluster.
            allServerSocketsInCluster = true;

            factory = parallel.internal.pool.PoolEndpointFactory.create( ...
                UseSecureMatlabPool=useSecureMatlabPool, ...
                UseOptimizedBroadcast=optimizedBroadcastDefault, ...
                AllServerSocketsInCluster=allServerSocketsInCluster);
        end
    end

    % --------------------------------------------------------------------------
    % Implementation of some abstract/protected methods
    methods ( Sealed, Access = protected )
        function jl = getJobs( obj )
            if numel( obj ) > 1
                jl = arrayfun( @(x) x.Support.getJobs( ...
                        x, x.hGetSupportID() ), obj, ...
                        'UniformOutput', false );
            else
                jl = obj.Support.getJobs( obj, obj.hGetSupportID() );
            end
        end
        function job = buildIndependentJob( obj )
            variant = parallel.internal.types.Variant.IndependentJob;
            job     = obj.Support.buildJob( obj, obj.SupportID, variant );
        end
        function job = buildCommunicatingJob( obj, variant, executionMode )
            job = obj.Support.buildJob( obj, obj.SupportID, variant );
            % Communicating jobs need to have their NumWorkersRange set
            job.NumWorkersRange = [1 obj.NumWorkers];
            isInteractive = (executionMode == parallel.internal.types.ExecutionMode.Interactive);
            job.hSetInteractive(isInteractive);
        end
        function varargout = getJobsByState( obj )
            import parallel.internal.types.States;
            statesStr = { States.Pending.Name, ...
                          States.Queued.Name, ...
                          States.Running.Name, ...
                          States.Finished.Name, ...
                          States.Failed.Name };
            if numel( obj ) > 1
                pqrff = arrayfun( @(x) x.Support.getJobsByState( ...
                            x, x.hGetSupportID(), statesStr ), obj, ...
                            'UniformOutput', false );
                pqrff = vertcat( pqrff{:} );
            else
                pqrff = obj.Support.getJobsByState( obj, obj.hGetSupportID(), statesStr );
            end
            % We think that the 2nd branch of the if-statement below is dead code.
            % We want to verify that with this assertion.
            assert( nargout == 4, 'getJobsByState called with < 4 output arguments, this is unexpected' );
            % Need to concatenate Finished/Failed
            if nargout == 4
                % To be consistent with the API and to allow concatenation in calling functions
                % each cell of the varargout must be a 1x1 cell containing a row vector of jobs.
                % pqrff is an Nx5 cell where N == numel(obj).
                varargout = { {[pqrff{:,1}]}, {[pqrff{:,2}]}, {[pqrff{:,3}]}, {[pqrff{:,4:end}]} };
            else
                varargout = { [ pqrff(:,1:3), {[pqrff{:,4:end}]} ] };
            end
        end

        function tf = isEquivalentStorage( obj, cluster )
            % For CJS clusters, it is sufficient to compare the
            % JobStorageLocation
            import parallel.internal.cluster.AbstractStorage.getLocationsFromDirInfo
            tf = isa( cluster, 'parallel.cluster.CJSCluster' ) ...
                && strcmp( getLocationsFromDirInfo(obj.JobStorageLocation), getLocationsFromDirInfo(cluster.JobStorageLocation) );
        end
    end
    
    methods ( Access = protected )
        function [connMgr, strategy] = buildConnectionManager( ~ )
            % For CJS clusters, default is to use optimized broadcast.
            optimizedBroadcastDefault = true;
            [~, undoc] = pctconfig();
            [connMgr, strategy] = parallel.internal.apishared.ConnMgrBuilder.buildForCJS(optimizedBroadcastDefault, undoc.reliableconnections);
        end
    end

    % Used by concrete cluster classes instantiated with property maps
    methods ( Static, Access = protected )
        function [dataLoc, dataLocValType] = chooseDataLocation( propsFromArgs )
        % Look through the arguments we've got and see if there's a JobStorageLocation, else
        % derive the default. We will remove the JobStorageLocation from the propsFromArgs
        % map.
            if propsFromArgs.isKey( 'JobStorageLocation' )
                dataLocAndType = propsFromArgs('JobStorageLocation');
                dataLoc        = dataLocAndType{1};
                dataLocValType = dataLocAndType{2};
                propsFromArgs.remove( 'JobStorageLocation' );
            else
                dataLoc        = pwd;
                dataLocValType = parallel.internal.settings.ValueType.Factory;
            end
        end
    end
    
    methods ( Access = protected )
        function obj = CJSCluster( type, cProf, storageOrDataLoc, storageLocValType, jobStorageLocation )
            % Can either be given a storage object or just a job storage data
            % location (a directory or similar), from which a FileStorage
            % object will be created in this constructor:
            %
            % 3rd arg can be either a storage object or a job data directory
            % 4th arg is always JobStorageLocation's ValueType enum (User,
            % Profile or Factory)
            %
            % 5th arg is the JobStorageLocation
            %
            % Usage:
            %     CJSCluster( type, cProf, dataLoc, dataLocValType, jobStorageLocation )
            %     CJSCluster( type, cProf, storage, storageLocValType, jobStorageLocation )
            
            import parallel.internal.apishared.OsType
            
            obj@parallel.Cluster( type, cProf );
  
            % We must seed all our PCTGetSet properties into PropStorage
            factorySharedFilesystem = true;
            factoryMatlabroot       = matlabroot;
            factoryNumWorkers       = Inf;
            % Factory 'OperatingSystem' is local OS.
            clientOsEnum            = OsType.forClient();
            factoryOs               = clientOsEnum.Api2Name;

            factoryRequiresOnlineLicensing = false;
            factoryLicenseNumber    = '';
            
            try
                % Storage object supplied
                if isa( storageOrDataLoc, 'parallel.internal.cluster.AbstractStorage' ) 
                    stor = storageOrDataLoc;
                % For compatibility with all CJSCluster-based schedulers
                else
                    storageType = parallel.internal.types.StorageType.FileStorage;                      
                    stor = parallel.internal.cluster.StorageFactory.createStorage( storageType, storageOrDataLoc );
                end
                if isempty( jobStorageLocation )
                    jobStorageLocation = stor.StorageLocation;
                end
                obj.Support   = parallel.internal.cluster.CJSSupport( stor, jobStorageLocation );
                obj.SupportID = stor.getRootLocation();
            catch E
                throwAsCaller( E );    
            end

            obj.PropStorage.addFactoryValue( 'ClusterMatlabRoot', factoryMatlabroot );
            obj.PropStorage.addFactoryValue( 'NumWorkers', factoryNumWorkers );
            obj.PropStorage.addFactoryValue( 'JobStorageLocation', jobStorageLocation );
            obj.PropStorage.setValueType(    'JobStorageLocation', storageLocValType );
            obj.PropStorage.addFactoryValue( 'HasSharedFilesystem', factorySharedFilesystem );
            obj.PropStorage.addFactoryValue( 'OperatingSystem', factoryOs );
            obj.PropStorage.addFactoryValue( 'RequiresOnlineLicensing', factoryRequiresOnlineLicensing );
            obj.PropStorage.addFactoryValue( 'LicenseNumber', factoryLicenseNumber );
        end
        
        function onInvalidatingClusters( obj )
            % If the cluster is using online licensing at the
            % time of logout, invalidate the cluster object and all of its
            % children (jobs/tasks)
            if obj.RequiresOnlineLicensing
                obj.Support.invalidateAllChildren();
                obj.delete();
            end
        end
    end

    methods ( Access = private )
        function delete( obj )
            delete@handle( obj );
        end
    end

    % --------------------------------------------------------------------------
    % Environment Variable methods
    methods ( Sealed, Access = protected )
        function [names, values] = getClusterEnvVarsForJobSubmission( obj )
        % Get environment variables that are required for job submission.
        % This is called by the hSubmit*Job methods.
            [licNames,  licVars]       = obj.getWebLicensingEnvVars();
            [clusNames, clusValues]    = obj.getClusterSpecificEnvVars();
            [debugNames,  debugValues] = obj.getDebugEnvVars();
            % PCT_JT_CLEANUP g2364333 - Remove
            % ExperimentalVersion switching code and java paths
            [expNames,  expValues]     = obj.getFeatureSwitchEnvVars();

            names  = [ licNames(:); clusNames(:); debugNames(:); expNames(:); ];
            values = [ licVars(:); clusValues(:); debugValues(:); expValues(:); ];
        end
    end

    methods ( Abstract, Access = protected )
        % Return names/values of environment variables that this cluster must
        % send down to workers
        [names, values] = getClusterSpecificEnvVars( obj )
    end

    methods ( Access = private )
        function [names, values] = getDebugEnvVars( ~ )
             % Forward the debug value only if it is set.  If we forward it
            % regardless, then we may overwrite existing system or cluster
            % environment variables.
            names  = {};
            values = {};
            [debugSetting, flagName] = parallel.internal.logging.getWorkerDebugSetting();
            if ~isempty(debugSetting)
                names  = {flagName};
                values = {debugSetting};
            end
        end
        
        function [names, values] = getWebLicensingEnvVars( obj )
            names  = {};
            values = {};
            if ~obj.RequiresOnlineLicensing
                return;
            end

            obj.ensureLicenseNumberIsSetOrError;
            currentUser = parallel.internal.mwa.ensureLoggedIn();
            names  = { 'MLM_WEB_LICENSE', ...
                       'MLM_WEB_USER_CRED', ...
                       'MLM_WEB_ID', ...
                       'PARALLEL_SERVER_LICENSE_NUMBER'};
            values = { 'true', ...
                       currentUser.getToken(), ...
                       obj.LicenseEntitlement.Id, ...
                       obj.LicenseEntitlement.LicenseNumber };
        end
        
        function [names, values] = getFeatureSwitchEnvVars( ~ )
            names = {...
                parallel.internal.apishared.getJavaTransportEnvironment(), ... % Java transition
                };
            values = cellfun(@getenv, names, 'UniformOutput', false);
            isSet = ~cellfun(@isempty, values);
            names = names(isSet);
            values = values(isSet);
        end
    end
    
    methods ( Static, Hidden )
        function SchedulerIDs = hSimplifySchedulerIDs( ~, SchedulerIDs )
            % Simplify a list of task SchedulerIDs. How to do this depends
            % on the scheduler, so this method may be overridden in
            % subclasses if a more efficient implementation is possible,
            % e.g. all the tasks can be described via a single ID.
            % In the general case we don't know what to do, so do nothing.
        end
    end

    % --------------------------------------------------------------------------
    % User-visible methods - getDebugLog.
    methods (Sealed)
        function varargout = getDebugLog( cluster, jobOrTask )
        %getDebugLog Read output messages from communicating job or independent task
        %
        %   str = getDebugLog(myCluster, jobOrTask) returns any output
        %   written to the standard output or standard error stream by the
        %   communicating job or independent task identified by jobOrTask,
        %   being run by the cluster identified by myCluster.
        %
        %   Example:
        %   % Use batch to run a script that uses a the local cluster and
        %   % a parallel pool of size 2.
        %   myJob = batch('aScript', 'Profile', 'local', 'Pool', '2');
        %   wait(myJob);
        %
        %   % Look at the debug log.
        %   myCluster = parcluster('local');
        %   getDebugLog(myCluster, myJob);

            import parallel.internal.types.Variant
            validateattributes( cluster, {'parallel.cluster.CJSCluster'}, ...
                                {'scalar'}, 'getDebugLog', 'cluster', 1 );

            narginchk( 2, 2 );
            if numel( jobOrTask ) ~= 1
                error(message('parallel:cluster:GetDebugLogScalarJobOrTask'));
            end
            switch jobOrTask.Variant
              case { Variant.CommunicatingPoolJob, ...
                     Variant.CommunicatingSPMDJob }
                job        = jobOrTask;
                jobCluster = job.Parent;
                fcnH       = @() getDebugLogForCommunicatingJob( ...
                    cluster, job );
              case Variant.Task
                task       = jobOrTask;
                job        = task.Parent;
                jobCluster = job.Parent;
                fcnH       = @() getDebugLogForIndependentTask( ...
                    cluster, job, task );
                if job.Variant ~= Variant.IndependentJob
                    error(message('parallel:cluster:GetDebugLogJobMustBeIndependent'));
                end
              otherwise
                error(message('parallel:cluster:GetDebugLogScalarJobOrTask'));
            end
            if ~strcmp(cluster.Type, jobCluster.Type) || ~cluster.isEquivalentStorage( jobCluster )
                error(message('parallel:cluster:GetDebugLogWrongCluster'));
            end
            if ~cluster.wasSubmittedByThisType( job )
                % No debug log / not our job etc.
                varargout = cell( 1, nargout );
                return;
            end
            % Actually invoke the underlying method. Note we have captured the
            % different arg-lists into the function handle.
            try
                logData = fcnH();
                if nargout > 0
                    varargout = { logData };
                else
                    disp( logData );
                end
            catch E
                throw( E );
            end
        end
    end

    % --------------------------------------------------------------------------
    % API2(orig) compatibility for distcomp_evaluate_filetask
    properties ( Hidden, Dependent )
        SharedFilesystem
        % For compatibility continue to accept RequiresMathWorksHostedLicensing
        RequiresMathWorksHostedLicensing
    end
    methods
        function set.SharedFilesystem( obj, val )
            obj.HasSharedFilesystem = val;
        end
        function val = get.SharedFilesystem( obj )
            val = obj.HasSharedFilesystem;
        end
        function set.RequiresMathWorksHostedLicensing( obj, val )
            obj.RequiresOnlineLicensing = val;
        end
        function val = get.RequiresMathWorksHostedLicensing( obj )
            val = obj.RequiresOnlineLicensing;
        end
    end

    % --------------------------------------------------------------------------
    % API1 compatibility for RemoteClusterAccess
    methods ( Hidden )
        function result = pGetJobFilesInfo( obj, job )
            assert( job.Parent == obj );
            result = obj.Support.getJobFilesInfo( job.hGetSupportID() );
        end
    end
end
%#ok<*ATUNK> custom attributes

function clusterState = iSanitiseClusterJobState(job, clusterState, jobStateInStorage, stateFromTasks)
% The cluster states of 'Finished', 'Failed' and 'Unknown' should be treated the
% same way. The Generic plugins often return 'Finished' if the job can't be
% found on the scheduler because it isn't possible to ask the scheduler about
% completed jobs. For example, the 'squeue' command on Slurm and 'bjobs' command
% on LSF will only display queued and running jobs.
%
% It's also possible that the job has only just been submitted and doesn't yet
% appear in the output of the scheduler's queue command. This becomes more
% likely if the scheduler has been set up to have multiple submission nodes, as
% the daemons need time to coordinate.
%
% Sometimes a worker can complete a job successfully but appear failed on the
% scheduler: for example, if MATLAB experiences a problem during shutdown and
% exits with a non-zero code.

import parallel.internal.types.States

if ~ismember(clusterState, [States.Unknown, States.Finished, States.Failed])
    % Non-terminal states are OK
    return
end

if jobStateInStorage.isTerminal()
    % State on disk is terminal as expected, so use that.
    clusterState = jobStateInStorage;
    return
end

if stateFromTasks.isTerminal()
    % The state deduced from the tasks is terminal but the stored job state on
    % disk isn't. This happens if we haven't checked the job state since the
    % last worker finished.
    clusterState = stateFromTasks;
    return
end

% Check the task states on disk once more in case they've updated since we last
% checked. This commonly happens if getting the state from the cluster took a
% long time, e.g. a call to the scheduler over SSH, or if file mirroring is
% involved.
stateFromTasks = job.hGetStateFromTasks();

if stateFromTasks.isTerminal()
    % Task states now agree with the cluster that the job is terminal.
    clusterState = stateFromTasks;
    return
end

% Before a worker exits, it will update its state on disk to finished. Only once
% the worker exits will the cluster report the task as finished. If the task
% states on disk say the job is still running but the cluster thinks the job has
% finished or doesn't know about it then the job has probably crashed.
%
% We mustn't be too hasty in assuming the job has failed. We could be reading
% from a laggy filesystem and haven't yet seen the state files update. If we're
% a Generic cluster in nonshared mode, file mirroring might still be busy
% copying the new files down.
%
% To be safe, ask the JobStateHelper. It stores the time we last asked it about
% a particular job and the state it was in at that time. It allows the job a
% grace period to update its task states before we assume it has failed.
clusterState = parallel.internal.cluster.JobStateHelper.getJobState(job, jobStateInStorage);

end
