This is an automated email from the ASF dual-hosted git repository. xingtanzjr pushed a commit to branch speed_up_restart_recover in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit eb0f82e9130b7fd6f33843f2b03ccb56c964da91 Author: Zhijia Cao <[email protected]> AuthorDate: Fri Oct 27 19:49:37 2023 +0800 recover --- .../main/java/org/apache/iotdb/SessionExample.java | 17 ++++++------ .../db/storageengine/dataregion/DataRegion.java | 30 ++++++++++++++-------- .../dataregion/flush/FlushManager.java | 6 +++-- .../dataregion/memtable/TsFileProcessor.java | 13 ++++++---- .../storageengine/dataregion/wal/WALManager.java | 6 ++--- 5 files changed, 43 insertions(+), 29 deletions(-) diff --git a/example/session/src/main/java/org/apache/iotdb/SessionExample.java b/example/session/src/main/java/org/apache/iotdb/SessionExample.java index 60bf282ed8f..8d1deb764c5 100644 --- a/example/session/src/main/java/org/apache/iotdb/SessionExample.java +++ b/example/session/src/main/java/org/apache/iotdb/SessionExample.java @@ -26,7 +26,6 @@ import org.apache.iotdb.isession.template.Template; import org.apache.iotdb.isession.util.Version; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; -import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.session.Session; import org.apache.iotdb.session.template.MeasurementNode; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; @@ -78,13 +77,13 @@ public class SessionExample { // set session fetchSize session.setFetchSize(10000); - try { - session.createDatabase("root.sg1"); - } catch (StatementExecutionException e) { - if (e.getStatusCode() != TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode()) { - throw e; - } - } + // try { + //// session.createDatabase("root.sg1"); + // } catch (StatementExecutionException e) { + // if (e.getStatusCode() != TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode()) { + // throw e; + // } + // } // createTemplate(); createTimeseries(); @@ -401,7 +400,7 @@ public class SessionExample { // Method 1 to add tablet data long timestamp = System.currentTimeMillis(); - for (long row = 0; row < 100; row++) { + for (long row = 0; row < 10000000; row++) { int rowIndex = tablet.rowSize++; tablet.addTimestamp(rowIndex, timestamp); for (int s = 0; s < 3; s++) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index b5bf310c6e7..87d72ab0692 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -134,6 +134,8 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; import java.util.TreeMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -1363,21 +1365,21 @@ public class DataRegion implements IDataRegionForQuery { * @param sequence whether this tsfile processor is sequence or not * @param tsFileProcessor tsfile processor */ - public void asyncCloseOneTsFileProcessor(boolean sequence, TsFileProcessor tsFileProcessor) { + public Future<?> asyncCloseOneTsFileProcessor(boolean sequence, TsFileProcessor tsFileProcessor) { // for sequence tsfile, we update the endTimeMap only when the file is prepared to be closed. // for unsequence tsfile, we have maintained the endTimeMap when an insertion comes. if (closingSequenceTsFileProcessor.contains(tsFileProcessor) || closingUnSequenceTsFileProcessor.contains(tsFileProcessor) || tsFileProcessor.alreadyMarkedClosing()) { - return; + return null; } logger.info( "Async close tsfile: {}", tsFileProcessor.getTsFileResource().getTsFile().getAbsolutePath()); - + Future<?> future; if (sequence) { closingSequenceTsFileProcessor.add(tsFileProcessor); - tsFileProcessor.asyncClose(); + future = tsFileProcessor.asyncClose(); workSequenceTsFileProcessors.remove(tsFileProcessor.getTimeRangeId()); // if unsequence files don't contain this time range id, we should remove it's version @@ -1388,7 +1390,7 @@ public class DataRegion implements IDataRegionForQuery { logger.info("close a sequence tsfile processor {}", databaseName + "-" + dataRegionId); } else { closingUnSequenceTsFileProcessor.add(tsFileProcessor); - tsFileProcessor.asyncClose(); + future = tsFileProcessor.asyncClose(); workUnsequenceTsFileProcessors.remove(tsFileProcessor.getTimeRangeId()); // if sequence files don't contain this time range id, we should remove it's version @@ -1397,6 +1399,7 @@ public class DataRegion implements IDataRegionForQuery { timePartitionIdVersionControllerMap.remove(tsFileProcessor.getTimeRangeId()); } } + return future; } /** @@ -1591,7 +1594,7 @@ public class DataRegion implements IDataRegionForQuery { public void syncCloseAllWorkingTsFileProcessors() { synchronized (closeStorageGroupCondition) { try { - asyncCloseAllWorkingTsFileProcessors(); + List<Future<?>> futures = asyncCloseAllWorkingTsFileProcessors(); long startTime = System.currentTimeMillis(); while (!closingSequenceTsFileProcessor.isEmpty() || !closingUnSequenceTsFileProcessor.isEmpty()) { @@ -1603,7 +1606,12 @@ public class DataRegion implements IDataRegionForQuery { (System.currentTimeMillis() - startTime) / 1000); } } - } catch (InterruptedException e) { + for (Future<?> f : futures) { + if (f != null) { + f.get(); + } + } + } catch (InterruptedException | ExecutionException e) { logger.error( "CloseFileNodeCondition error occurs while waiting for closing the storage " + "group {}", @@ -1615,23 +1623,25 @@ public class DataRegion implements IDataRegionForQuery { } /** close all working tsfile processors */ - public void asyncCloseAllWorkingTsFileProcessors() { + public List<Future<?>> asyncCloseAllWorkingTsFileProcessors() { writeLock("asyncCloseAllWorkingTsFileProcessors"); + List<Future<?>> futures = new ArrayList<>(); try { logger.info("async force close all files in database: {}", databaseName + "-" + dataRegionId); // to avoid concurrent modification problem, we need a new array list for (TsFileProcessor tsFileProcessor : new ArrayList<>(workSequenceTsFileProcessors.values())) { - asyncCloseOneTsFileProcessor(true, tsFileProcessor); + futures.add(asyncCloseOneTsFileProcessor(true, tsFileProcessor)); } // to avoid concurrent modification problem, we need a new array list for (TsFileProcessor tsFileProcessor : new ArrayList<>(workUnsequenceTsFileProcessors.values())) { - asyncCloseOneTsFileProcessor(false, tsFileProcessor); + futures.add(asyncCloseOneTsFileProcessor(false, tsFileProcessor)); } } finally { writeUnlock(); } + return futures; } /** force close all working tsfile processors */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushManager.java index f00d85676a5..08a6a6dd624 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushManager.java @@ -33,6 +33,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.Future; @SuppressWarnings("squid:S6548") public class FlushManager implements FlushManagerMBean, IService { @@ -120,7 +121,7 @@ public class FlushManager implements FlushManagerMBean, IService { * @param tsFileProcessor tsFileProcessor to be flushed */ @SuppressWarnings("squid:S2445") - public void registerTsFileProcessor(TsFileProcessor tsFileProcessor) { + public Future<?> registerTsFileProcessor(TsFileProcessor tsFileProcessor) { synchronized (tsFileProcessor) { if (tsFileProcessor.isManagedByFlushManager()) { LOGGER.debug( @@ -138,7 +139,7 @@ public class FlushManager implements FlushManagerMBean, IService { tsFileProcessorQueue.size()); } tsFileProcessor.setManagedByFlushManager(true); - flushPool.submit(new FlushThread()); + return flushPool.submit(new FlushThread()); } else { if (LOGGER.isDebugEnabled()) { LOGGER.debug( @@ -148,6 +149,7 @@ public class FlushManager implements FlushManagerMBean, IService { } } } + return null; } private FlushManager() {} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 1363128e181..5215f10e842 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -85,6 +85,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Future; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -854,7 +855,7 @@ public class TsFileProcessor { } /** async close one tsfile, register and close it by another thread */ - public void asyncClose() { + public Future<?> asyncClose() { flushQueryLock.writeLock().lock(); if (logger.isDebugEnabled()) { logger.debug( @@ -883,7 +884,7 @@ public class TsFileProcessor { } if (shouldClose) { - return; + return null; } // when a flush thread serves this TsFileProcessor (because the processor is submitted by // registerTsFileProcessor()), the thread will seal the corresponding TsFile and @@ -904,9 +905,10 @@ public class TsFileProcessor { // When invoke closing TsFile after insert data to memTable, we shouldn't flush until invoke // flushing memTable in System module. - addAMemtableIntoFlushingList(tmpMemTable); + Future<?> future = addAMemtableIntoFlushingList(tmpMemTable); logger.info("Memtable {} has been added to flushing list", tmpMemTable); shouldClose = true; + return future; } catch (Exception e) { logger.error( "{}: {} async close failed, because", @@ -921,6 +923,7 @@ public class TsFileProcessor { FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName()); } } + return null; } /** @@ -1009,7 +1012,7 @@ public class TsFileProcessor { * queue, set the current working memtable as null and then register the tsfileProcessor into the * flushManager again. */ - private void addAMemtableIntoFlushingList(IMemTable tobeFlushed) throws IOException { + private Future<?> addAMemtableIntoFlushingList(IMemTable tobeFlushed) throws IOException { Map<String, Long> lastTimeForEachDevice = new HashMap<>(); if (sequence) { lastTimeForEachDevice = tobeFlushed.getMaxTime(); @@ -1048,7 +1051,7 @@ public class TsFileProcessor { totalMemTableSize += tobeFlushed.memSize(); } workMemTable = null; - FlushManager.getInstance().registerTsFileProcessor(this); + return FlushManager.getInstance().registerTsFileProcessor(this); } /** put back the memtable to MemTablePool and make metadata in writer visible */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java index 6ccdb2aa3d3..031fafc4fd3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java @@ -21,7 +21,6 @@ package org.apache.iotdb.db.storageengine.dataregion.wal; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; -import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.exception.StartupException; import org.apache.iotdb.commons.service.IService; @@ -274,8 +273,9 @@ public class WALManager implements IService { private void registerScheduleTask(long initDelayMs, long periodMs) { walDeleteThread = IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(ThreadName.WAL_DELETE.getName()); - ScheduledExecutorUtil.safelyScheduleWithFixedDelay( - walDeleteThread, this::deleteOutdatedFiles, initDelayMs, periodMs, TimeUnit.MILLISECONDS); + // ScheduledExecutorUtil.safelyScheduleWithFixedDelay( + // walDeleteThread, this::deleteOutdatedFiles, initDelayMs, periodMs, + // TimeUnit.MILLISECONDS); } @TestOnly
