From 64705f612d227a2ec7f8a17faac6b5edd6f8eeb1 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Fri, 2 Jun 2023 14:20:32 +0400 Subject: [PATCH] We should always send only best status form Multistream.observeStatus --- .../dshackle/upstream/Multistream.kt | 31 +++------ .../dshackle/upstream/MultistreamSpec.groovy | 67 +++++++++---------- 2 files changed, 40 insertions(+), 58 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 17665770..263f0e62 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -40,10 +40,8 @@ import reactor.core.publisher.Mono import reactor.core.publisher.Sinks import reactor.util.function.Tuples import java.time.Duration -import java.time.Instant -import java.util.concurrent.atomic.AtomicReference +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.locks.ReentrantLock -import java.util.function.Predicate import kotlin.concurrent.withLock /** @@ -232,8 +230,8 @@ abstract class Multistream( ).map { UpstreamStatus(up, it) } } return Flux.merge(upstreamsFluxes) - .filter(FilterBestAvailability()) - .map { it.status } + .map(FilterBestAvailability()) + .distinct() } override fun isAvailable(): Boolean { @@ -452,24 +450,13 @@ abstract class Multistream( // -------------------------------------------------------------------------------------------------------- - class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) + class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability) - class FilterBestAvailability : Predicate { - private val lastRef = AtomicReference() - - override fun test(t: UpstreamStatus): Boolean { - val curr = lastRef.updateAndGet { last -> - val changed = last == null || - t.status < last.status || - (last.upstream == t.upstream && t.status != last.status) || - last.ts.isBefore(t.ts - Duration.ofSeconds(60)) - if (changed) { - t - } else { - last - } - } - return curr == t + class FilterBestAvailability : java.util.function.Function { + val map = ConcurrentHashMap() + override fun apply(t: UpstreamStatus): UpstreamAvailability { + map[t.upstream.getId()] = t.status + return map.values.min() } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index 4d5c8353..94968d56 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -70,87 +70,83 @@ class MultistreamSpec extends Specification { def up = TestingCommons.upstream() def filter = new Multistream.FilterBestAvailability() def update1 = new Multistream.UpstreamStatus( - up, UpstreamAvailability.LAGGING, Instant.now() + up, UpstreamAvailability.LAGGING ) when: - def act = filter.test(update1) + def status = filter.apply(update1) then: - act + status == UpstreamAvailability.LAGGING } def "Filter Best Status accepts better input"() { setup: def up1 = TestingCommons.upstream("test-1") def up2 = TestingCommons.upstream("test-2") - def time0 = Instant.now() - Duration.ofSeconds(60) def filter = new Multistream.FilterBestAvailability() def update0 = new Multistream.UpstreamStatus( - up1, UpstreamAvailability.LAGGING, time0 + up1, UpstreamAvailability.LAGGING ) def update1 = new Multistream.UpstreamStatus( - up2, UpstreamAvailability.OK, time0 + up2, UpstreamAvailability.OK ) when: - filter.test(update0) - def act = filter.test(update1) + filter.apply(update0) + def status = filter.apply(update1) then: - act + status == UpstreamAvailability.OK } def "Filter Best Status declines worse input"() { setup: def up1 = TestingCommons.upstream("test-1") def up2 = TestingCommons.upstream("test-2") - def time0 = Instant.now() - Duration.ofSeconds(60) def filter = new Multistream.FilterBestAvailability() def update0 = new Multistream.UpstreamStatus( - up1, UpstreamAvailability.LAGGING, time0 + up1, UpstreamAvailability.LAGGING ) def update1 = new Multistream.UpstreamStatus( - up2, UpstreamAvailability.IMMATURE, time0 + up2, UpstreamAvailability.IMMATURE ) when: - filter.test(update0) - def act = filter.test(update1) + filter.apply(update0) + def status = filter.apply(update1) then: - !act + status == UpstreamAvailability.LAGGING } def "Filter Best Status accepts worse input from same upstream"() { setup: def up = TestingCommons.upstream("test-1") - def time0 = Instant.now() - Duration.ofSeconds(60) def filter = new Multistream.FilterBestAvailability() def update0 = new Multistream.UpstreamStatus( - up, UpstreamAvailability.LAGGING, time0 + up, UpstreamAvailability.LAGGING ) def update1 = new Multistream.UpstreamStatus( - up, UpstreamAvailability.IMMATURE, time0 + up, UpstreamAvailability.IMMATURE ) when: - filter.test(update0) - def act = filter.test(update1) + filter.apply(update0) + def status = filter.apply(update1) then: - act + status == UpstreamAvailability.IMMATURE } def "Filter Best Status accepts any input if existing is outdated"() { setup: def up1 = TestingCommons.upstream("test-1") def up2 = TestingCommons.upstream("test-2") - def time0 = Instant.now() - Duration.ofSeconds(90) def filter = new Multistream.FilterBestAvailability() def update0 = new Multistream.UpstreamStatus( - up1, UpstreamAvailability.OK, time0 + up1, UpstreamAvailability.OK ) def update1 = new Multistream.UpstreamStatus( - up2, UpstreamAvailability.IMMATURE, time0 + Duration.ofSeconds(65) + up2, UpstreamAvailability.IMMATURE ) when: - filter.test(update0) - def act = filter.test(update1) + filter.apply(update0) + def status = filter.apply(update1) then: - act + status == UpstreamAvailability.OK } def "Filter Best Status declines same status"() { @@ -158,28 +154,27 @@ class MultistreamSpec extends Specification { def up1 = TestingCommons.upstream("test-1") def up2 = TestingCommons.upstream("test-2") def up3 = TestingCommons.upstream("test-3") - def time0 = Instant.now() - Duration.ofSeconds(60) def filter = new Multistream.FilterBestAvailability() def update0 = new Multistream.UpstreamStatus( - up1, UpstreamAvailability.OK, time0 + up1, UpstreamAvailability.OK ) def update1 = new Multistream.UpstreamStatus( - up2, UpstreamAvailability.OK, time0 + up2, UpstreamAvailability.OK ) def update2 = new Multistream.UpstreamStatus( - up3, UpstreamAvailability.OK, time0 + Duration.ofSeconds(10) + up3, UpstreamAvailability.OK ) when: - filter.test(update0) - def act = filter.test(update1) + filter.apply(update0) + def status = filter.apply(update1) then: - !act + status == UpstreamAvailability.OK when: - act = filter.test(update2) + status = filter.apply(update2) then: - !act + status == UpstreamAvailability.OK }