% Implementation of AbstractRemoteSpmdController backed by MCOS
% class parallel.internal.pool.SPMDBlockController.

% Copyright 2020 The MathWorks, Inc.

classdef RemoteSpmdControllerImpl < parallel.internal.pool.AbstractRemoteSpmdController

    properties (Constant, Hidden)
        InvalidProcessIndex = -2;
    end
    
    properties (SetAccess = immutable, GetAccess = private)
        % MCOS implementation of SPMDBlockController
        MCOSSpmdBlockController
    end
    
    properties (Access = private)
        % Cache the following results as fetching them moves them out of
        % the controller.
        BodyResult;
        CleanupResult;
    end
    
    methods (Access = private)
        function obj = RemoteSpmdControllerImpl(SpmdBlockController)
            obj.MCOSSpmdBlockController = SpmdBlockController;
            obj.BodyResult = containers.Map('KeyType', 'int32', 'ValueType', 'any');
            obj.CleanupResult = containers.Map('KeyType', 'int32', 'ValueType', 'any');
        end
        
        function result = getBodyResult(obj, processIndex)
            if ~isKey(obj.BodyResult, processIndex)
                obj.BodyResult(processIndex) = obj.MCOSSpmdBlockController.getBlockBodyResult(int32(processIndex));
            end
            result = obj.BodyResult(processIndex);
        end
    end
    
    methods (Static)
        function obj = create(session, processIndices, compositeKeysToClearMap)
            % Create spf plr backed SpmdBlockController and acquire
            % the given processes. These process will be released on delete.
            % session - parallel.internal.pool.SpfClientSession
            % processIndices - array of worker indices to acquire
            % compositeKeysToClearMap - instance of parallel.internal.pool.ProcessKeyMap
            try
                controller = parallel.internal.pool.SPMDBlockController.createForSession(...
                    session, int32(processIndices), compositeKeysToClearMap);
                obj = parallel.internal.pool.RemoteSpmdControllerImpl(controller);
            catch E
                except = MException(message('parallel:lang:spmd:SpmdCreationError'));
                except = addCause(except, E);
                throw(except);
            end
        end
    end
    
    % Implement AbstractRemoteSpmdController methods
    methods
        function initiateRemoteSpmdBlock(obj, remoteCommsCell, remoteFcnHandles)
            obj.MCOSSpmdBlockController.initiateRemoteSpmdBlock(remoteCommsCell, remoteFcnHandles);
        end
        
        function completed = awaitBlock(obj, timeoutMilliseconds)
            completed = obj.MCOSSpmdBlockController.awaitBlockBody(timeoutMilliseconds);
        end
        
        function completed = awaitCleanup(obj, timeoutMilliseconds)
            completed = obj.MCOSSpmdBlockController.awaitBlockCleanup(timeoutMilliseconds);
        end
        
        function throwAnyBlockBodyError(obj, processIndex)
            result = obj.getBodyResult(processIndex);
            if result.Exception.IsError
                ex = parallel.internal.pool.handleRemoteException(result.Exception);
                dctSchedulerMessage(0, 'Failed to get result from SPMD block:', ex);
            else
                if ~isempty(result.Result) && ~result.MatlabOKFlag
                    throwAsCaller(result.Result);
                end
            end
        end
        
        function matlabResult = getCleanupResult(obj, processIndex)
            if ~isKey(obj.CleanupResult, processIndex)
                obj.CleanupResult(processIndex) = obj.MCOSSpmdBlockController.getBlockCleanupResult(int32(processIndex));
            end
            result = obj.CleanupResult(processIndex);
            if isempty(result)
                % This is really serious - there was no remoteBlockCleanup return from the
                % process, which may mean that the parallel pool is stuck.
                throw(MException(message('parallel:lang:spmd:NoRemoteCleanup', processIndex)));
            end
            
            matlabResult = result.Result;
            if ~isempty(matlabResult)
                if ~result.MatlabOKFlag
                    % Throw MATLAB error
                    throwAsCaller(matlabResult);
                end
                % Successful MATLAB result
                return
            end
            
            ex = parallel.internal.pool.handleRemoteException(result.Exception);
            if ~isempty(ex)
                dctSchedulerMessage(0, 'Failed to get result from SPMD cleanup:', ex);
                throw(ex);
            end
        end
        
        function processIndex = getProcessCausingInterrupt(obj)
            processIndex = double(obj.MCOSSpmdBlockController.getProcessCausingInterrupt());
        end
        
        function throwInterruptedProcessException(obj)
            processIndex = obj.getProcessCausingInterrupt();
            if processIndex ~= obj.InvalidProcessIndex
                result = obj.getBodyResult(processIndex);
                matlabResult = result.Result;
                if ~isempty(matlabResult)
                    if ~result.MatlabOKFlag
                        % Throw MATLAB error
                        throwAsCaller(matlabResult);
                    end
                end

                ex = parallel.internal.pool.handleRemoteException(result.Exception);
                if ~isempty(ex)
                    dctSchedulerMessage(0, 'Exception causing spmd interrupt:', ex);
                    throw(ex);
                end
            end
        end
        
        function drainIO(obj, lastcall)
            obj.MCOSSpmdBlockController.flushIO(lastcall);
        end
        
        function interrupt(obj)
            obj.MCOSSpmdBlockController.interrupt();
        end

        function dispose(obj)
            obj.MCOSSpmdBlockController.dispose();
        end
    end
end


