%AbstractInteractiveClient Client object for interactive parallel pool

% Copyright 2012-2021 The MathWorks, Inc.
classdef (Abstract) AbstractInteractiveClient < parallel.internal.pool.AbstractClient
    properties ( SetAccess = immutable )
        Tag
        Type
        UserName
    end

    methods (Static)
        function out = getSetJobCheckIntervalSeconds(i)
            persistent JobCheckIntervalSeconds;
            if isempty(JobCheckIntervalSeconds)
                JobCheckIntervalSeconds = 10;
            end
            if nargin
                JobCheckIntervalSeconds = i;
            end
            out = JobCheckIntervalSeconds;
        end
    end

    methods (Access = protected)
        function jobs = findJobsOnCluster(obj, clus)
            jobs = clus.findJob('ApiTag', obj.Tag, 'Username', obj.UserName);
        end
        function deleteJobs(~, sched, jobs, timeout)
            iDeleteJobs(sched, jobs, timeout);
        end
    end

    methods (Access = protected, Abstract)
        postStartActions(obj, session)
        postSessionDestroyedActions(obj)
        postShutdown(obj, stopFlag)
    end
    
    methods
        function obj = AbstractInteractiveClient(type, tag)
            import parallel.internal.cluster.ParpoolJobConstants
            
            obj.UserName = distcomp.pGetDefaultUsername();
            obj.JobStartupTimeout = Inf;
            obj.Type = type;
            obj.Tag = tag;
        end

        function running = isPossiblyRunning( obj )
            %isPossiblyRunning Check if an interactive session is currently running.
            %   Return true if obj's status indicates that we have an open connection to
            %   lab 1.
            %   Note that this method does not check whether the connection is intact and
            %   the labs are running and responsive.

            % Note that we must be very careful not to make this test strict because the
            % user cannot run 'pmode exit' when this test returns false.
            % Say we are running if the ParallelJob is non-empty
            running = ~isempty(obj.ParallelJob);

            if ~running
                % Check the last possibility: The session object may exist.
                session = obj.Session;
                running = ~isempty(session) && isvalid(session) && session.isSessionRunning;
            end
        end


        function spmdInitialized = start(obj, sessionBuildFcn, sessionInfo, nlabs, clus, filedependencies, spmdMode,...
                                         timeoutInMinutes, restartOnClusterChange, ...
                                         restartOnPreferredWorkerChange, ...
                                         autoAddClientPath, environmentVariables)
            %start Start interactive session
            %   Start interactive session, or detect the presence of an active session.  Perform
            %   all the setup and message display needed.

            if isempty(sessionInfo)
                sessionInfo = parallel.internal.pool.ISessionNotifier.getInstance().buildSessionInfo();
            end
            
            try
                enableSpmd = spmdMode;
                idleTimeoutInSeconds = timeoutInMinutes * 60;
                
                % If the idleTimeout is Inf, then make it intmax
                if ~isfinite(idleTimeoutInSeconds)
                    idleTimeoutInSeconds = intmax('int64');
                end
                sessionInfo = parallel.internal.pool.ISessionNotifier.getInstance().notifyStarting(...
                    sessionInfo, clus.Type, clus.Profile, idleTimeoutInSeconds, restartOnClusterChange, restartOnPreferredWorkerChange);

                % ANY error in this function will trigger this DisarmableOncleanup which calls
                % pStopLabsAndDisconnect - it is only at the very end of this function that we say
                % startup is complete and disarm.
                cleanupIfStartupFailed = parallel.internal.general.DisarmableOncleanup(...
                    @() iCleanupIfStartupFailed(obj));

                % Build the session wrapper, and get the connection function for the job.
                [session, connFcn] = sessionBuildFcn(clus);

                % Remove finished and failed jobs.  Warn about queued and pending jobs.
                try
                    obj.pRemoveOldJobs(clus);
                catch err
                    % Need to close the ConnectionManager
                    iThrowWithCause( 'parallel:convenience:FailedToDeleteOldJobs', err );
                end

                % At this point, we know that the configuration worked, so we can print messages
                % to the user.  Make sure we don't leave this function without printing the
                % final newline.
                cleanup = onCleanup(@() parallel.internal.display.parallelPrint(1, '\n'));

                % Create and store the parallel job.
                try
                    spmdInitialized = iCreateCommunicatingJob(obj, clus, nlabs, connFcn, ...
                                                              enableSpmd, autoAddClientPath);
                    if ~isempty(filedependencies)
                        % We *prepend* extra attached files
                        obj.ParallelJob.AttachedFiles = [filedependencies(:); ...
                                            obj.ParallelJob.AttachedFiles(:)];
                    end
                    if ~isempty(environmentVariables)
                        % But *append* EnvironmentVariables.
                        obj.ParallelJob.EnvironmentVariables = [obj.ParallelJob.EnvironmentVariables(:); ...
                            environmentVariables(:)];
                    end
                    submit(obj.ParallelJob);
                catch err
                    iThrowWithCause( 'parallel:convenience:FailedToStartInteractiveSession', ...
                        err, obj.Type );
                end

                % Ask the session wrapper to complete startup
                try
                    session.waitForConnections(spmdInitialized, sessionInfo, obj.buildCheckFcn());
                    % Once created we should loop waiting for all labs to finish starting
                    % up
                    while ~session.waitForSessionToStart(0.1)
                    end
                catch err
                    iThrowWithCause( 'parallel:convenience:FailedToInitializeInteractiveSession', err );
                end

                obj.Session = session;
                
                try
                    % Give the client a chance to perform any actions that might be required now
                    % that the underlying session has been set up.
                    obj.postStartActions(session);
                catch err
                    iThrowWithCause( 'parallel:convenience:FailedToInitializeInteractiveSession', err );
                end

                try
                    parallel.internal.display.parallelPrint(1, '%s', getString( message( ...
                        'parallel:convenience:ConnectedToWorkers', session.getPoolSize() ) ) );
                    cleanupIfStartupFailed.disarm();
                catch err %#ok<NASGU>
                end
            catch err
                if strcmp(err.identifier, 'parallel:cloud:ForcefullyStopped')
                    % If the cluster was forcefully stopped, just display the message.
                    errorReport = err.message;
                else
                    errorReport = err.getReport('extended','hyperlinks','off');
                end
                dctSchedulerMessage(0, 'Session failed to start when creating InteractiveClient. Error: %s', errorReport);
                parallel.internal.pool.ISessionNotifier.getInstance().notifyFailedToStart(...
                        sessionInfo, errorReport);
                rethrow(err)
            end
        end

        function stopLabsAndDisconnect(obj, stopFlag)
            %stopLabsAndDisconnect Stop all the labs and perform client cleanup
            %  Send the stop signal to all the labs so they exit and the parallel job
            %  finishes.  Clean up sockets and streams.  Closes the GUI unless closeGUI
            %  is false.
            %  stopFlag might be 'leaveguiopen'.


            % We don't need to stop if it isn't already running.
            if ~obj.isPossiblyRunning()
                return;
            end

            if nargin < 2
                stopFlag = '';
            end

            obj.pStopLabsAndDisconnect(stopFlag);
        end
    end
    methods (Hidden) % for testing
        function pRemoveOldJobs(obj, clus)
            %Remove old jobs that are either finished or failed.
            %   Also warn about jobs that are in the scheduler's queue.

            % This method is tested, hence Hidden rather than private
            [~, undoc] = pctconfig();
            if undoc.preservejobs
                return;
            end

            jobs = obj.findJobsOnCluster(clus);

            if isempty(jobs)
                return;
            end
            state = jobs.get({'State'});
            old = strcmp(state, parallel.internal.types.States.Finished.Name) | ...
                  strcmp(state, parallel.internal.types.States.Failed.Name);

            % When deleting old jobs, do not use the timeout.
            timeoutForRemovingOldJobs = 0;
            if any(old)
                iDeleteJobs( clus, jobs(old), timeoutForRemovingOldJobs );
            end
        end
    end
    methods (Access = private)
        function fcn = buildCheckFcn(obj)
        % Build a function that will be called during the session wrapper's "waitForConnections"
            startTime = tic;
            lastJobCheckTime = tic;
            function nDoCheck()
                if toc(startTime) > obj.JobStartupTimeout
                    iThrowTimeoutError(obj);
                end
                if toc(lastJobCheckTime) > obj.getSetJobCheckIntervalSeconds
                    dctSchedulerMessage(3, 'Checking communicating job status.');
                    iThrowIfBadParallelJobStatus(obj);
                    lastJobCheckTime = tic;
                end
            end
            fcn = @nDoCheck;
        end
        function pStopLabsAndDisconnect(obj, stopFlag)
            %stopLabsAndDisconnect Stop all the labs and perform client cleanup
            %  Send the stop signal to all the labs so they exit and the parallel job
            %  finishes.

            % This method does not throw any errors.

            constants = parallel.internal.pool.Constants();
            timeToWait = constants.clientTimeBetweenStopAndDestroyJob;

            session = obj.Session;
            canLabsStop = false;
            if ~isempty(session) && isvalid(session)
                % If we have a session, always ask it to destroyClientSession. We require that this
                % method always waits for the workers to complete if necessary.
                canLabsStop = session.destroyClientSession();
                % Finished with Session - forget about it, and make sure that it gets
                % cleaned up completely even if other classes (e.g. FileManager)
                % have references to the same handle.
                delete(obj.Session);
            end

            obj.postSessionDestroyedActions();

            if ~isempty(obj.ParallelJob)
                if canLabsStop
                    % Give the labs a few seconds to finish their labBarrier and exit before
                    % destroying the job.
                    try
                        obj.ParallelJob.wait(parallel.internal.types.States.Finished.Name, timeToWait);
                    catch err
                        dctSchedulerMessage(2, ['Failed to wait for job to finish due to the ', ...
                            'following error:\n%s'], err.message);
                    end
                end

                try
                    [~, undoc] = pctconfig();
                    if ~undoc.preservejobs
                        % Use the same timeout as between stop and wait.
                        iDeleteJobs( obj.ParallelJob.Parent, ...
                                     obj.ParallelJob, timeToWait );
                    else
                        cancel(obj.ParallelJob);
                    end
                catch err
                    dctSchedulerMessage(2, ['Failed to destroy the job due to the ', ...
                        'following error:\n%s'], err.message);
                end
            end
            % We have finished with this parallel job - so forget about it
            obj.ParallelJob = [];
            obj.postShutdown(stopFlag);
        end
    end

    methods ( Access = private )
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function iThrowWithCause(id, err, varargin)
if strcmp(err.identifier, 'parallel:cloud:ForcefullyStopped')
    throw(err);
