【Flink】Flink 1.12.2 SlotManager

冷不防 2022-09-03 12:15 90阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70]

# 1.概述 #

转载：[Flink 1.12.2 源码浅析 :SlotManager][Flink 1.12.2 _ _SlotManager]

`SlotManager` 位于`ResourceManager`中.

SlotManager 负责维护所有已注册的任务管理器slot、它们的分配和所有挂起的slot请求的视图。

无论何时注册新slot或释放分配的slot，它都会尝试执行另一个挂起的slot请求。 每当没有足够的可用slot时，slot管理器将通过\{@link ResourceActions\#allocateResource（WorkerResourceSpec）\}通知资源管理器。

为了释放资源并避免资源泄漏，空闲任务管理器（其slot当前未使用的任务管理器）和挂起的slot请求分别超时，从而触发其释放和失败。

# 二 .SlotManager #

## 2.1. 介绍 ##

SlotManager 是一个接口类,定义了ResourceManager中管理Slot的相关操作.  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 1]

## 2.2. 接口清单 ##

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>start</td> 
   <td>启动</td> 
  </tr> 
  <tr> 
   <td>suspend</td> 
   <td>挂起</td> 
  </tr> 
  <tr> 
   <td>int getNumberRegisteredSlots()</td> 
   <td>获取注册slot数量</td> 
  </tr> 
  <tr> 
   <td>int getNumberRegisteredSlotsOf(InstanceID instanceId)</td> 
   <td>根据InstanceID获取slot数量</td> 
  </tr> 
  <tr> 
   <td>int getNumberFreeSlots()</td> 
   <td>获取空闲slot数量</td> 
  </tr> 
  <tr> 
   <td>int getNumberFreeSlotsOf(InstanceID instanceId)</td> 
   <td>根据InstanceID获取空闲slot数量</td> 
  </tr> 
  <tr> 
   <td>Map&lt;WorkerResourceSpec, Integer&gt; getRequiredResources();</td> 
   <td>获取从{@link ResourceActions}请求的尚未完成的workers SlotManager的数量。</td> 
  </tr> 
  <tr> 
   <td>ResourceProfile getRegisteredResource();</td> 
   <td>获取注册的Resource</td> 
  </tr> 
  <tr> 
   <td>ResourceProfile getRegisteredResourceOf(InstanceID instanceID);</td> 
   <td>根据InstanceId 获取注册的Resource</td> 
  </tr> 
  <tr> 
   <td>ResourceProfile getFreeResource();</td> 
   <td>获取空闲Resource</td> 
  </tr> 
  <tr> 
   <td>ResourceProfile getFreeResourceOf(InstanceID instanceID);</td> 
   <td>根据InstanceID获取空闲的Resource</td> 
  </tr> 
  <tr> 
   <td>int getNumberPendingSlotRequests();</td> 
   <td>挂起的SlotRequests数量</td> 
  </tr> 
  <tr> 
   <td>void processResourceRequirements(ResourceRequirements resourceRequirements);</td> 
   <td>通知slot manager关于 job需要的资源信息</td> 
  </tr> 
  <tr> 
   <td>boolean registerSlotRequest(SlotRequest slotRequest)</td> 
   <td>发送slot请求.</td> 
  </tr> 
  <tr> 
   <td>boolean unregisterSlotRequest(AllocationID allocationId)</td> 
   <td>取消发送slot请求.</td> 
  </tr> 
  <tr> 
   <td>registerTaskManager</td> 
   <td>注册TaskManager</td> 
  </tr> 
  <tr> 
   <td>unregisterTaskManager</td> 
   <td>取消注册TaskManager</td> 
  </tr> 
  <tr> 
   <td>boolean reportSlotStatus(InstanceID instanceId, SlotReport slotReport);</td> 
   <td>报告 SlotStatus</td> 
  </tr> 
  <tr> 
   <td>void freeSlot(SlotID slotId, AllocationID allocationId);</td> 
   <td>释放Slot</td> 
  </tr> 
  <tr> 
   <td>void setFailUnfulfillableRequest(boolean failUnfulfillableRequest);</td> 
   <td>设置失败未实现的请求.</td> 
  </tr> 
 </tbody> 
</table>

# 三 .SlotManagerImpl #

SlotManagerImpl是SlotManager接口的实现类.

## 3.1. 属性清单 ##

