% RemoteSpmdExecutor - control class to manage remote execution of an SPMD
% block This class manages the M side of dealing with the controller,
% and in particular attempts to deal correctly with M-side interrupts.

% Copyright 2008-2020 The MathWorks, Inc.

classdef RemoteSpmdExecutor < spmdlang.AbstractSpmdExecutor

    properties (Access = private, Hidden, Transient)
        % The remote controller
        RemoteSpmdController;
        % Number of remote processes
        NumLabs;
        % A cell array of communicators for those processes
        RemoteCommCell;
        % The session object
        Session;
        % Flag to indicate whether we have sent the block for execution
        SentBlock = false;
        % Stash how many output arguments to expect (can't access superclass data in destructor)
        SpmdNargout = 0;

        % Have we warned about the interrupt yet?
        WarnedAboutError = false;
        % Time at which we first spotted there was an error - we don't warn for a while
        TimeErrorFirstSeen = [];
    end
    
    properties (GetAccess = private, Constant)
        % This is the timeout after noticing that a process has encountered an error
        % (and therefore sending an interrupt) before we warn the users.
        TIMEOUT_BEFORE_WARNING_ABOUT_ERROR = 5; % seconds
        
        COMPOSITE_WARNING_CLASSES = {'distributed', 'Composite'};
    end

    methods (Access = private, Hidden)
        function errorIfNoSession(rse)
        % Method used to bail out of various situations if we discover that the
        % session has shut down.
            if ~rse.Session.isSessionRunning()
                except = ParallelException.hBuildLikeMException(message('parallel:lang:spmd:SessionNotRunning'));
                if rse.SentBlock
                    % Tell the RemoteSpmdController that we're out of here. This
                    % is required so that we can extract information about the
                    % block results, even though the block is not necessarily
                    % complete.
                    rse.RemoteSpmdController.dispose();
                    rse.Disposed = true;
                    excepts      = rse.makeBlockExceptionCauses();
                    if ~isempty(excepts)
                        if numel(excepts) == 1
                            except = ParallelException.hCloneWithNewMessage(excepts{1}, ...
                                'parallel:lang:spmd:SessionNotRunningSingleWorker', excepts{1}.message);
                        else
                            for ii = 1:length(excepts)
                                except = except.addCause(excepts{ii});
                            end
                        end
                    end
                end
                throw(except);
            end
        end
    end

    methods

        function obj = RemoteSpmdExecutor(resSetH, bodyF, assignOutF, getOutF, unpackInF, initialOuts, session)

            % assert that we've got a remote resource set
            resSet = resSetH.getResourceSet();
            if ~isa(resSet, 'spmdlang.RemoteResourceSet')
                error(message('parallel:lang:spmd:RemoteInconsistency'));
            end

            % Build the parent class - manages the closures/
            obj = obj@spmdlang.AbstractSpmdExecutor(resSetH, bodyF, assignOutF, ....
                                                     getOutF, unpackInF, initialOuts);

            % Build the controller which will actually do the work
            obj.RemoteSpmdController = buildController(resSet.numlabs, session);
            obj.NumLabs              = resSet.numlabs;
            obj.RemoteCommCell       = resSet.RemoteCommCell;

            % stash the session
            obj.Session              = session;
            obj.SpmdNargout          = length(initialOuts);
        end


        function initiateComputation(obj)
            % Make sure we can detect save/load of Remote objects
            [listener, getListFcn] = ...
                parallel.internal.general.SerializationNotifier.createAccumulatingListener(); %#ok<ASGLU>

            fcns = { obj.BodyFcn, obj.GetOutFcn, obj.UnpackInFcn };
            
            warnStates = spmdlang.AbstractSpmdExecutor.disableRemoteSaveWarnings();
            try
                % Actually initiate the remote computation. This will also
                % do any required serialization
                obj.RemoteSpmdController.initiateRemoteSpmdBlock(obj.RemoteCommCell, fcns);
            catch E
                spmdlang.AbstractSpmdExecutor.restoreRemoteSaveWarnings(warnStates);
                rethrow(E);
            end
            spmdlang.AbstractSpmdExecutor.restoreRemoteSaveWarnings(warnStates);
            
            % Was the serialization OK?
            serialisationClasses = getListFcn();
            if ~isempty(serialisationClasses)
                if any(ismember(serialisationClasses, obj.COMPOSITE_WARNING_CLASSES))
                    pOneHiddenCompositeWarning;
                end
            end

            % Remember that we did actually send stuff - see delete()
            obj.SentBlock = true;
        end

        function tf = isComputationComplete(obj)
            obj.errorIfNoSession();

            % Simply query with a shortish timeout
            tf = obj.RemoteSpmdController.awaitBlock(200);

            % Take the opportunity to drain IO here - "false" because we expect to call
            % this again.
            obj.RemoteSpmdController.drainIO(false);

            obj.maybeWarnIfInterruptedAndWaiting();
        end

        function throwBlockExceptions(obj)
            % Look through what we received from executing the block, see if there are
            % any errors to throw.

            if ~obj.SentBlock
                % Can't possibly be in error.
                return;
            end

            % Retrieve the exceptions from the controller
            [excepts, exceptProcesses] = obj.makeBlockExceptionCauses();

            if ~isempty(excepts)
                processStr = sprintf('%d ', exceptProcesses);
                processStr = processStr(1:end-1);
                
                % Create generic SPMD error, add all excepts as cause and
                % as remote cause, and throw.
                except = pWrapExceptionsIntoParallelException(excepts, processStr);
                throw(except);
            end
        end

        function dispose(obj)
            
            % If we didn't even send the block, get out of here
            if ~obj.SentBlock
                if ~isempty(obj.RemoteSpmdController)
                    obj.RemoteSpmdController.dispose();
                end
                return;
            end

            if obj.Disposed
                return;
            end

            % Ensure we dispose of the controller however we leave this
            % delete() method.
            rsc = obj.RemoteSpmdController;
            x = onCleanup(@() dispose(rsc));

            % Set disposed to true - we don't need to go through here again.
            obj.Disposed = true;
            
            % First, we must ascertain whether or not we completed correctly.
            if ~obj.isComputationComplete()
                obj.errorIfNoSession();
                % The client must have hit CTRL-C or similar - send the
                % interrupt to the processes
                obj.RemoteSpmdController.interrupt();
                
                % carry on waiting - for the block + interrupts to complete. If the user
                % gets bored, they might keep hitting CTRL-C, but the
                % remote end will continue.
                while ~obj.isComputationComplete()
                    % Wait for the interrupt to take its course
                    obj.errorIfNoSession();
                end
            end

            % Normal completion, attempt to retrieve output arguments - long timeout, as
            % we're not doing anything really useful.
            while ~obj.RemoteSpmdController.awaitCleanup(1000)
                % Wait a while
                obj.errorIfNoSession();
            end

            % Retrieve the deserialized parcels
            [deserParcels, E] = obj.extractReturnParcels();

            obj.RemoteSpmdController.drainIO(true);

            % Let the controller know that we're done with it.
            obj.callAssignOuts(deserParcels);

            % Now that we've overwritten the old values, flush any remotes - this makes
            % a remote call to remoteClear - but only on those processes that need it.
            reset = obj.ResourceSetHolder.getResourceSet();
            reset.forceClear();
            
            % Throw any errors relating to calling the cleanup method.
            if ~isempty(E)
                throw(E);
            end
        end
    end

    methods (Access = private)

        function [excepts, exceptProcesses] = makeBlockExceptionCauses(obj)
        % Look through what we received from executing the block, see if there are
        % any errors to throw.

        % Retrieve the exceptions from the controller
            nProcess   = obj.NumLabs;
            excepts    = cell(1, nProcess);
            gotExcept  = false(1, nProcess);

            for idx = 1:nProcess
                % Get any error thrown during remoteBlockBody
                try
                    obj.RemoteSpmdController.throwAnyBlockBodyError(idx);
                catch E
                    thisExcept = iBuildParallelException(E);
                    excepts{idx}   = obj.maybeTransformMissingSourceException(thisExcept);
                    gotExcept(idx) = true;
                end
            end
            excepts = excepts(gotExcept);
            exceptProcesses = find(gotExcept);
        end
        
        function maybeWarnIfInterruptedAndWaiting(obj)
        % If an error has occurred on the processes, warn the users if we've been waiting
        % a long time for the processes to come back.
            
            if obj.WarnedAboutError
                % Seen an error, already warned about it, get out of here
                return;
            end
            
            if isempty(obj.TimeErrorFirstSeen)
                % Haven't seen an error, let's check
                processID = obj.RemoteSpmdController.getProcessCausingInterrupt();
                if processID >= 0
                    obj.TimeErrorFirstSeen = tic;
                end
                % Can't possibly be time to warn, so get out of here
                return;
            end
            
            % Have seen an error - we must be in possession of a TimeErrorFirstSeen - was
            % it long enough ago for us to warn?
            if (toc(obj.TimeErrorFirstSeen) >= ...
                 spmdlang.RemoteSpmdExecutor.TIMEOUT_BEFORE_WARNING_ABOUT_ERROR) 
                
                % Ensure we don't come back through
                obj.WarnedAboutError = true;
                
                % Get the information from the SpmdController about the exception
                processID = obj.RemoteSpmdController.getProcessCausingInterrupt();
                if processID >= 0
                    try
                        obj.RemoteSpmdController.throwInterruptedProcessException();
                    catch E
                        except  = iBuildParallelException(E);
                        warning(message('parallel:lang:spmd:WaitingForBlockInterruption', double(processID), except.getReport()));
                    end
                end
            end
        end

        function E = maybeTransformMissingSourceException(obj, origE)
        % If the process threw a SourceCodeNotAvailableOnLab exception, transform that to
        % describe the source code that wasn't found.
            if strcmp(origE.identifier, 'parallel:lang:spmd:SourceCodeNotAvailableOnLab')
                E = ParallelException.hCloneWithNewMessage(origE, ...
                        'parallel:lang:spmd:SourceCodeNotAvailable', obj.getBodyFunctionFile());
                E = E.addCause(origE.remotecause{1});
            elseif strcmp(origE.identifier, 'MATLAB:UndefinedFunction') && ...
                ~isempty(origE.remotecause{1}.arguments)
                undefinedFunction = origE.remotecause{1}.arguments{1};
                E = ParallelException.hCloneWithNewMessage(origE, ...
                        'parallel:lang:spmd:UndefinedFunctionOnWorker', undefinedFunction, undefinedFunction);
                E = E.addCause(origE.remotecause{1});    
            else
                % No transformation
                E = origE;
            end
        end
        
        function [deserParcels, E] = extractReturnParcels(obj)
        % Process the results of the cleanup - return the parcels, and an error that
        % we think ought to be thrown.
            
            nProcesses = obj.NumLabs;
            E     = [];
            
            % Pre-allocate the return:
            deserParcels = cell(1, obj.SpmdNargout);
            for argout = 1:obj.SpmdNargout
                deserParcels{argout} = cell(1, nProcesses);
            end
            
            for idx = 1:nProcesses
                try
                    thisProcessReturn = obj.RemoteSpmdController.getCleanupResult(idx);
                    if ~iscell(thisProcessReturn) || isempty(thisProcessReturn)
                        % If thisProcessReturn is completely empty - this means that the process didn't
                        % execute any of the lines needed to produce
                        % output. If it's not even a cell, then the
                        % remoteBlockBody>iStashOutputs didn't get called.
                    else
                        for argout = 1:obj.SpmdNargout
                            deserParcels{argout}{idx} = thisProcessReturn{argout};
                        end
                    end
                catch ex
                    E = obj.handleBadCleanupResult(E, ex);
                end
            end
        end

        function E = handleBadCleanupResult(~, E, ex)
        % Called during extractReturnParcels() to handle an errors
        % returned from a process which indicates problems
            if isempty(E)
                E = ParallelException.hBuildLikeMException(...
                    message('parallel:lang:spmd:RemoteCleanupErrored'));
            end
            E = iAddRemoteCause(E, ex);
        end

    end
end

function E = iAddRemoteCause(E, cause)
    E = addCause(E, cause);
    E = hAddRemoteCause(E, cause);
end

function ctrl = buildController(numProcesses, session)
    ctrl = session.createSpmdController(1:numProcesses);
end

function E = iBuildParallelException(origErr)
    clientStackToIgnore = 'spmd_feval';
    E = ParallelException.hBuildRemoteParallelException(...
        origErr, clientStackToIgnore);
end
