function [OK, out] = remoteBlockExecution(action, varargin)
%remoteBlockExecution controls the SPMD block execution on the workers
% All arguments and results are serialized

% Copyright 2008-2022 The MathWorks, Inc.

% Delegate to remoteBlockExecutionPlain. This function just does any
% required adaption of arguments and results.

switch action
    case 'prelude'
        [commHCell, fcnHandles, keysToClear, preludeExcept] = iUnpackPreludeArgs(varargin{:});
        [OK, out] = spmdlang.remoteBlockExecutionPlain(action, commHCell, fcnHandles, keysToClear, preludeExcept);
    otherwise
        [OK, out] = spmdlang.remoteBlockExecutionPlain(action, varargin{:});
end
try
    out = distcompserialize(out);
catch E
    OK = false;
    stackFilesToIgnore = {'LocalSpmdExecutor', 'remoteBlockExecution'};
    out = distcompserialize(ParallelException.hBuildFromRemoteException(E, stackFilesToIgnore));
end
    
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iUnpackPreludeArgs - convert java arguments and serialized data into
% suitable arguments for remoteBlockExecutionPlain
function [commHCell, fcnHandles, keysToClear, preludeExcept] = iUnpackPreludeArgs(serCommHandleCell, serializedFcnHandles, arrayHolderMap)
    % Fetch any keys that are no longer in use. We need to find out our 'worker
    % index' rather than our world spmdIndex to handle the differences between
    % an interactive and a batch mode pool.
    sessionObj   = parallel.internal.pool.workerSession();
    workerKey = java.lang.Integer(sessionObj.WorkerIndex);
    keysToClear = int64.empty();
    if arrayHolderMap.containsKey(workerKey)
        keysForWorker = arrayHolderMap.get(workerKey);
        if ~isempty(keysForWorker)
            keysToClear = keysForWorker.array();
        end
    end

    % Unpack the data we've been given
    [commHCell, fcnHandles, preludeExcept] = ...
        iDeserializeInputs(serCommHandleCell, serializedFcnHandles);
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iDeserializeInputs - unpack everything sent over, and return along with a
% flag indicating whether any Remote type objects were sent.
function [commHCell, fcnHandles, deserExc] = iDeserializeInputs(serCommHandleCell, serializedFcnHandles)
    state = [];
    % Since we're returning an exception, give these default invalid values.
    commHCell = []; fcnHandles = [];
    try
        commHCell = parallel.internal.pool.deserialize(serCommHandleCell);
        fcnHandles = ...
            parallel.internal.pool.deserializeFunction(serializedFcnHandles);

        E = [];
    catch E
        % E will be rethrown later.
    end

    if ~isempty(state)
        warning(state);
    end

    % Return the error if we've got one
    deserExc = E;
end