end

throw( ...
    addCause( ...
    MException( id, '%s', ...
    getString( message( id, varargin{:} ) ) ), ...
    err ) );
end

% --------------------------------------------------------------------------
function logMsg = iGetDebugLog( pjob )
clus = pjob.Parent;
if ismethod(clus, 'getDebugLog')
    logMsg = sprintf('%s\n%s', ...
        getString( message( 'parallel:convenience:CommunicatingJobDebugLog' ) ), ...
        clus.getDebugLog(pjob));
else
    logMsg = getString( message( 'parallel:convenience:NoDebugLogAvailable', 'cluster' ) );
end
end


% --------------------------------------------------------------------------
function enableSpmd = iCreateCommunicatingJob( obj, sched, desiredNumlabs, connFcn, ...
                                               enableSpmd, autoAddClientPath )
% Create parallel job to attach to this client - note that configuration is
% first in the list so that what follows overrides it's values
if ~isempty(desiredNumlabs)
    numWorkersArgs = { 'NumWorkersRange', desiredNumlabs };
else
    numWorkersArgs = {};
end

% NB: we do not pass on the 'Profile' from the cluster as this is implicit in
% API-2.

% Create parallel job to attach to this client - note that configuration is
% first in the list so that what follows overrides its values

% Resolve what to do with enableSpmd in terms of cluster support. If cluster
% doesn't support concurrent jobs, always enable SPMD.
canUseConcurrent = sched.hIsVariantSupported( ...
    parallel.internal.types.Variant.ConcurrentJob );
