function job = createCommunicatingJob( obj, varargin )
%createCommunicatingJob Create a new communicating job
%    job = createCommunicatingJob(cluster) creates a communicating job object
%    for the identified cluster.
%
%    job = createCommunicatingJob(..., 'p1', v1, 'p2', v2, ...) creates a
%    communicating job object with the specified property values.
%
%    job = createCommunicatingJob(..., 'Type', 'Pool', ...) creates a
%    'Pool'-type communicating job. This is the default if 'Type' is not
%    specified. A 'Pool'-type job runs the specified task function with a MATLAB
%    Pool available to run the body of PARFOR loops or SPMD blocks.
%
%    job = createCommunicatingJob(..., 'Type', 'SPMD', ...) creates an
%    'SPMD'-type communicating job where the specified task function is run
%    simultaneously on each worker, and lab* functions can be used for
%    communication.
%
%    job = createCommunicatingJob(..., 'Profile', 'profileName',...)  creates a
%    communicating job object with the property values specified in the profile
%    'profileName'.  If not specified and the cluster has a 'Profile' specified,
%    then the cluster's profile is automatically applied.
%
%    Example:
%    % Consider the function 'myFunction' which uses PARFOR:
%    function result = myFunction(N)
%        result = 0;
%        parfor ii=1:N
%           result = result + max(eig(rand(ii)));
%        end
%    end
%
%    % Create a communicating job object to evaluate myFunction on the cluster.
%    myCluster = parcluster; % create the default cluster object
%    j = createCommunicatingJob(myCluster); % use the default 'Pool' type
%
%    % Add the task to the job, supplying an input argument
%    createTask(j, @myFunction, 1, {100});
%
%    % Set the number of workers required for parallel execution.
%    j.NumWorkersRange = [5 10];
%
%    % Run the job.
%    submit(j);
%
%    % Wait until the job is finished.
%    wait(j);
%
%    % Retrieve and display task results.
%    out = fetchOutputs(j)
%
%    % Delete the job.
%    delete(j);
%
%    See also parcluster, parallel.Job/createTask, parallel.Job/submit,
%             parallel.Cluster/createJob.

% Copyright 2011-2023 The MathWorks, Inc.

import parallel.internal.types.Variant
import parallel.internal.cluster.types.SpmdEnabled
import parallel.internal.customattr.PropSet

% Convert any string inputs to character vectors
[varargin{:}] = parallel.internal.general.convertPVPairsToChars(varargin{:});

validateattributes( obj, {'parallel.Cluster'}, {'scalar'}, 'createCommunicatingJob', 'cluster', 1 );

try
    % 'Type' is "special", we need to tease that out before we can even
    % build the job.
    [gotVariant, userVariant, otherArgs] = PropSet.extractOneProperty( ...
        'Type', varargin{:} );
    if gotVariant
        variant = Variant.fromNameI( userVariant );
        if ~ ( variant == Variant.CommunicatingPoolJob || ...
               variant == Variant.CommunicatingSPMDJob )
            error(message('parallel:cluster:InvalidType', userVariant));
        end
    else
        variant = Variant.CommunicatingPoolJob;
    end

    % 'ExecutionMode' is a hidden property that specifies if this is an
    % interactive or batch job.
    [gotExecutionMode, executionMode, otherArgs] = PropSet.extractOneProperty( ...
        'ExecutionMode', otherArgs{:} );
    if ~gotExecutionMode
        % If execution mode was not specified, the user is calling
        % createCommunicatingJob directly, which can only happen in
        % a batch workflow.
        executionMode = parallel.internal.types.ExecutionMode.Batch;
    end

    % SpmdEnabled gets resolved here.
    buildFcn = @(obj) buildCommunicatingJob( obj, variant, executionMode );
    [gotSpmdEnabled, spmdEnabled, otherArgs] = PropSet.extractOneProperty( ...
        'SpmdEnabled', otherArgs{:} );
    if gotSpmdEnabled
        spmdEnabledEnum = SpmdEnabled.fromLogical(spmdEnabled);
        if variant == Variant.CommunicatingSPMDJob && spmdEnabledEnum ~= SpmdEnabled.True
            error(message('parallel:job:InvalidSpmdEnabledValue'));
        end
        otherArgs = [otherArgs, {'SpmdEnabled', spmdEnabled}];
    elseif variant == Variant.CommunicatingSPMDJob
        otherArgs = [otherArgs, {'SpmdEnabled', true}];
    else
        otherArgs = [otherArgs, {'SpmdEnabledEnum', SpmdEnabled.Auto}];
    end
    obj.preCreateJob( variant, otherArgs{:} );
    job = buildFcn( obj );
    obj.applyJobProperties( job, otherArgs{:} );
    dctSchedulerMessage(2, 'Created Communicating %s', parallel.internal.logging.LogHelper.getJobsString(job));
catch E
    throw( E );
end
end