/** * 所有当前可用slot的索引。 * Map for all registered slots. */
        private final HashMap<SlotID, TaskManagerSlot> slots;
    
        /** Index of all currently free slots. */
        private final LinkedHashMap<SlotID, TaskManagerSlot> freeSlots;
    
        /** * 所有当前注册的任务管理器。 * All currently registered task managers. */
        private final HashMap<InstanceID, TaskManagerRegistration> taskManagerRegistrations;
    
        /** * 用于请求重复数据消除的已完成和活动分配的映射。 * Map of fulfilled and active allocations for request deduplication purposes. * */
        private final HashMap<AllocationID, SlotID> fulfilledSlotRequests;
    
        /** * 挂起/未完成的slot分配请求的映射 * Map of pending/unfulfilled slot allocation requests. */
        private final HashMap<AllocationID, PendingSlotRequest> pendingSlotRequests;
    
        
        
        /** Scheduled executor for timeouts. */
        private final ScheduledExecutor scheduledExecutor;
    
        /** Timeout for slot requests to the task manager. */
        private final Time taskManagerRequestTimeout;
    
        /** Timeout after which an allocation is discarded. */
        private final Time slotRequestTimeout;
    
        /** Timeout after which an unused TaskManager is released. */
        private final Time taskManagerTimeout;
    
        private final HashMap<TaskManagerSlotId, PendingTaskManagerSlot> pendingSlots;
    
        private final SlotMatchingStrategy slotMatchingStrategy;
    
        /** ResourceManager's id. */
        private ResourceManagerId resourceManagerId;
    
        /** Executor for future callbacks which have to be "synchronized". */
        private Executor mainThreadExecutor;
    
        /** Callbacks for resource (de-)allocations. */
        private ResourceActions resourceActions;
    
        private ScheduledFuture<?> taskManagerTimeoutsAndRedundancyCheck;
    
        private ScheduledFuture<?> slotRequestTimeoutCheck;
    
        /** True iff the component has been started. */
        private boolean started;
    
        /** * Release task executor only when each produced result partition is either consumed or failed. */
        private final boolean waitResultConsumedBeforeRelease;
    
        /** Defines the max limitation of the total number of slots. */
        private final int maxSlotNum;
    
        /** Defines the number of redundant taskmanagers. */
        private final int redundantTaskManagerNum;
    
        /** * If true, fail unfulfillable slot requests immediately. Otherwise, allow unfulfillable request * to pend. A slot request is considered unfulfillable if it cannot be fulfilled by neither a * slot that is already registered (including allocated ones) nor a pending slot that the {@link * ResourceActions} can allocate. */
        private boolean failUnfulfillableRequest = true;
    
        /** The default resource spec of workers to request. */
        private final WorkerResourceSpec defaultWorkerResourceSpec;
    
        private final int numSlotsPerWorker;
    
        private final ResourceProfile defaultSlotResourceProfile;
    
        private final SlotManagerMetricGroup slotManagerMetricGroup;

## 3.2. 构造方法 ##

构造方法就是普通的赋值&初始化的过程…

public SlotManagerImpl(
                ScheduledExecutor scheduledExecutor,
                SlotManagerConfiguration slotManagerConfiguration,
                SlotManagerMetricGroup slotManagerMetricGroup) { 
    
            this.scheduledExecutor = Preconditions.checkNotNull(scheduledExecutor);
    
            Preconditions.checkNotNull(slotManagerConfiguration);
            this.slotMatchingStrategy = slotManagerConfiguration.getSlotMatchingStrategy();
            this.taskManagerRequestTimeout = slotManagerConfiguration.getTaskManagerRequestTimeout();
            this.slotRequestTimeout = slotManagerConfiguration.getSlotRequestTimeout();
            this.taskManagerTimeout = slotManagerConfiguration.getTaskManagerTimeout();
            this.waitResultConsumedBeforeRelease =
                    slotManagerConfiguration.isWaitResultConsumedBeforeRelease();
            this.defaultWorkerResourceSpec = slotManagerConfiguration.getDefaultWorkerResourceSpec();
            this.numSlotsPerWorker = slotManagerConfiguration.getNumSlotsPerWorker();
            this.defaultSlotResourceProfile =
                    generateDefaultSlotResourceProfile(defaultWorkerResourceSpec, numSlotsPerWorker);
            this.slotManagerMetricGroup = Preconditions.checkNotNull(slotManagerMetricGroup);
            this.maxSlotNum = slotManagerConfiguration.getMaxSlotNum();
            this.redundantTaskManagerNum = slotManagerConfiguration.getRedundantTaskManagerNum();
    
            slots = new HashMap<>(16);
            freeSlots = new LinkedHashMap<>(16);
            taskManagerRegistrations = new HashMap<>(4);
            fulfilledSlotRequests = new HashMap<>(16);
            pendingSlotRequests = new HashMap<>(16);
            pendingSlots = new HashMap<>(16);
    
            resourceManagerId = null;
            resourceActions = null;
            mainThreadExecutor = null;
            taskManagerTimeoutsAndRedundancyCheck = null;
            slotRequestTimeoutCheck = null;
    
            started = false;
        }

