% FileStorage - Stores data objects in files. 
%
% FileStorage fulfills the responsibilities of an AbstractStorage object
% using the file system as a backing store for its data.
%
% A FileStorage object also provides some "file-specific" functions to CJS
% support (getTaskDiaryFile, serializeForSubmission, etc).

% Copyright 2014-2021 The MathWorks, Inc.

classdef (Hidden) FileStorage < parallel.internal.cluster.AbstractStorage

    properties (Constant, Access = private)
        FileSep = filesep() % OS-specific file separator
        
        % Timeout after which if fail to create a new job folder we give
        % up.
        JobCreationTimeoutSeconds = 60;
        JobCreationTimeout = seconds(parallel.internal.cluster.FileStorage.JobCreationTimeoutSeconds);
    end
    
    properties (Access = private)
        % This is a containers.Map that is used to ensure that each job from this
        % storage will get an incrementing value, even if the previous one has
        % been destroyed. This was originally a workaround to NFS cache problems
        % for mpiexec particularly, where the machine on which the job is about
        % to run still has the previous job and task in NFS cache memory. Thus
        % they are unable to run the new one until the cache has flushed. If the
        % name changes then this problem does not occur - however we need to
        % consider the possibility that multiple machines will be accessing the
        % same file storage simultaneously.
        %
        % As per g1199342, this is also used to improve performance of creating
        % tasks in a loop by obviating the need to perform a directory listing
        % to calculate the last known job or task.
        LastJobID
        LastTaskIDMap

        % A file system object to use to make file operations. This can be swapped
        % out for a test-implementation to allow testing without going directly to
        % the OS.
        FileSystem
    end
    
    % Private constructor (use static factory method to create instances)
    methods (Access = private)
        function obj = FileStorage(storageDir, fileSystem)
            import parallel.internal.apishared.FilenameUtils
            import parallel.internal.cluster.AbstractStorage

            storageDir = FilenameUtils.getAbsolutePath(storageDir); 
            storageDir = AbstractStorage.convertMappedDrivePathToUncPath(storageDir);
            
            % Warn if the storage directory is not valid
            isValidStorageDir = iIsWritable(storageDir);
            if ~isValidStorageDir
                warning(message('parallel:cluster:FileStorageNotWritable', storageDir));
            end

            % Call inherited constructor
            obj = obj@parallel.internal.cluster.AbstractStorage(storageDir);

            obj.IsValidStorageDir = isValidStorageDir;
            obj.LastJobID = [];
            obj.LastTaskIDMap = containers.Map('KeyType', 'char', 'ValueType', 'any');
            
            obj.RemoteBuilderFcn = 'makeFileStorageObject';
            obj.FileSystem = fileSystem;
        end
    end
    
    methods (Static)
        % Static factory method to create instance
        function obj = createFileStorage(dirinfo, fileSystem)
            import parallel.internal.cluster.AbstractStorage
            import parallel.internal.cluster.FileStorage
            import parallel.internal.files.MATLABFileSystem
            
            % Use a persistent map of job storage locations to FileStorage
            % objects to ensure that multiple cluster objects use the same
            % FileStorage object and hence, crucially, the same LastJobID
            % and LastTaskIDMap.
            persistent MAP
            if isempty(MAP)
                MAP = containers.Map('KeyType', 'char', 'ValueType', 'any');
            end
            
            % Check that we have a valid directory
            storageDir = AbstractStorage.getLocationsFromDirInfo(dirinfo);
            AbstractStorage.checkLocationIsString(storageDir);
            if ~exist(storageDir, 'dir')
                error(message('parallel:cluster:FileStorageInvalidFolder', storageDir));
            end
            
            if MAP.isKey(storageDir)
                obj = MAP(storageDir);
            else
                if nargin < 2
                    fileSystem = MATLABFileSystem();
                end
                obj = FileStorage(storageDir, fileSystem);
                MAP(storageDir) = obj;
            end
        end
        
        % Static method to create dummy task files if absent. This method
        % should only be used by built-in CJS scheduler types that rely on
        % the existence of these task files for file transfer code to work.
        function createDefaultTaskFilesIfAbsent(taskLocationPrefix)
            import parallel.internal.cluster.FileStorage
            
            allProps = enumeration('parallel.internal.types.TaskProp');
            [~, uniqueExtensionIndices] = iUnique(arrayfun(@(prop) prop.Extension, allProps));
            sampleProps = allProps(uniqueExtensionIndices);
            
            for ii = 1:numel(sampleProps)
                prop = sampleProps(ii);
                filename = taskLocationPrefix + prop.Extension;
                if ~exist(filename, 'file')
                    data = struct();
                    data.(prop.StorageName) = prop.DefaultValue;
                    prop.FileFormat.save(obj.FileSystem, filename, data);
                end
            end
        end
    end
    
    % Methods that involve parsing text with regular expressions. These
    % methods are static and hidden for unit testing.
    methods (Static, Hidden)
        function jobIDs = parseJobIDs(entryNames)
            assert(iscellstr(entryNames), 'parseJobIDs() only accepts a cellstr'); %#ok<ISCLSTR>
            
            if isempty(entryNames)
                jobIDs = []; % No entry names
                return
            end
            
            jobIDs = regexp(entryNames, 'Job([0-9]+)', 'tokens', 'once');
            jobIDs = [jobIDs{:}]; % Flatten cell array of cell arrays
            jobIDs = cellfun(@str2double, jobIDs);
        end
        
        function taskRanges = parseTaskRanges(entryNames)
            assert(iscellstr(entryNames), 'parseTaskRanges() only accepts a cellstr'); %#ok<ISCLSTR>
            
            if isempty(entryNames)
                taskRanges = {}; % No entry names
                return
            end
            
            % Convert {'Task1-4.in.mat', 'Task5.in.mat', ...} to
            % {{'1', '4'}, {'5', ''}, ...}
            taskRanges = regexp(entryNames, 'Task([0-9]+)-?([0-9]+)?', 'tokens');
            taskRanges = [taskRanges{:}]; % Flatten cell array of cell arrays

            % Convert each cellstr pair to numbers using str2double so
            % {{'1', '4'}, {'5', ''}, ...} --> {[1 4], 5, ...}
            % NB: This is done using double(string(...)) calls in a for
            % loop to achieve the best performance.
            for ii = 1:numel(taskRanges)
                taskRange = taskRanges{ii};
                if isempty(taskRange{2})
                    taskRanges{ii} = double(string(taskRange{1}));
                else
                    taskRanges{ii} = double(string(taskRange));
                end
            end
        end
        
        function ID = parseID(sId)
            assert(ischar(sId), 'parseID() only accepts a character vector');
            
            % Extract the ID from the sId - it's always the digits at the end
            strID = regexp(sId, '[0-9]*$', 'match', 'once');
            if isempty(strID)
                error(message('parallel:cluster:FileStorageEntityEndInID'));
            end
            ID = str2double(strID);
        end
    end
    
    % Implementation of create and destroy methods for job and tasks
    methods
        function [jobID, jobSId] = createJob(obj, variant)
            import parallel.internal.types.FileFormat
            import parallel.internal.types.JobProp
            
            obj.errorIfInvalid();

            deadline = datetime('now') + parallel.internal.cluster.FileStorage.JobCreationTimeout;
            while datetime('now') < deadline
                % Allocate the new job ID
                [jobID, jobSId] = obj.allocateNextJobID();

                % Create the job folder
                jobFolder = obj.getJobLocation(jobSId);
                [OK, errorMessage, errorID] = mkdir(jobFolder);
                if strcmp(errorID, 'MATLAB:MKDIR:DirectoryExists')
                    % Job folder already exists. This means that another
                    % client has created the folder since
                    % allocateNextJobID. Try again.
                    dctSchedulerMessage(1, 'Job folder %s already created. Trying again.', jobFolder);
                    OK = false;
                    continue;
                end
                if ~OK
                    mkdirError = MException(errorID, '%s', errorMessage);
                    ex = MException(message('parallel:cluster:FileStorageCannotCreateFolder', jobFolder));
                    ex = ex.addCause(mkdirError);
                    throw(ex);
                end
                % Successfully created job folder
                break;
            end
            
            if ~OK
                % Timed out creating job folder. This can only happen if
                % lots of clients are creating jobs concurrently in the
                % same job storage location.
                mkdirError = MException(errorID, '%s', errorMessage);
                ex = MException(message('parallel:cluster:FileStorageTimedOutCreatingFolder', jobFolder, ...
                    parallel.internal.cluster.FileStorage.JobCreationTimeoutSeconds));
                ex = ex.addCause(mkdirError);
                throw(ex);
            end
            
            % Create the job metadata file
            metadataFile = jobFolder + JobProp.Variant.Extension;
            data = struct(JobProp.Variant.StorageName, variant.Name);
            FileFormat.Text.save(obj.FileSystem, metadataFile, data);
        end
        
        function [taskIDs, taskSIds] = createTasks(obj, jobSId, taskInfos)
            import parallel.internal.types.FileFormat
            import parallel.internal.types.TaskProp
            
            obj.errorIfInvalid();
            
            % Allocate the new task IDs
            numTasks = numel(taskInfos);
            [taskIDs, taskSIds] = obj.allocateNextTaskIDs(jobSId, numTasks);
            
            % Create the task .in.mat files
            createDateTime = datetime('now', 'TimeZone', 'local');
            for ii = 1:numTasks
                taskID = taskIDs(ii);
                taskSId = taskSIds{ii};
                taskInfo = taskInfos{ii};
                
                data = struct();
                data.(TaskProp.Name.StorageName)               = "Task" + taskID;
                data.(TaskProp.Function.StorageName)           = taskInfo.Function;
                data.(TaskProp.NumOutputArguments.StorageName) = taskInfo.NumOutputArguments;
                data.(TaskProp.InputArguments.StorageName)     = taskInfo.InputArguments;
                data.(TaskProp.CaptureDiary.StorageName)       = taskInfo.CaptureDiary;
                data.(TaskProp.CreateDateTime.StorageName)     = createDateTime;
                if isfield(taskInfo, 'AutoAttachedFiles')
                    data.(TaskProp.DependentFiles.StorageName) = string(taskInfo.AutoAttachedFiles);
                else
                    data.(TaskProp.DependentFiles.StorageName) = string.empty();
                end
                data.(TaskProp.SchedulerID.StorageName)        = "";

                taskInFile = obj.getTaskLocation(taskSId) + TaskProp.Name.Extension;
                FileFormat.Mat.save(obj.FileSystem, taskInFile, data);
            end
        end
        
        function [taskIDs, taskSIds] = createTasksVec(obj, jobSId, tasksVecInfo)
            import parallel.internal.types.FileFormat
            import parallel.internal.types.TaskProp
            
            % Revert to the non-vectorized case if there is only 1 task.
            % NB: We can reuse tasksVecInfo for taskInfo so long as we
            % convert the input arguments to the non-grouped form
            numTasks = tasksVecInfo.NumTasks;
            if numTasks == 1
                taskInfo = tasksVecInfo;
                taskInfo.InputArguments = iGetArgsIn(1, tasksVecInfo.InputArguments, tasksVecInfo.SlicedIndices);
                [taskIDs, taskSIds] = createTasks(obj, jobSId, {taskInfo});
                return
            end
            
            obj.errorIfInvalid();
            
            % Allocate the new task IDs
            [taskIDs, taskSIds] = obj.allocateNextTaskIDs(jobSId, numTasks);
            taskNames = "Task" + taskIDs;
            
            % Create the new task group
            data = struct();
            data.(TaskProp.Name.StorageName)               = taskNames;
            data.(TaskProp.Function.StorageName)           = tasksVecInfo.Function;
            data.(TaskProp.NumOutputArguments.StorageName) = tasksVecInfo.NumOutputArguments;
            data.(TaskProp.InputArguments.StorageName)     = tasksVecInfo.InputArguments;
            data.(TaskProp.SlicedIndices.StorageName)      = tasksVecInfo.SlicedIndices;
            data.(TaskProp.CaptureDiary.StorageName)       = tasksVecInfo.CaptureDiary;
            data.(TaskProp.CreateDateTime.StorageName)     = datetime('now', 'TimeZone', 'local');
            data.(TaskProp.DependentFiles.StorageName)     = tasksVecInfo.AutoAttachedFiles;
            data.(TaskProp.SchedulerID.StorageName)        = "";
            
            groupedTaskInFile = obj.getGroupedTaskLocation(jobSId, taskIDs) + TaskProp.Name.Extension;
            FileFormat.Mat.save(obj.FileSystem, groupedTaskInFile, data);
        end

        function submitJob(obj, jobSId)
            import parallel.internal.types.FileFormat
            import parallel.internal.types.JobProp

            obj.errorIfInvalid();

            jobData = struct();
            jobData.(JobProp.IsSubmitted.StorageName) = true;
            jobLocation = obj.getJobLocation(jobSId);
            jobInFile = jobLocation + JobProp.Name.Extension;
            FileFormat.Mat.save(obj.FileSystem, jobInFile, jobData);
        end

        function waitForSubmitComplete(obj, jobSId, timeout)
            import parallel.internal.types.FileFormat
            import parallel.internal.types.JobProp

            obj.errorIfInvalid();

            jobLocation = obj.getJobLocation(jobSId);
            jobInFile = jobLocation + JobProp.Name.Extension;
            isSubmitted = iLoadWithRetry(FileFormat.Mat.LoadFunction, obj.FileSystem, jobInFile, JobProp.IsSubmitted);
            isSubmitted = isSubmitted{:};
            if isSubmitted
                return;
            end

            t = tic;
            while ~isSubmitted && toc(t) < timeout
                pause(1);
                isSubmitted = iLoadWithRetry(FileFormat.Mat.LoadFunction, obj.FileSystem, jobInFile, JobProp.IsSubmitted);
                isSubmitted = isSubmitted{:};
            end
            if ~isSubmitted
                error(message('parallel:cluster:JobSubmitFailedTimeout', timeout));
            end
        end
        
        function destroyJob(obj, jobSId)
            obj.errorIfInvalid();
            
            % Delete all job files
            jobLocation = obj.getJobLocation(jobSId);
            iDeleteExtensions(jobLocation);
            
            % Delete the job folder
            iDeleteFolder(jobLocation);
            
            % Remove the entry for this job from LastTaskIDMap, if present
            if obj.LastTaskIDMap.isKey(jobSId)
                obj.LastTaskIDMap.remove(jobSId);
            end
        end
        
        function destroyTasks(obj, taskSIds)
            obj.errorIfInvalid();
            
            % First group the tasks by jobSIds
            jobSIds = iGetJobSId(taskSIds);
            [uniqueJobSIds, ~, I] = iUnique(jobSIds);
            for ii = 1:numel(uniqueJobSIds)
                jobSId = uniqueJobSIds{ii};
                taskSIdsForThisJob = taskSIds(I == ii);
                
                % Find the grouped tasks and their indices
                taskRanges = obj.getTaskRangesFromDisk(jobSId);
                allowErrors = true;
                [taskInFiles, taskIndexInFiles] = obj.getTaskInFileFromRanges(jobSId, taskSIdsForThisJob, taskRanges, allowErrors);
                
                % Split any grouped task in files, as necessary
                [uniqueTaskInFiles, ~, J] = iUnique(taskInFiles);
                for jj = 1:numel(uniqueTaskInFiles)
                    taskInFile = uniqueTaskInFiles(jj);
                    
                    % Skip over task in files that could not be found
                    if taskInFile == ""
                        continue
                    end
                    
                    taskIndicesInFile = taskIndexInFiles(J == jj);
                    createSplitIndexFiles = false; % Don't bother creating the files for the tasks that we're about to delete!
                    iSplitTaskInFileByIndices(obj.FileSystem, taskInFile, taskIndicesInFile, createSplitIndexFiles);
                end
            end
            
            % Delete all task files
            for ii = 1:numel(taskSIds)
                taskSId = taskSIds{ii};
                taskLocation = obj.getTaskLocation(taskSId);
                iDeleteExtensions(taskLocation);
            end
        end
    end
    
    methods (Access = private)
        function errorIfInvalid(obj)
            if ~obj.IsValidStorageDir
                error(message('parallel:cluster:FileStorageNotWritableChangeDir', obj.StorageLocation));
            end
        end
        
        function [newJobID, newJobSId] = allocateNextJobID(obj)
            % Find the last used job ID. NB: This should always be the
            % maximum of LastJobID from the cache (to prevent NFS cache
            % problems) and the largest job ID from disk (to better support
            % multiple MATLAB clients using the same storage location).
            lastJobID = max([0, obj.getLastJobIDFromDisk(), obj.LastJobID]);
            
            newJobID = lastJobID + 1;
            newJobSId = char("Job" + newJobID);
            
            % Update the last job ID in the cache
            obj.LastJobID = newJobID;
        end

        function lastJobID = getLastJobIDFromDisk(obj)
            jobIDs = obj.getSortedJobIDsFromDisk(true); % Match any job file
            lastJobID = max([0, jobIDs]);
        end

        function jobIDs = getSortedJobIDsFromDisk(obj, anyFile)
            import parallel.internal.cluster.FileStorage
            import parallel.internal.types.JobProp
            
            if anyFile
                entries = dir([obj.StorageLocation FileStorage.FileSep 'Job*']);
            else
                entries = dir([obj.StorageLocation FileStorage.FileSep 'Job*'] + JobProp.Variant.Extension);
            end
            
            jobIDs = FileStorage.parseJobIDs({entries.name});
            jobIDs = iUnique(jobIDs); % Makes the jobIDs unique *and* sorted
        end
        
        function jobLocation = getJobLocation(obj, jobSId)
            jobLocation = [obj.StorageLocation parallel.internal.cluster.FileStorage.FileSep jobSId];
        end
        
        function [newTaskIDs, newTaskSIds] = allocateNextTaskIDs(obj, jobSId, numTasks)
            if obj.LastTaskIDMap.isKey(jobSId)
                lastTaskID = obj.LastTaskIDMap(jobSId);
            else
                lastTaskID = max([0, obj.getLastTaskIDFromDisk(jobSId)]);
                obj.LastTaskIDMap(jobSId) = lastTaskID;
            end
            
            newTaskIDs = lastTaskID + (1:numTasks);
            newTaskSIds = cellstr(jobSId + "/Task" + newTaskIDs); % Always use '/' for taskSIds
            
            % Update the last task ID in the cache
            obj.LastTaskIDMap(jobSId) = newTaskIDs(end);
        end
        
        function lastTaskID = getLastTaskIDFromDisk(obj, jobSId)
            taskIDs = obj.getSortedTaskIDsFromDisk(jobSId);
            lastTaskID = max([0, taskIDs]);
        end
        
        function taskIDs = getSortedTaskIDsFromDisk(obj, jobSId)
            taskRanges = obj.getTaskRangesFromDisk(jobSId);
            
            % Expand any grouped task range, e.g. [1 4], into [1 2 3 4]
            expandedTaskRanges = cellfun(@(range) range(1):range(end), taskRanges, 'UniformOutput', false);
            
            taskIDs = sort([expandedTaskRanges{:}]);
        end
        
        function taskRanges = getTaskRangesFromDisk(obj, jobSId)
            import parallel.internal.cluster.FileStorage
            import parallel.internal.types.TaskProp
            
            % Find all entries matching 'Task*.in.mat'
            entries = dir([obj.getJobLocation(jobSId) FileStorage.FileSep 'Task*'] + TaskProp.Name.Extension);
            
            % Extract the task ranges from the entry names
            taskRanges = FileStorage.parseTaskRanges({entries.name});
        end
        
        function groupedTask = isGroupedTask(obj, taskSId)
            % A task, say Task 3, is grouped if and only if Task3.in.mat
            % doesn't exist; a grouped Task 3 would be stored in a file
            % like Task1-4.in.mat.
            taskLocation = obj.getTaskLocation(taskSId);
            ungroupedTaskInFile = taskLocation + parallel.internal.types.TaskProp.Name.Extension;
            groupedTask = ~exist(ungroupedTaskInFile, 'file');
        end
        
        function taskLocation = getTaskLocation(obj, taskSId)
            taskLocation = obj.getLocation(taskSId);
        end
        
        function location = getLocation(obj, sId)
            import parallel.internal.cluster.FileStorage
            
            sId = strrep(sId, "/", FileStorage.FileSep); % Simply convert slash to this OS's file separator
            location = [obj.StorageLocation FileStorage.FileSep] + sId;
        end
        
        function groupedTaskLocation = getGroupedTaskLocation(obj, jobSId, taskIDRange)
            groupedTaskLocation = [obj.getJobLocation(jobSId) parallel.internal.cluster.FileStorage.FileSep] ...
                + "Task" + taskIDRange(1) + "-" + taskIDRange(end);
        end
    end
    
    % These methods only used by CJSSupport and CJSCluster
    methods
        function ID = getID(~, sId)
            ID = parallel.internal.cluster.FileStorage.parseID(sId);
        end
        
        function serializeForSubmission(obj, jobSId)
            import parallel.internal.types.JobProp
            import parallel.internal.types.TaskProp
            
            % Zip job files into Job1.zip
            jobExtensions = [JobProp.Name.Extension; JobProp.Variant.Extension];
            jobFiles = strcat(jobSId, jobExtensions);
            zipFilename = fullfile(obj.StorageLocation, jobSId + ".zip");
            zip(zipFilename, jobFiles, obj.StorageLocation);
            
            % And the task files into Task1.zip, Task2.zip, etc.
            % NB: Grouped tasks must be split apart into individual .zip
            % files in order to be processed by our submission commands
            taskRanges = obj.getTaskRangesFromDisk(jobSId);
            for ii = 1:numel(taskRanges)
                taskRange = taskRanges{ii};
                numTasksInRange = taskRange(end) - taskRange(1) + 1;
                if numTasksInRange > 1
                    taskInFile = obj.getGroupedTaskLocation(jobSId, taskRange) + TaskProp.Name.Extension;
                    taskIndicesInFile = 1:numTasksInRange;
                    iSplitTaskInFileByIndices(obj.FileSystem, taskInFile, taskIndicesInFile);
                end
            end
            
            taskIDs = obj.getSortedTaskIDsFromDisk(jobSId);
            for ii = 1:numel(taskIDs)
                taskID = taskIDs(ii);
                taskSId = char(jobSId + "/Task" + taskID); % Always use '/' for taskSIds
                zipFilename = obj.getTaskLocation(taskSId) + ".zip";
                zip(zipFilename, taskSId + TaskProp.Name.Extension, obj.StorageLocation);
            end
        end
        
        function fileName = getTaskDiaryFile(obj, taskSId)
            fileName = char(obj.getTaskLocation(taskSId) + parallel.internal.types.TaskProp.Diary.Extension);
        end
        
        function location = getRootLocation(~)
            location = '';
        end
    end
    
    % These methods only used by CJSSupport
    methods
        function info = getJobFilesInfo(obj, jobSId)
            import parallel.internal.types.JobProp
            
            info = struct();
            info.RootDirectory = obj.StorageLocation; % Name of root directory on this client
            
            jobInExtensions = [JobProp.Name.Extension; JobProp.Variant.Extension];
            jobOutExtensions = [JobProp.StartDateTime.Extension; JobProp.FinishDateTime.Extension];
            jobStateExtensions = JobProp.StateEnum.Extension;
            
            info.JobInputFiles  = cellstr(jobSId + jobInExtensions);    % Files not modified during job execution
            info.JobOutputFiles = cellstr(jobSId + jobOutExtensions);   % Files modified during job execution
            info.JobStateFiles  = cellstr(jobSId + jobStateExtensions); % State indicator file, maybe modified during job execution
            info.TaskFiles      = {jobSId};                             % Files relating to tasks (actually just the directory containing all tasks)
        end
        
        function [jobSIds, jobIDs] = getJobSIds(obj)
            jobIDs = obj.getSortedJobIDsFromDisk(false); % Only match job metadata.txt files
            jobSIds = cellstr("Job" + jobIDs);
        end
        
        function [taskSIds, taskIDs] = getTaskSIds(obj, jobSId)
            taskIDs = obj.getSortedTaskIDsFromDisk(jobSId);
            taskSIds = cellstr(jobSId + "/Task" + taskIDs); % Always use '/' for taskSIds
        end
        
        function values = getFields(obj, type, sIds, props)
            import parallel.internal.types.JobProp
            import parallel.internal.types.TaskProp
            
            if ~iscell(sIds)
                sIds = {sIds};
            end
            
            if iscell(props)
                props = [props{:}];
            end
            
            % If there are no entities or names, return an empty array
            numEntities = numel(sIds);
            numProps = numel(props);
            if numEntities == 0 || numProps == 0
                values = cell(numEntities, numProps);
                return
            end
            
            % Ensure the props are rows
            props = reshape(props, 1, numProps);
            
            % Pre-create output array
            values = cell(numEntities, numProps);

            % Order by extension - the output index is the index into the new
            % extensions array for the equivalent name
            [extensions, ~, I] = iUnique([props.Extension]);

            % Now actually get the fields and values
            for ii = 1:numel(extensions)
                extension = extensions(ii);
                
                % Which names are saved in this file
                indexToLoad = find(I == ii);
                propsToLoad = props(indexToLoad);
                
                % Special behaviour is necessary for task input files
                isTaskInExtension = extension == TaskProp.Name.Extension && strcmp(type, 'task');
                
                if isTaskInExtension
                    taskInValues = obj.getTaskInFields(sIds, propsToLoad);
                    values(:, indexToLoad) = [taskInValues{:}]';
                else
                    % Whether the extension corresponds to a file that must
                    % exist. For such an extension, an error is thrown if
                    % the file does not exist.
                    fileMustExist = extension == JobProp.Variant.Extension || extension == JobProp.Name.Extension;
                    
                    fileFormat = propsToLoad(1).FileFormat;
                    
                    filenames = obj.getLocation(sIds) + extension;
                    
                    % If there is only one entity, there is only one unique
                    % folder; don't bother extracting that one folder via
                    % fileparts only to reconstruct the filename later
                    if numEntities == 1
                        uniqueFolders = filenames;
                        J = 1;
                    else
                        folders = strings(1, numel(filenames));
                        for jj = 1:numel(filenames)
                            folders(jj) = fileparts(filenames(jj));
                        end
                        [uniqueFolders, ~, J] = iUnique(folders);
                    end

                    for jj = 1:numel(uniqueFolders)
                        uniqueFolder = uniqueFolders{jj};
                        indicesForThisFolder = find(J == jj);
                        numIndicesForThisFolder = numel(indicesForThisFolder);
                        
                        % NB: There is an important optimization here to
                        % list multiple files in the folder with the
                        % desired extension only once and try loading from
                        % only those files that exist. Without this
                        % optimization, vectorized gets on entities with
                        % default properties will be O(N), not O(1).
                        if numIndicesForThisFolder > 1
                            folderPrefix = [uniqueFolder obj.FileSep];
                            entries = dir(folderPrefix + "*" + extension);
                            existingFilesWithExtension = strcat(folderPrefix, {entries.name});
                        end
                        
                        for kk = 1:numel(indicesForThisFolder)
                            indexForThisFolder = indicesForThisFolder(kk);
                            filename = filenames(indexForThisFolder);
                            if numIndicesForThisFolder > 1
                                fileExists = ismember(filename, existingFilesWithExtension);
                            else
                                fileExists = obj.FileSystem.exist(filename, 'file');
                            end
                            if fileExists
                                try
                                    data = iLoadWithRetry(fileFormat.LoadFunction, obj.FileSystem, filename, propsToLoad);
                                catch E
                                    % Failed to load the information from the
                                    % existing file. For all files *except* a
                                    % state file, throw an error. For state
                                    % files, use 'unavailable'.
                                    if ~strcmp(extension, TaskProp.StateEnum.Extension)
                                        error(message('parallel:cluster:CouldNotOpenFile', filename, E.message));
                                    end
                                    data = {parallel.internal.types.States.Unavailable.Name};
                                end
                            else
                                % A .metadata.txt or a .in.mat file must exist,
                                % so error if it doesn't. For all other files,
                                % the lack of a file implies default property
                                % values.
                                if fileMustExist
                                    errorString = getString(message('MATLAB:load:couldNotReadFile', filename));
                                    error(message('parallel:cluster:CouldNotOpenFile', filename, errorString));
                                end
                                data = {propsToLoad.DefaultValue};
                            end

                            values(indexForThisFolder, indexToLoad) = data;
                        end
                    end
                end
            end
        end
        
        function setFields(obj, type, sIds, props, values)
            import parallel.internal.types.TaskProp
            
            if ~iscell(sIds)
                sIds = {sIds};
            end
            
            if iscell(props)
                props = [props{:}];
            end
            
            numSIds = numel(sIds);
            numProps = numel(props);
            
            % Return early if there is nothing to set
            if numSIds == 0 || numProps == 0
                return
            end
            
            numValues = numel(values);
            if numProps ~= numValues
                error(message('parallel:cluster:FileSerializerNumFieldsValuesNotEqual'));
            end
            
            % Ensure the props and values are columns
            props = reshape(props, numProps, 1);
            names = {props.StorageName};
            values = reshape(values, numValues, 1);

            % Order by common extension - the output index is the index
            % into the new extensions array for the equivalent name
            [extensions, ~, J] = iUnique([props.Extension]);

            % Now actually save the fields and values
            for ii = 1:numel(extensions)
                extension = extensions(ii);
                
                % Which props are saved in this file
                indexToSave = find(J == ii);
                propsToSave = props(indexToSave);
                valuesToSave = values(indexToSave);
                
                % Special behaviour is necessary for task input files
                isTaskInExtension = extension == TaskProp.Name.Extension && strcmp(type, 'task');
                
                if isTaskInExtension
                    obj.setTaskInFields(sIds, propsToSave, valuesToSave);
                else
                    namesToSave = names(indexToSave);
                    fileFormat = propsToSave(1).FileFormat;
                
                    % Loop over the supplied entities
                    for jj = 1:numSIds
                        sId = sIds{jj};
                        
                        % For a single sId, valuesToSave is a cell array.
                        % For multiple sIds, valuesToSave is a cell array
                        % of cell arrays.
                        if numSIds == 1
                            valuesToSaveForSId = valuesToSave;
                        else
                            valuesToSaveForSId = cellfun(@(v) v{jj}, valuesToSave, 'UniformOutput', false);
                        end
                        
                        % Construct a structure to save from
                        structToSave = cell2struct(valuesToSaveForSId, namesToSave);
                
                        % Get the complete name of this location
                        filename = obj.getLocation(sId) + extension;
                        fileFormat.save(obj.FileSystem, filename, structToSave);
                    end
                end
            end
        end
    end
    
    methods (Access = private)
        function values = getTaskInFields(obj, taskSIds, props)
            import parallel.internal.types.FileFormat
            import parallel.internal.types.TaskProp
            
            assert(iscell(taskSIds), 'Expecting taskSIds to be a cell!');
            numTasks = numel(taskSIds);
            
            % Single ungrouped task optimization
            if numTasks == 1 && ~obj.isGroupedTask(taskSIds{1})
                values = obj.getTaskInFieldsForSingleUngroupedTask(taskSIds{1}, props);
                return
            end
            
            % If InputArguments is one of the props, we must also load
            % SlicedIndices for grouped tasks to determine which input
            % arguments should be sliced and which ones shouldn't be
            names = {props.StorageName}';
            loadSlicedIndices = ismember(TaskProp.InputArguments, props);
            
            % Preallocate the return value
            values = cell(numTasks, 1);
            
            % Group the tasks by their jobID first
            jobSIds = iGetJobSId(taskSIds);
            [uniqueJobSIds, ~, I] = iUnique(jobSIds);
            
            for ii = 1:numel(uniqueJobSIds)
                jobSId = uniqueJobSIds{ii};
                taskIndicesForThisJob = find(I == ii);
                taskRanges = obj.getTaskRangesFromDisk(jobSId);
                
                % Next group the tasks by their filename
                [filenames, taskIndices, taskRangesArray] = obj.getTaskInFileFromRanges(jobSId, taskSIds(taskIndicesForThisJob), taskRanges);
                [uniqueFilenames, ~, J] = iUnique(filenames);
                
                for jj = 1:numel(uniqueFilenames)
                    filename = uniqueFilenames(jj);
                    taskRange = taskRangesArray{jj};
                    indicesForThisFilename = find(J == jj);
                    
                    singleTaskFile = taskRange(1) == taskRange(end);
                    if singleTaskFile
                        taskIndexForThisJob = taskIndicesForThisJob(indicesForThisFilename(1));
                        try
                            data = iLoadWithRetry(FileFormat.Mat.LoadFunction, obj.FileSystem, filename, props);
                        catch E
                            error(message('parallel:cluster:CouldNotOpenFile', filename, E.message));
                        end
                        values{taskIndexForThisJob} = data;
                    else
                        % Load names from the file, including
                        % 'slicedindices' if necessary
                        if loadSlicedIndices
                            allNames = [names; {TaskProp.SlicedIndices.StorageName}];
                            fileData = iLoadWithRetry(@iLoad, obj.FileSystem, filename, allNames{:});
                        else
                            fileData = iLoadWithRetry(@iLoad, obj.FileSystem, filename, names{:});
                        end

                        for kk = 1:numel(indicesForThisFilename)
                            indexForThisFilename = indicesForThisFilename(kk);
                            taskIndex = taskIndices(indexForThisFilename);
                            taskIndexForThisJob = taskIndicesForThisJob(indexForThisFilename);

                            data = fileData;
                            % Remove 'slicedindices' field if it was loaded; it
                            % shouldn't be returned as part of data
                            if loadSlicedIndices
                                slicedIndices = data.(TaskProp.SlicedIndices.StorageName);
                                data = rmfield(data, TaskProp.SlicedIndices.StorageName);
                            end

                            data = struct2cell(data);

                            % Extract the values for the task
                            for ll = 1:numel(data)
                                name = names{ll};
                                dataValue = data{ll};

                                % Extract the sliced as necessary
                                if strcmp(name, TaskProp.InputArguments.StorageName)
                                    data{ll} = iGetArgsIn(taskIndex, dataValue, slicedIndices);
                                elseif strcmp(name, TaskProp.DependentFiles.StorageName)
                                    % Do nothing; autoattachedfiles are never
                                    % sliced
                                else
                                    if numel(dataValue) > 1
                                        data{ll} = dataValue(taskIndex);
                                    end
                                end
                            end

                            values{taskIndexForThisJob} = data;
                        end
                    end
                end
            end
        end
        
        function values = getTaskInFieldsForSingleUngroupedTask(obj, taskSId, props)
            filename = obj.getTaskLocation(taskSId) + parallel.internal.types.TaskProp.Name.Extension;
            try
                data = iLoadWithRetry(parallel.internal.types.FileFormat.Mat.LoadFunction, obj.FileSystem, filename, props);
            catch E
                error(message('parallel:cluster:CouldNotOpenFile', filename, E.message));
            end
            values = {data};
        end
        
        function setTaskInFields(obj, taskSIds, props, values)
            import parallel.internal.types.FileFormat
            
            assert(iscell(taskSIds), 'Expecting taskSIds to be a cell!');
            numTasks = numel(taskSIds);
            
            % Group the tasks by their jobID first
            jobSIds = iGetJobSId(taskSIds);
            [uniqueJobSIds, ~, I] = iUnique(jobSIds);
            
            for ii = 1:numel(uniqueJobSIds)
                jobSId = uniqueJobSIds{ii};
                taskIndicesForThisJob = find(I == ii);
                taskRanges = obj.getTaskRangesFromDisk(jobSId);
                
                % Next group the tasks by their filename
                [filenames, taskIndicesInFile, taskRangesArray] = obj.getTaskInFileFromRanges(jobSId, taskSIds(taskIndicesForThisJob), taskRanges);
                [uniqueFilenames, ~, J] = iUnique(filenames);
                
                for jj = 1:numel(uniqueFilenames)
                    filename = uniqueFilenames(jj);
                    taskRange = taskRangesArray{jj};
                    indicesForThisFilename = find(J == jj);
                    
                    singleTaskFile = taskRange(1) == taskRange(end);
                    if singleTaskFile
                        taskIndexForThisJob = taskIndicesForThisJob(indicesForThisFilename(1));
                        % For a single task, values is a cell array. For
                        % multiple tasks, values is a cell array of cell
                        % arrays.
                        if numTasks == 1
                            valuesToSave = values;
                        else
                            valuesToSave = cellfun(@(v) v{taskIndexForThisJob}, values, 'UniformOutput', false);
                        end
                        
                        % Construct a structure to save to disk
                        names = {props.StorageName};
                        structToSave = cell2struct(valuesToSave, names);
                        FileFormat.Mat.save(obj.FileSystem, filename, structToSave);
                    else
                        taskIndicesInFileForThisFilename = taskIndicesInFile(indicesForThisFilename);
                        groupedTaskIndicesForThisJob = taskIndicesForThisJob(indicesForThisFilename);
                
                        % Perform grouped task property sets
                        groupedTaskSIds = taskSIds(groupedTaskIndicesForThisJob);

                        % For a single task, values is a cell array. For
                        % multiple tasks, values is a cell array of cell
                        % arrays.
                        if numTasks == 1
                            valuesToSave = values;
                        else
                            valuesToSave = cellfun(@(v) v(groupedTaskIndicesForThisJob), values, 'UniformOutput', false);
                        end
                        
                        obj.setGroupedTaskInFields(filename, taskIndicesInFileForThisFilename, groupedTaskSIds, props, valuesToSave);
                    end
                end
            end
        end
        
        function setGroupedTaskInFields(obj, taskInFile, taskIndicesInFile, taskSIds, props, values)
            import parallel.internal.types.TaskProp
            
            if any(ismember(props, [TaskProp.Function, TaskProp.InputArguments]))
                % Setting these properties, in general, prevents us from
                % using the optimized grouped task in file, so split the
                % grouped task in file and set the properties normally.
                iSplitTaskInFileByIndices(obj.FileSystem, taskInFile, taskIndicesInFile);
                obj.setTaskInFields(taskSIds, props, values);
            else
                % Otherwise we should be able to maintain the grouped task
                % in file
                names = {props.StorageName};
                data = iLoadWithRetry(@iLoad, obj.FileSystem, taskInFile, names{:});
                
                for ii = 1:numel(props)
                    prop = props(ii);
                    name = names{ii};
                    value = values{ii};
                    
                    % No task in fields can be stored as character vectors
                    % so that we can treat all task in fields uniformly as
                    % arrays using numel/paren-indexing
                    assert(~ischar(value), 'No character vectors allowed for task in data');
                    
                    % Convert values to an array, if necessary; vectorized
                    % property sets always come in as cell arrays
                    if iscell(value)
                        value = [value{:}];
                    end
                    
                    if prop == TaskProp.DependentFiles
                        % DependentFiles are never sliced, so just take all
                        % unique values. NB: For vectorized sets, the
                        % dependent files for each task should be the same,
                        % so this should be equivalent to taking the files
                        % for the first task.
                        data.(name) = unique(value);
                    else
                        oldValue = data.(name);
                        if numel(oldValue) > 1
                            oldValue(taskIndicesInFile) = value;
                            data.(name) = oldValue;
                        else
                            taskRange = iGetTaskRangeFromTaskInFile(taskInFile);
                            newValue = repmat(oldValue, 1, taskRange(end) - taskRange(1) + 1);
                            newValue(taskIndicesInFile) = value;
                            data.(name) = newValue;
                        end
                    end
                end
                
                parallel.internal.types.FileFormat.Mat.save(obj.FileSystem, taskInFile, data);
            end
        end
        
        function [taskInFiles, taskIndicesInFile, taskRangesForFile] = getTaskInFileFromRanges(obj, jobSId, taskSIds, taskRanges, allowErrors)
            assert(iscell(taskSIds), 'Expecting taskSIds to be a cell!');
            numTasks = numel(taskSIds);
            
            taskIDs = iGetTaskID(taskSIds);
            taskInFiles = strings(1, numTasks);
            taskIndicesInFile = zeros(1, numTasks);
            taskRangesForFile = cell(1, numTasks);
            extension = parallel.internal.types.TaskProp.Name.Extension;
            
            for ii = 1:numel(taskRanges)
                taskRange = taskRanges{ii};
                matches = taskRange(1) <= taskIDs & taskIDs <= taskRange(end);
                if any(matches)
                    if taskRange(1) == taskRange(end)
                        taskInFiles(matches) = obj.getTaskLocation(taskSIds(matches)) + extension;
                        taskIndicesInFile(matches) = 1;
                    else
                        taskInFiles(matches) = obj.getGroupedTaskLocation(jobSId, taskRange) + extension;
                        taskIndicesInFile(matches) = 1 + taskIDs(matches) - taskRange(1);
                    end
                    taskRangesForFile(matches) = {taskRange};
                    if all(taskIndicesInFile) % If all task indices in file have been populated
                        return
                    end
                end
            end
            
            % Log or error, as appropriate, for files that did not exist
            if nargin < 5
                allowErrors = false;
            end
            
            if allowErrors
                onError = @(taskSId) dctSchedulerMessage(1, 'Unable to find the .in.mat file for ''%s''.', taskSId);
            else
                onError = @(taskSId) error(message('parallel:cluster:CouldNotOpenFile', obj.getTaskLocation(taskSId) + extension, 'The file does not exist'));
            end
            
            missingTaskSIds = taskSIds(taskIndicesInFile == 0);
            cellfun(onError, missingTaskSIds);
        end
    end
