% MJSCommunicatingJob Communicating job for MJS clusters
%
%   parallel.job.MJSCommunicatingJob methods:
%      cancel                - Cancel a pending, queued, or running job
%      createTask            - Create a new task in a job
%      delete                - Remove a job object from its cluster and memory
%      diary                 - Display or save text of batch job
%      fetchOutputs          - Retrieve output arguments from all tasks in a job
%      findTask              - Find task objects belonging to a job
%      listAutoAttachedFiles - List the files that are automatically attached to the job
%      load                  - Load workspace variables from batch job
%      recreate              - Create a new job from an existing job
%      submit                - Submit job for execution in the cluster
%      wait                  - Wait for job execution to change state
%
%   parallel.job.MJSCommunicatingJob properties:
%      AdditionalPaths       - Folders to add to MATLAB search path of workers
%      AttachedFiles         - Files and folders that are sent to the workers
%      AuthorizedUsers       - Users authorized to access job
%      AutoAddClientPath     - Determines whether user-added-entries on the client's path are added to each worker's path
%      AutoAttachFiles       - Determines whether dependent files will be automatically sent to the workers
%      CreateDateTime        - Date and time at which this job was created
%      EnvironmentVariables  - Names of environment variables that are sent to the workers
%      FinishDateTime        - Date and time at which this job finished running
%      FinishedFcn           - Callback executed when the job or task finishes
%      ID                    - Job's numeric identifier
%      JobData               - Data made available to all workers for job's tasks
%      Name                  - Name of this job
%      NumThreads            - Number of computational threads for workers
%      NumWorkersRange       - Specify limits for number of workers
%      Parent                - Cluster object containing this job
%      QueuedFcn             - Callback executed when the job is queued
%      RestartWorker         - True if MATLAB workers are restarted before evaluating the job's tasks
%      RunningFcn            - Callback executed when the job or task starts running
%      RunningDuration       - The duration the job has been running for, if it has started
%      StartDateTime         - Date and time at which this job started running
%      State                 - State of the job: pending, queued, running, finished, or failed
%      SubmitDateTime        - Date and time at which this job was submitted
%      Tag                   - Label associated with this job
%      Tasks                 - Array of task objects contained in this job
%      Timeout               - Time limit to complete job
%      Type                  - Job's type: Independent, Pool, or SPMD
%      UserData              - Data associated with a job object
%      Username              - Name of the user who created this job

% Copyright 2011-2021 The MathWorks, Inc.

