Write memory control (#196)

Author: jt2594838

iotdb/conf/iotdb-engine.properties

@@ -56,12 +56,12 @@ period_time_for_merge_in_second=7200
 # default value is +08:00
 # eg. +08:00, -01:00
 time_zone=+08:00
+        
+# if memory used by write reaches this threshold, auto flush will be triggered, percentile of Java heap memory
+mem_threshold_warning=0.8
 
-# if memory used by write reaches this threshold, auto flush will be triggered, in byte, 8GB by default
-mem_threshold_warning=23622320128
-
-# if memory used by write reaches this threshold, write will be blocked, in byte, 16GB by default
-mem_threshold_dangerous=25769803776
+# if memory used by write reaches this threshold, write will be blocked, percentile of Java heap memory
+mem_threshold_dangerous=0.9
 
 # every such interval, a thread will check if memory exceeds mem_threshold_warning
 # if do exceed, auto flush will be triggered, in ms, 1s by default
@@ -72,8 +72,25 @@ mem_monitor_interval=1000
 # 1 is JVMMemController, which use JVM heap memory as threshold.
 mem_controller_type=1
 
+# When a bufferwrite's metadata size (in byte) exceed this, the bufferwrite is forced closed.
+bufferwrite_meta_size_threshold=209715200
+
+# When a bufferwrite's file size (in byte) exceed this, the bufferwrite is forced closed.
+bufferwrite_file_size_threshold=2147483648
+
+# When a overflow's metadata size (in byte) exceed this, the bufferwrite is forced closed.
+overflow_meta_size_threshold=209715200
+
+# When a overflow's file size (in byte) exceed this, the bufferwrite is forced closed.
+overflow_file_size_threshold=2147483648
+
+# How many thread can concurrently flush. When <= 0, use CPU core number.
+concurrent_flush_thread=0
+
+
 # Statistics Monitor configuration
 # default monitor is enabled, and write statistics info to IoTDB every 5 seconds
 # Choose to change the back_loop_period >= 1 seconds
 enable_stat_monitor = true
-back_loop_period = 5
+back_loop_period = 5
+

iotdb/conf/iotdb-env.sh

@@ -110,5 +110,6 @@ fi
 IOTDB_DERBY_OPTS="-Dderby.stream.error.field=cn.edu.tsinghua.iotdb.auth.dao.DerbyUtil.DEV_NULL"
 
 IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -Xloggc:${IOTDB_HOME}/gc.log -XX:+PrintGCDateStamps -XX:+PrintGCDetails"
+
 IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -Xms${HEAP_NEWSIZE}"
 IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -Xmx${MAX_HEAP_SIZE}"

iotdb/conf/logback.xml

@@ -96,7 +96,7 @@
 
 	<logger name="cn.edu.tsinghua.iotdb.service" level="info" />
 	<logger name="cn.edu.tsinghua.iotdb.conf" level="info" />
-	<root level="ERROR">
+	<root level="info">
 		<appender-ref ref="FILEDEBUG" />
 		<appender-ref ref="FILEINFO" />
 		<appender-ref ref="FILEWARN" />

pom.xml

@@ -4,7 +4,7 @@
 
 	<groupId>cn.edu.tsinghua</groupId>
 	<artifactId>IoTDB</artifactId>
-	<version>0.3.1</version>
+	<version>0.4.0</version>
 	<packaging>jar</packaging>
 
 	<name>IoTDB</name>
@@ -23,7 +23,7 @@
 		<dependency>
 			<groupId>cn.edu.tsinghua</groupId>
 			<artifactId>iotdb-jdbc</artifactId>
-			<version>0.3.1</version>
+			<version>0.4.0</version>
 		</dependency>
 		<dependency>
 			<groupId>cn.edu.fudan.dsm</groupId>

src/main/java/cn/edu/tsinghua/iotdb/conf/TsFileDBConstant.java

@@ -4,12 +4,21 @@ public class TsFileDBConstant {
     public static final String ENV_FILE_NAME = "iotdb-env";
     public static final String IOTDB_CONF = "IOTDB_CONF";
 	public static final String GLOBAL_DB_NAME = "IoTDB";
-	public static final String VERSION = "0.3.1";
+	public static final String VERSION = "0.4.0";
     public static final String REMOTE_JMX_PORT_NAME = "com.sun.management.jmxremote.port";
     public static final String TSFILEDB_LOCAL_JMX_PORT_NAME = "iotdb.jmx.local.port";
     public static final String TSFILEDB_REMOTE_JMX_PORT_NAME = "iotdb.jmx.remote.port";
     public static final String SERVER_RMI_ID = "java.rmi.server.randomIDs";
     public static final String RMI_SERVER_HOST_NAME = "java.rmi.server.hostname";
     public static final String JMX_REMOTE_RMI_PORT = "com.sun.management.jmxremote.rmi.port";
+
+    public static final long GB = 1024 * 1024 * 1024L;
+    public static final long MB = 1024 * 1024L;
+    public static final long KB = 1024L;
+
+    public static final long MEM_THRESHOLD_WARNING_DEFAULT = 8 * GB;
+    public static final long MEM_THRESHOLD_DANGEROUS_DEFAULT = 16 * GB;
+
     public static final String IOTDB_HOME = "IOTDB_HOME";
+
 }

src/main/java/cn/edu/tsinghua/iotdb/conf/TsfileDBConfig.java

@@ -117,6 +117,50 @@ public class TsfileDBConfig {
 	public DateTimeZone timeZone = DateTimeZone.getDefault();
 
 	/**
+	 *  BufferWriteProcessor and OverflowProcessor will immediately flush if this threshold is reached.
+	 */
+	public long memThresholdWarning = (long) (0.8 * Runtime.getRuntime().maxMemory());
+
+	/**
+	 * No more insert is allowed if this threshold is reached.
+	 */
+	public long memThresholdDangerous = (long) (0.9 * Runtime.getRuntime().maxMemory());
+
+	/**
+	 * MemMonitorThread will check every such interval. If memThresholdWarning is reached, MemMonitorThread
+	 * will inform FileNodeManager to flush.
+	 */
+	public long memMonitorInterval = 1000;  // in ms
+
+	/**
+	 * Decide how to control memory used by inserting data.
+	 * 0 is RecordMemController, which count the size of every record (tuple).
+	 * 1 is JVMMemController, which use JVM heap memory as threshold.
+	 */
+	public int memControllerType = 1;
+
+	/**
+	 * When a bufferwrite's metadata size (in byte) exceed this, the bufferwrite is forced closed.
+	 */
+	public long bufferwriteMetaSizeThreshold = 200 * 1024 * 1024L;
+
+	/**
+	 * When a bufferwrite's file size (in byte) exceed this, the bufferwrite is forced closed.
+	 */
+	public long bufferwriteFileSizeThreshold = 2 * 1024 * 1024 * 1024L;
+
+	/**
+	 * When a overflow's metadata size (in byte) exceed this, the overflow is forced closed.
+	 */
+	public long overflowMetaSizeThreshold = 200 * 1024 * 1024L;
+
+	/**
+	 * When a overflow's file size (in byte) exceed this, the overflow is forced closed.
+	 */
+	public long overflowFileSizeThreshold = 2 * 1024 * 1024 * 1024L;
+
+
+	/*
 	 * The statMonitor's BackLoop period, 5s is enough
 	 */
 	public int backLoopPeriod = 5;
@@ -129,6 +173,7 @@ public class TsfileDBConfig {
 	 * the maximum number of writing instances existing in same time.
 	 */
 
+
 	public TsfileDBConfig() {}
 
 	public void updateDataPath() {

src/main/java/cn/edu/tsinghua/iotdb/conf/TsfileDBDescriptor.java

@@ -7,6 +7,7 @@
 import java.io.InputStream;
 import java.util.Properties;
 
+import cn.edu.tsinghua.iotdb.engine.memcontrol.BasicMemController;
 import org.joda.time.DateTimeZone;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -87,6 +88,25 @@ private void loadProps() {
 			conf.periodTimeForFlush = Long.parseLong(properties.getProperty("period_time_for_flush_in_second", conf.periodTimeForFlush+"").trim());
 			conf.periodTimeForMerge = Long.parseLong(properties.getProperty("period_time_for_merge_in_second", conf.periodTimeForMerge+"").trim());
 
+			conf.memThresholdWarning = (long) (Runtime.getRuntime().maxMemory() * Double.parseDouble(properties.getProperty("mem_threshold_warning", conf.memThresholdWarning+"").trim()) );
+			conf.memThresholdDangerous = (long) (Runtime.getRuntime().maxMemory() * Double.parseDouble(properties.getProperty("mem_threshold_dangerous", conf.memThresholdDangerous+"").trim()));
+
+			conf.memMonitorInterval = Long.parseLong(properties.getProperty("mem_monitor_interval", conf.memMonitorInterval+"").trim());
+
+			conf.memControllerType = Integer.parseInt(properties.getProperty("mem_controller_type", conf.memControllerType+"").trim());
+			conf.memControllerType = conf.memControllerType >= BasicMemController.CONTROLLER_TYPE.values().length ? 0 : conf.memControllerType;
+
+			conf.bufferwriteMetaSizeThreshold = Long.parseLong(properties.getProperty("bufferwrite_meta_size_threshold", conf.bufferwriteMetaSizeThreshold + "").trim());
+			conf.bufferwriteFileSizeThreshold = Long.parseLong(properties.getProperty("bufferwrite_file_size_threshold", conf.bufferwriteFileSizeThreshold + "").trim());
+
+			conf.overflowMetaSizeThreshold = Long.parseLong(properties.getProperty("overflow_meta_size_threshold", conf.overflowMetaSizeThreshold + "").trim());
+			conf.overflowFileSizeThreshold = Long.parseLong(properties.getProperty("overflow_file_size_threshold", conf.overflowFileSizeThreshold + "").trim());
+
+			if(conf.memThresholdWarning <= 0)
+				conf.memThresholdWarning = TsFileDBConstant.MEM_THRESHOLD_WARNING_DEFAULT;
+			if(conf.memThresholdDangerous < conf.memThresholdWarning)
+				conf.memThresholdDangerous = Math.max(conf.memThresholdWarning, TsFileDBConstant.MEM_THRESHOLD_DANGEROUS_DEFAULT);
+
 			conf.concurrentFlushThread  = Integer.parseInt(properties.getProperty("concurrent_flush_thread", conf.concurrentFlushThread + ""));
 			if(conf.concurrentFlushThread <= 0)
 				conf.concurrentFlushThread = Runtime.getRuntime().availableProcessors();

src/main/java/cn/edu/tsinghua/iotdb/engine/Processor.java

@@ -155,6 +155,8 @@ public boolean equals(Object obj) {
 	 */
 	public abstract boolean canBeClosed();
 
+	public abstract void flush() throws IOException;
+
 	/**
 	 * Close the processor.<br>
 	 * Notice: Thread is not safe
@@ -163,4 +165,6 @@ public boolean equals(Object obj) {
 	 * @throws ProcessorException
 	 */
 	public abstract void close() throws ProcessorException;
+	
+	public abstract long memoryUsage();
 }