Merge pull request #111 from p2p-org/fix-extra-cpu-usage-1

Small improvements after dynamic head introduction

src/main/kotlin/io/emeraldpay/dshackle/commons/DynamicMergeFlux.kt

@@ -8,7 +8,7 @@ import java.util.concurrent.ConcurrentHashMap
 
 class DynamicMergeFlux<K : Any, T>(private val scheduler: Scheduler) {
 
-    private val merge = Sinks.many().multicast().onBackpressureBuffer<T>()
+    private val merge = Sinks.many().multicast().directBestEffort<T>()
     private val sources = ConcurrentHashMap<K, Disposable>()
 
     fun add(flux: Flux<T>, id: K) {
@@ -30,4 +30,6 @@ class DynamicMergeFlux<K : Any, T>(private val scheduler: Scheduler) {
         sources.forEach { (_, d) -> d.dispose() }
         merge.emitComplete { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
     }
+
+    fun getKeys(): List<K> = sources.keys().toList()
 }

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<Multistream> {
         return Chain.values()
             .filterNot { it == Chain.UNSPECIFIED }
             .mapNotNull { chain ->
                 when (BlockchainType.from(chain)) {
-                    BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory)
+                    BlockchainType.EVM_POS -> ethereumPosMultistream(chain, cachesFactory, headScheduler)
-                    BlockchainType.EVM_POW -> ethereumMultistream(chain, cachesFactory)
+                    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) }
     }
 

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"))
 

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)
     }
 }

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<EthereumUpstream>,
-    caches: Caches
+    caches: Caches,
+    headScheduler: Scheduler
 ) : Multistream(chain, upstreams as MutableList<Upstream>, 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<String, Head> =

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<EthereumPosUpstream>,
-    caches: Caches
+    caches: Caches,
+    headScheduler: Scheduler
 ) : Multistream(chain, upstreams as MutableList<Upstream>, 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())

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<EthereumPosRpcUpstream> upstreams, @NotNull Caches caches) {
-            super(chain, upstreams, caches)
+            super(chain, upstreams, caches, Schedulers.boundedElastic())
         }
 
         EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumPosRpcUpstream> upstreams) {

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()
 }

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<EthereumPosUpstream> upstreams, @NotNull Caches caches) {
-            super(chain, upstreams, caches)
+            super(chain, upstreams, caches, Schedulers.boundedElastic())
         }
 
         @NotNull