function distcomp_evaluate_task(outputWriterStack, workerProxy, performJobInit, ...
                                justStarted, serializedCredentials, taskExecutionInfo, postFcns)
% Accepts a worker, job, and task and computes and submits
% the results of the task.
%
% performJobInit indicates whether this is the first task that has been run
% from this job by this worker.
%
% justStarted indicates whether this is
% the first task that has been run by this worker since it was last
% started

% Copyright 2004-2019 The MathWorks, Inc.

if nargin < 7 || isempty(postFcns)
    postFcns = {};
end

mlock; % so that the original path persistent variable is saved (see iSetup)

% Remember to set the scheduler message handler.
setSchedulerMessageHandler(javaWorkerMessageHandler());

% Announce that we've started
dctSchedulerMessage(3, 'Begin distcomp_evaluate_task');

% Get the job manager proxy to use for evaluating this task
jobManagerProxy = taskExecutionInfo.getJobManager();

% Retrieve information from taskExecutionInfo
jobAndTask = taskExecutionInfo.getJobAndTaskID();
jobType = taskExecutionInfo.getJobMLType();
portSelectionInformation = taskExecutionInfo.getPortSelectionInformation();
attempt = jobAndTask.getTaskAttemptIdentifier();

hbListener = iSetupHeartbeatListener(workerProxy);

handlers = struct('abortFcn', @nAbortFunction, ...
                  'exitFcn',  @nExitFunction, ...
                  'errorFcn', @nErrorFunction);

% Performs MATLAB Parallel Server specific job initialization work
try    
    runprop = iSetup(jobManagerProxy, workerProxy, jobAndTask, jobType, ...
                 performJobInit, justStarted, serializedCredentials, handlers, outputWriterStack, portSelectionInformation);
catch err
    handlers.errorFcn(err, 'Job setup failed - MATLAB will now exit and restart.');
end

try
    % Define any post processing tasks that we want to evaluate and send them in
    % to the run function - this allows us to ensure that these will get run when
    % the finish function is called.
    % This function could be called AFTER a task has been dispatched to the
    % worker (which would happen when pSubmitResult is called).
    postFcns = [ ...
        {{@iRemoveHeartbeatListener, hbListener}}, ...
        postFcns, ...
        {{@iMakeJavaWorkerIdle, workerProxy, jobAndTask}}];

    % Test if we should remove the DependencyDirectory at the end
    if runprop.CleanUpDependencyDirOnTaskFinish
        postFcns = [ ...
            {{@removeDirectory, runprop.DependencyDirectory}}, ...
            postFcns];
    end
    % Always clean up the per-host directory
    postFcns = [ ...
        {{@removeDirectory, runprop.PerHostDirectory}}, ...
        postFcns];
    
    % Always clean up parallel pool state if applicable.
    postFcns = [ ...
        {{@iCleanupPoolState}}, ...
        postFcns];

    % Actually do the task evaluation stuff
    iDoTask(handlers, attempt, postFcns);
    % NOTE - do not add any code after the call to iDoTask as this will NOT get
    % executed in the case of an interactive parallel job. Any code that needs to
    % run after the users code has executed should be added into iFinishTask below.
catch err
    handlers.abortFcn(err, 'Unexpected error in DoTask - MATLAB will now exit and restart.');
end

dctSchedulerMessage(1, 'End distcomp_evaluate_task');


%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
    function nErrorFunction(err, varargin)
        % Helper function that will exit matlab if the last error that we detected
        % was related to not finding a job or task. The reasoning behind this is that
        % if a job or task is not found then we are about to be shutdown by the
        % jobmanager, so we might as well do it ourselves.
        try
            if any(strcmp(err.identifier, ...
                {'parallel:task:MJSTaskNotFound', ...
                'parallel:job:MJSJobNotFound', ...
                'parallel:cluster:MJSTaskOrJobNotFound'}))
                dctSchedulerMessage(2, 'MATLAB will now exit and restart as a job or task was destroyed while this worker was executing it');
                handlers.exitFcn();
            else
                handlers.abortFcn(err, varargin{:});
            end
        catch e %#ok<NASGU> - don't want this function to EVER error
            handlers.abortFcn(err, varargin{:});
        end
    end
