
%

%   Copyright 2012-2020 The MathWorks, Inc.

classdef JobManagerFinder
    
    properties ( Constant, GetAccess = private )
        EnforceInitJavaRMI = parallel.internal.cluster.JobManagerFinder.initJavaRMI();
    end
    
    methods ( Static )
        
        function info = discoverMJS( timeoutSeconds, host )
            import com.mathworks.toolbox.distcomp.mjs.service.ServiceAccessor
            if nargin < 3
                % Passing in an empty string for the host in order to do
                % multicast lookup.
                host = '';
            end
            dctSchedulerMessage( 5, '>> JobManagerFinder.discoverMJS()' );
            try
                info = ServiceAccessor.discoverMJS( host, timeoutSeconds );
            catch E
                iHandleExceptionFromInterruptibleJavaMethodCall( E, 'ServiceAccessor.discoverMJS()', timeoutSeconds );
                info = [];
            end
            dctSchedulerMessage( 5, '<< JobManagerFinder.discoverMJS()' );
        end
        
        function jmprx = getOneJobManagerProxy( host, name, testProxy )
            if nargin < 3
                testProxy = true;
            end
            jmprx =  iGetJobManagerProxy( host, name, testProxy );
        end
        
        function jmprx = getOneJobManagerProxyInterruptibly( host, name, timeoutSeconds )
            jmprx = iGetJobManagerProxyInterruptibly( host, name, timeoutSeconds );
        end
        
        % Initialize Java RMI used by job manager proxies.
        % NB: This is a static method so that it can be called from
        % initworker even when we don't need to "find" a job manager
        % (because the job manager proxy is passed in).
        function success = initJavaRMI()
            % Set JVM properties for RMI and DNS
            RMI_connectionTimeoutMillis = '10000';
            RMI_readTimeoutMillis = '300000';
            DNS_lookupIntervalSecs = '300';
            try
                java.lang.System.setProperty('sun.rmi.transport.connectionTimeout', RMI_connectionTimeoutMillis);
            catch err
                % It's OK if we can't set the custom value for the timeout - we'll just use
                % the default value of 15000ms.
                logFcn('Failed to set RMI Connection Timeout.  Reason: %s', err.getReport());
            end
            try
                java.lang.System.setProperty('sun.rmi.transport.tcp.readTimeout', RMI_readTimeoutMillis);
            catch err
                % It's OK if we can't set the custom value for the timeout - we'll just use
                % the default value of 2 hours
                logFcn('Failed to set RMI Read Timeout.  Reason: %s', err.getReport());
            end
            try
                java.lang.System.setProperty('sun.net.inetaddr.ttl', DNS_lookupIntervalSecs);
            catch err
                % It's OK if we can't set the custom value for the ttl - we'll just use
                % the default value of -1, which means "cache forever"
                logFcn('Failed to set DNS Lookup Interval Timeout.  Reason: %s', err.getReport());
            end

            % Make sure we never try and load code from anywhere but local classpath
            java.lang.System.setProperty('java.rmi.server.codebase', '');
            java.lang.System.setProperty('java.rmi.server.useCodebaseOnly', 'true');

            % Perform client-only initialization
            if ~feature('isdmlworker')
                iInitClientJavaRMI();
            end

            success = true;
        end
    end
    
end

% --------------------------------------------------------------------------
% iGetJobManagerProxy - returns one proxy or errors
function jmprx = iGetJobManagerProxy( host, name, testProxy )
[proxyManagers, badProxyManagers] = iGetJobManagerProxies( host, name, testProxy );

% If we don't get any good proxies, we retry once, to account for 'stale'
% proxies, see g1955791
if numel(proxyManagers) == 0 && numel(badProxyManagers) > 0
    [proxyManagers, badProxyManagers] = iGetJobManagerProxies( host, name, testProxy );
end

jmprx = iGetOneProxyOrError( proxyManagers, badProxyManagers, host, name );

% Make sure cached proxies are disposed of at MCR shutdown
jmprx.addShutdownHookForCurrentMCR();
end

% --------------------------------------------------------------------------
% iGetJobManagerProxies
function [proxyManagers, badProxyManagers] = iGetJobManagerProxies( host, name, testProxy )
    import com.mathworks.toolbox.distcomp.mjs.service.ServiceAccessor
    allProxyManagers = ServiceAccessor.getJobManagers( host, name, testProxy );
    okManagersIdx = arrayfun( @(x) x.isOk(), allProxyManagers );
    proxyManagers = allProxyManagers( okManagersIdx );
    badProxyManagers = allProxyManagers( ~okManagersIdx );
end

