Flink 源码阅读笔记（6）- 计算资源管理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

class TaskSlot {
	/** Index of the task slot. */
	private final int index;

	/** State of this slot. */
	private TaskSlotState state;

	/** Resource characteristics for this slot. */
	private final ResourceProfile resourceProfile;

	/** Tasks running in this slot. */
	//在一个 Slot 中可能执行多个 Task
	private final Map<ExecutionAttemptID, Task> tasks;

	/** Job id to which the slot has been allocated; null if not allocated. */
	private JobID jobId;

	/** Allocation id of this slot; null if not allocated. */
	private AllocationID allocationId;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class TaskSlotTable implements TimeoutListener<AllocationID> {
/** Timer service used to time out allocated slots. */
	private final TimerService<AllocationID> timerService;

	/** The list of all task slots. */
	private final List<TaskSlot> taskSlots;

	/** Mapping from allocation id to task slot. */
	private final Map<AllocationID, TaskSlot> allocationIDTaskSlotMap;

	/** Mapping from execution attempt id to task and task slot. */
	private final Map<ExecutionAttemptID, TaskSlotMapping> taskSlotMappings;

	/** Mapping from job id to allocated slots for a job. */
	private final Map<JobID, Set<AllocationID>> slotsPerJob;

	/** Interface for slot actions, such as freeing them or timing them out. */
	private SlotActions slotActions;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

class TaskSlotTable {
	public boolean allocateSlot(int index, JobID jobId, AllocationID allocationId, Time slotTimeout) {
		checkInit();
		TaskSlot taskSlot = taskSlots.get(index);
		boolean result = taskSlot.allocate(jobId, allocationId);
		if (result) {
			// update the allocation id to task slot map
			allocationIDTaskSlotMap.put(allocationId, taskSlot);
			// register a timeout for this slot since it's in state allocated
			timerService.registerTimeout(allocationId, slotTimeout.getSize(), slotTimeout.getUnit());
			// add this slot to the set of job slots
			Set<AllocationID> slots = slotsPerJob.get(jobId);
			if (slots == null) {
				slots = new HashSet<>(4);
				slotsPerJob.put(jobId, slots);
			}
			slots.add(allocationId);
		}
		return result;
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

class TaskSlotTable {
    public boolean markSlotActive(AllocationID allocationId) throws SlotNotFoundException {
		checkInit();
		TaskSlot taskSlot = getTaskSlot(allocationId);
		if (taskSlot != null) {
			if (taskSlot.markActive()) {
				// unregister a potential timeout
				LOG.info("Activate slot {}.", allocationId);
				timerService.unregisterTimeout(allocationId);
				return true;
			} else {
				return false;
			}
		} else {
			throw new SlotNotFoundException(allocationId);
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

class TaskExecutor {
	private void establishResourceManagerConnection(
			ResourceManagerGateway resourceManagerGateway,
			ResourceID resourceManagerResourceId,
			InstanceID taskExecutorRegistrationId,
			ClusterInformation clusterInformation) {
		//首次建立连接，向 RM 报告 slot 信息
		final CompletableFuture<Acknowledge> slotReportResponseFuture = resourceManagerGateway.sendSlotReport(
			getResourceID(),
			taskExecutorRegistrationId,
			taskSlotTable.createSlotReport(getResourceID()),
			taskManagerConfiguration.getTimeout());

		//.........
	}

	private class ResourceManagerHeartbeatListener implements HeartbeatListener<Void, SlotReport> {
		//心跳信息
		@Override
		public CompletableFuture<SlotReport> retrievePayload(ResourceID resourceID) {
			return callAsync(
					() -> taskSlotTable.createSlotReport(getResourceID()),
					taskManagerConfiguration.getTimeout());
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

class TaskExecutor {
	@Override
	public CompletableFuture<Acknowledge> requestSlot(
		final SlotID slotId,
		final JobID jobId,
		final AllocationID allocationId,
		final String targetAddress,
		final ResourceManagerId resourceManagerId,
		final Time timeout) {
		try {
			//判断发送请求的 RM 是否是当前 TaskExecutor 注册的
			if (!isConnectedToResourceManager(resourceManagerId)) {
				final String message = String.format("TaskManager is not connected to the resource manager %s.", resourceManagerId);
				log.debug(message);
				throw new TaskManagerException(message);
			}

			if (taskSlotTable.isSlotFree(slotId.getSlotNumber())) {
				//如果 slot 是 Free 状态，则分配 slot
				if (taskSlotTable.allocateSlot(slotId.getSlotNumber(), jobId, allocationId, taskManagerConfiguration.getTimeout())) {
					log.info("Allocated slot for {}.", allocationId);
				} else {
					log.info("Could not allocate slot for {}.", allocationId);
					throw new SlotAllocationException("Could not allocate slot.");
				}
			} else if (!taskSlotTable.isAllocated(slotId.getSlotNumber(), jobId, allocationId)) {
				//如果 slot 已经被分配了，则抛出异常
				final String message = "The slot " + slotId + " has already been allocated for a different job.";
				log.info(message);
				final AllocationID allocationID = taskSlotTable.getCurrentAllocation(slotId.getSlotNumber());
				throw new SlotOccupiedException(message, allocationID, taskSlotTable.getOwningJob(allocationID));
			}

			//将分配的 slot 提供给发送请求的 JobManager
			if (jobManagerTable.contains(jobId)) {
				//如果和对应的 JobManager 已经建立了连接，则向 JobManager 提供 slot
				offerSlotsToJobManager(jobId);
			} else {
				//否则，先和JobManager 建立连接，连接建立后会调用 offerSlotsToJobManager(jobId) 方法
				try {
					jobLeaderService.addJob(jobId, targetAddress);
				} catch (Exception e) {
					// free the allocated slot
					try {
						taskSlotTable.freeSlot(allocationId);
					} catch (SlotNotFoundException slotNotFoundException) {
						// slot no longer existent, this should actually never happen, because we've
						// just allocated the slot. So let's fail hard in this case!
						onFatalError(slotNotFoundException);
					}

					// release local state under the allocation id.
					localStateStoresManager.releaseLocalStateForAllocationId(allocationId);
					// sanity check
					if (!taskSlotTable.isSlotFree(slotId.getSlotNumber())) {
						onFatalError(new Exception("Could not free slot " + slotId));
					}
					throw new SlotAllocationException("Could not add job to job leader service.", e);
				}
			}
		} catch (TaskManagerException taskManagerException) {
			return FutureUtils.completedExceptionally(taskManagerException);
		}

		return CompletableFuture.completedFuture(Acknowledge.get());
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

class TaskExecutor {
	private void offerSlotsToJobManager(final JobID jobId) {
		final JobManagerConnection jobManagerConnection = jobManagerTable.get(jobId);
		if (jobManagerConnection == null) {
			log.debug("There is no job manager connection to the leader of job {}.", jobId);
		} else {
			if (taskSlotTable.hasAllocatedSlots(jobId)) {
				log.info("Offer reserved slots to the leader of job {}.", jobId);
				final JobMasterGateway jobMasterGateway = jobManagerConnection.getJobManagerGateway();
				//获取分配给当前 Job 的 slot，这里只会取得状态为 allocated 的 slot
				final Iterator<TaskSlot> reservedSlotsIterator = taskSlotTable.getAllocatedSlots(jobId);
				final JobMasterId jobMasterId = jobManagerConnection.getJobMasterId();
				final Collection<SlotOffer> reservedSlots = new HashSet<>(2);
				while (reservedSlotsIterator.hasNext()) {
					SlotOffer offer = reservedSlotsIterator.next().generateSlotOffer();
					reservedSlots.add(offer);
				}
				//通过 RPC 调用，将slot提供给 JobMaster
				CompletableFuture<Collection<SlotOffer>> acceptedSlotsFuture = jobMasterGateway.offerSlots(
					getResourceID(),
					reservedSlots,
					taskManagerConfiguration.getTimeout());

				acceptedSlotsFuture.whenCompleteAsync(
					(Iterable<SlotOffer> acceptedSlots, Throwable throwable) -> {
						if (throwable != null) {
							//超时，则重试
							if (throwable instanceof TimeoutException) {
								log.info("Slot offering to JobManager did not finish in time. Retrying the slot offering.");
								// We ran into a timeout. Try again.
								offerSlotsToJobManager(jobId);
							} else {
								log.warn("Slot offering to JobManager failed. Freeing the slots " +
									"and returning them to the ResourceManager.", throwable);
								// 发生异常，则释放所有的 slot
								for (SlotOffer reservedSlot: reservedSlots) {
									freeSlotInternal(reservedSlot.getAllocationId(), throwable);
								}
							}
						} else {
							//调用成功
							// check if the response is still valid
							if (isJobManagerConnectionValid(jobId, jobMasterId)) {
								// mark accepted slots active
								//对于被 JobMaster 确认接受的 slot， 标记为 Active 状态
								for (SlotOffer acceptedSlot : acceptedSlots) {
									try {
										if (!taskSlotTable.markSlotActive(acceptedSlot.getAllocationId())) {
											// the slot is either free or releasing at the moment
											final String message = "Could not mark slot " + jobId + " active.";
											log.debug(message);
											jobMasterGateway.failSlot(
												getResourceID(),
												acceptedSlot.getAllocationId(),
												new FlinkException(message));
										}
									} catch (SlotNotFoundException e) {
										final String message = "Could not mark slot " + jobId + " active.";
										jobMasterGateway.failSlot(
											getResourceID(),
											acceptedSlot.getAllocationId(),
											new FlinkException(message));
									}

									reservedSlots.remove(acceptedSlot);
								}

								final Exception e = new Exception("The slot was rejected by the JobManager.");
								//释放剩余没有被接受的 slot
								for (SlotOffer rejectedSlot : reservedSlots) {
									freeSlotInternal(rejectedSlot.getAllocationId(), e);
								}
							} else {
								// discard the response since there is a new leader for the job
								log.debug("Discard offer slot response since there is a new leader " +
									"for the job {}.", jobId);
							}
						}
					},
					getMainThreadExecutor());
			} else {
				log.debug("There are no unassigned slots for the job {}.", jobId);
			}
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

class TaskExecutor {
	@Override
	public CompletableFuture<Acknowledge> freeSlot(AllocationID allocationId, Throwable cause, Time timeout) {
		freeSlotInternal(allocationId, cause);
		return CompletableFuture.completedFuture(Acknowledge.get());
	}

	private void freeSlotInternal(AllocationID allocationId, Throwable cause) {
		checkNotNull(allocationId);
		try {
			final JobID jobId = taskSlotTable.getOwningJob(allocationId);
			//尝试释放 allocationId 绑定的 slot
			final int slotIndex = taskSlotTable.freeSlot(allocationId, cause);

			if (slotIndex != -1) {
				//成功释放 slot
				if (isConnectedToResourceManager()) {
					//告知 ResourceManager 当前 slot 可用
					// the slot was freed. Tell the RM about it
					ResourceManagerGateway resourceManagerGateway = establishedResourceManagerConnection.getResourceManagerGateway();
					resourceManagerGateway.notifySlotAvailable(
						establishedResourceManagerConnection.getTaskExecutorRegistrationId(),
						new SlotID(getResourceID(), slotIndex),
						allocationId);
				}
				if (jobId != null) {
					// 如果和 allocationID 绑定的 Job 已经没有分配的 slot 了，那么可以断开和 JobMaster 的连接了
					// check whether we still have allocated slots for the same job
					if (taskSlotTable.getAllocationIdsPerJob(jobId).isEmpty()) {
						// we can remove the job from the job leader service
						try {
							jobLeaderService.removeJob(jobId);
						} catch (Exception e) {
							log.info("Could not remove job {} from JobLeaderService.", jobId, e);
						}

						closeJobManagerConnection(
							jobId,
							new FlinkException("TaskExecutor " + getAddress() +
								" has no more allocated slots for job " + jobId + '.'));
					}
				}
			}
		} catch (SlotNotFoundException e) {
			log.debug("Could not free slot for allocation id {}.", allocationId, e);
		}
		localStateStoresManager.releaseLocalStateForAllocationId(allocationId);
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

class SlotManager {
	/** 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;

	/** Map of pending/unfulfilled slot allocation requests. */
	private final HashMap<AllocationID, PendingSlotRequest> pendingSlotRequests;

	// 资源不足的时候会通过 ResourceActions#allocateResource(ResourceProfile) 申请新的资源（可能启动新的 TaskManager，也可能什么也不做），
	// 这些新申请的资源会被封装为 PendingTaskManagerSlot
	private final HashMap<TaskManagerSlotId, PendingTaskManagerSlot> pendingSlots;

	/** ResourceManager's id. */
	private ResourceManagerId resourceManagerId;

	/** Callbacks for resource (de-)allocations. */
	private ResourceActions resourceActions;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

class SlotManager {
	public void registerTaskManager(final TaskExecutorConnection taskExecutorConnection, SlotReport initialSlotReport) {
		checkInit();
		LOG.debug("Registering TaskManager {} under {} at the SlotManager.", taskExecutorConnection.getResourceID(), taskExecutorConnection.getInstanceID());
		// we identify task managers by their instance id
		if (taskManagerRegistrations.containsKey(taskExecutorConnection.getInstanceID())) {
			reportSlotStatus(taskExecutorConnection.getInstanceID(), initialSlotReport);
		} else {
			// 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);

			// 依次注册所有的 slot
			// next register the new slots
			for (SlotStatus slotStatus : initialSlotReport) {
				registerSlot(
					slotStatus.getSlotID(),
					slotStatus.getAllocationID(),
					slotStatus.getJobID(),
					slotStatus.getResourceProfile(),
					taskExecutorConnection);
			}
		}
	}

	//注册一个slot
	private void registerSlot(
			SlotID slotId,
			AllocationID allocationId,
			JobID jobId,
			ResourceProfile resourceProfile,
			TaskExecutorConnection taskManagerConnection) {
		if (slots.containsKey(slotId)) {
			// remove the old slot first
			removeSlot(slotId);
		}

		//创建一个 TaskManagerSlot 对象，并加入 slots 中
		final TaskManagerSlot slot = createAndRegisterTaskManagerSlot(slotId, resourceProfile, taskManagerConnection);

		final PendingTaskManagerSlot pendingTaskManagerSlot;
		if (allocationId == null) {
			//这个 slot 还没有被分配，则找到和当前 slot 的计算资源相匹配的 PendingTaskManagerSlot
			pendingTaskManagerSlot = findExactlyMatchingPendingTaskManagerSlot(resourceProfile);
		} else {
			//这个 slot 已经被分配了
			pendingTaskManagerSlot = null;
		}
		if (pendingTaskManagerSlot == null) {
			//两种可能： 1）slot已经被分配了 2）没有匹配的 PendingTaskManagerSlot
			updateSlot(slotId, allocationId, jobId);
		} else {
			// 新注册的 slot 能够满足 PendingTaskManagerSlot 的要求
			pendingSlots.remove(pendingTaskManagerSlot.getTaskManagerSlotId());
			final PendingSlotRequest assignedPendingSlotRequest = pendingTaskManagerSlot.getAssignedPendingSlotRequest();
			// PendingTaskManagerSlot 可能有关联的 PedningSlotRequest
			if (assignedPendingSlotRequest == null) {
				//没有关联的 PedningSlotRequest，则将 slot 是 Free 状态
				handleFreeSlot(slot);
			} else {
				//有关联的 PedningSlotRequest，则这个 request 可以被满足，分配 slot
				assignedPendingSlotRequest.unassignPendingTaskManagerSlot();
				allocateSlot(slot, assignedPendingSlotRequest);
			}
		}
	}

	private void handleFreeSlot(TaskManagerSlot freeSlot) {
		Preconditions.checkState(freeSlot.getState() == TaskManagerSlot.State.FREE);
		//先查找是否有能够满足的 PendingSlotRequest
		PendingSlotRequest pendingSlotRequest = findMatchingRequest(freeSlot.getResourceProfile());
		if (null != pendingSlotRequest) {
			//如果有匹配的 PendingSlotRequest，则分配slot
			allocateSlot(freeSlot, pendingSlotRequest);
		} else {
			freeSlots.put(freeSlot.getSlotId(), freeSlot);
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

class SlotManager {
	public boolean registerSlotRequest(SlotRequest slotRequest) throws SlotManagerException {
		checkInit();
		if (checkDuplicateRequest(slotRequest.getAllocationId())) {
			LOG.debug("Ignoring a duplicate slot request with allocation id {}.", slotRequest.getAllocationId());
			return false;
		} else {
			//将请求封装为 PendingSlotRequest
			PendingSlotRequest pendingSlotRequest = new PendingSlotRequest(slotRequest);
			pendingSlotRequests.put(slotRequest.getAllocationId(), pendingSlotRequest);
			try {
				//执行请求分配slot的逻辑
				internalRequestSlot(pendingSlotRequest);
			} catch (ResourceManagerException e) {
				// requesting the slot failed --> remove pending slot request
				pendingSlotRequests.remove(slotRequest.getAllocationId());
				throw new SlotManagerException("Could not fulfill slot request " + slotRequest.getAllocationId() + '.', e);
			}
			return true;
		}
	}

	private void internalRequestSlot(PendingSlotRequest pendingSlotRequest) throws ResourceManagerException {
		final ResourceProfile resourceProfile = pendingSlotRequest.getResourceProfile();
		//首先从 FREE 状态的已注册的 slot 中选择符合要求的 slot
		TaskManagerSlot taskManagerSlot = findMatchingSlot(resourceProfile);

		if (taskManagerSlot != null) {
			//找到了符合条件的slot，分配
			allocateSlot(taskManagerSlot, pendingSlotRequest);
		} else {
			//从 PendingTaskManagerSlot 中选择
			Optional<PendingTaskManagerSlot> pendingTaskManagerSlotOptional = findFreeMatchingPendingTaskManagerSlot(resourceProfile);

			//如果连 PendingTaskManagerSlot 中都没有
			if (!pendingTaskManagerSlotOptional.isPresent()) {
				//请求 ResourceManager 分配资源，通过 ResourceActions#allocateResource(ResourceProfile) 回调进行
				pendingTaskManagerSlotOptional = allocateResource(resourceProfile);
			}

			//将 PendingTaskManagerSlot 指派给 PendingSlotRequest
			pendingTaskManagerSlotOptional.ifPresent(pendingTaskManagerSlot -> assignPendingTaskManagerSlot(pendingSlotRequest, pendingTaskManagerSlot));
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

class SlotManager {
	private void allocateSlot(TaskManagerSlot taskManagerSlot, PendingSlotRequest pendingSlotRequest) {
		Preconditions.checkState(taskManagerSlot.getState() == TaskManagerSlot.State.FREE);
		TaskExecutorConnection taskExecutorConnection = taskManagerSlot.getTaskManagerConnection();
		TaskExecutorGateway gateway = taskExecutorConnection.getTaskExecutorGateway();

		final CompletableFuture<Acknowledge> completableFuture = new CompletableFuture<>();
		final AllocationID allocationId = pendingSlotRequest.getAllocationId();
		final SlotID slotId = taskManagerSlot.getSlotId();
		final InstanceID instanceID = taskManagerSlot.getInstanceId();

		//taskManagerSlot 状态变为 PENDING
		taskManagerSlot.assignPendingSlotRequest(pendingSlotRequest);
		pendingSlotRequest.setRequestFuture(completableFuture);

		//如果有 PendingTaskManager 指派给当前 pendingSlotRequest，要先解除关联
		returnPendingTaskManagerSlotIfAssigned(pendingSlotRequest);

		TaskManagerRegistration taskManagerRegistration = taskManagerRegistrations.get(instanceID);
		if (taskManagerRegistration == null) {
			throw new IllegalStateException("Could not find a registered task manager for instance id " +
				instanceID + '.');
		}
		taskManagerRegistration.markUsed();

		// RPC call to the task manager
		// 通过 RPC 调用向 TaskExecutor 请求 slot
		CompletableFuture<Acknowledge> requestFuture = gateway.requestSlot(
			slotId,
			pendingSlotRequest.getJobId(),
			allocationId,
			pendingSlotRequest.getTargetAddress(),
			resourceManagerId,
			taskManagerRequestTimeout);

		//RPC调用的请求完成
		requestFuture.whenComplete(
			(Acknowledge acknowledge, Throwable throwable) -> {
				if (acknowledge != null) {
					completableFuture.complete(acknowledge);
				} else {
					completableFuture.completeExceptionally(throwable);
				}
			});

		//PendingSlotRequest 请求完成的回调函数
		//PendingSlotRequest 请求完成可能是由于上面 RPC 调用完成，也可能是因为 PendingSlotRequest 被取消
		completableFuture.whenCompleteAsync(
			(Acknowledge acknowledge, Throwable throwable) -> {
				try {
					if (acknowledge != null) {
						//如果请求成功，则取消 pendingSlotRequest，并更新 slot 状态 PENDING -> ALLOCATED
						updateSlot(slotId, allocationId, pendingSlotRequest.getJobId());
					} else {
						if (throwable instanceof SlotOccupiedException) {
							//这个 slot 已经被占用了，更新状态
							SlotOccupiedException exception = (SlotOccupiedException) throwable;
							updateSlot(slotId, exception.getAllocationId(), exception.getJobId());
						} else {
							//请求失败，将 pendingSlotRequest 从 TaskManagerSlot 中移除
							removeSlotRequestFromSlot(slotId, allocationId);
						}

						if (!(throwable instanceof CancellationException)) {
							//slot request 请求失败，会进行重试
							handleFailedSlotRequest(slotId, allocationId, throwable);
						} else {
							//主动取消
							LOG.debug("Slot allocation request {} has been cancelled.", allocationId, throwable);
						}
					}
				} catch (Exception e) {
					LOG.error("Error while completing the slot allocation.", e);
				}
			},
			mainThreadExecutor);
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

class SlotManager {
	public boolean unregisterSlotRequest(AllocationID allocationId) {
		checkInit();
		//从 pendingSlotRequests 中移除
		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;
		}
	}

	private void cancelPendingSlotRequest(PendingSlotRequest pendingSlotRequest) {
		CompletableFuture<Acknowledge> request = pendingSlotRequest.getRequestFuture();
		returnPendingTaskManagerSlotIfAssigned(pendingSlotRequest);
		if (null != request) {
			request.cancel(false);
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class SlotManager {
		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;

		//检查 TaskExecutor 是否长时间处于 idle 状态
		taskManagerTimeoutCheck = scheduledExecutor.scheduleWithFixedDelay(
			() -> mainThreadExecutor.execute(
				() -> checkTaskManagerTimeouts()),
			0L,
			taskManagerTimeout.toMilliseconds(),
			TimeUnit.MILLISECONDS);

		//检查 slot request 是否超时
		slotRequestTimeoutCheck = scheduledExecutor.scheduleWithFixedDelay(
			() -> mainThreadExecutor.execute(
				() -> checkSlotRequestTimeouts()),
			0L,
			slotRequestTimeout.toMilliseconds(),
			TimeUnit.MILLISECONDS);
	}
}

1
2
3
4
5
6
7

class SlotManager {
	private void releaseTaskExecutor(InstanceID timedOutTaskManagerId) {
		final FlinkException cause = new FlinkException("TaskExecutor exceeded the idle timeout.");
		LOG.debug("Release TaskExecutor {} because it exceeded the idle timeout.", timedOutTaskManagerId);
		resourceActions.releaseResource(timedOutTaskManagerId, cause);
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class SlotManager {
	private void checkSlotRequestTimeouts() {
		if (!pendingSlotRequests.isEmpty()) {
			long currentTime = System.currentTimeMillis();
			Iterator<Map.Entry<AllocationID, PendingSlotRequest>> slotRequestIterator = pendingSlotRequests.entrySet().iterator();
			while (slotRequestIterator.hasNext()) {
				PendingSlotRequest slotRequest = slotRequestIterator.next().getValue();
				if (currentTime - slotRequest.getCreationTimestamp() >= slotRequestTimeout.toMilliseconds()) {
					slotRequestIterator.remove();

					if (slotRequest.isAssigned()) {
						cancelPendingSlotRequest(slotRequest);
					}

					resourceActions.notifyAllocationFailure(
						slotRequest.getJobId(),
						slotRequest.getAllocationId(),
						new TimeoutException("The allocation could not be fulfilled in time."));
				}
			}
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

interface ResouceManagerGateway {
	CompletableFuture<Acknowledge> requestSlot(
		JobMasterId jobMasterId,
		SlotRequest slotRequest,
		@RpcTimeout Time timeout);

	void cancelSlotRequest(AllocationID allocationID);

	CompletableFuture<Acknowledge> sendSlotReport(
		ResourceID taskManagerResourceId,
		InstanceID taskManagerRegistrationId,
		SlotReport slotReport,
		@RpcTimeout Time timeout);

	void notifySlotAvailable(
		InstanceID instanceId,
		SlotID slotID,
		AllocationID oldAllocationId);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

class ResourceManager {
	private class ResourceActionsImpl implements ResourceActions {
		@Override
		public void releaseResource(InstanceID instanceId, Exception cause) {
			validateRunsInMainThread();
			//释放资源
			ResourceManager.this.releaseResource(instanceId, cause);
		}

		@Override
		public Collection<ResourceProfile> allocateResource(ResourceProfile resourceProfile) {
			validateRunsInMainThread();
			//申请新的资源，具体行为和不同的 ResourceManager 的实现有关。其返回的列表相当于是承诺即将分配的资源
			return startNewWorker(resourceProfile);
		}

		@Override
		public void notifyAllocationFailure(JobID jobId, AllocationID allocationId, Exception cause) {
			validateRunsInMainThread();
			JobManagerRegistration jobManagerRegistration = jobManagerRegistrations.get(jobId);
			if (jobManagerRegistration != null) {
				jobManagerRegistration.getJobManagerGateway().notifyAllocationFailure(allocationId, cause);
			}
		}
	}

	protected void releaseResource(InstanceID instanceId, Exception cause) {
		WorkerType worker = null;

		// TODO: Improve performance by having an index on the instanceId
		for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> entry : taskExecutors.entrySet()) {
			if (entry.getValue().getInstanceID().equals(instanceId)) {
				worker = entry.getValue().getWorker();
				break;
			}
		}

		if (worker != null) {
			//停止对应的worker，具体行为和不同的 ResourceManager 的实现有关
			if (stopWorker(worker)) {
				closeTaskManagerConnection(worker.getResourceID(), cause);
			} else {
				log.debug("Worker {} could not be stopped.", worker.getResourceID());
			}
		} else {
			// unregister in order to clean up potential left over state
			slotManager.unregisterTaskManager(instanceId);
		}
	}

	public abstract Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile);
	public abstract boolean stopWorker(WorkerType worker);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class YarnResourceManager {
	@Override
	public Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile) {
		Preconditions.checkArgument(
			ResourceProfile.UNKNOWN.equals(resourceProfile),
			"The YarnResourceManager does not support custom ResourceProfiles yet. It assumes that all containers have the same resources.");
		//申请 container
		requestYarnContainer();

		return slotsPerWorker;
	}

	@Override
	public boolean stopWorker(final YarnWorkerNode workerNode) {
		final Container container = workerNode.getContainer();
		log.info("Stopping container {}.", container.getId());
		try {
			nodeManagerClient.stopContainer(container.getId(), container.getNodeId());
		} catch (final Exception e) {
			log.warn("Error while calling YARN Node Manager to stop container", e);
		}
		//释放container
		resourceManagerClient.releaseAssignedContainer(container.getId());
		workerNodeMap.remove(workerNode.getResourceID());
		return true;
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

public interface PhysicalSlot extends SlotContext {
	boolean tryAssignPayload(Payload payload);

	/**
	 * Payload which can be assigned to an {@link AllocatedSlot}.
	 */
	interface Payload {
		/**
		 * Releases the payload
		 *
		 * @param cause of the payload release
		 */
		void release(Throwable cause);
	}
}

class AllocatedSlot implements PhysicalSlot {
	/** The ID under which the slot is allocated. Uniquely identifies the slot. */
	private final AllocationID allocationId;

	/** The location information of the TaskManager to which this slot belongs */
	private final TaskManagerLocation taskManagerLocation;

	/** The resource profile of the slot provides */
	private final ResourceProfile resourceProfile;

	/** RPC gateway to call the TaskManager that holds this slot */
	private final TaskManagerGateway taskManagerGateway;

	/** The number of the slot on the TaskManager to which slot belongs. Purely informational. */
	private final int physicalSlotNumber;

	private final AtomicReference<Payload> payloadReference;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

public interface LogicalSlot {
	TaskManagerLocation getTaskManagerLocation();
	TaskManagerGateway getTaskManagerGateway();
	int getPhysicalSlotNumber();

	AllocationID getAllocationId();
	SlotRequestId getSlotRequestId();

	Locality getLocality();

	CompletableFuture<?> releaseSlot(@Nullable Throwable cause);

	@Nullable
	SlotSharingGroupId getSlotSharingGroupId();

	boolean tryAssignPayload(Payload payload);
	Payload getPayload();

	/**
	 * Payload for a logical slot.
	 */
	interface Payload {
		void fail(Throwable cause);
		CompletableFuture<?> getTerminalStateFuture();
	}
}

public class SingleLogicalSlot implements LogicalSlot, PhysicalSlot.Payload {
	private final SlotRequestId slotRequestId;

	private final SlotContext slotContext;

	// null if the logical slot does not belong to a slot sharing group, otherwise non-null
	@Nullable
	private final SlotSharingGroupId slotSharingGroupId;

	// locality of this slot wrt the requested preferred locations
	private final Locality locality;

	// owner of this slot to which it is returned upon release
	private final SlotOwner slotOwner;

	private final CompletableFuture<Void> releaseFuture;

	private volatile State state;

	// LogicalSlot.Payload of this slot
	private volatile Payload payload;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class TaskSlot {
		// every TaskSlot has an associated slot request id
		private final SlotRequestId slotRequestId;

		// all task slots except for the root slots have a group id assigned
		// 除了 root 节点，每个节点都有一个 groupId 用来区分一个 TaskSlot。可能是 JobVertexID，也可能是 CoLocationGroup 的 ID
		@Nullable
		private final AbstractID groupId;

		public boolean contains(AbstractID groupId) {
			return Objects.equals(this.groupId, groupId);
		}

		public abstract void release(Throwable cause);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

public final class MultiTaskSlot extends TaskSlot implements PhysicalSlot.Payload {
		private final Map<AbstractID, TaskSlot> children;

		// the root node has its parent set to null
		@Nullable
		private final MultiTaskSlot parent;

		// underlying allocated slot
		private final CompletableFuture<? extends SlotContext> slotContextFuture;

		// slot request id of the allocated slot
		@Nullable
		private final SlotRequestId allocatedSlotRequestId;

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

public final class SingleTaskSlot extends TaskSlot {
		private final MultiTaskSlot parent;
// future containing a LogicalSlot which is completed once the underlying SlotContext future is completed
		private final CompletableFuture<SingleLogicalSlot> singleLogicalSlotFuture;

		private SingleTaskSlot(
				SlotRequestId slotRequestId,
				AbstractID groupId,
				MultiTaskSlot parent,
				Locality locality) {
			super(slotRequestId, groupId);

			this.parent = Preconditions.checkNotNull(parent);

			Preconditions.checkNotNull(locality);
			singleLogicalSlotFuture = parent.getSlotContextFuture()
				.thenApply(
					(SlotContext slotContext) -> {
						//在父节点被分配了 PhysicalSlot 后，创建 SingleLogicalSlot
						LOG.trace("Fulfill single task slot [{}] with slot [{}].", slotRequestId, slotContext.getAllocationId());
						return new SingleLogicalSlot(
							slotRequestId,
							slotContext,
							slotSharingGroupId,
							locality,
							slotOwner);
					});
		}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

class SlotPoolImpl implements SlotPool {
	/** The book-keeping of all allocated slots. */
	//所有分配给当前 JobManager 的 slots
	private final AllocatedSlots allocatedSlots;

	/** The book-keeping of all available slots. */
	//所有可用的 slots（已经分配给该 JobManager，但还没有装载 payload）
	private final AvailableSlots availableSlots;

	/** All pending requests waiting for slots. */
	//所有处于等待状态的slot request（已经发送请求给 ResourceManager）
	private final DualKeyMap<SlotRequestId, AllocationID, PendingRequest> pendingRequests;

	/** The requests that are waiting for the resource manager to be connected. */
	//处于等待状态的 slot request （还没有发送请求给 ResourceManager，此时没有和 ResourceManager 建立连接）
	private final HashMap<SlotRequestId, PendingRequest> waitingForResourceManager;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

class SlotPoolImpl implements SlotPool {
	//列出当前可用的 slot
	@Override
	public Collection<SlotInfo> getAvailableSlotsInformation() {
		return availableSlots.listSlotInfo();
	}

	//将 allocationID 关联的 slot 分配给 slotRequestId 对应的请求
	@Override
	public Optional<PhysicalSlot> allocateAvailableSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull AllocationID allocationID) {

		componentMainThreadExecutor.assertRunningInMainThread();

		//从 availableSlots 中移除
		AllocatedSlot allocatedSlot = availableSlots.tryRemove(allocationID);
		if (allocatedSlot != null) {
			//加入已分配的映射关系中
			allocatedSlots.add(slotRequestId, allocatedSlot);
			return Optional.of(allocatedSlot);
		} else {
			return Optional.empty();
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

class SlotPoolImpl implements SlotPool {

	//向RM申请新的 slot
	@Override
	public CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull ResourceProfile resourceProfile,
		Time timeout) {

		return requestNewAllocatedSlotInternal(slotRequestId, resourceProfile, timeout)
			.thenApply((Function.identity()));
	}

	private CompletableFuture<AllocatedSlot> requestNewAllocatedSlotInternal(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull ResourceProfile resourceProfile,
		@Nonnull Time timeout) {

		componentMainThreadExecutor.assertRunningInMainThread();

		//构造一个 PendingRequest
		final PendingRequest pendingRequest = new PendingRequest(
			slotRequestId,
			resourceProfile);

		// register request timeout
		FutureUtils
			.orTimeout(
				pendingRequest.getAllocatedSlotFuture(),
				timeout.toMilliseconds(),
				TimeUnit.MILLISECONDS,
				componentMainThreadExecutor)
			.whenComplete(
				(AllocatedSlot ignored, Throwable throwable) -> {
					if (throwable instanceof TimeoutException) {
						//超时处理
						timeoutPendingSlotRequest(slotRequestId);
					}
				});

		if (resourceManagerGateway == null) {
			//如果当前没有和 RM 建立连接，则需要等待 RM 建立连接
			stashRequestWaitingForResourceManager(pendingRequest);
		} else {
			//当前已经和 RM 建立了连接，向 RM 申请slot
			requestSlotFromResourceManager(resourceManagerGateway, pendingRequest);
		}
		return pendingRequest.getAllocatedSlotFuture();
	}

	//如果当前没有和 RM 建立连接，则需要等待 RM 建立连接，加入 waitingForResourceManager
	//一旦和 RM 建立连接，就会向 RM 发送请求
	private void stashRequestWaitingForResourceManager(final PendingRequest pendingRequest) {
		log.info("Cannot serve slot request, no ResourceManager connected. " +
				"Adding as pending request [{}]",  pendingRequest.getSlotRequestId());
		waitingForResourceManager.put(pendingRequest.getSlotRequestId(), pendingRequest);
	}

	//当前已经和 RM 建立了连接，向 RM 申请slot
	private void requestSlotFromResourceManager(
			final ResourceManagerGateway resourceManagerGateway,
			final PendingRequest pendingRequest) {
		//生成一个 AllocationID，后面分配的 slot 通过 AllocationID 区分
		final AllocationID allocationId = new AllocationID();

		//添加到等待处理的请求中
		pendingRequests.put(pendingRequest.getSlotRequestId(), allocationId, pendingRequest);

		pendingRequest.getAllocatedSlotFuture().whenComplete(
			(AllocatedSlot allocatedSlot, Throwable throwable) -> {
				if (throwable != null || !allocationId.equals(allocatedSlot.getAllocationId())) {
					// cancel the slot request if there is a failure or if the pending request has
					// been completed with another allocated slot
					resourceManagerGateway.cancelSlotRequest(allocationId);
				}
			});

		//通过 RPC 调用向 RM 请求 slot，RM 的处理流程在前面已经介绍过
		CompletableFuture<Acknowledge> rmResponse = resourceManagerGateway.requestSlot(
			jobMasterId,
			new SlotRequest(jobId, allocationId, pendingRequest.getResourceProfile(), jobManagerAddress),
			rpcTimeout);

		FutureUtils.whenCompleteAsyncIfNotDone(
			rmResponse,
			componentMainThreadExecutor,
			(Acknowledge ignored, Throwable failure) -> {
				// on failure, fail the request future
				if (failure != null) {
					slotRequestToResourceManagerFailed(pendingRequest.getSlotRequestId(), failure);
				}
			});
	}
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112

class SlotPoolImpl {
	//向 SlotPool 分配 slot，返回已经被接受的 slot 集合。没有被接受的 slot，RM 可以分配给其他 Job。
	@Override
	public Collection<SlotOffer> offerSlots(
			TaskManagerLocation taskManagerLocation,
			TaskManagerGateway taskManagerGateway,
			Collection<SlotOffer> offers) {
		ArrayList<SlotOffer> result = new ArrayList<>(offers.size());
		//SlotPool 可以确定是否接受每一个 slot
		for (SlotOffer offer : offers) {
			if (offerSlot(
				taskManagerLocation,
				taskManagerGateway,
				offer)) {

				result.add(offer);
			}
		}
		return result;
	}

	boolean offerSlot(
			final TaskManagerLocation taskManagerLocation,
			final TaskManagerGateway taskManagerGateway,
			final SlotOffer slotOffer) {

		componentMainThreadExecutor.assertRunningInMainThread();

		// check if this TaskManager is valid
		final ResourceID resourceID = taskManagerLocation.getResourceID();
		final AllocationID allocationID = slotOffer.getAllocationId();

		if (!registeredTaskManagers.contains(resourceID)) {
			log.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}",
					slotOffer.getAllocationId(), taskManagerLocation);
			return false;
		}

		// check whether we have already using this slot
		// 如果当前 slot 关联的 AllocationID 已经在 SlotPool 中出现
		AllocatedSlot existingSlot;
		if ((existingSlot = allocatedSlots.get(allocationID)) != null ||
			(existingSlot = availableSlots.get(allocationID)) != null) {

			// we need to figure out if this is a repeated offer for the exact same slot,
			// or another offer that comes from a different TaskManager after the ResourceManager
			// re-tried the request

			// we write this in terms of comparing slot IDs, because the Slot IDs are the identifiers of
			// the actual slots on the TaskManagers
			// Note: The slotOffer should have the SlotID
			final SlotID existingSlotId = existingSlot.getSlotId();
			final SlotID newSlotId = new SlotID(taskManagerLocation.getResourceID(), slotOffer.getSlotIndex());

			if (existingSlotId.equals(newSlotId)) {
				//这个 slot 在之前已经被 SlotPool 接受了，相当于 TaskExecutor 发送了一个重复的 offer
				log.info("Received repeated offer for slot [{}]. Ignoring.", allocationID);

				// return true here so that the sender will get a positive acknowledgement to the retry
				// and mark the offering as a success
				return true;
			} else {
				//已经有一个其他的 AllocatedSlot 和 这个 AllocationID 关联了，因此不能接受当前的这个 slot
				// the allocation has been fulfilled by another slot, reject the offer so the task executor
				// will offer the slot to the resource manager
				return false;
			}
		}

		//这个 slot 关联的 AllocationID 此前没有出现过

		//新建一个 AllocatedSlot 对象，表示新分配的 slot
		final AllocatedSlot allocatedSlot = new AllocatedSlot(
			allocationID,
			taskManagerLocation,
			slotOffer.getSlotIndex(),
			slotOffer.getResourceProfile(),
			taskManagerGateway);

		// check whether we have request waiting for this slot
		// 检查是否有一个 request 和 这个 AllocationID 关联
		PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID);
		if (pendingRequest != null) {
			// we were waiting for this!
			//有一个pending request 正在等待这个 slot
			allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot);

			//尝试去完成那个等待的请求
			if (!pendingRequest.getAllocatedSlotFuture().complete(allocatedSlot)) {
				// we could not complete the pending slot future --> try to fulfill another pending request
				//失败了
				allocatedSlots.remove(pendingRequest.getSlotRequestId());
				//尝试去满足其他在等待的请求
				tryFulfillSlotRequestOrMakeAvailable(allocatedSlot);
			} else {
				log.debug("Fulfilled slot request [{}] with allocated slot [{}].", pendingRequest.getSlotRequestId(), allocationID);
			}
		}
		else {
			//没有请求在等待这个slot，可能请求已经被满足了
			// we were actually not waiting for this:
			//   - could be that this request had been fulfilled
			//   - we are receiving the slots from TaskManagers after becoming leaders
			//尝试去满足其他在等待的请求
			tryFulfillSlotRequestOrMakeAvailable(allocatedSlot);
		}

		// we accepted the request in any case. slot will be released after it idled for
		// too long and timed out
		return true;
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

class SlotManagerImpl implements SlotPool {
	private void tryFulfillSlotRequestOrMakeAvailable(AllocatedSlot allocatedSlot) {
		Preconditions.checkState(!allocatedSlot.isUsed(), "Provided slot is still in use.");

		//查找和当前 AllocatedSlot 的计算资源相匹配的还在等待的请求
		final PendingRequest pendingRequest = pollMatchingPendingRequest(allocatedSlot);

		if (pendingRequest != null) {
			//如果有匹配的请求，那么将 AllocatedSlot 分配给等待的请求
			log.debug("Fulfilling pending slot request [{}] early with returned slot [{}]",
				pendingRequest.getSlotRequestId(), allocatedSlot.getAllocationId());

			allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot);
			pendingRequest.getAllocatedSlotFuture().complete(allocatedSlot);
		} else {
			//如果没有，那么这个 AllocatedSlot 变成 available 的
			log.debug("Adding returned slot [{}] to available slots", allocatedSlot.getAllocationId());
			availableSlots.add(allocatedSlot, clock.relativeTimeMillis());
		}
	}

	private PendingRequest pollMatchingPendingRequest(final AllocatedSlot slot) {
		final ResourceProfile slotResources = slot.getResourceProfile();

		// try the requests sent to the resource manager first
		for (PendingRequest request : pendingRequests.values()) {
			if (slotResources.isMatching(request.getResourceProfile())) {
				pendingRequests.removeKeyA(request.getSlotRequestId());
				return request;
			}
		}

		// try the requests waiting for a resource manager connection next
		for (PendingRequest request : waitingForResourceManager.values()) {
			if (slotResources.isMatching(request.getResourceProfile())) {
				waitingForResourceManager.remove(request.getSlotRequestId());
				return request;
			}
		}
		// no request pending, or no request matches
		return null;
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

class SlotPoolImpl {
	private void checkIdleSlot() {
		// The timestamp in SlotAndTimestamp is relative
		final long currentRelativeTimeMillis = clock.relativeTimeMillis();

		final List<AllocatedSlot> expiredSlots = new ArrayList<>(availableSlots.size());

		for (SlotAndTimestamp slotAndTimestamp : availableSlots.availableSlots.values()) {
			if (currentRelativeTimeMillis - slotAndTimestamp.timestamp > idleSlotTimeout.toMilliseconds()) {
				expiredSlots.add(slotAndTimestamp.slot);
			}
		}

		final FlinkException cause = new FlinkException("Releasing idle slot.");

		for (AllocatedSlot expiredSlot : expiredSlots) {
			final AllocationID allocationID = expiredSlot.getAllocationId();
			if (availableSlots.tryRemove(allocationID) != null) {

				//将空闲的 slot 归还给 TaskManager
				log.info("Releasing idle slot [{}].", allocationID);
				final CompletableFuture<Acknowledge> freeSlotFuture = expiredSlot.getTaskManagerGateway().freeSlot(
					allocationID,
					cause,
					rpcTimeout);

				FutureUtils.whenCompleteAsyncIfNotDone(
					freeSlotFuture,
					componentMainThreadExecutor,
					(Acknowledge ignored, Throwable throwable) -> {
						if (throwable != null) {
							if (registeredTaskManagers.contains(expiredSlot.getTaskManagerId())) {
								log.debug("Releasing slot [{}] of registered TaskExecutor {} failed. " +
										"Trying to fulfill a different slot request.", allocationID, expiredSlot.getTaskManagerId(),
									throwable);
								tryFulfillSlotRequestOrMakeAvailable(expiredSlot);
							} else {
								log.debug("Releasing slot [{}] failed and owning TaskExecutor {} is no " +
									"longer registered. Discarding slot.", allocationID, expiredSlot.getTaskManagerId());
							}
						}
					});
			}
		}

		scheduleRunAsync(this::checkIdleSlot, idleSlotTimeout);
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

class SlotSharingManager {
	private final SlotSharingGroupId slotSharingGroupId;

	/** Actions to release allocated slots after a complete multi task slot hierarchy has been released. */
	private final AllocatedSlotActions allocatedSlotActions;

	/** Owner of the slots to which to return them when they are released from the outside. */
	private final SlotOwner slotOwner;

	//所有的 TaskSlot，包括 root 和 inner 和 leaf
	private final Map<SlotRequestId, TaskSlot> allTaskSlots;

	//root MultiTaskSlot，但底层的 Physical Slot 还没有分配好
	/** Root nodes which have not been completed because the allocated slot is still pending. */
	private final Map<SlotRequestId, MultiTaskSlot> unresolvedRootSlots;

	//root MultiTaskSlot，底层的 physical slot 也已经分配好了，按照两层 map 的方式组织，
	//可以通过已分配的 Physical slot 所在的TaskManager 的位置进行查找
	/** Root nodes which have been completed (the underlying allocated slot has been assigned). */
	private final Map<TaskManagerLocation, Map<AllocationID, MultiTaskSlot>> resolvedRootSlots;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

class SlotSharingManager {
	MultiTaskSlot createRootSlot(
			SlotRequestId slotRequestId,
			CompletableFuture<? extends SlotContext> slotContextFuture,
			SlotRequestId allocatedSlotRequestId) {
		final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot(
			slotRequestId,
			slotContextFuture,
			allocatedSlotRequestId);

		LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId);

		allTaskSlots.put(slotRequestId, rootMultiTaskSlot);

		//先加入到 unresolvedRootSlots 中
		unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot);

		// add the root node to the set of resolved root nodes once the SlotContext future has
		// been completed and we know the slot's TaskManagerLocation
		slotContextFuture.whenComplete(
			(SlotContext slotContext, Throwable throwable) -> {
				if (slotContext != null) {
					//一旦physical slot完成分配，就从 unresolvedRootSlots 中移除，加入到 resolvedRootSlots 中
					final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId);

					if (resolvedRootNode != null) {
						final AllocationID allocationId = slotContext.getAllocationId();
						LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId);

						final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent(
							slotContext.getTaskManagerLocation(),
							taskManagerLocation -> new HashMap<>(4));

						MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode);
						Preconditions.checkState(previousValue == null);
					}
				} else {
					rootMultiTaskSlot.release(throwable);
				}
			});
		return rootMultiTaskSlot;
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

class SlotSharingManager {
	@Nonnull
	public Collection<SlotInfo> listResolvedRootSlotInfo(@Nullable AbstractID groupId) {
		//列出已经分配了physical slot 的root MultiTaskSlot，但要求 MultiTaskSlot 不包含指定的 groupId
		return resolvedRootSlots
			.values()
			.stream()
			.flatMap((Map<AllocationID, MultiTaskSlot> map) -> map.values().stream())
			.filter((MultiTaskSlot multiTaskSlot) -> !multiTaskSlot.contains(groupId))
			.map((MultiTaskSlot multiTaskSlot) -> (SlotInfo) multiTaskSlot.getSlotContextFuture().join())
			.collect(Collectors.toList());
	}

	@Nullable
	public MultiTaskSlot getResolvedRootSlot(@Nonnull SlotInfo slotInfo) {
		//根据 SlotInfo（TasManagerLocation 和 AllocationId）找到 MultiTaskSlot
		Map<AllocationID, MultiTaskSlot> forLocationEntry = resolvedRootSlots.get(slotInfo.getTaskManagerLocation());
		return forLocationEntry != null ? forLocationEntry.get(slotInfo.getAllocationId()) : null;
	}

	/**
	 * Gets an unresolved slot which does not yet contain the given groupId. An unresolved
	 * slot is a slot whose underlying allocated slot has not been allocated yet.
	 *
	 * @param groupId which the returned slot must not contain
	 * @return the unresolved slot or null if there was no root slot with free capacities
	 */
	@Nullable
	MultiTaskSlot getUnresolvedRootSlot(AbstractID groupId) {
		//找到一个不包含指定 groupId 的 root MultiTaskSlot
		for (MultiTaskSlot multiTaskSlot : unresolvedRootSlots.values()) {
			if (!multiTaskSlot.contains(groupId)) {
				return multiTaskSlot;
			}
		}
		return null;
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

public interface SlotProvider {
	//申请 slot，返回值一个 LogicalSlot 的 future
	CompletableFuture<LogicalSlot> allocateSlot(
		SlotRequestId slotRequestId,
		ScheduledUnit scheduledUnit,
		SlotProfile slotProfile,
		boolean allowQueuedScheduling,
		Time allocationTimeout);

	void cancelSlotRequest(
		SlotRequestId slotRequestId,
		@Nullable SlotSharingGroupId slotSharingGroupId,
		Throwable cause);
}

public interface Scheduler extends SlotProvider, SlotOwner {
	void start(@Nonnull ComponentMainThreadExecutor mainThreadExecutor);
	boolean requiresPreviousExecutionGraphAllocations();
}

1
2
3
4
5
6
7

class SchedulerImpl implements Scheduler {
	private final SlotSelectionStrategy slotSelectionStrategy;

	private final SlotPool slotPool;

	private final Map<SlotSharingGroupId, SlotSharingManager> slotSharingManagers;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

class SchedulerImpl {
	public CompletableFuture<LogicalSlot> allocateSlot(
		SlotRequestId slotRequestId,
		ScheduledUnit scheduledUnit,
		SlotProfile slotProfile,
		boolean allowQueuedScheduling,
		Time allocationTimeout) {
		log.debug("Received slot request [{}] for task: {}", slotRequestId, scheduledUnit.getTaskToExecute());

		componentMainThreadExecutor.assertRunningInMainThread();

		final CompletableFuture<LogicalSlot> allocationResultFuture = new CompletableFuture<>();

		//如果没有指定 SlotSharingGroupId，说明这个任务不运行 slot 共享，要独占一个 slot
		CompletableFuture<LogicalSlot> allocationFuture = scheduledUnit.getSlotSharingGroupId() == null ?
			allocateSingleSlot(slotRequestId, slotProfile, allowQueuedScheduling, allocationTimeout) :
			allocateSharedSlot(slotRequestId, scheduledUnit, slotProfile, allowQueuedScheduling, allocationTimeout);

		allocationFuture.whenComplete((LogicalSlot slot, Throwable failure) -> {
			if (failure != null) {
				cancelSlotRequest(
					slotRequestId,
					scheduledUnit.getSlotSharingGroupId(),
					failure);
				allocationResultFuture.completeExceptionally(failure);
			} else {
				allocationResultFuture.complete(slot);
			}
		});
		return allocationResultFuture;
	}

	@Override
	public void cancelSlotRequest(
		SlotRequestId slotRequestId,
		@Nullable SlotSharingGroupId slotSharingGroupId,
		Throwable cause) {

		componentMainThreadExecutor.assertRunningInMainThread();

		if (slotSharingGroupId != null) {
			releaseSharedSlot(slotRequestId, slotSharingGroupId, cause);
		} else {
			slotPool.releaseSlot(slotRequestId, cause);
		}
	}

	@Override
	public void returnLogicalSlot(LogicalSlot logicalSlot) {
		SlotRequestId slotRequestId = logicalSlot.getSlotRequestId();
		SlotSharingGroupId slotSharingGroupId = logicalSlot.getSlotSharingGroupId();
		FlinkException cause = new FlinkException("Slot is being returned to the SlotPool.");
		cancelSlotRequest(slotRequestId, slotSharingGroupId, cause);
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

class SchedulerImpl {
	private CompletableFuture<LogicalSlot> allocateSingleSlot(
		SlotRequestId slotRequestId,
		SlotProfile slotProfile,
		boolean allowQueuedScheduling,
		Time allocationTimeout) {

		//先尝试从 SlotPool 可用的 AllocatedSlot 中获取
		Optional<SlotAndLocality> slotAndLocality = tryAllocateFromAvailable(slotRequestId, slotProfile);

		if (slotAndLocality.isPresent()) {
			//如果有已经有可用的了，就创建一个 SingleLogicalSlot，并作为 AllocatedSlot 的payload
			// already successful from available
			try {
				return CompletableFuture.completedFuture(
					completeAllocationByAssigningPayload(slotRequestId, slotAndLocality.get()));
			} catch (FlinkException e) {
				return FutureUtils.completedExceptionally(e);
			}
		} else if (allowQueuedScheduling) {
			//暂时没有可用的，如果允许排队的话，可以要求 SlotPool 向 RM 申请一个新的 slot
			// we allocate by requesting a new slot
			return slotPool
				.requestNewAllocatedSlot(slotRequestId, slotProfile.getResourceProfile(), allocationTimeout)
				.thenApply((PhysicalSlot allocatedSlot) -> {
					try {
						return completeAllocationByAssigningPayload(slotRequestId, new SlotAndLocality(allocatedSlot, Locality.UNKNOWN));
					} catch (FlinkException e) {
						throw new CompletionException(e);
					}
				});
		} else {
			// failed to allocate
			return FutureUtils.completedExceptionally(
				new NoResourceAvailableException("Could not allocate a simple slot for " + slotRequestId + '.'));
		}
	}

	private Optional<SlotAndLocality> tryAllocateFromAvailable(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull SlotProfile slotProfile) {

		Collection<SlotInfo> slotInfoList = slotPool.getAvailableSlotsInformation();

		Optional<SlotSelectionStrategy.SlotInfoAndLocality> selectedAvailableSlot =
			slotSelectionStrategy.selectBestSlotForProfile(slotInfoList, slotProfile);

		return selectedAvailableSlot.flatMap(slotInfoAndLocality -> {
			Optional<PhysicalSlot> optionalAllocatedSlot = slotPool.allocateAvailableSlot(
				slotRequestId,
				slotInfoAndLocality.getSlotInfo().getAllocationId());

			return optionalAllocatedSlot.map(
				allocatedSlot -> new SlotAndLocality(allocatedSlot, slotInfoAndLocality.getLocality()));
		});
	}
}