diff --git a/common/src/main/java/org/tron/common/parameter/CommonParameter.java b/common/src/main/java/org/tron/common/parameter/CommonParameter.java index a73158a718a..2638c3e8c90 100644 --- a/common/src/main/java/org/tron/common/parameter/CommonParameter.java +++ b/common/src/main/java/org/tron/common/parameter/CommonParameter.java @@ -459,6 +459,9 @@ public class CommonParameter { @Getter @Setter public int jsonRpcMaxBlockFilterNum = 50000; + @Getter + @Setter + public int jsonRpcMaxLogFilterNum = 20000; @Getter @Setter diff --git a/framework/src/main/java/org/tron/core/config/args/Args.java b/framework/src/main/java/org/tron/core/config/args/Args.java index 83d7fd2c63d..11e8063b4ab 100644 --- a/framework/src/main/java/org/tron/core/config/args/Args.java +++ b/framework/src/main/java/org/tron/core/config/args/Args.java @@ -265,6 +265,11 @@ public static void applyConfigParams( config.getInt(ConfigKey.NODE_JSONRPC_MAX_BLOCK_FILTER_NUM); } + if (config.hasPath(ConfigKey.NODE_JSONRPC_MAX_LOG_FILTER_NUM)) { + PARAMETER.jsonRpcMaxLogFilterNum = + config.getInt(ConfigKey.NODE_JSONRPC_MAX_LOG_FILTER_NUM); + } + if (config.hasPath(ConfigKey.VM_MIN_TIME_RATIO)) { PARAMETER.minTimeRatio = config.getDouble(ConfigKey.VM_MIN_TIME_RATIO); } diff --git a/framework/src/main/java/org/tron/core/config/args/ConfigKey.java b/framework/src/main/java/org/tron/core/config/args/ConfigKey.java index b21c9c440a4..9d2751fa825 100644 --- a/framework/src/main/java/org/tron/core/config/args/ConfigKey.java +++ b/framework/src/main/java/org/tron/core/config/args/ConfigKey.java @@ -150,6 +150,8 @@ private ConfigKey() { public static final String NODE_JSONRPC_MAX_SUB_TOPICS = "node.jsonrpc.maxSubTopics"; public static final String NODE_JSONRPC_MAX_BLOCK_FILTER_NUM = "node.jsonrpc.maxBlockFilterNum"; + public static final String NODE_JSONRPC_MAX_LOG_FILTER_NUM = + "node.jsonrpc.maxLogFilterNum"; // node - dns public static final String NODE_DNS_TREE_URLS = "node.dns.treeUrls"; diff --git a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpc.java b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpc.java index 115df6ef9da..42da36a7811 100644 --- a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpc.java +++ b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpc.java @@ -291,9 +291,10 @@ CompilationResult ethSubmitHashrate(String hashrate, String id) @JsonRpcErrors({ @JsonRpcError(exception = JsonRpcMethodNotFoundException.class, code = -32601, data = "{}"), @JsonRpcError(exception = JsonRpcInvalidParamsException.class, code = -32602, data = "{}"), + @JsonRpcError(exception = JsonRpcExceedLimitException.class, code = -32005, data = "{}"), }) String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException, - JsonRpcMethodNotFoundException; + JsonRpcMethodNotFoundException, JsonRpcExceedLimitException; @JsonRpcMethod("eth_newBlockFilter") @JsonRpcErrors({ diff --git a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java index de939bdfff4..8fe7d3cb1b7 100644 --- a/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java +++ b/framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpcImpl.java @@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; import java.util.regex.Pattern; import lombok.Getter; @@ -115,6 +116,7 @@ public enum RequestSource { private static final String FILTER_NOT_FOUND = "filter not found"; public static final int EXPIRE_SECONDS = 5 * 60; private static final int maxBlockFilterNum = Args.getInstance().getJsonRpcMaxBlockFilterNum(); + private static final int maxLogFilterNum = Args.getInstance().getJsonRpcMaxLogFilterNum(); private static final Cache logElementCache = CacheBuilder.newBuilder() .maximumSize(300_000L) // 300s * tps(1000) * 1 log/tx ≈ 300_000 @@ -169,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 */ @@ -222,53 +226,68 @@ public static void handleBLockFilter(BlockFilterCapsule blockFilterCapsule) { * append LogsFilterCapsule's LogFilterElement list to each filter if matched */ public static void handleLogsFilter(LogsFilterCapsule logsFilterCapsule) { - Iterator> it; + long t1 = System.currentTimeMillis(); + Map eventFilterMap; if (logsFilterCapsule.isSolidified()) { - it = getEventFilter2ResultSolidity().entrySet().iterator(); + eventFilterMap = getEventFilter2ResultSolidity(); } else { - it = getEventFilter2ResultFull().entrySet().iterator(); + eventFilterMap = getEventFilter2ResultFull(); } - while (it.hasNext()) { - Entry entry = it.next(); - if (entry.getValue().isExpire()) { - it.remove(); - continue; - } + 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.debug("handleLogsFilter {} cost {}, filter size {}", + logsFilterCapsule.isSolidified() ? "Solidity" : "Full", t2 - t1, eventFilterMap.size()); + } + + private static void processLogFilterEntry( + Map.Entry entry, + Map eventFilterMap, + LogsFilterCapsule logsFilterCapsule) { + LogFilterAndResult logFilterAndResult = entry.getValue(); + if (logFilterAndResult.isExpire()) { + eventFilterMap.remove(entry.getKey()); + return; + } - LogFilterAndResult logFilterAndResult = entry.getValue(); - long fromBlock = logFilterAndResult.getLogFilterWrapper().getFromBlock(); - long toBlock = logFilterAndResult.getLogFilterWrapper().getToBlock(); - if (!(fromBlock <= logsFilterCapsule.getBlockNumber() - && logsFilterCapsule.getBlockNumber() <= toBlock)) { - continue; - } + 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())) { - continue; - } + if (logsFilterCapsule.getBloom() != null && !logFilterAndResult.getLogFilterWrapper() + .getLogFilter().matchBloom(logsFilterCapsule.getBloom())) { + return; + } - LogFilter logFilter = logFilterAndResult.getLogFilterWrapper().getLogFilter(); - List elements = - LogMatch.matchBlock(logFilter, logsFilterCapsule.getBlockNumber(), - logsFilterCapsule.getBlockHash(), logsFilterCapsule.getTxInfoList(), - logsFilterCapsule.isRemoved()); + LogFilter logFilter = logFilterAndResult.getLogFilterWrapper().getLogFilter(); + List elements = + LogMatch.matchBlock(logFilter, blockNumber, logsFilterCapsule.getBlockHash(), + logsFilterCapsule.getTxInfoList(), logsFilterCapsule.isRemoved()); - 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().add(cachedElement); + List 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); } + logFilterAndResult.getResult().addAll(localResults); } @Override @@ -1406,7 +1425,7 @@ public CompilationResult ethSubmitHashrate(String hashrate, String id) @Override public String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException, - JsonRpcMethodNotFoundException { + JsonRpcMethodNotFoundException, JsonRpcExceedLimitException { disableInPBFT("eth_newFilter"); // not supports finalized as block parameter @@ -1421,7 +1440,10 @@ public String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException, } else { eventFilter2Result = eventFilter2ResultSolidity; } - + if (eventFilter2Result.size() >= maxLogFilterNum) { + throw new JsonRpcExceedLimitException( + "exceed max log filters: " + maxLogFilterNum + ", try again later"); + } long currentMaxFullNum = wallet.getNowBlock().getBlockHeader().getRawData().getNumber(); LogFilterAndResult logFilterAndResult = new LogFilterAndResult(fr, currentMaxFullNum, wallet); String filterID = generateFilterId(); @@ -1573,6 +1595,7 @@ public static Object[] getFilterResult(String filterId, Map matchedLog = matchBlock(logFilterWrapper.getLogFilter(), blockNum, blockHash, transactionInfoList, false); + if (!matchedLog.isEmpty()) { + if (logFilterElementList.size() + matchedLog.size() > LogBlockQuery.MAX_RESULT) { + throw new JsonRpcTooManyResultException( + "query returned more than " + LogBlockQuery.MAX_RESULT + " results"); + } logFilterElementList.addAll(matchedLog); } - - if (logFilterElementList.size() > LogBlockQuery.MAX_RESULT) { - throw new JsonRpcTooManyResultException( - "query returned more than " + LogBlockQuery.MAX_RESULT + " results"); - } } return logFilterElementList.toArray(new LogFilterElement[0]); diff --git a/framework/src/main/resources/config.conf b/framework/src/main/resources/config.conf index 369924074bc..2e2418cbe47 100644 --- a/framework/src/main/resources/config.conf +++ b/framework/src/main/resources/config.conf @@ -375,6 +375,8 @@ node { maxSubTopics = 1000 # Allowed maximum number for blockFilter maxBlockFilterNum = 50000 + # Allowed maximum number for newFilter + maxLogFilterNum = 20000 } # Disabled api list, it will work for http, rpc and pbft, both FullNode and SolidityNode, diff --git a/framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java b/framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java index fdd9cb44222..b1d5c40e835 100644 --- a/framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java +++ b/framework/src/test/java/org/tron/core/jsonrpc/WalletCursorTest.java @@ -87,6 +87,7 @@ public void testDisableInSolidity() { TronJsonRpcImpl tronJsonRpc = new TronJsonRpcImpl(nodeInfoService, wallet, dbManager); try { tronJsonRpc.buildTransaction(buildArguments); + tronJsonRpc.close(); } catch (Exception e) { Assert.assertEquals("the method buildTransaction does not exist/is not available in " + "SOLIDITY", e.getMessage()); @@ -136,6 +137,7 @@ public void testEnableInFullNode() { try { tronJsonRpc.buildTransaction(buildArguments); + tronJsonRpc.close(); } catch (Exception e) { Assert.fail(); } diff --git a/framework/src/test/resources/config-localtest.conf b/framework/src/test/resources/config-localtest.conf index f1f40dead76..1f3b8a6f3af 100644 --- a/framework/src/test/resources/config-localtest.conf +++ b/framework/src/test/resources/config-localtest.conf @@ -168,6 +168,7 @@ node { # maxBlockRange = 5000 # maxSubTopics = 1000 # maxBlockFilterNum = 30000 + # maxLogFilterNum = 20000 } } diff --git a/framework/src/test/resources/config-test-mainnet.conf b/framework/src/test/resources/config-test-mainnet.conf index d39f432ac36..9f968c5628d 100644 --- a/framework/src/test/resources/config-test-mainnet.conf +++ b/framework/src/test/resources/config-test-mainnet.conf @@ -95,6 +95,7 @@ node { # maxBlockRange = 5000 # maxSubTopics = 1000 # maxBlockFilterNum = 50000 + # maxLogFilterNum = 20000 } rpc { diff --git a/framework/src/test/resources/config-test.conf b/framework/src/test/resources/config-test.conf index 71e93f84db5..21cebbfeef4 100644 --- a/framework/src/test/resources/config-test.conf +++ b/framework/src/test/resources/config-test.conf @@ -119,6 +119,7 @@ node { # maxBlockRange = 5000 # maxSubTopics = 1000 # maxBlockFilterNum = 30000 + # maxLogFilterNum = 20000 } # use your ipv6 address for node discovery and tcp connection, default false