%FileSystemTools
% Several tools to allow copying of data to HDFS. This should only be used
% for copying small files, for example error files, between local and HDFS.

%   Copyright 2014-2018 The MathWorks, Inc.

classdef (Sealed, Hidden) FileSystemTools < handle
    
    properties (Access = private, Constant)
        ErrorSubFolder = '.matlaberror'; % The name of the subfolder to store mapreduce errors.
    end
    
    methods
        % Construct an instance of FileSystemTools that use the given
        % Hadoop configuration.
        function this = FileSystemTools(~)
        end
        
        % Helper function that writes the given error to a special
        % matlaberror folder inside the output folder.
        function saveErrorToOutputFolder(this, error, outputFolderIRI, taskAttemptID)
            import parallel.internal.mapreduce.TempLocalFolder;
            
            tempLocalFolder = TempLocalFolder;
            localFilename = fullfile(tempLocalFolder.Path, 'error.mat');
            iSaveError(error, localFilename);
            
            errorFolderIRI = [outputFolderIRI '/' this.ErrorSubFolder];
            remoteFileIRI = [errorFolderIRI '/' char(taskAttemptID.toString()) '.mat'];
            this.createFolder(errorFolderIRI);
            this.copyFile(iResolveLocalPath(localFilename), remoteFileIRI);
        end
        
        % Helper function that reads all errors from the given output
        % folder.
        function errors = loadErrorsFromOutputFolder(this, outputFolderIRI)
            import parallel.internal.mapreduce.TempLocalFolder;
            
            this.throwIfOutputFolderDoesNotExists(outputFolderIRI);
            
            tempLocalFolder = TempLocalFolder;
            
            errors = containers.Map('KeyType', 'char', 'ValueType', 'any');
            
            errorFolderIRI = [outputFolderIRI '/' this.ErrorSubFolder];
            files = this.listFilesInFolder(errorFolderIRI);
            for ii = 1:numel(files)
                [~, filenameWithoutExt, ext] = fileparts(files{ii});
                filename = [filenameWithoutExt, ext];
                
                localFilename = fullfile(tempLocalFolder.Path, filename);
                this.copyFile(files{ii}, iResolveLocalPath(localFilename));
            
                try
                    switch ext
                        case '.mat'
                            errors(filenameWithoutExt) = iLoadError(localFilename);
                        case '.err'
                            errors(filenameWithoutExt) = iReadTextErrorID(localFilename);
                    end
                catch err
                    % This can happen if Hadoop decides to kill a job in
                    % response to too many errors while one of the workers
                    % is attempting to write its error instance to file.
                    dctSchedulerMessage( 2, 'Failed to load an error file from a MATLAB Worker. Error thrown was:\n%s\n', ...
                        err.getReport );
                end
            end
        end
        
        % Helper function that throws if the given output folder does not
        % exist as a directory.
        function throwIfOutputFolderDoesNotExists(~, outputFolderIRI)
            import matlab.io.datastore.internal.ContextClassLoaderGuard;
            import matlab.io.datastore.internal.HadoopConfiguration;
            import matlab.io.datastore.internal.buildHadoopPath;
            guard = ContextClassLoaderGuard; %#ok<NASGU>
            path = buildHadoopPath(outputFolderIRI);
            fileSystem = HadoopConfiguration.getGlobalFileSystem(path.toUri());
            if ~fileSystem.exists(path) || ~fileSystem.isDirectory(path)
                error(message('parallel:mapreduce:InvalidHadoopOutputFolder', outputFolderIRI));
            end
        end
        
        % Helper function for creating a directory on Hadoop.
        function createFolder(~, path)
            import matlab.io.datastore.internal.ContextClassLoaderGuard;
            import matlab.io.datastore.internal.buildHadoopPath;
            import matlab.io.datastore.internal.HadoopConfiguration;
            guard = ContextClassLoaderGuard; %#ok<NASGU>
            path = buildHadoopPath(path);
            fileSystem = HadoopConfiguration.getGlobalFileSystem(path.toUri());
            
            fileSystem.mkdirs(path);
        end
        
        % Helper function to list files in the given folder.
        function [files, isFileEmpty] = listFilesInFolder(~, path)
            import matlab.io.datastore.internal.ContextClassLoaderGuard;
            import matlab.io.datastore.internal.HadoopConfiguration;
            import matlab.io.datastore.internal.buildHadoopPath;
            guard = ContextClassLoaderGuard; %#ok<NASGU>
            path = buildHadoopPath(path);
            fileSystem = HadoopConfiguration.getGlobalFileSystem(path.toUri());
            
            try
                fileInfos = fileSystem.listStatus(path);
            catch
                files = {};
                isFileEmpty = false(0,0);
                return;
            end
            files = cell(numel(fileInfos), 1);
            isFileEmpty = false(size(files));
            for ii = numel(fileInfos):-1:1
                fileInfo = fileInfos(ii);
                if fileInfo.isDir()
                    files(ii) = [];
                    isFileEmpty(ii) = [];
                else
                    files{ii} = char(fileInfo.getPath().toUri());
                    isFileEmpty(ii) = fileInfo.getLen() == 0;
                end
            end
        end
        
        % Helper function for copying between local and HDFS.
        function copyFile(~, path1, path2)
            import org.apache.commons.io.IOUtils;
            import matlab.io.datastore.internal.ContextClassLoaderGuard;
            import matlab.io.datastore.internal.HadoopConfiguration;
            import matlab.io.datastore.internal.buildHadoopPath;
            guard = ContextClassLoaderGuard; %#ok<NASGU>
            
            path1 = buildHadoopPath(path1);
            fileSystem1 = HadoopConfiguration.getGlobalFileSystem(path1.toUri());
            stream1 = fileSystem1.open(path1);
            stream1Cleanup = onCleanup(@stream1.close);
            
            path2 = buildHadoopPath(path2);
            fileSystem2 = HadoopConfiguration.getGlobalFileSystem(path2.toUri());
            stream2 = fileSystem2.create(path2);
            stream2Cleanup = onCleanup(@stream2.close);
            
            IOUtils.copy(stream1, stream2);
        end
    end
    
    methods (Static)
        % Helper function to assist with opening local text files.
        function [fid, fidCleanup] = openLocalTextFile(filename, permissions)
            for retry = 1:3
                [fid, msg] = fopen(filename, permissions, 'n', 'UTF-8');
                if fid > 0
                    fidCleanup = onCleanup(@()fclose(fid));
                    return;
                end
            end
            error(message('parallel:cluster:CouldNotOpenFile', filename, msg));
        end
    end
    
end

% Helper function to save an error to the given local file
function iSaveError(error, localFilename)
matFile = matfile(localFilename);
matFile.Error = error;
end

% Helper function to load an error from the given local file
function error = iLoadError(localFilename)
matFile = matfile(localFilename);
error = matFile.Error;
end

% Helper function that resolves a local path to a file:/path IRI.
function iri = iResolveLocalPath(path)
import parallel.internal.mapreduce.PathTools;
iri = PathTools.ensureIsIri(path);
end

% Helper function that reads a text error file and constructs an error.
% The contents is expected to be a list of inputs to message, delimited by
% the new-line character.
function err = iReadTextErrorID(localFilename)
import parallel.internal.hadoop.FileSystemTools;
[fid, fidCleanup] = FileSystemTools.openLocalTextFile(localFilename, 'r'); %#ok<ASGLU>

data = strsplit(fread(fid, [1, Inf], '*char'), '\n', 'CollapseDelimiters', false);
% These files will always contain a trailing new-line character. Due to
% this, the last element of data will be empty and not intended for message.
err = MException(message(data{1 : end - 1}));
end
