You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2024/09/24 02:32:53 UTC
(flink) 04/04: [FLINK-35411][State] Support process write state requests within coordinator thread
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8bb2f4d7479cd91c5c1dc52148cf7a31afc6d0b4
Author: Zakelly <za...@gmail.com>
AuthorDate: Mon Sep 23 18:36:38 2024 +0800
[FLINK-35411][State] Support process write state requests within coordinator thread
---
.../state/forst/ForStConfigurableOptions.java | 2 +
.../flink/state/forst/ForStKeyedStateBackend.java | 1 +
.../org/apache/flink/state/forst/ForStOptions.java | 16 ++++-
.../flink/state/forst/ForStResourceContainer.java | 4 ++
.../flink/state/forst/ForStStateExecutor.java | 77 ++++++++++++++--------
.../flink/state/forst/ForStStateExecutorTest.java | 4 +-
6 files changed, 72 insertions(+), 32 deletions(-)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
index 322373e97b7..53b2300e1b3 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
@@ -40,6 +40,7 @@ import static org.apache.flink.configuration.ConfigOptions.key;
import static org.apache.flink.configuration.description.LinkElement.link;
import static org.apache.flink.configuration.description.TextElement.code;
import static org.apache.flink.state.forst.ForStOptions.EXECUTOR_READ_IO_PARALLELISM;
+import static org.apache.flink.state.forst.ForStOptions.EXECUTOR_WRITE_IO_INLINE;
import static org.apache.flink.state.forst.ForStOptions.EXECUTOR_WRITE_IO_PARALLELISM;
import static org.rocksdb.CompactionStyle.FIFO;
import static org.rocksdb.CompactionStyle.LEVEL;
@@ -325,6 +326,7 @@ public class ForStConfigurableOptions implements Serializable {
static final ConfigOption<?>[] CANDIDATE_CONFIGS =
new ConfigOption<?>[] {
// configurable forst executor
+ EXECUTOR_WRITE_IO_INLINE,
EXECUTOR_WRITE_IO_PARALLELISM,
EXECUTOR_READ_IO_PARALLELISM,
// configurable DBOptions
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
index 4d4fae93463..9b46918f1d6 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java
@@ -276,6 +276,7 @@ public class ForStKeyedStateBackend<K> implements AsyncKeyedStateBackend<K> {
}
StateExecutor stateExecutor =
new ForStStateExecutor(
+ optionsContainer.isWriteInline(),
optionsContainer.getReadIoParallelism(),
optionsContainer.getWriteIoParallelism(),
db,
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
index 7dc55903937..782b3061da0 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
@@ -157,17 +157,27 @@ public class ForStOptions {
+ "of rocksdb timer service, but consumes more heap memory at the same time.",
TIMER_SERVICE_FACTORY.key(), ForStDB.name()));
+ public static final ConfigOption<Boolean> EXECUTOR_WRITE_IO_INLINE =
+ ConfigOptions.key("state.backend.forst.executor.inline-write")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to let write requests executed within the coordinator thread.");
+
public static final ConfigOption<Integer> EXECUTOR_READ_IO_PARALLELISM =
- ConfigOptions.key("state.backend.forst.memory.executor-read-io-parallelism")
+ ConfigOptions.key("state.backend.forst.executor.read-io-parallelism")
.intType()
.defaultValue(3)
.withDescription(
"The number of threads used for read IO operations in the executor.");
public static final ConfigOption<Integer> EXECUTOR_WRITE_IO_PARALLELISM =
- ConfigOptions.key("state.backend.forst.memory.executor-write-io-parallelism")
+ ConfigOptions.key("state.backend.forst.executor.write-io-parallelism")
.intType()
.defaultValue(1)
.withDescription(
- "The number of threads used for write IO operations in the executor.");
+ "The number of threads used for write IO operations in the executor."
+ + " Only valid when '"
+ + EXECUTOR_WRITE_IO_INLINE.key()
+ + "' is false.");
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
index 5b48ef3711b..0e72717bcee 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java
@@ -277,6 +277,10 @@ public final class ForStResourceContainer implements AutoCloseable {
}
}
+ public boolean isWriteInline() {
+ return configuration.get(ForStOptions.EXECUTOR_WRITE_IO_INLINE);
+ }
+
public int getReadIoParallelism() {
return configuration.get(ForStOptions.EXECUTOR_READ_IO_PARALLELISM);
}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java
index 1366e688b59..6b2ecf0e747 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateExecutor.java
@@ -61,6 +61,9 @@ public class ForStStateExecutor implements StateExecutor {
private final int readThreadCount;
+ /** Whether the write thread is sharing with others. */
+ private final boolean sharedWriteThread;
+
private final RocksDB db;
private final WriteOptions writeOptions;
@@ -71,27 +74,49 @@ public class ForStStateExecutor implements StateExecutor {
private final AtomicLong ongoing;
public ForStStateExecutor(
- int readIoParallelism, int writeIoParallelism, RocksDB db, WriteOptions writeOptions) {
- Preconditions.checkState(readIoParallelism > 0 || writeIoParallelism > 0);
- this.coordinatorThread =
- Executors.newSingleThreadScheduledExecutor(
- new ExecutorThreadFactory("ForSt-StateExecutor-Coordinator"));
- if (readIoParallelism <= 0 || writeIoParallelism <= 0) {
- this.readThreadCount = Math.max(readIoParallelism, writeIoParallelism);
- this.readThreads =
- Executors.newFixedThreadPool(
- readThreadCount, new ExecutorThreadFactory("ForSt-StateExecutor-IO"));
- this.writeThreads = null;
- } else {
+ boolean isWriteInline,
+ int readIoParallelism,
+ int writeIoParallelism,
+ RocksDB db,
+ WriteOptions writeOptions) {
+ if (isWriteInline) {
+ Preconditions.checkState(readIoParallelism > 0);
+ this.coordinatorThread =
+ Executors.newSingleThreadScheduledExecutor(
+ new ExecutorThreadFactory("ForSt-StateExecutor-Coordinator-And-Write"));
this.readThreadCount = readIoParallelism;
this.readThreads =
Executors.newFixedThreadPool(
readIoParallelism,
new ExecutorThreadFactory("ForSt-StateExecutor-read-IO"));
this.writeThreads =
- Executors.newFixedThreadPool(
- writeIoParallelism,
- new ExecutorThreadFactory("ForSt-StateExecutor-write-IO"));
+ org.apache.flink.util.concurrent.Executors.newDirectExecutorService();
+ this.sharedWriteThread = true;
+ } else {
+ Preconditions.checkState(readIoParallelism > 0 || writeIoParallelism > 0);
+ this.coordinatorThread =
+ Executors.newSingleThreadScheduledExecutor(
+ new ExecutorThreadFactory("ForSt-StateExecutor-Coordinator"));
+ if (readIoParallelism <= 0 || writeIoParallelism <= 0) {
+ this.readThreadCount = Math.max(readIoParallelism, writeIoParallelism);
+ this.readThreads =
+ Executors.newFixedThreadPool(
+ readThreadCount,
+ new ExecutorThreadFactory("ForSt-StateExecutor-IO"));
+ this.writeThreads = readThreads;
+ this.sharedWriteThread = true;
+ } else {
+ this.readThreadCount = readIoParallelism;
+ this.readThreads =
+ Executors.newFixedThreadPool(
+ readIoParallelism,
+ new ExecutorThreadFactory("ForSt-StateExecutor-read-IO"));
+ this.writeThreads =
+ Executors.newFixedThreadPool(
+ writeIoParallelism,
+ new ExecutorThreadFactory("ForSt-StateExecutor-write-IO"));
+ this.sharedWriteThread = false;
+ }
}
this.db = db;
this.writeOptions = writeOptions;
@@ -122,17 +147,6 @@ public class ForStStateExecutor implements StateExecutor {
() -> {
long startTime = System.currentTimeMillis();
List<CompletableFuture<Void>> futures = new ArrayList<>(3);
- List<ForStDBPutRequest<?, ?, ?>> putRequests =
- stateRequestClassifier.pollDbPutRequests();
- if (!putRequests.isEmpty()) {
- ForStWriteBatchOperation writeOperations =
- new ForStWriteBatchOperation(
- db,
- putRequests,
- writeOptions,
- writeThreads == null ? readThreads : writeThreads);
- futures.add(writeOperations.process());
- }
if (!getRequests.isEmpty()) {
ForStGeneralMultiGetOperation getOperations =
@@ -152,6 +166,15 @@ public class ForStStateExecutor implements StateExecutor {
futures.add(iterOperations.process());
}
+ List<ForStDBPutRequest<?, ?, ?>> putRequests =
+ stateRequestClassifier.pollDbPutRequests();
+ if (!putRequests.isEmpty()) {
+ ForStWriteBatchOperation writeOperations =
+ new ForStWriteBatchOperation(
+ db, putRequests, writeOptions, writeThreads);
+ futures.add(writeOperations.process());
+ }
+
FutureUtils.combineAll(futures)
.thenAcceptAsync(
(e) -> {
@@ -204,7 +227,7 @@ public class ForStStateExecutor implements StateExecutor {
@Override
public void shutdown() {
readThreads.shutdown();
- if (writeThreads != null) {
+ if (!sharedWriteThread) {
writeThreads.shutdown();
}
coordinatorThread.shutdown();
diff --git a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java
index df8170962a4..48618678877 100644
--- a/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java
+++ b/flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateExecutorTest.java
@@ -47,7 +47,7 @@ class ForStStateExecutorTest extends ForStDBOperationTestBase {
@SuppressWarnings("unchecked")
void testExecuteValueStateRequest() throws Exception {
ForStStateExecutor forStStateExecutor =
- new ForStStateExecutor(3, 1, db, new WriteOptions());
+ new ForStStateExecutor(false, 3, 1, db, new WriteOptions());
ForStValueState<Integer, VoidNamespace, String> state1 =
buildForStValueState("value-state-1");
ForStValueState<Integer, VoidNamespace, String> state2 =
@@ -131,7 +131,7 @@ class ForStStateExecutorTest extends ForStDBOperationTestBase {
@Test
void testExecuteMapStateRequest() throws Exception {
ForStStateExecutor forStStateExecutor =
- new ForStStateExecutor(3, 1, db, new WriteOptions());
+ new ForStStateExecutor(false, 3, 1, db, new WriteOptions());
ForStMapState<Integer, VoidNamespace, String, String> state =
buildForStMapState("map-state");
StateRequestContainer stateRequestContainer =