
%

%   Copyright 2014-2020 The MathWorks, Inc.

classdef WorkerWaiter < parallel.internal.cloud.Waiter
    properties (SetAccess = immutable)
        NumRequestedWorkers
        Cluster
        Notifier = []
    end

    methods
        function obj = WorkerWaiter(waitDelay, numRequestedWorkers, cluster, notifier)
            obj@parallel.internal.cloud.Waiter(waitDelay)
            obj.NumRequestedWorkers = numRequestedWorkers;
            obj.Cluster = cluster;
            if nargin > 3
                obj.Notifier = notifier;
            end
        end
    end

    methods (Access = protected)
        function ok = doCheck(obj)
            ok = false;
            
            % we want to abort on the spot if the cluster was instructed to shut down
            parallel.cluster.MJSCloudCluster.hErrorIfClusterShuttingDown(obj.Cluster.Identifier);
            
            % There is a potential race condition here, as we are making separate
            % calls to get the number of idle and busy workers independently.
            % However for MHC we are only supporting one client using the cluster,
            % so no workers can be made busy, as the client is blocking waiting
            % for the cluster to report that it is ready.
            % We need to get the actual number of workers from the cluster. 
            % We use the getClusterProps method instead of the NumWorkers
            % property because the property does not reflect the actual state of
            % the cluster.
            numWorkers = obj.Cluster.hGetClusterProps('NumWorkers');
            numRequestedWorkers = obj.NumRequestedWorkers;
            if numWorkers >= numRequestedWorkers
                % We've got enough workers
                ok = true;
                dctSchedulerMessage(1, ...
                    'Requested number of workers registered with job manager, requested workers: %d, actual workers: %d', ...
                    numRequestedWorkers, numWorkers);
            else
                % Check if the cluster is not going to start any more
                % workers for us; error out in this case.
                iErrorIfInsufficientWorkers(obj.Cluster, numRequestedWorkers);
                
                dctSchedulerMessage(4, ...
                    'Not all workers registered with job manager, requested workers: %d, actual workers: %d', ...
                    numRequestedWorkers, numWorkers);
            end
            if ~isempty(obj.Notifier)
                obj.Notifier.notifyWorkersRegistered(obj.Cluster.Identifier, numWorkers, numRequestedWorkers);
            end
        end
    end
end

function iErrorIfInsufficientWorkers(cluster, numRequestedWorkers)
% Always make sure we have the latest state before deciding if we
% should error.
cluster.hRefreshState();
state = cluster.hGetCloudConsoleState();
numAvailableWorkers = cluster.hGetAvailableNumWorkers();
numTotalWorkers = cluster.hGetTotalNumWorkers();
% If the cluster has reached the "READY" state, but reports fewer available
% workers than the total number of workers, that tells us we're not going
% to be able to start a pool unless the requested number of workers is less
% than the number of available workers. And we should never get to here if
% that is the case, because the checks above verify this condition already.
if state == parallel.internal.types.CloudCenterClusterState.Ready && ...
        numAvailableWorkers < numTotalWorkers
    dctSchedulerMessage(1, ...
        ['Could not acquire requested workers for pool formation: ' ...
         'requested workers: %d, available workers: %d'], ...
        numRequestedWorkers, numAvailableWorkers);
    error(message('parallel:cluster:MJSComputeCloud:RequestedMoreWorkersThanAvailable', ...
        numRequestedWorkers, numAvailableWorkers));
end
end
