Flink 基于 MailBox 实现的 StreamTask 线程模型

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

BlockingQueue<Runnable> mailbox = ...
void runMailboxProcessing() {
//TODO: can become a cancel-event through mailbox eventually
    Runnable letter;
    while (isRunning()) {
        while ((letter = mailbox.poll()) != null) { letter.run();
            letter.run();
        }
        defaultAction();
    }
}
void defaultAction() {
// e.g. event-processing from an input
}

1

while (running && inputProcessor.processInput())

1

inputProcessor.processInput() // 每次触发，都相当于处理一个有限流

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

// org.apache.flink.streaming.runtime.tasks.StreamTask
public final void invoke() throws Exception {
    try {
        beforeInvoke();

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

        // let the task do its work
        isRunning = true;
        runMailboxLoop();

        // if this left the run() method cleanly despite the fact that this was canceled,
        // make sure the "clean shutdown" is not attempted
        if (canceled) {
            throw new CancelTaskException();
        }

        afterInvoke();
    }
    finally {
        cleanUpInvoke();
    }
}

private void runMailboxLoop() throws Exception {
    //note: mailbox 处理
    try {
        mailboxProcessor.runMailboxLoop();
    }
    catch (Exception e) {
        Optional<InterruptedException> interruption = ExceptionUtils.findThrowable(e, InterruptedException.class);
        if (interruption.isPresent()) {
            if (!canceled) {
                Thread.currentThread().interrupt();
                throw interruption.get();
            }
        } else if (canceled) {
            LOG.warn("Error while canceling task.", e);
        }
        else {
            throw e;
        }
    }
}

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

//org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor
/**
 * Runs the mailbox processing loop. This is where the main work is done.
 * note: mailbox 处理核心流程
 */
public void runMailboxLoop() throws Exception {

    final TaskMailbox localMailbox = mailbox;

    Preconditions.checkState(
        localMailbox.isMailboxThread(),
        "Method must be executed by declared mailbox thread!");

    //note: MailBox 的状态必须是 OPEN，才能继续循环
    assert localMailbox.getState() == TaskMailbox.State.OPEN : "Mailbox must be opened!";

    final MailboxController defaultActionContext = new MailboxController(this);

    while (processMail(localMailbox)) { //note: 如果有 mail 需要处理，这里会进行相应的处理，处理完才会进行下面的 event processing
        //note: 进行 task 的 default action，也就是调用 processInput()
        mailboxDefaultAction.runDefaultAction(defaultActionContext); // lock is acquired inside default action as needed
    }
}

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

//org.apache.flink.streaming.runtime.tasks.StreamTask
/**
 * This method implements the default action of the task (e.g. processing one event from the input). Implementations
 * should (in general) be non-blocking.
 * note: 这个方法执行这个 task 默认的 action
 *
 * @param controller controller object for collaborative interaction between the action and the stream task.
 * @throws Exception on any problems in the action.
 */
protected void processInput(MailboxDefaultAction.Controller controller) throws Exception {
    InputStatus status = inputProcessor.processInput(); //note: event 处理
    if (status == InputStatus.MORE_AVAILABLE && recordWriter.isAvailable()) {
        //note: 如果输入还有数据，并且 writer 是可用的，这里就直接返回了
        return;
    }
    if (status == InputStatus.END_OF_INPUT) {
        //note: 输入已经处理完了，会调用这个方法
        controller.allActionsCompleted();
        return;
    }
    CompletableFuture<?> jointFuture = getInputOutputJointFuture(status);
    //note: 告诉 MailBox 先暂停 loop
    MailboxDefaultAction.Suspension suspendedDefaultAction = controller.suspendDefaultAction();
    //note: 等待 future 完成后，继续 mailbox loop（等待 input 和 output 可用后，才会继续）
    jointFuture.thenRun(suspendedDefaultAction::resume);
}

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

//org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor
@Override
public Future<Boolean> triggerCheckpointAsync(
        CheckpointMetaData checkpointMetaData,
        CheckpointOptions checkpointOptions,
boolean advanceToEndOfEventTime) {

    //note: checkpoint 触发时，提交相应的 task
    return mailboxProcessor.getMainMailboxExecutor().submit(
            () -> triggerCheckpoint(checkpointMetaData, checkpointOptions, advanceToEndOfEventTime),
            "checkpoint %s with %s",
        checkpointMetaData,
        checkpointOptions);
}

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

//org.apache.flink.streaming.runtime.tasks.SourceStreamTask
@Override
protected void processInput(MailboxDefaultAction.Controller controller) throws Exception {

    //note: 告诉 MailBox 先暂停 loop
    controller.suspendDefaultAction();

    // Against the usual contract of this method, this implementation is not step-wise but blocking instead for
    // compatibility reasons with the current source interface (source functions run as a loop, not in steps).
    sourceThread.setTaskDescription(getName());
    sourceThread.start();
    sourceThread.getCompletionFuture().whenComplete((Void ignore, Throwable sourceThreadThrowable) -> {
        if (sourceThreadThrowable == null || isFinished) {
            //note: sourceThread 完成后，没有抛出异常或 task 完成的情况下
            mailboxProcessor.allActionsCompleted();
        } else {
            //note: 没有完成但结束了或者抛出异常的情况下
            mailboxProcessor.reportThrowable(sourceThreadThrowable);
        }
    });
}


/**
 * Runnable that executes the the source function in the head operator.
 * note: source 产生 data 的一个线程
 */
private class LegacySourceFunctionThread extends Thread {

    private final CompletableFuture<Void> completionFuture;

    LegacySourceFunctionThread() {
        this.completionFuture = new CompletableFuture<>();
    }

    @Override
    public void run() {
        try {
            //note: 调用 source Operator 的 run
            headOperator.run(getCheckpointLock(), getStreamStatusMaintainer(), operatorChain);
            completionFuture.complete(null);
        } catch (Throwable t) {
            // Note, t can be also an InterruptedException
            completionFuture.completeExceptionally(t);
        }
    }

    public void setTaskDescription(final String taskDescription) {
        setName("Legacy Source Thread - " + taskDescription);
    }

    CompletableFuture<Void> getCompletionFuture() {
        return completionFuture;
    }
}