From 39ac0e38e8ab3df91b1eeda79d348519ef1360eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 24 Sep 2018 15:57:49 +0300 Subject: [PATCH] eth, freezer: add immutable chain freezer --- eth/backend.go | 9 ++ freezer/freezer.go | 213 +++++++++++++++++++++++++++++++++ freezer/mmap.go | 86 ++++++++++++++ freezer/table.go | 290 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 598 insertions(+) create mode 100644 freezer/freezer.go create mode 100644 freezer/mmap.go create mode 100644 freezer/table.go diff --git a/eth/backend.go b/eth/backend.go index 6a136182ab..deed195563 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -24,6 +24,7 @@ import ( "runtime" "sync" "sync/atomic" + "time" "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/common" @@ -41,6 +42,7 @@ import ( "github.com/ethereum/go-ethereum/eth/gasprice" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/freezer" "github.com/ethereum/go-ethereum/internal/ethapi" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/miner" @@ -69,6 +71,7 @@ type Ethereum struct { // Handlers txPool *core.TxPool blockchain *core.BlockChain + freezer *freezer.Freezer protocolManager *ProtocolManager lesServer LesServer @@ -168,6 +171,12 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { if err != nil { return nil, err } + eth.freezer, err = freezer.New(ctx.ResolvePath("freezer")) + if err != nil { + return nil, err + } + go eth.freezer.Freeze(eth.blockchain, chainDb, time.Minute, 60000) + // Rewind the chain in case of an incompatible config upgrade. if compat, ok := genesisErr.(*params.ConfigCompatError); ok { log.Warn("Rewinding chain to upgrade configuration", "err", compat) diff --git a/freezer/freezer.go b/freezer/freezer.go new file mode 100644 index 0000000000..0c6951c37a --- /dev/null +++ b/freezer/freezer.go @@ -0,0 +1,213 @@ +// Copyright 2018 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 freezer implements an append-only immutable mmap chain database. +package freezer + +import ( + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" + "github.com/ethereum/go-ethereum/rlp" +) + +// 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 { + frozen uint64 // Number of blocks already frozen + + headers *table // Data table for storing the block headers + bodies *table // Data table for storing the block bodies + receipts *table // Data table for storing the block receipts + diffs *table // Data table for storing the block tds + + logger log.Logger // Contextual logger for the freezer database +} + +// New creates a chain freezer that moves ancient chain data into immutable flat +// file containers. +func New(datadir string) (*Freezer, error) { + // Create the initial freezer object + var ( + freezer = &Freezer{ + logger: log.New("path", datadir), + } + readMeter = metrics.NewRegisteredMeter("eth/db/freezer/read", nil) + writeMeter = metrics.NewRegisteredMeter("eth/db/freezer/write", nil) + err error + ) + // Open all the supported data tables + if freezer.headers, err = newTable(datadir, "headers", readMeter, writeMeter); err != nil { + return nil, err + } + if freezer.bodies, err = newTable(datadir, "bodies", readMeter, writeMeter); err != nil { + freezer.headers.Close() + return nil, err + } + if freezer.receipts, err = newTable(datadir, "receipts", readMeter, writeMeter); err != nil { + freezer.bodies.Close() + freezer.headers.Close() + return nil, err + } + if freezer.diffs, err = newTable(datadir, "diffs", readMeter, writeMeter); err != nil { + freezer.receipts.Close() + freezer.bodies.Close() + freezer.headers.Close() + return nil, err + } + return freezer, nil +} + +// Close terminates the chain freezer, unmapping all the data files. +func (f *Freezer) Close() error { + f.diffs.Close() + f.receipts.Close() + f.bodies.Close() + f.headers.Close() + + 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(chain *core.BlockChain, db ethdb.Database, recheck time.Duration, delay uint64) { + for { + // Retrieve the freezing threshold. In theory we're interested only in full + // blocks post-sync, but that would keep the live database enormous during + // dast sync. By picking the fast block, we still get to deep freeze all the + // final immutable data without having to wait for sync to finish. + head := chain.CurrentFastBlock() + if head == nil { + log.Error("Current fast block is nil") + time.Sleep(recheck) + continue + } + if head.NumberU64() < delay { + log.Debug("Current block not old enough", "number", head.Number(), "hash", head.Hash(), "age", common.PrettyAge(time.Unix(head.Time().Int64(), 0)), "delay", delay) + time.Sleep(recheck) + continue + } + limit := head.NumberU64() - delay + if limit <= f.frozen { + log.Debug("Ancient blocks frozen already") + time.Sleep(recheck) + continue + } + // Seems we have data ready to be frozen, process in usable batches + if limit-f.frozen > 30000 { + limit = f.frozen + 30000 + } + var ( + start = time.Now() + first = f.frozen + last *types.Block + ) + for f.frozen < limit { + // Deep freeze the next canonical block if it's available + if block := chain.GetBlockByNumber(f.frozen); block != nil { + // Deep freeze the block header and body + blob, _ := rlp.EncodeToBytes(block.Header()) + if err := f.headers.Append(f.frozen, blob); err != nil { + log.Error("Failed to deep freeze header", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)), "err", err) + break + } + blob, _ = rlp.EncodeToBytes(block.Body()) + if err := f.bodies.Append(f.frozen, blob); err != nil { + log.Error("Failed to deep freeze body", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)), "err", err) + break + } + // Deep freeze the block receipts and total difficulty + if receipts := chain.GetReceiptsByHash(block.Hash()); receipts != nil { + blob, _ = rlp.EncodeToBytes(receipts) + if err := f.receipts.Append(f.frozen, blob); err != nil { + log.Error("Failed to deep freeze receipts", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)), "err", err) + break + } + } + if td := chain.GetTd(block.Hash(), block.NumberU64()); td != nil { + blob, _ = rlp.EncodeToBytes(td) + if err := f.diffs.Append(f.frozen, blob); err != nil { + log.Error("Failed to deep freeze difficulty", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0)), "err", err) + break + } + } + log.Trace("Deep froze ancient block", "number", block.Number(), "hash", block.Hash(), "age", common.PrettyAge(time.Unix(block.Time().Int64(), 0))) + f.frozen++ + + // If it's the last block, save for reporting + if f.frozen == limit-1 { + last = block + } + } + } + // Batch of blocks have been frozen, flush them before wiping from leveldb + if err := f.headers.Flush(); err != nil { + f.logger.Error("Failed to flush frozen headers", "err", err) + time.Sleep(recheck) + continue + } + if err := f.bodies.Flush(); err != nil { + f.logger.Error("Failed to flush frozen bodies", "err", err) + time.Sleep(recheck) + continue + } + if err := f.receipts.Flush(); err != nil { + f.logger.Error("Failed to flush frozen receipts", "err", err) + time.Sleep(recheck) + continue + } + if err := f.diffs.Flush(); err != nil { + f.logger.Error("Failed to flush frozen diffs", "err", err) + time.Sleep(recheck) + continue + } + // Wipe out all data from the active database + for number := first; number < f.frozen; number++ { + if number == 0 { + // Skip deleting the genesis for the PoC + continue + } + rawdb.DeleteBlock(db, rawdb.ReadCanonicalHash(db, number), number) + } + // Log something friendly for the user + context := []interface{}{ + "count", f.frozen - first, "elapsed", common.PrettyDuration(time.Since(start)), "number", f.frozen - 1, + } + if last != nil { + context = append(context, []interface{}{"hash", last.Hash(), "age", common.PrettyAge(time.Unix(last.Time().Int64(), 0))}...) + } + log.Info("Deep froze chain segment", context...) + + // Avoid database thrashing with tiny writes + if f.frozen-first < 30000 { + time.Sleep(recheck) + } + } +} diff --git a/freezer/mmap.go b/freezer/mmap.go new file mode 100644 index 0000000000..40d3836fe8 --- /dev/null +++ b/freezer/mmap.go @@ -0,0 +1,86 @@ +// Copyright 2018 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 freezer + +import ( + "os" + "reflect" + "unsafe" + + "github.com/edsrzf/mmap-go" +) + +// openWithSize opens a file and ensures it is at least the provided bytes in +// size, growing it if necessary. +func openWithSize(path string, size uint64) (*os.File, error) { + // Open the file for writing, potentially creating it + file, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0644) + if err != nil { + return nil, err + } + // Ensure the file's size is at least as much as requested + info, err := file.Stat() + if err != nil { + file.Close() + return nil, err + } + if info.Size() < int64(size) { + if err := file.Truncate(int64(size)); err != nil { + file.Close() + return nil, err + } + } + // File is correctly open and of the correct size + return file, err +} + +// mmapBytes tries to memory map a file, creating it if it's non existent. +func mmapBytes(path string, size uint64) (*os.File, mmap.MMap, []byte, error) { + // Open the file to memory map and ensure it's large enough + file, err := openWithSize(path, size) + if err != nil { + return nil, nil, nil, err + } + // Memory map the file, cast to a byte slice and return + mem, err := mmap.Map(file, mmap.RDWR, 0) + if err != nil { + file.Close() + return nil, nil, nil, err + } + header := *(*reflect.SliceHeader)(unsafe.Pointer(&mem)) + return file, mem, *(*[]byte)(unsafe.Pointer(&header)), nil +} + +// mmapUints tries to memory map a file, creating it if it's non existent. +func mmapUints(path string, size uint64) (*os.File, mmap.MMap, []uint64, error) { + // Open the file to memory map and ensure it's large enough + file, err := openWithSize(path, size) + if err != nil { + return nil, nil, nil, err + } + // Memory map the file, cast to an uint64 slice and return + mem, err := mmap.Map(file, mmap.RDWR, 0) + if err != nil { + file.Close() + return nil, nil, nil, err + } + header := *(*reflect.SliceHeader)(unsafe.Pointer(&mem)) + header.Len /= 8 + header.Cap /= 8 + + return file, mem, *(*[]uint64)(unsafe.Pointer(&header)), nil +} diff --git a/freezer/table.go b/freezer/table.go new file mode 100644 index 0000000000..81aa673c94 --- /dev/null +++ b/freezer/table.go @@ -0,0 +1,290 @@ +// Copyright 2018 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 freezer + +import ( + "errors" + "os" + "path/filepath" + "sync" + + "github.com/edsrzf/mmap-go" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" + "github.com/golang/snappy" +) + +const ( + growthIndex uint64 = 1024 * 1024 // Growth rate of the index file when remapping + growthData uint64 = 128 * 1024 * 1024 // Growth rate of the data file when remapping +) + +var ( + // errAlreadyExists is returned if the user attempts to append an item to the + // freezer table that already exists (i.e. it's offset is lower or equal to + // the head of the immutable file). + errAlreadyExists = errors.New("item already exists") + + // errGappedWrite is returned if the user attempts to append an item to the + // freezer table that would produce a data gap (i.e. it's offset is larger than + // the head of the immutable file). + errGappedWrite = errors.New("item produces data gap") + + // errTableInaccessible is returned if a previously well functioning freezer + // table becomes non-accessible after a memory remap (resize). + errTableInaccessible = errors.New("table not accessible") + + // errOutOfBounds is returned if the item requested is not contained within the + // freezer table. + errOutOfBounds = errors.New("out of bounds") +) + +// table represents a single chained data table within the freezer (e.g. blocks). +// It consists of a data file (snappy encoded arbitrary data blobs) and an index +// file (uncompressed 64 bit indices into the data file). In addition a counter +// (binary) file is also created which simply contains the number of items. +type table struct { + path string // Database folder to store the files into + name string // Table name to multiplex multiple tables into the same folder + + fileData *os.File // File descriptor for the data region + fileIndex *os.File // File descriptor for the offset region + fileCounter *os.File // File descriptor for the counter region + + mmapData mmap.MMap // Memory mapping for the data region + mmapIndex mmap.MMap // Memory mapping for the offset region + mmapCounter mmap.MMap // Memory mapping for the counter region + + rawData []byte // Direct memory region for the data file + rawIndex []uint64 // Direct memory region for the offset file + rawCounter []uint64 // Direct memory region for the counter (single item) + + readMeter metrics.Meter // Meter for measuring the effective amount of data read + writeMeter metrics.Meter // Meter for measuring the effective amount of data written + + logger log.Logger // Logger with database path and table name ambedded + lock sync.RWMutex // Lock protecting the reads from remaps +} + +// newTable attempts to new freezer table by memory mapping the composing files. +// If no file exists, it will create new ones. +func newTable(path string, name string, readMeter metrics.Meter, writeMeter metrics.Meter) (*table, error) { + if err := os.MkdirAll(path, 0755); err != nil { + return nil, err + } + tab := &table{ + path: path, + name: name, + readMeter: readMeter, + writeMeter: writeMeter, + logger: log.New("path", path, "table", name), + } + if err := tab.ensure(); err != nil { + return nil, err + } + return tab, nil +} + +// ensure checks all the memory region limits and ensures that they conform to +// the required data counts. If anything is off, this method will recreate and +// remap accordingly. +func (t *table) ensure() error { + // Figure out if we need to remap or not + // If the regions are already open and large enough, leave as is + if t.rawCounter != nil { + // Ensure the data file is large enough, unmap it otherwise + wantData := (t.rawIndex[t.rawCounter[0]]/growthData + 1) * growthData + if uint64(len(t.rawData)) < wantData { + if err := t.mmapData.Unmap(); err != nil { + t.logger.Error("Failed to unmap freezer datastore", "err", err) + } + if err := t.fileData.Close(); err != nil { + t.logger.Error("Failed to close freezer datastore", "err", err) + } + t.fileData, t.mmapData, t.rawData = nil, nil, nil + } + // Ensure the index file is large enough, unmap it otherwise + wantIndex := (((t.rawCounter[0]+1)*8)/growthIndex + 1) * growthIndex + if uint64(len(t.rawIndex)) < wantIndex { + if err := t.mmapIndex.Unmap(); err != nil { + t.logger.Error("Failed to unmap freezer index", "err", err) + } + if err := t.fileIndex.Close(); err != nil { + t.logger.Error("Failed to close freezer index", "err", err) + } + t.fileIndex, t.mmapIndex, t.rawIndex = nil, nil, nil + } + } + // Memory map the counter and retrieve the size of the offset file + var err error + + if t.rawCounter == nil { + if t.fileCounter, t.mmapCounter, t.rawCounter, err = mmapUints(filepath.Join(t.path, t.name+".len"), 8); err != nil { + return err + } + } + // Memory map the index file and retrieve the size of the data file + if t.rawIndex == nil { + size := (((t.rawCounter[0]+1)*8)/growthIndex + 1) * growthIndex + if t.fileIndex, t.mmapIndex, t.rawIndex, err = mmapUints(filepath.Join(t.path, t.name+".idx"), size); err != nil { + if err := t.mmapCounter.Unmap(); err != nil { + t.logger.Error("Failed to unmap freezer counter", "err", err) + } + if err := t.fileCounter.Close(); err != nil { + t.logger.Error("Failed to close freezer counter", "err", err) + } + t.fileCounter, t.mmapCounter, t.rawCounter = nil, nil, nil + return err + } + } + // Memory map the data file and return the final freezer table + if t.rawData == nil { + size := growthData + if t.rawCounter[0] > 0 { + size = (t.rawIndex[t.rawCounter[0]]/growthData + 1) * growthData + } + if t.fileData, t.mmapData, t.rawData, err = mmapBytes(filepath.Join(t.path, t.name+".dat"), size); err != nil { + if err := t.mmapIndex.Unmap(); err != nil { + t.logger.Error("Failed to unmap freezer index", "err", err) + } + if err := t.fileIndex.Close(); err != nil { + t.logger.Error("Failed to close freezer index", "err", err) + } + t.fileIndex, t.mmapIndex, t.rawIndex = nil, nil, nil + + if err := t.mmapCounter.Unmap(); err != nil { + t.logger.Error("Failed to unmap freezer counter", "err", err) + } + if err := t.fileCounter.Close(); err != nil { + t.logger.Error("Failed to close freezer counter", "err", err) + } + t.fileCounter, t.mmapCounter, t.rawCounter = nil, nil, nil + + return err + } + } + return nil +} + +// Close unmaps all active memory mapped regions. +func (t *table) Close() error { + t.lock.Lock() + defer t.lock.Unlock() + + if t.mmapData != nil { + if err := t.mmapData.Unmap(); err != nil { + t.logger.Error("Failed to unmap freezer datastore", "err", err) + } + if err := t.fileData.Close(); err != nil { + t.logger.Error("Failed to close freezer datastore", "err", err) + } + t.fileData, t.mmapData, t.rawData = nil, nil, nil + } + if t.mmapIndex != nil { + if err := t.mmapIndex.Unmap(); err != nil { + t.logger.Error("Failed to unmap freezer index", "err", err) + } + if err := t.fileIndex.Close(); err != nil { + t.logger.Error("Failed to close freezer index", "err", err) + } + t.fileIndex, t.mmapIndex, t.rawIndex = nil, nil, nil + } + if t.mmapCounter != nil { + if err := t.mmapCounter.Unmap(); err != nil { + t.logger.Error("Failed to unmap freezer counter", "err", err) + } + if err := t.fileCounter.Close(); err != nil { + t.logger.Error("Failed to close freezer counter", "err", err) + } + t.fileCounter, t.mmapCounter, t.rawCounter = nil, nil, nil + } + return nil +} + +// Append injects a binary blob at the end of the freezer table. The item index +// is a precautionary parameter to ensure data correctness, but the table will +// reject already existing data. +// +// Note, this method will *not* flush any data to disk (unless the files need to +// be resized). Be sure to explicitly flush before irreversibly deleting data +// from the fast chain database. +func (t *table) Append(item uint64, blob []byte) error { + t.lock.Lock() + defer t.lock.Unlock() + + // Ensure the table is still accessible + if t.rawCounter == nil { + if err := t.ensure(); err != nil { + t.logger.Error("Failed to re-access table", "err", err) + return errTableInaccessible + } + } + // Ensure only the next item can be written, nothing else + items := t.rawCounter[0] + if items > item { + return errAlreadyExists + } + if items < item { + return errGappedWrite + } + // Encode the blob and write it into the data file + blob = snappy.Encode(nil, blob) + + t.rawIndex[items+1] = t.rawIndex[items] + uint64(len(blob)) + copy(t.rawData[t.rawIndex[items]:], blob) + t.rawCounter[0]++ + + t.writeMeter.Mark(int64(len(blob))) + + // Ensure we have enough space for future appends + return t.ensure() +} + +// Retrieve looks up the data offset of an item with the given index and retrieves +// the raw binary blob from the data file. +func (t *table) Retrieve(item uint64) ([]byte, error) { + t.lock.RLock() + defer t.lock.RUnlock() + + // Ensure the table and the item is accessible + if t.rawCounter == nil { + return nil, errTableInaccessible + } + if t.rawCounter[0] <= item { + return nil, errOutOfBounds + } + // Item reachable, retrive and return to the user + blob := t.rawData[t.rawIndex[item]:t.rawIndex[item+1]] + t.writeMeter.Mark(int64(len(blob))) + + return snappy.Decode(nil, blob) +} + +// Flush pushes any pending data from memory out to disk. This is an expensive +// operation, so use it with care. +func (t *table) Flush() error { + if err := t.mmapData.Flush(); err != nil { + return err + } + if err := t.mmapIndex.Flush(); err != nil { + return err + } + if err := t.mmapCounter.Flush(); err != nil { + return err + } + return nil +}