function distcomp_evaluate_filetask_core(outputWriterStack, mdceDebugEnabled, isSyncTaskEvaluation, varargin)
; %#ok Undocumented
% Accepts a job and task and computes and returns
% the results of the task and then exits matlab
%

% Copyright 2012-2020 The MathWorks, Inc.

mlock;

dctSchedulerMessage(6, 'Enter distcomp_evaluate_filetask_core');

if isdeployed
    handlers = struct(  'abortFcn',     @iDeployedAbortFunction, ...
                        'exitFcn',      @iDeployedExitFunction, ...
                        'errorFcn',     @iDeployedAbortFunction);
else
    handlers = struct(  'abortFcn',     @iAbortFunction, ...
                        'exitFcn',      @iExitFunction, ...
                        'errorFcn',     @iAbortFunction);
end

% Performs MATLAB Parallel Server specific job initialization work
try
    runprop = iSetup(handlers, mdceDebugEnabled, outputWriterStack, isSyncTaskEvaluation, varargin);
catch err
    handlers.abortFcn(err, 'Job setup failed - MATLAB will now exit.');
end
% Ensure that this function always exits if necessary
try
    postFcns = {};
    
    % Test if we should remove the DependencyDirectory at the end
    if runprop.CleanUpDependencyDirOnTaskFinish && ...
            ~exist(runprop.DependencyDirectory, 'dir')
        postFcns = [ ...
            {{@removeDirectory, runprop.DependencyDirectory}}, ...
            postFcns];
    end
    % Always clean up the per-host directory
    postFcns = [ ...
        {{@removeDirectory, runprop.PerHostDirectory}}, ...
        postFcns];
        
    if runprop.ExitOnTaskFinish 
        postFcns = [ ...
            postFcns ...
            {{handlers.exitFcn 'About to exit MATLAB normally'}}];
    end
    
    % Actually do the task evaluation stuff
    iDoTask(handlers, postFcns);
    % NOTE - do not add any code after the call to iDoTask as this will NOT get
    % executed if runprop.ExitOnTaskFinish is true. Generally this is true!
catch err
    handlers.abortFcn(err, 'Unexpected error in DoTask - MATLAB will now exit and restart.');
end
end

%--------------------------------------------------------------------------
% Define an abort function.
%--------------------------------------------------------------------------
function iAbortFunction(err, varargin)
    try
        if nargin > 1
            % Display some textual output if requested
            dctSchedulerMessage(0, varargin{:})
        end
    catch e %#ok<NASGU> - don't want this function to EVER error
    end
    try
        dctSchedulerMessage(0, err);
    catch e %#ok<NASGU> - don't want this function to EVER error
    end
    % Always exit - no matter what
    parallel.internal.cluster.exitOnException(1);
end

%--------------------------------------------------------------------------
% Define an exit function.
%--------------------------------------------------------------------------
function iExitFunction(varargin)
    try
        if nargin > 0
            % Display some textual output if requested
            dctSchedulerMessage(0, varargin{:})
        end
    catch e %#ok<NASGU> - don't want this function to EVER error
    end
    % Always exit - no matter what
    parallel.internal.cluster.exitOnException(0);
end

%--------------------------------------------------------------------------
% Define an abort function for a deployed worker.
%--------------------------------------------------------------------------
function iDeployedAbortFunction(err, varargin)
    try
        iAllowDeployedWorkerToExit();
    catch e
        try
            dctSchedulerMessage(0, e);
        catch e %#ok<NASGU> - don't want this function to EVER error
        end
    end
    % Log that we're about to abort
    dctSchedulerMessage(0, 'Deployed worker about to abort.');
    parallel.internal.logging.logStack(4, 'Aborting');

    % Always call the normal abort function
    iAbortFunction(err, varargin{:});
end

%--------------------------------------------------------------------------
% Define an exit function for a deployed worker.
%--------------------------------------------------------------------------
function iDeployedExitFunction(varargin)
    try
        if nargin > 0
            % Display some textual output if requested
            dctSchedulerMessage(0, varargin{:})
        end
    catch e %#ok<NASGU> - don't want this function to EVER error
    end
    
    try
        iAllowDeployedWorkerToExit();
    catch err
        try
            dctSchedulerMessage(0, err);
        catch e %#ok<NASGU> - don't want this function to EVER error
        end
    end
    % Don't need to do anything else - the process should now exit.
    dctSchedulerMessage(0, 'Deployed worker about to exit normally.');
    parallel.internal.logging.logStack(4, 'Exiting');
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iDoTask(handlers, postFcns)
% Need to know what the original path of matlab is - because the user might
% change this during the course of running their code and stop us removing
% some directories later.
persistent originalPath;
if isempty(originalPath)
    originalPath = path;
