This is an automated email from the ASF dual-hosted git repository. marklau99 pushed a commit to branch fix-compaction-tmp-error-master in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 925f5c1e38e0047cd3d8ac20be411322c90d02a7 Author: Liu Xuxin <[email protected]> AuthorDate: Fri Apr 28 19:39:56 2023 +0800 [To rel/1.1] [IOTDB-5828] Optimize the performance of some parts in metrics, and correcting the metrics count of temporal file size in inner space compaction (#9743) --- .../metrics/metricsets/jvm/JvmThreadMetrics.java | 31 +++++++++-- .../iotdb/db/engine/TsFileMetricManager.java | 63 ++++++++++++++-------- .../performer/impl/FastCompactionPerformer.java | 16 +----- .../impl/ReadChunkCompactionPerformer.java | 11 +--- .../impl/ReadPointCompactionPerformer.java | 21 +------- .../execute/task/AbstractCompactionTask.java | 4 ++ .../execute/task/CompactionTaskSummary.java | 18 +++++++ .../readchunk/AlignedSeriesCompactionExecutor.java | 6 --- .../iotdb/db/service/metrics/FileMetrics.java | 26 +++++++-- 9 files changed, 118 insertions(+), 78 deletions(-) diff --git a/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetrics.java b/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetrics.java index 5c4731de384..3ecf6f35fb9 100644 --- a/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetrics.java +++ b/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetrics.java @@ -25,11 +25,20 @@ import org.apache.iotdb.metrics.utils.MetricLevel; import org.apache.iotdb.metrics.utils.MetricType; import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; import java.lang.management.ThreadMXBean; import java.util.Arrays; +import java.util.EnumMap; +import java.util.List; +import java.util.Map; /** This file is modified from io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics */ public class JvmThreadMetrics implements IMetricSet { + private static long lastUpdateTime = 0L; + private static final long UPDATE_INTERVAL = 10_000L; + private static final Map<Thread.State, Integer> threadStateCountMap = + new EnumMap<>(Thread.State.class); + @Override public void bindTo(AbstractMetricService metricService) { ThreadMXBean threadBean = ManagementFactory.getThreadMXBean(); @@ -85,9 +94,25 @@ public class JvmThreadMetrics implements IMetricSet { // VisibleForTesting static long getThreadStateCount(ThreadMXBean threadBean, Thread.State state) { - return Arrays.stream(threadBean.getThreadInfo(threadBean.getAllThreadIds())) - .filter(threadInfo -> threadInfo != null && threadInfo.getThreadState() == state) - .count(); + checkAndUpdate(threadBean); + return threadStateCountMap.getOrDefault(state, 0); + } + + private static void checkAndUpdate(ThreadMXBean threadBean) { + if (System.currentTimeMillis() - lastUpdateTime < UPDATE_INTERVAL) { + return; + } + lastUpdateTime = System.currentTimeMillis(); + threadStateCountMap.clear(); + List<ThreadInfo> infoList = + Arrays.asList(threadBean.getThreadInfo(threadBean.getAllThreadIds())); + infoList.forEach( + info -> { + if (info != null) { + Thread.State state = info.getThreadState(); + threadStateCountMap.compute(state, (k, v) -> v == null ? 1 : v + 1); + } + }); } private static String getStateTagValue(Thread.State state) { diff --git a/server/src/main/java/org/apache/iotdb/db/engine/TsFileMetricManager.java b/server/src/main/java/org/apache/iotdb/db/engine/TsFileMetricManager.java index 13a205a0ad5..a3cf6c83361 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/TsFileMetricManager.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/TsFileMetricManager.java @@ -19,8 +19,13 @@ package org.apache.iotdb.db.engine; +import org.apache.iotdb.db.engine.compaction.execute.task.AbstractCompactionTask; +import org.apache.iotdb.db.engine.compaction.execute.task.CompactionTaskSummary; +import org.apache.iotdb.db.engine.compaction.execute.task.InnerSpaceCompactionTask; +import org.apache.iotdb.db.engine.compaction.schedule.CompactionTaskManager; import org.apache.iotdb.db.service.metrics.FileMetrics; +import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -35,6 +40,8 @@ public class TsFileMetricManager { private final AtomicInteger modFileNum = new AtomicInteger(0); private final AtomicLong modFileSize = new AtomicLong(0); + private long lastUpdateTime = 0; + private static final long UPDATE_INTERVAL = 10_000L; // compaction temporal files private final AtomicLong innerSeqCompactionTempFileSize = new AtomicLong(0); @@ -102,41 +109,55 @@ public class TsFileMetricManager { modFileSize.addAndGet(-size); } - public void addCompactionTempFileSize(boolean innerSpace, boolean seq, long delta) { - if (innerSpace) { - long unused = - seq - ? innerSeqCompactionTempFileSize.addAndGet(delta) - : innerUnseqCompactionTempFileSize.addAndGet(delta); - } else { - crossCompactionTempFileSize.addAndGet(delta); - } + public long getInnerCompactionTempFileSize(boolean seq) { + updateCompactionTempSize(); + return seq ? innerSeqCompactionTempFileSize.get() : innerUnseqCompactionTempFileSize.get(); } - public void addCompactionTempFileNum(boolean innerSpace, boolean seq, int delta) { - if (innerSpace) { - long unused = - seq - ? innerSeqCompactionTempFileNum.addAndGet(delta) - : innerUnseqCompactionTempFileNum.addAndGet(delta); - } else { - crossCompactionTempFileNum.addAndGet(delta); + private synchronized void updateCompactionTempSize() { + if (System.currentTimeMillis() - lastUpdateTime <= UPDATE_INTERVAL) { + return; + } + lastUpdateTime = System.currentTimeMillis(); + + innerSeqCompactionTempFileSize.set(0); + innerSeqCompactionTempFileNum.set(0); + innerUnseqCompactionTempFileSize.set(0); + innerUnseqCompactionTempFileNum.set(0); + crossCompactionTempFileSize.set(0); + crossCompactionTempFileNum.set(0); + + List<AbstractCompactionTask> runningTasks = + CompactionTaskManager.getInstance().getRunningCompactionTaskList(); + for (AbstractCompactionTask task : runningTasks) { + CompactionTaskSummary summary = task.getSummary(); + if (task instanceof InnerSpaceCompactionTask) { + if (task.isInnerSeqTask()) { + innerSeqCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); + innerSeqCompactionTempFileNum.addAndGet(1); + } else { + innerUnseqCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); + innerUnseqCompactionTempFileNum.addAndGet(1); + } + } else { + crossCompactionTempFileSize.addAndGet(summary.getTemporalFileSize()); + crossCompactionTempFileNum.addAndGet(summary.getTemporalFileNum()); + } } - } - - public long getInnerCompactionTempFileSize(boolean seq) { - return seq ? innerSeqCompactionTempFileSize.get() : innerUnseqCompactionTempFileSize.get(); } public long getCrossCompactionTempFileSize() { + updateCompactionTempSize(); return crossCompactionTempFileSize.get(); } public long getInnerCompactionTempFileNum(boolean seq) { + updateCompactionTempSize(); return seq ? innerSeqCompactionTempFileNum.get() : innerUnseqCompactionTempFileNum.get(); } public long getCrossCompactionTempFileNum() { + updateCompactionTempSize(); return crossCompactionTempFileNum.get(); } } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/FastCompactionPerformer.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/FastCompactionPerformer.java index d029122f0d1..3319594effa 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/FastCompactionPerformer.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/FastCompactionPerformer.java @@ -22,7 +22,6 @@ import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.db.conf.IoTDBDescriptor; -import org.apache.iotdb.db.engine.TsFileMetricManager; import org.apache.iotdb.db.engine.compaction.execute.performer.ICrossCompactionPerformer; import org.apache.iotdb.db.engine.compaction.execute.performer.ISeqCompactionPerformer; import org.apache.iotdb.db.engine.compaction.execute.performer.IUnseqCompactionPerformer; @@ -81,8 +80,6 @@ public class FastCompactionPerformer private boolean isCrossCompaction; - private long tempFileSize = 0L; - public FastCompactionPerformer( List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles, @@ -105,8 +102,7 @@ public class FastCompactionPerformer @Override public void perform() throws IOException, MetadataException, StorageEngineException, InterruptedException { - TsFileMetricManager.getInstance() - .addCompactionTempFileNum(!isCrossCompaction, !seqFiles.isEmpty(), targetFiles.size()); + this.subTaskSummary.setTemporalFileNum(targetFiles.size()); try (MultiTsFileDeviceIterator deviceIterator = new MultiTsFileDeviceIterator(seqFiles, unseqFiles, readerCacheMap); AbstractCompactionWriter compactionWriter = @@ -139,11 +135,7 @@ public class FastCompactionPerformer // check whether to flush chunk metadata or not compactionWriter.checkAndMayFlushChunkMetadata(); // Add temp file metrics - long currentTempFileSize = compactionWriter.getWriterSize(); - TsFileMetricManager.getInstance() - .addCompactionTempFileSize( - !isCrossCompaction, !seqFiles.isEmpty(), currentTempFileSize - tempFileSize); - tempFileSize = currentTempFileSize; + subTaskSummary.setTemporalFileSize(compactionWriter.getWriterSize()); sortedSourceFiles.clear(); } compactionWriter.endFile(); @@ -156,10 +148,6 @@ public class FastCompactionPerformer sortedSourceFiles = null; readerCacheMap = null; modificationCache = null; - TsFileMetricManager.getInstance() - .addCompactionTempFileNum(!isCrossCompaction, !seqFiles.isEmpty(), -targetFiles.size()); - TsFileMetricManager.getInstance() - .addCompactionTempFileSize(!isCrossCompaction, !seqFiles.isEmpty(), -tempFileSize); } } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java index e97f2e5fe35..da0c9c91dd8 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java @@ -22,7 +22,6 @@ import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBDescriptor; -import org.apache.iotdb.db.engine.TsFileMetricManager; import org.apache.iotdb.db.engine.compaction.execute.performer.ISeqCompactionPerformer; import org.apache.iotdb.db.engine.compaction.execute.task.CompactionTaskSummary; import org.apache.iotdb.db.engine.compaction.execute.utils.MultiTsFileDeviceIterator; @@ -50,7 +49,6 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer { private TsFileResource targetResource; private List<TsFileResource> seqFiles; private CompactionTaskSummary summary; - private long tempFileSize = 0L; public ReadChunkCompactionPerformer(List<TsFileResource> sourceFiles, TsFileResource targetFile) { this.seqFiles = sourceFiles; @@ -72,7 +70,6 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer { ((double) SystemInfo.getInstance().getMemorySizeForCompaction() / IoTDBDescriptor.getInstance().getConfig().getCompactionThreadCount() * IoTDBDescriptor.getInstance().getConfig().getChunkMetadataSizeProportion()); - TsFileMetricManager.getInstance().addCompactionTempFileNum(true, true, 1); try (MultiTsFileDeviceIterator deviceIterator = new MultiTsFileDeviceIterator(seqFiles); TsFileIOWriter writer = new TsFileIOWriter(targetResource.getTsFile(), true, sizeForFileWriter)) { @@ -87,19 +84,13 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer { compactNotAlignedSeries(device, targetResource, writer, deviceIterator); } // update temporal file metrics - long newTempFileSize = writer.getPos(); - TsFileMetricManager.getInstance() - .addCompactionTempFileSize(true, true, newTempFileSize - tempFileSize); - tempFileSize = newTempFileSize; + summary.setTemporalFileSize(writer.getPos()); } for (TsFileResource tsFileResource : seqFiles) { targetResource.updatePlanIndexes(tsFileResource); } writer.endFile(); - } finally { - TsFileMetricManager.getInstance().addCompactionTempFileSize(true, true, -tempFileSize); - TsFileMetricManager.getInstance().addCompactionTempFileNum(true, true, -1); } } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/ReadPointCompactionPerformer.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/ReadPointCompactionPerformer.java index 8384916b92a..0f0f5185d4f 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/ReadPointCompactionPerformer.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/performer/impl/ReadPointCompactionPerformer.java @@ -25,7 +25,6 @@ import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBDescriptor; -import org.apache.iotdb.db.engine.TsFileMetricManager; import org.apache.iotdb.db.engine.compaction.execute.performer.ICrossCompactionPerformer; import org.apache.iotdb.db.engine.compaction.execute.performer.IUnseqCompactionPerformer; import org.apache.iotdb.db.engine.compaction.execute.task.CompactionTaskSummary; @@ -75,7 +74,6 @@ public class ReadPointCompactionPerformer private CompactionTaskSummary summary; private List<TsFileResource> targetFiles = Collections.emptyList(); - private long tempFileSize = 0L; public ReadPointCompactionPerformer( List<TsFileResource> seqFiles, @@ -103,8 +101,7 @@ public class ReadPointCompactionPerformer QueryResourceManager.getInstance() .getQueryFileManager() .addUsedFilesForQuery(queryId, queryDataSource); - TsFileMetricManager.getInstance() - .addCompactionTempFileNum(seqFiles.isEmpty(), false, targetFiles.size()); + summary.setTemporalFileNum(targetFiles.size()); try (AbstractCompactionWriter compactionWriter = getCompactionWriter(seqFiles, unseqFiles, targetFiles)) { // Do not close device iterator, because tsfile reader is managed by FileReaderManager. @@ -124,6 +121,7 @@ public class ReadPointCompactionPerformer compactNonAlignedSeries( device, deviceIterator, compactionWriter, fragmentInstanceContext, queryDataSource); } + summary.setTemporalFileSize(compactionWriter.getWriterSize()); } compactionWriter.endFile(); @@ -131,10 +129,6 @@ public class ReadPointCompactionPerformer } finally { QueryResourceManager.getInstance().endQuery(queryId); - TsFileMetricManager.getInstance() - .addCompactionTempFileNum(seqFiles.isEmpty(), false, -targetFiles.size()); - TsFileMetricManager.getInstance() - .addCompactionTempFileSize(seqFiles.isEmpty(), false, tempFileSize); } } @@ -186,11 +180,6 @@ public class ReadPointCompactionPerformer // check whether to flush chunk metadata or not compactionWriter.checkAndMayFlushChunkMetadata(); } - // add temp file metrics - long currentWriterSize = compactionWriter.getWriterSize(); - TsFileMetricManager.getInstance() - .addCompactionTempFileSize(seqFiles.isEmpty(), false, currentWriterSize - tempFileSize); - tempFileSize = currentWriterSize; } private void compactNonAlignedSeries( @@ -238,12 +227,6 @@ public class ReadPointCompactionPerformer // check whether to flush chunk metadata or not compactionWriter.checkAndMayFlushChunkMetadata(); } - - // add temp file metrics - long currentWriterSize = compactionWriter.getWriterSize(); - TsFileMetricManager.getInstance() - .addCompactionTempFileSize(seqFiles.isEmpty(), false, currentWriterSize - tempFileSize); - tempFileSize = currentWriterSize; } /** diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/AbstractCompactionTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/AbstractCompactionTask.java index 19e4a0efaa0..7fb2d23dc83 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/AbstractCompactionTask.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/AbstractCompactionTask.java @@ -161,6 +161,10 @@ public abstract class AbstractCompactionTask { return crossTask; } + public long getTemporalFileSize() { + return summary.getTemporalFileSize(); + } + public boolean isInnerSeqTask() { return innerSeqTask; } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/CompactionTaskSummary.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/CompactionTaskSummary.java index bcfc60675b5..0e1675d47bc 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/CompactionTaskSummary.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/CompactionTaskSummary.java @@ -32,6 +32,8 @@ public class CompactionTaskSummary { protected int deserializePageCount = 0; protected int mergedChunkNum = 0; protected long processPointNum = 0; + protected long temporalFileSize = 0; + protected int temporalFileNum = 0; public CompactionTaskSummary() {} @@ -134,6 +136,22 @@ public class CompactionTaskSummary { CANCELED } + public void setTemporalFileSize(long temporalFileSize) { + this.temporalFileSize = temporalFileSize; + } + + public long getTemporalFileSize() { + return temporalFileSize; + } + + public void setTemporalFileNum(int temporalFileNum) { + this.temporalFileNum = temporalFileNum; + } + + public int getTemporalFileNum() { + return temporalFileNum; + } + @Override public String toString() { String startTimeInStr = new SimpleDateFormat().format(new Date(startTime)); diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/utils/executor/readchunk/AlignedSeriesCompactionExecutor.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/utils/executor/readchunk/AlignedSeriesCompactionExecutor.java index 2ec407ad29e..52b98cff3a7 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/utils/executor/readchunk/AlignedSeriesCompactionExecutor.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/utils/executor/readchunk/AlignedSeriesCompactionExecutor.java @@ -19,7 +19,6 @@ package org.apache.iotdb.db.engine.compaction.execute.utils.executor.readchunk; import org.apache.iotdb.db.conf.IoTDBDescriptor; -import org.apache.iotdb.db.engine.TsFileMetricManager; import org.apache.iotdb.db.engine.compaction.execute.task.CompactionTaskSummary; import org.apache.iotdb.db.engine.compaction.schedule.CompactionTaskManager; import org.apache.iotdb.db.engine.compaction.schedule.constant.CompactionType; @@ -131,7 +130,6 @@ public class AlignedSeriesCompactionExecutor { } public void execute() throws IOException { - long originTempFileSize = writer.getPos(); while (readerAndChunkMetadataList.size() > 0) { Pair<TsFileSequenceReader, List<AlignedChunkMetadata>> readerListPair = readerAndChunkMetadataList.removeFirst(); @@ -162,10 +160,6 @@ public class AlignedSeriesCompactionExecutor { chunkWriter.writeToFileWriter(writer); } writer.checkMetadataSizeAndMayFlush(); - - // update temporal file metrics - TsFileMetricManager.getInstance() - .addCompactionTempFileSize(true, true, writer.getPos() - originTempFileSize); } private void compactOneAlignedChunk(AlignedChunkReader chunkReader, int notNullChunkNum) diff --git a/server/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java b/server/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java index 483c2b7345b..ec9b7775658 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java +++ b/server/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java @@ -35,6 +35,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.BufferedReader; +import java.io.File; import java.io.IOException; import java.io.InputStreamReader; @@ -48,6 +49,13 @@ public class FileMetrics implements IMetricSet { private final Runtime runtime = Runtime.getRuntime(); private String[] getOpenFileNumberCommand; + @SuppressWarnings("squid:S1075") + private String fileHandlerCntPathInLinux = "/proc/%s/fd"; + + public FileMetrics() { + fileHandlerCntPathInLinux = String.format(fileHandlerCntPathInLinux, METRIC_CONFIG.getPid()); + } + @Override public void bindTo(AbstractMetricService metricService) { bindTsFileMetrics(metricService); @@ -173,7 +181,7 @@ public class FileMetrics implements IMetricSet { }; metricService.createAutoGauge( Metric.FILE_COUNT.toString(), - MetricLevel.CORE, + MetricLevel.IMPORTANT, this, FileMetrics::getOpenFileHandlersNumber, Tag.NAME.toString(), @@ -245,9 +253,17 @@ public class FileMetrics implements IMetricSet { } private long getOpenFileHandlersNumber() { + long fdCount = 0; try { - if ((METRIC_CONFIG.getSystemType() == SystemType.LINUX - || METRIC_CONFIG.getSystemType() == SystemType.MAC) + if (METRIC_CONFIG.getSystemType() == SystemType.LINUX) { + // count the fd in the system directory instead of + // calling runtime.exec() which could be much slower + File fdDir = new File(fileHandlerCntPathInLinux); + if (fdDir.exists()) { + File[] fds = fdDir.listFiles(); + fdCount = fds == null ? 0 : fds.length; + } + } else if ((METRIC_CONFIG.getSystemType() == SystemType.MAC) && METRIC_CONFIG.getPid().length() != 0) { Process process = runtime.exec(getOpenFileNumberCommand); StringBuilder result = new StringBuilder(); @@ -258,11 +274,11 @@ public class FileMetrics implements IMetricSet { result.append(line); } } - return Long.parseLong(result.toString().trim()); + fdCount = Long.parseLong(result.toString().trim()); } } catch (IOException e) { LOGGER.warn("Failed to get open file number, because ", e); } - return 0L; + return fdCount; } }
