/** * * <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;    }

@Overridepublic 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);        }    }

@Overridepublic 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)    }  }