Added compression of data which saving in HistoryStore

hilltracer · Jan 20, 2022 · 852f9f8 · 852f9f8
1 parent 54faa48
commit 852f9f8
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 6 deletions.
diff --git a/rspace/src/main/scala/coop/rchain/rspace/RSpace.scala b/rspace/src/main/scala/coop/rchain/rspace/RSpace.scala
@@ -232,7 +232,7 @@ class RSpace[F[_]: Concurrent: ContextShift: Log: Metrics: Span, C, P, A, K](
 
         h = historyRepositoryAtom.get()
         debugMessage <- if (blockNumber != 0L) for {
-                         historyNumAndSize <- h.numRecordsAndSizeBytesHistory
+                         historyNumAndSize <- (0, 0L).pure
                          stroreNum         = h.numRecordsStore()
                          stroreSize        = h.sizeBytesStore()
                          r = "[blockNumber, storeNumNodes, storeSizeBytes, historyNumNodes, historySizeBytes],  " + blockNumber.toString +

diff --git a/rspace/src/main/scala/coop/rchain/rspace/history/RadixTree.scala b/rspace/src/main/scala/coop/rchain/rspace/history/RadixTree.scala
@@ -3,6 +3,8 @@ package coop.rchain.rspace.history
 import cats.effect.Sync
 import cats.syntax.all._
 import cats.{Monad, Parallel}
+import coop.rchain.shared.Compression
+import net.jpountz.lz4.{LZ4CompressorWithLength, LZ4DecompressorWithLength}
 import scodec.bits.ByteVector
 
 import scala.annotation.tailrec
@@ -440,16 +442,31 @@ object RadixTree {
 
   class RadixTreeImpl[F[_]: Sync: Parallel](store: RadixStore[F]) {
 
+    // Compression
+
+    private val compressor = new LZ4CompressorWithLength(Compression.factory.fastCompressor())
+    // val compressor = new LZ4CompressorWithLength(factory.highCompressor(17)) // Max compression
+    private val decompressor = new LZ4DecompressorWithLength(Compression.factory.fastDecompressor())
+
+    def compressBytes(bytes: ByteVector): ByteVector =
+      ByteVector(compressor.compress(bytes.toArray))
+
+    def decompressBytes(bytes: ByteVector): F[ByteVector] =
+      Sync[F].delay(ByteVector(decompressor.decompress(bytes.toArray))).handleErrorWith { ex =>
+        new Exception("Decompress of block failed.", ex).raiseError
+      }
+
     /**
       * Load and decode serializing data from KVDB.
       */
     private def loadNodeFromStore(nodePtr: ByteVector): F[Option[Node]] =
       for {
         nodeOpt <- store.get(Seq(nodePtr))
-        r = nodeOpt.head match {
-          case None       => None
-          case Some(node) => Some(codecs.decode(node))
-        }
+        r <- nodeOpt.head match {
+              case None => None.pure
+              case Some(nodeCompress) =>
+                decompressBytes(nodeCompress).map(node => Some(codecs.decode(node)))
+            }
       } yield r
 
     /**
@@ -513,7 +530,7 @@ object RadixTree {
           )
         case None => cacheR.update(hash, node)
       }
-      cacheW.update(hash, bytes)
+      cacheW.update(hash, compressBytes(bytes))
       hash
     }