% InvocationFuture - NotifierFuture which wraps another
% NotifierFuture with an extra function to modify the result.
% Extends NotifierFuture which is defined in C++.
% NotifierFuture class defines:
% - 'Completed' Event
% - abstract hIsFinished() method
% - abstract hGet() method
%
%   InvocationFuture methods:
%      cancel           - Cancel an InvocationFuture
%      addContinuations - Specify function to invoke after each InvocationFuture completes
%      fetchOutputs     - Retrieve all InvocationFutures outputs
%      wait             - Wait for InvocationFutures to complete

% Copyright 2020 The MathWorks, Inc.

classdef InvocationFuture < parallel.internal.apishared.BaseNotifierFuture
       
    properties (Access = private)
        TriggerNotifierFuture
        
        Result = {};
        
        Exception = MException.empty();
        
        Finished = false;
        
        TriggerListener
    end
    
    methods
        function obj = InvocationFuture(triggerNotifierFuture, resultAdaptorFcn, numArgOut)
            % OBJ = INVOCATIONFUTURE(TRIGGERNOTIFIERFUTURE, RESULTSADAPTORFCN, NUMARGOUT)
            % TRIGGERNOTIFIERFUTURE - BaseNotifierFuture which will trigger this
            % future to complete
            % RESULTSADAPTORFCN - Function handle which accepts the result
            % of TRIGGERNOTIFIERFUTURE as an input and the outputs sets the result
            % of this listener.
            % NUMARGOUT - Number of outputs arguments of RESULTSADAPTORFCN
            
            import parallel.internal.cluster.InvocationFuture;
            
            validateattributes(triggerNotifierFuture, {'parallel.internal.apishared.BaseNotifierFuture'}, {'scalar'});
            validateattributes(resultAdaptorFcn, {'function_handle'}, {});
            validateattributes(numArgOut, {'numeric'}, {'scalar','nonnegative','integer'});
                        
            obj.TriggerNotifierFuture = triggerNotifierFuture;
            
            % This check-then-act on the state then attaching a listener,
            % works because the triggerNotifierFuture state is always set on the
            % MATLAB thread.
            if obj.TriggerNotifierFuture.hIsFinished()
                obj.setResult(resultAdaptorFcn, numArgOut);
            else
                obj.TriggerListener = event.listener(obj.TriggerNotifierFuture, ...
                    InvocationFuture.CompletedEventName, @(~, ~) obj.setResult(resultAdaptorFcn, numArgOut));
            end
        end
                
        function cancel(objOrObjs)
            % CANCEL(OBJOROBJS)
            % Cancel future(s). This does not stop the remote invocation,
            % but cancels all client side event handling and callbacks
            arrayfun(@(o) cancelScalar(o), objOrObjs);
        end
        
        function addContinuations(objOrObjs, resultsFcn, errorFcn)
            % ADDCONTINUATIONS(OBJOROBJS, RESULTSFCN, ERRORFCN)
            % Add functions to invoke when this future
            % completes. If the future is already finished the function is
            % invoked at once, otherwise a listener is created to invoke on
            % completion
            % RESULTSFCN is a function handle of the form:
            %   fcn(varargin)
            % and invoked on successful completion
            % ERRORFCN is a function handle of the form:
            %   fcn(exception)
            % and invoked on completion with an error
            
            validateattributes(resultsFcn, {'function_handle'}, {'scalar'});
            validateattributes(errorFcn, {'function_handle'}, {'scalar'});
            
            arrayfun(@(o) addContinuationsScalar(o, resultsFcn, errorFcn), objOrObjs);
        end
        
        function results = fetchOutputs(objOrObjs)
            % RESULTS = FETCHOUTPUTS(OBJOROBJS)
            % Wait for all futures to complete and return results.
            % If any of the futures have an error, instead a collective
            % error is thrown with the individual future errors as causes.
            % Results of each future is returned as a cell array.
            
            wait(objOrObjs);
            
            errors = arrayfun(@(o) o.Exception, objOrObjs, 'UniformOutput', false);
            errors = [errors{:}];
            if ~isempty(errors)
                err = MException(message('MATLAB:parallel:future:ExecutionErrors'));
                for error = errors
                    err = addCause(err, error);
                end
                throw(err);
            end
            
            if isscalar(objOrObjs)
                results = objOrObjs.Result;
            else
                results = arrayfun(@(o) o.Result, objOrObjs, 'UniformOutput', false);
            end
        end
        
        function ok = wait(objOrObjs, timeout)
            % WAIT(OBJOROBJS)
            % Wait for all futures to complete. 
            % OK = WAIT(OBJOROBJS, TIMEOUT) blocks execution for a maximum 
            % of TIMEOUT, where TIMEOUT is a duration. OK is FALSE if 
            % TIMEOUT is exceeded before all futures complete.
            
            import parallel.internal.cluster.InvocationFuture;
            
            if nargin < 2
                timeout = seconds(Inf);
            end
           
            validateattributes(timeout, {'duration'}, {'scalar'});
            deadline = datetime('now') + timeout;
            ok = all(arrayfun(@(o) waitScalar(o, deadline), ...
                   objOrObjs));
        end
    end
    
    methods (Hidden)
        function tf = hIsFinished(obj)
            tf = obj.Finished;
        end
        
        function result = hGet(obj)
            % Clients should not use this method directly as the
            % asynchronous evaluation might not have finished. 
            % Use addContinuations instead.
            
            obj.wait();
            if ~isempty(obj.Exception)
                throw(obj.Exception);
            end
            
            result = obj.Result;
        end
    end
    
    methods (Access = private)
        function setResult(obj, resultAdaptorFcn, numArgOut)
            import parallel.internal.cluster.InvocationFuture;
            
            assert(~obj.hIsFinished(), 'Cannot set result for completion notifier which has already finished');
            try
                % resultAdaptorFcn is evaluated on setting the result to
                % ensure it runs even if nobody ever fetches the result.
                
                % Capture all outputs from trigger future
                triggerResult = obj.TriggerNotifierFuture.hGet();
                if ~iscell(triggerResult)
                    triggerResult = {triggerResult};
                end
                % If resultAdaptorFcn has multiple outputs, capture them
                % all as a cell array in result.
                [result{1:numArgOut}] = resultAdaptorFcn(triggerResult{:});
                obj.Result = result;
            catch E
                obj.Exception = E;
            end
            obj.Finished = true;
            obj.notify(InvocationFuture.CompletedEventName);
        end
        
        function addContinuationsScalar(obj, resultsFcn, errorFcn)
            import parallel.internal.cluster.InvocationFuture;
                        
            if obj.hIsFinished()
                obj.invokeContinuations(resultsFcn, errorFcn);
            else
                obj.addlistener(InvocationFuture.CompletedEventName, ...
                    @(~, ~) obj.invokeContinuations(resultsFcn, errorFcn));
            end
        end
        
        function cancelScalar(obj)
            obj.TriggerListener = [];
            obj.TriggerNotifierFuture = [];
            obj.Finished = true;
            obj.Exception = MException('parallel:fevalqueue:ExecutionCancelled', ...
                message('MATLAB:parallel:future:ExecutionCancelled'));
        end
        
        function ok = waitScalar(obj, deadline)
            import parallel.internal.cluster.InvocationFuture;
            
            if obj.Finished
                ok = true;
                return;
            end
            
            waitingListener = event.listener(obj, InvocationFuture.CompletedEventName, @iNoOpCallback);
            waiter = distcomp.eventwaiter(waitingListener);
            ok = waiter.waitForEvent(@() true, deadline);
        end
            
        function invokeContinuations(obj, resultsFcn, errorFcn)
            if ~isempty(obj.Exception)
                errorFcn(obj.Exception);
            else
                resultsFcn(obj.Result{:});
            end
        end
    end
end 

function iNoOpCallback(~, ~)    
end
