mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-27 23:26:44 +00:00
core, eth: drop old sequential key db upgrader
This commit is contained in:
Péter Szilágyi
parent
328bf78a3a
commit
655a1802fa
3 changed files with 9 additions and 268 deletions
|
|
@ -53,15 +53,9 @@ var (
|
|||
|
||||
configPrefix = []byte("ethereum-config-") // config prefix for the db
|
||||
|
||||
// used by old (non-sequential keys) db, now only used for conversion
|
||||
oldBlockPrefix = []byte("block-")
|
||||
oldHeaderSuffix = []byte("-header")
|
||||
oldTdSuffix = []byte("-td") // headerPrefix + num (uint64 big endian) + hash + tdSuffix -> td
|
||||
oldBodySuffix = []byte("-body")
|
||||
oldBlockNumPrefix = []byte("block-num-")
|
||||
oldBlockReceiptsPrefix = []byte("receipts-block-")
|
||||
oldReceiptsPrefix = []byte("receipts-")
|
||||
oldTxMetaSuffix = []byte{0x01}
|
||||
// used by old db, now only used for conversion
|
||||
oldReceiptsPrefix = []byte("receipts-")
|
||||
oldTxMetaSuffix = []byte{0x01}
|
||||
|
||||
ErrChainConfigNotFound = errors.New("ChainConfig not found") // general config not found error
|
||||
|
||||
|
|
@ -90,10 +84,7 @@ func encodeBlockNumber(number uint64) []byte {
|
|||
func GetCanonicalHash(db ethdb.Database, number uint64) common.Hash {
|
||||
data, _ := db.Get(append(append(headerPrefix, encodeBlockNumber(number)...), numSuffix...))
|
||||
if len(data) == 0 {
|
||||
data, _ = db.Get(append(oldBlockNumPrefix, big.NewInt(int64(number)).Bytes()...))
|
||||
if len(data) == 0 {
|
||||
return common.Hash{}
|
||||
}
|
||||
return common.Hash{}
|
||||
}
|
||||
return common.BytesToHash(data)
|
||||
}
|
||||
|
|
@ -107,15 +98,7 @@ const missingNumber = uint64(0xffffffffffffffff)
|
|||
func GetBlockNumber(db ethdb.Database, hash common.Hash) uint64 {
|
||||
data, _ := db.Get(append(blockHashPrefix, hash.Bytes()...))
|
||||
if len(data) != 8 {
|
||||
data, _ := db.Get(append(append(oldBlockPrefix, hash.Bytes()...), oldHeaderSuffix...))
|
||||
if len(data) == 0 {
|
||||
return missingNumber
|
||||
}
|
||||
header := new(types.Header)
|
||||
if err := rlp.Decode(bytes.NewReader(data), header); err != nil {
|
||||
log.Crit("Failed to decode block header", "err", err)
|
||||
}
|
||||
return header.Number.Uint64()
|
||||
return missingNumber
|
||||
}
|
||||
return binary.BigEndian.Uint64(data)
|
||||
}
|
||||
|
|
@ -158,9 +141,6 @@ func GetHeadFastBlockHash(db ethdb.Database) common.Hash {
|
|||
// if the header's not found.
|
||||
func GetHeaderRLP(db ethdb.Database, hash common.Hash, number uint64) rlp.RawValue {
|
||||
data, _ := db.Get(append(append(headerPrefix, encodeBlockNumber(number)...), hash.Bytes()...))
|
||||
if len(data) == 0 {
|
||||
data, _ = db.Get(append(append(oldBlockPrefix, hash.Bytes()...), oldHeaderSuffix...))
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
|
|
@ -182,9 +162,6 @@ func GetHeader(db ethdb.Database, hash common.Hash, number uint64) *types.Header
|
|||
// GetBodyRLP retrieves the block body (transactions and uncles) in RLP encoding.
|
||||
func GetBodyRLP(db ethdb.Database, hash common.Hash, number uint64) rlp.RawValue {
|
||||
data, _ := db.Get(append(append(bodyPrefix, encodeBlockNumber(number)...), hash.Bytes()...))
|
||||
if len(data) == 0 {
|
||||
data, _ = db.Get(append(append(oldBlockPrefix, hash.Bytes()...), oldBodySuffix...))
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
|
|
@ -208,10 +185,7 @@ func GetBody(db ethdb.Database, hash common.Hash, number uint64) *types.Body {
|
|||
func GetTd(db ethdb.Database, hash common.Hash, number uint64) *big.Int {
|
||||
data, _ := db.Get(append(append(append(headerPrefix, encodeBlockNumber(number)...), hash[:]...), tdSuffix...))
|
||||
if len(data) == 0 {
|
||||
data, _ = db.Get(append(append(oldBlockPrefix, hash.Bytes()...), oldTdSuffix...))
|
||||
if len(data) == 0 {
|
||||
return nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
td := new(big.Int)
|
||||
if err := rlp.Decode(bytes.NewReader(data), td); err != nil {
|
||||
|
|
@ -246,10 +220,7 @@ func GetBlock(db ethdb.Database, hash common.Hash, number uint64) *types.Block {
|
|||
func GetBlockReceipts(db ethdb.Database, hash common.Hash, number uint64) types.Receipts {
|
||||
data, _ := db.Get(append(append(blockReceiptsPrefix, encodeBlockNumber(number)...), hash[:]...))
|
||||
if len(data) == 0 {
|
||||
data, _ = db.Get(append(oldBlockReceiptsPrefix, hash.Bytes()...))
|
||||
if len(data) == 0 {
|
||||
return nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
storageReceipts := []*types.ReceiptForStorage{}
|
||||
if err := rlp.DecodeBytes(data, &storageReceipts); err != nil {
|
||||
|
|
|
|||
|
|
@ -19,240 +19,18 @@ package eth
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"math/big"
|
||||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/core"
|
||||
"github.com/ethereum/go-ethereum/core/types"
|
||||
"github.com/ethereum/go-ethereum/ethdb"
|
||||
"github.com/ethereum/go-ethereum/log"
|
||||
"github.com/ethereum/go-ethereum/rlp"
|
||||
)
|
||||
|
||||
var useSequentialKeys = []byte("dbUpgrade_20160530sequentialKeys")
|
||||
var deduplicateData = []byte("dbUpgrade_20170714deduplicateData")
|
||||
|
||||
// upgradeSequentialKeys checks the chain database version and
|
||||
// starts a background process to make upgrades if necessary.
|
||||
// Returns a stop function that blocks until the process has
|
||||
// been safely stopped.
|
||||
func upgradeSequentialKeys(db ethdb.Database) (stopFn func()) {
|
||||
data, _ := db.Get(useSequentialKeys)
|
||||
if len(data) > 0 && data[0] == 42 {
|
||||
return nil // already converted
|
||||
}
|
||||
|
||||
if data, _ := db.Get([]byte("LastHeader")); len(data) == 0 {
|
||||
db.Put(useSequentialKeys, []byte{42})
|
||||
return nil // empty database, nothing to do
|
||||
}
|
||||
|
||||
log.Warn("Upgrading chain database to use sequential keys")
|
||||
|
||||
stopChn := make(chan struct{})
|
||||
stoppedChn := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
stopFn := func() bool {
|
||||
select {
|
||||
case <-time.After(time.Microsecond * 100): // make sure other processes don't get starved
|
||||
case <-stopChn:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
err, stopped := upgradeSequentialCanonicalNumbers(db, stopFn)
|
||||
if err == nil && !stopped {
|
||||
err, stopped = upgradeSequentialBlocks(db, stopFn)
|
||||
}
|
||||
if err == nil && !stopped {
|
||||
err, stopped = upgradeSequentialOrphanedReceipts(db, stopFn)
|
||||
}
|
||||
if err == nil && !stopped {
|
||||
log.Info("Database conversion successful")
|
||||
db.Put(useSequentialKeys, []byte{42})
|
||||
}
|
||||
if err != nil {
|
||||
log.Error("Database conversion failed", "err", err)
|
||||
}
|
||||
close(stoppedChn)
|
||||
}()
|
||||
|
||||
return func() {
|
||||
close(stopChn)
|
||||
<-stoppedChn
|
||||
}
|
||||
}
|
||||
|
||||
// upgradeSequentialCanonicalNumbers reads all old format canonical numbers from
|
||||
// the database, writes them in new format and deletes the old ones if successful.
|
||||
func upgradeSequentialCanonicalNumbers(db ethdb.Database, stopFn func() bool) (error, bool) {
|
||||
prefix := []byte("block-num-")
|
||||
it := db.(*ethdb.LDBDatabase).NewIterator()
|
||||
defer func() {
|
||||
it.Release()
|
||||
}()
|
||||
it.Seek(prefix)
|
||||
cnt := 0
|
||||
for bytes.HasPrefix(it.Key(), prefix) {
|
||||
keyPtr := it.Key()
|
||||
if len(keyPtr) < 20 {
|
||||
cnt++
|
||||
if cnt%100000 == 0 {
|
||||
it.Release()
|
||||
it = db.(*ethdb.LDBDatabase).NewIterator()
|
||||
it.Seek(keyPtr)
|
||||
log.Info("Converting canonical numbers", "count", cnt)
|
||||
}
|
||||
number := big.NewInt(0).SetBytes(keyPtr[10:]).Uint64()
|
||||
newKey := []byte("h12345678n")
|
||||
binary.BigEndian.PutUint64(newKey[1:9], number)
|
||||
if err := db.Put(newKey, it.Value()); err != nil {
|
||||
return err, false
|
||||
}
|
||||
if err := db.Delete(keyPtr); err != nil {
|
||||
return err, false
|
||||
}
|
||||
}
|
||||
|
||||
if stopFn() {
|
||||
return nil, true
|
||||
}
|
||||
it.Next()
|
||||
}
|
||||
if cnt > 0 {
|
||||
log.Info("converted canonical numbers", "count", cnt)
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// upgradeSequentialBlocks reads all old format block headers, bodies, TDs and block
|
||||
// receipts from the database, writes them in new format and deletes the old ones
|
||||
// if successful.
|
||||
func upgradeSequentialBlocks(db ethdb.Database, stopFn func() bool) (error, bool) {
|
||||
prefix := []byte("block-")
|
||||
it := db.(*ethdb.LDBDatabase).NewIterator()
|
||||
defer func() {
|
||||
it.Release()
|
||||
}()
|
||||
it.Seek(prefix)
|
||||
cnt := 0
|
||||
for bytes.HasPrefix(it.Key(), prefix) {
|
||||
keyPtr := it.Key()
|
||||
if len(keyPtr) >= 38 {
|
||||
cnt++
|
||||
if cnt%10000 == 0 {
|
||||
it.Release()
|
||||
it = db.(*ethdb.LDBDatabase).NewIterator()
|
||||
it.Seek(keyPtr)
|
||||
log.Info("Converting blocks", "count", cnt)
|
||||
}
|
||||
// convert header, body, td and block receipts
|
||||
var keyPrefix [38]byte
|
||||
copy(keyPrefix[:], keyPtr[0:38])
|
||||
hash := keyPrefix[6:38]
|
||||
if err := upgradeSequentialBlockData(db, hash); err != nil {
|
||||
return err, false
|
||||
}
|
||||
// delete old db entries belonging to this hash
|
||||
for bytes.HasPrefix(it.Key(), keyPrefix[:]) {
|
||||
if err := db.Delete(it.Key()); err != nil {
|
||||
return err, false
|
||||
}
|
||||
it.Next()
|
||||
}
|
||||
if err := db.Delete(append([]byte("receipts-block-"), hash...)); err != nil {
|
||||
return err, false
|
||||
}
|
||||
} else {
|
||||
it.Next()
|
||||
}
|
||||
|
||||
if stopFn() {
|
||||
return nil, true
|
||||
}
|
||||
}
|
||||
if cnt > 0 {
|
||||
log.Info("Converted blocks", "count", cnt)
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// upgradeSequentialOrphanedReceipts removes any old format block receipts from the
|
||||
// database that did not have a corresponding block
|
||||
func upgradeSequentialOrphanedReceipts(db ethdb.Database, stopFn func() bool) (error, bool) {
|
||||
prefix := []byte("receipts-block-")
|
||||
it := db.(*ethdb.LDBDatabase).NewIterator()
|
||||
defer it.Release()
|
||||
it.Seek(prefix)
|
||||
cnt := 0
|
||||
for bytes.HasPrefix(it.Key(), prefix) {
|
||||
// phase 2 already converted receipts belonging to existing
|
||||
// blocks, just remove if there's anything left
|
||||
cnt++
|
||||
if err := db.Delete(it.Key()); err != nil {
|
||||
return err, false
|
||||
}
|
||||
|
||||
if stopFn() {
|
||||
return nil, true
|
||||
}
|
||||
it.Next()
|
||||
}
|
||||
if cnt > 0 {
|
||||
log.Info("Removed orphaned block receipts", "count", cnt)
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// upgradeSequentialBlockData upgrades the header, body, td and block receipts
|
||||
// database entries belonging to a single hash (doesn't delete old data).
|
||||
func upgradeSequentialBlockData(db ethdb.Database, hash []byte) error {
|
||||
// get old chain data and block number
|
||||
headerRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-header")...))
|
||||
if len(headerRLP) == 0 {
|
||||
return nil
|
||||
}
|
||||
header := new(types.Header)
|
||||
if err := rlp.Decode(bytes.NewReader(headerRLP), header); err != nil {
|
||||
return err
|
||||
}
|
||||
number := header.Number.Uint64()
|
||||
bodyRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-body")...))
|
||||
tdRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-td")...))
|
||||
receiptsRLP, _ := db.Get(append([]byte("receipts-block-"), hash...))
|
||||
// store new hash -> number association
|
||||
encNum := make([]byte, 8)
|
||||
binary.BigEndian.PutUint64(encNum, number)
|
||||
if err := db.Put(append([]byte("H"), hash...), encNum); err != nil {
|
||||
return err
|
||||
}
|
||||
// store new chain data
|
||||
if err := db.Put(append(append([]byte("h"), encNum...), hash...), headerRLP); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(tdRLP) != 0 {
|
||||
if err := db.Put(append(append(append([]byte("h"), encNum...), hash...), []byte("t")...), tdRLP); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if len(bodyRLP) != 0 {
|
||||
if err := db.Put(append(append([]byte("b"), encNum...), hash...), bodyRLP); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if len(receiptsRLP) != 0 {
|
||||
if err := db.Put(append(append([]byte("r"), encNum...), hash...), receiptsRLP); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// upgradeDeduplicateData checks the chain database version and
|
||||
// starts a background process to make upgrades if necessary.
|
||||
// Returns a stop function that blocks until the process has
|
||||
|
|
|
|||
|
|
@ -177,16 +177,8 @@ func (pool *TxPool) checkMinedTxs(ctx context.Context, hash common.Hash, number
|
|||
}
|
||||
// If some transactions have been mined, write the needed data to disk and update
|
||||
if list != nil {
|
||||
// Retrieve all the receipts belonging to this block
|
||||
receipts, err := GetBlockReceipts(ctx, pool.odr, hash, number)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Write the transactions, receipts and lookup table to disk (all needed for lookups)
|
||||
if err := core.WriteBlock(pool.chainDb, block); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := core.WriteBlockReceipts(pool.chainDb, hash, number, receipts); err != nil {
|
||||
// Retrieve all the receipts belonging to this block and write the loopup table
|
||||
if _, err := GetBlockReceipts(ctx, pool.odr, hash, number); err != nil { // ODR caches, ignore results
|
||||
return err
|
||||
}
|
||||
if err := core.WriteLookupEntries(pool.chainDb, block); err != nil {
|
||||
|
|
|
|||
Loading…
Reference in a new issue