end

function tf = iIsWritable(storageDir)
    % Check whether the directory is writable by attempting to create and
    % view a file.
    %
    % NB: Checking file attributes does not work since fileattrib() reports
    % the read/write status for the user who owns the directory, not the
    % user who owns this MATLAB process. See g2200579.
    
    % Check that we can create a new file
    file = tempname(storageDir);
    if exist(file, 'file')
        % It is very unlikely that a file with the temp name already
        % exists, but if it does then just skip this part of the check
        dctSchedulerMessage(1, 'Test file ''%s'' already exists, check for directory being writable may not be accurate.', file);
    else
        fid = fopen(file, 'a');
        if fid < 0
            tf = false;
            return
        end
        fclose(fid);
        fileCleanup = onCleanup(@() iDeleteFileNoError(file));
    end
    
    % Check that we can see the file in the directory
    entries = dir(storageDir);
    tf = any(arrayfun(@(entry) strcmp(fullfile(entry.folder, entry.name), file), entries));
end

%--------------------------------------------------------------------------
% File/folder deletion helper methods
%--------------------------------------------------------------------------
        
function iDeleteExtensions(fileLocation)
    % Disable expected warnings temporarily so that the user doesn't see them
    warningState1 = warning('off', 'MATLAB:DELETE:FileNotFound');
    cleanupObj1 = onCleanup(@() warning(warningState1));
    warningState2 = warning('off', 'MATLAB:DELETE:Permission');
    cleanupObj2 = onCleanup(@() warning(warningState2));
    warningState3 = warning('off', 'MATLAB:DELETE:DirectoryDeletion');
    cleanupObj3 = onCleanup(@() warning(warningState3));
    
    % It is possible that MATLAB workers have open file handles to these
    % files, in which case they will fail to be deleted on Windows. To
    % avoid drooling files, retry as necessary.
    attempt = 1;
    while true
        % Get the files to delete
        files = iGetFiles(fileLocation);
        
        % Try to delete all the files, returning if all were deleted
        % successfully
        if all(cellfun(@iDeleteFileNoError, files))
            return
        end
        
        % Lazily load the number of attempts
        if attempt == 1
            numAttempts = iGetNumAttempts();
        end

        % If there are attempts remaining, try again after waiting 1 second
        if attempt < numAttempts
            pause(1);
            attempt = attempt + 1;
        else
            break
        end
    end
