/**
 *
 * <h1>Worker allocation steps</h1>
 *
 * <ol>
 *     <li>The resource manager decides to request more workers. This can happen in order
 *         to fill the initial pool, or as a result of the JobManager requesting more workers.</li>
 *
 *     <li>The resource master calls {@link #requestNewWorkers(int)}, which triggers requests
 *         for more containers. After that, the {@link #getNumWorkerRequestsPending()}
 *         should reflect the pending requests.</li>
 *
 *     <li>The concrete framework may acquire containers and then trigger to start TaskManagers
 *         in those containers. That should be reflected in {@link #getNumWorkersPendingRegistration()}.</li>
 *
 *     <li>At some point, the TaskManager processes will have started and send a registration
 *         message to the JobManager. The JobManager will perform
 *         a lookup with the ResourceManager to check if it really started this TaskManager.
 *         The method {@link #workerStarted(ResourceID)} will be called
 *         to inform about a registered worker.</li>
 * </ol>
 *
 */
public abstract class FlinkResourceManager<WorkerType extends ResourceIDRetrievable> extends FlinkUntypedActor {

    /** The service to find the right leader JobManager (to support high availability) */
    private final LeaderRetrievalService leaderRetriever; //用于发现leader jobmanager和当leader切换时收到通知

    /** Map which contains the workers from which we know that they have been successfully started
     * in a container. This notification is sent by the JM when a TM tries to register at it. */
    private final Map<ResourceID, WorkerType> startedWorkers; //已经成功启动的Workers，当他启动成功注册到JM的时候，JM会发出通知


    /** The JobManager that the framework master manages resources for */
    private ActorRef jobManager;

    /** Our JobManager's leader session */
    private UUID leaderSessionID;

    /** The size of the worker pool that the resource master strives to maintain */
    private int designatedPoolSize; //resource pool大小

