%ParallelMapReducer
% matlab.mapreduce.ParallelMapReducer is the MapReducer class that represents
% executing mapreduce on a parallel pool. MapReducer objects can be constructed
% using the MAPREDUCER function.
%
%   matlab.mapreduce.ParallelMapReducer properties:
%      Pool - The corresponding parallel pool object to be used by this MapReducer.
%
% See also mapreducer, parallel.Pool.

%   Copyright 2014-2022 The MathWorks, Inc.

classdef ParallelMapReducer < matlab.mapreduce.MapReducer
    properties (SetAccess = immutable)
        Pool; % The corresponding parallel pool object to be used by this MapReducer.
    end
    
    properties (Constant, Access = private)
        MaxNumPartitions = 100; % There is no point partitioning into more than 100 as the display
                                % cannot do finer than 1% at a time.
    end
    
    methods
        % Construct a ParallelMapReducer that performs mapreduce on a
        % parallel.Pool.
        %
        function this = ParallelMapReducer(pool)
            if nargin >= 1
                validateattributes(pool, {'parallel.Pool'}, {'nonempty'}, 'mapreduce', 'Pool');
                if ~isvalid(pool)
                    error(message('MATLAB:class:InvalidHandle'));
                elseif ~pool.Connected
                    error(message('parallel:lang:pool:InvalidPool'));
                end
            else
                pool = gcp('nocreate');
                if isempty(pool)
                    pool = parpool;
                end
            end

            % If the given pool is based on a cluster, we require
            % MPI to communicate data inbetween machines.
            mpiRequired = matlab.mapreduce.ParallelMapReducer.isMPIRequired(pool);
            if mpiRequired
                spmdIsEnabled = pool.hInitializeSpmd();
                if ~spmdIsEnabled
                    error(message('parallel:mapreduce:SpmdEnabledRequired'));
                end
            end
            this.Pool = pool;
        end
    end
    
    methods (Hidden)
        % Check if this MapReducer is valid to be used as the current MapReducer.
        function isValid = isValidForCurrentMapReducer(obj)
            isValid = strcmp(obj.ObjectVisibility, 'On') && isvalid(obj.Pool) && obj.Pool.Connected;
        end
        
        % The internal entry point to mapreduce for ParallelMapReducer.
        function outds = execMapReduce(obj, inputDS, mapFcn, reduceFcn, settings)
            import parallel.internal.mapreduce.SqliteShuffleSortStrategy;
            import parallel.internal.mapreduce.SqliteRemoteCopyShuffleSortStrategy;
            import parallel.internal.mapreduce.ReduceStageStrategy;
            import parallel.internal.mapreduce.Settings;
            
            % LocalThreads does not support mapreduce
            if isa(obj.Pool, "parallel.ThreadPool")
                error(message('parallel:lang:pool:UnsupportedFeature'));
            end
            
            % Without a pool, this object becomes invalid and cannot be
            % used for mapreduce.
            if ~obj.Pool.Connected
                error(message('parallel:lang:pool:InvalidPool'));
            end
            
            % We do not support any datastore that does not support
            % Partitionable.
            matlab.io.datastore.internal.validateParallelBigdata(...
                inputDS, "Partitionable", obj.getName());
            
            try
                settings = Settings(settings);
                reduceStageStrategy = ReduceStageStrategy(obj.Pool.NumWorkers, settings.getOutputFolderAsLocalPath());
                numReduceTasks = reduceStageStrategy.getNumReduceTasks();
                if isa(obj.Pool.Cluster, 'parallel.cluster.Local')
                    shuffleSortStrategy = SqliteShuffleSortStrategy(numReduceTasks);
                else
                    shuffleSortStrategy = SqliteRemoteCopyShuffleSortStrategy(numReduceTasks);
                end
                
                outds = obj.pExecMapReduce(inputDS, mapFcn, reduceFcn,...
                    shuffleSortStrategy, reduceStageStrategy, settings);
            catch err
                if isa(err, 'ParallelException')
                    % Exceptions that come from the users mapper or reducer
                    % code shouldn't display the stack in between the call
                    % to mapreduce and the user code.
                    throw(ParallelException.hBuildRemoteParallelException(err, mfilename));
                else
                    throw(err);
                end
            end
        end
    end
    
    methods (Hidden, Static)
        function tf = isMPIRequired(pool)
            % Check if mapreduce/tall on the given pool requires MPI.
            tf = ~isa(pool, 'parallel.ThreadPool');
        end
        
        function isSupported = isPoolSupported(pool)
            % Check if the given pool is supported by mapreduce/tall
            
            % If the given pool is based on a cluster, we require
            % MPI to communicate data inbetween machines. Otherwise we only
            % require that the pool is valid.
            mpiRequired = matlab.mapreduce.ParallelMapReducer.isMPIRequired(pool);
            isSupported = isvalid(pool) && pool.Connected ...
                && (~mpiRequired || pool.SpmdEnabled);
        end
    end
    
    methods (Access = private)
        % The implementation of mapreduce for ParallelMapReducer.
        function outds = pExecMapReduce(this, inputDS, mapFcn, reduceFcn,...
                shuffleSortStrategy, reduceStageStrategy, settings)
            import parallel.internal.mapreduce.MapReduceRemoteCache;
            
            if strcmp(settings.Display, 'On')
                iDisplayHeader;
            end
            
            outputFolderScheme = settings.getOutputFolderScheme();
            if ~strcmp(outputFolderScheme, 'file')
                error(message('parallel:mapreduce:UnsupportedOutputSchemeForParallelMapReducer', outputFolderScheme, settings.OutputFolder));
            end
            
            outputFolder = settings.getOutputFolderAsLocalPath();
            if ~exist(outputFolder, 'dir')
                iCreateOutputFolder(outputFolder);
            end
            
            resultTimeStamp = char(datetime("now", "Format", "dd-MMM-yyyy_HH-mm-ss_SSS"));
            
            % All details to be cached on the worker
            remoteMapReduceDetails = MapReduceRemoteCache(inputDS, mapFcn, reduceFcn, shuffleSortStrategy);
            
            % Map stage
            this.pExecuteMapTasks(inputDS, remoteMapReduceDetails, shuffleSortStrategy, reduceStageStrategy, settings);
                        
            % Reduce stage
            outputFiles = this.pExecuteReduceTasks(remoteMapReduceDetails, resultTimeStamp, reduceStageStrategy, settings);
            
            outds = iCreateOutputDatastore(outputFiles, settings.OutputFolder, settings.OutputType);
        end
        
        % Execute the Map-Stage
        function pExecuteMapTasks(this, inputDS, remoteMapReduceDetails, shuffleSortStrategy, reduceStageStrategy, settings)
            import matlab.mapreduce.ParallelMapReducer;
            import parallel.internal.mapreduce.MapTask;
            import parallel.internal.mapreduce.MapTaskResult;
            import parallel.internal.mapreduce.ParfevalTaskRunner;
            
            maxNumPartitions = max(this.Pool.NumWorkers * 3, ParallelMapReducer.MaxNumPartitions);
            numPartitions = min(numpartitions(inputDS), maxNumPartitions);
            
            shuffleSortStepIds = iGetShuffleSortStepIds(numPartitions, shuffleSortStrategy.isAsyncShuffleSortSupported());
            
            taskRunner = ParfevalTaskRunner(this.Pool, ...
                numPartitions + numel(shuffleSortStepIds) + 2, ...
                shuffleSortStrategy.isAsyncShuffleSortSupported());
            taskRunner.TaskResultCombiner = reduceStageStrategy;
            mrprogress = matlab.mapreduce.internal.MapReduceProgress(strcmp(settings.Display, 'On'), 1);
            mrprogress.PrintHeader();
            mrprogress.PrintZero();
            setReduceProgress(mrprogress, 0);
            taskRunner.ProgressCallback = mrprogress;
            
            % We launch a shuffle-sort task first to initialize mpiShuffler
            % at the same time for all workers. This is to guard against
            % the scenario that initialize is called for some workers and
            % not others.
            if shuffleSortStrategy.isAsyncShuffleSortSupported()
                taskRunner.launchTask(reduceStageStrategy.createShuffleSortStepTask(remoteMapReduceDetails));
            end
            
            for id = 1:numPartitions
                mapTask = MapTask(id, {numPartitions, id}, remoteMapReduceDetails);
                taskRunner.launchTask(mapTask);
                
                if any(id == shuffleSortStepIds)
                    taskRunner.launchTask(reduceStageStrategy.createShuffleSortStepTask(remoteMapReduceDetails));
                end
            end
            taskRunner.launchTask(reduceStageStrategy.createFinalShuffleSortStepTask(remoteMapReduceDetails));
            taskRunner.wait();
            mrprogress.PrintMapHundred();
        end
                
        % Execute the Reduce-Stage
        function outputFiles = pExecuteReduceTasks(this, remoteMapReduceDetails, resultTimeStamp, reduceStageStrategy, settings)
            import parallel.internal.mapreduce.ReduceOutputServer;
            import parallel.internal.mapreduce.ReduceTask;
            import parallel.internal.mapreduce.ParfevalTaskRunner;
            import parallel.internal.mapreduce.ReduceTaskResultCombiner;
            
            
            outputServer = [];
            outputSinkFactory = iCreateOutputSinkFactory(settings.OutputType, settings.getOutputFolderAsLocalPath(), resultTimeStamp);
            if ~reduceStageStrategy.useSharedFilesystem()
                % Non-shared filesystem code-path
                dctSchedulerMessage(6, 'Using non-shared filesystem code-path for output location.\n');
                outputServer = ReduceOutputServer(outputSinkFactory, this.Pool);
                outputSinkFactory = outputServer.getRemoteOutputSinkFactory();
            end
            
            reduceTasks = reduceStageStrategy.createReduceTasks(remoteMapReduceDetails, outputSinkFactory);

            taskRunner = ParfevalTaskRunner(this.Pool, numel(reduceTasks));
            taskRunner.TaskResultCombiner = ReduceTaskResultCombiner();
            taskRunner.OutputServer = outputServer;
            
            % Set up displayable monitoring of MapReduce
            mrprogress = matlab.mapreduce.internal.MapReduceProgress(strcmp(settings.Display, 'On'), 1);
            setMapProgress(mrprogress, 100);
            taskRunner.ProgressCallback = mrprogress;
            
            for ii = 1:numel(reduceTasks)
                taskRunner.launchTask(reduceTasks{ii});
            end
            taskRunner.wait();
            mrprogress.PrintHundred();
            
            if ~isempty(outputServer)
                % Non-shared filesystem code-path
                outputServer.commit();
                taskRunner.TaskResultCombiner.append(outputServer.getOutputMetadata());
            end
            outputFiles = taskRunner.TaskResultCombiner.getOutputFiles();
        end
    end
    
    methods (Access = protected)
        function executor = createPartitionedArrayExecutor(obj)
            if ~obj.Pool.Connected
                error(message('parallel:lang:pool:InvalidPool'));
            end
            if isa(obj.Pool, "parallel.ThreadPool")
                executor = parallel.internal.bigdata.NonCommPoolExecutor(obj.Pool);
            else
                executor = parallel.internal.bigdata.CommPoolExecutor(obj.Pool);
            end
        end
        
        % Create a memento struct that can recreate this ParallelMapReducer.
        function m = doGetMemento(obj)
            if isa(obj.Pool, "parallel.ThreadPool")
                m = struct('PoolType', {class(obj.Pool)});
            else
                m = struct(...
                    'PoolType', {class(obj.Pool)}, ...
                    'Cluster', {obj.Pool.Cluster}, ...
                    'NumWorkers', {obj.Pool.NumWorkers}, ...
                    'AttachedFiles', {obj.Pool.AttachedFiles}, ...
                    'SpmdEnabled', {obj.Pool.SpmdEnabled}, ...
                    'IdleTimeout', {obj.Pool.IdleTimeout}, ...
                    'ObjectVisibility', {obj.ObjectVisibility});
            end
        end
        
        % Check whether the given memento is equivalent to one created fresh
        % from this class.
        function tf = doCheckMemento(obj, memento)
            tf = matlab.mapreduce.ParallelMapReducer.checkMementoAgainstPool(obj.Pool, memento);
        end
    end
    
    methods (Access = protected, Static)
        % Check whether the given memento is equivalent to one created fresh
        % from the given pool object.
        function tf = checkMementoAgainstPool(pool, memento)
            if memento.PoolType == "parallel.ThreadPool"
                tf = isequal(class(pool), memento.PoolType);
            else
                tf = isequal(pool.Cluster, memento.Cluster) ...
                    && pool.NumWorkers == memento.NumWorkers ...
                    && pool.SpmdEnabled == memento.SpmdEnabled;
            end
        end
        
        % Recreate a ParallelMapReducer object from a memento struct.
        function obj = doCreateFromMemento(m)
            p = gcp('nocreate');
            if isempty(p)
                if m.PoolType == "parallel.LocalThreads"
                    p = parpool("LocalThreads");
                else
                    p = parpool(m.Cluster, m.NumWorkers, 'AttachedFiles', m.AttachedFiles, ...
                        'SpmdEnabled', m.SpmdEnabled, 'IdleTimeout', m.IdleTimeout);
                end
            else
                if ~matlab.mapreduce.ParallelMapReducer.checkMementoAgainstPool(p, m)
                    obj = [];
                    return;
                end
                addAttachedFiles(p, setdiff(m.AttachedFiles, p.AttachedFiles));
            end
            obj = matlab.mapreduce.ParallelMapReducer(p);
            obj.ObjectVisibility = m.ObjectVisibility;
        end
    end
    
    methods (Access = private)
        % Get a user visible string that represents this execution
        % environment.
        function name = getName(obj)
            if isa(obj.Pool, 'parallel.ThreadPool')
                profileName = 'LocalThreads';
            else
                profileName = obj.Pool.Cluster.Profile;
            end
            if isempty(profileName)
                name = getString(message('parallel:bigdata:ParallelPoolName'));
            else
                name = getString(message('parallel:bigdata:ParallelPoolWithProfileName', profileName));
            end
        end
    end
