Flink Execute Implements (三)

def runJobManager(
      configuration: Configuration,
      executionMode: JobManagerMode,
      listeningAddress: String,
      listeningPort: Int)
    : Unit = {

    val numberProcessors = Hardware.getNumberCPUCores()

    val futureExecutor = Executors.newScheduledThreadPool(
      numberProcessors,
      new ExecutorThreadFactory("jobmanager-future"))

    val ioExecutor = Executors.newFixedThreadPool(
      numberProcessors,
      new ExecutorThreadFactory("jobmanager-io"))

    val timeout = AkkaUtils.getTimeout(configuration)

    // we have to first start the JobManager ActorSystem because this determines the port if 0
    // was chosen before. The method startActorSystem will update the configuration correspondingly.
    val jobManagerSystem = startActorSystem(
      configuration,
      listeningAddress,
      listeningPort)

    val highAvailabilityServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(
      configuration,
      ioExecutor,
      AddressResolution.NO_ADDRESS_RESOLUTION)

    val metricRegistry = new MetricRegistryImpl(
      MetricRegistryConfiguration.fromConfiguration(configuration))

    metricRegistry.startQueryService(jobManagerSystem, null)

    val (_, _, webMonitorOption, _) = try {
      startJobManagerActors(
        jobManagerSystem,
        configuration,
        executionMode,
        listeningAddress,
        futureExecutor,
        ioExecutor,
        highAvailabilityServices,
        metricRegistry,
        classOf[JobManager],
        classOf[MemoryArchivist],
        Option(classOf[StandaloneResourceManager])
      )
    } catch {
      case t: Throwable =>
        futureExecutor.shutdownNow()
        ioExecutor.shutdownNow()

        throw t
    }

    // block until everything is shut down
    jobManagerSystem.awaitTermination()

    .......
}

