% MJSSupport - holds access proxies, invokes methods on them; caches MCOS
% job/task objects.

% Copyright 2011-2021 The MathWorks, Inc.

classdef (Hidden, Sealed) MJSSupport < handle

    properties ( Access = private )
        CachedName % The Name of the jobmanager, used as root key into Cache
    end
    
    properties ( Transient, Access = private )
        Cache      % TwoLevelCache for caching Job/Task objects and Workers
        MJSAccess   % Access for MJS properties/methods
        JobAccess  % Access for independent job properties/methods
        SJobAccess % Access for communicating & concurrent job properties/methods
        TaskAccess % Access for Task properties / methods
        MJSAccessFactory
        MJSAccessCleanup
    end

    properties ( Access = private )
        % Local instance of JobStorageEventManager for firing events on creating and
        % modifying jobs.
        JobStorageEventManager
    end
    
    methods( Access = public )
        function obj = MJSSupport( cache, accessFactory )
            obj.Cache = cache;
            obj.MJSAccessFactory = accessFactory;
            obj.initAccess();
            obj.JobStorageEventManager = parallel.internal.cluster.currentJobStorageEventManager();
        end
    end

    methods ( Access = {?parallel.cluster.MJS, ...
            ?parallel.internal.cloud.CloudSupport})
        function invalidateAllChildren( obj )
            valsc = obj.Cache.getAllValues();
            cellfun( @hInvalidate, valsc );
            obj.Cache.clear();
        end
    end
    
    methods ( Access = private )
        function initAccess( obj )
            [ mjsAccess, accessCleanup ] = obj.MJSAccessFactory.getOrCreate();
            if mjsAccess.equals(obj.MJSAccess)
                %MJSAccess is already initialized therefore we return early
                return;
            end            
            [ jobAccess, sJobAccess, taskAccess ] = iGetWritableJobAndTaskAccess( mjsAccess );
            obj.initAccessCommon( mjsAccess, jobAccess, sJobAccess, taskAccess, accessCleanup );
        end
        
        function initAccessCommon( obj, mjsAccess, jobAccess, sjobAccess, taskAccess, accessCleanup )
            % Dispose previous MJSAccess
            obj.disposeAccess();
            
            % Now old MJSAccess has been disposed and its associated
            % cleanup functions have been triggered we can now store the
            % new MJSAccess and its cleanup function
            obj.MJSAccess   = mjsAccess;
            obj.MJSAccessCleanup = accessCleanup;
            
            % Now we have everything OK, update the other fields           
            obj.JobAccess  = jobAccess;
            obj.SJobAccess = sjobAccess;
            obj.TaskAccess = taskAccess;
            obj.CachedName = char( obj.MJSAccess.getName() );
        end
        
        function disposeAccess( obj )
            if isempty( obj.MJSAccess )
                return;
            end
            % Release reference to any Java objects
            obj.MJSAccessCleanup = [];
            obj.MJSAccess.dispose();
            obj.MJSAccess = [];
        end
    end
    
    methods
       
        function delete( obj )
            obj.disposeAccess();
        end
        
        function updated = updateMJSAccess( obj )
            [updated, mjsAccess, accessCleanup] = obj.MJSAccessFactory.updateMJSAccess();
            if updated
                [ jobAccess, sJobAccess, taskAccess ] = iGetWritableJobAndTaskAccess( mjsAccess );
                obj.initAccessCommon( mjsAccess, jobAccess, sJobAccess, taskAccess, accessCleanup );
                
            end
        end
    end
    
    methods ( Static, Access = ?parallel.internal.cluster.JmAccessFactory )
        function wrapPerformSets( access, uuid, variant, propNameOrNames, valOrVals )
            % Simple wrapper around MJSMethods.performSets
            import parallel.internal.cluster.MJSMethods
            
            pcell = iscell( propNameOrNames );
            vcell = iscell( valOrVals );
            
            if pcell
                assert( vcell );
                propNames = propNameOrNames;
                values    = valOrVals;
            else
                % In this case, valOrVals might already be a cell -
                % e.g. AdditionalPaths.
                propNames = { propNameOrNames };
                values    = { valOrVals };
            end
            assert( numel( propNames ) == numel( values ) );
            
            MJSMethods.performSets( access, uuid, variant, propNames, values );
        end
    end
    
    methods ( Static, Access = private )
        % Wrappers around MJSMethods.performSets/performGets
        
        function v = wrapPerformGets( access, uuid, variant, propNameOrNames )
            % Simple wrapper around performGets to deal with vectorization of
            % propNameOrNames.
            import parallel.internal.cluster.MJSMethods
            
            if iscell(propNameOrNames)
                propNames = propNameOrNames;
            else
                propNames = { propNameOrNames };
            end
            v = MJSMethods.performGets( access, uuid, variant, propNames );
            
            returnCell = iscell( propNameOrNames ) || numel(uuid) > 1;
            if ~returnCell
                v = v{1};
            end
        end
    end
    
    methods % ClusterSupport
        function pauseQueue( obj )
            obj.MJSAccess.pauseQueue();
        end
        
        function resumeQueue( obj )
            obj.MJSAccess.resumeQueue();
        end
        
        function logout( obj, varargin )
            % logout is a client only operation therefore no need to call switchToWritable
            userIdentity = iGetUserIdentity( varargin{:} );
            obj.MJSAccess.revokeAuthentication( userIdentity );
        end
        
        function changePassword( obj, varargin )
            userIdentity = iGetUserIdentity( varargin{:} );
            obj.MJSAccess.changeCredentialsOfExistingUser( userIdentity );
        end
        
        function promoteJob( obj, jobSId )
            [~, jobUUID] = obj.unpackJobAccessInfo( jobSId );
            obj.MJSAccess.promote( jobUUID(1) );
        end
        
        function demoteJob( obj, jobSId )
            [~, jobUUID] = obj.unpackJobAccessInfo( jobSId );
            obj.MJSAccess.demote( jobUUID(1) );
        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
        
        function w = getWorkers( obj, mjs, type )
            % getWorkers - type must be 'BusyWorkers' or 'IdleWorkers'.
            % Returns a list of workers of the given type.
            % Caches worker objects.
            import parallel.cluster.MJSWorker
            import parallel.internal.types.Variant
            import parallel.internal.cluster.MJSSupport
            
            validatestring( type, {'BusyWorkers', 'IdleWorkers'} );
            
            workerProxArray = MJSSupport.wrapPerformGets( obj.MJSAccess, [], ...
                Variant.Cluster, type );
            if numel( workerProxArray ) == 0
                w = MJSWorker.empty();
            else
                workersCell = cell( 1, length( workerProxArray ) );
                for ii = 1:length( workerProxArray )
                    workerUuidStr = char( workerProxArray(ii).getID() );
                    if ~obj.Cache.isKey( workerUuidStr )
                        workersCell{ii} = iCreateMJSWorker( mjs, workerProxArray(ii) );
                        obj.Cache.add( obj.CachedName, workerUuidStr, workersCell{ii} );
                    else
                        workersCell{ii} = obj.Cache.retrieve( workerUuidStr );
                    end
                end
                % Return a column (N-by-1) vector
                w = [ workersCell{:} ].';
            end
        end
        
        function val = getOperatingSystem( obj )
            import parallel.internal.apishared.OsType
            
            operatingSystem = obj.MJSAccess.getClusterOperatingSystem();
            if strcmp( operatingSystem, 'MIXED' )
                val = OsType.MIXED.Api2Name;
            elseif strncmp( operatingSystem, 'PC', 2 ) % This is same test 'ispc' uses
                val = OsType.PC.Api2Name;
            else
                val = OsType.UNIX.Api2Name;
            end
        end
        
        function val = getClusterProperties( obj, propNames )
            % Get properties of the jobmanager itself.
            
            import parallel.internal.cluster.MJSSupport
            import parallel.internal.types.Variant
            
            val = MJSSupport.wrapPerformGets( obj.MJSAccess, [], ...
                Variant.Cluster, propNames );
        end
        
        function setClusterProperties( obj, propNames, propVals )
            % Set properties of the jobmanager itself.
            
            import parallel.internal.cluster.MJSSupport
            import parallel.internal.types.Variant
            
            MJSSupport.wrapPerformSets( obj.MJSAccess, [], ...
                Variant.Cluster, ...
                propNames, propVals );
        end
        
        function val = requiresMathWorksHostedLicensing( obj )
            val = obj.MJSAccess.requireWebLicensing();
        end
    end
    methods
        function job = getJobFromStruct( obj, mjs, jobStruct )
            job = obj.getJobFromUUID( mjs, jobStruct.jobUUID, jobStruct.jobType );
        end
        
        function task = getTaskFromStruct( obj, job, jobStruct, taskStruct )
            task = obj.getTaskFromUUID( job, jobStruct.jobUUID, taskStruct.taskUUID );
        end
        
        function jl = getJobs( obj, mjs )
            % Return a list of jobs (or a roughly appropriately typed empty) for a
            % given MJS.
            
            %{
            validateattributes( mjs, {'parallel.cluster.MJS','parallel.cluster.MJSComputeCloud'}, {'scalar'} );
            %}
            typeAndIdArray = obj.MJSAccess.getJobs();
            jl = getJobsFromTypeAndIdArray( obj, mjs, typeAndIdArray );
            if isempty( jl )
                % Choose one of the subclasses.
                jl = parallel.job.MJSIndependentJob.empty();
            else
                jl = [ jl{:} ];
            end
            
        end
        
        function varargout = getJobsByState( obj, mjs )
            
            %{
            validateattributes( mjs, {'parallel.cluster.MJS','parallel.cluster.MJSComputeCloud'}, {'scalar'} );
            %}
            import com.mathworks.toolbox.distcomp.mjs.workunit.WorkUnit;
            
            states = [WorkUnit.PENDING_STATE, WorkUnit.QUEUED_STATE, ...
                WorkUnit.RUNNING_STATE, WorkUnit.FINISHED_STATE];
            
            % Get jobs in all states.
            typeAndIdArrays = obj.MJSAccess.getJobs( states );
            jobsCell = cell( numel( states ), 1 );
            
            % Loop over each state and convert job UUIDs into job objects.
            for ii = 1:numel( states )
                jobsCell{ii} = getJobsFromTypeAndIdArray( obj, mjs, typeAndIdArrays(ii) );
                if isempty( jobsCell{ii} )
                    % Choose one of the subclasses.
                    jobsCell{ii} = { parallel.job.MJSIndependentJob.empty() };
                end
                
            end
            
            if nargout == 4
                varargout = jobsCell;
            else
                varargout = { jobsCell(:) };
            end
            
        end
        
        function [ p, q, r, f ] = enumerateJobsInStates( obj )
            % Return the number of jobs in Pending, Queued, Running and Finished states
            
            import com.mathworks.toolbox.distcomp.mjs.workunit.WorkUnit
            
            statesToRetrieve = [ WorkUnit.PENDING_STATE, WorkUnit.QUEUED_STATE, WorkUnit.RUNNING_STATE, WorkUnit.FINISHED_STATE ];
            JobIdsAndTypes = cell(obj.MJSAccess.getJobs(statesToRetrieve));
            if isempty(JobIdsAndTypes)
                [p,q,r,f]=deal(0,0,0,0);
            else
                p = numel(JobIdsAndTypes{1});
                q = numel(JobIdsAndTypes{2});
                r = numel(JobIdsAndTypes{3});
                f = numel(JobIdsAndTypes{4});
            end
        end
        
        function job = buildJob( obj, mjs, variant, entitlement, optUserName )
            % Construct a brand new job given a job variant and optionally a
            % Username. Returns the job object.
            
            import parallel.internal.cluster.MJSSupport
            import parallel.internal.cluster.DefaultProperties
            
            jobProps                        = DefaultProperties.getNewJobPropStruct();
            jobProps.Timeout                = intmax( 'int64' );
            jobProps.MaxWorkers             = intmax('int32');
            jobProps.MinWorkers             = 1;
            jobProps.Restart                = false;
            jobProps.AttachedFileData       = []; % Sadly, really does need to be double-empty
            jobProps.Name                   = []; % Instruct java to calculate name
            
            %% Job owner
            %  Get userName from job arguments: check for input argument pair where the
            % key is 'UserName', get the username value and remove the pair from the
            % argument list (otherwise it would get set again later on).
            if nargin > 4
                jobProps.Username = optUserName;
            else
                %% If userName is unset try to set it to the user currently set in the
                %  proxy object (which must never be null or empty).
                userIdentity = obj.MJSAccess.getCurrentUser();
                if isempty(userIdentity)
                    error(message('parallel:cluster:MJSInternalError'));
                end
                jobProps.Username = userIdentity.getSimpleUsername();
            end
            jobUUID = iBuildJob( jobProps, variant, obj.MJSAccess, entitlement );
            job     = obj.getJobFromUUID( mjs, jobUUID, variant );
            % Fire event for building job
            obj.JobStorageEventManager.notifyCreateJob( job );
        end
        
        function [ok, msg] = checkTwoWayCommunications( obj, forceCheck )
            [ok, msg] = obj.MJSAccessFactory.checkTwoWayCommunications( forceCheck );
        end
        
        function [connMgr, strategy] = buildConnectionManager( obj )
            % c.f.: @distcomp/@jobmanager/pCreateConnectionManager

            % We must get these properties after we have switched to writeable,
            % as they are defined by the job manager.
            props = obj.getClusterProperties({'HasSecureCommunication', 'WorkerProxiesPoolConnections'});
            brokerConnectInfo = obj.MJSAccess.getBrokerServerSocketConnectInfo;
            workerAcceptInfoTemplate = obj.MJSAccess.getWorkerAcceptInfoTemplate;
            useSecureMatlabPool = props{1};
            workerProxiesPoolConnections = props{2};
            
            [connMgr, strategy] = parallel.internal.apishared.ConnMgrBuilder.buildForMJS( ...
                brokerConnectInfo, workerAcceptInfoTemplate, ...
                useSecureMatlabPool, workerProxiesPoolConnections );
        end
        
    end
    
    methods ( Access = private )
        function job = getJobFromUUID( obj, mjs, jobUUID, jobTypeOrVariant )
            % Used in all places that need to build a Job object.
            
            import parallel.internal.types.Variant
            
            %{
            validateattributes( mjs, {'parallel.cluster.MJS', 'parallel.cluster.MJSComputeCloud'}, {'scalar'} );
            validateattributes( jobUUID, {'net.jini.id.Uuid'}, {'scalar'} );
            validateattributes( jobTypeOrVariant, {'numeric', ...
                                'parallel.internal.types.Variant'}, {'scalar'} );
            %}
            
            uuidStr = iGetJavaStr( jobUUID );
            [found, job] = retrieveIfPresent( obj.Cache, uuidStr );
            if ~found
                job = createJobFromUUID( obj, mjs, jobUUID, jobTypeOrVariant );
                obj.Cache.add( obj.CachedName, uuidStr, job );
            end
        end
        
        function tasks = getTaskFromUUID( obj, job, jobUUID, taskUUIDs )
            % Used by all places that need to build a Task.
            % This is vectorized over taskUUIDs, but not jobUUID.
            
            %{
            validateattributes( job, {'parallel.Job'}, {'scalar'} );
            validateattributes( jobUUID, {'net.jini.id.Uuid'}, {'scalar'} );
            validateattributes( taskUUIDs, {'net.jini.id.Uuid'}, {'scalar'} );
            %}
            taskUUIDStrs = cellstr(string(taskUUIDs));
            [found, foundTasks] = retrieveArrayIfPresent( obj.Cache, taskUUIDStrs );
            tasks = cell( 1, numel( taskUUIDs ) );
            tasks( found ) = foundTasks;
            
            if any( ~found )
                uncachedTaskUUIDs = taskUUIDs( ~found );
                uncachedTaskIDs = obj.retrieveUncachedTaskIDs( uncachedTaskUUIDs );
                
                uncachedTaskIndices = find( ~found );
                for ii = 1:numel( uncachedTaskIDs )
                    tasks{uncachedTaskIndices(ii)} = obj.buildTaskAndCache( ...
                        job, jobUUID, uncachedTaskUUIDs(ii), uncachedTaskIDs{ii} );
                end
            end
            tasks = [tasks{:}]; % Convert to array
        end
        
        function jobIds = convertUuidToJobId(obj, variants, uuidArr)
            % This method takes a vector of variants and a vector of uuids
            % of the same length and returns the job IDs
            assert( isvector( variants ) );
            assert( isvector( uuidArr ) );
            assert( length( variants ) == length( uuidArr ));
            
            % Determine the unique variants and where in the array they
            % are
            [ uniqueVariants, ~, iUniqueVariants ] = unique( { variants.Name }, 'stable' );
            
            % Iterate over the variants and get the job IDs for each
            % variant in a vectorised call
            jobIds = cell( numel( variants ), 1 );
            for jj = 1:numel( uniqueVariants )
                currentUniqueVariant = uniqueVariants{jj};
                idxForCurrentVariant = iUniqueVariants == jj;
                uuidForVariant = uuidArr( idxForCurrentVariant );
                % Need to make sure we still have a Java array here to
                % pass to the Java API
                if numel(uuidForVariant) == 1
                    uuidArrForVariant = javaArray( 'net.jini.id.Uuid', 1 );
                    uuidArrForVariant(1) = uuidForVariant;
                else
                    uuidArrForVariant = uuidForVariant;
                end
                jobSIds = { parallel.internal.types.Variant.fromNameI( currentUniqueVariant ), uuidArrForVariant };
                jobIds( idxForCurrentVariant ) = obj.getJobPropertyNoErrors( jobSIds, 'ID' );
            end
        end
        
        function jobProperties = getJobPropertyNoErrors( obj, jobSIds, propName )
            % Helper to get a property for an array of jobs in a robust way. 
            % For any jobs where we fail to get the value, jobProperties
            % will contain an empty entry.
            try
                % Calling getJobProperties with propName in a cell array because
                % we need the return value to be a cell array.
                jobProperties = obj.getJobProperties( jobSIds, {propName} );
            catch E
                dctSchedulerMessage(1, 'Error caught getting job property. Fall back to for loop. Error: %s', E.getReport());
                % Revert to a for-loop iteration to try and return
                % non-corrupted job property
                variant = jobSIds{1};
                jobUUIDs = jobSIds{2};
                jobProperties = cell( numel( jobUUIDs ), 1 );
                for idx = 1:numel( jobUUIDs )
                    try
                        jobSId = { variant, iGetJavaUUIDArray(jobUUIDs(idx)) };
                        jobProperties( idx ) = obj.getJobProperties( jobSId, {propName} );
                    catch err
                        dctSchedulerMessage(1, 'Error caught getting single job property. Leaving return value empty. Error: %s', err.getReport());
                        % Swallow error and leave jobProperties for entry
                        % empty
                    end
                end
            end
        end
        
        function [variant, jobUUID, jobAccess] = unpackJobAccessInfo( obj, jobSId )
            % Helper for Job support methods to pick the right access.
            import parallel.internal.types.Variant
            
            [variant, jobUUID] = iUnpackWorkUnitSupportID( jobSId );
            switch variant
                case Variant.IndependentJob
                    jobAccess = obj.JobAccess;
                case { Variant.ConcurrentJob, ...
                        Variant.CommunicatingSPMDJob, ...
                        Variant.CommunicatingPoolJob }
                    jobAccess = obj.SJobAccess;
                otherwise
                    assert( false );
            end
        end
        
        function jobsCell = getJobsFromTypeAndIdArray( obj, mjs, typeAndIdArray )
            [uuids, types, uuidsCellStr] = iGetIdAndType( typeAndIdArray );            
            jobsCell = cell( 1, length( uuids ) );
            
            % Would like to optimize here for the case where most of the
            % jobs are already in the cache. Get them all in one go.
            [found, cachedJobs] = retrieveArrayIfPresent( obj.Cache, uuidsCellStr );
            jobsCell(found) = cachedJobs;
            
            if all(found)
                % return early if all jobs have been found
                return;
            end
            
            jobsNotFound = find(~found);
            % Need to determine the uuid and corresponding variant for each
            % job that was not found in the cache
            [variants, uuidArr] = iRetrieveVariantAndUuidForIndex( jobsNotFound, types, uuids );
            
            % Now use the variant and uuids to get the job IDs
            jobsNotFoundIds = obj.convertUuidToJobId( variants, uuidArr );
            
            % Now we can iterate over each job not found, construct it
            % and add it to the cache
            for kk = 1:numel(jobsNotFound)
                jobsNotFoundId = jobsNotFoundIds{kk};
                if isempty(jobsNotFoundId)
                    % There was an error finding this job so do not
                    % populate it
                    continue
                end
                indexIntoJobCell = jobsNotFound(kk);
                uuidArrForBuild = javaArray( 'net.jini.id.Uuid', 1 );
                uuidArrForBuild(1) = uuidArr(kk);
                aJob = mjs.hBuildChild( variants(kk), jobsNotFoundIds{kk}, {variants(kk), uuidArrForBuild} );
                add( obj.Cache, obj.CachedName, uuidsCellStr{indexIntoJobCell}, aJob );
                jobsCell{indexIntoJobCell} = aJob;
            end
            
            % Remove any jobs which we failed to populate
            jobsCell = jobsCell(~cellfun(@isempty, jobsCell));
        end
    end

    methods( Hidden )
        function hErrorIfNumPoolWorkersInvalid( obj, numWorkers, clusterNumWorkers )
            % Accessing element 1 in case numWorkers has 2 elements
            minWorkers = numWorkers(1);
            if minWorkers > clusterNumWorkers
                if isempty( obj.getClusterProperties('Name') )
                    error(message('parallel:cluster:mjs:NumWorkersExceeded', minWorkers, clusterNumWorkers));
                else
                    error(message('parallel:cluster:mjs:ClusterNumWorkersExceeded', obj.getClusterProperties('Name'), minWorkers, clusterNumWorkers));
                end
            end
        end
        
        function hAddRemoteInvocationListener( obj, listener )
            % Add listener for job manager peer session events
            proxy = obj.MJSAccess.getJobManager();
            proxy.addPeerSessionProfileListener( listener );
        end
        
        function hRemoveRemoteInvocationListener( obj, listener )
            % Remove listener for job manager peer session events
            proxy = obj.MJSAccess.getJobManager();
            proxy.removePeerSessionProfileListener( listener );
        end
        
        function provider = hCreateJobStatusProvider( obj, cluster )
            import com.mathworks.toolbox.distcomp.ui.jobmonitor.JobManagerUpdateProvider
            jobManager = obj.MJSAccess;
            jobAccess = obj.JobAccess;
            userName = cluster.Username;
            provider = JobManagerUpdateProvider( ...
                userName, jobManager, jobAccess, cluster.RequiresMathWorksHostedLicensing );
        end
        
        function jobIdentifier = hCreateJobUpdateIdentifier( ~, jobSId, job )
            import com.mathworks.toolbox.distcomp.ui.jobmonitor.JobIdentifier
            jobUUID = jobSId{2};
            jobIdentifier = JobIdentifier( job.ID, jobUUID );
        end
    end

    methods  % Callback support
        function registerForJobEvents( obj, mjs, jobSId )
            [~, jobUUID, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            try
                obj.MJSAccess.attachToListenableObject( jobAccess, jobUUID );
                obj.stashCluster( mjs );
            catch err
                throw( distcomp.handleJavaException( mjs, err ) );
            end
        end
        function unregisterForJobEvents( obj, mjs, jobSId )
            [~, jobUUID, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            try
                obj.unstashCluster( mjs );
                obj.MJSAccess.detachFromListenableObject( jobAccess, jobUUID );
            catch err %#ok<NASGU>
                % We should not treat unregistering for events as an error.
                % This can be triggered if someone else deletes a job or
                % task and we are still registered for eventing.
            end
        end
        function registerForTaskEvents( obj, mjs, taskSId )
            [~, taskUUID, taskAccess] = obj.unpackTaskAccessInfo( taskSId );
            try
                obj.MJSAccess.attachToListenableObject( taskAccess, taskUUID );
                obj.stashCluster( mjs );
            catch err
                throw( distcomp.handleJavaException( mjs, err ) );
            end
        end
        function unregisterForTaskEvents( obj, mjs, taskSId )
            [~, taskUUID, taskAccess] = obj.unpackTaskAccessInfo( taskSId );
            try
                obj.unstashCluster( mjs );
                obj.MJSAccess.detachFromListenableObject( taskAccess, taskUUID );
            catch err %#ok<NASGU>
                % We should not treat unregistering for events as an error.
                % This can be triggered if someone else deletes a job or
                % task and we are still registered for eventing.
            end
        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
        
        function job = createJobFromUUID( obj, mjs, jobUUID, jobTypeOrVariant )
            % This is creating the MATLAB side of a job. The Java side of
            % the job has already been created.
            if isa( jobTypeOrVariant, 'parallel.internal.types.Variant' )
                variant = jobTypeOrVariant;
            else
                variant = iGetJobVariantForType( jobTypeOrVariant(1) );
            end
            uuidArr    = javaArray( 'net.jini.id.Uuid', 1 );
            uuidArr(1) = jobUUID;
            jobSId     = { variant, uuidArr };
            jobId      = obj.getJobProperties( jobSId, 'ID' );
            job        = mjs.hBuildChild( variant, jobId, jobSId );
        end        
        
    end
    
    methods % JobSupport
        function v = getJobProperties( obj, jobSId, propName )
            import parallel.internal.cluster.MJSSupport
            
            [variant, jobUUID, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            
            v = MJSSupport.wrapPerformGets( jobAccess, jobUUID, variant, propName );
        end
        
        function setJobProperties( obj, jobSId, propName, val )
            import parallel.internal.cluster.MJSSupport
            
            [variant, jobUUID, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            MJSSupport.wrapPerformSets( jobAccess, jobUUID, variant, propName, val );
        end
        
        function numTasksPerJob = getTaskCount( obj, jobSId )
            [~, jobUUIDs, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            jobTaskUUIDs = jobAccess.getTasks( jobUUIDs );
            numTasksPerJob = arrayfun( @numel, jobTaskUUIDs );
        end
        
        function tl = getTasks( obj, jobs, jobSId )
            [~, jobUUIDs, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            jobTaskUUIDs = jobAccess.getTasks( jobUUIDs );
            numTasksPerJob = arrayfun( @numel, jobTaskUUIDs );
            
            % Retrieve all tasks that are present in the cache
            allTaskUUIDs = jobTaskUUIDs(:);
            allTaskUUIDStrs = cellstr(string(allTaskUUIDs));
            [found, foundTasks] = retrieveArrayIfPresent( obj.Cache, allTaskUUIDStrs );
            allTasks = cell( 1, numel( allTaskUUIDs ) );
            allTasks( found ) = foundTasks;
            
            % Retrieve all tasks that were not in the cache in a single
            % vectorized call
            if any( ~found )
                uncachedTaskUUIDs = allTaskUUIDs( ~found );
                uncachedTaskIDs = obj.retrieveUncachedTaskIDs( uncachedTaskUUIDs );
                
                % Construct a "map" of task index to job index
                jobIndexMap = cell2mat( arrayfun( @(jobIndex) repmat( jobIndex, 1, numTasksPerJob(jobIndex) ), ...
                    1:numel( numTasksPerJob ), 'UniformOutput', false ) );
                
                % Build the uncached tasks and put them in the cache
                uncachedTaskIndices = find( ~found );
                for ii = 1:numel( uncachedTaskIndices )
                    uncachedTaskID = uncachedTaskIDs{ii};
                    if isempty(uncachedTaskID)
                        % There was an error finding this task so populate
                        % it with an empty one
                        allTasks{uncachedTaskIndices(ii)} = parallel.task.MJSTask.empty();
                        continue
                    end
                    uncachedTaskIndex = uncachedTaskIndices(ii);
                    jobIndex = jobIndexMap(uncachedTaskIndex);
                    
                    allTasks{uncachedTaskIndex} = obj.buildTaskAndCache( ...
                        jobs(jobIndex), jobUUIDs(jobIndex), uncachedTaskUUIDs(ii), uncachedTaskIDs{ii} );
                end
            end
            
            % Construct the cell array of tasks to return
            tl = iSplitCellArray( allTasks, numTasksPerJob, parallel.task.MJSTask.empty() );
            
            % Maintain backwards compatibility with non-vectorized case
            if numel( tl ) <= 1
                tl = [tl{:}];
            end
        end

        function [taskIDs, taskSIds] = buildTasksVec( obj, job, jobSId, tasksVecInfo )
            numTasks = tasksVecInfo.NumTasks;
            taskIDs = 1:numTasks;
            taskSIds = 1:numTasks;
            
            % Extract the input arguments from tasksVecInfo using sliced indices
            allTaskArgs = cell( 1, numTasks );
            slicedIndices = tasksVecInfo.SlicedIndices;
            numSlicedIndices = numel( slicedIndices );
            for ii = 1:numTasks
                taskArgs = tasksVecInfo.InputArguments;
                for jj = 1:numSlicedIndices
                    slicedIndex = slicedIndices(jj);
                    argsFull = taskArgs{slicedIndex};
                    if iscell(argsFull)
                        taskArgs{slicedIndex} = argsFull{ii};
                    else
                        taskArgs{slicedIndex} = argsFull(ii);
                    end
                end
                allTaskArgs{ii} = taskArgs;
            end
            
            taskInfoStructs = struct(...
                'Function', tasksVecInfo.Function, ...
                'NumOutputArguments', tasksVecInfo.NumOutputArguments, ...
                'InputArguments', allTaskArgs, ...
                'Name', '', ...
                'Timeout', intmax( 'int64' ), ...
                'CaptureDiary', tasksVecInfo.CaptureDiary, ...
                'MaximumRetries', 1, ...
                'ContainsCallbacks', false);
            
            sameFcn = true;
            obj.buildTasks(job, jobSId, sameFcn, taskInfoStructs );
        end
        
        function tasks = buildTasks( obj, job, jobSId, sameFcn, taskInfoStructs )
            import com.mathworks.toolbox.distcomp.mjs.workunit.TaskInfo
            import com.mathworks.toolbox.distcomp.mjs.datastore.ByteBufferItem
            import com.mathworks.toolbox.distcomp.util.ArraySerializer
            
            numTasks = numel( taskInfoStructs );
            
            % Optimization to only serialize the task function once if it
            % is the same for all tasks
            if sameFcn && numTasks >= 1
                serializedFcn = distcompserialize64( taskInfoStructs(1).Function );
            end
            
            usingRealWorkers = isempty( getenv( 'MDCE_QE_DUMMYWORKERS' ) );
            
            % Build the cell array of TaskInfos
            taskInfoCell = cell( 1, numTasks );
            for ii = 1:numTasks
                taskInfoStruct = taskInfoStructs(ii);
                
                if ~sameFcn
                    serializedFcn = distcompserialize64( taskInfoStruct.Function );
                end
    
                % If we're using dummyworkers, we can't use
                % distcompserialize, since it can only be called from
                % MATLAB. We use distcomp.util.ArraySerializer instead.
                argsIn = taskInfoStruct.InputArguments;
                if usingRealWorkers
                    backing{1} = distcompserialize64( argsIn );
                else
                    arrayList = java.util.ArrayList;

                    for jj = 1:numel( argsIn )
                        arrayList.add( argsIn{jj} );
                    end

                    backing{1} = ArraySerializer.serialize( arrayList.toArray() );
                    % A Java byte[] becomes a column vector. Transposing to keep it
                    % consistent with the output of distcompserialize.
                    backing{1} = backing{1}';
                end
                
                backing{2} = serializedFcn;
                
                timeout = 1000 * int64( taskInfoStruct.Timeout ); % Convert timeout to milliseconds
                inData   = ByteBufferItem( distcompWrapInByteBufferHandles( backing{1} ) );
                funcData = ByteBufferItem( distcompWrapInByteBufferHandles( backing{2} ) );
                funcName = char( taskInfoStruct.Function );
                
                taskInfo = TaskInfo( taskInfoStruct.Name, ...
                                     timeout, ...
                                     taskInfoStruct.CaptureDiary, ...
                                     taskInfoStruct.MaximumRetries, ...
                                     taskInfoStruct.NumOutputArguments, ...
                                     inData, ...
                                     funcData, ...
                                     funcName, ...
                                     [] );
                                 
                taskInfoCell{ii} = struct('Info', taskInfo, 'Backing', {backing});
                
                if taskInfoStruct.ContainsCallbacks
                    % Create a ListenerInfo array from the ProxyToMatlabObjectAdaptor and
                    % put it in the TaskInfo holder. This has to happen before the
                    % task is created so that callbacks are fired (g794117).
                    listenerInfoArray = obj.createListenerInfoArrayForAllEvents;
                    taskInfoCell{ii}.Info.setListenerInfo( listenerInfoArray );
                end
            end
            
            % Use the JobAccess to create the tasks on the job manager
            [~, jobUUID, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            taskInfoArr = javaArray( ...
                'com.mathworks.toolbox.distcomp.mjs.workunit.TaskInfo', numTasks );
            for ii = 1:numTasks
                ti = taskInfoCell{ii};
                taskInfoArr(ii) = ti.Info;
            end
            jobUUIDs = javaArray( 'net.jini.id.Uuid', numTasks );
            jobUUIDs(:) = jobUUID(1);
            taskUUIDs = jobAccess.createTask( jobUUIDs, taskInfoArr );
            
            % Only build the task objects if necessary
            if nargout > 0
                tasks = obj.getTaskFromUUID( job, jobUUID(1), taskUUIDs );
            else
                tasks = parallel.Task.empty();
            end
            
            % Fire event for building tasks
            obj.JobStorageEventManager.notifyCreateTasks(job);
        end
        
        function cancelJob( obj, jobSId, cancelException )
            [~, jobUUID, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            jobAccess.cancel( jobUUID, {cancelException.message} );
        end
        
        function notifyCancelJob( obj, job )
            % Fire event for canceling job
            obj.JobStorageEventManager.notifyCancelJob(job);
        end
        
        function destroyJob( obj, jobSId )
            [~, jobUUID, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            
            % Invalidate any cached task objects
            jobTaskUUIDs = jobAccess.getTasks( jobUUID );
            allTaskUUIDs = jobTaskUUIDs(:);
            allTaskUUIDStrs = cellstr(string(allTaskUUIDs));
            [~, oldTasks] = obj.Cache.retrieveArrayIfPresent( allTaskUUIDStrs );
            cellfun(@parallel.internal.cluster.invalidateTask, oldTasks);
            
            % Destroy and remove job itself
            jobAccess.ensureDestroyed( jobUUID );
            arrayfun(@(uuid) obj.Cache.remove( obj.CachedName, iGetJavaStr( uuid )), jobUUID );
        end
 
        function notifyDeleteJob( obj, job )
            % Fire event for deleting job
            obj.JobStorageEventManager.notifyDeleteJob( job );
        end
        
        function [jobUUID, jobAccess] = prepareJobForSubmission( obj, job, jobSId )
            [~, jobUUID, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            mjsPrepareForSubmission( job, jobAccess, jobUUID );
        end
        
        function submitJob( obj, job, jobSId )
            [jobUUID, jobAccess] = obj.prepareJobForSubmission( job, jobSId );
            jobAccess.submit( jobUUID, [] );
            % Fire event for submitting job
            obj.JobStorageEventManager.notifySubmitJob(job);
        end
        
        function info = getJobDataStoreInfo( obj )
            info = iGetDataStoreInfo( obj.JobAccess );
        end
        
        function statusData = getJobStatusData( obj, jobSId )
            [~, jobUUIDs, jobAccess] = obj.unpackJobAccessInfo( jobSId );
            statusData = jobAccess.getStatusData( jobUUIDs );
            statusData = arrayfun( @iAdaptJobStatusData, statusData );
        end
    end
    
    methods ( Access = private )
        function listenerInfoArray = createListenerInfoArrayForAllEvents( obj )
            listenerInfoArray = obj.MJSAccess.createListenerInfoArrayForAllEvents();
        end
        
        function [variant, taskUUID, taskAccess] = unpackTaskAccessInfo( obj, taskSId )
            [variant, taskUUID] = iUnpackWorkUnitSupportID( taskSId );
            taskAccess = obj.TaskAccess;
        end
        
        function taskIDs = retrieveUncachedTaskIDs( obj, taskUUIDs )
            % Make a vectorized call to get all uncached task properties
            taskSId = iGetTaskSId( taskUUIDs );
            taskIDs = obj.getTaskPropertyNoErrors( taskSId, 'ID' );
        end
        
        function taskProperties = getTaskPropertyNoErrors( obj, taskSIds, propName )
            % Helper to get a property for an array of tasks in a robust way. 
            % For any tasks where we fail to get the value, taskProperties
            % will contain an empty entry.
            try
                % Calling getTaskProperties with propName in a cell array because
                % we need the return value to be a cell array.
                taskProperties = obj.getTaskProperties( taskSIds, {propName} );
            catch E
                dctSchedulerMessage(1, 'Error caught getting task property. Fall back to for loop. Error: %s', E.getReport());
                % Revert to a for-loop iteration to try and return
                % non-corrupted task property
                variant = taskSIds{1};
                taskUUIDs = taskSIds{2};
                taskProperties = cell( numel( taskUUIDs ), 1 );
                for idx = 1:numel( taskUUIDs )
                    try
                        taskSId = { variant, iGetJavaUUIDArray(taskUUIDs(idx)) };
                        taskProperties( idx ) = obj.getTaskProperties( taskSId, {propName} );
                    catch err
                        dctSchedulerMessage(1, 'Error caught getting single task property. Leaving return value empty. Error: %s', err.getReport());
                        % Swallow error and leave taskProperties for entry
                        % empty
                    end
                end
            end
        end
        
        function task = buildTaskAndCache( obj, job, jobUUID, taskUUID, taskID )
            % Build the task
            taskSId = iGetTaskSId( taskUUID );
            task = job.hBuildChild( taskID, taskSId );
            
            % Store it in the cache
            jobUUIDStr = iGetJavaStr( jobUUID );
            taskUUIDStr = iGetJavaStr( taskUUID );
            add( obj.Cache, jobUUIDStr, taskUUIDStr, task );
        end
        
        % Event handler functions for MJS State events
        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
    
    methods % TaskSupport
        % NB: 'taskSId' is a 2-cell containing { variant, taskUUID }
        function w = getWorkerForTask( obj, mjs, taskSId )
            % Retrieve the MJSWorker object for a task.
            import parallel.cluster.MJSWorker
            
            workerProp = obj.getTaskProperties( taskSId, 'WorkerProperties' );
            
            if isempty( workerProp )
                w = MJSWorker.empty();
            else
                workerUuidStr = char( workerProp.getID() );
                if ~obj.Cache.isKey( workerUuidStr )
                    w = iCreateMJSWorker( mjs, workerProp );
                    obj.Cache.add( obj.CachedName, workerUuidStr, w );
                else
                    w = obj.Cache.retrieve( workerUuidStr );
                end
            end
        end
        
        function v = getTaskProperties( obj, taskSId, propName )
            % Get one or more properties of a task
            import parallel.internal.cluster.MJSSupport
            
            [variant, taskUUID, taskAccess] = obj.unpackTaskAccessInfo( taskSId );
            v = MJSSupport.wrapPerformGets( taskAccess, taskUUID, variant, propName );
        end
        
        function setTaskProperties( obj, taskSId, propName, val )
            % Set one or more properties of a task
            import parallel.internal.cluster.MJSSupport
            
            [variant, taskUUID, taskAccess] = obj.unpackTaskAccessInfo( taskSId );
            MJSSupport.wrapPerformSets( taskAccess, taskUUID, variant, ...
                propName, val );
        end
        
        function destroyOneTask( obj, jobSId, taskSId )
            % Destroy a given task
            [~, taskUUID, taskAccess] = obj.unpackTaskAccessInfo( taskSId );
            [~, jobUUID]              = obj.unpackJobAccessInfo( jobSId );
            taskAccess.ensureDestroyed( taskUUID );
            obj.Cache.remove( iGetJavaStr( jobUUID(1) ), iGetJavaStr( taskUUID(1) ) );
        end
        
        function notifyDeleteTask( obj, job )
            % Fire event for deleting task
            obj.JobStorageEventManager.notifyDeleteTask( job );
        end
        
        function cancelOneTask( obj, taskSId, cancelException )
            [~, taskUUID, taskAccess] = obj.unpackTaskAccessInfo( taskSId );
            taskAccess.cancel( taskUUID, cancelException.message );
        end

        function notifyCancelTask( obj, job )
            % Fire event for canceling task
            obj.JobStorageEventManager.notifyCancelTask( job );
        end
        
        function submitTaskResult( obj, taskSId, attempt, output, exception, warnings, diaryOutput)
            import parallel.internal.cluster.ByteBufferItemWrapper
            import com.mathworks.toolbox.distcomp.mjs.workunit.TaskAttemptIdentifier
            
            % Serialize the output
            outputWrapper = ByteBufferItemWrapper.wrapInByteBufferItem( output );
            outputArray = outputWrapper.get();
            outputValue = outputArray(1);

            % Serialize the command window output
            encodedDiary = unicode2native( diaryOutput, 'UTF-16LE' );
            diaryWrapper = ByteBufferItemWrapper.wrapInByteBufferItemNoSerialization( encodedDiary );
            diaryArray = diaryWrapper.get();
            diaryValue = diaryArray(1);

            % Serialize the exception and extract the identifier and message
            if isempty( exception )
                serializedException = [];
                errorIdentifier = '';
                errorMessage = '';
            else
                serializedException = distcompserialize( exception );
                errorIdentifier = exception.identifier;
                errorMessage = exception.message;
            end

            % Serialize the warnings
            if isempty( warnings )
                % To save data transfer we create and send an empty uint8 array, since
                % serializing an empty Warning is still fairly large
                emptyBytes = uint8.empty(0);
                warningsWrapper = ByteBufferItemWrapper.wrapInByteBufferItemNoSerialization( emptyBytes );
            else
                warningsWrapper = ByteBufferItemWrapper.wrapInByteBufferItem( warnings );
            end
            warningsArray = warningsWrapper.get();
            warningsValue = warningsArray(1);

            % Submit the result
            [~, taskUUID, taskAccess] = obj.unpackTaskAccessInfo( taskSId );
            attemptID = TaskAttemptIdentifier( taskUUID(1), attempt );
            taskAccess.submitResult( taskUUID(1), attemptID, outputValue, serializedException, ...
                errorMessage, errorIdentifier, warningsValue, diaryValue );
        end
        
        function info = getTaskDataStoreInfo( obj )
            info = iGetDataStoreInfo( obj.TaskAccess );
        end
    end
end


% --------------------------------------------------------------------------
% Given a structure containing all the job properties we need, a variant and the
% MJSAccess, actually invoke the appropriate create*Job on the access to build a
% new job. Return the new job's UUID.
function jobUUID = iBuildJob( jobProps, variant, MJSAccess, entitlement )
import com.mathworks.toolbox.distcomp.mjs.workunit.JobInfo
import com.mathworks.toolbox.distcomp.mjs.workunit.JobMLType
import parallel.internal.types.Variant

switch variant
    case Variant.IndependentJob
        jobProps.Type = JobMLType.CLUSTER_STANDARD_JOB;
    case Variant.CommunicatingSPMDJob
        jobProps.Type = JobMLType.CLUSTER_PARALLEL_JOB;
    case Variant.CommunicatingPoolJob
        jobProps.Type = JobMLType.CLUSTER_MATLABPOOL_JOB;
    case Variant.ConcurrentJob
        jobProps.Type = JobMLType.CLUSTER_CONCURRENT_JOB;
    otherwise
        assert( false );
end

jobInfo = JobInfo( jobProps.Name, jobProps.Timeout, jobProps.Tag, ...
    jobProps.MaxWorkers, jobProps.MinWorkers, jobProps.Restart, ...
    jobProps.AutoAttachFiles, jobProps.DependentFiles, ...
    jobProps.AttachedFiles, jobProps.AttachedFilePaths, ...
    jobProps.MATLABDrivePathMap, jobProps.AdditionalPaths, ...
    jobProps.NumPathsToAppend, jobProps.AutoAddClientPath, ...
    jobProps.ProductKeys, jobProps.AttachedFileData, jobProps.JobData, ...
    jobProps.Type, jobProps.ExecutionMode, jobProps.Username, [] );

jobInfo.setMATLABRelease( pctGetMATLABRelease() );

if MJSAccess.requireWebLicensing
    jobInfo = iAddWebLicensingInfo( jobInfo, entitlement );
end

switch variant
    case Variant.IndependentJob
        jobUUID = MJSAccess.createJob( jobInfo );
    case Variant.CommunicatingSPMDJob
        jobUUID = MJSAccess.createParallelJob( jobInfo );
    case Variant.ConcurrentJob
        jobUUID = MJSAccess.createConcurrentJob( jobInfo );
    case Variant.CommunicatingPoolJob
        serializedNopFunction = distcompserialize(@parallel.internal.pool.poolWorkerFcn);
        serializedEmptyCell   = distcompserialize({});
        jobUUID = MJSAccess.createMatlabPoolJob( ...
            jobInfo, serializedNopFunction, serializedEmptyCell);
    otherwise
        assert( false );
end
end

% --------------------------------------------------------------------------
% Add web licensing information to a job info
function jobInfo = iAddWebLicensingInfo(jobInfo, entitlement)
import com.mathworks.toolbox.distcomp.mjs.auth.credentials.WebLicenseCredentials;
import com.mathworks.toolbox.distcomp.mjs.auth.credentials.CredentialRole;
import com.mathworks.toolbox.distcomp.mjs.auth.credentials.UserIdentity;
userIdentity = UserIdentity( jobInfo.getUserName );

currentUser = parallel.internal.mwa.ensureLoggedIn();
token = currentUser.getToken();
username = currentUser.getUsername();
webId = entitlement.Id;
licenseNumber = entitlement.LicenseNumber;
% Create a web license credentials object to pass securely to the job manager.
licenseInfo = WebLicenseCredentials(userIdentity, username, ...
    token, webId, licenseNumber);
credentialRole = CredentialRole.WEB_LICENSE_INFO;
jobInfo.addCredentials(credentialRole, licenseInfo);
end

% --------------------------------------------------------------------------
% Return a java array of UUIDs and a double-array of numeric type values given a
% typeAndIdArray as returned by the MJSAccess.getJobs() method.
function [uuids, types, uuidsCellStr] = iGetIdAndType(typeAndIdArray)
numJobs = numel(typeAndIdArray);
if numJobs > 0
    % Create the UUID array to be of the correct type
    uuids = javaArray('net.jini.id.Uuid', numJobs);
else
    % Return a java null which is the empty array
    uuids = [];
end
types = zeros(size(uuids));
uuidsCellStr = cell(size(uuids));

for i = 1:numJobs
    uuid = getJobID(typeAndIdArray(i));
    uuids(i) = uuid;
    types(i) = getJobType(typeAndIdArray(i));
    uuidsCellStr{i} = iGetJavaStr(uuid);
end
end

% --------------------------------------------------------------------------
% iGetJobVariantForType - given a list of types as returned by the MJSAccess,
% convert that to a cell array of Variant.
function var = iGetJobVariantForType( type )
import parallel.internal.types.Variant
import com.mathworks.toolbox.distcomp.mjs.workunit.JobMLType

if JobMLType.isStandardJob( type )
    var = Variant.IndependentJob;
elseif JobMLType.isParallelJob( type )
    var = Variant.CommunicatingSPMDJob;
elseif JobMLType.isMatlabpoolJob( type )
    var = Variant.CommunicatingPoolJob;
elseif type == JobMLType.CLUSTER_CONCURRENT_JOB
    var = Variant.ConcurrentJob;
else
    assert( false );
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'
        dctSchedulerMessage(4, 'Stashing cluster Name: %s, Host %s', cluster.Name, cluster.Host);
        c = STASHED_CLUSTERS.stash( cluster ); %#ok<NASGU>
    case 'unstash'
        dctSchedulerMessage(4, 'Unstashing cluster Name: %s, Host %s', cluster.Name, cluster.Host);
        c = STASHED_CLUSTERS.unstash( cluster ); %#ok<NASGU>
end
end

%--------------------------------------------------------------------------
% Detects SSLHandshake errors, and therefore problems with the certificate.
function iHandlePossibleSSLHandshakeFailure( E, jmprx )
[isJavaEx, exceptionType, causes] = isJavaException( E );
if isJavaEx ...
        && strcmp( exceptionType, 'java.rmi.ConnectIOException' ) && ~isempty( causes ) ...
        && strcmp( causes{1}, 'javax.net.ssl.SSLHandshakeException' )
    error(message('parallel:cluster:SSLHandshakeFailed', char( jmprx.getName ), char( jmprx.getHostName )));
else
    rethrow( E );
end
end

%-------------------------------------------------------------------------
% Get data store info back from a job or task access object
function info = iGetDataStoreInfo( access )
info = struct('DataStoreSize', NaN, 'DataStoreExportPort', NaN);
info.DataStoreSize = access.getDataStoreSize;
info.DataStoreExportPort = access.getDataStoreExportPort;
end

%-------------------------------------------------------------------------
% Get the job and task accesses from an mjsAccess
function [jobAccess, sJobAccess, taskAccess] = iGetWritableJobAndTaskAccess(mjsAccess)
try
    % Get the job/task accesses from the mjsAccess. If these remote calls fail
    % we don't update anything in obj.
    jobAccess  = mjsAccess.getJobAccess();
    sJobAccess = mjsAccess.getSimultaneousTasksJobAccess();
    taskAccess = mjsAccess.getTaskAccess();
catch E
    % These are the first remote calls to the JM we've made, so
    % it's the opportunity to detect certificate problems.
    iHandlePossibleSSLHandshakeFailure( E, mjsAccess );
end
end

function [variants, uuidArr] = iRetrieveVariantAndUuidForIndex(jobIndexArr ,types, uuids)
numJobsNotFound = numel(jobIndexArr);
% First we build an Java array of uuids and corresponding job variants
uuidArr = javaArray( 'net.jini.id.Uuid', numJobsNotFound );
variants = parallel.internal.types.Variant.empty();
for ii = 1:numJobsNotFound
    indexIntoJobCell = jobIndexArr(ii);
    if isa( types(indexIntoJobCell), 'parallel.internal.types.Variant' )
        variants(ii) = types(indexIntoJobCell);
    else
        type = types(indexIntoJobCell);
        variants(ii) = iGetJobVariantForType( type(1) );
    end
    uuidArr(ii) = uuids(indexIntoJobCell);
    
end
end

%--------------------------------------------------------------------------
function taskSId = iGetTaskSId( taskUUIDs )
taskUUIDArray = iGetJavaUUIDArray( taskUUIDs );
taskVariant = parallel.internal.types.Variant.Task;
taskSId = { taskVariant, taskUUIDArray };
end

function uuidArray = iGetJavaUUIDArray( uuids )
if isscalar( uuids )
    % Construct a Java UUID array for the single UUID; otherwise it will
    % remain a single object in Java
    uuidArray = javaArray( 'net.jini.id.Uuid', 1 );
    uuidArray(1) = uuids;
else
    uuidArray = uuids; % Automatically converted to Java UUID array
end
end

function uuidStr = iGetJavaStr( uuid )
% NOTE calling toString before calling char is significantly
% quicker that single char call.
uuidStr = char( toString( uuid ) );
end

function val = iSplitCellArray( array, sizes, emptyValue )
val = cell( 1, numel( sizes ) );

arrayIndex = 1;
for valIndex = 1:numel( val )
    size = sizes(valIndex);
    if size == 0
        val{valIndex} = emptyValue;
    else
        valArray = array(arrayIndex:arrayIndex + size - 1);
        val{valIndex} = [valArray{:}]; % Convert cell array to array
    end
    arrayIndex = arrayIndex + size;
end
end

function [variant, uuid] = iUnpackWorkUnitSupportID( supportID )
    % The first column is the Variant. They are all jobs or all tasks, so
    % just take the Variant of the first one.
    %
    % The second column are the Uuids of the jobs or tasks, we want to
    % concatenate all these so that we can make a vectorized call to the
    % job manager.
    variant = supportID{1,1};
    uuid = [supportID{:,2}];
end

function mjsWorker = iCreateMJSWorker( mjs, javaWorker )
    hostName = char( javaWorker.getHostName() );
    computerMLType = char( javaWorker.getComputerMLType() );
    allHostAddresses = cell( javaWorker.getAllHostAddresses() );
    name = char( javaWorker.getName() );
    mjsWorker = parallel.cluster.MJSWorker( mjs, hostName, computerMLType, allHostAddresses, name );
end

function jobStatusData = iAdaptJobStatusData( jobStatusData )
    import parallel.internal.general.DateTimeUtils
    
    % Convert an empty value to an empty JobStatusData
    if isempty( jobStatusData )
        jobStatusData = parallel.internal.cluster.JobStatusData();
        return
    end
    
    jobStatusData = parallel.internal.cluster.JobStatusData( ...
        jobStatusData.getId(), ...
        parallel.internal.cluster.types.JobType.fromOrdinal( jobStatusData.getRawJobType().ordinal() ), ...
        char( jobStatusData.getUsername() ), ...
        DateTimeUtils.getMatlabDateTime( jobStatusData.getCreateTime() ), ...
        DateTimeUtils.getMatlabDateTime( jobStatusData.getSubmitTime() ), ...
        DateTimeUtils.getMatlabDateTime( jobStatusData.getFinishTime() ), ...
        char( jobStatusData.getState() ), ...
        jobStatusData.getTotalPendingTasks(), ...
        jobStatusData.getTotalRunningTasks(), ...
        jobStatusData.getTotalFinishedTasks(), ...
        jobStatusData.getTotalTasks(), ...
        char( jobStatusData.getTag() ), ...
        char( jobStatusData.getApiTag() ), ...
        jobStatusData.getTotalSuccessfulFinishedTasks(), ...
        double( jobStatusData.getTaskIdsWithErrors() ), ...
        double( jobStatusData.getTaskIdsWithWarnings() ), ...
        char( jobStatusData.getFirstTaskMLFunctionName() ), ...
        transpose( arrayfun( @iAdaptCorruptionInfo, jobStatusData.getCorruptionInfo() ) ) );
end

function corruptionInfo = iAdaptCorruptionInfo( corruptionInfo )
    level = parallel.internal.cluster.types.JobCorruptionLevel.fromOrdinal( corruptionInfo.getLevel().ordinal() );
    message = char( corruptionInfo.getMessage() );
    corruptionInfo = parallel.internal.cluster.JobCorruptionInfo( level, message );
end

function userIdentity = iGetUserIdentity( username )
    import com.mathworks.toolbox.distcomp.mjs.auth.credentials.UserIdentity;

    if nargin < 1
        userIdentity = []; % converts to null
    else
        userIdentity = UserIdentity( username );
    end
end
