You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2021/07/21 03:22:11 UTC
[iotdb] 01/01: clear wal
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch clear_wal
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5f215d344b3577cca529d44808f7200919681045
Author: qiaojialin <64...@qq.com>
AuthorDate: Wed Jul 21 11:19:59 2021 +0800
clear wal
---
.../db/writelog/node/ExclusiveWriteLogNode.java | 75 +++-------------------
1 file changed, 8 insertions(+), 67 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java b/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
index 2d10381..26f8398 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
@@ -105,12 +105,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
if (deleted) {
throw new IOException("WAL node deleted");
}
- long start = System.currentTimeMillis();
lock.lock();
- long elapse = System.currentTimeMillis() - start;
- if (elapse > 5000) {
- logger.warn("[wal] ExclusiveWriteLogNode lock got cost: {}ms", elapse);
- }
try {
putLog(plan);
if (bufferedLogNum >= config.getFlushWalThreshold()) {
@@ -243,9 +238,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
}
private void forceWal() {
- logger.warn("[wal] {} forceWal try to start", this.hashCode());
lock.lock();
- logger.warn("[wal] {} forceWal start", this.hashCode());
try {
try {
if (currentFileWriter != null) {
@@ -256,14 +249,11 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
}
} finally {
lock.unlock();
- logger.warn("[wal] {} forceWal end", this.hashCode());
}
}
private void sync() {
- logger.warn("[wal] {} sync try to start", this.hashCode());
lock.lock();
- logger.warn("[wal] {} sync start", this.hashCode());
try {
if (bufferedLogNum == 0) {
return;
@@ -271,8 +261,6 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
switchBufferWorkingToFlushing();
ILogWriter currWriter = getCurrentFileWriter();
FLUSH_BUFFER_THREAD_POOL.submit(() -> flushBuffer(currWriter));
- switchBufferIdleToWorking();
-
bufferedLogNum = 0;
logger.debug("Log node {} ends sync.", identifier);
} catch (InterruptedException e) {
@@ -282,13 +270,10 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
logger.warn("can not found file {}", identifier, e);
} finally {
lock.unlock();
- logger.warn("[wal] {} sync end", this.hashCode());
}
}
private void flushBuffer(ILogWriter writer) {
- logger.warn("[wal] {} flushBuffer start", this.hashCode());
- long start = System.currentTimeMillis();
try {
writer.write(logBufferFlushing);
} catch (ClosedChannelException e) {
@@ -297,75 +282,31 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
logger.warn("Log node {} sync failed, change system mode to read-only", identifier, e);
IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
return;
- } finally {
- long elapse = System.currentTimeMillis() - start;
- if (elapse > 5000) {
- logger.warn("[wal] flushBuffer cost: {}ms", elapse);
- }
}
- logBufferFlushing.clear();
- try {
- switchBufferFlushingToIdle();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ // switch buffer flushing to idle and notify the sync thread
+ synchronized (switchBufferCondition) {
+ logBufferIdle = logBufferFlushing;
+ logBufferFlushing = null;
+ switchBufferCondition.notifyAll();
}
- logger.warn("[wal] {} flushBuffer end", this.hashCode());
}
private void switchBufferWorkingToFlushing() throws InterruptedException {
- logger.warn("[wal] {} switchBufferWorkingToFlushing start", this.hashCode());
long start = System.currentTimeMillis();
synchronized (switchBufferCondition) {
while (logBufferFlushing != null && !deleted) {
switchBufferCondition.wait(100);
}
logBufferFlushing = logBufferWorking;
- logBufferWorking = null;
- switchBufferCondition.notifyAll();
- }
- long elapse = System.currentTimeMillis() - start;
- if (elapse > 5000) {
- logger.warn("[wal] {} switch Working -> Flushing cost: {}ms", this.hashCode(), elapse);
- }
- logger.warn("[wal] {} switchBufferWorkingToFlushing end", this.hashCode());
- }
-
- private void switchBufferIdleToWorking() throws InterruptedException {
- logger.warn("[wal] {} switchBufferIdleToWorking start", this.hashCode());
- long start = System.currentTimeMillis();
- synchronized (switchBufferCondition) {
- while (logBufferIdle == null && !deleted) {
- switchBufferCondition.wait(100);
- }
logBufferWorking = logBufferIdle;
+ logBufferWorking.clear();
logBufferIdle = null;
- switchBufferCondition.notifyAll();
}
long elapse = System.currentTimeMillis() - start;
- if (elapse > 5000) {
- logger.warn("[wal] {} switch Idle -> Working cost: {}ms", this.hashCode(), elapse);
- }
- logger.warn("[wal] {} switchBufferIdleToWorking end", this.hashCode());
- }
-
- private void switchBufferFlushingToIdle() throws InterruptedException {
- logger.warn("[wal] {} switchBufferFlushingToIdle start", this.hashCode());
- long start = System.currentTimeMillis();
- synchronized (switchBufferCondition) {
- while (logBufferIdle != null && !deleted) {
- switchBufferCondition.wait(100);
- }
- logBufferIdle = logBufferFlushing;
- logBufferIdle.clear();
- logBufferFlushing = null;
- switchBufferCondition.notifyAll();
- }
- long elapse = System.currentTimeMillis() - start;
- if (elapse > 5000) {
- logger.warn("[wal] {} switch Flushing -> Idle cost: {}ms", this.hashCode(), elapse);
+ if (elapse > 3000) {
+ logger.warn("[wal] {} switch Working -> Flushing cost: {}ms", this.hashCode(), elapse);
}
- logger.warn("[wal] {} switchBufferFlushingToIdle end", this.hashCode());
}
private ILogWriter getCurrentFileWriter() throws FileNotFoundException {