RocksDB (#3830)

Author: ivan-mashonskiy

benchmark/build.sbt

@@ -3,7 +3,8 @@ enablePlugins(JmhPlugin)
 Jmh / version := "1.33"
 
 libraryDependencies ++= Seq(
-  "org.scodec" %% "scodec-core" % "1.11.10"
+  "org.scodec" %% "scodec-core" % "1.11.10",
+  "org.eclipse.collections"                % "eclipse-collections"   % "11.1.0"
 ) ++ Dependencies.logDeps
 
 // https://github.com/ktoso/sbt-jmh#adding-to-your-project

benchmark/src/main/scala/com/wavesplatform/state/DBState.scala

@@ -5,13 +5,12 @@ import java.io.File
 import com.wavesplatform.Application
 import com.wavesplatform.account.AddressScheme
 import com.wavesplatform.common.state.ByteStr
-import com.wavesplatform.database.{LevelDBWriter, openDB}
+import com.wavesplatform.database.{RDB, RocksDBWriter}
 import com.wavesplatform.lang.directives.DirectiveSet
 import com.wavesplatform.settings.WavesSettings
 import com.wavesplatform.transaction.smart.WavesEnvironment
 import com.wavesplatform.utils.ScorexLogging
 import monix.eval.Coeval
-import org.iq80.leveldb.DB
 import org.openjdk.jmh.annotations.{Param, Scope, State, TearDown}
 
 @State(Scope.Benchmark)
@@ -21,28 +20,29 @@ abstract class DBState extends ScorexLogging {
 
   lazy val settings: WavesSettings = Application.loadApplicationConfig(Some(new File(configFile)).filter(_.exists()))
 
-  lazy val db: DB = openDB(settings.dbSettings.directory)
+  lazy val rdb: RDB = RDB.open(settings.dbSettings)
 
-  lazy val levelDBWriter: LevelDBWriter =
-    LevelDBWriter.readOnly(
-      db,
-      settings.copy(dbSettings = settings.dbSettings.copy(maxCacheSize = 1))
+  lazy val rocksDBWriter: RocksDBWriter =
+    new RocksDBWriter(
+      rdb,
+      settings.blockchainSettings,
+      settings.dbSettings.copy(maxCacheSize = 1)
     )
 
   AddressScheme.current = new AddressScheme { override val chainId: Byte = 'W' }
 
   lazy val environment = new WavesEnvironment(
     AddressScheme.current.chainId,
     Coeval.raiseError(new NotImplementedError("`tx` is not implemented")),
-    Coeval(levelDBWriter.height),
-    levelDBWriter,
+    Coeval(rocksDBWriter.height),
+    rocksDBWriter,
     null,
     DirectiveSet.contractDirectiveSet,
     ByteStr.empty
   )
 
   @TearDown
   def close(): Unit = {
-    db.close()
+    rdb.close()
   }
 }

benchmark/src/test/scala/com/wavesplatform/RollbackBenchmark.scala

@@ -1,29 +1,29 @@
 package com.wavesplatform
 
 import java.io.File
+
 import com.google.common.primitives.Ints
 import com.google.protobuf.ByteString
 import com.wavesplatform.account.{Address, AddressScheme, KeyPair}
 import com.wavesplatform.block.Block
 import com.wavesplatform.common.state.ByteStr
 import com.wavesplatform.common.utils.*
-import com.wavesplatform.database.{LevelDBWriter, openDB}
+import com.wavesplatform.database.{RDB, RocksDBWriter}
 import com.wavesplatform.protobuf.transaction.PBRecipients
 import com.wavesplatform.state.{Diff, Portfolio}
-import com.wavesplatform.transaction.{GenesisTransaction, Proofs, TxDecimals, TxPositiveAmount}
 import com.wavesplatform.transaction.Asset.IssuedAsset
 import com.wavesplatform.transaction.assets.IssueTransaction
+import com.wavesplatform.transaction.{GenesisTransaction, Proofs, TxDecimals, TxPositiveAmount}
 import com.wavesplatform.utils.{NTP, ScorexLogging}
-import monix.reactive.Observer
 
 import scala.collection.immutable.VectorMap
 
 object RollbackBenchmark extends ScorexLogging {
   def main(args: Array[String]): Unit = {
     val settings      = Application.loadApplicationConfig(Some(new File(args(0))))
-    val db            = openDB(settings.dbSettings.directory)
+    val rdb           = RDB.open(settings.dbSettings)
     val time          = new NTP(settings.ntpServer)
-    val levelDBWriter = LevelDBWriter(db, Observer.stopped, settings)
+    val rocksDBWriter = new RocksDBWriter(rdb, settings.blockchainSettings, settings.dbSettings)
 
     val issuer = KeyPair(new Array[Byte](32))
 
@@ -73,7 +73,7 @@ object RollbackBenchmark extends ScorexLogging {
     } yield address -> Portfolio(assets = map)
 
     log.info("Appending genesis block")
-    levelDBWriter.append(
+    rocksDBWriter.append(
       Diff(portfolios = portfolios.toMap),
       0,
       0,
@@ -89,13 +89,13 @@ object RollbackBenchmark extends ScorexLogging {
     val nextDiff = Diff(portfolios = addresses.map(_ -> Portfolio(1, assets = VectorMap(IssuedAsset(assets.head.id()) -> 1L))).toMap)
 
     log.info("Appending next block")
-    levelDBWriter.append(nextDiff, 0, 0, None, ByteStr.empty, nextBlock)
+    rocksDBWriter.append(nextDiff, 0, 0, None, ByteStr.empty, nextBlock)
 
     log.info("Rolling back")
     val start = System.nanoTime()
-    levelDBWriter.rollbackTo(1)
+    rocksDBWriter.rollbackTo(1)
     val end = System.nanoTime()
     log.info(f"Rollback took ${(end - start) * 1e-6}%.3f ms")
-    levelDBWriter.close()
+    rdb.close()
   }
 }

benchmark/src/test/scala/com/wavesplatform/lang/v1/EvaluatorV2Benchmark.scala

@@ -1,15 +1,14 @@
 package com.wavesplatform.lang.v1
 
 import java.util.concurrent.TimeUnit
-
 import cats.Id
 import com.wavesplatform.lang.Common
 import com.wavesplatform.lang.directives.values.{V1, V3}
 import com.wavesplatform.lang.v1.EvaluatorV2Benchmark.*
 import com.wavesplatform.lang.v1.compiler.Terms.{EXPR, IF, TRUE}
 import com.wavesplatform.lang.v1.compiler.TestCompiler
 import com.wavesplatform.lang.v1.evaluator.EvaluatorV2
-import com.wavesplatform.lang.v1.evaluator.ctx.{EvaluationContext, LoggedEvaluationContext}
+import com.wavesplatform.lang.v1.evaluator.ctx.{DisabledLogEvaluationContext, EvaluationContext}
 import com.wavesplatform.lang.v1.evaluator.ctx.impl.PureContext
 import com.wavesplatform.lang.v1.traits.Environment
 import org.openjdk.jmh.annotations.*
@@ -20,7 +19,7 @@ import scala.annotation.tailrec
 object EvaluatorV2Benchmark {
   val pureContext: CTX[Environment]                       = PureContext.build(V1, useNewPowPrecision = true).withEnvironment[Environment]
   val pureEvalContext: EvaluationContext[Environment, Id] = pureContext.evaluationContext(Common.emptyBlockchainEnvironment())
-  val evaluatorV2: EvaluatorV2                            = new EvaluatorV2(LoggedEvaluationContext(_ => _ => (), pureEvalContext), V1, true, true)
+  val evaluatorV2: EvaluatorV2                            = new EvaluatorV2(DisabledLogEvaluationContext(pureEvalContext), V1, true, true, false)
 }
 
 @OutputTimeUnit(TimeUnit.MILLISECONDS)

benchmark/src/test/scala/com/wavesplatform/lang/v1/FractionIntBenchmark.scala

@@ -20,22 +20,22 @@ import org.openjdk.jmh.infra.Blackhole
 @Measurement(iterations = 10, time = 1)
 class FractionIntBenchmark {
   @Benchmark
-  def fraction1(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr1, LogExtraInfo(), V5, true, true))
+  def fraction1(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr1, LogExtraInfo(), V5, true, true, false))
 
   @Benchmark
-  def fraction2(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr2, LogExtraInfo(), V5, true, true))
+  def fraction2(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr2, LogExtraInfo(), V5, true, true, false))
 
   @Benchmark
-  def fraction3(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr3, LogExtraInfo(), V5, true, true))
+  def fraction3(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr3, LogExtraInfo(), V5, true, true, false))
 
   @Benchmark
