[源码解析] Flink的Slot究竟是什么？(2)

-  这时候的 report如下： slotReport = {SlotReport@9633}     0 = {SlotStatus@8969} "SlotStatus{slotID=40d390ec-7d52-4f34-af86-d06bb515cc48_0, resourceProfile=ResourceProfile{managedMemory=64.000mb (67108864 bytes), networkMemory=32.000mb (33554432 bytes)}, allocationID=null, jobID=null}"    slotID = {SlotID@8629} "40d390ec-7d52-4f34-af86-d06bb515cc48_0"    resourceProfile = {ResourceProfile@4194} "ResourceProfile{managedMemory=64.000mb (67108864 bytes), networkMemory=32.000mb (33554432 bytes)}"    allocationID = null    jobID = null        1 = {SlotStatus@9638} "SlotStatus{slotID=40d390ec-7d52-4f34-af86-d06bb515cc48_1, resourceProfile=ResourceProfile{managedMemory=64.000mb (67108864 bytes), networkMemory=32.000mb (33554432 bytes)}, allocationID=null, jobID=null}"    slotID = {SlotID@9643} "40d390ec-7d52-4f34-af86-d06bb515cc48_1"    resourceProfile = {ResourceProfile@4194} "ResourceProfile{managedMemory=64.000mb (67108864 bytes), networkMemory=32.000mb (33554432 bytes)}"    allocationID = null    jobID = null
 

-  这时候代码 & 变量如下，我们可以看到，就是把TM的Slot信息注册到SlotManager中： private TaskManagerSlot createAndRegisterTaskManagerSlot(SlotID slotId, ResourceProfile resourceProfile, TaskExecutorConnection taskManagerConnection) {    final TaskManagerSlot slot = new TaskManagerSlot(                                     slotId, resourceProfile, taskManagerConnection);    slots.put(slotId, slot);    return slot; }  slot = {TaskManagerSlot@13322}   slotId = {SlotID@8629} "40d390ec-7d52-4f34-af86-d06bb515cc48_0"  resourceProfile = {ResourceProfile@4194}    cpuCores = {CPUResource@11616} "Resource(CPU: 89884656743115785...0)"   taskHeapMemory = {MemorySize@11617} "4611686018427387903 bytes"   taskOffHeapMemory = {MemorySize@11618} "4611686018427387903 bytes"   managedMemory = {MemorySize@11619} "64 mb"   networkMemory = {MemorySize@11620} "32 mb"   extendedResources = {HashMap@11621}  size = 0  taskManagerConnection = {WorkerRegistration@11121}   allocationId = null  jobId = null  assignedSlotRequest = null  state = {TaskManagerSlot$State@13328} "FREE"
 

this = {SlotManagerImpl@11120} 
 scheduledExecutor = {ActorSystemScheduledExecutorAdapter@11125} 
 slotRequestTimeout = {Time@11127} "300000 ms"
 taskManagerTimeout = {Time@11128} "30000 ms"
 slots = {HashMap@11122}  size = 2
  {SlotID@9643} "40d390ec-7d52-4f34-af86-d06bb515cc48_1" -> {TaskManagerSlot@19206} 
  {SlotID@8629} "40d390ec-7d52-4f34-af86-d06bb515cc48_0" -> {TaskManagerSlot@13322} 
 freeSlots = {LinkedHashMap@11129}  size = 2
  {SlotID@8629} "40d390ec-7d52-4f34-af86-d06bb515cc48_0" -> {TaskManagerSlot@13322} 
  {SlotID@9643} "40d390ec-7d52-4f34-af86-d06bb515cc48_1" -> {TaskManagerSlot@19206} 
 taskManagerRegistrations = {HashMap@11130}  size = 1
 fulfilledSlotRequests = {HashMap@11131}  size = 0
 pendingSlotRequests = {HashMap@11132}  size = 0
 pendingSlots = {HashMap@11133}  size = 0
 slotMatchingStrategy = {AnyMatchingSlotMatchingStrategy@11134} "INSTANCE"
 slotRequestTimeoutCheck = {ActorSystemScheduledExecutorAdapter$ScheduledFutureTask@11139} 

-  slotReport = {SlotReport@8718}   slotsStatus = {ArrayList@8717}  size = 2   0 = {SlotStatus@9025} "SlotStatus{slotID=d99e16d7-a30c-4e21-b270-f82884b1813f\_0, resourceProfile=ResourceProfile{managedMemory=64.000mb (67108864 bytes), networkMemory=32.000mb (33554432 bytes)}, allocationID=null, jobID=null}"    slotID = {SlotID@9032} "d99e16d7-a30c-4e21-b270-f82884b1813f\_0"    resourceProfile = {ResourceProfile@4194} "ResourceProfile{managedMemory=64.000mb (67108864 bytes), networkMemory=32.000mb (33554432 bytes)}"    allocationID = null    jobID = null   1 = {SlotStatus@9026} "SlotStatus{slotID=d99e16d7-a30c-4e21-b270-f82884b1813f\_1, resourceProfile=ResourceProfile{managedMemory=64.000mb (67108864 bytes), networkMemory=32.000mb (33554432 bytes)}, allocationID=null, jobID=null}"    slotID = {SlotID@9029} "d99e16d7-a30c-4e21-b270-f82884b1813f\_1"    resourceProfile = {ResourceProfile@4194} "ResourceProfile{managedMemory=64.000mb (67108864 bytes), networkMemory=32.000mb (33554432 bytes)}"    allocationID = null    jobID = null

-  freeSlots.put(freeSlot.getSlotId(), freeSlot);

-  int numTaskVertices = vertexParallelism > 0 ? vertexParallelism : defaultParallelism; this.taskVertices = new ExecutionVertex[numTaskVertices];

