% MJS - Perform discovery for MJS

%    Copyright 2012-2020 The MathWorks, Inc.

classdef MJS < parallel.internal.discover.AbstractDiscoverer
    
    properties ( Constant = true )
        %DEFAULT_TOTAL_DISCOVERY_TIMEOUT_SECONDS
        % Default for how long we wait overall for discovery before aborting.
        DEFAULT_TOTAL_DISCOVERY_TIMEOUT_SECONDS = 120

        %QUERY_JM_PROPERTIES_TIMEOUT_SECONDS
        % Defines how long wait for a job manager query before aborting.
        QUERY_JM_PROPERTIES_TIMEOUT_SECONDS = 10
    end
    
    methods
        function obj = MJS()
            import parallel.internal.types.SchedulerType
            obj = obj@parallel.internal.discover.AbstractDiscoverer( SchedulerType.MJS );
        end
    end
    
    methods % overrides of Abstract methods in AbstractDiscoverer
        function discover( obj, terminateEarlyFcn )
            % discover - discover MJS clusters
            % Fires ClusterFound events whenever a new cluster
            % is found.  Terminates discovery if the supplied
            % terminateEarlyFcn returns true.
            validateattributes(terminateEarlyFcn, {'function_handle'}, {'scalar'});
            obj.doMulticastLookup( terminateEarlyFcn );
        end
    end
    
    methods ( Static, Hidden )
        function oldVal = currentMJSDiscoveryTimeout( newVal )
            persistent multicastTimeout;
            if isempty( multicastTimeout )
                multicastTimeout = parallel.internal.discover.MJS.DEFAULT_TOTAL_DISCOVERY_TIMEOUT_SECONDS;
            end
            oldVal = multicastTimeout;
            if nargin > 0
                validateattributes( newVal, {'numeric'}, {'integer'} );
                multicastTimeout = newVal;
            end
        end
        
        function [ success, evtData ] = queryJobManager( jmInfo, timeoutSeconds )
            dctSchedulerMessage( 2, '>> queryJobManager() timeout: %d', timeoutSeconds );
            evtData = {};
            success = false;
            try
                matlabRelease = char( jmInfo.getRelease() );
                isCompatible = parallel.internal.version.Version.Current.isCompatibleWithServer( ...
                    jmInfo.getVersionNum(), jmInfo.getMinimumCompatibleVersionNum() );
                
                propNames = { 'Name', 'Host', 'SecurityLevel' };
                % Need to use getLookupURLOrHost to support non-default
                % baseports. See g1332704 and g988845 and g1097409.
                host = char( parallel.internal.cluster.getLookupURLOrHost(jmInfo) );
                name = char( jmInfo.getName() );
                propVals = { name, host, jmInfo.getSecurityLevel() };
                dctSchedulerMessage( 5, '-- queryJobManager matlabRelease: %s host: %s name: %s isCompatible %d', ...
                    matlabRelease, host, name, isCompatible );
                if isCompatible
                    jmprx = parallel.internal.cluster.JobManagerFinder.getOneJobManagerProxyInterruptibly( host, name, timeoutSeconds );
                    numWorkers = jmprx.getNumIdleWorkers() + jmprx.getNumBusyWorkers();
                    propNames{end+1} = 'NumWorkers';
                    propVals{end+1} = numWorkers;
                    dctSchedulerMessage( 5, '-- queryJobManager numWorkers %d', numWorkers );
                end
                
                % Filter out the clusters that require client certificates, as these are
                % not discoverable, since there is no way to transfer the certificate to
                % the client via the discovery service.
                % TODO (g1661610): allow discovery, but not profile creation for these clusters
                requireClientCertificate = jmInfo.getRequireClientCertificate();
                if requireClientCertificate
                    dctSchedulerMessage( 5, '-- queryJobManager cluster filtered because requireClientCertificate was true' );
                    return
                end
                evtData = parallel.internal.discover.ClusterFoundEventData.createForMJS( ...
                    propNames, propVals, {'Host', 'Name'}, matlabRelease, isCompatible );
                success = true;
            catch err
                % Ignore clusters that error at this stage, as a user won't be able to use them.
                dctSchedulerMessage( 2,  'A cluster was found but threw an exception when we tried to get information from it, the exception was', err );
            end
            dctSchedulerMessage( 2, '<< queryJobManager()');
        end
    end
    
    methods ( Access = private )
        
        function doMulticastLookup( obj, terminateEarlyFcn )
            dctSchedulerMessage( 5, '>> doMulticastLookup()' );
            foundJMs = containers.Map();
            
            % Use multicast to find all the job managers on the network. More than
            % one multicast lookup will be required.
            t = tic;
            while ( toc( t ) < parallel.internal.discover.MJS.currentMJSDiscoveryTimeout ...
                    && ~terminateEarlyFcn() )
                try
                    % Providing an empty host and name will do a multicast lookup on the local subnet.
                    jmInfo = ...
                        parallel.internal.cluster.JobManagerFinder.discoverMJS( ...
                        parallel.internal.discover.MJS.DEFAULT_TOTAL_DISCOVERY_TIMEOUT_SECONDS );
                catch err
                    % If there's a multicast lookup failure here we don't want it to bubble up so
                    % just log it and carry on.
                    if strcmp( err.identifier, 'parallel:cluster:jobmanager:MulticastServiceNotFound' )
                        dctSchedulerMessage( 2, 'Multicast lookup failed, the exception was', err );
                        continue;
                    else
                        rethrow( err );
                    end
                end
                if terminateEarlyFcn()
                    break;
                end
                % If we've found some job managers then notify our listeners.
                if ~isempty( jmInfo )
                    dctSchedulerMessage( 5, 'Found %d job managers', numel( jmInfo ) );
                    for ii = 1:numel( jmInfo )
                        thisJM = jmInfo(ii);
                        thisID = char( thisJM.getID() );
                        dctSchedulerMessage( 5, 'Checking if job manager already found: %s', thisID ); 
                        if ~foundJMs.isKey( thisID )
                            foundJMs(thisID) = thisJM;
                            [ success, jmData ] = parallel.internal.discover.MJS.queryJobManager( thisJM, ...
                                parallel.internal.discover.MJS.QUERY_JM_PROPERTIES_TIMEOUT_SECONDS );
                            if success
                                obj.notifyFound( jmData );
                            end
                        end
                        % Stop if we've taken too long or the user cancels
                        if toc( t ) >= parallel.internal.discover.MJS.currentMJSDiscoveryTimeout ...
                                || terminateEarlyFcn()
                            dctSchedulerMessage( 2, 'currentMJSDiscoveryTimeout expired: %d out of %d Job Managers queried', ii, numel( jmInfo ) );
                            break;
                        end
                    end
                end
                % Don't spin hard
                pause(1);
            end
            dctSchedulerMessage( 5, '<< doMulticastLookup()' );
        end
                
        function notifyFound( obj, evtData )
            try
                dctSchedulerMessage( 5, '>> notifyFound()' );
                obj.notify( 'ClusterFound', evtData );
                dctSchedulerMessage( 5, '<< notifyFound()' );
            catch err
                % Ignore clusters that error at this stage, as a user won't be able to use them.
                dctSchedulerMessage( 2, 'Failed to notify listeners of ClusterFound event. Error', err );
            end
        end
    end
end
