[iotdb] branch ml_test_1_async updated: change offer to put do disable wal

[xingtanzjr]

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

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


The following commit(s) were added to refs/heads/ml_test_1_async by this push:
     new b3413871ee change offer to put do disable wal
b3413871ee is described below

commit b3413871ee2c6eac4bd0ab297e166961a3d18e1e
Author: Jinrui.Zhang <xingtan...@gmail.com>
AuthorDate: Tue Aug 2 15:46:01 2022 +0800

    change offer to put do disable wal
---
 .../multileader/logdispatcher/LogDispatcher.java   | 40 ++++++++++++++--------
 1 file changed, 26 insertions(+), 14 deletions(-)

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 da5ad41aaa..faffa7bf60 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
@@ -20,6 +20,7 @@
 package org.apache.iotdb.consensus.multileader.logdispatcher;
 
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.StepTracker;
 import org.apache.iotdb.commons.client.IClientManager;
 import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
 import org.apache.iotdb.consensus.common.Peer;
@@ -114,21 +115,32 @@ public class LogDispatcher {
               "{}: Push a log to the queue, where the queue length is {}",
               impl.getThisNode().getGroupId(),
               thread.getPendingRequest().size());
-          if (!thread
-              .getPendingRequest()
-              .offer(new IndexedConsensusRequest(serializedRequest, 
request.getSearchIndex()))) {
-            logger.info(
-                "{}: Log queue to {} is full. skip current request: {}",
-                impl.getThisNode().getGroupId(),
-                thread.getPeer().getEndpoint().getIp(),
-                request.getSearchIndex());
-            logger.debug(
-                "{}: Log queue of {} is full, ignore the log to this node",
-                impl.getThisNode().getGroupId(),
-                thread.getPeer());
-          } else {
-            thread.countQueueUsage(request.getSearchIndex());
+          long startTime = System.nanoTime();
+          try {
+            thread
+                .getPendingRequest()
+                .put(new IndexedConsensusRequest(serializedRequest, 
request.getSearchIndex()));
+          } catch (InterruptedException e) {
+            e.printStackTrace();
+          } finally {
+            StepTracker.trace("putToQueue", startTime, System.nanoTime());
           }
+          //          if (!thread
+          //              .getPendingRequest()
+          //              .put(new IndexedConsensusRequest(serializedRequest,
+          // request.getSearchIndex()))) {
+          //            logger.info(
+          //                "{}: Log queue to {} is full. skip current 
request: {}",
+          //                impl.getThisNode().getGroupId(),
+          //                thread.getPeer().getEndpoint().getIp(),
+          //                request.getSearchIndex());
+          //            logger.debug(
+          //                "{}: Log queue of {} is full, ignore the log to 
this node",
+          //                impl.getThisNode().getGroupId(),
+          //                thread.getPeer());
+          //          } else {
+          //            thread.countQueueUsage(request.getSearchIndex());
+          //          }
         });
   }