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 4480055038 add StepTracker
4480055038 is described below
commit 4480055038af92e7f2d219a74c2332ea9c620768
Author: Jinrui.Zhang <[email protected]>
AuthorDate: Tue Aug 2 13:24:56 2022 +0800
add StepTracker
---
.../multileader/MultiLeaderServerImpl.java | 50 ++++++++++++----------
.../service/MultiLeaderRPCServiceProcessor.java | 3 +-
.../org/apache/iotdb/commons}/StepTracker.java | 27 +++++++++---
.../iotdb/db/engine/storagegroup/DataRegion.java | 5 +++
4 files changed, 57 insertions(+), 28 deletions(-)
diff --git
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
index 3e8cbbfb3b..f8a08aa553 100644
---
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
+++
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.consensus.multileader;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.StepTracker;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.consensus.IStateMachine;
import org.apache.iotdb.consensus.common.DataSet;
@@ -110,29 +111,34 @@ public class MultiLeaderServerImpl {
*/
public TSStatus write(IConsensusRequest request) {
synchronized (stateMachine) {
- IndexedConsensusRequest indexedConsensusRequest =
- buildIndexedConsensusRequestForLocalRequest(request);
- if (indexedConsensusRequest.getSearchIndex() % 1000 == 0) {
- logger.info(
- "DataRegion[{}]: index after build: safeIndex:{}, searchIndex: {}",
- thisNode.getGroupId(),
- getCurrentSafelyDeletedSearchIndex(),
- indexedConsensusRequest.getSearchIndex());
+ long startTime = System.nanoTime();
+ try {
+ IndexedConsensusRequest indexedConsensusRequest =
+ buildIndexedConsensusRequestForLocalRequest(request);
+ if (indexedConsensusRequest.getSearchIndex() % 1000 == 0) {
+ logger.info(
+ "DataRegion[{}]: index after build: safeIndex:{}, searchIndex:
{}",
+ thisNode.getGroupId(),
+ getCurrentSafelyDeletedSearchIndex(),
+ indexedConsensusRequest.getSearchIndex());
+ }
+ // TODO wal and memtable
+ TSStatus result = stateMachine.write(indexedConsensusRequest);
+ if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ logDispatcher.offer(indexedConsensusRequest);
+ } else {
+ logger.debug(
+ "{}: write operation failed. searchIndex: {}. Code: {}",
+ thisNode.getGroupId(),
+ indexedConsensusRequest.getSearchIndex(),
+ result.getCode());
+ index.decrementAndGet();
+ }
+
+ return result;
+ } finally {
+ StepTracker.trace("MultiLeaderWrite", startTime, System.nanoTime());
}
- // TODO wal and memtable
- TSStatus result = stateMachine.write(indexedConsensusRequest);
- if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- logDispatcher.offer(indexedConsensusRequest);
- } else {
- logger.debug(
- "{}: write operation failed. searchIndex: {}. Code: {}",
- thisNode.getGroupId(),
- indexedConsensusRequest.getSearchIndex(),
- result.getCode());
- index.decrementAndGet();
- }
-
- return result;
}
}
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 acb62967f3..08d6a3a9f4 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
@@ -20,6 +20,7 @@
package org.apache.iotdb.consensus.multileader.service;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.StepTracker;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest;
import org.apache.iotdb.consensus.common.request.IConsensusRequest;
@@ -101,7 +102,7 @@ public class MultiLeaderRPCServiceProcessor implements
MultiLeaderConsensusIServ
} catch (Exception e) {
resultHandler.onError(e);
} finally {
-
+ StepTracker.trace("syncLog", 25, startTime, System.nanoTime());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java
b/node-commons/src/main/java/org/apache/iotdb/commons/StepTracker.java
similarity index 75%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java
rename to node-commons/src/main/java/org/apache/iotdb/commons/StepTracker.java
index b5e1fc88ad..1aef5cf11a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/StepTracker.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/StepTracker.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan;
+package org.apache.iotdb.commons;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -27,18 +27,24 @@ import java.util.Map;
class Metric {
private static final Logger logger = LoggerFactory.getLogger(Metric.class);
- private static final int PRINT_RATE = 1000;
+ private static final int DEFAULT_PRINT_RATE = 1000;
public String stepName;
public long invokeCount;
public long totalTime;
public long lastCycleTime;
+ public int printRate;
public Metric(String stepName) {
+ this(stepName, DEFAULT_PRINT_RATE);
+ }
+
+ public Metric(String stepName, int printRate) {
this.stepName = stepName;
this.invokeCount = 0;
this.totalTime = 0;
this.lastCycleTime = 0;
+ this.printRate = printRate;
}
public void trace(long startTime, long endTime) {
@@ -47,7 +53,7 @@ class Metric {
}
public void tryPrint() {
- if (invokeCount % PRINT_RATE == 0) {
+ if (invokeCount % printRate == 0) {
logger.info(
String.format(
"step metrics [%d]-[%s] - Total: %d, SUM: %.2fms, AVG: %fms,
Last%dAVG: %fms",
@@ -56,8 +62,8 @@ class Metric {
invokeCount,
totalTime * 1.0 / 1000000,
totalTime * 1.0 / 1000000 / invokeCount,
- PRINT_RATE,
- (totalTime * 1.0 - lastCycleTime) / 1000000 / PRINT_RATE));
+ printRate,
+ (totalTime * 1.0 - lastCycleTime) / 1000000 / printRate));
lastCycleTime = totalTime;
}
}
@@ -74,6 +80,17 @@ public class StepTracker {
metrics.get().get(stepName).tryPrint();
}
+ public static void trace(String stepName, int printRate, long startTime,
long endTime) {
+ if (metrics.get() == null) {
+ metrics.set(new HashMap<>());
+ }
+ metrics
+ .get()
+ .computeIfAbsent(stepName, key -> new Metric(stepName, printRate))
+ .trace(startTime, endTime);
+ metrics.get().get(stepName).tryPrint();
+ }
+
public static void cleanup() {
metrics.set(null);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
index 3c08cb50cb..4e03ce4f64 100755
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.engine.storagegroup;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.StepTracker;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
@@ -1060,7 +1061,9 @@ public class DataRegion {
if (enableMemControl) {
StorageEngineV2.blockInsertionIfReject(null);
}
+ long startTimeBeforeLock = System.nanoTime();
writeLock("insertTablet");
+ long startTimeAfterLock = System.nanoTime();
try {
TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
@@ -1144,6 +1147,8 @@ public class DataRegion {
// TODO: trigger // fire trigger after insertion
// TriggerEngine.fire(TriggerEvent.AFTER_INSERT, insertTabletPlan,
firePosition);
} finally {
+ StepTracker.trace("lockAndInsert", startTimeBeforeLock,
System.nanoTime());
+ StepTracker.trace("InsertTabletWithoutLock", startTimeAfterLock,
System.nanoTime());
writeUnlock();
}
}