% --------------------------------------------------------------------------
% iGetJobManagerProxyInterruptibly - returns one proxy or errors
function jmprx = iGetJobManagerProxyInterruptibly( host, name, timeoutSeconds )
[proxyManagers, badProxyManagers] = iGetJobManagerProxiesInterruptibly( host, name, timeoutSeconds );

% If we don't get any good proxies, we retry once, to account for 'stale'
% proxies, see g1955791
if numel(proxyManagers) == 0 && numel(badProxyManagers) > 0
    [proxyManagers, badProxyManagers] = iGetJobManagerProxiesInterruptibly( host, name, timeoutSeconds );
end

jmprx = iGetOneProxyOrError( proxyManagers, badProxyManagers, host, name );

% Make sure cached proxies are disposed of at MCR shutdown
jmprx.addShutdownHookForCurrentMCR();
end

% --------------------------------------------------------------------------
% iGetJobManagerProxiesInterruptibly
function [proxyManagers, badProxyManagers] = iGetJobManagerProxiesInterruptibly( host, name, timeoutSeconds )
import com.mathworks.toolbox.distcomp.mjs.service.ServiceAccessor
try
    allProxyManagers = ServiceAccessor.getJobManagersInterruptibly( host, name, timeoutSeconds );
catch E
    iHandleExceptionFromInterruptibleJavaMethodCall( E, 'ServiceAccessor.getJobManagersInterruptibly()', timeoutSeconds );
    allProxyManagers = [];
end
okManagersIdx = arrayfun( @(x) x.isOk(), allProxyManagers );
proxyManagers = allProxyManagers( okManagersIdx );
badProxyManagers = allProxyManagers( ~okManagersIdx );
end

% --------------------------------------------------------------------------
% iGetOneProxyOrError - if the resulting list of proxies is exactly 1 element
% long, return that or else error.
function proxy = iGetOneProxyOrError( proxyManagers, badProxyManagers, host, name )
ex = MException.empty();
switch numel( proxyManagers )
    case 0
        ex = iGetNoCompatibleProxiesException( badProxyManagers, host, name );
    case 1
        % ok
    otherwise
        descriptionFcn  = @(jmp)sprintf('Name: %s, Host: %s', char(jmp.getName), char(jmp.getHostName) );
        descriptionCell = arrayfun( descriptionFcn, proxyManagers, 'UniformOutput', false );
        descriptionStr  = sprintf('%s\n', descriptionCell{:});
        ex = MException( 'parallel:cluster:MJSMultipleFound', ...
            'Multiple MJS clusters were found:\n%s', descriptionStr );
end
if ~isempty( ex )
    throwAsCaller( ex );
end
proxy = proxyManagers(1);
end

% --------------------------------------------------------------------------
% iGetNoCompatibleProxiesException - No proxies were found, return an appropriate exception.
function ex = iGetNoCompatibleProxiesException( badProxyManagers, host, name )

matlabRelease = pctGetMATLABRelease();
if isempty( host ) && isempty( name )
    mjsNotFoundMsg = message( 'parallel:cluster:MJSNotFound', matlabRelease );
elseif ~isempty( host ) && isempty( name )
    mjsNotFoundMsg = message( 'parallel:cluster:MJSNotFoundOnHost', host, matlabRelease );
elseif ~isempty( name ) && isempty ( host )
    mjsNotFoundMsg = message( 'parallel:cluster:MJSNotFoundWithName', name, matlabRelease );
else
    mjsNotFoundMsg = message( 'parallel:cluster:MJSNotFoundFullySpecified', name, host, matlabRelease );
end

% Discover if other incompatible jobmanagers are available.
jmInfo = parallel.internal.cluster.JobManagerFinder.discoverMJS( ...
    parallel.internal.discover.MJS.DEFAULT_TOTAL_DISCOVERY_TIMEOUT_SECONDS, host );

% If a name was specified, we only mention job managers with that name.
if ~isempty( name )
    jmInfo = iFilterByName( name, jmInfo );
end

% If we've found some "bad" proxies, we remove them from the list of discovered
% proxies so we don't mention them twice.
if ~isempty( badProxyManagers )
    jmInfo = iFilterAllByHostAndName( badProxyManagers, jmInfo );
end

