% MJSSupportWrapper - Holds an MJSSupportv2 for invoking methods on the job
% manager, caches MCOS worker/job/task objects, and handles job/task event
% notification

% Copyright 2020-2021 The MathWorks, Inc.

classdef (Hidden, Sealed) MJSSupportWrapper < handle
    
    properties (GetAccess = private, SetAccess = immutable)
        % Used for invoking methods on the job manager
        MJSAccess
        
        % The job manager name, used as root key into Cache
        CacheName
        
        % Used for firing events on creating and modifying jobs
        JobStorageEventManager = parallel.internal.cluster.currentJobStorageEventManager()
    end
    
    properties (Transient, GetAccess = private, SetAccess = immutable)
        % Used for storing job/task/worker objects
        Cache
    end

    methods
        function obj = MJSSupportWrapper(mjsAccess, cache)
            obj.MJSAccess = mjsAccess;
            obj.CacheName = obj.getClusterProperties('Name');
            obj.Cache = cache;
        end
    end
    
    methods (Access = {?parallel.cluster.MJS, ...
            ?parallel.internal.cloud.CloudSupport})
        function invalidateAllChildren(obj)
            allChildren = obj.Cache.getAllValues();
            cellfun(@hInvalidate, allChildren);
            obj.Cache.clear();
        end
    end
    
    % ClusterSupport methods
    methods
        function values = getClusterProperties(obj, propNames)
            getFcn = @(propNames) obj.MJSAccess.getClusterProperties(propNames);
            values = iWrapGets(getFcn, propNames);
        end
        
        function setClusterProperties(obj, propNames, values)
            setFcn = @(propNames, values) obj.MJSAccess.setClusterProperties(propNames, values);
            iWrapSets(setFcn, propNames, values);
        end
        
        function pauseQueue(obj)
            obj.MJSAccess.pauseQueue();
        end
        
        function resumeQueue(obj)
            obj.MJSAccess.resumeQueue();
        end
        
        function logout(obj, varargin)
            obj.MJSAccess.logout(varargin{:});
        end
        
        function changePassword(obj, varargin)
            obj.MJSAccess.changePassword(varargin{:});
        end
        
        function promoteJob(obj, jobSID)
            obj.MJSAccess.promoteJob(jobSID);
        end
        
        function demoteJob(obj, jobSID)
            obj.MJSAccess.demoteJob(jobSID);
        end
        
        function getClusterLogs(obj, saveLocation)
            import parallel.internal.apishared.FilenameUtils
            
            % Use a temporary hidden file to prevent it appearing in the
            % desktop browser before it has been fully written
            [saveDir, filename, ext] = fileparts(saveLocation);
            saveDir = FilenameUtils.getAbsolutePath(saveDir);
            if ~exist(saveDir, 'dir')
                error(message('parallel:cluster:ClusterLogDirNotExist', saveDir));
            end
            if ~FilenameUtils.isWritable(saveDir)
                error(message('parallel:cluster:ClusterLogDirNotWritable', saveDir));
            end
            tempName = ['.' filename ext '.part'];
            tempSaveLocation = fullfile(saveDir, tempName);
            
            % The job manager zips its logs and writes to the given location
            obj.MJSAccess.getClusterLogs(tempSaveLocation);
            if exist(tempSaveLocation, 'file')
                movefile(tempSaveLocation, saveLocation);
            end
        end
        
        % Gets workers of the given type, which must be 'BusyWorkers',
        % 'IdleWorkers', or 'Workers'
        function workers = getWorkers(obj, mjs, type)
            validatestring(type, {'BusyWorkers', 'IdleWorkers', 'Workers'});
            
            workerStructs = obj.getClusterProperties(type);
            workers = obj.getWorkersFromCacheOrCreate(mjs, workerStructs);
        end
        
        function operatingSystem = getOperatingSystem(obj)
            operatingSystem = obj.getClusterProperties('ClusterOperatingSystem');
        end
        
        function value = requiresMathWorksHostedLicensing(obj)
            value = obj.getClusterProperties('RequiresOnlineLicensing');
        end
        
        function job = getJobFromStruct(obj, mjs, jobStruct)
            job = obj.getJobsFromCacheOrCreate(mjs, jobStruct);
        end
        
        function jobs = getJobs(obj, mjs)
            jobStructs = obj.getClusterProperties('Jobs');
            jobs = obj.getJobsFromCacheOrCreate(mjs, jobStructs);
        end
        
        function varargout = getJobsByState(obj, mjs)
            import parallel.internal.cluster.types.WorkUnitState
            
            % Get jobs in all states
            states = [WorkUnitState.Pending, WorkUnitState.Queued, WorkUnitState.Running, WorkUnitState.Finished];
            jobsByState = obj.MJSAccess.getJobsByState(states);
            
            % Convert job structs to jobs
            jobsCell = cellfun(@(jobStructs) num2cell(obj.getJobsFromCacheOrCreate(mjs, jobStructs)), jobsByState, 'UniformOutput', false);
            varargout = jobsCell(1:nargout);
        end
        
        % Return the number of jobs in Pending, Queued, Running and Finished states
        function [p, q, r, f] = enumerateJobsInStates(obj)
            import parallel.internal.cluster.types.WorkUnitState
            
            states = [WorkUnitState.Pending, WorkUnitState.Queued, WorkUnitState.Running, WorkUnitState.Finished];
            jobsByState = obj.MJSAccess.getJobsByState(states);
            if isempty(jobsByState)
                [p, q, r, f] = deal(0, 0, 0, 0);
            else
                p = numel(jobsByState{1});
                q = numel(jobsByState{2});
                r = numel(jobsByState{3});
                f = numel(jobsByState{4});
            end
        end
        
        % Create and return a new job
        function job = buildJob(obj, mjs, variant, entitlement)
            import parallel.internal.types.Variant
            
            if ~obj.getClusterProperties('RequiresOnlineLicensing')
                % Only send entitlement if required by remote end
                entitlement = [];
            end
            
            switch variant
                case Variant.IndependentJob
                    jobStruct = obj.MJSAccess.createJob(entitlement);
                    job = parallel.job.MJSIndependentJob(mjs, jobStruct.num, jobStruct.jobID, obj);
                case Variant.CommunicatingSPMDJob
                    jobStruct = obj.MJSAccess.createCommunicatingSPMDJob(entitlement);
                    job = parallel.job.MJSCommunicatingJob(mjs, jobStruct.num, variant, jobStruct.jobID, obj);
                case Variant.CommunicatingPoolJob
                    jobStruct = obj.MJSAccess.createCommunicatingPoolJob(entitlement);
                    job = parallel.job.MJSCommunicatingJob(mjs, jobStruct.num, variant, jobStruct.jobID, obj);
                case Variant.ConcurrentJob
                    jobStruct = obj.MJSAccess.createConcurrentJob(entitlement);
                    job = parallel.job.MJSConcurrentJob(mjs, jobStruct.num, jobStruct.jobID, obj);
                otherwise
                    assert(false, "Variant '" + jobVariant.Name + "' is not a job type");
            end
            
            % Put the job in the cache
            cacheName = iGetJobCacheNames(jobStruct.jobID);
            obj.Cache.add(obj.CacheName, cacheName, job);
            
            % Fire event for building job
            obj.JobStorageEventManager.notifyCreateJob(job);
        end
    end
    
    methods
        function [connMgr, strategy] = buildConnectionManager(~) %#ok<STOUT> Function always asserts
            % PCT_Java_Transition_TODO (g2421835): Implement SPF connection manager or remove this
            assert(false, 'Connection manager for SPF PLR does not exist');
        end
    end
    
    methods (Access = private)
        function workers = getWorkersFromCacheOrCreate(obj, mjs, workerStructs)
            if isempty(workerStructs)
                workers = parallel.cluster.MJSWorker.empty();
            else
                workerCacheNames = iGetWorkerCacheNames(workerStructs);
                createFcn = @(ii) parallel.cluster.MJSWorker(mjs, ...
                    workerStructs(ii).hostName, ...
                    workerStructs(ii).computerMLType, ...
                    workerStructs(ii).allHostAddresses, ...
                    workerStructs(ii).name);
                parentNameFcn = @(~) obj.CacheName;
                workers = obj.getFromCacheOrCreate(workerCacheNames, createFcn, parentNameFcn);
                workers = reshape(workers, numel(workers), 1);
            end
        end
        
        function jobs = getJobsFromCacheOrCreate(obj, mjs, jobStructs)
            if isempty(jobStructs)
                jobs = parallel.job.MJSIndependentJob.empty();
            else
                jobSIDs = [jobStructs.jobID];
                jobCacheNames = iGetJobCacheNames(jobSIDs);
                createFcn = @(ii) obj.createJobFromStruct(mjs, jobStructs(ii));
                parentNameFcn = @(~) obj.CacheName;
                jobs = obj.getFromCacheOrCreate(jobCacheNames, createFcn, parentNameFcn);
            end
        end
        
        function job = createJobFromStruct(obj, mjs, jobStruct)
            import parallel.internal.types.Variant
            
            jobSID = jobStruct.jobID;
            jobID = jobStruct.num;
            jobMLType = jobStruct.mlType;
            variant = jobMLType.Variant;
            
            switch variant
                case Variant.IndependentJob
                    job = parallel.job.MJSIndependentJob(mjs, jobID, jobSID, obj);
                case Variant.CommunicatingSPMDJob
                    job = parallel.job.MJSCommunicatingJob(mjs, jobID, variant, jobSID, obj);
                case Variant.CommunicatingPoolJob
                    job = parallel.job.MJSCommunicatingJob(mjs, jobID, variant, jobSID, obj);
                case Variant.ConcurrentJob
                    job = parallel.job.MJSConcurrentJob(mjs, jobID, jobSID, obj);
                otherwise
                    assert(false, "Variant '" + jobVariant.Name + "' is not a job type");
            end
        end
        
        function objects = getFromCacheOrCreate(obj, cacheNames, createFcn, parentNameFcn)
            numObjects = numel(cacheNames);

            % Quick exit if there are no objects to retrieve
            if numObjects == 0
                objects = [];
                return
            end
            
            % Retrieve as many objects from the cache as possible
            [found, foundObjects] = obj.Cache.retrieveArrayIfPresent(cacheNames);
            
            if all(found)
                % Optimization when all objects are found
                objects = foundObjects;
            else
                % Otherwise build a partially-filled cell array of the
                % objects that were found, populate the missing entries
                % with objects created using the given createFcn, and cache
                % the newly-created objects using the given parentNameFcn
                objects = cell(1, numObjects);
                objects(found) = foundObjects;
                for missingIndex = find(~found)
                    objects{missingIndex} = createFcn(missingIndex);
                    obj.Cache.add(parentNameFcn(missingIndex), cacheNames{missingIndex}, objects{missingIndex});
                end
            end
            
            % Convert to array
            objects = [objects{:}];
        end
    end

    methods (Hidden)
        function hErrorIfNumPoolWorkersInvalid(obj, numWorkers, clusterNumWorkers)
            % Accessing element 1 in case numWorkers has 2 elements
            minWorkers = numWorkers(1);
            if minWorkers > clusterNumWorkers
                clusterName = obj.getClusterProperties('Name');
                if isempty(clusterName)
                    error(message('parallel:cluster:mjs:NumWorkersExceeded', minWorkers, clusterNumWorkers));
                else
                    error(message('parallel:cluster:mjs:ClusterNumWorkersExceeded', clusterName, minWorkers, clusterNumWorkers));
                end
            end
        end
        
        function provider = hCreateJobStatusProvider(obj, cluster) %#ok<INUSD,STOUT>
            % PCT_Java_Transition_TODO (g2194224): Implement job monitor functionality
            error('Not yet implemented');
        end
        
        function jobIdentifier = hCreateJobUpdateIdentifier(~, jobSID, job) %#ok<INUSD,STOUT>
            % PCT_Java_Transition_TODO (g2194224): Implement job monitor functionality
            error('Not yet implemented');
        end
        
        % The following methods are public for testing purposes
        function hCloseSession(obj)
            obj.MJSAccess.hCloseSession();
        end

        function hCloseClientSession(obj)
            obj.MJSAccess.hCloseClientSession();
        end

        function id = hGetClientSessionID(obj)
            id = obj.MJSAccess.hGetClientSessionID();
        end
    end
    
    %% Test methods
    methods (Hidden)
        function prepareJobForSubmission(obj, job, jobSID)
            mjsPrepareForSubmission2(job, obj.MJSAccess, jobSID);
        end
        
        function hAddRemoteInvocationListener(obj, listener)
            obj.MJSAccess.hAddRemoteInvocationListener(listener);
        end
        
        function hRemoveRemoteInvocationListener(~, ~)
            % noop
        end
    end

    % CallbackSupport methods
    methods
        function registerForJobEvents(obj, mjs, jobSID)
            obj.MJSAccess.addEventListener(jobSID);
            obj.stashCluster(mjs);
        end
        
        function unregisterForJobEvents(obj, mjs, jobSID)
            obj.MJSAccess.removeEventListener(jobSID);
            obj.unstashCluster(mjs);
        end
        
        function registerForTaskEvents(obj, mjs, taskSID)
            obj.MJSAccess.addEventListener(taskSID);
            obj.stashCluster(mjs);
        end
        
        function unregisterForTaskEvents(obj, mjs, taskSID)
            obj.MJSAccess.removeEventListener(taskSID);
            obj.unstashCluster(mjs);
        end
        
        function registerStateListeners(obj, job)
            job.setSupportQueuedCallbackFcn(@obj.handleJobQueued);
            job.setSupportRunningCallbackFcn(@obj.handleJobRunning);
            job.setSupportFinishedCallbackFcn(@obj.handleJobFinished);
        end
        
        function stashCluster(~, mjs)
            iCallbackStash('stash', mjs);
        end
        
        function unstashCluster(~, mjs)
            iCallbackStash('unstash', mjs);
        end
    end
    
    methods (Access = private)
        function handleJobQueued(obj, job, ~)
            obj.JobStorageEventManager.notifyJobQueued(job);
        end
        
        function handleJobRunning(obj, job, ~)
            obj.JobStorageEventManager.notifyJobRunning(job);
        end
        
        function handleJobFinished(obj, job, ~)
            obj.JobStorageEventManager.notifyJobFinished(job);
        end
    end
    
    % JobSupport methods
    methods
        function values = getJobProperties(obj, jobSIDs, propNames)
            getFcn = @(propNames) obj.MJSAccess.getJobProperties(jobSIDs, propNames);
            values = iWrapGets(getFcn, propNames, numel(jobSIDs));
        end
        
        function setJobProperties(obj, jobSIDs, propNames, values)
            setFcn = @(propNames, values) obj.MJSAccess.setJobProperties(jobSIDs, propNames, values);
            iWrapSets(setFcn, propNames, values, numel(jobSIDs))
        end
        
        function numTasksPerJob = getTaskCount(obj, jobSIDs)
            taskStructsPerJob = obj.getJobProperties(jobSIDs, {'Tasks'});
            numTasksPerJob = cellfun(@numel, taskStructsPerJob);
        end
        
        function tasks = getTasks(obj, jobs, jobSIDs)
            numJobs = numel(jobs);
            taskStructsPerJob = obj.getJobProperties(jobSIDs, {'Tasks'});
            numTasksPerJob = cellfun(@numel, taskStructsPerJob);
            
            if all(numTasksPerJob == 0)
                tasks = repmat({parallel.task.MJSTask.empty()}, 1, numJobs);
            else
                % Create a map of task index to job index
                jobIndex = arrayfun(@(ii) repmat(ii, 1, numTasksPerJob(ii)), 1:numJobs, 'UniformOutput', false);
                jobIndex = [jobIndex{:}];

                % Get the tasks from the cache or create them
                taskStructs = taskStructsPerJob(numTasksPerJob > 0); % Only keep non-empty structs
                taskStructs = vertcat(taskStructs{:});
                taskSIDs = [taskStructs.taskID];
                taskCacheNames = iGetTaskCacheNames(taskSIDs);
                createFcn = @(ii) obj.createTaskFromStruct(jobs(jobIndex(ii)), jobSIDs(jobIndex(ii)), taskStructs(ii));
                parentNameFcn = @(ii) iGetJobCacheNames(jobSIDs(jobIndex(ii)));
                tasks = obj.getFromCacheOrCreate(taskCacheNames, createFcn, parentNameFcn);
                
                % Group the tasks by their original job
                tasks = arrayfun(@(ii) tasks(ii == jobIndex), 1:numel(jobs), 'UniformOutput', false);
            end
            
            % Maintain backwards compatibility with non-vectorized case
            if numel(tasks) <= 1
                tasks = [tasks{:}];
            end
        end
        
        function task = getTaskFromStruct(obj, job, jobStruct, taskStruct)
            task = obj.getTasksFromCacheOrCreate(job, jobStruct.jobID, taskStruct);
        end

        function [taskIDs, taskSIDs] = buildTasksVec(obj, job, jobSID, tasksVecInfo)
            numTasks = tasksVecInfo.NumTasks;
            
            % Extract properties from tasksVecInfo
            taskFcns = repmat({tasksVecInfo.Function}, 1, numTasks);
            numArgsOut = repmat(tasksVecInfo.NumOutputArguments, 1, numTasks);
            taskNames = repmat({''}, 1, numTasks); % Use default names
            timeouts = repmat(intmax('int64'), 1, numTasks);
            captureDiary = repmat(tasksVecInfo.CaptureDiary, 1, numTasks);
            maxNumRetries = ones(1, numTasks);
            funcNames = repmat({char(tasksVecInfo.Function)}, 1, numTasks);
            containsCallbacks = false(1, numTasks);
            
            % Expand input arguments using sliced indices
            argsInFull = cell(1, numTasks);
            slicedIndices = tasksVecInfo.SlicedIndices;
            numSlicedIndices = numel(slicedIndices);
            for ii = 1:numTasks
                argsIn = tasksVecInfo.InputArguments;
                for jj = 1:numSlicedIndices
                    slicedIndex = slicedIndices(jj);
                    argInFull = argsIn{slicedIndex};
                    if iscell(argInFull)
                        argsIn{slicedIndex} = argInFull{ii};
                    else
                        argsIn{slicedIndex} = argInFull(ii);
                    end
                end
                argsInFull{ii} = argsIn;
            end
            
            % Create the tasks
            jobSIDs = repmat(jobSID, 1, numTasks);
            taskStructs = obj.MJSAccess.createTask(jobSIDs, taskFcns, numArgsOut, argsInFull, taskNames, timeouts, captureDiary, maxNumRetries, funcNames, containsCallbacks);
            taskSIDs = [taskStructs.taskID];
            taskIDs = [taskStructs.num];
            
            % Fire event for building tasks
            obj.JobStorageEventManager.notifyCreateTasks(job);
        end
        
        function tasks = buildTasks(obj, job, jobSID, ~, taskInfoStructs)
            taskFcns = {taskInfoStructs.Function};
            numArgsOut = [taskInfoStructs.NumOutputArguments];
            argsIn = {taskInfoStructs.InputArguments};
            taskNames = {taskInfoStructs.Name};
            timeouts = [taskInfoStructs.Timeout];
            captureDiary = [taskInfoStructs.CaptureDiary];
            maximumRetries = [taskInfoStructs.MaximumRetries];
            containsCallbacks = [taskInfoStructs.ContainsCallbacks];
            funcNames = cellfun(@char, taskFcns, 'UniformOutput', false);
        
            jobSIDs = repmat(jobSID, 1, numel(taskInfoStructs));
            taskStructs = obj.MJSAccess.createTask(jobSIDs, taskFcns, numArgsOut, argsIn, taskNames, timeouts, captureDiary, maximumRetries, funcNames, containsCallbacks);
            tasks = arrayfun(@(taskStruct) obj.createTaskFromStruct(job, jobSID, taskStruct), taskStructs);
            
            % Put the tasks in the cache
            jobCacheName = iGetJobCacheNames(jobSID);
            taskSIDs = [taskStructs.taskID];
            taskCacheNames = iGetTaskCacheNames(taskSIDs);
            arrayfun(@(task, taskCacheName) obj.Cache.add(jobCacheName, taskCacheName, task), tasks, taskCacheNames);

            % Fire event for building tasks
            obj.JobStorageEventManager.notifyCreateTasks(job);
        end
        
        function cancelJob(obj, jobSID, cancelException)
            obj.MJSAccess.cancelJobs(jobSID, {cancelException.message});
        end
        
        function notifyCancelJob(obj, job)
            obj.JobStorageEventManager.notifyCancelJob(job);
        end
        
        function destroyJob(obj, jobSIDs)
            taskStructsPerJob = obj.getJobProperties(jobSIDs, {'Tasks'});
            taskStructs = taskStructsPerJob(~cellfun(@isempty, taskStructsPerJob));
            taskStructs = vertcat(taskStructs{:});

            % Invalidate any cached task objects
            if ~isempty(taskStructs)
                taskSIDs = [taskStructs.taskID];
                taskCacheNames = iGetTaskCacheNames(taskSIDs);
                [~, oldTasks] = obj.Cache.retrieveArrayIfPresent(taskCacheNames);
                cellfun(@parallel.internal.cluster.invalidateTask, oldTasks);
            end
            
            % Destroy and remove job itself
            obj.MJSAccess.destroyJobs(jobSIDs);
            obj.Cache.remove(obj.CacheName, iGetJobCacheNames(jobSIDs));
        end
 
        function notifyDeleteJob(obj, job)
            obj.JobStorageEventManager.notifyDeleteJob(job);
        end
        
        function submitJob(obj, job, jobSID)
            mjsPrepareForSubmission2(job, obj, jobSID);
            obj.MJSAccess.submitJobs(jobSID);
            
            % Fire event for submitting job
            obj.JobStorageEventManager.notifySubmitJob(job);
        end
        
        function statusData = getJobStatusData(obj, jobSIDs)
            future = obj.MJSAccess.getJobStatusDataAsync(jobSIDs);
            outputs = future.fetchOutputs();
            statusData = [outputs{1}{:}];
        end
    end
    
    methods (Access = private)
        function tasks = getTasksFromCacheOrCreate(obj, jobs, jobSIDs, taskStructs)
            if isempty(taskStructs)
                tasks = parallel.job.MJSTask.empty();
            else
                taskSIDs = [taskStructs.taskID];
                taskCacheNames = iGetTaskCacheNames(taskSIDs);
                createFcn = @(ii) obj.createTaskFromStruct(jobs(ii), jobSIDs(ii), taskStructs(ii));
                parentNameFcn = @(ii) iGetJobCacheNames(jobSIDs(ii));
                tasks = obj.getFromCacheOrCreate(taskCacheNames, createFcn, parentNameFcn);
            end
        end
        
        function task = createTaskFromStruct(obj, job, jobSID, taskStruct)
            task = parallel.task.MJSTask(job, taskStruct.num, taskStruct.taskID, obj, jobSID);
        end
    end
    
    % TaskSupport methods
    methods
        function values = getTaskProperties(obj, taskSIDs, propNames)
            getFcn = @(propNames) obj.MJSAccess.getTaskProperties(taskSIDs, propNames);
            values = iWrapGets(getFcn, propNames, numel(taskSIDs));
        end
        
        function setTaskProperties(obj, taskSIDs, propNames, values)
            setFcn = @(propNames, values) obj.MJSAccess.setTaskProperties(taskSIDs, propNames, values);
            iWrapSets(setFcn, propNames, values, numel(taskSIDs))
        end
        
        function worker = getWorkerForTask(obj, mjs, taskSID)
            workerStruct = obj.getTaskProperties(taskSID, 'WorkerProperties');
            worker = obj.getWorkersFromCacheOrCreate(mjs, workerStruct);
        end
        
        function destroyOneTask(obj, jobSID, taskSID)
            obj.MJSAccess.destroyTasks(taskSID);
            obj.Cache.remove(iGetJobCacheNames(jobSID), iGetTaskCacheNames(taskSID));
        end
        
        function notifyDeleteTask(obj, job)
            obj.JobStorageEventManager.notifyDeleteTask(job);
        end
        
        function cancelOneTask(obj, taskSID, cancelException)
            obj.MJSAccess.cancelTasks(taskSID, {cancelException.message});
        end

        function notifyCancelTask(obj, job)
            obj.JobStorageEventManager.notifyCancelTask(job);
        end
        
        function submitTaskResult(obj, taskID, attempt, result, error, warnings, diary)
            obj.MJSAccess.submitTaskResult(taskID, attempt, result, error, warnings, diary);
        end
    end
    
    % Cloud support methods
    methods
        function updated = updateMJSAccess(obj)
            updated = obj.MJSAccess.refreshClient();
        end
        
        function [ok, msg] = checkTwoWayCommunications(obj, forceCheck)
            ok = true;
            msg = '';
            if forceCheck
                [ok, msg] = obj.MJSAccess.checkTwoWayCommunications();
            end
        end
    end
