function distcomp_evaluate_task( outputWriterStack, jobManagerLookupURL, jobManagerName, workerProxy, ...
                                 jobAndTask, jobType, performJobInit, justStarted, ...
                                 taskLogLevel, serializedCredentials , portSelectionInformation )
; %#ok Undocumented
% 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-2017 The MathWorks, Inc.

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 jobmanager proxy for this worker
jobManagerProxy = parallel.internal.cluster.JobManagerFinder.getOneJobManagerProxy(jobManagerLookupURL, jobManagerName);

% Configure the TaskHandler. Cleaned up in iFinishTask()
iSetupTaskHandler(jobManagerProxy, jobAndTask, taskLogLevel);

hbListener = iSetupHeartbeatListener(workerProxy, jobAndTask);

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

% Performs MATLAB Distributed Computing 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}, ...
                {@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];

    % 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 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, postFcns)

try
    finishFcn = @(job, task, out) iFinishTask(handlers, job, task, 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, out)

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 ~isempty(out.errOutput)
            try
                dctSchedulerMessage(4, 'About to pSubmitResult from parallel task that errored');
                task.pSubmitResult(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. Worker must not submit a result.
            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 pSubmitResult');
    task.pSubmitResult(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

iClearTaskHandler();

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

    if performJobInit                
        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
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 = 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');
    % Whatever mode the JobManager is in wrt RMI exporting, we want to use plain RMI
    jobManagerProxy.forceVanillaRmi(true);

    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
    % The communication check will already have been performed when constructing the MJS,
    % so use the cached results.
    forceCheck = false;
    if ~jobmanager.hCheckTwoWayCommunications( forceCheck )
        dctSchedulerMessage(1, 'Communication with jobmanager failed')
        error( message('parallel:cluster:EvaluateTaskCommunicationFailure') );
    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.NullEvaluator(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 iSetupTaskHandler(jobManagerProxy, jobAndTask, taskLogLevel)
dctSchedulerMessage(4, 'Entering iSetupTaskHandler');
taskHandler = iGetTaskHandler();
taskHandler.setTaskAccess(jobManagerProxy.getTaskAccess());
taskHandler.setTaskIDAndLevel(jobAndTask.getTaskID(), taskLogLevel);
end

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

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iClearTaskHandler()
taskHandler = iGetTaskHandler();
taskHandler.flush();
taskHandler.clearTaskIDAndLevel();
end

%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function iRemoveHeartbeatListener(hbListener)
hbListener.unregister();
end
%--------------------------------------------------------------------------
%
%--------------------------------------------------------------------------
function taskHandler = iGetTaskHandler()
persistent theTaskHandler;
if isempty(theTaskHandler)
    % First try and find one attached to the logger
    logger = com.mathworks.toolbox.distcomp.mjs.worker.PackageInfo.LOGGER;
    handlers = logger.getHandlers();
    for n = 1:length(handlers)
        if isa(handlers(n), 'com.mathworks.toolbox.distcomp.mjs.worker.TaskHandler')
            dctSchedulerMessage(6, 'Found existing TaskHandler')
            theTaskHandler = handlers(n);
            break;
        end    
    end
    % If failed to find one, create a new one and add it to the logger.
    if isempty(theTaskHandler)        
        dctSchedulerMessage(6, 'Creating new TaskHandler')
        theTaskHandler = com.mathworks.toolbox.distcomp.mjs.worker.TaskHandler();
        logger.addHandler(theTaskHandler);
    end
end
taskHandler = theTaskHandler;
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
workerProxy.makeJavaWorkerIdle(jobAndTask);
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