classdef ( ConstructOnLoad = true ) MJSCommunicatingJob < parallel.CommunicatingJob & ...
                             parallel.internal.cluster.MJSQueuedCallbackMixin & ...
                             parallel.internal.cluster.MJSRunningCallbackMixin & ...
                             parallel.internal.cluster.MJSFinishedCallbackMixin & ...
                             parallel.internal.cluster.MJSJobMixin

    properties ( PCTGetSet, Transient, PCTConstraint = 'logicalscalar|presubmission' )       
        %RestartWorker True if MATLAB workers are restarted before evaluating the job's tasks        
        %   In some cases, you might want to restart MATLAB on the workers
        %   before they evaluate any tasks in a job. This action resets
        %   defaults, clears the workspace, frees available memory, and so
        %   on.
        %
        %   Set RestartWorker to true if you want the job to restart the
        %   MATLAB session on any workers before they evaluate their first
        %   task for that job. The workers are not reset between tasks of
        %   the same job. Set RestartWorker to false if you do not want
        %   MATLAB restarted on any workers. The default value is false,
        %   which does not restart the workers.
        %
        %   See also parallel.job.MJSCommunicatingJob.submit
        RestartWorker
    end

    properties ( PCTGetSet, Transient, PCTConstraint = 'positivescalar|presubmission' )
        %Timeout Time limit to complete job        
        %   Timeout holds a double value specifying the number of seconds
        %   to wait before giving up on a job. The time for timeout begins
        %   counting when the job state property value changes from queued
        %   to running. When a job times out, the behavior of the job is
        %   the same as if the job were stopped using the cancel function,
        %   except all pending and running tasks are treated as having
        %   timed out.
        %
        %   The default value for Timeout is large enough so that in
        %   practice, jobs will never time out. You should set the value of
        %   Timeout to the number of seconds you want to allow for
        %   completion of jobs.
        %
        %   See also parallel.job.MJSCommunicatingJob.State,
        %            parallel.job.MJSCommunicatingJob.submit.
        Timeout
    end
    
    properties ( PCTGetSet, Transient, PCTConstraint = 'cellstr' )
        %AuthorizedUsers Users authorized to access job
        %   The AuthorizedUsers property value is a cell array of strings
        %   which identify the users who are allowed to access the job.
        %   This controls who can set properties on the job, add tasks,
        %   destroy the job, etc. The person identified as the owner by the
        %   job's Username property does not have to be listed in the
        %   AuthorizedUsers property value.
        %
        %   The effect of AuthorizedUsers at different security levels is:
        %   0       - No effect. All users can access the job without
        %             hindrance.
        %   1       - For users included in the property value, the system
        %             suppresses the dialog box that requires
        %             acknowledgment that the job belongs to another user.
        %             All other users must acknowledge job ownership every
        %             time they access the job.
        %   2 and 3 - Only users who are authenticated in this session and
        %             are listed in AuthorizedUsers can access the job.
        %
        %   See also parallel.job.MJSCommunicatingJob.Username,
        %            parallel.cluster.MJS.SecurityLevel.
        AuthorizedUsers
    end
    
    properties (Constant, Access = private)
        StateEnumToEventName = iBuildStateEnumToEventMap();
        
        % This delay should be long enough that we don't overwhelm the job
        % manager even if all workers in the cluster are polling at the
        % same time. Since the job manager can process ~10,000 calls/second
        % and we want to support up to ~10,000 workers at once, a value of
        % ~1 second is appropriate.
        PollDelaySeconds = 1;
    end
    
    methods ( Hidden )
        function obj = MJSCommunicatingJob( parent, jobID, pvar, supportID, support )
            parallel.Job.loadCheck( nargin );
            obj@parallel.CommunicatingJob( parent, jobID, pvar );
            obj@parallel.internal.cluster.MJSJobMixin( support, supportID );
        end
        function t = hBuildChild( obj, tid, uuid )
            try
                t = parallel.task.MJSTask( obj, tid, uuid, obj.Support, obj.SupportID );
            catch E
                throw( distcomp.handleJavaException( obj, E ) );
            end
        end
        function hPromote( obj )
            obj.Support.promoteJob( obj.SupportID );
        end        
        function hDemote( obj )
            obj.Support.demoteJob( obj.SupportID );
        end
        function [jobPropertyMap, propNames]  = hGetDisplayItems(obj, diFactory)
            jobPropertyMap = hGetDisplayItems@parallel.CommunicatingJob(obj, diFactory);

            % The order the properties appear in the propNames array is the
            % order in which they will be displayed.
            propNames = {...
                'ID', ...
                'Type', ...
                'Username', ...
                'State', ...
                'SubmitDateTime', ...
                'StartDateTime', ...
                'RunningDuration', ...
                'NumWorkersRange', ...
                'NumThreads', ...
                'Separator', ...
                'AutoAttachFiles', ...
                'Auto Attached Files', ...
                'AutoAddClientPath', ...
                'AttachedFiles', ...
                'AdditionalPaths',...
                'EnvironmentVariables',...
                };
        end
    end

    % MJSCallbackMixin methods
    methods ( Access = protected )
        function doRegisterForEvents( obj )
            obj.Support.registerForJobEvents( obj.Parent, obj.SupportID );
            obj.enableAutoEventUnregistration();
        end
        function doUnregisterForEvents( obj )
            obj.Support.unregisterForJobEvents( obj.Parent, obj.SupportID );
            obj.disableAutoEventUnregistration();
        end
        function doUnregisterForEventsAfterDestroyed( obj )
            obj.Support.unstashCluster( obj.Parent )
        end
    end

    methods( Hidden )
        function hFireStateEvent( obj, stateEnum )
            notify( obj, obj.StateEnumToEventName(stateEnum.Name) );
        end
    end

    methods ( Hidden )
        function hMpiInit( job, task )
            iMpiInit( job, task );
        end
    end

    methods ( Access = protected )
        function t = createTaskOnOneJob( job, varargin )
            if isempty( job.Tasks )
                allowMultipleTasks = false;
                t = job.doCreateTaskOnOneJob( allowMultipleTasks, varargin{:} );
            else
                error(message('parallel:job:CommunicatingJobOneTask'));
            end
        end

        function preSubmit( obj )
            obj.hErrorIfReleaseMismatch( 'parallel:job:SubmitVersionMismatch' );
            preSubmit@parallel.Job( obj );
        end

        function submitOneJob( job )
            job.Parent.hJobPreSubmitCheck( job );
            try
                job.Support.submitJob( job, job.SupportID );
            catch E
                throw( distcomp.handleJavaException( job, E ) );
            end
        end
        
        function newJob = recreateOneJob( oldJob, varargin )
            cluster = oldJob.Parent;
            % Type is necessary for communicating jobs only
            jobPVPairs = extractPVPairsFromExistingObject( oldJob, [ {'Type'}, getMJSJobPropsForJobRecreation() ] );
            newJob = cluster.createCommunicatingJob( jobPVPairs{:} );
            jobHiddenPVPairs = extractPVPairsFromExistingObject( oldJob, getCommonHiddenJobPropsForRecreation() ); 
            if numel(jobHiddenPVPairs) >= 2
                newJob.hSetProperty( jobHiddenPVPairs(1:2:end), jobHiddenPVPairs(2:2:end) );
            end
            % Communicating jobs only have one task
            taskPVPairs = extractCreationArgsFromExistingTask( oldJob.Tasks(1), getMJSTaskPropsForJobRecreation() );
            newJob.createTask( taskPVPairs{:});
        end

        function destroyJob( jobOrJobs )
            hDelete( jobOrJobs );
            arrayfun(@unregisterForEventsAfterDestroyed, jobOrJobs );
        end

        function cancelJob( jobOrJobs, cancelExceptions )
            hCancel( jobOrJobs, cancelExceptions );
        end

        function ok = doWait( job, stateEnum, timeout )
            waitRequiredFcn = iCreateStateComparisonFcn( stateEnum );
            ok = job.waitForEvent( job.StateEnumToEventName(stateEnum.Name), timeout, waitRequiredFcn );
        end
        
        function attachFileDataToJob( job, filedata )
            try
                job.Support.setJobProperties( job.SupportID, 'AttachedFileData', filedata );
            catch E
                throw( distcomp.handleJavaException( job, E ) );
            end
        end
        
        function supportVec = checkVectorizedSetSupport( ~ )
            % Override checkVectorizedSetSupport method inherited from
            % CustomGetSet to indicate that vectorized set calls are
            % supported. This is done to minimise remote calls.
            supportVec = true;
        end
    end

    % API1 compatibility
    methods ( Hidden )
        function pPreJobEvaluate( job, task )
            % Perform parallel language initialization
            preMJSCommunicatingJob( job, task );
        end

        function pPostJobEvaluate( job )
            if ~job.PoolShutdownSuccessful
                % NB: never get here for 'SPMD' jobs
                dctSchedulerMessage( 0, 'About to exit MATLAB because pool shutdown was unsuccessful' );
                parallel.internal.cluster.exitOnException( 0 );
            end
            
            % Clean up MPI if it got initialized
            mpiCleanup();
        end

        function info = pGetDataStoreInfo( job )
            info = job.Support.getJobDataStoreInfo;
        end
    end
