remove Lazy tronJsonRpcImpl in Manager

Author: 317787106

common/src/main/java/org/tron/common/es/ExecutorServiceManager.java

@@ -4,11 +4,14 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import lombok.extern.slf4j.Slf4j;
 import org.tron.common.exit.ExitManager;
 
@@ -35,6 +38,17 @@ public static ScheduledExecutorService newSingleThreadScheduledExecutor(String n
         new ThreadFactoryBuilder().setNameFormat(name).setDaemon(isDaemon).build());
   }
 
+  public static ForkJoinPool newForkJoinPool(String name, int parallelism) {
+    AtomicInteger counter = new AtomicInteger(0);
+    ForkJoinPool.ForkJoinWorkerThreadFactory factory = pool -> {
+      ForkJoinWorkerThread thread =
+          ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
+      thread.setName(name + "-" + counter.getAndIncrement());
+      return thread;
+    };
+    return new ForkJoinPool(parallelism, factory, null, false);
+  }
+
   public static ExecutorService newFixedThreadPool(String name, int fixThreads) {
     return newFixedThreadPool(name, fixThreads, false);
   }

framework/src/main/java/org/tron/common/logsfilter/capsule/BlockFilterCapsule.java

@@ -5,7 +5,6 @@
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 import org.tron.core.capsule.BlockCapsule;
-import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
 
 @Slf4j(topic = "API")
 @ToString
