% Implementation of AbstractRemoteSpmdController backed by java classes

% Copyright 2020 The MathWorks, Inc.

classdef JavaRemoteSpmdController < parallel.internal.pool.AbstractRemoteSpmdController

    properties (SetAccess = immutable, GetAccess = private)
        % Java implmentation of RemoteSpmdController
        RemoteSpmdController
    end
    
    methods (Access = private)
        function obj = JavaRemoteSpmdController(remoteSpmdController)
            obj.RemoteSpmdController = remoteSpmdController;
        end
    end
    
    methods (Static)
        function controller = create(jctrl, processIndices)           
            try
                % If we successfully built the controller, must ensure
                % that it gets disposed if acquireProcesses fails.
                tmpCleanupCtrl = parallel.internal.general.DisarmableOncleanup(...
                    @() jctrl.dispose());
                controller = parallel.internal.pool.JavaRemoteSpmdController(jctrl);
                jctrl.acquireProcesses(processIndices);
            catch E
                % If this is a java exception that someone has provided a non-generic ID for
                if isJavaException(E) &&  ~ isequal(E.identifier , 'Matlab:Java:GenericException')
                    rethrow(E);
                else
                    except = MException(message('parallel:lang:spmd:SpmdCreationError'));
                    except = addCause(except, E);
                end

                throw(except);
            end
            tmpCleanupCtrl.disarm();
        end
    end
    
    % Implement AbstractRemoteSpmdController methods
    methods       
        function initiateRemoteSpmdBlock(obj, remoteCommsCell, remoteFcnCell)
            % Serialize communicator handles and function handles for transmission
            comms = parallel.internal.pool.serialize(remoteCommsCell);
            fcns  = parallel.internal.pool.serialize(remoteFcnCell);
            obj.RemoteSpmdController.initiateRemoteSpmdBlock(comms, fcns);
        end
        
        function completed = awaitBlock(obj, timeoutMilliseconds)
            import java.util.concurrent.TimeUnit;
            completed = obj.RemoteSpmdController.awaitBlock(timeoutMilliseconds, TimeUnit.MILLISECONDS);
        end
        
        function completed = awaitCleanup(obj, timeoutMilliseconds)
            import java.util.concurrent.TimeUnit;
            completed = obj.RemoteSpmdController.awaitCleanup(timeoutMilliseconds, TimeUnit.MILLISECONDS);
        end
        
        function throwAnyBlockBodyError(obj, processIndex)
            result = obj.RemoteSpmdController.getBlockResult(processIndex);
            
            if ~isempty(result) && result.isError()
                ex = result.getException();
                if ~isempty(ex)
                    [isMvmException] = ...
                        parallel.internal.general.transformIfMvmException(ex);
                    if isMvmException
                        % 'ex' is a Java exception, log it.
                        parallel.internal.general.logJavaException(...
                            0, 'Failed to get result from SPMD block.', ex);
                    else
                        thisExcept = distcompdeserialize(ex);
                        throwAsCaller(thisExcept);
                    end
                end
            end
        end
        
        function matlabValue = getCleanupResult(obj, processIndex)
            % Retrieve result corresponding to remoteBlockCleanup
            result = obj.RemoteSpmdController.getCleanupResult(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.getResult();
            if ~isempty(matlabResult)
                matlabError = ~logical(matlabResult(1));
                matlabValue =  distcompdeserialize(matlabResult(2));
                if matlabError
                    % Throw MATLAB error
                    throwAsCaller(matlabValue);
                end
                % Successful MATLAB result
                return
            end
            
            % Java exception
            javaException = result.getException();
            parallel.internal.general.logJavaException(...
                0, 'Failed to get result from SPMD cleanup.', javaException);    
            [isMvmException, parException] = ...
                parallel.internal.general.transformIfMvmException(javaException);
            if isMvmException
               throw(parException);
            end
        end
        
        function processIndex = getProcessCausingInterrupt(obj)
            processIndex = obj.RemoteSpmdController.getProcessCausingInterrupt();
        end
        
        function throwInterruptedProcessException(obj)
            msg = obj.RemoteSpmdController.getMessageCausingInterrupt();
            if ~isempty(msg) && msg.isError() && ~isempty(msg.getException())
                except = distcompdeserialize(msg.getException());
                throwAsCaller(except);
            end
        end
        
        function drainIO(obj, lastcall)
            obj.RemoteSpmdController.drainIO(lastcall);
        end
        
        function interrupt(obj)
            obj.RemoteSpmdController.interrupt();
        end
        
        function dispose(obj)
            obj.RemoteSpmdController.dispose();
        end
    end
end

