You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2023/11/21 08:42:34 UTC
(iotdb) 01/03: Speed up recover stage when DataNode restarting (#11421)
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch yongfeng_rel12_1121
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit fdd7580e9760a4d74a6fbcb97e9d7a8c8f2aa317
Author: Zhang.Jinrui <xi...@gmail.com>
AuthorDate: Tue Oct 31 12:02:11 2023 +0800
Speed up recover stage when DataNode restarting (#11421)
---
.../apache/iotdb/db/service/IoTDBShutdownHook.java | 5 ++-
.../iotdb/db/storageengine/StorageEngine.java | 5 +++
.../db/storageengine/dataregion/DataRegion.java | 36 +++++++++++++---------
.../dataregion/flush/FlushManager.java | 7 +++--
.../dataregion/memtable/TsFileProcessor.java | 14 ++++++---
5 files changed, 45 insertions(+), 22 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/IoTDBShutdownHook.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/IoTDBShutdownHook.java
index 97d4efa204f..ad13ff320b9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/IoTDBShutdownHook.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/IoTDBShutdownHook.java
@@ -68,7 +68,10 @@ public class IoTDBShutdownHook extends Thread {
WALManager.getInstance().waitAllWALFlushed();
// flush data to Tsfile and remove WAL log files
- if (!IoTDBDescriptor.getInstance().getConfig().isClusterMode()) {
+ if (!IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getDataRegionConsensusProtocolClass()
+ .equals(ConsensusFactory.RATIS_CONSENSUS)) {
StorageEngine.getInstance().syncCloseAllProcessor();
}
WALManager.getInstance().deleteOutdatedWALFiles();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index d1b8cd8c1ff..a644b310596 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -309,6 +309,11 @@ public class StorageEngine implements IService {
}
recover();
+ for (DataRegion dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null) {
+ dataRegion.initCompaction();
+ }
+ }
ttlCheckThread =
IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(ThreadName.TTL_CHECK.getName());
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 0b8ad3e5a3f..96fac15d66f 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
@@ -133,6 +133,9 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+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;
@@ -544,9 +547,6 @@ public class DataRegion implements IDataRegionForQuery {
throw new DataRegionException(e);
}
- // recover and start timed compaction thread
- initCompaction();
-
if (StorageEngine.getInstance().isAllSgReady()) {
logger.info("The data region {}[{}] is created successfully", databaseName, dataRegionId);
} else {
@@ -568,7 +568,7 @@ public class DataRegion implements IDataRegionForQuery {
}
}
- private void initCompaction() {
+ public void initCompaction() {
if (!config.isEnableSeqSpaceCompaction()
&& !config.isEnableUnseqSpaceCompaction()
&& !config.isEnableCrossSpaceCompaction()) {
@@ -1349,21 +1349,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 CompletableFuture.completedFuture(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
@@ -1374,7 +1374,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
@@ -1383,6 +1383,7 @@ public class DataRegion implements IDataRegionForQuery {
timePartitionIdVersionControllerMap.remove(tsFileProcessor.getTimeRangeId());
}
}
+ return future;
}
/**
@@ -1585,7 +1586,7 @@ public class DataRegion implements IDataRegionForQuery {
public void syncCloseAllWorkingTsFileProcessors() {
synchronized (closeStorageGroupCondition) {
try {
- asyncCloseAllWorkingTsFileProcessors();
+ List<Future<?>> tsFileProcessorsClosingFutures = asyncCloseAllWorkingTsFileProcessors();
long startTime = System.currentTimeMillis();
while (!closingSequenceTsFileProcessor.isEmpty()
|| !closingUnSequenceTsFileProcessor.isEmpty()) {
@@ -1597,7 +1598,12 @@ public class DataRegion implements IDataRegionForQuery {
(System.currentTimeMillis() - startTime) / 1000);
}
}
- } catch (InterruptedException e) {
+ for (Future<?> f : tsFileProcessorsClosingFutures) {
+ if (f != null) {
+ f.get();
+ }
+ }
+ } catch (InterruptedException | ExecutionException e) {
logger.error(
"CloseFileNodeCondition error occurs while waiting for closing the storage "
+ "group {}",
@@ -1609,23 +1615,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..87f0440c3be 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
@@ -32,7 +32,9 @@ import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.Future;
@SuppressWarnings("squid:S6548")
public class FlushManager implements FlushManagerMBean, IService {
@@ -120,7 +122,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 +140,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 +150,7 @@ public class FlushManager implements FlushManagerMBean, IService {
}
}
}
+ return CompletableFuture.completedFuture(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 eaae14fce3a..f74ae4b8f9e 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
@@ -83,8 +83,10 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
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;
@@ -852,7 +854,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(
@@ -881,7 +883,7 @@ public class TsFileProcessor {
}
if (shouldClose) {
- return;
+ return CompletableFuture.completedFuture(null);
}
// when a flush thread serves this TsFileProcessor (because the processor is submitted by
// registerTsFileProcessor()), the thread will seal the corresponding TsFile and
@@ -903,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",
@@ -920,6 +923,7 @@ public class TsFileProcessor {
FLUSH_QUERY_WRITE_RELEASE, storageGroupName, tsFileResource.getTsFile().getName());
}
}
+ return CompletableFuture.completedFuture(null);
}
/**
@@ -1008,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();
@@ -1047,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 */