end

function files = iGetFiles(fileLocation)
    entries = dir(fileLocation + ".*"); % Includes the job folder on Windows!
    entries = entries(~[entries.isdir]); % Remove the job folder, if present
    files = arrayfun(@(entry) fullfile(entry.folder, entry.name), entries, 'UniformOutput', false);
end

function success = iDeleteFileNoError(file, numAttempts)
    if nargin < 2
        numAttempts = 1;
    end
    
    for ii = 1:numAttempts
        try
            delete(file)
            success = true;
            return
        catch E
            dctSchedulerMessage(5, 'Unable to delete job or task file ''%s''. Original error:\n%s', file, E.getReport());
        end
        
        % Try again after waiting 1 second
        if ii < numAttempts
            pause(1);
        end
    end
    
    success = false;
end
        
function iDeleteFolder(folder)
    % Quick exit if the folder doesn't exist
    if ~exist(folder, 'dir')
        return
    end
    
    % It is possible that MATLAB workers have open file handles to files in
    % the folder, in which case the job folder will fail to be deleted on
    % Windows. To avoid drooling the folder, retry as necessary.
    numAttempts = iGetNumAttempts();
    for ii = 1:numAttempts
        dctSchedulerMessage(5, 'Attempting to delete job or task information folder. \n%s', folder);
        try
            % rmdir has different error behaviour depending on which output
            % arguments are collected; we want to force all three to be
            % generated. See g943440.
            [ok, msg, ~] = rmdir(folder, 's');
            if ~ok
                dctSchedulerMessage(5, 'Unable to delete job or task information folder. \n%s\n%s', folder, msg);
            end
        catch E
            dctSchedulerMessage(5, 'Unable to delete job or task folder ''%s''. Original error:\n%s', folder, E.getReport());
        end
        
        if ~exist(folder, 'dir')
            break
        end
        
        % Try again after waiting 1 second
        if ii < numAttempts
            pause(1);
        end
    end
