%Generic Interact with a Generic cluster type
%   The Generic cluster type allows you to specify MATLAB functions to
%   interact with your cluster. You can specify functions to define how to
%   submit jobs, how to query job states, how to cancel running jobs, and
%   how to perform any additional clean-up required when deleting a job.
%
%   When you submit an independent job to a cluster, the function named
%   'independentSubmitFcn.m' executes in the MATLAB client session. This
%   function is located in the folder defined by the property
%   PluginScriptsLocation. You set the cluster's PluginScriptsLocation
%   property to identify the submit function.
%
%   Likewise, when you submit a communicating job to a cluster, the
%   function named 'communicatingSubmitFcn.m' executes in the MATLAB client
%   session. This function is located in the folder defined by the property
%   PluginScriptsLocation.
%
%   Any user-defined variables can be passed into the plugin scripts
%   located in PluginScriptsLocation by adding them as fields to the
%   AdditionalProperties property. This is available in all plugin scripts
%   and can be used to modify their behaviour.
%
%   parallel.cluster.Generic methods:
%      Generic                - Create a generic cluster instance
%      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
%      getDebugLog            - Read output messages from communicating job or independent task
%      getJobClusterData      - Get specific userdata for a job on a generic cluster
%      getJobFolder           - Return folder on the client where jobs are stored
%      getJobFolderOnCluster  - Return folder on cluster where jobs are stored
%      getLogLocation         - Get log location for a job or task
%      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
%      setJobClusterData      - Set specific userdata for a job on a generic cluster
%
%   parallel.cluster.Generic properties:
%      AdditionalProperties    - Additional properties for plugin scripts
%      ClusterMatlabRoot       - Path to MATLAB on the cluster
%      HasSharedFilesystem     - Specify whether the client and cluster nodes share JobStorageLocation
%      Host                    - Identifies the headnode of this cluster
%      JobStorageLocation      - Folder 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 in the cluster
%      OperatingSystem         - Operating System of the cluster
%      PluginScriptsLocation   - Folder containing scheduler plugin scripts
%      PreferredPoolNumWorkers - Preferred number of workers in a parallel pool
%      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
%
%   See also parallel.Cluster, parcluster.

% Copyright 2011-2023 The MathWorks, Inc.

