function [ERROR, out] = remoteParallelFunction(init, data, isFinalInterval)
; %#ok Undocumented

% Copyright 2007-2020 The MathWorks, Inc.

persistent channel initData initException
mlock;

try
    notFound = false;
    % Channel construction data supplied
    if ~isempty( init ) && init(1).get.capacity() > 0
        % Make sure that numlabs and labindex are 1 during a parfor, and that
        % Matlab files are re-interpreted as necessary.
        pctPreRemoteEvaluation( 'mpi_mi' );

        % Make absolutely sure that we recreate the channel.
        channel = [];

        % Clear out initException - we'll store something here if we fail to
        % deserialize the init data we've been sent. We need to store it in a
        % persistent to handle the case where no actual interval data is being
        % sent.
        initException = [];

        % Always deserialize the initData as soon as we see it - this might overlap with
        % communication.
        try
            [initData{1:2}] = parallel.internal.pool.deserializeFunction(init);
        catch E
            % If we get here, store away the problem so that in the case where there is no
            % interval to be run right now, we throw this error when we do
            % have an interval.
            dctSchedulerMessage(1, 'Error deserializing init data: %s', getReport(E));
            initException = E;
            rethrow(E);
        end
    end

    % Iterate data supplied
    if ~isempty(data) && data(1).get.capacity() > 0
        if isempty(channel)
            % By the time we get here, initData must be a 2-element cell array, the first
            % element of which is a cell array containing the relevant
            % information to reconstruct the channel - the form is @func =
            % {@func, args}. The second element of initData is the flag from
            % deserializeFunction indicating whether a "function not found"
            % warning was issued during deserialization.
            
            if ~iscell(initData) || ~isequal(size(initData), [1, 2])
                dctSchedulerMessage(1, 'Invalid initData: %s', class(initData));
                if isa(initException, 'MException')
                    % We've been left an exception. Let's throw that.
                    throw(initException);
                else
                    error(message('parallel:lang:parfor:InvalidInitData'));
                end
            end
            [channelData, notFound] = deal(initData{:});
            if isempty(channelData) || ~iscell(channelData)
                error(message('parallel:lang:parfor:InvalidChannelData'));
            end
            channel = feval(channelData{:});
            initData = [];
        end
        channelArgs = parallel.internal.pool.deserialize(data);
        out = parallel.internal.pool.serialize(feval(channel, channelArgs{:}));
    else
        % No iterate data - perhaps we're just being sent broadcast data.
        out = parallel.internal.pool.serialize([]);
    end

    % Indicate that no error occurred
    ERROR = false;
catch err
    ERROR = true;
    isMissingSourceError = parallel.internal.pool.missingSourcesFromException(err);
    if notFound || isMissingSourceError
        % Build the Source Code Not Available error and throw it so that we get
        % some error stack associated with it.
        try
            sourceCodeErr = MException(message('parallel:lang:parfor:SourceCodeNotAvailable'));
            sourceCodeErr = addCause(sourceCodeErr, err);
            throw(sourceCodeErr);
        catch theErr
        end
    else
        theErr = err;
    end
    stackToIgnore = 'parallel_function';
    errToSerialize = ParallelException.hBuildFromRemoteException(theErr, stackToIgnore);
    out = parallel.internal.pool.serialize(errToSerialize);
end

if isFinalInterval || ERROR
    channel  = [];
    initData = [];
    % Reset the MPI layer after a parfor
    dctRegisterMpiFunctions('mwmpi');
end
if isFinalInterval
    initException = [];
end
