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

problem: doesn't use cache for block-by-height requests

Browse Source
This commit is contained in:
Igor Artamonov
2020-02-07 20:21:05 -05:00
parent d69ee41e7a
commit d13397260f
10 changed files with 334 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/main/kotlin/io/emeraldpay/dshackle/cache/BlockByHeight.kt vendored Normal file
View File

@@ -0,0 +1,27 @@
package io.emeraldpay.dshackle.cache
import io.emeraldpay.dshackle.reader.Reader
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
/**
* Connects two caches to read through them. First is cache height->hash, second is hash->block.
*/
class BlockByHeight(
private val heights: Reader<Long, BlockHash>,
private val blocks: Reader<BlockHash, BlockJson<TransactionRefJson>>
): Reader<Long, BlockJson<TransactionRefJson>> {
companion object {
private val log = LoggerFactory.getLogger(BlockByHeight::class.java)
}
override fun read(key: Long): Mono<BlockJson<TransactionRefJson>> {
return heights.read(key)
.flatMap { blocks.read(it) }
}
}

7
src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt vendored
View File

@@ -15,6 +15,7 @@
*/
package io.emeraldpay.dshackle.cache
import io.emeraldpay.dshackle.reader.Reader
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
@@ -25,13 +26,13 @@ import java.util.concurrent.ConcurrentLinkedQueue
class BlocksMemCache(
val maxSize: Int = 64
) {
): Reader<BlockHash, BlockJson<TransactionRefJson>> {
private val mapping = ConcurrentHashMap<BlockHash, BlockJson<TransactionRefJson>>()
private val queue = ConcurrentLinkedQueue<BlockHash>()
fun get(hash: BlockHash): Mono<BlockJson<TransactionRefJson>> {
return Mono.justOrEmpty(mapping[hash])
override fun read(key: BlockHash): Mono<BlockJson<TransactionRefJson>> {
return Mono.justOrEmpty(mapping[key])
}
fun add(block: BlockJson<TransactionRefJson>) {

38
src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt vendored Normal file
View File

@@ -0,0 +1,38 @@
package io.emeraldpay.dshackle.cache
import io.emeraldpay.dshackle.reader.Reader
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.util.concurrent.ConcurrentHashMap
/**
* Memory cache for blocks heights, keeps mapping height->hash.
*/
class HeightCache(
val maxSize: Int = 256
): Reader<Long, BlockHash> {
companion object {
private val log = LoggerFactory.getLogger(HeightCache::class.java)
}
private val heights = ConcurrentHashMap<Long, BlockHash>()
override fun read(key: Long): Mono<BlockHash> {
return Mono.justOrEmpty(heights[key])
}
fun add(block: BlockJson<TransactionRefJson>) {
heights[block.number] = block.hash
// evict old numbers if full
var dropHeight = block.number - maxSize
while (heights.size > maxSize && dropHeight < block.number) {
heights.remove(dropHeight)
dropHeight++
}
}
}

32
src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt
View File

@@ -1,32 +0,0 @@
/**
* Copyright (c) 2019 ETCDEV GmbH
*
* 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.reader
import io.emeraldpay.dshackle.cache.BlocksMemCache
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import reactor.core.publisher.Mono
class BlockCacheReader(
val cache: BlocksMemCache
): Reader<BlockHash, BlockJson<TransactionRefJson>> {
override fun read(key: BlockHash): Mono<BlockJson<TransactionRefJson>> {
return cache.get(key)
}
}

22
src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt
View File

@@ -16,18 +16,11 @@
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.BlockByHeight
import io.emeraldpay.dshackle.cache.BlocksMemCache
import io.emeraldpay.dshackle.cache.HeightCache
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.BlockCacheReader
import io.emeraldpay.dshackle.reader.CompoundReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.reactivestreams.Publisher
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import reactor.core.publisher.Flux
@@ -42,11 +35,9 @@ abstract class AggregatedUpstream(
val objectMapper: ObjectMapper
): Upstream, Lifecycle {
private val blocksCache = BlocksMemCache()
private var cacheSubscription: Disposable? = null
private val blockReader: Reader<BlockHash, BlockJson<TransactionRefJson>> = CompoundReader(
listOf(BlockCacheReader(blocksCache))
)
private val blockReaderByHash = BlocksMemCache()
private val blockReaderByHeight = HeightCache()
var cache: CachingEthereumApi = CachingEthereumApi.empty()
private val reconfigLock = ReentrantLock()
private var callMethods: CallMethods? = null
@@ -118,9 +109,10 @@ abstract class AggregatedUpstream(
reconfigLock.withLock {
cacheSubscription?.dispose()
cacheSubscription = head.getFlux().subscribe {
blocksCache.add(it)
blockReaderByHash.add(it)
blockReaderByHeight.add(it)
}
cache = CachingEthereumApi(objectMapper, blockReader, head)
cache = CachingEthereumApi(objectMapper, blockReaderByHash, BlockByHeight(blockReaderByHeight, blockReaderByHash), head)
}
}

19
src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt
View File

@@ -29,11 +29,13 @@ import io.infinitape.etherjar.rpc.json.ResponseJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.math.BigInteger
import java.util.function.Function
open class CachingEthereumApi(
private val objectMapper: ObjectMapper,
private val cache: Reader<BlockHash, BlockJson<TransactionRefJson>>,
private val cacheHeight: Reader<Long, BlockJson<TransactionRefJson>>,
private val head: EthereumHead
): EthereumApi(objectMapper) {
@@ -42,7 +44,7 @@ open class CachingEthereumApi(
@JvmStatic
fun empty(): CachingEthereumApi {
return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyEthereumHead())
return CachingEthereumApi(ObjectMapper(), EmptyReader(), EmptyReader(), EmptyEthereumHead())
}
}
@@ -63,6 +65,21 @@ open class CachingEthereumApi(
Mono.empty()
}
else Mono.empty()
"eth_getBlockByNumber" ->
if (params.size == 2 && (params[1] == "false" || params[1] == false))
Mono.just(params[0])
.map { HexQuantity.from(it as String) }
.filter {
it.value < BigInteger.valueOf(Long.MAX_VALUE)
}
.map { it.value.toLong() }
.flatMap(cacheHeight::read)
.map(toJson(id))
.onErrorResume { t ->
log.warn("Error during read from cache", t)
Mono.empty()
}
else Mono.empty()
else ->
Mono.empty()
}

138
src/test/groovy/io/emeraldpay/dshackle/cache/BlockByHeightSpec.groovy vendored Normal file
View File

@@ -0,0 +1,138 @@
package io.emeraldpay.dshackle.cache
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import spock.lang.Specification
class BlockByHeightSpec extends Specification {
String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33"
def "Fetch with all data available"() {
setup:
def blocks = new BlocksMemCache()
def heights = new HeightCache()
def block = new BlockJson<TransactionRefJson>()
block.number = 100
block.hash = BlockHash.from(hash1)
blocks.add(block)
heights.add(block)
def blocksByHeight = new BlockByHeight(heights, blocks)
when:
def act = blocksByHeight.read(100).block()
then:
act == block
}
def "Fetch correct blocks if multiple"() {
setup:
def blocks = new BlocksMemCache()
def heights = new HeightCache()
def block1 = new BlockJson<TransactionRefJson>()
block1.number = 100
block1.hash = BlockHash.from(hash1)
def block2 = new BlockJson<TransactionRefJson>()
block2.number = 101
block2.hash = BlockHash.from(hash2)
blocks.add(block1)
heights.add(block1)
blocks.add(block2)
heights.add(block2)
def blocksByHeight = new BlockByHeight(heights, blocks)
when:
def act = blocksByHeight.read(100).block()
then:
act == block1
when:
act = blocksByHeight.read(101).block()
then:
act == block2
}
def "Fetch last block if updated"() {
setup:
def blocks = new BlocksMemCache()
def heights = new HeightCache()
def block1 = new BlockJson<TransactionRefJson>()
block1.number = 100
block1.hash = BlockHash.from(hash1)
def block2 = new BlockJson<TransactionRefJson>()
block2.number = 100
block2.hash = BlockHash.from(hash2)
blocks.add(block1)
heights.add(block1)
blocks.add(block2)
heights.add(block2)
def blocksByHeight = new BlockByHeight(heights, blocks)
when:
def act = blocksByHeight.read(100).block()
then:
act == block2
}
def "Fetch nothing if block expired"() {
setup:
def blocks = new BlocksMemCache()
def heights = new HeightCache()
def block = new BlockJson<TransactionRefJson>()
block.number = 100
block.hash = BlockHash.from(hash1)
// add only to heights
heights.add(block)
def blocksByHeight = new BlockByHeight(heights, blocks)
when:
def act = blocksByHeight.read(100).block()
then:
act == null
}
def "Fetch nothing if height expired"() {
setup:
def blocks = new BlocksMemCache()
def heights = new HeightCache()
def block = new BlockJson<TransactionRefJson>()
block.number = 100
block.hash = BlockHash.from(hash1)
// add only to blocks
blocks.add(block)
def blocksByHeight = new BlockByHeight(heights, blocks)
when:
def act = blocksByHeight.read(100).block()
then:
act == null
}
def "Fetch nothing if both empty"() {
setup:
def blocks = new BlocksMemCache()
def heights = new HeightCache()
def blocksByHeight = new BlockByHeight(heights, blocks)
when:
def act = blocksByHeight.read(100).block()
then:
act == null
}
}

16
src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy vendored
View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.cache
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import spock.lang.Specification
class BlocksMemCacheSpec extends Specification {
@@ -30,13 +31,13 @@ class BlocksMemCacheSpec extends Specification {
def "Add and read"() {
setup:
def cache = new BlocksMemCache()
def block = new BlockJson<TransactionId>()
def block = new BlockJson<TransactionRefJson>()
block.number = 100
block.hash = BlockHash.from(hash1)
when:
cache.add(block)
def act = cache.get(BlockHash.from(hash1)).block()
def act = cache.read(BlockHash.from(hash1)).block()
then:
act == block
}
@@ -48,20 +49,21 @@ class BlocksMemCacheSpec extends Specification {
when:
[hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i ->
def block = new BlockJson<TransactionId>()
def block = new BlockJson<TransactionRefJson>()
block.number = 100 + i
block.hash = BlockHash.from(hash)
cache.add(block)
}
def act1 = cache.get(BlockHash.from(hash1)).block()
def act2 = cache.get(BlockHash.from(hash2)).block()
def act3 = cache.get(BlockHash.from(hash3)).block()
def act4 = cache.get(BlockHash.from(hash4)).block()
def act1 = cache.read(BlockHash.from(hash1)).block()
def act2 = cache.read(BlockHash.from(hash2)).block()
def act3 = cache.read(BlockHash.from(hash3)).block()
def act4 = cache.read(BlockHash.from(hash4)).block()
then:
act2.hash.toHex() == hash2
act3.hash.toHex() == hash3
act4.hash.toHex() == hash4
act1 == null
}
}

61
src/test/groovy/io/emeraldpay/dshackle/cache/HeightCacheSpec.groovy vendored Normal file
View File

@@ -0,0 +1,61 @@
package io.emeraldpay.dshackle.cache
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import spock.lang.Specification
class HeightCacheSpec extends Specification {
String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33"
String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
def "Add and read"() {
setup:
def cache = new HeightCache()
when:
[hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i ->
def block = new BlockJson<TransactionRefJson>()
block.number = 100 + i
block.hash = BlockHash.from(hash)
cache.add(block)
}
def act1 = cache.read(100).block()
def act2 = cache.read(101).block()
def act3 = cache.read(102).block()
def act4 = cache.read(103).block()
then:
act1.toHex() == hash1
act2.toHex() == hash2
act3.toHex() == hash3
act4.toHex() == hash4
}
def "Keeps only configured amount"() {
setup:
def cache = new HeightCache(3)
[hash1]
when:
[hash1, hash2, hash3, hash4].eachWithIndex{ String hash, int i ->
def block = new BlockJson<TransactionRefJson>()
block.number = 100 + i
block.hash = BlockHash.from(hash)
cache.add(block)
}
def act1 = cache.read(100).block()
def act2 = cache.read(101).block()
def act3 = cache.read(102).block()
def act4 = cache.read(103).block()
then:
act1 == null
act2.toHex() == hash2
act3.toHex() == hash3
act4.toHex() == hash4
}
}

35
src/test/groovy/io/emeraldpay/dshackle/upstream/CachingEthereumApiSpec.groovy
View File

@@ -1,7 +1,8 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.cache.BlockByHeight
import io.emeraldpay.dshackle.cache.BlocksMemCache
import io.emeraldpay.dshackle.reader.BlockCacheReader
import io.emeraldpay.dshackle.cache.HeightCache
import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
@@ -23,6 +24,7 @@ class CachingEthereumApiSpec extends Specification {
def api = new CachingEthereumApi(
TestingCommons.objectMapper(),
new EmptyReader<BlockHash, BlockJson<TransactionRefJson>>(),
new EmptyReader<>(),
head
)
1 * head.getFlux() >> Flux.just(new BlockJson<TransactionRefJson>(number: 100))
@@ -42,6 +44,7 @@ class CachingEthereumApiSpec extends Specification {
def api = new CachingEthereumApi(
TestingCommons.objectMapper(),
new EmptyReader<BlockHash, BlockJson<TransactionRefJson>>(),
new EmptyReader<>(),
head
)
when:
@@ -53,13 +56,14 @@ class CachingEthereumApiSpec extends Specification {
.verify(Duration.ofSeconds(3))
}
def "Return block when cached"() {
def "Return block by hash when cached"() {
setup:
def cache = new BlocksMemCache();
def head = Mock(EthereumHead.class)
def api = new CachingEthereumApi(
TestingCommons.objectMapper(),
new BlockCacheReader(cache),
cache,
new EmptyReader<>(),
head
)
cache.add(new BlockJson<TransactionRefJson>(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58")))
@@ -74,4 +78,29 @@ class CachingEthereumApiSpec extends Specification {
.verify(Duration.ofSeconds(3))
}
def "Return block by height when cached"() {
setup:
def blocksCache = new BlocksMemCache()
def heightCache = new HeightCache()
def head = Mock(EthereumHead.class)
def api = new CachingEthereumApi(
TestingCommons.objectMapper(),
blocksCache,
new BlockByHeight(heightCache, blocksCache),
head
)
def block = new BlockJson<TransactionRefJson>(number: 100, hash: BlockHash.from("0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58"))
heightCache.add(block)
blocksCache.add(block)
when:
def act = api.execute(1, "eth_getBlockByNumber", ["0x64", false]).map { new String(it)}
then:
StepVerifier.create(act)
.expectNext('{"jsonrpc":"2.0","id":1,"result":{"number":"0x64","hash":"0x5b4590a9905fa1c9cc273f32e6dc63b4c512f0ee14edc6fa41c26b416a7b5d58","transactions":[],"uncles":[]}}')
.expectComplete()
.verify(Duration.ofSeconds(3))
}
}