end

function numAttempts = iGetNumAttempts()
    [~, undoc] = pctconfig();
    numAttempts = undoc.filestoragenumattempts;
end

function jobSId = iGetJobSId(taskSId)
    jobSId = extractBefore(taskSId, '/'); % taskSIds always use '/'
end

function taskID = iGetTaskID(taskSId)
    taskID = str2double(extractAfter(taskSId, 'Task'));
end

% Wrapper around the MATLAB unique function that is optimized for the
% one-element case
function [C, IA, IC] = iUnique(A)
    if isscalar(A)
        C = A;
        IA = 1;
        IC = 1;
    else
        [C, IA, IC] = unique(A);
    end
end

function data = iLoad(fileSystem, filename, varargin)
    data = fileSystem.load(filename, varargin{:});
end

function data = iLoadWithRetry(loadFunction, fileSystem, filename, varargin)
    attempt = 1;
    while true
        try
            data = loadFunction(fileSystem, filename, varargin{:});
            return
        catch E
            dctSchedulerMessage(4, 'Unable to load file ''%s''. Original error:\n%s', filename, E.getReport());
            
            % Lazily load the number of attempts
            if attempt == 1
                numAttempts = iGetNumAttempts();
            end
            
            % Break out of the loop if there are no attempts left
            if attempt >= numAttempts
                rethrow(E);
            end
            
            % Otherwise try again after waiting 1 second
            pause(1);
            attempt = attempt + 1;
        end
    end
