%%
% LocalSpmdExecutor - control class to manage local execution of an SPMD block

% Copyright 2008-2021 The MathWorks, Inc.

classdef LocalSpmdExecutor < spmdlang.AbstractSpmdExecutor

    properties ( Access = private, Hidden )
        % In cases where we are using LocalSpmdExecutor on a worker, we
        % should perform a cleanup operation to restore the previous
        % communicator.
        RestoreCommunicator;

        % If the block encountered an exception, store that here.
        BlockException = [];
    end

    methods

        function obj = LocalSpmdExecutor( resSetH, bodyF, assignOutF, getOutF, unpackInF, initialOuts )
        % Simulate transfer by going through serialize/deserialize cycle. This also
        % allows us to disallow hidden variant transfers.

            [listener, getListFcn] = ...
                parallel.internal.general.SerializationNotifier.createAccumulatingListener(); %#ok<ASGLU>

            warnStates = spmdlang.AbstractSpmdExecutor.disableRemoteSaveWarnings();
            try
                transferred = parallel.internal.pool.mxSerializeDeserialize( ...
                        { bodyF, getOutF, unpackInF } );
            catch E
                spmdlang.AbstractSpmdExecutor.restoreRemoteSaveWarnings( warnStates );
                rethrow( E );
            end
            spmdlang.AbstractSpmdExecutor.restoreRemoteSaveWarnings( warnStates );

            if any(ismember(getListFcn(), {'distributed', 'Composite'}))
                pOneHiddenCompositeWarning;
            end

            obj = obj@spmdlang.AbstractSpmdExecutor( resSetH, transferred{1}, assignOutF, ...
                                                     transferred{2}, transferred{3}, initialOuts );

        end

        % For local execution, we actually perform all the computation here.
        function initiateComputation( obj )
            % If we are using LocalSpmdExecutor on a worker, we need to
            % perform some communicator setup and cleanup.
            if parallel.internal.pool.isPoolThreadWorker()
                prevComm = parallel.internal.mpi.getThreadCommunicator();
                selfComm = parallel.internal.mpi.buildSelfThreadCommunicator();
                obj.RestoreCommunicator = onCleanup( @() iThreadCommCleanup( prevComm , selfComm ) );
                parallel.internal.mpi.setThreadCommunicator( selfComm );
            elseif mpiInitialized
                prevComm = mpiCommManip('queryState');
                obj.RestoreCommunicator = onCleanup( @() mpiCommManip('select', prevComm) );
                mpiCommManip( 'select', 'self' );
            end

            % Unpack the inputs and execute the block
            obj.UnpackInFcn( @spmdlang.AbstractSpmdExecutor.unpack );
            try
                obj.BodyFcn();
            catch cause

                clientStackToIgnore = {'spmd_feval', 'spmd_feval_impl', 'spmd_feval_fcn', ...
                    'distributedSpmdWrapper', 'LocalSpmdExecutor'};
                % Create a new ParallelException from cause with a cleaned stack.
                except = ParallelException.hBuildLocalParallelException(cause, clientStackToIgnore);

                % Stash this for later.
                obj.BlockException = except;
            end
        end

        function throwBlockExceptions( obj )
            if ~isempty( obj.BlockException )
                % Create generic SPMD error, add obj.BlockException as
                % cause and as remote cause, and throw.
                except = pWrapExceptionsIntoParallelException({obj.BlockException}, '1');
                throw( except );
            end
        end

        function yes = isComputationComplete( obj ) %#ok<MANU>
        % All the work is done in "initiateComputation" for local execution, so simply return true.
            yes = true;
        end

        function dispose( obj )
                        % Set disposed to "true" now - we don't need to go through here again under
            % any circumstances.
            obj.Disposed = true;

            outCell = obj.GetOutFcn();
            outCellDeref = cell( 1, length( outCell ) );

            %% "LAB"-side
            % ii is which variable we're dealing with
            for ii=1:length( outCell )
                if isempty( outCell{ii} )
                    outCellDeref{ii}{1} = [];
                else
                    % During remote execution, this parceling is done in remoteBlockBody
                    data = outCell{ii}{1};
                    key  = spmdlang.ValueStore.store( data );
                    [fcn, data] = getRemoteFromSPMD( data );
                    outCellDeref{ii}{1} = spmdlang.ReturnableParcel( key, fcn, data );
                end
            end

            obj.callAssignOuts( outCellDeref );
        end
    end
end

function iThreadCommCleanup( prevComm , selfComm )
% Thread Communicators must alwasys be cleaned up, including the 'self'
% communicator.
parallel.internal.mpi.setThreadCommunicator( prevComm );
parallel.internal.mpi.invalidateThreadCommunicator( selfComm );

end