function pool = parpool(aCluster, varargin)
%PARPOOL Create a parallel pool of workers on a cluster and return a pool object
%
%   pool = PARPOOL(cluster) creates and returns a pool on the specified
%   cluster object with its NumWorkers in the range [1, preferredNumWorkers]
%   for running parallel language features (parfor, spmd and distributed). 
%   preferredNumWorkers is the value defined in your parallel preferences.
%
%   pool = PARPOOL(cluster, numWorkers) creates and returns a pool on the
%   specified cluster with the specified number of workers. numWorkers can
%   be a positive integer or a range specified as a 2-element vector of 
%   integers. If NumWorkers is a range, the resulting pool has size as large
%   as possible in the range requested.
%
%   pool = PARPOOL(..., P1, V1, ..., Pn, Vn) allows additional
%   parameter-value pairs that specify further properties of the created
%   pool. The accepted parameters are:
%
%   - 'AttachedFiles' - a character vector, string, string array, or a cell array of
%   character vectors specifying files to be attached to the pool at the time of its creation.
%   Attach files to the pool to allow them to be used by the workers.
%
%   - 'SpmdEnabled' - a logical value (true or false) that enables the use
%   of the SPMD language construct. By default 'SpmdEnabled' is set to
%   true. If SpmdEnabled is set to false then you will not be able to use   
%   the SPMD language construct, however the parallel pool created should be
%   more robust against workers crashing.
%
%   - 'IdleTimeout' - an integer greater than zero that sets the time in
%   minutes after which the pool will automatically shut itself down if it is
%   idle. A pool is idle if it is not running code on the workers. By 
%   default 'IdleTimeout' is the same as the value in your parallel preferences.
%
%   - 'EnvironmentVariables' - A character vector, string, string array, or cell
%   array of character vectors that defines the names of environment variables
%   which will be copied from the client session to the workers.
%
%   Examples:
%   % Create a pool using the cluster myCluster.
%   % First create myCluster using 'parcluster'.
%   myCluster  = parcluster();
%   % Now create the pool.
%   myPool = parpool(myCluster);
%
%   % Create a pool using the cluster myCluster and attach the file 'sum.m'
%   to it.
%   myPool = parpool(myCluster, 'AttachedFiles', {'sum.m'});
%
%   % Create a pool using the cluster myCluster and disable the use of SPMD.
%   myPool = parpool(myCluster, 'SpmdEnabled', false);
%
%   % Create a pool using the cluster myCluster and an IdleTimeout of 2 hours.
%   myPool = parpool(myCluster, 'IdleTimeout', 120);
%
%   See also parpool, parfor, batch, gcp, parcluster, parallel.Pool, 
%   parallel.Pool/addAttachedFiles, parallel.Pool/updateAttachedFiles,
%   parallel.Pool/listAutoAttachedFiles, parallel.Pool/delete, 

%   Copyright 2013-2020 The MathWorks, Inc.

% This cleanupObj will notify listeners of the termination of this function
% by the user.
notifier = parallel.internal.pool.ISessionNotifier.getInstance();
% Since we do not yet have a pool and therefore no Session or
% SessionInfo we need to create a default SessionInfo in order to notify
% listeners in the event of any errors. The notifier gets to choose the
% detailed type of info.
sessionInfo = notifier.buildSessionInfo();
cleanupObj = parallel.internal.general.DisarmableOncleanup(...
    @() notifier.notifyAbortByUser(sessionInfo));

try
    pool = parallel.internal.pool.doParpool(aCluster, varargin{:});
catch err
    % The following nested try-catch block is necessary in order that the
    % error passed to the SessionFactory notifier includes parpool in the stack trace
    try
        throw(err)
    catch parpoolErr
        if ~strcmp(parpoolErr.identifier, 'parallel:convenience:ConnectionOpen')
            dctSchedulerMessage(0, 'Failed to open pool. Error:', parpoolErr);
            parallel.internal.pool.ISessionNotifier.getInstance().notifyFailedToStart(sessionInfo, parpoolErr);
        end
        % Since a more specific notification can be made, we can disarm the
        % original cleanupObj
        cleanupObj.disarm();
    end
    throw(err);
end
% Now completed pool creation, the cleanupObj can be disarmed
cleanupObj.disarm();
end
