This is an automated email from the ASF dual-hosted git repository. xingtanzjr pushed a commit to branch multiLeader_out_of_order in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b76335a36b51393406b9ea8f35700badbbe1769f Author: stormbroken <[email protected]> AuthorDate: Tue Jul 26 11:13:53 2022 +0800 add throttle of statemachine. --- .../consensus/multileader/MultiLeaderServerImpl.java | 19 +++++++++++++++++++ .../multileader/logdispatcher/LogDispatcher.java | 17 +++++++++++++---- 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java index 5573f6ba9a..0ff7532cc4 100644 --- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java +++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java @@ -53,6 +53,9 @@ public class MultiLeaderServerImpl { private static final String CONFIGURATION_FILE_NAME = "configuration.dat"; private final Logger logger = LoggerFactory.getLogger(MultiLeaderServerImpl.class); + private static final int THROTTLE_UPPER_BOUND = 1000; + private static final int THROTTLE_LOWER_BOUND = 300; + private static final long TIME_OUT = 60 * 1000L; private final Peer thisNode; private final IStateMachine stateMachine; @@ -110,6 +113,14 @@ public class MultiLeaderServerImpl { */ public TSStatus write(IConsensusRequest request) { synchronized (stateMachine) { + if (needToThrottleDown()) { + logger.info("[Throttle Down] index:{}, safeIndex:{}", getIndex(), getCurrentSafelyDeletedSearchIndex()); + try { + stateMachine.wait(TIME_OUT); + } catch (InterruptedException e) { + logger.error("Failed to wait", e); + } + } IndexedConsensusRequest indexedConsensusRequest = buildIndexedConsensusRequestForLocalRequest(request); if (indexedConsensusRequest.getSearchIndex() % 1000 == 0) { @@ -218,4 +229,12 @@ public class MultiLeaderServerImpl { public MultiLeaderConfig getConfig() { return config; } + + public boolean needToThrottleDown() { + return getIndex() - getCurrentSafelyDeletedSearchIndex() > THROTTLE_UPPER_BOUND; + } + + public boolean needToThrottleUp() { + return getIndex() - getCurrentSafelyDeletedSearchIndex() < THROTTLE_LOWER_BOUND; + } } 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 71ce7caa45..9307827ee9 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 @@ -192,10 +192,13 @@ public class LogDispatcher { pendingRequest.poll(PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC, TimeUnit.SECONDS); if (request != null) { bufferedRequest.add(request); - // If write pressure is low, we simply sleep a little to reduce the number of RPC - if (pendingRequest.size() <= config.getReplication().getMaxRequestPerBatch()) { - Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs()); - } + // If write pressure is low, we simply sleep a little to reduce the + // number of RPC + // if (pendingRequest.size() <= + // config.getReplication().getMaxRequestPerBatch()) { + // + // Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs()); + // } } } // we may block here if the synchronization pipeline is full @@ -216,6 +219,12 @@ public class LogDispatcher { // indicating that insert nodes whose search index are before this value can be deleted // safely reader.setSafelyDeletedSearchIndex(impl.getCurrentSafelyDeletedSearchIndex()); + // notify + if (impl.needToThrottleUp()) { + synchronized (impl.getStateMachine()) { + impl.getStateMachine().notifyAll(); + } + } } public PendingBatch getBatch() {