end
%#ok<*ATUNK> custom attributes

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% c.f.: @distcomp/@paralleljob/pMpiInit.m
function iMpiInit( job, task )

% Force the "sock" option, as this is the only one that will work under the
% jobmanager.
    mpiLibrary = job.Parent.MpiLibraryName;
    mpiInit( mpiLibrary );
    dctSchedulerMessage(4, 'mpiInit(''%s'') completed ok', mpiLibrary);

    % This is a new parallel session
    mpiParallelSessionStarting;

    % "Leading task" is assigned by the JM during dispatch of the job.
    % Find out if this task is the leading task.
    leadingTaskID = job.hGetProperty( 'LeadingTaskNum' );
    assert( leadingTaskID > 0, 'The leading task ID should be greater than 0' );
    isleadingtask = task.ID == leadingTaskID;

    if isleadingtask
        USING_LINEAR_CONN   = strcmpi(getenv('MDCE_CONNECTACCEPT'), 'linear');
        USING_BIN_TREE_CONN = ~USING_LINEAR_CONN;

        if USING_LINEAR_CONN
            % Open the port and set the parallel tag
            port = mpigateway( 'openport' );
            connectMethod = 'linear';
        elseif USING_BIN_TREE_CONN
            port = 'no port';
            connectMethod = 'binary';
        end
        % Set the parallel tag
        job.hSetPropertyNoCheck( 'ParallelTag', ...
                                 sprintf( '%s\n%s\n%s', connectMethod, port, computer() ) );
        dctSchedulerMessage(4, 'Leading task parallel tag ... method : %s, port : %s, comp : %s', connectMethod, port, computer());
    else
        % Get the parallel tag, waiting until it has been set
        parallelTag = job.hGetProperty( 'ParallelTag' );
        while isempty( parallelTag )
            pause( parallel.job.MJSCommunicatingJob.PollDelaySeconds ); % A shortish wait
            parallelTag = job.hGetProperty( 'ParallelTag' );
        end

        try
            [connectMethod, port, comp] = iDecodeParallelTag( parallelTag );
            dctSchedulerMessage(4, 'parallel tag decode: method : %s, port : %s, comp : %s', connectMethod, port, comp);
        catch err
            dctSchedulerMessage(4, 'Error in iDecodeParallelTag:', err );
            rethrow(err);
        end
        
        iCompatCheck( comp );
    end
    
    % Based on the parallel tag we decide if we are doing a binary or linear
    % connection method.
    if strcmpi(connectMethod, 'binary')
        iBinaryTreeConnectAccept(job, task, leadingTaskID);
    elseif strcmpi(connectMethod, 'linear')
        iLinearConnectAccept(job, task, port, isleadingtask);
    else
        error(message('parallel:job:MpiUnknownConnectionMethod', connectMethod));
    end
