% Remote resource set object - gateway to remote SPMD execution. Manages a
% cell array of MPI communicator handles, and a corresponding array of
% process objects and can build the appropriate SPMD block executor


% Copyright 2008-2016 The MathWorks, Inc.

classdef RemoteResourceSet < spmdlang.AbstractResourceSet

    properties ( SetAccess = private, Transient )
        % Cell array of remote communicator handles. (Might be character vectors or uint64s).
        RemoteCommCell;
        
        % Java array of process instances
        RemoteProcessArray;
    end
    
    properties ( Access = private, Transient )
        % Am I the current "world"
        IsWorld = false;
        
        % Parallel pool
        Pool = parallel.Pool.empty();
    end

    methods ( Access = public, Static )

        % This function exists mainly to support parallel pool jobs, where the effective
        % world isn't 'world'. 
        function x = getWorldCommsCell( nlabs )
            [wc, labIdFirst] = spmdlang.commForWorld( 'get' );
            
            % Need to pad the cell array so that the labs can index into it by their
            % "World" labindex.
            padCells = cell( 1, labIdFirst - 1 );
            
            if ischar( wc ) % probably 'world'
                x = repmat( {wc}, 1, nlabs );
            else
                x = wc;
            end
            x = [padCells, x]; % in parallel pool job, this is e.g.: {[], 6, 6, 6}
        end
        
        % Bootstrap - this static method builds the "world" set from the pool, and
        % requires no communication or remote execution to achieve that.
        function worldSet = buildWorldSet(pool)
            if isempty(pool)
                error(message('parallel:lang:spmd:NoSession'));
            end
            ps = pool.NumWorkers;
            worldSet = spmdlang.RemoteResourceSet( ...
                1:ps, spmdlang.RemoteResourceSet.getWorldCommsCell( ps ), pool );
            
            worldSet.IsWorld = true;
        end
    
    end
    
    methods ( Access = public, Hidden )
        function obj = RemoteResourceSet( processIndices, remoteCommCell, pool )
            obj = obj@spmdlang.AbstractResourceSet( length( processIndices ) );
            obj.RemoteCommCell = remoteCommCell;
            
            % Build the process instance array from the "labindex" vector.
            import com.mathworks.toolbox.distcomp.pmode.poolmessaging.ProcessInstance;
            allLabs = ProcessInstance.getAllLabs( max( processIndices ) );
            obj.RemoteProcessArray = allLabs( processIndices );
            obj.Pool = pool;
        end
        
        % "Factory method" to build a remote resource set of a specific size from the
        % open pool. It does that by splitting the base set. 
        % "sz" is simply the required size of the subset
        function subSetObj = buildRemoteResourceSet( obj, sz )
            
            if sz == obj.numlabs
                subSetObj = obj;
            elseif sz < obj.numlabs

                % Build from the base set, 
                newComms = spmd_feval_fcn( @iBuildCommHandle, { sz }, obj );

                % Pad as necessary to handle mismatch between MPI 'world' labindex and
                % processes. (Happens in parallel pool jobs)
                [~, labIdFirst] = spmdlang.commForWorld( 'get' );
                padCells = cell( 1, labIdFirst - 1 );
                newComms = {padCells{:}, newComms{:}}; %#ok<CCAT> Dereference the Composite returned from the spmd_feval_fcn
                
                subSetObj = spmdlang.RemoteResourceSet( 1:sz, newComms(:), obj.Pool );
            else
                % Should never get here because there's already a check that will not let us
                % build a resource set greater than the size of the base set
                error(message('parallel:lang:spmd:InvalidResourceSelection', sz));
            end
        end

        
        function tf = isValid( obj )
            pool = obj.Pool;
            tf = ~isempty(pool) && isvalid(pool) && pool.Connected;
        end
        
        function tf = canAccessLabs( obj )
            % Need to test if we are allowed to access the lab and ask it
            % to do and operation - currently this is simply the same as
            % asking are we in parallel_function but might become much more
            % complex in the future when labs (MUE's) have actual
            % allocation undertaken.
            if obj.isValid()
                tf = matlab.internal.getParallelFunctionDepth == 0;
            else
                tf = false;
            end
        end
        
        function tf = satisfiesConstraints( obj, minN, maxN )
            obj.errorIfInvalid();
            tf = ( obj.numlabs <= maxN && obj.numlabs >= minN );
        end
        
        function val = getFromLab( obj, labidx, key )
        % Talk to the lab, get the value
            obj.errorIfInvalid();

            % Workaround for g1184951 - build a remote executor to keep the labs busy while
            % we're getting data.
            dummyBlockEx = spmdlang.RemoteSpmdExecutor( spmdlang.ResourceSetHolder( obj ), ...
                                                        [], [], [], ...
                                                        [], [], obj.Pool.hGetSession() );
            cleanupBlockEx = onCleanup( @() dispose( dummyBlockEx ) );
            
            % Make sure we can detect transfer of Remote objects
            spmdlang.AbstractRemote.saveLoadCount( 'clear' );

            val = remoteRetrieval( obj, obj.RemoteProcessArray( labidx ), key );

            if spmdlang.AbstractRemote.saveLoadCount( 'get' ) ~= 0
                warning(message('parallel:lang:spmd:RemoteObjectRetrievedFromLabs'));
            end
        end
        
        function newKey = setOnLab( obj, labidx, newValue )
        % Talk to the lab, send the value, retrieve the key
            obj.errorIfInvalid();

            % Workaround for g1184951 - build a remote executor to keep the labs busy while
            % we're setting data.
            dummyBlockEx = spmdlang.RemoteSpmdExecutor( spmdlang.ResourceSetHolder( obj ), ...
                                                        [], [], [], ...
                                                        [], [], obj.Pool.hGetSession() );
            cleanupBlockEx = onCleanup( @() dispose( dummyBlockEx ) );

            % Make sure we can detect transfer of Remote objects
            spmdlang.AbstractRemote.saveLoadCount( 'clear' );

            newKey = remoteSet( obj, obj.RemoteProcessArray( labidx ), newValue );

            if spmdlang.AbstractRemote.saveLoadCount( 'get' ) ~= 0
                warning(message('parallel:lang:spmd:RemoteObjectSentToLabs'));
            end
        end
        
        function keyUnreferenced( obj, labidx, key )
        % Push the key onto the stack of things to clear for the particular lab -
        % don't worry if we're not valid (that means the lab has probably
        % gone away)
            if isValid( obj )
                remoteClear( obj, obj.RemoteProcessArray( labidx ), key );
            end
        end
        
        function blockEx = buildBlockExecutor( obj, bodyFcn, assignOutFcn, getOutFcn, ...
                                               unpackInFcn, initialOuts )
            obj.errorIfInvalid();
            blockEx = spmdlang.RemoteSpmdExecutor( spmdlang.ResourceSetHolder( obj ), ...
                                                  bodyFcn, assignOutFcn, getOutFcn, ...
                                                  unpackInFcn, initialOuts, obj.Pool.hGetSession() );
        end
        
        function invalidate( obj, labsStillAvailable )
            if ~obj.IsWorld && labsStillAvailable
                % Use SPMD itself to release the communicators
                spmd_feval_fcn( @iFreeMyComm, {}, obj );
            end
            % Finally, call the superclass invalidation
            invalidate@spmdlang.AbstractResourceSet( obj, labsStillAvailable );
        end

        function tf = shouldRetryIfFileNotFound( ~ )
        % Always retry the computation a file not found error is thrown by the resource set.
            tf = true;
        end
    end
    
    %% Methods to do with Remote retrieval
    methods ( Access = private, Hidden )
        function errorIfSessionClosed( obj )
            if ~obj.isValid
                error(message('parallel:lang:spmd:PoolSessionHasClosed'));
            end
        end
        
        function val = remoteRetrieval( obj, remoteProc, key )
            import java.util.concurrent.TimeUnit;
            obj.errorIfSessionClosed();

            obs = obj.Pool.hGetSession().getCompositeAssistant.retrieveCompositeValue( remoteProc, ...
                                                              distcompserialize( key ) );
            
            % Note that disposing the observer frees the byte buffers
            x = onCleanup( @()(obs.dispose()) );
            while ~obs.await( 100, TimeUnit.MILLISECONDS )
                % Allow CTRL-C out of lengthy retrieval. The byte buffer will be disposed of
                % immediately.
                obj.errorIfSessionClosed();
            end
            
            result = obs.getResult().getResult();
            % Also note - don't call "free" here since the observer will do that for me.
            doFree = false;
            thing  = parallel.internal.pool.deserialize( result(2), doFree );

            if logical( result(1) )
                val = thing;
            else
                except = thing;
                ME = MException(message('parallel:lang:spmd:RemoteRetrievalError'));
                ME = addCause( ME, except );
                throw( ME );
            end
        end
        
        function newKey = remoteSet( obj, remoteProc, newValue )
            import java.util.concurrent.TimeUnit;
            obj.errorIfSessionClosed();
            
            rca = obj.Pool.hGetSession().getCompositeAssistant;
            
            bbufh = parallel.internal.pool.serialize( newValue );
            obs = rca.sendCompositeValue( remoteProc, bbufh );
            
            % Note that disposing the observer frees the byte buffer
            x = onCleanup( @()(obs.dispose()) );
            while ~obs.await( 100, TimeUnit.MILLISECONDS )
                % Allow CTRL-C out of lengthy set. 
                obj.errorIfSessionClosed();
            end
            
            result = obs.getResult().getResult();

            % Do not free result here - composite assistant will do that.
            doFree = false;
            thing  = parallel.internal.pool.deserialize( result(2), doFree );
            if logical( result(1) )
                newKey = thing;
            else
                except = thing;
                ME = MException(message('parallel:lang:spmd:RemoteSet'));
                ME = addCause( ME, except );
                throw( ME );
            end
        end

        function remoteClear( obj, remoteProc, key )
            obj.errorIfSessionClosed();
            import com.mathworks.toolbox.distcomp.pmode.RemoteCompositeAssistant;
            obj.Pool.hGetSession().getCompositeAssistant.notifyRemoteClear( remoteProc, key );
        end
    end
    
end

% Used by RemoteResourceSet::invalidate on the labs to free a given communicator.
function iFreeMyComm()
    % Use a special back-door to free communicators - the block cleanup needs to
    % use them for error checking.
    spmdlang.remoteBlockExecution( 'doFreeComm' );
end

% Used by RemoteResourceSet::buildRemoteResourceSet to split the world communicator.
function val = iBuildCommHandle( newSz )
    if newSz > numlabs
        % Should never get here - two layers of previous checks would have to fail
        % to catch this...
        error(message('parallel:lang:spmd:InvalidComms', newSz, numlabs));
    end
    
    % All labs <= labindex get color 1, others get color -1 (so they get COMM_NULL)
    color = (2 * double(labindex <= newSz)) - 1;
    key    = labindex;
    
    val    = mpiCommManip( 'split', color, key );
end