end

%--------------------------------------------------------------------------
% Grouped task .in.mat helper methods
%--------------------------------------------------------------------------
        
function iSplitTaskInFileByIndices(fileSystem, taskInFile, taskIndicesInFile, createSplitIndexFiles)
    import parallel.internal.cluster.FileStorage
    import parallel.internal.types.TaskProp
    
    if nargin < 4
        createSplitIndexFiles = true;
    end

    dataOrig = iLoadWithRetry(@iLoad, fileSystem, taskInFile);
    jobLocation = char(fileparts(taskInFile)); % Ensure jobLocation is a char (for concatenation) even if taskInFile is a string
    taskRange = iGetTaskRangeFromTaskInFile(taskInFile);
    taskOffset = taskRange(1) - 1;
    indexRange = taskRange - taskOffset;
    
    % Compute the new split task ranges
    newTaskRanges = iGetSplitTaskRanges(indexRange, taskIndicesInFile, createSplitIndexFiles);

    % Create the new task files
    filesToDeleteOnError = string.empty(); % If something goes wrong, rollback
    try
        for ii = 1:numel(newTaskRanges)
            newTaskRange = newTaskRanges{ii};
            data = iGetTaskInSubData(dataOrig, newTaskRange);
            newTaskFilePrefix = iGetTaskInFilePrefix(newTaskRange + taskOffset);
            newTaskFile = [jobLocation FileStorage.FileSep] + newTaskFilePrefix + TaskProp.Name.Extension;
            save(newTaskFile, '-struct', 'data');
            filesToDeleteOnError(end + 1) = newTaskFile; %#ok<AGROW>
        end

        % Delete the original file
        delete(taskInFile);
    catch E
        maxNumAttempts = iGetNumAttempts();
        arrayfun(@(file) iDeleteFileNoError(file, maxNumAttempts), filesToDeleteOnError);
        rethrow(E);
    end