## 3.3. 方法详单 ##

### 3.3.1. start ###

启动

/** * Starts the slot manager with the given leader id and resource manager actions. * * @param newResourceManagerId to use for communication with the task managers * @param newMainThreadExecutor to use to run code in the ResourceManager's main thread * @param newResourceActions to use for resource (de-)allocations */
        @Override
        public void start(
                ResourceManagerId newResourceManagerId,
                Executor newMainThreadExecutor,
                ResourceActions newResourceActions) { 
            LOG.info("Starting the SlotManager.");
    
            this.resourceManagerId = Preconditions.checkNotNull(newResourceManagerId);
            mainThreadExecutor = Preconditions.checkNotNull(newMainThreadExecutor);
            resourceActions = Preconditions.checkNotNull(newResourceActions);
    
            started = true;
    
            taskManagerTimeoutsAndRedundancyCheck =
                    scheduledExecutor.scheduleWithFixedDelay(
                            () ->
                                    mainThreadExecutor.execute(
                                            () -> checkTaskManagerTimeoutsAndRedundancy()),
                            0L,
                            taskManagerTimeout.toMilliseconds(),
                            TimeUnit.MILLISECONDS);
    
            slotRequestTimeoutCheck =
                    scheduledExecutor.scheduleWithFixedDelay(
                            () -> mainThreadExecutor.execute(() -> checkSlotRequestTimeouts()),
                            0L,
                            slotRequestTimeout.toMilliseconds(),
                            TimeUnit.MILLISECONDS);
    
            registerSlotManagerMetrics();
        }

## 3.3.2. suspend ##

挂起

/** Suspends the component. This clears the internal state of the slot manager. */
        @Override
        public void suspend() { 
            LOG.info("Suspending the SlotManager.");
    
            // stop the timeout checks for the TaskManagers and the SlotRequests
            if (taskManagerTimeoutsAndRedundancyCheck != null) { 
                taskManagerTimeoutsAndRedundancyCheck.cancel(false);
                taskManagerTimeoutsAndRedundancyCheck = null;
            }
    
            if (slotRequestTimeoutCheck != null) { 
                slotRequestTimeoutCheck.cancel(false);
                slotRequestTimeoutCheck = null;
            }
    
            for (PendingSlotRequest pendingSlotRequest : pendingSlotRequests.values()) { 
                cancelPendingSlotRequest(pendingSlotRequest);
            }
    
            pendingSlotRequests.clear();
    
            ArrayList<InstanceID> registeredTaskManagers =
                    new ArrayList<>(taskManagerRegistrations.keySet());
    
            for (InstanceID registeredTaskManager : registeredTaskManagers) { 
                unregisterTaskManager(
                        registeredTaskManager,
                        new SlotManagerException("The slot manager is being suspended."));
            }
    
            resourceManagerId = null;
            resourceActions = null;
            started = false;
        }

### 3.3.3. int getNumberRegisteredSlots() ###

获取注册slot数量

@Override
        public int getNumberRegisteredSlots() { 
            return slots.size();
        }

### 3.3.4. int getNumberRegisteredSlotsOf(InstanceID instanceId) ###

根据InstanceID获取slot数量

@Override
        public int getNumberRegisteredSlotsOf(InstanceID instanceId) { 
            TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(instanceId);
    
            if (taskManagerRegistration != null) { 
                return taskManagerRegistration.getNumberRegisteredSlots();
            } else { 
                return 0;
            }
        }

### 3.3.5. int getNumberFreeSlots() ###

获取空闲slot数量

@Override
        public int getNumberFreeSlots() { 
            return freeSlots.size();
        }

