Merge pull request #366 from ergoplatform/i332

Use cache in HistoryStorage

build.sbt

@@ -5,7 +5,7 @@ organization := "org.ergoplatform"
 
 name := "ergo"
 
-version := "1.5.1"
+version := "1.5.2-SNAPSHOT"
 
 scalaVersion := "2.12.6"
 
@@ -17,6 +17,7 @@ resolvers ++= Seq("Sonatype Releases" at "https://oss.sonatype.org/content/repos
 val scorexVersion = "941a63a8-SNAPSHOT"
 
 libraryDependencies ++= Seq(
+  "com.google.guava" % "guava" % "21.0",
   "org.scorexfoundation" %% "scrypto" % "2.1.2",
   "org.scorexfoundation" %% "sigma-state" % "0.9.4",
   "org.scala-lang.modules" %% "scala-async" % "0.9.7",

lock.sbt

@@ -61,4 +61,4 @@ dependencyOverrides in ThisBuild ++= Seq(
   "org.typelevel" % "macro-compat_2.12" % "1.1.1",
   "org.whispersystems" % "curve25519-java" % "0.5.0"
 )
-// LIBRARY_DEPENDENCIES_HASH 20410da74eb095d244152f4fabc1f8990cdda9c9
+// LIBRARY_DEPENDENCIES_HASH 6481de067f686370800c1dc0427c1b7fd6837b7d

src/main/resources/application.conf

@@ -44,6 +44,11 @@ ergo {
     keepPoolSize = 1
   }
 
+  cache {
+    # Number of recently used modifiers that will be kept in memory
+    historyStorageCacheSize = 1000
+  }
+
   # Chain-specific settings. Change only if you are going to launch a new chain!
   chain {
 

src/main/scala/org/ergoplatform/nodeView/history/ErgoHistory.scala

@@ -185,7 +185,7 @@ object ErgoHistory extends ScorexLogging {
     historyFolder.mkdirs()
     val indexStore = new LSMStore(historyFolder, keepVersions = 0)
     val objectsStore = new FilesObjectsStore(historyFolder.getAbsolutePath)
-    val db = new HistoryStorage(indexStore, objectsStore)
+    val db = new HistoryStorage(indexStore, objectsStore, settings.cacheSettings)
     val nodeSettings = settings.nodeSettings
 
     val history: ErgoHistory = (nodeSettings.verifyTransactions, nodeSettings.PoPoWBootstrap) match {

src/main/scala/org/ergoplatform/nodeView/history/storage/HistoryStorage.scala

@@ -1,26 +1,45 @@
 package org.ergoplatform.nodeView.history.storage
 
+import com.google.common.cache.CacheBuilder
 import io.iohk.iodb.{ByteArrayWrapper, Store}
 import org.ergoplatform.modifiers.ErgoPersistentModifier
 import org.ergoplatform.modifiers.history.HistoryModifierSerializer
+import org.ergoplatform.settings.{Algos, CacheSettings}
 import scorex.core.ModifierId
 import scorex.core.utils.{ScorexEncoding, ScorexLogging}
 
-import scala.util.{Failure, Success}
+import scala.util.Failure
 
-class HistoryStorage(indexStore: Store, objectsStore: ObjectsStore) extends ScorexLogging with AutoCloseable
-  with ScorexEncoding {
+class HistoryStorage(indexStore: Store, objectsStore: ObjectsStore, config: CacheSettings) extends ScorexLogging
+  with AutoCloseable with ScorexEncoding {
 
-  def modifierById(id: ModifierId): Option[ErgoPersistentModifier] = objectsStore.get(id)
-    .flatMap { bBytes =>
-      HistoryModifierSerializer.parseBytes(bBytes) match {
-        case Success(b) =>
-          Some(b)
-        case Failure(e) =>
-          log.warn(s"Failed to parse modifier ${encoder.encode(id)} from db (bytes are: ${bBytes.mkString("-")}): ", e)
-          None
-      }
+  private val modifiersCache = CacheBuilder.newBuilder()
+    .maximumSize(config.historyStorageCacheSize)
+    .build[String, ErgoPersistentModifier]
+
+  private def keyById(id: ModifierId): String = Algos.encode(id)
+
+  def modifierById(id: ModifierId): Option[ErgoPersistentModifier] = {
+    val key = keyById(id)
+    Option(modifiersCache.getIfPresent(key)) match {
+      case Some(e) =>
+        log.trace(s"Got modifier $key from cache")
+        Some(e)
+      case None =>
+        objectsStore.get(id).flatMap { bBytes =>
+          HistoryModifierSerializer.parseBytes(bBytes).recoverWith { case e =>
+            log.warn(s"Failed to parse modifier ${encoder.encode(id)} from db (bytes are: ${Algos.encode(bBytes)})")
+            Failure(e)
+          }.toOption match {
+            case Some(pm) =>
+              log.trace(s"Cache miss for existing modifier $key")
+              modifiersCache.put(key, pm)
+              Some(pm)
+            case None => None
+          }
+        }
     }
+  }
 
   def getIndex(id: ByteArrayWrapper): Option[ByteArrayWrapper] = indexStore.get(id)
 
@@ -38,7 +57,10 @@ class HistoryStorage(indexStore: Store, objectsStore: ObjectsStore) extends Scor
       indexesToInsert)
   }
 
-  def remove(idsToRemove: Seq[ModifierId]): Unit = idsToRemove.foreach(id => objectsStore.delete(id))
+  def remove(idsToRemove: Seq[ModifierId]): Unit = idsToRemove.foreach { id =>
+    modifiersCache.invalidate(keyById(id))
+    objectsStore.delete(id)
+  }
 
   override def close(): Unit = {
     log.warn("Closing history storage...")

src/main/scala/org/ergoplatform/settings/CacheSettings.scala

@@ -0,0 +1,11 @@
+package org.ergoplatform.settings
+
+/**
+  * Configuration file for different caches
+  * @see src/main/resources/application.conf for parameters description
+  */
+case class CacheSettings(historyStorageCacheSize: Int)
+
+object CacheSettings {
+  val default: CacheSettings = CacheSettings(100)
+}

src/main/scala/org/ergoplatform/settings/ErgoSettings.scala

@@ -14,6 +14,7 @@ case class ErgoSettings(directory: String,
                         chainSettings: ChainSettings,
                         testingSettings: TestingSettings,
                         nodeSettings: NodeConfigurationSettings,
+                        cacheSettings: CacheSettings,
                         scorexSettings: ScorexSettings)
 
 object ErgoSettings extends ScorexLogging
@@ -35,14 +36,15 @@ object ErgoSettings extends ScorexLogging
     val nodeSettings = config.as[NodeConfigurationSettings](s"$configPath.node")
     val chainSettings = config.as[ChainSettings](s"$configPath.chain")
     val testingSettings = config.as[TestingSettings](s"$configPath.testing")
+    val cacheSettings = config.as[CacheSettings](s"$configPath.cache")
     val scorexSettings = config.as[ScorexSettings](scorexConfigPath)
 
     if (nodeSettings.stateType == Digest && nodeSettings.mining) {
       log.error("Malformed configuration file was provided! Mining is not possible with digest state. Aborting!")
       ErgoApp.forceStopApplication()
     }
 
-    consistentSettings(ErgoSettings(directory, chainSettings, testingSettings, nodeSettings, scorexSettings))
+    consistentSettings(ErgoSettings(directory, chainSettings, testingSettings, nodeSettings, cacheSettings, scorexSettings))
   }
 
   private def readConfigFromPath(userConfigPath: Option[String]): Config = {

src/test/scala/org/ergoplatform/api/routes/Stubs.scala

@@ -149,7 +149,7 @@ trait Stubs extends ErgoGenerators with ErgoTestHelpers with ChainGenerator with
 
     val dir = createTempDir
     val fullHistorySettings: ErgoSettings = ErgoSettings(dir.getAbsolutePath, chainSettings, testingSettings,
-      nodeSettings, scorexSettings)
+      nodeSettings, CacheSettings.default, scorexSettings)
 
     ErgoHistory.readOrGenerate(fullHistorySettings, timeProvider)
   }

src/test/scala/org/ergoplatform/utils/HistorySpecification.scala

@@ -51,7 +51,7 @@ trait HistorySpecification extends ErgoPropertyTest {
 
     val dir = createTempDir
     val fullHistorySettings: ErgoSettings = ErgoSettings(dir.getAbsolutePath, chainSettings, testingSettings,
-      nodeSettings, scorexSettings)
+      nodeSettings, CacheSettings.default, scorexSettings)
 
     ErgoHistory.readOrGenerate(fullHistorySettings, timeProvider)
   }