-  this.currentExecution = new Execution(  			getExecutionGraph().getFutureExecutor(), 			this,	0, initialGlobalModVersion,	createTimestamp, timeout);

-  allocateSlotsAndDeployExecutionVertices(schedulingTopology.getVertices());

-  private void allocateSlotsAndDeployExecutionVertices(       final Iterable<? extends SchedulingExecutionVertex<?, ?>> vertices) {    // 取出状态是CREATED，且输入Ready的 ExecutionVertex 	 final Set<ExecutionVertexID> verticesToDeploy = IterableUtils.toStream(vertices) 			.filter(IS\_IN\_CREATED\_EXECUTION\_STATE.and(isInputConstraintSatisfied())) 			.map(SchedulingExecutionVertex::getId) 			.collect(Collectors.toSet());    // 根据 ExecutionVertex 建立 DeploymentOption    final List<ExecutionVertexDeploymentOption> vertexDeploymentOptions = ...;    // 分配资源并且部署    schedulerOperations.allocateSlotsAndDeploy(vertexDeploymentOptions); }

@Override
public void allocateSlotsAndDeploy(final List<ExecutionVertexDeploymentOption> executionVertexDeploymentOptions) {
   validateDeploymentOptions(executionVertexDeploymentOptions);

   final Map<ExecutionVertexID, ExecutionVertexDeploymentOption> deploymentOptionsByVertex =
      groupDeploymentOptionsByVertexId(executionVertexDeploymentOptions);

   final List<ExecutionVertexID> verticesToDeploy = executionVertexDeploymentOptions.stream()
      .map(ExecutionVertexDeploymentOption::getExecutionVertexId)
      .collect(Collectors.toList());

   final Map<ExecutionVertexID, ExecutionVertexVersion> requiredVersionByVertex =
      executionVertexVersioner.recordVertexModifications(verticesToDeploy);

   transitionToScheduled(verticesToDeploy);

   // 分配Slot，其实这时候并没有分配，而是建立一系列Future，然后根据Future返回SlotExecutionVertexAssignment列表
   final List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments =
      allocateSlots(executionVertexDeploymentOptions);

   // 根据SlotExecutionVertexAssignment建立DeploymentHandle
   final List<DeploymentHandle> deploymentHandles = createDeploymentHandles(
      requiredVersionByVertex,
      deploymentOptionsByVertex,
      slotExecutionVertexAssignments);
  
   // 根据deploymentHandles进行部署，其实是根据Future把部署搭建起来，具体如何部署需要在slot分配成功之后再执行
   if (isDeployIndividually()) {
      deployIndividually(deploymentHandles);
   } else {
      waitForAllSlotsAndDeploy(deploymentHandles);
   }
}

 *  Run in Job Manager
 *
 *    DefaultScheduler#allocateSlotsAndDeploy 
 *        |
 *        +----> DefaultScheduler#allocateSlots
 *        |     //把ExecutionVertex转化为ExecutionVertexSchedulingRequirements
 *        |     
 *        +----> DefaultExecutionSlotAllocator#allocateSlotsFor（ 调用 1 开始 ）
 *        |     // 得到 我们的第一个 CompletableFuture，我们称之为 Future 1
 *        |  
 *        |    
 *        +--------------> NormalSlotProviderStrategy#allocateSlot 
 *        |    
 *        |        
 *        +--------------> SchedulerImpl#allocateSlotInternal
 *        |     // 生成了第一个 CompletableFuture，以后称之为 allocationResultFuture
 *        |  
 *  ┌────────────┐   
 *  │  Future 1  │ 生成 allocationResultFuture
 *  └────────────┘ 
 *        │     
 *        │            
 *        +----> SchedulerImpl#internalAllocateSlot（ 调用 2 开始 ）  
 *        |      // Future 1 做为参数被传进来，这里会继续调用，生成 Future 2， Future 3
 *        |     
 *        |       
 *        +-----------> SchedulerImpl#allocateSharedSlot（ 调用 3 开始 ）
 *        |    	// 这里涉及到 MultiTaskSlot 和 SingleTaskSlot
 *        |        
 *        +-----------> SchedulerImpl # allocateMultiTaskSlot （ 调用 4 开始 ）
 *        |  
 *        |       
 *        +--------------------> SchedulerImpl # requestNewAllocatedSlot
 *        |    
 *        |    
 *        +--------------------> SlotPoolImpl#requestNewAllocatedSlot
 *        |    	// 这里生成一个 PendingRequest
 *        | 	// PendingRequest的构造函数中有 new CompletableFuture<>()，
 *        |     // 所以这里是生成了第三个 Future，注意这里的 Future 是针对 PhysicalSlot  
 *        |  
 *        |  
 *  ┌────────────┐   
 *  │  Future 3  │ 生成 Future<PhysicalSlot>，这个 Future 3 实际是对用户不可见的。
 *  └────────────┘      
 *        |  
 *        | 
 *        +-----------> SchedulerImpl # allocateMultiTaskSlot（ 调用 4 结束 ）
 *        |      // 回到 （ 调用 4 ） 这里，得倒 Future 3
 *        |      // 这里得倒了第三个 Future<PhysicalSlot> 
 *        |      // 第三是因为从用户角度看，它是第三个出现的     
 *        |     
 *        +-----------------------> slotSharingManager # createRootSlot  
 *        |      // 把 Future 3 做为参数传进去     
 *        |      // 这里马上生成 Future 2
 *        |      // Future 2 被设置为 multiTaskSlot 的成员变量 slotContextFuture;  
 *        |      // 然后forward . thenApply 语句 会 设置为 Future 3 回调 Future 2 的回调函数
 *        |     
 *        |       
 *        +-----------> SchedulerImpl#allocateSharedSlot
 *        |    	// 回到 （ 调用 3 ） 这里   
 *        |     
 *        | 
 *        +-----------------------> SlotSharingManager#allocateSingleTaskSlo  
 *        |  // 在 rootMultiTaskSlot 之上生成一个 SingleTaskSlot leaf加入到allTaskSlots。    
 *        |  // leaf.getLogicalSlotFuture(); 这个就是Future 2，设置好的
 *        |   
 *        |        
 *        +-----------> SchedulerImpl#allocateSharedSlot
 *        |    	// 还在 （ 调用 3 ） 这里  
 *        |     // return leaf.getLogicalSlotFuture(); 返回 Future 2   
 *        | 
 *        |       
 *  ┌────────────┐   
 *  │  Future 2  │
 *  └────────────┘  
 *        |     
 *        |       
 *        |       
 *        +----> SchedulerImpl#internalAllocateSlot    
 *        |      // 回到 （ 调用 2 ） 这里 
 *        |      // 设置，在 Future 2 的回调函数中会调用 Future 1    
 *        |  
 *        |      
 *        +----> DefaultExecutionSlotAllocator#allocateSlotsFor 
 *        |     // 回到 （ 调用 1 ） 这里 
 *        |
 *        |  
 *        |       
 *  ┌────────────┐   
 *  │  Future 1  │ 
 *  └────────────┘    
 *        |  
 *        |      
 *        +---->  createDeploymentHandles    
 *        |  // 生成 DeploymentHandle
 *        |     
 *        |       
 *        +-----------> deployIndividually(deploymentHandles);    
 *        |           // 这里会给 Future 1 再加上两个 回调函数，作为 部署回调
 *        | 

-  public List<SlotExecutionVertexAssignment> allocateSlotsFor(...) { 		for (ExecutionVertexSchedulingRequirements schedulingRequirements : executionVertexSchedulingRequirements) {              // 得到第一个 CompletableFuture，具体是在 calculatePreferredLocations 中通过  			CompletableFuture<LogicalSlot> slotFuture =          calculatePreferredLocations(...).thenCompose(...) -> 								slotProviderStrategy.allocateSlot( // 函数里面生成了第一个CompletableFuture 									slotRequestId, 									new ScheduledUnit(...), 									SlotProfile.priorAllocation(...)));  			SlotExecutionVertexAssignment slotExecutionVertexAssignment = 					new SlotExecutionVertexAssignment(executionVertexId, slotFuture);  			slotFuture.whenComplete( 					(ignored, throwable) -> { // 第一个CompletableFuture的回调函数，里其实只是异常处理，后续有人会调用到这里 						pendingSlotAssignments.remove(executionVertexId); 						if (throwable != null) { 							slotProviderStrategy.cancelSlotRequest(slotRequestId, slotSharingGroupId, throwable); 						} 					});  			slotExecutionVertexAssignments.add(slotExecutionVertexAssignment); 		}  		return slotExecutionVertexAssignments; }

-  private CompletableFuture<LogicalSlot> allocateSlotInternal(...) {     // 这里生成了第一个 CompletableFuture，我们以后称之为 allocationResultFuture 		final CompletableFuture<LogicalSlot> allocationResultFuture = new CompletableFuture<>();     // allocationResultFuture 会传送进去继续处理 		internalAllocateSlot(allocationResultFuture, slotRequestId, scheduledUnit,                           slotProfile, allocationTimeout);     // 返回 allocationResultFuture 		return allocationResultFuture; }

-  private void internalAllocateSlot( 			CompletableFuture<LogicalSlot> allocationResultFuture, ...) {   	// 这里得到第二个 CompletableFuture，我们以后称为 allocationFuture，注意目前只是得到，不是生成。 		CompletableFuture<LogicalSlot> allocationFuture = scheduledUnit.getSlotSharingGroupId() == null ? 			allocateSingleSlot(slotRequestId, slotProfile, allocationTimeout) : 			allocateSharedSlot(slotRequestId, scheduledUnit, slotProfile, allocationTimeout); 		// 第二个Future，allocationFuture的回调函数。注意，CompletableFuture可以连续调用多个whenComplete。 		allocationFuture.whenComplete((LogicalSlot slot, Throwable failure) -> { 			if (failure != null) { // 异常处理 				cancelSlotRequest(...); 				allocationResultFuture.completeExceptionally(failure); 			} else { 				allocationResultFuture.complete(slot); // 它将回调第一个 allocationResultFuture的回调函数 			} 		}); }

-  private CompletableFuture<LogicalSlot> allocateSharedSlot(...) {  		// allocate slot with slot sharing  		final SlotSharingManager multiTaskSlotManager = slotSharingManagers.computeIfAbsent(  			scheduledUnit.getSlotSharingGroupId(),  			id -> new SlotSharingManager(id,slotPool,this)); // 生成 SlotSharingManager    		final SlotSharingManager.MultiTaskSlotLocality multiTaskSlotLocality;    			if (scheduledUnit.getCoLocationConstraint() != null) {  				multiTaskSlotLocality = allocateCoLocatedMultiTaskSlot(...);  			} else {  				multiTaskSlotLocality = allocateMultiTaskSlot(...); // 这里生成 MultiTaskSlot  			}      	// 这里生成 SingleTaskSlot  		final SlotSharingManager.SingleTaskSlot leaf = multiTaskSlotLocality.getMultiTaskSlot().allocateSingleTaskSlot(...);      		return leaf.getLogicalSlotFuture(); // 返回 SingleTaskSlot 的 future，就是第二个Future，具体生成我们在下面会详述  	}

-  private SlotSharingManager.MultiTaskSlotLocality allocateMultiTaskSlot(...) {    		SlotSharingManager.MultiTaskSlot multiTaskSlot = slotSharingManager.getUnresolvedRootSlot(groupId);    		if (multiTaskSlot == null) {        // requestNewAllocatedSlot 会调用 SlotPoolImpl 的同名函数       // 得到第 三 个 Future，注意，这个 Future 针对的是 PhysicalSlot  			final CompletableFuture<PhysicalSlot> slotAllocationFuture = requestNewAllocatedSlot(...);         // 使用 第 三 个 Future 来构建 multiTaskSlot  			multiTaskSlot = slotSharingManager.createRootSlot(...,slotAllocationFuture,...);        // 第 三 个 Future的回调函数，这里会把payload赋值给slot  			slotAllocationFuture.whenComplete(  				(PhysicalSlot allocatedSlot, Throwable throwable) -> {  					final SlotSharingManager.TaskSlot taskSlot = slotSharingManager.getTaskSlot(multiTaskSlotRequestId);    					if (taskSlot != null) {              // 会把payload赋值给slot  							if (!allocatedSlot.tryAssignPayload(((SlotSharingManager.MultiTaskSlot) taskSlot))) {...}  					}   				});  		}    		return SlotSharingManager.MultiTaskSlotLocality.of(multiTaskSlot, Locality.UNKNOWN);  	}

-  public CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(...) {    		// 生成 PendingRequest  		final PendingRequest pendingRequest = PendingRequest.createStreamingRequest(slotRequestId, resourceProfile);    		// 添加 PendingRequest 到 waitingForResourceManager，然后返回Future  		return requestNewAllocatedSlotInternal(pendingRequest)  			.thenApply((Function.identity()));  	}

-  PendingRequest的构造函数中有 new CompletableFuture<>()，所以这里是生成了第三个 Future，注意这里的 Future 是针对 PhysicalSlot

-  requestNewAllocatedSlotInternal

    -  private CompletableFuture<AllocatedSlot> requestNewAllocatedSlotInternal(PendingRequest pendingRequest) {     if (resourceManagerGateway == null) {       // 就是把 pendingRequest 加到 waitingForResourceManager 之中       stashRequestWaitingForResourceManager(pendingRequest);    } else {       requestSlotFromResourceManager(resourceManagerGateway, pendingRequest);    }    return pendingRequest.getAllocatedSlotFuture(); // 第三个Future }

-  MultiTaskSlot createRootSlot(       SlotRequestId slotRequestId,       CompletableFuture<? extends SlotContext> slotContextFuture, // 参数是第三个Future       SlotRequestId allocatedSlotRequestId) {     // 生成第二个Future<SlotContext>    final CompletableFuture<SlotContext> slotContextFutureAfterRootSlotResolution = new CompletableFuture<>();       final MultiTaskSlot rootMultiTaskSlot = createAndRegisterRootSlot(...       slotContextFutureAfterRootSlotResolution); // 第二个Future 在 createAndRegisterRootSlot 函数中 被赋值为 MultiTaskSlot的 slotContextFuture 成员变量     FutureUtils.forward(       slotContextFuture.thenApply( // 第三个Future进一步回调时候，会回调第二个Future          (SlotContext slotContext) -> {             // 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             tryMarkSlotAsResolved(slotRequestId, slotContext);             return slotContext;          }),       slotContextFutureAfterRootSlotResolution); // 在这里回调第二个Future     return rootMultiTaskSlot; }

-  SingleTaskSlot allocateSingleTaskSlot( 				SlotRequestId slotRequestId, ResourceProfile resourceProfile, 				AbstractID groupId, Locality locality) {  			final SingleTaskSlot leaf = new SingleTaskSlot( 				slotRequestId, resourceProfile, groupId, this, locality);  			children.put(groupId, leaf);  			// register the newly allocated slot also at the SlotSharingManager 			allTaskSlots.put(slotRequestId, leaf);  			reserveResource(resourceProfile);  			return leaf; }

-  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() {           singleLogicalSlotFuture = parent.getSlotContextFuture()             .thenApply(               (SlotContext slotContext) -> {                 return new SingleLogicalSlot( // 未来回调时候才会生成                   slotRequestId,                   slotContext,                   slotSharingGroupId,                   locality,                   slotOwner);               });         }      CompletableFuture<LogicalSlot> getLogicalSlotFuture() {        return singleLogicalSlotFuture.thenApply(Function.identity());     }   }

-  private void deployIndividually(final List<DeploymentHandle> deploymentHandles) {    for (final DeploymentHandle deploymentHandle : deploymentHandles) {       FutureUtils.assertNoException(          deploymentHandle             .getSlotExecutionVertexAssignment()             .getLogicalSlotFuture()             .handle(assignResourceOrHandleError(deploymentHandle))             .handle(deployOrHandleError(deploymentHandle)));    } }

-  private BiFunction<LogicalSlot, Throwable, Void> assignResourceOrHandleError(final DeploymentHandle deploymentHandle) {       final ExecutionVertexVersion requiredVertexVersion = deploymentHandle.getRequiredVertexVersion();    final ExecutionVertexID executionVertexId = deploymentHandle.getExecutionVertexId();     return (logicalSlot, throwable) -> {       if (throwable == null) {          final ExecutionVertex executionVertex = getExecutionVertex(executionVertexId);          final boolean sendScheduleOrUpdateConsumerMessage = deploymentHandle.getDeploymentOption().sendScheduleOrUpdateConsumerMessage();          executionVertex             .getCurrentExecutionAttempt()             .registerProducedPartitions(logicalSlot.getTaskManagerLocation(), sendScheduleOrUpdateConsumerMessage);          executionVertex.tryAssignResource(logicalSlot);       } else {          handleTaskDeploymentFailure(executionVertexId, maybeWrapWithNoResourceAvailableException(throwable));       }       return null;    }; }

-  private BiFunction<Object, Throwable, Void> deployOrHandleError(final DeploymentHandle deploymentHandle) {       final ExecutionVertexVersion requiredVertexVersion = deploymentHandle.getRequiredVertexVersion();    final ExecutionVertexID executionVertexId = requiredVertexVersion.getExecutionVertexId();     return (ignored, throwable) -> {       if (throwable == null) {          deployTaskSafe(executionVertexId);       } else {          handleTaskDeploymentFailure(executionVertexId, throwable);       }       return null;    }; }

-  private void requestSlotFromResourceManager( 			final ResourceManagerGateway resourceManagerGateway, 			final PendingRequest pendingRequest) {     // 生成一个 AllocationID，这个会传到 TM 那里，注册到 TaskSlot上。     final AllocationID allocationId = new AllocationID();     // 生成一个SlotRequest，并且向 RM 进行 RPC 请求。 	CompletableFuture<Acknowledge> rmResponse =          				resourceManagerGateway.requestSlot(                                 jobMasterId,                                 new SlotRequest(jobId, allocationId,                                       		pendingRequest.getResourceProfile(),                                  jobManagerAddress),                                 rpcTimeout); }

-  pendingSlotRequests.put

-  private void internalRequestSlot(PendingSlotRequest pendingSlotRequest) throws ResourceManagerException {    final ResourceProfile resourceProfile = pendingSlotRequest.getResourceProfile();    OptionalConsumer.of(findMatchingSlot(resourceProfile))       .ifPresent(taskManagerSlot -> allocateSlot(taskManagerSlot, pendingSlotRequest))       .ifNotPresent(() -> fulfillPendingSlotRequestWithPendingTaskManagerSlot(pendingSlotRequest)); }

-  TaskExecutorGateway gateway = taskExecutorConnection.getTaskExecutorGateway(); CompletableFuture<Acknowledge> requestFuture = gateway.requestSlot( 			slotId, 			pendingSlotRequest.getJobId(), 			allocationId, 			pendingSlotRequest.getResourceProfile(), 			pendingSlotRequest.getTargetAddress(), 			resourceManagerId, 			taskManagerRequestTimeout);

-  public boolean allocateSlot(int index, JobID jobId, AllocationID allocationId, 			ResourceProfile resourceProfile,Time slotTimeout) {     taskSlot = new TaskSlot<>(index, resourceProfile, memoryPageSize, jobId, allocationId);     taskSlots.put(index, taskSlot);     allocatedSlots.put(allocationId, taskSlot);     slots.add(allocationId); }

---------- Task Executor ----------
       │ 
       │ 
┌─────────────┐   
│  TaskSlot   │  requestSlot
└─────────────┘     
       │ 
       │                  
┌──────────────┐   
│  SlotOffer   │  offerSlotsToJobManager
└──────────────┘       
       │ 
       │      
------------- Job Manager -------------
       │ 
       │       
┌──────────────┐   
│  SlotOffer   │  JobMaster#offerSlots(taskManagerId,slots)
└──────────────┘     
       │ //taskManager = registeredTaskManagers.get(taskManagerId);     
       │ //taskManagerLocation = taskManager.f0;     
       │ //taskExecutorGateway = taskManager.f1;     
       │    
       │       
┌──────────────┐   
│  SlotOffer   │  SlotPoolImpl#offerSlots
└──────────────┘       
       │ 
       │      
┌───────────────┐   
│ AllocatedSlot │  SlotPoolImpl#offerSlot
└───────────────┘      
       │ 
       │      
┌───────────────┐   
│ 回调 Future 3  │ SlotSharingManager#createRootSlot
└───────────────┘      
       │ 
       │      
┌───────────────┐   
│ 回调 Future 2  │  SingleTaskSlot#SingleTaskSlot 
└───────────────┘      
       │ 
       │      
┌───────────────────┐   
│ SingleLogicalSlot │ new SingleLogicalSlot
└───────────────────┘    
       │ 
       │     
┌───────────────────┐   
│ SingleLogicalSlot │  
│ 回调 Future 1      │ allocationResultFuture.complete()
└───────────────────┘   
       │    
       │        
┌───────────────────────────────┐  
│     SingleLogicalSlot         │  
│回调 assignResourceOrHandleError│ 
└───────────────────────────────┘
       │    
       │        
┌────────────────┐   
│ ExecutionVertex│ tryAssignResource
└────────────────┘    
       │    
       │        
┌────────────────┐   
│    Execution   │ tryAssignResource
└────────────────┘       
       │    
       │        
┌──────────────────┐   
│ SingleLogicalSlot│ tryAssignPayload
└──────────────────┘  
       │    
       │        
┌───────────────────────┐   
│   SingleLogicalSlot   │      
│ 回调deployOrHandleError│ 
└───────────────────────┘   
       │    
       │       
┌────────────────┐   
│ ExecutionVertex│ deploy
└────────────────┘    
       │    
       │        
┌────────────────┐   
│    Execution   │ deploy // 关键点
└────────────────┘        
       │  
       │    
       │        
 ---------- Task Executor ----------
       │    
       │     
┌────────────────┐   
│  TaskExecutor  │ submitTask
└────────────────┘     
       │    
       │        
┌────────────────┐   
│  TaskExecutor  │ startTaskThread
└────────────────┘         

-  private void offerSlotsToJobManager(final JobID jobId) { 				final Iterator<TaskSlot<Task>> 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); 				}     			// 把 SlotOffer 通过RPC发给 JM 				CompletableFuture<Collection<SlotOffer>> acceptedSlotsFuture =                      jobMasterGateway.offerSlots(                             getResourceID(),                             reservedSlots,                             taskManagerConfiguration.getTimeout());     }

-  public CompletableFuture<Collection<SlotOffer>> offerSlots( 			final ResourceID taskManagerId, 			final Collection<SlotOffer> slots, 			final Time timeout) {  		Tuple2<TaskManagerLocation, TaskExecutorGateway> taskManager = registeredTaskManagers.get(taskManagerId);  		final TaskManagerLocation taskManagerLocation = taskManager.f0; 		final TaskExecutorGateway taskExecutorGateway = taskManager.f1;  		final RpcTaskManagerGateway rpcTaskManagerGateway = new RpcTaskManagerGateway(taskExecutorGateway, getFencingToken());  		return CompletableFuture.completedFuture( 			slotPool.offerSlots( 				taskManagerLocation, 				rpcTaskManagerGateway, 				slots)); 	}

-  boolean offerSlot( 			final TaskManagerLocation taskManagerLocation, 			final TaskManagerGateway taskManagerGateway, 			final SlotOffer slotOffer) {              // 根据 SlotOffer 的信息生成一个 AllocatedSlot，对于 AllocatedSlot 来说，有效信息就是 slotIndex, resourceProfile 	final AllocatedSlot allocatedSlot = new AllocatedSlot(        allocationID,        taskManagerLocation,        slotOffer.getSlotIndex(),        slotOffer.getResourceProfile(),        taskManagerGateway);          allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot); 		if (pendingRequest != null) { 			allocatedSlots.add(pendingRequest.getSlotRequestId(), allocatedSlot);        // 这里取出了 pendingRequest 的 Future, 就是我们之前的 Future 3，进行回调 			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); 			}  		} }

-  FutureUtils.forward(    slotContextFuture.thenApply(       (SlotContext slotContext) -> {          // 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          tryMarkSlotAsResolved(slotRequestId, slotContext); // 运行到这里          return slotContext;       }),    slotContextFutureAfterRootSlotResolution); // 然后到这里

-  singleLogicalSlotFuture = parent.getSlotContextFuture()    .thenApply(       (SlotContext slotContext) -> {          return new SingleLogicalSlot( // 回调生成了 SingleLogicalSlot             slotRequestId,             slotContext,             slotSharingGroupId,             locality,             slotOwner);       });

-  allocationFuture.whenComplete((LogicalSlot slot, Throwable failure) -> {    if (failure != null) {       cancelSlotRequest(          slotRequestId,          scheduledUnit.getSlotSharingGroupId(),          failure);       allocationResultFuture.completeExceptionally(failure);    } else {       allocationResultFuture.complete(slot); // 代码在这里    } });

-  private BiFunction<LogicalSlot, Throwable, Void> assignResourceOrHandleError(final DeploymentHandle deploymentHandle) {    		return (logicalSlot, throwable) -> {  			if (executionVertexVersioner.isModified(requiredVertexVersion)) {    			if (throwable == null) {  				final ExecutionVertex executionVertex = getExecutionVertex(executionVertexId);  				final boolean sendScheduleOrUpdateConsumerMessage = deploymentHandle.getDeploymentOption().sendScheduleOrUpdateConsumerMessage();  				executionVertex  					.getCurrentExecutionAttempt()  					.registerProducedPartitions(logicalSlot.getTaskManagerLocation(), sendScheduleOrUpdateConsumerMessage);  				executionVertex.tryAssignResource(logicalSlot); // 运行到这里  			}   			return null;  		};  	}

-  回调函数会深入调用 executionVertex.tryAssignResource，

-  ExecutionVertex  # tryAssignResource

-  Execution  # tryAssignResource

-  SingleLogicalSlot# tryAssignPayload(this)，**这里会把 Execution 自己 赋值给Slot.payload**，最后 Execution 在 runtime 的变量举例如下：

    -  payload = {Execution@10669} "Attempt #0 (CHAIN DataSource (at getDefaultTextLineDataSet(WordCountData.java:47) (org.apache.flink.api.java.io.CollectionInputFormat)) -> FlatMap (FlatMap at main(WordCount.java:64)) -> Combine (SUM(1), at main(WordCount.java:67) (1/1)) @ org.apache.flink.runtime.jobmaster.slotpool.SingleLogicalSlot@61c7928f - [SCHEDULED]"  executor = {ScheduledThreadPoolExecutor@5928} "java.util.concurrent.ScheduledThreadPoolExecutor@6a2c6c71[Running, pool size = 3, active threads = 0, queued tasks = 1, completed tasks = 2]"  vertex = {ExecutionVertex@10534} "CHAIN DataSource (at getDefaultTextLineDataSet(WordCountData.java:47) (org.apache.flink.api.java.io.CollectionInputFormat)) -> FlatMap (FlatMap at main(WordCount.java:64)) -> Combine (SUM(1), at main(WordCount.java:67) (1/1)"  attemptId = {ExecutionAttemptID@10792} "2f8b6c7297527225ee4c8036c457ba27"  globalModVersion = 1  stateTimestamps = {long[9]@10793}   attemptNumber = 0  rpcTimeout = {Time@5924} "18000000 ms"  partitionInfos = {ArrayList@10794}  size = 0  terminalStateFuture = {CompletableFuture@10795} "java.util.concurrent.CompletableFuture@2eb8f94c[Not completed]"  releaseFuture = {CompletableFuture@10796} "java.util.concurrent.CompletableFuture@7c794914[Not completed]"  taskManagerLocationFuture = {CompletableFuture@10797} "java.util.concurrent.CompletableFuture@2e11ac18[Not completed]"  state = {ExecutionState@10789} "SCHEDULED"  assignedResource = {SingleLogicalSlot@10507}   failureCause = null  taskRestore = null  assignedAllocationID = null  accumulatorLock = {Object@10798}   userAccumulators = null  ioMetrics = null  producedPartitions = {LinkedHashMap@10799}  size = 1

-  private BiFunction<Object, Throwable, Void> deployOrHandleError(final DeploymentHandle deploymentHandle) {     return (ignored, throwable) -> {       if (executionVertexVersioner.isModified(requiredVertexVersion)) {        if (throwable == null) {          deployTaskSafe(executionVertexId); // 在这里部署       } else {          handleTaskDeploymentFailure(executionVertexId, throwable);       }       return null;    }; }

-  回调函数深入调用其他函数

-  DefaultScheduler  # deployTaskSafe

-  ExecutionVertex  # deploy

-  Execution # deploy。每次调度ExecutionVertex，都会有一个Execution，在此阶段会将Execution的状态变更为DEPLOYING状态，并且为该ExecutionVertex生成对应的部署描述信息，然后从对应的slot中获取对应的TaskManagerGateway，以便向对应的TaskManager提交Task。其中，ExecutionVertex.createDeploymentDescriptor方法中，包含了从Execution Graph到真正物理执行图的转换。如将IntermediateResultPartition转化成ResultPartition，ExecutionEdge转成InputChannelDeploymentDescriptor（最终会在执行时转化成InputGate）。

    -  // 这里一个关键点是：Execution 部署时候，是 从 SingleLogicalSlot ---> AllocatedSlot ---> TaskManagerGateway 这个顺序获取了 TaskManager 的 RPC 网关，然后通过 taskManagerGateway.submitTask 才能提交任务的。这样就把 Execution 部署阶段和执行阶段联系起来了 public void deploy() throws JobException { 			final TaskDeploymentDescriptor deployment = TaskDeploymentDescriptorFactory 			.fromExecutionVertex(vertex, attemptNumber) 				.createDeploymentDescriptor( 				slot.getAllocationId(), 					slot.getPhysicalSlotNumber(), 					taskRestore, 					producedPartitions.values());        	// 这里就是关键点   		final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();          // 在这里通过RPC提交task给了TaskManager   		CompletableFuture.supplyAsync(() -> taskManagerGateway.submitTask(deployment, rpcTimeout), executor).thenCompose(Function.identity()) }

-  // 这个方法会创建真正的Task，然后调用task.startTaskThread();开始task的执行。 public CompletableFuture<Acknowledge> submitTask(       TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, Time timeout) {   		// taskSlot.getMemoryManager(); 会获取slot的内存管理器，这里就是分割内存的部分功能   		memoryManager = taskSlotTable.getTaskMemoryManager(tdd.getAllocationId());   		// 在Task构造函数中，会根据输入的参数，创建InputGate, ResultPartition, ResultPartitionWriter等。 			Task task = new Task( 				jobInformation, 				taskInformation, 				tdd.getExecutionAttemptId(), 				tdd.getAllocationId(), 				tdd.getSubtaskIndex(), 				tdd.getAttemptNumber(), 				tdd.getProducedPartitions(), 				tdd.getInputGates(), 				tdd.getTargetSlotNumber(), 				memoryManager, 				taskExecutorServices.getIOManager(), 				taskExecutorServices.getShuffleEnvironment(), 				taskExecutorServices.getKvStateService(), 				taskExecutorServices.getBroadcastVariableManager(), 				taskExecutorServices.getTaskEventDispatcher(), 				taskStateManager, 				taskManagerActions, 				inputSplitProvider, 				checkpointResponder, 				aggregateManager, 				blobCacheService, 				libraryCache, 				fileCache, 				taskManagerConfiguration, 				taskMetricGroup, 				resultPartitionConsumableNotifier, 				partitionStateChecker, 				getRpcService().getExecutor());          taskAdded = taskSlotTable.addTask(task);   		task.startTaskThread(); }

-  开始了线程了。而`startTaskThread`方法，则会执行`executingThread.start`，从而调用`Task.run`方法。

    -  public void startTaskThread() {    executingThread.start(); }

-  // 这里的invokable即为operator对象实例，通过反射创建。 private void doRun() { 	AbstractInvokable invokable = null; 	invokable = loadAndInstantiateInvokable(userCodeClassLoader, nameOfInvokableClass, env); 	// run the invokable   invokable.invoke(); }

public class WordCount {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setString("heartbeat.timeout", "18000000");
        conf.setString("resourcemanager.job.timeout", "18000000");
        conf.setString("resourcemanager.taskmanager-timeout", "18000000");
        conf.setString("slotmanager.request-timeout", "18000000");
        conf.setString("slotmanager.taskmanager-timeout", "18000000");
        conf.setString("slot.request.timeout", "18000000");
        conf.setString("slot.idle.timeout", "18000000");
        conf.setString("akka.ask.timeout", "18000000");
        conf.setString("taskmanager.numberOfTaskSlots", "1");

        final LocalEnvironment env = ExecutionEnvironment.createLocalEnvironment(conf);
        final MultipleParameterTool params = MultipleParameterTool.fromArgs(args);
        env.getConfig().setGlobalJobParameters(params);

        // get input data
        DataSet<String> text = null;
        if (params.has("input")) {
            // union all the inputs from text files
            for (String input : params.getMultiParameterRequired("input")) {
                if (text == null) {
                    text = env.readTextFile(input);
                } else {
                    text = text.union(env.readTextFile(input));
                }
            }
        } else {
            // get default test text data
            text = WordCountData.getDefaultTextLineDataSet(env);
        }

        DataSet<Tuple2<String, Integer>> counts =
                // split up the lines in pairs (2-tuples) containing: (word,1)
                text.flatMap(new Tokenizer())
                        // group by the tuple field "0" and sum up tuple field "1"
                        .groupBy(0)
                        .sum(1);

        // emit result
        if (params.has("output")) {
            counts.writeAsCsv(params.get("output"), "\n", " ");
            env.execute("WordCount Example");
        } else {
            counts.print();
        }
    }

    // *************************************************************************
    //     USER FUNCTIONS
    // *************************************************************************
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            // normalize and split the line
            String[] tokens = value.toLowerCase().split("\\W+");

            // emit the pairs
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}

// 这里一个关键点是：Execution 部署时候，是 从 SingleLogicalSlot ---> AllocatedSlot ---> TaskManagerGateway 这个顺序获取了 TaskManager 的 RPC 网关，然后通过 taskManagerGateway.submitTask 才能提交任务的。这样就把 Execution 部署阶段和执行阶段联系起来了
public void deploy() throws JobException {
			final TaskDeploymentDescriptor deployment = TaskDeploymentDescriptorFactory
			.fromExecutionVertex(vertex, attemptNumber)
				.createDeploymentDescriptor(
				slot.getAllocationId(),
					slot.getPhysicalSlotNumber(),
					taskRestore,
					producedPartitions.values());

   	// 这里就是关键点
  		final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();

      // 在这里通过RPC提交task给了TaskManager
  		CompletableFuture.supplyAsync(() -> taskManagerGateway.submitTask(deployment, rpcTimeout), executor).thenCompose(Function.identity())
}

// set up the input splits, if the vertex has any
try {
			InputSplitSource<InputSplit> splitSource = (InputSplitSource<InputSplit>) jobVertex.getInputSplitSource();

			if (splitSource != null) {
				try {
					inputSplits = splitSource.createInputSplits(numTaskVertices);
					if (inputSplits != null) {
						splitAssigner = splitSource.getInputSplitAssigner(inputSplits);
					}
				} 
}
            
// 此时splitSource如下：
splitSource = {CollectionInputFormat@7603} "[To be, or not to be,--that is the question:--, Whether 'tis nobler in the mind to suffer, The slings and arrows of outrageous fortune, ...]"
 serializer = {StringSerializer@7856} 
 dataSet = {ArrayList@7857}  size = 35
 iterator = null
 partitionNumber = 0
 runtimeContext = null    

Task.run()
 |-- invokable.invoke()
 |    |-- StreamTask.invoke()
 |    |    |-- beforeInvoke()
 |    |    |    |-- init()
 |    |    |    |    |-- SourceStreamTask.init()
 |    |    |    |-- initializeStateAndOpen()
 |    |    |    |    |-- operator.initializeState()
 |    |    |    |    |-- operator.open()
 |    |    |    |    |    |-- SourceStreamTask.LegacySourceFunctionThread.run()
 |    |    |    |    |    |    |-- StreamSource.run()
 |    |    |    |    |    |    |    |-- userFunction.run(ctx)
 |    |    |    |    |    |    |    |    |-- ContinuousFileMonitoringFunction.run()
 |    |    |    |    |    |    |    |    |    |-- RebalancePartitioner.selectChannel()
 |    |    |    |    |    |    |    |    |    |-- RecordWriter.emit()

Task.run()
 |-- invokable.invoke()
 |    |-- StreamTask.invoke()
 |    |    |-- beforeInvoke()
 |    |    |    |-- init()
 |    |    |    |    |--OneInputStreamTask.init()
 |    |    |    |-- initializeStateAndOpen()
 |    |    |    |    |-- operator.initializeState()
 |    |    |    |    |    |-- ContinuousFileReaderOperator.initializeState()
 |    |    |    |    |-- operator.open()
 |    |    |    |    |    |-- ContinuousFileReaderOperator.open()
 |    |    |    |    |    |    |-- ContinuousFileReaderOperator.SplitReader.run()
 |    |    |-- runMailboxLoop()
 |    |    |    |-- StreamTask.processInput()
 |    |    |    |    |-- StreamOneInputProcessor.processInput()
 |    |    |    |    |    |-- StreamTaskNetworkInput.emitNext() while循环不停的处理输入数据
 |    |    |    |    |    |    |-- ContinuousFileReaderOperator.processElement()
 |    |    |-- afterInvoke()    

-  while (!this.taskCanceled && splitIterator.hasNext())

-  CompletableFuture<SerializedInputSplit> futureInputSplit = jobMasterGateway.requestNextInputSplit(   jobVertexID,   executionAttemptID);

-  public SerializedInputSplit requestNextInputSplit(JobVertexID vertexID, ExecutionAttemptID executionAttempt) throws IOException {  		final Execution execution = executionGraph.getRegisteredExecutions().get(executionAttempt);  		final ExecutionJobVertex vertex = executionGraph.getJobVertex(vertexID);  		final InputSplit nextInputSplit = execution.getNextInputSplit();  		final byte[] serializedInputSplit = InstantiationUtil.serializeObject(nextInputSplit);          		return new SerializedInputSplit(serializedInputSplit); }

-  这里  execution.getNextInputSplit() 就会调用 Slot，可以看到，这里先获取Slot，然后从Slot获取其 TaskManager 的host。再从 Vertiex 获取 InputSplit。

    -  public InputSplit getNextInputSplit() { 		final LogicalSlot slot = this.getAssignedResource(); 		final String host = slot != null ? slot.getTaskManagerLocation().getHostname() : null; 		return this.vertex.getNextInputSplit(host); }

    -  public InputSplit getNextInputSplit(String host) { 		final int taskId = getParallelSubtaskIndex(); 		synchronized (inputSplits) { 			final InputSplit nextInputSplit = jobVertex.getSplitAssigner().getNextInputSplit(host, taskId); 			if (nextInputSplit != null) { 				inputSplits.add(nextInputSplit); 			} 			return nextInputSplit; 		} }  // runtime 信息如下 inputSplits = {GenericInputSplit[1]@13113}   0 = {GenericInputSplit@13121} "GenericSplit (0/1)"   partitionNumber = 0   totalNumberOfPartitions = 1

-  final InputSplit split = splitIterator.next(); final InputFormat<OT, InputSplit> format = this.format;			 // open input format // open还没开始真正的读数据，只是定位，设置当前切片信息(切片的开始位置，切片长度)，和定位开始位置。把第一个换行符，分到前一个分片，自己从第二个换行符开始读取数据 format.open(split);