override def handleMessage: Receive = {
  case GrantLeadership(newLeaderSessionID) =>
    log.info(s"JobManager $getAddress was granted leadership with leader session ID " +
      s"$newLeaderSessionID.")
    leaderSessionID = newLeaderSessionID

    .......

private def submitJob(jobGraph: JobGraph, jobInfo: JobInfo, isRecovery: Boolean = false): Unit = {

    ......

    executionGraph = ExecutionGraphBuilder.buildGraph(
          executionGraph,
          jobGraph,
          FlinkConfiguration,
          futureExecutor,
          ioExecutor,
          scheduler,
          userCodeLoader,
          checkpointRecoveryFactory,
          Time.of(timeout.length, timeout.unit),
          restartStrategy,
          jobMetrics,
          numSlots,
          blobServer,
          log.logger)

    ......

    if (leaderElectionService.hasLeadership) {
            log.info(s"Scheduling job $jobId ($jobName).")

            executionGraph.scheduleForExecution()

          } else {
            self ! decorateMessage(RemoveJob(jobId, removeJobFromStateBackend = false))

            log.warn(s"Submitted job $jobId, but not leader. The other leader needs to recover " +
              "this. I am not scheduling the job for execution.")

    ......
}

public void deploy() throws JobException {

        ......

		try {
			// good, we are allowed to deploy
			if (!slot.setExecutedVertex(this)) {
				throw new JobException("Could not assign the ExecutionVertex to the slot " + slot);
			}

			// race double check, did we fail/cancel and do we need to release the slot?
			if (this.state != DEPLOYING) {
				slot.releaseSlot();
				return;
			}

			if (LOG.isInfoEnabled()) {
				LOG.info(String.format("Deploying %s (attempt #%d) to %s", vertex.getTaskNameWithSubtaskIndex(),
						attemptNumber, getAssignedResourceLocation().getHostname()));
			}

			final TaskDeploymentDescriptor deployment = vertex.createDeploymentDescriptor(
				attemptId,
				slot,
				taskState,
				attemptNumber);

			final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();

			final CompletableFuture<Acknowledge> submitResultFuture = taskManagerGateway.submitTask(deployment, timeout);

            ......
		}
		catch (Throwable t) {
			markFailed(t);
			ExceptionUtils.rethrow(t);
		}
	}

val task = new Task(
        jobInformation,
        taskInformation,
        tdd.getExecutionAttemptId,
        tdd.getAllocationId,
        tdd.getSubtaskIndex,
        tdd.getAttemptNumber,
        tdd.getProducedPartitions,
        tdd.getInputGates,
        tdd.getTargetSlotNumber,
        tdd.getTaskStateHandles,
        memoryManager,
        ioManager,
        network,
        bcVarManager,
        taskManagerConnection,
        inputSplitProvider,
        checkpointResponder,
        blobCache,
        libCache,
        fileCache,
        config,
        taskMetricGroup,
        resultPartitionConsumableNotifier,
        partitionStateChecker,
        context.dispatcher)

// all good, we kick off the task, which performs its own initialization
task.startTaskThread()

sender ! decorateMessage(Acknowledge.get())

for (ResultPartitionDeploymentDescriptor desc: resultPartitionDeploymentDescriptors) {
	ResultPartitionID partitionId = new ResultPartitionID(desc.getPartitionId(), executionId);

	this.producedPartitions[counter] = new ResultPartition(
	    taskNameWithSubtaskAndId,
		this,
		jobId,
		partitionId,
		desc.getPartitionType(),
		desc.getNumberOfSubpartitions(),
		desc.getMaxParallelism(),
		networkEnvironment.getResultPartitionManager(),
		resultPartitionConsumableNotifier,
		ioManager,
		desc.sendScheduleOrUpdateConsumersMessage());		
	//为每个partition初始化对应的writer
	writers[counter] = new ResultPartitionWriter(producedPartitions[counter]);

	++counter;
}

// Consumed intermediate result partitions
this.inputGates = new SingleInputGate[inputGateDeploymentDescriptors.size()];
this.inputGatesById = new HashMap<>();

counter = 0;

for (InputGateDeploymentDescriptor inputGateDeploymentDescriptor: inputGateDeploymentDescriptors) {
	SingleInputGate gate = SingleInputGate.create(
		taskNameWithSubtaskAndId,
		jobId,
		executionId,
		inputGateDeploymentDescriptor,
		networkEnvironment,
		this,
		metricGroup.getIOMetricGroup());

	inputGates[counter] = gate;
	inputGatesById.put(gate.getConsumedResultId(), gate);

	++counter;
}

      while (true) {
	ExecutionState current = this.executionState;
	////如果当前的执行状态为CREATED，则将其设置为DEPLOYING状态
	if (current == ExecutionState.CREATED) {
		if (transitionState(ExecutionState.CREATED, ExecutionState.DEPLOYING)) {
			// success, we can start our work
			break;
		}
	}
	//如果当前执行状态为FAILED，则发出通知并退出run方法
	else if (current == ExecutionState.FAILED) {
		// we were immediately failed. tell the TaskManager that we reached our final state
		notifyFinalState();
		if (metrics != null) {
			metrics.close();
		}
		return;
	}
	//如果当前执行状态为CANCELING，则将其修改为CANCELED状态，并退出run
	else if (current == ExecutionState.CANCELING) {
		if (transitionState(ExecutionState.CANCELING, ExecutionState.CANCELED)) {
			// we were immediately canceled. tell the TaskManager that we reached our final state
			notifyFinalState();
			if (metrics != null) {
				metrics.close();
			}
			return;
		}
	}
	//否则说明发生了异常
	else {
		if (metrics != null) {
			metrics.close();
		}
		throw new IllegalStateException("Invalid state for beginning of operation of task " + this + '.');
	}
}

invokable.invoke();

// now load the task's invokable code
//通过反射生成对象
invokable = loadAndInstantiateInvokable(userCodeClassLoader, nameOfInvokableClass);

public final void invoke() throws Exception {

	boolean disposed = false;
    try {
			// -------- Initialize ---------
			//先做一些赋值操作
            ......

	// if the clock is not already set, then assign a default TimeServiceProvider
	//处理timer
	if (timerService == null) {
		ThreadFactory timerThreadFactory =
			new DispatcherThreadFactory(TRIGGER_THREAD_GROUP, "Time Trigger for " + getName());

		timerService = new SystemProcessingTimeService(this, getCheckpointLock(), timerThreadFactory);
	}

    //把之前JobGraph串起来的chain的信息形成实现
	operatorChain = new OperatorChain<>(this);
	headOperator = operatorChain.getHeadOperator();

	// task specific initialization
	//这个init操作的起名非常诡异，因为这里主要是处理算子采用了自定义的checkpoint检查机制的情况，但是起了一个非常大众脸的名字
	init();

	// save the work of reloading state, etc, if the task is already canceled
	if (canceled) {
		throw new CancelTaskException();
	}

	// -------- Invoke --------
	LOG.debug("Invoking {}", getName());

	// we need to make sure that any triggers scheduled in open() cannot be
	// executed before all operators are opened
	synchronized (lock) {

		// both the following operations are protected by the lock
		// so that we avoid race conditions in the case that initializeState()
		// registers a timer, that fires before the open() is called.

        //初始化操作符状态，主要是一些state啥的
		initializeState();
		//对于富操作符，执行其open操作
		openAllOperators();
	}

	// final check to exit early before starting to run
	f (canceled) {
	    throw new CancelTaskException();
	}

	// let the task do its work
	//真正开始执行的代码
	isRunning = true;
	run();

protected void run() throws Exception {
	// cache processor reference on the stack, to make the code more JIT friendly
	final StreamInputProcessor<IN> inputProcessor = this.inputProcessor;

	while (running && inputProcessor.processInput()) {
		// all the work happens in the "processInput" method
	}
}

public void init() throws Exception {
	StreamConfig configuration = getConfiguration();

	TypeSerializer<IN> inSerializer = configuration.getTypeSerializerIn1(getUserCodeClassLoader());
	int numberOfInputs = configuration.getNumberOfInputs();

	if (numberOfInputs > 0) {
		InputGate[] inputGates = getEnvironment().getAllInputGates();

		inputProcessor = new StreamInputProcessor<>(
				inputGates,
				inSerializer,
				this,
				configuration.getCheckpointMode(),
				getCheckpointLock(),
				getEnvironment().getIOManager(),
				getEnvironment().getTaskManagerInfo().getConfiguration(),
				getStreamStatusMaintainer(),
				this.headOperator);

		// make sure that stream tasks report their I/O statistics
		inputProcessor.setMetricGroup(getEnvironment().getMetricGroup().getIOMetricGroup());
	}
}

public boolean processInput() throws Exception {
	if (isFinished) {
		return false;
	}
	if (numRecordsIn == null) {
		numRecordsIn = ((OperatorMetricGroup) streamOperator.getMetricGroup()).getIOMetricGroup().getNumRecordsInCounter();
	}

      //这个while是用来处理单个元素的（不要想当然以为是循环处理元素的）
	while (true) {
	    //注意 1在下面
	    //2.接下来，会利用这个反序列化器得到下一个数据记录，并进行解析（是用户数据还是watermark等等），然后进行对应的操作
		if (currentRecordDeserializer != null) {
			DeserializationResult result = currentRecordDeserializer.getNextRecord(deserializationDelegate);

			if (result.isBufferConsumed()) {
				currentRecordDeserializer.getCurrentBuffer().recycle();
				currentRecordDeserializer = null;
			}

			if (result.isFullRecord()) {
				StreamElement recordOrMark = deserializationDelegate.getInstance();

                  //如果元素是watermark，就准备更新当前channel的watermark值（并不是简单赋值，因为有乱序存在），
				if (recordOrMark.isWatermark()) {
					// handle watermark
					statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);
					continue;
				} else if (recordOrMark.isStreamStatus()) {
				//如果元素是status，就进行相应处理。可以看作是一个flag，标志着当前stream接下来即将没有元素输入（idle），或者当前即将由空闲状态转为有元素状态（active）。同时，StreamStatus还对如何处理watermark有影响。通过发送status，上游的operator可以很方便的通知下游当前的数据流的状态。
					// handle stream status
					statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);
					continue;
				} else if (recordOrMark.isLatencyMarker()) {
				//LatencyMarker是用来衡量代码执行时间的。在Source处创建，携带创建时的时间戳，流到Sink时就可以知道经过了多长时间
					// handle latency marker
					synchronized (lock) {
						streamOperator.processLatencyMarker(recordOrMark.asLatencyMarker());
					}
					continue;
				} else {
				//这里就是真正的，用户的代码即将被执行的地方。从章节1到这里足足用了三万字，有点万里长征的感觉
					// now we can do the actual processing
					StreamRecord<IN> record = recordOrMark.asRecord();
					synchronized (lock) {
						numRecordsIn.inc();
						streamOperator.setKeyContextElement1(record);
						streamOperator.processElement(record);
					}
					return true;
				}
			}
		}

          //1.程序首先获取下一个buffer
          //这一段代码是服务于Flink的FaultTorrent机制的，后面我会讲到，这里只需理解到它会尝试获取buffer，然后赋值给当前的反序列化器
		final BufferOrEvent bufferOrEvent = barrierHandler.getNextNonBlocked();
		if (bufferOrEvent != null) {
			if (bufferOrEvent.isBuffer()) {
				currentChannel = bufferOrEvent.getChannelIndex();
				currentRecordDeserializer = recordDeserializers[currentChannel];
				currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());
			}
			else {
				// Event received
				final AbstractEvent event = bufferOrEvent.getEvent();
				if (event.getClass() != EndOfPartitionEvent.class) {
					throw new IOException("Unexpected event: " + event);
				}
			}
		}
		else {
			isFinished = true;
			if (!barrierHandler.isEmpty()) {
				throw new IllegalStateException("Trailing data in checkpoint barrier handler.");
			}
			return false;
		}
	}
}