This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch optimize/ml_memory_control
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/optimize/ml_memory_control by
this push:
new 4f65653b88 remove parameter for queue size
4f65653b88 is described below
commit 4f65653b88bdd361259cd4ed6d53977ec4c0d326
Author: Jinrui.Zhang <[email protected]>
AuthorDate: Fri Sep 30 16:03:16 2022 +0800
remove parameter for queue size
---
.../apache/iotdb/consensus/config/MultiLeaderConfig.java | 14 --------------
.../consensus/multileader/logdispatcher/LogDispatcher.java | 13 ++++++-------
.../logdispatcher/MultiLeaderMemoryManager.java | 6 +++---
3 files changed, 9 insertions(+), 24 deletions(-)
diff --git
a/consensus/src/main/java/org/apache/iotdb/consensus/config/MultiLeaderConfig.java
b/consensus/src/main/java/org/apache/iotdb/consensus/config/MultiLeaderConfig.java
index a14866c188..cd0d58c4ae 100644
---
a/consensus/src/main/java/org/apache/iotdb/consensus/config/MultiLeaderConfig.java
+++
b/consensus/src/main/java/org/apache/iotdb/consensus/config/MultiLeaderConfig.java
@@ -196,7 +196,6 @@ public class MultiLeaderConfig {
}
public static class Replication {
- private final int maxPendingRequestNumPerNode;
private final int maxRequestPerBatch;
private final int maxPendingBatch;
private final int maxWaitingTimeForAccumulatingBatchInMs;
@@ -208,7 +207,6 @@ public class MultiLeaderConfig {
private final Long allocateMemoryForConsensus;
private Replication(
- int maxPendingRequestNumPerNode,
int maxRequestPerBatch,
int maxPendingBatch,
int maxWaitingTimeForAccumulatingBatchInMs,
@@ -218,7 +216,6 @@ public class MultiLeaderConfig {
long throttleTimeOutMs,
long checkpointGap,
long allocateMemoryForConsensus) {
- this.maxPendingRequestNumPerNode = maxPendingRequestNumPerNode;
this.maxRequestPerBatch = maxRequestPerBatch;
this.maxPendingBatch = maxPendingBatch;
this.maxWaitingTimeForAccumulatingBatchInMs =
maxWaitingTimeForAccumulatingBatchInMs;
@@ -230,10 +227,6 @@ public class MultiLeaderConfig {
this.allocateMemoryForConsensus = allocateMemoryForConsensus;
}
- public int getMaxPendingRequestNumPerNode() {
- return maxPendingRequestNumPerNode;
- }
-
public int getMaxRequestPerBatch() {
return maxRequestPerBatch;
}
@@ -275,7 +268,6 @@ public class MultiLeaderConfig {
}
public static class Builder {
- private int maxPendingRequestNumPerNode = 600;
private int maxRequestPerBatch = 30;
// (IMPORTANT) Value of this variable should be the same with
MAX_REQUEST_CACHE_SIZE
// in DataRegionStateMachine
@@ -288,11 +280,6 @@ public class MultiLeaderConfig {
private long checkpointGap = 500;
private long allocateMemoryForConsensus;
- public Replication.Builder setMaxPendingRequestNumPerNode(int
maxPendingRequestNumPerNode) {
- this.maxPendingRequestNumPerNode = maxPendingRequestNumPerNode;
- return this;
- }
-
public Replication.Builder setMaxRequestPerBatch(int maxRequestPerBatch)
{
this.maxRequestPerBatch = maxRequestPerBatch;
return this;
@@ -336,7 +323,6 @@ public class MultiLeaderConfig {
public Replication build() {
return new Replication(
- maxPendingRequestNumPerNode,
maxRequestPerBatch,
maxPendingBatch,
maxWaitingTimeForAccumulatingBatchInMs,
diff --git
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
index 2ad72b2fbd..86f7186c6c 100644
---
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
+++
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
@@ -47,9 +47,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.stream.Collectors;
@@ -191,8 +191,7 @@ public class LogDispatcher {
public LogDispatcherThread(Peer peer, MultiLeaderConfig config, long
initialSyncIndex) {
this.peer = peer;
this.config = config;
- this.pendingRequest =
- new
ArrayBlockingQueue<>(config.getReplication().getMaxPendingRequestNumPerNode());
+ this.pendingRequest = new LinkedBlockingQueue<>();
this.controller =
new IndexController(
impl.getStorageDir(),
@@ -244,7 +243,7 @@ public class LogDispatcher {
}
/** try to remove a request from queue with memory control */
- private void remove(IndexedConsensusRequest indexedConsensusRequest) {
+ private void releaseReservedMemory(IndexedConsensusRequest
indexedConsensusRequest) {
multiLeaderMemoryManager.free(indexedConsensusRequest.getSerializedSize());
}
@@ -335,7 +334,7 @@ public class LogDispatcher {
IndexedConsensusRequest request = iterator.next();
if (request.getSearchIndex() < startIndex) {
iterator.remove();
- remove(request);
+ releaseReservedMemory(request);
} else {
break;
}
@@ -366,7 +365,7 @@ public class LogDispatcher {
constructBatchIndexedFromConsensusRequest(prev, logBatches);
endIndex = prev.getSearchIndex();
iterator.remove();
- remove(prev);
+ releaseReservedMemory(prev);
while (iterator.hasNext()
&& logBatches.size() <=
config.getReplication().getMaxRequestPerBatch()) {
IndexedConsensusRequest current = iterator.next();
@@ -391,7 +390,7 @@ public class LogDispatcher {
// current function, but that's fine, we'll continue processing
these elements in the
// bufferedRequest the next time we go into the function, they're
never lost
iterator.remove();
- remove(current);
+ releaseReservedMemory(current);
}
batch = new PendingBatch(startIndex, endIndex, logBatches);
logger.debug(
diff --git
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/MultiLeaderMemoryManager.java
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/MultiLeaderMemoryManager.java
index 8968bdeae0..4abfde8fd2 100644
---
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/MultiLeaderMemoryManager.java
+++
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/MultiLeaderMemoryManager.java
@@ -34,7 +34,7 @@ public class MultiLeaderMemoryManager {
public boolean reserve(long size) {
synchronized (this) {
if (size > maxMemorySizeInByte - memorySizeInByte.get()) {
- logger.info(
+ logger.debug(
"consensus memory limited. required: {}, used: {}, total: {}",
size,
memorySizeInByte.get(),
@@ -43,7 +43,7 @@ public class MultiLeaderMemoryManager {
}
memorySizeInByte.addAndGet(size);
}
- logger.info(
+ logger.debug(
"{} add {} bytes, total memory size: {} bytes.",
Thread.currentThread().getName(),
size,
@@ -53,7 +53,7 @@ public class MultiLeaderMemoryManager {
public void free(long size) {
long currentUsedMemory = memorySizeInByte.addAndGet(-size);
- logger.info(
+ logger.debug(
"{} free {} bytes, total memory size: {} bytes.",
Thread.currentThread().getName(),
size,