end

try
    finishFcn = @(job, task, out) iFinishTask(handlers, job, task, out, originalPath);
    dctEvaluateTask(postFcns, finishFcn);
catch e
    if isa(e, 'distcomp.ExitException')
        m = message( 'parallel:cluster:UnexpectedExitFailure', e.message );
        handlers.errorFcn( e.CauseException, '%s', m.getString() );
    else
        m = message( 'parallel:cluster:UnexpectedExitFailureTaskNoRestart' );
        handlers.errorFcn( e, '%s', m.getString() );
    end
end

end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iFinishTask(handlers, job, task, out, originalPath)
% If we are going to clean up the dependency directory later then we need to
% reset the matlab path because this might stop us deleting some directories.
try
    runprop = get(distcomp.getdistcompobjectroot, 'CurrentRunprop');
    if runprop.CleanUpDependencyDirOnTaskFinish
        path(originalPath);
    end
catch e
    dctSchedulerMessage(1, 'Unable to revert to original path - continuing\nNested Error', e);
end

try
    dctSchedulerMessage(4, 'About to call pPostTaskEvaluate');
    task.pPostTaskEvaluate(out.output, out.errOutput, out.textOutput, out.warnOutput);
catch e
    handlers.errorFcn(e, 'Unable to submit task result - MATLAB will now exit.');
end

try
    dctSchedulerMessage(4, 'About to call pPostJobEvaluate');
    job.pPostJobEvaluate;
catch e 
    % Do nothing if this throws an error
    dctSchedulerMessage(1, 'Error during pPostJobEvaluate - continuing\nNested Error', e);
end

end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function [fcn, fcnString] = iGetAndCacheDecodeFunction()
% Only deduce the decodeFunction once per MATLAB session. NOTE that
% this whole file is mlocked and so this persistent variable cannot be
% cleared.
persistent decodeFunction decodeFunctionString
if isempty(decodeFunction)
    decodeFunctionString = '';
    try
        decodeFunctionString = parallel.internal.cluster.getParallelServerEnv('PARALLEL_SERVER_DECODE_FUNCTION');
        % How should we interpret the environment?
        decodeFunction = str2func(decodeFunctionString);
        % We'll check decodeFunction is a valid function handle later on
        % when we call it.
    catch err
        dctSchedulerMessage(0, ['Error converting the environment variable PARALLEL_SERVER_DECODE_FUNCTION to a function handle.\n' ...
                            'This is probably because the environment variable (PARALLEL_SERVER_DECODE_FUNCTION) does not exist.\n' ...
                            'The PARALLEL_SERVER_DECODE_FUNCTION variable''s current value is "%s"'], decodeFunctionString);
        dctSchedulerMessage(0, 'Error returned was:\n', err);
        parallel.internal.cluster.exitOnException(1);
    end
end
fcn = decodeFunction;
fcnString = decodeFunctionString;
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iSetupProcessExitAndMatlabRef()
try
    % Make sure this matlab cannot core dump and exits on SEGV - this might
    % be changed by internal code so ensure it is done every job or task
    dct_psfcns('ensureProcessExitsOnFault');
    % Initialize the MatlabRefStore
    com.mathworks.toolbox.distcomp.util.MatlabRefStore.initMatlabRef();
    dctSchedulerMessage(6, 'This process will exit on any fault.');
catch err
    dctSchedulerMessage(0, 'Unexpected error occurred during setup.');
    dctSchedulerMessage(0, 'Error returned was:\n', err);
    parallel.internal.cluster.exitOnException(1);
end
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iMaybeSetupProcessMonitoringThreads()
% Since we are running on a third-party scheduler there is a good chance that
% we have been launched by worker or worker.bat. Under these circumstances it
% is often hard for the scheduler to kill the matlab. Thus we may well want to
% fire up a process watching thread that will kill this matlab if the relevant
% process being watched dies.
persistent processMonitoringThreadStarted;
try
    if isempty(processMonitoringThreadStarted) || ~processMonitoringThreadStarted
        iSetupProcessMonitoringThreads;
        processMonitoringThreadStarted = true;
        dctSchedulerMessage(6, 'This process will exit when its parent process dies.');
    end
catch err
    dctSchedulerMessage(0, 'Unexpected error setting up process monitor. Error returned:\n', err);
    parallel.internal.cluster.exitOnException(1);