end

function taskRange = iGetTaskRangeFromTaskInFile(taskInFile)
    [~, filename] = fileparts(taskInFile);
    taskRange = parallel.internal.cluster.FileStorage.parseTaskRanges(cellstr(filename)); % Input must be a cellstr, even if filename is a string
    taskRange = taskRange{1}; % Unwrap cellstr
end

function splitTaskRanges = iGetSplitTaskRanges(indexRange, indicesToSplit, keepIndicesToSplit)
    % Handle the trivial cases separately
    if isempty(indicesToSplit)
        splitTaskRanges = {indexRange};
        return
    end
    if indexRange(1) == indexRange(end)
        if keepIndicesToSplit
            splitTaskRanges = {indexRange};
        else
            splitTaskRanges = {};
        end
        return
    end
    
    % Ensure that the indices to split are unique and sorted
    indicesToSplit = iUnique(indicesToSplit);

    % Compute all pairs of ranges, including invalid ones. For example, for
    % indexRange = [1 10] and indicesToSplit = [1 4 9], create
    %     splitTaskRanges = {[1,0], 1, [2,3], 4, [5,8], 9, [10,10]}
    % if keepIndicesToSplit is true and
    %     splitTaskRanges = {[1,0], [2,3], [5,8], [10,10]}
    % if keepIndicesToSplit is false.
    %
    % NB: Invalid ranges will be removed later to leave the (valid) split
    % ranges 1, 2-3, 4, 5-8, 9, and 10-10 (keepIndicesToSplit == true) or
    % 2-3, 5-8, 10-10 (keepIndicesToSplit == false).
    if keepIndicesToSplit
        splitTaskRanges = cell(1, numel(indicesToSplit) * 2 + 1);
    else
        splitTaskRanges = cell(1, numel(indicesToSplit) + 1);
    end
    endIndex = 1;
    
    lowIndex = indexRange(1);
    for ii = 1:numel(indicesToSplit)
        indexToSplit = indicesToSplit(ii);
        
        splitTaskRanges{endIndex} = [lowIndex, indexToSplit - 1];
        endIndex = endIndex + 1;
        
        if keepIndicesToSplit
            splitTaskRanges{endIndex} = indexToSplit;
            endIndex = endIndex + 1;
        end
        lowIndex = indexToSplit + 1;
    end
    splitTaskRanges{endIndex} = [lowIndex, indexRange(end)];
    
    % Remove all invalid entries. For the example above,
    %     splitTaskRanges = {1, [2,3], 4, [5,8], 9, [10,10]}
    % if keepIndicesSplit is true and
    %     splitTaskRanges = {[2,3], [5,8], [10,10]}
    % if keepIndicesToSplit is false.
    invalidIndices = cellfun(@(range) range(1) > range(end), splitTaskRanges);
    splitTaskRanges(invalidIndices) = [];
