From b69bdc2a4f8efab7cb99934652e2d9b2508c4544 Mon Sep 17 00:00:00 2001
From: Martin Holst Swende <martin@swende.se>
Date: Tue, 26 Mar 2019 12:28:23 +0100
Subject: [PATCH] freezer: implement split files for data
* freezer: implement split files for data
* freezer: add tests
* freezer: close old head-file when opening next
* freezer: fix truncation
* freezer: more testing around close/open
* rawdb/freezer: address review concerns
* freezer: fix minor review concerns
* freezer: fix remaining concerns + testcases around truncation
* freezer: docs
* freezer: implement multithreading
* core/rawdb: fix freezer nitpicks + change offsets to uint32
* freezer: preopen files, simplify lock constructs
* freezer: delete files during truncation
---
core/rawdb/freezer_table.go | 399 ++++++++++++++++++------
core/rawdb/freezer_table_test.go | 511 +++++++++++++++++++++++++++++++
2 files changed, 815 insertions(+), 95 deletions(-)
create mode 100644 core/rawdb/freezer_table_test.go
diff --git a/core/rawdb/freezer_table.go b/core/rawdb/freezer_table.go
index 546db0c65..313ac8b78 100644
--- a/core/rawdb/freezer_table.go
+++ b/core/rawdb/freezer_table.go
@@ -1,4 +1,4 @@
-// Copyright 2018 The go-ethereum Authors
+// 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
@@ -23,6 +23,7 @@ import (
"os"
"path/filepath"
"sync"
+ "sync/atomic"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
@@ -40,16 +41,47 @@ var (
errOutOfBounds = errors.New("out of bounds")
)
+// indexEntry contains the number/id of the file that the data resides in, aswell as the
+// offset within the file to the end of the data
+// In serialized form, the filenum is stored as uint16.
+type indexEntry struct {
+ filenum uint32 // stored as uint16 ( 2 bytes)
+ offset uint32 // stored as uint32 ( 4 bytes)
+}
+
+const indexEntrySize = 6
+
+// unmarshallBinary deserializes binary b into the rawIndex entry.
+func (i *indexEntry) unmarshalBinary(b []byte) error {
+ i.filenum = uint32(binary.BigEndian.Uint16(b[:2]))
+ i.offset = binary.BigEndian.Uint32(b[2:6])
+ return nil
+}
+
+// marshallBinary serializes the rawIndex entry into binary.
+func (i *indexEntry) marshallBinary() []byte {
+ b := make([]byte, indexEntrySize)
+ binary.BigEndian.PutUint16(b[:2], uint16(i.filenum))
+ binary.BigEndian.PutUint32(b[2:6], i.offset)
+ return b
+}
+
// freezerTable 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
+// It consists of a data file (snappy encoded arbitrary data blobs) and an indexEntry
// file (uncompressed 64 bit indices into the data file).
type freezerTable struct {
- content *os.File // File descriptor for the data content of the table
- offsets *os.File // File descriptor for the index file of the table
+ noCompression bool // if true, disables snappy compression. Note: does not work retroactively
+ maxFileSize uint32 // Max file size for data-files
+ name string
+ path string
- items uint64 // Number of items stored in the table
- bytes uint64 // Number of content bytes stored in the table
+ head *os.File // File descriptor for the data head of the table
+ files map[uint32]*os.File // open files
+ headId uint32 // number of the currently active head file
+ index *os.File // File descriptor for the indexEntry file of the table
+ items uint64 // Number of items stored in the table
+ headBytes uint32 // Number of bytes written to the head file
readMeter metrics.Meter // Meter for measuring the effective amount of data read
writeMeter metrics.Meter // Meter for measuring the effective amount of data written
@@ -57,149 +89,231 @@ type freezerTable struct {
lock sync.RWMutex // Mutex protecting the data file descriptors
}
-// newTable opens a freezer table, creating the data and index files if they are
+// newTable opens a freezer table with default settings - 2G files and snappy compression
+func newTable(path string, name string, readMeter metrics.Meter, writeMeter metrics.Meter) (*freezerTable, error) {
+ return newCustomTable(path, name, readMeter, writeMeter, 2*1000*1000*1000, false)
+}
+
+// newCustomTable opens a freezer table, creating the data and index files if they are
// non existent. Both files are truncated to the shortest common length to ensure
// they don't go out of sync.
-func newTable(path string, name string, readMeter metrics.Meter, writeMeter metrics.Meter) (*freezerTable, error) {
- // Ensure the containing directory exists and open the two data files
+func newCustomTable(path string, name string, readMeter metrics.Meter, writeMeter metrics.Meter, maxFilesize uint32, noCompression bool) (*freezerTable, error) {
+ // Ensure the containing directory exists and open the indexEntry file
if err := os.MkdirAll(path, 0755); err != nil {
return nil, err
}
- content, err := os.OpenFile(filepath.Join(path, name+".dat"), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
- if err != nil {
- return nil, err
+ var idxName string
+ if noCompression {
+ // raw idx
+ idxName = fmt.Sprintf("%s.ridx", name)
+ } else {
+ // compressed idx
+ idxName = fmt.Sprintf("%s.cidx", name)
}
- offsets, err := os.OpenFile(filepath.Join(path, name+".idx"), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
+ offsets, err := os.OpenFile(filepath.Join(path, idxName), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
if err != nil {
- content.Close()
return nil, err
}
// Create the table and repair any past inconsistency
tab := &freezerTable{
- content: content,
- offsets: offsets,
- readMeter: readMeter,
- writeMeter: writeMeter,
- logger: log.New("database", path, "table", name),
+ index: offsets,
+ files: make(map[uint32]*os.File),
+ readMeter: readMeter,
+ writeMeter: writeMeter,
+ name: name,
+ path: path,
+ logger: log.New("database", path, "table", name),
+ noCompression: noCompression,
+ maxFileSize: maxFilesize,
}
if err := tab.repair(); err != nil {
- offsets.Close()
- content.Close()
+ tab.Close()
return nil, err
}
return tab, nil
}
-// repair cross checks the content and the offsets file and truncates them to
+// repair cross checks the head and the index file and truncates them to
// be in sync with each other after a potential crash / data loss.
func (t *freezerTable) repair() error {
- // Create a temporary offset buffer to init files with and read offsts into
- offset := make([]byte, 8)
+ // Create a temporary offset buffer to init files with and read indexEntry into
+ buffer := make([]byte, indexEntrySize)
- // If we've just created the files, initialize the offsets with the 0 index
- stat, err := t.offsets.Stat()
+ // If we've just created the files, initialize the index with the 0 indexEntry
+ stat, err := t.index.Stat()
if err != nil {
return err
}
if stat.Size() == 0 {
- if _, err := t.offsets.Write(offset); err != nil {
+ if _, err := t.index.Write(buffer); err != nil {
return err
}
}
- // Ensure the offsets are a multiple of 8 bytes
- if overflow := stat.Size() % 8; overflow != 0 {
- t.offsets.Truncate(stat.Size() - overflow) // New file can't trigger this path
+ // Ensure the index is a multiple of indexEntrySize bytes
+ if overflow := stat.Size() % indexEntrySize; overflow != 0 {
+ t.index.Truncate(stat.Size() - overflow) // New file can't trigger this path
}
// Retrieve the file sizes and prepare for truncation
- if stat, err = t.offsets.Stat(); err != nil {
+ if stat, err = t.index.Stat(); err != nil {
return err
}
offsetsSize := stat.Size()
- if stat, err = t.content.Stat(); err != nil {
+ // Open the head file
+ var (
+ lastIndex indexEntry
+ contentSize int64
+ contentExp int64
+ )
+ t.index.ReadAt(buffer, offsetsSize-indexEntrySize)
+ lastIndex.unmarshalBinary(buffer)
+ t.head, err = t.openFile(lastIndex.filenum, os.O_RDWR|os.O_CREATE|os.O_APPEND)
+ if err != nil {
+ return err
+ }
+ if stat, err = t.head.Stat(); err != nil {
return err
}
- contentSize := stat.Size()
+ contentSize = stat.Size()
// Keep truncating both files until they come in sync
- t.offsets.ReadAt(offset, offsetsSize-8)
- contentExp := int64(binary.LittleEndian.Uint64(offset))
+ contentExp = int64(lastIndex.offset)
for contentExp != contentSize {
- // Truncate the content file to the last offset pointer
+ // Truncate the head file to the last offset pointer
if contentExp < contentSize {
- t.logger.Warn("Truncating dangling content", "indexed", common.StorageSize(contentExp), "stored", common.StorageSize(contentSize))
- if err := t.content.Truncate(contentExp); err != nil {
+ t.logger.Warn("Truncating dangling head", "indexed", common.StorageSize(contentExp), "stored", common.StorageSize(contentSize))
+ if err := t.head.Truncate(contentExp); err != nil {
return err
}
contentSize = contentExp
}
- // Truncate the offsets to point within the content file
+ // Truncate the index to point within the head file
if contentExp > contentSize {
- t.logger.Warn("Truncating dangling offsets", "indexed", common.StorageSize(contentExp), "stored", common.StorageSize(contentSize))
- if err := t.offsets.Truncate(offsetsSize - 8); err != nil {
+ t.logger.Warn("Truncating dangling indexes", "indexed", common.StorageSize(contentExp), "stored", common.StorageSize(contentSize))
+ if err := t.index.Truncate(offsetsSize - indexEntrySize); err != nil {
return err
}
- offsetsSize -= 8
-
- t.offsets.ReadAt(offset, offsetsSize-8)
- contentExp = int64(binary.LittleEndian.Uint64(offset))
+ offsetsSize -= indexEntrySize
+ t.index.ReadAt(buffer, offsetsSize-indexEntrySize)
+ var newLastIndex indexEntry
+ newLastIndex.unmarshalBinary(buffer)
+ // We might have slipped back into an earlier head-file here
+ if newLastIndex.filenum != lastIndex.filenum {
+ // release earlier opened file
+ t.releaseFile(lastIndex.filenum)
+ t.head, err = t.openFile(newLastIndex.filenum, os.O_RDWR|os.O_CREATE|os.O_APPEND)
+ if stat, err = t.head.Stat(); err != nil {
+ // TODO, anything more we can do here?
+ // A data file has gone missing...
+ return err
+ }
+ contentSize = stat.Size()
+ }
+ lastIndex = newLastIndex
+ contentExp = int64(lastIndex.offset)
}
}
// Ensure all reparation changes have been written to disk
- if err := t.offsets.Sync(); err != nil {
+ if err := t.index.Sync(); err != nil {
return err
}
- if err := t.content.Sync(); err != nil {
+ if err := t.head.Sync(); err != nil {
return err
}
// Update the item and byte counters and return
- t.items = uint64(offsetsSize/8 - 1) // last index points to the end of the data file
- t.bytes = uint64(contentSize)
+ t.items = uint64(offsetsSize/indexEntrySize - 1) // last indexEntry points to the end of the data file
+ t.headBytes = uint32(contentSize)
+ t.headId = lastIndex.filenum
- t.logger.Debug("Chain freezer table opened", "items", t.items, "size", common.StorageSize(t.bytes))
+ // Close opened files and preopen all files
+ if err := t.preopen(); err != nil {
+ return err
+ }
+ t.logger.Debug("Chain freezer table opened", "items", t.items, "size", common.StorageSize(t.headBytes))
return nil
}
+// preopen opens all files that the freezer will need. This method should be called from an init-context,
+// since it assumes that it doesn't have to bother with locking
+// The rationale for doing preopen is to not have to do it from within Retrieve, thus not needing to ever
+// obtain a write-lock within Retrieve.
+func (t *freezerTable) preopen() (err error) {
+ // The repair might have already opened (some) files
+ t.releaseFilesAfter(0, false)
+ // Open all except head in RDONLY
+ for i := uint32(0); i < t.headId; i++ {
+ if _, err = t.openFile(i, os.O_RDONLY); err != nil {
+ return err
+ }
+ }
+ // Open head in read/write
+ t.head, err = t.openFile(t.headId, os.O_RDWR|os.O_CREATE|os.O_APPEND)
+ return err
+}
+
// truncate discards any recent data above the provided threashold number.
func (t *freezerTable) truncate(items uint64) error {
+ t.lock.Lock()
+ defer t.lock.Unlock()
// If out item count is corrent, don't do anything
- if t.items <= items {
+ if atomic.LoadUint64(&t.items) <= items {
return nil
}
// Something's out of sync, truncate the table's offset index
t.logger.Warn("Truncating freezer table", "items", t.items, "limit", items)
- if err := t.offsets.Truncate(int64(items+1) * 8); err != nil {
+ if err := t.index.Truncate(int64(items+1) * indexEntrySize); err != nil {
return err
}
// Calculate the new expected size of the data file and truncate it
- offset := make([]byte, 8)
- t.offsets.ReadAt(offset, int64(items)*8)
- expected := binary.LittleEndian.Uint64(offset)
+ buffer := make([]byte, indexEntrySize)
+ if _, err := t.index.ReadAt(buffer, int64(items*indexEntrySize)); err != nil {
+ return err
+ }
+ var expected indexEntry
+ expected.unmarshalBinary(buffer)
+ // We might need to truncate back to older files
+ if expected.filenum != t.headId {
+ // If already open for reading, force-reopen for writing
+ t.releaseFile(expected.filenum)
+ newHead, err := t.openFile(expected.filenum, os.O_RDWR|os.O_CREATE|os.O_APPEND)
+ if err != nil {
+ return err
+ }
+ // release any files _after the current head -- both the previous head
+ // and any files which may have been opened for reading
+ t.releaseFilesAfter(expected.filenum, true)
+ // set back the historic head
+ t.head = newHead
+ atomic.StoreUint32(&t.headId, expected.filenum)
+ }
- if err := t.content.Truncate(int64(expected)); err != nil {
+ if err := t.head.Truncate(int64(expected.offset)); err != nil {
return err
}
// All data files truncated, set internal counters and return
- t.items, t.bytes = items, expected
+ atomic.StoreUint64(&t.items, items)
+ atomic.StoreUint32(&t.headBytes, expected.offset)
return nil
}
-// Close unmaps all active memory mapped regions.
+// Close closes all opened files.
func (t *freezerTable) Close() error {
t.lock.Lock()
defer t.lock.Unlock()
var errs []error
- if err := t.offsets.Close(); err != nil {
+ if err := t.index.Close(); err != nil {
errs = append(errs, err)
}
- t.offsets = nil
+ t.index = nil
- if err := t.content.Close(); err != nil {
- errs = append(errs, err)
+ for _, f := range t.files {
+ if err := f.Close(); err != nil {
+ errs = append(errs, err)
+ }
}
- t.content = nil
+ t.head = nil
if errs != nil {
return fmt.Errorf("%v", errs)
@@ -207,78 +321,173 @@ func (t *freezerTable) Close() error {
return nil
}
-// Append injects a binary blob at the end of the freezer table. The item index
+// openFile assumes that the write-lock is held by the caller
+func (t *freezerTable) openFile(num uint32, flag int) (f *os.File, err error) {
+ var exist bool
+ if f, exist = t.files[num]; !exist {
+ var name string
+ if t.noCompression {
+ name = fmt.Sprintf("%s.%d.rdat", t.name, num)
+ } else {
+ name = fmt.Sprintf("%s.%d.cdat", t.name, num)
+ }
+ f, err = os.OpenFile(filepath.Join(t.path, name), flag, 0644)
+ if err != nil {
+ return nil, err
+ }
+ t.files[num] = f
+ }
+ return f, err
+}
+
+// releaseFile closes a file, and removes it from the open file cache.
+// Assumes that the caller holds the write lock
+func (t *freezerTable) releaseFile(num uint32) {
+ if f, exist := t.files[num]; exist {
+ delete(t.files, num)
+ f.Close()
+ }
+}
+
+// releaseFilesAfter closes all open files with a higher number, and optionally also deletes the files
+func (t *freezerTable) releaseFilesAfter(num uint32, remove bool) {
+ for fnum, f := range t.files {
+ if fnum > num {
+ delete(t.files, fnum)
+ f.Close()
+ if remove {
+ os.Remove(f.Name())
+ }
+ }
+ }
+}
+
+// Append injects a binary blob at the end of the freezer table. The item number
// 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 so be sure to explicitly
// fsync before irreversibly deleting data from the database.
func (t *freezerTable) Append(item uint64, blob []byte) error {
+ // Read lock prevents competition with truncate
+ t.lock.RLock()
// Ensure the table is still accessible
- if t.offsets == nil || t.content == nil {
+ if t.index == nil || t.head == nil {
return errClosed
}
// Ensure only the next item can be written, nothing else
- if t.items != item {
+ if atomic.LoadUint64(&t.items) != item {
panic(fmt.Sprintf("appending unexpected item: want %d, have %d", t.items, item))
}
// Encode the blob and write it into the data file
- blob = snappy.Encode(nil, blob)
- if _, err := t.content.Write(blob); err != nil {
- return err
+ if !t.noCompression {
+ blob = snappy.Encode(nil, blob)
}
- t.bytes += uint64(len(blob))
+ bLen := uint32(len(blob))
+ if t.headBytes+bLen < bLen ||
+ t.headBytes+bLen > t.maxFileSize {
+ // we need a new file, writing would overflow
+ t.lock.RUnlock()
+ t.lock.Lock()
+ nextId := atomic.LoadUint32(&t.headId) + 1
+ // We open the next file in truncated mode -- if this file already
+ // exists, we need to start over from scratch on it
+ newHead, err := t.openFile(nextId, os.O_RDWR|os.O_CREATE|os.O_TRUNC)
+ if err != nil {
+ t.lock.Unlock()
+ return err
+ }
+ // Close old file, and reopen in RDONLY mode
+ t.releaseFile(t.headId)
+ t.openFile(t.headId, os.O_RDONLY)
- offset := make([]byte, 8)
- binary.LittleEndian.PutUint64(offset, t.bytes)
- if _, err := t.offsets.Write(offset); err != nil {
- return err
+ // Swap out the current head
+ t.head = newHead
+ atomic.StoreUint32(&t.headBytes, 0)
+ atomic.StoreUint32(&t.headId, nextId)
+ t.lock.Unlock()
+ t.lock.RLock()
}
- t.items++
- t.writeMeter.Mark(int64(len(blob) + 8)) // 8 = 1 x 8 byte offset
+ defer t.lock.RUnlock()
+
+ if _, err := t.head.Write(blob); err != nil {
+ return err
+ }
+ newOffset := atomic.AddUint32(&t.headBytes, bLen)
+ idx := indexEntry{
+ filenum: atomic.LoadUint32(&t.headId),
+ offset: newOffset,
+ }
+ // Write indexEntry
+ t.index.Write(idx.marshallBinary())
+ t.writeMeter.Mark(int64(bLen + indexEntrySize))
+ atomic.AddUint64(&t.items, 1)
return nil
}
-// Retrieve looks up the data offset of an item with the given index and retrieves
+// getBounds returns the indexes for the item
+// returns start, end, filenumber and error
+func (t *freezerTable) getBounds(item uint64) (uint32, uint32, uint32, error) {
+ var startIdx, endIdx indexEntry
+ buffer := make([]byte, indexEntrySize)
+ if _, err := t.index.ReadAt(buffer, int64(item*indexEntrySize)); err != nil {
+ return 0, 0, 0, err
+ }
+ startIdx.unmarshalBinary(buffer)
+ if _, err := t.index.ReadAt(buffer, int64((item+1)*indexEntrySize)); err != nil {
+ return 0, 0, 0, err
+ }
+ endIdx.unmarshalBinary(buffer)
+ if startIdx.filenum != endIdx.filenum {
+ // If a piece of data 'crosses' a data-file,
+ // it's actually in one piece on the second data-file.
+ // We return a zero-indexEntry for the second file as start
+ return 0, endIdx.offset, endIdx.filenum, nil
+ }
+ return startIdx.offset, endIdx.offset, endIdx.filenum, nil
+}
+
+// Retrieve looks up the data offset of an item with the given number and retrieves
// the raw binary blob from the data file.
func (t *freezerTable) Retrieve(item uint64) ([]byte, error) {
- t.lock.RLock()
- defer t.lock.RUnlock()
// Ensure the table and the item is accessible
- if t.offsets == nil || t.content == nil {
+ if t.index == nil || t.head == nil {
return nil, errClosed
}
- if t.items <= item {
+ if atomic.LoadUint64(&t.items) <= item {
return nil, errOutOfBounds
}
- // Item reachable, retrieve the data content boundaries
- offset := make([]byte, 8)
- if _, err := t.offsets.ReadAt(offset, int64(item*8)); err != nil {
+ t.lock.RLock()
+ startOffset, endOffset, filenum, err := t.getBounds(item)
+ if err != nil {
return nil, err
}
- start := binary.LittleEndian.Uint64(offset)
-
- if _, err := t.offsets.ReadAt(offset, int64((item+1)*8)); err != nil {
- return nil, err
+ dataFile, exist := t.files[filenum]
+ if !exist {
+ return nil, fmt.Errorf("missing data file %d", filenum)
}
- end := binary.LittleEndian.Uint64(offset)
-
// Retrieve the data itself, decompress and return
- blob := make([]byte, end-start)
- if _, err := t.content.ReadAt(blob, int64(start)); err != nil {
+ blob := make([]byte, endOffset-startOffset)
+ if _, err := dataFile.ReadAt(blob, int64(startOffset)); err != nil {
+ t.lock.RUnlock()
return nil, err
}
- t.readMeter.Mark(int64(len(blob) + 16)) // 16 = 2 x 8 byte offset
+ t.lock.RUnlock()
+ t.readMeter.Mark(int64(len(blob) + 2*indexEntrySize))
+
+ if t.noCompression {
+ return blob, nil
+ }
return snappy.Decode(nil, blob)
}
// Sync pushes any pending data from memory out to disk. This is an expensive
// operation, so use it with care.
func (t *freezerTable) Sync() error {
- if err := t.offsets.Sync(); err != nil {
+ if err := t.index.Sync(); err != nil {
return err
}
- return t.content.Sync()
+ return t.head.Sync()
}
diff --git a/core/rawdb/freezer_table_test.go b/core/rawdb/freezer_table_test.go
new file mode 100644
index 000000000..d6ce6e93c
--- /dev/null
+++ b/core/rawdb/freezer_table_test.go
@@ -0,0 +1,511 @@
+// 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 <http://www.gnu.org/licenses/>.
+
+package rawdb
+
+import (
+ "bytes"
+ "fmt"
+ "github.com/ethereum/go-ethereum/metrics"
+ "math/rand"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+)
+
+func init() {
+ rand.Seed(time.Now().Unix())
+}
+
+// Gets a chunk of data, filled with 'b'
+func getChunk(size int, b byte) []byte {
+ data := make([]byte, size)
+ for i, _ := range data {
+ data[i] = b
+ }
+ return data
+}
+
+func print(t *testing.T, f *freezerTable, item uint64) {
+ a, err := f.Retrieve(item)
+ if err != nil {
+ t.Fatal(err)
+ }
+ fmt.Printf("db[%d] = %x\n", item, a)
+}
+
+// TestFreezerBasics test initializing a freezertable from scratch, writing to the table,
+// and reading it back.
+func TestFreezerBasics(t *testing.T) {
+ t.Parallel()
+ // set cutoff at 50 bytes
+ f, err := newCustomTable(os.TempDir(),
+ fmt.Sprintf("unittest-%d", rand.Uint64()),
+ metrics.NewMeter(), metrics.NewMeter(), 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ // Write 15 bytes 255 times, results in 85 files
+ for x := byte(0); x < 255; x++ {
+ data := getChunk(15, x)
+ f.Append(uint64(x), data)
+ }
+
+ //print(t, f, 0)
+ //print(t, f, 1)
+ //print(t, f, 2)
+ //
+ //db[0] = 000000000000000000000000000000
+ //db[1] = 010101010101010101010101010101
+ //db[2] = 020202020202020202020202020202
+
+ for y := byte(0); y < 255; y++ {
+ exp := getChunk(15, y)
+ got, err := f.Retrieve(uint64(y))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(got, exp) {
+ t.Fatalf("test %d, got \n%x != \n%x", y, got, exp)
+ }
+ }
+}
+
+// TestFreezerBasicsClosing tests same as TestFreezerBasics, but also closes and reopens the freezer between
+// every operation
+func TestFreezerBasicsClosing(t *testing.T) {
+ t.Parallel()
+ // set cutoff at 50 bytes
+ var (
+ fname = fmt.Sprintf("basics-close-%d", rand.Uint64())
+ m1, m2 = metrics.NewMeter(), metrics.NewMeter()
+ f *freezerTable
+ err error
+ )
+ f, err = newCustomTable(os.TempDir(), fname, m1, m2, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Write 15 bytes 255 times, results in 85 files
+ for x := byte(0); x < 255; x++ {
+ data := getChunk(15, x)
+ f.Append(uint64(x), data)
+ f.Close()
+ f, err = newCustomTable(os.TempDir(), fname, m1, m2, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ defer f.Close()
+
+ for y := byte(0); y < 255; y++ {
+ exp := getChunk(15, y)
+ got, err := f.Retrieve(uint64(y))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(got, exp) {
+ t.Fatalf("test %d, got \n%x != \n%x", y, got, exp)
+ }
+ f.Close()
+ f, err = newCustomTable(os.TempDir(), fname, m1, m2, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// TestFreezerRepairDanglingHead tests that we can recover if index entries are removed
+func TestFreezerRepairDanglingHead(t *testing.T) {
+ t.Parallel()
+ wm, rm := metrics.NewMeter(), metrics.NewMeter()
+ fname := fmt.Sprintf("dangling_headtest-%d", rand.Uint64())
+
+ { // Fill table
+ f, err := newCustomTable(os.TempDir(), fname, rm, wm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Write 15 bytes 255 times
+ for x := byte(0); x < 0xff; x++ {
+ data := getChunk(15, x)
+ f.Append(uint64(x), data)
+ }
+ // The last item should be there
+ if _, err = f.Retrieve(0xfe); err != nil {
+ t.Fatal(err)
+ }
+ f.Close()
+ }
+ // open the index
+ idxFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s.ridx", fname)), os.O_RDWR, 0644)
+ if err != nil {
+ t.Fatalf("Failed to open index file: %v", err)
+ }
+ // Remove 4 bytes
+ stat, err := idxFile.Stat()
+ if err != nil {
+ t.Fatalf("Failed to stat index file: %v", err)
+ }
+ idxFile.Truncate(stat.Size() - 4)
+ idxFile.Close()
+ // Now open it again
+ {
+ f, err := newCustomTable(os.TempDir(), fname, rm, wm, 50, true)
+ // The last item should be missing
+ if _, err = f.Retrieve(0xff); err == nil {
+ t.Errorf("Expected error for missing index entry")
+ }
+ // The one before should still be there
+ if _, err = f.Retrieve(0xfd); err != nil {
+ t.Fatalf("Expected no error, got %v", err)
+ }
+ }
+}
+
+// TestFreezerRepairDanglingHeadLarge tests that we can recover if very many index entries are removed
+func TestFreezerRepairDanglingHeadLarge(t *testing.T) {
+ t.Parallel()
+ wm, rm := metrics.NewMeter(), metrics.NewMeter()
+ fname := fmt.Sprintf("dangling_headtest-%d", rand.Uint64())
+
+ { // Fill a table and close it
+ f, err := newCustomTable(os.TempDir(), fname, wm, rm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Write 15 bytes 255 times
+ for x := byte(0); x < 0xff; x++ {
+ data := getChunk(15, x)
+ f.Append(uint64(x), data)
+ }
+ // The last item should be there
+ if _, err = f.Retrieve(f.items - 1); err == nil {
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ f.Close()
+ }
+ // open the index
+ idxFile, err := os.OpenFile(filepath.Join(os.TempDir(), fmt.Sprintf("%s.ridx", fname)), os.O_RDWR, 0644)
+ if err != nil {
+ t.Fatalf("Failed to open index file: %v", err)
+ }
+ // Remove everything but the first item, and leave data unaligned
+ // 0-indexEntry, 1-indexEntry, corrupt-indexEntry
+ idxFile.Truncate(indexEntrySize + indexEntrySize + indexEntrySize/2)
+ idxFile.Close()
+ // Now open it again
+ {
+ f, err := newCustomTable(os.TempDir(), fname, rm, wm, 50, true)
+ // The first item should be there
+ if _, err = f.Retrieve(0); err != nil {
+ t.Fatal(err)
+ }
+ // The second item should be missing
+ if _, err = f.Retrieve(1); err == nil {
+ t.Errorf("Expected error for missing index entry")
+ }
+ // We should now be able to store items again, from item = 1
+ for x := byte(1); x < 0xff; x++ {
+ data := getChunk(15, ^x)
+ f.Append(uint64(x), data)
+ }
+ f.Close()
+ }
+ // And if we open it, we should now be able to read all of them (new values)
+ {
+ f, _ := newCustomTable(os.TempDir(), fname, rm, wm, 50, true)
+ for y := byte(1); y < 255; y++ {
+ exp := getChunk(15, ^y)
+ got, err := f.Retrieve(uint64(y))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(got, exp) {
+ t.Fatalf("test %d, got \n%x != \n%x", y, got, exp)
+ }
+ }
+ }
+}
+
+// TestSnappyDetection tests that we fail to open a snappy database and vice versa
+func TestSnappyDetection(t *testing.T) {
+ t.Parallel()
+ wm, rm := metrics.NewMeter(), metrics.NewMeter()
+ fname := fmt.Sprintf("snappytest-%d", rand.Uint64())
+ // Open with snappy
+ {
+ f, err := newCustomTable(os.TempDir(), fname, wm, rm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Write 15 bytes 255 times
+ for x := byte(0); x < 0xff; x++ {
+ data := getChunk(15, x)
+ f.Append(uint64(x), data)
+ }
+ f.Close()
+ }
+ // Open without snappy
+ {
+ f, err := newCustomTable(os.TempDir(), fname, wm, rm, 50, false)
+ if _, err = f.Retrieve(0); err == nil {
+ f.Close()
+ t.Fatalf("expected empty table")
+ }
+ }
+
+ // Open with snappy
+ {
+ f, err := newCustomTable(os.TempDir(), fname, wm, rm, 50, true)
+ // There should be 255 items
+ if _, err = f.Retrieve(0xfe); err != nil {
+ f.Close()
+ t.Fatalf("expected no error, got %v", err)
+ }
+ }
+
+}
+func assertFileSize(f string, size int64) error {
+ stat, err := os.Stat(f)
+ if err != nil {
+ return err
+ }
+ if stat.Size() != size {
+ return fmt.Errorf("error, expected size %d, got %d", size, stat.Size())
+ }
+ return nil
+
+}
+
+// TestFreezerRepairDanglingIndex checks that if the index has more entries than there are data,
+// the index is repaired
+func TestFreezerRepairDanglingIndex(t *testing.T) {
+ t.Parallel()
+ wm, rm := metrics.NewMeter(), metrics.NewMeter()
+ fname := fmt.Sprintf("dangling_indextest-%d", rand.Uint64())
+
+ { // Fill a table and close it
+ f, err := newCustomTable(os.TempDir(), fname, wm, rm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Write 15 bytes 9 times : 150 bytes
+ for x := byte(0); x < 9; x++ {
+ data := getChunk(15, x)
+ f.Append(uint64(x), data)
+ }
+ // The last item should be there
+ if _, err = f.Retrieve(f.items - 1); err != nil {
+ f.Close()
+ t.Fatal(err)
+ }
+ f.Close()
+ // File sizes should be 45, 45, 45 : items[3, 3, 3)
+ }
+ // Crop third file
+ fileToCrop := filepath.Join(os.TempDir(), fmt.Sprintf("%s.2.rdat", fname))
+ // Truncate third file: 45 ,45, 20
+ {
+ if err := assertFileSize(fileToCrop, 45); err != nil {
+ t.Fatal(err)
+ }
+ file, err := os.OpenFile(fileToCrop, os.O_RDWR, 0644)
+ if err != nil {
+ t.Fatal(err)
+ }
+ file.Truncate(20)
+ file.Close()
+ }
+ // Open db it again
+ // It should restore the file(s) to
+ // 45, 45, 15
+ // with 3+3+1 items
+ {
+ f, err := newCustomTable(os.TempDir(), fname, wm, rm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if f.items != 7 {
+ f.Close()
+ t.Fatalf("expected %d items, got %d", 7, f.items)
+ }
+ if err := assertFileSize(fileToCrop, 15); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+func TestFreezerTruncate(t *testing.T) {
+
+ t.Parallel()
+ wm, rm := metrics.NewMeter(), metrics.NewMeter()
+ fname := fmt.Sprintf("truncation-%d", rand.Uint64())
+
+ { // Fill table
+ f, err := newCustomTable(os.TempDir(), fname, rm, wm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Write 15 bytes 30 times
+ for x := byte(0); x < 30; x++ {
+ data := getChunk(15, x)
+ f.Append(uint64(x), data)
+ }
+ // The last item should be there
+ if _, err = f.Retrieve(f.items - 1); err != nil {
+ t.Fatal(err)
+ }
+ f.Close()
+ }
+ // Reopen, truncate
+ {
+ f, err := newCustomTable(os.TempDir(), fname, rm, wm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ f.truncate(10) // 150 bytes
+ if f.items != 10 {
+ t.Fatalf("expected %d items, got %d", 10, f.items)
+ }
+ // 45, 45, 45, 15 -- bytes should be 15
+ if f.headBytes != 15 {
+ t.Fatalf("expected %d bytes, got %d", 15, f.headBytes)
+ }
+
+ }
+
+}
+
+// TestFreezerRepairFirstFile tests a head file with the very first item only half-written.
+// That will rewind the index, and _should_ truncate the head file
+func TestFreezerRepairFirstFile(t *testing.T) {
+ t.Parallel()
+ wm, rm := metrics.NewMeter(), metrics.NewMeter()
+ fname := fmt.Sprintf("truncationfirst-%d", rand.Uint64())
+ { // Fill table
+ f, err := newCustomTable(os.TempDir(), fname, rm, wm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Write 80 bytes, splitting out into two files
+ f.Append(0, getChunk(40, 0xFF))
+ f.Append(1, getChunk(40, 0xEE))
+ // The last item should be there
+ if _, err = f.Retrieve(f.items - 1); err != nil {
+ t.Fatal(err)
+ }
+ f.Close()
+ }
+ // Truncate the file in half
+ fileToCrop := filepath.Join(os.TempDir(), fmt.Sprintf("%s.1.rdat", fname))
+ {
+ if err := assertFileSize(fileToCrop, 40); err != nil {
+ t.Fatal(err)
+ }
+ file, err := os.OpenFile(fileToCrop, os.O_RDWR, 0644)
+ if err != nil {
+ t.Fatal(err)
+ }
+ file.Truncate(20)
+ file.Close()
+ }
+ // Reopen
+ {
+ f, err := newCustomTable(os.TempDir(), fname, wm, rm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if f.items != 1 {
+ f.Close()
+ t.Fatalf("expected %d items, got %d", 0, f.items)
+ }
+ // Write 40 bytes
+ f.Append(1, getChunk(40, 0xDD))
+ f.Close()
+ // Should have been truncated down to zero and then 40 written
+ if err := assertFileSize(fileToCrop, 40); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// TestFreezerReadAndTruncate tests:
+// - we have a table open
+// - do some reads, so files are open in readonly
+// - truncate so those files are 'removed'
+// - check that we did not keep the rdonly file descriptors
+func TestFreezerReadAndTruncate(t *testing.T) {
+ t.Parallel()
+ wm, rm := metrics.NewMeter(), metrics.NewMeter()
+ fname := fmt.Sprintf("read_truncate-%d", rand.Uint64())
+ { // Fill table
+ f, err := newCustomTable(os.TempDir(), fname, rm, wm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Write 15 bytes 30 times
+ for x := byte(0); x < 30; x++ {
+ data := getChunk(15, x)
+ f.Append(uint64(x), data)
+ }
+ // The last item should be there
+ if _, err = f.Retrieve(f.items - 1); err != nil {
+ t.Fatal(err)
+ }
+ f.Close()
+ }
+ // Reopen and read all files
+ {
+ f, err := newCustomTable(os.TempDir(), fname, wm, rm, 50, true)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if f.items != 30 {
+ f.Close()
+ t.Fatalf("expected %d items, got %d", 0, f.items)
+ }
+ for y := byte(0); y < 30; y++ {
+ f.Retrieve(uint64(y))
+ }
+ // Now, truncate back to zero
+ f.truncate(0)
+ // Write the data again
+ for x := byte(0); x < 30; x++ {
+ data := getChunk(15, ^x)
+ if err := f.Append(uint64(x), data); err != nil {
+ t.Fatalf("error %v", err)
+ }
+ }
+ f.Close()
+ }
+}
+
+// TODO (?)
+// - test that if we remove several head-files, aswell as data last data-file,
+// the index is truncated accordingly
+// Right now, the freezer would fail on these conditions:
+// 1. have data files d0, d1, d2, d3
+// 2. remove d2,d3
+//
+// However, all 'normal' failure modes arising due to failing to sync() or save a file should be
+// handled already, and the case described above can only (?) happen if an external process/user
+// deletes files from the filesystem.
--
GitLab