%AfterAllFuture Future for function invoked after all preceding futures
%   Create an AfterAllFuture by calling the afterAll method on
%   parallel.internal.queue.Future
%
%   parallel.internal.queue.AfterAllFuture methods:
%      afterAll     - Specify function to invoke after all Futures complete
%      afterEach    - Specify function to invoke after each Future completes
%      cancel       - Cancel a pending, queued, or running Future
%      fetchOutputs - Retrieve all Futures outputs
%      isequal      - true if futures have the same ID
%      wait         - Wait for Futures to complete
%
%   parallel.internal.queue.AfterAllFuture properties:
%      CreateDateTime     - Date and time at which this future was created
%      Error              - Future error information
%      FinishDateTime     - Date and time at which this future finished running
%      Function           - Function to evaluate
%      ID                 - Future's numeric identifier
%      InputArguments     - Input arguments to function
%      NumOutputArguments - Number of arguments returned by function
%      OutputArguments    - Output arguments from running Function on a worker
%      Predecessors       - The preceding futures
%      StartDateTime      - Date and time at which this future started running
%      State              - Current state of future
%
%   See also parallel.internal.queue.Future.afterEach,
%            parallel.internal.queue.Future.afterAll, parfeval, parfevalOnAll.

% Copyright 2017-2020 The MathWorks, Inc.

classdef AfterAllFuture < parallel.internal.queue.AbstractContinuationFuture

    properties(Access = private)
        NumTriggered = 0;
    end
    
    properties(Access = protected)
        ContinuationError;                
        ContinuationOutputArguments;
        ContinuationInputArguments;
    end
    
    methods
        function obj = AfterAllFuture(varargin)
            obj@parallel.internal.queue.AbstractContinuationFuture(varargin{:});
            
            obj.ContinuationError = ParallelException.empty;
            obj.ContinuationOutputArguments = {};
            obj.ContinuationInputArguments = {};
            
            obj.listenOnPredecessorsAndRun()
        end
    end
        
    methods (Access = protected)
        
        function setCancellationError(obj, E)
            obj.ContinuationError = E;
        end
         
        function throwIfInvalidFuturesForFetchOutputsScalar(~)
            % Should work for AfterAllFuture, no error thrown.
        end
              
        function state = runImpl(obj, completedPredecessorIdx)            
            % Ignore if all predecessors have finished before
            if all(obj.Triggered)
                state = 'finished';
                return
            end
            
            % Check if a new predecessor has finished
            if ismember(obj.Predecessors(completedPredecessorIdx).State, {'finished', 'failed'}) && ~obj.Triggered(completedPredecessorIdx)
                obj.NumTriggered = obj.NumTriggered + 1;
                obj.Triggered(completedPredecessorIdx) = true;
            end
            
            % Check if all our predecessors are completed, otherwise
            % return, without changing state
            if obj.NumTriggered < numel(obj.Predecessors)
                state = obj.State;
                return
            end
            
                     
            % If any of the predecessors have errors, and PassFuture=false,
            % we don't execute the function, but set an error and return.
            whichPredecessorsHaveErrors = obj.getWhichPredecessorsHaveErrors();
            if ~obj.UseFutureAsArg && any(whichPredecessorsHaveErrors(:))
                obj.ContinuationError = obj.getExceptionFromPredecessorErrors(whichPredecessorsHaveErrors);
                state = 'finished';
                return
            end
            
            % All the predecessors are completed, error free, and we haven't been executed
            % before. Execute the continuation function            
            obj.Triggered = true(1, numel(obj.Predecessors));
            obj.ContinuationStarted = datetime('now', 'TimeZone', 'local');
            
            try
                state = obj.doRun();
            catch E
                ME = MException(message('MATLAB:parallel:future:ExecutionErrors'));
                ME = addCause(ME, E);
                obj.ContinuationError = ME;
                state = 'finished';
            end
        end
                
        function c = getAllOutputsCell(F, states)
            c = cell(size(F));
            for idx = 1:numel(F)
                c{idx} = getOutputArguments(F(idx), states(idx));
            end
        end
        
        function tf = hasAnyErrors(F, ~)
            for idx = 1:numel(F)
                if ~isempty(F(idx).ContinuationError)
                    tf = true;
                    return;
                end
            end
            tf = false;
        end
        
        function c = getAllOutputsScalar(f, ~)
            c = f.OutputArguments;
        end
        
        function c = getAllErrorsCellScalar(f)
            c = {f.ContinuationError};
        end
        
        function setReadScalar(~)
            % Nothing to do for AfterAllFuture
        end
        
        function val = getError(obj, ~)
            val = obj.ContinuationError;
        end
    end
    
    methods(Access=private)
        function state = doRun(obj)

            if obj.UseFutureAsArg
                args = {obj.Predecessors};
            else
                args = cell(1, min([obj.Predecessors.NumOutputArguments]));
                [args{:}] = fetchOutputs(obj.Predecessors);                    
            end

            obj.ContinuationInputArguments = args;
            obj.ContinuationOutputArguments = obj.evaluateFunction(args);

            state = 'finished';
        end
        
        function whichPredecessorsHaveErrors = getWhichPredecessorsHaveErrors(obj)
            for idx = numel(obj.Predecessors):-1:1
                states(idx) = getStateEnum(obj.Predecessors(idx));
            end
            states = reshape(states, size(obj.Predecessors));
            
            whichPredecessorsHaveErrors = arrayfun(@hasAnyErrors, obj.Predecessors, states);
        end
        
        % Creates the Exception set as error on this future, when there are
        % errors with one or more of the predecessors. The errors of the
        % failing predecessors are added as causes.
        function ME = getExceptionFromPredecessorErrors(obj, predecessorsWithErrorsIndices)
            failingIDs = [obj.Predecessors(predecessorsWithErrorsIndices).ID];
            
            ME = MException(message('MATLAB:parallel:future:PrecedingExecutionErrors', int2str(failingIDs)));

            % For each of the failing predecessors, add its error as a cause
            predecessorsWithErrors = obj.Predecessors(predecessorsWithErrorsIndices);
            for idx = 1:numel(predecessorsWithErrors)
                pred = predecessorsWithErrors(idx);
                                
                predException = obj.getExceptionFromError(pred.Error);
                ME = addCause(ME, predException);
            end
        end
    end
end
