function job = hBatchVec(obj, taskFcn, numArgsOut, argsIn, varargin)
%HBATCHVEC Run a MATLAB function multiple times on workers
%   j = HBATCHVEC(cluster, fcn, N, {x1, ..., xn}, 'NumTasks', M) runs the
%   function specified by a function handle or function name, fcn, with
%   input arguments specified by the cell array, {x1, ..., xn}, on workers of the
%   identified cluster.  The function returns j, a handle to the job object
%   that runs the function M times using M tasks. Each task evaluates the
%   function with the given arguments, x1,...,xn, returning N output
%   arguments.  The function file for fcn is copied to the worker.  If the
%   cluster object's Profile property is not empty, the profile is applied
%   to the job and tasks that run the function.
%
%   j = HBATCHVEC( ..., P1, V1, ..., Pn, Vn) allows additional parameter-
%   value pairs that modify the behavior of the job.  Accepted parameters
%   include:
%
%   - 'AdditionalPaths' - A character vector, string, string array, or cell
%     array of character vectors specifying paths to be added to the
%     workers' MATLAB path before the function is executed.
%
%   - 'AttachedFiles' - A character vector, string, string array, or cell
%     array of character vectors specifying the files and/or folders to be
%     transferred to the workers.
%
%   - 'AutoAddClientPath' - A logical value (true or false) specifying
%     whether user-added entries on the client's path are added to each 
%     workers' path.  The default is true.
%
%   - 'AutoAttachFiles' - A logical value specifying whether code files
%     are automatically attached to the job. If true, the function is
%     analyzed and the code files it depends upon are automatically
%     transferred to the workers. The default is true.
%
%   - 'CaptureDiary' - A logical value specifying whether diary output
%     is retrieved from the function call. See the DIARY function for how
%     to return this information to the client. The default is true.
%
%   - 'EnvironmentVariables' - A character vector, string, string array, or
%     cell array of character vectors specifying names of the environment
%     variables to copy from the client session to the workers.
%
%   - 'SlicedIndices' - An array of positive integers specifying the
%     indices of the input arguments {x1, ..., xn} that should be "sliced"
%     when evaluated by the workers. If the i-th input argument is sliced,
%     xi must be an array or a cell array with M elements. The m-th task
%     will evaluate the function using the m-th element of xi as the i-th
%     input argument.
%
%   All parameter-value pairs not listed above are passed into
%   parallel.Cluster/createJob.
%
%   Examples:
%   % Run a batch function that generates a random number 1000 times on a
%   % cluster
%   myCluster = parcluster; % creates the default cluster
%   j = hBatchVec(myCluster, @rand, 1, {1}, 'NumTasks', 1000);
%
%   % Run a batch function that generates random matrices rand(1), rand(2),
%   % ..., rand(1000) on a cluster
%   j = hBatchVec(myCluster, @rand, 1, {1:1000}, 'SlicedIndices', 1);
%
%   % Run a batch function that generates random column vectors rand(1, 1),
%   % rand(2, 1), ..., rand(1000, 1) on a cluster
%   j = hBatchVec(myCluster, 'rand', 1, {1:1000, 1}, 'SlicedIndices', 1);
%   
%   See also batch, parcluster, parallel.Cluster/createJob,
%            parallel.Job/createTask.