classdef (ConstructOnLoad = true) Generic < parallel.cluster.CJSCluster

    properties (PCTGetSet, Transient, PCTConstraint = 'callback')
        %IndependentSubmitFcn Function to run when submitting an independent job
        %   IndependentSubmitFcn identifies the function to run when you submit
        %   an independent job to the generic cluster. The function runs in the
        %   MATLAB client. This user-defined submit function provides certain
        %   job and task data for the MATLAB worker, and identifies a
        %   corresponding decode function for the MATLAB worker to run.
        IndependentSubmitFcn = [];

        %CommunicatingSubmitFcn Function to run when submitting a communicating job
        %   CommunicatingSubmitFcn identifies the function to run when you submit
        %   an communicating job to the generic cluster. The function runs in the
        %   MATLAB client. This user-defined submit function provides certain
        %   job and task data for the MATLAB worker, and identifies a
        %   corresponding decode function for the MATLAB worker to run.
        CommunicatingSubmitFcn = [];

        %GetJobStateFcn Function to run when querying job state
        %   Called when you call when you call wait, or any other function that
        %   queries the state of a job running on a generic cluster. This
        %   function lets you communicate with the cluster, to provide any
        %   instructions beyond the normal toolbox action of retrieving the job
        %   state from disk. To identify the job for the cluster, the function
        %   should include a call to getJobClusterData.
        GetJobStateFcn = [];

        %CancelJobFcn Function to run when cancelling a job
        %   Called when you call CANCEL for a job running on a generic
        %   cluster. This function lets you communicate with the cluster, to
        %   provide any instructions beyond the normal toolbox action of
        %   changing the state of the job. To identify the job for the cluster,
        %   the function should include a call to getJobClusterData.
        CancelJobFcn = [];

        %CancelTaskFcn Function to run when cancelling a task
        %   Called when you call CANCEL for a task running on a generic
        %   cluster. This function lets you communicate with the cluster, to
        %   provide any instructions beyond the normal toolbox action of
        %   changing the state of the task. To identify the task for the cluster,
        %   the function should include a call to getJobClusterData.
        CancelTaskFcn = [];

        %DeleteJobFcn Function to run when deleting a job
        %   Called when you call DELETE for a job running on a generic
        %   cluster. This function lets you communicate with the cluster, to
        %   provide any instructions beyond the normal toolbox action of
        %   deleting the job from disk. To identify the job for the cluster,
        %   the function should include a call to getJobClusterData.
        DeleteJobFcn = [];

        %DeleteTaskFcn Function to run when deleting a task
        %   Called when you call DELETE for a task running on a generic
        %   cluster. This function lets you communicate with the cluster, to
        %   provide any instructions beyond the normal toolbox action of
        %   deleting the task from disk. To identify the task for the cluster,
        %   the function should include a call to getJobClusterData.
        DeleteTaskFcn = [];
    end

    properties (PCTGetSet, Transient, PCTConstraint = 'string')
        %PluginScriptsLocation Folder containing scheduler plugin scripts
        %   This folder will be searched at runtime for plugin scripts
        %   with their associated default name. Functions in this folder
        %   will take priority over any others which exist on the path.
        PluginScriptsLocation = '';
    end

    properties (PCTGetSet, Transient, PCTConstraint = 'additionalproperties')
        %AdditionalProperties Additional properties for plugin scripts
        %   Stores user defined properties. Allowed values are character
        %   vector, logical scalar, and numeric scalar. Values stored here
        %   are available in all plugin scripts and allow the behaviour
        %   of the plugin scripts to be modified.
        AdditionalProperties
    end

    properties (Hidden, Transient, SetAccess = immutable)
        % Internal attribute specifying which version of the generic
        % cluster API we present to the user. This is either set with an
        % undocumented name-value pair at construction, or by the
        % scheduler component.
        Version = parallel.cluster.generic.GenericVersion.Current;
    end

    properties (Hidden, Dependent, Transient, PCTConstraint = 'string')
        %IntegrationScriptsLocation Folder containing scheduler plugin scripts
        %   IntegrationScriptsLocation is not recommended. Use
        %   PluginScriptsLocation instead.
        %
        %   This folder will be searched at runtime for plugin scripts
        %   with their associated default name. Functions in this folder
        %   will take priority over any others which exist on the path.
        IntegrationScriptsLocation;
    end
    methods
        function set.IntegrationScriptsLocation(obj, val)
            obj.PluginScriptsLocation = val;
        end
        function value = get.IntegrationScriptsLocation(obj)
            value = obj.PluginScriptsLocation;
        end
    end

    properties (Constant, GetAccess = private)
        AllPluginFunctionProperties = { ...
            'IndependentSubmitFcn', ...
            'CommunicatingSubmitFcn', ...
            'GetJobStateFcn', ...
            'CancelJobFcn', ...
            'CancelTaskFcn', ...
            'DeleteJobFcn', ...
            'DeleteTaskFcn'};

        %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 = [ ...
            "OperatingSystem", ...
            "HasSharedFilesystem", ...
            "Version" ];

        %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.
        %
        % It is important that the memento is small as it will be
        % serialized and displayed in the command window as a link. Thus it
        % will not include AdditionalProperties or the properties listed in
        % AllPluginFunctionProperties.
        FullMementoProperties = "ClusterMatlabRoot";

        % A string array that stores all the properties needed for a
        % minimal cluster memento for this class. It should include all
        % those defined in MinimalMementoProperties in this class and in
        % all superclasses.
        AllNonVersionDependentPropertiesForMinimalMemento = parallel.Cluster.hDetermineMementoProperties( ...
            ?parallel.cluster.Generic, parallel.internal.types.ClusterMementoType.Minimal);

        % A string array that stores all the properties needed for a full
        % cluster memento for this class. It should include all those
        % defined in MinimalMementoProperties and FullMementoProperties in
        % this class and in all superclasses.
        AllNonVersionDependentPropertiesForFullMemento = parallel.Cluster.hDetermineMementoProperties( ...
            ?parallel.cluster.Generic, parallel.internal.types.ClusterMementoType.Full);
    end

    properties (Transient, Access = private)
        JobsWithGetJobStateFcnRunning = [];

        % Listener for AdditionalProperties which allows us to recognize when the
        % user modifies it.
        AdditionalPropertiesListener = [];
    end

    % Version specific flags which modify the behaviour of the Generic
    % cluster
    properties (Dependent, Access = private)
        % Internal attribute specifying whether to use the default plugin
        % scripts in PluginScriptsLocation or use the plugin script
        % properties.
        UsePluginScriptsLocation

        % Internal attribute specifying whether AdditionalProperties is
        % supported.
        SupportsAdditionalProperties

        % Internal attribute specifying whether postConstructFcn is
        % supported.
        SupportsPostConstructFcn

        % Internal attribute specifying whether PluginScriptsLocation
        % refers to a MATLAB namespace/package
        IsPluginScriptsLocationAPackage
    end
    methods
        function usePluginScriptsLocation = get.UsePluginScriptsLocation(obj)
            import parallel.cluster.generic.GenericVersion
            usePluginScriptsLocation = obj.Version > GenericVersion.R2016b;
        end

        function supportsAdditionalProperties = get.SupportsAdditionalProperties(obj)
            import parallel.cluster.generic.GenericVersion
            supportsAdditionalProperties = obj.Version > GenericVersion.R2016b;
        end

        function supportsPostConstructFcn = get.SupportsPostConstructFcn(obj)
            import parallel.cluster.generic.GenericVersion
            supportsPostConstructFcn = obj.Version > GenericVersion.R2016b;
        end

        function isPluginScriptsLocationAPackage = get.IsPluginScriptsLocationAPackage(obj)
            isPluginScriptsLocationAPackage = iIsExistingPackage(obj.PluginScriptsLocation);
        end
    end

    properties (SetAccess = immutable)
        %Host Identifies the headnode of this cluster
        Host
    end

    methods
        function obj = Generic(varargin)
            %Generic Create a generic cluster instance
            %   parallel.cluster.Generic(p1, v1, p2, v2, ...) builds a generic
            %   cluster with the specified property values. The properties can
            %   include any of the properties of the Generic cluster class, or a
            %   profile.
            %
            %   Example:
            %   % Create a Generic cluster using the 'myGeneric' profile.
            %   myCluster = parallel.cluster.Generic('Profile', 'myGeneric');
            %
            %   See also parcluster, parallel.Cluster.

            import parallel.internal.cluster.ConstructorArgsHelper
            import parallel.internal.cluster.StorageFactory
            import parallel.cluster.CJSCluster
            import parallel.cluster.Generic

            try
                % For child classes of Generic, the 1st argin may be the
                % MetaClass of that class for use in interpreting the ctr
                % args and for our Type property.
                if nargin > 0 && iIsMetaClassScalar(varargin{1})
                    metaClass = varargin{1};
                    varargin = varargin(2:end);
                else
                    metaClass = ?parallel.cluster.Generic;
                end
                type = extractAfter(metaClass.Name, metaClass.ContainingPackage.Name + ".");

                [propsFromArgs, cProf] = ...
                    ConstructorArgsHelper.interpretClusterCtorArgs( ...
                    metaClass, varargin{:});
            catch E
                throw(E);
            end

            [storLoc, storLocValType] = CJSCluster.chooseDataLocation(propsFromArgs);
            storageType = iChooseStorageType(propsFromArgs);

            % If storageType is not recognized it errors with InvalidStorageType
            storage = StorageFactory.createStorage(storageType, storLoc);

            obj@parallel.cluster.CJSCluster(type, cProf, storage, storLocValType, storLoc);

            obj.Host = parallel.internal.general.HostNameUtils.getLocalCanonicalHostName();
            obj.Version = iExtractGenericVersion(propsFromArgs);

            % Seed factory values of plugin script properties
            cellfun(@(x) obj.PropStorage.addFactoryValue(x, []), ...
                Generic.AllPluginFunctionProperties);

            obj.PropStorage.addFactoryValue('PluginScriptsLocation', '');

            if obj.SupportsAdditionalProperties
                obj.PropStorage.addFactoryValue('AdditionalProperties', parallel.cluster.AdditionalProperties());
                obj.setAdditionalPropertiesListener();
            else
                obj.PropStorage.addFactoryValue('AdditionalProperties', parallel.cluster.AdditionalProperties.empty());
            end

            obj.applyConstructorArgs(propsFromArgs, true);

            if obj.SupportsPostConstructFcn
                obj.runPostConstructFcn();
            end
        end

        function setJobClusterData(cluster, job, userdata)
            %setJobClusterData Set specific userdata for a job on a generic cluster
            %   setJobClusterData(C, J, DATA) stores DATA for the job J that is
            %   running under the generic cluster C. You can later retrieve this
            %   information with the function getJobClusterData. For example, it
            %   might be useful to store the third-party cluster's external ID for
            %   this job, so that the function specified in GetJobStateFcn can later
            %   query the cluster about the state of the job.
            %
            %   See also parallel.cluster.Generic/getJobClusterData.

            validateattributes(cluster, {'parallel.Cluster'}, {'scalar'});
            validateattributes(job, {'parallel.job.CJSIndependentJob', ...
                'parallel.job.CJSCommunicatingJob'}, {'scalar'});
            try
                job.JobSchedulerData = struct('type', 'generic', ...
                    'userdata', {userdata});
            catch E
                throw(E);
            end
        end

        function userdata = getJobClusterData(cluster, job)
            %getJobClusterData Get specific userdata for a job on a generic cluster
            %   userdata = getJobClusterData(C, J) returns data stored for
            %   the job J that was derived from the generic cluster C. The
            %   information was originally stored with the function
            %   setJobClusterData. For example, it might be useful to store the
            %   third-party scheduler's external ID for this job, so that the
            %   function specified in GetJobStateFcn can later query the scheduler
            %   about the state of the job.
            %
            %
            %   See also parallel.cluster.Generic/setJobClusterData.

            validateattributes(cluster, {'parallel.Cluster'}, {'scalar'});
            validateattributes(job, {'parallel.job.CJSIndependentJob', ...
                'parallel.job.CJSCommunicatingJob'}, {'scalar'});

            userdata = [];
            try
                if cluster.wasSubmittedByThisType(job)
                    struc    = job.JobSchedulerData;
                    userdata = struc.userdata;
                end
            catch E
                throw(E);
            end
        end

        function fname = getLogLocation(cluster, jobOrTask)
            %getLogLocation Get log location for a job or task
            %   logfile = getLogLocation(C, cJob) for a generic cluster C and
            %   communicating job cJob returns the location where the log data should
            %   be stored for the whole job cJob.
            %
            %   logfile = getLogLocation(C, IT) for a generic cluster C and task IT of
            %   an independent job returns the location where the log data should
            %   be stored for the task IT.
            %
            %   This function can be useful during submission to instruct the
            %   third-party cluster to put worker output logs in the correct
            %   location.

            validateattributes( cluster, {'parallel.cluster.Generic'}, {'scalar'});

            if numel(jobOrTask) == 1
                if isa(jobOrTask, 'parallel.CommunicatingJob')
                    jobSupportID = jobOrTask.hGetSupportID();
                    fname = cluster.Support.getCommJobLogFile(jobSupportID);
                    return
                elseif isa(jobOrTask, 'parallel.Task')
                    task = jobOrTask;
                    job  = task.Parent;
                    if isa(job, 'parallel.IndependentJob')
                        jobSupportID = job.hGetSupportID();
                        fname = cluster.Support.getIndJobTaskLogFile( ...
                            jobSupportID, task.hGetSupportID());
                        return
                    end
                end
            end
            % If we get here, there's a problem
            error(message('parallel:cluster:GenericLogLocation'));
        end

        function dirname = getJobFolder(cluster, job)
            %getJobFolder Return folder on the client where jobs are stored
            %   getJobFolder(c, j) for a generic cluster c and job j returns the
            %   folder on disk where files relating to job j are stored. This folder
            %   is valid only at the MATLAB client.
            %
            %   See also parallel.cluster.Generic/getJobFolderOnCluster.

            validateattributes(cluster, {'parallel.cluster.Generic'}, {'scalar'});
            validateattributes(job, {'parallel.Job'}, {'scalar'});
            dirname = cluster.Support.getJobDirOnClient(job.hGetSupportID());
        end

        function dirname = getJobFolderOnCluster(cluster, job)
            %getJobFolderOnCluster Return folder on cluster where jobs are stored
            %   getJobFolderOnCluster(c, j) for a generic cluster c and job j returns the
            %   folder on disk where files relating to job j are stored. This folder
            %   is valid on the cluster. An error is reported if the
            %   HasSharedFilesystem property of the cluster is FALSE.
            %
            %   See also parallel.cluster.Generic/getJobFolder.

            validateattributes(cluster, {'parallel.cluster.Generic'}, {'scalar'});
            validateattributes(job, {'parallel.Job'}, {'scalar'});
            if ~cluster.HasSharedFilesystem
                error(message('parallel:cluster:GenericJobDirOnCluster'));
            end
            try
                dirname = cluster.Support.getJobDirForOs( ...
                    job.hGetSupportID(), cluster.OperatingSystem);
            catch E
                throw(E);
            end
        end
    end

    methods (Hidden)
        % Add Cluster Properties to Display
        function [clusterPropertyMap, propNames] = hGetDisplayItems(obj, diFactory)
            clusterPropertyMap = hGetDisplayItems@parallel.Cluster(obj, diFactory);

            % The order the properties appear in the propNames array is the
            % order in which they will be displayed.
            propNames = {...
                'Profile', ...
                'Modified', ...
                'Host', ...
                'NumWorkers', ...
                'NumThreads', ...
                'Separator', ...
                'JobStorageLocation', ...
                'ClusterMatlabRoot', ...
                'OperatingSystem', ...
                'Separator', ...
                'RequiresOnlineLicensing', ...
                'PreferredPoolNumWorkers'};

            [propNames, clusterPropertyMap] = obj.appendVersionSpecificDisplay(diFactory, propNames, clusterPropertyMap);
        end

        function hDestroyJob(cluster, job)
            cluster.deleteJobOrTask('DeleteJobFcn', job, job);
        end

        function hDestroyTask(cluster, task)
            cluster.deleteJobOrTask('DeleteTaskFcn', task.Parent, task);
        end

        function OK = hCancelJob(cluster, job)
            OK = cluster.cancelJobOrTask('CancelJobFcn', job, job);
        end

        function OK = hCancelTask(cluster, task)
            OK = cluster.cancelJobOrTask('CancelTaskFcn', task.Parent, task);
        end

        function hSubmitIndependentJob(cluster, job, jobSupport, jobSupportID)
            import parallel.internal.cluster.SubmissionChecks
            import parallel.internal.types.GenericPluginFunctions

            pluginFcnType = GenericPluginFunctions.IndependentSubmitFcn;
            pluginFcnHandle = cluster.getPluginFcnHandle(pluginFcnType);

            % Error if required plugin unspecified
            if isempty(pluginFcnHandle)
                error(cluster.getIndependentSubmitFcnRequiredMessage(pluginFcnType));
            end

            SubmissionChecks.checkIndependentTasksBeforeSubmission(job);

            jobIsCommunicating = false;
            cluster.submitJobCommon(job, jobSupport, jobSupportID, ...
                jobIsCommunicating, pluginFcnHandle, pluginFcnType);
        end

        function hSubmitCommunicatingJob(cluster, job, jobSupport, jobSupportID)
            import parallel.internal.cluster.CJSJobMethods
            import parallel.internal.cluster.SubmissionChecks
            import parallel.internal.types.GenericPluginFunctions

            pluginFcnType = GenericPluginFunctions.CommunicatingSubmitFcn;
            pluginFcnHandle = cluster.getPluginFcnHandle(pluginFcnType);

            % Error if required plugin unspecified
            if isempty(pluginFcnHandle)
                error(cluster.getCommunicatingSubmitFcnRequiredMessage(pluginFcnType));
            end

            SubmissionChecks.checkSimultaneousTasksBeforeSubmission(job);
            maxW = SubmissionChecks.checkWorkerLimsForSimultaneousJob( ...
                cluster, job);
            CJSJobMethods.duplicateTasksOfSimultaneousJob( ...
                job, jobSupport, jobSupportID, maxW);

            jobIsCommunicating = true;
            cluster.submitJobCommon(job, jobSupport, jobSupportID, ...
                jobIsCommunicating, pluginFcnHandle, pluginFcnType);
        end

        function hSetProperty(obj, names, values)
            if ~iscell(names)
                names  = {names};
                values = {values};
            end

            % Check for setting properties which are incompatible with
            % Version.
            [legacyProps, futureProps] = obj.getInvalidProperties();
            for ii = 1:length(names)
                if ismember(names{ii}, futureProps) && obj.isNotCurrentValue(names{ii}, values{ii})
                    error(message('parallel:cluster:GenericCannotSetFutureProperty', names{ii}));
                elseif ismember(names{ii}, legacyProps) && obj.isNotCurrentValue(names{ii}, values{ii})
                    error(message('parallel:cluster:GenericCannotSetLegacyProperty', names{ii}));
                end
            end

            hSetProperty@parallel.cluster.CJSCluster(obj, names, values);
        end

        function name = hGetSchedulerComponentName(obj)
            if class(obj) == "parallel.cluster.Generic"
                % Generic cluster is associated with different
                % SchedulerComponents depending on Version.
                name = obj.Version.SchedulerName;
            else
                % If we're derived from Generic, just use the scheduler
                % Type.
                name = obj.Type;
            end
        end

        % Return flags indicating the visibility of certain operations
        % between a client and a running worker. The output flags are
        % expected to be:
        % {workersSeeClientWrites, workersSeeClientRemoves,
        % clientSeesWorkerWrites, clientSeesWorkerRemoves}
        % Override the default implementation in
        % parallel.cluster.CJSCluster which is all true.
        function flags = hGetOperationVisibilityFlags(obj)
            if obj.HasSharedFilesystem
                flags = num2cell(true(4,1));
            else
                % Default implementation of Generic with a nonshared file
                % system assumes file mirroring allows the client to see
                % updates made by workers
                flags = num2cell(false(4,1));
                flags{3} = true;
            end

            % Allow AdditionalProperties to override any visibility flags
            propertyNames = {...
                'WorkersSeeClientUpdates', ...
                'WorkersSeeClientRemoves', ...
                'ClientsSeeWorkerUpdates', ...
                'ClientsSeeWorkerRemoves'};
            isDefined = ismember(propertyNames, properties(obj.AdditionalProperties));
            for idx = 1:numel(propertyNames)
                if ~isDefined(idx)
                    continue
                end
                definedValue = obj.AdditionalProperties.(propertyNames{idx});
                if islogical(definedValue)
                    flags{idx} = definedValue;
                end
            end
        end

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

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

            % For Generic clusters, in order to use server sockets in the
            % cluster the workers need to tell us how to connect to them
            % via the ValueStore. This can only happen if we (the client)
            % can see worker updates. We allow the plugin scripts to
            % override this value via an AdditionalProperty.
            addProps = obj.AdditionalProperties;
            if isprop(addProps, "ClientConnectsToWorkers") && ...
                    islogical(addProps.ClientConnectsToWorkers) && ...
                    isscalar(addProps.ClientConnectsToWorkers)
                allServerSocketsInCluster = addProps.ClientConnectsToWorkers;
            else
                vf = obj.hGetOperationVisibilityFlags();
                allServerSocketsInCluster = vf{3};
            end

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

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

        function [propNames, clusterPropertyMap] = appendVersionSpecificDisplay(cluster, diFactory, propNames, clusterPropertyMap)
            % Adjust cluster display based on Version flags
            import parallel.cluster.Generic

            if cluster.UsePluginScriptsLocation
                propNames = [propNames {'PluginScriptsLocation'}];
                clusterPropertyMap('PluginScriptsLocation') = diFactory.createDefaultItem(cluster.hGetDisplayPropertiesNoError('PluginScriptsLocation'));
            else
                propNames = [propNames Generic.AllPluginFunctionProperties];
                values = diFactory.makeMultipleItems(@createDefaultItem, cluster.hGetDisplayPropertiesNoError(Generic.AllPluginFunctionProperties));
                clusterPropertyMap = [clusterPropertyMap; containers.Map(Generic.AllPluginFunctionProperties, values)];
            end

            if cluster.SupportsAdditionalProperties
                propNames = [propNames {'AdditionalProperties'}];
                displayValue = getString(message('parallel:cluster:AdditionalPropertiesLink'));
                clusterPropertyMap('AdditionalProperties') = diFactory.createLinkItem(displayValue, cluster.hGetDisplayPropertiesNoError('AdditionalProperties'));
            end
        end

        function [legacyProperties, futureProperties] = getInvalidProperties(cluster)
            % Get properties which are not available for the clusterVersion
            import parallel.cluster.Generic
            import parallel.cluster.generic.GenericVersion

            legacyProperties = {};
            futureProperties = {};
            if cluster.Version > GenericVersion.R2016b
                legacyProperties = [legacyProperties Generic.AllPluginFunctionProperties];
            elseif cluster.Version < GenericVersion.Current
                currentProps = {'AdditionalProperties', 'IntegrationScriptsLocation', 'PluginScriptsLocation'};
                futureProperties = [futureProperties currentProps];
            end
        end

        function tf = isSubmittedGenericClusterJob(cluster, job)
            % isSubmittedGenericClusterJob - job submitted by generic cluster
            % c.f. pIsSubmittedGenericSchedulerJob
            import parallel.internal.types.States

            jobStateEnumInStorage = job.hGetProperty('StateEnum');
            tf = cluster.wasSubmittedByThisType(job) && ...
                jobStateEnumInStorage > States.Pending;
        end

        function submitJobCommon(cluster, job, jobSupport, jobSupportID, ...
                jobIsCommunicating, pluginFcnHandle, pluginFcnType)
            % Common code for job submission, calls the appropriate user-supplied function.

            cluster.ensureLicenseNumberIsSetOrError();
            taskSIds = arrayfun(@hGetSupportID, job.Tasks, UniformOutput=false);
            jobSupport.prepareJobForSubmission(job, jobSupportID);

            % TODO:later move this?
            storedEnv = distcomp.pClearEnvironmentBeforeSubmission();
            cleanup = onCleanup(@() distcomp.pRestoreEnvironmentAfterSubmission(storedEnv));

            % NB: we still use distcomp.setprop to transmit information to the SubmitFcn.
            environmentProperties = jobSupport.getJobSubmissionSetprop(jobSupportID, taskSIds);
            [~, matlabExe, matlabArgs] = parallel.internal.apishared.WorkerCommand.defaultCommand( ...
                cluster.OperatingSystem, cluster.ClusterMatlabRoot, jobIsCommunicating);
            environmentProperties.pSetExecutableStrings(matlabExe, matlabArgs);
            token = '';
            webID = '';
            licenseNumber = '';
            if cluster.RequiresOnlineLicensing
                currentUser = parallel.internal.mwa.ensureLoggedIn();
                token = currentUser.getToken();
                webID = cluster.LicenseEntitlement.Id;
                licenseNumber = cluster.LicenseEntitlement.LicenseNumber;
            end
            environmentProperties.pSetLicensingInformation(cluster.RequiresOnlineLicensing, ...
                token, webID, licenseNumber);

            cluster.setJobClusterData(job, []);
            [pluginFcnHandle, userArgs] = iExtractFunctionArgs(pluginFcnHandle);

            try
                feval(pluginFcnHandle, cluster, job, environmentProperties, userArgs{:});
            catch submitError
                cluster.setJobClusterData(job, []);
                ME = MException(cluster.getSubmissionFailedMessage(pluginFcnType));
                ME = addCause(ME, submitError);
                throw(ME);
            end
        end

        function OK = cancelJobOrTask(cluster, pluginFunctionProp, job, entity)
            import parallel.internal.types.GenericPluginFunctions
            pluginFcnType = GenericPluginFunctions.(pluginFunctionProp);
            pluginFcnHandle = cluster.getPluginFcnHandle(pluginFcnType);

            % Do NOT call the user fcn if any of the following:
            % 1. No plugin script defined or found
            % 2. The job is not a submitted Generic scheduler job
            shouldShortcut = isempty(pluginFcnHandle) || ...
                ~cluster.isSubmittedGenericClusterJob(job);
            if shouldShortcut
                OK = true;
                return
            end

            [pluginFcnHandle, userArgs] = iExtractFunctionArgs(pluginFcnHandle);

            try
                % Call the user supplied function
                OK = feval(pluginFcnHandle, cluster, entity, userArgs{:});
            catch err
                OK = false;
                parallel.internal.warningNoBackTrace( ...
                    cluster.getCancelErrorMessage(pluginFcnType, err));
            end

            % Check that the user has returned a logical
            if ~(islogical(OK) && isscalar(OK))
                parallel.internal.warningNoBackTrace( ...
                    cluster.getCancelReturnErrorMessage(pluginFcnType));
                OK = false;
            end
        end

        function deleteJobOrTask(cluster, pluginFunctionProp, job, entity)
            import parallel.internal.types.GenericPluginFunctions
            pluginFcnType = GenericPluginFunctions.(pluginFunctionProp);
            pluginFcnHandle = cluster.getPluginFcnHandle(pluginFcnType);

            % Do NOT call the user fcn if any of the following:
            % 1. No plugin script defined or found
            % 2. The job is not a submitted Generic scheduler job
            shouldShortcut = isempty(pluginFcnHandle) || ...
                ~cluster.isSubmittedGenericClusterJob(job);
            if shouldShortcut
                return
            end

            [pluginFcnHandle, userArgs] = iExtractFunctionArgs(pluginFcnHandle);

            try
                % Call the user supplied function
                feval(pluginFcnHandle, cluster, entity, userArgs{:});
            catch err
                parallel.internal.warningNoBackTrace( ...
                    cluster.getDeleteErrorMessage(pluginFcnType, err));
            end
        end

        function runPostConstructFcn(cluster)
            import parallel.internal.types.GenericPluginFunctions
            pluginFcnType = GenericPluginFunctions.PostConstructFcn;
            pluginFcnHandle = cluster.getPluginFcnHandle(pluginFcnType);

            % If no plugin script defined or found we cannot proceed
            if isempty(pluginFcnHandle)
                return
            end

            [pluginFcnHandle, userArgs] = iExtractFunctionArgs(pluginFcnHandle);

            try
                % Call the user supplied function
                feval(pluginFcnHandle, cluster, userArgs{:});
            catch err
                parallel.internal.warningNoBackTrace(message( ...
                    'parallel:cluster:GenericDefaultFcnError', ...
                    pluginFcnType.DefaultFunction, err.getReport()));
            end
        end

        function fcnHandle = getPluginFcnHandle(cluster, pluginFcnType)
            if cluster.UsePluginScriptsLocation
                if cluster.IsPluginScriptsLocationAPackage
                    fcnHandle = iGetHandleToFcnInPkg(pluginFcnType.DefaultFunction, cluster.PluginScriptsLocation);
                else
                    [pluginFcnName, pluginFcnParentDirectory] = iGetDefaultPluginFcnAndParentDirectory(pluginFcnType, cluster.PluginScriptsLocation);
                    if isempty(pluginFcnName)
                        fcnHandle = [];
                        return
                    end
                    iAddPluginFcnParentDirectoryToTopOfPathIfNeccessary(pluginFcnName, pluginFcnParentDirectory);
                    fcnHandle = iGetFunctionHandle(pluginFcnName);
                end
            else
                % Using pre-17a API, return value of the appropriate plugin
                % function property
                fcnHandle = cluster.(pluginFcnType.Name);
            end
        end

        function setAdditionalPropertiesListener(cluster)
            cluster.AdditionalPropertiesListener = event.listener(cluster.AdditionalProperties, ...
                parallel.cluster.AdditionalProperties.ModifiedEventName, @cluster.setAdditionalPropertiesUserModified);
        end

        function setAdditionalPropertiesUserModified(cluster, ~, ~)
            % If AdditionalProperties is modified PropStorage needs to be
            % notified for it to be saved to the profile. This is because
            % the value of AdditionalProperties is a handle class so
            % modification does not invoke the set method on the cluster object.
            if cluster.SupportsAdditionalProperties
                cluster.PropStorage.setUserModified('AdditionalProperties');
            else
                error(message('parallel:cluster:GenericCannotSetFutureProperty', 'AdditionalProperties'));
            end
        end
    end

    methods (Access = protected)
        function jobState = getJobState(cluster, job, jobStateInStorage)
            import parallel.internal.types.GenericPluginFunctions
            import parallel.internal.types.States

            pluginFcnType = GenericPluginFunctions.GetJobStateFcn;
            pluginFcnHandle = cluster.getPluginFcnHandle(pluginFcnType);

            % Do NOT call the user fcn if any of the following:
            % 1. No plugin function defined or found
            %    (this check also protects against calling a user fcn on a
            %    cluster object created from a memento)
            % 2. The job is not a submitted generic scheduler job
            % 3. If the job state on disk is already terminal
            % 4. If we are already in the user function for this job
            shouldShortcut = isempty(pluginFcnHandle) || ...
                ~cluster.isSubmittedGenericClusterJob(job) || ...
                jobStateInStorage.isTerminal() || ...
                ismember(job.Id, cluster.JobsWithGetJobStateFcnRunning);
            if shouldShortcut
                jobState = jobStateInStorage;
                return
            end

            % If we are going to run the user function we need to ensure we
            % don't re-enter the user code, so store the job Id in
            % JobsWithGetJobStateFcnRunning and make sure we remove it at
            % the end.
            cluster.JobsWithGetJobStateFcnRunning(end+1) = job.Id;
            cleanup = onCleanup(@() removeJobFromRunningList(cluster, job.Id));

            [pluginFcnHandle, userArgs] = iExtractFunctionArgs(pluginFcnHandle);

            try
                userSuppliedStateStr = feval(pluginFcnHandle, cluster, job, ...
                    jobStateInStorage.Name, userArgs{:});
            catch err
                parallel.internal.warningNoBackTrace(cluster.getJobStateErrorMessage(pluginFcnType, err));
                userSuppliedStateStr = jobStateInStorage.Name;
            end

            % Check that the user has returned a string
            userSuppliedStateStr = convertStringsToChars(userSuppliedStateStr);
            if ~iIsCharRowVec(userSuppliedStateStr)
                parallel.internal.warningNoBackTrace(cluster.getInvalidJobStateTypeMessage(pluginFcnType));
                userSuppliedStateStr = jobStateInStorage.Name;
            end

            % Check that the user has returned an allowed state
            try
                jobState = States.fromName(userSuppliedStateStr);
                isStateAllowed = iIsAllowedState(jobState);
            catch
                isStateAllowed = false;
            end

            % If the state is not allowed, warn and return the state from
            % storage
            if ~isStateAllowed
                parallel.internal.warningNoBackTrace(cluster.getInvalidJobStateStringMessage(pluginFcnType, userSuppliedStateStr));
                jobState = jobStateInStorage;
            end
        end

        function tf = wasSubmittedByThisType(cluster, job)
            validateattributes(cluster, {'parallel.cluster.Generic'}, {'scalar'});
            validateattributes(job, {'parallel.job.CJSIndependentJob', ...
                'parallel.job.CJSCommunicatingJob'}, {'scalar'});

            jobData = job.JobSchedulerData;
            tf = ~isempty(jobData) && isequal(jobData.type, 'generic');
        end

        function [names, values] = getClusterSpecificEnvVars(~)
            % Return names/values of environment variables that this cluster must
            % send down to workers.  Since all of this work is done in the user's
            % submit functions, there is nothing to do here.
            names = {};
            values = {};
        end

        % Get all the properties needed for the ClusterMemento
        function props = getAllMementoProperties(obj, clusterMementoType)
            % It is important that the memento is small as it will be
            % serialized and displayed in the command window as a link.
            % Thus it will not include AdditionalProperties or properties
            % listed in AllPluginFunctionProperties. We also do not include
            % GetJobStateFcn or the plugin scripts as we cannot guarantee
            % those will work without the AdditionalProperties: e.g. we
            % cannot use getJobStateFcn for remote and nonshared without
            % ClusterHost.

            import parallel.internal.types.ClusterMementoType
            switch clusterMementoType
                case ClusterMementoType.Minimal
                    props = obj.AllNonVersionDependentPropertiesForMinimalMemento;
                case ClusterMementoType.Full
                    props = obj.AllNonVersionDependentPropertiesForFullMemento;
                otherwise
                    % Default to the full memento
                    props = obj.AllNonVersionDependentPropertiesForFullMemento;
            end
        end

        function val = additionalPropertyOrDefault(obj, prop, default)
            % If the specified AdditionalProperty PROP exists, return that.
            % Otherwise, return DEFAULT.
            if isprop(obj.AdditionalProperties, prop)
                val = obj.AdditionalProperties.(prop);
            else
                val = default;
            end
        end

        function overrideFactoryValue(obj, name, value)
            obj.PropStorage.setValue(name, ...
                parallel.internal.settings.ValueType.Factory, value);
        end
    end

    % Customized messages depending on value of UsePluginScriptsLocation.
    methods (Access = private)
        function msg = getJobStateErrorMessage(cluster, pluginFcnType, err)
            if cluster.UsePluginScriptsLocation
                msg = message('parallel:cluster:GenericDefaultFcnError', ...
                    pluginFcnType.DefaultFunction, err.getReport());
            else
                msg = message('parallel:cluster:GenericGetJobStateFcnError', ...
                    cluster.pluginAsString(pluginFcnType), err.getReport());
            end
        end

        function msg = getInvalidJobStateTypeMessage(cluster, pluginFcnType)
            if cluster.UsePluginScriptsLocation
                msg = message('parallel:cluster:GenericInvalidStateTypeDefault', ...
                    pluginFcnType.DefaultFunction);
            else
                msg = message('parallel:cluster:GenericInvalidStateType', ...
                    cluster.pluginAsString(pluginFcnType));
            end
        end

        function msg = getInvalidJobStateStringMessage(cluster, pluginFcnType, userSuppliedStateStr)
            allowedStatesNames = strjoin(iAllAllowedStates);
            if cluster.UsePluginScriptsLocation
                msg = message('parallel:cluster:GenericInvalidStateStringDefault', ...
                    pluginFcnType.DefaultFunction, userSuppliedStateStr, allowedStatesNames);
            else
                msg = message('parallel:cluster:GenericInvalidStateString', ...
                    cluster.pluginAsString(pluginFcnType), userSuppliedStateStr, allowedStatesNames);
            end
        end

        function msg = getIndependentSubmitFcnRequiredMessage(cluster, pluginFcnType)
            if cluster.UsePluginScriptsLocation
                if isfolder(cluster.PluginScriptsLocation)
                    msg = message('parallel:cluster:GenericIndependentSubmitFcnRequiredDefault', ...
                        pluginFcnType.DefaultFunction);
                else
                    msg = message('parallel:cluster:GenericPluginScriptsLocationDoesNotExistIndependent', ...
                        cluster.PluginScriptsLocation, pluginFcnType.DefaultFunction);
                end
            else
                msg = message('parallel:cluster:GenericIndependentSubmitFcnRequired');
            end
        end

        function msg = getCommunicatingSubmitFcnRequiredMessage(cluster, pluginFcnType)
            if cluster.UsePluginScriptsLocation
                if isfolder(cluster.PluginScriptsLocation)
                    msg = message('parallel:cluster:GenericCommunicatingSubmitFcnRequiredDefault', ...
                        pluginFcnType.DefaultFunction);
                else
                    msg = message('parallel:cluster:GenericPluginScriptsLocationDoesNotExistCommunicating', ...
                        cluster.PluginScriptsLocation, pluginFcnType.DefaultFunction);
                end
            else
                msg = message('parallel:cluster:GenericCommunicatingSubmitFcnRequired');
            end
        end

        function msg = getSubmissionFailedMessage(cluster, pluginFcnType)
            if cluster.UsePluginScriptsLocation
                msg = message('parallel:cluster:GenericSubmissionFailedDefault', ...
                    pluginFcnType.DefaultFunction);
            else
                msg = message('parallel:cluster:GenericSubmissionFailed', ...
                    pluginFcnType.Name, cluster.pluginAsString(pluginFcnType));
            end
        end

        function msg = getCancelErrorMessage(cluster, pluginFcnType, err)
            if cluster.UsePluginScriptsLocation
                msg = message('parallel:cluster:GenericDefaultFcnError', ...
                    pluginFcnType.DefaultFunction, err.getReport());
            else
                msg = message('parallel:cluster:GenericCancelFcnError', ...
                    pluginFcnType.Name, cluster.pluginAsString(pluginFcnType), err.getReport());
            end
        end

        function msg = getCancelReturnErrorMessage(cluster, pluginFcnType)
            if cluster.UsePluginScriptsLocation
                msg = message('parallel:cluster:GenericDefaultReturnError', ...
                    pluginFcnType.DefaultFunction);
            else
                msg = message('parallel:cluster:GenericCancelFcnReturnError', ...
                    pluginFcnType.Name, cluster.pluginAsString(pluginFcnType));
            end
        end

        function msg = getDeleteErrorMessage(cluster, pluginFcnType, err)
            if cluster.UsePluginScriptsLocation
                msg = message('parallel:cluster:GenericDefaultFcnError', ...
                    pluginFcnType.DefaultFunction, err.getReport());
            else
                msg = message('parallel:cluster:GenericDeleteFcnError', ...
                    pluginFcnType.Name, cluster.pluginAsString(pluginFcnType), err.getReport());
            end
        end

        function str = pluginAsString(cluster, pluginFcnType)
            fcnHandle = cluster.(pluginFcnType.Name);
            if isempty(fcnHandle)
                str = '';
            else
                fcnHandle = iExtractFunctionArgs(fcnHandle);
                str = func2str(fcnHandle);
            end
        end

        function removeJobFromRunningList(cluster, jobId)
            cluster.JobsWithGetJobStateFcnRunning = setdiff(cluster.JobsWithGetJobStateFcnRunning, jobId);
        end
    end
