function hSubmitCommunicatingJob( obj, job, jobSupport, jobSupportID )
% hSubmitCommunicatingJob - start workers for local pools

% Copyright 2011-2023 The MathWorks, Inc.

import parallel.cluster.Local;
import parallel.internal.cluster.SubmissionChecks;
import parallel.internal.cluster.CJSJobMethods;
import parallel.internal.apishared.SmpdGateway;


% Figure out which variant of communicating job we are, which will determine
% how we launch the workers. There are three main variants (the names
% correspond to the names of the relevant java classes:
%
% -ConcurrentJob
% -MpiExecJob
% -ParallelJob
% 
% ConcurrentJobs correspond to pools started with
% SpmdEnabled=auto or SpmdEnabled=false. For SmpdEnabled=true, we're a
% MpiExecJob, unless distcomp.feature('LocalUseMpiexec') has been set to
% false, in which case we're a parallel job.
%
% To complicate things, there are multiple mpi implementations. On Windows, we ship two mpi implementations, usually
% we're using msmpi, but this can potentially be overridden with
% distcomp.feature('LocalUseMsmpi'). On mac and linux we use mpich2.
% 
% We also include mpich3 for linux, although it is not used at the moment
% due to g1877160. To work around a bug in mpich3 connect/accept, concurrent and
% parallel jobs on linux (which is the only place we're using mpich3)
% use mpiexec (despite the name).

% Use concurrent scheduling for SpmdEnabled auto or false.
jobIsConcurrent = job.hGetProperty('SpmdEnabledEnum') ~= parallel.internal.cluster.types.SpmdEnabled.True;

USE_MPIEXEC = false;
USE_MSMPI   = false;
USE_MPICH3  = false;

if ~jobIsConcurrent && distcomp.feature('LocalUseMpiexec')
    if distcomp.feature('LocalUseMsmpi')
        USE_MPIEXEC = true;
        USE_MSMPI   = true;
    else
        % Checks if we can launch smpd, which we generally expect to be
        % able to do.
        USE_MPIEXEC = SmpdGateway.canUseSmpd(obj.LocalSched);
    end
end
    
SubmissionChecks.checkSimultaneousTasksBeforeSubmission( job );

workerLims = job.NumWorkersRange;
minW       = workerLims(1);

% Local cluster has a complicated series of tests on NumWorkers.
iLocalNumWorkersCheck( obj, minW );

% If we get here, we know that minW <= obj.NumWorkers, so we can be sure maxW >=
% minW:
maxW = min( workerLims(2), obj.NumWorkers );

CJSJobMethods.duplicateTasksOfSimultaneousJob( job, jobSupport, jobSupportID, maxW );
jobSupport.prepareJobForSubmission( job, jobSupportID );

% Get the duplicated tasks
tasks    = job.Tasks;
numTasks = numel( tasks );

% discover the directory to use for temporary stuff relating to this job
configDir = jobSupport.getJobDirOnClient( jobSupportID );

if USE_MPIEXEC && USE_MSMPI
    configDir = distcompGetShortPath(configDir);
    % distcompGetShortPath does not work if with non-ASCII folder names
    % 8 characters or shorter.
    if iIsNonASCII(configDir)
        configDir = tempname;
        createConfigDir = true;
    else
        createConfigDir = false;
    end
end

% Communicating jobs need the license information in the command if they're
% being launched directly. The SmpdGateway handles licenses for mpiexec jobs.
needLicenseArgInCommand = ~USE_MPIEXEC;

% Set up the environment variables
[jsNames, jsVals]     = jobSupport.getJobSubmissionEnvVars( jobSupportID );
[cNames,  cVals ]     = obj.getClusterEnvVarsForJobSubmission();
[cjNames, cjVals]     = iCommJobEnvVars( USE_MPIEXEC, USE_MSMPI, jobIsConcurrent, configDir );
envNames              = [jsNames(:); cNames(:); cjNames(:)];
envValues             = [jsVals(:); cVals(:); cjVals(:)];
commandArray          = obj.calculateMatlabCommandForJob( job, needLicenseArgInCommand );
[absLog, relLog, ext] = jobSupport.getCommJobLogFile( jobSupportID );
absLogNoExt           = absLog(1:length(absLog) - length(ext));
numWorkersRange       = cast([minW, maxW], 'like', zeros('int32'));

import parallel.internal.cluster.ParpoolJobConstants
startedByParpool      = ParpoolJobConstants.isParpoolJob(job.ApiTag);

if jobIsConcurrent
    if USE_MPICH3
         mpiexecHydraCmd = { fullfile( matlabroot, 'bin', dct_arch, 'mpiexec' ), ...
             '-genvlist', strjoin(envNames, ','), ...
             '-n', '1', '-launcher', 'fork'}; %#ok<UNRCH> MPICH3 hardcoded to false due to g1877160
         commandArray = [mpiexecHydraCmd, commandArray];
    end
    
    jobArgs = parallel.internal.local.ConcurrentJobArgs();
    jobArgs.WorkerCmdLine = commandArray;
    jobArgs.setEnvironment(envNames, envValues);
    jobArgs.LogFile = absLog;
    jobArgs.LogFileNoExtension = absLogNoExt;
    jobArgs.LogFileExtension = ext;
    jobArgs.MinNumWorkers = numWorkersRange(1);
    jobArgs.MaxNumWorkers = numWorkersRange(2);
    jobArgs.StartedByParpool = startedByParpool;
    uuid = obj.LocalSched.createShrinkableConcurrentJob(jobArgs);
    
elseif USE_MPIEXEC && USE_MSMPI

    % Workaround for msmpi limitation: (see g1296083 and g2207655) working
    % directory for msmpi cannot contain unicode characters. So, start the
    % mpiexec process in a presumed-safe location (the root directory of the
    % drive where the OS is installed), and use the "-sd" startup option to
    % ensure the worker MATLABs start in the client's current directory.
    mpiexecShortWorkDir = [getenv('SystemDrive'), '\'];
    commandArray = [commandArray, '-sd', pwd];
    
    % We use '-l' to tell mpiexec to prepend log output with the MPI rank
    % (i.e. spmdIndex-1)
    mpiexecCmdArray = { fullfile(matlabroot, 'bin', dct_arch, 'msmpi', 'mpiexec.exe'), ...
                        '-genvlist', strjoin(envNames, ','), '-l'};
    
    jobArgs = parallel.internal.local.MpiexecJobArgs;
    jobArgs.MpiexecCmdLine = mpiexecCmdArray;
    jobArgs.MpiexecWorkDir = mpiexecShortWorkDir;
    jobArgs.WorkerCmdLine = commandArray;
    jobArgs.setEnvironment(envNames, envValues);
    jobArgs.ConfigDir = configDir;
    jobArgs.CreateConfigDir = createConfigDir;
    jobArgs.LogFile = absLog;
    jobArgs.MinNumWorkers = numWorkersRange(1);
    jobArgs.MaxNumWorkers = numWorkersRange(2);
    % We need to close the input stream on Windows as MPIEXEC can
    % sometimes hang on close if the stream is still open. Note, we
    % don't do this for Unix as Unix MPIEXEC appears to spin at 100%
    % when we close the input stream.
    jobArgs.CloseInputStream = ispc();
    jobArgs.StartedByParpool = startedByParpool;
    uuid = obj.LocalSched.createMsMpiexecJob(jobArgs);
elseif USE_MPIEXEC && ~USE_MSMPI
    jobArgs = parallel.internal.local.MpiexecJobArgs;
    jobArgs.WorkerCmdLine = commandArray;
    jobArgs.setEnvironment(envNames, envValues);
    jobArgs.ConfigDir = configDir;
    jobArgs.LogFile = absLog;
    jobArgs.MinNumWorkers = numWorkersRange(1);
    jobArgs.MaxNumWorkers = numWorkersRange(2);
    jobArgs.CloseInputStream = ispc();
    jobArgs.StartedByParpool = startedByParpool;
    if USE_MPICH3
        jobArgs.MpiexecCmdLine = getHydraMpiexecArgs( envNames ); %#ok<UNRCH> MPICH3 hardcoded to false due to g1877160
        uuid = obj.LocalSched.createHydraMpiexecJob(jobArgs);
    else
        jobArgs.MpiexecCmdLine = SmpdGateway.getMpiexecArgs( envNames );
        uuid = obj.LocalSched.createSmpdMpiexecJob(jobArgs);
    end
else
    if USE_MPICH3
         mpiexecHydraCmd = { fullfile( matlabroot, 'bin', dct_arch, 'mpiexec' ), ...
             '-genvlist', strjoin(envNames, ','), ...
             '-n', '1', '-launcher', 'fork'};  %#ok<UNRCH> MPICH3 hardcoded to false due to g1877160
         commandArray = [mpiexecHydraCmd, commandArray];
    end
    
    jobArgs = parallel.internal.local.ConcurrentJobArgs();
    jobArgs.WorkerCmdLine = commandArray;
    jobArgs.setEnvironment(envNames, envValues);
    jobArgs.LogFile = absLog;
    jobArgs.LogFileNoExtension = absLogNoExt;
    jobArgs.LogFileExtension = ext;
    jobArgs.MinNumWorkers = numWorkersRange(1);
    jobArgs.MaxNumWorkers = numWorkersRange(2);
    jobArgs.StartedByParpool = startedByParpool;
    uuid = obj.LocalSched.createFixedConcurrentJob(jobArgs);
end

uuids = obj.LocalSched.createTaskViews(uuid);

assert(numel(uuids) == numTasks);

job.JobSchedulerData = struct('type', 'local', ...
                              'taskUUIDs', {uuids} , ...
                              'taskIDs', {1:numTasks}, ...
                              'submitProcInfo', obj.ProcessInformation, ...
                              'logRelToStorage', {relLog});

obj.LocalSched.submit(uuid);

end

function cmdArray = getHydraMpiexecArgs( envNames ) %#ok<DEFNU> MPICH3 hardcoded to false due to g1877160
    % getHydraMpiexecArgs - return a cell array of arguments for use with
    % mpiexec for hydra/mpich3
    % - envNames: the list of environment variable names to forward
    % returns
    % - cmdArray: cell array of arguments to mpiexec.

    % Calculate the full path to the mpiexec executable
    cmd = fullfile( matlabroot, 'bin', dct_arch, 'mpiexec' );

    % Build the genvlist portion
    eNames = sprintf( '%s,', envNames{:} );
    eArgs = { '-genvlist', eNames(1:end-1) };

    % Concoct the complete argument list
    cmdArray = [cmd, eArgs, { '-l'  }];
end

function tf = iIsNonASCII( c )
% Return false if the character array, c, only contains printable ASCII characters,
% true otherwise.
tf = any(c > 126 | c < 32);
end

% --------------------------------------------------------------------------
% iCommJobEnvVars - communicating job environment variables, dependent on
% whether mpiexec is being used.
function [ names, vals ] = iCommJobEnvVars( useMpiExec, useMsmpi, jobIsConcurrent, configDir )

names = {};
vals  = {};

if ~jobIsConcurrent
    % Indicate to distcomp_evaluate_filetask_core that we wish to have mpiInit
    % called for us.
    names = { 'MDCE_PARALLEL'};
    vals  = { 'true' };
end

loopbackAddress = parallel.internal.general.HostNameUtils.getLoopbackHostname();
if isempty( getenv( 'MDCE_OVERRIDE_CLIENT_HOST' ) )
    names{end+1} = 'MDCE_OVERRIDE_CLIENT_HOST'; % append these to envNames
    vals{end+1}  = loopbackAddress;
end
if isempty( getenv( 'MPICH_INTERFACE_HOSTNAME' ) )
    names{end+1} = 'MPICH_INTERFACE_HOSTNAME';
    vals{end+1}  = loopbackAddress;
end

% Use func2str on the decode function to allow dependencies to be tracked.
if useMpiExec
    % Just define extraNames /values in one go here
    names{end+1} = 'PARALLEL_SERVER_DECODE_FUNCTION';
    vals{end+1}  = func2str( @parallel.internal.decode.localMpiexecTask );
    if useMsmpi
        names = [ names, {'MDCE_FORCE_MPI_OPTION', ...
                          'MSMPI_PRINT_ENVIRONMENT_BLOCK', ...
                          'MSMPI_PRINT_ENVIRONMENT_BLOCK_FILE'} ];
        vals  = [ vals, {'msmpi', ...
                         '1', ...
                         fullfile(configDir, 'workerEnvBlock')} ];
        % Note that for MSMPI we must append the licensing environment variables
        % here to those specified by mpiexec - for MPICH2 they're specified
        % as part of the smpd environment because of an MPICH2 limitation.
        [names, vals] = parallel.internal.cluster.appendLicenseEnvVars(names, vals);
    end
elseif jobIsConcurrent
    names{end+1} = 'PARALLEL_SERVER_DECODE_FUNCTION';
    vals{end+1}  = func2str( @parallel.internal.decode.localConcurrentTask );
else
    % Use the connect/accept decode function, and force the "sock" MPI build.
    names = [names, { 'PARALLEL_SERVER_DECODE_FUNCTION', ...
                      'MDCE_SENTINAL_LOCATION', ...
                      'MDCE_FORCE_MPI_OPTION' } ];

    vals  = [vals, { func2str( @parallel.internal.decode.localParallelTask ), ...
                     fullfile(configDir, 'mpi'), ...
                     'sock' } ];
end
end

% --------------------------------------------------------------------------
% iLocalNumWorkersCheck - check the minimum number of workers requested for the
% job and throw an error if necessary.
function iLocalNumWorkersCheck( obj, minW )
if minW > obj.MaxAllowedNumWorkers
    % Never going to run
    error(message('parallel:cluster:LocalMaxAllowedWorkersExceeded', minW, obj.MaxAllowedNumWorkers));
elseif minW > obj.NumWorkers
    % NumWorkers must be < obj.MaxAllowedNumWorkers, so could be increased
    if isempty( obj.Profile )
        error(message('parallel:cluster:LocalNumWorkersExceeded', obj.NumWorkers, minW, obj.MaxAllowedNumWorkers));
    else
        error(message('parallel:cluster:LocalProfileNumWorkersExceeded', obj.Profile, obj.NumWorkers, minW, obj.MaxAllowedNumWorkers));
    end
end
end
