[iotdb] branch iotdb-3791 updated: remove lock in MultiLeaderProcessor

[xingtanzjr]

This is an automated email from the ASF dual-hosted git repository.

xingtanzjr pushed a commit to branch iotdb-3791
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/iotdb-3791 by this push:
     new 7adcbc0739 remove lock in MultiLeaderProcessor
7adcbc0739 is described below

commit 7adcbc073964c4765abcc720903d9481237d79bf
Author: Jinrui.Zhang <xingtan...@gmail.com>
AuthorDate: Thu Jul 14 21:57:22 2022 +0800

    remove lock in MultiLeaderProcessor
---
 .../iotdb/consensus/config/MultiLeaderConfig.java  |  2 +-
 .../service/MultiLeaderRPCServiceProcessor.java    | 28 +++++++++++-----------
 2 files changed, 15 insertions(+), 15 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 06cc2e453c..6383684545 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
@@ -249,7 +249,7 @@ public class MultiLeaderConfig {
     public static class Builder {
       private int maxPendingRequestNumPerNode = 200;
       private int maxRequestPerBatch = 40;
-      private int maxPendingBatch = 6;
+      private int maxPendingBatch = 10;
       private int maxWaitingTimeForAccumulatingBatchInMs = 500;
       private long basicRetryWaitTimeMs = TimeUnit.MILLISECONDS.toMillis(100);
       private long maxRetryWaitTimeMs = TimeUnit.SECONDS.toMillis(20);
diff --git 
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
 
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
index e0ddd7f256..36c040c983 100644
--- 
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
+++ 
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
@@ -73,21 +73,21 @@ public class MultiLeaderRPCServiceProcessor implements 
MultiLeaderConsensusIServ
       StepTracker.trace("syncLogPrepare", 10, prepareStartTime, 
System.nanoTime());
       // We use synchronized to ensure atomicity of executing multiple logs
       long lockWaitingStartTime = System.nanoTime();
-      synchronized (impl.getStateMachine()) {
-        StepTracker.trace("syncLogWaitingLock", 10, lockWaitingStartTime, 
System.nanoTime());
-        StepTracker.trace("req.getBatches().size()", 10, 0, 
req.getBatches().size() * 1000_000L);
-        for (TLogBatch batch : req.getBatches()) {
-          long writeOneBatch = System.nanoTime();
-          statuses.add(
-              impl.getStateMachine()
-                  .write(
-                      impl.buildIndexedConsensusRequestForRemoteRequest(
-                          batch.isFromWAL()
-                              ? new MultiLeaderConsensusRequest(batch.data)
-                              : new ByteBufferConsensusRequest(batch.data))));
-          StepTracker.trace("writeOneBatch", 400, writeOneBatch, 
System.nanoTime());
-        }
+      //      synchronized (impl.getStateMachine()) {
+      StepTracker.trace("syncLogWaitingLock", 10, lockWaitingStartTime, 
System.nanoTime());
+      StepTracker.trace("req.getBatches().size()", 10, 0, 
req.getBatches().size() * 1000_000L);
+      for (TLogBatch batch : req.getBatches()) {
+        long writeOneBatch = System.nanoTime();
+        statuses.add(
+            impl.getStateMachine()
+                .write(
+                    impl.buildIndexedConsensusRequestForRemoteRequest(
+                        batch.isFromWAL()
+                            ? new MultiLeaderConsensusRequest(batch.data)
+                            : new ByteBufferConsensusRequest(batch.data))));
+        StepTracker.trace("writeOneBatch", 400, writeOneBatch, 
System.nanoTime());
       }
+      //      }
       logger.debug("Execute TSyncLogReq for {} with result {}", 
req.consensusGroupId, statuses);
       resultHandler.onComplete(new TSyncLogRes(statuses));
     } catch (Exception e) {