You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2025/04/02 10:38:10 UTC
(iotdb) branch random-opt-pipe created (now bb9895fc8bf)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch random-opt-pipe
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at bb9895fc8bf Pipe: Set thread name for pipe receiver
This branch includes the following new commits:
new 919ed99abef Pipe: Optimized insert node cache hit possibility
new bb9895fc8bf Pipe: Set thread name for pipe receiver
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 02/02: Pipe: Set thread name for pipe receiver
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch random-opt-pipe
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit bb9895fc8bfa9e8ad13adf8d4ecafc03e4dc4b95
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Apr 2 18:37:56 2025 +0800
Pipe: Set thread name for pipe receiver
---
.../org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
index a733fcde0a6..a8c65a93808 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
@@ -115,6 +115,10 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver {
}
receiverId.set(RECEIVER_ID_GENERATOR.incrementAndGet());
+ Thread.currentThread()
+ .setName(
+ String.format(
+ "Pipe-Receiver-%s-%s:%s", receiverId.get(), getSenderHost(), getSenderPort()));
// Clear the original receiver file dir if exists
if (receiverFileDirWithIdSuffix.get() != null) {
(iotdb) 01/02: Pipe: Optimized insert node cache hit possibility
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch random-opt-pipe
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 919ed99abef8be34b5d00eab2d35fe72d2c3aeda
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Apr 2 18:09:51 2025 +0800
Pipe: Optimized insert node cache hit possibility
---
.../db/storageengine/dataregion/wal/utils/WALEntryHandler.java | 10 ++++++++--
.../storageengine/dataregion/wal/utils/WALEntryPosition.java | 4 ++--
.../storageengine/dataregion/wal/utils/WALInsertNodeCache.java | 6 ++++++
3 files changed, 16 insertions(+), 4 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java
index 333842e38ef..f5d7406f5a6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.exception.MemTablePinExc
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
+import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -90,8 +91,13 @@ public class WALEntryHandler {
public InsertNode getInsertNodeViaCacheIfPossible() {
try {
final WALEntryValue finalValue = value;
- return finalValue instanceof InsertNode ? (InsertNode) finalValue : null;
- } catch (Exception e) {
+ if (finalValue instanceof InsertNode) {
+ return (InsertNode) finalValue;
+ }
+ final Pair<ByteBuffer, InsertNode> byteBufferInsertNodePair =
+ walEntryPosition.getByteBufferOrInsertNodeIfPossible();
+ return byteBufferInsertNodePair == null ? null : byteBufferInsertNodePair.getRight();
+ } catch (final Exception e) {
logger.warn("Fail to get insert node via cache. {}", this, e);
throw e;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java
index 438e6a897e2..1370745cea2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java
@@ -64,8 +64,8 @@ public class WALEntryPosition {
* Try to read the wal entry directly from the cache. No need to check if the wal entry is ready
* for read.
*/
- public Pair<ByteBuffer, InsertNode> readByteBufferOrInsertNodeViaCacheDirectly() {
- return cache.getByteBufferOrInsertNode(this);
+ public Pair<ByteBuffer, InsertNode> getByteBufferOrInsertNodeIfPossible() {
+ return cache.getByteBufferOrInsertNodeIfPossible(this);
}
/**
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java
index 21571c1ed98..0823c7e7b6e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java
@@ -229,6 +229,12 @@ public class WALInsertNodeCache {
return pair;
}
+ public Pair<ByteBuffer, InsertNode> getByteBufferOrInsertNodeIfPossible(
+ final WALEntryPosition position) {
+ hasPipeRunning = true;
+ return lruCache.getIfPresent(position);
+ }
+
public void cacheInsertNodeIfNeeded(
final WALEntryPosition walEntryPosition, final InsertNode insertNode) {
// reduce memory usage