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

Refactor RBF WAL to only only checkpoint-in-full.

Browse source
This commit is contained in:
Ben Johnson 2020-11-01 10:03:31 -07:00
parent ea3732fa62
commit 731a1ef25e
3 changed files with 39 additions and 165 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
go.sum
View file

@ -161,7 +161,6 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU=
github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
github.com/pilosa/pilosa v1.4.1 h1:zSNyS/MqXTfRDNRBBNCApsOu2oUnTEj2uymCaqYSNx8=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

158
rbf/db.go
View file

@ -146,7 +146,7 @@ func (db *DB) Open() (err error) {
// Open write-ahead log & checkpoint to the end since no transactions are open.
if err := db.openWALSegments(); err != nil {
return fmt.Errorf("wal open: %w", err)
} else if err := db.checkpoint(true, &nopLocker{}); err != nil {
} else if err := db.checkpoint(true); err != nil {
return fmt.Errorf("checkpoint: %w", err)
}
@ -204,32 +204,13 @@ func (db *DB) updateWALSegment(s WALSegment) {
db.segments = segments
}
// Checkpoint copies pages from WAL segments into the main DB file. This can
// only copy pages that aren't in use by an active transaction. The page map
// is rebuilt as well for all WAL pages still in use.
//
// If exclusive is true, all WAL writes are flushed to disk.
func (db *DB) Checkpoint() error {
return db.checkpoint(false, &db.mu)
}
// checkpoint moves WAL segments to the main DB file.
//
// Note that mu should db.mu when called through DB.Checkpoint() but it
// can be &nopLocker if called under lock. The external API will be used
// to periodically checkpoint outside of a transaction and the locking
// must be used only in the beginning (to obtain the segment list) and at
// the end (when removing old segments from the list). If the entire function
// were to obtain a lock then it would block all new read & write transactions.
func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
// Obtain a snapshot of WAL segments at the start.
mu.Lock()
opened := db.opened
segments := db.segments
mu.Unlock()
if !opened {
// Must be called by a write transaction while under db.mu lock.
func (db *DB) checkpoint(exclusive bool) error {
if !db.opened {
return nil
} else if len(db.txs) > 0 {
return nil // skip if transactions open
}
// Determine last checkpointed WAL ID.
@ -239,15 +220,11 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
}
walID := readMetaWALID(page)
// Determine the high water mark for WAL pages that can be copied.
writerWALID := db.writerWALID()
// Loop over each transaction
walID++
pageMap := immutable.NewMap(&uint32Hasher{})
for {
// Determine last page of transaction.
metaWALID, err := findNextWALMetaPage(segments, walID)
metaWALID, err := findNextWALMetaPage(db.segments, walID)
if err == io.EOF {
break
} else if err != nil {
@ -256,9 +233,7 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
// Loop over pages in the transaction.
for ; walID <= metaWALID; walID++ {
canCheckpoint := exclusive || writerWALID == 0 || walID < writerWALID
page, err := readWALPage(segments, walID)
page, err := readWALPage(db.segments, walID)
if err != nil {
return err
}
@ -274,16 +249,10 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
pgno = readPageNo(page)
}
// If we can no longer checkpoint, map the page number to the WAL page.
if !canCheckpoint {
pageMap = pageMap.Set(pgno, walID)
continue
}
// 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(segments, walID); err != nil {
if page, err = readWALPage(db.segments, walID); err != nil {
return err
}
}
@ -299,97 +268,21 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
if err := db.fsync(db.file); err != nil {
return fmt.Errorf("db file sync: %w", err)
}
mu.Lock()
db.pageMap = pageMap
mu.Unlock()
db.pageMap = immutable.NewMap(&uint32Hasher{})
// Remove WAL segments that have been checkpointed.
minPageMapWALID := db.minPageMapWALID()
for _, segment := range segments {
if minPageMapWALID != 0 && segment.MaxWALID() > minPageMapWALID {
break
}
if err := func() error {
mu.Lock()
defer mu.Unlock()
return db.removeWALSegment(segment.Path)
}(); err != nil {
for _, segment := range db.segments {
if err := segment.Close(); err != nil {
return err
} else if err := os.Remove(segment.Path); err != nil {
return err
}
}
// Ensure all segments are flushed and there is no remapped pages.
if exclusive {
mu.Lock()
assert(len(db.segments) == 0)
assert(pageMap.Len() == 0)
mu.Unlock()
}
db.segments = nil
return nil
}
// removeWALSegment closes and deletes the segment with the given path.
//
// The DB's segment list is entirely replaced so that transactions with
// a reference to the old list can continue to use it without a lock.
func (db *DB) removeWALSegment(path string) error {
newSegments := make([]WALSegment, 0, len(db.segments))
for _, segment := range db.segments {
// Close and remove if path matches.
if segment.Path == path {
if err := segment.Close(); err != nil {
return err
} else if err := os.Remove(segment.Path); err != nil {
return err
}
continue
}
// Otherwise append to new slice of segments.
newSegments = append(newSegments, segment)
}
// Replace entire slice of segments.
db.segments = newSegments
return nil
}
// minPageMapWALID returns the lowest WAL ID referenced by an active page map.
func (db *DB) minPageMapWALID() int64 {
// Use the db's page map because that is the state of the map when the
// writer transaction started. We can't use the writer transaction's map
// because it can change.
min := pageMapMinWALID(db.pageMap)
for tx := range db.txs {
// If a write transaction is active, ensure the min is at least the starting WAL.
if tx.writable {
if min == 0 || tx.walID < min {
min = tx.walID
}
continue
}
// Record the min WAL ID referenced by the reader's page map.
if walID := pageMapMinWALID(tx.pageMap); min == 0 || walID < min {
min = walID
}
}
return min
}
// writerWALID returns the starting WAL ID of the active writer tx.
func (db *DB) writerWALID() int64 {
for tx := range db.txs {
if tx.writable {
return tx.walID
}
}
return 0
}
// Close closes the database.
func (db *DB) Close() (err error) {
// TODO(bbj): Add wait group to hang until last Tx is complete.
@ -632,7 +525,7 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
// Flush all WAL writes to disk before an exclusive writer so that we can
// work directly with the on-disk database.
if exclusive {
if err := db.checkpoint(true, &nopLocker{}); err != nil {
if err := db.checkpoint(true); err != nil {
cleanup()
db.mu.Unlock()
return nil, err
@ -709,7 +602,7 @@ func (db *DB) removeTx(tx *Tx) error {
// checkpoint eagerly.
if tx.writable {
if db.cfg.CheckpointEveryDur == 0 || time.Since(db.lastCheckpoint) > db.cfg.CheckpointEveryDur {
if err := db.checkpoint(false, &nopLocker{}); err != nil {
if err := db.checkpoint(false); err != nil {
return fmt.Errorf("checkpoint: %w", err)
}
db.lastCheckpoint = time.Now()
@ -750,20 +643,3 @@ func (db *DB) readMetaPage() ([]byte, error) {
}
return db.readDBPage(0)
}
type nopLocker struct{}
func (*nopLocker) Lock() {}
func (*nopLocker) Unlock() {}
// pageMapMinWALID returns the lowest WAL ID
func pageMapMinWALID(m *immutable.Map) int64 {
var min int64
for itr := m.Iterator(); !itr.Done(); {
_, v := itr.Next()
if walID := v.(int64); min == 0 || walID < min {
min = walID
}
}
return min
}

45
rbf/db_test.go
View file

@ -307,14 +307,32 @@ func TestDB_MultiTx(t *testing.T) {
// Run multiple readers in separate goroutines.
ctx, cancel := context.WithCancel(context.Background())
g, ctx := errgroup.WithContext(ctx)
for i := 0; i < 10; i++ {
for i := 0; i < 4; i++ {
g.Go(func() error {
for {
if ctx.Err() != nil {
return nil // cancelled, return no error
} else if err := testDB_MultiTx_reader(db); err != nil {
} else if err := func() error {
tx, err := db.Begin(false)
if err != nil {
return err
}
defer tx.Rollback()
time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
for i := 0; i < rand.Intn(1000); i++ {
v := rand.Intn(1 << 20)
if _, err := tx.Contains("x", uint64(v)); err != nil {
return err
}
}
return nil
}(); err != nil {
return err
}
time.Sleep(time.Duration(rand.Intn(int(100 * time.Millisecond))))
}
})
}
@ -328,11 +346,12 @@ func TestDB_MultiTx(t *testing.T) {
}
defer tx.Rollback()
for j := 0; j < rand.Intn(10); j++ {
for j := 0; j < rand.Intn(100); j++ {
v := rand.Intn(1 << 20)
if _, err := tx.Add("x", uint64(v)); err != nil {
t.Fatal(err)
}
}
if err := tx.Commit(); err != nil {
@ -348,26 +367,6 @@ func TestDB_MultiTx(t *testing.T) {
}
}
// testDB_MultiTx_reader checks if a bitmap contains a random set of bits.
func testDB_MultiTx_reader(db *rbf.DB) error {
tx, err := db.Begin(false)
if err != nil {
return err
}
defer tx.Rollback()
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
for i := 0; i < rand.Intn(1000); i++ {
v := rand.Intn(1 << 20)
if _, err := tx.Contains("x", uint64(v)); err != nil {
return err
}
}
return nil
}
// better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests.
func TestMain(m *testing.M) {
port := getAvailPort()