%HadoopMapReduceJob
% Internal class that represents a Hadoop job in the PCT framework.

%   Copyright 2014-2018 The MathWorks, Inc.

classdef (Sealed) HadoopMapReduceJob < parallel.Job & parallel.internal.hadoop.HadoopJobMixin
    
    properties
        DisplayObject % An object that encapsulates displaying to the command line.
    end
    
    methods
        % Construct a HadoopMapReduceJob from the given cluster and construction
        % arguments for the info object.
        function obj = HadoopMapReduceJob(cluster, info)
            % ID is unused.
            ID = 0;
            obj = obj@parallel.Job(cluster, ID, parallel.internal.types.Variant.IndependentJob);
            obj = obj@parallel.internal.hadoop.HadoopJobMixin(info);
        end
    end
    
    methods (Access = protected)
        % getTasks must return a list of parallel.Task objects
        function tl = getTasks( obj )
            import parallel.internal.hadoop.HadoopTask;
            tl = [HadoopTask(obj, 1, obj.Info.MapFunction); ...
                HadoopTask(obj, 2, obj.Info.ReduceFunction)];
        end
        
        % getTaskCount return the number of tasks
        function tc = getTaskCount( obj )
            tc = numel(obj.getTasks);
        end
        
        % Submit a scalar job. This will currently block until that job has
        % completed.
        function doSubmit(obj, workerDataFiles)
            import parallel.internal.mapreduce.TempLocalFolder;
            import parallel.internal.mapreduce.PathTools;
            import parallel.internal.hadoop.HadoopCommandRunner;
            import parallel.internal.hadoop.HadoopMapReduceJobInfo;
            import parallel.internal.hadoop.HadoopWorker;
            
            % This is the folder that will be used as the current working
            % directory of the submit process. For Hadoop 1.x in local
            % mode, this requires to hold all attached data files.
            startFolder = TempLocalFolder();
            if obj.Parent.HadoopVersion == 1
                for ii = 1:numel(workerDataFiles)
                    copyfile(workerDataFiles{ii}, startFolder.Path);
                end
            end
            
            workerDataFiles = strjoin(cellfun(@PathTools.ensureIsIri, ...
                workerDataFiles, 'UniformOutput', false), ',');

            % We set LD_LIBRARY_PATH to the MATLAB Linux library path in
            % case the cluster is on Linux machine that uses an old version
            % of libstdc++. If Hadoop is setup to use native libraries,
            % then those libraries will load the old version of libstdc++
            % inside a reduce task before instantiating any of our classes.
            % Note, Hadoop will prepend this LD_LIBRARY_PATH to the
            % LD_LIBRARY_PATH set in mapred.admin.user.env.
            glnxa64LibraryPath = obj.getWorkerLibraryPath('glnxa64');
            childEnvironment = obj.pAppendToProperty('mapred.child.env', ...
                sprintf('LD_LIBRARY_PATH=%s', glnxa64LibraryPath), ',');
            
            % The options are written to a file in order to avoid having to
            % escape command line syntaxes. We currently have to specify
            % mapred.child.env and mapred.child.java.opts in order to
            % ensure librarypath is set.
            optionsFile = fullfile(startFolder.Path, 'options.in');
            obj.pWriteOptionsToFile(optionsFile, workerDataFiles, childEnvironment);
            
            commandRunner = HadoopCommandRunner(obj.Parent.HadoopInstallFolder, obj.Parent.HadoopVersion);
            commandRunner.Display = strcmp(obj.Info.Display, 'On');
            
            % For the purposes of local mode, which potentially makes the current
            % folder dirty by placing temporary resources in it.
            commandRunner.StartInPath = startFolder.Path;
            
            % For the purposes of local mode, which does not start any
            % child processes and ignores mapred.child.env set with childEnvironment
            % above.
            commandRunner.Environment('JAVA_LIBRARY_PATH') = strrep(obj.getClientLibraryPath(), ',', pathsep);
            commandRunner.Environment('HADOOP_CLASSPATH') = strjoin(obj.pDetermineClasspath(true), pathsep);
            
            outputStatusFile = fullfile(startFolder.Path, 'status.out');
            displayFcn = obj.pGetDisplayFunction(outputStatusFile);
            if ~isempty(displayFcn)
                commandRunner.DisplayFcn = displayFcn;
            end
            
            status = commandRunner.runTool('JobLaunchTool', optionsFile, outputStatusFile);
            if status ~= 0
                obj.pHandleFailure(outputStatusFile);
            end
        end
    end
    
    methods (Access = private)
        % Function that returns a Display function handle for the command
        % runner.
        function displayFcn = pGetDisplayFunction(obj, outputStatusFile)
            displayFcn = [];
            
            displayObject = obj.DisplayObject;
            if ~isempty(displayObject)
                displayFcn = @(~)iDisplayStatusFromFile(displayObject, outputStatusFile);
            end
        end
        
        % Convert a job failure into the appropriate error.
        function pHandleFailure(obj, outputStatusFile)
            % Attempt to read from the output status file. This will throw if
            % the output status file does not exist or does not contain
            % valid information.
            [jobId, submitErrorMessage] = iParseOutputStatusFile(outputStatusFile);
            
            if ~isempty(submitErrorMessage)
                error(message('parallel:mapreduce:HadoopJobSubmitFailure', submitErrorMessage));
            end
            
            % If there are attached error files in the output folder, there
            % were user visible errors generated on the worker side.
            iThrowIfTaskError(obj.Info.FinalOutputFolder);
            
            % Otherwise...
            error(message('parallel:mapreduce:HadoopJobOtherFailure', jobId));
        end
        
        % Write out all options to pass to Hadoop to the given file.
        function pWriteOptionsToFile(obj, optionsFile, attachedDataFiles, childEnvironment)
            import parallel.internal.hadoop.FileSystemTools;
            import parallel.internal.mapreduce.PathTools;
            
            envOptions = obj.getMatlabWorkerEnvironmentOptions();
            envOptions = cellfun(@iHadoopProperty, envOptions(:, 1), envOptions(:, 2));
            
            options = [...
                ... % Generic options from parallel.cluster.Hadoop
                obj.pGetHadoopConfFileOption(), ...
                obj.pGetHadoopPropertyOptions(), ...
                ... % The Job data to be sent to the worker nodes
                {'-files', attachedDataFiles}, ...
                ... % The environment variables to pass to the taskrunner.
                iHadoopProperty('mapred.child.env', childEnvironment), ...
                ... % Whether the child worker should use WorkerPoolManager.
                obj.getReuseWorkerFlag(), ...
                ... % Default Job Name
                obj.getDefaultJobNameOption(), ...
                ... % MATLAB Environment Options
                envOptions(:)', ...
                ... % The PID of this process
                iHadoopProperty('mathworks.toolbox.parallel.hadoop.ParentMatlabPid', num2str(feature('getpid'))), ...
                ... % The bin folder of this process
                iHadoopProperty('mathworks.toolbox.parallel.hadoop.ParentMatlabBinFolder', fullfile(matlabroot, 'bin', computer('arch'))), ...
                ... % The input paths to read from.
                obj.pGetInputArguments(), ...
                ... % The output path to write into.
                obj.pGetOutputPath()];
            
            [fid, fidCleanup] = FileSystemTools.openLocalTextFile(optionsFile, 'w'); %#ok<ASGLU>
            fprintf(fid, '%s\n', options{:});
        end
        
        % Return the option that determines whether WorkerPoolManager is
        % used.
        function option = getReuseWorkerFlag(obj)
            switch lower(distcomp.feature('HadoopReuseWorker'))
                case 'yes'
                    reuseWorkerFlag = 'true';
                case 'no'
                    reuseWorkerFlag = 'false';
                otherwise
                    if obj.Parent.HadoopVersion == 1
                        reuseWorkerFlag = 'false';
                    else
                        reuseWorkerFlag = 'true';
                    end
            end
            
            option = iHadoopProperty('mathworks.toolbox.parallel.hadoop.ReuseWorker', reuseWorkerFlag);
        end
        
        % Return the option for default job name. The submit code will use
        % this in absence of a user set job name.
        function option = getDefaultJobNameOption(obj)
            hadoopUseMapperReducerJobName = distcomp.feature('HadoopUseMapperReducerJobName');
            if hadoopUseMapperReducerJobName
                defaultJobName = sprintf('mapper = %s; reducer = %s;', ...
                    char(obj.Info.MapFunction), char(obj.Info.ReduceFunction));
            else
                defaultJobName = getString(message('parallel:mapreduce:HadoopDefaultJobName'));
            end
            
            option = iHadoopProperty('mathworks.toolbox.parallel.hadoop.DefaultJobName', defaultJobName);
        end
        
        % If there is an application defined configuration file, this is
        % the option that tells Hadoop about it.
        function option = pGetHadoopConfFileOption(obj)
            import parallel.internal.mapreduce.PathTools;
            hadoopConfigurationFile = obj.Parent.HadoopConfigurationFile;
            if isempty(hadoopConfigurationFile)
                option = {};
            else
                option = {'-conf', PathTools.ensureIsIri(hadoopConfigurationFile)};
            end
        end
        
        % Return the hadoop properties as a cell array of strings of the
        % form '-Dname=value'.
        function options = pGetHadoopPropertyOptions(obj)
            names = obj.Parent.HadoopProperties.keys;
            values = obj.Parent.HadoopProperties.values;
            if isempty(names)
                options = {};
            else
                options = cellfun(@(name, value){sprintf('-D%s=%s', name, value)}, names, values);
            end
        end
        
        % Return the input location information to pass to the Hadoop
        % scheduler.
        function inputArguments = pGetInputArguments(obj)
            inputArguments = matlab.bigdata.internal.hadoop.getSerializedLocation(...
                obj.Info.InputDatastore, @distcompserialize);
            inputArguments = inputArguments(:)';
        end
        
        % Return the output folder as a cell
        function path = pGetOutputPath(obj)
            path = obj.Info.FinalOutputFolder;
            if isempty(path)
                path = {'output'};
            else
                path = {path};
            end
        end
        
        % Get the local jar files to be either added to the local LaunchJobTool
        % or to be attached to the job to be sent to the worker.
        function classpathEntries = pDetermineClasspath(~, requiresDistcompJar)
            targetMatlabRoot = matlabroot;
            classpathEntries = {};
            
            if requiresDistcompJar
                classpathEntries = [classpathEntries, {fullfile(targetMatlabRoot, 'java', 'jar', 'toolbox', 'distcomp.jar')}];
            end
        end
        
        % Return the full property value of taking what is currently set in
        % the cluster object and appending the given value.
        function fullValue = pAppendToProperty(obj, propName, value, joinStr, optionalDefault)
            fullValue = obj.Parent.hGetHadoopProperty(propName);
            if isempty(fullValue) && nargin >= 5
                fullValue = optionalDefault;
            end
            if isempty(fullValue)
                fullValue = value;
            else
                fullValue = [fullValue, joinStr value];
            end
        end
        
    end