end

function [fcn, args] = iExtractFunctionArgs(clusterFcn)
% If it is a cell array then get the first argument which is a function handle
% or string and subsequent arguments are the additional args to the function.
if iscell(clusterFcn)
    fcn  = clusterFcn{1};
    args = clusterFcn(2:end);
else
    fcn  = clusterFcn;
    args = {};
end
end

function fcn = iGetFunctionHandle(str)
% If input is not already a function handle, convert it.
fcn = [];
if ischar(str)
    [~, name, ~] = fileparts(str);
    fcn = str2func(name);
elseif isa(str, 'function_handle')
    fcn = str;
end
end

function [pluginFcnName, pluginFcnParentDirectory] = iGetDefaultPluginFcnAndParentDirectory(pluginFcnType, pluginScriptsLocation)
% Search for default plugin function.
%
% This function looks for the default plugin function and returns its name
% and its parent directory if it can be found. Otherwise it will return
% empty arrays. The parent directory is returned so that it can be added to
% the path immediately before executing the plugin function to ensure the
% correct function is used (see
% iAddPluginFcnParentDirectoryToTopOfPathIfNeccessary). The detailed
% behaviour of this function depends on whether it is being run in
% deployed or MATLAB mode.
%
% In MATLAB mode, it will look for the plugin function in the
% PluginScriptsLocation. If it can be found, the function's name and
% its parent directory (i.e PluginScriptsLocation) will be returned.
%
% In deployed mode, it will first look for the plugin function on the
% search path (it is expected that users will have added the
% PluginScriptsLocation when compiling the app via the -a flag to mcc,
% and so the plugin scripts will be on the path). If it can be found,
% the function's name will be returned.  In addition, if
% PluginScriptsLocation starts with ctfroot, it will also search for
% the plugin function in the PluginScriptsLocation.  If it can be
% found there, then the PluginScriptsLocation will be returned as the
% parent directory.  Otherwise it will be returned as an empty array.

