%FutureOutputHelper for C++ parfeval

% Copyright 2021 The MathWorks, Inc.
classdef CppFutureOutputHelper < handle
properties (GetAccess = private, SetAccess = immutable)
IsBroadcast
% For a broadcast-type FilteringFuture, a list of worker indices to select for output.
FilteredWorkerIndices
end
properties (Access = private)
CppTask
OutputCache = {};
ErrorCache = ParallelException.empty();
DiaryCache = '';
WorkersCache
CacheValid = false;

% Time fields stored internally as 'int64' values to avoid paying overhead of
% datetime construction unless absolutely necessary.
CreateTimeCache = parallel.internal.timestamps.invalidTime();
StartTimeCache = parallel.internal.timestamps.invalidTime();
FinishTimeCache = parallel.internal.timestamps.invalidTime();
end

methods (Access = private)
function val = getAndCacheTimeField(obj, fieldName, taskFcn, taskState)
currVal = obj.(fieldName);
if currVal ~= parallel.internal.timestamps.invalidTime()
val = currVal;
else
if isTerminal(taskState)
updateCache(obj, taskState);
% Retrieve value again now cache is updated
val = obj.(fieldName);
else
task = obj.CppTask;
if ~isempty(task)
val = taskFcn(task);
obj.(fieldName) = val;
else
val = parallel.internal.timestamps.invalidTime();
end
end
end
end
end

methods (Access = ?parallel.internal.queue.Future)

function val = getCreateTime(obj, taskState)
val = epochMillisToDatetime(getAndCacheTimeField(obj, 'CreateTimeCache', @getCreateTime, taskState));
end

function val = getStartTime(obj, taskState)
val = epochMillisToDatetime(getAndCacheTimeField(obj, 'StartTimeCache', @getStartTime, taskState));
end

function val = getFinishTime(obj, taskState)
if ~obj.CacheValid && isTerminal(taskState)
updateCache(obj, taskState);
end
val = epochMillisToDatetime(obj.FinishTimeCache);
end

function val = getDiary(obj, taskState)
if isTerminal(taskState)
% Task is finished, get Diary for the final time.
updateCache(obj, taskState);
val = obj.DiaryCache;
else
task = obj.CppTask;
if ~isempty(task)
val = getDiaryImpl(obj, task);
else
% Never get here.
if obj.IsBroadcast
val = {};
else
val = '';
end
end
end
end

function val = getError(obj, taskState)
if ~obj.CacheValid && isTerminal(taskState)
updateCache(obj, taskState);
end
val = obj.ErrorCache;
end

function val = getOutputArguments(obj, taskState)
if ~obj.CacheValid && isTerminal(taskState)
updateCache(obj, taskState);
end
val = obj.OutputCache;
end

function val = getWorkers(obj, taskState)
if ~obj.CacheValid && isTerminal(taskState)
updateCache(obj, taskState);
end
val = obj.WorkersCache;
end
end

methods (Access = private)

function txtOrCell = getDiaryImpl(obj, optCppTask, workers)
if nargin < 2
cppTask = obj.CppTask;
else
cppTask = optCppTask;
end
assert(~isempty(cppTask));

if nargin < 3
workers = cppTask.getWorkers();
end

if obj.IsBroadcast
if isempty(workers)
% we just want an cell of empty strings
[txtOrCell{1:numel(workers)}] = deal('');
else
txtOrCell = arrayfun(@(w) cppTask.getDiary(w), workers, ...
'UniformOutput', false);
end
else
if isempty(workers)
txtOrCell = '';
else
txtOrCell = cppTask.getDiary(workers(1));
end
end
end

function updateCache(obj, taskState)
if obj.CacheValid
% Nothing to do
else
cppTask = obj.CppTask;
assert(~isempty(cppTask));
create = int64(cppTask.getCreateTime());
start = int64(cppTask.getStartTime());
stop = int64(cppTask.getFinishTime());

workers = cppTask.getWorkers();

% Make sure workers are a row, since output, diary should
% be rows
workers = reshape(workers, [1,numel(workers)]);

% Get output sorted by process index
workers = sort(workers);

if isempty(workers)
% Task cancelled before workers were assigned.
% IInterpertTaskResult will handle empty worker ID
[out, err] = iInterpretTaskResult(cppTask, [], taskState);
elseif obj.IsBroadcast
[out, err] = arrayfun(@(w) iInterpretTaskResult(cppTask, w, taskState), ...
workers, 'UniformOutput', false);
else
[out, err] = iInterpretTaskResult(cppTask, workers(1), taskState);
end

