StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #100 from p2p-org/move-to-caffeine

Browse Source 
Move from homemade solutions to Caffeine
This commit is contained in:
a10zn8
2022-12-23 14:50:01 +04:00
committed by GitHub
parent 9fb302729f 08b47d128c
commit 5f474678e4
6 changed files with 26 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

152
src/main/kotlin/io/emeraldpay/dshackle/commons/ExpiringSet.kt
View File

@@ -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
}
}
}

40
src/main/kotlin/io/emeraldpay/dshackle/data/RingSet.kt
View File

@@ -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)
}
}

23
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/AggregatedPendingTxes.kt
View File

@@ -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>(
Duration.ofSeconds(30),
HexDataComparator() as Comparator<TransactionId>,
10_000
)
private val track = Caffeine.newBuilder()
.expireAfterWrite(Duration.ofSeconds(30))
.maximumSize(10_000)
.build<TransactionId, Boolean>()
override fun connect(matcher: Selector.Matcher): Flux<TransactionId> {
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
}
}
}

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/NoChoiceWithPriorityForkChoice.kt
View File

@@ -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)
}
}

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/forkchoice/PriorityForkChoice.kt
View File

@@ -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
}
}

104
src/test/groovy/io/emeraldpay/dshackle/commons/ExpiringSetSpec.groovy
View File

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