pluginFcnName = [];
pluginFcnParentDirectory = [];

if isdeployed
    if iDoesPluginFcnExist(pluginFcnType.DefaultFunction)
        pluginFcnName = pluginFcnType.DefaultFunction;
    end
    if ~startsWith(pluginScriptsLocation, ctfroot)
        return
    end
end

if isempty(pluginScriptsLocation)
    return
end

fullPath = fullfile(pluginScriptsLocation, pluginFcnType.DefaultFunction);
if iDoesPluginFcnExist(fullPath)
    pluginFcnName = pluginFcnType.DefaultFunction;
    pluginFcnParentDirectory = pluginScriptsLocation;
end
end

function tf = iDoesPluginFcnExist(pluginFcnFile)
% This function checks whether a file of the suitable type for a plugin
% function exists. pluginFcnFile can be either a full path to a file or
% just a file name that you want to check is on the search path.
fileExistsCode = exist(pluginFcnFile, 'file');
tf = fileExistsCode > 1 && fileExistsCode ~= 7;
end

function versionEnum = iExtractGenericVersion(propsFromArgs)
% Extract and return GenericVersion from constructor props if present, otherwise
% return default value
import parallel.cluster.generic.GenericVersion
versionEnum = GenericVersion.Current;
if isKey(propsFromArgs, 'Version')
    value = propsFromArgs('Version');
    version = value{1};
    if isa(version, 'parallel.cluster.generic.GenericVersion')
        versionEnum = version;
    else
        versionEnum = GenericVersion.fromName(version);
    end
    % Remove field from propsFromArgs
    remove(propsFromArgs, 'Version');