enableSpmd = enableSpmd || ~canUseConcurrent;

if enableSpmd
    pjob = sched.createCommunicatingJob(...
        'Tag', obj.Tag, ...
        numWorkersArgs{:}, ...
        'Type', 'Pool', ...
        'ExecutionMode', parallel.internal.types.ExecutionMode.Interactive, ...
        'AutoAddClientPath', autoAddClientPath);
else
    pjob = sched.createConcurrentJob(...
        'Tag', obj.Tag, ...
        numWorkersArgs{:}, ...
        'AutoAddClientPath', autoAddClientPath);
end

% 'ApiTag' is Hidden, and cannot be specified as a constructor argument.
pjob.ApiTag = obj.Tag;

pjob.hSetInteractive(true);
obj.ParallelJob = pjob;

% The task to run will connect back using the connection function below
pjob.createTask(@parallel.internal.pool.poolWorkerFcn, 0, connFcn);
end

%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
function iThrowIfBadParallelJobStatus(obj)
% Throw an error if the parallel job is not running.
jobState = iGetJobState(obj.ParallelJob);
if strcmpi(jobState, parallel.internal.types.States.Queued.Name)
    % Job is queued, so there is nothing to do here.
    return;
end
if strcmpi(jobState, parallel.internal.types.States.Running.Name)
    % If any of the tasks have already finished or failed, the pool won't
    % form, so we should error. We only need to do this check for
    % concurrent jobs (i.e. SpmdEnabled = auto/false), otherwise the job is
    % started with MPI, and a single worker failure will cause the entire
    % job to fail.
    if obj.ParallelJob.Variant == parallel.internal.types.Variant.ConcurrentJob
        taskStates = hGetProperty(obj.ParallelJob.Tasks, {'StateEnum'});
        taskStates = [taskStates{:}];
        hasFailedTasks = any(taskStates == parallel.internal.types.States.Failed);
        hasFinishedTasks = any(taskStates == parallel.internal.types.States.Finished);

        if hasFinishedTasks || hasFailedTasks
            msg = iGetErrorMessage(obj.ParallelJob);
            if ~isempty(msg)
                error(message('parallel:convenience:InteractiveJobFailedWithMessage', msg))
            else
                error(message('parallel:convenience:InteractiveJobFailedNoMessage'));
            end
        end
    end
    
    % No errors found, return.
    return;
