You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2023/05/05 03:03:29 UTC
[iotdb] branch 55_wal_accumulation created (now 5baa4b524e6)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a change to branch 55_wal_accumulation
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 5baa4b524e6 optimize iotconsensus
This branch includes the following new commits:
new 5baa4b524e6 optimize iotconsensus
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: optimize iotconsensus
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch 55_wal_accumulation
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5baa4b524e6abb0af29409bcc19551b9ad9ab80e
Author: OneSizeFitQuorum <ta...@apache.org>
AuthorDate: Fri May 5 11:03:12 2023 +0800
optimize iotconsensus
Signed-off-by: OneSizeFitQuorum <ta...@apache.org>
---
.../iotdb/consensus/config/IoTConsensusConfig.java | 23 +++++++++++++++++-----
.../consensus/iot/logdispatcher/LogDispatcher.java | 8 ++------
2 files changed, 20 insertions(+), 11 deletions(-)
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java b/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java
index 9c676fdaa9c..2fe821a7cb0 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java
@@ -235,6 +235,8 @@ public class IoTConsensusConfig {
private final int maxLogEntriesNumPerBatch;
private final int maxSizePerBatch;
private final int maxPendingBatchesNum;
+
+ private final int maxQueueLength;
private final long maxWaitingTimeForWaitBatchInMs;
private final int maxWaitingTimeForAccumulatingBatchInMs;
private final long basicRetryWaitTimeMs;
@@ -249,6 +251,7 @@ public class IoTConsensusConfig {
int maxLogEntriesNumPerBatch,
int maxSizePerBatch,
int maxPendingBatchesNum,
+ int maxQueueLength,
long maxWaitingTimeForWaitBatchInMs,
int maxWaitingTimeForAccumulatingBatchInMs,
long basicRetryWaitTimeMs,
@@ -261,6 +264,7 @@ public class IoTConsensusConfig {
this.maxLogEntriesNumPerBatch = maxLogEntriesNumPerBatch;
this.maxSizePerBatch = maxSizePerBatch;
this.maxPendingBatchesNum = maxPendingBatchesNum;
+ this.maxQueueLength = maxQueueLength;
this.maxWaitingTimeForWaitBatchInMs = maxWaitingTimeForWaitBatchInMs;
this.maxWaitingTimeForAccumulatingBatchInMs = maxWaitingTimeForAccumulatingBatchInMs;
this.basicRetryWaitTimeMs = basicRetryWaitTimeMs;
@@ -284,6 +288,10 @@ public class IoTConsensusConfig {
return maxPendingBatchesNum;
}
+ public int getMaxQueueLength() {
+ return maxQueueLength;
+ }
+
public long getMaxWaitingTimeForWaitBatchInMs() {
return maxWaitingTimeForWaitBatchInMs;
}
@@ -326,13 +334,12 @@ public class IoTConsensusConfig {
public static class Builder {
- private int maxLogEntriesNumPerBatch = 30;
+ private int maxLogEntriesNumPerBatch = 1024;
private int maxSizePerBatch = 16 * 1024 * 1024;
- // (IMPORTANT) Value of this variable should be the same with MAX_REQUEST_CACHE_SIZE
- // in DataRegionStateMachine
- private int maxPendingBatchesNum = 5;
+ private int maxPendingBatchesNum = 16;
+ private int maxQueueLength = 4096;
private long maxWaitingTimeForWaitBatchInMs = 10 * 1000L;
- private int maxWaitingTimeForAccumulatingBatchInMs = 500;
+ private int maxWaitingTimeForAccumulatingBatchInMs = 100;
private long basicRetryWaitTimeMs = TimeUnit.MILLISECONDS.toMillis(100);
private long maxRetryWaitTimeMs = TimeUnit.SECONDS.toMillis(20);
private long walThrottleThreshold = 50 * 1024 * 1024 * 1024L;
@@ -356,6 +363,11 @@ public class IoTConsensusConfig {
return this;
}
+ public Builder setMaxQueueLength(int maxQueueLength) {
+ this.maxQueueLength = maxQueueLength;
+ return this;
+ }
+
public Replication.Builder setMaxWaitingTimeForWaitBatchInMs(
long maxWaitingTimeForWaitBatchInMs) {
this.maxWaitingTimeForWaitBatchInMs = maxWaitingTimeForWaitBatchInMs;
@@ -408,6 +420,7 @@ public class IoTConsensusConfig {
maxLogEntriesNumPerBatch,
maxSizePerBatch,
maxPendingBatchesNum,
+ maxQueueLength,
maxWaitingTimeForWaitBatchInMs,
maxWaitingTimeForAccumulatingBatchInMs,
basicRetryWaitTimeMs,
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java b/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
index f6da0efd52a..cd70efe655b 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java
@@ -46,9 +46,9 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.OptionalLong;
+import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
@@ -214,7 +214,7 @@ public class LogDispatcher {
public LogDispatcherThread(Peer peer, IoTConsensusConfig config, long initialSyncIndex) {
this.peer = peer;
this.config = config;
- this.pendingEntries = new LinkedBlockingQueue<>();
+ this.pendingEntries = new ArrayBlockingQueue<>(config.getReplication().getMaxQueueLength());
this.controller =
new IndexController(
impl.getStorageDir(),
@@ -314,10 +314,6 @@ public class LogDispatcher {
pendingEntries.poll(PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC, TimeUnit.SECONDS);
if (request != null) {
bufferedEntries.add(request);
- // If write pressure is low, we simply sleep a little to reduce the number of RPC
- if (pendingEntries.size() <= config.getReplication().getMaxLogEntriesNumPerBatch()) {
- Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs());
- }
}
}
MetricService.getInstance()