%JavaBackedFuture A parallel.internal.queue.Future backed directly by a java task
%   JavaBackedFuture is the base-class for parallel.internal.queue.FevalFuture and
%   parallel.internal.queue.FevalOnAllFuture, and deals with everything to do with the
%   underlying java task, and the input/output caches.

% Copyright 2016-2021 The MathWorks, Inc.

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

    properties (Transient, Dependent, Hidden)
        WorkerID
    end
    properties(Dependent, SetAccess=private)
        % Diary Text produced by execution of future object's function.
        %   (read-only)
        Diary
    end
    properties (Transient, SetAccess = protected)
        %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 JavaBackedFutures
        % for the save javaTask, with the same ID. This unique id is used
        % to tell them apart.
        UID = parallel.internal.queue.Future.InvalidID;
    end

    properties (Transient, GetAccess = protected, SetAccess = immutable)
        JavaTask = [];
    end
    properties (Transient, GetAccess = private, SetAccess = immutable)
        InputCache = [];
        OutputCache = [];
    end
    properties (Access = protected)
        IsDefinitelyFinished = 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, javaQueue);
    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.JavaBackedFuture.DefaultTimeField);
        end
    end
    
    methods (Access = protected)
        function val = getStateEnum(obj)
            import parallel.internal.types.States
            
            persistent FINISHED_STATE
            if isempty(FINISHED_STATE)
                FINISHED_STATE = States.Finished;
            end
            
            
            if obj.IsDefinitelyFinished
                % Shortcut once the state is latched into the 'finished' state.
                val = FINISHED_STATE;
            elseif ~obj.IsValid
                % Default constructed.
                val = States.Unavailable;
            else
                if isempty(obj.JavaTask.getIfNotDisposed())
                    % Shouldn't really get here, since disposal of JavaTask should set
                    % IsDefinitelyFinished.
                    val = FINISHED_STATE;
                    obj.IsDefinitelyFinished = true;
                else
                    stateFromJava = obj.JavaTask.getResource().getState();

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

                    if isTerminal(val)
                        % Dispose the java object and ensure we never check again.
                        obj.JavaTask.disposeIfNecessary();
                        obj.IsDefinitelyFinished = true;
                    end
                end
            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.IsValid
                val = getError(obj.OutputCache, state);
            else
                val = ParallelException.empty();
            end
            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.
           obj.Parent.addFuncToBeDispatchedOnFinish(obj.ID, obj.UID, @obj.handleFutureCompleted);
        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))
                cancellation = ParallelException.hBuildLikeMException(...
                    'parallel:fevalqueue:ExecutionCancelled', ...
                    message('MATLAB:parallel:future:ExecutionCancelled'));
                setCancelled(obj.OutputCache, cancellation);
                task = obj.JavaTask.getIfNotDisposed();
                if ~isempty(obj.Parent)
                    obj.Parent.removeFuncToBeDispatchedOnFinish(obj.ID, obj.UID);
                end
                if ~isempty(task)
                    javaException = ...
                        com.mathworks.toolbox.distcomp.pmode.taskqueue.TaskCancelledByUserException();
                    task.cancel(javaException);
                    obj.JavaTask.disposeIfNecessary();
                end
                
                % 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.JavaTask.getIfNotDisposed())
                % Invalid future or already finished and disposed.
                OK = true;
                return
            end

            % Actually need to wait.
            deadlineExpired = false;
            millisUnit      = java.util.concurrent.TimeUnit.MILLISECONDS;
            javaTask        = obj.JavaTask.getResource();

            while ~predicate(obj) && ~deadlineExpired
                millisToWait    = min(waitGranularity, ...
                                      parallel.internal.timestamps.millisRemaining(deadline));
                deadlineExpired = parallel.internal.timestamps.hasDeadlineExpired(deadline);
                javaTask.await(millisToWait, millisUnit);
                if ~predicate(obj)
                    iJavaLog(4, 'parallel.internal.queue.Future waiting for ID: %d', obj.ID);
                    % 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
    end
    
    methods (Sealed, Hidden, Access = ?parallel.internal.queue.JavaFevalQueue)
        function submitImpl(objOrObjs, queue, javaQueue)
        %SUBMIT submit futures to queue for execution
        
            for idx = 1:numel(objOrObjs)
                obj = objOrObjs(idx);
                

                
                submitScalar(obj, javaQueue);
                obj.Parent = queue;
            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().hGetJavaQueue());
        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
            % Java heap 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. This should allow
                % ~1400 tasks with full diary in heap memory.
                maxDiary = 250000;
            end
            if nargin < 1
                val = maxDiary;
                return 
            end
            maxDiary = 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.
        
            % We don't want an invalid reference to this obj in the QueueFinishedDispatcher
            % map. (That reference usually keeps us alive, this is needed if we're explicitly
            % deleted.)
            if ~isempty(obj.Parent) && isvalid(obj.Parent)
               obj.Parent.removeFuncToBeDispatchedOnFinish(obj.ID, obj.UID);
            end
            
            if ~isempty(obj.JavaTask)
                obj.JavaTask.disposeIfNecessary();
            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 = JavaBackedFuture(outputCacheCtor, jtaskCtor, varargin)
            import parallel.internal.queue.FutureCreation
            
            javaClassName = 'com.mathworks.toolbox.parallel.util.resourcemanagement.DisposableObjectWrapper';
            % default invalid construction
            id = parallel.internal.queue.Future.InvalidID;
            uid = parallel.internal.queue.Future.InvalidID;
            javaTask = {[]};
            inputCache = {[]};
            outputCache = {[]};
            numOutputs = 0;
            parent = [];
            if nargin == 2
                % Do nothing - default construction, leave in invalid state.
            elseif nargin == 4 && ...
                       isa(varargin{1}, javaClassName) && ...
                       isa(varargin{2}, 'parallel.FevalQueue')
                % Construction from the running/pending lists from parallel.FevalQueue.
                origWrapper = varargin{1};
                javaTask    = {origWrapper.duplicate()};
                inputCache  = {parallel.internal.queue.FutureInputHelper(origWrapper.duplicate())};
                outputCache = {outputCacheCtor(origWrapper.duplicate())};
                id          = double(origWrapper.getResource().getID());
                uid = double(parallel.internal.queue.FutureID.reserveUIDs(numel(origWrapper)));
                parent      = varargin{2};
                if id ~= parallel.internal.queue.Future.InvalidID
                    numOutputs = inputCache{1}.NumOutputArguments;
                end
            else
                if nargin == 3 && ...
                    isa(varargin{1}, 'parallel.internal.queue.FutureCreation')
                    % Ordinary public construction
                    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)));

                for idx = 1:numel(id)
                    inputProps = {taskFcn{idx}, numOutputs(idx), argsIn{idx}};
                    data       = parallel.internal.pool.serialize(inputProps);
                    javaTask{idx} = jtaskCtor(id(idx), data, parallel.internal.queue.JavaBackedFuture.maxDiaryOutput());
                    inputCache{idx} = parallel.internal.queue.FutureInputHelper(inputProps{:});
                    outputCache{idx} = outputCacheCtor(javaTask{idx}.duplicate());
                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).JavaTask    = javaTask{idx};    %#ok<AGROW>
                obj(idx).InputCache  = inputCache{idx};  %#ok<AGROW>
                obj(idx).OutputCache = outputCache{idx}; %#ok<AGROW>
            end
        end
    end
    
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iJavaLog - log to the pmode logger. Note we do not use dctSchedulerMessage
% here as that prints output to the command window.
function iJavaLog(level, fmt, varargin)
    try
        loglvl = com.mathworks.toolbox.parallel.pctutil.logging.DistcompLevel.getLevelFromValue(level);
        LOGGER = com.mathworks.toolbox.distcomp.pmode.taskqueue.Log.LOGGER;
        LOGGER.log(loglvl, sprintf(fmt, varargin{:}));
    catch E %#ok<NASGU>
            % ignore any logging problems
    end
end