end
dctSchedulerMessage(1, iGetJobStatusDescription(obj.ParallelJob));
if strcmpi(jobState, parallel.internal.types.States.Destroyed.Name)
    error( message( 'parallel:convenience:InteractiveJobWasDeleted' ) );
end
if strcmpi(jobState, parallel.internal.types.States.Unavailable.Name)
    error( message( 'parallel:convenience:InteractiveJobUnavailable' ) );
end
% We now know that the job has completed and that we have a valid job object.
msg = iGetErrorMessage(obj.ParallelJob);

try
    failed = strcmpi(jobState, parallel.internal.types.States.Failed.Name);
catch err %#ok<NASGU>
    failed = false;
end
if ~isempty(msg)
    error( message( 'parallel:convenience:InteractiveJobFailedWithMessage', msg ) );
end
if failed
    error( message( 'parallel:convenience:InteractiveJobFailedNoMessage' ) );
else
    error( message( 'parallel:convenience:InteractiveJobFinishedNoMessage' ) );
end
end
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
function iThrowTimeoutError(obj)
% Throw an error due to timeout exceeded.
try
    running = strcmpi(obj.ParallelJob.State, parallel.internal.types.States.Running.Name);
catch err %#ok<NASGU>
    running = false;
end
if running
    error( message( 'parallel:convenience:ConnectionTimeoutExceeded', ...
        obj.JobStartupTimeout ) );
else
    error( message( 'parallel:convenience:JobStartTimeoutExceeded', ...
        obj.JobStartupTimeout ) );