end
end

function storageType = iChooseStorageType(propsFromArgs)
% Extract StorageType from constructor's AdditionalProperties prop if present,
% otherwise return the default value - Filesystem
% NOTE: This function doesn't check if StorageType is valid, the
%       parallel.internal.cluster.StorageFactory is responsible for that
if propsFromArgs.isKey('AdditionalProperties')
    addPropsAndValType = propsFromArgs('AdditionalProperties');
    % We don't care about ValueType
    addProps = addPropsAndValType{1};
    if isprop(addProps, 'StorageType')
        storageType = addProps.StorageType;
        return
    end
end
storageType = parallel.internal.types.StorageType.FileStorage;
end

function iAddPluginFcnParentDirectoryToTopOfPathIfNeccessary(script, scriptParentDirectory)
% Ensure the script we are about to call has scriptParentDirectory as its
% parent directory. If not, we add scriptParentDirectory to the front of
% the path.

if ~isempty(scriptParentDirectory) && isfolder(scriptParentDirectory)
    desiredScript = fullfile(scriptParentDirectory, script);
    scriptOnPath = which(script);
    if isempty(scriptOnPath) || ~iIsSameFile(desiredScript, scriptOnPath)
        % Script currently in scope either does not exist or does not have
        % scriptParentDirectory as parent directory. Need to modify the
        % path.
        addpath(scriptParentDirectory, '-begin');
    end
