[Retention] (1) Added retention (2) Stop implemented

blockchain/blockchain.go

@@ -109,13 +109,14 @@ const (
 )
 
 const (
-	DefaultTriesInMemory    = 128
-	DefaultBlockInterval    = 128
-	DefaultPruningRetention = 172800 // 2*params.DefaultStakeUpdateInterval
-	MaxPrefetchTxs          = 20000
-	DefaultChunkBlockSize   = uint64(10000)
-	MB_1                    = uint64(1000000)
-	DefaultCompressChunkCap = MB_1 * 100 // 100MB
+	DefaultTriesInMemory     = 128
+	DefaultBlockInterval     = 128
+	DefaultPruningRetention  = 172800 // 2*params.DefaultStakeUpdateInterval
+	MaxPrefetchTxs           = 20000
+	MB_1                     = uint64(1000000)
+	DefaultChunkBlockSize    = uint64(10000)
+	DefaultCompressChunkCap  = MB_1 * 100 // 100MB
+	DefaultCompressRetention = 86400
 
 	// BlockChainVersion ensures that an incompatible database forces a resync from scratch.
 	// Changelog:

cmd/utils/config.go

@@ -591,6 +591,7 @@ func (kCfg *KaiaConfig) SetKaiaConfig(ctx *cli.Context, stack *node.Node) {
 
 	cfg.CompressBlockChunkSize = ctx.Uint64(CompressBlockChunkSizeFlag.Name)
 	cfg.CompressChunkCap = ctx.Uint64(CompressChunkCapFlag.Name)
+	cfg.CompressRetention = ctx.Uint64(CompressRetentionFlag.Name)
 
 	if gcmode := ctx.String(GCModeFlag.Name); gcmode != "full" && gcmode != "archive" {
 		log.Fatalf("--%s must be either 'full' or 'archive'", GCModeFlag.Name)

cmd/utils/flaggroup.go

@@ -116,6 +116,7 @@ var FlagGroups = []FlagGroup{
 			DBNoPerformanceMetricsFlag,
 			CompressBlockChunkSizeFlag,
 			CompressChunkCapFlag,
+			CompressRetentionFlag,
 			TxPruningFlag,
 			TxPruningRetentionFlag,
 			ReceiptPruningFlag,

cmd/utils/flags.go

@@ -489,21 +489,29 @@ var (
 		Category: "DATABASE",
 	}
 	CompressBlockChunkSizeFlag = &cli.Uint64Flag{
-		Name:     "database.compress-check-block-size",
+		Name:     "database.compress-chunk-block-size",
 		Usage:    "Number of blocks which chunk can accomodate",
 		Value:    blockchain.DefaultChunkBlockSize,
 		Aliases:  []string{},
 		EnvVars:  []string{"KLAYTN_COMPRESS_BLOCK_CHUNK_SIZE", "KAIA_COMPRESS_BLOCK_CHUNK_SIZE"},
 		Category: "DATABASE",
 	}
 	CompressChunkCapFlag = &cli.Uint64Flag{
-		Name:     "database.compress-check-cap",
+		Name:     "database.compress-chunk-cap",
 		Usage:    "Byte size which chunk can accomodate",
 		Value:    blockchain.DefaultCompressChunkCap,
 		Aliases:  []string{},
 		EnvVars:  []string{"KLAYTN_COMPRESS_CHUNK_CAP", "KAIA_COMPRESS_CHUNK_CAP"},
 		Category: "DATABASE",
 	}
+	CompressRetentionFlag = &cli.Uint64Flag{
+		Name:     "database.compress-retention",
+		Usage:    "Number of blocks from the latest block where compression should not be performed",
+		Value:    blockchain.DefaultCompressRetention,
+		Aliases:  []string{},
+		EnvVars:  []string{"KLAYTN_COMPRESS_RETENTIONP", "KAIA_COMPRESS_RETENTION"},
+		Category: "DATABASE",
+	}
 	SenderTxHashIndexingFlag = &cli.BoolFlag{
 		Name:     "sendertxhashindexing",
 		Usage:    "Enables storing mapping information of senderTxHash to txHash",

cmd/utils/nodeflags.go

@@ -199,6 +199,7 @@ var CommonNodeFlags = []cli.Flag{
 	altsrc.NewBoolFlag(LevelDBNoBufferPoolFlag),
 	altsrc.NewUint64Flag(CompressBlockChunkSizeFlag),
 	altsrc.NewUint64Flag(CompressChunkCapFlag),
+	altsrc.NewUint64Flag(CompressRetentionFlag),
 	altsrc.NewBoolFlag(DBNoPerformanceMetricsFlag),
 	altsrc.NewBoolFlag(RocksDBSecondaryFlag),
 	altsrc.NewUint64Flag(RocksDBCacheSizeFlag),

kaiax/compress/impl/compress.go

@@ -34,6 +34,22 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+const SEC_TEN = time.Second * 10
+
+func (c *CompressModule) stopCompress() {
+	for range allCompressTypes {
+		c.terminateCompress <- struct{}{}
+	}
+	for {
+		if len(c.terminateCompress) == 0 {
+			for _, compressTyp := range allCompressTypes {
+				c.setIdleState(compressTyp, nil)
+			}
+			return
+		}
+	}
+}
+
 func (c *CompressModule) Compress() {
 	go c.compressHeader()
 	go c.compressBody()
@@ -52,30 +68,68 @@ func (c *CompressModule) compressReceipts() {
 	c.compress(ReceiptCompressType, compressReceipts)
 }
 
+func (c *CompressModule) idle(compressTyp CompressionType, nBlocks uint64) bool {
+	idle := false
+	for {
+		select {
+		case <-c.terminateCompress:
+			logger.Info("[Compression] Stop signal received", "type", compressTyp.String())
+			return true
+		case <-time.After(SEC_TEN):
+			return false
+		default:
+			if !idle {
+				idealIdleTime := time.Second * time.Duration(nBlocks)
+				c.setIdleState(compressTyp, &IdleState{true, idealIdleTime})
+				logger.Info("[Compression] Enter idle state", "type", compressTyp.String(), "idle", SEC_TEN, "ideal idle time", idealIdleTime)
+				idle = true
+			}
+		}
+	}
+}
+
 func (c *CompressModule) compress(compressTyp CompressionType, compressFn CompressFn) {
-	var (
-		SEC_TEN     = time.Second * 10
-		totalChunks = 0
-	)
+	totalChunks := 0
 	for {
+		select {
+		case <-c.terminateCompress:
+			logger.Info("[Compression] Stop signal received", "type", compressTyp.String())
+			return
+		default:
+		}
+
 		var (
-			curBlkNum             = c.Chain.CurrentBlock().NumberU64()
-			residualBlkCnt        = curBlkNum % c.getCompressChunk()
-			nextCompressionBlkNum = readSubsequentCompressionBlkNumber(c.Dbm, compressTyp)
+			curBlkNum               = c.Chain.CurrentBlock().NumberU64()
+			residualBlkCnt          = curBlkNum % c.getCompressChunk()
+			nextCompressionBlkNum   = readSubsequentCompressionBlkNumber(c.Dbm, compressTyp)
+			nextCompressionDistance = curBlkNum - nextCompressionBlkNum
 			// Do not wait if next compression block number is far awway. Start migration right now
-			noWait     = curBlkNum > nextCompressionBlkNum && curBlkNum-nextCompressionBlkNum > c.getCompressChunk()
+			noWait     = curBlkNum > nextCompressionBlkNum && nextCompressionDistance > c.getCompressChunk()
 			originFrom = readSubsequentCompressionBlkNumber(c.Dbm, compressTyp)
 			from       = originFrom
 		)
+		// 1. Idle check
 		if curBlkNum < c.getCompressChunk() || (residualBlkCnt != 0 && !noWait) {
-			idealIdleTime := time.Second * time.Duration(c.getCompressChunk()-residualBlkCnt)
-			c.setIdleState(compressTyp, &IdleState{true, idealIdleTime})
-			logger.Info("[Compression] Enter idle state", "type", compressTyp.String(), "idle", SEC_TEN, "ideal idle time", idealIdleTime)
-			time.Sleep(SEC_TEN)
+			if c.idle(compressTyp, c.getCompressChunk()-residualBlkCnt) {
+				return
+			}
+			continue
+		}
+		if c.getCompressRetention() > nextCompressionDistance {
+			if c.idle(compressTyp, c.getCompressRetention()-nextCompressionDistance) {
+				return
+			}
 			continue
 		}
 		c.setIdleState(compressTyp, nil)
+		// 2. Main loop (compression)
 		for {
+			select {
+			case <-c.terminateCompress:
+				logger.Info("[Compression] Stop signal received", "type", compressTyp.String())
+				return
+			default:
+			}
 			subsequentBlkNumber, compressedSize, err := compressFn(c.Dbm, from, 0, curBlkNum, c.getCompressChunk(), c.getChunkCap(), true)
 			if err != nil {
 				logger.Warn("[Compression] failed to compress chunk", "type", compressTyp.String(), "err", err)
@@ -88,6 +142,7 @@ func (c *CompressModule) compress(compressTyp CompressionType, compressFn Compre
 			from = subsequentBlkNumber
 			totalChunks++
 		}
+
 	}
 }
 
@@ -128,7 +183,7 @@ func (c *CompressModule) FindReceiptsFromChunkWithBlkHash(dbm database.DBManager
 }
 
 func (c *CompressModule) restoreFragmentByRewind() {
-	for _, compressTyp := range []CompressionType{HeaderCompressType, BodyCompressType, ReceiptCompressType} {
+	for _, compressTyp := range allCompressTypes {
 		var (
 			lastCompressDeleteKeyPrefix, lastCompressDeleteValuePrefix = getLsatCompressionDeleteKeyPrefix(compressTyp), getLsatCompressionDeleteValuePrefix(compressTyp)
 			miscDB                                                     = c.Dbm.GetMiscDB()
@@ -297,7 +352,7 @@ func (c *CompressModule) testCompressPerformance(from, to uint64) error {
 		totalBodyFinderElapsedTime         time.Duration
 		totalReceiptsFinderElapsedTime     time.Duration
 	)
-	for _, compressTyp := range []CompressionType{HeaderCompressType, BodyCompressType, ReceiptCompressType} {
+	for _, compressTyp := range allCompressTypes {
 		from = originFrom
 		writeSubsequentCompressionBlkNumber(c.Dbm, compressTyp, from)
 		for {
@@ -415,7 +470,7 @@ func TestCompressFinder(t *testing.T, setup func(t *testing.T) (*blockchain.Bloc
 		originBodyElapsed     time.Duration
 		originReceiptsElapsed time.Duration
 	)
-	for _, compressTyp := range []CompressionType{HeaderCompressType, BodyCompressType, ReceiptCompressType} {
+	for _, compressTyp := range allCompressTypes {
 		var (
 			r    = rand.Uint64() % (5000)
 			hash = mCompress.Dbm.ReadCanonicalHash(r)

kaiax/compress/impl/compress_test.go

@@ -20,12 +20,13 @@ import (
 	"math/big"
 	"testing"
 
+	"github.com/kaiachain/kaia/blockchain"
 	"github.com/kaiachain/kaia/blockchain/types"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestCompressStorage(t *testing.T) {
-	for _, compressTyp := range []CompressionType{HeaderCompressType, BodyCompressType, ReceiptCompressType} {
+	for _, compressTyp := range allCompressTypes {
 		switch compressTyp {
 		case HeaderCompressType:
 			testHeaderCompress(t)
@@ -38,7 +39,7 @@ func TestCompressStorage(t *testing.T) {
 }
 
 func TestDecompressStorage(t *testing.T) {
-	for _, compressTyp := range []CompressionType{HeaderCompressType, BodyCompressType, ReceiptCompressType} {
+	for _, compressTyp := range allCompressTypes {
 		switch compressTyp {
 		case HeaderCompressType:
 			testHeaderDecompress(t)
@@ -51,7 +52,7 @@ func TestDecompressStorage(t *testing.T) {
 }
 
 func TestDeleteStorage(t *testing.T) {
-	for _, compressTyp := range []CompressionType{HeaderCompressType, BodyCompressType, ReceiptCompressType} {
+	for _, compressTyp := range allCompressTypes {
 		switch compressTyp {
 		case HeaderCompressType:
 			testCompressedHeaderDelete(t)
@@ -88,14 +89,14 @@ func TestRewind(t *testing.T) {
 								next compression block number
 			Chunks = C1|C2|C3|C4|C5|C6|C7|C8|C9|C10
 
-		[Phase2: Rewind]
+		[Phase2: Rewind] (`Stop` called)
 			Once `setHead` is invoked,
 			0 ------------ 50 ---- 55
 							^
 				next compression block number
 			Chunks = C1|C2|C3|C4|C5
 
-		[Phase3: Compress again]
+		[Phase3: Compress again] (`Start` called)
 			compressed data range 50-59 is restored and Sync is started from 55. Finally,
 			0 ------------ 50 ------------ 99
 			Chunks = C1|C2|C3|C4|C5|C6|C7|C8|C9|C10
@@ -107,6 +108,7 @@ func TestRewind(t *testing.T) {
 		mCompress, dbm, headers, bodies, receipts = runCompress(t, nBlocks)
 	)
 
+	mCompress.Stop()
 	for i := nBlocks - 1; i >= setHeadTo; i-- {
 		num := uint64(i)
 		hash := dbm.ReadCanonicalHash(num)
@@ -147,14 +149,54 @@ func TestRewind(t *testing.T) {
 	// expected next compression block number should be equal or less than `setHeadTo`.
 	// Given the value of `setHeadTo` is 55 and chunk size is 10,
 	// The rewinded next compression block number should be 50.
-	nextCompressionNumber := readSubsequentCompressionBlkNumber(dbm, HeaderCompressType)
-	assert.Equal(t, int(nextCompressionNumber), setHeadTo-(setHeadTo%int(mCompress.getCompressChunk())))
+	for _, compressTyp := range allCompressTypes {
+		nextCompressionNumber := readSubsequentCompressionBlkNumber(dbm, compressTyp)
+		assert.Equal(t, int(nextCompressionNumber), setHeadTo-(setHeadTo%int(mCompress.getCompressChunk())))
+	}
 
-	go mCompress.Start() // fragment restore invoked before starting compression
+	mCompress.Start() // fragment restore invoked before starting compression
 	waitCompression(mCompress)
 	checkCompressedIntegrity(t, dbm, 0, nBlocks-1, canonicalHeaders, canonicalBodies, canonicalReceipts, false)
 
 	// Once completed the compression, next compression block number reaches to `nBlocks - 1`
-	nextCompressionNumber = readSubsequentCompressionBlkNumber(dbm, HeaderCompressType)
-	assert.Equal(t, int(nextCompressionNumber), nBlocks-1)
+	for _, compressTyp := range allCompressTypes {
+		nextCompressionNumber := readSubsequentCompressionBlkNumber(dbm, compressTyp)
+		assert.Equal(t, int(nextCompressionNumber), nBlocks-1)
+	}
+}
+
+func TestRetention(t *testing.T) {
+	var (
+		nBlocks    = 100
+		chain, dbm = initMock(t, nBlocks)
+		mCompress  = NewCompression()
+		err        = mCompress.Init(&InitOpts{
+			ChunkBlockSize: blockchain.DefaultChunkBlockSize,
+			ChunkCap:       blockchain.DefaultCompressChunkCap,
+			Chain:          chain,
+			Dbm:            dbm,
+		})
+	)
+	assert.Nil(t, err)
+	dbm.SetCompressModule(mCompress)
+	mCompress.setCompressChunk(10)
+	mCompress.setCompressRetention(uint64(nBlocks))
+	go mCompress.Start()
+	waitCompression(mCompress)
+
+	// compression is not performed by retention
+	for _, compressTyp := range allCompressTypes {
+		nextCompressionNumber := readSubsequentCompressionBlkNumber(dbm, compressTyp)
+		assert.Equal(t, nextCompressionNumber, uint64(0))
+	}
+	mCompress.Stop()
+
+	// compress work by reset retention
+	mCompress.setCompressRetention(0)
+	mCompress.Start()
+	waitCompression(mCompress)
+	for _, compressTyp := range allCompressTypes {
+		nextCompressionNumber := readSubsequentCompressionBlkNumber(dbm, compressTyp)
+		assert.Equal(t, int(nextCompressionNumber), nBlocks-1)
+	}
 }