%HadoopWorker
% The main entry points for calling into MATLAB from Hadoop.

%   Copyright 2014-2019 The MathWorks, Inc.

classdef (Sealed, Hidden) HadoopWorker < handle
    
    properties (Constant)
        HadoopJobInfoFile = 'job.mat'; % The file that includes job information such as map/reduce handles.
        HadoopJobAttachFile = 'attach.mat'; % The file that includes attached files/additional paths information.
    end
    
    properties (Constant)
        Instance = parallel.internal.hadoop.HadoopWorker; % A persistent instance that will contain job level persistent state.
    end
    
    properties (SetAccess = private)
        Cluster; % The internal Cluster object associated with this job.
        Job; % The internal Job object associated with this job.
        FileDependenciesFolder; % The temporary folder containing file dependencies.
    end
    
    methods (Static)
        function configure(jobDataFolder)
            import parallel.internal.hadoop.HadoopWorker;
            import parallel.internal.mapreduce.TempLocalFolder;
            
            dctSchedulerMessage(6, 'Entering parallel.internal.hadoop.HadoopWorker/configure');
            
            % Explicitly set the number of computational threads to 1 on a worker. The user
            % may choose to override this using jobStartup.
            numThreadsPerWorker = 1;
            iSetupComputationalThreads(numThreadsPerWorker);
            
            % This will load the job singleton from the current folder (this is where
            % Hadoop will soft-link the corresponding file).
            attach = load(fullfile(jobDataFolder, HadoopWorker.HadoopJobAttachFile));
            
            % Ensure job persistent state is set
            instance = HadoopWorker.Instance;
            instance.FileDependenciesFolder = TempLocalFolder;
            
            % Deserialize any attached file data and add this to the MATLAB
            % path alongside any additional paths. We do this first so that
            % handles loaded later are valid.
            fileDependenciesPath = instance.FileDependenciesFolder.Path;
            iDeserializeFileData(attach.FileData, fileDependenciesPath);
            
            % TODO: (g1442287) Add support for MATLAB Drive path translation for Hadoop workers.
            attachedFilePaths = fullfile(fileDependenciesPath, attach.AttachedFilePaths.values);
            parallel.internal.general.addpathWithFoldersFilteredIfDeployed(attachedFilePaths{:});
            parallel.internal.general.addpathWithFoldersFilteredIfDeployed(attach.AdditionalPaths{:});
            
            % Ensure MHLM is setup so we can create a cluster object.
            [reallyUsingMHLM, entitlement] = iSetupMhlmAndDesktopClient();
            
            cluster = parallel.internal.hadoop.HadoopCluster;
            if reallyUsingMHLM
                cluster.RequiresMathWorksHostedLicensing = true;
                hSetEntitlementNoCheck(cluster, entitlement);
            end
            
            % As the final step, we can now load the job info that includes
            % the Map/Reduce function handles.
            info = load(fullfile(jobDataFolder, HadoopWorker.HadoopJobInfoFile));
            instance.Job = info.JobInfo.wrapAsJob(cluster);
            pPreHadoopJobExecution(instance.Job);
        end
    end
    
    methods (Access = private)
        function obj = HadoopWorker()
            if feature('isDMLWorker')
                mlock;
            end
        end
    end
end

% Helper class that deserializes file data to a path.
function iDeserializeFileData(fileData, outputPath)
import parallel.internal.mapreduce.TempLocalFolder;

if isempty(fileData)
    % This is the case when no files were attached.
    return;
end

tempFolder = TempLocalFolder;
zipFile = fullfile(tempFolder.Path, 'dependentfiles.zip');
[fid, msg] = fopen(zipFile, 'w');
if fid < 0
    error(message('parallel:cluster:CouldNotOpenFile', zipFile, msg));
end
fwrite(fid, fileData, 'int8');
fclose(fid);
unzip(zipFile, outputPath);
end

% Setup MHLM in the same way as CJS.
function [reallyUsingMHLM, entitlement] = iSetupMhlmAndDesktopClient()
% Deal with MHLM and set the desktop client appropriately.
% We must do this before constructing the cluster object.
useMLM = getenv('MLM_WEB_LICENSE');
reallyUsingMHLM = ~isempty(useMLM) && strcmpi(useMLM, 'true');
if reallyUsingMHLM
    token = getenv('MLM_WEB_USER_CRED');
    licenseWebId = getenv('MLM_WEB_ID');
    licenseNumber = parallel.internal.cluster.getParallelServerEnv('PARALLEL_SERVER_LICENSE_NUMBER');
    desktopClient = parallel.internal.mwa.MDCSDesktopClient.buildWithFixedToken(token);
    entitlement = parallel.internal.mwa.Entitlement(licenseWebId, licenseNumber, '');
else
    desktopClient = parallel.internal.mwa.MDCSDesktopClient();
    entitlement = [];
end
parallel.internal.mwa.currentDesktopClient(desktopClient);
end

function iSetupComputationalThreads(num)
try
    feature('numthreads', num);
catch err
    dctSchedulerMessage( 1, 'Warning - failed to set number of computational threads to %d. Error thrown was:\n%s\n', ...
        num, err.getReport );
end
end
