% Static helper class for discovering job managers

%   Copyright 2020 The MathWorks, Inc.

classdef CachingJobManagerFinder

    methods (Static)
        function clearCache()
            % Available for tests to wipe away all cached clients
            iGetOrSetOrClearCachedClient();
        end
        
        function client = getOneJobManager(timeoutSeconds, hostname, name, testCachedClient, useCache)
            if nargin < 4
                % By default, test any cached client
                testCachedClient = true;
            end
            
            if nargin < 5
                % By default, use any cached clients
                useCache = true;
            end
            
            if isempty(hostname)
                % PCT_Java_Transition_TODO (g2210330): Implement discovery for SPF services
                error('Unsupported operation');
            end
            
            connectUrl = iCreateConnectUrl(hostname);
            client = parallel.internal.cluster.CachingJobManagerFinder.getOneJobManagerFromUrl(...
                timeoutSeconds, connectUrl, name, testCachedClient, useCache);
        end

        function client = getOneJobManagerFromUrl(timeoutSeconds, connectUrl, name, testCachedClient, useCache)
            dctSchedulerMessage(4, 'MJS connect string: %s', connectUrl);
            
            % Check cache for existing client
            if useCache && ~isempty(connectUrl) && ~isempty(name)
                key = iCreateCacheKey(connectUrl, name);
                client = iGetOrSetOrClearCachedClient(key);
                if ~isempty(client)
                    if testCachedClient && ~iTestClient(client)
                        % Cached client is invalid. Remove from cache and
                        % continue.
                        iGetOrSetOrClearCachedClient(key, []);
                    else
                        % Found and successfully tested a cached client. Return
                        % that.
                        return
                    end
                end
            end
            
            if isempty(name)
                finder = parallel.internal.cluster.JobManagerFinderImpl(connectUrl);
            else
                finder = parallel.internal.cluster.JobManagerFinderImpl(connectUrl, name);
            end
            clients = finder.find(timeoutSeconds * 1000);
            client = iGetOneMJSOrError(clients, connectUrl, name);
            
            % Add new client to cache
            name = client.getProperty('Name');
            key = iCreateCacheKey(connectUrl, name);
            iGetOrSetOrClearCachedClient(key, client);
        end
    end
    
end

% --------------------------------------------------------------------------
% iCreateConnectUrl - Verify that hostname has the required protocol and
% port parts of the form:
% protocol://host:port
% If not, add the default values to create the connect url.
% Port here is the baseport, so will be shifted by the exchange port offset
% to create the connect string
function connectUrl = iCreateConnectUrl(hostname)
    import parallel.internal.cluster.PortConfig;
    
    connectUrl = parallel.internal.cluster.createConnectUrl(hostname, ...
        PortConfig.getExchangeProtocol(), ...
        PortConfig.getDefaultBasePort(), ...
        PortConfig.getExchangePortOffset);
end

% --------------------------------------------------------------------------
% iCreateCacheKey - Create cache key from connect url and name
function key = iCreateCacheKey(connectUrl, name)
    key = [connectUrl '@' name];
end

% --------------------------------------------------------------------------
% iGetOrSetCachedClient - Get or add a cached client
function client = iGetOrSetOrClearCachedClient(key, client)
    mlock;
    persistent CLIENT_CACHE;
    if isempty(CLIENT_CACHE)
        CLIENT_CACHE = containers.Map();
    end
    
    if nargin < 1
        % Special no arg call to clear cache
        dctSchedulerMessage(4, 'Clearing job manager client cache');
        CLIENT_CACHE = containers.Map();
        return
    end
    
    if nargin < 2
        % Getting cached value
        if isKey(CLIENT_CACHE, key)
            dctSchedulerMessage(4, 'Found cached job manager client for key: %s', key);
            client = CLIENT_CACHE(key);
        else
            dctSchedulerMessage(4, 'Did not find cached job manager client for key: %s', key);
            client = [];
        end
    else
        if isempty(client)
            % Removing a cached value
            dctSchedulerMessage(4, 'Removing cached job manager client for key: %s', key);
            CLIENT_CACHE.remove(key);
            return
        end
        % Setting cached value
        if isKey(CLIENT_CACHE, key)
            dctSchedulerMessage(4, 'Replacing cached job manager client for key: %s', key);
            CLIENT_CACHE(key) = client;
        else
            dctSchedulerMessage(4, 'Adding new cached job manager client for key: %s', key);
            CLIENT_CACHE(key) = client;
        end
    end
end

% --------------------------------------------------------------------------
% iTestClient - Test given client to ensure remote end is still valid
function tf = iTestClient(client)
    try
        refreshFcn = @() assert(false);
        access = parallel.internal.cluster.MJSAccess(client, refreshFcn);
        access.ping();
        tf = true;
    catch E
        dctSchedulerMessage(1, 'Failed to ping cached job manager client', E);
        tf = false;
    end
end

% --------------------------------------------------------------------------
% iGetOneMJSOrError - if the resulting list of clients is exactly 1 element
% long, return that or else error.
function client = iGetOneMJSOrError(clients, host, name)
% Clients found using different but equivalent endpoints will
% appear multiple times. We therefore filter based on SessionID to
% count the number of actually different services.
[~, uniqueIndices] = unique(arrayfun(@(c) c.getProperty('SessionID'), clients, 'UniformOutput', false));
clients = clients(uniqueIndices);
switch numel(clients)
    case 0
        throwAsCaller(iGetNoCompatibleMJSException(host, name));
    case 1
        % ok
    otherwise
        descriptionCell = arrayfun(@iGetMJSDescription, clients, 'UniformOutput', false);
        descriptionCell = descriptionCell(~cellfun(@isempty, descriptionCell));
        descriptionStr = sprintf('%s\n', descriptionCell{:});
        ex = MException('parallel:cluster:MJSMultipleFound', ...
            'Multiple MJS clusters were found:\n%s', descriptionStr);
        throwAsCaller(ex);
end

client = clients(1);
end

% --------------------------------------------------------------------------
% iGetMJSDescription - Get description of found client for populating error message.
function description = iGetMJSDescription(client)
    description = sprintf('Name: %s, Host: %s', client.getProperty('Name'), client.getProperty('HostName'));
end

% --------------------------------------------------------------------------
% iGetNoCompatibleMJSException - No job managers were found, return an appropriate exception.
function ex = iGetNoCompatibleMJSException(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

    % PCT_Java_Transition_TODO (g2210330): Use discovery to suggest other job
    % managers found with the same name.
    ex = MException(mjsNotFoundMsg);
end
