repair full gc

ctripcorp · Sep 20, 2024 · 556e38c · 556e38c
1 parent b42baac
commit 556e38c
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 10 deletions.
diff --git a/...is-keeper/src/main/java/com/ctrip/xpipe/redis/keeper/applier/command/SequenceCommand.java b/...is-keeper/src/main/java/com/ctrip/xpipe/redis/keeper/applier/command/SequenceCommand.java
@@ -11,7 +11,6 @@
 
 /**
  * @author Slight
- *
  * Make sure to execute this command on stateThread.
  * Jan 31, 2022 1:06 PM
  */
@@ -44,7 +43,7 @@ public SequenceCommand(Collection<SequenceCommand<?>> pasts, Command<V> inner, E
 
     private void executeSelf() {
         CommandFuture<V> future = inner.execute(workerThreads);
-        future.addListener((f)->{
+        future.addListener((f) -> {
             if (f.isSuccess()) {
                 stateThread.execute(() -> {
                     try {
@@ -65,10 +64,13 @@ private void executeSelf() {
     }
 
     private void nextAfter(Collection<SequenceCommand<?>> pasts) {
-        for (SequenceCommand<?> past : pasts) {
-            past.future().addListener((f)->{
+        final int size = pasts.size();
+        List<SequenceCommand> temp = Lists.newArrayList(pasts);
+        for (SequenceCommand<?> past : temp) {
+            past.future().addListener((f) -> {
                 if (f.isSuccess()) {
-                    if (++complete == pasts.size()) {
+                    pasts.remove(past);
+                    if (++complete == size) {
                         executeSelf();
                     }
                 } else {

diff --git a/...rc/main/java/com/ctrip/xpipe/redis/keeper/applier/sequence/DefaultSequenceController.java b/...rc/main/java/com/ctrip/xpipe/redis/keeper/applier/sequence/DefaultSequenceController.java
@@ -154,8 +154,8 @@ private void submitNoneKeyCommand(RedisOpCommand<?> command, long commandOffset)
 
         /* do some stuff when finish */
 
-        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
         increaseOffsetWhenSuccess(current, commandOffset);
+        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
 
         /* run self */
 
@@ -189,8 +189,8 @@ private void submitSingleKeyCommand(RedisOpDataCommand<?> command, long commandO
 
         /* do some stuff when finish */
 
-        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
         increaseOffsetWhenSuccess(current, commandOffset);
+        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
 
         /* run self */
 
@@ -212,8 +212,8 @@ private void submitMultiKeyCommand(RedisOpDataCommand<?> command, long commandOf
             forgetWhenDone(current, key);
         }
 
-        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
         increaseOffsetWhenSuccess(current, commandOffset);
+        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
 
         current.execute();
     }
@@ -240,8 +240,8 @@ private void submitObstacle(RedisOpCommand<?> command, long commandOffset) {
 
         /* do some stuff when finish */
 
-        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
         increaseOffsetWhenSuccess(current, commandOffset);
+        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
 
         /* run self */
 
@@ -266,7 +266,7 @@ private void forgetWhenDone(SequenceCommand<?> sequenceCommand, RedisKey key) {
 
 
     private void releaseMemoryThresholdWhenDone(SequenceCommand<?> sequenceCommand, long memory) {
-        sequenceCommand.future().addListener((f)->{
+        sequenceCommand.future().addListener((f) -> {
             concurrencyThreshold.release();
             memoryThreshold.release(memory);
         });

diff --git a/...per/src/main/java/com/ctrip/xpipe/redis/keeper/applier/sync/DefaultCommandDispatcher.java b/...per/src/main/java/com/ctrip/xpipe/redis/keeper/applier/sync/DefaultCommandDispatcher.java
@@ -128,6 +128,7 @@ public void onRdbData(ByteBuf rdbData) {
 
     @Override
     public void endReadRdb(EofType eofType, GtidSet emptyGtidSet, long rdbOffset) {
+        logger.info("[endReadRdb] eofType={}", eofType);
     }
 
     @Override