StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #81 from p2p-org/optimise-block-by-number

Browse Source 
optimise default getBlockByNumber - skip processing requests of tx bo…
This commit is contained in:
a10zn8
2022-12-13 17:06:16 +04:00
committed by GitHub
parent f6296f63ea 817925ea7d
commit bbf8dc707f
5 changed files with 27 additions and 631 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt vendored
View File

@@ -24,7 +24,6 @@ import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.CompoundReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.EthereumFullBlocksReader
import io.emeraldpay.etherjar.rpc.json.BlockJson
import io.emeraldpay.etherjar.rpc.json.TransactionJson
import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
@@ -195,14 +194,6 @@ open class Caches(
return txsByHash
}
fun getFullBlocks(): Reader<BlockId, BlockContainer> {
return EthereumFullBlocksReader(blocksByHash, txsByHash)
}
fun getFullBlocksByHeight(): Reader<Long, BlockContainer> {
return BlockByHeight(blocksByHeight, EthereumFullBlocksReader(blocksByHash, txsByHash))
}
fun getReceipts(): Reader<TxId, ByteArray> {
return receiptByHash
}

145
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReader.kt
View File

@@ -1,145 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.Reader
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.util.function.Tuple2
import reactor.util.function.Tuples
import java.nio.ByteBuffer
import java.util.function.BiFunction
/**
* Reads blocks with full transactions details. Based on data contained in readers for blocks
* and transactions, i.e. two separate readers that must be provided.
*
* If source block, with just transaction hashes is not available, it returns empty
* If any of the expected block transactions is not available it returns empty
*/
class EthereumFullBlocksReader(
private val blocks: Reader<BlockId, BlockContainer>,
private val txes: Reader<TxId, TxContainer>
) : Reader<BlockId, BlockContainer> {
companion object {
private val log = LoggerFactory.getLogger(EthereumFullBlocksReader::class.java)
}
private val accumulate: BiFunction<ByteBuffer, ByteArray, ByteBuffer> = BiFunction { buf, x ->
if (buf.remaining() < x.size) {
val resize = ByteBuffer.allocate(buf.capacity() + buf.capacity() / 4 + x.size)
resize.put(buf.flip()).put(x)
} else {
buf.put(x)
}
}
override fun read(key: BlockId): Mono<BlockContainer> {
return blocks.read(key).flatMap { block ->
if (block.transactions.isEmpty()) {
// in fact it's not necessary to create a copy, made just for code clarity but it may be a performance loss
val fullBlock = BlockContainer(
block.height, block.hash, block.difficulty, block.timestamp,
true,
block.json,
block.parsed,
block.transactions
)
return@flatMap Mono.just(fullBlock)
}
val blockSplit = splitByTransactions(block.json!!)
val transactions = Flux.fromIterable(block.transactions)
.flatMapSequential { txes.read(it) }
.collectList()
return@flatMap transactions.flatMap { transactionsData ->
// make sure that all transaction are loaded, otherwise just return empty because cannot make full block data
if (transactionsData.size != block.transactions.size) {
log.warn("No data to fill the block")
Mono.empty()
} else {
joinWithTransactions(
blockSplit.t1,
blockSplit.t2,
Flux.fromIterable(transactionsData).map { it.json!! }
)
.reduce(ByteBuffer.allocate(block.json.size * 4), accumulate)
.map(this@EthereumFullBlocksReader::extractContent)
.map { json ->
BlockContainer(
block.height, block.hash, block.difficulty, block.timestamp,
true,
json,
null,
block.transactions
)
}
}
}
}
}
fun extractContent(it: ByteBuffer): ByteArray {
val pos = it.position()
val result = ByteArray(pos)
it.flip().get(result, 0, pos)
return result
}
fun splitByTransactions(json: ByteArray): Tuple2<ByteArray, ByteArray> {
// TODO find a lib that implements Knuth-Morris-Pratt Pattern Matching Algorithm for byte arrays
// and reimplement without making a string copy from bytes
val s = String(json)
val fieldStart = s.indexOf("\"transactions\"")
val arrayStart = s.indexOf("[", fieldStart)
val arrayEnd = s.indexOf("]", arrayStart)
val head = s.substring(0, arrayStart + 1)
val tail = s.substring(arrayEnd, s.length)
return Tuples.of(head.toByteArray(), tail.toByteArray())
}
fun joinWithTransactions(head: ByteArray, tail: ByteArray, transactions: Flux<ByteArray>): Flux<ByteArray> {
val separator = Flux.range(0, Integer.MAX_VALUE)
.map { it != 0 }
val transactionsWithSeparator = transactions.zipWith(separator)
.flatMap {
val tx = Flux.just(it.t1)
if (it.t2) {
Flux.concat(Flux.just(",".toByteArray()), tx)
} else {
tx
}
}
return Flux.concat(
Flux.just(head),
transactionsWithSeparator,
Flux.just(tail)
)
}
}