### 3.3.6. int getNumberFreeSlotsOf(InstanceID instanceId) ###

根据InstanceID获取空闲slot数量

@Override
        public int getNumberFreeSlotsOf(InstanceID instanceId) { 
            TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(instanceId);
    
            if (taskManagerRegistration != null) { 
                return taskManagerRegistration.getNumberFreeSlots();
            } else { 
                return 0;
            }
        }

### 3.3.7. Map<WorkerResourceSpec, Integer> getRequiredResources(); ###

获取从\{@link ResourceActions\}请求的尚未完成的workers SlotManager的数量。

@Override
        public Map<WorkerResourceSpec, Integer> getRequiredResources() { 
            final int pendingWorkerNum =
                    MathUtils.divideRoundUp(pendingSlots.size(), numSlotsPerWorker);
            return pendingWorkerNum > 0
                    ? Collections.singletonMap(defaultWorkerResourceSpec, pendingWorkerNum)
                    : Collections.emptyMap();
        }

### 3.3.8. ResourceProfile getRegisteredResource(); ###

获取注册的Resource

@Override
        public ResourceProfile getRegisteredResource() { 
            return getResourceFromNumSlots(getNumberRegisteredSlots());
        }

### 3.3.9. ResourceProfile getRegisteredResourceOf(InstanceID instanceID); ###

根据InstanceId 获取注册的Resource

@Override
        public ResourceProfile getRegisteredResourceOf(InstanceID instanceID) { 
            return getResourceFromNumSlots(getNumberRegisteredSlotsOf(instanceID));
        }

### 3.3.10. ResourceProfile getFreeResource(); ###

获取空闲Resource

@Override
        public ResourceProfile getFreeResource() { 
            return getResourceFromNumSlots(getNumberFreeSlots());
        }

### 3.3.11. ResourceProfile getFreeResourceOf(InstanceID instanceID); ###

根据InstanceID获取空闲的Resource

@Override
        public ResourceProfile getFreeResourceOf(InstanceID instanceID) { 
            return getResourceFromNumSlots(getNumberFreeSlotsOf(instanceID));
        }

### 3.3.12. int getNumberPendingSlotRequests(); ###

挂起的SlotRequests数量

@Override
        public ResourceProfile getFreeResource() { 
            return getResourceFromNumSlots(getNumberFreeSlots());
        }

### 3.3.13. void processResourceRequirements(ResourceRequirements resourceRequirements); ###

通知slot manager关于 job需要的资源信息

@Override
        public void processResourceRequirements(ResourceRequirements resourceRequirements) { 
            // no-op; don't throw an UnsupportedOperationException here because there are code paths
            // where the resource
            // manager calls this method regardless of whether declarative resource management is used
            // or not
        }

### 3.3.14. boolean registerSlotRequest(SlotRequest slotRequest) ###

发送slot请求.

/** * Requests a slot with the respective resource profile. * * @param slotRequest specifying the requested slot specs * @return true if the slot request was registered; false if the request is a duplicate * @throws ResourceManagerException if the slot request failed (e.g. not enough resources left) */
        @Override
        public boolean registerSlotRequest(SlotRequest slotRequest) throws ResourceManagerException { 
            checkInit();
    
            if (checkDuplicateRequest(slotRequest.getAllocationId())) { 
                LOG.debug(
                        "Ignoring a duplicate slot request with allocation id {}.",
                        slotRequest.getAllocationId());
    
                return false;
            } else { 
                //构建请求
                PendingSlotRequest pendingSlotRequest = new PendingSlotRequest(slotRequest);
    
                pendingSlotRequests.put(slotRequest.getAllocationId(), pendingSlotRequest);
    
                try { 
                    // 请求操作
                    internalRequestSlot(pendingSlotRequest);
                } catch (ResourceManagerException e) { 
                    // requesting the slot failed --> remove pending slot request
                    pendingSlotRequests.remove(slotRequest.getAllocationId());
    
                    throw new ResourceManagerException(
                            "Could not fulfill slot request " + slotRequest.getAllocationId() + '.', e);
                }
    
                return true;
            }
        }
    
    
        /** * Tries to allocate a slot for the given slot request. If there is no slot available, the * resource manager is informed to allocate more resources and a timeout for the request is * registered. * * @param pendingSlotRequest to allocate a slot for * @throws ResourceManagerException if the slot request failed or is unfulfillable */
        private void internalRequestSlot(PendingSlotRequest pendingSlotRequest)
                throws ResourceManagerException { 
            // 获取配置...
            final ResourceProfile resourceProfile = pendingSlotRequest.getResourceProfile();
    
            OptionalConsumer.of(findMatchingSlot(resourceProfile))
                    .ifPresent(taskManagerSlot -> { 
                        // taskManagerSlot 存在操作
                        allocateSlot(taskManagerSlot, pendingSlotRequest);
                    })
                    .ifNotPresent(() -> { 
                                // taskManagerSlot 不存在操作 ==> 启动 TaskManager
                                fulfillPendingSlotRequestWithPendingTaskManagerSlot( pendingSlotRequest);
                            });
        }

