core/rawdb: introduce flush offset in freezer

Author: rjl493456442

core/blockchain.go

@@ -112,21 +112,28 @@ const (
 	//   * the `BlockNumber`, `TxHash`, `TxIndex`, `BlockHash` and `Index` fields of log are deleted
 	//   * the `Bloom` field of receipt is deleted
 	//   * the `BlockIndex` and `TxIndex` fields of txlookup are deleted
+	//
 	// - Version 5
 	//  The following incompatible database changes were added:
 	//    * the `TxHash`, `GasCost`, and `ContractAddress` fields are no longer stored for a receipt
 	//    * the `TxHash`, `GasCost`, and `ContractAddress` fields are computed by looking up the
 	//      receipts' corresponding block
+	//
 	// - Version 6
 	//  The following incompatible database changes were added:
 	//    * Transaction lookup information stores the corresponding block number instead of block hash
+	//
 	// - Version 7
 	//  The following incompatible database changes were added:
 	//    * Use freezer as the ancient database to maintain all ancient data
+	//
 	// - Version 8
 	//  The following incompatible database changes were added:
 	//    * New scheme for contract code in order to separate the codes and trie nodes
-	BlockChainVersion uint64 = 8
+	//
+	// - Version 9
+	//    * The metadata structure of freezer is changed by adding 'flushOffset'
+	BlockChainVersion uint64 = 9
 )
 
 // CacheConfig contains the configuration values for the trie database

core/rawdb/accessors_chain_test.go

@@ -879,6 +879,7 @@ func TestHeadersRLPStorage(t *testing.T) {
 		t.Fatalf("failed to create database with ancient backend")
 	}
 	defer db.Close()
+
 	// Create blocks
 	var chain []*types.Block
 	var pHash common.Hash
@@ -894,7 +895,7 @@ func TestHeadersRLPStorage(t *testing.T) {
 		chain = append(chain, block)
 		pHash = block.Hash()
 	}
-	var receipts []types.Receipts = make([]types.Receipts, 100)
+	receipts := make([]types.Receipts, 100)
 	// Write first half to ancients
 	WriteAncientBlocks(db, chain[:50], receipts[:50], big.NewInt(100))
 	// Write second half to db

core/rawdb/ancient_scheme.go

@@ -50,10 +50,25 @@ var chainFreezerNoSnappy = map[string]bool{
 	ChainFreezerDifficultyTable: true,
 }
 