-  def fraction1Round(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr1Round, LogExtraInfo(), V5, true, true))
+  def fraction1Round(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr1Round, LogExtraInfo(), V5, true, true, false))
 
   @Benchmark
-  def fraction2Round(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr2Round, LogExtraInfo(), V5, true, true))
+  def fraction2Round(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr2Round, LogExtraInfo(), V5, true, true, false))
 
   @Benchmark
-  def fraction3Round(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr3Round, LogExtraInfo(), V5, true, true))
+  def fraction3Round(bh: Blackhole, s: St): Unit = bh.consume(EvaluatorV2.applyCompleted(s.ctx, s.expr3Round, LogExtraInfo(), V5, true, true, false))
 }
 
 @State(Scope.Benchmark)

benchmark/src/test/scala/com/wavesplatform/lang/v1/package.scala

@@ -31,5 +31,5 @@ package object v1 {
       expr: EXPR,
       stdLibVersion: StdLibVersion
   ): (Log[Id], Int, Either[ExecutionError, Terms.EVALUATED]) =
-    EvaluatorV2.applyCompleted(ctx, expr, LogExtraInfo(), stdLibVersion, newMode = true, correctFunctionCallScope = true)
+    EvaluatorV2.applyCompleted(ctx, expr, LogExtraInfo(), stdLibVersion, newMode = true, correctFunctionCallScope = true, enableExecutionLog = false)
 }

