Added fixed channel executor thread pool

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(
+        return if (monitoringConfig.enableExtended)
-            if (monitoringConfig.enableExtended)
+            ExecutorServiceMetrics.monitor(
-                ExecutorServiceMetrics.monitor(
+                Metrics.globalRegistry,
-                    Metrics.globalRegistry,
+                pool,
-                    pool,
+                name,
-                    name,
+                prefix
-                    prefix
+            )
-                )
+        else
-            else
+            pool
-                pool
-        )
     }
 }

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
         }

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

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

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,