From 4f2959a75f10eee099955133047f5429dfe1d0ae Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Wed, 17 Dec 2025 16:38:12 +0400 Subject: [PATCH] Perf improvements (#758) --- src/main/kotlin/io/emeraldpay/dshackle/Global.kt | 3 +++ .../dshackle/config/context/SchedulersConfig.kt | 10 ++++++++++ .../configure/GenericConnectorFactoryCreator.kt | 4 ++++ .../configure/RestConnectorFactoryCreator.kt | 4 ++++ .../dshackle/upstream/BasicHttpFactory.kt | 6 ++++-- .../upstream/ethereum/WsConnectionFactory.kt | 3 ++- .../upstream/ethereum/WsConnectionImpl.kt | 8 +++++--- .../upstream/restclient/RestHttpReader.kt | 8 +++++--- .../upstream/rpcclient/JsonRpcHttpReader.kt | 15 +++++++++++---- .../emeraldpay/dshackle/test/ApiReaderMock.groovy | 10 +++++++++- .../ethereum/WsConnectionImplRealSpec.groovy | 1 + .../upstream/ethereum/WsConnectionImplSpec.groovy | 9 ++++++--- .../rpcclient/JsonRpcHttpReaderSpec.groovy | 7 ++++--- .../GenericConnectorFactoryCreatorTest.kt | 3 +++ 14 files changed, 71 insertions(+), 20 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index 8d5ee45f..a9775dbb 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt @@ -35,6 +35,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.domain.TransactionId import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.TransactionIdSerializer import io.emeraldpay.dshackle.upstream.ton.TonMasterchainInfo import io.emeraldpay.dshackle.upstream.ton.TonMasterchainInfoDeserializer +import reactor.netty.resources.LoopResources import java.math.BigInteger import java.text.SimpleDateFormat import java.util.Locale @@ -46,6 +47,8 @@ class Global { companion object { + val wsLoops: LoopResources = LoopResources.create("reactor-ws") + val nullValue: ByteArray = "null".toByteArray() var metricsExtended = false 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 4ad9a48e..f3ad28a2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt @@ -76,6 +76,16 @@ open class SchedulersConfig { return makeScheduler("auth-scheduler", 4, monitoringConfig) } + @Bean + open fun httpScheduler(monitoringConfig: MonitoringConfig): Scheduler { + return makeScheduler("http-scheduler", 30, monitoringConfig) + } + + @Bean + open fun eventsScheduler(monitoringConfig: MonitoringConfig): Scheduler { + return makeScheduler("ws-events-scheduler", 30, monitoringConfig) + } + private fun makeScheduler(name: String, size: Int, monitoringConfig: MonitoringConfig): Scheduler { return Schedulers.fromExecutorService(makePool(name, size, monitoringConfig)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt index 8dc32db5..f85aa742 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt @@ -26,6 +26,8 @@ open class GenericConnectorFactoryCreator( private val wsScheduler: Scheduler, private val headLivenessScheduler: Scheduler, private val monitoringCfg: MonitoringConfig, + private val httpScheduler: Scheduler, + private val eventsScheduler: Scheduler, ) : ConnectorFactoryCreator { protected val log = LoggerFactory.getLogger(this::class.java) @@ -75,6 +77,7 @@ open class GenericConnectorFactoryCreator( conn.basicAuth, tls, monitoringCfg.nettyMetricsConfig.enabled, + httpScheduler, ) } } @@ -92,6 +95,7 @@ open class GenericConnectorFactoryCreator( endpoint.url, endpoint.origin ?: URI("http://localhost"), wsScheduler, + eventsScheduler, ).apply { config = endpoint basicAuth = endpoint.basicAuth diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/RestConnectorFactoryCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/RestConnectorFactoryCreator.kt index 321c8eeb..7a58eebe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/RestConnectorFactoryCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/RestConnectorFactoryCreator.kt @@ -22,6 +22,8 @@ class RestConnectorFactoryCreator( private val headScheduler: Scheduler, private val headLivenessScheduler: Scheduler, monitoringCfg: MonitoringConfig, + httpScheduler: Scheduler, + eventsScheduler: Scheduler, ) : GenericConnectorFactoryCreator( fileResolver, Schedulers.single(), @@ -29,6 +31,8 @@ class RestConnectorFactoryCreator( Schedulers.single(), headLivenessScheduler, monitoringCfg, + httpScheduler, + eventsScheduler, ) { override fun createConnectorFactory( id: String, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt index 7db34b1e..5a9b90b6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt @@ -10,6 +10,7 @@ import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.Tag import io.micrometer.core.instrument.Timer import org.slf4j.LoggerFactory +import reactor.core.scheduler.Scheduler class BasicHttpFactory( private val url: String, @@ -18,6 +19,7 @@ class BasicHttpFactory( private val basicAuth: AuthConfig.ClientBasicAuth?, private val tls: ByteArray?, private val nettyMetricsEnabled: Boolean, + private val httpScheduler: Scheduler, ) : HttpFactory { private val log = LoggerFactory.getLogger(this::class.java) @@ -44,8 +46,8 @@ class BasicHttpFactory( ) if (chain.type.apiType == ApiType.REST) { - return RestHttpReader(url, maxConnections, queueSize, metrics, basicAuth, tls) + return RestHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, basicAuth, tls) } - return JsonRpcHttpReader(url, maxConnections, queueSize, metrics, basicAuth, tls) + return JsonRpcHttpReader(url, maxConnections, queueSize, metrics, httpScheduler, basicAuth, tls) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionFactory.kt index 4ef0914c..6179ac33 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionFactory.kt @@ -17,6 +17,7 @@ open class WsConnectionFactory( private val uri: URI, private val origin: URI, private val scheduler: Scheduler, + private val eventsScheduler: Scheduler, ) { var basicAuth: AuthConfig.ClientBasicAuth? = null @@ -45,7 +46,7 @@ open class WsConnectionFactory( } open fun createWsConnection(connIndex: Int = 0): WsConnection = - WsConnectionImpl(uri, origin, basicAuth, metrics(connIndex), scheduler).also { ws -> + WsConnectionImpl(uri, origin, basicAuth, metrics(connIndex), scheduler, eventsScheduler).also { ws -> config?.frameSize?.let { ws.frameSize = it } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt index d9c35433..02aae582 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt @@ -29,7 +29,6 @@ import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsMessage import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser import io.micrometer.core.instrument.Metrics -import io.netty.buffer.ByteBufInputStream import io.netty.handler.codec.http.HttpHeaderNames import io.netty.resolver.DefaultAddressResolverGroup import org.reactivestreams.Publisher @@ -66,6 +65,7 @@ open class WsConnectionImpl( private val basicAuth: AuthConfig.ClientBasicAuth?, private val requestMetrics: RequestMetrics?, private val scheduler: Scheduler, + private val eventsScheduler: Scheduler, ) : AutoCloseable, WsConnection, Cloneable { companion object { @@ -193,6 +193,7 @@ open class WsConnectionImpl( log.info("Connecting to WebSocket: $uri") connection?.dispose() connection = HttpClient.create() + .runOn(Global.wsLoops) .resolver(DefaultAddressResolverGroup.INSTANCE) .doOnDisconnected { disconnects.tryEmitNext(Instant.now()) @@ -251,8 +252,9 @@ open class WsConnectionImpl( var read = false val consumer = inbound .aggregateFrames(msgSizeLimit) - .receiveFrames() - .map { ByteBufInputStream(it.content()).readAllBytes() } + .receive() + .asByteArray() + .publishOn(eventsScheduler) .filter { it.isNotEmpty() } .flatMap { try { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt index 1b0a2b24..78276960 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt @@ -17,6 +17,7 @@ import io.netty.handler.codec.http.HttpMethod import org.apache.commons.lang3.time.StopWatch import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.scheduler.Scheduler import reactor.kotlin.core.publisher.switchIfEmpty import java.util.concurrent.TimeUnit @@ -25,6 +26,7 @@ class RestHttpReader( maxConnections: Int, queueSize: Int, metrics: RequestMetrics, + private val httpScheduler: Scheduler, basicAuth: AuthConfig.ClientBasicAuth? = null, tlsCAAuth: ByteArray? = null, ) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth) { @@ -86,7 +88,7 @@ class RestHttpReader( response.response { header, bytes -> val statusCode = header.status().code() - bytes.aggregate().asByteArray().map { + bytes.aggregate().asByteArray().publishOn(httpScheduler).map { AggregateResponse(it, statusCode) }.switchIfEmpty { Mono.just(AggregateResponse(ByteArray(0), statusCode)) @@ -95,13 +97,13 @@ class RestHttpReader( } else { response.responseConnection { t, u -> if (t.status().code() != 200) { - u.inbound().receive().aggregate().asByteArray() + u.inbound().receive().aggregate().asByteArray().publishOn(httpScheduler) .map { AggregateResponse(it, t.status().code()) } } else { Mono.just( StreamResponse( Flux.concat( - u.inbound().receive().asByteArray() + u.inbound().receive().asByteArray().publishOn(httpScheduler) .map { Chunk(it, false) }, Mono.just(Chunk(ByteArray(0), true)), ), diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt index bdf09ad9..4de3c065 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt @@ -29,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.stream.StreamResponse import io.netty.buffer.Unpooled import org.apache.commons.lang3.time.StopWatch import reactor.core.publisher.Mono +import reactor.core.scheduler.Scheduler import java.util.concurrent.TimeUnit import java.util.function.Function @@ -40,6 +41,7 @@ class JsonRpcHttpReader( maxConnections: Int, queueSize: Int, metrics: RequestMetrics, + private val httpScheduler: Scheduler, basicAuth: AuthConfig.ClientBasicAuth? = null, tlsCAAuth: ByteArray? = null, ) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth) { @@ -59,15 +61,20 @@ class JsonRpcHttpReader( response.response { header, bytes -> val statusCode = header.status().code() - bytes.aggregate().asByteArray().map { - AggregateResponse(it, statusCode) - } + bytes.aggregate().asByteArray() + .publishOn(httpScheduler) + .map { + AggregateResponse(it, statusCode) + } }.single() } else { response.responseConnection { t, u -> streamParser.streamParse( t.status().code(), - u.inbound().receive().asByteArray(), + u.inbound() + .receive() + .asByteArray() + .publishOn(httpScheduler), ) }.single() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy index 89b0fec7..4a4067e6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy @@ -227,7 +227,15 @@ class ApiReaderMock implements Reader { @Override ByteBufFlux receive() { - throw new UnsupportedOperationException() + return ByteBufFlux + .fromString( + Flux.merge( + jsonResponses, + responses.map { + Global.objectMapper.writeValueAsString(it) + } + ) + ) } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy index 969b65fc..d2fa7308 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplRealSpec.groovy @@ -43,6 +43,7 @@ class WsConnectionImplRealSpec extends Specification { Chain.ETHEREUM__MAINNET, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI(), + Schedulers.boundedElastic(), Schedulers.boundedElastic() ) ).create(upstream).getConnection() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy index eeff55ee..18601bae 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy @@ -46,7 +46,8 @@ class WsConnectionImplSpec extends Specification { Chain.ETHEREUM__MAINNET, new URI("http://localhost"), new URI("http://localhost"), - Schedulers.boundedElastic() + Schedulers.boundedElastic(), + Schedulers.boundedElastic(), ) ) def apiMock = TestingCommons.api() @@ -81,7 +82,8 @@ class WsConnectionImplSpec extends Specification { Chain.ETHEREUM__MAINNET, new URI("http://localhost"), new URI("http://localhost"), - Schedulers.boundedElastic() + Schedulers.boundedElastic(), + Schedulers.boundedElastic(), ) ) def apiMock = TestingCommons.api() @@ -114,7 +116,8 @@ class WsConnectionImplSpec extends Specification { Chain.ETHEREUM__MAINNET, new URI("http://localhost"), new URI("http://localhost"), - Schedulers.boundedElastic() + Schedulers.boundedElastic(), + Schedulers.boundedElastic(), ) ) def apiMock = TestingCommons.api() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy index 8589b780..977c3ec1 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy @@ -27,6 +27,7 @@ import org.mockserver.integration.ClientAndServer import org.mockserver.model.HttpRequest import org.mockserver.model.HttpResponse import org.springframework.util.SocketUtils +import reactor.core.scheduler.Schedulers import spock.lang.Specification import java.time.Duration @@ -52,7 +53,7 @@ class JsonRpcHttpReaderSpec extends Specification { def "Make a request"() { setup: - JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics,null, null) + JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, Schedulers.boundedElastic(),null, null) def resp = '{' + ' "jsonrpc": "2.0",' + ' "result": "0x98de45",' + @@ -73,7 +74,7 @@ class JsonRpcHttpReaderSpec extends Specification { def "Produces RPC Exception on error status code"() { setup: - def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, null, null) + def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, Schedulers.boundedElastic(), null, null) mockServer.when( HttpRequest.request() @@ -97,7 +98,7 @@ class JsonRpcHttpReaderSpec extends Specification { def "Tries to extract message if HTTP error if it still contains a JSON RPC message"() { setup: - def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, null, null) + def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, Schedulers.boundedElastic(), null, null) mockServer.when( HttpRequest.request() diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactoryCreatorTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactoryCreatorTest.kt index d79b7e1e..b2327106 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactoryCreatorTest.kt +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/connectors/GenericConnectorFactoryCreatorTest.kt @@ -19,6 +19,7 @@ import org.junit.jupiter.params.provider.MethodSource import org.mockito.Mockito.mockConstruction import org.mockito.kotlin.doReturn import org.mockito.kotlin.mock +import reactor.core.scheduler.Schedulers import reactor.core.scheduler.Schedulers.immediate import java.io.File import java.net.URI @@ -39,6 +40,8 @@ class GenericConnectorFactoryCreatorTest { immediate(), immediate(), MonitoringConfig.default(), + Schedulers.boundedElastic(), + Schedulers.boundedElastic(), ) var args: List<*>? = null