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 d432620 Make the Wal buffer pool parameters to be configurable (#2745)
d432620 is described below
commit d43262010aca8d672e9374b5580a2ca307120946
Author: Jackie Tien <[email protected]>
AuthorDate: Mon Mar 1 20:40:50 2021 +0800
Make the Wal buffer pool parameters to be configurable (#2745)
---
.../resources/conf/iotdb-engine.properties | 10 +++++++
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 20 ++++++++++++++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 31 +++++++++++++++++-----
.../engine/storagegroup/StorageGroupProcessor.java | 18 +++++++------
4 files changed, 64 insertions(+), 15 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties
b/server/src/assembly/resources/conf/iotdb-engine.properties
index 81e209e..1a0a00b 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -414,6 +414,16 @@ enable_stat_monitor=false
enable_monitor_series_write=false
####################
+### WAL Direct Buffer Pool Configuration
+####################
+# the interval to trim the wal pool
+wal_pool_trim_interval_ms=10000
+
+# the max number of wal bytebuffer can be allocated for each time partition,
if there is no unseq data you can set it to 4.
+# it should be an even number
+max_wal_bytebuffer_num_for_each_partition=6
+
+####################
### External sort Configuration
####################
# Is external sort enable
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 0c98992..00c2e8a 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
@@ -167,6 +167,10 @@ public class IoTDBConfig {
*/
private int walBufferSize = 16 * 1024 * 1024;
+ private int maxWalBytebufferNumForEachPartition = 6;
+
+ private long walPoolTrimIntervalInMS = 10_000;
+
private int estimatedSeriesSize = 300;
/**
@@ -1144,6 +1148,22 @@ public class IoTDBConfig {
this.walBufferSize = walBufferSize;
}
+ public int getMaxWalBytebufferNumForEachPartition() {
+ return maxWalBytebufferNumForEachPartition;
+ }
+
+ public void setMaxWalBytebufferNumForEachPartition(int
maxWalBytebufferNumForEachPartition) {
+ this.maxWalBytebufferNumForEachPartition =
maxWalBytebufferNumForEachPartition;
+ }
+
+ public long getWalPoolTrimIntervalInMS() {
+ return walPoolTrimIntervalInMS;
+ }
+
+ public void setWalPoolTrimIntervalInMS(long walPoolTrimIntervalInMS) {
+ this.walPoolTrimIntervalInMS = walPoolTrimIntervalInMS;
+ }
+
public int getEstimatedSeriesSize() {
return estimatedSeriesSize;
}
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 f26a1b5..b5bb472 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
@@ -260,13 +260,6 @@ public class IoTDBDescriptor {
conf.setWalDir(properties.getProperty("wal_dir", conf.getWalDir()));
- int walBufferSize =
- Integer.parseInt(
- properties.getProperty("wal_buffer_size",
Integer.toString(conf.getWalBufferSize())));
- if (walBufferSize > 0) {
- conf.setWalBufferSize(walBufferSize);
- }
-
int mlogBufferSize =
Integer.parseInt(
properties.getProperty(
@@ -819,6 +812,30 @@ public class IoTDBDescriptor {
properties.getProperty(
"enable_discard_out_of_order_data",
Boolean.toString(conf.isEnableDiscardOutOfOrderData()))));
+
+ int walBufferSize =
+ Integer.parseInt(
+ properties.getProperty("wal_buffer_size",
Integer.toString(conf.getWalBufferSize())));
+ if (walBufferSize > 0) {
+ conf.setWalBufferSize(walBufferSize);
+ }
+
+ int maxWalBytebufferNumForEachPartition =
+ Integer.parseInt(
+ properties.getProperty(
+ "max_wal_bytebuffer_num_for_each_partition",
+
Integer.toString(conf.getMaxWalBytebufferNumForEachPartition())));
+ if (maxWalBytebufferNumForEachPartition > 0) {
+
conf.setMaxWalBytebufferNumForEachPartition(maxWalBytebufferNumForEachPartition);
+ }
+
+ long poolTrimIntervalInMS =
+ Integer.parseInt(
+ properties.getProperty(
+ "wal_pool_trim_interval_ms",
Long.toString(conf.getWalPoolTrimIntervalInMS())));
+ if (poolTrimIntervalInMS > 0) {
+ conf.setWalPoolTrimIntervalInMS(poolTrimIntervalInMS);
+ }
}
private void loadAutoCreateSchemaProps(Properties properties) {
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 61b692c..c08d616 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
@@ -254,11 +254,6 @@ public class StorageGroupProcessor {
private static final int WAL_BUFFER_SIZE =
IoTDBDescriptor.getInstance().getConfig().getWalBufferSize() / 2;
- private static final int MAX_WAL_BYTEBUFFER_NUM =
-
IoTDBDescriptor.getInstance().getConfig().getConcurrentWritingTimePartition() *
4;
-
- private static final long DEFAULT_POOL_TRIM_INTERVAL_MILLIS = 10_000;
-
private final Deque<ByteBuffer> walByteBufferPool = new LinkedList<>();
private int currentWalPoolSize = 0;
@@ -277,6 +272,10 @@ public class StorageGroupProcessor {
public ByteBuffer[] getWalDirectByteBuffer() {
ByteBuffer[] res = new ByteBuffer[2];
synchronized (walByteBufferPool) {
+ long startTime = System.nanoTime();
+ int MAX_WAL_BYTEBUFFER_NUM =
+ config.getConcurrentWritingTimePartition()
+ * config.getMaxWalBytebufferNumForEachPartition();
while (walByteBufferPool.isEmpty() && currentWalPoolSize + 2 >
MAX_WAL_BYTEBUFFER_NUM) {
try {
walByteBufferPool.wait();
@@ -288,6 +287,9 @@ public class StorageGroupProcessor {
virtualStorageGroupId,
e);
}
+ logger.info(
+ "Waiting {} ms for wal direct byte buffer.",
+ (System.nanoTime() - startTime) / 1_000_000);
}
// If the queue is not empty, it must have at least two.
if (!walByteBufferPool.isEmpty()) {
@@ -337,7 +339,7 @@ public class StorageGroupProcessor {
// we will trim the size to expectedSize until the pool is empty
while (expectedSize < currentWalPoolSize
&& !walByteBufferPool.isEmpty()
- && poolNotEmptyIntervalInMS >= DEFAULT_POOL_TRIM_INTERVAL_MILLIS) {
+ && poolNotEmptyIntervalInMS >= config.getWalPoolTrimIntervalInMS()) {
MmapUtil.clean((MappedByteBuffer) walByteBufferPool.removeLast());
MmapUtil.clean((MappedByteBuffer) walByteBufferPool.removeLast());
currentWalPoolSize -= 2;
@@ -380,8 +382,8 @@ public class StorageGroupProcessor {
ScheduledExecutorService executorService =
Executors.newSingleThreadScheduledExecutor();
executorService.scheduleWithFixedDelay(
this::trimTask,
- DEFAULT_POOL_TRIM_INTERVAL_MILLIS,
- DEFAULT_POOL_TRIM_INTERVAL_MILLIS,
+ config.getWalPoolTrimIntervalInMS(),
+ config.getWalPoolTrimIntervalInMS(),
TimeUnit.MILLISECONDS);
recover();
}