end

% Wrapper to vectorize single property gets and return values without a
% cell array when appropriate
function values = iWrapGets(getFcn, propNames, numObjects)
    if nargin < 3
        numObjects = 1;
    end
    
    % Convert propNames to a cell array, if necessary
    noCell = ~iscell(propNames);
    if noCell
        propNames = {propNames};
    end
    
    % Retrieve values as an NxM cell array where
    %   N = number of objects
    %   M = number of properties
    values = getFcn(propNames);
    values = reshape(iFlattenCell(values), numObjects, numel(propNames));

    % Unwrap a single value if propNames was specified without a cell array
    if noCell && numel(values) == 1
        values = values{1};
    end
end

% Wrapper to vectorize single property sets that are specified without
% using a cell array
function iWrapSets(setFcn, propNames, values, numObjects)
    if nargin < 4
        numObjects = 1;
    end
    
    % Convert propNames and values to cell arrays, if necessary
    if iscell(propNames)
        assert(iscell(values), 'Property values must be specified as a cell array when property names are');
    else
        propNames = {propNames};
        values = {values}; % Convert values to a cell array too
    end
    
    numProps = numel(propNames);
    numValues = numel(values);
    assert(numProps == numValues, 'Number of property names must match number of values');

    % Wrap values in an extra cell array for setFcn, if necessary
    if numObjects == 1
        values = cellfun(@(x) {x}, values, 'UniformOutput', false);
    end

    setFcn(propNames, values);
