%FutureOutputHelper
%   Implementation helper for parallel.internal.queue.Future.

% Copyright 2013-2021 The MathWorks, Inc.
classdef FutureOutputHelper < handle
    properties (GetAccess = private, SetAccess = immutable)
        JavaTask
        IsBroadcast
        % For a broadcast-type FilteringFuture, a list of worker indices to select for output.
        WorkerIndexFilter
    end
    properties (Access = private)
        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 = int64(-1);
        StartTimeCache = int64(-1);
        FinishTimeCache = int64(-1);
    end
    
    methods (Access = private)
        function val = getAndCacheTimeField(obj, fieldName, taskFcn, taskState)
            currVal = obj.(fieldName);
            if currVal ~= -1
                val = currVal;
            else
                if isTerminal(taskState)
                    updateCache(obj, taskState);
                    % Retrieve value again now cache is updated
                    val = obj.(fieldName);
                else
                    task = obj.JavaTask.getIfNotDisposed();
                    if ~isempty(task)
                        val = taskFcn(task);
                        obj.(fieldName) = val;
                    else
                        val = int64(-1);
                    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.JavaTask.getIfNotDisposed();
                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 wCell = getWorkersCell(obj, javaTask)
        % Get a cell array of workers (applies to all types)
            if nargin < 2
                javaTask = obj.JavaTask.getIfNotDisposed();
            end
            assert(~isempty(javaTask));
            if obj.IsBroadcast
                wList = javaTask.getWorkers();
                % Convert java array to cell array.
                numWorkers = double(wList.size());
                wCell = arrayfun(@(idx) wList.get(idx), 0:(numWorkers-1), ...
                                    'UniformOutput', false);
            else
                wCell = {javaTask.getWorker()};
            end
        end

        function txtOrCell = getDiaryImpl(obj, optJavaTask, workersCell)
            if nargin < 2
                javaTask = obj.JavaTask.getIfNotDisposed();
            else
                javaTask = optJavaTask;
            end
            assert(~isempty(javaTask));
            if nargin < 3
                workersCell = getWorkersCell(obj);
            end
            if obj.IsBroadcast
                txtOrCell = cellfun(@(w) char(javaTask.getDiary(w)), workersCell, ...
                                    'UniformOutput', false);
            else
                txtOrCell = char(javaTask.getDiary());
            end
        end

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

                workersCell = getWorkersCell(obj, javaTask);
                if obj.IsBroadcast
                    [out, err] = cellfun(@(w) iInterpret(javaTask.getResult(w), taskState), ...
                                         workersCell, 'UniformOutput', false);
                else
                    [out, err] = iInterpret(javaTask.getResult(), taskState);
                end
                obj.DiaryCache = getDiaryImpl(obj, javaTask, workersCell);
                
                % Worker index is effectively 'labindex'.
                workers = cellfun(@iGetWorkerIndex, workersCell);
                if ~any(isnan(workers))
                    [workers, ordering] = sort(workers);
                else
                    ordering = 1:numel(workers);
                end
                
                if ~isempty(obj.WorkerIndexFilter)
                    % If we have a worker index filter, we need to work out which elements of
                    % 'workers' are present in the WorkerIndexFilter, and then
                    % update 'workers' and 'ordering'.
                    validMask = ismember(workers, obj.WorkerIndexFilter);
                    ordering  = ordering(validMask);
                    workers   = workers(validMask);
                end

                if obj.IsBroadcast
                    out = out(ordering);
                    err = err(ordering);
                    if all(cellfun(@isempty, err))
                        % Override cell array of N empties with empty cell.
                        err = {};
                    end
                    obj.DiaryCache = obj.DiaryCache(ordering);
                end
                obj.OutputCache  = out;
                obj.ErrorCache   = err;
                obj.CacheValid   = true;
                obj.WorkersCache = workers;
                obj.CreateTimeCache = create;
                obj.StartTimeCache = start;
                obj.FinishTimeCache = stop;
                obj.JavaTask.dispose();
            end
        end
    end

    methods
        function delete(obj)
            obj.JavaTask.disposeIfNecessary();
        end
    end

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

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

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

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

    methods (Access = private)
        function obj = FutureOutputHelper(jTask, isBroadcast, filter)
            obj.JavaTask = jTask;
            obj.IsBroadcast = isBroadcast;
            if isBroadcast
                obj.OutputCache = {{}};
                obj.ErrorCache  = {};
                obj.DiaryCache  = {};
            else
                % Do nothing - the default values are correct.
            end
            if nargin > 2
                obj.WorkerIndexFilter = filter;
            else
                obj.WorkerIndexFilter = [];
            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 cancelled.
            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;
            obj.JavaTask.disposeIfNecessary();
        end
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Is the java exception due to lack of resources?
function tf = iIsLackOfResources(javaException)
    executionExClass = 'java.util.concurrent.ExecutionException';
    noResourceExClass = 'com.mathworks.toolbox.distcomp.pmode.taskqueue.ResourcesUnavailableException';
    tf = isa(javaException, noResourceExClass) || ...
         (isa(javaException, executionExClass) && ...
          isa(javaException.getCause(), noResourceExClass));
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Is the java exception due to a broadcast worker disappearing?
function tf = iIsBroadcastFailed(javaException)
    executionExClass = 'java.util.concurrent.ExecutionException';
    noSuchDestination = 'com.mathworks.toolbox.distcomp.pmode.shared.NoSuchDestinationException';
    tf = isa(javaException, noSuchDestination) || ...
         (isa(javaException, executionExClass) && ...
          isa(javaException.getCause(), noSuchDestination));
