From ed82aa373a4ddfd049dc75424b2a5bb85d778207 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Tue, 17 Jan 2023 19:55:07 +0400 Subject: [PATCH] Added fixed channel executor thread pool --- .../config/context/SchedulersConfig.kt | 31 ++++++++++++------- .../dshackle/startup/ConfiguredUpstreams.kt | 9 ++++-- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 2 -- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 5 ++- .../startup/ConfiguredUpstreamsSpec.groovy | 17 +++++++--- 5 files changed, 43 insertions(+), 21 deletions(-) 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 1bfc04a1..b0758f64 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt @@ -8,6 +8,8 @@ import org.springframework.context.annotation.Configuration import org.springframework.scheduling.concurrent.CustomizableThreadFactory import reactor.core.scheduler.Scheduler import reactor.core.scheduler.Schedulers +import java.util.concurrent.Executor +import java.util.concurrent.ExecutorService import java.util.concurrent.Executors @Configuration @@ -27,19 +29,26 @@ open class SchedulersConfig { return makeScheduler("head-scheduler", "head_merge", 5, monitoringConfig) } + @Bean + open fun grpcChannelExecutor(monitoringConfig: MonitoringConfig): Executor { + return makePool("grpc-client-channel", "grpc_client_channel", 10, monitoringConfig) + } + private fun makeScheduler(name: String, prefix: String, size: Int, monitoringConfig: MonitoringConfig): Scheduler { + return Schedulers.fromExecutorService(makePool(name, prefix, size, monitoringConfig)) + } + + private fun makePool(name: String, prefix: String, size: Int, monitoringConfig: MonitoringConfig): ExecutorService { val pool = Executors.newFixedThreadPool(size, CustomizableThreadFactory("$name-")) - return Schedulers.fromExecutorService( - if (monitoringConfig.enableExtended) - ExecutorServiceMetrics.monitor( - Metrics.globalRegistry, - pool, - name, - prefix - ) - else - pool - ) + return if (monitoringConfig.enableExtended) + ExecutorServiceMetrics.monitor( + Metrics.globalRegistry, + pool, + name, + prefix + ) + else + pool } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index d612bfab..c84b8637 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -42,6 +42,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Qualifier import org.springframework.boot.ApplicationArguments import org.springframework.boot.ApplicationRunner import org.springframework.context.ApplicationEventPublisher @@ -50,6 +51,7 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.scheduler.Schedulers import java.net.URI +import java.util.concurrent.Executor import java.util.concurrent.atomic.AtomicInteger import java.util.function.Function import kotlin.math.abs @@ -59,7 +61,9 @@ open class ConfiguredUpstreams( private val fileResolver: FileResolver, private val config: UpstreamsConfig, private val callTargets: CallTargetsHolder, - private val eventPublisher: ApplicationEventPublisher + private val eventPublisher: ApplicationEventPublisher, + @Qualifier("grpcChannelExecutor") + private val channelExecutor: Executor ) : ApplicationRunner { private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) @@ -308,7 +312,8 @@ open class ConfiguredUpstreams( endpoint.auth, fileResolver, endpoint.upstreamRating, - config.labels + config.labels, + channelExecutor ).apply { timeout = options.timeout } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index 6195f69e..142c3dd1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -40,7 +40,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.rpc.RpcException -import io.grpc.ManagedChannel import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import reactor.core.publisher.Flux @@ -120,7 +119,6 @@ open class EthereumPosGrpcUpstream( } override fun stop() { - } override fun update(conf: BlockchainOuterClass.DescribeChain) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 1d5cbc36..30485f60 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -45,6 +45,7 @@ import reactor.core.Disposable import reactor.core.publisher.Flux import java.io.IOException import java.time.Duration +import java.util.concurrent.Executor import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock @@ -58,7 +59,8 @@ class GrpcUpstreams( private val auth: AuthConfig.ClientTlsAuth? = null, private val fileResolver: FileResolver, private val nodeRating: Int, - private val labels: UpstreamsConfig.Labels + private val labels: UpstreamsConfig.Labels, + private val executor: Executor ) { private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) @@ -73,6 +75,7 @@ class GrpcUpstreams( // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces) .maxInboundMessageSize(Defaults.maxMessageSize) .enableRetry() + .executor(executor) .maxRetryAttempts(3) if (auth != null && StringUtils.isNotEmpty(auth.ca)) { chanelBuilder diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index ec3cd2da..870d9430 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy @@ -11,6 +11,8 @@ import io.emeraldpay.dshackle.Chain import org.springframework.context.ApplicationEventPublisher import spock.lang.Specification +import java.util.concurrent.Executors + class ConfiguredUpstreamsSpec extends Specification { def "Applied quorum to extra methods"() { @@ -20,7 +22,8 @@ class ConfiguredUpstreamsSpec extends Specification { Stub(FileResolver), Stub(UpstreamsConfig), callTargetsHolder, - Mock(ApplicationEventPublisher) + Mock(ApplicationEventPublisher), + Executors.newFixedThreadPool(1) ) def methods = new UpstreamsConfig.Methods( [ @@ -45,7 +48,8 @@ class ConfiguredUpstreamsSpec extends Specification { Stub(FileResolver), Stub(UpstreamsConfig), callTargetsHolder, - Mock(ApplicationEventPublisher) + Mock(ApplicationEventPublisher), + Executors.newFixedThreadPool(1) ) def methods = new UpstreamsConfig.Methods( [ @@ -69,7 +73,8 @@ class ConfiguredUpstreamsSpec extends Specification { Stub(FileResolver), Stub(UpstreamsConfig), callTargetsHolder, - Mock(ApplicationEventPublisher) + Mock(ApplicationEventPublisher), + Executors.newFixedThreadPool(1) ) expect: configurer.getHash(node, src) == expected @@ -88,7 +93,8 @@ class ConfiguredUpstreamsSpec extends Specification { Stub(FileResolver), Stub(UpstreamsConfig), callTargetsHolder, - Mock(ApplicationEventPublisher) + Mock(ApplicationEventPublisher), + Executors.newFixedThreadPool(1) ) when: def h1 = configurer.getHash(null, "hohoho") @@ -112,7 +118,8 @@ class ConfiguredUpstreamsSpec extends Specification { Stub(FileResolver), Stub(UpstreamsConfig), callTargetsHolder, - Mock(ApplicationEventPublisher) + Mock(ApplicationEventPublisher), + Executors.newFixedThreadPool(1) ) def methodsGroup = new UpstreamsConfig.MethodGroups( ["filter"] as Set,