% JavaFevalQueue Java backed IFevalQueueImpl implementation

% Copyright 2021 The MathWorks, Inc.
classdef JavaFevalQueue < parallel.internal.queue.IFevalQueueImpl
    
    properties (GetAccess = private, SetAccess = immutable, Transient)
        JavaQueueObj
        WaitHelper
        ParentQueueID;
    end
    
    properties (Access = private, Transient)
        CallbackHelper
        % Holds on to functions to be run when futures complete, used for
        % afterEach/All
        QueueFinishedDispatcher
    end
    
    methods(Access = ?parallel.internal.pool.JavaBackedSession)
        function obj = JavaFevalQueue(javaSession, parentQueueId)
            % Instances of this class may exist on exit, so mlock
            % to avoid warnings.
            mlock;
            obj.ParentQueueID = parentQueueId;
            [~,u] = pctconfig;
            obj.JavaQueueObj = javaSession.getTaskQueue(u.parfevalMaxWorkerQueueLength);
            obj.WaitHelper = ...
                com.mathworks.toolbox.distcomp.pmode.taskqueue.TaskCompletionWaiter.build(obj.JavaQueueObj);
            obj.CallbackHelper = []; % disabled by default
            obj.QueueFinishedDispatcher = []; % disabled by default
        end
    end
    
    methods (Access = {?parallel.FevalQueue, ?parallel.internal.queue.IFevalQueueImpl})
        function taskVector = getFuturesInState(obj, parentQueue, state)
            if ~hIsValid(obj)
                taskVector = parallel.FevalFuture.empty();
                return
            end
            % Get a list of outstanding tasks - we are handed new
            % duplicates.
            javaTaskList = obj.JavaQueueObj.getOutstandingTasks();
            disposer     = onCleanup(@() iDisposeList(javaTaskList));
            
            numTasks = double(javaTaskList.size());
            taskCell = cell(1, numTasks);
            for idx = 1:numTasks
                thisTask    = javaTaskList.get(idx-1);
                if isequal(char(thisTask.getResource().getState()), state)
                    taskConstructor = iGetConstructor(thisTask);
                    taskCell{idx}   = taskConstructor(thisTask, parentQueue);
                end
            end
            taskVector = [ taskCell{:} ];
            
            if isempty(taskVector)
                taskVector = parallel.FevalFuture.empty();
            else
                % Return futures in ID order. In the absence of separate construction and
                % submission, this will turn out to be submission order.
                taskIDs      = [taskVector.ID];
                [~, reorder] = sort(taskIDs);
                taskVector   = taskVector(reorder);
            end
        end
        
        function t = parfeval(obj, parentQueue, fcn, numArgsOut, varargin)
            import parallel.internal.queue.FutureCreation
            
            fc = FutureCreation.resolveScalarCreationArgs(fcn, numArgsOut, varargin);
            try
                t = parallel.internal.queue.FevalFuture(fc);
                submitImpl(t, parentQueue, obj.JavaQueueObj);
                t = parallel.Future.hBuild(t);
            catch E
                throw(E);
            end
        end
        
        function bt = parfevalOnAll(obj, parentQueue, fcn, numArgsOut, varargin)
            import parallel.internal.queue.FutureCreation
                      
            fc = FutureCreation.resolveScalarCreationArgs(fcn, numArgsOut, varargin);
            try
                bt = parallel.internal.queue.FevalOnAllFuture(fc);
                submitImpl(bt, parentQueue, obj.JavaQueueObj);
                bt = parallel.Future.hBuild(bt);
            catch E
                throw(E);
            end
        end
        
        function wasEnabled = toggleCallbacks(obj, newVal)
            wasEnabled = ~isempty(obj.CallbackHelper);
            if newVal ~= wasEnabled
                if newVal
                    obj.CallbackHelper = ...
                        com.mathworks.toolbox.distcomp.pmode.taskqueue.CallMatlabOnTaskCompletion.build(...
                        obj.JavaQueueObj, obj.ParentQueueID);
                else
                    obj.CallbackHelper.dispose();
                    obj.CallbackHelper = [];
                end
            end
        end
        
        function notifyFutureCompleted(obj, taskId)
            if ~isempty(obj.QueueFinishedDispatcher)
                obj.QueueFinishedDispatcher.dispatch(taskId);
            end
        end
        
        function turnOnQueueFinishedDispatcher(obj)
            if isempty(obj.QueueFinishedDispatcher) % Only turn on once
                obj.QueueFinishedDispatcher = parallel.internal.queue.QueueFinishedDispatcher;
                obj.toggleCallbacks(true);
            end
        end
        
        % Add function to be called to the QueueFinishedDispatcher for a
        % given task. When this task finishes, the function is called.
        function addFuncToBeDispatchedOnFinish(obj, ID, UID, func_handle)
            obj.turnOnQueueFinishedDispatcher();
            obj.QueueFinishedDispatcher.addFuncToDispatch(ID, UID, func_handle);
        end
        
        function removeFuncToBeDispatchedOnFinish(obj, ID, UID)
            % Only remove if this has been turned on, cf. functions above
            if(~isempty(obj.QueueFinishedDispatcher))
                obj.QueueFinishedDispatcher.removeFuncToDispatch(ID, UID);
            end
        end
        
        function f = getFutureByID(obj, parentQueue, id)
            javaTaskWrapper = obj.JavaQueueObj.getOutstandingTask(id);
            if isempty(javaTaskWrapper)
                f = parallel.FevalFuture.empty();
                return;
            end
            disposer = onCleanup(@() javaTaskWrapper.dispose());
            taskConstructor = iGetConstructor(javaTaskWrapper);
            f = taskConstructor(javaTaskWrapper, parentQueue);
        end
    end
    
    methods (Hidden)
        function jq = hGetJavaQueue(obj)
            jq = obj.JavaQueueObj;
        end
        
        function waiter = hGetWaiter(obj)
            waiter = obj.WaitHelper;
        end
        
        function tf = hIsValid(obj)
            tf = ~isempty(obj.JavaQueueObj) && obj.JavaQueueObj.isValid();
        end
    end
    
    methods
        function delete(obj)
            %DELETE Object destructor.
            
            % Simply clean up our listener objects.
            if ~isempty(obj.WaitHelper)
                obj.WaitHelper.dispose();
            end
            if ~isempty(obj.CallbackHelper)
                obj.CallbackHelper.dispose();
            end
        end
    end
end


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Given a java task wrapper, work out the correct MATLAB constructor to use.
function ctor = iGetConstructor(javaTask)
isBroadcast = isa(javaTask.getResource(), ...
    'com.mathworks.toolbox.distcomp.pmode.taskqueue.FevalBroadcastTask');
if isBroadcast
    ctor = @(varargin) parallel.Future.hBuild(...
        parallel.internal.queue.FevalOnAllFuture(varargin{:}));
else
    ctor = @(varargin) parallel.Future.hBuild(...
        parallel.internal.queue.FevalFuture(varargin{:}));
end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Dispose of a java array of wrappers
function iDisposeList(javaTaskList)
for ii = 0:(double(javaTaskList.size())-1)
    try
        javaTaskList.get(ii).dispose();
    catch E %#ok<NASGU>
    end
end
end