### 3.3.15. boolean unregisterSlotRequest(AllocationID allocationId) ###

取消发送slot请求.

/** * Cancels and removes a pending slot request with the given allocation id. If there is no such * pending request, then nothing is done. * * @param allocationId identifying the pending slot request * @return True if a pending slot request was found; otherwise false */
        @Override
        public boolean unregisterSlotRequest(AllocationID allocationId) { 
            checkInit();
    
            PendingSlotRequest pendingSlotRequest = pendingSlotRequests.remove(allocationId);
    
            if (null != pendingSlotRequest) { 
                LOG.debug("Cancel slot request {}.", allocationId);
    
                cancelPendingSlotRequest(pendingSlotRequest);
    
                return true;
            } else { 
                LOG.debug(
                        "No pending slot request with allocation id {} found. Ignoring unregistration request.",
                        allocationId);
    
                return false;
            }
        }

### 3.3.16. registerTaskManager ###

注册TaskManager

/** * * 在 slot manager中注册一个新的task manager * * 从而是 task managers slots 可以被感知/调度 * * Registers a new task manager at the slot manager. * This will make the task managers slots known and, thus, available for allocation. * * @param taskExecutorConnection for the new task manager * @param initialSlotReport for the new task manager * @return True if the task manager has not been registered before and is registered * successfully; otherwise false */
        @Override
        public boolean registerTaskManager(
                final TaskExecutorConnection taskExecutorConnection, SlotReport initialSlotReport) { 
    
            // 初始化检查
            // The slot manager has not been started.
            checkInit();
    
            LOG.debug(
                    "Registering TaskManager {} under {} at the SlotManager.",
                    taskExecutorConnection.getResourceID().getStringWithMetadata(),
                    taskExecutorConnection.getInstanceID());
    
            // 我们通过任务管理器的实例id来识别它们
            // we identify task managers by their instance id
    
            if (taskManagerRegistrations.containsKey(taskExecutorConnection.getInstanceID())) { 
    
                // 之间已经连接过, 直接报搞slot的状态.
                reportSlotStatus(taskExecutorConnection.getInstanceID(), initialSlotReport);
                return false;
            } else { 
    
                if (isMaxSlotNumExceededAfterRegistration(initialSlotReport)) { 
                    // 是否查过最大的 slot 数量...
                    LOG.info(
                            "The total number of slots exceeds the max limitation {}, release the excess resource.",
                            maxSlotNum);
                    resourceActions.releaseResource(
                            taskExecutorConnection.getInstanceID(),
                            new FlinkException(
                                    "The total number of slots exceeds the max limitation."));
                    return false;
                }
    
                // 第一次注册TaskManager
                // first register the TaskManager
                ArrayList<SlotID> reportedSlots = new ArrayList<>();
    
                for (SlotStatus slotStatus : initialSlotReport) { 
                    reportedSlots.add(slotStatus.getSlotID());
                }
    
                TaskManagerRegistration taskManagerRegistration =
                        new TaskManagerRegistration(taskExecutorConnection, reportedSlots);
    
                taskManagerRegistrations.put(
                        taskExecutorConnection.getInstanceID(), taskManagerRegistration);
    
    
                // next register the new slots
                for (SlotStatus slotStatus : initialSlotReport) { 
    
                    // 开始注册slots
                    registerSlot(
                            slotStatus.getSlotID(),
                            slotStatus.getAllocationID(),
                            slotStatus.getJobID(),
                            slotStatus.getResourceProfile(),
                            taskExecutorConnection);
                }
    
                return true;
            }
        }

### 3.3.17. unregisterTaskManager ###

取消注册TaskManager

