This is an automated email from the ASF dual-hosted git repository.
qiaojialin 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 d951e82 [Mem control] Move one of the insert check out from the sg
lock (#2154)
d951e82 is described below
commit d951e82f9509b3772ff9a9fe3bf1444f406d4afc
Author: Haonan <[email protected]>
AuthorDate: Thu Dec 3 16:49:13 2020 +0800
[Mem control] Move one of the insert check out from the sg lock (#2154)
---
.../resources/conf/iotdb-engine.properties | 8 ++--
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 14 +++---
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 6 +--
.../org/apache/iotdb/db/engine/StorageEngine.java | 20 +++++++-
.../engine/storagegroup/StorageGroupProcessor.java | 12 +++++
.../db/engine/storagegroup/TsFileProcessor.java | 53 ++++++----------------
6 files changed, 58 insertions(+), 55 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties
b/server/src/assembly/resources/conf/iotdb-engine.properties
index b9b698a..5c7851d 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -245,11 +245,11 @@ storage_group_report_threshold=16777216
# it's just an advised value, the real limitation will be the smaller one
between this and the one we calculated
max_deduplicated_path_num=1000
-# When an inserting is rejected, waiting time (in ms) to check system again, 0
by default.
-waiting_time_when_insert_blocked=0
+# When an inserting is rejected, waiting period (in ms) to check system again,
50 by default.
+check_period_when_insert_blocked=50
-# When the waiting time (in ms) of an inserting exceeds this, throw an
exception. 0 by default.
-max_waiting_time_when_insert_blocked=0
+# When the waiting time (in ms) of an inserting exceeds this, throw an
exception. 10000 by default.
+max_waiting_time_when_insert_blocked=10000
# estimated metadata size (in byte) of one timeseries in Mtree
estimated_series_size=300
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index a1a922b..aebe4a8 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -162,14 +162,14 @@ public class IoTDBConfig {
private long storageGroupSizeReportThreshold = 16 * 1024 * 1024L;
/**
- * When inserting rejected, waiting this time to check system again
+ * When inserting rejected, waiting period to check system again
*/
- private int waitingTimeWhenInsertBlockedInMs = 0;
+ private int checkPeriodWhenInsertBlocked = 50;
/**
* When inserting rejected exceeds this, throw an exception
*/
- private int maxWaitingTimeWhenInsertBlockedInMs = 0;
+ private int maxWaitingTimeWhenInsertBlockedInMs = 10000;
/**
* Is the write ahead log enable.
*/
@@ -2014,12 +2014,12 @@ public class IoTDBConfig {
this.maxQueryDeduplicatedPathNum = maxQueryDeduplicatedPathNum;
}
- public int getWaitingTimeWhenInsertBlocked() {
- return waitingTimeWhenInsertBlockedInMs;
+ public int getCheckPeriodWhenInsertBlocked() {
+ return checkPeriodWhenInsertBlocked;
}
- public void setWaitingTimeWhenInsertBlocked(int
waitingTimeWhenInsertBlocked) {
- this.waitingTimeWhenInsertBlockedInMs = waitingTimeWhenInsertBlocked;
+ public void setCheckPeriodWhenInsertBlocked(int
checkPeriodWhenInsertBlocked) {
+ this.checkPeriodWhenInsertBlocked = checkPeriodWhenInsertBlocked;
}
public int getMaxWaitingTimeWhenInsertBlocked() {
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 570272a..eea0d35 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -264,9 +264,9 @@ public class IoTDBDescriptor {
.getProperty("avg_series_point_number_threshold",
Integer.toString(conf.getAvgSeriesPointNumberThreshold()))));
- conf.setWaitingTimeWhenInsertBlocked(Integer.parseInt(properties
- .getProperty("waiting_time_when_insert_blocked",
- Integer.toString(conf.getWaitingTimeWhenInsertBlocked()))));
+ conf.setCheckPeriodWhenInsertBlocked(Integer.parseInt(properties
+ .getProperty("check_period_when_insert_blocked",
+ Integer.toString(conf.getCheckPeriodWhenInsertBlocked()))));
conf.setMaxWaitingTimeWhenInsertBlocked(Integer.parseInt(properties
.getProperty("max_waiting_time_when_insert_blocked",
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 ddb0895..5540fbb 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
@@ -67,13 +67,13 @@ import
org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
-import org.apache.iotdb.db.monitor.MonitorConstants;
import org.apache.iotdb.db.monitor.StatMonitor;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryFileManager;
+import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.service.ServiceType;
@@ -839,4 +839,22 @@ public class StorageEngine implements IService {
public void mergeUnLock(List<StorageGroupProcessor> list) {
list.forEach(storageGroupProcessor ->
storageGroupProcessor.getTsFileManagement().readUnLock());
}
+
+ /**
+ * block insertion if the insertion is rejected by memory control
+ */
+ public static void blockInsertionIfReject() throws WriteProcessException {
+ long startTime = System.currentTimeMillis();
+ while (SystemInfo.getInstance().isRejected()) {
+ try {
+ TimeUnit.MILLISECONDS.sleep(config.getCheckPeriodWhenInsertBlocked());
+ if (System.currentTimeMillis() - startTime >
config.getMaxWaitingTimeWhenInsertBlocked()) {
+ throw new WriteProcessException("System rejected over " +
config.getMaxWaitingTimeWhenInsertBlocked() +
+ "ms");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 69aec66..2911fa6 100755
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -676,6 +676,9 @@ public class StorageGroupProcessor {
if (!isAlive(insertRowPlan.getTime())) {
throw new OutOfTTLException(insertRowPlan.getTime(),
(System.currentTimeMillis() - dataTTL));
}
+ if (enableMemControl) {
+ StorageEngine.blockInsertionIfReject();
+ }
writeLock();
try {
// init map
@@ -710,6 +713,15 @@ public class StorageGroupProcessor {
*/
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
public void insertTablet(InsertTabletPlan insertTabletPlan) throws
BatchInsertionException {
+ if (enableMemControl) {
+ try {
+ StorageEngine.blockInsertionIfReject();
+ } catch (WriteProcessException e) {
+ TSStatus[] results = new TSStatus[insertTabletPlan.getRowCount()];
+ Arrays.fill(results,
RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR));
+ throw new BatchInsertionException(results);
+ }
+ }
writeLock();
try {
TSStatus[] results = new TSStatus[insertTabletPlan.getRowCount()];
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 4ed4d31..c18ec90 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
@@ -27,7 +27,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedDeque;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -35,6 +34,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.adapter.CompressionRatio;
+import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.flush.CloseFileListener;
import org.apache.iotdb.db.engine.flush.FlushListener;
import org.apache.iotdb.db.engine.flush.FlushManager;
@@ -83,8 +83,6 @@ public class TsFileProcessor {
private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private final boolean enableMemControl = config.isEnableMemControl();
- private final int waitingTimeWhenInsertBlocked =
config.getWaitingTimeWhenInsertBlocked();
- private final int maxWaitingTimeWhenInsertBlocked =
config.getMaxWaitingTimeWhenInsertBlocked();
private StorageGroupInfo storageGroupInfo;
private TsFileProcessorInfo tsFileProcessorInfo;
@@ -177,7 +175,6 @@ public class TsFileProcessor {
workMemTable = new PrimitiveMemTable(enableMemControl);
}
if (enableMemControl) {
- blockInsertionIfReject();
checkMemCostAndAddToTspInfo(insertRowPlan);
}
@@ -219,12 +216,11 @@ public class TsFileProcessor {
if (workMemTable == null) {
workMemTable = new PrimitiveMemTable(enableMemControl);
}
- if (enableMemControl) {
- blockInsertionIfReject();
- checkMemCostAndAddToTspInfo(insertTabletPlan, start, end);
- }
try {
+ if (enableMemControl) {
+ checkMemCostAndAddToTspInfo(insertTabletPlan, start, end);
+ }
workMemTable.insertTablet(insertTabletPlan, start, end);
if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
insertTabletPlan.setStart(start);
@@ -285,22 +281,8 @@ public class TsFileProcessor {
textDataIncrement += MemUtils.getBinarySize((Binary)
insertRowPlan.getValues()[i]);
}
}
- memTableIncrement += textDataIncrement;
- storageGroupInfo.addStorageGroupMemCost(memTableIncrement);
- tsFileProcessorInfo.addTSPMemCost(unsealedResourceIncrement +
chunkMetadataIncrement);
- if (storageGroupInfo.needToReportToSystem()) {
- SystemInfo.getInstance().reportStorageGroupStatus(storageGroupInfo);
- try {
- blockInsertionIfReject();
- } catch (WriteProcessException e) {
- storageGroupInfo.releaseStorageGroupMemCost(memTableIncrement);
- tsFileProcessorInfo.releaseTSPMemCost(unsealedResourceIncrement +
chunkMetadataIncrement);
- SystemInfo.getInstance().resetStorageGroupStatus(storageGroupInfo,
false);
- throw e;
- }
- }
- workMemTable.addTVListRamCost(memTableIncrement);
- workMemTable.addTextDataSize(textDataIncrement);
+ updateMemoryInfo(memTableIncrement, unsealedResourceIncrement,
+ chunkMetadataIncrement, textDataIncrement);
}
private void checkMemCostAndAddToTspInfo(InsertTabletPlan insertTabletPlan,
int start, int end)
@@ -349,13 +331,19 @@ public class TsFileProcessor {
textDataIncrement += MemUtils.getBinaryColumnSize(column, start, end);
}
}
+ updateMemoryInfo(memTableIncrement, unsealedResourceIncrement,
+ chunkMetadataIncrement, textDataIncrement);
+ }
+
+ private void updateMemoryInfo(long memTableIncrement, long
unsealedResourceIncrement,
+ long chunkMetadataIncrement, long textDataIncrement) throws
WriteProcessException {
memTableIncrement += textDataIncrement;
storageGroupInfo.addStorageGroupMemCost(memTableIncrement);
tsFileProcessorInfo.addTSPMemCost(unsealedResourceIncrement +
chunkMetadataIncrement);
if (storageGroupInfo.needToReportToSystem()) {
SystemInfo.getInstance().reportStorageGroupStatus(storageGroupInfo);
try {
- blockInsertionIfReject();
+ StorageEngine.blockInsertionIfReject();
} catch (WriteProcessException e) {
storageGroupInfo.releaseStorageGroupMemCost(memTableIncrement);
tsFileProcessorInfo.releaseTSPMemCost(unsealedResourceIncrement +
chunkMetadataIncrement);
@@ -367,21 +355,6 @@ public class TsFileProcessor {
workMemTable.addTextDataSize(textDataIncrement);
}
- private void blockInsertionIfReject() throws WriteProcessException {
- long startTime = System.currentTimeMillis();
- while (SystemInfo.getInstance().isRejected()) {
- try {
- TimeUnit.MILLISECONDS.sleep(waitingTimeWhenInsertBlocked);
- if (System.currentTimeMillis() - startTime >
maxWaitingTimeWhenInsertBlocked) {
- throw new WriteProcessException("System rejected over " +
maxWaitingTimeWhenInsertBlocked + "ms");
- }
- } catch (InterruptedException e) {
- logger.error("Failed when waiting for getting memory for insertion ",
e);
- Thread.currentThread().interrupt();
- }
- }
- }
-
/**
* Delete data which belongs to the timeseries `deviceId.measurementId` and
the timestamp of which
* <= 'timestamp' in the deletion. <br/>
