Move from homemade solutions to Caffeine

src/main/kotlin/io/emeraldpay/dshackle/commons/ExpiringSet.kt

@@ -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<T>(
-    ttl: Duration,
-    comparator: Comparator<T>,
-    val limit: Int,
-) : MutableSet<T> {
-
-    companion object {
-        private val log = LoggerFactory.getLogger(ExpiringSet::class.java)
-    }
-
-    private val tree = TreeSet<T>(comparator)
-    private val lock = ReentrantLock()
-    private val journal = LinkedList<JournalItem<T>>()
-    private var count = 0
-
-    private val ttl = ttl.toMillis()
-
-    data class JournalItem<T>(
-        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<T>): 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<T> {
-        // not mutable
-        return UnmodifiableIterator.unmodifiableIterator(tree.iterator())
-    }
-
-    override fun retainAll(elements: Collection<T>): 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<T>): 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<T>): 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
-        }
-    }
-}

src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt

@@ -1,40 +0,0 @@
-package io.emeraldpay.dshackle.data
-
-import java.util.concurrent.atomic.AtomicReference
-
-class RingSet<T>(
-    private val maxSize: Int
-) : Set<T> {
-    private var setRef: AtomicReference<LinkedHashSet<T>> = AtomicReference(LinkedHashSet<T>())
-    override val size: Int
-        get() = setRef.get().size
-
-    fun add(element: T) {
-        setRef.getAndUpdate { set ->
-            if (!set.contains(element)) {
-                val copyset = LinkedHashSet<T>(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<T> {
-        return setRef.get().iterator()
-    }
-
-    override fun containsAll(elements: Collection<@UnsafeVariance T>): Boolean {
-        return setRef.get().containsAll(elements)
-    }
-}

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<TransactionId>(
+    private val track = Caffeine.newBuilder()
-        Duration.ofSeconds(30),
+        .expireAfterWrite(Duration.ofSeconds(30))
-        HexDataComparator() as Comparator<TransactionId>,
+        .maximumSize(10_000)
-        10_000
+        .build<TransactionId, Boolean>()
-    )
 
     override fun connect(matcher: Selector.Matcher): Flux<TransactionId> {
         return Flux.merge(
-            sources.map { it.connect(matcher) } // todo check
+            sources.map { it.connect(matcher) }
-        ).filter(track::add)
+        ).filter {
+            val res = track.getIfPresent(it)
+            if (res == null) {
+                track.put(it, true)
+                return@filter true
+            }
+            return@filter false
+        }
     }
 }

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<BlockContainer>(null)
-    private val seenBlocks = RingSet<BlockId>(100)
+    private val seenBlocks = CacheBuilder.newBuilder()
+        .maximumSize(100)
+        .build<BlockId, Boolean>()
 
     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)
             }
         }

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<BlockContainer>(null)
-    private val seenBlocks = RingSet<BlockId>(10)
+    private val seenBlocks = CacheBuilder.newBuilder()
+        .maximumSize(10)
+        .build<BlockId, Boolean>()
 
     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
             }
         }

src/test/groovy/io/emeraldpay/dshackle/commons/ExpiringSetSpec.groovy

@@ -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<TransactionId>, 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<TransactionId>, 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<TransactionId>, 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"))
-    }
-
-}