scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Refactor RBF to use a single WAL file

Browse source
This commit is contained in:
Ben Johnson 2020-11-05 15:26:24 -07:00
parent e99f7f0fc3
commit 3554048877
10 changed files with 185 additions and 829 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
rbf/cfg/cfg.go
View file

@ -27,6 +27,9 @@ type Config struct {
// The maximum allowed database size. Required by mmap.
MaxSize int64
// The maximum allowed WAL size. Required by mmap.
MaxWALSize int64
// Set before calling db.Open()
FsyncEnabled bool
@ -38,26 +41,27 @@ type Config struct {
// after CheckpointEveryDur since the previous.
CheckpointEveryDur time.Duration
// Maximum size of a single WAL segment.
// May exceed by one page if last page is a bitmap header + bitmap.
MaxWALSegmentFileSize int
// Maximum size of a WAL write cache.
MaxWALWriteCacheSize int
}
func NewDefaultConfig() *Config {
return &Config{
MaxSize: DefaultMaxSize,
FsyncEnabled: true,
CheckpointEveryDur: time.Millisecond,
MaxWALSegmentFileSize: 1 << 20,
MaxSize: DefaultMaxSize,
MaxWALSize: DefaultMaxWALSize,
FsyncEnabled: true,
CheckpointEveryDur: time.Millisecond,
MaxWALWriteCacheSize: 1 << 20,
}
}
func (cfg *Config) DefineFlags(flags *pflag.FlagSet) {
default0 := NewDefaultConfig()
flags.IntVar(&cfg.MaxWALSegmentFileSize, "rbf-max-wal", default0.MaxWALSegmentFileSize, "RBF write-Ahead-Log file size in bytes")
flags.IntVar(&cfg.MaxWALWriteCacheSize, "rbf-max-write-cache-size", default0.MaxWALWriteCacheSize, "RBF write cache size, in bytes")
flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", default0.CheckpointEveryDur,
"RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.")
flags.Int64Var(&cfg.MaxSize, "rbf-max-db-size", default0.MaxSize, "RBF maximum size in bytes of a database file (distinct from a WAL file)")
flags.Int64Var(&cfg.MaxWALSize, "rbf-max-wal-size", default0.MaxWALSize, "RBF maximum size in bytes of a WAL file (distinct from a DB file)")
// renamed from --rbf-fsync to just --fsync because now it applies to all Tx backends.
flags.BoolVar(&cfg.FsyncEnabled, "fsync", default0.FsyncEnabled, "enable fsync fully safe flush-to-disk")

5
rbf/cfg/os.go
View file

@ -20,3 +20,8 @@ package cfg
// size of the database. The size can be increased by updating the DB.MaxSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxSize = 4 * (1 << 30)
// DefaultMaxWALSize is the default mmap size and therefore the maximum allowed
// size of the WAL. The size can be increased by updating the DB.MaxWALSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxWALSize = 2 * (1 << 30)

5
rbf/cfg/os_386.go
View file

@ -18,3 +18,8 @@ package cfg
// size of the database. The size can be increased by updating the DB.MaxSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxSize = 256 * (1 << 20) // 256MB
// DefaultMaxWALSize is the default mmap size and therefore the maximum allowed
// size of the WAL. The size can be increased by updating the DB.MaxWALSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxWALSize = 64 * (1 << 20) // 64MB

1
rbf/cursor_test.go
View file

@ -787,7 +787,6 @@ func TestCursor_RLEConversion(t *testing.T) {
}()...)
if err != nil {
t.Fatalf("ERR adding bits: %v\n", err)
}
if err := c.First(); err != nil {

276
rbf/db.go
View file

@ -18,7 +18,6 @@ import (
"errors"
"fmt"
"io"
"io/ioutil"
"os"
"path/filepath"
"sync"
@ -40,15 +39,17 @@ var (
type DB struct {
cfg rbfcfg.Config
data []byte // mmap data
file *os.File // file descriptor
data []byte // database mmap
file *os.File // database file descriptor
rootRecords []*RootRecord // cached root records
pageMap *immutable.Map // pgno-to-WALID mapping
txs map[*Tx]struct{} // active transactions
opened bool // true if open
wcache []byte // wal write cache
segments []WALSegment // write-ahead log
wal []byte // wal mmap
walFile *os.File // wal file descriptor
walPageN int // wal page count
wcache []byte // wal write cache
mu sync.RWMutex // general mutex
rwmu sync.Mutex // mutex for restricting single writer
@ -70,7 +71,7 @@ func NewDB(path string, cfg *rbfcfg.Config) *DB {
cfg: *cfg,
txs: make(map[*Tx]struct{}),
pageMap: immutable.NewMap(&uint32Hasher{}),
wcache: make([]byte, cfg.MaxWALSegmentFileSize+PageSize),
wcache: make([]byte, 0, cfg.MaxWALWriteCacheSize),
Path: path,
}
return db
@ -81,7 +82,7 @@ func (db *DB) DataPath() string {
return filepath.Join(db.Path, "data")
}
// WALPath returns the path to the WAL directory.
// WALPath returns the path to the WAL file.
func (db *DB) WALPath() string {
return filepath.Join(db.Path, "wal")
}
@ -115,7 +116,7 @@ func (db *DB) Open() (err error) {
return fmt.Errorf("open file: %w", err)
}
// Open read-only mmap.
// Open read-only database mmap.
if f, err := os.OpenFile(db.DataPath(), os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open mmap file: %w", err)
} else if db.data, err = syswrap.Mmap(int(f.Fd()), 0, int(db.cfg.MaxSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
@ -136,15 +137,10 @@ func (db *DB) Open() (err error) {
// TODO(BBJ): Obtain advisory lock on file.
// Ensure WAL directory exists.
if err := os.MkdirAll(db.WALPath(), 0777); err != nil {
return fmt.Errorf("create wal dir: %w", err)
}
db.opened = true
// Open write-ahead log & checkpoint to the end since no transactions are open.
if err := db.openWALSegments(); err != nil {
if err := db.openWAL(); err != nil {
return fmt.Errorf("wal open: %w", err)
} else if err := db.checkpoint(true); err != nil {
return fmt.Errorf("checkpoint: %w", err)
@ -153,58 +149,49 @@ func (db *DB) Open() (err error) {
return nil
}
func (db *DB) openWALSegments() error {
fis, err := ioutil.ReadDir(db.WALPath())
if err != nil {
return fmt.Errorf("read dir: %w", err)
func (db *DB) openWAL() (err error) {
// Open WAL file writer.
if db.walFile, err = os.OpenFile(db.WALPath(), os.O_WRONLY|os.O_CREATE, 0666); err != nil {
return fmt.Errorf("open wal file: %w", err)
}
// Open all WAL segments.
for _, fi := range fis {
if filepath.Ext(fi.Name()) != ".wal" {
continue
}
// Open read-only mmap.
if f, err := os.OpenFile(db.WALPath(), os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open wal mmap file: %w", err)
} else if db.wal, err = syswrap.Mmap(int(f.Fd()), 0, int(db.cfg.MaxWALSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
return fmt.Errorf("open wal mmap file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("cannot close wal mmap file: %w", err)
}
segment := db.NewWALSegment(filepath.Join(db.WALPath(), fi.Name()))
if err := segment.Open(); err != nil {
_ = db.closeWALSegments()
// Determine the number of whole pages in the WAL.
var pageN int
if fi, err := db.walFile.Stat(); err != nil {
return fmt.Errorf("wal stat: %w", err)
} else {
pageN = int(fi.Size() / PageSize)
}
// Read backwards through the WAL to find the last valid meta page.
for ; pageN > 0; pageN-- {
if page, err := db.readWALPageAt(pageN - 1); err != nil {
return err
} else if IsMetaPage(page) {
break
}
db.segments = append(db.segments, segment)
}
// Truncate everything after the last successful meta page.
if walID, err := findLastWALMetaPage(db.segments); err != nil {
return err
} else if db.segments, err = db.truncateWALAfter(db.segments, walID); err != nil {
return err
// Truncate WAL to the last valid meta page.
if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil {
return fmt.Errorf("wal truncate: %w", err)
}
db.walPageN = pageN
return nil
}
// updateWALSegment updates or adds a segment.
func (db *DB) updateWALSegment(s WALSegment) {
segments := make([]WALSegment, len(db.segments), len(db.segments)+1)
copy(segments, db.segments)
// Find the matching segment using the path.
segment := walSegmentByPath(segments, s.Path)
// Update existing segment if it already exists.
// Otherwise append segment to the end.
if segment != nil {
*segment = s
} else {
assert(len(segments) == 0 || segments[len(segments)-1].MinWALID < s.MinWALID)
segments = append(segments, s)
}
// Replace DB segment list.
db.segments = segments
}
// checkpoint moves WAL segments to the main DB file.
// checkpoint moves all WAL pages to the main DB file.
// Must be called by a write transaction while under db.mu lock.
func (db *DB) checkpoint(exclusive bool) error {
if !db.opened {
@ -213,105 +200,43 @@ func (db *DB) checkpoint(exclusive bool) error {
return nil // skip if transactions open
}
// Determine last checkpointed WAL ID.
page, err := db.readDBPage(0)
if err != nil {
return err
}
walID := readMetaWALID(page)
// INVAR: walID represents everything already in the DB, and
// any wal page k > walID is in the WAL not the DB.
// Loop over each transaction
// We could be looking at a recovery. When there is no new
// meta page further down in the WAL, then there was power
// failure or the process was killed. So we have to search
// and find the next meta page, if present.
//
// Read ahead to the next meta page, if present, in the WAL. If we
// we find it, then we must ensure the pages between
// [walID, next_meta_page.walID] are committed.
// If there is NOT another meta page after, then those writes get
// rolled back.
walID++
for {
// Determine last page of transaction.
metaWALID, err := findNextWALMetaPage(db.segments, walID)
if err == ErrNoMetaFound {
break
} else if err != nil {
for i := 0; i < db.walPageN; i++ {
page, err := db.readWALPageAt(i)
if err != nil {
return err
}
// Loop over pages in the transaction.
for ; walID <= metaWALID; walID++ {
page, err := readWALPage(db.segments, walID)
if err != nil {
// Determine page number. Meta pages are always on zero & bitmap
// headers specify the page number of the next page in the WAL.
// All other pages have their page number in the page data.
var pgno uint32
if IsBitmapHeader(page) {
pgno = readPageNo(page)
if page, err = db.readWALPageAt(i + 1); err != nil {
return err
}
isBitmapHeader := IsBitmapHeader(page)
i++ // bitmaps in WAL are two pages
} else if !IsMetaPage(page) {
pgno = readPageNo(page)
}
// Determine page number. Meta pages are always on zero & bitmap
// headers specify the page number of the next page in the WAL.
// All other pages have their page number in the page data.
var pgno uint32
if isBitmapHeader {
pgno, walID = readPageNo(page), walID+1 // skip next page
} else if !IsMetaPage(page) {
pgno = readPageNo(page)
}
// Ensure we actually read the bitmap data in when we checkpoint.
// NOTE: The walID variable is incremented above in the pgno check.
if isBitmapHeader {
if page, err = readWALPage(db.segments, walID); err != nil {
return err
}
}
// TODO: address this problem: if we write a WAL meta page to database page 0 before fsyncing
// the transactions updates from the WAL into the DB, then (upon
// power failure in the middle of a fsync), the meta page might
// get updated before all of the databases pages that included the changes
// that the meta page represents. The only way to have a strict
// ordering that the meta page is updated only after the other
// pages is to fsync it in a 2nd fsync that follows the
// the first. SSDs and HDs both exhibit these "unsynchronized writes".
// reference https://www.usenix.org/system/files/conference/fast13/fast13-final80.pdf
//
// needed pattern:
// 1) write tx-content pages;
// 2) fsync the tx-content pages;
// 3) write meta page;
// 4) fsync the meta page.
// The OS can also be inserting fsyncs at any point (e.g. due to memory pressure)
// and so we have to be certain that the meta page is written after a separate fsync.
// Write page data into main db file.
if err := db.writeDBPage(pgno, page); err != nil {
return err
}
// Write data to the data file.
if err := db.writeDBPage(pgno, page); err != nil {
return err
}
}
// Ensure WAL pages are fully copied & synced to DB file.
// Ensure database file is synced and then truncate the WAL file.
if err := db.fsync(db.file); err != nil {
return fmt.Errorf("db file sync: %w", err)
} else if err := db.walFile.Truncate(0); err != nil {
return fmt.Errorf("truncate wal file: %w", err)
} else if err := db.fsync(db.walFile); err != nil {
return fmt.Errorf("wal file sync: %w", err)
}
db.walPageN = 0
db.pageMap = immutable.NewMap(&uint32Hasher{})
// Remove WAL segments that have been checkpointed.
for _, segment := range db.segments {
if err := segment.Close(); err != nil {
return err
} else if err := os.Remove(segment.Path); err != nil {
return err
}
}
db.segments = nil
return nil
}
@ -344,21 +269,22 @@ func (db *DB) Close() (err error) {
db.file = nil
}
if e := db.closeWALSegments(); e != nil && err == nil {
err = e
}
return err
}
// closeWALSegments closes the WAL and all its segments.
func (db *DB) closeWALSegments() (err error) {
for _, s := range db.segments {
if e := s.Close(); e != nil && err == nil {
// Close WAL mmap handle.
if db.wal != nil {
if e := syswrap.Munmap(db.wal); e != nil && err == nil {
err = e
}
db.wal = nil
}
db.segments = nil
// Close wal writer handler.
if db.walFile != nil {
if e := db.walFile.Close(); e != nil && err == nil {
err = e
}
db.walFile = nil
}
return err
}
@ -431,23 +357,18 @@ func (db *DB) Size() (int64, error) {
if err != nil {
return 0, err
}
return walSize(db.segments) + fi.Size(), nil
return db.walSize() + fi.Size(), nil
}
// WALSize returns the size of all WAL segments, in bytes.
// WALSize returns the size of the WAL, in bytes.
func (db *DB) WALSize() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return walSize(db.segments)
return db.walSize()
}
// WALSegments returns the WAL segments currently on the DB.
func (db *DB) WALSegments() []WALSegment {
db.mu.RLock()
defer db.mu.RUnlock()
other := make([]WALSegment, len(db.segments))
copy(other, db.segments)
return other
func (db *DB) walSize() int64 {
return int64(db.walPageN * PageSize)
}
// init initializes a new database file.
@ -568,23 +489,17 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
db: db,
rootRecords: db.rootRecords,
pageMap: db.pageMap,
walPageN: db.walPageN,
writable: writable,
exclusive: exclusive,
DeleteEmptyContainer: true,
}
if writable {
tx.wcache = db.wcache[:0]
}
// Copy list of WAL segments so they can be altered by the tx.
// Add last segment to the list of segments that will be updated/added.
if len(db.segments) != 0 {
tx.segments = make([]WALSegment, len(db.segments))
copy(tx.segments, db.segments)
tx.updatedSegmentPaths = []string{tx.segments[len(tx.segments)-1].Path}
}
// Copy meta page into transaction's buffer.
// This page is only written at the end of a dirty transaction.
page, err := db.readMetaPage()
@ -628,10 +543,7 @@ func (db *DB) removeTx(tx *Tx) error {
// Write pages from WAL to DB.
// TODO(bbj): Move this to an async goroutine.
// TODO(jea): Make the time-based checkpointing work at all, and update the
// comment in cfg/cfg.go for CheckpointEveryDur. Seems that
// wal.go readWALPage() can receive a request for a walID that
// comes before the segments it is passed if we do not
// checkpoint eagerly.
// comment in cfg/cfg.go for CheckpointEveryDur.
if tx.writable {
if db.cfg.CheckpointEveryDur == 0 || time.Since(db.lastCheckpoint) > db.cfg.CheckpointEveryDur {
if err := db.checkpoint(false); err != nil {
@ -669,9 +581,25 @@ func (db *DB) readDBPage(pgno uint32) ([]byte, error) {
return db.data[offset : offset+PageSize], nil
}
// baseWALID returns the WAL ID stored in the database file meta page.
func (db *DB) baseWALID() int64 {
return readMetaWALID(db.data)
}
// readWALPageByID reads a WAL page by WAL ID.
func (db *DB) readWALPageByID(id int64) ([]byte, error) {
return db.readWALPageAt(int(id - db.baseWALID() - 1))
}
// readWALPageAt reads the i-th page in the WAL file.
func (db *DB) readWALPageAt(i int) ([]byte, error) {
offset := int64(i) * PageSize
return db.wal[offset : offset+PageSize], nil
}
func (db *DB) readMetaPage() ([]byte, error) {
if walID, ok := db.pageMap.Get(uint32(0)); ok {
return readWALPage(db.segments, walID.(int64))
return db.readWALPageByID(walID.(int64))
}
return db.readDBPage(0)
}

44
rbf/db_test.go
View file

@ -39,45 +39,6 @@ func TestDB_Open(t *testing.T) {
}
}
/* optimization of wal size means there may certainly be more than 2 WAL segments.
func TestDB_Checkpoint(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
}
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Create bitmap.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
// Create a bunch of transactions to generate WAL segments.
rand := rand.New(rand.NewSource(0))
for i := 0; i < 1000; i++ {
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", rand.Uint64()); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", rand.Uint64()); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}
// Ensure there is no more than two WAL segments.
if n := len(db.WALSegments()); n > 2 {
t.Fatalf("expected two or fewer WAL segments, got %d", n)
}
}
*/
func TestDB_Recovery(t *testing.T) {
// Ensure a bitmap header written without a bitmap is truncated.
t.Run("TruncPartialWALBitmap", func(t *testing.T) {
@ -121,11 +82,10 @@ func TestDB_Recovery(t *testing.T) {
tx1.Rollback()
// Close database & truncate WAL to remove commit page & bitmap data page.
segments := db.WALSegments()
segment := segments[len(segments)-1]
walPath, walSize := db.WALPath(), db.WALSize()
if err := db.Close(); err != nil {
t.Fatal(err)
} else if err := os.Truncate(segment.Path, segment.Size()-(2*rbf.PageSize)); err != nil {
} else if err := os.Truncate(walPath, walSize-(2*rbf.PageSize)); err != nil {
t.Fatal(err)
}

24
rbf/rbf.go
View file

@ -694,3 +694,27 @@ func (db *DB) fsync(f *os.File) error {
}
return f.Sync()
}
// uint32Hasher implements Hasher for uint32 keys.
type uint32Hasher struct{}
// Hash returns a hash for key.
func (h *uint32Hasher) Hash(key interface{}) uint32 {
return hashUint64(uint64(key.(uint32)))
}
// Equal returns true if a is equal to b. Otherwise returns false.
// Panics if a and b are not ints.
func (h *uint32Hasher) Equal(a, b interface{}) bool {
return a.(uint32) == b.(uint32)
}
// hashUint64 returns a 32-bit hash for a 64-bit value.
func hashUint64(value uint64) uint32 {
hash := value
for value > 0xffffffff {
value /= 0xffffffff
hash ^= value
}
return uint32(hash)
}

124
rbf/tx.go
View file

@ -17,8 +17,6 @@ import (
"fmt"
"io"
"math"
"os"
"path/filepath"
"sort"
"strings"
"sync"
@ -33,13 +31,13 @@ var _ = txkey.ToString
// Tx represents a transaction.
type Tx struct {
mu sync.RWMutex
db *DB // parent db
segments []WALSegment // copy of WAL segments
updatedSegmentPaths []string // updated or added segment paths
meta [PageSize]byte // copy of current meta page
walID int64 // max WAL ID at start of tx
rootRecords []*RootRecord // read-only cache of root records
mu sync.RWMutex
db *DB // parent db
meta [PageSize]byte // copy of current meta page
walID int64 // max WAL ID at start of tx
walPageN int // wal page count
wcache []byte // write cache
rootRecords []*RootRecord // read-only cache of root records
// pageMap holds WAL pages that have not yet been transferred
// into the database pages. So it can be empty, if the whole previous
@ -49,8 +47,6 @@ type Tx struct {
exclusive bool // if true, tx writes directly to db file (no wal)
dirty bool // if true, changes have been made
wcache []byte // write cache
// If Rollback() has already completed, don't do it again.
// Note db == nil means that commit has already been done.
rollbackDone bool
@ -91,8 +87,9 @@ func (tx *Tx) Commit() error {
return err
} else if err := tx.flushWALWriter(); err != nil {
return err
} else if err := tx.db.fsync(tx.db.walFile); err != nil {
return fmt.Errorf("sync wal: %w", err)
}
// future plan: after checkpoint is moved to background
// or not every removeTx, then we can move the
// tx.db.rootRecords = tx.rootRecords into removeTx().
@ -102,12 +99,7 @@ func (tx *Tx) Commit() error {
tx.db.mu.Lock()
tx.db.rootRecords = tx.rootRecords
tx.db.pageMap = tx.pageMap
for _, path := range tx.updatedSegmentPaths {
segment := walSegmentByPath(tx.segments, path)
assert(segment != nil)
//lint:ignore SA5011 the assert above prevents this from being nil
tx.db.updateWALSegment(*segment) //nolint:staticcheck
}
tx.db.walPageN = tx.walPageN
tx.db.mu.Unlock()
}
@ -131,12 +123,15 @@ func (tx *Tx) Rollback() {
// TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen?
// Remove any writes to the WAL from this transaction.
if tx.dirty {
if _, err := tx.db.truncateWALAfter(tx.segments, tx.walID); err != nil {
panicOn(err)
}
tx.segments = nil
tx.updatedSegmentPaths = nil
func() {
tx.db.mu.Lock()
defer tx.db.mu.Unlock()
if err := tx.db.walFile.Truncate(int64(tx.db.walPageN * PageSize)); err != nil {
panicOn(fmt.Errorf("rollback truncate: %w", err))
}
}()
}
// Disconnect transaction from DB.
@ -955,14 +950,14 @@ func (tx *Tx) readPage(pgno uint32) ([]byte, error) {
walID64 := walID.(int64)
// Read from write cache if not yet flushed to disk.
maxWALID := activeWALSegment(tx.segments).MaxWALID()
maxWALID := tx.db.baseWALID() + int64(tx.walPageN)
if walID64 > maxWALID {
offset := (walID64 - maxWALID - 1) * PageSize
return tx.wcache[offset : offset+PageSize], nil
}
// Otherwise return remapped page from WAL segment.
return readWALPage(tx.segments, walID64)
// Otherwise return remapped page from WAL.
return tx.db.readWALPageByID(walID64)
}
return tx.db.readDBPage(pgno)
@ -1615,37 +1610,25 @@ func (tx *Tx) flushWALWriter() error {
return nil
}
// Determine active WAL segment.
assert(len(tx.segments) != 0)
segment := &tx.segments[len(tx.segments)-1]
// Open write handle to active segment.
w, err := os.OpenFile(segment.Path, os.O_WRONLY, 0666)
if err != nil {
return fmt.Errorf("open wal segment write handle: %w", err)
}
defer w.Close()
// Flush cache to writer.
if _, err := w.WriteAt(tx.wcache, int64(segment.PageN)*PageSize); err != nil {
return fmt.Errorf("write wal segment: %w", err)
} else if err := tx.db.fsync(w); err != nil {
return fmt.Errorf("sync wal segment: %w", err)
} else if err := w.Close(); err != nil {
return fmt.Errorf("close wal segment: %w", err)
if _, err := tx.db.walFile.WriteAt(tx.wcache, int64(tx.walPageN)*PageSize); err != nil {
return fmt.Errorf("write wal: %w", err)
}
// Increase the size of the last WAL segment & clear cache.
// Increase the size of the WAL & clear cache.
assert(len(tx.wcache)%PageSize == 0)
segment.PageN += len(tx.wcache) / PageSize
tx.walPageN += len(tx.wcache) / PageSize
tx.wcache = tx.wcache[:0]
return nil
}
func (tx *Tx) writeWALPage(page []byte, isMeta bool) (walID int64, err error) {
if err := tx.ensureWritableWALSegment(); err != nil {
return 0, err
// Flush WAL cache if there is not enough space to write the page.
if len(tx.wcache) == cap(tx.wcache) {
if err := tx.flushWALWriter(); err != nil {
return 0, err
}
}
// Determine next WAL ID from cached meta page.
@ -1661,8 +1644,11 @@ func (tx *Tx) writeWALPage(page []byte, isMeta bool) (walID int64, err error) {
}
func (tx *Tx) writeBitmapWALPage(pgno uint32, page []byte) (walID int64, err error) {
if err := tx.ensureWritableWALSegment(); err != nil {
return 0, err
// Flush WAL cache if there is not enough space to write the header & page.
if len(tx.wcache)+PageSize >= cap(tx.wcache) {
if err := tx.flushWALWriter(); err != nil {
return 0, err
}
}
// Write header page for next bitmap page.
@ -1678,46 +1664,6 @@ func (tx *Tx) writeBitmapWALPage(pgno uint32, page []byte) (walID int64, err err
return tx.writeWALPage(page, false)
}
func (tx *Tx) ensureWritableWALSegment() error {
// Ignore if we still have space in the write cache.
writeCacheSize := int64(len(tx.wcache))
if len(tx.segments) != 0 && activeWALSegment(tx.segments).Size()+writeCacheSize < int64(tx.db.cfg.MaxWALSegmentFileSize) {
return nil
}
// Flush write cache out to file before adding new segment.
if err := tx.flushWALWriter(); err != nil {
return err
}
// If we have a current active WAL segment then close it and start the
// next segment from the next WAL ID. If there is no existing WAL segments,
// read the last checkpointed WAL ID from the DB and start after that.
var base int64
if len(tx.segments) != 0 {
base = activeWALSegment(tx.segments).MaxWALID() + 1
} else {
page, err := tx.readPage(0)
if err != nil {
return err
}
base = readMetaWALID(page) + 1
}
// Create new segment file.
s := tx.db.NewWALSegment(filepath.Join(tx.db.WALPath(), FormatWALSegmentPath(base)))
if err := s.Open(); err != nil {
return fmt.Errorf("add wal segment: %w", err)
}
// Track all segments that need to be added back to DB.
// The DB can remove segments in the background so we don't want to replace.
tx.segments = append(tx.segments, s)
tx.updatedSegmentPaths = append(tx.updatedSegmentPaths, s.Path)
return nil
}
// Pages returns meta & record data for a list of pages.
func (tx *Tx) Pages(pgnos []uint32) ([]Page, error) {
// Read page info for all pages in the database.

329
rbf/wal.go
View file

@ -1,329 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package rbf
import (
"fmt"
"os"
"path/filepath"
"sort"
"syscall"
"github.com/pilosa/pilosa/v2/syswrap"
)
var _ = sort.Search
// WALSegment represents a single file in the WAL.
type WALSegment struct {
db *DB
Path string // path to file
MinWALID int64 // base WALID; calculated from path
PageN int // number of written pages
data []byte // read-only mmap data
}
// NewWALSegment returns a new instance of WALSegment for a given path.
func (db *DB) NewWALSegment(path string) WALSegment {
return WALSegment{
db: db,
Path: path,
}
}
// MaxWALID returns the maximum WAL ID of the segment. Only available after Open().
func (s WALSegment) MaxWALID() int64 {
return s.MinWALID + int64(s.PageN) - 1
}
// Size returns the current size of the segment, in bytes.
func (s WALSegment) Size() int64 {
return int64(s.PageN) * PageSize
}
func (s *WALSegment) Open() (err error) {
// Extract base WAL ID and validate path.
if s.MinWALID, err = ParseWALSegmentPath(s.Path); err != nil {
return err
}
// Determine file size & create if necessary.
var sz int64
if fi, err := os.Stat(s.Path); os.IsNotExist(err) {
if f, err := os.OpenFile(s.Path, os.O_RDWR|os.O_CREATE, 0666); err != nil {
return fmt.Errorf("touch wal segment file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("close touched wal segment file: %w", err)
}
} else if err != nil {
return fmt.Errorf("stat wal segment file: %w", err)
} else {
sz = fi.Size()
}
// Determine page count & truncate if a partial page is written.
s.PageN = int(sz / PageSize)
if sz%PageSize != 0 {
sz = int64(s.PageN * PageSize)
if err := s.db.truncate(s.Path, sz); err != nil {
return fmt.Errorf("truncate wal file: %w", err)
}
}
// Default the mmap size to the max size plus a page of padding for bitmap pages.
// If the actual size is larger, then increase to that size.
mmapSize := int64(s.db.cfg.MaxWALSegmentFileSize + PageSize)
if sz > mmapSize {
mmapSize = sz
}
// Open file as a read-only memory map.
if f, err := os.OpenFile(s.Path, os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open wal segment file: %w", err)
} else if s.data, err = syswrap.Mmap(int(f.Fd()), 0, int(mmapSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
return fmt.Errorf("mmap wal segment: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("close wal segment mmap file: %w", err)
}
return nil
}
// Close closes the write handle and the read-only mmap.
func (s *WALSegment) Close() error {
if s.data != nil {
if err := syswrap.Munmap(s.data); err != nil {
return err
}
s.data = nil
}
return nil
}
// ReadWALPage reads a single page at the given WAL ID.
func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
// Ensure requested ID is contained in this file.
if walID < s.MinWALID || walID > s.MinWALID+int64(s.PageN) {
return nil, fmt.Errorf("wal segment page read out of range: id=%d base=%d pageN=%d", walID, s.MinWALID, s.PageN)
}
offset := (walID - s.MinWALID) * PageSize
return s.data[offset : offset+PageSize], nil
}
func walSegmentByPath(segments []WALSegment, path string) *WALSegment {
for i := range segments {
if segments[i].Path == path {
return &segments[i]
}
}
return nil
}
func activeWALSegment(segments []WALSegment) WALSegment {
if len(segments) == 0 {
return WALSegment{}
}
return segments[len(segments)-1]
}
func minWALID(segments []WALSegment) int64 {
if len(segments) == 0 {
return 0
}
return segments[0].MinWALID
}
func maxWALID(segments []WALSegment) int64 {
if len(segments) == 0 {
return 0
}
s := segments[len(segments)-1]
return s.MaxWALID()
}
func walSize(segments []WALSegment) int64 {
var sz int64
for _, s := range segments {
sz += s.Size()
}
return sz
}
// readWALPage reads a single page at the given WAL ID.
func readWALPage(segments []WALSegment, walID int64) ([]byte, error) {
n := len(segments)
i := sort.Search(n, func(i int) bool {
return walID < segments[i].MinWALID
})
minWALID, maxWALID := int64(-1), int64(-1)
if i > 0 {
s := segments[i-1]
minWALID = s.MinWALID
maxWALID = s.MaxWALID()
if walID >= minWALID && walID <= maxWALID {
return s.ReadWALPage(walID)
}
}
// ok, we're about to error, which should never happen.
// So we can afford to provide detailed diagnostics.
// Report min and max WALID over all supplied segments.
for _, s := range segments {
if minWALID < 0 || s.MinWALID < minWALID {
minWALID = s.MinWALID
}
max := s.MaxWALID()
if maxWALID < 0 || max > maxWALID {
maxWALID = max
}
}
// show all the current segments too.
detail := WALSegmentsAsString(segments)
return nil, fmt.Errorf("cannot find segment containing WAL page: %d; over all supplied segments, minWALID=%v, maxWALID=%v; detail='%v'", walID, minWALID, maxWALID, detail)
}
var ErrNoMetaFound = fmt.Errorf("no meta page found")
func findNextWALMetaPage(segments []WALSegment, walID int64) (metaWALID int64, err error) {
maxWALID := maxWALID(segments)
for ; walID <= maxWALID; walID++ {
// Read page data from WAL and return if it is a meta page.
page, err := readWALPage(segments, walID)
if err != nil {
return walID, err
} else if IsMetaPage(page) {
return walID, nil
}
// Skip over next page if this is a bitmap header.
if IsBitmapHeader(page) {
walID++
}
}
return -1, ErrNoMetaFound
}
func findLastWALMetaPage(segments []WALSegment) (walID int64, err error) {
if len(segments) == 0 {
return 0, nil
}
var maxMetaWALID int64
maxWALID := maxWALID(segments)
for walID := minWALID(segments); walID <= maxWALID; walID++ {
if page, err := readWALPage(segments, walID); err != nil {
return walID, err
} else if IsBitmapHeader(page) {
walID++ // skip next page for bitmap headers
} else if IsMetaPage(page) {
maxMetaWALID = walID // save max meta WAL ID
}
}
return maxMetaWALID, nil
}
// truncateWALAfter removes all pages in the WAL after walID.
func (db *DB) truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) {
var newSegments []WALSegment
for i := range segments {
segment := &segments[i]
// Append entire segment if WAL range entirely before target WAL ID.
if walID > segment.MaxWALID() {
newSegments = append(newSegments, *segment)
continue
}
// If we only remove some of the WAL pages then truncate and append.
if segment.MinWALID < walID {
newSegment := *segment
newSegment.PageN = int((walID - newSegment.MinWALID) + 1)
if err := db.truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
return segments, err
}
newSegments = append(newSegments, newSegment)
continue
}
// Drop entire segment if all pages are after WAL ID.
if err := segment.Close(); err != nil {
return segments, err
} else if err := os.Remove(segment.Path); err != nil {
return segments, err
}
}
return newSegments, nil
}
func DumpWALSegments(segments []WALSegment) {
fmt.Printf("WAL (%d segments)\n", len(segments))
for i, s := range segments {
fmt.Printf("[%d] WALIDs=(%d-%d) PageN=%d\n", i, s.MinWALID, s.MaxWALID(), s.PageN)
}
}
func WALSegmentsAsString(segments []WALSegment) (r string) {
r = fmt.Sprintf("WAL (%d segments)\n", len(segments))
for i, s := range segments {
r += fmt.Sprintf("[%d] WALIDs=(%d-%d) PageN=%d\n", i, s.MinWALID, s.MaxWALID(), s.PageN)
}
return
}
// FormatWALSegmentPath returns a path for a WAL segment using a WAL ID.
func FormatWALSegmentPath(walID int64) string {
return fmt.Sprintf("%016x.wal", walID)
}
// ParseWALSegmentPath returns the WAL ID for a given WAL segment path.
func ParseWALSegmentPath(s string) (walID int64, err error) {
if _, err = fmt.Sscanf(filepath.Base(s), "%016x.wal", &walID); err != nil {
return 0, fmt.Errorf("invalid WAL path: %s", s)
}
return walID, nil
}
// uint32Hasher implements Hasher for uint32 keys.
type uint32Hasher struct{}
// Hash returns a hash for key.
func (h *uint32Hasher) Hash(key interface{}) uint32 {
return hashUint64(uint64(key.(uint32)))
}
// Equal returns true if a is equal to b. Otherwise returns false.
// Panics if a and b are not ints.
func (h *uint32Hasher) Equal(a, b interface{}) bool {
return a.(uint32) == b.(uint32)
}
// hashUint64 returns a 32-bit hash for a 64-bit value.
func hashUint64(value uint64) uint32 {
hash := value
for value > 0xffffffff {
value /= 0xffffffff
hash ^= value
}
return uint32(hash)
}

186
rbf/wal_test.go
View file

@ -1,186 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package rbf_test
import (
// "bytes"
// "encoding/hex"
"io/ioutil"
// "math/rand"
"os"
"path/filepath"
"testing"
"github.com/pilosa/pilosa/v2/rbf"
)
func TestWALSegment_Open(t *testing.T) {
t.Run("OK", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
s := MustOpenWALSegment(t, db, 10)
defer MustCloseWALSegment(t, s)
if got, want := s.MinWALID, int64(10); got != want {
t.Fatalf("Base()=%d, want %d", got, want)
} else if got, want := s.PageN, 0; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
})
// TODO(BBJ): Test open w/ partially written pages.
}
/*
func TestWALSegment_WritePage(t *testing.T) {
rand := rand.New(rand.NewSource(0))
s := MustOpenWALSegment(t, 10)
defer MustCloseWALSegment(t, s)
pages := [][]byte{
make([]byte, rbf.PageSize),
make([]byte, rbf.PageSize),
}
rand.Read(pages[0])
rand.Read(pages[1])
// Write first page.
if walID, err := s.WriteWALPage(pages[0], false); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(10); got != want {
t.Fatalf("WALID=%d, want %d", got, want)
} else if got, want := s.PageN(), 1; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
// Write second page.
if walID, err := s.WriteWALPage(pages[1], false); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(11); got != want {
t.Fatalf("WALID=%d, want %d", got, want)
} else if got, want := s.PageN(), 2; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
// Read & verify first page.
if buf, err := s.ReadWALPage(10); err != nil {
t.Fatal(err)
} else if !bytes.Equal(pages[0], buf) {
t.Fatalf("unexpected first page:\n%s", hex.Dump(buf))
}
// Read & verify second page.
if buf, err := s.ReadWALPage(11); err != nil {
t.Fatal(err)
} else if !bytes.Equal(pages[1], buf) {
t.Fatal("unexpected second page")
}
}
*/
func TestFormatWALSegmentPath(t *testing.T) {
if got, want := rbf.FormatWALSegmentPath(1234), "00000000000004d2.wal"; got != want {
t.Fatalf("FormatWALSegmentPath()=%q, want %q", got, want)
}
}
func TestParseWALSegmentPath(t *testing.T) {
t.Run("OK", func(t *testing.T) {
if walID, err := rbf.ParseWALSegmentPath("/tmp/00000000000004d2.wal"); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(1234); got != want {
t.Fatalf("ParseWALSegmentPath()=%q, want %q", got, want)
}
})
t.Run("ErrInvalidWALPath", func(t *testing.T) {
if _, err := rbf.ParseWALSegmentPath("/tmp/xyz"); err == nil || err.Error() != "invalid WAL path: /tmp/xyz" {
t.Fatalf("unexpected error: %#v", err)
}
})
}
/*
func BenchmarkWALSegment_WriteWALPage(b *testing.B) {
b.Run("8KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 8*(1<<10)) })
b.Run("16KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 16*(1<<10)) })
b.Run("64KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 64*(1<<10)) })
b.Run("256KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 256*(1<<10)) })
b.Run("1MB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, (1 << 20)) })
b.Run("10MB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 10*(1<<20)) })
}
func benchmarkWALSegment_WriteWALPage(b *testing.B, flushSize int) {
page := make([]byte, rbf.PageSize)
for i := 0; i < b.N; i++ {
func() {
s := MustOpenWALSegment(b, 0)
defer MustCloseWALSegment(b, s)
// Fill the segment but stop after each flush interval to flush the write buffer.
for j := 0; j < rbf.MaxWALSegmentFileSize; j += rbf.PageSize {
if _, err := s.WriteWALPage(page, false); err != nil {
b.Fatal(err)
}
// Flush write buffer.
if j != 0 && j%flushSize == 0 {
if err := s.Flush(); err != nil {
b.Fatal(err)
}
}
}
// Fsync to disk at the end.
if err := s.Sync(); err != nil {
b.Fatal(err)
}
}()
}
b.SetBytes(rbf.MaxWALSegmentFileSize)
}
*/
// MustOpenWALSegment opens a WAL segment in a temporary path. Fails on error.
func MustOpenWALSegment(tb testing.TB, db *rbf.DB, walID int64) rbf.WALSegment {
tb.Helper()
dir, err := ioutil.TempDir("", "")
if err != nil {
tb.Fatal(err)
}
path := filepath.Join(dir, rbf.FormatWALSegmentPath(walID))
if err := ioutil.WriteFile(path, nil, 0666); err != nil {
tb.Fatal(err)
}
s := db.NewWALSegment(path)
if err := s.Open(); err != nil {
tb.Fatal(err)
}
return s
}
// MustCloseWALSegment closes s. Fails on error.
func MustCloseWALSegment(tb testing.TB, s rbf.WALSegment) {
tb.Helper()
if err := s.Close(); err != nil {
tb.Fatal(err)
} else if err := os.Remove(s.Path); err != nil {
tb.Fatal(err)
}
}