%AbstractContinuationFuture A continuation parallel.internal.queue.Future 
%   AbstractContinuationFuture is the base-class for
%   parallel.internal.queue.AfterEachFuture and parallel.internal.queue.AfterAllFuture, and
%   deals with listening on the completion of preceding futures.

% Copyright 2017-2021 The MathWorks, Inc.

classdef AbstractContinuationFuture < parallel.internal.queue.Future
    
    properties (SetAccess = immutable)
        % The futures this continuation should be executed after
        Predecessors
    end

    properties (SetAccess = immutable, GetAccess = protected)
        UseFutureAsArg
        ContinuationFunction
    end
    
    properties (Access = private)
        % Listeners for the 'FutureCompleted' events of the predecessors.
        % We only add listeners if the predecessor is not finished when the
        % continuation is added. The length of this array will there be
        % between 0 and the number of predecessors
        Listener = event.listener.empty;
        
        ContinuationState = 'queued';
        ContinuationFinished = datetime.empty;
        ContinuationCreated = datetime.empty;
    end
    
    properties (Access = protected)
        
        % Logical array with one element for each predecessor. Each element
        % is initially false, and switched to true when the corresponding
        % predecessor notifies the continuation.
        Triggered;
        
        % Map from the taskID of the predecessor futures to their index in
        % the Predecessors array
        PredecessorMap;
        
        IsDefinitelyFinished = false;
        
        ContinuationStarted = datetime.empty;
    end
      
    properties (Abstract, Access = protected)
        ContinuationError;                
        ContinuationOutputArguments;
        ContinuationInputArguments;
    end

    methods (Abstract, Access = protected)
        state = runImpl(obj, predecessorIdx);
        setCancellationError(obj, cancellationError);
    end
   
    methods (Access = protected)
        function obj = AbstractContinuationFuture(underlyingFutures, continuationFcn, continuationNargsout, useFutureAsArg)
            import parallel.internal.types.States
            
            for idx = 1:numel(underlyingFutures)
                if getStateEnum(underlyingFutures(idx)) == States.Pending
                    error(message('MATLAB:parallel:future:ContinuationOnPending'));
                end
            end
            
            id = parallel.internal.queue.FutureID.buildNextID();          
            obj@parallel.internal.queue.Future(id, continuationNargsout);
            
            obj.ContinuationCreated = datetime('now', 'TimeZone', 'local');
            obj.Predecessors = underlyingFutures;
            obj.PredecessorMap = obj.createAndFillPredecessorMap(underlyingFutures);
            obj.Triggered = false(1, numel(underlyingFutures));
            obj.ContinuationFunction = continuationFcn;
            obj.UseFutureAsArg = useFutureAsArg;
            
            for i = 1:numel(underlyingFutures)
                underlyingFutures(i).appendToSuccessors(obj);
            end
        end
        
        function listenOnPredecessorsAndRun(obj)
            
            % In case the continuation was added after the predecessors
            % finished
            obj.handleFutureCompleted(1:numel(obj.Predecessors));
            
            % Unless the run above finished us, we listen on our
            % predecessors
            if ~strcmp(obj.ContinuationState, 'finished')
                for i = 1:numel(obj.Predecessors)
                    % We only listen on the predecessors which have not
                    % finished.

                    if ~obj.Triggered(i)
                        obj.Predecessors(i).enableCompletionEvent();
                        obj.Listener(end + 1) = event.listener(obj.Predecessors(i), 'FutureCompleted', @obj.handleFutureCompletedHandler);
                        
                        % In case the predecessor finished before we turned
                        % on the listener (but after the first loop above),
                        % and we did not catch the event
                        obj.handleFutureCompleted(i);
                    end
                end
            end           
        end

        function handleFutureCompletedHandler(obj, src, eventData)                       
            % Make sure we've got the right kind of event
            if ~strcmp(eventData.EventName, 'FutureCompleted')
                return
            end
            
            % Make sure the source is one of our predecessors
            if ~isKey(obj.PredecessorMap, src.ID)
                return;
            end
            
            predecessorIdxs = obj.PredecessorMap(src.ID);
            obj.handleFutureCompleted(predecessorIdxs)
        end
        
        function handleFutureCompleted(obj, predecessorIdxs)
            
            for i = 1:numel(predecessorIdxs)
                obj.ContinuationState = obj.runImpl(predecessorIdxs(i));
            end
                        
            % If we're finished, notify our potential successors, and stop
            % listening on our predecessors
            if strcmp(obj.ContinuationState, 'finished')
                obj.ContinuationFinished = datetime('now', 'TimeZone', 'local');
                notify(obj, 'FutureCompleted')
                delete(obj.Listener)
                obj.disableCompletionEvent();                
            end
        end
        
        function out = evaluateFunction(obj, args)
            out = {};
            
            if obj.NumOutputArguments == 0
                feval(obj.Function, args{:});
            else
                out = cell(1, obj.NumOutputArguments);
                [out{:}] = feval(obj.Function, args{:});
            end
        end 
        
        function cancelScalar(obj)
            % No longer listen for predecessors
            delete(obj.Listener);
            
            % Switch state to finished, and add cancellation error (unless
            % we're already finished)
            if ~strcmp(obj.ContinuationState, 'finished')
                obj.ContinuationState = 'finished';
                cancellationError = MException('parallel:fevalqueue:ExecutionCancelled', ...
                    message('MATLAB:parallel:future:ExecutionCancelled'));
                obj.setCancellationError(cancellationError); 
            end
            
            % Notify successors
            notify(obj, 'FutureCompleted');
        end

        function OK = waitScalar(obj, predicate, granularity, deadline)
            if ~obj.IsValid 
                % Invalid future.
                OK = true;
                return
            end

            % Actually need to wait.
            deadlineExpired = false;

            while ~predicate(obj) && ~deadlineExpired
                millisToDeadline = parallel.internal.timestamps.millisRemaining(deadline);
                waitMillis = min(granularity, millisToDeadline);
                deadlineExpired = parallel.internal.timestamps.hasDeadlineExpired(deadline);
                pause(waitMillis/1000);
                if ~predicate(obj)
                    % Ensure wait() is re-entrant by incorporating a drawnow.
                    drawnow();
                end
            end
            OK = predicate(obj);
        end

        function val = getFunction(obj)
            val = obj.ContinuationFunction;
        end

        function val = getInputArguments(obj)
            val = obj.ContinuationInputArguments;
        end
        
        function val = getOutputArguments(obj, ~)
            val = obj.ContinuationOutputArguments;
        end

        function val = getState(obj)
            stateEnum = getStateEnum(obj);
            val = stateEnum.Name;
        end

        function val = getStateEnum(obj)
            import parallel.internal.types.States
            
            if strcmp(obj.ContinuationState, 'finished')
                val = States.Finished;
            elseif strcmp(obj.ContinuationState, 'running')
                val = States.Running;
            elseif strcmp(obj.ContinuationState, 'queued')
                val = States.Queued;
            else
                val = States.Unknown;
            end
        end
        
        function val = getCreateDateTime(obj, ~)
            val = obj.ContinuationCreated;
        end

        function val = getStartDateTime(obj, ~)
            val = obj.ContinuationStarted;
        end

        function val = getFinishDateTime(obj, ~)
            val = obj.ContinuationFinished;
        end
        
        function enableCompletionEvent(~)
           % Nothing to do for AbstractContinuationFutures
        end
        
        function disableCompletionEvent(obj)
            for i = 1:numel(obj.Predecessors)
                obj.Predecessors(i).disableCompletionEvent();
            end
        end
        
        function exception = getExceptionFromError(~, error)     
            % If the error is a single (subclass of) MException we return
            % it
            if isa(error, 'MException')
                exception = error;
                return
            end
            
            % Otherwise, it's a cell array of (subclasses of) MException.
            % We create a dummy exception, and add them as causes.
            assert(iscell(error), 'Error must be an MException or cell array of MExceptions');
            dummyException = MException(message('MATLAB:parallel:future:ExecutionErrors'));
            for i = 1:numel(error)
                assert(isa(error{i}, 'MException'), 'Error must be an MException or cell array of MExceptions');
                % We cannot add ParallelException.empty as a cause
                if ~isempty(error{i})
                    dummyException = addCause(dummyException, error{i});
                end
            end
            exception = dummyException;
        end
    end
    
    methods(Static, Access = private)
        function predecessorMap = createAndFillPredecessorMap(underlyingFutures)
            predecessorMap = containers.Map('KeyType', 'int32', 'ValueType', 'any');
            for i = 1:numel(underlyingFutures)
                predId = underlyingFutures(i).ID;
                if isKey(predecessorMap, predId)
                    existingPredecessorIndices = predecessorMap(predId);
                    predecessorMap(predId) = [existingPredecessorIndices, i];
                else
                    predecessorMap(predId) = i;
                end
            end
        end
    end
    
    methods(Static)
        function parser = parseInput(F, fcn, nout, varargin)
            parser = inputParser;
            parser.KeepUnmatched = false;
            parser.PartialMatching = false;
            
            addRequired(parser, 'F', @(x) validateattributes(x, {'parallel.internal.queue.Future'}, {'nonempty'}));
            addRequired(parser, 'fcn', @(x) validateattributes(x, {'function_handle'}, {'scalar'}));
            addRequired(parser, 'nout', @(x) validateattributes(x, {'numeric'}, {'finite', 'scalar', 'integer', 'real', 'nonnegative'}));
            addParameter(parser, 'PassFuture', false, @(x) validateattributes(x, {'logical'}, {'scalar'}));

            parse(parser, F, fcn, nout, varargin{:});
        end
    end
end