numDiscovered = numel( jmInfo );
numFailedTests = numel( badProxyManagers );
numberOfMJS = numDiscovered + numFailedTests;
if numberOfMJS > 0
    % Here we build a list of extra information about incompatible clusters
    % on this host, with the requested name.
    availableJMString = iConvertInfoToString( jmInfo, @iGetMJSInfoFromDiscovery );
    errorJMString = iConvertInfoToString( badProxyManagers, @iGetMJSInfoFromErroredProxy );
    if numFailedTests == 0
        ex = MException( message( 'parallel:cluster:AvailableMJSClusters', ...
            mjsNotFoundMsg.getString(), availableJMString ) );
    elseif numDiscovered == 0
        ex = MException( message( 'parallel:cluster:AvailableMJSClustersOnlyFailures', ...
            mjsNotFoundMsg.getString(), errorJMString ) );
    else
        ex = MException( message( 'parallel:cluster:AvailableMJSClustersWithFailures', ...
            mjsNotFoundMsg.getString(), availableJMString, errorJMString ) );
    end
else
    ex = MException( mjsNotFoundMsg );
end
end

function infoString = iConvertInfoToString( info, conversionFcn )
infoCell = cell( numel( info ), 1 );
for ii = 1:numel( info )
    infoCell{ii} = conversionFcn( info(ii) );
end
infoString = strjoin( infoCell, '\n' );
end

function mjsInfo = iGetMJSInfoFromDiscovery( discoveryInfo )
matlabRelease = char( discoveryInfo.getRelease() );
name = char( discoveryInfo.getName() );
host = char( discoveryInfo.getHostName() );
mjsInfo = message( 'parallel:cluster:MJSInformation', ...
    name, host, matlabRelease ).getString();
end

function mjsInfo = iGetMJSInfoFromErroredProxy( proxy )
name = char( proxy.getName() );
host = char( proxy.getHostName() );
errorMsg = char( proxy.getErrorMessage() );
mjsInfo = message( 'parallel:cluster:MJSInformationWithError', ...
    name, host, errorMsg ).getString();
end

function filteredJMInfo = iFilterByName( name, jmInfo )
index = arrayfun( @(x) strcmp( char( x.getName() ), name ), jmInfo );
filteredJMInfo = jmInfo( index );
end

function filteredJMInfo = iFilterByHostAndName( name, host, jmInfo )
index = arrayfun( @(x) strcmp( char( x.getName() ), name ) && ...
    strcmp( char( x.getHostName() ), host ), jmInfo );
filteredJMInfo = jmInfo( ~index );
end

function filteredJMInfo = iFilterAllByHostAndName( badProxies, jmInfo )
    filteredJMInfo = jmInfo;
    for ii = 1:numel( badProxies )
        filteredJMInfo = iFilterByHostAndName( ...
            char( badProxies(ii).getName() ), ...
            char( badProxies(ii).getHostName() ), ...
            filteredJMInfo );
    end
end

function iHandleExceptionFromInterruptibleJavaMethodCall( E, methodName, timeoutSeconds )
    err = distcomp.handleJavaException( [], E );
    % If the interruptible method call timed out or was cancelled, log it and
    % then return
    if strcmp( err.identifier, 'parallel:internal:JavaMethodTimeout' )
        dctSchedulerMessage( 4, '%s timed out after %d seconds.', methodName, timeoutSeconds );
        return
    elseif strcmp( err.identifier, 'parallel:internal:JavaMethodCancellation ')
        dctSchedulerMessage( 4, '%s was cancelled.', methodName );
        return
    end
    throw(err);
end

function iInitClientJavaRMI()
    % Set the values of the export port and hostname.
    config = pctconfig;

    % The conditions are necessary because iInitJavaRMI may be called multiple times.
    if ~com.mathworks.toolbox.distcomp.mjs.service.ExportConfigInfo.checkIfPortSet()
        portrange = config.portrange;
        % portrange will be either [minPort, maxPort] or 0
        if isscalar(portrange)
            com.mathworks.toolbox.distcomp.mjs.service.ExportConfigInfo.useEphemeralPort();
        else
            com.mathworks.toolbox.distcomp.mjs.service.ExportConfigInfo.setPortRange(...
                portrange(1), portrange(2));
        end
    end
    export_hostname = config.hostname;
    java.lang.System.setProperty('java.rmi.server.hostname', export_hostname);

    % Turn down the JINI logging levels to SEVERE, unless they have already
    % been set to some other level. This would ensure that the second time this
    % function is called we don't change the level.
    try
        logger = java.util.logging.Logger.getLogger('net.jini');
        if isempty(logger.getLevel)
            logger.setLevel(java.util.logging.Level.SEVERE);
        end
    catch err
        % It's OK to proceed if we couldn't change the log level.
        dctSchedulerMessage(6, 'Failed to change JINI log level to severe.  Reason: %s', ...
            err.getReport());
    end

    % Tell pctconfig that we have used values from it
    pctconfig('initclienthasrun', true);
end