end

function subData = iGetTaskInSubData(data, indexRange)
% Given task.in.mat data for a group of tasks, return the corresponding
% data for the subset of tasks in the given index range
    import parallel.internal.types.TaskProp
    
    names = fieldnames(data);
    for ii = 1:numel(names)
        name = names{ii};
        
        switch name
            case TaskProp.InputArguments.StorageName
                % Special extraction of input arguments for the tasks that
                % will remain based on the sliced indices
                subData.(name) = iGetArgsIn(indexRange, data.(name), data.(TaskProp.SlicedIndices.StorageName));
            case TaskProp.SlicedIndices.StorageName
                % Copy over sliced indices since these are the same for all
                % tasks, including the remaining ones (except when there is
                % only one task left, in which case we can drop the sliced
                % indices)
                if indexRange(1) ~= indexRange(end)
                    subData.(name) = data.(name);
                end
            case TaskProp.DependentFiles.StorageName
                % Copy over auto-attached files since these are the same
                % for all tasks, including the remaining ones
                subData.(name) = data.(name);
            otherwise
                % All remaining properties are standard arrays; extract
                % only the property values for the tasks that remain
                subData.(name) = iGetTaskInSubValue(data.(name), indexRange);
        end
    end
end

function argsin = iGetArgsIn(taskIndexRange, allArgsIn, slicedIndices)
% For grouped tasks, retrieve the input arguments for the given task range
% from the grouped input arguments for all tasks in the group. The returned
% argsin is in the grouped task form, which happens to reduce to the single
% task form when there is only one task.

    numArgsIn = numel(allArgsIn);
    argsin = cell(0, numArgsIn);
    isSliced = ismember(1:numArgsIn, slicedIndices);
    
    % Set all broadcast input arguments
    argsin(~isSliced) = allArgsIn(~isSliced);
    
    % Set all sliced input arguments
    low = taskIndexRange(1);
    high = taskIndexRange(end);
    for ii = find(isSliced)
        slicedValues = allArgsIn{ii};
        
        % NB: When taskIndexRange corresponds to only one task, index the
        % sliced input arguments using {} for cells and () for arrays
        if iscell(slicedValues)
            if low == high
                argsin{ii} = slicedValues{low}; % Not a cell array
            else
                argsin{ii} = slicedValues(low:high); % Still a cell array
            end
        else
            argsin{ii} = slicedValues(low:high);
        end
    end
end

function subValue = iGetTaskInSubValue(value, indexRange)
% Given an array of values, returns the subset of values corresponding to
% the given index range
    if numel(value) > 1
        subValue = value(indexRange(1):indexRange(end));
    else
        subValue = value;
    end
end

function taskInFilePrefix = iGetTaskInFilePrefix(taskRange)
    if taskRange(1) == taskRange(end)
        taskInFilePrefix = "Task" + taskRange(1);
    else
        taskInFilePrefix = "Task" + taskRange(1) + "-" + taskRange(end);
    end
end
