Skip to content

Commit 114bbec

Browse files
317787106317787106
authored
Optimize memory allocation for the filter interface of JSON-RPC, eth_newFilter, eth_newBlockFilter (#6495)
1 parent 0779184 commit 114bbec

File tree

18 files changed

+184
-9
lines changed

18 files changed

+184
-9
lines changed

‎chainbase/src/main/java/org/tron/core/db2/core/SnapshotManager.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -423,6 +423,7 @@ public List<String> getCheckpointList() {
423423

424424
private void deleteCheckpoint() {
425425
if(checkTmpStore == null) {
426+
// only occurs in mock test. TODO fix test
426427
return;
427428
}
428429
try {

‎common/src/main/java/org/tron/common/parameter/CommonParameter.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -527,6 +527,9 @@ public class CommonParameter {
527527
@Getter
528528
@Setter
529529
public int jsonRpcMaxSubTopics = 1000;
530+
@Getter
531+
@Setter
532+
public int jsonRpcMaxBlockFilterNum = 50000;
530533

531534
@Getter
532535
@Setter

‎common/src/main/java/org/tron/core/Constant.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,7 @@ public class Constant {
154154
public static final String NODE_JSONRPC_HTTP_PBFT_PORT = "node.jsonrpc.httpPBFTPort";
155155
public static final String NODE_JSONRPC_MAX_BLOCK_RANGE = "node.jsonrpc.maxBlockRange";
156156
public static final String NODE_JSONRPC_MAX_SUB_TOPICS = "node.jsonrpc.maxSubTopics";
157+
public static final String NODE_JSONRPC_MAX_BLOCK_FILTER_NUM = "node.jsonrpc.maxBlockFilterNum";
157158

158159
public static final String NODE_DISABLED_API_LIST = "node.disabledApi";
159160

‎common/src/main/java/org/tron/core/exception/jsonrpc/JsonRpcExceedLimitException.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package org.tron.core.exception.jsonrpc;
2+
3+
public class JsonRpcExceedLimitException extends JsonRpcException {
4+
5+
public JsonRpcExceedLimitException() {
6+
super();
7+
}
8+
9+
public JsonRpcExceedLimitException(String message) {
10+
super(message);
11+
}
12+
13+
public JsonRpcExceedLimitException(String message, Throwable cause) {
14+
super(message, cause);
15+
}
16+
}

‎framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -546,6 +546,7 @@ public boolean isBusy() {
546546
}
547547
int queueSize = 0;
548548
if (eventListeners == null || eventListeners.isEmpty()) {
549+
// only occurs in mock test. TODO fix test
549550
return false;
550551
}
551552
for (IPluginEventListener listener : eventListeners) {

‎framework/src/main/java/org/tron/core/config/args/Args.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -207,6 +207,7 @@ public static void clearParam() {
207207
PARAMETER.jsonRpcHttpPBFTNodeEnable = false;
208208
PARAMETER.jsonRpcMaxBlockRange = 5000;
209209
PARAMETER.jsonRpcMaxSubTopics = 1000;
210+
PARAMETER.jsonRpcMaxBlockFilterNum = 50000;
210211
PARAMETER.nodeMetricsEnable = false;
211212
PARAMETER.metricsStorageEnable = false;
212213
PARAMETER.metricsPrometheusEnable = false;
@@ -491,6 +492,11 @@ public static void setParam(final Config config) {
491492
config.getInt(Constant.NODE_JSONRPC_MAX_SUB_TOPICS);
492493
}
493494

495+
if (config.hasPath(Constant.NODE_JSONRPC_MAX_BLOCK_FILTER_NUM)) {
496+
PARAMETER.jsonRpcMaxBlockFilterNum =
497+
config.getInt(Constant.NODE_JSONRPC_MAX_BLOCK_FILTER_NUM);
498+
}
499+
494500
if (config.hasPath(Constant.VM_MIN_TIME_RATIO)) {
495501
PARAMETER.minTimeRatio = config.getDouble(Constant.VM_MIN_TIME_RATIO);
496502
}

‎framework/src/main/java/org/tron/core/net/peer/PeerStatusCheck.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ public void statusCheck() {
4343
long now = System.currentTimeMillis();
4444

4545
if (tronNetDelegate == null) {
46-
//only occurs in mock test. TODO fix test
46+
// only occurs in mock test. TODO fix test
4747
return;
4848
}
4949
tronNetDelegate.getActivePeer().forEach(peer -> {

‎framework/src/main/java/org/tron/core/services/jsonrpc/TronJsonRpc.java‎

Lines changed: 37 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import com.googlecode.jsonrpc4j.JsonRpcMethod;
88
import java.io.IOException;
99
import java.util.List;
10+
import java.util.Objects;
1011
import java.util.concurrent.ExecutionException;
1112
import lombok.AllArgsConstructor;
1213
import lombok.Getter;
@@ -18,6 +19,7 @@
1819
import org.tron.common.utils.ByteArray;
1920
import org.tron.core.exception.BadItemException;
2021
import org.tron.core.exception.ItemNotFoundException;
22+
import org.tron.core.exception.jsonrpc.JsonRpcExceedLimitException;
2123
import org.tron.core.exception.jsonrpc.JsonRpcInternalException;
2224
import org.tron.core.exception.jsonrpc.JsonRpcInvalidParamsException;
2325
import org.tron.core.exception.jsonrpc.JsonRpcInvalidRequestException;
@@ -29,6 +31,9 @@
2931
import org.tron.core.services.jsonrpc.types.TransactionReceipt;
3032
import org.tron.core.services.jsonrpc.types.TransactionResult;
3133

34+
/**
35+
* Error code refers to https://www.quicknode.com/docs/ethereum/error-references
36+
*/
3237
@Component
3338
public interface TronJsonRpc {
3439

@@ -292,9 +297,10 @@ String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException,
292297

293298
@JsonRpcMethod("eth_newBlockFilter")
294299
@JsonRpcErrors({
300+
@JsonRpcError(exception = JsonRpcExceedLimitException.class, code = -32005, data = "{}"),
295301
@JsonRpcError(exception = JsonRpcMethodNotFoundException.class, code = -32601, data = "{}"),
296302
})
297-
String newBlockFilter() throws JsonRpcMethodNotFoundException;
303+
String newBlockFilter() throws JsonRpcExceedLimitException, JsonRpcMethodNotFoundException;
298304

299305
@JsonRpcMethod("eth_uninstallFilter")
300306
@JsonRpcErrors({
@@ -472,5 +478,35 @@ public LogFilterElement(String blockHash, Long blockNum, String txId, Integer tx
472478
}
473479
this.removed = removed;
474480
}
481+
482+
@Override
483+
public boolean equals(Object o) {
484+
if (this == o) {
485+
return true;
486+
}
487+
if (o == null || this.getClass() != o.getClass()) {
488+
return false;
489+
}
490+
LogFilterElement item = (LogFilterElement) o;
491+
if (!Objects.equals(blockHash, item.blockHash)) {
492+
return false;
493+
}
494+
if (!Objects.equals(transactionHash, item.transactionHash)) {
495+
return false;
496+
}
497+
if (!Objects.equals(transactionIndex, item.transactionIndex)) {
498+
return false;
499+
}
500+
if (!Objects.equals(logIndex, item.logIndex)) {
501+
return false;
502+
}
503+
return removed == item.removed;
504+
}
505+
506+
@Override
507+
public int hashCode() {
508+
return Objects.hash(blockHash, transactionHash, transactionIndex, logIndex, removed);
509+
}
510+
475511
}
476512
}

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

Lines changed: 53 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111
import static org.tron.core.services.jsonrpc.JsonRpcApiUtil.triggerCallContract;
1212

1313
import com.alibaba.fastjson.JSON;
14+
import com.google.common.cache.Cache;
15+
import com.google.common.cache.CacheBuilder;
1416
import com.google.protobuf.ByteString;
1517
import com.google.protobuf.GeneratedMessageV3;
1618
import java.io.Closeable;
@@ -25,10 +27,10 @@
2527
import java.util.concurrent.ConcurrentHashMap;
2628
import java.util.concurrent.ExecutionException;
2729
import java.util.concurrent.ExecutorService;
30+
import java.util.concurrent.TimeUnit;
2831
import java.util.regex.Pattern;
2932
import lombok.Getter;
3033
import lombok.extern.slf4j.Slf4j;
31-
import org.apache.commons.collections4.CollectionUtils;
3234
import org.apache.commons.lang3.StringUtils;
3335
import org.bouncycastle.util.encoders.Hex;
3436
import org.springframework.beans.factory.annotation.Autowired;
@@ -51,6 +53,7 @@
5153
import org.tron.core.Wallet;
5254
import org.tron.core.capsule.BlockCapsule;
5355
import org.tron.core.capsule.TransactionCapsule;
56+
import org.tron.core.config.args.Args;
5457
import org.tron.core.db.Manager;
5558
import org.tron.core.db2.core.Chainbase;
5659
import org.tron.core.exception.BadItemException;
@@ -59,6 +62,7 @@
5962
import org.tron.core.exception.HeaderNotFound;
6063
import org.tron.core.exception.ItemNotFoundException;
6164
import org.tron.core.exception.VMIllegalException;
65+
import org.tron.core.exception.jsonrpc.JsonRpcExceedLimitException;
6266
import org.tron.core.exception.jsonrpc.JsonRpcInternalException;
6367
import org.tron.core.exception.jsonrpc.JsonRpcInvalidParamsException;
6468
import org.tron.core.exception.jsonrpc.JsonRpcInvalidRequestException;
@@ -110,6 +114,17 @@ public enum RequestSource {
110114

111115
private static final String FILTER_NOT_FOUND = "filter not found";
112116
public static final int EXPIRE_SECONDS = 5 * 60;
117+
private static final int maxBlockFilterNum = Args.getInstance().getJsonRpcMaxBlockFilterNum();
118+
private static final Cache<LogFilterElement, LogFilterElement> logElementCache =
119+
CacheBuilder.newBuilder()
120+
.maximumSize(300_000L) // 300s * tps(1000) * 1 log/tx ≈ 300_000
121+
.expireAfterWrite(EXPIRE_SECONDS, TimeUnit.SECONDS)
122+
.recordStats().build(); //LRU cache
123+
private static final Cache<String, String> blockHashCache =
124+
CacheBuilder.newBuilder()
125+
.maximumSize(60_000L) // 300s * 200 block/s when syncing
126+
.expireAfterWrite(EXPIRE_SECONDS, TimeUnit.SECONDS)
127+
.recordStats().build(); //LRU cache
113128
/**
114129
* for log filter in Full Json-RPC
115130
*/
@@ -181,16 +196,31 @@ public static void handleBLockFilter(BlockFilterCapsule blockFilterCapsule) {
181196
it = getBlockFilter2ResultFull().entrySet().iterator();
182197
}
183198

199+
if (!it.hasNext()) {
200+
return;
201+
}
202+
final String originalBlockHash = ByteArray.toJsonHex(blockFilterCapsule.getBlockHash());
203+
String cachedBlockHash;
204+
try {
205+
// compare with hashcode() first, then with equals(). If not exist, put it.
206+
cachedBlockHash = blockHashCache.get(originalBlockHash, () -> originalBlockHash);
207+
} catch (ExecutionException e) {
208+
logger.error("Getting/loading blockHash from cache failed", e); // never happen
209+
cachedBlockHash = originalBlockHash;
210+
}
184211
while (it.hasNext()) {
185212
Entry<String, BlockFilterAndResult> entry = it.next();
186213
if (entry.getValue().isExpire()) {
187214
it.remove();
188215
continue;
189216
}
190-
entry.getValue().getResult().add(ByteArray.toJsonHex(blockFilterCapsule.getBlockHash()));
217+
entry.getValue().getResult().add(cachedBlockHash);
191218
}
192219
}
193220

221+
/**
222+
* append LogsFilterCapsule's LogFilterElement list to each filter if matched
223+
*/
194224
public static void handleLogsFilter(LogsFilterCapsule logsFilterCapsule) {
195225
Iterator<Entry<String, LogFilterAndResult>> it;
196226

@@ -226,8 +256,17 @@ public static void handleLogsFilter(LogsFilterCapsule logsFilterCapsule) {
226256
LogMatch.matchBlock(logFilter, logsFilterCapsule.getBlockNumber(),
227257
logsFilterCapsule.getBlockHash(), logsFilterCapsule.getTxInfoList(),
228258
logsFilterCapsule.isRemoved());
229-
if (CollectionUtils.isNotEmpty(elements)) {
230-
logFilterAndResult.getResult().addAll(elements);
259+
260+
for (LogFilterElement element : elements) {
261+
LogFilterElement cachedElement;
262+
try {
263+
// compare with hashcode() first, then with equals(). If not exist, put it.
264+
cachedElement = logElementCache.get(element, () -> element);
265+
} catch (ExecutionException e) {
266+
logger.error("Getting/loading LogFilterElement from cache fails", e); // never happen
267+
cachedElement = element;
268+
}
269+
logFilterAndResult.getResult().add(cachedElement);
231270
}
232271
}
233272
}
@@ -797,7 +836,7 @@ public TransactionReceipt getTransactionReceipt(String txId)
797836
long blockNum = blockCapsule.getNum();
798837
TransactionInfoList transactionInfoList = wallet.getTransactionInfoByBlockNum(blockNum);
799838
long energyFee = wallet.getEnergyFee(blockCapsule.getTimeStamp());
800-
839+
801840
// Find transaction context
802841
TransactionReceipt.TransactionContext context
803842
= findTransactionContext(transactionInfoList,
@@ -806,7 +845,7 @@ public TransactionReceipt getTransactionReceipt(String txId)
806845
if (context == null) {
807846
return null; // Transaction not found in block
808847
}
809-
848+
810849
return new TransactionReceipt(blockCapsule, transactionInfo, context, energyFee);
811850
}
812851

@@ -1391,7 +1430,8 @@ public String newFilter(FilterRequest fr) throws JsonRpcInvalidParamsException,
13911430
}
13921431

13931432
@Override
1394-
public String newBlockFilter() throws JsonRpcMethodNotFoundException {
1433+
public String newBlockFilter() throws JsonRpcMethodNotFoundException,
1434+
JsonRpcExceedLimitException {
13951435
disableInPBFT("eth_newBlockFilter");
13961436

13971437
Map<String, BlockFilterAndResult> blockFilter2Result;
@@ -1400,6 +1440,10 @@ public String newBlockFilter() throws JsonRpcMethodNotFoundException {
14001440
} else {
14011441
blockFilter2Result = blockFilter2ResultSolidity;
14021442
}
1443+
if (blockFilter2Result.size() >= maxBlockFilterNum) {
1444+
throw new JsonRpcExceedLimitException(
1445+
"exceed max block filters: " + maxBlockFilterNum + ", try again later");
1446+
}
14031447

14041448
BlockFilterAndResult filterAndResult = new BlockFilterAndResult();
14051449
String filterID = generateFilterId();
@@ -1529,6 +1573,8 @@ public static Object[] getFilterResult(String filterId, Map<String, BlockFilterA
15291573

15301574
@Override
15311575
public void close() throws IOException {
1576+
logElementCache.invalidateAll();
1577+
blockHashCache.invalidateAll();
15321578
ExecutorServiceManager.shutdownAndAwaitTermination(sectionExecutor, esName);
15331579
}
15341580

‎framework/src/main/resources/config-localtest.conf‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -163,6 +163,7 @@ node {
163163
# httpPBFTPort = 8565
164164
# maxBlockRange = 5000
165165
# maxSubTopics = 1000
166+
# maxBlockFilterNum = 30000
166167
}
167168

168169
}

0 commit comments

Comments
 (0)