aboutsummaryrefslogtreecommitdiff
path: root/core/rawdb/freezer.go
diff options
context:
space:
mode:
Diffstat (limited to 'core/rawdb/freezer.go')
-rw-r--r--core/rawdb/freezer.go393
1 files changed, 393 insertions, 0 deletions
diff --git a/core/rawdb/freezer.go b/core/rawdb/freezer.go
new file mode 100644
index 0000000..ce2e879
--- /dev/null
+++ b/core/rawdb/freezer.go
@@ -0,0 +1,393 @@
+// Copyright 2019 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package rawdb
+
+import (
+ "errors"
+ "fmt"
+ "math"
+ "os"
+ "path/filepath"
+ "sync/atomic"
+ "time"
+
+ "github.com/ava-labs/coreth/params"
+ "github.com/ava-labs/go-ethereum/common"
+ "github.com/ava-labs/go-ethereum/ethdb"
+ "github.com/ava-labs/go-ethereum/log"
+ "github.com/ava-labs/go-ethereum/metrics"
+ "github.com/prometheus/tsdb/fileutil"
+)
+
+var (
+ // errUnknownTable is returned if the user attempts to read from a table that is
+ // not tracked by the freezer.
+ errUnknownTable = errors.New("unknown table")
+
+ // errOutOrderInsertion is returned if the user attempts to inject out-of-order
+ // binary blobs into the freezer.
+ errOutOrderInsertion = errors.New("the append operation is out-order")
+
+ // errSymlinkDatadir is returned if the ancient directory specified by user
+ // is a symbolic link.
+ errSymlinkDatadir = errors.New("symbolic link datadir is not supported")
+)
+
+const (
+ // freezerRecheckInterval is the frequency to check the key-value database for
+ // chain progression that might permit new blocks to be frozen into immutable
+ // storage.
+ freezerRecheckInterval = time.Minute
+
+ // freezerBatchLimit is the maximum number of blocks to freeze in one batch
+ // before doing an fsync and deleting it from the key-value store.
+ freezerBatchLimit = 30000
+)
+
+// freezer is an memory mapped append-only database to store immutable chain data
+// into flat files:
+//
+// - The append only nature ensures that disk writes are minimized.
+// - The memory mapping ensures we can max out system memory for caching without
+// reserving it for go-ethereum. This would also reduce the memory requirements
+// of Geth, and thus also GC overhead.
+type freezer struct {
+ // WARNING: The `frozen` field is accessed atomically. On 32 bit platforms, only
+ // 64-bit aligned fields can be atomic. The struct is guaranteed to be so aligned,
+ // so take advantage of that (https://golang.org/pkg/sync/atomic/#pkg-note-BUG).
+ frozen uint64 // Number of blocks already frozen
+
+ tables map[string]*freezerTable // Data tables for storing everything
+ instanceLock fileutil.Releaser // File-system lock to prevent double opens
+}
+
+// newFreezer creates a chain freezer that moves ancient chain data into
+// append-only flat file containers.
+func newFreezer(datadir string, namespace string) (*freezer, error) {
+ // Create the initial freezer object
+ var (
+ readMeter = metrics.NewRegisteredMeter(namespace+"ancient/read", nil)
+ writeMeter = metrics.NewRegisteredMeter(namespace+"ancient/write", nil)
+ sizeCounter = metrics.NewRegisteredCounter(namespace+"ancient/size", nil)
+ )
+ // Ensure the datadir is not a symbolic link if it exists.
+ if info, err := os.Lstat(datadir); !os.IsNotExist(err) {
+ if info.Mode()&os.ModeSymlink != 0 {
+ log.Warn("Symbolic link ancient database is not supported", "path", datadir)
+ return nil, errSymlinkDatadir
+ }
+ }
+ // Leveldb uses LOCK as the filelock filename. To prevent the
+ // name collision, we use FLOCK as the lock name.
+ lock, _, err := fileutil.Flock(filepath.Join(datadir, "FLOCK"))
+ if err != nil {
+ return nil, err
+ }
+ // Open all the supported data tables
+ freezer := &freezer{
+ tables: make(map[string]*freezerTable),
+ instanceLock: lock,
+ }
+ for name, disableSnappy := range freezerNoSnappy {
+ table, err := newTable(datadir, name, readMeter, writeMeter, sizeCounter, disableSnappy)
+ if err != nil {
+ for _, table := range freezer.tables {
+ table.Close()
+ }
+ lock.Release()
+ return nil, err
+ }
+ freezer.tables[name] = table
+ }
+ if err := freezer.repair(); err != nil {
+ for _, table := range freezer.tables {
+ table.Close()
+ }
+ lock.Release()
+ return nil, err
+ }
+ log.Info("Opened ancient database", "database", datadir)
+ return freezer, nil
+}
+
+// Close terminates the chain freezer, unmapping all the data files.
+func (f *freezer) Close() error {
+ var errs []error
+ for _, table := range f.tables {
+ if err := table.Close(); err != nil {
+ errs = append(errs, err)
+ }
+ }
+ if err := f.instanceLock.Release(); err != nil {
+ errs = append(errs, err)
+ }
+ if errs != nil {
+ return fmt.Errorf("%v", errs)
+ }
+ return nil
+}
+
+// HasAncient returns an indicator whether the specified ancient data exists
+// in the freezer.
+func (f *freezer) HasAncient(kind string, number uint64) (bool, error) {
+ if table := f.tables[kind]; table != nil {
+ return table.has(number), nil
+ }
+ return false, nil
+}
+
+// Ancient retrieves an ancient binary blob from the append-only immutable files.
+func (f *freezer) Ancient(kind string, number uint64) ([]byte, error) {
+ if table := f.tables[kind]; table != nil {
+ return table.Retrieve(number)
+ }
+ return nil, errUnknownTable
+}
+
+// Ancients returns the length of the frozen items.
+func (f *freezer) Ancients() (uint64, error) {
+ return atomic.LoadUint64(&f.frozen), nil
+}
+
+// AncientSize returns the ancient size of the specified category.
+func (f *freezer) AncientSize(kind string) (uint64, error) {
+ if table := f.tables[kind]; table != nil {
+ return table.size()
+ }
+ return 0, errUnknownTable
+}
+
+// AppendAncient injects all binary blobs belong to block at the end of the
+// append-only immutable table files.
+//
+// Notably, this function is lock free but kind of thread-safe. All out-of-order
+// injection will be rejected. But if two injections with same number happen at
+// the same time, we can get into the trouble.
+func (f *freezer) AppendAncient(number uint64, hash, header, body, receipts, td []byte) (err error) {
+ // Ensure the binary blobs we are appending is continuous with freezer.
+ if atomic.LoadUint64(&f.frozen) != number {
+ return errOutOrderInsertion
+ }
+ // Rollback all inserted data if any insertion below failed to ensure
+ // the tables won't out of sync.
+ defer func() {
+ if err != nil {
+ rerr := f.repair()
+ if rerr != nil {
+ log.Crit("Failed to repair freezer", "err", rerr)
+ }
+ log.Info("Append ancient failed", "number", number, "err", err)
+ }
+ }()
+ // Inject all the components into the relevant data tables
+ if err := f.tables[freezerHashTable].Append(f.frozen, hash[:]); err != nil {
+ log.Error("Failed to append ancient hash", "number", f.frozen, "hash", hash, "err", err)
+ return err
+ }
+ if err := f.tables[freezerHeaderTable].Append(f.frozen, header); err != nil {
+ log.Error("Failed to append ancient header", "number", f.frozen, "hash", hash, "err", err)
+ return err
+ }
+ if err := f.tables[freezerBodiesTable].Append(f.frozen, body); err != nil {
+ log.Error("Failed to append ancient body", "number", f.frozen, "hash", hash, "err", err)
+ return err
+ }
+ if err := f.tables[freezerReceiptTable].Append(f.frozen, receipts); err != nil {
+ log.Error("Failed to append ancient receipts", "number", f.frozen, "hash", hash, "err", err)
+ return err
+ }
+ if err := f.tables[freezerDifficultyTable].Append(f.frozen, td); err != nil {
+ log.Error("Failed to append ancient difficulty", "number", f.frozen, "hash", hash, "err", err)
+ return err
+ }
+ atomic.AddUint64(&f.frozen, 1) // Only modify atomically
+ return nil
+}
+
+// Truncate discards any recent data above the provided threshold number.
+func (f *freezer) TruncateAncients(items uint64) error {
+ if atomic.LoadUint64(&f.frozen) <= items {
+ return nil
+ }
+ for _, table := range f.tables {
+ if err := table.truncate(items); err != nil {
+ return err
+ }
+ }
+ atomic.StoreUint64(&f.frozen, items)
+ return nil
+}
+
+// sync flushes all data tables to disk.
+func (f *freezer) Sync() error {
+ var errs []error
+ for _, table := range f.tables {
+ if err := table.Sync(); err != nil {
+ errs = append(errs, err)
+ }
+ }
+ if errs != nil {
+ return fmt.Errorf("%v", errs)
+ }
+ return nil
+}
+
+// freeze is a background thread that periodically checks the blockchain for any
+// import progress and moves ancient data from the fast database into the freezer.
+//
+// This functionality is deliberately broken off from block importing to avoid
+// incurring additional data shuffling delays on block propagation.
+func (f *freezer) freeze(db ethdb.KeyValueStore) {
+ nfdb := &nofreezedb{KeyValueStore: db}
+
+ for {
+ // Retrieve the freezing threshold.
+ hash := ReadHeadBlockHash(nfdb)
+ if hash == (common.Hash{}) {
+ log.Debug("Current full block hash unavailable") // new chain, empty database
+ time.Sleep(freezerRecheckInterval)
+ continue
+ }
+ number := ReadHeaderNumber(nfdb, hash)
+ switch {
+ case number == nil:
+ log.Error("Current full block number unavailable", "hash", hash)
+ time.Sleep(freezerRecheckInterval)
+ continue
+
+ case *number < params.ImmutabilityThreshold:
+ log.Debug("Current full block not old enough", "number", *number, "hash", hash, "delay", params.ImmutabilityThreshold)
+ time.Sleep(freezerRecheckInterval)
+ continue
+
+ case *number-params.ImmutabilityThreshold <= f.frozen:
+ log.Debug("Ancient blocks frozen already", "number", *number, "hash", hash, "frozen", f.frozen)
+ time.Sleep(freezerRecheckInterval)
+ continue
+ }
+ head := ReadHeader(nfdb, hash, *number)
+ if head == nil {
+ log.Error("Current full block unavailable", "number", *number, "hash", hash)
+ time.Sleep(freezerRecheckInterval)
+ continue
+ }
+ // Seems we have data ready to be frozen, process in usable batches
+ limit := *number - params.ImmutabilityThreshold
+ if limit-f.frozen > freezerBatchLimit {
+ limit = f.frozen + freezerBatchLimit
+ }
+ var (
+ start = time.Now()
+ first = f.frozen
+ ancients = make([]common.Hash, 0, limit)
+ )
+ for f.frozen < limit {
+ // Retrieves all the components of the canonical block
+ hash := ReadCanonicalHash(nfdb, f.frozen)
+ if hash == (common.Hash{}) {
+ log.Error("Canonical hash missing, can't freeze", "number", f.frozen)
+ break
+ }
+ header := ReadHeaderRLP(nfdb, hash, f.frozen)
+ if len(header) == 0 {
+ log.Error("Block header missing, can't freeze", "number", f.frozen, "hash", hash)
+ break
+ }
+ body := ReadBodyRLP(nfdb, hash, f.frozen)
+ if len(body) == 0 {
+ log.Error("Block body missing, can't freeze", "number", f.frozen, "hash", hash)
+ break
+ }
+ receipts := ReadReceiptsRLP(nfdb, hash, f.frozen)
+ if len(receipts) == 0 {
+ log.Error("Block receipts missing, can't freeze", "number", f.frozen, "hash", hash)
+ break
+ }
+ td := ReadTdRLP(nfdb, hash, f.frozen)
+ if len(td) == 0 {
+ log.Error("Total difficulty missing, can't freeze", "number", f.frozen, "hash", hash)
+ break
+ }
+ log.Trace("Deep froze ancient block", "number", f.frozen, "hash", hash)
+ // Inject all the components into the relevant data tables
+ if err := f.AppendAncient(f.frozen, hash[:], header, body, receipts, td); err != nil {
+ break
+ }
+ ancients = append(ancients, hash)
+ }
+ // Batch of blocks have been frozen, flush them before wiping from leveldb
+ if err := f.Sync(); err != nil {
+ log.Crit("Failed to flush frozen tables", "err", err)
+ }
+ // Wipe out all data from the active database
+ batch := db.NewBatch()
+ for i := 0; i < len(ancients); i++ {
+ // Always keep the genesis block in active database
+ if first+uint64(i) != 0 {
+ DeleteBlockWithoutNumber(batch, ancients[i], first+uint64(i))
+ DeleteCanonicalHash(batch, first+uint64(i))
+ }
+ }
+ if err := batch.Write(); err != nil {
+ log.Crit("Failed to delete frozen canonical blocks", "err", err)
+ }
+ batch.Reset()
+ // Wipe out side chain also.
+ for number := first; number < f.frozen; number++ {
+ // Always keep the genesis block in active database
+ if number != 0 {
+ for _, hash := range ReadAllHashes(db, number) {
+ DeleteBlock(batch, hash, number)
+ }
+ }
+ }
+ if err := batch.Write(); err != nil {
+ log.Crit("Failed to delete frozen side blocks", "err", err)
+ }
+ // Log something friendly for the user
+ context := []interface{}{
+ "blocks", f.frozen - first, "elapsed", common.PrettyDuration(time.Since(start)), "number", f.frozen - 1,
+ }
+ if n := len(ancients); n > 0 {
+ context = append(context, []interface{}{"hash", ancients[n-1]}...)
+ }
+ log.Info("Deep froze chain segment", context...)
+
+ // Avoid database thrashing with tiny writes
+ if f.frozen-first < freezerBatchLimit {
+ time.Sleep(freezerRecheckInterval)
+ }
+ }
+}
+
+// repair truncates all data tables to the same length.
+func (f *freezer) repair() error {
+ min := uint64(math.MaxUint64)
+ for _, table := range f.tables {
+ items := atomic.LoadUint64(&table.items)
+ if min > items {
+ min = items
+ }
+ }
+ for _, table := range f.tables {
+ if err := table.truncate(min); err != nil {
+ return err
+ }
+ }
+ atomic.StoreUint64(&f.frozen, min)
+ return nil
+}
Powered by cgit, Themed by Ted Yin.