You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by zy...@apache.org on 2024/01/11 10:58:00 UTC
(iotdb) branch master updated: Fix PBTree metric and support async flushAll (#11884)
This is an automated email from the ASF dual-hosted git repository.
zyk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5e8d43f2faa Fix PBTree metric and support async flushAll (#11884)
5e8d43f2faa is described below
commit 5e8d43f2faa50376021c99e0a99d1914df5e9a64
Author: Chen YZ <43...@users.noreply.github.com>
AuthorDate: Thu Jan 11 18:57:55 2024 +0800
Fix PBTree metric and support async flushAll (#11884)
---
.../mtree/impl/pbtree/flush/Scheduler.java | 110 ++++++++++-----------
.../impl/pbtree/memory/ReleaseFlushMonitor.java | 3 +-
.../mtree/impl/pbtree/schemafile/SchemaFile.java | 7 +-
.../schemafile/pagemgr/BTreePageManager.java | 10 +-
.../pbtree/schemafile/pagemgr/IPageManager.java | 3 +
.../pbtree/schemafile/pagemgr/PageManager.java | 16 +--
.../schemaRegion/SchemaStatisticsTest.java | 2 +-
7 files changed, 72 insertions(+), 79 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/flush/Scheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/flush/Scheduler.java
index 38282d12c2a..579db96c5e8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/flush/Scheduler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/flush/Scheduler.java
@@ -80,7 +80,7 @@ public class Scheduler {
* MNodes will be placed into node cache. This method will return synchronously after all stores
* are successfully flushed.
*/
- public synchronized void forceFlushAll() {
+ public synchronized CompletableFuture<Void> forceFlushAll() {
List<Map.Entry<Integer, CachedMTreeStore>> flushEngineList = new ArrayList<>();
for (Map.Entry<Integer, CachedMTreeStore> entry : regionToStore.entrySet()) {
if (flushingRegionSet.contains(entry.getKey())) {
@@ -89,61 +89,59 @@ public class Scheduler {
flushingRegionSet.add(entry.getKey());
flushEngineList.add(entry);
}
- CompletableFuture.allOf(
- flushEngineList.stream()
- .map(
- entry ->
- CompletableFuture.runAsync(
- () -> {
- int regionId = entry.getKey();
- CachedMTreeStore store = entry.getValue();
- if (store == null) {
- // store has been closed
- return;
- }
- IMemoryManager memoryManager = store.getMemoryManager();
- ISchemaFile file = store.getSchemaFile();
- LockManager lockManager = store.getLockManager();
- long startTime = System.currentTimeMillis();
- AtomicLong flushNodeNum = new AtomicLong(0);
- AtomicLong flushMemSize = new AtomicLong(0);
- try {
- lockManager.globalReadLock();
- if (!regionToStore.containsKey(regionId)) {
- // double check store have not been closed
- return;
- }
- PBTreeFlushExecutor flushExecutor =
- new PBTreeFlushExecutor(memoryManager, file, lockManager);
- flushExecutor.flushVolatileNodes(flushNodeNum, flushMemSize);
- } catch (MetadataException e) {
- LOGGER.warn(
- "Error occurred during MTree flush, current SchemaRegionId is {} because {}",
- regionId,
- e.getMessage(),
- e);
- } finally {
- long time = System.currentTimeMillis() - startTime;
- if (time > 10_000) {
- LOGGER.info(
- "It takes {}ms to flush MTree in SchemaRegion {}",
- time,
- regionId);
- } else {
- LOGGER.debug(
- "It takes {}ms to flush MTree in SchemaRegion {}",
- time,
- regionId);
- }
- store.recordFlushMetrics(
- time, flushNodeNum.get(), flushMemSize.get());
- lockManager.globalReadUnlock();
- flushingRegionSet.remove(regionId);
- }
- },
- workerPool))
- .toArray(CompletableFuture[]::new))
- .join();
+ return CompletableFuture.allOf(
+ flushEngineList.stream()
+ .map(
+ entry ->
+ CompletableFuture.runAsync(
+ () -> {
+ int regionId = entry.getKey();
+ CachedMTreeStore store = entry.getValue();
+ if (store == null) {
+ // store has been closed
+ return;
+ }
+ IMemoryManager memoryManager = store.getMemoryManager();
+ ISchemaFile file = store.getSchemaFile();
+ LockManager lockManager = store.getLockManager();
+ long startTime = System.currentTimeMillis();
+ AtomicLong flushNodeNum = new AtomicLong(0);
+ AtomicLong flushMemSize = new AtomicLong(0);
+ try {
+ lockManager.globalReadLock();
+ if (!regionToStore.containsKey(regionId)) {
+ // double check store have not been closed
+ return;
+ }
+ PBTreeFlushExecutor flushExecutor =
+ new PBTreeFlushExecutor(memoryManager, file, lockManager);
+ flushExecutor.flushVolatileNodes(flushNodeNum, flushMemSize);
+ } catch (MetadataException e) {
+ LOGGER.warn(
+ "Error occurred during MTree flush, current SchemaRegionId is {} because {}",
+ regionId,
+ e.getMessage(),
+ e);
+ } finally {
+ long time = System.currentTimeMillis() - startTime;
+ if (time > 10_000) {
+ LOGGER.info(
+ "It takes {}ms to flush MTree in SchemaRegion {}",
+ time,
+ regionId);
+ } else {
+ LOGGER.debug(
+ "It takes {}ms to flush MTree in SchemaRegion {}",
+ time,
+ regionId);
+ }
+ store.recordFlushMetrics(time, flushNodeNum.get(), flushMemSize.get());
+ lockManager.globalReadUnlock();
+ flushingRegionSet.remove(regionId);
+ }
+ },
+ workerPool))
+ .toArray(CompletableFuture[]::new));
}
/**
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/memory/ReleaseFlushMonitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/memory/ReleaseFlushMonitor.java
index 5f584a9061e..cf367416968 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/memory/ReleaseFlushMonitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/memory/ReleaseFlushMonitor.java
@@ -121,7 +121,6 @@ public class ReleaseFlushMonitor {
if (releaseFlushStrategy.isExceedReleaseThreshold()) {
scheduler.forceFlushAll();
regionToTraverserTime.values().forEach(RecordList::clear);
- scheduler.scheduleRelease(false);
}
synchronized (blockObject) {
// invoke the notifyAll() method to wake up the thread waiting for the release
@@ -239,7 +238,7 @@ public class ReleaseFlushMonitor {
}
}
if (needFlush) {
- scheduler.forceFlushAll();
+ scheduler.forceFlushAll().join();
scheduler.scheduleRelease(true);
} else {
break;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java
index 0de07f4958f..1b59a45cf53 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java
@@ -79,7 +79,6 @@ public class SchemaFile implements ISchemaFile {
// attributes for file
private File pmtFile;
private FileChannel channel;
- private SchemaRegionCachedMetric metric = null;
private final IMNodeFactory<ICachedMNode> nodeFactory =
MNodeFactoryLoader.getInstance().getCachedMNodeIMNodeFactory();
@@ -359,7 +358,7 @@ public class SchemaFile implements ISchemaFile {
ReadWriteIOUtils.write(sgNodeTemplateIdWithState, headerContent);
ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent);
lastSGAddr = 0L;
- pageManager = new BTreePageManager(channel, pmtFile, -1, logPath, metric);
+ pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
} else {
channel.read(headerContent);
headerContent.clear();
@@ -374,7 +373,7 @@ public class SchemaFile implements ISchemaFile {
throw new MetadataException("SchemaFile with wrong version, please check or upgrade.");
}
- pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath, metric);
+ pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath);
}
}
@@ -451,7 +450,7 @@ public class SchemaFile implements ISchemaFile {
}
public void setMetric(SchemaRegionCachedMetric metric) {
- this.metric = metric;
+ pageManager.setMetric(metric);
}
// endregion
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/BTreePageManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/BTreePageManager.java
index 4e1b347460c..ddb47c97361 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/BTreePageManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/BTreePageManager.java
@@ -21,7 +21,6 @@ package org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafi
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.db.exception.metadata.schemafile.ColossalRecordException;
-import org.apache.iotdb.db.schemaengine.metric.SchemaRegionCachedMetric;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.ICachedMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.ISchemaPage;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.ISegmentedPage;
@@ -43,14 +42,9 @@ import static org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.sc
public class BTreePageManager extends PageManager {
- public BTreePageManager(
- FileChannel channel,
- File pmtFile,
- int lastPageIndex,
- String logPath,
- SchemaRegionCachedMetric metric)
+ public BTreePageManager(FileChannel channel, File pmtFile, int lastPageIndex, String logPath)
throws IOException, MetadataException {
- super(channel, pmtFile, lastPageIndex, logPath, metric);
+ super(channel, pmtFile, lastPageIndex, logPath);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/IPageManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/IPageManager.java
index d1014550407..bb9baf86fa7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/IPageManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/IPageManager.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.pagemgr;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.db.schemaengine.metric.SchemaRegionCachedMetric;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.ICachedMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaFile;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaPage;
@@ -57,4 +58,6 @@ public interface IPageManager {
int getLastPageIndex();
void inspect(PrintWriter pw) throws IOException, MetadataException;
+
+ void setMetric(SchemaRegionCachedMetric metric);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java
index 5fecad18d6d..97be15f4ef6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java
@@ -95,18 +95,13 @@ public abstract class PageManager implements IPageManager {
private final AtomicInteger logCounter;
private SchemaFileLogWriter logWriter;
- private final SchemaRegionCachedMetric metric;
+ private SchemaRegionCachedMetric metric = null;
// flush strategy is dependent on consensus protocol, only check protocol on init
protected FlushPageStrategy flushDirtyPagesStrategy;
protected SinglePageFlushStrategy singlePageFlushStrategy;
- PageManager(
- FileChannel channel,
- File pmtFile,
- int lastPageIndex,
- String logPath,
- SchemaRegionCachedMetric metric)
+ PageManager(FileChannel channel, File pmtFile, int lastPageIndex, String logPath)
throws IOException, MetadataException {
this.pageInstCache =
Collections.synchronizedMap(new LinkedHashMap<>(SchemaFileConfig.PAGE_CACHE_SIZE, 1, true));
@@ -739,7 +734,7 @@ public abstract class PageManager implements IPageManager {
ByteBuffer newBuf = ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH);
if (metric != null) {
- metric.recordFlushPageNum(1);
+ metric.recordLoadPageNum(1);
}
loadFromFile(newBuf, pageIdx);
page = ISchemaPage.loadSchemaPage(newBuf);
@@ -1172,4 +1167,9 @@ public abstract class PageManager implements IPageManager {
return null;
}
}
+
+ @Override
+ public void setMetric(SchemaRegionCachedMetric metric) {
+ this.metric = metric;
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaStatisticsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaStatisticsTest.java
index 5b7c4dd6589..da8d88b34c3 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaStatisticsTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/schemaRegion/SchemaStatisticsTest.java
@@ -124,7 +124,7 @@ public class SchemaStatisticsTest extends AbstractSchemaRegionTest {
IMNodeFactory<?> nodeFactory = MNodeFactoryLoader.getInstance().getCachedMNodeIMNodeFactory();
// wait release and flush task
- Thread.sleep(1000);
+ Thread.sleep(6500);
// schemaRegion1
IMNode<?> sg1 =
nodeFactory.createDatabaseMNode(