end
end

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

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iCallMpiInitIfParallel()
% Deal with MDCE_PARALLEL, if set - this is the one place where the MPI
% code is told to initialize and hang off MPI_COMM_WORLD. Note that
% local scheduler job will connect/accept off this later in the decode
% function. Thus it is essential that this occur before we call the decode
% function
if ~isempty( getenv( 'MDCE_PARALLEL' ) )
    try
        dctSchedulerMessage(6, 'About to initialize MPI.');
        mpiInit;
        dctSchedulerMessage(6, 'About to call mpiParallelSessionStarting.');
        mpiParallelSessionStarting;
        dctSchedulerMessage(6, 'Finished initializing MPI.');
    catch err
        dctSchedulerMessage(0,  'Error initializing MPI', err );
        parallel.internal.cluster.exitOnException(1);
    end
end
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function runprop = iBuildRunProp(decodeArguments)
% Create a runprop object to hold the information necessary to create the
% taskrunner and proxy objects
runprop = distcomp.runprop;
% File tasks should default to exiting when finished
runprop.ExitOnTaskFinish = true;
% Since we are going to exit when the task is finished, we will need to also clean
% up our dependency dir, otherwise it's going to get left behind
runprop.CleanUpDependencyDirOnTaskFinish = true;
runprop.DecodeArguments = decodeArguments;
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iCallDecodeFunction(decodeFunction, decodeFunctionString, runprop)
try
    dctSchedulerMessage(6, 'About to call decode function.');
    % The decode function will fill in the correct parts of the runprop object
    % and we can then use this to make storage and task runner objects
    decodeFunction(runprop);
catch err
    if strcmp(err.identifier, 'MATLAB:UndefinedFunction') && isempty(err.stack)
        % Turns out that the function handle wasn't valid, so log this information.
        % NB if the decodeFunction itself contains code that caused an UndefinedFunction
        % error, the error stack will not be empty, and the "Error caught in decode function"
        % will appear in the log.
        dctSchedulerMessage(0, ['Decode function "%s" does not represent a valid MATLAB ', ...
                            'function on the current matlab path'], decodeFunctionString);
    else
        dctSchedulerMessage(0, 'Error caught in decode function ("%s")', decodeFunctionString);
        dctSchedulerMessage(0, 'Error returned was:\n', err);
    end
    parallel.internal.cluster.exitOnException(1);
end
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
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 storage = iBuildStorage(decodeFunctionString, runprop)
dctSchedulerMessage(2, 'About to construct the storage object using constructor "%s" and location "%s"', runprop.StorageConstructor, runprop.StorageLocation);
try
    storageConstructor = str2func(runprop.StorageConstructor);
    % We'll test that we got a valid function handle later when 
    % we call it.
catch err
    dctSchedulerMessage(0, ['Error converting the StorageConstructor to a function handle.\n' ...
                        'This is probably because the value supplied in the decode function ("%s") was empty.\n' ...
                        'The StorageConstructor''s current value is "%s"'], decodeFunctionString, runprop.StorageConstructor);
    dctSchedulerMessage(0, 'Error returned was:\n', err);
    parallel.internal.cluster.exitOnException(1);
end
try
    storage = storageConstructor(runprop.StorageLocation);
catch err
    if strcmp(err.identifier, 'MATLAB:UndefinedFunction') && isempty(err.stack)
        % Turns out that the function handle wasn't valid, so log this information.
        % NB if the storageConstructor itself contains code that caused an UndefinedFunction
        % error, the error stack will not be empty, and the "Error finding the StorageLocation"
        % will appear in the log.
        dctSchedulerMessage(0, ['Returned value for StorageConstructor ("%s") does not represent a valid\n' ...
                            'MATLAB function on the current matlab path\n' ...
                            'This value was supplied by the decode function "%s"'], runprop.StorageConstructor, decodeFunctionString);
    else
        dctSchedulerMessage(0, sprintf(['Error finding the StorageLocation. This is probably \n'...
                            'because the StorageLocation ("%s") \n'...
                            'from the decode function does not exist.\n'...
                            'The error thrown by the StorageConstructor was :'], runprop.StorageLocation), err);
    end
    parallel.internal.cluster.exitOnException(1);
end
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function sched = iCreateSchedulerWithFallback(runprop, storage)
schedulerType = runprop.LocalSchedulerName;
fallbackSchedType = runprop.FallbackSchedulerName;
try
    sched = iCreateSchedulerObject(schedulerType, storage);
