function hSubmitIndependentJob( obj, job, jobSupport, jobSupportID )
% hSubmitIndependentJob - submit tasks for execution

% Copyright 2011-2023 The MathWorks, Inc.

import parallel.cluster.Local;
import parallel.internal.cluster.SubmissionChecks;
import parallel.internal.apishared.FilenameUtils;

% Independent jobs always need the license information in the command
needLicenseArgInCommand = true;

if isempty( job.BatchVecData )
    [tasks, numTasks] = SubmissionChecks.checkIndependentTasksBeforeSubmission( job );
    taskSIDs = arrayfun( @hGetSupportID, tasks, 'UniformOutput', false );
    taskIDs = arrayfun( @(task) task.Id, tasks );
else
    numTasks = job.BatchVecData.NumTasks;
    taskSIDs = job.BatchVecData.TaskSIds;
    taskIDs  = job.BatchVecData.TaskIDs;
end

jobSupport.prepareJobForSubmission( job, jobSupportID );
[jsNames, jsVals]   = jobSupport.getJobSubmissionEnvVars( jobSupportID );
[cNames,  cVals ]   = obj.getClusterEnvVarsForJobSubmission();
[ijNames, ijVals]   = iIndJobEnvVars();
jobNames            = [ jsNames(:); cNames(:); ijNames ];
jobVals             = [ jsVals(:); cVals(:); ijVals ];
commandArray        = obj.calculateMatlabCommandForJob( job, needLicenseArgInCommand );
taskUUIDs           = cell(numTasks, 1);
taskLogs = jobSupport.getIndJobTaskLogFile( jobSupportID, taskSIDs );

taskArgs = parallel.internal.local.IndependentTaskArgs;
taskArgs.WorkerCmdLine = commandArray;
for ii = 1:numTasks
    % Set the task location environment variable correctly
    taskSID = taskSIDs{ii};
    [taskNames, taskVals] = jobSupport.getTaskSubmissionEnvVars( taskSID );
    taskArgs.setEnvironment([jobNames; taskNames(:)], [jobVals; taskVals(:)]);
    taskArgs.LogFile = taskLogs{ii};
    
    % Mechanism to track the UUID against the actual task
    taskUUIDs{ii} = obj.LocalSched.createIndependentTask(taskArgs);
end


% Data returned from local -
% localTaskUUIDs : the local task Id for each task - NOTE for parallel jobs where
%                  there is only one actual task every real task will map to
%                  the same local task
% taskIDs        : the Id of the parallel.task.CJSTask associated with a local
%                  task in the field above. It is an error for taskIDs and
%                  localTaskUUIDs to be of different lengths.

% TODO:later remove this abomination.
logRelToStorage = FilenameUtils.fixSlashes(regexprep(taskSIDs{1}, '[0-9]*$', ''));

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

% Now submit the actual java tasks
%
for ii = 1:numTasks
    obj.LocalSched.submit(taskUUIDs{ii});
end
end

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
function [ names, vals ] = iIndJobEnvVars()
names = { 'PARALLEL_SERVER_DECODE_FUNCTION' };
vals  = { func2str( @parallel.internal.decode.localSingleTask ) };
end
