diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 5f4606d9..9e281b38 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -229,7 +229,7 @@ open class NativeCall( .forMethod(method) .forLabels(Selector.convertToMatcher(request.selector)) - val callQuorum = availableMethods.getQuorumFor(method) // can be null in tests + val callQuorum = availableMethods.createQuorumFor(method) // can be null in tests callQuorum.init(upstream.getHead()) // for NotLaggingQuorum it makes sense to select compatible upstreams before the call diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethods.kt index eae9f0b7..034748d5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethods.kt @@ -37,10 +37,10 @@ class AggregatedCallMethods( /** * Finds first delegate that has Allowed that method and returns its Quorum */ - override fun getQuorumFor(method: String): CallQuorum { + override fun createQuorumFor(method: String): CallQuorum { return delegates.find { it.isCallable(method) || it.isHardcoded(method) - }?.getQuorumFor(method) ?: throw IllegalStateException("No executor delegate for $method") + }?.createQuorumFor(method) ?: throw IllegalStateException("No executor delegate for $method") } /** diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallMethods.kt index d0abc4fa..e895fc65 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallMethods.kt @@ -24,9 +24,11 @@ import io.emeraldpay.dshackle.quorum.CallQuorum interface CallMethods { /** + * For a stateful CallQuorum it _MUST CREATE_ a new instance of each time to avoid using a shared state between different requests + * * @return CallQuorum configured for the specified method */ - fun getQuorumFor(method: String): CallQuorum + fun createQuorumFor(method: String): CallQuorum /** * Check if the method can be called on an upstream. Doesn't include Hardcoded methods diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt index 35ce7939..25b51f51 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt @@ -59,7 +59,7 @@ class DefaultBitcoinMethods : CallMethods { private val allowedMethods = (freshMethods + anyResponseMethods + headVerifiedMethods + broadcastMethods).sorted() - override fun getQuorumFor(method: String): CallQuorum { + override fun createQuorumFor(method: String): CallQuorum { return when { Collections.binarySearch(hardcodedMethods, method) >= 0 -> AlwaysQuorum() Collections.binarySearch(anyResponseMethods, method) >= 0 -> NonEmptyQuorum() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt index 532692e9..20bea42f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt @@ -112,7 +112,7 @@ class DefaultEthereumMethods( getChainSpecificMethods(chain) } - override fun getQuorumFor(method: String): CallQuorum { + override fun createQuorumFor(method: String): CallQuorum { return when { filterMethods.contains(method) -> NotLaggingQuorum(1) hardcodedMethods.contains(method) -> AlwaysQuorum() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DirectCallMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DirectCallMethods.kt index 0f46d376..c60e1aa3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DirectCallMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DirectCallMethods.kt @@ -28,7 +28,7 @@ open class DirectCallMethods(private val methods: Set) : CallMethods { constructor() : this(emptySet()) constructor(methods: Collection) : this(methods.toSet()) - override fun getQuorumFor(method: String): CallQuorum { + override fun createQuorumFor(method: String): CallQuorum { return AlwaysQuorum() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt index 3b53499b..4259656e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt @@ -42,7 +42,7 @@ class EthereumCallSelector( "eth_getBalance", "eth_getCode", "eth_getTransactionCount", - // no "eth_getStorageAt" because it's has different structure, and therefore separate logic + // no "eth_getStorageAt" because it has different structure, and therefore separate logic "eth_call" ).sorted() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt index e2ffc47f..0b27ede2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt @@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.NonEmptyQuorum import io.emeraldpay.dshackle.quorum.NotLaggingQuorum +import org.apache.commons.collections4.Factory import org.slf4j.LoggerFactory import java.io.IOException import java.util.Collections @@ -38,14 +39,16 @@ class ManagedCallMethods( companion object { private val log = LoggerFactory.getLogger(ManagedCallMethods::class.java) - private val defaultQuorum = AlwaysQuorum() + private val defaultQuorum: Factory = Factory { + AlwaysQuorum() + } } private val delegated = delegate.getSupportedMethods().sorted() private val allAllowed: Set = Collections.unmodifiableSet( enabled + delegated - disabled ) - private val quorum: MutableMap = HashMap() + private val quorum: MutableMap> = HashMap() private val staticResponse: MutableMap = HashMap() private val redefined = delegated.filter(enabled::contains).sorted() @@ -57,9 +60,9 @@ class ManagedCallMethods( fun setQuorum(method: String, quorumId: String) { val quorum = when (quorumId) { - "always" -> AlwaysQuorum() - "no-lag", "not-lagging", "no_lag", "not_lagging" -> NotLaggingQuorum(0) - "not-empty", "not_empty", "non-empty", "non_empty" -> NonEmptyQuorum() + "always" -> Factory { AlwaysQuorum() } + "no-lag", "not-lagging", "no_lag", "not_lagging" -> Factory { NotLaggingQuorum(0) } + "not-empty", "not_empty", "non-empty", "non_empty" -> Factory { NonEmptyQuorum() } else -> { log.warn("Unknown quorum: $quorumId for custom method $method") return @@ -72,13 +75,13 @@ class ManagedCallMethods( this.staticResponse[method] = response } - override fun getQuorumFor(method: String): CallQuorum { + override fun createQuorumFor(method: String): CallQuorum { return when { - isDelegated(method) && !isRedefined(method) -> delegate.getQuorumFor(method) - enabled.contains(method) -> quorum[method] ?: defaultQuorum + isDelegated(method) && !isRedefined(method) -> delegate.createQuorumFor(method) + enabled.contains(method) -> quorum[method]?.create() ?: defaultQuorum.create() else -> { log.warn("Getting quorum for unknown method") - defaultQuorum + defaultQuorum.create() } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt index 15bc299c..429c135c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt @@ -132,8 +132,12 @@ class EthereumDirectReader( */ private fun readWithQuorum(request: JsonRpcRequest): Mono { return quorumReaderFactory - // we do not use Signer for internal requests because it doesn't make much sense - .create(up.getApiSource(Selector.empty), callMethodsFactory.create().getQuorumFor(request.method), null) + .create( + up.getApiSource(Selector.empty), + callMethodsFactory.create().createQuorumFor(request.method), + // we do not use Signer for internal requests because it doesn't make much sense + null + ) .read(request) .map { it.value } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy index 286458c1..8393dbef 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy @@ -1,7 +1,6 @@ package io.emeraldpay.dshackle.startup import io.emeraldpay.dshackle.FileResolver -import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.quorum.NonEmptyQuorum import io.emeraldpay.dshackle.upstream.CallTargetsHolder @@ -36,7 +35,7 @@ class ConfiguredUpstreamsSpec extends Specification { def act = configurer.buildMethods(upstream, Chain.ETHEREUM) then: act instanceof ManagedCallMethods - act.getQuorumFor("foo_bar") instanceof NonEmptyQuorum + act.createQuorumFor("foo_bar") instanceof NonEmptyQuorum } def "Got static response from extra methods"() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index 2f54b1b0..fb36e45b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -58,9 +58,9 @@ class MultistreamSpec extends Specification { act.isCallable("eth_test1") act.isCallable("eth_test2") act.isCallable("eth_test3") - act.getQuorumFor("eth_test1") instanceof AlwaysQuorum - act.getQuorumFor("eth_test2") instanceof AlwaysQuorum - act.getQuorumFor("eth_test3") instanceof AlwaysQuorum + act.createQuorumFor("eth_test1") instanceof AlwaysQuorum + act.createQuorumFor("eth_test2") instanceof AlwaysQuorum + act.createQuorumFor("eth_test3") instanceof AlwaysQuorum } def "Filter Best Status accepts any input when none available "() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethodsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethodsSpec.groovy index 2109feae..236589d7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethodsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethodsSpec.groovy @@ -31,11 +31,11 @@ class AggregatedCallMethodsSpec extends Specification { def delegate2 = Mock(CallMethods) { _ * getSupportedMethods() >> ["eth_test", "foo_bar"] 1 * isCallable("eth_test") >> true - 1 * getQuorumFor("eth_test") >> quorum + 1 * createQuorumFor("eth_test") >> quorum } def aggregate = new AggregatedCallMethods([delegate1, delegate2]) when: - def act = aggregate.getQuorumFor("eth_test") + def act = aggregate.createQuorumFor("eth_test") then: act == quorum } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy index 8c1d5afd..c9261504 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethodsSpec.groovy @@ -18,14 +18,15 @@ package io.emeraldpay.dshackle.upstream.calls import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.BroadcastQuorum +import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.NonEmptyQuorum -import io.emeraldpay.dshackle.quorum.NonceQuorum import io.emeraldpay.dshackle.quorum.NotLaggingQuorum -import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.Chain import spock.lang.Specification +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit + class ManagedCallMethodsSpec extends Specification { def "Gets quorum for enabled method"() { @@ -36,7 +37,7 @@ class ManagedCallMethodsSpec extends Specification { [] as Set ) when: - def act = managed.getQuorumFor("eth_test") + def act = managed.createQuorumFor("eth_test") then: act instanceof AlwaysQuorum } @@ -72,7 +73,7 @@ class ManagedCallMethodsSpec extends Specification { def delegated = ["eth_test", "eth_test2"] as Set def delegate = Mock(CallMethods) { _ * it.getSupportedMethods() >> delegated - 0 * it.getQuorumFor("eth_test") >> new BroadcastQuorum() + 0 * it.createQuorumFor("eth_test") >> new BroadcastQuorum() } def managed = new ManagedCallMethods( delegate, @@ -80,7 +81,7 @@ class ManagedCallMethodsSpec extends Specification { ["foo_bar"] as Set ) when: - def act = managed.getQuorumFor("eth_test") + def act = managed.createQuorumFor("eth_test") then: act != null act instanceof AlwaysQuorum @@ -96,18 +97,40 @@ class ManagedCallMethodsSpec extends Specification { managed.setQuorum("eth_test", "not_empty") managed.setQuorum("eth_foo", "not_lagging") when: - def act = managed.getQuorumFor("eth_test") + def act = managed.createQuorumFor("eth_test") then: act instanceof NonEmptyQuorum when: - act = managed.getQuorumFor("eth_foo") + act = managed.createQuorumFor("eth_foo") then: act instanceof NotLaggingQuorum when: - act = managed.getQuorumFor("eth_bar") + act = managed.createQuorumFor("eth_bar") then: act instanceof AlwaysQuorum } + + def "Doesn't reuse same instance"() { + def managed = new ManagedCallMethods( + new DefaultEthereumMethods(Chain.ETHEREUM), + ["eth_test"] as Set, + [] as Set + ) + def parallel = Executors.newFixedThreadPool(16) + when: + List instances = [] + 50.times { + instances << managed.createQuorumFor("eth_test") + } + parallel.shutdown() + parallel.awaitTermination(5, TimeUnit.SECONDS) + + def ids = instances.collect { System.identityHashCode(it) } + + then: + instances.size() == 50 + ids.toSet().size() == 50 + } }