You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2025/03/06 06:13:46 UTC
(iotdb) branch fix_sync_lag_1,3 created (now 7dd33d483e0)
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a change to branch fix_sync_lag_1,3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 7dd33d483e0 Fix frozen IoTConsensus sync lag (#15023)
This branch includes the following new commits:
new 7dd33d483e0 Fix frozen IoTConsensus sync lag (#15023)
The 1 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) 01/01: Fix frozen IoTConsensus sync lag (#15023)
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch fix_sync_lag_1,3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7dd33d483e0ea4ef059f430f3b4ff9c3f5a168bd
Author: Jiang Tian <jt...@163.com>
AuthorDate: Thu Mar 6 13:03:05 2025 +0800
Fix frozen IoTConsensus sync lag (#15023)
* Fix frozen IoTConsensus sync lag
* spotless
(cherry picked from commit f9f0574b0f7b4a22793b87b43f020bffbb55c511)
---
.../db/storageengine/dataregion/wal/buffer/IWALBuffer.java | 10 ++++++++++
.../db/storageengine/dataregion/wal/buffer/WALBuffer.java | 13 +++++++++++++
.../iotdb/db/storageengine/dataregion/wal/node/WALNode.java | 8 ++++++--
3 files changed, 29 insertions(+), 2 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/IWALBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/IWALBuffer.java
index f3da2cd7143..978ea781162 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/IWALBuffer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/IWALBuffer.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
/**
* This class serializes and flushes {@link WALEntry}. If search is enabled, the order of search
@@ -52,6 +53,15 @@ public interface IWALBuffer extends AutoCloseable {
*/
void waitForFlush() throws InterruptedException;
+ /**
+ * Wait for next flush operation done, if the predicate == true after entering a locked
+ * environment. Otherwise, return directly.
+ *
+ * @param waitPredicate the condition which should be satisfied before waiting.
+ * @throws InterruptedException when interrupted by the flush thread
+ */
+ public void waitForFlush(Predicate<WALBuffer> waitPredicate) throws InterruptedException;
+
/**
* Wait for next flush operation done.
*
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
index 17fbeff0c23..872a116d5c9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
@@ -63,6 +63,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Predicate;
import static org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode.DEFAULT_SEARCH_INDEX;
@@ -653,6 +654,18 @@ public class WALBuffer extends AbstractWALBuffer {
}
}
+ @Override
+ public void waitForFlush(Predicate<WALBuffer> waitPredicate) throws InterruptedException {
+ buffersLock.lock();
+ try {
+ if (waitPredicate.test(this)) {
+ idleBufferReadyCondition.await();
+ }
+ } finally {
+ buffersLock.unlock();
+ }
+ }
+
@Override
public boolean waitForFlush(long time, TimeUnit unit) throws InterruptedException {
buffersLock.lock();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
index d0781f12d98..35432a8fd6f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
@@ -813,20 +813,24 @@ public class WALNode implements IWALNode {
@Override
public void waitForNextReady() throws InterruptedException {
boolean walFileRolled = false;
+ long bufferLastSearchIndex = 0;
while (!hasNext()) {
if (!walFileRolled) {
boolean timeout =
!buffer.waitForFlush(WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC, TimeUnit.SECONDS);
if (timeout) {
+ bufferLastSearchIndex = buffer.getCurrentSearchIndex();
logger.info(
"timeout when waiting for next WAL entry ready, execute rollWALFile. Current search index in wal buffer is {}, and next target index is {}",
- buffer.getCurrentSearchIndex(),
+ bufferLastSearchIndex,
nextSearchIndex);
rollWALFile();
walFileRolled = true;
}
} else {
- buffer.waitForFlush();
+ // only wait when the search index of the buffer remains the same as the previous check
+ long finalBufferLastSearchIndex = bufferLastSearchIndex;
+ buffer.waitForFlush(buf -> buf.getCurrentSearchIndex() == finalBufferLastSearchIndex);
}
}
}