[iotdb] branch IOTDB-3164 updated: remove bug in chooseMemtableToFlush

[ma...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

marklau99 pushed a commit to branch IOTDB-3164
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/IOTDB-3164 by this push:
     new 7364451bd8 remove bug in chooseMemtableToFlush
7364451bd8 is described below

commit 7364451bd8da78c52c72d51bdd467e885b0f13e8
Author: Liu Xuxin <li...@outlook.com>
AuthorDate: Thu Jul 21 18:16:38 2022 +0800

    remove bug in chooseMemtableToFlush
---
 .../src/main/java/org/apache/iotdb/db/engine/StorageEngine.java  | 2 +-
 .../org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java | 3 ---
 .../java/org/apache/iotdb/db/rescon/memory/MemoryController.java | 2 +-
 .../org/apache/iotdb/db/rescon/memory/WriteMemoryController.java | 9 ++++++++-
 4 files changed, 10 insertions(+), 6 deletions(-)

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 701e5acc96..5ea98f43a8 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
@@ -195,7 +195,7 @@ public class StorageEngine implements IService {
   public static void blockInsertionIfReject(TsFileProcessor tsFileProcessor)
       throws WriteProcessRejectException {
     long startTime = System.currentTimeMillis();
-    while (WriteMemoryController.getInstance().isRejected()) {
+    while (WriteMemoryController.getInstance().checkRejected()) {
       if (tsFileProcessor != null && tsFileProcessor.shouldFlush()) {
         break;
       }
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 156b640caf..1f041b8285 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -786,7 +786,6 @@ public class TsFileProcessor {
     tsFileProcessorInfo.addTSPMemCost(chunkMetadataIncrement);
     WriteMemoryController controller = WriteMemoryController.getInstance();
     boolean allocateSuccess = false;
-    long startTime = System.nanoTime();
     try {
       while (workMemTable.needToAllocate(memTableIncrement)) {
         allocateSuccess =
@@ -796,8 +795,6 @@ public class TsFileProcessor {
         } else {
           storageGroupInfo.addAllocateSize(WriteMemoryController.FRAME_SIZE);
           workMemTable.addAllocatedMemSize(WriteMemoryController.FRAME_SIZE);
-          logger.error(
-              "Allocation time for {} is {} ns", storageGroupName, System.nanoTime() - startTime);
         }
       }
     } catch (WriteProcessRejectException e) {
diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/memory/MemoryController.java b/server/src/main/java/org/apache/iotdb/db/rescon/memory/MemoryController.java
index b6fa801547..71e7b4d96f 100644
--- a/server/src/main/java/org/apache/iotdb/db/rescon/memory/MemoryController.java
+++ b/server/src/main/java/org/apache/iotdb/db/rescon/memory/MemoryController.java
@@ -137,7 +137,7 @@ public class MemoryController<T> {
     }
   }
 
-  private void checkTrigger(long newUsage, T triggerParam) {
+  protected void checkTrigger(long newUsage, T triggerParam) {
     if (newUsage >= triggerThreshold && trigger != null) {
       if (triggerRunning.compareAndSet(false, true)) {
         try {
diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/memory/WriteMemoryController.java b/server/src/main/java/org/apache/iotdb/db/rescon/memory/WriteMemoryController.java
index 1581ecd7ac..bb4bb32c0c 100644
--- a/server/src/main/java/org/apache/iotdb/db/rescon/memory/WriteMemoryController.java
+++ b/server/src/main/java/org/apache/iotdb/db/rescon/memory/WriteMemoryController.java
@@ -56,6 +56,9 @@ public class WriteMemoryController extends MemoryController<TsFileProcessor> {
 
   public boolean tryAllocateMemory(long size, StorageGroupInfo info, TsFileProcessor processor) {
     boolean success = super.tryAllocateMemory(size, processor);
+    if (!success) {
+      checkTrigger(memoryUsage.get(), processor);
+    }
     if (memoryUsage.get() > REJECT_THRESHOLD && !rejected) {
       logger.info(
           "Change system to reject status. Triggered by: logical SG ({}), mem cost delta ({}), totalSgMemCost ({}).",
@@ -112,7 +115,10 @@ public class WriteMemoryController extends MemoryController<TsFileProcessor> {
         ((double) memoryUsage.get()) / 1024.0d / 1024.0d);
   }
 
-  public boolean isRejected() {
+  public boolean checkRejected() {
+    if (rejected) {
+      checkTrigger(memoryUsage.get(), null);
+    }
     return rejected;
   }
 
@@ -168,6 +174,7 @@ public class WriteMemoryController extends MemoryController<TsFileProcessor> {
       }
       if (selectedTsFileProcessor.getWorkMemTable() == null
           || selectedTsFileProcessor.getWorkMemTable().shouldFlush()) {
+        allTsFileProcessors.poll();
         continue;
       }
       long memUsageForThisMemTable = selectedTsFileProcessor.getWorkMemTableAllocateSize();