@Override
        public boolean unregisterTaskManager(InstanceID instanceId, Exception cause) { 
            checkInit();
    
            LOG.debug("Unregister TaskManager {} from the SlotManager.", instanceId);
    
            TaskManagerRegistration taskManagerRegistration =
                    taskManagerRegistrations.remove(instanceId);
    
            if (null != taskManagerRegistration) { 
                internalUnregisterTaskManager(taskManagerRegistration, cause);
    
                return true;
            } else { 
                LOG.debug(
                        "There is no task manager registered with instance ID {}. Ignoring this message.",
                        instanceId);
    
                return false;
            }
        }

### 3.3.18. boolean reportSlotStatus(InstanceID instanceId, SlotReport slotReport); ###

报告 SlotStatus

### 3.3.19. void freeSlot(SlotID slotId, AllocationID allocationId); ###

释放Slot

/** * Reports the current slot allocations for a task manager identified by the given instance id. * * @param instanceId identifying the task manager for which to report the slot status * @param slotReport containing the status for all of its slots * @return true if the slot status has been updated successfully, otherwise false */
        @Override
        public boolean reportSlotStatus(InstanceID instanceId, SlotReport slotReport) { 
            checkInit();
    
            TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(instanceId);
    
            if (null != taskManagerRegistration) { 
                // Received slot report from instance
                // 5ad0a12f8bb03f9d016f8d1e18380563 :
                // SlotReport{ 
                // SlotStatus{ 
                // slotID=container_1615446205104_0025_01_000002_0,
                // allocationID=3755cb8f9962a9a7738db04f2a02084c,
                // jobID=694474d11da6100e82744c9e47e2f511,
                // resourceProfile=ResourceProfile{ 
                // cpuCores=1.0000000000000000,
                // taskHeapMemory=384.000mb (402653174 bytes),
                // taskOffHeapMemory=0 bytes,
                // managedMemory=512.000mb (536870920 bytes),
                // networkMemory=128.000mb (134217730 bytes)
                // }
                // }
                // }.
                LOG.debug("Received slot report from instance {}: {}.", instanceId, slotReport);
    
                for (SlotStatus slotStatus : slotReport) { 
                    updateSlot(
                            slotStatus.getSlotID(),
                            slotStatus.getAllocationID(),
                            slotStatus.getJobID());
                }
    
                return true;
            } else { 
                LOG.debug(
                        "Received slot report for unknown task manager with instance id {}. Ignoring this report.",
                        instanceId);
    
                return false;
            }
        }

### 3.3.20. void setFailUnfulfillableRequest(boolean failUnfulfillableRequest); ###

设置失败未实现的请求.

@Override
        public void setFailUnfulfillableRequest(boolean failUnfulfillableRequest) { 
            if (!this.failUnfulfillableRequest && failUnfulfillableRequest) { 
                // fail unfulfillable pending requests
                Iterator<Map.Entry<AllocationID, PendingSlotRequest>> slotRequestIterator =
                        pendingSlotRequests.entrySet().iterator();
                while (slotRequestIterator.hasNext()) { 
                    PendingSlotRequest pendingSlotRequest = slotRequestIterator.next().getValue();
                    if (pendingSlotRequest.getAssignedPendingTaskManagerSlot() != null) { 
                        continue;
                    }
                    if (!isFulfillableByRegisteredOrPendingSlots(
                            pendingSlotRequest.getResourceProfile())) { 
                        slotRequestIterator.remove();
                        resourceActions.notifyAllocationFailure(
                                pendingSlotRequest.getJobId(),
                                pendingSlotRequest.getAllocationId(),
                                new UnfulfillableSlotRequestException(
                                        pendingSlotRequest.getAllocationId(),
                                        pendingSlotRequest.getResourceProfile()));
                    }
                }
            }
            this.failUnfulfillableRequest = failUnfulfillableRequest;
        }

## 3.4. freeSlot ##

