% CppBackedFuture abstract parent class for C++ based parfeval futures

% Copyright 2021 The MathWorks, Inc.

classdef (Abstract) CppBackedFuture < parallel.internal.queue.Future

    properties (Transient, Dependent, Hidden)
        % The worker on which we were executed. Empty until the future is
        % finished.
        WorkerID
    end
    
    properties(Dependent, SetAccess=private)
        % Diary Text produced by execution of future object's function.
        % (read-only)
        Diary
    end
    
    properties (Transient, SetAccess = private)
        % Parent FevalQueue containing this future. Empty if the future has
        % not yet been submitted. (read-only)
        Parent = parallel.FevalQueue.empty();
    end
    
    properties(Transient, Access=private)
        % Accessing the pool.FevalQueue creates multiple CppBackedFutures
        % for the save cppTask, with the same ID. This unique id is used
        % to tell them apart.
        UID = parallel.internal.queue.Future.InvalidID;
    end


    properties (Transient, Access = private)
        % The underlying C++ task. Empty after the
        % outputs have been read and moved to the output cache.
        CppTask = [];
        
        CompletionEventEnabled = false;
    end
    
    % Contains the unserialized input and output to/from the
    % parfeval. Once these are populated, we no longer need to go to
    % the underlying C++ task to get the data.
    properties (Transient, GetAccess = private, SetAccess = immutable)
        InputCache = [];
        OutputCache = [];
    end
    
    properties (Access = protected)
        % Set to true when this future is finished, to avoid more expensive
        % checks.
        IsDefinitelyFinished = false;
        
        % Set to true when this future is read, to avoid more expensive
        % checks.
        IsDefinitelyRead = false;

        % true only if this future is known to be complete and error-free. This enables
        % 'hasAnyErrors' to be much more efficient for many cases.
        CompletedErrorFree = false;
    end
    
    properties (Constant, GetAccess = private)
        % Avoid building 'datetime.empty()' repeatedly as it is expensive.
        DefaultTimeField = datetime.empty();
    end
    
    methods (Abstract, Access = protected)
        submitScalar(obj, cppQueue);
    end
    
    methods (Access = private)
        function val = getIfValid(obj, fctor, elseValue)
            if obj.IsValid
                val = fctor();
            else
                val = elseValue;
            end
        end
        
        function val = getTimeField(obj, cacheFcn, state)
            val = getIfValid(obj, ...
                             @() cacheFcn(obj.OutputCache, state), ...
                             parallel.internal.queue.CppBackedFuture.DefaultTimeField);
        end
    end
    
    methods (Access = protected)
        
        function [helper, indexMap] = getFetchNextHelper(objOrObjs, ids)
            % Build a C++ FetchNextHelper for use with this array of tasks.
            
            % Cache the FetchNextHelper in case fetchNext is called again
            % with the exact same futures
            persistent IDS HELPER INDEXMAP
            
            if ~isequal(ids, IDS)
                queuesCell = {objOrObjs.Parent};
                queueIsEmpty = cellfun(@isempty, queuesCell);
                if all(queueIsEmpty)
                    throwAsCaller(MException(message(...
                        'MATLAB:parallel:future:WaitOnUnsubmittedFuture')));
                else
                    nonEmptyQueues = [queuesCell{:}];
                    queue = nonEmptyQueues(1);
                    for idx = 2:numel(nonEmptyQueues)
                        if queue ~= nonEmptyQueues(idx)
                            throwAsCaller(MException(message(...
                                'MATLAB:parallel:future:FetchNextUnequalQueues')));
                        end
                    end
                end

                % To handle disposed CppTasks, the tasks array passed the
                % fetchNextHelper does not include them.
                indexMap = find(~cellfun(@isempty, {objOrObjs.CppTask}));
                tasks  = [objOrObjs(indexMap).CppTask];
                HELPER = parallel.internal.queue.FetchNextHelper(tasks);
                IDS    = ids;
                INDEXMAP = indexMap;
            end
            
            helper = HELPER;
            indexMap = INDEXMAP;
        end
        
        function val = getStateEnum(obj)
            import parallel.internal.types.States            
            
            if obj.IsDefinitelyFinished
                % Shortcut once the state is latched into the 'finished' state.
                val = States.Finished;
            elseif ~obj.IsValid
                % Default constructed.
                val = States.Unavailable;
            else
                if isempty(obj.CppTask)
                    % Shouldn't really get here, since disposal of CppTask should set
                    % IsDefinitelyFinished.
                    val = States.Finished;
                    obj.IsDefinitelyFinished = true;
                else
                    stateFromCpp = obj.CppTask.getState();

                    if isempty(obj.Parent) || hIsValid(obj.Parent)
                        % Use cpp state if unsubmitted or valid parent.
                        val = States.fromName(char(stateFromCpp));
                    else
                        switch char(stateFromCpp)
                          case {'running', 'queued'}
                            val = States.Failed;
                          case 'finished'
                            val = States.Finished;
                          otherwise
                            % Do not expect to get here.
                            val = States.Finished;
                        end
                    end

                    if isTerminal(val)
                        obj.IsDefinitelyFinished = true;
                    end
                end
            end
        end
        
        function val = getIsRead(obj)
            if obj.IsDefinitelyRead
                val = true;
                return;
            end
            
            if ~isempty(obj.CppTask)
                val = obj.CppTask.getReadFlag();
                return
            end
            
            val = false;
        end
        
        function setReadScalar(obj)
            if ~obj.IsDefinitelyRead
                % Don't need to hold on to this any longer, but we must
                % flip the flag first, in case it's currently in use by a
                % fetchNextHelper
                
                if ~isempty(obj.CppTask)
                    obj.CppTask.setReadFlag(true);
                    obj.CppTask = [];
                end
                obj.IsDefinitelyRead = true;
            end
        end
        
        function val = getFunction(obj)
            val = getIfValid(obj, @() obj.InputCache.Function, []);
        end
        
        function val = getInputArguments(obj)
            val = getIfValid(obj, @() obj.InputCache.InputArguments, {});
        end
        
        function val = getState(obj)
            stateEnum = getStateEnum(obj);
            val = stateEnum.Name;
        end
        
        function val = getDiary(obj)
            val = getIfValid(obj, @() getDiary(...
                obj.OutputCache, getStateEnum(obj)), '');
        end
        
        function val = getError(obj, state)
            val = getIfValid(obj, @() getError(obj.OutputCache, state), ...
                ParallelException.empty());
                        
            if obj.IsDefinitelyFinished && isempty(val)
                obj.CompletedErrorFree = true;
            end
        end
        
        function val = getOutputArguments(obj, state)
        % Note this method doesn't use 'getIfValid' purely as a performance
        % optimization.
            if obj.IsValid
                val = getOutputArguments(obj.OutputCache, state);
            else
                val = {};
            end
        end
        
        function val = getCreateDateTime(obj, state)
            val = getTimeField(obj, @getCreateTime, state);
        end
        
        function val = getStartDateTime(obj, state)
            val = getTimeField(obj, @getStartTime, state);
        end
        
        function val = getFinishDateTime(obj, state)
            val = getTimeField(obj, @getFinishTime, state);
        end
        

        function enableCompletionEvent(obj)
           % Allow us to notify our continuations (if present) when
           % this future completes.
           if ~isempty(obj.Parent)
               obj.doEnableCompletionEvent(obj.Parent);
           end
        end
        
        function disableCompletionEvent(obj)
            if ~any(event.hasListener(obj,'FutureCompleted'))
                if ~isempty(obj.Parent)
                    obj.Parent.removeFuncToBeDispatchedOnFinish(obj.ID, obj.UID);
                end                
            end
        end
        
        function cancelScalar(obj)
            % Unwrapping the 'cancel' method.
            if obj.IsValid && ~isTerminal(getStateEnum(obj))
                
                if ~isempty(obj.Parent)
                    obj.Parent.removeFuncToBeDispatchedOnFinish(obj.ID, obj.UID);
                end
                
                task = obj.CppTask;
                if ~isempty(task)
                    task.cancel();
                    obj.CppTask = [];
                end

                cancellationMessage = ParallelException.hBuildLikeMException(...
                    'parallel:fevalqueue:ExecutionCancelled', ...
                    message('MATLAB:parallel:future:ExecutionCancelled'));
                setCancelled(obj.OutputCache, cancellationMessage);
                               
                % Notify continuations
                obj.IsDefinitelyFinished = true;
                notify(obj, 'FutureCompleted');
            end
        end

        function OK = waitScalar(obj, predicate, waitGranularity, deadline)
            % Unwrapping the 'wait' method.
            if isempty(obj.Parent)
                error(message('MATLAB:parallel:future:WaitOnUnsubmittedFuture'));
            end
        
            if ~obj.IsValid || isempty(obj.CppTask)
                % Invalid future or already finished and disposed.
                OK = true;
                return
            end

            % Actually need to wait.
            deadlineExpired = false;
            while ~predicate(obj) && ~deadlineExpired
                millisToWait    = min(waitGranularity, ...
                                      parallel.internal.timestamps.millisRemaining(deadline));
                deadlineExpired = parallel.internal.timestamps.hasDeadlineExpired(deadline);
                obj.CppTask.await(millisToWait);
                if ~predicate(obj)
                    % Ensure wait() is re-entrant by incorporating a drawnow.
                    drawnow();
                end
            end
            
            % Yield one final time to ensure all events directly generated by
            % code in the PARFEVAL wait are executed before we give control back
            % to the user.
            parallel.internal.pool.yield();
            OK = predicate(obj);
        end
        
        function broadcast(obj, cppQueue)
            cppQueue.broadcast(obj.CppTask);
        end
        
        function submitToQueue(obj, cppQueue)
            cppQueue.submit(obj.CppTask);
        end
    end
    
    methods (Sealed, Hidden, Access = ?parallel.internal.queue.CppFevalQueue)
        function submitImpl(objOrObjs, queue, cppQueue)
        %SUBMIT submit futures to queue for execution
        
            for idx = 1:numel(objOrObjs)
                obj = objOrObjs(idx);
                obj.Parent = queue;
                submitScalar(obj, cppQueue);
            end
        end
    end
    
    methods (Sealed, Hidden)
        function submit(objOrObjs, queue)
        % Separate submission - used only in tests.
            if any(~cellfun(@isempty, {objOrObjs.Parent})) || ...
                    any(~hIsValid(objOrObjs))
                error(message('MATLAB:parallel:future:InvalidStateForSubmission'));
            end

            submitImpl(objOrObjs, queue, queue.hGetQueueImpl().hGetCppQueue());
        end
    end 
    
    methods (Hidden, Access = {?parallel.internal.queue.CppFevalQueue, ...
            ?parallel.internal.queue.CppBackedFuture})
        function doEnableCompletionEvent(obj, queue)
           % Allow us to notify our continuations (if present) when
           % this future completes.
           
           if ~obj.CompletionEventEnabled
               obj.CompletionEventEnabled = true;
               obj.CppTask.enableCallMatlabOnCompletion(queue.QueueID);
               queue.addFuncToBeDispatchedOnFinish(obj.ID, obj.UID, @obj.handleFutureCompleted);
           end
        end
    end

    methods (Hidden, Static)
        function val = maxDiaryOutput(val)
            % Set or get the maximum dairy we will store in a single task.
            % Diary produced beyond this limit will result in the oldest diary being
            % discarded. This is done to address the issue of running out of
            % memory (g1914592). This is hidden but not private to
            % allow users to edit this if absolutely necessary.
            persistent maxDiary;
            if isempty(maxDiary)
                % Default limit is 250000 characters.
                maxDiary = 250000;
            end
            if nargin < 1
                val = maxDiary;
                return 
            end
            maxDiary = val;
        end
        
        function val = maxNumRetryAttempts(val)
            % Set or get the maximum number of times a this task will be retried on a
            % new worker after failing (failures are often caused by the
            % task crashing on a worker). This is hidden but not private to
            % allow users to edit this if absolutely necessary.
            persistent maxRetries;
            if isempty(maxRetries)
                % Default value. Essentially infinite, since the max pool
                % size will be 1000 for the foreseeable future
                maxRetries = 10000;
            end
            if nargin < 1
                val = maxRetries;
                return 
            end
            maxRetries = val;
        end
    end
    
    methods
        function val = get.Diary(obj)
            val = getDiary(obj);
        end
        function val = get.WorkerID(obj)
            val = getIfValid(obj, @() getWorkers(...
                obj.OutputCache, getStateEnum(obj)), []);
        end

        function delete(obj)
            %DELETE Object destructor
            % Release resources associated with this Future.
            % 'obj' is guaranteed always to be a scalar instance, so we must defend only
            % against invalid objects.
            
            obj.CppTask = [];
                    
            if ~isempty(obj.Parent)
                obj.Parent.removeFuncToBeDispatchedOnFinish(obj.ID, obj.UID);
            end
        end
        
        function handleFutureCompleted(obj, id)
            % We only care about events from our predecessors
            if id == obj.ID
                notify(obj, 'FutureCompleted')
            end
        end
        
        function obj = CppBackedFuture(outputCacheCtor, cppTaskCtor, varargin)
            import parallel.internal.queue.FutureCreation
            
            % default invalid construction
            id = parallel.internal.queue.Future.InvalidID;
            uid = parallel.internal.queue.Future.InvalidID;
            cppTask = {[]};
            inputCache = {[]};
            outputCache = {[]};
            numOutputs = 0;
            parent = [];
            
            if nargin == 2
                % Do nothing - default construction, leave in invalid state.
            elseif nargin == 4 && ...
                       isa(varargin{1}, 'parallel.internal.queue.FevalTask') && ...
                       isa(varargin{2}, 'parallel.FevalQueue')
                % Construction from the running/pending lists from parallel.internal.queue.CppFevalQueue.
                cppTask = varargin{1};
                inputCache = {parallel.internal.queue.CppFutureInputHelper(cppTask)};
                outputCache = {outputCacheCtor(cppTask)};
                id  = double(cppTask.getId());
                uid = double(parallel.internal.queue.FutureID.reserveUIDs(numel(cppTask)));
                parent = varargin{2};
                if id ~= parallel.internal.queue.Future.InvalidID
                    numOutputs = inputCache{1}.NumOutputArguments;
                end
                cppTask = {cppTask};
            else
                if nargin == 3 && isa(varargin{1}, 'parallel.internal.queue.FutureCreation')
                    % Ordinary public construction, get the futureCreation
                    % with the parsed arguments
                    fc = varargin{1};
                else
                    % This must be the undocumented direct-construction route, so we need to resolve
                    % the arguments while permitting vectorized creation.
                    try
                        fc = FutureCreation.resolveArgs(varargin{:});
                    catch E
                        throwAsCaller(E);
                    end
                end

                taskFcn    = fc.RequestFcn;
                numOutputs = fc.NumArgsOut;
                argsIn     = fc.ArgsIn;
                parent     = parallel.FevalQueue.empty();

                % Vectorized construction
                id = double(parallel.internal.queue.FutureID.reserveIDs(numel(taskFcn)));
                uid = double(parallel.internal.queue.FutureID.reserveUIDs(numel(taskFcn)));
                
                cppTask = cell(1, numel(id));
                for idx = 1:numel(id)
                    inputProps = {taskFcn{idx}, numOutputs(idx), argsIn{idx}};
                    
                    cppTask{idx} = cppTaskCtor(id(idx), inputProps, ...
                        parallel.internal.queue.CppBackedFuture.maxDiaryOutput(), ...
                        parallel.internal.queue.CppBackedFuture.maxNumRetryAttempts());
                    
                    inputCache{idx} = parallel.internal.queue.CppFutureInputHelper(inputProps{:});
                    outputCache{idx} = outputCacheCtor(cppTask{idx});
                end
            end
            
            obj@parallel.internal.queue.Future(id, numOutputs);
                     
            for idx = 1:numel(id)
                obj(idx).UID         = uid(idx);         %#ok<AGROW>
                obj(idx).Parent      = parent;           %#ok<AGROW>
                obj(idx).CppTask     = cppTask{idx};    %#ok<AGROW>
                obj(idx).InputCache  = inputCache{idx};  %#ok<AGROW>
                obj(idx).OutputCache = outputCache{idx}; %#ok<AGROW>
            end
        end
    end
    
end