catch err
    dctSchedulerMessage(0, sprintf('Error constructing local scheduler object of type %s. The error thrown was:' , schedulerType), err);
    % See if there is a sensible fall back option for this particular scheduler/cluster type.
    if ~isempty(fallbackSchedType)
        dctSchedulerMessage(2, 'Trying to construct the fallback scheduler object of type %s.', fallbackSchedType)
        try
            sched = iCreateSchedulerObject(fallbackSchedType, storage);
        catch err
            dctSchedulerMessage(0, 'Error constructing local scheduler object. The error thrown was ', err);
            parallel.internal.cluster.exitOnException(1);
        end
    else
        parallel.internal.cluster.exitOnException(1);
    end
end
sched.HasSharedFilesystem = runprop.HasSharedFilesystem;
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iApplyEntitlement(sched, entitlement)
try
    dctSchedulerMessage( 2, 'About to set license number on current cluster to ''%s'' with web id ''%s''', ...
                         entitlement.LicenseNumber, entitlement.Id );
    sched.RequiresOnlineLicensing = true;
    sched.hSetLicenseEntitlementNoCheck(entitlement);
catch err
    dctSchedulerMessage(0, 'Error setting license information on cluster.');
    dctSchedulerMessage(0, 'Error returned was', err);
    dctSchedulerMessage(0, 'Job submission on this worker will fail using the current cluster.');
    % Don't call exitOnException here because it's pretty likely that 
    % the job will run successfully so long as it doesn't try to submit
    % a job using the current cluster.  If they do, an error will be
    % thrown by the MDCSDesktopClient.
end
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function [job, task] = iSetupJobAndTaskOnRunProp(handlers, mdceDebugEnabled, outputWriterStack, ...
                                   isSyncTaskEvaluation, root, runprop, sched)
try
    dctSchedulerMessage( 2, ['About to find job and task using ', ...
                        'locations "%s" and "%s"'], ...
                         runprop.JobLocation, runprop.TaskLocation );
    [job, task] = sched.hGetJobAndTaskFromLocations( runprop.JobLocation, ...
                                                     runprop.TaskLocation );
    task.hSetMatlabLogDir();
    worker = parallel.cluster.CJSWorker.hBuildForCurrentProcess();

    % Select which evaluator to use
    captureDiary = task.CaptureDiary;
    if isSyncTaskEvaluation
        % TODO: This is a workaround for a problem with SOA jobs. See g846800 and g846801.
        runprop.TaskEvaluator = parallel.internal.evaluator.EvalcEvaluator( captureDiary );
    else
        if captureDiary
            filename = task.hGetDiaryFile( runprop.TaskLocation );
            dctSchedulerMessage( 4, 'Setting the TaskEvaluator to the CJSStreamingEvaluator');
            runprop.TaskEvaluator = parallel.internal.evaluator.CJSStreamingEvaluator( outputWriterStack, filename );
        else
            dctSchedulerMessage( 4, 'Setting the TaskEvaluator to the NullRedirectionEvaluator');
            runprop.TaskEvaluator = parallel.internal.evaluator.NullRedirectionEvaluator( outputWriterStack );
        end
    end

    root.setCurrentTaskInfo(sched, worker, job, task, runprop, handlers);
    filename = sched.hGetTaskJavaLogIfPossible( task );
    if mdceDebugEnabled && ~isempty( filename )
        iEnableJavaLogging( filename );
    end
    
    % Set environment variables attached to the job.
    runprop.EnvironmentVariablesMap = job.EnvironmentCredentials;
catch err
    dctSchedulerMessage(0, 'Error setting current job and task info');
    dctSchedulerMessage(0, 'Error returned was', err);
    parallel.internal.cluster.exitOnException(1);
end
end
%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function runprop = iSetup(handlers, mdceDebugEnabled, outputWriterStack, isSyncTaskEvaluation, decodeArguments)
% NOTE - this function can be called multiple times in an SOA type
% infrastructure and it is essential that it deal with these multiple calls
dctSchedulerMessage(6, 'Enter distcomp_evaluate_filetask_core/iSetup');

iExecuteHook('MDCE_QE_PRE_JOB_SETUP');

%%%%%%%%%%%%%%%%%%%%
% Per-worker-process setup phases:
% Turn off beep for workers.
beep('off');
iSetupProcessExitAndMatlabRef();
iMaybeSetupProcessMonitoringThreads();
[decodeFunction, decodeFunctionString] = iGetAndCacheDecodeFunction();

%%%%%%%%%%%%%%%%%%%%