/** * Free the given slot from the given allocation. If the slot is still allocated by the given * allocation id, then the slot will be marked as free and will be subject to new slot requests. * * @param slotId identifying the slot to free * @param allocationId with which the slot is presumably allocated */
        @Override
        public void freeSlot(SlotID slotId, AllocationID allocationId) { 
            checkInit();
    
            TaskManagerSlot slot = slots.get(slotId);
    
            if (null != slot) { 
                if (slot.getState() == SlotState.ALLOCATED) { 
                    if (Objects.equals(allocationId, slot.getAllocationId())) { 
    
                        TaskManagerRegistration taskManagerRegistration =
                                taskManagerRegistrations.get(slot.getInstanceId());
    
                        if (taskManagerRegistration == null) { 
                            throw new IllegalStateException(
                                    "Trying to free a slot from a TaskManager "
                                            + slot.getInstanceId()
                                            + " which has not been registered.");
                        }
    
                        updateSlotState(slot, taskManagerRegistration, null, null);
                    } else { 
                        LOG.debug(
                                "Received request to free slot {} with expected allocation id {}, "
                                        + "but actual allocation id {} differs. Ignoring the request.",
                                slotId,
                                allocationId,
                                slot.getAllocationId());
                    }
                } else { 
                    LOG.debug("Slot {} has not been allocated.", allocationId);
                }
            } else { 
                LOG.debug(
                        "Trying to free a slot {} which has not been registered. Ignoring this message.",
                        slotId);
            }
        }

## 3.5. updateSlotState ##

// 根据slot的状态进行不同的操作...
        private void updateSlotState(
                TaskManagerSlot slot,
                TaskManagerRegistration taskManagerRegistration,
                @Nullable AllocationID allocationId,
                @Nullable JobID jobId) { 
            if (null != allocationId) { 
                switch (slot.getState()) { 
                    // 挂起...
                    case PENDING:
                        // we have a pending slot request --> check whether we have to reject it
                        PendingSlotRequest pendingSlotRequest = slot.getAssignedSlotRequest();
    
                        if (Objects.equals(pendingSlotRequest.getAllocationId(), allocationId)) { 
                            // we can cancel the slot request because it has been fulfilled
                            cancelPendingSlotRequest(pendingSlotRequest);
    
                            // remove the pending slot request, since it has been completed
                            pendingSlotRequests.remove(pendingSlotRequest.getAllocationId());
    
                            slot.completeAllocation(allocationId, jobId);
                        } else { 
                            // we first have to free the slot in order to set a new allocationId
                            slot.clearPendingSlotRequest();
                            // set the allocation id such that the slot won't be considered for the
                            // pending slot request
                            slot.updateAllocation(allocationId, jobId);
    
                            // remove the pending request if any as it has been assigned
                            final PendingSlotRequest actualPendingSlotRequest =
                                    pendingSlotRequests.remove(allocationId);
    
                            if (actualPendingSlotRequest != null) { 
                                cancelPendingSlotRequest(actualPendingSlotRequest);
                            }
    
                            // this will try to find a new slot for the request
                            rejectPendingSlotRequest(
                                    pendingSlotRequest,
                                    new Exception(
                                            "Task manager reported slot "
                                                    + slot.getSlotId()
                                                    + " being already allocated."));
                        }
    
                        taskManagerRegistration.occupySlot();
                        break;
                    case ALLOCATED:
                        // 分配
                        if (!Objects.equals(allocationId, slot.getAllocationId())) { 
                            slot.freeSlot();
                            slot.updateAllocation(allocationId, jobId);
                        }
                        break;
                    case FREE:
                        // 释放..
                        // the slot is currently free --> it is stored in freeSlots
                        freeSlots.remove(slot.getSlotId());
                        slot.updateAllocation(allocationId, jobId);
                        taskManagerRegistration.occupySlot();
                        break;
                }
    
                fulfilledSlotRequests.put(allocationId, slot.getSlotId());
            } else { 
                // no allocation reported
                switch (slot.getState()) { 
                    case FREE:
                        // 处理空闲
                        handleFreeSlot(slot);
                        break;
                    case PENDING:
                        // 不要做任何事情，因为我们还有一个挂起的slot请求
                        // don't do anything because we still have a pending slot request
                        break;
                    case ALLOCATED:
                        // 分配操作...
                        AllocationID oldAllocation = slot.getAllocationId();
                        slot.freeSlot();
                        fulfilledSlotRequests.remove(oldAllocation);
                        taskManagerRegistration.freeSlot();
    
                        handleFreeSlot(slot);
                        break;
                }
            }
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70]: /images/20220829/0fcd45e8dca24b15b69056935bbaf1c8.png
[Flink 1.12.2 _ _SlotManager]: https://zhangboyi.blog.csdn.net/article/details/116229016
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 1]: /images/20220829/7d5f9b516c1c4c059b5f4f72f69d9fb9.png