StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #68 from p2p-org/schedulers-config

Browse Source 
add schedulers config
This commit is contained in:
a10zn8
2022-12-07 17:22:41 +04:00
committed by GitHub
parent 3a606fe0fe d35fd31d9a
commit 6a77a754ba
5 changed files with 68 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/main/kotlin/io/emeraldpay/dshackle/GrpcServer.kt
View File

@@ -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<io.grpc.BindableService>,
@Autowired val mainConfig: MainConfig,
@Autowired val tlsSetup: TlsSetup,
@Autowired val accessHandler: AccessHandlerGrpc
private val rpcs: List<io.grpc.BindableService>,
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)

40
src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt Normal file
View File

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

19
src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
View File

@@ -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<Int, String>()
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<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
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<BlockchainOuterClass.TxStatus>(IllegalStateException("Internal Error"))
Flux.error(IllegalStateException("Internal Error"))
}
}
}
override fun subscribeBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
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<BlockchainOuterClass.AddressBalance>(IllegalStateException("Internal Error"))
Flux.error(IllegalStateException("Internal Error"))
}
}
}
override fun getBalance(requestMono: Mono<BlockchainOuterClass.BalanceRequest>): Flux<BlockchainOuterClass.AddressBalance> {
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<BlockchainOuterClass.EstimateFeeRequest>): Mono<BlockchainOuterClass.EstimateFeeResponse> {
return request
.subscribeOn(Schedulers.boundedElastic())
.subscribeOn(scheduler)
.flatMap {
val chain = Chain.byId(it.chainValue)
val metrics = chainMetrics.get(chain)

9
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt
View File

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

15
src/test/groovy/io/emeraldpay/dshackle/rpc/TrackEthereumTxSpec.groovy
View File

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