refactor(framework): decouple Manager from TronJsonRpcImpl

framework/src/main/java/org/tron/common/application/ApplicationImpl.java

@@ -10,6 +10,7 @@
 import org.tron.core.db.Manager;
 import org.tron.core.net.TronNetService;
 import org.tron.core.services.event.EventService;
+import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
 import org.tron.program.SolidityNode;
 
 @Slf4j(topic = "app")
@@ -37,6 +38,9 @@ public class ApplicationImpl implements Application {
   @Autowired(required = false)
   private SolidityNode solidityNode;
 
+  @Autowired
+  private TronJsonRpcImpl tronJsonRpc;
+
   private final CountDownLatch shutdown = new CountDownLatch(1);
 
   /**
@@ -62,6 +66,9 @@ public void shutdown() {
     if (solidityNode != null) {
       solidityNode.close();
     }
+    // Stop the json-rpc filter consumer explicitly, once the producers above are stopped,
+    // rather than in Spring's destroy phase.
+    tronJsonRpc.close();
     dbManager.close();
     shutdown.countDown();
   }

framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java

@@ -0,0 +1,32 @@
+package org.tron.common.logsfilter.queue;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+import org.springframework.stereotype.Component;
+import org.tron.common.logsfilter.capsule.FilterTriggerCapsule;
+
+/**
+ * Queue between the block-processing producer (Manager) and the json-rpc filter
+ * consumer (TronJsonRpcImpl), so that neither side references the other.
+ */
+@Component
+public class FilterCapsuleQueue {
+
+  private final BlockingQueue<FilterTriggerCapsule> queue = new LinkedBlockingQueue<>();
+
+  public boolean offer(FilterTriggerCapsule capsule) {
+    return queue.offer(capsule);
+  }
+
+  public FilterTriggerCapsule poll(long timeout, TimeUnit unit) throws InterruptedException {
+    return queue.poll(timeout, unit);
+  }
+
+  @VisibleForTesting
+  public Stream<FilterTriggerCapsule> stream() {
+    return queue.stream();
+  }
+}

framework/src/main/java/org/tron/core/db/Manager.java

@@ -48,7 +48,6 @@
 import org.apache.commons.collections4.CollectionUtils;
 import org.bouncycastle.util.encoders.Hex;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
 import org.tron.api.GrpcAPI;
 import org.tron.api.GrpcAPI.TransactionInfoList;
@@ -62,11 +61,11 @@
 import org.tron.common.logsfilter.capsule.BlockFilterCapsule;
 import org.tron.common.logsfilter.capsule.BlockLogTriggerCapsule;
 import org.tron.common.logsfilter.capsule.ContractTriggerCapsule;
-import org.tron.common.logsfilter.capsule.FilterTriggerCapsule;
 import org.tron.common.logsfilter.capsule.LogsFilterCapsule;
 import org.tron.common.logsfilter.capsule.SolidityTriggerCapsule;
 import org.tron.common.logsfilter.capsule.TransactionLogTriggerCapsule;
 import org.tron.common.logsfilter.capsule.TriggerCapsule;
+import org.tron.common.logsfilter.queue.FilterCapsuleQueue;
 import org.tron.common.logsfilter.trigger.ContractEventTrigger;
 import org.tron.common.logsfilter.trigger.ContractLogTrigger;
 import org.tron.common.logsfilter.trigger.ContractTrigger;
@@ -143,7 +142,6 @@
 import org.tron.core.service.MortgageService;
 import org.tron.core.service.RewardViCalService;
 import org.tron.core.services.event.exception.EventException;
-import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
 import org.tron.core.store.AccountAssetStore;
 import org.tron.core.store.AccountIdIndexStore;
 import org.tron.core.store.AccountIndexStore;
@@ -253,8 +251,8 @@ public class Manager {
   @Getter
   private BlockingQueue<TriggerCapsule> triggerCapsuleQueue;
   // log filter
-  private boolean isRunFilterProcessThread = true;
-  private BlockingQueue<FilterTriggerCapsule> filterCapsuleQueue;
+  @Autowired
+  private FilterCapsuleQueue filterCapsuleQueue;
 
   @Getter
   private volatile long latestSolidityNumShutDown;
@@ -273,16 +271,10 @@ public class Manager {
   private static final String rePushEsName = "repush";
   private ExecutorService triggerEs;
   private static final String triggerEsName = "event-trigger";
-  private ExecutorService filterEs;
-  private static final String filterEsName = "filter";
 
   @Autowired
   private RewardViCalService rewardViCalService;
 
-  @Lazy
-  @Autowired
-  private TronJsonRpcImpl tronJsonRpcImpl;
-
   /**
    * Cycle thread to rePush Transactions
    */
@@ -334,26 +326,6 @@ public class Manager {
         }
       };
 
-  private Runnable filterProcessLoop =
-      () -> {
-        while (isRunFilterProcessThread) {
-          try {
-            FilterTriggerCapsule filterCapsule = filterCapsuleQueue.poll(1, TimeUnit.SECONDS);
-            if (filterCapsule instanceof LogsFilterCapsule) {
-              tronJsonRpcImpl.handleLogsFilter((LogsFilterCapsule) filterCapsule);
-            } else if (filterCapsule instanceof BlockFilterCapsule) {
-              tronJsonRpcImpl.handleBLockFilter((BlockFilterCapsule) filterCapsule);
-            }
-          } catch (InterruptedException e) {
-            logger.error("FilterProcessLoop get InterruptedException, error is {}.",
-                    e.getMessage());
-            Thread.currentThread().interrupt();
-          } catch (Throwable throwable) {
-            logger.error("Unknown throwable happened in filterProcessLoop. ", throwable);
-          }
-        }
-      };
-
   private Comparator downComparator = (Comparator<TransactionCapsule>) (o1, o2) -> Long
       .compare(o2.getOrder(), o1.getOrder());
 
@@ -476,11 +448,6 @@ public void stopRePushTriggerThread() {
     ExecutorServiceManager.shutdownAndAwaitTermination(triggerEs, triggerEsName);
   }
 
-  public void stopFilterProcessThread() {
-    isRunFilterProcessThread = false;
-    ExecutorServiceManager.shutdownAndAwaitTermination(filterEs, filterEsName);
-  }
-
   public void stopValidateSignThread() {
     ExecutorServiceManager.shutdownAndAwaitTermination(validateSignService, "validate-sign");
   }
@@ -510,7 +477,6 @@ public void init() {
       this.rePushTransactions = new LinkedBlockingQueue<>();
     }
     this.triggerCapsuleQueue = new LinkedBlockingQueue<>();
-    this.filterCapsuleQueue = new LinkedBlockingQueue<>();
     chainBaseManager.setMerkleContainer(getMerkleContainer());
     chainBaseManager.setMortgageService(mortgageService);
     this.initGenesis();
@@ -584,12 +550,6 @@ public void init() {
       ExecutorServiceManager.submit(triggerEs, triggerCapsuleProcessLoop);
     }
 
-    // start json rpc filter process
-    if (CommonParameter.getInstance().isJsonRpcFilterEnabled()) {
-      filterEs = ExecutorServiceManager.newSingleThreadExecutor(filterEsName);
-      ExecutorServiceManager.submit(filterEs, filterProcessLoop);
-    }
-
     //initStoreFactory
     prepareStoreFactory();
     //initActuatorCreator
@@ -2344,9 +2304,7 @@ private void reApplyBlockEvents(List<KhaosBlock> newBranch) {
   private void postBlockFilter(final BlockCapsule blockCapsule, boolean solidified) {
     BlockFilterCapsule blockFilterCapsule =
         new BlockFilterCapsule(blockCapsule, solidified);
-    if (!filterCapsuleQueue.offer(blockFilterCapsule)) {
-      logger.info("Too many filters, block filter lost: {}.", blockCapsule.getBlockId());
-    }
+    filterCapsuleQueue.offer(blockFilterCapsule);
   }
 
   private void postLogsFilter(final BlockCapsule blockCapsule, boolean solidified,
@@ -2359,9 +2317,7 @@ private void postLogsFilter(final BlockCapsule blockCapsule, boolean solidified,
           blockCapsule.getBlockId().toString(), blockCapsule.getBloom(), transactionInfoList,
           solidified, removed);
 
-      if (!filterCapsuleQueue.offer(logsFilterCapsule)) {
-        logger.info("Too many filters, logs filter lost: {}.", blockNumber);
-      }
+      filterCapsuleQueue.offer(logsFilterCapsule);
     }
   }
 
@@ -2658,7 +2614,6 @@ public void close() {
     stopRePushThread();
     stopRePushTriggerThread();
     EventPluginLoader.getInstance().stopPlugin();
-    stopFilterProcessThread();
     stopValidateSignThread();
     chainBaseManager.shutdown();
     revokingStore.shutdown();

framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java

@@ -20,10 +20,10 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
+import com.google.common.util.concurrent.Uninterruptibles;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.GeneratedMessageV3;
 import java.io.Closeable;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -36,7 +36,9 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Pattern;
+import javax.annotation.PostConstruct;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -53,7 +55,9 @@
 import org.tron.common.es.ExecutorServiceManager;
 import org.tron.common.logsfilter.ContractEventParser;
 import org.tron.common.logsfilter.capsule.BlockFilterCapsule;
+import org.tron.common.logsfilter.capsule.FilterTriggerCapsule;
 import org.tron.common.logsfilter.capsule.LogsFilterCapsule;
+import org.tron.common.logsfilter.queue.FilterCapsuleQueue;
 import org.tron.common.parameter.CommonParameter;
 import org.tron.common.runtime.vm.DataWord;
 import org.tron.common.utils.ByteArray;
@@ -193,20 +197,50 @@ public enum RequestSource {
   private final ExecutorService sectionExecutor;
   private final NodeInfoService nodeInfoService;
   private final Wallet wallet;
-  @Autowired
-  private Manager manager;
+  private final Manager manager;
   private final String esName = "query-section";
 
   @Autowired
-  public TronJsonRpcImpl(@Autowired NodeInfoService nodeInfoService, @Autowired Wallet wallet) {
+  private FilterCapsuleQueue filterCapsuleQueue;
+  private ExecutorService filterEs;
+  private static final String filterEsName = "filter";
+  private final AtomicBoolean closed = new AtomicBoolean(false);
+
+  @Autowired
+  public TronJsonRpcImpl(NodeInfoService nodeInfoService, Wallet wallet, Manager manager) {
     this.nodeInfoService = nodeInfoService;
     this.wallet = wallet;
+    this.manager = manager;
     this.sectionExecutor = ExecutorServiceManager.newFixedThreadPool(esName, 5);
   }
 
-  @VisibleForTesting
-  public void setManager(Manager manager) {
-    this.manager = manager;
+  @PostConstruct
+  private void start() {
+    if (CommonParameter.getInstance().isJsonRpcFilterEnabled()) {
+      filterEs = ExecutorServiceManager.newSingleThreadExecutor(filterEsName, true);
+      ExecutorServiceManager.submit(filterEs, this::filterProcessLoop);
+    }
+  }
+
+  private void filterProcessLoop() {
+    while (!closed.get()) {
+      try {
+        FilterTriggerCapsule filterCapsule = filterCapsuleQueue.poll(1, TimeUnit.SECONDS);
+        if (filterCapsule instanceof LogsFilterCapsule) {
+          handleLogsFilter((LogsFilterCapsule) filterCapsule);
+        } else if (filterCapsule instanceof BlockFilterCapsule) {
+          handleBLockFilter((BlockFilterCapsule) filterCapsule);
+        } else if (filterCapsule != null) {
+          logger.warn("Unknown FilterTriggerCapsule: {}", filterCapsule.getClass().getName());
+        }
+      } catch (InterruptedException e) {
+        logger.error("FilterProcessLoop get InterruptedException, error is {}.", e.getMessage());
+        Thread.currentThread().interrupt();
+        return;
+      } catch (Throwable throwable) {
+        logger.error("Unknown throwable happened in filterProcessLoop. ", throwable);
+      }
+    }
   }
 
   @VisibleForTesting
@@ -1613,7 +1647,17 @@ public Object[] getFilterResult(String filterId, Map<String, BlockFilterAndResul
   }
 
   @Override
-  public void close() throws IOException {
+  public void close() {
+    if (!closed.compareAndSet(false, true)) {
+      return;
+    }
+    // The consumer loop submits to logsFilterPool (over-threshold path), so it must
+    // terminate before the pool shuts down, even if the closing thread is interrupted.
+    ExecutorServiceManager.shutdownAndAwaitTermination(filterEs, filterEsName);
+    if (filterEs != null && !filterEs.isTerminated()
+        && !Uninterruptibles.awaitTerminationUninterruptibly(filterEs, 2, TimeUnit.SECONDS)) {
+      logger.warn("Pool {} did not terminate before logs-filter-pool shutdown", filterEsName);
+    }
     ExecutorServiceManager.shutdownAndAwaitTermination(logsFilterPool, "logs-filter-pool");
     logElementCache.invalidateAll();
     blockHashCache.invalidateAll();

framework/src/test/java/org/tron/common/runtime/vm/Create2Test.java

@@ -209,8 +209,7 @@ private void testJsonRpc(byte[] actualContract, long loop) {
     NodeInfoService nodeInfoService;
     nodeInfoService = context.getBean(NodeInfoService.class);
     Wallet wallet = context.getBean(Wallet.class);
-    tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet);
-    tronJsonRpc.setManager(manager);
+    tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, manager);
     try {
       String res =
           tronJsonRpc.getStorageAt(ByteArray.toHexString(actualContract), "0", "latest");

framework/src/test/java/org/tron/core/db/ManagerTest.java

@@ -45,10 +45,10 @@
 import org.tron.common.logsfilter.EventPluginLoader;
 import org.tron.common.logsfilter.capsule.BlockFilterCapsule;
 import org.tron.common.logsfilter.capsule.BlockLogTriggerCapsule;
-import org.tron.common.logsfilter.capsule.FilterTriggerCapsule;
 import org.tron.common.logsfilter.capsule.LogsFilterCapsule;
 import org.tron.common.logsfilter.capsule.TransactionLogTriggerCapsule;
 import org.tron.common.logsfilter.capsule.TriggerCapsule;
+import org.tron.common.logsfilter.queue.FilterCapsuleQueue;
 import org.tron.common.logsfilter.trigger.ContractLogTrigger;
 import org.tron.common.parameter.CommonParameter;
 import org.tron.common.runtime.RuntimeImpl;
@@ -1694,8 +1694,8 @@ public void adjustBalance(AccountStore accountStore, byte[] accountAddress, long
   @Test
   public void switchForkShouldPostFullNodeFilterForNewBranch() throws Exception {
     CommonParameter.getInstance().jsonRpcHttpFullNodeEnable = true;
-    // filterProcessLoop only starts when isJsonRpcFilterEnabled() held at Manager.init() time; it
-    // was false then, so filterCapsuleQueue is produce-only here and fully observable.
+    // The consumer thread only starts when isJsonRpcFilterEnabled() held at context startup; it
+    // was false then, so the FilterCapsuleQueue bean is produce-only here and fully observable.
 
     // bootstrap a head with a known witness
     String key = PublicMethod.getRandomPrivateKey();
@@ -1733,9 +1733,7 @@ public void switchForkShouldPostFullNodeFilterForNewBranch() throws Exception {
     dbManager.pushBlock(p);
 
     long expiration = t + 1_000_000L;
-    BlockingQueue<FilterTriggerCapsule> queue =
-        ReflectUtils.getFieldValue(dbManager, "filterCapsuleQueue");
-    queue.clear();
+    FilterCapsuleQueue queue = context.getBean(FilterCapsuleQueue.class);
 
     // old branch: A carries a transfer; applied via the normal extend path
     BlockCapsule a = blockWithTransfer(t + 6000, base + 2, p.getBlockId().getByteString(), keys,
@@ -1868,7 +1866,7 @@ private BlockCapsule blockWithTransfer(long time, long number, ByteString parent
     return blockCapsule;
   }
 
-  private boolean hasLogsFilterCapsule(BlockingQueue<FilterTriggerCapsule> queue, BlockCapsule b,
+  private boolean hasLogsFilterCapsule(FilterCapsuleQueue queue, BlockCapsule b,
       boolean removed) {
     String blockHash = b.getBlockId().toString();
     return queue.stream()
@@ -1878,7 +1876,7 @@ private boolean hasLogsFilterCapsule(BlockingQueue<FilterTriggerCapsule> queue,
             && blockHash.equals(c.getBlockHash()));
   }
 
-  private boolean hasBlockFilterCapsule(BlockingQueue<FilterTriggerCapsule> queue,
+  private boolean hasBlockFilterCapsule(FilterCapsuleQueue queue,
       BlockCapsule b) {
     String blockHash = b.getBlockId().toString();
     return queue.stream()

framework/src/test/java/org/tron/core/jsonrpc/ConcurrentHashMapTest.java

@@ -23,7 +23,7 @@
 @Slf4j
 public class ConcurrentHashMapTest {
   private static final String EXECUTOR_NAME = "jsonrpc-concurrent-map-test";
-  private final TronJsonRpcImpl jsonRpc = new TronJsonRpcImpl(null, null);
+  private final TronJsonRpcImpl jsonRpc = new TronJsonRpcImpl(null, null, null);
 
   private static int randomInt(int minInt, int maxInt) {
     return (int) round(random(true) * (maxInt - minInt) + minInt, true);