Perf improvements (#758)

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

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

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

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,

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

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
             }

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()
+            .receive()
-            .map { ByteBufInputStream(it.content()).readAllBytes() }
+            .asByteArray()
+            .publishOn(eventsScheduler)
             .filter { it.isNotEmpty() }
             .flatMap {
                 try {

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

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 {
+                bytes.aggregate().asByteArray()
-                    AggregateResponse(it, statusCode)
+                    .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()
         }

src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy

@@ -227,7 +227,15 @@ class ApiReaderMock implements Reader<ChainRequest, ChainResponse> {
 
         @Override
         ByteBufFlux receive() {
-            throw new UnsupportedOperationException()
+            return ByteBufFlux
+                    .fromString(
+                        Flux.merge(
+                            jsonResponses,
+                            responses.map {
+                                Global.objectMapper.writeValueAsString(it)
+                            }
+                        )
+                    )
         }
 
         @Override

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

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

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

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