classdef RemoteClusterAccess < handle
    %RemoteClusterAccess Connect to clusters when the cluster's client utilities are not available locally
    %   RemoteClusterAccess allows you to establish a connection and run
    %   commands on a remote host. This class is intended for use with the
    %   generic scheduler interface when using remote submission of jobs or
    %   on nonshared file systems.
    %
    %   r = RemoteClusterAccess(username) uses the supplied username when
    %   connecting to the remote host. You will be prompted for a password
    %   when establishing the connection.
    %
    %   r = RemoteClusterAccess(username,P1,V1,...,Pn,Vn) allows
    %   additional parameter-value pairs that modify the behavior of the
    %   connection. The accepted parameters are:
    %    - 'AuthenticationMode' - Authentication mode you use when you connect
    %      to the cluster, specified as a string scalar. Valid options are:
    %      - 'Password' - Enter your SSH password when prompted by MATLAB.
    %      - 'IdentityFile' - Use an identity file on disk. If you specify
    %        a file using the 'IdentityFileName' parameter, you use that file.
    %        Otherwise, MATLAB prompts you to specify the full path to an
    %        identity file when you connect.
    %      - 'Agent' - Interface with an SSH agent running on the client machine.
    %      - 'Multifactor' - Enable the cluster to prompt you for input one
    %        or more times. For example, if two-factor authentication (2FA)
    %        is enabled on the cluster, the cluster will request your password
    %        and a response for the second authentication factor.
    %      If you specify the parameter 'IdentityFilename', the default is
    %      'IdentityFile'. Otherwise, the default is 'Password'.
    %    - 'IdentityFilename' - String scalar containing the full path
    %      to the identity file to use when connecting to a remote host.
    %    - 'IdentityFileHasPassphrase' - Flag indicating if the identity file
    %      requires a passphrase, specified as true or false. If true, you
    %      will be prompted for a password when establishing a connection.
    %      If an identity file is not supplied, this property is not used.
    %      This value is false by default.
    %    - 'Port' - Number specifying the port on the cluster to connect
    %      to, specified as an integer scalar between 1 and 65535. The
    %      default value is 22.
    %
    %   Examples:
    %
    %       Mirror files from the remote data location. Assume the job object
    %       represents the job on the generic scheduler.
    %
    %           remoteConnection = parallel.cluster.RemoteClusterAccess('testname');
    %           connect(remoteConnection,'headnode1','/tmp/filemirror');
    %           startMirrorForJob(remoteConnection,job);
    %           submit(job);
    %           % Wait for the job to finish
    %           wait(job);
    %           % Ensure that all the local files are up to date and remove the
    %           % remote files
    %           doLastMirrorForJob(remoteConnection,job);
    %           % Get the output arguments for the job
    %           results = fetchOutputs(job);
    %
    %   For more information and detailed examples, see the integration scripts
    %   provided via the Add On Explorer and MATLAB(R) Central File Exchange:
    %
    %   Grid Engine (formerly SGE):
    %   https://www.mathworks.com/matlabcentral/fileexchange/52816-parallel-computing-toolbox-integration-for-matlab-distributed-computing-server-with-grid-engine
    %
    %   HTCondor:
    %   https://www.mathworks.com/matlabcentral/fileexchange/78823-parallel-computing-toolbox-integration-for-matlab-distributed-computing-server-with-htcondor
    %
    %   LSF:
    %   https://www.mathworks.com/matlabcentral/fileexchange/52817-parallel-computing-toolbox-integration-for-matlab-distributed-computing-server-with-ibm-platform-lsf
    %
    %   PBS:
    %   https://www.mathworks.com/matlabcentral/fileexchange/52815-parallel-computing-toolbox-integration-for-matlab-distributed-computing-server-with-pbs
    %
    %   SLURM:
    %   https://www.mathworks.com/matlabcentral/fileexchange/52807-parallel-computing-toolbox-integration-for-matlab-distributed-computing-server-with-slurm
    %
    %   parallel.cluster.RemoteClusterAccess methods:
    %      RemoteClusterAccess          - Connect to clusters when the cluster's client utilities are not available locally
    %      connect                      - Connect to a remote host
    %      copyFileFromRemote           - Copy file or folder from the remote host
    %      copyFileToRemote             - Copy file or folder to the remote host
    %      disconnect                   - Disconnect from a remote host
    %      doLastMirrorForJob           - Perform a last mirror for the supplied job
    %      getRemoteJobLocation         - Get the remote location for the supplied job ID
    %      isJobUsingConnection         - Query the connection to see if the job is being mirrored
    %      remoteDelete                 - Delete a file or folder on the remote host
    %      resumeMirrorForJob           - Resume the mirroring of files for the supplied job
    %      runCommand                   - Run the supplied command on the remote host
    %      startMirrorForJob            - Start the mirroring of files
    %      stopMirrorForJob             - Stop the mirroring of files for the supplied job
    %      getConnectedAccess           - Return a connected RemoteClusterAccess instance
    %      getConnectedAccessWithMirror - Return a connected RemoteClusterAccess instance
    %
    %   parallel.cluster.RemoteClusterAccess properties:
    %      AuthenticationMode        - How to authenticate when you connect to the remote host
    %      Hostname                  - Name of the remote host to access
    %      IdentityFileHasPassphrase - True if the identity file requires a passphrase
    %      IdentityFilename          - Full path to the identity file used when connecting to the remote host
    %      IsConnected               - True if there is an active connection to the remote host
    %      IsFileMirrorSupported     - True if file mirroring is supported for this connection
    %      JobStorageLocation        - Location on the remote host for files that are being mirrored
    %      Port                      - Port to connect to on the remote host
    %      UseIdentityFile           - True if an identity file should be used when connecting to the remote host
    %      Username                  - Username for connecting to the remote host
    %
    %   See also parallel.Job/submit, parallel.Job/fetchOutputs, parcluster,
    %            parallel.cluster.Generic.

    %  Copyright 2010-2023 The MathWorks, Inc.

    properties (SetAccess = private)
        %Hostname Name of the remote host to access
        %
        %   See also parallel.cluster.RemoteClusterAccess/connect.
        Hostname char = ''

        %JobStorageLocation Location on the remote host for files that are being mirrored
        %
        %   See also parallel.cluster.RemoteClusterAccess/connect.
        JobStorageLocation char = ''
        % Note: we store the remote JobStorageLocation using slashes as
        % provided by the user.

        %Username for connecting to the remote host
        %
        %   See also parallel.cluster.RemoteClusterAccess.
        Username char

        %AuthenticationMode How to authenticate when connecting to the remote host
        AuthenticationMode

        %IdentityFilename Full path to the identity file (public key) used when connecting to the remote host
        %
        %   See also parallel.cluster.RemoteClusterAccess.UseIdentityFile,
        %            parallel.cluster.RemoteClusterAccess.IdentityFileHasPassphrase,
        %            parallel.cluster.RemoteClusterAccess.
        IdentityFilename

        %IdentityFileHasPassphrase True if the identity file requires a passphrase
        %
        %   See also parallel.cluster.RemoteClusterAccess.UseIdentityFile,
        %            parallel.cluster.RemoteClusterAccess.IdentityFilename,
        %            parallel.cluster.RemoteClusterAccess.RemoteClusterAccess.
        IdentityFileHasPassphrase (1,1) logical

        %Port to connect to on the remote host
        Port (1,1) double
    end

    properties (Dependent)
        %UseIdentityFile True if an identity file should be used when connecting to the remote host
        %
        %   See also parallel.cluster.RemoteClusterAccess.IdentityFilename,
        %            parallel.cluster.RemoteClusterAccess.IdentityFileHasPassphrase,
        %            parallel.cluster.RemoteClusterAccess.RemoteClusterAccess.
        UseIdentityFile (1,1) logical

        %IsFileMirrorSupported True if file mirroring is supported for this connection
        %   File mirroring is not supported if no remote JobStorageLocation is supplied
        %   to the connect() method.
        %
        %   See also parallel.cluster.RemoteClusterAccess/connect.
        IsFileMirrorSupported (1,1) logical

        %IsConnected True if there is an active connection to the remote host
        %
        %   See also parallel.cluster.RemoteClusterAccess/connect,
        %            parallel.cluster.RemoteClusterAccess/disconnect.
        IsConnected (1,1) logical
    end

    properties
        %AutoReconnect True if the SSH session should be recreated if running a remote command errors
        AutoReconnect (1,1) logical = true
    end

    properties (Hidden)
        %PreserveRemoteFiles Whether the remote copies of job files should
        %   be kept after the job is complete. By default, the remote
        %   copies of files pertaining to completed jobs are deleted. The
        %   local copies are only deleted once the job object is deleted.
        PreserveRemoteFiles (1,1) logical = false
    end

    properties (Hidden, Dependent)
        %FileArchiveType Specifies which type of file archive will be used
        %   by the file mirror to transfer files. Allowed values are
        %   'None', 'Tar' and 'Zip'.
        FileArchiveType (1,1) parallel.internal.remoteaccess.FileArchiveType

        %InitialMirrorWait Initial wait between mirrors, in sec.
        InitialMirrorWait (1,1) double

        %MinimumMirrorWait Smallest allowable wait between mirrors, in sec.
        MinimumMirrorWait (1,1) double

        %MaximumMirrorWait Largest allowable wait between mirrors, in sec.
        MaximumMirrorWait (1,1) double

        %MirrorBackoffFactor Factor to decrease/increase the next mirror
        %   wait by if any files are/are not mirrored respectively.
        MirrorBackoffFactor (1,1) double

        %KeepaliveInterval Time interval between sending keepalives over
        %   member SSH connections, in sec. Set to 0 to disable, otherwise
        %   must be at least 2.
        KeepaliveInterval (1,1) double

        %UseSSHAgent True if ssh-agent should be used when connecting to the remote host
        UseSSHAgent (1,1) logical

        %UseMultifactor True if we're using multifactor authentication when connecting to the remote host
        UseMultifactor (1,1) logical
    end

    properties (Access = private)
        AuthenticationInfo
    end

    methods
        function value = get.UseIdentityFile(obj)
            value = any(obj.AuthenticationMode == "IdentityFile");
        end

        function value = get.UseSSHAgent(obj)
            value = any(obj.AuthenticationMode == "Agent");
        end

        function value = get.UseMultifactor(obj)
            value = any(obj.AuthenticationMode == "Multifactor");
        end

        function value = get.IsFileMirrorSupported(obj)
            value = ~isempty(obj.JobStorageLocation);
        end

        function value = get.IsConnected(obj)
            value = ~isempty(obj.Hostname);
        end

        function value = get.FileArchiveType(obj)
            value = obj.getMirrorProperty('FileArchiveType');
        end

        function set.FileArchiveType(obj, value)
            arguments
                obj
                value (1,1) parallel.internal.remoteaccess.FileArchiveType
            end
            obj.setMirrorProperty('FileArchiveType', value);
        end

        function value = get.InitialMirrorWait(obj)
            value = obj.getMirrorProperty('InitialMirrorWait');
        end

        function set.InitialMirrorWait(obj, value)
            arguments
                obj
                value (1,1) double {mustBeNonnegative}
            end
            obj.setMirrorProperty('InitialMirrorWait', value);
        end

        function value = get.MinimumMirrorWait(obj)
            value = obj.getMirrorProperty('MinimumMirrorWait');
        end

        function set.MinimumMirrorWait(obj, value)
            arguments
                obj
                value (1,1) double {mustBeNonnegative}
            end
            obj.setMirrorProperty('MinimumMirrorWait', value);
        end

        function value = get.MaximumMirrorWait(obj)
            value = obj.getMirrorProperty('MaximumMirrorWait');
        end

        function set.MaximumMirrorWait(obj, value)
            arguments
                obj
                value (1,1) double {mustBeNonnegative}
            end
            obj.setMirrorProperty('MaximumMirrorWait', value);
        end

        function value = get.MirrorBackoffFactor(obj)
            value = obj.getMirrorProperty('MirrorBackoffFactor');
        end

        function set.MirrorBackoffFactor(obj, value)
            arguments
                obj
                value (1,1) double {mustBeGreaterThanOrEqual(value, 1)}
            end
            obj.setMirrorProperty('MirrorBackoffFactor', value);
        end

        function value = get.KeepaliveInterval(obj)
            obj.errorIfNotConnected();
            value = obj.SSHClient.KeepaliveInterval;
        end

        function set.KeepaliveInterval(obj, value)
            arguments
                obj
                value (1,1) uint32 {mustBeKeepaliveInterval}
            end
            obj.errorIfNotConnected();
            obj.SSHClient.KeepaliveInterval = value;
            if ~isempty(obj.FileMirror) && obj.FileMirror ~= obj.SSHClient
                obj.FileMirror.KeepaliveInterval = value;
            end
        end
    end

    properties (Access = private)
        %SSHClient matlabshared.internal.sshaccess.SSHClient used to execute
        %   commands on the remote machine.
        SSHClient = []

        %FileMirror parallel.internal.remoteaccess.FileMirror used to
        %   upload job input files and copy back output files.
        FileMirror = []
    end

    properties (Constant, Access = private)
        % Prefix to add to all dctSchedulerMessages
        SchedulerMessagePrefix = 'RemoteClusterAccess:';

        % File separator on Windows
        WindowsSlash = '\';

        % File separator on Unix-based systems
        UnixSlash = '/';

        % A map of connected RemoteClusterAccess instances. Populated via
        % the static method 'getConnectedAccess'. When 'disconnect' is
        % called, objects should remove themselves from this map.
        ConnectedMap = containers.Map('KeyType', 'char', 'ValueType', 'any');
    end

    methods (Static, Access = private)
        function key = createKey(clusterHost, remoteJobStorageLocation, username)
            %createKey Return the key into the ConnectedMap for a given
            % combination of clusterHost, remoteJobStorageLocation and
            % username.
            key = sprintf('%s|%s|%s', clusterHost, remoteJobStorageLocation, username);
        end

        function [access, key] = getConnectedAccessFromCache(clusterHost, remoteJobStorageLocation, username)
            %getConnectedAccessFromCache
            % For this combination of clusterHost, remoteJobStorageLocation
            % and username, return a connected RemoteClusterAccess instance
            % if one already exists. Otherwise returns empty.

            import parallel.cluster.RemoteClusterAccess

            key = RemoteClusterAccess.createKey(clusterHost, remoteJobStorageLocation, username);

            if RemoteClusterAccess.ConnectedMap.isKey(key)
                access = RemoteClusterAccess.ConnectedMap(key);
                if isvalid(access) && access.IsConnected
                    return
                else
                    % Instances *should* remove themselves from the cache
                    % when they are disconnected or destroyed, but if we
                    % have found a disconnected or deleted instance we'll
                    % remove it now.
                    RemoteClusterAccess.ConnectedMap.remove(key);
                end
            end

            % If we get this far then there is not already a connected
            % RemoteClusterAccess instance in the cache.
            access = [];
        end

        function obj = getConnectedAccessImpl(clusterHost, remoteJobStorageLocation, username, varargin)
            %getConnectedAccessImpl Return a connected RemoteClusterAccess
            %   instance. If remoteJobStorageLocation is not empty,
            %   mirroring will be enabled.
            [clusterHost, remoteJobStorageLocation, username] = convertStringsToChars(clusterHost, remoteJobStorageLocation, username);

            import parallel.cluster.RemoteClusterAccess

            % Get a connected instance from the cache, if one already
            % exists
            [obj, key] = RemoteClusterAccess.getConnectedAccessFromCache(clusterHost, remoteJobStorageLocation, username);
            if ~isempty(obj)
                return
            end

            % Not in cache, need to build
            obj = RemoteClusterAccess(username, varargin{:});
            obj.connect(clusterHost, remoteJobStorageLocation);

            % Add ourselves to the cache
            obj.ConnectedMap(key) = obj;
        end
    end

    methods (Static, Access = public)
        function obj = getConnectedAccess(clusterHost, username, varargin)
            %getConnectedAccess Return a connected RemoteClusterAccess instance.
            %
            %   RemoteClusterAccess.getConnectedAccess(host,username)
            %   returns a RemoteClusterAccess instance that is connected to
            %   the supplied host. This function may return a previously
            %   constructed RemoteClusterAccess instance if one exists.
            %
            %   RemoteClusterAccess.getConnectedAccess(...,P1,V1,...PN,VN)
            %   passes the additional parameters to the RemoteClusterAccess
            %   constructor.
            %
            %   See also parallel.cluster.RemoteClusterAccess/connect.
            arguments
                clusterHost {iIsStr(clusterHost, 'clusterHost')}
                username {iIsStr(username, 'username')}
            end
            arguments (Repeating)
                varargin
            end

            import parallel.cluster.RemoteClusterAccess

            try
                obj = RemoteClusterAccess.getConnectedAccessImpl( ...
                    clusterHost, '', username, varargin{:});
            catch E
                throw(E);
            end
        end

        function obj = getConnectedAccessWithMirror(clusterHost, remoteJobStorageLocation, username, varargin)
            %getConnectedAccessWithMirror Return a connected RemoteClusterAccess instance
            %   RemoteClusterAccess.getConnectedAccessWithMirror(host,location,username)
            %   returns a RemoteClusterAccess instance that is connected to the
            %   supplied host, using the location as the mirror location. This function
            %   may return a previously constructed RemoteClusterAccess instance
            %   if one exists.
            %
            %   RemoteClusterAccess.getConnectedAccessWithMirror(...,P1,V1,...PN,VN)
            %   passes the additional parameters to the RemoteClusterAccess
            %   constructor.
            %
            %   See also parallel.cluster.RemoteClusterAccess/connect.
            arguments
                clusterHost {iIsStr(clusterHost, 'clusterHost')}
                remoteJobStorageLocation {iIsStr(remoteJobStorageLocation, 'remoteJobStorageLocation')}
                username {iIsStr(username, 'username')}
            end
            arguments (Repeating)
                varargin
            end

            import parallel.cluster.RemoteClusterAccess

            try
                obj = RemoteClusterAccess.getConnectedAccessImpl( ...
                    clusterHost, remoteJobStorageLocation, username, varargin{:});
            catch E
                throw(E);
            end
        end
    end

    methods
        function obj = RemoteClusterAccess(username, PVPairs)
            arguments
                username {iIsStr(username, 'username')}
                PVPairs.AuthenticationMode {mustBeText, ...
                    mustBeMember(PVPairs.AuthenticationMode, ["Password", "IdentityFile", "Agent", "Multifactor"])} = {}
                PVPairs.IdentityFilename {mustBeText, iCheckIdentityFile} = {}
                PVPairs.IdentityFileHasPassphrase (1,1) logical {mustBeNumericOrLogical} = false
                PVPairs.Port (1,1) double {mustBeInteger, mustBePositive} = 22
            end

            % Convert to string row vectors
            authModes = convertCharsToStrings(PVPairs.AuthenticationMode(:).');
            idFiles = convertCharsToStrings(PVPairs.IdentityFilename(:).');

            % If authentication mode unspecified, choose based on other inputs
            if isempty(authModes)
                if isempty(PVPairs.IdentityFilename)
                    authModes = "Password";
                else
                    authModes = "IdentityFile";
                end
            end

            % Verify we have as many identity files as IdentityFile steps
            numIdentityFilesNeeded = nnz("IdentityFile" == authModes);
            if numIdentityFilesNeeded > 0
                numIdentityFilesProvided = length(idFiles);
                if numIdentityFilesProvided < numIdentityFilesNeeded
                    error(message('parallel:cluster:RemoteClusterAccessNotEnoughIdentityFiles', ...
                        numIdentityFilesNeeded, numIdentityFilesProvided));
                end
            else
                idFiles = "";
            end

            % Assign properties to object
            obj.AuthenticationMode = convertStringsToChars(authModes);
            obj.Username = username;
            obj.IdentityFilename = convertStringsToChars(idFiles);
            obj.IdentityFileHasPassphrase = PVPairs.IdentityFileHasPassphrase;
            obj.Port = PVPairs.Port;
        end

        function connect(obj, clusterHost, remoteJobStorageLocation)
            %connect Connect to a remote host
            %   connect(remoteClusterAccess, clusterHost) establishes a connection to the
            %   specified host using the user credential options supplied in the constructor.
            %   File Mirroring is not supported.
            %
            %   connect(remoteClusterAccess, clusterHost, remoteJobStorageLocation) establishes a
            %   connection to the specified host using the user credential options supplied
            %   in the constructor.  remoteJobStorageLocation identifies a folder on the clusterHost
            %   that is used for file mirroring.  The user credentials supplied in the
            %   constructor must have write access to this folder.
            %
            %   See also parallel.cluster.RemoteClusterAccess/disconnect,
            %            parallel.cluster.RemoteClusterAccess/startMirrorForJob,
            %            parallel.cluster.RemoteClusterAccess/runCommand.
            arguments
                obj (1,1)
                clusterHost {iIsStr(clusterHost, 'clusterHost')}
                remoteJobStorageLocation {iIsStr(remoteJobStorageLocation, 'remoteJobStorageLocation')} = ''
            end
            [clusterHost, remoteJobStorageLocation] = convertStringsToChars(clusterHost, remoteJobStorageLocation);

            % Error if already connected to a remote machine
            obj.errorIfConnected(clusterHost);

            obj.Hostname = clusterHost;
            obj.JobStorageLocation = remoteJobStorageLocation;

            % Populate authentication info object
            obj.AuthenticationInfo = obj.getAuthInfo();

            obj.createConnections();

            % Enable keepalives with a 1 minute interval
            obj.KeepaliveInterval = 60;
        end

        function disconnect(obj)
            %disconnect Disconnect from a remote host
            %
            %   RemoteClusterAccess.disconnect() disconnects the existing remote connection.
            %   The connect method must have already been called.
            %
            %   See also parallel.cluster.RemoteClusterAccess/connect
            arguments
                obj (1,1)
            end

            % Can only disconnect if we are connected!
            obj.errorIfNotConnected();

            % Remove ourselves from the cache, if we were added there
            key = obj.createKey(obj.Hostname, obj.JobStorageLocation, obj.Username);
            map = obj.ConnectedMap;
            if ~isempty(map) && isvalid(map) && map.isKey(key)
                map.remove(key);
            end

            % Destroy our SSHClient and FileMirror objects
            obj.FileMirror = [];
            obj.SSHClient = [];

            % Reset connection data
            obj.Hostname = '';
            obj.JobStorageLocation = '';
        end

        function delete(obj)
            if obj.IsConnected
                obj.disconnect();
            end
        end

        function startMirrorForJob(obj, job)
            %startMirrorForJob Start the mirroring of files
            %   startMirrorForJob(remoteClusterAccess, job) copies all the job files
            %   from the local JobStorageLocation to the remote JobStorageLocation,
            %   and starts mirroring files so that any changes to the files in the
            %   remote JobStorageLocation are copied back to the local
            %   JobStorageLocation.  The CONNECT method must have already been
            %   called.
            %
            %   See also parallel.cluster.RemoteClusterAccess/connect,
            %            parallel.cluster.RemoteClusterAccess/doLastMirrorForJob.
            arguments
                obj (1,1)
                job (1,1) parallel.Job
            end

            % Check we're connected, mirroring is supported and mirroring
            % has not been previously started for this job
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotStartMirroringNotSupported');
            obj.errorIfAlreadyBeingMirrored('parallel:cluster:RemoteClusterAccessCannotStartJobAlreadyMirrored', job.ID);

            % Gather job info into a MirrorRequest
            mirrorRequest = obj.createMirrorRequest(job);

            try
                obj.FileMirror.addToMirror(mirrorRequest);
            catch err
                % If anything went wrong, try to remove the remote files
                dctSchedulerMessage(6, '%s Removing files for job %d because of error when starting mirror', ...
                    obj.SchedulerMessagePrefix, job.ID);
                try
                    obj.FileMirror.removeFromMirrorNow(job.ID);
                catch removeErr
                    % Warn if we also managed to fail at removing the files
                    warning(message('parallel:cluster:RemoteClusterAccessFailedToRemovePossiblyDelete', job.ID, obj.JobStorageLocation, removeErr.message));
                end
                % Throw an error for failing to add the job to the mirror
                ex = MException(message('parallel:cluster:RemoteClusterAccessStartMirrorFailed', job.ID));
                ex = ex.addCause(err);
                throw(ex);
            end
        end

        function stopMirrorForJob(obj, job)
            %stopMirrorForJob Stop the mirroring of files for the supplied job
            %   stopMirrorForJob(remoteClusterAccess, job) immediately stops the
            %   mirroring of files from the remote JobStorageLocation to the local
            %   JobStorageLocation for the supplied job.  The startMirrorForJob or
            %   resumeMirrorForJob method must have already been called.  This
            %   cancels the running mirror and removes the files for the job from
            %   the remote location.  This is similar to doLastMirrorForJob except
            %   that stopMirrorForJob makes no attempt to ensure that the local job
            %   files are up-to-date.  For normal mirror stoppage, use
            %   doLastMirrorForJob.
            %
            %   See also parallel.cluster.RemoteClusterAccess/startMirrorForJob,
            %            parallel.cluster.RemoteClusterAccess/resumeMirrorForJob,
            %            parallel.cluster.RemoteClusterAccess/doLastMirrorForJob.
            arguments
                obj (1,1)
                job (1,1) parallel.Job
            end

            % Check we're connected, supporting mirroring and this job is
            % currently mirrored
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotStopMirroringNotSupported');
            obj.errorIfNotBeingMirrored('parallel:cluster:RemoteClusterAccessCannotStopJobNotMirrored', job.ID);

            dctSchedulerMessage(6, '%s Cancelling mirror and removing remote files for job %d.', ...
                obj.SchedulerMessagePrefix, job.ID);

            try
                obj.FileMirror.removeFromMirrorNow(job.ID);
            catch err
                % We are always able to remove the job from the mirror, but
                % we might fail to remove the remote files. It's OK to fail
                % to remove any nfs hidden files. For other files, warn.
                errMessages = splitlines(err.message);
                rmMessages = errMessages(startsWith(errMessages, "rm: "));
                nfsFilePattern = "/.nfs" + alphanumericsPattern + "': ";
                if ~all(contains(rmMessages, nfsFilePattern))
                    warning(message('parallel:cluster:RemoteClusterAccessFailedToRemovePossiblyDelete', job.ID, obj.JobStorageLocation, err.message));
                end
            end
        end

        function resumeMirrorForJob(obj, job)
            %resumeMirrorForJob Resume the mirroring of files for the supplied job
            %   resumeMirrorForJob(remoteClusterAccess, job) resumes the mirroring
            %   of files from the remote JobStorageLocation to the local
            %   JobStorageLocation for the supplied job.  This is similar to the
            %   startMirrorForJob method but does not first copy the files from the
            %   local JobStorageLocation to the remote JobStorageLocation.  The
            %   CONNECT method must have already been called.  This is useful if the
            %   original client MATLAB session has ended, and you are accessing the
            %   same files from a new client session.
            %
            %   See also parallel.cluster.RemoteClusterAccess/connect,
            %            parallel.cluster.RemoteClusterAccess/startMirrorForJob,
            %            parallel.cluster.RemoteClusterAccess/doLastMirrorForJob.
            arguments
                obj (1,1)
                job (1,1) parallel.Job
            end

            % Check we're connected, mirroring is supported and mirroring
            % has not been previously started for this job
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotResumeMirroringNotSupported');
            obj.errorIfAlreadyBeingMirrored('parallel:cluster:RemoteClusterAccessCannotResumeJobAlreadyMirrored', job.ID);

            % Gather job info into a MirrorRequest
            mirrorRequest = obj.createMirrorRequest(job, false);

            try
                obj.FileMirror.addToMirror(mirrorRequest);
            catch err
                % Don't attempt any cleanup at the remote end if resume
                % errored.  We should preserve the files on the remote end
                % until user asks us to stop the mirror.
                ex = MException(message('parallel:cluster:RemoteClusterAccessResumeMirrorFailed', job.ID));
                ex = ex.addCause(err);
                throw(ex);
            end
        end

        function doLastMirrorForJob(obj, job, lastMirrorDelay)
            %doLastMirrorForJob Perform a last mirror for the supplied job
            %   doLastMirrorForJob(remoteClusterAccess, job) performs a
            %   final copy of changed files from the remote
            %   JobStorageLocation to the local JobStorageLocation for the
            %   supplied job. Any running mirrors for the job then stop
            %   and, if the final mirror was successful, the job files are
            %   removed from the remote JobStorageLocation. This provides a
            %   graceful termination of the mirroring process.
            %
            %   The job does not have to be being mirrored to call this
            %   function. If the job is not currently mirrored,
            %   resumeMirrorForJob is called first to provide the final
            %   mirror.
            %
            %   See also parallel.cluster.RemoteClusterAccess/startMirrorForJob,
            %            parallel.cluster.RemoteClusterAccess/resumeMirrorForJob,
            %            parallel.cluster.RemoteClusterAccess/stopMirrorForJob.
            arguments
                obj (1,1)
                job (1,1) parallel.Job
                lastMirrorDelay (1,1) double {mustBeNonnegative} = 300 % seconds to continue mirroring files
            end

            % Check we're connected and supporting mirroring.
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotDoLastMirroringNotSupported');

            % It is OK to do the last mirror even if the job is not
            % currently being mirrored. This can happen if a job submitted
            % by a previous MATLAB session has finished and we need to do a
            % final download of its remote files. In this case, resume
            % mirroring for the job and then immediately remove it from the
            % mirror by setting it to expire after 0 seconds.
            if ~obj.FileMirror.isBeingMirrored(job.ID)
                obj.resumeMirrorForJob(job);
                lastMirrorDelay = 0;
            end

            try
                obj.FileMirror.removeFromMirrorAfter(job.ID, lastMirrorDelay);

                % Let any final files come over before returning. Also
                % gives time for the job to be immediately removed from the
                % mirror if we set the expiry to 0.
                obj.synchronizeNow();
            catch err
                ex = MException(message('parallel:cluster:RemoteClusterAccessDoLastMirrorFailed', job.ID));
                ex = ex.addCause(err);
                throw(ex);
            end
        end

        function isMirrored = isJobUsingConnection(obj, jobID)
            %isJobUsingConnection Query the connection to see if the job is being mirrored
            %
            %   RemoteClusterAccess.isJobUsingConnection(jobID) returns
            %   true if the job's files are currently being mirrored.
            %   Returns false if file mirroring has not been configured.
            %
            %   See also startMirrorForJob, doLastMirrorForJob
            arguments
                obj (1,1)
                jobID (1,1) double
            end

            % Must be connected
            obj.errorIfNotConnected();

            isMirrored = obj.IsFileMirrorSupported && obj.FileMirror.isBeingMirrored(jobID);
        end

        function remoteJobDirectory = getRemoteJobLocation(obj, jobID, remoteOS)
            %getRemoteJobLocation Get the remote location for the supplied job ID
            %   RemoteClusterAccess.getRemoteJobLocation(jobID, remoteOS) returns the
            %   full path to the remote job location for the supplied jobID.  Valid
            %   values of remoteOS are 'windows', 'pc' and 'unix'.
            arguments
                obj (1,1)
                jobID (1,1) double
                remoteOS (1,:) char {iIsOS}
            end

            % Must be connected and supporting mirroring
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotGetRemoteLocationMirroringNotSupported');

            % Select appropriate filesep for the remote OS
            if strcmpi(remoteOS, 'unix')
                remoteFileSeparator = obj.UnixSlash;
            else
                remoteFileSeparator = obj.WindowsSlash;
            end

            % Build up the remote directory using the remote OS's slashes.
            % We assume that the user provided the correctly slashed
            % JobStorageLocation from their cluster OS type in the first place.
            remoteJobDirectory = sprintf('%s%sJob%d', obj.JobStorageLocation, remoteFileSeparator, jobID);
        end

        function [cmdStatus, cmdOut] = runCommand(obj, commandToRun)
            % runCommand Run the supplied command on the remote host
            %
            %   [status, result] = RemoteClusterAccess.runCommand(command) runs the
            %   supplied command on the remote host and returns the resulting status
            %   and standard output.  The CONNECT method must have already been called.
            %
            %   See also CONNECT
            arguments
                obj (1,1)
                commandToRun {iIsStr(commandToRun, 'commandToRun')}
            end
            commandToRun = convertStringsToChars(commandToRun);

            % Must be connected
            obj.errorIfNotConnected();

            % Try running the command
            try
                if obj.AutoReconnect
                    try
                        result = obj.SSHClient.runCommand(commandToRun);
                    catch
                        obj.reconnect();
                        result = obj.SSHClient.runCommand(commandToRun);
                    end
                else
                    result = obj.SSHClient.runCommand(commandToRun);
                end
            catch err
                ex = MException(message('parallel:cluster:RemoteClusterAccessFailedToRunCommand', commandToRun, obj.Hostname));
                ex = ex.addCause(err);
                throw(ex);
            end

            % Running the command was successful, though the remote command
            % itself might not have been. Prepare the result for output.
            cmdStatus = double(result.exitcode);
            cmdOut = strip(result.stdout, 'right');
            cmdErr = strip(result.stderr, 'right');
            if ~isempty(cmdErr)
                cmdOut = sprintf('%s\n%s', cmdOut, cmdErr);
            end
        end

        function reconnect(obj)
            %reconnect Recreate underlying SSH connections

            % connect must have been called previously so we have the
            % connection and authentication info; the connections don't
            % have to be valid, as we will recreate them
            obj.errorIfNotConnected();

            % Clear any existing connection objects and create new ones
            obj.SSHClient = [];
            obj.FileMirror = [];
            obj.createConnections();
        end
    end

    methods (Hidden)
        function varargout = copyFileToRemote(obj, source, destination)
            %copyFileToRemote Copy file or folder to the remote host
            arguments
                obj (1,1)
                source {iIsStr(source, 'source')}
                destination {iIsStr(destination, 'destination')} = obj.JobStorageLocation
            end
            nargoutchk(0, 3);
            [varargout{1:nargout}] = obj.copyFileImpl("copyToRemote", source, destination);
        end

        function varargout = copyFileFromRemote(obj, source, destination)
            %copyFileFromRemote Copy file or folder from the remote host
            arguments
                obj (1,1)
                source {iIsStr(source, 'source')}
                destination {iIsStr(destination, 'destination')} = pwd
            end
            nargoutchk(0, 3);
            [varargout{1:nargout}] = obj.copyFileImpl("copyFromRemote", source, destination);
        end

        function remoteDelete(obj, filepath)
            %remoteDelete Delete file or folder on the remote host
            arguments
                obj (1,1)
            end
            arguments (Repeating)
                filepath {mustBeTextScalar}
            end

            % Must be connected and supporting mirroring
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotDeleteFilesMirroringNotSupported');

            for idx = 1:numel(filepath)
                try
                    obj.FileMirror.remoteDelete(filepath{idx});
                catch err
                    % If there's a problem, warn and continue
                    parallel.internal.warningNoBackTrace(err.identifier, strip(err.message));
                end
            end
        end

        function synchronizeNow(obj)
            %synchronizeNow Carry out an immediate synchronization of
            %   changed files.

            % Must be connected and supporting mirroring
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotSynchronizeMirroringNotSupported');

            % Trigger a sync and wait until it finishes.
            obj.FileMirror.syncNow();
        end

        function deleteRemoteJobStorageLocation(obj)
            %deleteRemoteJobStorageLocation Deletes the JobStorageLocation
            %   on the remote host. There must not be any jobs being
            %   mirrored.

            % Must be connected and supporting mirroring
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotSynchronizeMirroringNotSupported');

            % Attempt to delete the remote folder. This will throw if there
            % are any mirrored jobs.
            obj.FileMirror.deleteRemoteJobStorageLocation();
        end
    end

    methods (Static, Hidden)
        function count = cacheCount()
            %cacheCount Number of connected RemoteClusterAccess instances cached.
            count = double(parallel.cluster.RemoteClusterAccess.ConnectedMap.Count);
        end
    end

    methods (Access = private)
        function createConnections(obj)
            %createConnections Create the SSHClient and (if nonshared) FileMirror

            try
                % Only need a FileMirror if in nonshared submission mode
                if ~isempty(obj.JobStorageLocation)
                    try
                        obj.FileMirror = parallel.internal.remoteaccess.FileMirror( ...
                            obj.JobStorageLocation, obj.Hostname, obj.Port, obj.AuthenticationInfo);
                    catch cause
                        e = MException(message('parallel:cluster:remoteaccess:ConnectionError', obj.Hostname));
                        e = e.addCause(cause);
                        throw(e);
                    end

                    % Verify that the remote shell isn't writing to stdout
                    result = obj.FileMirror.runCommand('');
                    if ~isempty(result.stdout)
                        error(message('parallel:cluster:RemoteClusterAccessExtraBytesFromShell', result.stdout));
                    end

                    % If we're using keyboard-interactive authentication, we
                    % don't want to establish two connections to the cluster as
                    % the user will be prompted to authenticate twice. Instead,
                    % set our FileMirror object as the SSHClient to run
                    % commands through that.
                    if obj.UseMultifactor
                        obj.SSHClient = obj.FileMirror;
                    end
                end

                % Need an SSHClient for both remote and nonshared submission modes
                if isempty(obj.SSHClient)
                    try
                        obj.SSHClient = matlabshared.internal.sshaccess.SSHClient( ...
                            obj.Hostname, obj.Port, obj.AuthenticationInfo);
                    catch cause
                        e = MException(message('parallel:cluster:remoteaccess:ConnectionError', obj.Hostname));
                        e = e.addCause(cause);
                        throw(e);
                    end
                end
            catch err
                % If there's a problem, unset the connection config
                obj.Hostname = '';
                obj.JobStorageLocation = '';
                throwAsCaller(err)
            end
        end

        function errorIfConnected(obj, clusterHost)
            % Throws if this instance is already connected to a remote host.
            if obj.IsConnected
                ex = MException(message('parallel:cluster:RemoteClusterAccessAlreadyConnected', clusterHost, obj.Hostname));
                throwAsCaller(ex);
            end
        end

        function errorIfNotConnected(obj)
            % Throws if this instance is not connected to a remote host.
            if ~obj.IsConnected
                ex = MException(message('parallel:cluster:RemoteClusterAccessNotConnected'));
                throwAsCaller(ex);
            end
        end

        function errorIfMirroringNotSupported(obj, errID, varargin)
            % Throws with the provided error message ID if this instance
            % has not been set up to support file mirroring.
            if ~obj.IsFileMirrorSupported
                ex = MException(message(errID, varargin{:}));
                throwAsCaller(ex);
            end
        end

        function errorIfAlreadyBeingMirrored(obj, errID, jobID)
            % Throws with the provided error message ID if a job with the
            % provided ID is already being mirrored.
            if obj.FileMirror.isBeingMirrored(jobID)
                ex = MException(message(errID, jobID));
                throwAsCaller(ex);
            end
        end

        function errorIfNotBeingMirrored(obj, errID, jobID)
            % Throws with the provided error message ID if a job with the
            % provided ID is not currently being mirrored.
            if ~obj.FileMirror.isBeingMirrored(jobID)
                ex = MException(message(errID, jobID));
                throwAsCaller(ex);
            end
        end

        function authVec = getAuthInfo(obj)
            authVec = matlabshared.internal.sshaccess.SSHAuthInfoVector;
            numIdentityFiles = 0;
            authModes = string(obj.AuthenticationMode);
            idFiles = string(obj.IdentityFilename);
            for authMode = authModes
                switch authMode
                    case "Agent"
                        % ssh-agent
                        authInfo = matlabshared.internal.sshaccess.AgentAuth(obj.Username);
                    case "IdentityFile"
                        % Keypair authentication
                        numIdentityFiles = numIdentityFiles + 1;
                        authInfo = obj.getIdentityFileAuth(idFiles(numIdentityFiles));
                    case "Multifactor"
                        % keyboard-interactive, used for two-factor auth
                        authInfo = matlabshared.internal.sshaccess.KeyboardInteractiveAuth(obj.Username, 'parallel.internal.keyboardInteractiveCallback');
                    case "Password"
                        % Username/password authentication
                        authInfo = obj.getPasswordAuth();
                end
                authVec.add(authInfo);
            end
        end

        function auth = getIdentityFileAuth(obj, idFile)
            if obj.IdentityFileHasPassphrase
                passphraseTitle = getString(message('parallel:cluster:EnterPassphrase'));
                passphraseMsg = getString(message('parallel:cluster:EnterPassphraseForIdentityFile', idFile));
                keyfilePassphrase = matlabshared.internal.sshaccess.solicitPassword(passphraseTitle, passphraseMsg);
            else
                keyfilePassphrase = '';
            end
            auth = matlabshared.internal.sshaccess.IdentityFileAuth(obj.Username, idFile, keyfilePassphrase);
        end

        function auth = getPasswordAuth(obj)
            passwordTitle = getString(message('parallel:cluster:EnterPassword'));
            passwordMsg = getString(message('parallel:cluster:EnterPasswordForUserOnMachine', obj.Username, obj.Hostname));
            password = matlabshared.internal.sshaccess.solicitPassword(passwordTitle, passwordMsg);
            auth = matlabshared.internal.sshaccess.PasswordAuth(obj.Username, password);
        end

        function value = getMirrorProperty(obj, prop)
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotGetMirrorPropertyMirroringNotSupported', prop);
            value = obj.FileMirror.(prop);
        end

        function setMirrorProperty(obj, prop, value)
            obj.errorIfNotConnected();
            obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotSetMirrorPropertyMirroringNotSupported', prop);
            obj.FileMirror.(prop) = value;
        end

        function request = createMirrorRequest(obj, job, doUpload)
            %createMirrorRequest
            %   Creates a parallel.internal.remoteaccess.MirrorRequest for
            %   the provided job.
            arguments
                obj (1,1)
                job (1,1) parallel.Job
                doUpload (1,1) logical = true
            end

            jobFilesInfo = job.Parent.pGetJobFilesInfo(job);
            request = parallel.internal.remoteaccess.MirrorRequest();

            % Job ID, used to refer to this mirror request later
            request.ID = job.ID;

            % The local JobStorageLocation
            request.LocalStorageLocation = jobFilesInfo.RootDirectory;

            % Paths to upload from local to remote before job submission.
            % Only need to do this for new jobs - when resuming an existing
            % job, doUpload should be false.
            if doUpload
                uploadPaths = [jobFilesInfo.JobInputFiles; jobFilesInfo.TaskFiles];
                dctSchedulerMessage(6, '%s Files to upload for job %d:\n%s', ...
                    obj.SchedulerMessagePrefix, job.ID, sprintf('\t%s\n', uploadPaths{:}));
                request.UploadFilepaths = uploadPaths;
            end

            % Paths (excluding any state files) to mirror during the job
            dataFiles = [jobFilesInfo.JobOutputFiles; jobFilesInfo.TaskFiles];
            dctSchedulerMessage(6, '%s Data files to mirror for job %d:\n%s', ...
                obj.SchedulerMessagePrefix, job.ID, sprintf('\t%s\n', dataFiles{:}));
            request.MirroredFilepaths = dataFiles;

            % Paths to explicitly exclude from mirroring. Can't overwrite
            % readonly files and don't want to overwrite task input files
            % as they don't change on the remote end but can on local.
            readOnlyFiles = obj.getReadOnlyFiles( ...
                jobFilesInfo.RootDirectory, dataFiles );
            excludedFiles = strrep([jobFilesInfo.JobInputFiles; readOnlyFiles], '\', '/');
            excludedFiles = strrep(excludedFiles, '.', '\.');
            taskInputFilePattern = [sprintf('Job%d', job.ID) '/Task.*\.in\.mat'];
            excludedFiles = [excludedFiles; {taskInputFilePattern}];
            excludedFiles = join(excludedFiles, '|');
            request.ExcludedPattern = excludedFiles{1};

            % Whether to delete the remote copies of the files at the end
            % of the job
            request.RemoveRemoteFilesAtEnd = ~obj.PreserveRemoteFiles;
        end

        function varargout = copyFileImpl(obj, fcn, source, destination)
            try
                % Must be connected and supporting mirroring
                obj.errorIfNotConnected();
                obj.errorIfMirroringNotSupported('parallel:cluster:RemoteClusterAccessCannotCopyFilesMirroringNotSupported');

                obj.FileMirror.(fcn)(source, destination);

                if nargout > 0
                    varargout = {true, '', ''};
                end
            catch err
                if nargout == 0
                    err.rethrow;
                else
                    varargout = {false, err.message, err.identifier};
                end
            end
        end
    end

    methods (Static, Access = private)
        function readOnlyFiles = getReadOnlyFiles(dirPrefix, filenames)
            %getReadOnlyFiles Returns the readOnlyFiles relative the dirPrefix.
            readOnlyFiles = {};
            for ii = 1:numel(filenames)
                currFile = filenames{ii};
                currFullFile = fullfile(dirPrefix, currFile);
                if isfolder(currFullFile)
                    listing = dir(currFullFile);
                    filesInDir = {listing.name};
                    % Remove '.' and '..' from the listing
                    filesInDir = filesInDir(~strcmp('.', filesInDir));
                    filesInDir = filesInDir(~strcmp('..', filesInDir));
                    % And recursively call this method to get the read only files
                    readOnlyInDir = parallel.cluster.RemoteClusterAccess.getReadOnlyFiles(currFullFile, filesInDir);
                    % Make sure we put the current relative dir back onto the list of files
                    if ~isempty(readOnlyInDir)
                        readOnlyFiles = [readOnlyFiles; strcat(currFile, filesep, readOnlyInDir)]; %#ok<AGROW>
                    end
                else
                    [~, attribs] = fileattrib(currFullFile);
                    if isstruct(attribs) && ~attribs.UserWrite
                        readOnlyFiles = [readOnlyFiles; currFile];  %#ok<AGROW>
                    end
                end
            end
        end
    end
end

function iIsStr(input, propname)
% Validates input is "a string" or 'char array'. Avoids the arguments block
% auto-converting inappropriate datatypes, like ints, into chars/strings.
if ~ischar(input) && ~isStringScalar(input)
    error(message('parallel:cluster:RemoteClusterAccessArgumentNotString', propname));
end
end

function iCheckIdentityFile(idFiles)
% Verify each identity file exists.
idFiles = convertCharsToStrings(idFiles(:).');
for filename = idFiles
    if ~isfile(filename)
        error(message('parallel:cluster:RemoteClusterAccessNonExistentIdentityFile', filename));
    end
    if isunix
        % On Unix, check the file permissions look sensible. If the private
        % key is visible to other users, the server may reject it.
        [~, perms] = fileattrib(filename);
        if perms.OtherRead || perms.OtherWrite || perms.OtherExecute ...
                || perms.GroupRead || perms.GroupWrite || perms.GroupExecute
            warning(message('parallel:cluster:RemoteClusterAccessWrongIdentityFilePermsOnLinux', filename));
        end
    end
end
end

function iIsOS(OS)
allowedOS = ["unix", "pc", "windows"];
if ~ismember(OS, allowedOS)
    error(message('parallel:cluster:RemoteClusterAccessIncorrectRemoteOS', join(allowedOS, ', ')));
end
end

function mustBeKeepaliveInterval(val)
if val == 1
    error(message('parallel:cluster:RemoteClusterAccessInvalidKeepaliveInterval'));
end
end