% Copyright 2019-2020 The MathWorks, Inc.
    
    % Validate inputs
    validateattributes(obj, {'parallel.Cluster'}, {'scalar'}, mfilename, 'cluster', 1 );
    isFunction = isa(taskFcn, 'function_handle');
    if isFunction
        validateattributes(taskFcn, {'function_handle'}, {'scalar'}, mfilename, 'taskFcn', 2 );
    else
        validateattributes(taskFcn, {'char'}, {'scalartext'}, mfilename, 'taskFcn', 2 );
    end
    validateattributes(numArgsOut, {'numeric'}, {'nonnegative', 'scalar', 'integer'}, mfilename, 'numArgsOut', 3 );
    validateattributes(argsIn, {'cell'}, {}, mfilename, '', 4 );

    % Parse name/value pairs
    p = inputParser();
    p.addOptional('NumTasks', 1);
    p.addOptional('SlicedIndices', []);
    p.addOptional('CaptureDiary', true);
    p.addOptional('AutoAttachFiles', true);
    
    p.KeepUnmatched = true; % Unmatched arguments will be passed to createJob()
    p.parse(varargin{:});
    results = p.Results;

    numTasks = results.NumTasks;
    slicedIndices = results.SlicedIndices;
    captureDiary = results.CaptureDiary;
    autoAttachFiles = results.AutoAttachFiles;
    
    % Validate name/value pairs
    validateattributes(numTasks, {'numeric'}, {'positive', 'scalar', 'integer'}, mfilename, 'NumTasks');
    validateattributes(slicedIndices, {'numeric'}, {'positive', 'integer'}, mfilename, 'SlicedIndices');
    validateattributes(captureDiary, {'logical'}, {'scalar'}, mfilename, 'CaptureDiary');
    validateattributes(autoAttachFiles, {'logical'}, {'scalar'}, mfilename, 'AutoAttachFiles');
    
    % Determine the number of tasks to create, which can either be
    % specified directly via 'NumTasks' or inferred via 'SlicedIndices'
    if max(slicedIndices) > numel(argsIn)
        error(message('parallel:batch:SlicedIndicesTooLarge'));
    end
    numelPerSlicedArg = cellfun(@numel, argsIn(slicedIndices));
    if numel(unique(numelPerSlicedArg)) > 1
        error(message('parallel:batch:InconsistentSlicedIndices'));
    end
    numTasksInferred = max(1, max(numelPerSlicedArg));
    numTasksSpecified = ~any(strcmp(p.UsingDefaults, 'NumTasks'));
    slicedIndicesSpecified = ~any(strcmp(p.UsingDefaults, 'SlicedIndices'));
    if numTasksSpecified && slicedIndicesSpecified
        % Ensure that the 'NumTasks' and sliced arguments are consistent
        if numTasks ~= numTasksInferred
            error(message('parallel:batch:InconsistentNumTasks'));
        end
    elseif numTasksSpecified
        % Do nothing; proceed with specified numTasks
    elseif slicedIndicesSpecified
        numTasks = numTasksInferred;
    else
        % Do nothing; proceed with default numTasks = 1
    end
    
    % If automatically attaching files, perform the dependency analysis
    % once here for all tasks rather than for each task later
    if autoAttachFiles
        if isFunction
            functionsToAnalyze = parallel.internal.apishared.AttachedFiles.convertFunctionHandleForDependencyAnalysis( taskFcn );
        else
            functionsToAnalyze = {taskFcn};
        end
        files = cellfun( @parallel.internal.apishared.FilenameUtils.safeWhich, functionsToAnalyze, 'UniformOutput', false );
        autoAttachedFiles = parallel.internal.apishared.AttachedFiles.calculateAttachedFiles( files );
        if isempty(autoAttachedFiles)
            autoAttachedFiles = {};
        else
            autoAttachedFiles = autoAttachedFiles(:); % Make sure autoAttachedFiles is a column vector
        end
    else
        autoAttachedFiles = {};
    end
    
    % Create the job, passing any arguments not processed above (e.g.
    % 'AdditionalPaths', 'AttachedFiles', 'AutoAddClientPath',
    % 'EnvironmentVariables') to createJob()
    unmatched = p.Unmatched;
    names = fields(unmatched);
    extraArgs = cell(1, 2 * numel(names));
    for ii = 1:numel(names)
        extraArgs{2 * ii - 1} = names{ii};
        extraArgs{2 * ii} = unmatched.(names{ii});
    end
    job = obj.createJob('AutoAttachFiles', autoAttachFiles, extraArgs{:});
    
    % Create the tasks
    jobSId = job.hGetSupportID();
    tasksVecInfo = struct();
    tasksVecInfo.NumTasks = numTasks;
    tasksVecInfo.Function = taskFcn;
    tasksVecInfo.NumOutputArguments = numArgsOut;
    tasksVecInfo.InputArguments = argsIn;
    tasksVecInfo.SlicedIndices = slicedIndices;
    tasksVecInfo.CaptureDiary = captureDiary;
    tasksVecInfo.AutoAttachedFiles = autoAttachedFiles;
    [taskIDs, taskSIds] = obj.hBuildTasksVec(job, jobSId, tasksVecInfo);
    
    % Create and store a cache of batchVec-related data on the job object,
    % which will allow us to take advantage of as many performance
    % optimizations as possible when submitting the job
    batchVecData = struct();
    batchVecData.NumTasks = numTasks;
    batchVecData.AutoAttachedFiles = autoAttachedFiles;
    batchVecData.TaskIDs = taskIDs;
    batchVecData.TaskSIds = taskSIds;
    job.BatchVecData = batchVecData;
    
    % Submit the job
    job.submit();
end
