use ForkJoinPool as necessary

Author: 317787106

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

@@ -171,6 +171,8 @@ public enum RequestSource {
   private static final String NO_BLOCK_HEADER_BY_HASH = "header for hash not found";
 
   private static final String ERROR_SELECTOR = "08c379a0"; // Function selector for Error(string)
+  private static final int FILTER_PARALLEL_THRESHOLD = 10000;
+  private static final ForkJoinPool LOGS_FILTER_POOL = new ForkJoinPool(2);
   /**
    * thread pool of query section bloom store
    */
@@ -233,49 +235,59 @@ public static void handleLogsFilter(LogsFilterCapsule logsFilterCapsule) {
       eventFilterMap = getEventFilter2ResultFull();
     }
 
-    ForkJoinPool pool = new ForkJoinPool(2); //parallelStream default num(CPU) -1
-    pool.submit(() -> eventFilterMap.entrySet().parallelStream().forEach(entry -> {
+    if (eventFilterMap.size() <= FILTER_PARALLEL_THRESHOLD) {
+      eventFilterMap.entrySet().forEach(
+          entry -> processLogFilterEntry(entry, eventFilterMap, logsFilterCapsule));
+    } else {
+      LOGS_FILTER_POOL.submit(() -> eventFilterMap.entrySet().parallelStream()
+          .forEach(entry -> processLogFilterEntry(entry, eventFilterMap, logsFilterCapsule))
+      ).join();
+    }
+    long t2 = System.currentTimeMillis();
+    logger.info("handleLogsFilter {} cost {}, filter size {}",
+        logsFilterCapsule.isSolidified() ? "Solidity" : "Full", t2 - t1, eventFilterMap.size());
+  }
 
-      LogFilterAndResult logFilterAndResult = entry.getValue();
-      if (logFilterAndResult.isExpire()) {
-        eventFilterMap.remove(entry.getKey());
-        return;
-      }
+  private static void processLogFilterEntry(
+      Map.Entry<String, LogFilterAndResult> entry,
+      Map<String, LogFilterAndResult> eventFilterMap,
+      LogsFilterCapsule logsFilterCapsule) {
+    LogFilterAndResult logFilterAndResult = entry.getValue();
+    if (logFilterAndResult.isExpire()) {
+      eventFilterMap.remove(entry.getKey());
+      return;
+    }
 
-      long blockNumber = logsFilterCapsule.getBlockNumber();
-      long fromBlock = logFilterAndResult.getLogFilterWrapper().getFromBlock();
-      long toBlock = logFilterAndResult.getLogFilterWrapper().getToBlock();
-      if (!(fromBlock <= blockNumber && blockNumber <= toBlock)) {
-        return;
-      }
+    long blockNumber = logsFilterCapsule.getBlockNumber();
+    long fromBlock = logFilterAndResult.getLogFilterWrapper().getFromBlock();
+    long toBlock = logFilterAndResult.getLogFilterWrapper().getToBlock();
+    if (!(fromBlock <= blockNumber && blockNumber <= toBlock)) {
+      return;
+    }
 
-      if (logsFilterCapsule.getBloom() != null && !logFilterAndResult.getLogFilterWrapper()
-          .getLogFilter().matchBloom(logsFilterCapsule.getBloom())) {
-        return;
-      }
+    if (logsFilterCapsule.getBloom() != null && !logFilterAndResult.getLogFilterWrapper()
+        .getLogFilter().matchBloom(logsFilterCapsule.getBloom())) {
+      return;
+    }
 
-      LogFilter logFilter = logFilterAndResult.getLogFilterWrapper().getLogFilter();
-      List<LogFilterElement> elements =
-          LogMatch.matchBlock(logFilter, blockNumber, logsFilterCapsule.getBlockHash(),
-              logsFilterCapsule.getTxInfoList(), logsFilterCapsule.isRemoved());
+    LogFilter logFilter = logFilterAndResult.getLogFilterWrapper().getLogFilter();
+    List<LogFilterElement> elements =
+        LogMatch.matchBlock(logFilter, blockNumber, logsFilterCapsule.getBlockHash(),
+            logsFilterCapsule.getTxInfoList(), logsFilterCapsule.isRemoved());
 
-      List<LogFilterElement> localResults = new ArrayList<>(elements.size());
-      for (LogFilterElement element : elements) {
-        LogFilterElement cachedElement;
-        try {
-          // compare with hashcode() first, then with equals(). If not exist, put it.
-          cachedElement = logElementCache.get(element, () -> element);
-        } catch (ExecutionException e) {
-          logger.error("Getting/loading LogFilterElement from cache fails", e); // never happen
-          cachedElement = element;
-        }
-        localResults.add(cachedElement);
+    List<LogFilterElement> localResults = new ArrayList<>(elements.size());
+    for (LogFilterElement element : elements) {
+      LogFilterElement cachedElement;
+      try {
+        // compare with hashcode() first, then with equals(). If not exist, put it.
+        cachedElement = logElementCache.get(element, () -> element);
+      } catch (ExecutionException e) {
+        logger.error("Getting/loading LogFilterElement from cache fails", e); // never happen
+        cachedElement = element;
       }
-      logFilterAndResult.getResult().addAll(localResults);
-    })).join();
-    long t2 = System.currentTimeMillis();
-    logger.info("handleLogsFilter {} cost {}, filter size {}",
-        logsFilterCapsule.isSolidified() ? "Solidity" : "Full", t2 - t1, eventFilterMap.size());
+      localResults.add(cachedElement);
+    }
+    logFilterAndResult.getResult().addAll(localResults);
   }
 
   @Override
@@ -1428,7 +1440,7 @@ public String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException,
     } else {
       eventFilter2Result = eventFilter2ResultSolidity;
     }
-    if (eventFilter2Result.size() > maxLogFilterNum) {
+    if (eventFilter2Result.size() >= maxLogFilterNum) {
       throw new JsonRpcExceedLimitException(
           "exceed max log filters: " + maxLogFilterNum + ", try again later");
     }