%--------------------------------------------------------------------------
% Define a nested abort function that retains the relevant information we
% want, such as jWorker, jJobID and jTaskID
%--------------------------------------------------------------------------
    function nAbortFunction(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.getReport);
        catch e %#ok<NASGU> - don't want this function to EVER error
        end
        % Always exit - no matter what
        iExitOnException(workerProxy, jobAndTask, varargin{:});
    end
%--------------------------------------------------------------------------
% Define a nested exit function that retains the relevant information we
% want, such as jWorker, jJobID and jTaskID
%--------------------------------------------------------------------------
    function nExitFunction(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
        iExitOnException(workerProxy, jobAndTask, varargin{:});
    end

end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iDoTask(handlers, attempt, postFcns)

try
    finishFcn = @(job, task, out) iFinishTask(handlers, job, task, attempt, out);
    dctSchedulerMessage(4, 'About to evaluate task.'); 
    dctEvaluateTask(postFcns, finishFcn);
    dctSchedulerMessage(4, 'Finished evaluating task.'); 
catch e
    if isa(e, 'distcomp.ExitException')
        m = message('parallel:cluster:UnexpectedExitFailure', e.getReport);
        handlers.errorFcn(e.CauseException, '%s', m.getString());
    else
        m = message('parallel:cluster:UnexpectedExitFailureTaskNoCause');
        handlers.errorFcn(e, '%s', m.getString());
    end
end

end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iFinishTask(handlers, job, task, attempt, out)

% Set whether the task errored or not so that this information can be
% retrieved (e.g. by MJSCommunicatingJob) without going to the job manager
taskErrored = ~isempty(out.errOutput);
task.hSetErrored(taskErrored);

try
    % If the task is part of a paralleljob and it errored, we try and submit
    % the result and then exit.
    if isa(job, 'distcomp.paralleljob') || isa(job, 'parallel.CommunicatingJob')
        if taskErrored
            try
                dctSchedulerMessage(4, 'About to submit result from parallel task that errored in user code');
                parallel.internal.cluster.submitTaskResult(task, attempt, ...
                    out.output, out.errOutput, out.textOutput, out.warnOutput);
            catch e %#ok<NASGU>
                % Don't worry about being unable to cancel the current task,
                % this might fail if the current task no longer exists
                % We still want to exit normally without leaving log messages
                % because this is a normal code path
            end
            % Normal termination
            handlers.exitFcn();
        end
    end
catch e
    handlers.errorFcn(e, 'Unexpected error in parallel job cleanup - MATLAB will now exit and restart.');
end

try
    dctSchedulerMessage(4, 'About to pPostJobEvaluate');
    job.pPostJobEvaluate();
catch e
    handlers.errorFcn(e, 'Unexpected error in PostJobEvaluate - MATLAB will now exit and restart.');
end

try
    dctSchedulerMessage(4, 'About to submit task result');
    parallel.internal.cluster.submitTaskResult(task, attempt, ...
        out.output, out.errOutput, out.textOutput, out.warnOutput);
catch e
    handlers.errorFcn(e, 'Unable to submit task result - MATLAB will now exit and restart.');
end

pctSetmcrappkeys('', '');

try
    % Only need to do this clean-up in the API1 case
    if isa(task, 'distcomp.task')
        % Need to free up the cached tasks so we don't leak memory
        root = distcomp.getdistcompobjectroot;
        root.removeObjectFromHashtable(task.pReturnUUID);
        delete(task);
    end
catch e % Failure to delete from cache is not catastrophic
    dctSchedulerMessage(1, 'Warning - unable to remove task object from cache:\n%s', e.getReport);
end

end % iFinishTask

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iExitOnException(worker, jobAndTask, varargin)
% This function errored out and the integrity of the running MATLAB can no longer be assumed.
% Terminate MATLAB to force it to be restarted. Print an error message and inform the Java worker
% that MATLAB is being terminated. Give the Java workers about a minute to terminate MATLAB.
% If that does not work, the exit command should be called.
try
    if nargin>2
        msg = sprintf(varargin{:});
    else
        % Fallback message - typically this is from a call to handlers.exitFcn() with no message.
        msg = getString(message('parallel:cluster:DistcompEvaluateTaskTerminateMatlab'));
    end
    % An asynchronous call to inform the worker that the task errored.
    % The worker should notify the jobmanager and terminate this process.
    %
    % The comment line below will be replace by QE in order to enable
    % successful code coverage report generation. This is required
    % due to corruption of profile reports when the workers are
    % aborted and not exited in the normal way - DO NOT DELETE.
    % QE_PCT_WORKER_FINISH
    worker.notifyTaskErrored(jobAndTask, msg);
catch e %#ok<NASGU> - don't want this function to EVER error
end

% Give the Java workers about 60 seconds to terminate MATLAB.
%
% Use Thread.sleep to ensure that any pending IQM requests cannot preempt
% this! pause allows other code to run.
java.lang.Thread.sleep(60);
% Attempt to terminate gracefully by calling the exit command.
dct_psfcns('fastexit', 0)
end % iExitOnException

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function runprop = iSetup(jobManagerProxy, workerProxy, jobAndTask, jobType, ...
                          performJobInit, justStarted, serializedCredentials, ...
                          handlers, outputWriterStack, portSelectionInformation)

dctSchedulerMessage(4, 'Entering iSetup with performJobInit=%d and justStarted=%d', performJobInit, justStarted);

persistent originalpath;
persistent originalWarningState;

% Check the worker and MATLAB agree on the computer type.
workerComputerType = char(workerProxy.getComputerMLType());
if ~strcmp(workerComputerType, computer)
    dctSchedulerMessage(1, 'Worker and MATLAB don''t agree on computer type.');
    error(message('parallel:cluster:ComputerTypeMismatch', workerComputerType, computer));
end

iExecuteHook('MDCE_QE_PRE_JOB_SETUP');

rootDependencyDir = char(workerProxy.getFileDependencyDir);
workerDir = char(workerProxy.getWorkerDir);

if justStarted || performJobInit
    if isempty(originalpath)
        dctSchedulerMessage(4, 'Storing original path and warning state');
        % initialize the original path
        originalpath = path;
        originalWarningState = warning('query', 'all');
    else
        % restore the original path before executing the next job
        dctSchedulerMessage(4, 'Restoring original path and warning state');
        path(originalpath);
        warning(originalWarningState);
    end
              
    dctSchedulerMessage(4, 'Performing job initialization');
    % Always try and delete the file dependencies directory if this is
    % the first task we are running.
    if exist(rootDependencyDir, 'dir')
        try 
            dctSchedulerMessage(4, 'Removing old dependency directory %s', rootDependencyDir);
            rmdir(rootDependencyDir, 's');
        catch err
            % Don't worry too much if we can't do this as it is actually
            % quite likely to happen on windows if the matlab hasn't
            % restarted. At some point we are going to be able to delete
            % the contents of this dir.
            dctSchedulerMessage(6, 'Removing dependency directory failed: %s', err.message);
        end
    end

    % Make sure the worker directory exists.
    if ~exist(workerDir, 'dir')
        dctSchedulerMessage(4, 'Making new worker directory %s', workerDir);
        try 
            mkdir(workerDir); 
        catch err
            dctSchedulerMessage(1, 'Failed to make worker directory');
            rethrow(err);
        end

    end
end

% Ensure that any changes made by mpiprofile to general profiler settings
% have been reverted.
try
    dctSchedulerMessage(4, 'Resetting mpiprofile');
    mpiprofile('reset');
catch err
    dctSchedulerMessage(1, 'Warning - failed to reset profiler state. Error thrown was:\n%s\n', ...
                         err.message);
end

% We wish to ensure that the current working directory of matlab is not a
% UNC path on windows, and possibly that it is writable by this process.
% So lets ensure that we change directory to the local work directory
dctSchedulerMessage(4, 'Changing directory to %s', workerDir);
cd(workerDir);

% Create the correct runprop data
runprop = distcomp.runprop;
% Add PathDependencies if the worker has just restarted or if this is the
% first task in a job this worker has processed
runprop.AppendPathDependencies = justStarted || performJobInit;
% Add FileDependencies if this is the first task in a job
runprop.AppendFileDependencies = justStarted || performJobInit;
% Is this the first task in this job that the worker has executed
runprop.IsFirstTask = justStarted || performJobInit;

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

runprop.PortSelectionInformation = portSelectionInformation;

try
    root = distcomp.getdistcompobjectroot;
    
    % If this isn't the first task in the job that the worker has executed
    % and we're not going to rebuild the AttachedFilesPathMap for some
    % other reason, keep the AttachedFilesPathMap computed previously.
    % See g1565643.
    if ~runprop.IsFirstTask && ~runprop.AppendFileDependencies
        runprop.AttachedFilesPathMap = root.CurrentRunprop.AttachedFilesPathMap;
    end
    
    % Make sure that all the proxies are correctly rooted in my hierarchy
    dctSchedulerMessage(4, 'Creating jobmanager from proxy');
    % Use whichever form of RMI the job manager is configured with. This allows use
    % of PeerRMI which has greater chance of successful SSH handshakes with future java 
    % versions (g2048280)
    jobManagerProxy.forceVanillaRmi(false);

    import com.mathworks.toolbox.distcomp.mjs.auth.credentials.CredentialRole
    % These credentials were serialized without encryption by the worker.
    credentials = serializedCredentials.unpack([],[]);
    
    % Create the appropriate type of desktop client.
    webLicenseInfo = credentials.getCredentialsForRole(CredentialRole.WEB_LICENSE_INFO);
    usingMHLM = ~isempty(webLicenseInfo);
    if usingMHLM
        licenseWebId = char(webLicenseInfo.getLicenseID);
        licenseNumber = char(webLicenseInfo.getLicenseNumber());
        desktopClient = parallel.internal.mwa.MDCSDesktopClient.buildWithFixedToken(...
            char(webLicenseInfo.getWebLicenseToken), char(webLicenseInfo.getWebUserName));
        dctSchedulerMessage(6, 'About to create entitlement using web id %s and license number %s', licenseWebId, licenseNumber);
        entitlement = parallel.internal.mwa.Entitlement(licenseWebId, licenseNumber, '');
    else
        desktopClient = parallel.internal.mwa.MDCSDesktopClient();
    end
    parallel.internal.mwa.currentDesktopClient(desktopClient);
    
    % Set environment variables attached to the job.
    if credentials.hasCredentialsForRole(CredentialRole.ENVIRONMENT_VARIABLES)
        serializedCredentials = credentials.getCredentialsForRole(CredentialRole.ENVIRONMENT_VARIABLES);
        runprop.EnvironmentVariablesMap = iFromJavaMap(serializedCredentials.getObject);
    end

    % Switch between 'cluster' and 'scheduler' based on the jobType.
    import com.mathworks.toolbox.distcomp.mjs.workunit.JobMLType
    % Build an MJS 'Cluster'; because of the different caching strategy, no
    % need to clear up previous jobs.
    jobmanager = parallel.cluster.MJS(jobManagerProxy, credentials);
    if usingMHLM
        jobmanager.hSetLicenseEntitlementNoCheck(entitlement);
    end
    worker = parallel.cluster.MJSWorker(jobmanager, workerProxy);
    dctSchedulerMessage(4, 'Finding job %s and task %s', ...
                        char(jobAndTask.getJobID.toString()), ...
                        char(jobAndTask.getTaskID.toString()));
    [job, task] = jobmanager.hGetJobAndTaskFromUUIDs(jobAndTask.getJobID(), ...
                                                      jobType, ...
                                                      jobAndTask.getTaskID());
    iSetupComputationalThreads(job.NumThreads);
catch err
    dctSchedulerMessage(1, 'Caught error during iSetup: %s', err.message);
    rethrow(err);
end

% Tell it where the DependencyDirectory is located
runprop.DependencyDirectory = fullfile(rootDependencyDir, num2str(job.ID));

% Clean up dependency directory only if the job is a concurrent type.
runprop.CleanUpDependencyDirOnTaskFinish = (isa(job, 'parallel.ConcurrentJob') || ...
                                            isa(job, 'parallel.CommunicatingJob'));

taskAccess = jobManagerProxy.getTaskAccess();
% Choose the evaluator to use.
try
    if task.CaptureCommandWindowOutput
        runprop.TaskEvaluator = parallel.internal.evaluator.MJSStreamingEvaluator(...
            outputWriterStack, jobAndTask.getTaskID, taskAccess);
    else
        runprop.TaskEvaluator = parallel.internal.evaluator.NullRedirectionEvaluator(outputWriterStack);
    end
catch err
    dctSchedulerMessage(1, 'Caught error creating task evaluator: %s', err.message);
    rethrow(err);
end

try
    % Set the properties on the root object
    dctSchedulerMessage(4, 'Setting current task info');
    setCurrentTaskInfo(root, jobmanager, worker, job, task, runprop, handlers);
catch err
    dctSchedulerMessage(1, 'Caught error setting current task info: %s', err.message);
    rethrow(err);
end

iExecuteHook('MDCE_QE_POST_JOB_SETUP');

end %iJobStartup

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

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function hbListener = iSetupHeartbeatListener(workerProxy)
% Create the heartbeat failure listener and give it to the heartbeat manager.
import com.mathworks.toolbox.distcomp.mjs.worker.WorkerNotifyingHeartbeatListener;
hbListener = WorkerNotifyingHeartbeatListener(workerProxy);
hbListener.register();
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iRemoveHeartbeatListener(hbListener)
hbListener.unregister();
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iCleanupPoolState()
try
    delete(gcp('nocreate'));
catch e
    dctSchedulerMessage(2, 'Error cleaning up pool state: %s', e.getReport);
end
end
%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iMakeJavaWorkerIdle(workerProxy, jobAndTask)
try
    preFcn = parallel.internal.apishared.getSetTaskCompleteFcn;
    if ~isempty(preFcn)
        dctSchedulerMessage(4, 'About to execute makeWorkerIdlePreFcn');
        preFcn();
    end
catch e
    dctSchedulerMessage(2, 'Error executing makeWorkerIdlePreFcn: %s', e.getReport);
end
% This MATLAB process can only run subsequent communicating jobs if MPI has never
% been initialized.
canRunCommunicatingJobs = ~feval('_mpiEverInitialized');
workerProxy.makeJavaWorkerIdle(jobAndTask, canRunCommunicatingJobs);
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iSetupComputationalThreads(num)
dctSchedulerMessage(4, 'Setting number of computational threads to %d.', 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

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function matlabMap = iFromJavaMap(javaMap)
% Convert Java Map to containers.Map
matlabMap = containers.Map('KeyType', 'char', 'ValueType', 'char');
keys = javaMap.keySet().toArray();
for idx = 1:numel(keys)
    key = keys(idx);
    value = char(javaMap.get(java.lang.String(key)));
    matlabMap(key) = value;
end
end
