%Composite Construct a Composite object
%   C = Composite() creates a Composite object on the client using workers from
%   the parallel pool. The actual number of workers referred to by this Composite
%   object depends on the size of the parallel pool, and any pre-existing
%   Composite objects. Generally, Composite objects should be constructed
%   outside any SPMD block.
%
%   C = Composite(NWORKERS) creates a Composite on the parallel resource set
%   that matches the specified constraint.  NWORKERS must be a vector of length
%   1 or 2, containing integers or Inf.  If NWORKERS is of length 1, it
%   specifies the exact number of workers to use. If NWORKERS is of size 2, it
%   specifies the minimum and maximum number of workers to use. The actual
%   number of workers used is the maximum number of workers compatible with the
%   size of the parallel pool, and other existing Composite objects.  An error
%   is thrown if the constraints on the number of workers cannot be met.
%
%   A Composite object has one entry for each worker, initially each entry will
%   contain no data. Use either indexing or SPMD to define values for the
%   entries.
%
%   Example: 
%   % Create a Composite object with no defined entries
%   c = Composite();
%   for ii = 1:length( c )
%       % Set the entry for each worker to zero
%       c{ii} = 0;
%   end
%   
%   Composite methods:
%      EXIST    - Determine if the Composite entry on a given worker contains data
%      LENGTH   - Return how many entries a Composite has
%      GATHER   - Retrieve all entries into a cell array on the client
%      SUBSREF  - Retrieve a subset of entries from workers
%      SUBSASGN - Set entries on workers
%
%   See also SPMD, PARPOOL

% Copyright 2008-2022 The MathWorks, Inc.