-const (
-	// stateHistoryTableSize defines the maximum size of freezer data files.
-	stateHistoryTableSize = 2 * 1000 * 1000 * 1000
+// chainFreezerSize configures the maximum size for each freezer table data files.
+var chainFreezerSize = map[string]uint32{
+	// The size of each item's value is roughly 650 bytes, about 2 millions
+	// items per data file.
+	ChainFreezerHeaderTable: 2 * 1000 * 1000 * 1000,
+
+	// The size of each item’s value is fixed at 32 bytes, 2 millions items
+	// per data file.
+	ChainFreezerHashTable: 64 * 1000 * 1000,
+
+	// The size of each item’s value is less than 10 bytes, 2 millions items
+	// per data file.
+	ChainFreezerDifficultyTable: 20 * 1000 * 1000,
+
+	ChainFreezerBodiesTable:  2 * 1000 * 1000 * 1000,
+	ChainFreezerReceiptTable: 2 * 1000 * 1000 * 1000,
+}
 
+const (
 	// stateHistoryAccountIndex indicates the name of the freezer state history table.
 	stateHistoryMeta         = "history.meta"
 	stateHistoryAccountIndex = "account.index"
@@ -62,6 +77,7 @@ const (
 	stateHistoryStorageData  = "storage.data"
 )
 
+// stateFreezerNoSnappy configures whether compression is disabled for the state freezer.
 var stateFreezerNoSnappy = map[string]bool{
 	stateHistoryMeta:         true,
 	stateHistoryAccountIndex: false,
@@ -70,6 +86,17 @@ var stateFreezerNoSnappy = map[string]bool{
 	stateHistoryStorageData:  false,
 }
 
+// stateFreezerSize configures the maximum size for each freezer table data files.
+var stateFreezerSize = map[string]uint32{
+	// The size of each item's value is fixed at 73 bytes, about 2 millions
+	// items per data file.
+	stateHistoryMeta:         128 * 1000 * 1000,
+	stateHistoryAccountIndex: 2 * 1000 * 1000 * 1000,
+	stateHistoryStorageIndex: 2 * 1000 * 1000 * 1000,
+	stateHistoryAccountData:  2 * 1000 * 1000 * 1000,
+	stateHistoryStorageData:  2 * 1000 * 1000 * 1000,
+}
+
 // The list of identifiers of ancient stores.
 var (
 	ChainFreezerName       = "chain"        // the folder name of chain segment ancient store.
@@ -96,5 +123,5 @@ func NewStateFreezer(ancientDir string, verkle bool, readOnly bool) (ethdb.Reset
 	} else {
 		name = filepath.Join(ancientDir, MerkleStateFreezerName)
 	}
-	return newResettableFreezer(name, "eth/db/state", readOnly, stateHistoryTableSize, stateFreezerNoSnappy)
+	return newResettableFreezer(name, "eth/db/state", readOnly, stateFreezerSize, stateFreezerNoSnappy)
 }

core/rawdb/ancient_utils.go

@@ -120,12 +120,13 @@ func InspectFreezerTable(ancient string, freezerName string, tableName string, s
 	var (
 		path   string
 		tables map[string]bool
+		sizes  map[string]uint32
 	)
 	switch freezerName {
 	case ChainFreezerName:
-		path, tables = resolveChainFreezerDir(ancient), chainFreezerNoSnappy
+		path, tables, sizes = resolveChainFreezerDir(ancient), chainFreezerNoSnappy, chainFreezerSize
 	case MerkleStateFreezerName, VerkleStateFreezerName:
-		path, tables = filepath.Join(ancient, freezerName), stateFreezerNoSnappy
+		path, tables, sizes = filepath.Join(ancient, freezerName), stateFreezerNoSnappy, stateFreezerSize
 	default:
 		return fmt.Errorf("unknown freezer, supported ones: %v", freezers)
 	}
@@ -137,7 +138,7 @@ func InspectFreezerTable(ancient string, freezerName string, tableName string, s
 		}
 		return fmt.Errorf("unknown table, supported ones: %v", names)
 	}
-	table, err := newFreezerTable(path, tableName, noSnappy, true)
+	table, err := newFreezerTable(path, tableName, noSnappy, true, sizes[tableName])
 	if err != nil {
 		return err
 	}

core/rawdb/chain_freezer.go

@@ -64,7 +64,7 @@ func newChainFreezer(datadir string, namespace string, readonly bool) (*chainFre
 	if datadir == "" {
 		freezer = NewMemoryFreezer(readonly, chainFreezerNoSnappy)
 	} else {
-		freezer, err = NewFreezer(datadir, namespace, readonly, freezerTableSize, chainFreezerNoSnappy)
+		freezer, err = NewFreezer(datadir, namespace, readonly, chainFreezerSize, chainFreezerNoSnappy)
 	}
 	if err != nil {
 		return nil, err

core/rawdb/freezer.go

@@ -49,9 +49,6 @@ var (
 	errSymlinkDatadir = errors.New("symbolic link datadir is not supported")
 )
 
-// freezerTableSize defines the maximum size of freezer data files.
-const freezerTableSize = 2 * 1000 * 1000 * 1000
-
 // Freezer is an append-only database to store immutable ordered data into
 // flat files:
 //
@@ -77,7 +74,7 @@ type Freezer struct {
 //
 // The 'tables' argument defines the data tables. If the value of a map
 // entry is true, snappy compression is disabled for the table.
-func NewFreezer(datadir string, namespace string, readonly bool, maxTableSize uint32, tables map[string]bool) (*Freezer, error) {
+func NewFreezer(datadir string, namespace string, readonly bool, maxTableSize map[string]uint32, tables map[string]bool) (*Freezer, error) {
 	// Create the initial freezer object
 	var (
 		readMeter  = metrics.NewRegisteredMeter(namespace+"ancient/read", nil)
@@ -116,7 +113,11 @@ func NewFreezer(datadir string, namespace string, readonly bool, maxTableSize ui
 
 	// Create the tables.
 	for name, disableSnappy := range tables {
-		table, err := newTable(datadir, name, readMeter, writeMeter, sizeGauge, maxTableSize, disableSnappy, readonly)
+		size, exist := maxTableSize[name]
+		if !exist {
+			return nil, fmt.Errorf("table size for %q is not defined", name)
+		}
+		table, err := newTable(datadir, name, readMeter, writeMeter, sizeGauge, size, disableSnappy, readonly)
 		if err != nil {
 			for _, table := range freezer.tables {
 				table.Close()

core/rawdb/freezer_batch.go

@@ -18,6 +18,7 @@ package rawdb
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common/math"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -188,9 +189,6 @@ func (batch *freezerTableBatch) commit() error {
 	if err != nil {
 		return err
 	}
-	if err := batch.t.head.Sync(); err != nil {
-		return err
-	}
 	dataSize := int64(len(batch.dataBuffer))
 	batch.dataBuffer = batch.dataBuffer[:0]
 
@@ -208,6 +206,12 @@ func (batch *freezerTableBatch) commit() error {
 	// Update metrics.
 	batch.t.sizeGauge.Inc(dataSize + indexSize)
 	batch.t.writeMeter.Mark(dataSize + indexSize)
+
+	// Periodically sync the table
+	if time.Since(batch.t.lastSync) > 10*time.Second {
+		batch.t.lastSync = time.Now()
+		return batch.t.syncWithNoLock()
+	}
 	return nil
 }