%

% Copyright 2012-2021 The MathWorks, Inc.
classdef ( ConstructOnLoad = true, Hidden ) MJSConcurrentJob < parallel.ConcurrentJob & ...
        parallel.internal.cluster.MJSQueuedCallbackMixin & ...
        parallel.internal.cluster.MJSRunningCallbackMixin & ...
        parallel.internal.cluster.MJSFinishedCallbackMixin & ...
        parallel.internal.cluster.MJSJobMixin

    properties ( PCTGetSet, Transient, PCTConstraint = 'logicalscalar|presubmission' )
        RestartWorker
    end

    properties ( PCTGetSet, Transient, PCTConstraint = 'positivescalar|presubmission' )
        Timeout
    end

    properties ( PCTGetSet, Transient, PCTConstraint = 'cellstr' )
        AuthorizedUsers
    end
    
    %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
    % Public / documented properties
    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 = MJSConcurrentJob( parent, jobID, supportID, support )
            parallel.Job.loadCheck( nargin );
            obj@parallel.ConcurrentJob( parent, jobID );
            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.ConcurrentJob(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', ...
                'Separator',...
                'AutoAttachFiles', ...
                'Auto Attached Files', ...
                'AttachedFiles', ...
                'AutoAddClientPath', ...
                '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

        function functor = getBroadcastFunctor( obj, task )
            if task.ID == 1
                functor = @(data) iSetParallelTag(obj, data);
            else
                functor = @() iGetParallelTag(obj);
            end
        end
    end

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

    methods ( Access = protected )
        function t = createTaskOnOneJob( job, varargin )
            if isempty( job.Tasks )
                allowMultipleTasks = false;
                t = job.doCreateTaskOnOneJob( allowMultipleTasks, varargin{:}, ...
                    'MaxNumRetries', 0 ); % Retrying a concurrent task is unlikely to be useful
            else
                error(message('parallel:job:ConcurrentJobOneTask'));
            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;
            jobPVPairs = extractPVPairsFromExistingObject( oldJob, getMJSJobPropsForJobRecreation() );
            newJob = cluster.createConcurrentJob( jobPVPairs{:} );
            jobHiddenPVPairs = extractPVPairsFromExistingObject( oldJob, getCommonHiddenJobPropsForRecreation() ); 
            if numel(jobHiddenPVPairs) >= 2
                newJob.hSetProperty( jobHiddenPVPairs(1:2:end), jobHiddenPVPairs(2:2:end) );
            end
            %Concurrent 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

    methods ( Hidden )
        function pPreJobEvaluate( job, task )
            preMJSIndependentJob( job, task );
        end
        function pPostJobEvaluate( ~ )
            % Clean up MPI if it got initialized
            mpiCleanup();
        end
        function info = pGetDataStoreInfo( job )
            info = job.Support.getJobDataStoreInfo;
        end
    end
end
%#ok<*ATUNK> custom attributes

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 iSetParallelTag(job, data)
% Safely serialize data into the parallel tag for the job.
    serialData = distcompserialize(data);
    serialDataString = matlab.net.base64encode(serialData);
    job.hSetPropertyNoCheck('ParallelTag', serialDataString);
end

function data = iGetParallelTag(job)
% Decode and deserialize the parallel tag when it becomes available.
    parallelTag = job.hGetProperty('ParallelTag');
    % Wait until setup info is retrieved from the job manager.
    while isempty(parallelTag)
        pause(parallel.job.MJSConcurrentJob.PollDelaySeconds); % A shortish wait
        parallelTag = job.hGetProperty('ParallelTag');
    end
    serialData = matlab.net.base64decode(parallelTag);
    data = distcompdeserialize(serialData);
end
