%NonCommPoolExecutor
% Parallel pool implementation of the PartitionArrayExecutor interface that
% uses parfeval without communication.
%

%   Copyright 2019 The MathWorks, Inc.

classdef (Sealed) NonCommPoolExecutor < parallel.internal.bigdata.ParallelPoolExecutor

    methods
        function obj = NonCommPoolExecutor(pool)
            obj = obj@parallel.internal.bigdata.ParallelPoolExecutor(pool);
            
            fetchOutputs(parfevalOnAll(pool, ...
                @matlab.bigdata.internal.util.TempFolder.tempPath, ...
                0, tempdir));
        end
    end
    
    % Overridden methods of ParallelPoolExecutor to ignore Connected and
    % IdleTimeout.
    methods
        %CHECKISVALID Check whether this executor is valid.
        function tf = checkIsValid(obj)
            tf = isvalid(obj.Pool);
        end
        
        %KEEPALIVE Notify to the executor that operations have just been
        %performed and it should reset any idle timeouts.
        function keepAlive(~)
        end
    end
    
    % Implementation of the internal interface of ParallelPoolExecutor.
    % See parallel.internal.bigdata.ParallelPoolExecutor for more
    % information on these methods and the respective contract.
    methods (Access = protected)
        % Convert the input task graph into an array of independent tasks
        % that can be executed one by one.
        function [stageTasks, broadcastMap, additionalArguments] ...
                = buildStageTasks(obj, taskGraph, ~, ~)
            import matlab.bigdata.internal.executor.BroadcastMap;
            broadcastMap = BroadcastMap();
            
            shuffleStrategyMap = containers.Map('KeyType', 'char', 'ValueType', 'any');
            stageTasks = matlab.bigdata.internal.executor.convertToIndependentTasks(taskGraph, ...
                'CreateShuffleStorageFunction', @(task) obj.createShuffleStorage(task, shuffleStrategyMap), ...
                'ResolvePartitionStrategyFunction', @obj.resolvePartitionStrategy);
            
            additionalArguments = {};
        end
        
        % Execute one stage task using the pool. This performs all
        % operations that require the pool itself.
        function readFailureSummary = executeStageTaskRemotely(obj, stageTask, broadcastMap, ~)
            partitionStrategy = stageTask.ExecutionPartitionStrategy;
            numExecutorPartitions = obj.numPartitions(partitionStrategy);
            
            import matlab.bigdata.internal.executor.ReadFailureAccumulator;
            readFailureAccumulator = ReadFailureAccumulator(partitionStrategy.MaxNumReadFailures);
            
            import parallel.internal.bigdata.ParfevalJob;
            job = ParfevalJob(obj.Pool, 2, numExecutorPartitions);
            
            import parallel.internal.bigdata.NonCommRunPartitionFunction;
            dataProcessorFactory = stageTask.DataProcessorFactory;
            outputBroadcastKeys = {stageTask.OutputBroadcasts.Id};
            for partitionIndex = 1:numExecutorPartitions
                job.addParfevalTask(NonCommRunPartitionFunction(...
                    dataProcessorFactory, partitionIndex, numExecutorPartitions, ...
                    partitionStrategy, broadcastMap, outputBroadcastKeys));
            end
            
            while ~job.isDone()
                [nextFailureSummary, nextBroadcastOutput] = fetchNext(job);
                readFailureAccumulator.append(nextFailureSummary);
                broadcastMap.append(nextBroadcastOutput);
            end
            
            readFailureSummary = readFailureAccumulator.Summary;
        end
        
        function name = getName(~)
            % Get a user visible string that represents this execution
            % environment.
            name = getString(message('parallel:bigdata:ParallelPoolName'));
        end
    end
    
    methods (Access = private)
        function [writerFactory, readerFactory] = createShuffleStorage(obj, task, shuffleStrategyMap)
            % Create a ShuffleSortStrategy that communicates the output of
            % a task into that task's OutputPartitionStrategy.
            strategy = task.OutputPartitionStrategy;
            strategy = obj.resolvePartitionStrategy(strategy);
            useSharedStorage = true;
            if obj.useMatShuffle()
                [shuffleSortStrategy, writerFactory, readerFactory] ...
                    = obj.createMatArrayShuffleStorage(...
                    task.OutputCommunicationType, strategy, useSharedStorage);
            else
                [shuffleSortStrategy, writerFactory, readerFactory] ...
                    = obj.createSqliteShuffleStorage(...
                    task.OutputCommunicationType, strategy, useSharedStorage);
            end
            shuffleStrategyMap(task.Id) = shuffleSortStrategy; %#ok<NASGU>
        end
    end
    
    methods (Static)
        function out = useMatShuffle(in)
            % Select whether to use MAT files for shuffle storage. If
            % false, default is to use SQLlite.
            persistent state
            if isempty(state)
                state = false;
            end
            if nargout
                out = state;
            end
            if nargin
                state = in;
            end
        end
    end
end
