Merge pull request #7 from p2p-org/fix_proxy_performance

Fix proxy performance

src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt

@@ -28,6 +28,7 @@ import io.micrometer.core.instrument.Metrics
 import io.micrometer.core.instrument.Timer
 import io.netty.channel.ChannelHandler
 import io.netty.channel.ChannelHandlerContext
+import io.netty.channel.nio.NioEventLoopGroup
 import org.slf4j.LoggerFactory
 import reactor.netty.http.server.HttpServer
 import reactor.netty.http.server.HttpServerRoutes
@@ -107,6 +108,7 @@ class ProxyServer(
 
         serverBuilder
             .route(this::setupRoutes)
+            .runOn(NioEventLoopGroup())
             .bindNow()
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumBlockValidator.kt

@@ -174,7 +174,7 @@ class EthereumBlockValidator : BlockValidator {
     private fun BigInteger.asUint64() = this.toByteArray().let { bytes ->
         ByteArray(8) {
             val index = bytes.size - it - 1
-            if (index < bytes.size) bytes[index] else 0
+            if (index < bytes.size && index >= 0) bytes[index] else 0
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt

@@ -119,7 +119,9 @@ open class WsConnection(
         .multicast()
         .directBestEffort<Instant>()
     private val sendIdSeq = AtomicInteger(IDS_START)
-    private val sendExecutor = Executors.newSingleThreadExecutor()
+    private val sendExecutor = Executors.newFixedThreadPool(
+        1.coerceAtLeast(Runtime.getRuntime().availableProcessors() / 2)
+    )
     private var keepConnection = true
     private var connection: Disposable? = null
     private val reconnecting = AtomicBoolean(false)
@@ -169,6 +171,7 @@ open class WsConnection(
 
     private fun connectInternal() {
         log.info("Connecting to WebSocket: $uri")
+        log.info("Available processors: ${Runtime.getRuntime().availableProcessors()}")
         connection?.dispose()
         connection = HttpClient.create()
             .resolver(DefaultAddressResolverGroup.INSTANCE)