26
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
View File

@@ -47,11 +47,6 @@ class LocalCallRouter(
private val log = LoggerFactory.getLogger(LocalCallRouter::class.java)
}
private val fullBlocksReader = EthereumFullBlocksReader(
reader.blocksByIdAsCont(),
reader.txByHashAsCont()
)
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
if (methods.isHardcoded(key.method)) {
return Mono.just(methods.executeHardcoded(key.method))
@@ -107,7 +102,7 @@ class LocalCallRouter(
}
val withTx = params[1].toString().toBoolean()
if (withTx) {
fullBlocksReader.read(hash).map { it.json!! }
null
} else {
reader.blocksByIdAsCont().read(hash).map { it.json!! }
}
@@ -136,6 +131,11 @@ class LocalCallRouter(
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
}
val number: Long
val withTx = params[1].toString().toBoolean()
if (withTx) {
// with Tx request much more efficient in remote call
return null
}
try {
val blockRef = params[0].toString()
when {
@@ -165,16 +165,8 @@ class LocalCallRouter(
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be a block number")
}
val withTx = params[1].toString().toBoolean()
var block = reader.blocksByHeightAsCont()
.read(number)
block = if (withTx) {
block.flatMap {
fullBlocksReader.read(it.hash)
}
} else {
block
}
return block.map { it.json!! }
return reader.blocksByHeightAsCont()
.read(number).map { it.json!! }
}
}

460
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReaderSpec.groovy
View File

@@ -1,460 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.core.util.DefaultPrettyPrinter
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.BlocksMemCache
import io.emeraldpay.dshackle.cache.TxMemCache
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.etherjar.domain.BlockHash
import io.emeraldpay.etherjar.domain.TransactionId
import io.emeraldpay.etherjar.rpc.json.BlockJson
import io.emeraldpay.etherjar.rpc.json.TransactionJson
import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
import org.apache.commons.codec.binary.Hex
import reactor.core.publisher.Mono
import spock.lang.Specification
import java.nio.ByteBuffer
import java.time.Duration
import java.time.Instant
class EthereumFullBlocksReaderSpec extends Specification {
// sorted
String hash1 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33"
String hash3 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
String hash4 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
ObjectMapper objectMapper = Global.objectMapper
def tx1 = new TransactionJson().with {
it.blockNumber = 100
it.blockHash = BlockHash.from(hash1)
it.hash = TransactionId.from(hash1)
it.nonce = 1
it
}
def tx2 = new TransactionJson().with {
it.blockNumber = 100
it.blockHash = BlockHash.from(hash1)
it.hash = TransactionId.from(hash2)
it.nonce = 2
it
}
def tx3 = new TransactionJson().with {
it.blockNumber = 101
it.blockHash = BlockHash.from(hash3)
it.hash = TransactionId.from(hash3)
it.nonce = 3
it
}
// no block
def tx4 = new TransactionJson().with {
it.blockNumber = 102
it.blockHash = BlockHash.from(hash4)
it.hash = TransactionId.from(hash4)
it.nonce = 4
it
}
// two transactiosn, tx1 and tx2
def block1 = new BlockJson().with {
it.number = 100
it.hash = BlockHash.from(hash1)
it.totalDifficulty = BigInteger.ONE
it.timestamp = Instant.now()
it.transactions = [
new TransactionRefJson(tx1.hash),
new TransactionRefJson(tx2.hash)
]
it
}
// one transaction, tx3
def block2 = new BlockJson().with {
it.number = 101
it.hash = BlockHash.from(hash3)
it.totalDifficulty = BigInteger.ONE
it.timestamp = Instant.now()
it.transactions = [
new TransactionRefJson(tx3.hash)
]
it
}
// no transactions
def block3 = new BlockJson().with {
it.number = 102
it.hash = BlockHash.from(hash4)
it.totalDifficulty = BigInteger.ONE
it.timestamp = Instant.now()
it.transactions = []
it
}
// all transaction2
def block4 = new BlockJson().with {
it.number = 103
it.hash = BlockHash.from(hash3)
it.totalDifficulty = BigInteger.ONE
it.timestamp = Instant.now()
it.transactions = [
new TransactionRefJson(tx1.hash),
new TransactionRefJson(tx2.hash),
new TransactionRefJson(tx3.hash),
new TransactionRefJson(tx4.hash)
]
it
}
def "Read transactions and return full block"() {
setup:
def txes = new TxMemCache()
def blocks = new BlocksMemCache()
txes.add(TxContainer.from(tx1))
txes.add(TxContainer.from(tx2))
txes.add(TxContainer.from(tx3))
txes.add(TxContainer.from(tx4))
blocks.add(BlockContainer.from(block1))
blocks.add(BlockContainer.from(block2))
blocks.add(BlockContainer.from(block3))
def full = new EthereumFullBlocksReader(blocks, txes)
when:
def act = full.read(BlockId.from(block1.hash)).block()
then:
act != null
when:
act = objectMapper.readValue(act.json, BlockJson)
then:
act.hash == BlockHash.from(hash1)
act.number == 100
act.transactions.size() == 2
when:
def transactions = act.transactions.sort { it.hash }
then:
transactions[0] instanceof TransactionJson
transactions[1] instanceof TransactionJson
//verify it didn't update original block
block1.transactions[0] instanceof TransactionRefJson
block1.transactions[1] instanceof TransactionRefJson
with(transactions[0]) {
hash == TransactionId.from(hash1)
nonce == 1
}
with(transactions[1]) {
hash == TransactionId.from(hash2)
nonce == 2
}
// request second block
when:
act = full.read(BlockId.from(block2.hash)).block()
then:
act != null
when:
act = objectMapper.readValue(act.json, BlockJson)
then:
act.hash == BlockHash.from(hash3)
act.number == 101
act.transactions.size() == 1
with(act.transactions[0]) {
hash == TransactionId.from(hash3)
nonce == 3
}
}
def "Produce block with original order of transactions"() {
setup:
def txes = Mock(Reader) {
1 * it.read(TxId.from(tx1.hash)) >> Mono.just(TxContainer.from(tx1)).delayElement(Duration.ofMillis(50))
1 * it.read(TxId.from(tx2.hash)) >> Mono.just(TxContainer.from(tx2)).delayElement(Duration.ofMillis(40))
1 * it.read(TxId.from(tx3.hash)) >> Mono.just(TxContainer.from(tx3)).delayElement(Duration.ofMillis(30))
1 * it.read(TxId.from(tx4.hash)) >> Mono.just(TxContainer.from(tx4)).delayElement(Duration.ofMillis(20))
}
def blocks = new BlocksMemCache()
blocks.add(BlockContainer.from(block4))
def full = new EthereumFullBlocksReader(blocks, txes)
when:
def act = full.read(BlockId.from(block4.hash)).block()
act = objectMapper.readValue(act.json, BlockJson)
def transactions = act.transactions
then:
transactions.size() == 4
transactions[0] instanceof TransactionJson
transactions[1] instanceof TransactionJson
transactions[0].hash == TransactionId.from(hash1)
transactions[1].hash == TransactionId.from(hash2)
transactions[2].hash == TransactionId.from(hash3)
transactions[3].hash == TransactionId.from(hash4)
}
def "Read block without transactions"() {
setup:
def txes = new TxMemCache()
def blocks = new BlocksMemCache()
txes.add(TxContainer.from(tx1))
txes.add(TxContainer.from(tx2))
txes.add(TxContainer.from(tx3))
txes.add(TxContainer.from(tx4))
blocks.add(BlockContainer.from(block1))
blocks.add(BlockContainer.from(block2))
blocks.add(BlockContainer.from(block3))
def full = new EthereumFullBlocksReader(blocks, txes)
when:
def act = full.read(BlockId.from(block3.hash)).block()
then:
act != null
act.hash == BlockId.from(hash4)
act.height == 102
act.transactions.size() == 0
}
def "Return nothing if some of tx are unavailable"() {
setup:
def txes = new TxMemCache()
def blocks = new BlocksMemCache()
txes.add(TxContainer.from(tx1))
blocks.add(BlockContainer.from(block1)) //missing tx2 in cache
def full = new EthereumFullBlocksReader(blocks, txes)
when:
def act = full.read(BlockId.from(block1.hash)).block()
then:
act == null
}
def "Return nothing if no block"() {
setup:
def txes = new TxMemCache()
def blocks = new BlocksMemCache()
txes.add(TxContainer.from(tx1))
txes.add(TxContainer.from(tx2))
txes.add(TxContainer.from(tx3))
def full = new EthereumFullBlocksReader(blocks, txes)
when:
def act = full.read(BlockId.from(block1.hash)).block()
then:
act == null
}
def "Doesn't change original cached json"() {
setup:
def txes = new TxMemCache()
def blocks = new BlocksMemCache()
def blockJson = '''
{
"number": "0x100001",
"hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
"timestamp": "0x56cc7b8c",
"totalDifficulty": "0x6baba0399a0f2e73",
"extraField": "extraValue",
"transactions": [
"0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77",
"0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"
],
"extraField2": "extraValue2"
}
'''
blocks.add(BlockContainer.fromEthereumJson(blockJson.bytes, "EthereumFullBlocksReaderSpec"))
def tx1 = '''
{
"hash": "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77",
"blockHash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
"blockNumber": "0x100001",
"extraField3": "extraValue3"
}
'''
def tx2 = '''
{
"hash": "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e",
"blockHash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
"blockNumber": "0x100001",
"from": "0x2a65aca4d5fc5b5c859090a6c34d164135398226",
"extraField4": "extraValue4"
}
'''
txes.add(TxContainer.from(tx1.bytes))
txes.add(TxContainer.from(tx2.bytes))
def full = new EthereumFullBlocksReader(blocks, txes)
def blockJsonExpected = '''
{
"number": "0x100001",
"hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
"timestamp": "0x56cc7b8c",
"totalDifficulty": "0x6baba0399a0f2e73",
"extraField": "extraValue",
"transactions": [
''' + tx1 + ', ' + tx2 +
''' ],
"extraField2": "extraValue2"
}
'''
blockJsonExpected = Global.objectMapper.readValue(blockJsonExpected, Map)
def prettyJson = Global.objectMapper.writer(new DefaultPrettyPrinter())
blockJsonExpected = prettyJson.writeValueAsString(blockJsonExpected)
when:
def act = full.read(BlockId.from("0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446")).block()
act = prettyJson.writeValueAsString(Global.objectMapper.readValue(act.json, Map))
then:
act.contains("extraValue")
act.contains("extraValue2")
act.contains("extraValue3")
act.contains("extraValue4")
act == blockJsonExpected
}
def "Split block with tx"() {
setup:
def blockJson = '''
{
"number": "0x100001",
"hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
"timestamp": "0x56cc7b8c",
"totalDifficulty": "0x6baba0399a0f2e73",
"extraField": "extraValue",
"transactions": [
"0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77",
"0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"
],
"extraField2": "extraValue2"
}
'''
def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader))
when:
def act = reader.splitByTransactions(blockJson.bytes)
then:
new String(act.getT1()).endsWith('"transactions": [')
new String(act.getT2()).startsWith('],')
new String(act.getT2()).trim().endsWith('}')
}
def "Split block with tx if formatted with space"() {
setup:
def blockJson = '''
{
"number": "0x100001",
"hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
"timestamp": "0x56cc7b8c",
"totalDifficulty": "0x6baba0399a0f2e73",
"extraField": "extraValue",
"transactions" : [
"0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77",
"0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"
] ,
"extraField2": "extraValue2"
}
'''
def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader))
when:
def act = reader.splitByTransactions(blockJson.bytes)
then:
new String(act.getT1()).endsWith('"transactions" : [')
new String(act.getT2()).startsWith('] ,')
new String(act.getT2()).trim().endsWith('}')
}
def "Split block with tx if formatted without space"() {
setup:
def blockJson = '{"extraField": "extraValue","transactions":["0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77","0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"]}'
def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader))
when:
def act = reader.splitByTransactions(blockJson.bytes)
then:
new String(act.getT1()).endsWith('"transactions":[')
new String(act.getT2()) == ']}'
}
def "Extract from buffer without trailing zeroes"() {
setup:
def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader))
when:
def buffer = ByteBuffer.allocate(8)
buffer.put(1 as byte)
buffer.put(2 as byte)
def act = reader.extractContent(buffer)
then:
act.size() == 2
Hex.encodeHexString(act) == "0102"
when:
buffer = ByteBuffer.allocate(8)
buffer.put(1 as byte)
buffer.put(2 as byte)
buffer.put(3 as byte)
act = reader.extractContent(buffer)
then:
act.size() == 3
Hex.encodeHexString(act) == "010203"
when:
buffer = ByteBuffer.allocate(8)
buffer.put(1 as byte)
buffer.put(2 as byte)
buffer.put(3 as byte)
buffer.put(4 as byte)
buffer.put(5 as byte)
buffer.put(6 as byte)
buffer.put(7 as byte)
buffer.put(8 as byte)
act = reader.extractContent(buffer)
then:
act.size() == 8
Hex.encodeHexString(act) == "0102030405060708"
}
}

18
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy
View File

@@ -138,4 +138,22 @@ class LocalCallRouterSpec extends Specification {
}
}
}
def "getBlockByNumber skips requests with tx bodies"() {
setup:
def head = Mock(Head)
def reader = Mock(EthereumCachingReader) {
_ * blocksByIdAsCont() >> new EmptyReader<>()
_ * txByHashAsCont() >> new EmptyReader<>()
_ * blocksByHeightAsCont() >> new EmptyReader<>()
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(reader, methods, head, true)
when:
def act = router.getBlockByNumber(["0x0", true])
then:
act == null
}
}