problem: ineffective mem cache implementation

src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt

@@ -16,36 +16,33 @@
  */
 package io.emeraldpay.dshackle.cache
 
+import com.github.benmanes.caffeine.cache.Caffeine
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.reader.Reader
 import reactor.core.publisher.Mono
-import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.ConcurrentLinkedQueue
 
 open class BlocksMemCache(
-        val maxSize: Int = 64
+        maxSize: Int = 64
 ) : Reader<BlockId, BlockContainer> {
 
-    private val mapping = ConcurrentHashMap<BlockId, BlockContainer>()
+    private val mapping = Caffeine.newBuilder()
-    private val queue = ConcurrentLinkedQueue<BlockId>()
+            .maximumSize(maxSize.toLong())
+            .build<BlockId, BlockContainer>()
 
     override fun read(key: BlockId): Mono<BlockContainer> {
-        return Mono.justOrEmpty(mapping[key])
+        return Mono.justOrEmpty(get(key))
     }
 
     open fun get(key: BlockId): BlockContainer? {
-        return mapping[key]
+        return mapping.getIfPresent(key)
     }
 
     open fun add(block: BlockContainer) {
         mapping.put(block.hash, block)
-        queue.add(block.hash)
-
-        while (queue.size > maxSize) {
-            val old = queue.remove()
-            mapping.remove(old)
-        }
     }
 
+    open fun purge() {
+        mapping.cleanUp()
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt

@@ -15,6 +15,7 @@
  */
 package io.emeraldpay.dshackle.cache
 
+import com.github.benmanes.caffeine.cache.Caffeine
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxContainer
@@ -22,8 +23,6 @@ import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.Reader
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
-import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.ConcurrentLinkedQueue
 
 /**
  * Memory cache for transactions
@@ -37,24 +36,25 @@ open class TxMemCache(
         private val log = LoggerFactory.getLogger(TxMemCache::class.java)
     }
 
-    private val mapping = ConcurrentHashMap<TxId, TxContainer>()
+    private val mapping = Caffeine.newBuilder()
-    private val queue = ConcurrentLinkedQueue<TxId>()
+            .maximumSize(maxSize.toLong())
+            .build<TxId, TxContainer>()
 
     override fun read(key: TxId): Mono<TxContainer> {
-        return Mono.justOrEmpty(mapping[key])
+        return Mono.justOrEmpty(mapping.getIfPresent(key))
     }
 
     open fun evict(block: BlockContainer) {
         block.transactions.forEach {
-            mapping.remove(it)
+            mapping.invalidate(it)
         }
     }
 
     open fun evict(block: BlockId) {
-        val ids = mapping.filter { it.value.blockId == block }
+        val ids = mapping.asMap()
-        ids.forEach {
+                .filter { it.value.blockId == block }
-            mapping.remove(it.key)
+                .map { it.key }
-        }
+        mapping.invalidateAll(ids)
     }
 
     open fun add(tx: TxContainer) {
@@ -63,12 +63,9 @@ open class TxMemCache(
             return
         }
         mapping.put(tx.hash, tx)
-        queue.add(tx.hash)
-
-        while (queue.size > maxSize) {
-            val old = queue.remove()
-            mapping.remove(old)
-        }
     }
 
+    open fun purge() {
+        mapping.cleanUp()
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy

@@ -71,6 +71,7 @@ class BlocksMemCacheSpec extends Specification {
 
             cache.add(BlockContainer.from(block))
         }
+        cache.purge()
 
         def act1 = cache.read(BlockId.from(hash1)).block()
         def act2 = cache.read(BlockId.from(hash2)).block()

src/test/groovy/io/emeraldpay/dshackle/cache/TxMemCacheSpec.groovy

@@ -67,6 +67,7 @@ class TxMemCacheSpec extends Specification {
             tx.hash = TransactionId.from(hash)
             cache.add(TxContainer.from(tx))
         }
+        cache.purge()
 
         def act1 = cache.read(TxId.from(hash1)).block()
         def act2 = cache.read(TxId.from(hash2)).block()