This is an automated email from the ASF dual-hosted git repository.
heiming pushed a commit to branch new_geely_car_0205
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/new_geely_car_0205 by this
push:
new df3e500162 add wal buffer num parameter
df3e500162 is described below
commit df3e500162a06a3e6faa5a196514ab4ee7524e3c
Author: HeimingZ <[email protected]>
AuthorDate: Mon Feb 6 18:35:11 2023 +0800
add wal buffer num parameter
---
.../resources/conf/iotdb-common.properties | 4 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 13 +++-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 8 +++
.../org/apache/iotdb/db/wal/buffer/WALBuffer.java | 78 +++++++++++++---------
4 files changed, 68 insertions(+), 35 deletions(-)
diff --git a/node-commons/src/assembly/resources/conf/iotdb-common.properties
b/node-commons/src/assembly/resources/conf/iotdb-common.properties
index 6d0ab7964b..b266bcc23d 100644
--- a/node-commons/src/assembly/resources/conf/iotdb-common.properties
+++ b/node-commons/src/assembly/resources/conf/iotdb-common.properties
@@ -662,7 +662,9 @@ cluster_name=defaultCluster
# Buffer size of each wal node
# If it's a value smaller than 0, use the default value 16 * 1024 * 1024 bytes
(16MB).
# Datatype: int
-# wal_buffer_size_in_byte=16777216
+# wal_buffer_size_in_byte=8388608
+
+#wal_buffer_num_per_node=3
# Blocking queue capacity of each wal buffer, restricts maximum number of
WALEdits cached in the blocking queue.
# Datatype: int
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 5298dc16f9..008e8c9f14 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
@@ -188,7 +188,10 @@ public class IoTDBConfig {
private volatile long fsyncWalDelayInMs = 3;
/** Buffer size of each wal node. Unit: byte */
- private int walBufferSize = 16 * 1024 * 1024;
+ private int walBufferSize = 8 * 1024 * 1024;
+
+ /** Buffer number of each wal node. Unit: byte */
+ private int walBufferNumPerNode = 3;
/** Buffer entry size of each wal buffer. Unit: byte */
private int walBufferEntrySize = 16 * 1024;
@@ -1656,6 +1659,14 @@ public class IoTDBConfig {
this.walBufferSize = walBufferSize;
}
+ public int getWalBufferNumPerNode() {
+ return walBufferNumPerNode;
+ }
+
+ public void setWalBufferNumPerNode(int walBufferNumPerNode) {
+ this.walBufferNumPerNode = walBufferNumPerNode;
+ }
+
public int getWalBufferEntrySize() {
return walBufferEntrySize;
}
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 128ea8233c..bce85bcc3e 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
@@ -1009,6 +1009,14 @@ public class IoTDBDescriptor {
conf.setWalBufferSize(walBufferSize);
}
+ int walBufferNumPerNode =
+ Integer.parseInt(
+ properties.getProperty(
+ "wal_buffer_num_per_node",
Integer.toString(conf.getWalBufferNumPerNode())));
+ if (walBufferNumPerNode > 0) {
+ conf.setWalBufferNumPerNode(walBufferNumPerNode);
+ }
+
int walBufferEntrySize =
Integer.parseInt(
properties.getProperty(
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
index db2d627ef5..0ac57da334 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/buffer/WALBuffer.java
@@ -43,6 +43,7 @@ import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
@@ -57,7 +58,6 @@ import static
org.apache.iotdb.db.wal.node.WALNode.DEFAULT_SEARCH_INDEX;
public class WALBuffer extends AbstractWALBuffer {
private static final Logger logger =
LoggerFactory.getLogger(WALBuffer.class);
private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
- private static final int HALF_WAL_BUFFER_SIZE = config.getWalBufferSize() /
2;
private static final double FSYNC_BUFFER_RATIO = 0.95;
private static final int QUEUE_CAPACITY = config.getWalBufferQueueCapacity();
@@ -73,10 +73,8 @@ public class WALBuffer extends AbstractWALBuffer {
/** two buffers switch between three statuses (there is always 1 buffer
working) */
// buffer in working status, only updated by serializeThread
private volatile ByteBuffer workingBuffer;
- // buffer in idle status
- private volatile ByteBuffer idleBuffer;
- // buffer in syncing status, serializeThread makes sure no more writes to
syncingBuffer
- private volatile ByteBuffer syncingBuffer;
+ // buffers in idle status
+ private final BlockingQueue<ByteBuffer> idleBuffers = new
LinkedBlockingQueue<>();
// endregion
/** file status of working buffer, updating file writer's status when
syncing */
protected volatile WALFileStatus currentFileStatus;
@@ -107,8 +105,10 @@ public class WALBuffer extends AbstractWALBuffer {
private void allocateBuffers() {
try {
- workingBuffer = ByteBuffer.allocateDirect(HALF_WAL_BUFFER_SIZE);
- idleBuffer = ByteBuffer.allocateDirect(HALF_WAL_BUFFER_SIZE);
+ workingBuffer = ByteBuffer.allocateDirect(config.getWalBufferSize());
+ for (int i = 1; i < config.getWalBufferNumPerNode(); ++i) {
+ idleBuffers.add(ByteBuffer.allocateDirect(config.getWalBufferSize()));
+ }
} catch (OutOfMemoryError e) {
logger.error("Fail to allocate wal node-{}'s buffer because out of
memory.", identifier, e);
close();
@@ -174,7 +174,7 @@ public class WALBuffer extends AbstractWALBuffer {
}
// try to get more WALEntries with blocking interface to enlarge write
batch
- while (totalSize < HALF_WAL_BUFFER_SIZE * FSYNC_BUFFER_RATIO) {
+ while (totalSize < config.getWalBufferSize() * FSYNC_BUFFER_RATIO) {
WALEntry walEntry = null;
try {
// for better fsync performance, wait a while to enlarge write batch
@@ -367,35 +367,37 @@ public class WALBuffer extends AbstractWALBuffer {
/** Notice: this method only called when buffer is exhausted by
SerializeTask. */
private void syncWorkingBuffer(long searchIndex, WALFileStatus fileStatus) {
- switchWorkingBufferToFlushing();
- syncBufferThread.submit(new SyncBufferTask(searchIndex, fileStatus,
false));
+ ByteBuffer buffer = switchWorkingBufferToFlushing();
+ syncBufferThread.submit(new SyncBufferTask(searchIndex, fileStatus, false,
buffer));
currentFileStatus = WALFileStatus.CONTAINS_NONE_SEARCH_INDEX;
}
/** Notice: this method only called at the last of SerializeTask. */
private void fsyncWorkingBuffer(long searchIndex, WALFileStatus fileStatus,
SerializeInfo info) {
- switchWorkingBufferToFlushing();
- syncBufferThread.submit(new SyncBufferTask(searchIndex, fileStatus, true,
info));
+ ByteBuffer buffer = switchWorkingBufferToFlushing();
+ syncBufferThread.submit(new SyncBufferTask(searchIndex, fileStatus, true,
info, buffer));
currentFileStatus = WALFileStatus.CONTAINS_NONE_SEARCH_INDEX;
}
// only called by serializeThread
- private void switchWorkingBufferToFlushing() {
+ private ByteBuffer switchWorkingBufferToFlushing() {
+ ByteBuffer idleBuffer = null;
+ try {
+ idleBuffer = idleBuffers.take();
+ } catch (InterruptedException e) {
+ logger.warn("Interrupted when waiting for next idle buffer.");
+ Thread.currentThread().interrupt();
+ }
+ ByteBuffer syncingBuffer;
buffersLock.lock();
try {
- while (idleBuffer == null) {
- idleBufferReadyCondition.await();
- }
syncingBuffer = workingBuffer;
workingBuffer = idleBuffer;
workingBuffer.clear();
- idleBuffer = null;
- } catch (InterruptedException e) {
- logger.warn("Interrupted When waiting for available working buffer.");
- Thread.currentThread().interrupt();
} finally {
buffersLock.unlock();
}
+ return syncingBuffer;
}
// endregion
@@ -408,17 +410,24 @@ public class WALBuffer extends AbstractWALBuffer {
private final WALFileStatus fileStatus;
private final boolean forceFlag;
private final SerializeInfo info;
+ private final ByteBuffer syncingBuffer;
- public SyncBufferTask(long searchIndex, WALFileStatus fileStatus, boolean
forceFlag) {
- this(searchIndex, fileStatus, forceFlag, null);
+ public SyncBufferTask(
+ long searchIndex, WALFileStatus fileStatus, boolean forceFlag,
ByteBuffer syncingBuffer) {
+ this(searchIndex, fileStatus, forceFlag, null, syncingBuffer);
}
public SyncBufferTask(
- long searchIndex, WALFileStatus fileStatus, boolean forceFlag,
SerializeInfo info) {
+ long searchIndex,
+ WALFileStatus fileStatus,
+ boolean forceFlag,
+ SerializeInfo info,
+ ByteBuffer syncingBuffer) {
this.searchIndex = searchIndex;
this.fileStatus = fileStatus;
this.forceFlag = forceFlag;
this.info = info == null ? new SerializeInfo() : info;
+ this.syncingBuffer = syncingBuffer;
}
@Override
@@ -440,7 +449,7 @@ public class WALBuffer extends AbstractWALBuffer {
"Fail to sync wal node-{}'s buffer, change system mode to error.",
identifier, e);
CommonDescriptor.getInstance().getConfig().handleUnrecoverableError();
} finally {
- switchSyncingBufferToIdle();
+ switchSyncingBufferToIdle(syncingBuffer);
}
boolean forceSuccess = false;
@@ -488,13 +497,12 @@ public class WALBuffer extends AbstractWALBuffer {
}
// only called by syncBufferThread
- private void switchSyncingBufferToIdle() {
+ private void switchSyncingBufferToIdle(ByteBuffer buffer) {
buffersLock.lock();
try {
// No need to judge whether idleBuffer is null because syncingBuffer is
not null
// and there is only one buffer can be null between syncingBuffer and
idleBuffer
- idleBuffer = syncingBuffer;
- syncingBuffer = null;
+ idleBuffers.add(buffer);
idleBufferReadyCondition.signalAll();
} finally {
buffersLock.unlock();
@@ -550,11 +558,13 @@ public class WALBuffer extends AbstractWALBuffer {
if (workingBuffer != null) {
MmapUtil.clean((MappedByteBuffer) workingBuffer);
}
- if (idleBuffer != null) {
- MmapUtil.clean((MappedByteBuffer) workingBuffer);
- }
- if (syncingBuffer != null) {
- MmapUtil.clean((MappedByteBuffer) syncingBuffer);
+ for (int i = 1; i < config.getWalBufferNumPerNode(); ++i) {
+ try {
+ MmapUtil.clean((MappedByteBuffer) idleBuffers.take());
+ } catch (InterruptedException e) {
+ logger.warn("Interrupted when waiting for next idle buffer.");
+ Thread.currentThread().interrupt();
+ }
}
}
@@ -574,7 +584,9 @@ public class WALBuffer extends AbstractWALBuffer {
public boolean isAllWALEntriesConsumed() {
buffersLock.lock();
try {
- return walEntries.isEmpty() && workingBuffer.position() == 0 &&
syncingBuffer == null;
+ return walEntries.isEmpty()
+ && workingBuffer.position() == 0
+ && idleBuffers.size() == config.getWalBufferNumPerNode() - 1;
} finally {
buffersLock.unlock();
}