end
end

function allowedStates = iAllAllowedStates()
allStates = enumeration('parallel.internal.types.States');
allowedStatesIdx = arrayfun(@iIsAllowedState, allStates);
allowedStates = {allStates(allowedStatesIdx).Name};
end

function tf = iIsAllowedState(state)
% Has the user given us the name of a valid state the job might be in? Note
% we do not accept 'pending' as to get here the job must have been
% submitted.
import parallel.internal.types.States
tf = States.isValidState(state) && state ~= States.Pending;
end

function tf = iIsSameFile(file1, file2)
tf = parallel.internal.apishared.FilenameUtils.isSameFile(file1, file2);
end

function tf = iIsCharRowVec(text)
tf = ischar(text) && isrow(text);
end

function tf = iIsExistingPackage(pkg)
tf = ~isempty(meta.package.fromName(pkg));
end

function fcnHandle = iGetHandleToFcnInPkg(fcn, pkg)
[~, fcn] = fileparts(fcn);
if iIsFunctionInPackage(fcn, pkg)
    fcnHandle = str2func(pkg + "." + fcn);
else
    fcnHandle = [];
end
end

function tf = iIsFunctionInPackage(fcn, pkg)
m = meta.package.fromName(pkg);
tf = ismember(fcn, {m.FunctionList.Name});
end

function tf = iIsMetaClassScalar(val)
tf = isa(val, "meta.class") && isscalar(val);
end
