This is an automated email from the ASF dual-hosted git repository. jiangtian pushed a commit to branch cluster_tpc in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e9081ccace9a0c9f9d17238c44d91584e599e0b7 Author: jt <[email protected]> AuthorDate: Thu Oct 8 14:26:43 2020 +0800 add timers --- .../org/apache/iotdb/cluster/log/applier/DataLogApplier.java | 7 +++++++ .../src/main/java/org/apache/iotdb/cluster/server/Timer.java | 9 +++++++++ server/pom.xml | 5 +++++ .../main/java/org/apache/iotdb/db/engine/StorageEngine.java | 12 ++++++++++++ 4 files changed, 33 insertions(+) diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/DataLogApplier.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/DataLogApplier.java index ab332d4..cdc1b89 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/DataLogApplier.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/applier/DataLogApplier.java @@ -25,6 +25,8 @@ import org.apache.iotdb.cluster.log.Log; import org.apache.iotdb.cluster.log.logtypes.CloseFileLog; import org.apache.iotdb.cluster.log.logtypes.PhysicalPlanLog; import org.apache.iotdb.cluster.partition.slot.SlotPartitionTable; +import org.apache.iotdb.cluster.server.Timer; +import org.apache.iotdb.cluster.server.Timer.Statistic; import org.apache.iotdb.cluster.server.member.DataGroupMember; import org.apache.iotdb.cluster.server.member.MetaGroupMember; import org.apache.iotdb.db.engine.StorageEngine; @@ -68,9 +70,14 @@ public class DataLogApplier extends BaseApplier { } } else if (log instanceof CloseFileLog) { CloseFileLog closeFileLog = ((CloseFileLog) log); + long start; + if (Timer.ENABLE_INSTRUMENTING) { + start = System.nanoTime(); + } StorageEngine.getInstance().closeProcessor(new PartialPath(closeFileLog.getStorageGroupName()), closeFileLog.getPartitionId(), closeFileLog.isSeq(), false); + Statistic.RAFT_SENDER_CLOSE_FILE_LOG_APPLY.addNanoFromStart(start); } else { logger.error("Unsupported log: {}", log); } diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/Timer.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/Timer.java index 6adf4fb..9c1c0d0 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/server/Timer.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/Timer.java @@ -131,6 +131,15 @@ public class Timer { RAFT_SENDER_DATA_LOG_APPLY( RAFT_MEMBER_SENDER, "apply data log", TIME_SCALE, true, RAFT_SENDER_COMMIT_WAIT_LOG_APPLY_V2), + RAFT_SENDER_CLOSE_FILE_LOG_APPLY( + RAFT_MEMBER_SENDER, "apply close file log", TIME_SCALE, true, + RAFT_SENDER_DATA_LOG_APPLY), + RAFT_SENDER_CLOSE_FILE_LOCK_PROCESSOR( + RAFT_MEMBER_SENDER, "close file - lock processor", TIME_SCALE, true, + RAFT_SENDER_CLOSE_FILE_LOG_APPLY), + RAFT_SENDER_CLOSE_FILE_CLOSE_PROCESSOR( + RAFT_MEMBER_SENDER, "close file - close processor", TIME_SCALE, true, + RAFT_SENDER_CLOSE_FILE_LOG_APPLY), RAFT_SENDER_LOG_FROM_CREATE_TO_ACCEPT( RAFT_MEMBER_SENDER, "log from create to accept", TIME_SCALE, RaftMember.USE_LOG_DISPATCHER, DATA_GROUP_MEMBER_LOCAL_EXECUTION), diff --git a/server/pom.xml b/server/pom.xml index 157e677..00a76d1 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -76,6 +76,11 @@ <scope>test</scope> </dependency> <dependency> + <groupId>org.apache.iotdb</groupId> + <artifactId>cluster</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-collections4</artifactId> </dependency> diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java index 7fc24dc..ef04d30 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.engine; import java.io.File; import java.io.IOException; +import java.sql.Time; import java.util.ArrayList; import java.util.ConcurrentModificationException; import java.util.HashMap; @@ -36,6 +37,8 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.io.FileUtils; +import org.apache.iotdb.cluster.server.Timer; +import org.apache.iotdb.cluster.server.Timer.Statistic; import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -441,7 +444,12 @@ public class StorageEngine implements IService { if (processor != null) { logger.info("async closing sg processor is called for closing {}, seq = {}, partitionId = {}", storageGroupPath, isSeq, partitionId); + long start; + if (Timer.ENABLE_INSTRUMENTING) { + start = System.nanoTime(); + } processor.writeLock(); + Statistic.RAFT_SENDER_CLOSE_FILE_LOCK_PROCESSOR.addNanoFromStart(start); // to avoid concurrent modification problem, we need a new array list List<TsFileProcessor> processors = isSeq ? new ArrayList<>(processor.getWorkSequenceTsFileProcessors()) : @@ -449,11 +457,15 @@ public class StorageEngine implements IService { try { for (TsFileProcessor tsfileProcessor : processors) { if (tsfileProcessor.getTimeRangeId() == partitionId) { + if (Timer.ENABLE_INSTRUMENTING) { + start = System.nanoTime(); + } if (isSync) { processor.syncCloseOneTsFileProcessor(isSeq, tsfileProcessor); } else { processor.asyncCloseOneTsFileProcessor(isSeq, tsfileProcessor); } + Statistic.RAFT_SENDER_CLOSE_FILE_CLOSE_PROCESSOR.addNanoFromStart(start); break; } }