end

% Format a name-value pair into something that can be passed as a Hadoop
% command line property
function out = iHadoopProperty(name, value)
if isempty(value)
    out = {};
elseif ischar(value)
    out = {sprintf('-D%s=%s', name, value)};
else
    out = {sprintf('-D%s=%g', name, value)};
end
end

% Handle errors coming from the execution of the job. This will attach any
% errors found in the output folder as causes for the failure error.
function iThrowIfTaskError(outputFolder)
import matlab.io.datastore.internal.HadoopConfiguration;
import parallel.internal.hadoop.FileSystemTools

fileSystemTools = FileSystemTools(HadoopConfiguration.getGlobalConfiguration());
errors = fileSystemTools.loadErrorsFromOutputFolder(outputFolder);
errors = errors.values;
if isempty(errors)
    return;
end

err = MException(message('parallel:mapreduce:HadoopJobFailure'));
isHot = matlab.internal.display.isHot;
for ii = 1:numel(errors)
    if isHot && isequal(errors{ii}.identifier, 'parallel:mapreduce:HadoopTaskInvalidHome')
        arguments = errors{ii}.arguments;
        arguments{5} = '<a href="matlab:helpview(''matlab-parallel-server'',''configure_hadoop'')">';
        arguments{6} = '</a>';
        errors{ii} = MException(message('parallel:mapreduce:HadoopTaskInvalidHome', arguments{:}));
    end
    err = err.addCause(errors{ii});
