[iotdb] branch rel/0.11 updated: [IOTDB-1247] [To rel/0.11] fix the insert blocked caused the bugs in mem control module (#2877)

[hx...@apache.org]

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

hxd pushed a commit to branch rel/0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/rel/0.11 by this push:
     new 16ba205  [IOTDB-1247] [To rel/0.11] fix the insert blocked caused the bugs in mem control module (#2877)
16ba205 is described below

commit 16ba20599490abc83b391df48d13d292e01c247b
Author: Haonan <hh...@outlook.com>
AuthorDate: Sat Mar 20 14:45:26 2021 +0800

    [IOTDB-1247] [To rel/0.11] fix the insert blocked caused the bugs in mem control module (#2877)
---
 RELEASE_NOTES.md                                   |  5 +-
 .../org/apache/iotdb/db/rescon/SystemInfo.java     | 62 +++++++++++-----------
 2 files changed, 34 insertions(+), 33 deletions(-)

diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md
index 1ac9bd3..b7c549e 100644
--- a/RELEASE_NOTES.md
+++ b/RELEASE_NOTES.md
@@ -33,7 +33,8 @@
 * ISSUE-2639 Fix possible NPE during end query process
 * Alter IT for An error is reported and the system is suspended occasionally
 * IOTDB-1149 print error for -e param when set maxPRC<=0
-* IOTDB-2648 Last query not right when having multiple devices
+* IOTDB-1247 Fix the insert blocked caused the bugs in mem control module
+* ISSUE-2648 Last query not right when having multiple devices
 * Delete mods files after compaction
 * ISSUE-2687 fix insert NaN bug
 * ISSUE-2598 Throw explicit exception when time series is unknown in where clause
@@ -651,4 +652,4 @@ If you use the previous unofficial version 0.7.0. It is incompatible with 0.8.0.
 * Import/export csv script bug
 * Log level and stack print in test
 * Bug in TsFile-Spark-Connector
-* A doc bug of QuickStart.md
\ No newline at end of file
+* A doc bug of QuickStart.md
diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/SystemInfo.java b/server/src/main/java/org/apache/iotdb/db/rescon/SystemInfo.java
index cbf76bd..dc314e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/rescon/SystemInfo.java
+++ b/server/src/main/java/org/apache/iotdb/db/rescon/SystemInfo.java
@@ -77,44 +77,44 @@ public class SystemInfo {
   }
 
   /**
-   * Report resetting the mem cost of sg to system.
-   * It will be called after flushing, closing and failed to insert
+   * Report resetting the mem cost of sg to system. It will be called after flushing, closing and
+   * failed to insert
    *
    * @param storageGroupInfo storage group
    */
-  public synchronized void resetStorageGroupStatus(StorageGroupInfo storageGroupInfo,
-      boolean shouldInvokeFlush) {
-    if (reportedSgMemCostMap.containsKey(storageGroupInfo)) {
-      this.totalSgMemCost -= (reportedSgMemCostMap.get(storageGroupInfo) -
-          storageGroupInfo.getMemCost());
-      storageGroupInfo.setLastReportedSize(storageGroupInfo.getMemCost());
-      reportedSgMemCostMap.put(storageGroupInfo, storageGroupInfo.getMemCost());
-      if (shouldInvokeFlush) {
-        checkSystemToInvokeFlush();
+  public void resetStorageGroupStatus(
+      StorageGroupInfo storageGroupInfo, boolean shouldInvokeFlush) {
+    boolean needForceAsyncFlush = false;
+    synchronized (this) {
+      if (reportedSgMemCostMap.containsKey(storageGroupInfo)) {
+        this.totalSgMemCost -=
+            (reportedSgMemCostMap.get(storageGroupInfo) - storageGroupInfo.getMemCost());
+        storageGroupInfo.setLastReportedSize(storageGroupInfo.getMemCost());
+        reportedSgMemCostMap.put(storageGroupInfo, storageGroupInfo.getMemCost());
       }
-    }
-  }
 
-  private void checkSystemToInvokeFlush() {
-    if (totalSgMemCost >= FLUSH_THERSHOLD && totalSgMemCost < REJECT_THERSHOLD) {
-      logger.debug("Some sg memory released but still exceeding flush proportion, call flush.");
-      if (rejected) {
-        logger.info("Some sg memory released, set system to normal status.");
+      if (totalSgMemCost >= FLUSH_THERSHOLD && totalSgMemCost < REJECT_THERSHOLD) {
+        logger.debug("Some sg memory released but still exceeding flush proportion, call flush.");
+        if (rejected) {
+          logger.info("Some sg memory released, set system to normal status.");
+        }
+        logCurrentTotalSGMemory();
+        rejected = false;
+        needForceAsyncFlush = true;
+      } else if (totalSgMemCost >= REJECT_THERSHOLD) {
+        logger.warn("Some sg memory released, but system is still in reject status.");
+        logCurrentTotalSGMemory();
+        rejected = true;
+        needForceAsyncFlush = true;
+
+      } else {
+        logger.debug("Some sg memory released, system is in normal status.");
+        logCurrentTotalSGMemory();
+        rejected = false;
       }
-      logCurrentTotalSGMemory();
-      rejected = false;
-      forceAsyncFlush();
     }
-    else if (totalSgMemCost >= REJECT_THERSHOLD) {
-      logger.warn("Some sg memory released, but system is still in reject status.");
-      logCurrentTotalSGMemory();
-      rejected = true;
+    if (shouldInvokeFlush && needForceAsyncFlush) {
       forceAsyncFlush();
-    } 
-    else {
-      logger.debug("Some sg memory released, system is in normal status.");
-      logCurrentTotalSGMemory();
-      rejected = false;
     }
   }
 
@@ -148,7 +148,7 @@ public class SystemInfo {
    * Be Careful!! This method can only be called by flush thread!
    */
   private void forceAsyncFlush() {
-    if (FlushManager.getInstance().getNumberOfWorkingTasks() > 0) {
+    if (FlushManager.getInstance().getNumberOfWorkingTasks() > 1) {
       return;
     }
     List<TsFileProcessor> processors = getTsFileProcessorsToFlush();