diff --git a/src/main/kotlin/io/emeraldpay/dshackle/commons/ExpiringSet.kt b/src/main/kotlin/io/emeraldpay/dshackle/commons/ExpiringSet.kt deleted file mode 100644 index 8686d8ee..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/commons/ExpiringSet.kt +++ /dev/null @@ -1,152 +0,0 @@ -package io.emeraldpay.dshackle.commons - -import org.apache.commons.collections4.iterators.UnmodifiableIterator -import org.slf4j.LoggerFactory -import java.time.Duration -import java.util.LinkedList -import java.util.TreeSet -import java.util.concurrent.locks.ReentrantLock -import kotlin.concurrent.withLock - -/** - * A naive implementation of a Set with a limit for elements and an expiration time. Supposed to be used a filter for uniqueness. - * Internally it uses a TreeSet and a journal of added elements, which is used ot remove elements when they expire or the list grows too large. - * It's tread safe, but may be suboptimal to use in multithreaded scenario because of internal locks. - */ -class ExpiringSet( - ttl: Duration, - comparator: Comparator, - val limit: Int, -) : MutableSet { - - companion object { - private val log = LoggerFactory.getLogger(ExpiringSet::class.java) - } - - private val tree = TreeSet(comparator) - private val lock = ReentrantLock() - private val journal = LinkedList>() - private var count = 0 - - private val ttl = ttl.toMillis() - - data class JournalItem( - val since: Long = System.currentTimeMillis(), - val value: T - ) { - fun isExpired(ttl: Long): Boolean { - return System.currentTimeMillis() > since + ttl - } - } - - override val size: Int - get() = count - - override fun clear() { - lock.withLock { - tree.clear() - journal.clear() - count = 0 - } - } - - override fun addAll(elements: Collection): Boolean { - var changed = false - elements.forEach { - changed = changed || add(it) - } - return changed - } - - override fun add(element: T): Boolean { - lock.withLock { - val added = tree.add(element) - if (added) { - journal.offer(JournalItem(value = element)) - count++ - shrink() - } - return added - } - } - - override fun isEmpty(): Boolean { - return count == 0 - } - - override fun iterator(): MutableIterator { - // not mutable - return UnmodifiableIterator.unmodifiableIterator(tree.iterator()) - } - - override fun retainAll(elements: Collection): Boolean { - lock.withLock { - var changed = false - val iter = tree.iterator() - while (iter.hasNext()) { - val next = iter.next() - if (!elements.contains(next)) { - changed = true - iter.remove() - count-- - } - } - return changed - } - } - - override fun removeAll(elements: Collection): Boolean { - var changed = false - elements.forEach { - changed = changed || remove(it) - } - return changed - } - - override fun remove(element: T): Boolean { - lock.withLock { - val changed = tree.remove(element) - if (changed) { - count-- - } - return changed - } - } - - override fun containsAll(elements: Collection): Boolean { - return elements.all { contains(it) } - } - - override fun contains(element: T): Boolean { - lock.withLock { - return tree.contains(element) - } - } - - fun shrink() { - lock.withLock { - val iter = journal.iterator() - val removeAtLeast = (count - limit).coerceAtLeast(0) - var removed = 0 - var stop = false - while (!stop && iter.hasNext()) { - val next = iter.next() - val overflow = removeAtLeast > removed - val expired = next.isExpired(ttl) - if (overflow || expired) { - iter.remove() - if (tree.remove(next.value)) { - removed++ - } - } - // we always delete expired elements so don't stop on that - if (!expired) { - // but if we already deleted all non-expired element (i.e., started because it grew too large) - // then we stop as soon as we don't have any overflow - stop = !overflow - } - } - count -= removed - } - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt deleted file mode 100644 index 26a03a5b..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt +++ /dev/null @@ -1,40 +0,0 @@ -package io.emeraldpay.dshackle.data - -import java.util.concurrent.atomic.AtomicReference - -class RingSet( - private val maxSize: Int -) : Set { - private var setRef: AtomicReference> = AtomicReference(LinkedHashSet()) - override val size: Int - get() = setRef.get().size - - fun add(element: T) { - setRef.getAndUpdate { set -> - if (!set.contains(element)) { - val copyset = LinkedHashSet(set) - copyset.add(element) - if (copyset.size > maxSize) { - copyset.remove(set.elementAt(0)) - } - copyset - } else { - set - } - } - } - - override fun isEmpty(): Boolean { - return setRef.get().isEmpty() - } - override fun contains(element: @UnsafeVariance T): Boolean { - return setRef.get().contains(element) - } - override fun iterator(): Iterator { - return setRef.get().iterator() - } - - override fun containsAll(elements: Collection<@UnsafeVariance T>): Boolean { - return setRef.get().containsAll(elements) - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/AggregatedPendingTxes.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/AggregatedPendingTxes.kt index 6cb7fee4..20175033 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/AggregatedPendingTxes.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/AggregatedPendingTxes.kt @@ -15,10 +15,9 @@ */ package io.emeraldpay.dshackle.upstream.ethereum.subscribe -import io.emeraldpay.dshackle.commons.ExpiringSet +import com.github.benmanes.caffeine.cache.Caffeine import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.etherjar.domain.TransactionId -import io.emeraldpay.etherjar.hex.HexDataComparator import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import java.time.Duration @@ -31,15 +30,21 @@ class AggregatedPendingTxes( private val log = LoggerFactory.getLogger(AggregatedPendingTxes::class.java) } - private val track = ExpiringSet( - Duration.ofSeconds(30), - HexDataComparator() as Comparator, - 10_000 - ) + private val track = Caffeine.newBuilder() + .expireAfterWrite(Duration.ofSeconds(30)) + .maximumSize(10_000) + .build() override fun connect(matcher: Selector.Matcher): Flux { return Flux.merge( - sources.map { it.connect(matcher) } // todo check - ).filter(track::add) + sources.map { it.connect(matcher) } + ).filter { + val res = track.getIfPresent(it) + if (res == null) { + track.put(it, true) + return@filter true + } + return@filter false + } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt index 81059e6d..da2c69c7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt @@ -1,8 +1,8 @@ package io.emeraldpay.dshackle.upstream.forkchoice +import com.google.common.cache.CacheBuilder import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.data.RingSet import org.slf4j.LoggerFactory import java.util.concurrent.atomic.AtomicReference @@ -11,7 +11,9 @@ class NoChoiceWithPriorityForkChoice( private val upstreamId: String ) : ForkChoice { private val head = AtomicReference(null) - private val seenBlocks = RingSet(100) + private val seenBlocks = CacheBuilder.newBuilder() + .maximumSize(100) + .build() companion object { private val log = LoggerFactory.getLogger(NoChoiceWithPriorityForkChoice::class.java) @@ -21,7 +23,7 @@ class NoChoiceWithPriorityForkChoice( } override fun filter(block: BlockContainer): Boolean { - return !seenBlocks.contains(block.hash) + return seenBlocks.getIfPresent(block.hash) == null } override fun choose(block: BlockContainer): ForkChoice.ChoiceResult { @@ -31,7 +33,7 @@ class NoChoiceWithPriorityForkChoice( log.debug("Already seen block ${block.height} from $upstreamId") curr } else { - seenBlocks.add(block.hash) + seenBlocks.put(block.hash, true) block.copyWithRating(nodeRating) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt index 4077ddb4..e68399df 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt @@ -1,14 +1,16 @@ package io.emeraldpay.dshackle.upstream.forkchoice +import com.google.common.cache.CacheBuilder import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId -import io.emeraldpay.dshackle.data.RingSet import org.slf4j.LoggerFactory import java.util.concurrent.atomic.AtomicReference class PriorityForkChoice : ForkChoice { private val head = AtomicReference(null) - private val seenBlocks = RingSet(10) + private val seenBlocks = CacheBuilder.newBuilder() + .maximumSize(10) + .build() companion object { private val log = LoggerFactory.getLogger(PriorityForkChoice::class.java) @@ -20,7 +22,7 @@ class PriorityForkChoice : ForkChoice { override fun filter(block: BlockContainer): Boolean { val curr = head.get() - return (curr == null || curr.nodeRating <= block.nodeRating) && !seenBlocks.contains(block.hash) + return (curr == null || curr.nodeRating <= block.nodeRating) && seenBlocks.getIfPresent(block.hash) == null } override fun choose(block: BlockContainer): ForkChoice.ChoiceResult { @@ -33,7 +35,7 @@ class PriorityForkChoice : ForkChoice { curr } else { log.debug("Preparing to accept block ${block.height}") - seenBlocks.add(block.hash) + seenBlocks.put(block.hash, true) block } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/commons/ExpiringSetSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/commons/ExpiringSetSpec.groovy deleted file mode 100644 index ec5e23a3..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/commons/ExpiringSetSpec.groovy +++ /dev/null @@ -1,104 +0,0 @@ -/** - * Copyright (c) 2022 EmeraldPay, Inc - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.commons - -import io.emeraldpay.etherjar.domain.TransactionId -import io.emeraldpay.etherjar.hex.HexDataComparator -import io.emeraldpay.etherjar.tx.Transaction -import spock.lang.Specification - -import java.time.Duration - -class ExpiringSetSpec extends Specification { - - def "Add and check item"() { - setup: - def set = new ExpiringSet(Duration.ofSeconds(60), new HexDataComparator() as Comparator, 100) - - when: - def firstAdded = set.add(TransactionId.from("0xc008b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c")) - def firstExists = set.contains(TransactionId.from("0xc008b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c")) - def secondAdded = set.add(TransactionId.from("0x424e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d")) - def secondExists = set.contains(TransactionId.from("0x424e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d")) - def thirdExists = set.contains(TransactionId.from("0xe218f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0")) - - then: - firstAdded - secondAdded - firstExists - secondExists - !thirdExists - - set.size == 2 - } - - def "Doesn't grow after limit"() { - setup: - def set = new ExpiringSet(Duration.ofSeconds(60), new HexDataComparator() as Comparator, 3) - - when: - set.add(TransactionId.from("0xc008b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c")) - set.add(TransactionId.from("0x424e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d")) - set.add(TransactionId.from("0xe218f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0")) - set.add(TransactionId.from("0xd3082daa344a64369c8aace137f22beb5085351bf111859202bac66b70b28bdd")) - - then: - set.size == 3 - - when: - def firstExists = set.contains(TransactionId.from("0xc008b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c")) - def secondExists = set.contains(TransactionId.from("0x424e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d")) - def thirdExists = set.contains(TransactionId.from("0xe218f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0")) - - then: - !firstExists - secondExists - thirdExists - } - - def "Remove expired"() { - setup: - def set = new ExpiringSet(Duration.ofMillis(100), new HexDataComparator() as Comparator, 100) - - when: - set.add(TransactionId.from("0x1118b506367e1f96fcfdf6b683e84601434f8b655334bc61dae7970f0fb7d02c")) - set.add(TransactionId.from("0x222e36776777ddd2877df0f9d278c37077d1e00afa84defcfb6b367880d8eb6d")) - - then: - set.size == 2 - - when: - Thread.sleep(60) - set.add(TransactionId.from("0x3338f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0")) - set.add(TransactionId.from("0x44482daa344a64369c8aace137f22beb5085351bf111859202bac66b70b28bdd")) - - then: - set.size == 4 - - when: - Thread.sleep(60) - set.add(TransactionId.from("0x55531d466acf4ef72f7e0fbc60a5c2c9d2845b90a3d27b6d7581575cb119cac9")) - set.add(TransactionId.from("0x6665e1f32cb21aee5f27d804cfc65781d5c140b002776bc073f9405479e8b1e5")) - - then: - set.size == 4 - set.contains(TransactionId.from("0x3338f09c3060099cb7302304e2a29ed7a8693bd5dfe3a55cce96fabce27012e0")) - set.contains(TransactionId.from("0x44482daa344a64369c8aace137f22beb5085351bf111859202bac66b70b28bdd")) - set.contains(TransactionId.from("0x55531d466acf4ef72f7e0fbc60a5c2c9d2845b90a3d27b6d7581575cb119cac9")) - set.contains(TransactionId.from("0x6665e1f32cb21aee5f27d804cfc65781d5c140b002776bc073f9405479e8b1e5")) - } - -}