%%%%%%%%%%%%%%%%%%%%
% Job pre-setup phase - MPI_Init and decode function:
iCallMpiInitIfParallel();
runprop = iBuildRunProp(decodeArguments);
iCallDecodeFunction(decodeFunction, decodeFunctionString, runprop);
%%%%%%%%%%%%%%%%%%%%

%%%%%%%%%%%%%%%%%%%%
% Build the cluster object:
% Get the root object we are going to set against
root = distcomp.getdistcompobjectroot;
% Only construct scheduler and storage if this is the first task -
% otherwise re-use the existing one
if runprop.IsFirstTask
    [reallyUsingMHLM, entitlement] = iSetupMhlmAndDesktopClient();
    storage = iBuildStorage(decodeFunctionString, runprop);
    sched   = iCreateSchedulerWithFallback(runprop, storage);
else
    % Get scheduler to use from the root.
    sched = root.CurrentJobmanager;
end
if runprop.IsFirstTask && reallyUsingMHLM
    iApplyEntitlement(sched, entitlement);
end
%%%%%%%%%%%%%%%%%%%%


%%%%%%%%%%%%%%%%%%%%
% Build the job and task.
% These are set up on the runprop object as well as returned.
job = iSetupJobAndTaskOnRunProp(handlers, mdceDebugEnabled, outputWriterStack, ...
    isSyncTaskEvaluation, root, runprop, sched);
%%%%%%%%%%%%%%%%%%%%

%%%%%%%%%%%%%%%%%%%%
% Set the number of threads to use.
iSetupComputationalThreads(job.NumThreads);
%%%%%%%%%%%%%%%

runprop.MATLABDriveEnabled = strcmpi(getenv('MDCS_MATLAB_DRIVE_ENABLED_ON_WORKER'), 'true');
dctSchedulerMessage(4, 'MATLAB Drive Enabled %d', runprop.MATLABDriveEnabled);

iExecuteHook('MDCE_QE_POST_JOB_SETUP');
dctSchedulerMessage(2, 'Completed pre-execution phase');
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function sched = iCreateSchedulerObject(type, storage)
	ctor  = str2func( type );
    % This will error with any type other than TaskRunner
	sched = ctor( storage );
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iExecuteHook(hookName)
try
    hookStr = getenv(hookName);
    % Nothing defined for this hook - return immediately
    if isempty(hookStr)
        return
    end
    dctSchedulerMessage(2, 'Hook %s is defined : About to eval string\n%s', hookName, hookStr);
    try
        eval(hookStr);
    catch err
        dctSchedulerMessage(1, sprintf('Hook %s threw an error. The error was:', hookName), err);
    end
catch err %#ok<NASGU>
    % Do nothing if there is an error - this is a QE function only
end
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iSetupProcessMonitoringThreads
% On windows we are likely to be parented by worker.bat, on unix we will have
% exec'd into the shell process. Thus under mpiexec we expect the following
% process trees
%
% win ...  smpd - worker.bat - matlab
% unix ... smpd - MATLAB - matlab_helper

% Have we been asked to watch a particular pid rather than our parent?
pidToWatch = getenv('MDCE_PID_TO_WATCH');
if ~isempty(pidToWatch)
    % Try converting to a double ...
    pidToWatch = str2double(pidToWatch);
    % Special PID to watch that says don't watch
    if pidToWatch == 0
        return
    end
    % Did we get anything sensible?
    if isfinite(pidToWatch) && uint32(pidToWatch) == pidToWatch && pidToWatch > 0
        % If we are successfully watching this one do we need to watch our parent?
        dct_psfcns('pidwatch', pidToWatch)
        return
    end
end

if ispc
    % Start the monitor to ensure that when our parent process terminates (which
    % will be the command-shell or shell/batch script running MATLAB), we
    % terminate the MATLAB process.
    dct_psfcns( 'winparentprocesscheck' );
end

end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iEnableJavaLogging(filename)
% Attach a FileHandler to the distcomp package logger. In particular this 
% means that log messages from the pmode infrastructure get captured.
append = true;
levelNum = 6; % Maximum amount of logging
handler = com.mathworks.toolbox.parallel.pctutil.logging.DistcompFileHandler(filename, append, levelNum);
logger = com.mathworks.toolbox.parallel.pctutil.logging.RootLog.LOG;
logger.addHandler(handler);
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iAllowDeployedWorkerToExit()
% Allow the deployed worker to exit. 
% If this was a parallel pool job this means the worker process can now exit. If 
% it was a normal job then the worker process will exit when 
% distcomp_evaluate_filetask returns.
if isdeployed
    parallel.internal.cluster.workerShutdown('allow');
end
end
