%InteractivePmodeClient - client for pmode

% Copyright 2020 The MathWorks, Inc.
classdef InteractivePmodeClient < parallel.internal.pool.AbstractInteractiveClient
    
    properties (SetAccess = immutable, GetAccess = private)
        ShowUI (1,1) logical
    end
    properties (Access = private)
        IsGUIOpen (1,1) logical = false
    end
    
    methods
        function obj = InteractivePmodeClient(guiMode) 
            obj@parallel.internal.pool.AbstractInteractiveClient(...
                'pmode', parallel.internal.cluster.ParpoolJobConstants.PmodeJobTag);
            obj.ShowUI = nargin >= 1 && strcmp(guiMode, 'opengui');
        end
        
        function cleanupPmode( obj, sched )
        %Remove all interactive jobs owned by the current user.

            % Give a current session a chance to shutdown.
            if obj.isPossiblyRunning()
                obj.stopLabsAndDisconnect();
            end
            jobs = obj.findJobsOnCluster( sched );

            if isempty(jobs)
                fprintf( '%s\n\n', getString( message( ...
                    'parallel:convenience:NoExtantInteractiveJobs', ...
                    obj.Type ) ) );
            else
                [~, undoc] = pctconfig();
                if ~undoc.preservejobs
                    fprintf( '%s\n\n', getString( message( ...
                        'parallel:convenience:DestroyingInteractiveJobs', ...
                        length(jobs), obj.Type ) ) );
                    % When running "cleanup()", do not use the timeout.
                    timeoutForCleanup = 0;
                    obj.deleteJobs(sched, jobs, timeoutForCleanup);
                else
                    % Respect the instruction to preserve the jobs. But since
                    % this is 'close force', try at least to cancel them
                    cancel( jobs );
                end
            end
        end
        
        function client2lab(obj, clientvarname, labidxes, labvarname)
        %client2lab Client code to transfer a serializable variable to labs.
        %   Caller should have made sure that:
        %   - labidxes is a vector of integers
        %   - isvarname(clientvarname) and isvarname(labvarname) return true
        %   - pmode is currently running

        % Verify that the variable exists in the base workspace.
            if ~evalin('base', sprintf('exist(''%s'', ''var'')', clientvarname))
                error( message( 'parallel:convenience:ClientToLabNoSuchVariable', clientvarname ) );
            end

            % Only send to one of the labs.  It will then use MPI to send to the others.
            firstLab = min(labidxes);
            try
                labs = obj.Session.pmodeGetLabs();
                transfer = obj.Session.pmodeGetTransfer();
                sender = transfer.getDataSender();
                manager = transfer.getTransferManager();
                % Prepare the tracker and sender objects for the transfer.
                % Note: During debug sessions, you must overwrite the value of the following
                % variable, otherwise the labs will halt the transfer on your first
                % breakpoint.
                isDebugging = false;
                monitor = manager.initiateTransfer(sender, firstLab, isDebugging);
            catch E
                error( message( 'parallel:convenience:PmodeNotRunning' ) );
            end

            % Create error messages that are specific to a transfer from client to
            % lab. Store function handles that can then be evaluated on the client.
            errorMsgStruct = struct('undefinedVariable', ...
                                    @() message( ...
                                        'parallel:convenience:VariableUndefinedInClient', clientvarname ), ...
                                    'failedToSerialize', ...
                                    @() message( ...
                                        'parallel:convenience:FailedToSerializeVariable', clientvarname ) );

            % Execute parallel.internal.pool.client2lab on all the labs.  This will cause firstLab to
            % send us a "ok to start sending" message.
            cmd = sprintf('parallel.internal.pool.client2lab(%s, [ %s ], ''%s'');', ...
                          num2str(monitor.getTransferSeqNumber()), ...
                          num2str(labidxes), labvarname);
            labs.eval(cmd, monitor.getLabsCompletionObserver());

            % From this point onwards, the data send operation is identical to that when
            % the client is sending data to a lab, so let a package method do the sending.
            parallel.internal.pool.doInteractiveSend(...
                clientvarname, sender, monitor, errorMsgStruct);
        end
        function lab2client(obj, labvarname, labidx, clientvarname)
        %lab2client Client code to transfers a variable from lab to client.
        %   Caller should have made sure that lab is an integer, and that
        %   isvarname(clientvarname) and isvarname(labvarname) return true.

            try
                labs = obj.Session.pmodeGetLabs();
                transfer = obj.Session.pmodeGetTransfer();
                manager = transfer.getTransferManager();
                receiver = transfer.getDataReceiver();
                % Prepare the tracker and receiver objects for the transfer.
                % Note: During debug sessions, you must overwrite the value of the following
                % variable, otherwise the labs will halt the transfer on your first
                % breakpoint.
                isDebugging = false;
                monitor = manager.initiateTransfer(receiver, labidx, isDebugging);
            catch E
                error( message( 'parallel:convenience:PmodeNotRunning' ) );
            end

            try
                % Execute lab2client on all the labs to start the send.
                cmd = sprintf('parallel.internal.pool.lab2client(%d, ''%s'', %d);', ...
                              monitor.getTransferSeqNumber(), labvarname, labidx);
                labs.eval(cmd, monitor.getLabsCompletionObserver());
            catch err
                monitor.setLocalMError(err.identifier, err.message);
                rethrow(err);
            end

            % At this point, the data receive operation is identical to that when
            % the lab is sending data to a client, so let a package method do the receive.
            serializedValue = parallel.internal.pool.doInteractiveReceive(receiver, monitor);

            try
                out = distcompdeserialize(serializedValue);
                assignin('base', clientvarname, out);
            catch E
                errID = 'parallel:convenience:TransferFailed';
                errMsgObj = message( errID, clientvarname );
                errMsg = getString( errMsgObj );
                monitor.setLocalMError(errID, errMsg);
                error( errMsgObj );
            end

            %
            % We are now done, and have to wait till the remote side is done.
            monitor.setLocalFinished();
            iLoopUntilErrorOrTrue(monitor, @() monitor.isRemoteFinished());
        end
    end
    
    methods (Access = protected)
        function postStartActions(obj, session)
            if obj.ShowUI
                com.mathworks.toolbox.distcomp.parallelui.ParallelUI.start(...
                    session.pmodeGetLabs());
                obj.IsGUIOpen = true;
            end
        end
        function postSessionDestroyedActions(~)
            try
                % Disable the command input window.
                com.mathworks.toolbox.distcomp.parallelui.ParallelUI.suspend();
            catch err
                dctSchedulerMessage(1, ['Failed to disable the GUI due to the following ', ...
                                    'error:\n%s'], err.message);
            end
        end
        function postShutdown(obj, stopFlag)
            if ~strcmp(stopFlag, 'leaveguiopen') && obj.IsGUIOpen
                try
                    com.mathworks.toolbox.distcomp.parallelui.ParallelUI.stop();
                catch err
                    dctSchedulerMessage(1, ['Failed to stop the GUI due to the ', ...
                                        'following error:\n%s'], err.message);
                end
                obj.IsGUIOpen = false;
            end
        end
    end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
function iLoopUntilErrorOrTrue(monitor, func)
% This function only returns normally when func returns true is true and there
% are no errors.
    while true
        err = monitor.getOutsideLocalMError();
        if ~isempty(err)
            error(char(err.getErrorIdentifier()), '%s', char(err.getErrorMessage()));
        end
        if func()
            break;
        end
        pause(0.01);
    end
end

%#ok<*JAPIMATHWORKS> transition class.
