[incubator-iotdb] branch refactor_mem_controller updated: record memory usage even in dangerous level

[ji...@apache.org]

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

jiangtian pushed a commit to branch refactor_mem_controller
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git


The following commit(s) were added to refs/heads/refactor_mem_controller by this push:
     new 12d9641  record memory usage even in dangerous level
12d9641 is described below

commit 12d96417e006e143b85c6296d77665727abd8cce
Author: 江天 <jt...@163.com>
AuthorDate: Fri Apr 19 13:36:04 2019 +0800

    record memory usage even in dangerous level
---
 .../db/engine/filenode/FileNodeProcessor.java      |  1 +
 .../db/engine/memcontrol/RecordMemController.java  | 54 ++++++++--------------
 2 files changed, 21 insertions(+), 34 deletions(-)

diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
index 47aca8e..98cd121 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
@@ -1525,6 +1525,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
 
     if (mergeFileWriter != null) {
       mergeFileWriter.endFile(fileSchema);
+      mergeFileWriter = null;
     }
     backupIntervalFile.setBaseDirIndex(directories.getTsFileFolderIndex(mergeBaseDir));
     backupIntervalFile.setRelativePath(mergeFileName);
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/RecordMemController.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/RecordMemController.java
index 80a8014..c9cb202 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/RecordMemController.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/memcontrol/RecordMemController.java
@@ -82,42 +82,18 @@ public class RecordMemController extends BasicMemController {
     AtomicLong userUsage = memMap.computeIfAbsent(user, k -> new AtomicLong(0));
     long oldUsage = userUsage.get();
 
-    long newTotUsage = totalMemUsed.get() + usage;
-    // check if the new usage will reach dangerous threshold
-    if (newTotUsage > dangerouseThreshold) {
-      logDangerous(newTotUsage, user);
+    long newTotUsage = totalMemUsed.addAndGet(usage);
+    userUsage.addAndGet(usage);
+    if (newTotUsage < warningThreshold) {
+      logSafe(newTotUsage, user, usage, oldUsage);
+      return UsageLevel.SAFE;
+    } else if (newTotUsage < dangerouseThreshold) {
+      logWarn(newTotUsage, user, usage, oldUsage);
+      return UsageLevel.WARNING;
+    } else {
+      logDangerous(newTotUsage, user, usage, oldUsage);
       return UsageLevel.DANGEROUS;
     }
-
-    if (newTotUsage < dangerouseThreshold) {
-      newTotUsage = totalMemUsed.addAndGet(usage);
-      // double check if updating will reach dangerous threshold
-      if (newTotUsage < warningThreshold) {
-        // still safe, action taken
-        userUsage.addAndGet(usage);
-        logSafe(newTotUsage, user, usage, oldUsage);
-        return UsageLevel.SAFE;
-      } else if (newTotUsage < dangerouseThreshold) {
-        // become warning because competition with other threads, still take the action
-        userUsage.addAndGet(usage);
-        logWarn(newTotUsage, user, usage, oldUsage);
-        return UsageLevel.WARNING;
-      } else {
-        logDangerous(newTotUsage, user);
-        // become dangerous because competition with other threads, discard this action
-        totalMemUsed.addAndGet(-usage);
-        return UsageLevel.DANGEROUS;
-      }
-    }
-    return null;
-  }
-
-  private void logDangerous(long newTotUsage, Object user) {
-    if (LOGGER.isWarnEnabled()) {
-      LOGGER.warn("Memory request from {} is denied, memory usage : {}",
-          user,
-          MemUtils.bytesCntToStr(newTotUsage));
-    }
   }
 
   private void logSafe(long newTotUsage, Object user, long usage, long oldUsage) {
@@ -138,6 +114,15 @@ public class RecordMemController extends BasicMemController {
     }
   }
 
+  private void logDangerous(long newTotUsage, Object user, long usage, long oldUsage) {
+    if (LOGGER.isWarnEnabled()) {
+      LOGGER.warn("Warning Threshold : {} allocated to {}, it is using {}, total usage {}",
+          MemUtils.bytesCntToStr(usage), user,
+          MemUtils.bytesCntToStr(oldUsage + usage),
+          MemUtils.bytesCntToStr(newTotUsage));
+    }
+  }
+
   /**
    * report the decreased memory usage of the object user.
    */
@@ -154,6 +139,7 @@ public class RecordMemController extends BasicMemController {
           .error("{} requests to free {} bytes while it only registered {} bytes", user,
               freeSize, usage);
       totalMemUsed.addAndGet(-usageLong);
+      usage.set(0);
     } else {
       long newTotalMemUsage = totalMemUsed.addAndGet(-freeSize);
       usage.addAndGet(-freeSize);