classdef Composite < spmdlang.BaseRemote

    properties ( Access = private, Hidden, Transient )
        % Store the client-side value
        ClientValue = [];
        % Is the client-side value valid?
        ClientValueValid = false;
    end

    methods ( Access = public, Hidden )
        % NB - called from AbstractSpmdExecutor
        function obj = setClientValue( obj, clientVal )
            obj.ClientValue = clientVal;
            obj.ClientValueValid = true;
        end
    end

    methods ( Access = protected, Hidden )
        function obj = pPostKeySet( obj )
        % Use this to clear the client value if the key vector is "full".
            gotKey = cellfun( @isempty, obj.KeyVector );
            if all( gotKey )
                obj.ClientValue = [];
                obj.ClientValueValid = false;
            end
        end
    end
    
    methods ( Access = public )

        % Public construction of a Composite is of the form:
        % Composite() or Composite( [numeric vector] )
        % Private construction is of the form
        % Composite( '<action>', args )
        % Currently, only valid action is 'empty' with no args.
        function obj = Composite( varargin )

        % Deal with args
            if nargin == 0
                action = 'public';
                workersC  = {};
            elseif nargin == 1 && isnumeric( varargin{1} )
                action = 'public';
                nWorkersV = varargin{1};
                switch length( nWorkersV )
                  case 1
                    workersC = {nWorkersV};
                  case 2
                    workersC = {nWorkersV(1), nWorkersV(2)};
                  otherwise
                    error(message('parallel:lang:spmd:CompositeCtorArgs'));
                end
            elseif ischar( varargin{1} )
                action = varargin{1};
            else
                error(message('parallel:lang:spmd:CompositeCtorArgs'));
            end

            % Build via superclass constructor
            obj = obj@spmdlang.BaseRemote();

            % Perform any post-superclass-construction actions
            switch action
              case 'public'
                try
                    rs = spmdlang.ResourceSetMgr.chooseResourceSet( {}, workersC{:} );
                catch E
                    if strcmp( E.identifier, 'parallel:lang:spmd:SpmdNotSupportedWithThisPool' )
                        throw( E );
                    else
                        EE = MException(message('parallel:lang:spmd:CompositeCtorArgs'));
                        EE = addCause( EE, E );
                        throw( EE );
                    end
                end
                % Initialize with an empty cell array of keys - i.e. no worker-side information
                % at all.
                obj = obj.init( cell( 1, rs.NumWorkers ), ...
                                spmdlang.ResourceSetHolder( rs ), ...
                                @spmdlang.plainCompositeBuilder );
              case 'empty'
                % This is the case used by plainCompositeBuilder -
                % init will be called later.
              otherwise
                error(message('parallel:lang:spmd:CompositeCtorArgs'));
            end
        end

        %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
        % disp and display

        function disp( obj )
            if isempty( obj.ResSetHolder )
                fprintf( '    %s\n', getString( message( 'parallel:lang:spmd:InvalidComposite' ) ) );
            elseif ~obj.isResourceSetOpen()
                fprintf( '    %s\n', getString( message( 'parallel:lang:spmd:InvalidCompositeNoPool' ) ) );
            else
                keyCellOrEmpty = getRawKeyCell( obj );
                
                if obj.ClientValueValid
                    % Send a string for the workers to use if they don't have a value of their
                    % own.
                    clientStr = iGetOneDisplayStr( obj.ClientValue );
                else
                    clientStr = '';
                end

                % Build the display strings on each of the workers. This
                % returns another Composite object.
                dispStr = spmd_feval_fcn( @iGetDisplayStrs, {keyCellOrEmpty, clientStr}, ...
                    obj.ResSetHolder.getResourceSet() );
                
                fprintf( 1, '%s', getValOrError( dispStr, dispStr.KeyVector{1} ) );
            end
        end

    end
    
    methods ( Access = public, Hidden )

        function packed = packForTransmission( obj )
        % Send the client value to SPMD - this method decorates the AbstractRemote
        % version by adding the client value if one is present.
            packed = packForTransmission@spmdlang.AbstractRemote( obj );
            if obj.ClientValueValid
                packed = packed.setClientValue( obj.ClientValue );
            end
        end
        
    end

    methods ( Access = private, Hidden, Sealed )
        
        % Ensure we don't use any remote operations inside a parfor() loop
        function errorIfRemoteOperationInProgress( obj )
            if obj.isResourceSetOpen()
                resSet = obj.ResSetHolder.getResourceSet();
                if ~resSet.canAccessWorkers()
                    error(message('parallel:lang:spmd:InvalidCompositeAccess'));
                end
            end
        end
        
        % Send data to the workers, update key vector.
        function obj = setValOrError( obj, workerIdx, value )
            % Always check that we're not in a parfor
            obj.errorIfRemoteOperationInProgress();
            if workerIdx >= 1 && workerIdx <= length( obj )
                
                % Actually send the data to the worker, and store the returned key.
                newKey = obj.ResSetHolder.getResourceSet().setOnWorker( workerIdx, value );
                obj.KeyVector{workerIdx} = spmdlang.KeyHolder( newKey, workerIdx, obj.ResSetHolder );
            else
                error(message('parallel:lang:spmd:CompositeSetValue', workerIdx));
            end
        end

        % Get data from a worker
        function val = getValOrError( obj, keyHolder, workerIdx )
            gotClient = obj.ClientValueValid;
            
            % Always check that we're not in a parfor loop
            obj.errorIfRemoteOperationInProgress();

            if ~isempty( keyHolder )
                val = keyHolder.getFromWorker();
            elseif gotClient
                val = obj.ClientValue;
            else
                error(message('parallel:lang:spmd:CompositeGetValue', workerIdx));
            end
        end
    end

    
    methods ( Access = public )

        % No need to doc "end", so keep it in here.
        function idx = end( obj, k, n )
            idx = builtin( 'end', obj.KeyVector, k, n );
        end
    end        

end

function oneStr = iGetOneDisplayStr( value )
    oneStr = sprintf( 'class = %s, size = [%s]', ...
                      class( value ), num2str( size( value ) ) );
end

% Internal function used by disp
function workerStr = iGetDisplayStrs( keyCell, clientStr )
    
% Mimic the SPMD output format - to do this effectively, we need the
% spmdIndex formatted to the correct width
    requiredDigits = 1 + floor( log10( spmdSize ) );
    formatStr = sprintf( 'Worker %%%dd: ', requiredDigits );
    workerIdxStr = sprintf( formatStr, spmdIndex );
    
    if ~isempty( keyCell{ spmdIndex } )
        data = spmdlang.ValueStore.retrieve( keyCell{spmdIndex} );
        workerStr = sprintf( '   %s%s', workerIdxStr, iGetOneDisplayStr( data ) );
    else
        if ~isempty( clientStr )
            workerStr = sprintf( '   %s%s', workerIdxStr, clientStr );
        else
            workerStr = sprintf( '   %sNo data', workerIdxStr );
        end
    end

    workerStrCell = spmdCat( {workerStr}, 1 );
    if spmdIndex == 1
        workerStr = sprintf( '%s\n', workerStrCell{:} );
    end
end