@@ -18,22 +17,13 @@ public class BlockFilterCapsule extends FilterTriggerCapsule {
   @Setter
   private boolean solidified;
 
-  private final TronJsonRpcImpl jsonRpc;
-
-  public BlockFilterCapsule(BlockCapsule block, boolean solidified, TronJsonRpcImpl jsonRpc) {
-    this(block.getBlockId().toString(), solidified, jsonRpc);
+  public BlockFilterCapsule(BlockCapsule block, boolean solidified) {
+    this(block.getBlockId().toString(), solidified);
   }
 
-  public BlockFilterCapsule(String blockHash, boolean solidified, TronJsonRpcImpl jsonRpc) {
+  public BlockFilterCapsule(String blockHash, boolean solidified) {
     this.blockHash = blockHash;
     this.solidified = solidified;
-    this.jsonRpc = jsonRpc;
-  }
-
-  @Override
-  public void processFilterTrigger() {
-    jsonRpc.handleBLockFilter(this);
   }
 
 }
-

framework/src/main/java/org/tron/common/logsfilter/capsule/LogsFilterCapsule.java

@@ -6,7 +6,6 @@
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 import org.tron.common.bloom.Bloom;
-import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
 import org.tron.protos.Protocol.TransactionInfo;
 
 @Slf4j(topic = "API")
@@ -32,22 +31,14 @@ public class LogsFilterCapsule extends FilterTriggerCapsule {
   @Setter
   private boolean removed;
 
-  private final TronJsonRpcImpl jsonRpc;
-
   public LogsFilterCapsule(long blockNumber, String blockHash, Bloom bloom,
-      List<TransactionInfo> txInfoList, boolean solidified, boolean removed,
-      TronJsonRpcImpl jsonRpc) {
+      List<TransactionInfo> txInfoList, boolean solidified, boolean removed) {
     this.blockNumber = blockNumber;
     this.blockHash = blockHash;
     this.bloom = bloom;
     this.txInfoList = txInfoList;
     this.solidified = solidified;
     this.removed = removed;
-    this.jsonRpc = jsonRpc;
   }
 
-  @Override
-  public void processFilterTrigger() {
-    jsonRpc.handleLogsFilter(this);
-  }
-}
+}

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;
@@ -278,7 +277,6 @@ public class Manager {
   @Autowired
   private RewardViCalService rewardViCalService;
 
-  @Lazy
   @Autowired
   private TronJsonRpcImpl tronJsonRpcImpl;
 
@@ -338,8 +336,10 @@ public class Manager {
         while (isRunFilterProcessThread) {
           try {
             FilterTriggerCapsule filterCapsule = filterCapsuleQueue.poll(1, TimeUnit.SECONDS);
-            if (filterCapsule != null) {
-              filterCapsule.processFilterTrigger();
+            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 {}.",
@@ -2271,7 +2271,7 @@ private void reOrgLogsFilter() {
 
   private void postBlockFilter(final BlockCapsule blockCapsule, boolean solidified) {
     BlockFilterCapsule blockFilterCapsule =
-        new BlockFilterCapsule(blockCapsule, solidified, tronJsonRpcImpl);
+        new BlockFilterCapsule(blockCapsule, solidified);
     if (!filterCapsuleQueue.offer(blockFilterCapsule)) {
       logger.info("Too many filters, block filter lost: {}.", blockCapsule.getBlockId());
     }
@@ -2285,7 +2285,7 @@ private void postLogsFilter(final BlockCapsule blockCapsule, boolean solidified,
               = getTransactionInfoByBlockNum(blockNumber).getTransactionInfoList();
       LogsFilterCapsule logsFilterCapsule = new LogsFilterCapsule(blockNumber,
           blockCapsule.getBlockId().toString(), blockCapsule.getBloom(), transactionInfoList,
-          solidified, removed, tronJsonRpcImpl);
+          solidified, removed);
 
       if (!filterCapsuleQueue.offer(logsFilterCapsule)) {
         logger.info("Too many filters, logs filter lost: {}.", blockNumber);

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

@@ -60,7 +60,7 @@ public class JsonRpcApiUtil {
   public static final String TAG_SAFE_SUPPORT_ERROR = "TAG safe not supported";
   public static final String BLOCK_NUM_ERROR = "invalid block number";
 
-  static SecureRandom random = new SecureRandom();
+  private static final SecureRandom random = new SecureRandom();
 
   public static byte[] convertToTronAddress(byte[] address) {
     byte[] newAddress = new byte[21];

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

@@ -34,6 +34,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 import lombok.Getter;
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.bouncycastle.util.encoders.Hex;
@@ -120,12 +121,12 @@ public enum RequestSource {
   public static final int EXPIRE_SECONDS = 5 * 60;
   private final int maxBlockFilterNum = Args.getInstance().getJsonRpcMaxBlockFilterNum();
   private final int maxLogFilterNum = Args.getInstance().getJsonRpcMaxLogFilterNum();
-  private final Cache<LogFilterElement, LogFilterElement> logElementCache =
+  private static final Cache<LogFilterElement, LogFilterElement> logElementCache =
       CacheBuilder.newBuilder()
           .maximumSize(300_000L) // 300s * tps(1000) * 1 log/tx ≈ 300_000
           .expireAfterWrite(EXPIRE_SECONDS, TimeUnit.SECONDS)
           .recordStats().build(); //LRU cache
-  private final Cache<String, String> blockHashCache =
+  private static final Cache<String, String> blockHashCache =
       CacheBuilder.newBuilder()
           .maximumSize(60_000L) // 300s * 200 block/s when syncing
           .expireAfterWrite(EXPIRE_SECONDS, TimeUnit.SECONDS)
@@ -174,22 +175,23 @@ public enum RequestSource {
    * processing for each block within the 3-second BLOCK_PRODUCED_INTERVAL. Increasing the thread
    * pool size too much may affect the performance of the main block processing thread.
    */
-  private final ForkJoinPool logsFilterPool = new ForkJoinPool(3);
+  private final ForkJoinPool logsFilterPool =
+      ExecutorServiceManager.newForkJoinPool("logs-filter-pool", 3);
   /**
    * thread pool of query section bloom store
    */
   private final ExecutorService sectionExecutor;
   private final NodeInfoService nodeInfoService;
   private final Wallet wallet;
-  private final Manager manager;
+  @Setter
+  @Autowired
+  private Manager manager;
   private final String esName = "query-section";
 
   @Autowired
-  public TronJsonRpcImpl(@Autowired NodeInfoService nodeInfoService, @Autowired Wallet wallet,
-                         @Autowired Manager manager) {
+  public TronJsonRpcImpl(@Autowired NodeInfoService nodeInfoService, @Autowired Wallet wallet) {
     this.nodeInfoService = nodeInfoService;
     this.wallet = wallet;
-    this.manager = manager;
     this.sectionExecutor = ExecutorServiceManager.newFixedThreadPool(esName, 5);
   }
 

framework/src/test/java/org/tron/common/logsfilter/capsule/BlockFilterCapsuleTest.java

@@ -6,19 +6,16 @@
 import org.junit.Test;
 import org.tron.common.utils.Sha256Hash;
 import org.tron.core.capsule.BlockCapsule;
-import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
 
 public class BlockFilterCapsuleTest {
 
   private BlockFilterCapsule blockFilterCapsule;
-  private TronJsonRpcImpl jsonRpc;
 
   @Before
   public void setUp() {
-    jsonRpc = new TronJsonRpcImpl(null, null, null);
     BlockCapsule blockCapsule = new BlockCapsule(1, Sha256Hash.ZERO_HASH,
         System.currentTimeMillis(), ByteString.EMPTY);
-    blockFilterCapsule = new BlockFilterCapsule(blockCapsule, false, jsonRpc);
+    blockFilterCapsule = new BlockFilterCapsule(blockCapsule, false);
   }
 
   @Test
@@ -32,9 +29,8 @@ public void testSetAndGetBlockHash() {
   @Test
   public void testSetAndIsSolidified() {
     blockFilterCapsule = new BlockFilterCapsule(
-        "e58f33f9baf9305dc6f82b9f1934ea8f0ade2defb951258d50167028c780351f", false, jsonRpc);
+        "e58f33f9baf9305dc6f82b9f1934ea8f0ade2defb951258d50167028c780351f", false);
     blockFilterCapsule.setSolidified(true);
-    blockFilterCapsule.processFilterTrigger();
     Assert.assertTrue(blockFilterCapsule.isSolidified());
   }
 }

framework/src/test/java/org/tron/common/logsfilter/capsule/LogsFilterCapsuleTest.java

@@ -6,19 +6,16 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.tron.common.bloom.Bloom;
-import org.tron.core.services.jsonrpc.TronJsonRpcImpl;
 
 public class LogsFilterCapsuleTest {
 
   private LogsFilterCapsule capsule;
-  private TronJsonRpcImpl jsonRpc;
 
   @Before
   public void setUp() {
-    jsonRpc = new TronJsonRpcImpl(null, null, null);
     capsule = new LogsFilterCapsule(0,
         "e58f33f9baf9305dc6f82b9f1934ea8f0ade2defb951258d50167028c780351f",
-        new Bloom(), new ArrayList<>(), true, false, jsonRpc);
+        new Bloom(), new ArrayList<>(), true, false);
   }
 
   @Test
@@ -30,7 +27,6 @@ public void testSetAndGetLogsFilterCapsule() {
     capsule.setRemoved(capsule.isRemoved());
     capsule.setTxInfoList(capsule.getTxInfoList());
     assertNotNull(capsule.toString());
-    capsule.processFilterTrigger();
   }
 
 }

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

@@ -209,7 +209,8 @@ 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, manager);
+    tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet);
+    tronJsonRpc.setManager(manager);
     try {
       String res =
           tronJsonRpc.getStorageAt(ByteArray.toHexString(actualContract), "0", "latest");

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, null);
+  private final TronJsonRpcImpl jsonRpc = new TronJsonRpcImpl(null, null);
 
   private static int randomInt(int minInt, int maxInt) {
     return (int) round(random(true) * (maxInt - minInt) + minInt, true);
@@ -71,7 +71,7 @@ public void testHandleBlockHash() {
 
           for (int j = 1 + (i - 1) * eachCount; j <= i * eachCount; j++) {
             BlockFilterCapsule blockFilterCapsule =
-                new BlockFilterCapsule(String.valueOf(j), false, jsonRpc);
+                new BlockFilterCapsule(String.valueOf(j), false);
             jsonRpc.handleBLockFilter(blockFilterCapsule);
           }
           try {