Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
flinkbot commented on PR #24667: URL: https://github.com/apache/flink/pull/24667#issuecomment-2056402383 ## CI report: * aea56668e33e7062e6b18dac4086a2f05dc36fc1 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run azure` re-run the last Azure build -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on PR #24667: URL: https://github.com/apache/flink/pull/24667#issuecomment-2058174500 @Zakelly @ljz2051 would you please take a look? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1566744878 ## flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java: ## @@ -1085,6 +1085,54 @@ public void setUseSnapshotCompression(boolean useSnapshotCompression) { configuration.set(ExecutionOptions.SNAPSHOT_COMPRESSION, useSnapshotCompression); } +// +// Asynchronous execution configurations +// + +@Internal Review Comment: It might be better to mark them as `@Experimental` instead of `@Internal`. - Internal: Annotation to mark methods within **stable**, public APIs as an internal developer API. - Experimental: Classes with this annotation are neither battle-tested **nor stable**, and may be changed or removed in future versions. ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,73 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +/** + * A flag to enable or disable async mode related components when tasks initialize. As long as + * this option is enabled, the state access of Async state APIs will be executed asynchronously. + * Otherwise, the state access of Async state APIs will be executed synchronously. For Sync + * state APIs, the state access is always executed synchronously, enable this option would bring + * some overhead. + * + * Note: This is an experimental feature(FLIP-425) under evaluation. + */ +@Experimental +public static final ConfigOption ASYNC_STATE_ENABLED = +ConfigOptions.key("execution.async-mode.enabled") +.booleanType() +.defaultValue(false) +.withDescription( +"A flag to enable or disable async mode related components when tasks initialize." ++ " As long as this option is enabled, the state access of Async state APIs will be executed asynchronously." ++ " Otherwise, the state access of Async state APIs will be executed synchronously." ++ " For Sync state APIs, the state access is always executed synchronously, enable this option would bring some overhead.\n" Review Comment: It might be better to remove this configuration for now and add it in future when such use cases are found. - Whether to enable async state access can be inferred automatically by the Flink infrastructure, depending on the location of state backends and the state API used in operators. - It is better not to expose implementation details, like the sync/async modes mentioned here, to end users. So long as the order to same-key records and the order of async state callbacks are guaranteed, that is enough for users. - Especially, the implementation in the sync mode might be altered in future, in order to improve performance in situations where sync state API + remote state backend is used. In this case the statement "the state access is always executed synchronously" should cause deprecation issues. ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,73 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +/** + * A flag to enable or disable async mode related components when tasks initialize. As long as + * this option is enabled, the state access of Async state APIs will be executed asynchronously. + * Otherwise, the state access of Async state APIs will be executed synchronously. For Sync + * state APIs, the state access is always executed synchronously, enable this option would bring + * some overhead. + * + * Note: This is an experimental feature(FLIP-425) under evaluation. + */ +@Experimental +public static final ConfigOption ASYNC_STATE_ENABLED = +ConfigOptions.key("execution.async-mode.enabled") +.booleanType() +.defaultValue(false) +.withDescription( +"A flag to enable or disable async mode related components when tasks initialize." ++ " As long as this option i
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1566742123 ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,73 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +/** + * A flag to enable or disable async mode related components when tasks initialize. As long as + * this option is enabled, the state access of Async state APIs will be executed asynchronously. + * Otherwise, the state access of Async state APIs will be executed synchronously. For Sync + * state APIs, the state access is always executed synchronously, enable this option would bring + * some overhead. + * + * Note: This is an experimental feature(FLIP-425) under evaluation. + */ +@Experimental +public static final ConfigOption ASYNC_STATE_ENABLED = +ConfigOptions.key("execution.async-mode.enabled") +.booleanType() +.defaultValue(false) +.withDescription( +"A flag to enable or disable async mode related components when tasks initialize." ++ " As long as this option is enabled, the state access of Async state APIs will be executed asynchronously." ++ " Otherwise, the state access of Async state APIs will be executed synchronously." ++ " For Sync state APIs, the state access is always executed synchronously, enable this option would bring some overhead.\n" Review Comment: It might be better to remove this configuration for now and add it in future when such use cases are found. - Whether to enable async state access can be inferred automatically by the Flink infrastructure, depending on the location of state backends and the state API used in operators. - It is better not to expose implementation details, like the sync/async modes mentioned here, to end users. So long as the order to same-key records and the order of async state callbacks are guaranteed, that is enough for users. - Especially, the implementation in the sync mode might be altered in future, in order to improve performance in situations where sync state API + remote state backend is used. In this case the statement "the state access is always executed synchronously" might cause deprecation issues. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1566744878 ## flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java: ## @@ -1085,6 +1085,54 @@ public void setUseSnapshotCompression(boolean useSnapshotCompression) { configuration.set(ExecutionOptions.SNAPSHOT_COMPRESSION, useSnapshotCompression); } +// +// Asynchronous execution configurations +// + +@Internal Review Comment: It might be better to mark them as `@Experimental` instead of `@Internal`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1568656329 ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,73 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +/** + * A flag to enable or disable async mode related components when tasks initialize. As long as + * this option is enabled, the state access of Async state APIs will be executed asynchronously. + * Otherwise, the state access of Async state APIs will be executed synchronously. For Sync + * state APIs, the state access is always executed synchronously, enable this option would bring + * some overhead. + * + * Note: This is an experimental feature(FLIP-425) under evaluation. + */ +@Experimental +public static final ConfigOption ASYNC_STATE_ENABLED = +ConfigOptions.key("execution.async-mode.enabled") +.booleanType() +.defaultValue(false) +.withDescription( +"A flag to enable or disable async mode related components when tasks initialize." ++ " As long as this option is enabled, the state access of Async state APIs will be executed asynchronously." ++ " Otherwise, the state access of Async state APIs will be executed synchronously." ++ " For Sync state APIs, the state access is always executed synchronously, enable this option would bring some overhead.\n" Review Comment: Thanks for your insight, after discussing offline, we think it would be better to expose it on the data stream API in some way so that users can enable async execution in a more fine-grained manner. So, this option is removed now. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1568659638 ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,73 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +/** + * A flag to enable or disable async mode related components when tasks initialize. As long as + * this option is enabled, the state access of Async state APIs will be executed asynchronously. + * Otherwise, the state access of Async state APIs will be executed synchronously. For Sync + * state APIs, the state access is always executed synchronously, enable this option would bring + * some overhead. + * + * Note: This is an experimental feature(FLIP-425) under evaluation. + */ +@Experimental +public static final ConfigOption ASYNC_STATE_ENABLED = +ConfigOptions.key("execution.async-mode.enabled") +.booleanType() +.defaultValue(false) +.withDescription( +"A flag to enable or disable async mode related components when tasks initialize." ++ " As long as this option is enabled, the state access of Async state APIs will be executed asynchronously." ++ " Otherwise, the state access of Async state APIs will be executed synchronously." ++ " For Sync state APIs, the state access is always executed synchronously, enable this option would bring some overhead.\n" ++ " Note: This is an experimental feature under evaluation."); + +/** + * The max limit of in-flight records number in async execution mode, 'in-flight' refers to the + * records that have entered the operator but have not yet been processed and emitted to the + * downstream. If the in-flight records number exceeds the limit, the newly records entering + * will be blocked until the in-flight records number drops below the limit. + */ +@Experimental +public static final ConfigOption ASYNC_INFLIGHT_RECORDS_LIMIT = +ConfigOptions.key("execution.async-mode.in-flight-records-limit") +.intType() +.defaultValue(6000) +.withDescription( +"The max limit of in-flight records number in async execution mode, 'in-flight' refers" ++ " to the records that have entered the operator but have not yet been processed and" ++ " emitted to the downstream. If the in-flight records number exceeds the limit," ++ " the newly records entering will be blocked until the in-flight records number drops below the limit."); + +/** + * The size of buffer under async execution mode. Async execution mode provides a buffer + * mechanism to reduce state access. When the number of state requests in the buffer exceeds the + * batch size, a batched state execution would be triggered. Larger batch sizes will bring + * higher end-to-end latency, this option works with {@link #ASYNC_BUFFER_TIMEOUT} to control + * the frequency of triggering. + */ +@Experimental +public static final ConfigOption ASYNC_BUFFER_SIZE = +ConfigOptions.key("execution.async-mode.buffer-size") +.intType() +.defaultValue(1000) +.withDescription( +"The size of buffer under async execution mode. Async execution mode provides a buffer mechanism to reduce state access." ++ " When the number of state requests in the active buffer exceeds the batch size," ++ " a batched state execution would be triggered. Larger batch sizes will bring higher end-to-end latency," ++ " this option works with 'execution.async-state.buffer-timeout' to control the frequency of triggering."); + +/** + * The timeout of buffer triggering in milliseconds. If the buffer has not reached the {@link + * #ASYNC_BUFFER_SIZE} within 'buffer-timeout' milliseconds, a trigger will perform actively. + */ +@Experimental +public static final ConfigOption ASYNC_BUFFER_TIMEOUT = +ConfigOptions.key("execution.async-state.buffer-timeout") Review Comment: Unify those options to `execution.async-mode`, cause we also provide `Record-order` mode to preserve the order of records that without state access. -- This is an automated messag
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on PR #24667: URL: https://github.com/apache/flink/pull/24667#issuecomment-2060999083 @yunfengzhou-hub @Zakelly Thanks for the detailed review, I have rebased this PR and addressed some comments, PTAL if you are free. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1569768663 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,19 +103,26 @@ public class AsyncExecutionController { final AtomicInteger inFlightRecordNum; public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); +this( +mailboxExecutor, +stateExecutor, +DEFAULT_BATCH_SIZE, +DEFAULT_BUFFER_TIMEOUT, +DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); Review Comment: Given that #24657 has been merged, it might be better to verify that the introduced configurations can pass the configured values into AEC through operator setups now. ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,73 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +/** + * A flag to enable or disable async mode related components when tasks initialize. As long as + * this option is enabled, the state access of Async state APIs will be executed asynchronously. + * Otherwise, the state access of Async state APIs will be executed synchronously. For Sync + * state APIs, the state access is always executed synchronously, enable this option would bring + * some overhead. + * + * Note: This is an experimental feature(FLIP-425) under evaluation. + */ +@Experimental +public static final ConfigOption ASYNC_STATE_ENABLED = +ConfigOptions.key("execution.async-mode.enabled") +.booleanType() +.defaultValue(false) +.withDescription( +"A flag to enable or disable async mode related components when tasks initialize." ++ " As long as this option is enabled, the state access of Async state APIs will be executed asynchronously." ++ " Otherwise, the state access of Async state APIs will be executed synchronously." ++ " For Sync state APIs, the state access is always executed synchronously, enable this option would bring some overhead.\n" ++ " Note: This is an experimental feature under evaluation."); + +/** + * The max limit of in-flight records number in async execution mode, 'in-flight' refers to the + * records that have entered the operator but have not yet been processed and emitted to the + * downstream. If the in-flight records number exceeds the limit, the newly records entering + * will be blocked until the in-flight records number drops below the limit. + */ +@Experimental +public static final ConfigOption ASYNC_INFLIGHT_RECORDS_LIMIT = +ConfigOptions.key("execution.async-mode.in-flight-records-limit") +.intType() +.defaultValue(6000) +.withDescription( +"The max limit of in-flight records number in async execution mode, 'in-flight' refers" ++ " to the records that have entered the operator but have not yet been processed and" ++ " emitted to the downstream. If the in-flight records number exceeds the limit," ++ " the newly records entering will be blocked until the in-flight records number drops below the limit."); + +/** + * The size of buffer under async execution mode. Async execution mode provides a buffer + * mechanism to reduce state access. When the number of state requests in the buffer exceeds the + * batch size, a batched state execution would be triggered. Larger batch sizes will bring + * higher end-to-end latency, this option works with {@link #ASYNC_BUFFER_TIMEOUT} to control + * the frequency of triggering. + */ +@Experimental +public static final ConfigOption ASYNC_BUFFER_SIZE = +ConfigOptions.key("execution.async-mode.buffer-size") +.intType() +.defaultValue(1000) +.withDescription( +"The size of buffer under async execution mode. Async execution mode provides a buffer mechanism to reduce state access." ++ " When the number of state requests in the active buffer exceeds the batch size," +
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1569768663 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,19 +103,26 @@ public class AsyncExecutionController { final AtomicInteger inFlightRecordNum; public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); +this( +mailboxExecutor, +stateExecutor, +DEFAULT_BATCH_SIZE, +DEFAULT_BUFFER_TIMEOUT, +DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); Review Comment: Given that #24657 has been merged, it might be better to verify that the introduced configurations can pass the configured values into AEC through operator setups now. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1569863724 ## flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java: ## @@ -1085,6 +1085,54 @@ public void setUseSnapshotCompression(boolean useSnapshotCompression) { configuration.set(ExecutionOptions.SNAPSHOT_COMPRESSION, useSnapshotCompression); } +// +// Asynchronous execution configurations +// + +@Internal Review Comment: Ah, I misread the comment, and changed them from `@Experimental` to `@Internal`, I have corrected them now. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1569864750 ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,73 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +/** + * A flag to enable or disable async mode related components when tasks initialize. As long as + * this option is enabled, the state access of Async state APIs will be executed asynchronously. + * Otherwise, the state access of Async state APIs will be executed synchronously. For Sync + * state APIs, the state access is always executed synchronously, enable this option would bring + * some overhead. + * + * Note: This is an experimental feature(FLIP-425) under evaluation. + */ +@Experimental +public static final ConfigOption ASYNC_STATE_ENABLED = +ConfigOptions.key("execution.async-mode.enabled") +.booleanType() +.defaultValue(false) +.withDescription( +"A flag to enable or disable async mode related components when tasks initialize." ++ " As long as this option is enabled, the state access of Async state APIs will be executed asynchronously." ++ " Otherwise, the state access of Async state APIs will be executed synchronously." ++ " For Sync state APIs, the state access is always executed synchronously, enable this option would bring some overhead.\n" ++ " Note: This is an experimental feature under evaluation."); + +/** + * The max limit of in-flight records number in async execution mode, 'in-flight' refers to the + * records that have entered the operator but have not yet been processed and emitted to the + * downstream. If the in-flight records number exceeds the limit, the newly records entering + * will be blocked until the in-flight records number drops below the limit. + */ +@Experimental +public static final ConfigOption ASYNC_INFLIGHT_RECORDS_LIMIT = +ConfigOptions.key("execution.async-mode.in-flight-records-limit") +.intType() +.defaultValue(6000) +.withDescription( +"The max limit of in-flight records number in async execution mode, 'in-flight' refers" ++ " to the records that have entered the operator but have not yet been processed and" ++ " emitted to the downstream. If the in-flight records number exceeds the limit," ++ " the newly records entering will be blocked until the in-flight records number drops below the limit."); + +/** + * The size of buffer under async execution mode. Async execution mode provides a buffer + * mechanism to reduce state access. When the number of state requests in the buffer exceeds the + * batch size, a batched state execution would be triggered. Larger batch sizes will bring + * higher end-to-end latency, this option works with {@link #ASYNC_BUFFER_TIMEOUT} to control + * the frequency of triggering. + */ +@Experimental +public static final ConfigOption ASYNC_BUFFER_SIZE = +ConfigOptions.key("execution.async-mode.buffer-size") +.intType() +.defaultValue(1000) +.withDescription( +"The size of buffer under async execution mode. Async execution mode provides a buffer mechanism to reduce state access." ++ " When the number of state requests in the active buffer exceeds the batch size," ++ " a batched state execution would be triggered. Larger batch sizes will bring higher end-to-end latency," ++ " this option works with 'execution.async-state.buffer-timeout' to control the frequency of triggering."); + +/** + * The timeout of buffer triggering in milliseconds. If the buffer has not reached the {@link + * #ASYNC_BUFFER_SIZE} within 'buffer-timeout' milliseconds, a trigger will perform actively. + */ +@Experimental +public static final ConfigOption ASYNC_BUFFER_TIMEOUT = +ConfigOptions.key("execution.async-state.buffer-timeout") Review Comment: Good pointπ, changed it to `execution.async-mode.state-buffer-timeout`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to Git
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
Zakelly commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1571771428 ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,73 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +/** + * A flag to enable or disable async mode related components when tasks initialize. As long as + * this option is enabled, the state access of Async state APIs will be executed asynchronously. + * Otherwise, the state access of Async state APIs will be executed synchronously. For Sync + * state APIs, the state access is always executed synchronously, enable this option would bring + * some overhead. + * + * Note: This is an experimental feature(FLIP-425) under evaluation. + */ +@Experimental +public static final ConfigOption ASYNC_STATE_ENABLED = +ConfigOptions.key("execution.async-mode.enabled") +.booleanType() +.defaultValue(false) +.withDescription( +"A flag to enable or disable async mode related components when tasks initialize." ++ " As long as this option is enabled, the state access of Async state APIs will be executed asynchronously." ++ " Otherwise, the state access of Async state APIs will be executed synchronously." ++ " For Sync state APIs, the state access is always executed synchronously, enable this option would bring some overhead.\n" ++ " Note: This is an experimental feature under evaluation."); + +/** + * The max limit of in-flight records number in async execution mode, 'in-flight' refers to the + * records that have entered the operator but have not yet been processed and emitted to the + * downstream. If the in-flight records number exceeds the limit, the newly records entering + * will be blocked until the in-flight records number drops below the limit. + */ +@Experimental +public static final ConfigOption ASYNC_INFLIGHT_RECORDS_LIMIT = +ConfigOptions.key("execution.async-mode.in-flight-records-limit") +.intType() +.defaultValue(6000) +.withDescription( +"The max limit of in-flight records number in async execution mode, 'in-flight' refers" ++ " to the records that have entered the operator but have not yet been processed and" ++ " emitted to the downstream. If the in-flight records number exceeds the limit," ++ " the newly records entering will be blocked until the in-flight records number drops below the limit."); + +/** + * The size of buffer under async execution mode. Async execution mode provides a buffer + * mechanism to reduce state access. When the number of state requests in the buffer exceeds the + * batch size, a batched state execution would be triggered. Larger batch sizes will bring + * higher end-to-end latency, this option works with {@link #ASYNC_BUFFER_TIMEOUT} to control + * the frequency of triggering. + */ +@Experimental +public static final ConfigOption ASYNC_BUFFER_SIZE = +ConfigOptions.key("execution.async-mode.buffer-size") +.intType() +.defaultValue(1000) +.withDescription( +"The size of buffer under async execution mode. Async execution mode provides a buffer mechanism to reduce state access." ++ " When the number of state requests in the active buffer exceeds the batch size," ++ " a batched state execution would be triggered. Larger batch sizes will bring higher end-to-end latency," ++ " this option works with 'execution.async-state.buffer-timeout' to control the frequency of triggering."); + +/** + * The timeout of buffer triggering in milliseconds. If the buffer has not reached the {@link + * #ASYNC_BUFFER_SIZE} within 'buffer-timeout' milliseconds, a trigger will perform actively. + */ +@Experimental +public static final ConfigOption ASYNC_BUFFER_TIMEOUT = +ConfigOptions.key("execution.async-state.buffer-timeout") Review Comment: Sry guys, I'm afraid I prefer the original proposal of `async-state` in FLIP, since this is only for the stateful operator and even the problem of `record-order` is caused by state
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on PR #24667: URL: https://github.com/apache/flink/pull/24667#issuecomment-2066106851 After offline discussion, FLIP-425 is only used for stateful operators, hence ,`async-mode` was changed back to `async-state`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1572344539 ## flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java: ## @@ -181,4 +182,62 @@ public class ExecutionOptions { + " operators. NOTE: It takes effect only in the BATCH runtime mode and requires sorted inputs" + SORT_INPUTS.key() + " to be enabled."); + +// - Async State Execution -- + +/** + * The max limit of in-flight records number in async state execution, 'in-flight' refers to the + * records that have entered the operator but have not yet been processed and emitted to the + * downstream. If the in-flight records number exceeds the limit, the newly records entering + * will be blocked until the in-flight records number drops below the limit. + */ +@Experimental +@Documentation.ExcludeFromDocumentation( +"This is an experimental option, internal use only for now.") +public static final ConfigOption ASYNC_INFLIGHT_RECORDS_LIMIT = +ConfigOptions.key("execution.async-state.in-flight-records-limit") +.intType() +.defaultValue(6000) +.withDescription( +"The max limit of in-flight records number in async state execution, 'in-flight' refers" ++ " to the records that have entered the operator but have not yet been processed and" ++ " emitted to the downstream. If the in-flight records number exceeds the limit," ++ " the newly records entering will be blocked until the in-flight records number drops below the limit."); + +/** + * The size of buffer under async state execution. Async state execution provides a buffer + * mechanism to reduce state access. When the number of state requests in the buffer exceeds the + * batch size, a batched state execution would be triggered. Larger batch sizes will bring + * higher end-to-end latency, this option works with {@link #ASYNC_STATE_BUFFER_TIMEOUT} to + * control the frequency of triggering. + */ +@Experimental +@Documentation.ExcludeFromDocumentation( +"This is an experimental option, internal use only for now.") +public static final ConfigOption ASYNC_STATE_BUFFER_SIZE = +ConfigOptions.key("execution.async-state.buffer-size") +.intType() +.defaultValue(1000) +.withDescription( +"The size of buffer under async state execution. Async state execution provides a buffer mechanism to reduce state access." ++ " When the number of state requests in the active buffer exceeds the batch size," ++ " a batched state execution would be triggered. Larger batch sizes will bring higher end-to-end latency," ++ " this option works with 'execution.async-state.buffer-timeout' to control the frequency of triggering."); + +/** + * The timeout of buffer triggering in milliseconds. If the buffer has not reached the {@link + * #ASYNC_STATE_BUFFER_SIZE} within 'buffer-timeout' milliseconds, a trigger will perform + * actively. + */ +@Experimental +@Documentation.ExcludeFromDocumentation( +"This is an experimental option, internal use only for now.") +public static final ConfigOption ASYNC_STATE_BUFFER_TIMEOUT = Review Comment: It might be better to make time configurations as `long` instead of as `int`. ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -50,15 +50,24 @@ public class AsyncExecutionController { private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class); -public static final int DEFAULT_BATCH_SIZE = 1000; -public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000; +private static final int DEFAULT_BATCH_SIZE = 1000; + +private static final int DEFAULT_BUFFER_TIMEOUT = 1000; +private static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000; /** * The batch size. When the number of state requests in the active buffer exceeds the batch * size, a batched state execution would be triggered. */ private final int batchSize; +/** + * The timeout of {@link StateRequestBuffer#activeQueue} triggering in milliseconds. If the + * activeQueue has not reached the {@link #batchSize} within 'buffer-timeout' milliseconds, a + * trigger will perform actively. + */ +private final int bufferTimeOut; Review
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1574293924 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -50,15 +50,24 @@ public class AsyncExecutionController { private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class); -public static final int DEFAULT_BATCH_SIZE = 1000; -public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000; +private static final int DEFAULT_BATCH_SIZE = 1000; + +private static final int DEFAULT_BUFFER_TIMEOUT = 1000; +private static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000; /** * The batch size. When the number of state requests in the active buffer exceeds the batch * size, a batched state execution would be triggered. */ private final int batchSize; +/** + * The timeout of {@link StateRequestBuffer#activeQueue} triggering in milliseconds. If the + * activeQueue has not reached the {@link #batchSize} within 'buffer-timeout' milliseconds, a + * trigger will perform actively. + */ +private final int bufferTimeOut; Review Comment: I added an implementation in second commit. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
masteryhx commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1574501889 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; Review Comment: Should AEC become `Closeable` since we maintain some resources like this internally ? ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); +this( +mailboxExecutor, +stateExecutor, +DEFAULT_BATCH_SIZE, +DEFAULT_BUFFER_TIMEOUT, +DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); } public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeOut, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeOut = bufferTimeOut; this.maxInFlightRecordNum = maxInFlightRecords; this.stateRequestsBuffer = new StateRequestBuffer<>(); this.inFlightRecordNum = new AtomicInteger(0); +this.timeoutFlag = new AtomicBoolean(false); + +// - initialize buffer timeout --- +this.currentScheduledFuture = null; +if (bufferTimeOut > 0) { +this.scheduledExecutor = +new ScheduledThreadPoolExecutor( +1, +new ThreadFactory() { +@Override +public Thread newThread(Runnable r) { +return new Thread(r, "AEC-scheduler"); +} +}); +this.scheduledExecutor.setRemoveOnCancelPolicy(true); + +// make sure shutdown removes all pending tasks + this.scheduledExecutor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); + this.scheduledExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); +} else { +this.scheduledExecutor = null; +} + LOG.info( "Create AsyncExecutionController: batchSize {}, maxInFlightRecordsNum {}", Review Comment: nit: Also add `bufferTimeOut` here BTW, `bufferTimeOut` -> `bufferTimeout` ? ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { Review Comment: Seems it's not used and default value in this class could be removed. -- This is an automated message f
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
Zakelly commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1574559188 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); +this( +mailboxExecutor, +stateExecutor, +DEFAULT_BATCH_SIZE, +DEFAULT_BUFFER_TIMEOUT, +DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); } public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeOut, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeOut = bufferTimeOut; this.maxInFlightRecordNum = maxInFlightRecords; this.stateRequestsBuffer = new StateRequestBuffer<>(); this.inFlightRecordNum = new AtomicInteger(0); +this.timeoutFlag = new AtomicBoolean(false); + +// - initialize buffer timeout --- +this.currentScheduledFuture = null; +if (bufferTimeOut > 0) { +this.scheduledExecutor = +new ScheduledThreadPoolExecutor( +1, +new ThreadFactory() { +@Override +public Thread newThread(Runnable r) { +return new Thread(r, "AEC-scheduler"); +} +}); +this.scheduledExecutor.setRemoveOnCancelPolicy(true); + +// make sure shutdown removes all pending tasks + this.scheduledExecutor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); + this.scheduledExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); +} else { +this.scheduledExecutor = null; +} + LOG.info( "Create AsyncExecutionController: batchSize {}, maxInFlightRecordsNum {}", batchSize, maxInFlightRecords); } +void scheduleTimeout() { +if (bufferTimeOut > 0) { +if (currentScheduledFuture != null +&& !currentScheduledFuture.isDone() +&& !currentScheduledFuture.isCancelled()) { +currentScheduledFuture.cancel(false); +} +currentScheduledFuture = +(ScheduledFuture) +scheduledExecutor.schedule( +() -> { +timeoutFlag.set(true); +mailboxExecutor.execute( +() -> triggerIfNeeded(false), "AEC-timeout"); Review Comment: I'd suggest a `sequence number` for each buffer to make sure the timeout trigger the intended one. ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); +
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1575573086 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; Review Comment: π I added a `close()` method, 2 things include: 1. drain all in-flight records 2. shutdown `scheduledExecutor` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1575603596 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); +this( +mailboxExecutor, +stateExecutor, +DEFAULT_BATCH_SIZE, +DEFAULT_BUFFER_TIMEOUT, +DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); } public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeOut, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeOut = bufferTimeOut; this.maxInFlightRecordNum = maxInFlightRecords; this.stateRequestsBuffer = new StateRequestBuffer<>(); this.inFlightRecordNum = new AtomicInteger(0); +this.timeoutFlag = new AtomicBoolean(false); + +// - initialize buffer timeout --- +this.currentScheduledFuture = null; +if (bufferTimeOut > 0) { +this.scheduledExecutor = +new ScheduledThreadPoolExecutor( +1, +new ThreadFactory() { +@Override +public Thread newThread(Runnable r) { +return new Thread(r, "AEC-scheduler"); +} +}); +this.scheduledExecutor.setRemoveOnCancelPolicy(true); + +// make sure shutdown removes all pending tasks + this.scheduledExecutor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); + this.scheduledExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); +} else { +this.scheduledExecutor = null; +} + LOG.info( "Create AsyncExecutionController: batchSize {}, maxInFlightRecordsNum {}", batchSize, maxInFlightRecords); } +void scheduleTimeout() { +if (bufferTimeOut > 0) { +if (currentScheduledFuture != null +&& !currentScheduledFuture.isDone() +&& !currentScheduledFuture.isCancelled()) { +currentScheduledFuture.cancel(false); +} +currentScheduledFuture = +(ScheduledFuture) +scheduledExecutor.schedule( +() -> { +timeoutFlag.set(true); +mailboxExecutor.execute( +() -> triggerIfNeeded(false), "AEC-timeout"); Review Comment: It might be simpler to remove `timeoutFlag` and invoke `triggerIfNeeded(true)` directly. ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -51,15 +56,24 @@ public class AsyncExecutionController { private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class); -public static final int DEFAULT_BATCH_SIZE = 1000; -public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000; +private static final int DEFAULT_BATCH_SIZE = 1000; + +private static final int DEFAULT_BUFFER_TIMEOUT = 1000; Review Comment: How about remove the constants and use `ASYNC_STATE_BUFFER_SIZE.defaultValue()` from the `ExecutionConfig`? This can help avoid maintaining the same value in two places. ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInt
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1575604054 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); +this( +mailboxExecutor, +stateExecutor, +DEFAULT_BATCH_SIZE, +DEFAULT_BUFFER_TIMEOUT, +DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); } public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeOut, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeOut = bufferTimeOut; Review Comment: The "O" in "BufferTimeOut" is upper-case while the "o" in timeoutFlag is lower-case. It might be better to get them unified to the same convention. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1575609479 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; Review Comment: Would it be better to reuse existing utility methods like `FutureUtils.delay()`? This way AEC won't need to maintain such resources. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
yunfengzhou-hub commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1575597004 ## flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java: ## @@ -346,6 +337,63 @@ public void testSyncPoint() { recordContext2.release(); } +@Test +void testBufferTimeout() throws InterruptedException { +batchSize = 5; +timeout = 1000; +setup(); +Runnable userCode = () -> valueState.asyncValue(); + +// basic timeout --- +for (int i = 0; i < batchSize - 1; i++) { +String record = String.format("key%d-r%d", i, i); +String key = String.format("key%d", batchSize + i); +RecordContext recordContext = aec.buildContext(record, key); +aec.setCurrentContext(recordContext); +userCode.run(); +} +assertThat(aec.timeoutFlag.get()).isFalse(); +assertThat(aec.currentScheduledFuture.isDone()).isFalse(); +assertThat(aec.inFlightRecordNum.get()).isEqualTo(batchSize - 1); + assertThat(aec.stateRequestsBuffer.activeQueueSize()).isEqualTo(batchSize - 1); +assertThat(aec.stateRequestsBuffer.blockingQueueSize()).isEqualTo(0); + +Thread.sleep(timeout + 100); Review Comment: It might be better to avoid using `Thread.sleep` in test cases, as it may increase the duration of CI and behave as flaky tests. How about introduce a `TestScheduledThreadPoolExecutor` so we can control when each step is triggered? An example of this is `org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1575735682 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +108,91 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; +/** + * The flag to indicate whether the {@link #bufferTimeOut} is reached, if yes, a trigger will + * perform actively when the next state request arrives even if the activeQueue has not reached + * the {@link #batchSize}. + */ +final AtomicBoolean timeoutFlag; + +/** The executor service that schedules and calls the triggers of this task. */ +final ScheduledThreadPoolExecutor scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); +this( +mailboxExecutor, +stateExecutor, +DEFAULT_BATCH_SIZE, +DEFAULT_BUFFER_TIMEOUT, +DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); } public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeOut, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeOut = bufferTimeOut; this.maxInFlightRecordNum = maxInFlightRecords; this.stateRequestsBuffer = new StateRequestBuffer<>(); this.inFlightRecordNum = new AtomicInteger(0); +this.timeoutFlag = new AtomicBoolean(false); + +// - initialize buffer timeout --- +this.currentScheduledFuture = null; +if (bufferTimeOut > 0) { +this.scheduledExecutor = +new ScheduledThreadPoolExecutor( +1, +new ThreadFactory() { +@Override +public Thread newThread(Runnable r) { +return new Thread(r, "AEC-scheduler"); +} +}); +this.scheduledExecutor.setRemoveOnCancelPolicy(true); + +// make sure shutdown removes all pending tasks + this.scheduledExecutor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); + this.scheduledExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); +} else { +this.scheduledExecutor = null; +} + LOG.info( "Create AsyncExecutionController: batchSize {}, maxInFlightRecordsNum {}", batchSize, maxInFlightRecords); } +void scheduleTimeout() { +if (bufferTimeOut > 0) { +if (currentScheduledFuture != null +&& !currentScheduledFuture.isDone() +&& !currentScheduledFuture.isCancelled()) { +currentScheduledFuture.cancel(false); +} +currentScheduledFuture = +(ScheduledFuture) +scheduledExecutor.schedule( +() -> { +timeoutFlag.set(true); +mailboxExecutor.execute( +() -> triggerIfNeeded(false), "AEC-timeout"); Review Comment: Added, `testBufferTimeoutSkip` is introduced to test it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1575787928 ## flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java: ## @@ -346,6 +337,63 @@ public void testSyncPoint() { recordContext2.release(); } +@Test +void testBufferTimeout() throws InterruptedException { +batchSize = 5; +timeout = 1000; +setup(); +Runnable userCode = () -> valueState.asyncValue(); + +// basic timeout --- +for (int i = 0; i < batchSize - 1; i++) { +String record = String.format("key%d-r%d", i, i); +String key = String.format("key%d", batchSize + i); +RecordContext recordContext = aec.buildContext(record, key); +aec.setCurrentContext(recordContext); +userCode.run(); +} +assertThat(aec.timeoutFlag.get()).isFalse(); +assertThat(aec.currentScheduledFuture.isDone()).isFalse(); +assertThat(aec.inFlightRecordNum.get()).isEqualTo(batchSize - 1); + assertThat(aec.stateRequestsBuffer.activeQueueSize()).isEqualTo(batchSize - 1); +assertThat(aec.stateRequestsBuffer.blockingQueueSize()).isEqualTo(0); + +Thread.sleep(timeout + 100); Review Comment: Thanks for the suggestion, removed `Thread.sleep` from `testBufferTimeout()` and `testBufferTimeoutSkip()`, used `ManuallyTriggeredScheduledExecutorService` as `ScheduledExecutor`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
Zakelly commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1577589826 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +106,80 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; -public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); -} +/** The executor service that schedules and calls the triggers of this task. */ +ScheduledExecutorService scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + +/** + * The current trigger sequence number, used to distinguish different triggers. Every time a + * trigger occurs, {@code currentTriggerSeq} increases by one. + */ +AtomicLong currentTriggerSeq; public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeout, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeout = bufferTimeout; this.maxInFlightRecordNum = maxInFlightRecords; this.stateRequestsBuffer = new StateRequestBuffer<>(); this.inFlightRecordNum = new AtomicInteger(0); +this.currentTriggerSeq = new AtomicLong(0); + +// - initialize buffer timeout --- +this.currentScheduledFuture = null; +if (bufferTimeout > 0) { +this.scheduledExecutor = +new ScheduledThreadPoolExecutor( +1, new ExecutorThreadFactory("AEC-timeout-scheduler")); +((ScheduledThreadPoolExecutor) this.scheduledExecutor).setRemoveOnCancelPolicy(true); +// make sure shutdown removes all pending tasks +((ScheduledThreadPoolExecutor) this.scheduledExecutor) + .setContinueExistingPeriodicTasksAfterShutdownPolicy(false); +((ScheduledThreadPoolExecutor) this.scheduledExecutor) +.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); +} else { +this.scheduledExecutor = null; +} + LOG.info( -"Create AsyncExecutionController: batchSize {}, maxInFlightRecordsNum {}", +"Create AsyncExecutionController: batchSize {}, bufferTimeout {}, maxInFlightRecordsNum {}", batchSize, +bufferTimeout, maxInFlightRecords); } +void scheduleTimeout(long triggerSeq) { +if (bufferTimeout > 0) { +if (currentScheduledFuture != null +&& !currentScheduledFuture.isDone() +&& !currentScheduledFuture.isCancelled()) { +currentScheduledFuture.cancel(false); +} +currentScheduledFuture = +(ScheduledFuture) +scheduledExecutor.schedule( +() -> { +if (triggerSeq != currentTriggerSeq.get()) { +// if any new trigger occurs, skip this schedule +return; +} +mailboxExecutor.execute( +() -> triggerIfNeeded(true), "AEC-timeout"); Review Comment: how about ``` mailboxExecutor.execute(() -> {if (triggerSeq == currentTriggerSeq.get()) {triggerIfNeeded(true);}, "AEC-timeout"); ``` ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -220,12 +283,17 @@ void insertBlockingBuffer(StateRequest request) { * @param force whether to trigger requests in force. */ void triggerIfNeeded(boolean force) { -// TODO: introduce a timeout mechanism for triggering. if (!force && stateRequestsBuffer.activeQueueSize() < batchSize) { +// if the active queue size is 1, it means that the current request is the oldest one in +// the active queue. +if (stateRequestsBuffer.activeQueueSize() == 1) { +scheduleTimeout(currentTriggerSeq.get()); +} Review Comment: Is it possible trigger multiple times for one seq? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1577695901 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -220,12 +283,17 @@ void insertBlockingBuffer(StateRequest request) { * @param force whether to trigger requests in force. */ void triggerIfNeeded(boolean force) { -// TODO: introduce a timeout mechanism for triggering. if (!force && stateRequestsBuffer.activeQueueSize() < batchSize) { +// if the active queue size is 1, it means that the current request is the oldest one in +// the active queue. +if (stateRequestsBuffer.activeQueueSize() == 1) { +scheduleTimeout(currentTriggerSeq.get()); +} Review Comment: No, because the seq number will increase by 1 per trigger. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1577700082 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +106,80 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; -public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); -} +/** The executor service that schedules and calls the triggers of this task. */ +ScheduledExecutorService scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + +/** + * The current trigger sequence number, used to distinguish different triggers. Every time a + * trigger occurs, {@code currentTriggerSeq} increases by one. + */ +AtomicLong currentTriggerSeq; public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeout, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeout = bufferTimeout; this.maxInFlightRecordNum = maxInFlightRecords; this.stateRequestsBuffer = new StateRequestBuffer<>(); this.inFlightRecordNum = new AtomicInteger(0); +this.currentTriggerSeq = new AtomicLong(0); + +// - initialize buffer timeout --- +this.currentScheduledFuture = null; +if (bufferTimeout > 0) { +this.scheduledExecutor = +new ScheduledThreadPoolExecutor( +1, new ExecutorThreadFactory("AEC-timeout-scheduler")); +((ScheduledThreadPoolExecutor) this.scheduledExecutor).setRemoveOnCancelPolicy(true); +// make sure shutdown removes all pending tasks +((ScheduledThreadPoolExecutor) this.scheduledExecutor) + .setContinueExistingPeriodicTasksAfterShutdownPolicy(false); +((ScheduledThreadPoolExecutor) this.scheduledExecutor) +.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); +} else { +this.scheduledExecutor = null; +} + LOG.info( -"Create AsyncExecutionController: batchSize {}, maxInFlightRecordsNum {}", +"Create AsyncExecutionController: batchSize {}, bufferTimeout {}, maxInFlightRecordsNum {}", batchSize, +bufferTimeout, maxInFlightRecords); } +void scheduleTimeout(long triggerSeq) { +if (bufferTimeout > 0) { +if (currentScheduledFuture != null +&& !currentScheduledFuture.isDone() +&& !currentScheduledFuture.isCancelled()) { +currentScheduledFuture.cancel(false); +} +currentScheduledFuture = +(ScheduledFuture) +scheduledExecutor.schedule( +() -> { +if (triggerSeq != currentTriggerSeq.get()) { +// if any new trigger occurs, skip this schedule +return; +} +mailboxExecutor.execute( +() -> triggerIfNeeded(true), "AEC-timeout"); Review Comment: This way may create an extra email and put it in the mailbox. I lean toward skipping it directly before submitting it to the mailbox. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
Zakelly commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1578767927 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,29 +106,80 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; -public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); -} +/** The executor service that schedules and calls the triggers of this task. */ +ScheduledExecutorService scheduledExecutor; + +ScheduledFuture currentScheduledFuture; + +/** + * The current trigger sequence number, used to distinguish different triggers. Every time a + * trigger occurs, {@code currentTriggerSeq} increases by one. + */ +AtomicLong currentTriggerSeq; public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeout, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeout = bufferTimeout; this.maxInFlightRecordNum = maxInFlightRecords; this.stateRequestsBuffer = new StateRequestBuffer<>(); this.inFlightRecordNum = new AtomicInteger(0); +this.currentTriggerSeq = new AtomicLong(0); + +// - initialize buffer timeout --- +this.currentScheduledFuture = null; +if (bufferTimeout > 0) { +this.scheduledExecutor = +new ScheduledThreadPoolExecutor( +1, new ExecutorThreadFactory("AEC-timeout-scheduler")); +((ScheduledThreadPoolExecutor) this.scheduledExecutor).setRemoveOnCancelPolicy(true); +// make sure shutdown removes all pending tasks +((ScheduledThreadPoolExecutor) this.scheduledExecutor) + .setContinueExistingPeriodicTasksAfterShutdownPolicy(false); +((ScheduledThreadPoolExecutor) this.scheduledExecutor) +.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); +} else { +this.scheduledExecutor = null; +} + LOG.info( -"Create AsyncExecutionController: batchSize {}, maxInFlightRecordsNum {}", +"Create AsyncExecutionController: batchSize {}, bufferTimeout {}, maxInFlightRecordsNum {}", batchSize, +bufferTimeout, maxInFlightRecords); } +void scheduleTimeout(long triggerSeq) { +if (bufferTimeout > 0) { +if (currentScheduledFuture != null +&& !currentScheduledFuture.isDone() +&& !currentScheduledFuture.isCancelled()) { +currentScheduledFuture.cancel(false); +} +currentScheduledFuture = +(ScheduledFuture) +scheduledExecutor.schedule( +() -> { +if (triggerSeq != currentTriggerSeq.get()) { +// if any new trigger occurs, skip this schedule +return; +} +mailboxExecutor.execute( +() -> triggerIfNeeded(true), "AEC-timeout"); Review Comment: Well, I mean we keep the `if (triggerSeq != currentTriggerSeq.get()) { return; }` before this, only change the mailbox processing part. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
Zakelly commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1578769079 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -220,12 +283,17 @@ void insertBlockingBuffer(StateRequest request) { * @param force whether to trigger requests in force. */ void triggerIfNeeded(boolean force) { -// TODO: introduce a timeout mechanism for triggering. if (!force && stateRequestsBuffer.activeQueueSize() < batchSize) { +// if the active queue size is 1, it means that the current request is the oldest one in +// the active queue. +if (stateRequestsBuffer.activeQueueSize() == 1) { +scheduleTimeout(currentTriggerSeq.get()); +} Review Comment: Ah... I've mispoken... Is it possible `scheduleTimeout` multiple times for one seq? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1578815645 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -220,12 +283,17 @@ void insertBlockingBuffer(StateRequest request) { * @param force whether to trigger requests in force. */ void triggerIfNeeded(boolean force) { -// TODO: introduce a timeout mechanism for triggering. if (!force && stateRequestsBuffer.activeQueueSize() < batchSize) { +// if the active queue size is 1, it means that the current request is the oldest one in +// the active queue. +if (stateRequestsBuffer.activeQueueSize() == 1) { +scheduleTimeout(currentTriggerSeq.get()); +} Review Comment: No, there will definitely be a trigger between two `stateRequestsBuffer.activeQueueSize() == 1`, so when the two `stateRequestsBuffer.activeQueueSize() == 1` conditions are met, seq number must be different. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
Zakelly commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1578825952 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -220,12 +283,17 @@ void insertBlockingBuffer(StateRequest request) { * @param force whether to trigger requests in force. */ void triggerIfNeeded(boolean force) { -// TODO: introduce a timeout mechanism for triggering. if (!force && stateRequestsBuffer.activeQueueSize() < batchSize) { +// if the active queue size is 1, it means that the current request is the oldest one in +// the active queue. +if (stateRequestsBuffer.activeQueueSize() == 1) { +scheduleTimeout(currentTriggerSeq.get()); +} Review Comment: I mean, the `triggerIfNeeded(false)` may be triggered somewhere else right? And even the `enqueueToActive` has two entry. I'd suggest move the `scheduleTimeout` part into the `StateRequestBuffer`, but the handler and seq maintaining part should stay in `AEC` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1578997828 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -220,12 +283,17 @@ void insertBlockingBuffer(StateRequest request) { * @param force whether to trigger requests in force. */ void triggerIfNeeded(boolean force) { -// TODO: introduce a timeout mechanism for triggering. if (!force && stateRequestsBuffer.activeQueueSize() < batchSize) { +// if the active queue size is 1, it means that the current request is the oldest one in +// the active queue. +if (stateRequestsBuffer.activeQueueSize() == 1) { +scheduleTimeout(currentTriggerSeq.get()); +} Review Comment: Currently, `triggerIfNeeded(false)` is only called in `handleRequest()`, I added a `schedulingCount==1` condition to avoid "trigger multiple times for one seq" in the future. And I moved `scheduleTimeout ` into `StateRequestBuffer`, thanks for the suggestion.π -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1578997828 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -220,12 +283,17 @@ void insertBlockingBuffer(StateRequest request) { * @param force whether to trigger requests in force. */ void triggerIfNeeded(boolean force) { -// TODO: introduce a timeout mechanism for triggering. if (!force && stateRequestsBuffer.activeQueueSize() < batchSize) { +// if the active queue size is 1, it means that the current request is the oldest one in +// the active queue. +if (stateRequestsBuffer.activeQueueSize() == 1) { +scheduleTimeout(currentTriggerSeq.get()); +} Review Comment: Currently, `triggerIfNeeded(false)` is only called in `handleRequest()`, I added a `schedulingCount==0` condition to avoid "trigger multiple times for one seq" in the future. And I moved `scheduleTimeout ` into `StateRequestBuffer`, thanks for the suggestion.π -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
Zakelly commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1579142292 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,26 +97,39 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; -public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); -} - public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeout, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeout = bufferTimeout; this.maxInFlightRecordNum = maxInFlightRecords; -this.stateRequestsBuffer = new StateRequestBuffer<>(); + this.inFlightRecordNum = new AtomicInteger(0); +this.stateRequestsBuffer = +new StateRequestBuffer<>( +bufferTimeout, +() -> +mailboxExecutor.execute( +() -> { +if (stateRequestsBuffer.currentTriggerSeq.get() +== stateRequestsBuffer.scheduledTriggerSeq Review Comment: This condition is always true. I suggest you pass the `triggered seq` into the handler lambda as parameter. ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java: ## @@ -53,17 +72,75 @@ public class StateRequestBuffer { /** The number of state requests in blocking queue. */ int blockingQueueSize; -public StateRequestBuffer() { +/** The timeout of {@link #activeQueue} triggering in milliseconds. */ +final long bufferTimeout; + +/** The handler to trigger when {@link #activeQueue} size is 1. */ +final Runnable timeoutHandler; + +/** The executor service that schedules and calls the triggers of this task. */ +ScheduledExecutorService scheduledExecutor; + +/** + * The current scheduled future, when the next scheduling occurs, the previous one that has not + * yet been executed will be canceled. + */ +ScheduledFuture currentScheduledFuture; + +/** + * The current scheduled trigger sequence number, a timeout trigger is scheduled only if {@code + * scheduledTriggerSeq} is less than {@code currentTriggerSeq}. + */ +AtomicLong scheduledTriggerSeq; + +/** + * The current trigger sequence number, used to distinguish different triggers. Every time a + * trigger occurs, {@code currentTriggerSeq} increases by one. + */ +AtomicLong currentTriggerSeq; + +public StateRequestBuffer(long bufferTimeout, Runnable timeoutHandler) { this.activeQueue = new LinkedList<>(); this.blockingQueue = new HashMap<>(); this.blockingQueueSize = 0; +this.bufferTimeout = bufferTimeout; +this.timeoutHandler = timeoutHandler; +this.scheduledTriggerSeq = new AtomicLong(-1); +this.currentTriggerSeq = new AtomicLong(0); +if (bufferTimeout > 0) { +this.scheduledExecutor = DELAYER; +} +} + +void advanceTriggerSeq() { +currentTriggerSeq.incrementAndGet(); } void enqueueToActive(StateRequest request) { if (request.getRequestType() == StateRequestType.SYNC_POINT) { request.getFuture().complete(null); } else { activeQueue.add(request); +// if the active queue size is 1, it means that the current request is the oldest one in Review Comment: nit. remove this? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
Zakelly commented on code in PR #24667: URL: https://github.com/apache/flink/pull/24667#discussion_r1580464018 ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java: ## @@ -94,26 +97,38 @@ public class AsyncExecutionController { */ final AtomicInteger inFlightRecordNum; -public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) { -this(mailboxExecutor, stateExecutor, DEFAULT_BATCH_SIZE, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM); -} - public AsyncExecutionController( MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int batchSize, +long bufferTimeout, int maxInFlightRecords) { this.keyAccountingUnit = new KeyAccountingUnit<>(maxInFlightRecords); this.mailboxExecutor = mailboxExecutor; this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor); this.stateExecutor = stateExecutor; this.batchSize = batchSize; +this.bufferTimeout = bufferTimeout; this.maxInFlightRecordNum = maxInFlightRecords; -this.stateRequestsBuffer = new StateRequestBuffer<>(); + this.inFlightRecordNum = new AtomicInteger(0); +this.stateRequestsBuffer = +new StateRequestBuffer<>( +bufferTimeout, +(scheduledTriggerSeq) -> +mailboxExecutor.execute( +() -> { +if (stateRequestsBuffer.currentTriggerSeq.get() Review Comment: How about providing a function `boolean stateRequestsBuffer#checkCurrentSeq(long seq)` ? ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java: ## @@ -53,17 +73,74 @@ public class StateRequestBuffer { /** The number of state requests in blocking queue. */ int blockingQueueSize; -public StateRequestBuffer() { +/** The timeout of {@link #activeQueue} triggering in milliseconds. */ +final long bufferTimeout; + +/** The handler to trigger when timeout. */ +final Consumer timeoutHandler; + +/** The executor service that schedules and calls the triggers of this task. */ +ScheduledExecutorService scheduledExecutor; + +/** + * The current scheduled future, when the next scheduling occurs, the previous one that has not + * yet been executed will be canceled. + */ +ScheduledFuture currentScheduledFuture; + +/** + * The current scheduled trigger sequence number, a timeout trigger is scheduled only if {@code + * scheduledTriggerSeq} is less than {@code currentTriggerSeq}. + */ +AtomicLong scheduledTriggerSeq; + +/** + * The current trigger sequence number, used to distinguish different triggers. Every time a + * trigger occurs, {@code currentTriggerSeq} increases by 1. + */ +AtomicLong currentTriggerSeq; + +public StateRequestBuffer(long bufferTimeout, Consumer timeoutHandler) { this.activeQueue = new LinkedList<>(); this.blockingQueue = new HashMap<>(); this.blockingQueueSize = 0; +this.bufferTimeout = bufferTimeout; +this.timeoutHandler = timeoutHandler; +this.scheduledTriggerSeq = new AtomicLong(-1); +this.currentTriggerSeq = new AtomicLong(0); +if (bufferTimeout > 0) { +this.scheduledExecutor = DELAYER; +} +} + +void advanceTriggerSeq() { +currentTriggerSeq.incrementAndGet(); } void enqueueToActive(StateRequest request) { if (request.getRequestType() == StateRequestType.SYNC_POINT) { request.getFuture().complete(null); } else { activeQueue.add(request); +if (bufferTimeout > 0 && currentTriggerSeq.get() > scheduledTriggerSeq.get()) { +if (currentScheduledFuture != null +&& !currentScheduledFuture.isDone() +&& !currentScheduledFuture.isCancelled()) { +currentScheduledFuture.cancel(false); +} +scheduledTriggerSeq.set(currentTriggerSeq.get()); Review Comment: I was thinking add `final long thisScheduledSeq = scheduledTriggerSeq.get();` here, and use `thisScheduledSeq == currentTriggerSeq.get()` condition at line 136. ## flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java: ## @@ -53,17 +73,74 @@ public class StateRequestBuffer { /** The number of state requests in blocking queue. */ int blockingQueueSize; -public StateRequestBuffer() { +/** The timeout of {@link #activeQueue} triggering in milliseconds. */ +final long bufferTimeout; + +/** The handler to trigger when timeout. */ +final Consumer timeoutHand
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia commented on PR #24667: URL: https://github.com/apache/flink/pull/24667#issuecomment-2078637067 @Zakelly Thanks for the review, I addressed the comments and rebased to master. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35026][runtime][config] Introduce async execution configurations [flink]
fredia merged PR #24667: URL: https://github.com/apache/flink/pull/24667 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org