diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 94c02abf..f113f886 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -124,7 +124,13 @@ data class UpstreamsConfig( val port: Int, ) - data class HttpEndpoint(val url: URI) { + data class HttpEndpoint( + val url: URI, + val maxConnections: Int, + val queueSize: Int, + ) { + constructor(url: URI) : this(url, DEFAULT_MAX_CONNECTIONS, DEFAULT_QUEUE_SIZE) + var basicAuth: AuthConfig.ClientBasicAuth? = null var tls: AuthConfig.ClientTlsAuth? = null } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 75b17b9a..982a1b5c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -27,6 +27,9 @@ import java.io.InputStream import java.net.URI import java.util.Locale +const val DEFAULT_MAX_CONNECTIONS = 1500 +const val DEFAULT_QUEUE_SIZE = 1000 + class UpstreamsConfigReader( private val fileResolver: FileResolver, private val optionsReader: ChainOptionsReader, @@ -138,7 +141,7 @@ class UpstreamsConfigReader( getMapping(connConfigNode, "esplora")?.let { node -> getValueAsString(node, "url")?.let { url -> - val http = UpstreamsConfig.HttpEndpoint(URI(url)) + val http = UpstreamsConfig.HttpEndpoint(URI(url), DEFAULT_MAX_CONNECTIONS, DEFAULT_QUEUE_SIZE) http.basicAuth = authConfigReader.readClientBasicAuth(node) http.tls = authConfigReader.readClientTls(node) connection.esplora = http @@ -168,8 +171,11 @@ class UpstreamsConfigReader( private fun readRpcConfig(connConfigNode: MappingNode): UpstreamsConfig.HttpEndpoint? { return getMapping(connConfigNode, "rpc")?.let { node -> + val maxConnections = getValueAsInt(node, "max-connections") ?: DEFAULT_MAX_CONNECTIONS + val queueSize = getValueAsInt(node, "queue-size") ?: DEFAULT_QUEUE_SIZE + getValueAsString(node, "url")?.let { url -> - val http = UpstreamsConfig.HttpEndpoint(URI(url)) + val http = UpstreamsConfig.HttpEndpoint(URI(url), maxConnections, queueSize) http.basicAuth = authConfigReader.readClientBasicAuth(node) http.tls = authConfigReader.readClientTls(node) http 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 661f1979..590359a8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericConnectorFactoryCreator.kt @@ -66,7 +66,7 @@ open class GenericConnectorFactoryCreator( } } urls?.add(endpoint.url) - BasicHttpFactory(endpoint.url.toString(), conn.basicAuth, tls) + BasicHttpFactory(endpoint.url.toString(), endpoint.maxConnections, endpoint.queueSize, conn.basicAuth, tls) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt index af1d749c..03ce8f1c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/BasicHttpFactory.kt @@ -9,13 +9,20 @@ import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Metrics import io.micrometer.core.instrument.Tag import io.micrometer.core.instrument.Timer +import org.slf4j.LoggerFactory class BasicHttpFactory( private val url: String, + private val maxConnections: Int, + private val queueSize: Int, private val basicAuth: AuthConfig.ClientBasicAuth?, private val tls: ByteArray?, ) : HttpFactory { + private val log = LoggerFactory.getLogger(this::class.java) + override fun create(id: String?, chain: Chain): HttpReader { + log.info("Creating http pool for {} with maxConnections {} and queueSize {}", url, maxConnections, queueSize) + val metricsTags = listOf( // "unknown" is not supposed to happen Tag.of("upstream", id ?: "unknown"), @@ -35,8 +42,8 @@ class BasicHttpFactory( ) if (chain.type.apiType == ApiType.REST) { - return RestHttpReader(url, metrics, basicAuth, tls) + return RestHttpReader(url, maxConnections, queueSize, metrics, basicAuth, tls) } - return JsonRpcHttpReader(url, metrics, basicAuth, tls) + return JsonRpcHttpReader(url, maxConnections, queueSize, metrics, basicAuth, tls) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt index af8deb96..ad76fefa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpReader.kt @@ -22,19 +22,21 @@ import java.util.function.Function abstract class HttpReader( protected val target: String, + maxConnections: Int, + queueSize: Int, protected val metrics: RequestMetrics?, basicAuth: AuthConfig.ClientBasicAuth? = null, tlsCAAuth: ByteArray? = null, ) : ChainReader { - constructor() : this("", null) + constructor() : this("", 1500, 1000, null) protected val httpClient: HttpClient init { val connectionProvider = ConnectionProvider.builder("dshackleConnectionPool") - .maxConnections(1500) - .pendingAcquireMaxCount(1000) + .maxConnections(maxConnections) + .pendingAcquireMaxCount(queueSize) .pendingAcquireTimeout(Duration.ofSeconds(10)) .build() 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 fd387cc9..1b0a2b24 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/restclient/RestHttpReader.kt @@ -22,10 +22,12 @@ import java.util.concurrent.TimeUnit class RestHttpReader( target: String, + maxConnections: Int, + queueSize: Int, metrics: RequestMetrics, basicAuth: AuthConfig.ClientBasicAuth? = null, tlsCAAuth: ByteArray? = null, -) : HttpReader(target, metrics, basicAuth, tlsCAAuth) { +) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth) { private val parser = ResponseRpcParser() private val requestParser = RestRequestParser 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 c4e8a710..bdf09ad9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReader.kt @@ -37,10 +37,12 @@ import java.util.function.Function */ class JsonRpcHttpReader( target: String, + maxConnections: Int, + queueSize: Int, metrics: RequestMetrics, basicAuth: AuthConfig.ClientBasicAuth? = null, tlsCAAuth: ByteArray? = null, -) : HttpReader(target, metrics, basicAuth, tlsCAAuth) { +) : HttpReader(target, maxConnections, queueSize, metrics, basicAuth, tlsCAAuth) { private val parser = ResponseRpcParser() private val streamParser = JsonRpcStreamParser() 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 c93476bc..172bb190 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy @@ -51,7 +51,7 @@ class JsonRpcHttpReaderSpec extends Specification { def "Make a request"() { setup: - JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${port}", metrics,null, null) + JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics,null, null) def resp = '{' + ' "jsonrpc": "2.0",' + ' "result": "0x98de45",' + @@ -72,7 +72,7 @@ class JsonRpcHttpReaderSpec extends Specification { def "Produces RPC Exception on error status code"() { setup: - def client = new JsonRpcHttpReader("localhost:${port}", metrics, null, null) + def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, null, null) mockServer.when( HttpRequest.request() @@ -96,7 +96,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}", metrics, null, null) + def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, null, null) mockServer.when( HttpRequest.request()