[FLINK-35024][Runtime/State] Implement the record buffer of AsyncExecutionController #24633

fredia · 2024-04-08T10:28:08Z

What is the purpose of the change

As a part of FLIP-425, this pr implements the record buffer of AsyncExecutionController.

Brief change log

Introduce activeBuffer , blockingBuffer and inFlightRecordNum into AsyncExecutionController.
Control the in-flight record number in AsyncExecutionController

Verifying this change

Add AsyncExecutionControllerTest#testInFlightRecordControl

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (yes)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (JavaDocs)

flinkbot · 2024-04-08T10:33:50Z

CI report:

e0a6f7e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Zakelly

Thanks for the PR! I post some concerns in advance!

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/RecordContext.java

...runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java

Zakelly · 2024-04-10T02:52:24Z

...runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java

+
+    /** Migrate the blocking requests to the active buffer. */
+    @VisibleForTesting
+    void migrateBlockingToActive() {


Personally I'd suggest a queueing mechanism, meaning that each blocked request will queue under one key. While the active request/context's reference counting reached zero, it pop one request from the corresponding queue of that key and put that into active buffer. I think this is a high-performance implementation.

Thanks for the suggestion, I add a new class StateRequestsBuffer which groups state requests in the blocking buffer by key. Migrating one state request from blocking buffer to active buffer will trigger in RecordContext#disposer.

Zakelly

Thanks for your update! I left some further comments. PTAL.

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestsBuffer.java

...runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java

Zakelly · 2024-04-10T10:44:46Z

...runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java

+            // 3. Ensure the currentContext is restored.
+            setCurrentContext(storedContext);
+            inFlightRecordNum.incrementAndGet();
+        } catch (InterruptedException e) {


IIUC, the InterruptedException should be ignored, otherwise it will produce a fatal error during TM normal exit (required by JM) ?

👍good catch, Thread.sleep(50); is deleted, so the InterruptedException wouldn't throw.

Zakelly · 2024-04-10T10:48:40Z

...runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java

+        // be less than the max in-flight record number.
+        // Note: the currentContext may be updated by {@code StateFutureFactory#build}.
+        try {
+            while (inFlightRecordNum.get() > maxInFlightRecordNum) {


I think we could clarify the definition of inFlightRecordNum... IIUC currently the inFlightRecordNum == keyAccountingUnit.occupiedCount. I'm not sure which size/count should we control, the record in AEC or the running record?

inFlightRecordNum is the records in AEC, including the records in active buffer and blocking buffer.
And the inFlightRecordNum == keyAccountingUnit.occupiedCount may not always be true.

I rewrited the description and added some asserts in testRecordsRunInOrder and testBasicRun.

fredia · 2024-04-11T03:15:10Z

@Zakelly Thanks for the suggestions, I updated and rebased this PR, please take a look again.

Zakelly

Thanks for your update!

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java

Zakelly · 2024-04-11T04:41:26Z

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java

+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+@NotThreadSafe


Better add some description about only manipulating this class within task thread

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java

...runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java

Zakelly

Thanks for the update. LGTM only one minor thing.

Zakelly · 2024-04-11T12:01:55Z

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java

+     * @param N the number of state requests to pop.
+     * @return A list of state requests.
+     */
+    List<StateRequest<?, ?, ?>> popActive(int N) {


Suggested change

List<StateRequest<?, ?, ?>> popActive(int N) {

List<StateRequest<?, ?, ?>> popActive(int n) {

fredia · 2024-04-12T02:19:29Z

@Zakelly Thanks for the review, updated and squashed.

masteryhx

Thanks for the PR.
I just left some minor suggestions. PTAL.

flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java

...ime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java

yunfengzhou-hub

Thanks for the PR! I checked the changes in production code and it LGTM.

masteryhx

Thanks for the update. LGTM.

…utionController

…utionController (apache#24633)

flinkbot added the component=Runtime/Task label Apr 8, 2024

fredia force-pushed the FLINK-35024 branch from 14fbde3 to 8c78128 Compare April 8, 2024 12:45

Zakelly reviewed Apr 10, 2024

View reviewed changes

fredia force-pushed the FLINK-35024 branch from 8c78128 to 242b161 Compare April 10, 2024 09:34

Zakelly reviewed Apr 10, 2024

View reviewed changes

fredia force-pushed the FLINK-35024 branch from 242b161 to f002d4b Compare April 11, 2024 02:27

fredia marked this pull request as ready for review April 11, 2024 02:27

fredia force-pushed the FLINK-35024 branch from f002d4b to 4a1aa87 Compare April 11, 2024 03:14

fredia requested review from curcur and masteryhx April 11, 2024 03:16

Zakelly reviewed Apr 11, 2024

View reviewed changes

Zakelly approved these changes Apr 11, 2024

View reviewed changes

fredia force-pushed the FLINK-35024 branch from e2d3038 to 0bb6d43 Compare April 12, 2024 02:18

fredia force-pushed the FLINK-35024 branch from 0bb6d43 to c87f422 Compare April 12, 2024 02:35

masteryhx reviewed Apr 12, 2024

View reviewed changes

yunfengzhou-hub approved these changes Apr 12, 2024

View reviewed changes

fredia force-pushed the FLINK-35024 branch from c87f422 to e51c471 Compare April 12, 2024 10:06

masteryhx approved these changes Apr 12, 2024

View reviewed changes

[FLINK-35024][Runtime/State] Implement the record buffer of AsyncExec…

e0a6f7e

…utionController

fredia force-pushed the FLINK-35024 branch from e51c471 to e0a6f7e Compare April 15, 2024 02:13

fredia merged commit 6d139a1 into apache:master Apr 15, 2024

fredia deleted the FLINK-35024 branch April 15, 2024 06:28

hanyuzheng7 pushed a commit to hanyuzheng7/flink that referenced this pull request May 6, 2024

[FLINK-35024][Runtime/State] Implement the record buffer of AsyncExec…

86e7f2d

…utionController (apache#24633)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-35024][Runtime/State] Implement the record buffer of AsyncExecutionController #24633

[FLINK-35024][Runtime/State] Implement the record buffer of AsyncExecutionController #24633

fredia commented Apr 8, 2024 •

edited

Loading

flinkbot commented Apr 8, 2024 •

edited

Loading

Zakelly left a comment

Zakelly Apr 10, 2024

fredia Apr 10, 2024

Zakelly left a comment

Zakelly Apr 10, 2024

fredia Apr 11, 2024

Zakelly Apr 10, 2024

fredia Apr 11, 2024

fredia commented Apr 11, 2024

Zakelly left a comment

Zakelly Apr 11, 2024

Zakelly left a comment

Zakelly Apr 11, 2024

fredia commented Apr 12, 2024

masteryhx left a comment

yunfengzhou-hub left a comment •

edited

Loading

masteryhx left a comment

	List<StateRequest<?, ?, ?>> popActive(int N) {
	List<StateRequest<?, ?, ?>> popActive(int n) {

[FLINK-35024][Runtime/State] Implement the record buffer of AsyncExecutionController #24633

[FLINK-35024][Runtime/State] Implement the record buffer of AsyncExecutionController #24633

Conversation

fredia commented Apr 8, 2024 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Apr 8, 2024 • edited Loading

CI report:

Zakelly left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Zakelly left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fredia commented Apr 11, 2024

Zakelly left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Zakelly left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fredia commented Apr 12, 2024

masteryhx left a comment

Choose a reason for hiding this comment

yunfengzhou-hub left a comment • edited Loading

Choose a reason for hiding this comment

masteryhx left a comment

Choose a reason for hiding this comment

fredia commented Apr 8, 2024 •

edited

Loading

flinkbot commented Apr 8, 2024 •

edited

Loading

yunfengzhou-hub left a comment •

edited

Loading