end

function cacheName = iGetWorkerCacheNames(workerStructs)
    cacheName = iGetUUIDStrs([workerStructs.id]);
end

function cacheName = iGetJobCacheNames(jobSIDs)
    cacheName = iGetUUIDStrs(jobSIDs);
end

function cacheName = iGetTaskCacheNames(taskSIDs)
    cacheName = iGetUUIDStrs(taskSIDs);
end

function uuidStr = iGetUUIDStrs(uuid)
    uuidStr = string([uuid.long1]) + string([uuid.long2]);
end

% Flattens a cell array (i.e. [val{:}]) but ensures an empty cell array
% is flattened to an empty cell array, not an empty double array []
function val = iFlattenCell(val)
    val = [val{:}];
    if isempty(val)
        val = {};
    end
end

% --------------------------------------------------------------------------
% iCallbackStash - stash/unstash clusters that have active callbacks
function iCallbackStash(action, cluster)
assert(isa(cluster, 'parallel.Cluster'), ...
    sprintf('Expected a parallel.Cluster but got a %s', class(cluster)));

mlock;
persistent STASHED_CLUSTERS
if isempty(STASHED_CLUSTERS)
    STASHED_CLUSTERS = parallel.internal.cluster.CountedStash();
end
switch action
    case 'stash'
        c = STASHED_CLUSTERS.stash(cluster); %#ok<NASGU>
    case 'unstash'
        c = STASHED_CLUSTERS.unstash(cluster); %#ok<NASGU>
end
end
