% CppFevalQueue C++ backed FevalQueue

% Copyright 2021 The MathWorks, Inc.

classdef CppFevalQueue < parallel.internal.queue.IFevalQueueImpl

    properties (GetAccess = private, SetAccess = immutable, Transient)
        CppQueueObj
    end

    properties (Transient, Hidden, Access = private)
        % Holds on to functions to be run when futures complete, used for
        % afterEach/All
        QueueFinishedDispatcher
        CallbacksEnabled = false;
    end
    
    methods (Access = ?parallel.internal.pool.CppBackedSession)
        function obj = CppFevalQueue(clientSession)
            % Instances of this class may exist on exit, so mlock
            % to avoid warnings.
            mlock;
            [~,u] = pctconfig;
            obj.CppQueueObj = parallel.internal.queue.TaskQueueImpl( ...
                clientSession, u.parfevalMaxWorkerQueueLength);
            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

            cppTaskList = obj.CppQueueObj.getAllOutstandingTasks();

            numTasks = numel(cppTaskList);
            taskCell = cell(1, numTasks);
            for idx = 1:numTasks
                thisTask    = cppTaskList(idx);
                if strcmp(thisTask.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.CppFevalFuture(fc);
                if obj.CallbacksEnabled
                    t.doEnableCompletionEvent(parentQueue);
                end
                submitImpl(t, parentQueue, obj.CppQueueObj);
                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.CppFevalOnAllFuture(fc);
                if obj.CallbacksEnabled
                    bt.doEnableCompletionEvent(parentQueue);
                end
                submitImpl(bt, parentQueue, obj.CppQueueObj);
                bt = parallel.Future.hBuild(bt);
            catch E
                throw(E);
            end
        end
        
        function wasEnabled = toggleCallbacks(obj, newValue, ~)
            wasEnabled = obj.CallbacksEnabled;
            obj.CallbacksEnabled = newValue;
        end
        
        function f = getFutureByID(obj, parentQueue, id)
            tasks = obj.CppQueueObj.getAllOutstandingTasks();
            idx = find(arrayfun(@(t) t.getId(), tasks) == id);
            if isempty(idx)
                f = parallel.FevalFuture.empty();
            else
                task = tasks(idx);
                taskConstructor = iGetConstructor(task);
                f = taskConstructor(task, parentQueue);
            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;
            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
    end
    
    methods (Hidden)
        function tf = hIsValid(obj)
            tf = ~isempty(obj.CppQueueObj);
        end
        
        function cq = hGetCppQueue(obj)
            cq = obj.CppQueueObj;
        end
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Given a cpp task, work out the correct MATLAB constructor to use.
function ctor = iGetConstructor(cppTask)
isBroadcast = cppTask.isBroadcast();
if isBroadcast
    ctor = @(varargin) parallel.Future.hBuild(...
        parallel.internal.queue.CppFevalOnAllFuture(varargin{:}));
else
    ctor = @(varargin) parallel.Future.hBuild(...
        parallel.internal.queue.CppFevalFuture(varargin{:}));
end
end