end
throw(err);
end

% Parses the output status file generated by JobLaunchTool.
function [jobId, submitErrorMessage, mapProgress, reduceProgress] = iParseOutputStatusFile(filename)
import parallel.internal.hadoop.FileSystemTools;
if ~exist(filename, 'file')
    error(message('parallel:mapreduce:HadoopJobSubmitExecutionFailure'));
end

[fid, fidCleanup] = FileSystemTools.openLocalTextFile(filename, 'r'); %#ok<ASGLU>

expression = ['JobId:(?<jobId>[^\n]*)\n' ...
    'MapProgress:(?<mapProgress>[^\n]*)\n' ...
    'ReduceProgress:(?<reduceProgress>[^\n]*)\n' ...
    '(?<submitErrorMessage>.*)'];
data = regexp(fread(fid, '*char')', expression, 'names', 'once');
if isempty(data)
    error(message('parallel:mapreduce:HadoopJobSubmitExecutionFailure'));
end
jobId = data.jobId;
submitErrorMessage = data.submitErrorMessage;
mapProgress = str2double(data.mapProgress);
reduceProgress = str2double(data.reduceProgress);
end

% Displays the status from the given file to the command line
function iDisplayStatusFromFile(displayObject, filename)
try
    [~,~, mapProgress, reduceProgress] = iParseOutputStatusFile(filename);
    displayObject.Progress(mapProgress * 100, reduceProgress * 100);
catch
    % We do nothing here because the worst that is lost is a progress
    % message.
end
end