obj.DiaryCache = getDiaryImpl(obj, cppTask, workers);

if ~isempty(obj.FilteredWorkerIndices)
assert(obj.IsBroadcast, "Only parfevalOnAll can have worker filter");

% If we have a worker index filter, we need to work out which elements of
% 'workers' are present in the FilteredWorkerIndices, and then
% update 'workers' and 'ordering'.
validMask = ismember(workers, obj.FilteredWorkerIndices);
if ~isempty(validMask)
workers = workers(validMask);
out = out(validMask);
err = err(validMask);
obj.DiaryCache = obj.DiaryCache(validMask);
end
end

if obj.IsBroadcast && ~isempty(workers)
if all(cellfun(@isempty, err))
% Override cell array of N empties with empty cell.
err = {};
end
end

obj.OutputCache = out;
obj.ErrorCache = err;
obj.CacheValid = true;
obj.WorkersCache = workers;
obj.CreateTimeCache = create;
obj.StartTimeCache = start;
obj.FinishTimeCache = stop;

% We've extracted all the data from the cppTask, no need to
% hold on to it longer.
obj.CppTask = [];
end
end
end

methods (Static)
function obj = buildForPlainRequest(cppTask)
% Factory method used to build an output helper for a plain request
isBroadcast = false;
obj = parallel.internal.queue.CppFutureOutputHelper(cppTask, isBroadcast);
end

function obj = buildForBroadcastRequest(cppTask)
% Factory method used to build an output helper for a broadcast request
isBroadcast = true;
obj = parallel.internal.queue.CppFutureOutputHelper(cppTask, isBroadcast);
end

function obj = buildForFilteringRequest(cppTask, filter)
isBroadcast = true;
obj = parallel.internal.queue.CppFutureOutputHelper(cppTask, isBroadcast, filter);
end
end

methods (Access = private)
function obj = CppFutureOutputHelper(cppTask, isBroadcast, filter)
obj.CppTask = cppTask;
obj.IsBroadcast = isBroadcast;
if isBroadcast
% Only needed for broadcast,
% the default values are correct otherwise
obj.OutputCache = {{}};
obj.ErrorCache = {};
obj.DiaryCache = {};
end
if nargin > 2
obj.FilteredWorkerIndices = filter;
else
obj.FilteredWorkerIndices = [];
end
end
end

methods (Access = ?parallel.internal.queue.Future)
function setCancelled(obj, errObj)
% We're being notified by the Future that we have been deliberately canceled.
if obj.IsBroadcast
fcn = @(x) {x};
else
fcn = @(x) x;
end

% Update time properties - these methods have side-effects which cache the
% values.
state = parallel.internal.types.States.Finished;
getCreateTime(obj, state);
getStartTime(obj, state);
obj.FinishTimeCache = parallel.internal.timestamps.currentTime();

% Overwrite other properties
obj.ErrorCache = fcn(errObj);
obj.DiaryCache = fcn('');
obj.OutputCache = fcn({});
obj.CacheValid = true;
end
end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% interpret a CppTask, and return the output and error.
function [out, err] = iInterpretTaskResult(cppTask, workerId, taskState)

out = [];
err = ParallelException.empty();

try
if isequal(taskState, parallel.internal.types.States.Failed) || ...
isempty(workerId)
% No result available.
err = ParallelException.hBuildLikeMException(message('MATLAB:parallel:future:NoResultAvailable'));
elseif cppTask.hasError()
remoteException = cppTask.getError();
err = parallel.internal.pool.handleRemoteException(remoteException);
err = ParallelException.hBuildLikeMException(err.identifier, err.message);
elseif cppTask.hasWorkerError(workerId)
remoteException = cppTask.getWorkerError(workerId);
err = parallel.internal.pool.handleRemoteException(remoteException);
err = ParallelException.hBuildLikeMException(err.identifier, err.message);
else
rawResult = cppTask.getResult(workerId);
if isempty(rawResult)
err = ParallelException.hBuildLikeMException(message('MATLAB:parallel:future:NoResultAvailable'));
return
end

if ~isempty(rawResult) && numel(rawResult) == 2
% expected format
if rawResult{1} == 0
out = rawResult{2};
else
% Got an error on the worker
err = rawResult{2};
end
else
err = ParallelException.hBuildLikeMException(message('MATLAB:parallel:future:InvalidExecutionResult'));
end
end

catch E
err = ParallelException.hBuildLikeMException(message('MATLAB:parallel:future:InvalidExecutionResult'));
err = addCause(err, E);
end

end