%HadoopJobMixin
% Internal mixin class that implements the parallel.Job abstract methods
% around an implementation of HadoopJobInfo.

%   Copyright 2016-2018 The MathWorks, Inc.

classdef (Abstract) HadoopJobMixin < handle
    
    properties (SetAccess = immutable)
        % The underlying information about the Job.
        Info;
    end
    
    properties (SetAccess = private, Hidden)
        % This is implemented here as this is abstract in parallel.Job.
        MinNumWorkers = 0;
    end
    
    methods
        % Construct a HadoopJob from the given cluster and construction
        % arguments for the info object.
        function obj = HadoopJobMixin(info)
            obj.Info = info;
        end
    end
    
    methods (Abstract, Access = protected)
        % Perform the submission, attaching the provided data files to the
        % Hadoop Job. This may block until the end of the job.
        doSubmit(obj, attachedDataFiles);
    end
    
    methods (Sealed, Access = protected)
        function submitOneJob(obj)
            import parallel.internal.mapreduce.TempLocalFolder;
            
            dctSchedulerMessage(6,'Entering parallel.internal.hadoop.HadoopMapReduceJobMixin/submitOneJob.');
            
            hPreJobSubmission(obj.Parent, obj);
            pPrepareHadoopJobForSubmission(obj);
            
            % Create the data folder that will be sent to all nodes
            tempFolder = TempLocalFolder;
            
            % The HadoopMapReduceJobInfo instance contains all the details needed by
            % the workers.
            attachedDataFiles = obj.writeJobData(tempFolder.Path);
            
            obj.doSubmit(attachedDataFiles);
        end
        
        % Attach the job's zipped file data.
        function attachFileDataToJob(obj, filedata)
            dctSchedulerMessage(6,'Attaching %i bytes of file data to the job.', numel(filedata));
            obj.Info.FileData = filedata;
        end
    end
    
    methods (Access = protected)
        % Get the library path for accessing Parallel Computing Toolbox
        % libraries from a process on the client machine.
        function libraryPath = getClientLibraryPath(~)
            libraryPath = strjoin({ ...
                fullfile(matlabroot, 'bin', computer('arch')); ...
                fullfile(matlabroot, 'sys', 'os', computer('arch')) }, pathsep);
        end
        
        % Get the library path for accessing Parallel Computing Toolbox
        % libraries from a process on the cluster machine of a given architecture.
        function libraryPath = getWorkerLibraryPath(obj, architecture, clusterPathsep)
            
            % This will work for both unix and windows.
            clusterFilesep = '/';
            
            if nargin < 3
                if strcmp(architecture, 'win64')
                    clusterPathsep = ';';
                else
                    clusterPathsep = ':';
                end
            end
            
            clusterMatlabRoot = obj.Parent.ClusterMatlabRoot;
            libraryPath = strjoin({ ...
                strjoin({clusterMatlabRoot, 'bin', architecture}, clusterFilesep); ...
                strjoin({clusterMatlabRoot, 'sys', 'os', architecture}, clusterFilesep)}, ...
                clusterPathsep);
        end
        
        % Get a list of name-value pairs that must be set in the Hadoop
        % Configuration to be able to launch a MATLAB Worker.
        function options = getMatlabWorkerEnvironmentOptions(obj)
            options = [...
                ... % The version of hadoop to set in configuration.
                {'mathworks.toolbox.parallel.hadoop.HadoopVersion', num2str(obj.Parent.HadoopVersion)}; ...
                {'mathworks.toolbox.parallel.hadoop.HadoopFullVersion', obj.Parent.HadoopFullVersion}; ...
                ... % The cluster root on the cluster.
                {'mathworks.toolbox.parallel.hadoop.ClusterMatlabRoot', obj.Parent.ClusterMatlabRoot}; ...
                ... % The version of hadoop to set in configuration.
                {'mathworks.toolbox.parallel.hadoop.ClientVersionString', iGetMatlabVersionString()}; ...
                ];
            
            jobInfo = obj.Info;
            if ~isempty(jobInfo.LicenseNumber)
                options = [options;...
                    {'mathworks.toolbox.parallel.hadoop.MLMWebLicense', 'true'}; ...
                    {'mathworks.toolbox.parallel.hadoop.MLMWebUserCred', jobInfo.WebUserCred}; ...
                    {'mathworks.toolbox.parallel.hadoop.MLMWebId', jobInfo.WebId}; ...
                    {'mathworks.toolbox.parallel.hadoop.LicenseNumber', jobInfo.LicenseNumber}; ...
                    ];
            end
        end
        
        % This will write out data that is required to be passed directly
        % to each MATLAB Worker for initialization.
        function paths = writeJobData(obj, location)
            import parallel.internal.hadoop.HadoopWorker;
            
            % The HadoopJobInfo instance contains all the details needed by
            % the workers.
            jobInfo = obj.Info;
            jobFile.JobInfo = jobInfo;
            jobInfoPath = fullfile(location, HadoopWorker.HadoopJobInfoFile);
            iSave(jobInfoPath, jobFile);
            
            % We save the file-data and init data as a separate file so that this can be
            % loaded before we load the map/reduce function handles.
            attachFile.FileData = obj.Info.FileData;
            attachFile.AttachedFilePaths = obj.AttachedFilePaths;
            attachFile.AdditionalPaths = obj.AdditionalPaths;
            attachFile.MATLABDrivePathMap = obj.MATLABDrivePathMap;
            jobAttachPath = fullfile(location, HadoopWorker.HadoopJobAttachFile);
            iSave(jobAttachPath, attachFile);
            
            paths = {jobAttachPath, jobInfoPath};
        end
    end
    
    methods
        function value = hGetProperty(obj, name)
            % hGetProperty supports vectorized calls for arrays of obj and 
            % cell arrays of name, so that remote calls can be minimized with MJS.
            returnCell = iscell(name);
            if ~returnCell
                name = {name};
            end
            value = cell(numel(obj), numel(name) );
            for ii = 1:numel(obj)
                for jj = 1:numel(name)
                    value{ii, jj} = obj(ii).Info.(name{jj});
                end
            end

            if numel( obj ) == 1 && ~returnCell
                value = value{1,1};
            end
        end
        
        function hSetPropertyNoCheck(obj, names, values)
            if ~iscell(names)
                names = {names};
                values = {values};
            end
            for ii = 1:length(names)
                obj.Info.(names{ii}) = values{ii};
            end
        end
    end
    
    methods (Access = protected)
        % HadoopJobMixin is only intended for internal use. These are
        % unimplemented.
        cancelJob(job, ex);
        destroyJob(job);
        newJob = recreateOneJob( ~, ~ );
        createTaskOnOneJob( job, varargin );
        getStatusData( jobOrJobs );
    end
    
    methods ( Hidden )
        % HadoopJobMixin is only intended for internal use.
        function tf = pCurrentTaskIsPartOfAPool(job) %#ok<MANU>
            tf = false;
        end
        
        % HadoopJobMixin is only intended for internal use.
        function pPreJobEvaluate(job, task) %#ok<INUSD>
        end
        
        % HadoopJobMixin is only intended for internal use.
        function pPostJobEvaluate(job) %#ok<MANU>
        end
        
        % HadoopJobMixin is only intended for internal use.
        function jobIdentifier = hCreateJobUpdateIdentifier( job ) %#ok<MANU>
            jobIdentifier = [];
        end
        
        % hGetStateEnum must return a parallel.internal.types.States for obj job
        function s = hGetStateEnum(obj) %#ok<MANU>
            s = parallel.internal.types.States.Pending;
        end
    end
end

% Saves the given struct as a mat-file
function iSave(path, dataStruct)
save(path, '-struct', 'dataStruct');
end

% Helper function to retrieve the release string of MATLAB.
function matlabVersionString = iGetMatlabVersionString()
matlabVersionString = pctGetMATLABRelease;
end