    @Override
    protected void handleMessage(Object message) {
        try {
            // --- messages about worker allocation and pool sizes

            if (message instanceof CheckAndAllocateContainers) {
                checkWorkersPool();
            }
            else if (message instanceof SetWorkerPoolSize) {
                SetWorkerPoolSize msg = (SetWorkerPoolSize) message;
                adjustDesignatedNumberOfWorkers(msg.numberOfWorkers());
            }
            else if (message instanceof RemoveResource) {
                RemoveResource msg = (RemoveResource) message;
                removeRegisteredResource(msg.resourceId());
            }

            // --- lookup of registered resources

            else if (message instanceof NotifyResourceStarted) {
                NotifyResourceStarted msg = (NotifyResourceStarted) message;
                handleResourceStarted(sender(), msg.getResourceID());
            }

            // --- messages about JobManager leader status and registration

            else if (message instanceof NewLeaderAvailable) {
                NewLeaderAvailable msg = (NewLeaderAvailable) message;
                newJobManagerLeaderAvailable(msg.leaderAddress(), msg.leaderSessionId());
            }
            else if (message instanceof TriggerRegistrationAtJobManager) {
                TriggerRegistrationAtJobManager msg = (TriggerRegistrationAtJobManager) message;
                triggerConnectingToJobManager(msg.jobManagerAddress());
            }
            else if (message instanceof RegisterResourceManagerSuccessful) {
                RegisterResourceManagerSuccessful msg = (RegisterResourceManagerSuccessful) message;
                jobManagerLeaderConnected(msg.jobManager(), msg.currentlyRegisteredTaskManagers());
            }

/**
 * This method causes the resource framework master to <b>synchronously</b>re-examine
 * the set of available and pending workers containers, and allocate containers
 * if needed.
 *
 * This method does not automatically release workers, because it is not visible to
 * this resource master which workers can be released. Instead, the JobManager must
 * explicitly release individual workers.
 */
private void checkWorkersPool() {
    int numWorkersPending = getNumWorkerRequestsPending();
    int numWorkersPendingRegistration = getNumWorkersPendingRegistration();

    // see how many workers we want, and whether we have enough
    int allAvailableAndPending = startedWorkers.size() +
        numWorkersPending + numWorkersPendingRegistration;

    int missing = designatedPoolSize - allAvailableAndPending;

    if (missing > 0) {
        requestNewWorkers(missing); //如果现有的worker不够，去requestNewWorker
    }
}

/**
 * Tells the ResourceManager that a TaskManager had been started in a container with the given
 * resource id.
 *
 * @param jobManager The sender (JobManager) of the message
 * @param resourceID The resource id of the started TaskManager
 */
private void handleResourceStarted(ActorRef jobManager, ResourceID resourceID) {
    if (resourceID != null) {
        // check if resourceID is already registered (TaskManager may send duplicate register messages)
        WorkerType oldWorker = startedWorkers.get(resourceID);
        if (oldWorker != null) { //看看该worker是否已经存在
            LOG.debug("Notification that TaskManager {} had been started was sent before.", resourceID);
        } else {
            WorkerType newWorker = workerStarted(resourceID); //取得worker

            if (newWorker != null) {
                startedWorkers.put(resourceID, newWorker); //注册新的worker
                LOG.info("TaskManager {} has started.", resourceID);
            } else {
                LOG.info("TaskManager {} has not been started by this resource manager.", resourceID);
            }
        }
    }

    // Acknowledge the resource registration
    jobManager.tell(decorateMessage(Acknowledge.get()), self()); //告诉jobManager，已经完成注册
}

public boolean scheduleForExecution(SlotProvider slotProvider, boolean queued) throws NoResourceAvailableException {

    final SlotSharingGroup sharingGroup = vertex.getJobVertex().getSlotSharingGroup();
    final CoLocationConstraint locationConstraint = vertex.getLocationConstraint();

    if (transitionState(CREATED, SCHEDULED)) {

        ScheduledUnit toSchedule = locationConstraint == null ?
            new ScheduledUnit(this, sharingGroup) :
            new ScheduledUnit(this, sharingGroup, locationConstraint);

        // IMPORTANT: To prevent leaks of cluster resources, we need to make sure that slots are returned
        //     in all cases where the deployment failed. we use many try {} finally {} clauses to assure that
        final Future<SimpleSlot> slotAllocationFuture = slotProvider.allocateSlot(toSchedule, queued); //异步去申请资源

        // IMPORTANT: We have to use the synchronous handle operation (direct executor) here so
        // that we directly deploy the tasks if the slot allocation future is completed. This is
        // necessary for immediate deployment.
        final Future<Void> deploymentFuture = slotAllocationFuture.handle(new BiFunction<SimpleSlot, Throwable, Void>() {
            @Override
            public Void apply(SimpleSlot simpleSlot, Throwable throwable) {
                if (simpleSlot != null) {
                    try {
                        deployToSlot(simpleSlot); //如果申请到，去部署
                    } catch (Throwable t) {
                        try {
                            simpleSlot.releaseSlot();
                        } finally {
                            markFailed(t);
                        }
                    }
                }
                else {
                    markFailed(throwable);
                }
                return null;
            }
        });
        
        return true;
    }

@Override
public Future<SimpleSlot> allocateSlot(ScheduledUnit task, boolean allowQueued)
        throws NoResourceAvailableException {

    final Object ret = scheduleTask(task, allowQueued);
    if (ret instanceof SimpleSlot) {
        return FlinkCompletableFuture.completed((SimpleSlot) ret); //如果是SimpleSlot，即已经分配成功，表示future结束
    }
    else if (ret instanceof Future) {
        return (Future) ret; //Future说明没有足够资源，申请还在异步中，继续future
    }
    else {
        throw new RuntimeException();
    }
}

/**
     * Returns either a {@link SimpleSlot}, or a {@link Future}.
     */
    private Object scheduleTask(ScheduledUnit task, boolean queueIfNoResource) throws NoResourceAvailableException {

        final ExecutionVertex vertex = task.getTaskToExecute().getVertex();
        
        final Iterable<TaskManagerLocation> preferredLocations = vertex.getPreferredLocations();
        final boolean forceExternalLocation = vertex.isScheduleLocalOnly() &&
                                    preferredLocations != null && preferredLocations.iterator().hasNext();
    
        synchronized (globalLock) { //全局锁
            
            SlotSharingGroup sharingUnit = task.getSlotSharingGroup();
            
            if (sharingUnit != null) { //如果是共享slot

                // 1)  === If the task has a slot sharing group, schedule with shared slots ===
                
                final SlotSharingGroupAssignment assignment = sharingUnit.getTaskAssignment();
                final CoLocationConstraint constraint = task.getLocationConstraint();
                
                // get a slot from the group, if the group has one for us (and can fulfill the constraint)
                final SimpleSlot slotFromGroup;
                if (constraint == null) {
                    slotFromGroup = assignment.getSlotForTask(vertex); //试图从现有的slots中找合适的
                }
                else {
                    slotFromGroup = assignment.getSlotForTask(vertex, constraint);
                }

                SimpleSlot newSlot = null;
                SimpleSlot toUse = null;

                // the following needs to make sure any allocated slot is released in case of an error
                try {
                    
                    // check whether the slot from the group is already what we want.
                    // any slot that is local, or where the assignment was unconstrained is good!
                    if (slotFromGroup != null && slotFromGroup.getLocality() != Locality.NON_LOCAL) { //如果可以找到合适的
                        updateLocalityCounters(slotFromGroup, vertex);
                        return slotFromGroup; //已经找到合适的slot，返回
                    }
                    
                    // the group did not have a local slot for us. see if we can one (or a better one)
                    newSlot = getNewSlotForSharingGroup(vertex, locations, assignment, constraint, localOnly); //试图申请一个新的slot

                    
                    if (slotFromGroup == null || !slotFromGroup.isAlive() || newSlot.getLocality() == Locality.LOCAL) {
                        // if there is no slot from the group, or the new slot is local,
                        // then we use the new slot
                        if (slotFromGroup != null) {
                            slotFromGroup.releaseSlot();
                        }
                        toUse = newSlot;
                    }
                    else {
                        // both are available and usable. neither is local. in that case, we may
                        // as well use the slot from the sharing group, to minimize the number of
                        // instances that the job occupies
                        newSlot.releaseSlot();
                        toUse = slotFromGroup;
                    }

                    // if this is the first slot for the co-location constraint, we lock
                    // the location, because we are going to use that slot
                    if (constraint != null && !constraint.isAssigned()) {
                        constraint.lockLocation();
                    }
                    
                    updateLocalityCounters(toUse, vertex);
                }
                
                return toUse; //返回申请的slot
            }
            else { //如果不是共享slot，比较简单
                
                // 2) === schedule without hints and sharing ===
                
                SimpleSlot slot = getFreeSlotForTask(vertex, preferredLocations, forceExternalLocation); //直接申请slot
                if (slot != null) {
                    updateLocalityCounters(slot, vertex);
                    return slot; //申请到了就返回slot
                }
                else {
                    // no resource available now, so queue the request
                    if (queueIfNoResource) { //如果可以queue
                        CompletableFuture<SimpleSlot> future = new FlinkCompletableFuture<>();
                        this.taskQueue.add(new QueuedTask(task, future)); //把task缓存起来，并把future对象返回，表示异步申请
                        return future;
                    }
                }
            }
        }
    }

/**
     * Gets a suitable instance to schedule the vertex execution to.
     * <p>
     * NOTE: This method does is not thread-safe, it needs to be synchronized by the caller.
     * 
     * @param vertex The task to run. 
     * @return The instance to run the vertex on, it {@code null}, if no instance is available.
     */
    protected SimpleSlot getFreeSlotForTask(ExecutionVertex vertex,
                                            Iterable<TaskManagerLocation> requestedLocations,
                                            boolean localOnly) {
        // we need potentially to loop multiple times, because there may be false positives
        // in the set-with-available-instances
        while (true) {
            Pair<Instance, Locality> instanceLocalityPair = findInstance(requestedLocations, localOnly); //找到分配slot的并符合location约束的instance

            if (instanceLocalityPair == null){
                return null; //没有合适的instance，分配失败
            }

            Instance instanceToUse = instanceLocalityPair.getLeft();
            Locality locality = instanceLocalityPair.getRight();

            try {
                SimpleSlot slot = instanceToUse.allocateSimpleSlot(vertex.getJobId()); //从instance分配出slot
                
                // if the instance has further available slots, re-add it to the set of available resources.
                if (instanceToUse.hasResourcesAvailable()) { //如果这个实例还有resources，放入instancesWithAvailableResources，下次可以继续分配
                    this.instancesWithAvailableResources.put(instanceToUse.getTaskManagerID(), instanceToUse);
                }
                
                if (slot != null) {
                    slot.setLocality(locality);
                    return slot; //成功就返回slot
                }
            }
            catch (InstanceDiedException e) {
                // the instance died it has not yet been propagated to this scheduler
                // remove the instance from the set of available instances
                removeInstance(instanceToUse);
            }
            
            // if we failed to get a slot, fall through the loop
        }
    }

/**
     * Tries to find a requested instance. If no such instance is available it will return a non-
     * local instance. If no such instance exists (all slots occupied), then return null.
     * 
     * <p><b>NOTE:</b> This method is not thread-safe, it needs to be synchronized by the caller.</p>
     *
     * @param requestedLocations The list of preferred instances. May be null or empty, which indicates that
     *                           no locality preference exists.   
     * @param localOnly Flag to indicate whether only one of the exact local instances can be chosen.  
     */
    private Pair<Instance, Locality> findInstance(Iterable<TaskManagerLocation> requestedLocations, boolean localOnly) {
        
        // drain the queue of newly available instances
        while (this.newlyAvailableInstances.size() > 0) { //把newlyAvailableInstances新加到instancesWithAvailableResources
            Instance queuedInstance = this.newlyAvailableInstances.poll();
            if (queuedInstance != null) {
                this.instancesWithAvailableResources.put(queuedInstance.getTaskManagerID(), queuedInstance);
            }
        }
        
        // if nothing is available at all, return null
        if (this.instancesWithAvailableResources.isEmpty()) { //如果没有instancesWithAvailableResources，直接返回失败
            return null;
        }

        Iterator<TaskManagerLocation> locations = requestedLocations == null ? null : requestedLocations.iterator();

        if (locations != null && locations.hasNext()) { //按照locality preference依次找instance
            // we have a locality preference

            while (locations.hasNext()) {
                TaskManagerLocation location = locations.next();
                if (location != null) {
                    Instance instance = instancesWithAvailableResources.remove(location.getResourceID());
                    if (instance != null) {
                        return new ImmutablePair<Instance, Locality>(instance, Locality.LOCAL);
                    }
                }
            }
            
            // no local instance available
            if (localOnly) {
                return null;
            }
            else {
                // take the first instance from the instances with resources
                Iterator<Instance> instances = instancesWithAvailableResources.values().iterator();
                Instance instanceToUse = instances.next();
                instances.remove();

                return new ImmutablePair<>(instanceToUse, Locality.NON_LOCAL);
            }
        }
        else {
            // no location preference, so use some instance
            Iterator<Instance> instances = instancesWithAvailableResources.values().iterator();
            Instance instanceToUse = instances.next();
            instances.remove();

            return new ImmutablePair<>(instanceToUse, Locality.UNCONSTRAINED);
        }
    }

@Override
public void newInstanceAvailable(Instance instance) {
    
    // synchronize globally for instance changes
    synchronized (this.globalLock) {
        
        try {
            // make sure we get notifications about slots becoming available
            instance.setSlotAvailabilityListener(this); //将Scheduler设为Instance的SlotAvailabilityListener
            
            // store the instance in the by-host-lookup
            String instanceHostName = instance.getTaskManagerLocation().getHostname();
            Set<Instance> instanceSet = allInstancesByHost.get(instanceHostName);
            if (instanceSet == null) {
                instanceSet = new HashSet<Instance>();
                allInstancesByHost.put(instanceHostName, instanceSet);
            }
            instanceSet.add(instance);

            // add it to the available resources and let potential waiters know
            this.instancesWithAvailableResources.put(instance.getTaskManagerID(), instance); //放入instancesWithAvailableResources

            // add all slots as available
            for (int i = 0; i < instance.getNumberOfAvailableSlots(); i++) {
                newSlotAvailable(instance);
            }
        }
    }
}

    @Override
    public void newSlotAvailable(final Instance instance) {
        
        // WARNING: The asynchrony here is necessary, because  we cannot guarantee the order
        // of lock acquisition (global scheduler, instance) and otherwise lead to potential deadlocks:
        // 
        // -> The scheduler needs to grab them (1) global scheduler lock
        //                                     (2) slot/instance lock
        // -> The slot releasing grabs (1) slot/instance (for releasing) and
        //                             (2) scheduler (to check whether to take a new task item
        // 
        // that leads with a high probability to deadlocks, when scheduling fast

        this.newlyAvailableInstances.add(instance); //加入到newlyAvailableInstances

        Futures.future(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                handleNewSlot(); //异步的处理queue中的task，当有新的slot要把queue中的task执行掉
                return null;
            }
        }, executionContext);
    }

    private void notifyNewInstance(Instance instance) {
        synchronized (this.instanceListeners) {
            for (InstanceListener listener : this.instanceListeners) {
                try {
                    listener.newInstanceAvailable(instance);
                }
                catch (Throwable t) {
                    LOG.error("Notification of new instance availability failed.", t);
                }
            }
        }
    }

case msg @ RegisterTaskManager(
      resourceId,
      connectionInfo,
      hardwareInformation,
      numberOfSlots) =>

  val taskManager = sender()

  currentResourceManager match {
    case Some(rm) => //如果有resourceManager
      val future = (rm ? decorateMessage(new NotifyResourceStarted(msg.resourceId)))(timeout) //通知ResourceMananger，某个resource已经成功启动
  }

  // ResourceManager is told about the resource, now let's try to register TaskManager
  if (instanceManager.isRegistered(resourceId)) { //如果已经注册过
    val instanceID = instanceManager.getRegisteredInstance(resourceId).getId

    taskManager ! decorateMessage(
      AlreadyRegistered(
        instanceID,
        libraryCacheManager.getBlobServerPort))
  } else { //新的resource
    try {
      val actorGateway = new AkkaActorGateway(taskManager, leaderSessionID.orNull)
      val taskManagerGateway = new ActorTaskManagerGateway(actorGateway)

      val instanceID = instanceManager.registerTaskManager( //向InstanceManager注册该TaskManager
        taskManagerGateway,
        connectionInfo,
        hardwareInformation,
        numberOfSlots)

      taskManagerMap.put(taskManager, instanceID) //在jobManager里面记录该taskManager

      taskManager ! decorateMessage(
        AcknowledgeRegistration(instanceID, libraryCacheManager.getBlobServerPort)) //通知taskManager完成注册

      // to be notified when the taskManager is no longer reachable
      context.watch(taskManager)
    }
  }