end
% -------------------------------------------------------------------------
function iBinaryTreeConnectAccept(job, task, leadingTaskID)
    import parallel.internal.cluster.ParallelPortInfo
    
    dctSchedulerMessage(4, 'Doing a binary tree connection');

    N = job.hGetProperty( 'NumLabs' );
    dctSchedulerMessage(4, 'Got parallel info with num labs: %d', N);

    % Create a CancelWatchdog to make sure it doesn't take us too long to
    % connect/accept
    cw = iBuildCancelWatchdog( N );

    try
        % Get my ID
        ID = task.Id;
        getPortFcn = @() iPollForParallelPortInfo( task.PreviousTask );
        openPortFcn = @() mpigateway( 'openport' );
        setPortFcn = @(portToUse) task.hSetProperty( 'ParallelPortInfo', ParallelPortInfo( task.Attempt, portToUse ) );

        iJoinRing( ID, N, getPortFcn, openPortFcn, setPortFcn, leadingTaskID );
        cw.ok();
    catch err
        dctSchedulerMessage(2, 'Caught error during iBinaryTreeConnectAccept. Report is:', err );
        cw.ok();
        rethrow(err)
    end
end

% --------------------------------------------------------------------------
function iJoinRing( myID, N, getPortFcn, openPortFcn, setPortFcn, leadingTaskID )
% This function joins an MPI ring together using oneBinaryTreeConnectAccept in a
% loop. A full description of the algorithm can be found in that function.
    

