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

% Copyright 2011-2022 The MathWorks, Inc.

import parallel.cluster.HPCServer
import parallel.internal.apishared.FilenameUtils
import parallel.internal.apishared.HPCServerUtils
import parallel.internal.apishared.WorkerCommand
import parallel.internal.cluster.SubmissionChecks

obj.ensureLicenseNumberIsSetOrError();
[tasks, numTasks] = SubmissionChecks.checkIndependentTasksBeforeSubmission(job);
jobSupport.prepareJobForSubmission(job, jobSupportID);
[cNames,  cVals ] = obj.getClusterEnvVarsForJobSubmission();
[jsNames, jsVals] = jobSupport.getJobSubmissionEnvVars(jobSupportID);
[ijNames, ijVals] = iIndJobEnvVars(obj);
allNames = [cNames(:); jsNames(:); ijNames(:)];
allVals  = [cVals(:);  jsVals(:);   ijVals(:)];
jobEnvironmentVariables = [allNames(:), allVals(:)];
numThreads = obj.NumThreads;

jobLocation = jobSupport.getJobDirOnClient(jobSupportID);
jobIsCommunicating = false;
[~, matlabExe, matlabArgs] = WorkerCommand.hpcServerCommand(obj.OperatingSystem, ...
    obj.ClusterMatlabRoot, obj.UseSOAJobSubmission, jobIsCommunicating);

taskNames = cell(numTasks, 1);
taskVals  = cell(numTasks, 1);
taskLogs  = cell(numTasks, 1);
taskIDs   = cell2mat(get(tasks, {'Id'}));
for ii = 1:numTasks
    taskSID = tasks(ii).hGetSupportID();
    [taskNames(ii), taskVals(ii)] = jobSupport.getTaskSubmissionEnvVars(taskSID);
    taskLogs{ii} = jobSupport.getIndJobTaskLogFile(jobSupportID, taskSID);
end
if numTasks > 1
    % Make sure all the tasks have the same environment variable name
    % because this is what the ServerConnection expects
    sameTaskEnvVarName = strcmp(taskNames{1}, taskNames(2:end));
    assert(all(sameTaskEnvVarName));
end

[username, password] = HPCServer.getDefaultUserCredentials();
% Ask the server connection to submit the job.
[schedulerJobIDs, schedulerTaskIDs, schedulerJobName] = obj.ServerConnection.submitJob(matlabExe, matlabArgs, ...
    jobEnvironmentVariables, taskNames{1}, ...
    job.Id, jobLocation, taskIDs, taskVals, taskLogs, ...
    username, password, numThreads);

% Set SchedulerIDs
schedulerAdditionalJobID = [];
% If the individual tasks do not have SchedulerIDs, this is an SOA job
if isempty(schedulerTaskIDs)
    % We may have either one or two jobIDs for an SOA job, depending on
    % the version of HPCServer
    if length(schedulerJobIDs) > 1
        schedulerAdditionalJobID = schedulerJobIDs(2);
    end
    taskSchedulerID = join(string(schedulerJobIDs), ",");
    schedulerIDs = repmat(taskSchedulerID, size(tasks));
else
    % format <jobID>.<taskID>
    schedulerIDs = schedulerJobIDs + "." + schedulerTaskIDs;
end
set(tasks, 'SchedulerID', schedulerIDs);

% TODO:later remove this abomination.
logTaskIDToken = '^taskID^';
logRelToStorage = sprintf('%s%s.log', regexprep(tasks(1).hGetSupportID(), '[0-9]*$', ''), logTaskIDToken);
logRelToStorage = FilenameUtils.fixSlashes(logRelToStorage);
schedulerData = distcomp.MicrosoftJobSchedulerData(obj.ClusterVersion, ...
    obj.ServerConnection.SchedulerVersion, obj.Host, ...
    obj.UseSOAJobSubmission, schedulerJobName, schedulerJobIDs(1), schedulerAdditionalJobID, ...
    schedulerTaskIDs, taskIDs, logRelToStorage, logTaskIDToken);
job.JobSchedulerData = schedulerData;
end

%-----------------------------------
function [names, vals] = iIndJobEnvVars(cluster)
decodePackage = 'parallel.internal.decode';
if cluster.UseSOAJobSubmission
    decodeFcnString = 'hpcserverSoaTask';
else
    decodeFcnString = 'hpcserverSingleTask';
end
names = {'PARALLEL_SERVER_DECODE_FUNCTION'};
vals  = {sprintf('%s.%s', decodePackage, decodeFcnString)};
end