end


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Is the java exception due to the user having cancelled the future?
function tf = isCancelledByUser(javaException)
    executionExClass = 'java.util.concurrent.ExecutionException';
    cancelledByUser = 'com.mathworks.toolbox.distcomp.pmode.taskqueue.TaskCancelledByUserException';
    tf = isa(javaException, cancelledByUser) || ...
         (isa(javaException, executionExClass) && ...
          isa(javaException.getCause(), cancelledByUser));
end


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% interpret a java result object, and return the output and error.
function [out, err] = iInterpret(javaResult, taskState)
% We never free, we rely on destruction of the object to release.
import com.mathworks.toolbox.distcomp.pmode.taskqueue.FevalEvaluationStatus

    persistent EVALUATION_OK
    if isempty(EVALUATION_OK)
        EVALUATION_OK = FevalEvaluationStatus.EVALUATION_OK;
    end
    
    doFree = false;
    out = [];
    err = ParallelException.empty();
    try
        if isequal(taskState, parallel.internal.types.States.Failed) || ...
                isempty(javaResult)
            % No result available.
            err = ParallelException.hBuildLikeMException(message('MATLAB:parallel:future:NoResultAvailable'));
        elseif javaResult.hasError()
            javaException = javaResult.getError();
            % Log the error at the client...
            parallel.internal.general.logJavaException(0, 'Failed to get result from PARFEVAL.', javaException);
            [isMvmException, mvmException] = ...
                parallel.internal.general.transformIfMvmException(javaException);
            if iIsLackOfResources(javaException)
                err = ParallelException.hBuildLikeMException(message('MATLAB:parallel:future:NoResourcesForQueue'));
            elseif iIsBroadcastFailed(javaException)
                err = ParallelException.hBuildLikeMException(message('MATLAB:parallel:future:FevalOnAllWorkerFailed'));
            elseif isCancelledByUser(javaException)
                err = ParallelException.hBuildLikeMException('parallel:fevalqueue:ExecutionCancelled', ...
                    message('MATLAB:parallel:future:ExecutionCancelled'));
            elseif isMvmException
                err = mvmException;
            else
                err = ParallelException.hBuildLikeMException(message('MATLAB:parallel:future:InvalidExecutionResult'));
            end
        else
            rawResult = javaResult.getResult();
            if ~isempty(rawResult) && numel(rawResult) == 2
                % expected format
                if rawResult(1) == EVALUATION_OK
                    out = ...
                        parallel.internal.pool.deserialize(rawResult(2), doFree);
                else
                    % Got an error on the worker
                    err = distcompdeserialize(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

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function idx = iGetWorkerIndex(workerObj)
    try
        idx = double(workerObj.getLabIndex());
    catch E %#ok<NASGU> never expect to get here.
        idx = NaN;
    end
end
