diff --git a/src/main/kotlin/io/emeraldpay/dshackle/commons/DynamicMergeFlux.kt b/src/main/kotlin/io/emeraldpay/dshackle/commons/DynamicMergeFlux.kt index 4ca80a8a..0a51b89d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/commons/DynamicMergeFlux.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/commons/DynamicMergeFlux.kt @@ -8,7 +8,7 @@ import java.util.concurrent.ConcurrentHashMap class DynamicMergeFlux(private val scheduler: Scheduler) { - private val merge = Sinks.many().multicast().onBackpressureBuffer() + private val merge = Sinks.many().multicast().directBestEffort() private val sources = ConcurrentHashMap() fun add(flux: Flux, id: K) { @@ -30,4 +30,6 @@ class DynamicMergeFlux(private val scheduler: Scheduler) { sources.forEach { (_, d) -> d.dispose() } merge.emitComplete { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } } + + fun getKeys(): List = sources.keys().toList() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt index c3925848..63a43b18 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/MultistreamsConfig.kt @@ -8,23 +8,27 @@ import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream +import org.springframework.beans.factory.annotation.Qualifier import org.springframework.beans.factory.config.ConfigurableListableBeanFactory import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration +import reactor.core.scheduler.Scheduler @Configuration open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) { @Bean open fun allMultistreams( cachesFactory: CachesFactory, - callTargetsHolder: CallTargetsHolder + callTargetsHolder: CallTargetsHolder, + @Qualifier("headMergedScheduler") + headScheduler: Scheduler ): List { return Chain.values() .filterNot { it == Chain.UNSPECIFIED } .mapNotNull { chain -> when (BlockchainType.from(chain)) { - BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory) - BlockchainType.EVM_POW -> ethereumMultistream(chain, cachesFactory) + BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory, headScheduler) + BlockchainType.EVM_POW -> ethereumMultistream(chain, cachesFactory, headScheduler) BlockchainType.BITCOIN -> bitcoinMultistream(chain, cachesFactory) else -> null } @@ -33,27 +37,31 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory) private fun ethereumMultistream( chain: Chain, - cachesFactory: CachesFactory + cachesFactory: CachesFactory, + headScheduler: Scheduler ): EthereumMultistream { val name = "multi-ethereum-$chain" return EthereumMultistream( chain, ArrayList(), - cachesFactory.getCaches(chain) + cachesFactory.getCaches(chain), + headScheduler ).also { register(it, name) } } open fun ethereumPosMultistream( chain: Chain, - cachesFactory: CachesFactory + cachesFactory: CachesFactory, + headScheduler: Scheduler ): EthereumPosMultiStream { val name = "multi-ethereum-pos-$chain" return EthereumPosMultiStream( chain, ArrayList(), - cachesFactory.getCaches(chain) + cachesFactory.getCaches(chain), + headScheduler ).also { register(it, name) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt index 4faf409b..a4ec2a0c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt @@ -22,6 +22,11 @@ open class SchedulersConfig { return makeScheduler("tracktx-scheduler", "tracktx", 5, monitoringConfig) } + @Bean + open fun headMergedScheduler(monitoringConfig: MonitoringConfig): Scheduler { + return makeScheduler("head-scheduler", "head_merge", 5, monitoringConfig) + } + private fun makeScheduler(name: String, prefix: String, size: Int, monitoringConfig: MonitoringConfig): Scheduler { val pool = Executors.newFixedThreadPool(size, CustomizableThreadFactory("$name-%d")) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt index 8d9871cb..967bd39b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DynamicMergedHead.kt @@ -40,10 +40,12 @@ open class DynamicMergedHead( } fun addHead(upstream: Upstream) { + log.debug("adding upstream head of [${upstream.getId()}] to dynamic head of [$label]. Current heads ${dynamicFlux.getKeys()}") dynamicFlux.add(upstream.getHead().getFlux(), upstream.getId()) } fun removeHead(id: String) { + log.debug("removing upstream head of [$id] from dynamic head $label. Current heads ${dynamicFlux.getKeys()}") dynamicFlux.remove(id) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 136d8db3..2ae8cebd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -32,13 +32,14 @@ import org.slf4j.LoggerFactory import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.scheduler.Schedulers +import reactor.core.scheduler.Scheduler @Suppress("UNCHECKED_CAST") open class EthereumMultistream( chain: Chain, val upstreams: MutableList, - caches: Caches + caches: Caches, + headScheduler: Scheduler ) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { companion object { @@ -48,7 +49,7 @@ open class EthereumMultistream( private var head: DynamicMergedHead = DynamicMergedHead( PriorityForkChoice(), "ETH Multistream", - Schedulers.boundedElastic() + headScheduler ) private val filteredHeads: MutableMap = diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index 81d1e55a..e9aa7d38 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -31,13 +31,14 @@ import org.slf4j.LoggerFactory import org.springframework.util.ConcurrentReferenceHashMap import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.scheduler.Schedulers +import reactor.core.scheduler.Scheduler @Suppress("UNCHECKED_CAST") open class EthereumPosMultiStream( chain: Chain, val upstreams: MutableList, - caches: Caches + caches: Caches, + headScheduler: Scheduler ) : Multistream(chain, upstreams as MutableList, caches), EthereumLikeMultistream { companion object { @@ -47,7 +48,7 @@ open class EthereumPosMultiStream( private var head: DynamicMergedHead = DynamicMergedHead( PriorityForkChoice(), "ETH Pos Multistream", - Schedulers.boundedElastic() + headScheduler ) private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory()) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 56942007..f94f8d4e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -34,6 +34,7 @@ import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Chain import org.jetbrains.annotations.NotNull +import reactor.core.scheduler.Schedulers class MultistreamHolderMock implements MultistreamHolder { @@ -50,7 +51,7 @@ class MultistreamHolderMock implements MultistreamHolder { if (up instanceof EthereumPosMultiStream) { upstreams[chain] = up } else if (up instanceof EthereumPosRpcUpstream) { - upstreams[chain] = new EthereumPosMultiStream(chain, [up as EthereumPosRpcUpstream], Caches.default()) + upstreams[chain] = new EthereumPosMultiStream(chain, [up as EthereumPosRpcUpstream], Caches.default(), Schedulers.boundedElastic()) } else { throw new IllegalArgumentException("Unsupported upstream type ${up.class}") } @@ -98,7 +99,7 @@ class MultistreamHolderMock implements MultistreamHolder { Head customHead = null EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches) + super(chain, upstreams, caches, Schedulers.boundedElastic()) } EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index b7cd076c..fd9ec67a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -38,6 +38,7 @@ import io.emeraldpay.dshackle.Chain import io.micrometer.core.instrument.MeterRegistry import io.micrometer.core.instrument.logging.LoggingMeterRegistry import org.apache.commons.lang3.StringUtils +import reactor.core.scheduler.Schedulers import java.time.Instant @@ -84,7 +85,7 @@ class TestingCommons { } static Multistream multistream(EthereumPosRpcUpstreamMock up) { - return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default()).tap { + return new EthereumPosMultiStream(Chain.ETHEREUM, [up], Caches.default(), Schedulers.boundedElastic()).tap { start() } } @@ -105,11 +106,11 @@ class TestingCommons { } static Multistream multistreamWithoutUpstreams(Chain chain) { - return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain)) + return new EthereumPosMultiStream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic()) } static Multistream multistreamClassicWithoutUpstreams(Chain chain) { - return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain)) + return new EthereumMultistream(chain, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic()) } static FileResolver fileResolver() { @@ -148,6 +149,4 @@ class TestingCommons { } static MeterRegistry meterRegistry = new LoggingMeterRegistry() - - static CallTargetsHolder callTargetsHolder = new CallTargetsHolder() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index 4a2b19b1..3c90ca15 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -37,6 +37,7 @@ import io.emeraldpay.dshackle.Chain import org.jetbrains.annotations.NotNull import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import spock.lang.Specification @@ -50,7 +51,7 @@ class MultistreamSpec extends Specification { setup: def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) def up2 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"])) - def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default()) + def aggr = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2], Caches.default(), Schedulers.boundedElastic()) when: aggr.onUpstreamsUpdated() def act = aggr.getMethods() @@ -186,7 +187,7 @@ class MultistreamSpec extends Specification { def up1 = TestingCommons.upstream("test-1", "internal") def up2 = TestingCommons.upstream("test-2", "external") def up3 = TestingCommons.upstream("test-3", "external") - def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default()) + def multistream = new EthereumPosMultiStream(Chain.ETHEREUM, [up1, up2, up3], Caches.default(), Schedulers.boundedElastic()) expect: multistream.getHead(new Selector.LabelMatcher("provider", ["internal"])).is(up1.ethereumHeadMock) @@ -254,7 +255,7 @@ class MultistreamSpec extends Specification { class TestEthereumPosMultistream extends EthereumPosMultiStream { TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { - super(chain, upstreams, caches) + super(chain, upstreams, caches, Schedulers.boundedElastic()) } @NotNull