% Make the IDs zero-based for simplicity
    ID = myID - leadingTaskID;

    % How many connect / accept rounds are going to be needed
    nRounds = ceil(log2(N));
    dctSchedulerMessage(5, 'Expecting %d rounds of connect/accept', nRounds);
    
    for i = 1:nRounds
        parallel.internal.mpi.oneBinaryTreeConnectAccept(ID, N, i, getPortFcn, openPortFcn, setPortFcn);
    end
    dctSchedulerMessage(5, 'Finished %d rounds of connect/accept', nRounds);

end

% --------------------------------------------------------------------------
function port = iPollForParallelPortInfo( task )
    parallelPortInfo = task.hGetProperty( 'ParallelPortInfo' );
    while isempty( parallelPortInfo ) || parallelPortInfo.Attempt ~= task.Attempt
        pause( parallel.job.MJSCommunicatingJob.PollDelaySeconds ); % A shortish wait
        parallelPortInfo = task.hGetProperty( 'ParallelPortInfo' );
    end
    port = parallelPortInfo.Value;
end

% -------------------------------------------------------------------------
function iLinearConnectAccept(job, task, port, isleadingtask)
    dctSchedulerMessage(4, 'Doing a linear connection');
    % Everyone needs to know the expected world size and the parallel tag.
    N = job.hGetProperty( 'NumLabs' );

    % Create a CancelWatchdog to make sure it doesn't take us too long to
    % connect/accept
    cw = iBuildCancelWatchdog( N );

    err = [];
    try
        if isleadingtask
            % accept the other connections - with timeout
            mpigateway( 'servaccept', port, N-1 );
        else
            % Connect up
            mpigateway( 'clientconn', port, N-1 );
        end
    catch exception
        err = exception;
    end
    % We're OK - so call off the watchdog.
    cw.ok();

    % If connect/accept threw an error, rethrow that.
    if ~isempty( err )
        rethrow( err );
    end

    % Set the labindex as desired
    tasksByIndex = gcat( task.Id );
    [junk, desOrder] = sort( tasksByIndex ); %#ok
    mpigateway( 'desiredlabordering', desOrder );
end
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% iCompatCheck - check compatibility of computers. The criteria are that the
% word sizes and endianness must match, but since MJS chooses same-OS
% workers for communicating jobs anyway, it is sufficient to just check
% that the computer types match.
function iCompatCheck( comp )

% We should never get a badly formatted tag, since the job is locked when
% being modified.
    if isempty( comp )
        error(message('parallel:job:MpiInitBadTag'));
    end

    mycomp = computer;
    compatible = strcmp(comp, mycomp);

    if ~compatible
        error(message('parallel:job:MpiCompatibilityCheckFailed', mycomp, comp));
    end
end

function [connectMethod, port, comp] = iDecodeParallelTag( ptag )
% Check that the this computer is compatible with the computer which opened
% the port
    strs = regexp(ptag, '\n+', 'split');
    [connectMethod, port, comp] = deal(strs{:});
end

function m = iBuildStateEnumToEventMap()
    import parallel.internal.types.States;
    m = containers.Map;
    m(States.Queued.Name)   = parallel.internal.cluster.MJSQueuedCallbackMixin.QueuedEventName;
    m(States.Running.Name)  = parallel.internal.cluster.MJSRunningCallbackMixin.RunningEventName;
    m(States.Finished.Name) = parallel.internal.cluster.MJSFinishedCallbackMixin.FinishedEventName;
end

function fcn = iCreateStateComparisonFcn( stateEnum )
% Returns a function that will check the state of the job when called.
% Created here to prevent the job object being sent as part of the
% workspace.
    fcn = @( obj ) obj.StateEnum < stateEnum;
end

function cw = iBuildCancelWatchdog( numLabs )
% Creates a CancelWatchdog used to make sure it doesn't take too long to
% connect/accept. Choose a unique-ish exit code for identifying this error
% mode and a timeout based on the expected number of labs.
    exitCode = 72;
    timeout = max( 120, 4 * numLabs );
    cw = parallel.internal.shared.CancelWatchdog( exitCode, timeout );
    dctSchedulerMessage( 4, 'MPI connect watchdog started with timeout %d seconds', timeout );
end