end

% Create a factory for the output type
function outputSinkFactory = iCreateOutputSinkFactory(outputType, outputFolder, resultTimeStamp)
import parallel.internal.mapreduce.LocalFileKeyValueSink;
outputSinkFactory = @(id)LocalFileKeyValueSink(outputType, outputFolder, id, resultTimeStamp);
end

% Display the first line when 'Display', 'on' is given as input.
function iDisplayHeader()
disp(getString(message('parallel:mapreduce:ParallelMapReducerDisplayHeader')));
end

% Create a datastore that represents the output.
function ds = iCreateOutputDatastore(outputFiles, outputFolder, outputType)
import matlab.mapreduce.internal.createTextDSWithKeyValueVarNames;
try
    switch outputType
        case 'Binary'
            ds = datastore(outputFiles, 'Type', 'KeyValue', 'FileType', 'mat');
        case 'TabularText'
            ds = createTextDSWithKeyValueVarNames(outputFiles, 'Type', 'TabularText');
        otherwise
            assert(false, 'Invalid value received in parsed OutputType property')
    end
catch err
    pErr = MException(message('parallel:mapreduce:CannotAccessOutputFiles', outputFolder));
    pErr = addCause(pErr, err);
    throw(pErr);
end
end

% Create the output folder of this mapreduce call. This will only be called
% if the folder does not already exist.
function iCreateOutputFolder(outputFolder)
[success, ~, ~] = mkdir(outputFolder);
if ~success
    error(message('parallel:mapreduce:CannotCreateOutputFolder', outputFolder));
end
end

% Return a list of map task Ids for which an shuffle-sort step should
% follow.
function shuffleSortStepIds = iGetShuffleSortStepIds(numPartitions, isAsyncShuffleSortSupported)
shuffleSortStepIds = [];
if ~isAsyncShuffleSortSupported
    return;
end
try
    shuffleSortSteps = distcomp.feature('MapReduceShuffleSortSteps');
catch
    return;
end

validateattributes(shuffleSortSteps, {'double'}, {'positive'});
shuffleSortStepIds = ceil(shuffleSortSteps * numPartitions);
shuffleSortStepIds(shuffleSortStepIds >= numPartitions) = [];
end