end
end
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
function msg = iGetErrorMessage(pjob)
% Return the error message of the task causing the job to fail.
% Never throw an error.
try
    msgs = get(pjob.Tasks, {'ErrorMessage'}); % Always returns a cell array.
    % The task that caused the job to failed is the one that should
    % not start with the string secondaryFailureMsg.  We also exclude empty
    % error messages.
    secondaryFailureMsg = 'The parallel job was cancelled because the task';
    realCause = cellfun(@isempty, regexp(msgs, secondaryFailureMsg)) ...
        & ~cellfun(@isempty, msgs);
    % If all is well, realCause is not empty.
    if any(realCause)
        msg = msgs{find(realCause, 1, 'first')};
    else
        % Our analysis failed, so just use the first non-empty error message
        % instead.
        msg = msgs{find(~cellfun(@isempty, msgs), 1, 'first')};
    end
catch err %#ok<NASGU>
    msg = '';
end
end
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
function jobState = iGetJobState(pjob)
try
    jobState = pjob.State;
catch err %#ok<NASGU>
    jobState = parallel.internal.types.States.Unavailable.Name;
end
end
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Delete jobs and wait for their completion, if necessary.
function iDeleteJobs(cluster, jobs, timeoutInSeconds)
    preserved = false(1, numel(jobs));
    for idx = 1:numel(jobs)
        job = jobs(idx);
        if job.hWorkerCrashedDuringExecution()
            preserved(idx) = true;
        else
            cluster.hDeleteOneJob(job, timeoutInSeconds);
        end
    end
    if any(preserved)
        preservedID = [jobs(preserved).ID];
        fprintf('%s\n', getString(message('parallel:convenience:PoolPreservingCrashedJobs', strjoin(string(preservedID)))));
        profileName = cluster.Profile;
        if isempty(profileName)
            fprintf('%s\n', getString(message('parallel:convenience:PoolNoProfileCleanup')));
        else
            fprintf('%s\n', getString(message('parallel:convenience:PoolProfileCleanup', profileName, profileName)));
        end
    end
end

%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
function msg = iGetJobStatusDescription(pjob)

try
    state = pjob.State;
    stateMsg = getString( message( 'parallel:convenience:JobInState', state ) );
catch err %#ok<NASGU>
    stateMsg = getString( message( 'parallel:convenience:CouldNotGetJobState' ) );
end
try
    msgs = get(pjob.Tasks, {'ErrorMessage'});
    ind = ~cellfun(@isempty, msgs);

    if any(ind)
        % Create a matrix containing alternating the ids and messages.
        ids = get( pjob.Tasks(ind), {'Id'} );
        msgs = msgs(ind);
        errorMsgsCell = cellfun( @(taskID, errMsg) ...
            getString( message( 'parallel:convenience:ErrorMessageForTask', ...
            taskID, errMsg ) ), ...
            ids, msgs, 'UniformOutput', false );
        errorMsgs = sprintf( '%s\n', errorMsgsCell{:} );
    else
        errorMsgs = getString( message( 'parallel:convenience:NoTaskErrorMessages' ) );
    end
catch err %#ok<NASGU>
    errorMsgs = getString( message( 'parallel:convenience:CouldNotAccessTaskErrorMessage' ) );
end

try
    logMsg = iGetDebugLog( pjob );
catch err %#ok<NASGU>
    logMsg = getString( message( 'parallel:convenience:CouldNotRetrieveDebugLog' ) );
end

msg = sprintf('%s\n', stateMsg, errorMsgs, logMsg);
end
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
function iCleanupIfStartupFailed(obj)
% Run pre-cancel hook if it exists
preCancelHook = getenv( 'MDCE_QE_PRE_CLIENT_CANCEL_FCN' );
if ~isempty( preCancelHook )
    try
        dctSchedulerMessage( 2, ['Pre-cancel hook is defined : About ', ...
                            'to eval string: %s'], preCancelHook );
        eval( preCancelHook );
    catch E
        dctSchedulerMessage( 2, 'Hook threw an error. Report:', E);
    end
end

obj.pStopLabsAndDisconnect('');
end
