%PoolClient Client object for parallel pool jobs

% Copyright 2012-2020 The MathWorks, Inc.
classdef PoolClient < parallel.internal.pool.AbstractClient
    properties ( Access = private )
        % Is this communicating job size 1 - i.e. no PoolWorkers.
        IsClientOnlySession = false
    end

    methods
        function obj = PoolClient
            obj.IsClientOnlySession = (numlabs < 2);
            obj.JobStartupTimeout   = Inf;
            obj.ParallelJob = getCurrentJob();
        end

        function spmdInitialized = start(obj, leadingTaskNum, sessionConstructorFcn)
        %START Start parallel pool job client.
            % If we are running just on the client none of the following is needed.
            if obj.IsClientOnlySession
                return
            end

            try
                [session, workerSessionFcn] = sessionConstructorFcn();
            catch err
                obj.pStopLabsAndDisconnect();
                rethrow(err);
            end

            % Share the hostname and port among all labs. The corresponding MPI call
            % resides in @matlabpoollab/start.m
            workerSessionFcn = labBroadcast(leadingTaskNum, workerSessionFcn); %#ok<NASGU>

            % Set up for SPMD execution - build the new communicator, and stash
            % that for later.
            try
                % Corresponding mpiCommManip is in the labs' "start" method
                newWorld = mpiCommManip( 'split', -1, labindex );
                worldVec = gcat( {newWorld} );
                spmdlang.commForWorld( 'set', worldVec(2:end), 2 );
                % Deadlock detection on for now while we don't have a good way of
                % interrupting non-dd communications
                mpiSettings( 'DeadlockDetection', 'on' );
            catch err
                obj.pStopLabsAndDisconnect();
                rethrow( err );
            end

            % Create the Java objects for this interactive session.
            try
                spmdInitialized = true; % always true for non-interactive pools.
                sessionInfo = parallel.internal.pool.SessionInfo();
                checkFcn = @() [];
                session.waitForConnections(spmdInitialized, sessionInfo, checkFcn);
                obj.Session = session;
                % Once created we should loop waiting for all labs to finish starting
                % up
                while ~obj.Session.waitForSessionToStart(0.1)
                end
            catch err
                obj.pStopLabsAndDisconnect();
                ex = MException(message('parallel:lang:pool:FailedToInitializeInteractiveSession'));
                ex = ex.addCause(err);
                throw(ex);
            end

            % For a parallel pool session tell the session to listen for Path and Clear
            % notification and send them onto the labs
            try
                session.startSendPathAndClearNotificationToLabs;
                mpiCommManip( 'select', 'self' );
            catch err
                obj.pStopLabsAndDisconnect();
                ex = MException(message('parallel:lang:pool:FailedToInitializeInteractiveSession'));
                ex = ex.addCause(err);
                throw(ex);
            end
        end

        function OK = stopLabsAndDisconnect(obj, ~)
        %stopLabsAndDisconnect Stop all the labs and perform client cleanup
        %   Send the stop signal to all the labs so they exit.  Clean up sockets
        %   and streams.
            OK = true;
            if ~obj.IsClientOnlySession
                OK = obj.pStopLabsAndDisconnect();
            end

            % Clean up memory associated with Composites
            spmdlang.ValueStore.clear();
            % Ensure we undo the spmdlang.commForWorld setting
            spmdlang.commForWorld('reset');
        end
    end

    methods ( Access = private )
        function OK = pStopLabsAndDisconnect(obj)
        %stopLabsAndDisconnect Stop all the labs and perform client cleanup
        %   Send the stop signal to all the labs so they exit.  Clean up sockets
        %   and streams.

            % Must ensure we are all back in the world communicator to clean up.
            mpiCommManip( 'select', 'world' );

            OK = true;
            if ~isempty(obj.Session)
                disp('Sending a stop signal to all the labs...');
                OK = obj.Session.destroyClientSession();
                %Finished with Session - forget about it
                obj.Session = [];
            end
        end
    end
end
