From 2d275c517028c0811586457fa0611c93a5ca8f9e Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 26 Aug 2019 22:36:45 -0400 Subject: [PATCH] problem: wastes 200ms even for a single call because takes next api before validation solution: manage apis flow right after value recorded --- .../dshackle/quorum/AlwaysQuorum.kt | 3 +- .../emeraldpay/dshackle/quorum/CallQuorum.kt | 2 +- .../dshackle/quorum/NotLaggingQuorum.kt | 4 +- .../dshackle/quorum/ValueAwareQuorum.kt | 3 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 25 +++++---- .../dshackle/upstream/FilteringApiIterator.kt | 2 +- .../dshackle/upstream/QuorumBasedMethods.kt | 1 + .../dshackle/rpc/NativeCallSpec.groovy | 54 +++++++++++++++++-- 8 files changed, 77 insertions(+), 17 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt index e7f5118c..619a36ef 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt @@ -32,9 +32,10 @@ open class AlwaysQuorum: CallQuorum { return resolved } - override fun record(response: ByteArray, upstream: Upstream) { + override fun record(response: ByteArray, upstream: Upstream): Boolean { result = response resolved = true + return true } override fun getResult(): ByteArray? { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt index 692c5d8b..920f7b3f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt @@ -28,7 +28,7 @@ interface CallQuorum { fun init(head: Head>) fun isResolved(): Boolean - fun record(response: ByteArray, upstream: Upstream) + fun record(response: ByteArray, upstream: Upstream): Boolean fun getResult(): ByteArray? companion object { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt index 2daac72a..bcdd88d9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt @@ -32,11 +32,13 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum { return result.get() != null } - override fun record(response: ByteArray, upstream: Upstream) { + override fun record(response: ByteArray, upstream: Upstream): Boolean { val lagging = upstream.getLag() > maxLag if (!lagging) { result.set(response) + return true } + return false } override fun getResult(): ByteArray { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt index ffb5618d..216504d9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt @@ -31,7 +31,7 @@ abstract class ValueAwareQuorum( return jacksonRpcConverter.fromJson(response.inputStream(), clazz) } - override fun record(response: ByteArray, upstream: Upstream) { + override fun record(response: ByteArray, upstream: Upstream): Boolean { try { val value = extractValue(response, clazz) recordValue(response, value, upstream) @@ -40,6 +40,7 @@ abstract class ValueAwareQuorum( } catch (e: Exception) { recordError(response, e.message, upstream) } + return isResolved(); } abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 713f4a9b..73ad3924 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -27,13 +27,11 @@ import org.apache.commons.lang3.StringUtils import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import reactor.core.publisher.toFlux -import reactor.core.publisher.toMono +import reactor.core.publisher.* import reactor.util.function.Tuples import java.lang.Exception import java.time.Duration +import java.util.concurrent.atomic.AtomicInteger import java.util.function.Predicate @Service @@ -124,17 +122,26 @@ class NativeCall( } fun executeOnRemote(ctx: CallContext): Mono> { - val p: Predicate = CallQuorum.untilResolved(ctx.callQuorum) val all = ctx.getApis().toFlux().share() - //execute on the first API immediately, and then make a delay between each call to not dos upstreams + //execute on the first API immediately, and then make a delay between each call to not overload upstreams val immediate = Flux.from(all).take(1) - val retries = Flux.from(all).delayElements(Duration.ofMillis(200)) + val repeatControl = EmitterProcessor.create()//TopicProcessor.create() + val retries = Flux.from(all).skip(1) + .zipWith(repeatControl).zipWith(Flux.interval(Duration.ofMillis(200))) + .map { it.t1.t1 } + return Flux.concat(immediate, retries) - .takeWhile(p) .flatMap { api -> api.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { Tuples.of(it, api.upstream!!) } } - .reduce(ctx.callQuorum, CallQuorum.asReducer()) + .reduce(ctx.callQuorum, {res, a -> + if (res.record(a.t1, a.t2)) { + repeatControl.onComplete() + } else { + repeatControl.onNext(true) + } + res + }) .filter { it.isResolved() } .map { val result = it.getResult() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt index a1eb48a5..8b6fd7a0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteringApiIterator.kt @@ -21,7 +21,7 @@ class FilteringApiIterator( private val upstreams: List, private var pos: Int, private val matcher: Selector.Matcher, - private val repeatLimit: Int = 3 + private val repeatLimit: Int = 5 ): Iterator { private var nextUpstream: Upstream? = null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/QuorumBasedMethods.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/QuorumBasedMethods.kt index 82bc9535..44197d79 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/QuorumBasedMethods.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/QuorumBasedMethods.kt @@ -80,6 +80,7 @@ class QuorumBasedMethods( override fun getQuorumFor(method: String): CallQuorum { return when { hardcodedMethods.contains(method) -> AlwaysQuorum() + firstValueMethods.contains(method) -> AlwaysQuorum() anyResponseMethods.contains(method) -> NotLaggingQuorum(6) headVerifiedMethods.contains(method) -> NotLaggingQuorum(1) specialMethods.contains(method) -> { diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index e37db800..3c97dd0e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -57,7 +57,6 @@ class NativeCallSpec extends Specification { def act = objectMapper.readValue(resp.payload, Map) then: act == [jsonrpc:"2.0", id:1, result: "foo"] - (2..3) * quorum.isResolved() // 2 times during api call (before and after) + 1 time in a filter after 1 * quorum.record(_, _) 1 * quorum.getResult() } @@ -86,11 +85,61 @@ class NativeCallSpec extends Specification { def act = objectMapper.readValue(resp.payload, Map) then: act == [jsonrpc:"2.0", id:1, result: "bar"] - (3..4) * quorum.isResolved() 2 * quorum.record(_, _) 1 * quorum.getResult() } + def "Have pause between repeats"() { + setup: + def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) + + def upstreams = Stub(Upstreams) + RpcClient rpcClient = Stub(RpcClient) + def apiMock = TestingCommons.api(rpcClient) + apiMock.upstream = Stub(Upstream) + + apiMock.answerOnce("eth_test", [], null) + apiMock.answerOnce("eth_test", [], "bar") + + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), + Selector.empty, quorum, + new NativeCall.ParsedCallDetails("eth_test", [])) + + + when: + def t1 = System.currentTimeMillis() + nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) + def delta = System.currentTimeMillis() - t1 + then: + delta >= 200 + } + + def "One call has no pause"() { + setup: + def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) + + def upstreams = Stub(Upstreams) + RpcClient rpcClient = Stub(RpcClient) + def apiMock = TestingCommons.api(rpcClient) + apiMock.upstream = Stub(Upstream) + + apiMock.answerOnce("eth_test", [], "bar") + + def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) + def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), + Selector.empty, quorum, + new NativeCall.ParsedCallDetails("eth_test", [])) + + + when: + def t1 = System.currentTimeMillis() + nativeCall.executeOnRemote(call).block(Duration.ofSeconds(2)) + def delta = System.currentTimeMillis() - t1 + then: + delta < 50 + } + def "Returns error if no quorum"() { setup: def quorum = Spy(new NonEmptyQuorum(TestingCommons.rpcConverter(), 3)) @@ -107,7 +156,6 @@ class NativeCallSpec extends Specification { def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(apiMock), Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_test", [])) - (4..5) * quorum.isResolved() 3 * quorum.record(_, _) 1 * quorum.getResult()