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

problem: doesn't use tx cache when requesting full block by number

Browse Source
This commit is contained in:
Igor Artamonov
2020-07-20 20:36:01 -04:00
parent d3a3b6606a
commit 3dd105eb1c
4 changed files with 243 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt
View File

@@ -28,6 +28,12 @@ class TxContainer(
) : SourceContainer(json, parsed) {
companion object {
@JvmStatic
fun from(raw: ByteArray): TxContainer {
val tx = Global.objectMapper.readValue(raw, TransactionJson::class.java)
return from(tx, raw)
}
@JvmStatic
fun from(tx: TransactionJson): TxContainer {
return from(tx, Global.objectMapper.writeValueAsBytes(tx))

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

@@ -15,20 +15,18 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
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.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import org.springframework.beans.BeanUtils
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
@@ -46,40 +44,91 @@ class EthereumFullBlocksReader(
private val log = LoggerFactory.getLogger(EthereumFullBlocksReader::class.java)
}
override fun read(key: BlockId): Mono<BlockContainer> {
return blocks.read(key).flatMap { block ->
val block = Global.objectMapper.readValue(block.json, BlockJson::class.java) as BlockJson<TransactionRefJson>
val fullBlock = if (block.transactions == null || 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 = BlockJson<TransactionJson>()
BeanUtils.copyProperties(block, fullBlock)
Mono.just(fullBlock)
} else {
Flux.fromIterable(block.transactions)
.map { TxId.from(it.hash) }
.flatMap { txes.read(it) }
.collectList()
.flatMap { list ->
if (block.transactions.size != list.size) {
Mono.empty<BlockJson<TransactionJson>>()
} else {
val fullBlock = BlockJson<TransactionJson>()
BeanUtils.copyProperties(block, fullBlock)
fullBlock.transactions = list.map {
Global.objectMapper.readValue(it.json, TransactionJson::class.java)
}
Mono.just(fullBlock)
}
}
}
fullBlock
.map { block ->
BlockContainer(block.number, BlockId.from(block.hash), block.totalDifficulty, block.timestamp, true,
Global.objectMapper.writeValueAsBytes(block),
block.transactions.map { tx -> TxId.from(tx) }
)
}
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)
.flatMap { 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 { it.flip().array() }
.map { json ->
BlockContainer(block.height, block.hash, block.difficulty, block.timestamp,
true,
json,
null,
block.transactions
)
}
}
}
}
}
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)
)
}
}

12
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt
View File

@@ -158,11 +158,15 @@ class NativeCallRouter(
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number")
}
val withTx = params[1].toString().toBoolean()
return if (withTx) {
log.warn("Block by number is not implemented")
null
var block = reader.blocksByHeightAsCont()
.read(number)
block = if (withTx) {
block.flatMap {
fullBlocksReader.read(it.hash)
}
} else {
reader.blocksByHeightAsCont().read(number).map { it.json!! }
block
}
return block.map { it.json!! }
}
}

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

@@ -15,6 +15,8 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.core.PrettyPrinter
import com.fasterxml.jackson.core.util.DefaultPrettyPrinter
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.BlocksMemCache
@@ -22,6 +24,8 @@ 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.reader.Reader
import io.emeraldpay.dshackle.test.ReaderMock
import io.emeraldpay.dshackle.test.TestingCommons
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
@@ -200,7 +204,7 @@ class EthereumFullBlocksReaderSpec extends Specification {
act.transactions.size() == 0
}
def "Return nothing if no transactions"() {
def "Return nothing if some of tx are unavailable"() {
setup:
def txes = new TxMemCache()
def blocks = new BlocksMemCache()
@@ -234,4 +238,140 @@ class EthereumFullBlocksReaderSpec extends Specification {
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.from(blockJson.bytes))
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()) == ']}'
}
}