% LocalUtils - utility functions used by Local cluster / localscheduler

% Copyright 2011-2020 The MathWorks, Inc.

classdef ( Hidden, Sealed ) LocalUtils

    methods ( Static, Access = private )
        function mpiAccept( fname_sentinel, fname_port, expectedNumlabs )
        % Ensure the sentinel exists
            fh = fopen( fname_sentinel, 'wt' );
            fclose( fh );

            % Open the port and set the parallel tag
            port = mpigateway( 'openport' );
            fh = fopen( fname_port, 'wt' );
            if fh == -1
                error(message('parallel:cluster:MpiFailToWritePortInfo', fname_port));
            end
            fprintf( fh, '%s\n', port );
            fclose( fh );

            % Delete the sentinel to indicate that we're ready
            delete( fname_sentinel );

            % Now, the other labs start looking out
            mpigateway( 'servaccept', port, expectedNumlabs - 1 );

            % Delete the port file because we're done with it now
            delete( fname_port );
        end

        function mpiConnect( fname_sentinel, fname_port, expectedNumlabs )
        % wait until the port file exists
            while ~exist( fname_port, 'file' )
                pause(0.5);
            end
            % now wait until the sentinel file has gone - this indicates that lab 1
            % has finished writing and closed the port file
            while exist( fname_sentinel, 'file' )
                pause( 0.5 );
            end
            % Read the port
            fh = fopen( fname_port, 'rt' );
            if fh == -1
                error(message('parallel:cluster:MpiFailToReadPortInfo', fname_port));
            end
            port = fgetl( fh );
            fclose( fh );

            % Perform the connection
            mpigateway( 'clientconn', port, expectedNumlabs - 1 );
        end
    end

    methods ( Static )
        function localSched = getScheduler()
        % Get the scheduler instance, setting things up correctly.
            persistent singletonScheduler
            if isempty(singletonScheduler)
                singletonScheduler = parallel.internal.local.LocalScheduler();
                factoryClusterSize = min(feature('numcores'), ...
                    double(singletonScheduler.getMaxAllowedNumWorkers()));
                singletonScheduler.setMaxNumWorkers( factoryClusterSize );
                mlock;
            end
            localSched = singletonScheduler;
        end
        
        function maxNum = getMaxAllowedNumWorkers()
        % Get the maximum number of workers allowed for local scheduler
            localSched = parallel.internal.apishared.LocalUtils.getScheduler();
            maxNum = double( localSched.getMaxAllowedNumWorkers() );
        end

        function installLocalSignalHandler()
        % Install the signal handler on UNIX, does nothing on other platforms.
            if isunix
                installLocalSignalHandler();
            end
        end

        function parallelConnectAccept( fnameroot, taskID, expectedNumlabs )
        % perform the MPI connect/accept using the filesystem for the local
        % scheduler.

            import parallel.internal.apishared.LocalUtils

            % Perform the parallel connect/accept for local scheduler jobs
            fname_sentinel = [fnameroot, '.lock'];
            fname_port     = [fnameroot, '.port'];

            % Create a CancelWatchdog to make sure it doesn't take us too long
            % to connect/accept.
            import parallel.internal.shared.CancelWatchdog;
            exitCode = 1;
            timeout  = 480;
            cw       = CancelWatchdog( exitCode, timeout );

            if taskID == 1
                LocalUtils.mpiAccept( fname_sentinel, fname_port, expectedNumlabs );
            else
                LocalUtils.mpiConnect( fname_sentinel, fname_port, expectedNumlabs );
            end

            % Indicate to the CancelWatchdog that we have passed the
            % time-critical point
            cw.ok();

            % Sort out the ordering
            tasksByIndex  = gcat( taskID );
            [~, desOrder] = sort( tasksByIndex );
            mpigateway( 'desiredlabordering', desOrder );
        end
    end
end
