%CppBackedSession Implementation of parallel.internal.pool.ISession
%

% Copyright 2020-2021 The MathWorks, Inc.

classdef CppBackedSession < parallel.internal.pool.ISession

    properties (Constant)
        ConnectScheme = 'tcp'
    end
    
    properties (SetAccess = immutable)
        UUID (1,1) string
        SpfSession
        URL
        
        % PCT_Java_Transition_TODO g2373577 this should depend
        % upon the connection topology
        SendInitDataSeparately (1,1) logical = true;
    end

    properties (SetAccess = private)
        SessionInfo
        CompositeKeysToClearMap = []
        PathAssistant
        FileDependenciesAssistant = []
        % CppFevalQueue hands us a cleanup object to remove itself from its cache.
        QueueCleanup = []
        IsShutdownComplete = false;
    end

    methods (Access = private)
        function obj = CppBackedSession(url, uuid, workerIdx, numWorkers)
            obj.URL = url;
            obj.UUID = uuid;
            if nargin == 2
                obj.SpfSession = parallel.internal.pool.SpfClientSession(url, uuid);
            else
                obj.URL = iProcessClientHostOverride(obj.URL);
                obj.SpfSession = parallel.internal.pool.SpfWorkerSession(obj.URL, uuid, workerIdx, numWorkers);
            end
        end
    end

    methods (Access = {?parallel.internal.pool.AbstractClusterPool, ...
                       ?parallel.internal.pool.ISession})
        function uuid = getSessionUUID(obj)
            uuid = obj.UUID;
        end

        function tf = canInitializeSpmd(obj)
            tf = obj.SpfSession.canInitializeSpmd();
        end

        function tf = initializeSpmd(obj)
            tf = obj.SpfSession.initializeSpmd();
        end

        function fq = getFevalQueue(obj, pool)
            fq = parallel.FevalQueue.getQueueForSession(pool, ...
               @(~) parallel.internal.queue.CppFevalQueue(obj.SpfSession));
        end
    end


    methods (Access = {?parallel.pool.TicBytesResult, ...
                       ?parallel.internal.pool.ISession})
        function info = getCurrentBytesTransferredToInstances(obj)
            assert(false, 'Unimplemented.');
            info = [];
        end
    end

    methods (Access = {?parallel.internal.pool.AbstractClient, ...
                       ?parallel.internal.pool.ISession})
        function out = destroyClientSession(obj)
            % This will wait for shutdown to complete
            obj.shutdown();
            out = true;
        end

        function startSendPathAndClearNotificationToLabs(obj)
            obj.PathAssistant = parallel.internal.pool.PathAssistant(obj.SpfSession);
        end

        function waitForConnections(obj, spmdEnabled, sessionInfo, checkFcn)

            % Pool clients end up here with a dummy sessionInfo, we need a real one.
            if strcmp(sessionInfo.State, 'CLOSED')
                % We don't have cluster or profile info here, but it is not needed.
                sessionInfo = parallel.internal.pool.ISessionNotifier.buildAboutToStartInfo('','');
                sessionInfo.starting('', '', intmax('int64'), false, false);
            end
            obj.SessionInfo = sessionInfo;
            obj.SpfSession.setSessionInfo(obj.SessionInfo);
            parallel.internal.pool.bootstrapClientServices(obj.SpfSession);
            
            % PCT_Java_Transition_TODO (g2421835) Remove this workaround which adds a simple
            % deadline for pool formation to avoid stalls.
            deadline = datetime('now') + seconds(300);
            
            while true
                assert(datetime('now') < deadline, 'Deadline for pool creation expired.');
                if obj.SpfSession.waitForPoolFormation(1000, spmdEnabled)
                    break
                end
                % checkFcn will abort this function with an error if necessary.
                checkFcn();
            end
        end

        function tf = waitForSessionToStart(~, ~)
        % We did everything in waitForConnections.
            tf = true;
        end
    end

    % Worker-only methods
    methods (Access = {?parallel.internal.pool.AbstractWorker, ...
                       ?parallel.internal.pool.ISession})
        function setDependencyDir(~, dependencyDir)
        % PCT_Java_Transition_TODO (g2381716): remove this once Java code is removed.
            dctSchedulerMessage(1, 'Ignoring request to setDependencyDir(%s)', dependencyDir)
        end
    end

    
    methods
        
        function evalOnClient(obj, cmd)
            parallel.internal.pool.evalOnClient(obj.SpfSession, cmd);
        end

        function fda = getFileDependenciesAssistant(obj)
            if isempty(obj.FileDependenciesAssistant)
                obj.FileDependenciesAssistant = ...
                    parallel.internal.pool.FileDependenciesAssistant(obj.SpfSession);
            end
            fda = obj.FileDependenciesAssistant;
        end

        function n = getPoolSize(obj)
            n = double(obj.SessionInfo.SessionSize);
        end

        function tf = isSessionRunning(obj)
            tf = obj.SpfSession.isSessionRunning();
        end

        function setRestartOnClusterChange(obj, tf)
            obj.SessionInfo.RestartOnClusterChange = tf;
        end

        function setShutdownIdleTimeout(obj, timeInSeconds)
            obj.SpfSession.setIdleTimeout(timeInSeconds);
        end

        function info = getClientSessionInfo(obj)
            info = parallel.internal.pool.CppBackedSessionNotifier.sessionInfoFromMf0(...
                obj.SessionInfo);
        end
        
        function rehashed = rehashIfNecessary(obj)
           rehashed = obj.SpfSession.getAndSetRequiresRehash(false);
           if rehashed
               rehash();
           end
        end
        
        function itemsToClear = drainItemsToClear(obj)
            itemsToClear = obj.SpfSession.drainItemsToClear();
            itemsToClear = [itemsToClear{:}]; % Unpack cell array into array of PathInfo
        end

        function releaseCurrentParforController(obj)
            assert(false, 'Unimplemented.');
        end

        function controller = createParforController(obj, maxBacklog, maxNumWorkers)
            controller = parallel.internal.pool.ParforController(...
                    obj.SpfSession, maxBacklog, obj.SendInitDataSeparately, maxNumWorkers);
        end

        function controller = createSpmdController(obj, labIndices)
            controller = parallel.internal.pool.RemoteSpmdControllerImpl.create(...
                obj.SpfSession, labIndices, obj.getCompositeKeysToClearMap());
        end
        function map = getCompositeKeysToClearMap(obj)
            if isempty(obj.CompositeKeysToClearMap)
                obj.CompositeKeysToClearMap = parallel.internal.pool.ProcessKeyMap();
            end
            map = obj.CompositeKeysToClearMap;
        end

        function assistant = getCompositeAssistant(obj, labIndices)
            assistant = parallel.internal.pool.RemoteCompositeAssistantImpl.createForSession(...
                obj.SpfSession, labIndices, obj.getCompositeKeysToClearMap());
        end

        function factory = createRemoteQueueFactory(obj, queueUuid)
        % When building the remote queue factory, we embed in the function
        % handle that we provide the necessary internal information needed to
        % build the RemoteSpfQueue.
            sessionUuid = obj.UUID;
            receiverProcessUuid = obj.SpfSession.ProcessUuid;
            buildFcn = @() parallel.internal.pool.CppBackedSession.buildRemoteSpfQueue(...
                sessionUuid, queueUuid, receiverProcessUuid);
            factory = parallel.internal.dataqueue.RemoteSpfQueueFactory(obj, buildFcn);
        end
        
        function shutdown(obj)
            if ~obj.IsShutdownComplete
                % Only call if we haven't attempted before.
                obj.SpfSession.shutdown();
                obj.IsShutdownComplete = true;
            end
        end
    end

    methods (Static, Access = private)
        function s = findSessionForUuid(sessionUuid)
        % Find instance of this class on client or worker if available. This
        % is used by DataQueue which needs to send a reference to a specific
        % pool session.
            s = [];
            sessionObject = parallel.internal.pool.workerSession();
            if isempty(sessionObject)
                pool = parallel.internal.pool.PoolArrayManager.getCurrentWithCleanup();
                if ~isempty(pool) && ~isa(pool, "parallel.ThreadPool")
                    sessionObject = pool.hGetClient();
                end
            end
            if isempty(sessionObject)
                return
            end
            possibleSession = sessionObject.Session;
            if possibleSession.isSessionRunning() && isequal(possibleSession.UUID, sessionUuid)
                s = possibleSession;
            end
        end
    end

    methods (Static)
        function rq = buildRemoteSpfQueue(sessionUuid, queueUuid, receiverProcessUuid)
            s = parallel.internal.pool.CppBackedSession.findSessionForUuid(sessionUuid);
            if isempty(s)
                error(message('MATLAB:parallel:dataqueue:InvalidProcess'));
            end
            rq = parallel.internal.dataqueue.RemoteSpfQueue(...
                queueUuid, receiverProcessUuid, s.SpfSession);
        end
        
        function [obj, taskArgs] = buildInteractiveClient(~, sessionInfo)
            [bindUrl, connectUrl] = iChooseURL('hostname');
            if isempty(sessionInfo)
                % Get here during profile validation
                uuid = parallel.internal.pool.SpfClientSession.allocateSessionId();
            else
                uuid = sessionInfo.SessionId;
            end
            obj = parallel.internal.pool.CppBackedSession(bindUrl, uuid);
            taskArgs = {@(workerIdx, numWorkers, broadcastFcn) ...
                        parallel.internal.pool.CppBackedSession.buildWorker( ...
                               connectUrl, workerIdx, numWorkers, broadcastFcn, uuid), ...
                       @() iYieldLoop(uuid)};
        end
        function obj = buildWorker(url, workerIdx, numWorkers, ~, sessionUuid)
            obj = parallel.internal.pool.CppBackedSession(url, sessionUuid, workerIdx, numWorkers);
            t = tic();
            logGranularity = 1; % Don't log more frequently than this
            maxLogGranularity = 16; % Max allowed log interval
            while true
                if obj.SpfSession.waitForClientFormation(100)
                    break
                end
                if toc(t) > logGranularity
                    dctSchedulerMessage(6, 'buildWorker waiting for client connection');
                    logGranularity = min(maxLogGranularity, 2*logGranularity);
                    t = tic();
                end
            end
            % Must not bootstrap worker services until we have a client
            % connection, otherwise the client might begin to use the
            % worker before the worker can send results back.
            parallel.internal.pool.bootstrapWorkerServices(obj.SpfSession);
            % Set up the persistent state for workerYield:
            parallel.internal.pool.CppBackedSession.workerYield(obj.SpfSession);
            % Tell parallel.internal.pool.poolWorkerFcn to call this function
            % for synchronous pool evaluation. This environment variable is
            % also read by dctEvaluateTask to discover that the pool execution
            % is synchronous.
            setenv('MDCS_SYNC_TASK_FCN', ...
                   'parallel.internal.pool.CppBackedSession.workerYield');
        end
        function [obj, workerBuild] = buildBatchClient()
            uuid = parallel.internal.pool.SpfClientSession.allocateSessionId();
            [bindUrl, connectUrl] = iChooseURL('internalhostname');
            obj = parallel.internal.pool.CppBackedSession(bindUrl, uuid);
            workerBuild = @(workerIdx, numWorkers, broadcastFcn) ...
                parallel.internal.pool.CppBackedSession.buildWorker( ...
                    connectUrl, workerIdx, numWorkers, broadcastFcn, uuid);
        end
        function workerYield(sessionObj)
            persistent SESSION_OBJ
            if nargin == 1
                dctSchedulerMessage(2, 'Setting up for pool session %s', sessionObj.UUID);
                SESSION_OBJ = sessionObj;
            else
                iYieldLoop(SESSION_OBJ);
            end
        end
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Try and use OS commands to deduce ports open for LISTEN. This is not
% sufficient, since this doesn't pick up CLOSE_WAIT ports - but they're a bit
% harder to work out.
function p = iPortsInUse()
    if isunix
        [s,w] = system('lsof -P -iTCP -sTCP:LISTEN | sed -n ''s/.*:\([0-9][0-9]*\) (LISTEN)/\1/p''');
        assert(s == 0, 'Failed to list ports in use.');
        p = rmmissing(str2double(strsplit(w)));
    else
        [s,w] = system('netstat -a');
        assert(s == 0, 'Failed to call netstat: %s', w);
        lines = strsplit(w, newline);
        % Pick only 'LISTENING' lines:
        lines = lines(endsWith(lines, 'LISTENING'));
        % Ignore IPv6 lines:
        lines = lines(~contains(lines, '[:'));
        % Pick out the port from a line like
        % " TCP  0.0.0.0:1234    hostname:0   LISTENING"
        p = rmmissing(str2double(extractAfter(extractBefore(extractAfter(...
            lines, digitBoundary('start')), ' '), ':')));
    end
    dctSchedulerMessage(4, 'Ports in use: %s', num2str(p));
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% PCT_Java_Transition_TODO (g2421835): this logic is in completely the wrong
% place. We need a connection manager on the "server socket" side that accepts
% the port range and works out for itself which port to use, and then let us
% know. Instead, for now, to unblock things, we're trying to work out which
% ports *should* be available.
function [bindUrl, connectUrl] = iChooseURL(whichHostname)
    persistent USED_PORTS RANDSTREAM
    
    if isempty(RANDSTREAM)
        RANDSTREAM = RandStream('threefry');
    end
    
    [c, undoc] = pctconfig();
    if strcmp(whichHostname, 'internalhostname') && ~isempty(undoc.internalhostname)
        % If we're asked to use the internal hostname, and one is set - apply that here.
        connectHost = undoc.internalhostname;
        dctSchedulerMessage(2, 'Using internalhostname ''%s'' for connect URL.', ...
                            connectHost);
    else
        connectHost = c.hostname;
    end
    
    % It seems insufficient to avoid only ports in LISTEN state, since some
    % appear to hang around connected. This isn't the right way to go about
    % this (clearly), but hopefully it will do for now.
    portsInUse = union(iPortsInUse(), USED_PORTS);
    portsToTry = c.portrange(1):c.portrange(end);
    if isequal(portsToTry, 0)
        % Ephemeral ports. From
        % https://en.wikipedia.org/wiki/Ephemeral_port#Range, it looks like
        % the following range ought to work everywhere.
        portsToTry = 49152:60000;
    end

    avail = setdiff(portsToTry, portsInUse);
    assert(~isempty(avail), 'No ports available.');
    % Extra futzing: pick a random available port.
    port = avail(randi(RANDSTREAM, [1, numel(avail)]));
    USED_PORTS(end+1) = port;
    connectUrl = parallel.internal.cluster.createConnectUrl(connectHost, ...
        parallel.internal.pool.CppBackedSession.ConnectScheme, port, 0);
    bindUrl = parallel.internal.cluster.createConnectUrl('*', ...
        parallel.internal.pool.CppBackedSession.ConnectScheme, port, 0);
    dctSchedulerMessage(4, 'Selected bind URL: %s; connect URL: %s', bindUrl, connectUrl);
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function iYieldLoop(workerSession)
    uuid = workerSession.UUID;
    dctSchedulerMessage(2, 'Worker is ready to execute commands for pool session %s', uuid);
    while workerSession.isSessionRunning()
        % Use simply 'pause' to process all IQM events in chunks.
        pause(0.1);
    end
    dctSchedulerMessage(2, 'Pool session %s no longer running.', uuid);
    
    % Begin worker shutdown
    dctFinishInteractiveSession();
    parallel.internal.pool.stopInteractiveWorker();
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function url = iProcessClientHostOverride(url)
% Special case code for the client address - if there happens to be an
% environment variable MDCE_OVERRIDE_CLIENT_HOST, use that in
% preference. This allows the local scheduler to defend against a
% variety of problems, including the client machine having either a
% broken name, or one which clashes with some other machine on the
% network.
    overrideClientHostname = getenv("MDCE_OVERRIDE_CLIENT_HOST");
    if ~isempty(overrideClientHostname)
        dctSchedulerMessage(2, 'Applying MDCE_OVERRIDE_CLIENT_HOST: %s', overrideClientHostname);
        url = parallel.internal.cluster.replaceHostname(url, ...
            parallel.internal.pool.CppBackedSession.ConnectScheme, overrideClientHostname);
    end
end

%#ok<*INUSD> unimplemented methods.
%#ok<*MANU> unimplemented methods.
