This is an automated email from the ASF dual-hosted git repository. xingtanzjr pushed a commit to branch test_wal_sync in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 4ccbd0387517c65c90323f601578f462d0d4059d Author: Jinrui.Zhang <[email protected]> AuthorDate: Mon Jun 27 21:14:21 2022 +0800 add time consumed metrics for wal construction --- .../multileader/logdispatcher/LogDispatcher.java | 46 +++++++++++++--------- 1 file changed, 28 insertions(+), 18 deletions(-) diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java index 9d592d004e..bfca8c4a8a 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java @@ -287,27 +287,37 @@ public class LogDispatcher { private long constructBatchFromWAL( long currentIndex, long maxIndex, List<TLogBatch> logBatches) { - if (iteratorIndex != currentIndex) { - walEntryiterator.skipTo(currentIndex); - iteratorIndex = currentIndex; - } - - while (currentIndex < maxIndex - && logBatches.size() < config.getReplication().getMaxRequestPerBatch()) { - try { - walEntryiterator.waitForNextReady(); - } catch (InterruptedException e) { - e.printStackTrace(); + long startTime = System.nanoTime(); + int count = 0; + try { + if (iteratorIndex != currentIndex) { + walEntryiterator.skipTo(currentIndex); + iteratorIndex = currentIndex; } - // TODO iterator - IConsensusRequest data = walEntryiterator.next(); - iteratorIndex++; - currentIndex++; - if (data != null) { - logBatches.add(new TLogBatch(data.serializeToByteBuffer())); + while (currentIndex < maxIndex + && logBatches.size() < config.getReplication().getMaxRequestPerBatch()) { + try { + walEntryiterator.waitForNextReady(); + } catch (InterruptedException e) { + e.printStackTrace(); + } + // TODO iterator + IConsensusRequest data = walEntryiterator.next(); + iteratorIndex++; + currentIndex++; + if (data != null) { + logBatches.add(new TLogBatch(data.serializeToByteBuffer())); + count++; + } } + return currentIndex - 1; + } finally { + double timeConsumed = (System.nanoTime() * 1.0 - startTime) / 1000_000; + logger.info( + String.format( + "[DataRegion[%s]->%s]construct batch[%d] time consumed: %.3f", + peer.getGroupId().getId(), peer.getEndpoint().ip, count, timeConsumed)); } - return currentIndex - 1; } private void constructBatchIndexedFromConsensusRequest(