benchmark/src/test/scala/com/wavesplatform/state/BaseState.scala

@@ -3,35 +3,35 @@ package com.wavesplatform.state
 import java.io.File
 import java.nio.file.Files
 
+import com.typesafe.config.ConfigFactory
 import com.wavesplatform.account.KeyPair
 import com.wavesplatform.block.Block
 import com.wavesplatform.common.utils.EitherExt2
-import com.wavesplatform.database.{LevelDBFactory, LevelDBWriter}
+import com.wavesplatform.database.{RDB, RocksDBWriter}
 import com.wavesplatform.lagonaki.mocks.TestBlock
 import com.wavesplatform.mining.MiningConstraint
-import com.wavesplatform.settings.FunctionalitySettings
+import com.wavesplatform.settings.{FunctionalitySettings, WavesSettings, loadConfig}
 import com.wavesplatform.state.diffs.BlockDiffer
-import com.wavesplatform.state.utils.TestLevelDB
+import com.wavesplatform.state.utils.TestRocksDB
 import com.wavesplatform.transaction.{GenesisTransaction, Transaction}
-import monix.execution.UncaughtExceptionReporter
-import monix.reactive.Observer
-import org.iq80.leveldb.{DB, Options}
 import org.openjdk.jmh.annotations.{Setup, TearDown}
 import org.scalacheck.{Arbitrary, Gen}
 
 trait BaseState {
-  import BaseState._
+  import BaseState.*
 
+  val benchSettings: Settings = Settings.fromConfig(ConfigFactory.load())
+  val wavesSettings: WavesSettings = {
+    val config = loadConfig(ConfigFactory.parseFile(new File(benchSettings.networkConfigFile)))
+    WavesSettings.fromRootConfig(config)
+  }
   private val fsSettings: FunctionalitySettings = updateFunctionalitySettings(FunctionalitySettings.TESTNET)
-  private val db: DB = {
-    val dir     = Files.createTempDirectory("state-synthetic").toAbsolutePath.toString
-    val options = new Options()
-    options.createIfMissing(true)
-    LevelDBFactory.factory.open(new File(dir), options)
+  private val rdb: RDB = {
+    val dir = Files.createTempDirectory("state-synthetic").toAbsolutePath.toString
+    RDB.open(wavesSettings.dbSettings.copy(directory = dir))
   }
 
-  private val portfolioChanges = Observer.empty(UncaughtExceptionReporter.default)
-  val state: LevelDBWriter     = TestLevelDB.withFunctionalitySettings(db, portfolioChanges, fsSettings)
+  val state: RocksDBWriter = TestRocksDB.withFunctionalitySettings(rdb, fsSettings)
 
   private var _richAccount: KeyPair = _
   def richAccount: KeyPair          = _richAccount
@@ -52,12 +52,11 @@ trait BaseState {
       transferTxs <- Gen.sequence[Vector[Transaction], Transaction]((1 to TxsInBlock).map { i =>
         txGenP(sender, base.header.timestamp + i)
       })
-    } yield
-      TestBlock.create(
-        time = transferTxs.last.timestamp,
-        ref = base.id(),
-        txs = transferTxs
-      )
+    } yield TestBlock.create(
+      time = transferTxs.last.timestamp,
+      ref = base.id(),
+      txs = transferTxs
+    )
 
   private val initGen: Gen[(KeyPair, Block)] = for {
     rich <- accountGen
@@ -98,7 +97,7 @@ trait BaseState {
 
   @TearDown
   def close(): Unit = {
-    db.close()
+    rdb.close()
   }
 }