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 <[email protected]> AuthorDate: Fri May 5 11:03:12 2023 +0800 optimize iotconsensus Signed-off-by: OneSizeFitQuorum <[email protected]> --- .../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()
