optimize: check timeout more efficient

Change-Id: I103c0aad5bcc313cce02e091f7cd11bf4dd0eb32
openmessaging · Aug 9, 2023 · 29f4daa · 29f4daa
1 parent e0969d3
commit 29f4daa
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 12 deletions.
diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java
@@ -216,21 +216,23 @@ public boolean completeResponseFuture(final ApplyEntry task) {
     /**
      * Check responseFutures timeout from {beginIndex} in currentTerm
      */
-    public void checkResponseFuturesTimeout() {
+    public void checkResponseFuturesTimeout(long beginIndex) {
         final long term = this.memberState.currTerm();
+        long maxIndex = this.memberState.getCommittedIndex() + dLedgerConfig.getMaxPendingRequestsNum() + 1;
+        if (maxIndex > this.memberState.getLedgerEndIndex()) {
+            maxIndex = this.memberState.getLedgerEndIndex() + 1;
+        }
         ConcurrentMap<Long, Closure> closureMap = this.pendingClosure.get(term);
         if (closureMap != null) {
-            for (Map.Entry<Long, Closure> futureEntry : closureMap.entrySet()) {
-                if (futureEntry == null) {
-                    continue;
-                }
-                Closure closure = futureEntry.getValue();
+            for (long i = beginIndex; i < maxIndex; i++) {
+                Closure closure = closureMap.get(i);
                 if (closure == null) {
-                    continue;
-                }
-                if (closure.isTimeOut()) {
+                    // index may be removed for complete, we should continue scan
+                } else if (closure.isTimeOut()) {
                     closure.done(Status.error(DLedgerResponseCode.WAIT_QUORUM_ACK_TIMEOUT));
-                    closureMap.remove(futureEntry.getKey());
+                    closureMap.remove(i);
+                } else {
+                    break;
                 }
             }
         }
@@ -306,7 +308,7 @@ public void doWork() {
                 }
                 if (DLedgerUtils.elapsed(lastCheckTimeoutTimeMs) > 1000) {
                     // clear the timeout pending closure should check all since it can timeout for different index
-                    checkResponseFuturesTimeout();
+                    checkResponseFuturesTimeout(DLedgerEntryPusher.this.memberState.getAppliedIndex() + 1);
                     lastCheckTimeoutTimeMs = System.currentTimeMillis();
                 }
                 if (!memberState.isLeader()) {

diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachineCaller.java b/dledger/src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachineCaller.java
@@ -203,7 +203,7 @@ private void doCommitted(final long committedIndex) {
         // Check response timeout.
         if (iter.getCompleteAckNums() == 0) {
             if (this.entryPusher != null) {
-                this.entryPusher.checkResponseFuturesTimeout();
+                this.entryPusher.checkResponseFuturesTimeout(this.memberState.getAppliedIndex() + 1);
             }
         }
     }