From d235e2c6a5788ec4a6cff15a16f56b38a3876a0d Mon Sep 17 00:00:00 2001 From: Determinant Date: Sun, 28 Jun 2020 14:47:41 -0400 Subject: ... --- core/rawdb/freezer.go | 393 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 393 insertions(+) create mode 100644 core/rawdb/freezer.go (limited to 'core/rawdb/freezer.go') 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 . + +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 +} -- cgit v1.2.3