diff --git a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt index 702a220e..2503a3fa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt @@ -16,7 +16,6 @@ */ package io.emeraldpay.dshackle -import com.google.common.util.concurrent.ThreadFactoryBuilder import io.emeraldpay.dshackle.config.MainConfig import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerGrpc import io.grpc.Server @@ -24,7 +23,7 @@ import io.grpc.netty.NettyServerBuilder import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics import org.slf4j.LoggerFactory -import org.springframework.beans.factory.annotation.Autowired +import org.springframework.scheduling.concurrent.CustomizableThreadFactory import org.springframework.stereotype.Service import java.net.InetSocketAddress import java.util.concurrent.Executors @@ -33,10 +32,10 @@ import javax.annotation.PreDestroy @Service open class GrpcServer( - @Autowired val rpcs: List, - @Autowired val mainConfig: MainConfig, - @Autowired val tlsSetup: TlsSetup, - @Autowired val accessHandler: AccessHandlerGrpc + private val rpcs: List, + private val mainConfig: MainConfig, + private val tlsSetup: TlsSetup, + private val accessHandler: AccessHandlerGrpc ) { private val log = LoggerFactory.getLogger(GrpcServer::class.java) @@ -67,7 +66,7 @@ open class GrpcServer( serverBuilder.addService(it) } - val pool = Executors.newFixedThreadPool(20, ThreadFactoryBuilder().setNameFormat("fixed-grpc-%d").build()) + val pool = Executors.newFixedThreadPool(20, CustomizableThreadFactory("fixed-grpc-%d")) serverBuilder.executor( if (mainConfig.monitoring.enableExtended) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt new file mode 100644 index 00000000..4faf409b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt @@ -0,0 +1,40 @@ +package io.emeraldpay.dshackle.config.context + +import io.emeraldpay.dshackle.config.MonitoringConfig +import io.micrometer.core.instrument.Metrics +import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics +import org.springframework.context.annotation.Bean +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.Executors + +@Configuration +open class SchedulersConfig { + @Bean + open fun rpcScheduler(monitoringConfig: MonitoringConfig): Scheduler { + return makeScheduler("blockchain-rpc-scheduler", "blockchain_rpc", 30, monitoringConfig) + } + + @Bean + open fun trackTxScheduler(monitoringConfig: MonitoringConfig): Scheduler { + return makeScheduler("tracktx-scheduler", "tracktx", 5, monitoringConfig) + } + + private fun makeScheduler(name: String, prefix: String, size: Int, monitoringConfig: MonitoringConfig): Scheduler { + val pool = Executors.newFixedThreadPool(size, CustomizableThreadFactory("$name-%d")) + + return Schedulers.fromExecutorService( + if (monitoringConfig.enableExtended) + ExecutorServiceMetrics.monitor( + Metrics.globalRegistry, + pool, + name, + prefix + ) + else + pool + ) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 13c46b26..d3321b46 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -26,11 +26,12 @@ import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.Timer import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Qualifier import org.springframework.context.annotation.DependsOn import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono -import reactor.core.scheduler.Schedulers +import reactor.core.scheduler.Scheduler import java.util.Locale import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.TimeUnit @@ -46,6 +47,8 @@ class BlockchainRpc( private val describe: Describe, private val subscribeStatus: SubscribeStatus, private val estimateFee: EstimateFee, + @Qualifier("rpcScheduler") + private val scheduler: Scheduler ) : ReactorBlockchainGrpc.BlockchainImplBase() { private val log = LoggerFactory.getLogger(BlockchainRpc::class.java) @@ -69,7 +72,7 @@ class BlockchainRpc( val idsMap = mutableMapOf() return nativeCall.nativeCall( request - .subscribeOn(Schedulers.boundedElastic()) + .subscribeOn(scheduler) .doOnNext { req -> metrics = chainMetrics.get(req.chain) metrics?.let { m -> @@ -115,7 +118,7 @@ class BlockchainRpc( } override fun subscribeTxStatus(requestMono: Mono): Flux { - return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request -> + return requestMono.subscribeOn(scheduler).flatMapMany { request -> val chain = Chain.byId(request.chainValue) val metrics = chainMetrics.get(chain) metrics.subscribeTxMetric.increment() @@ -128,13 +131,13 @@ class BlockchainRpc( } catch (t: Throwable) { log.error("Internal error during Tx Subscription", t) failMetric.increment() - Flux.error(IllegalStateException("Internal Error")) + Flux.error(IllegalStateException("Internal Error")) } } } override fun subscribeBalance(requestMono: Mono): Flux { - return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request -> + return requestMono.subscribeOn(scheduler).flatMapMany { request -> val chain = Chain.byId(request.asset.chainValue) val metrics = chainMetrics.get(chain) metrics.subscribeBalanceMetric.increment() @@ -151,13 +154,13 @@ class BlockchainRpc( } catch (t: Throwable) { log.error("Internal error during Balance Subscription", t) failMetric.increment() - Flux.error(IllegalStateException("Internal Error")) + Flux.error(IllegalStateException("Internal Error")) } } } override fun getBalance(requestMono: Mono): Flux { - return requestMono.subscribeOn(Schedulers.boundedElastic()).flatMapMany { request -> + return requestMono.subscribeOn(scheduler).flatMapMany { request -> val chain = Chain.byId(request.asset.chainValue) val metrics = chainMetrics.get(chain) metrics.getBalanceMetric.increment() @@ -186,7 +189,7 @@ class BlockchainRpc( override fun estimateFee(request: Mono): Mono { return request - .subscribeOn(Schedulers.boundedElastic()) + .subscribeOn(scheduler) .flatMap { val chain = Chain.byId(it.chainValue) val metrics = chainMetrics.get(chain) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt index d37af85b..2d5db6f4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt @@ -32,12 +32,11 @@ import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory -import org.springframework.beans.factory.annotation.Autowired +import org.springframework.beans.factory.annotation.Qualifier import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.scheduler.Scheduler -import reactor.core.scheduler.Schedulers import reactor.util.retry.Retry import java.math.BigInteger import java.time.Duration @@ -47,7 +46,9 @@ import kotlin.math.min @Service class TrackEthereumTx( - @Autowired private val multistreamHolder: MultistreamHolder + private val multistreamHolder: MultistreamHolder, + @Qualifier("trackTxScheduler") + private val scheduler: Scheduler ) : TrackTx { companion object { @@ -57,8 +58,6 @@ class TrackEthereumTx( private val NOT_MINED_TRACK_TTL = NOT_FOUND_TRACK_TTL.multipliedBy(2) } - var scheduler: Scheduler = Schedulers.boundedElastic() - private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java) override fun isSupported(chain: Chain): Boolean { diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy index e8d5c3fc..5efa26f2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy @@ -35,6 +35,7 @@ import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import reactor.core.publisher.Flux +import reactor.core.scheduler.Schedulers import reactor.test.StepVerifier import reactor.test.scheduler.VirtualTimeScheduler import spock.lang.Specification @@ -99,7 +100,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) apiMock.answer("eth_getTransactionByHash", [txId], txJson) apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson) @@ -122,9 +123,8 @@ class TrackEthereumTxSpec extends Specification { ((EthereumPosMultiStream) upstreams.getUpstream(Chain.ETHEREUM)).head = Mock(Head) { _ * getFlux() >> Flux.empty() } - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) def scheduler = VirtualTimeScheduler.create(true) - trackTx.scheduler = scheduler + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, scheduler) apiMock.answer("eth_getTransactionByHash", [txId], null) @@ -169,9 +169,8 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) def scheduler = VirtualTimeScheduler.create(true) - trackTx.scheduler = scheduler + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, scheduler) apiMock.answerOnce("eth_getTransactionByHash", [txId], null) apiMock.answer("eth_getTransactionByHash", [txId], txJson) @@ -194,7 +193,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) def block = new BlockContainer( @@ -216,7 +215,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) def tx = new TrackEthereumTx.TxDetails(Chain.ETHEREUM, Instant.now(), TransactionId.from(txId), 6) def block = new BlockContainer( @@ -289,7 +288,7 @@ class TrackEthereumTxSpec extends Specification { def apiMock = TestingCommons.api() def upstreamMock = TestingCommons.upstream(apiMock) MultistreamHolder upstreams = new MultistreamHolderMock(Chain.ETHEREUM, upstreamMock) - TrackEthereumTx trackTx = new TrackEthereumTx(upstreams) + TrackEthereumTx trackTx = new TrackEthereumTx(upstreams, Schedulers.boundedElastic()) apiMock.answerOnce("eth_getTransactionByHash", [txId], null) apiMock.answerOnce("eth_getTransactionByHash", [txId], txJsonBroadcasted)